text:
/**
 *    Copyright (C) 2014 MongoDB Inc.
 *
 *    This program is free software: you can redistribute it and/or  modify
 *    it under the terms of the GNU Affero General Public License, version 3,
 *    as published by the Free Software Foundation.
 *
 *
 *    This program is distributed in the hope that it will be useful,
 *    but WITHOUT ANY WARRANTY; without even the implied warranty of
 *    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 *    GNU Affero General Public License for more details.
 *
 *    You should have received a copy of the GNU Affero General Public License
 *    along with this program.  If not, see <http://www.gnu.org/licenses/>.
 *
 *    As a special exception, the copyright holders give permission to link the
 *    code of portions of this program with the OpenSSL library under certain
 *    conditions as described in each individual source file and distribute
 *    linked combinations including the program with the OpenSSL library. You
 *    must comply with the GNU Affero General Public License in all respects for
 *    all of the code used other than as permitted herein. If you modify file(s)
 *    with this exception, you may extend this exception to your version of the
 *    file(s), but you are not obligated to do so. If you do not wish to do so,
 *    delete this exception statement from your version. If you delete this
 *    exception statement from all source files in the program, then also delete
 *    it in the license file.
 */

#define MONGO_LOG_DEFAULT_COMPONENT ::mongo::logger::LogComponent::kStorage

#include "mongo/platform/basic.h"

#include "rocks_record_store.h"

#include <boost/scoped_array.hpp>
#include <boost/shared_ptr.hpp>
#include <memory>
#include <algorithm>

#include <rocksdb/comparator.h>
#include <rocksdb/db.h>
#include <rocksdb/experimental.h>
#include <rocksdb/options.h>
#include <rocksdb/slice.h>
#include <rocksdb/utilities/write_batch_with_index.h>

#include "mongo/base/checked_cast.h"
#include "mongo/bson/bsonobjbuilder.h"
#include "mongo/db/concurrency/write_conflict_exception.h"
#include "mongo/db/namespace_string.h"
#include "mongo/db/operation_context.h"
#include "mongo/db/storage/oplog_hack.h"
#include "mongo/platform/endian.h"
#include "mongo/util/background.h"
#include "mongo/util/log.h"
#include "mongo/util/mongoutils/str.h"

#include "rocks_counter_manager.h"
#include "rocks_engine.h"
#include "rocks_recovery_unit.h"
#include "rocks_util.h"

namespace mongo {

using boost::shared_ptr;
using std::string;

namespace {

class CappedInsertChange : public RecoveryUnit::Change {
public:
    CappedInsertChange(CappedVisibilityManager* cappedVisibilityManager, const RecordId& record)
        : _cappedVisibilityManager(cappedVisibilityManager), _record(record) {}

    virtual void commit() {
        _cappedVisibilityManager->dealtWithCappedRecord(_record);
    }

    virtual void rollback() {
        _cappedVisibilityManager->dealtWithCappedRecord(_record);
    }

private:
    CappedVisibilityManager* _cappedVisibilityManager;
    RecordId _record;
};
}  // namespace

void CappedVisibilityManager::addUncommittedRecord(OperationContext* txn, const RecordId& record) {
    boost::mutex::scoped_lock lk(_lock);
    _addUncommittedRecord_inlock(txn, record);
}

void CappedVisibilityManager::_addUncommittedRecord_inlock(OperationContext* txn,
                                                           const RecordId& record) {
    dassert(_uncommittedRecords.empty() || _uncommittedRecords.back() < record);
    _uncommittedRecords.push_back(record);
    txn->recoveryUnit()->registerChange(new CappedInsertChange(this, record));
    _oplog_highestSeen = record;
}

RecordId CappedVisibilityManager::getNextAndAddUncommittedRecord(OperationContext* txn,
                                                                 std::function<RecordId()> nextId) {
    boost::mutex::scoped_lock lk(_lock);
    RecordId record = nextId();
    _addUncommittedRecord_inlock(txn, record);
    return record;
}

void CappedVisibilityManager::dealtWithCappedRecord(const RecordId& record) {
    boost::mutex::scoped_lock lk(_lock);
    std::vector<RecordId>::iterator it =
        std::find(_uncommittedRecords.begin(), _uncommittedRecords.end(), record);
    invariant(it != _uncommittedRecords.end());
    _uncommittedRecords.erase(it);
}

bool CappedVisibilityManager::isCappedHidden(const RecordId& record) const {
    boost::mutex::scoped_lock lk(_lock);
    if (_uncommittedRecords.empty()) {
        return false;
    }
    return _uncommittedRecords.front() <= record;
}

void CappedVisibilityManager::updateHighestSeen(const RecordId& record) {
    if (record > _oplog_highestSeen) {
        boost::mutex::scoped_lock lk(_lock);
        if (record > _oplog_highestSeen) {
            _oplog_highestSeen = record;
        }
    }
}

RecordId CappedVisibilityManager::oplogStartHack() const {
    boost::mutex::scoped_lock lk(_lock);
    if (_uncommittedRecords.empty()) {
        return _oplog_highestSeen;
    } else {
        return _uncommittedRecords.front();
    }
}

// this object keeps track of keys in oplog. The format is this:
// <prefix>RecordId --> dataSize (small endian 32 bytes)
// <prefix> is oplog_prefix+1 (reserved by rocks_engine.cpp)
// That way we can cheaply delete old record in the oplog without actually reading oplog
// collection.
// All of the locking is done somewhere else -- we write exactly the same data as oplog, so we
// assume oplog already locked the relevant keys
class RocksOplogKeyTracker {
public:
    RocksOplogKeyTracker(std::string prefix) : _prefix(std::move(prefix)) {}
    void insertKey(RocksRecoveryUnit* ru, const RecordId& loc, int len) {
        uint32_t lenLittleEndian = endian::nativeToLittle(static_cast<uint32_t>(len));
        ru->writeBatch()->Put(RocksRecordStore::_makePrefixedKey(_prefix, loc),
                              rocksdb::Slice(reinterpret_cast<const char*>(&lenLittleEndian),
                                             sizeof(lenLittleEndian)));
    }
    void deleteKey(RocksRecoveryUnit* ru, const RecordId& loc) {
        ru->writeBatch()->Delete(RocksRecordStore::_makePrefixedKey(_prefix, loc));
    }
    rocksdb::Iterator* newIterator(RocksRecoveryUnit* ru) {
        return ru->NewIterator(_prefix, true);
    }
    int decodeSize(const rocksdb::Slice& value) {
        uint32_t size = endian::littleToNative(*reinterpret_cast<const uint32_t*>(value.data()));
        return static_cast<int>(size);
    }

private:
    std::string _prefix;
};

RocksRecordStore::RocksRecordStore(const StringData& ns,
                                   const StringData& id,
                                   rocksdb::DB* db,
                                   RocksCounterManager* counterManager,
                                   std::string prefix,
                                   bool isCapped,
                                   int64_t cappedMaxSize,
                                   int64_t cappedMaxDocs,
                                   CappedDocumentDeleteCallback* cappedDeleteCallback)
    : RecordStore(ns),
      _db(db),
      _counterManager(counterManager),
      _prefix(std::move(prefix)),
      _isCapped(isCapped),
      _cappedMaxSize(cappedMaxSize),
      _cappedMaxSizeSlack(std::min(cappedMaxSize / 10, int64_t(16 * 1024 * 1024))),
      _cappedMaxDocs(cappedMaxDocs),
      _cappedDeleteCallback(cappedDeleteCallback),
      _cappedDeleteCheckCount(0),
      _isOplog(NamespaceString::oplog(ns)),
      _oplogKeyTracker(_isOplog ? new RocksOplogKeyTracker(std::move(rocksGetNextPrefix(_prefix)))
                                : nullptr),
      _cappedOldestKeyHint(0),
      _cappedVisibilityManager((_isCapped || _isOplog) ? new CappedVisibilityManager() : nullptr),
      _ident(id.toString()),
      _dataSizeKey(std::string("\0\0\0\0", 4) + "datasize-" + id.toString()),
      _numRecordsKey(std::string("\0\0\0\0", 4) + "numrecords-" + id.toString()),
      _shuttingDown(false) {
    _oplogSinceLastCompaction.reset();

    LOG(1) << "opening collection " << ns << " with prefix "
           << rocksdb::Slice(_prefix).ToString(true);

    if (_isCapped) {
        invariant(_cappedMaxSize > 0);
        invariant(_cappedMaxDocs == -1 || _cappedMaxDocs > 0);
    } else {
        invariant(_cappedMaxSize == -1);
        invariant(_cappedMaxDocs == -1);
    }

    // Get next id
    boost::scoped_ptr<rocksdb::Iterator> iter(
        RocksRecoveryUnit::NewIteratorNoSnapshot(_db, _prefix));
    iter->SeekToLast();
    bool emptyCollection = !iter->Valid();
    if (iter->Valid()) {
        rocksdb::Slice lastSlice = iter->key();
        RecordId lastId = _makeRecordId(lastSlice);
        if (_isOplog || _isCapped) {
            _cappedVisibilityManager->updateHighestSeen(lastId);
        }
        _nextIdNum.store(lastId.repr() + 1);
    } else {
        // Need to start at 1 so we are always higher than RecordId::min()
        _nextIdNum.store(1);
    }

    // load metadata
    _numRecords.store(_counterManager->loadCounter(_numRecordsKey));
    _dataSize.store(_counterManager->loadCounter(_dataSizeKey));
    invariant(_dataSize.load() >= 0);
    invariant(_numRecords.load() >= 0);

    if (!emptyCollection && !_counterManager->crashSafe() &&
        _numRecords.load() < kCollectionScanOnCreationThreshold) {
        LOG(1) << "doing scan of collection " << ns << " to get info";

        _numRecords.store(0);
        _dataSize.store(0);

        long long numRecords = 0, dataSize = 0;
        for (iter->SeekToFirst(); iter->Valid(); iter->Next()) {
            numRecords++;
            dataSize += static_cast<long long>(iter->value().size());
        }

        _numRecords.store(numRecords);
        _dataSize.store(dataSize);

        rocksdb::WriteBatch wb;
        _counterManager->updateCounter(_numRecordsKey, numRecords, &wb);
        _counterManager->updateCounter(_dataSizeKey, dataSize, &wb);
        auto s = _db->Write(rocksdb::WriteOptions(), &wb);
        invariantRocksOK(s);
    }

    _hasBackgroundThread = RocksEngine::initRsOplogBackgroundThread(ns);
}

RocksRecordStore::~RocksRecordStore() {
    {
        boost::timed_mutex::scoped_lock lk(_cappedDeleterMutex);
        _shuttingDown = true;
    }
    delete _oplogKeyTracker;
}

int64_t RocksRecordStore::storageSize(OperationContext* txn,
                                      BSONObjBuilder* extraInfo,
                                      int infoLevel) const {
    // We need to make it multiple of 256 to make
    // jstests/concurrency/fsm_workloads/convert_to_capped_collection.js happy
    return static_cast<int64_t>(std::max(_dataSize.load() & (~255), static_cast<long long>(256)));
}

RecordData RocksRecordStore::dataFor(OperationContext* txn, const RecordId& loc) const {
    RecordData rd = _getDataFor(_db, _prefix, txn, loc);
    massert(28605, "Didn't find RecordId in RocksRecordStore", (rd.data() != nullptr));
    return rd;
}

void RocksRecordStore::deleteRecord(OperationContext* txn, const RecordId& dl) {
    std::string key(_makePrefixedKey(_prefix, dl));

    RocksRecoveryUnit* ru = RocksRecoveryUnit::getRocksRecoveryUnit(txn);
    if (!ru->transaction()->registerWrite(key)) {
        throw WriteConflictException();
    }

    std::string oldValue;
    auto status = ru->Get(key, &oldValue);
    invariantRocksOK(status);
    int oldLength = oldValue.size();

    ru->writeBatch()->Delete(key);
    if (_isOplog) {
        _oplogKeyTracker->deleteKey(ru, dl);
    }

    _changeNumRecords(txn, -1);
    _increaseDataSize(txn, -oldLength);
}

long long RocksRecordStore::dataSize(OperationContext* txn) const {
    RocksRecoveryUnit* ru = RocksRecoveryUnit::getRocksRecoveryUnit(txn);
    return _dataSize.load(std::memory_order::memory_order_relaxed) +
        ru->getDeltaCounter(_dataSizeKey);
}

long long RocksRecordStore::numRecords(OperationContext* txn) const {
    RocksRecoveryUnit* ru = RocksRecoveryUnit::getRocksRecoveryUnit(txn);
    return _numRecords.load(std::memory_order::memory_order_relaxed) +
        ru->getDeltaCounter(_numRecordsKey);
}

bool RocksRecordStore::cappedAndNeedDelete(long long dataSizeDelta,
                                           long long numRecordsDelta) const {
    invariant(_isCapped);

    if (_dataSize.load() + dataSizeDelta > _cappedMaxSize)
        return true;

    if ((_cappedMaxDocs != -1) && (_numRecords.load() + numRecordsDelta > _cappedMaxDocs))
        return true;

    return false;
}

int64_t RocksRecordStore::cappedDeleteAsNeeded(OperationContext* txn,
                                               const RecordId& justInserted) {
    if (!_isCapped) {
        return 0;
    }

    // We only want to do the checks occasionally as they are expensive.
    // This variable isn't thread safe, but has loose semantics anyway.
    dassert(!_isOplog || _cappedMaxDocs == -1);

    long long dataSizeDelta = 0, numRecordsDelta = 0;
    if (!_isOplog) {
        auto ru = RocksRecoveryUnit::getRocksRecoveryUnit(txn);
        dataSizeDelta = ru->getDeltaCounter(_dataSizeKey);
        numRecordsDelta = ru->getDeltaCounter(_numRecordsKey);
    }

    if (!cappedAndNeedDelete(dataSizeDelta, numRecordsDelta)) {
        return 0;
    }

    // ensure only one thread at a time can do deletes, otherwise they'll conflict.
    boost::timed_mutex::scoped_lock lock(_cappedDeleterMutex, boost::defer_lock);

    if (_cappedMaxDocs != -1) {
        lock.lock();  // Max docs has to be exact, so have to check every time.
    } else if (_hasBackgroundThread) {
        // We are foreground, and there is a background thread,

        // Check if we need some back pressure.
        if ((_dataSize.load() - _cappedMaxSize) < _cappedMaxSizeSlack) {
            return 0;
        }

        // Back pressure needed!
        // We're not actually going to delete anything, but we're going to syncronize
        // on the deleter thread.
        (void)lock.timed_lock(boost::posix_time::millisec(200));
        return 0;
    } else {
        if (!lock.try_lock()) {
            // Someone else is deleting old records. Apply back-pressure if too far behind,
            // otherwise continue.
            if ((_dataSize.load() - _cappedMaxSize) < _cappedMaxSizeSlack)
                return 0;

            if (!lock.timed_lock(boost::posix_time::millisec(200)))
                return 0;

            // If we already waited, let someone else do cleanup unless we are significantly
            // over the limit.
            if ((_dataSize.load() - _cappedMaxSize) < (2 * _cappedMaxSizeSlack))
                return 0;
        }
    }

    return cappedDeleteAsNeeded_inlock(txn, justInserted);
}

int64_t RocksRecordStore::cappedDeleteAsNeeded_inlock(OperationContext* txn,
                                                      const RecordId& justInserted) {
    // we do this is a sub transaction in case it aborts
    RocksRecoveryUnit* realRecoveryUnit =
        checked_cast<RocksRecoveryUnit*>(txn->releaseRecoveryUnit());
    invariant(realRecoveryUnit);
    txn->setRecoveryUnit(realRecoveryUnit->newRocksRecoveryUnit());

    int64_t dataSize = _dataSize.load() + realRecoveryUnit->getDeltaCounter(_dataSizeKey);
    int64_t numRecords = _numRecords.load() + realRecoveryUnit->getDeltaCounter(_numRecordsKey);

    int64_t sizeOverCap = (dataSize > _cappedMaxSize) ? dataSize - _cappedMaxSize : 0;
    int64_t sizeSaved = 0;
    int64_t docsOverCap = 0, docsRemoved = 0;
    if (_cappedMaxDocs != -1 && numRecords > _cappedMaxDocs) {
        docsOverCap = numRecords - _cappedMaxDocs;
    }
    BSONObj emptyBson;

    try {
        WriteUnitOfWork wuow(txn);
        auto ru = RocksRecoveryUnit::getRocksRecoveryUnit(txn);
        boost::scoped_ptr<rocksdb::Iterator> iter;
        if (_isOplog) {
            // we're using _oplogKeyTracker to find which keys to delete -- this is much faster
            // because we don't need to read any values. We theoretically need values to pass
            // the document to the cappedDeleteCallback, but the callback is only using
            // documents to remove them from indexes. opLog doesn't have indexes, so there
            // should be no need for us to reconstruct the document to pass it to the callback
            iter.reset(_oplogKeyTracker->newIterator(ru));
        } else {
            iter.reset(ru->NewIterator(_prefix));
        }
        int64_t storage;
        iter->Seek(RocksRecordStore::_makeKey(_cappedOldestKeyHint, &storage));

        RecordId newestOld;
        while ((sizeSaved < sizeOverCap || docsRemoved < docsOverCap) && (docsRemoved < 20000) &&
               iter->Valid()) {
            newestOld = _makeRecordId(iter->key());

            if (_cappedVisibilityManager->isCappedHidden(newestOld)) {
                // this means we have an older record that hasn't been committed yet. let's
                // wait until it gets committed before deleting
                break;
            }

            // don't go past the record we just inserted
            if (newestOld >= justInserted) {
                break;
            }

            if (_shuttingDown) {
                break;
            }

            std::string key(_makePrefixedKey(_prefix, newestOld));
            if (!ru->transaction()->registerWrite(key)) {
                log() << "got conflict truncating capped, total docs removed " << docsRemoved;
                break;
            }

            rocksdb::Slice oldValue;
            ++docsRemoved;
            if (_isOplog) {
                // trick the callback by giving it empty bson document
                oldValue = rocksdb::Slice(emptyBson.objdata(), emptyBson.objsize());
                // we keep data size in the value
                sizeSaved += _oplogKeyTracker->decodeSize(iter->value());
            } else {
                oldValue = iter->value();
                sizeSaved += oldValue.size();
            }

            if (_cappedDeleteCallback) {
                uassertStatusOK(_cappedDeleteCallback->aboutToDeleteCapped(
                    txn,
                    newestOld,
                    RecordData(static_cast<const char*>(oldValue.data()), oldValue.size())));
            }

            ru->writeBatch()->Delete(key);
            if (_isOplog) {
                _oplogKeyTracker->deleteKey(ru, newestOld);
            }

            iter->Next();
        }

        if (!iter->status().ok()) {
            log() << "RocksDB iterator failure when trying to delete capped, ignoring: "
                  << iter->status().ToString();
        }

        if (docsRemoved > 0) {
            _changeNumRecords(txn, -docsRemoved);
            _increaseDataSize(txn, -sizeSaved);
            wuow.commit();
        }

        if (iter->Valid()) {
            auto oldestAliveRecordId = _makeRecordId(iter->key());
            // we check if there's outstanding transaction that is older than
            // oldestAliveRecordId. If there is, we should not skip deleting that record next
            // time we clean up the capped collection. If there isn't, we know for certain this
            // is the record we'll start out deletions from next time
            if (!_cappedVisibilityManager->isCappedHidden(oldestAliveRecordId)) {
                _cappedOldestKeyHint = oldestAliveRecordId;
            }
        }
    } catch (const WriteConflictException& wce) {
        delete txn->releaseRecoveryUnit();
        txn->setRecoveryUnit(realRecoveryUnit);
        log() << "got conflict truncating capped, ignoring";
        return 0;
    } catch (...) {
        delete txn->releaseRecoveryUnit();
        txn->setRecoveryUnit(realRecoveryUnit);
        throw;
    }

    delete txn->releaseRecoveryUnit();
    txn->setRecoveryUnit(realRecoveryUnit);

    if (_isOplog) {
        if (_oplogSinceLastCompaction.minutes() >= kOplogCompactEveryMins) {
            log() << "Scheduling oplog compactions";
            _oplogSinceLastCompaction.reset();
            // schedule compaction for oplog
            std::string oldestAliveKey(_makePrefixedKey(_prefix, _cappedOldestKeyHint));
            rocksdb::Slice begin(_prefix), end(oldestAliveKey);
            rocksdb::experimental::SuggestCompactRange(_db, &begin, &end);

            // schedule compaction for oplog tracker
            std::string oplogKeyTrackerPrefix(rocksGetNextPrefix(_prefix));
            oldestAliveKey = _makePrefixedKey(oplogKeyTrackerPrefix, _cappedOldestKeyHint);
            begin = rocksdb::Slice(oplogKeyTrackerPrefix);
            end = rocksdb::Slice(oldestAliveKey);
            rocksdb::experimental::SuggestCompactRange(_db, &begin, &end);
        }
    }

    return docsRemoved;
}

StatusWith<RecordId> RocksRecordStore::insertRecord(OperationContext* txn,
                                                    const char* data,
                                                    int len,
                                                    bool enforceQuota) {
    if (_isCapped && len > _cappedMaxSize) {
        return StatusWith<RecordId>(ErrorCodes::BadValue, "object to insert exceeds cappedMaxSize");
    }

    RocksRecoveryUnit* ru = RocksRecoveryUnit::getRocksRecoveryUnit(txn);

    RecordId loc;
    if (_isOplog) {
        StatusWith<RecordId> status = oploghack::extractKey(data, len);
        if (!status.isOK()) {
            return status;
        }
        loc = status.getValue();
        _cappedVisibilityManager->updateHighestSeen(loc);
    } else if (_isCapped) {
        loc = _cappedVisibilityManager->getNextAndAddUncommittedRecord(txn,
                                                                       [&]() { return _nextId(); });
    } else {
        loc = _nextId();
    }

    // No need to register the write here, since we just allocated a new RecordId so no other
    // transaction can access this key before we commit
    ru->writeBatch()->Put(_makePrefixedKey(_prefix, loc), rocksdb::Slice(data, len));
    if (_isOplog) {
        _oplogKeyTracker->insertKey(ru, loc, len);
    }

    _changeNumRecords(txn, 1);
    _increaseDataSize(txn, len);

    cappedDeleteAsNeeded(txn, loc);

    return StatusWith<RecordId>(loc);
}

StatusWith<RecordId> RocksRecordStore::insertRecord(OperationContext* txn,
                                                    const DocWriter* doc,
                                                    bool enforceQuota) {
    const int len = doc->documentSize();
    boost::scoped_array<char> buf(new char[len]);
    doc->writeDocument(buf.get());

    return insertRecord(txn, buf.get(), len, enforceQuota);
}

StatusWith<RecordId> RocksRecordStore::updateRecord(OperationContext* txn,
                                                    const RecordId& loc,
                                                    const char* data,
                                                    int len,
                                                    bool enforceQuota,
                                                    UpdateNotifier* notifier) {
    std::string key(_makePrefixedKey(_prefix, loc));

    RocksRecoveryUnit* ru = RocksRecoveryUnit::getRocksRecoveryUnit(txn);
    if (!ru->transaction()->registerWrite(key)) {
        throw WriteConflictException();
    }

    std::string old_value;
    auto status = ru->Get(key, &old_value);
    invariantRocksOK(status);

    int old_length = old_value.size();

    ru->writeBatch()->Put(key, rocksdb::Slice(data, len));
    if (_isOplog) {
        _oplogKeyTracker->insertKey(ru, loc, len);
    }

    _increaseDataSize(txn, len - old_length);

    cappedDeleteAsNeeded(txn, loc);

    return StatusWith<RecordId>(loc);
}

bool RocksRecordStore::updateWithDamagesSupported() const {
    return false;
}

Status RocksRecordStore::updateWithDamages(OperationContext* txn,
                                           const RecordId& loc,
                                           const RecordData& oldRec,
                                           const char* damageSource,
                                           const mutablebson::DamageVector& damages) {
    invariant(false);
    return Status::OK();
}

RecordIterator* RocksRecordStore::getIterator(OperationContext* txn,
                                              const RecordId& start,
                                              const CollectionScanParams::Direction& dir) const {
    RecordId startIterator = start;

    if (_isOplog) {
        if (dir == CollectionScanParams::FORWARD) {
            auto ru = RocksRecoveryUnit::getRocksRecoveryUnit(txn);
            if (!ru->hasSnapshot() || ru->getOplogReadTill().isNull()) {
                // we don't have snapshot, we can update our view
                ru->setOplogReadTill(_cappedVisibilityManager->oplogStartHack());
            }
            if (start.isNull()) {
                startIterator = _cappedOldestKeyHint;
            }
        } else if (start.isNull()) {  // backward iterator + beginning not set
            startIterator = _cappedVisibilityManager->oplogStartHack();
        }
    } else if (_isCapped && start.isNull() && dir == CollectionScanParams::FORWARD) {
        // seek to first in capped collection. we know that there are no records smaller than
        // _cappedOldestKeyHint that are alive
        startIterator = _cappedOldestKeyHint;
    }

    return new Iterator(txn, _db, _prefix, _cappedVisibilityManager, dir, startIterator);
}

std::vector<RecordIterator*> RocksRecordStore::getManyIterators(OperationContext* txn) const {
    return {new Iterator(
        txn, _db, _prefix, _cappedVisibilityManager, CollectionScanParams::FORWARD, RecordId())};
}

Status RocksRecordStore::truncate(OperationContext* txn) {
    boost::scoped_ptr<RecordIterator> iter(getIterator(txn));
    while (!iter->isEOF()) {
        RecordId loc = iter->getNext();
        deleteRecord(txn, loc);
    }

    return Status::OK();
}

Status RocksRecordStore::compact(OperationContext* txn,
                                 RecordStoreCompactAdaptor* adaptor,
                                 const CompactOptions* options,
                                 CompactStats* stats) {
    std::string beginString(_makePrefixedKey(_prefix, RecordId()));
    std::string endString(_makePrefixedKey(_prefix, RecordId::max()));
    rocksdb::Slice beginRange(beginString);
    rocksdb::Slice endRange(endString);
    return rocksToMongoStatus(_db->CompactRange(&beginRange, &endRange));
}

Status RocksRecordStore::validate(OperationContext* txn,
                                  bool full,
                                  bool scanData,
                                  ValidateAdaptor* adaptor,
                                  ValidateResults* results,
                                  BSONObjBuilder* output) {
    long long nrecords = 0;
    long long dataSizeTotal = 0;
    if (scanData) {
        boost::scoped_ptr<RecordIterator> iter(getIterator(txn));
        results->valid = true;
        while (!iter->isEOF()) {
            ++nrecords;
            if (full) {
                size_t dataSize;
                RecordId loc = iter->curr();
                RecordData data = dataFor(txn, loc);
                Status status = adaptor->validate(data, &dataSize);
                if (!status.isOK()) {
                    results->valid = false;
                    results->errors.push_back(str::stream() << loc << " is corrupted");
                }
                dataSizeTotal += static_cast<long long>(dataSize);
            }
            iter->getNext();
        }

        if (full && results->valid) {
            long long storedNumRecords = numRecords(txn);
            long long storedDataSize = dataSize(txn);

            if (nrecords != storedNumRecords || dataSizeTotal != storedDataSize) {
                warning() << _ident << ": Existing record and data size counters ("
                          << storedNumRecords << " records " << storedDataSize << " bytes) "
                          << "are inconsistent with full validation results (" << nrecords
                          << " records " << dataSizeTotal << " bytes). "
                          << "Updating counters with new values.";
                if (nrecords != storedNumRecords) {
                    _changeNumRecords(txn, nrecords - storedNumRecords);
                    _increaseDataSize(txn, dataSizeTotal - storedDataSize);
                }
            }
        }
        output->appendNumber("nrecords", nrecords);
    } else {
        output->appendNumber("nrecords", numRecords(txn));
    }

    return Status::OK();
}

void RocksRecordStore::appendCustomStats(OperationContext* txn,
                                         BSONObjBuilder* result,
                                         double scale) const {
    string statsString;
    result->appendBool("capped", _isCapped);
    if (_isCapped) {
        result->appendIntOrLL("max", _cappedMaxDocs);
        result->appendIntOrLL("maxSize", _cappedMaxSize / scale);
    }
}

Status RocksRecordStore::oplogDiskLocRegister(OperationContext* txn, const OpTime& opTime) {
    invariant(_isOplog);
    StatusWith<RecordId> record = oploghack::keyForOptime(opTime);
    if (record.isOK()) {
        _cappedVisibilityManager->addUncommittedRecord(txn, record.getValue());
    }

    return record.getStatus();
}

void RocksRecordStore::updateStatsAfterRepair(OperationContext* txn,
                                              long long numRecords,
                                              long long dataSize) {
    _numRecords.store(numRecords);
    _dataSize.store(dataSize);
    rocksdb::WriteBatch wb;
    _counterManager->updateCounter(_numRecordsKey, numRecords, &wb);
    _counterManager->updateCounter(_dataSizeKey, dataSize, &wb);
    if (wb.Count() > 0) {
        auto s = _db->Write(rocksdb::WriteOptions(), &wb);
        invariantRocksOK(s);
    }
}

/**
 * Return the RecordId of an oplog entry as close to startingPosition as possible without
 * being higher. If there are no entries <= startingPosition, return RecordId().
 */
boost::optional<RecordId> RocksRecordStore::oplogStartHack(OperationContext* txn,
                                                           const RecordId& startingPosition) const {
    if (!_isOplog) {
        return boost::none;
    }

    auto ru = RocksRecoveryUnit::getRocksRecoveryUnit(txn);
    ru->setOplogReadTill(_cappedVisibilityManager->oplogStartHack());

    // we use _oplogKeyTracker, which contains exactly the same keys as oplog. the difference is
    // that values are different (much smaller), so reading is faster. in this case, we only
    // need keys (we never touch the values), so this works nicely
    boost::scoped_ptr<rocksdb::Iterator> iter(_oplogKeyTracker->newIterator(ru));
    int64_t storage;
    iter->Seek(_makeKey(startingPosition, &storage));
    if (!iter->Valid()) {
        iter->SeekToLast();
        if (iter->Valid()) {
            // startingPosition is bigger than everything else
            return _makeRecordId(iter->key());
        } else {
            invariantRocksOK(iter->status());
            // record store is empty
            return RecordId();
        }
    }

    // We're at or past target:
    // 1) if we're at -- return
    // 2) if we're past -- do a prev()
    RecordId foundKey = _makeRecordId(iter->key());
    int cmp = startingPosition.compare(foundKey);
    if (cmp != 0) {
        // RocksDB invariant -- iterator needs to land at or past target when Seek-ing
        invariant(cmp < 0);
        // we're past target -- prev()
        iter->Prev();
    }

    if (!iter->Valid()) {
        invariantRocksOK(iter->status());
        // there are no entries <= startingPosition
        return RecordId();
    }

    return _makeRecordId(iter->key());
}

void RocksRecordStore::temp_cappedTruncateAfter(OperationContext* txn,
                                                RecordId end,
                                                bool inclusive) {
    // copied from WiredTigerRecordStore::temp_cappedTruncateAfter()
    WriteUnitOfWork wuow(txn);
    boost::scoped_ptr<RecordIterator> iter(getIterator(txn, end));
    while (!iter->isEOF()) {
        RecordId loc = iter->getNext();
        if (end < loc || (inclusive && end == loc)) {
            deleteRecord(txn, loc);
        }
    }
    wuow.commit();
}

RecordId RocksRecordStore::_nextId() {
    invariant(!_isOplog);
    return RecordId(_nextIdNum.fetchAndAdd(1));
}

rocksdb::Slice RocksRecordStore::_makeKey(const RecordId& loc, int64_t* storage) {
    *storage = endian::nativeToBig(loc.repr());
    return rocksdb::Slice(reinterpret_cast<const char*>(storage), sizeof(*storage));
}

std::string RocksRecordStore::_makePrefixedKey(const std::string& prefix, const RecordId& loc) {
    int64_t storage;
    auto encodedLoc = _makeKey(loc, &storage);
    std::string key(prefix);
    key.append(encodedLoc.data(), encodedLoc.size());
    return key;
}

RecordId RocksRecordStore::_makeRecordId(const rocksdb::Slice& slice) {
    invariant(slice.size() == sizeof(int64_t));
    int64_t repr = endian::bigToNative(*reinterpret_cast<const int64_t*>(slice.data()));
    RecordId a(repr);
    return RecordId(repr);
}

bool RocksRecordStore::findRecord(OperationContext* txn,
                                  const RecordId& loc,
                                  RecordData* out) const {
    RecordData rd = _getDataFor(_db, _prefix, txn, loc);
    if (rd.data() == NULL)
        return false;
    *out = rd;
    return true;
}

RecordData RocksRecordStore::_getDataFor(rocksdb::DB* db,
                                         const std::string& prefix,
                                         OperationContext* txn,
                                         const RecordId& loc) {
    RocksRecoveryUnit* ru = RocksRecoveryUnit::getRocksRecoveryUnit(txn);

    std::string valueStorage;
    auto status = ru->Get(_makePrefixedKey(prefix, loc), &valueStorage);
    if (status.IsNotFound()) {
        return RecordData(nullptr, 0);
    }
    invariantRocksOK(status);

    SharedBuffer data = SharedBuffer::allocate(valueStorage.size());
    memcpy(data.get(), valueStorage.data(), valueStorage.size());
    return RecordData(data.moveFrom(), valueStorage.size());
}

void RocksRecordStore::_changeNumRecords(OperationContext* txn, int64_t amount) {
    RocksRecoveryUnit* ru = RocksRecoveryUnit::getRocksRecoveryUnit(txn);
    ru->incrementCounter(_numRecordsKey, &_numRecords, amount);
}

void RocksRecordStore::_increaseDataSize(OperationContext* txn, int64_t amount) {
    RocksRecoveryUnit* ru = RocksRecoveryUnit::getRocksRecoveryUnit(txn);
    ru->incrementCounter(_dataSizeKey, &_dataSize, amount);
}

// --------

RocksRecordStore::Iterator::Iterator(
    OperationContext* txn,
    rocksdb::DB* db,
    std::string prefix,
    boost::shared_ptr<CappedVisibilityManager> cappedVisibilityManager,
    const CollectionScanParams::Direction& dir,
    const RecordId& start)
    : _txn(txn),
      _db(db),
      _prefix(std::move(prefix)),
      _cappedVisibilityManager(cappedVisibilityManager),
      _dir(dir),
      _eof(true),
      _readUntilForOplog(RocksRecoveryUnit::getRocksRecoveryUnit(txn)->getOplogReadTill()),
      _iterator(RocksRecoveryUnit::getRocksRecoveryUnit(txn)
                    ->NewIterator(_prefix, /* isOplog */ !_readUntilForOplog.isNull())) {
    _locate(start);
}

bool RocksRecordStore::Iterator::isEOF() {
    return _eof;
}

RecordId RocksRecordStore::Iterator::curr() {
    if (_eof) {
        return RecordId();
    }

    return _curr;
}

RecordId RocksRecordStore::Iterator::getNext() {
    if (_eof) {
        return RecordId();
    }

    RecordId toReturn = _curr;

    if (_forward())
        _iterator->Next();
    else
        _iterator->Prev();

    if (_iterator->Valid()) {
        _curr = _decodeCurr();
        if (_cappedVisibilityManager.get()) {  // isCapped?
            if (_readUntilForOplog.isNull()) {
                // this is the normal capped case
                if (_cappedVisibilityManager->isCappedHidden(_curr)) {
                    _eof = true;
                }
            } else {
                // this is for oplogs
                if (_curr > _readUntilForOplog ||
                    (_curr == _readUntilForOplog &&
                     _cappedVisibilityManager->isCappedHidden(_curr))) {
                    _eof = true;
                }
            }
        }  // isCapped?
    } else {
        invariantRocksOK(_iterator->status());
        _eof = true;
        // we leave _curr as it is on purpose
    }

    _lastLoc = toReturn;
    return toReturn;
}

void RocksRecordStore::Iterator::invalidate(const RecordId& dl) {
    // this should never be called
}

void RocksRecordStore::Iterator::saveState() {
    _iterator.reset();
    _txn = nullptr;
}

bool RocksRecordStore::Iterator::restoreState(OperationContext* txn) {
    _txn = txn;
    if (_eof) {
        return true;
    }

    auto ru = RocksRecoveryUnit::getRocksRecoveryUnit(txn);
    _iterator.reset(ru->NewIterator(_prefix, /* isOplog */ !_readUntilForOplog.isNull()));

    RecordId saved = _lastLoc;
    _locate(_lastLoc);

    if (_eof) {
        _lastLoc = RecordId();
    } else if (_curr != saved) {
        // _cappedVisibilityManager is not-null when isCapped == true
        if (_cappedVisibilityManager.get() && saved != RecordId()) {
            // Doc was deleted either by cappedDeleteAsNeeded() or cappedTruncateAfter().
            // It is important that we error out in this case so that consumers don't
            // silently get 'holes' when scanning capped collections. We don't make
            // this guarantee for normal collections so it is ok to skip ahead in that case.
            _eof = true;
            return false;
        }
        // lastLoc was either deleted or never set (yielded before first call to getNext()),
        // so bump ahead to the next record.
    } else {
        // we found where we left off! we advanced to the next one
        getNext();
        _lastLoc = saved;
    }

    return true;
}

RecordData RocksRecordStore::Iterator::dataFor(const RecordId& loc) const {
    if (!_eof && loc == _curr && _iterator->Valid() && _iterator->status().ok()) {
        SharedBuffer data = SharedBuffer::allocate(_iterator->value().size());
        memcpy(data.get(), _iterator->value().data(), _iterator->value().size());
        return RecordData(data.moveFrom(), _iterator->value().size());
    }
    return RocksRecordStore::_getDataFor(_db, _prefix, _txn, loc);
}

void RocksRecordStore::Iterator::_locate(const RecordId& loc) {
    if (_forward()) {
        if (loc.isNull()) {
            _iterator->SeekToFirst();
        } else {
            int64_t locStorage;
            _iterator->Seek(RocksRecordStore::_makeKey(loc, &locStorage));
        }
    } else {  // backward iterator
        if (loc.isNull()) {
            _iterator->SeekToLast();
        } else {
            // lower bound on reverse iterator
            int64_t locStorage;
            _iterator->Seek(RocksRecordStore::_makeKey(loc, &locStorage));
            if (!_iterator->Valid()) {
                invariantRocksOK(_iterator->status());
                _iterator->SeekToLast();
            } else if (_decodeCurr() != loc) {
                _iterator->Prev();
            }
        }
    }
    _eof = !_iterator->Valid();
    if (_eof) {
        invariantRocksOK(_iterator->status());
        _curr = loc;
    } else {
        _curr = _decodeCurr();
    }
}

RecordId RocksRecordStore::Iterator::_decodeCurr() const {
    invariant(_iterator && _iterator->Valid());
    return _makeRecordId(_iterator->key());
}

bool RocksRecordStore::Iterator::_forward() const {
    return _dir == CollectionScanParams::FORWARD;
}
}

max_stars_repo_name: DavidAlphaFox/mongodb

max_issues_repo_name: DavidAlphaFox/mongodb

max_forks_repo_name: DavidAlphaFox/mongodb


text:
#pragma once

#include "ft/utils/Common.hpp"
#include "toolbox/util/Slot.hpp"
#include <boost/mp11/algorithm.hpp>
#include <boost/mp11/tuple.hpp>

namespace mp = boost::mp11;

namespace tb = toolbox;

namespace ft { inline namespace util {

template<typename T, typename ...ArgsT>
struct Signal : public tb::Signal<T, ArgsT...> {
    using Base = tb::Signal<T, ArgsT...>;
    using value_type = T;
    using Base::Base;
    using Base::operator();
    using Base::operator bool;
    using Base::connect;
    using Base::disconnect;
};

template<typename TypeListT, std::size_t size=1>
struct SignalTuple
{
    template<typename...Args>
    SignalTuple(Args...args)
    : impl_(std::forward<Args>(args)...) {}

    template<typename T>
    void connect(tb::BasicSlot<T> fn)
    {
        std::get< mp::mp_find<TypeListT, T>::value >(impl_).connect(fn);
    }

    template<typename T>
    void disconnect(tb::BasicSlot<T> fn)
    {
        std::get< mp::mp_find<TypeListT, T>::value >(impl_).disconnect(fn);
    }

    template<typename Fn>
    void for_each(Fn&& fn) {
        mp::tuple_for_each(impl_, fn);
    }
    mp::mp_rename<mp::mp_transform<Signal, TypeListT>, std::tuple> impl_;
};

}} // ft::util

ext: hpp

max_stars_repo_name: mmitkevich/freeticks-cpp

max_issues_repo_name: mmitkevich/freeticks-cpp

max_forks_repo_name: mmitkevich/freeticks-cpp


text:
/**
 * TitaniumKit
 *
 * Copyright (c) 2014-2015 by Appcelerator, Inc. All Rights Reserved.
 * Licensed under the terms of the Apache Public License.
 * Please see the LICENSE included with this distribution for details.
 */

#include "Titanium/GlobalObject.hpp"
#include "Titanium/detail/TiUtil.hpp"
#include "Titanium/detail/TiImpl.hpp"
#include "Titanium/Module.hpp"
#include <sstream>
#include <functional>
#include <boost/algorithm/string/predicate.hpp>
#include <boost/algorithm/string.hpp>
#include <boost/algorithm/string/join.hpp>

namespace Titanium
{
	using namespace Titanium::detail;

	const std::string GlobalObject::COMMONJS_SEPARATOR__{"/"};
	std::atomic<std::uint32_t> GlobalObject::timer_id_generator__;
	std::uint32_t GlobalObject::require_nested_count__ = 0;

	GlobalObject::GlobalObject(const JSContext& js_context) TITANIUM_NOEXCEPT
	    : JSExportObject(js_context)
		, currentDir__(COMMONJS_SEPARATOR__)
	{
		TITANIUM_LOG_DEBUG("GlobalObject:: ctor ", this);
	}

	void GlobalObject::postCallAsConstructor(const JSContext& js_context, const std::vector<JSValue>& arguments) {
		HAL_LOG_DEBUG("GlobalObject:: postCallAsConstructor ", this);
	}

	GlobalObject::~GlobalObject() TITANIUM_NOEXCEPT
	{
		TITANIUM_LOG_DEBUG("GlobalObject:: dtor ", this);
	}

	static std::vector<std::string> slice(const std::vector<std::string>& list, const size_t& begin, const size_t& end=std::string::npos)
	{
		std::vector<std::string> newlist;
		for (size_t i = begin; i < end; i++) {
			newlist.push_back(list[i]);
		}
		return newlist;
	}
	std::vector<std::string> GlobalObject::resolveRequirePaths(const std::string& dirname) const TITANIUM_NOEXCEPT
	{
		std::vector<std::string> paths;
		if (dirname != COMMONJS_SEPARATOR__) {
#pragma warning(push)
#pragma warning(disable:4996)
			std::vector<std::string> parts;
			boost::split(parts, dirname, boost::is_any_of(COMMONJS_SEPARATOR__));
			boost::split(parts, dirname, boost::is_any_of(COMMONJS_SEPARATOR__));
#pragma warning(pop)
			for (size_t i = parts.size() - 1; i > 0; i--) {
				if (parts[i] == "node_modules" || parts[i].empty()) {
					continue;
				}
				auto p = slice(parts, 0, i + 1);
				auto path =  boost::algorithm::join(p, COMMONJS_SEPARATOR__) + "/node_modules";
				paths.push_back(path);
			}
		}
		paths.push_back("/node_modules");
		return paths;
	}

	std::string GlobalObject::resolvePath(const std::string& path, const std::string& basedir) const TITANIUM_NOEXCEPT
	{
		std::string dir = basedir.length() == 1 || basedir.rfind(COMMONJS_SEPARATOR__) == basedir.length() ? basedir : basedir + COMMONJS_SEPARATOR__;
		const bool resolvedAbsolute = path.find(COMMONJS_SEPARATOR__)==0;

#pragma warning(push)
#pragma warning(disable:4996)
		std::vector<std::string> parts;
		boost::split(parts, path, boost::is_any_of(COMMONJS_SEPARATOR__));
#pragma warning(pop)

		size_t up = 0;
		const bool allowAboveRoot = !resolvedAbsolute;
		std::vector<std::string> newparts;
		for (std::vector<std::string>::reverse_iterator rit = parts.rbegin(); rit != parts.rend(); rit++) {
			std::string last = *rit;
			if (last == "") {
				// skip
			} else if (last == ".") {
				// skip
			} else if (last == "..") {
				// skip
				up++;
			} else if (up) {
				// skip
				up--;
			} else {
				std::vector<std::string>::iterator it = newparts.begin();
				newparts.insert(it, last);
			}
		}

		if (allowAboveRoot) {
			for (; up--;) {
				std::vector<std::string>::iterator it = newparts.begin();
				newparts.insert(it, "..");
			}
		}

		std::string newpath =  boost::algorithm::join(newparts, COMMONJS_SEPARATOR__);
		return resolvedAbsolute ? newpath : dir + newpath;
	}

	std::string GlobalObject::resolvePathAsFile(const JSObject& parent, const std::string& path) const TITANIUM_NOEXCEPT
	{
		std::vector<std::string> checks;
		checks.push_back(path);
		checks.push_back(path+".js");
		checks.push_back(path+".json");

		for (std::vector<std::string>::iterator i = checks.begin(); i!=checks.end(); i++) {
			auto check = *i;
			if (requiredModuleExists(check)) {
				return check;
			}
		}
		return std::string();
	}

	std::string GlobalObject::resolvePathAsDirectory(const JSObject& parent, const std::string& path) const TITANIUM_NOEXCEPT
	{
		const auto packageJSONFile = path + "/package.json";
		if (requiredModuleExists(packageJSONFile)) {
			TITANIUM_LOG_DEBUG("package.json exists");
			const auto content = readRequiredModule(parent, packageJSONFile);
			TITANIUM_LOG_DEBUG("Content: " + content);
			const auto result = parent.get_context().CreateValueFromJSON(content);
			if (result.IsObject()) {
				TITANIUM_LOG_DEBUG("Result from loading JSON is an object");
				const auto json = static_cast<JSObject>(result);
				auto mainValue = json.GetProperty("main");
				if (mainValue.IsString()) {
					TITANIUM_LOG_DEBUG("main: " + static_cast<std::string>(mainValue));
					const auto resolved = resolvePath(static_cast<std::string>(mainValue), path);
					std::string modulePath = resolvePathAsFile(parent, resolved);
					if (!modulePath.empty()) {
						return modulePath;
					}
					modulePath = resolvePathAsDirectory(parent, resolved);
					if (!modulePath.empty()) {
						return modulePath;
					}
				}
			}
		}
		const auto indexFile = path + "/index.js";
		if (requiredModuleExists(indexFile)) {
			return indexFile;
		}
		const auto indexJSON = path + "/index.json";
		if (requiredModuleExists(indexJSON)) {
			return indexJSON;
		}
		return std::string();
	}

	std::string GlobalObject::resolvePathAsModule(const JSObject& parent, const std::string& path, const std::string& dirname, const std::string& moduleId) const TITANIUM_NOEXCEPT
	{
		std::string modulePath;
		std::string resolvedPath = path;
		// First look for "legacy" CommonJS module as node_modules/moduleId/moduleId.js
		modulePath = resolvePathAsFile(parent, "node_modules" + COMMONJS_SEPARATOR__ + resolvedPath + COMMONJS_SEPARATOR__ + resolvedPath);
		if (!modulePath.empty()) {
			return modulePath;
		}

		// Now try looking up the node_modules paths
		auto reqPaths = resolveRequirePaths(dirname);
		if (resolvedPath.find(COMMONJS_SEPARATOR__)!=0) {
			resolvedPath = COMMONJS_SEPARATOR__ + resolvedPath;
		}
		for (size_t i = 0, len = reqPaths.size(); i < len; i++) {
			auto newResolvedPath = reqPaths[i] + COMMONJS_SEPARATOR__ + moduleId;
			modulePath = resolvePathAsFile(parent, newResolvedPath);
			if (!modulePath.empty()) {
				return modulePath;
			}
			modulePath = resolvePathAsDirectory(parent, newResolvedPath);
			if (!modulePath.empty()) {
				return modulePath;
			}
		}
		return modulePath;
	}

	std::string GlobalObject::requestResolveModule(const JSObject& parent, const std::string& moduleId, const std::string& dirname) TITANIUM_NOEXCEPT
	{
		// Create unique key for the dirname + moduleId, assuming moduleId does not contain whitespace.
		const auto modulePathCacheKey = dirname + " " + moduleId;
		const auto cached = module_path_cache__.find(modulePathCacheKey);
		if (cached != module_path_cache__.end()) {
			return cached->second;
		}

		auto isNodeModule = false;
		std::string rawPath;

		if (moduleId.find(COMMONJS_SEPARATOR__)==0) {
			rawPath = moduleId;
		} else if (moduleId.find("../") == 0 || moduleId.find("./") == 0) {
			rawPath = dirname + (dirname == COMMONJS_SEPARATOR__ ? "" : COMMONJS_SEPARATOR__) + moduleId;
		} else {
			isNodeModule = true;
			rawPath = moduleId;
		}

		const auto resolvedPath = resolvePath(rawPath ,dirname);
		std::string modulePath;

		if (isNodeModule) {
			modulePath = resolvePathAsModule(parent, resolvedPath, dirname, moduleId);
		} else {
			// load as file or load as directory
			modulePath = resolvePathAsFile(parent, resolvedPath);
			if (modulePath.empty()) {
				modulePath = resolvePathAsDirectory(parent, resolvedPath);
			}
		}
		// Functions return different paths, some return 'path/to/file.js', some
		// '/path/to/file.js', this converts them all to be '/path/to/file.js',
		// as long as a value was returned
		if (!modulePath.empty()) {
			modulePath = boost::algorithm::replace_all_copy("/" + modulePath, "//", COMMONJS_SEPARATOR__);
		}
		module_path_cache__.emplace(modulePathCacheKey, modulePath);

		return modulePath;
	}

	bool GlobalObject::requiredNativeModuleExists(const JSContext& js_context, const std::string& moduleId) const TITANIUM_NOEXCEPT
	{
		return false;
	}

	JSValue GlobalObject::requireNativeModule(const JSContext& js_context, const std::string& moduleId)
	{
		return js_context.CreateUndefined();
	}

	bool GlobalObject::requiredBuiltinModuleExists(const JSContext& js_context, const std::string& moduleId) const TITANIUM_NOEXCEPT
	{
		if (moduleId == "ti.map") {
			return true;
		}
		return false;
	}

	JSValue GlobalObject::requireBuiltinModule(const JSContext& js_context, const std::string& moduleId)
	{
		if (moduleId == "ti.map") {
			JSValue Titanium_property = js_context.get_global_object().GetProperty("Titanium");
			TITANIUM_ASSERT(Titanium_property.IsObject());  // precondition
			JSObject Titanium = static_cast<JSObject>(Titanium_property);

			JSValue Map_property = Titanium.GetProperty("Map");
			TITANIUM_ASSERT(Map_property.IsObject());  // precondition

			return Map_property;
		}

		return js_context.CreateUndefined();
	}

	JSValue GlobalObject::requireModule(const JSObject& parent, const std::string& moduleId, const bool& reload)
	{
		TITANIUM_GLOBALOBJECT_LOCK_GUARD;
		const auto js_context = parent.get_context();

		TITANIUM_LOG_DEBUG(std::string(require_nested_count__, '-') + " require(" + moduleId + ") from ", currentDir__);

		// check if we have special module such as ti.map
		if (requiredBuiltinModuleExists(js_context, moduleId)) {
			return requireBuiltinModule(js_context, moduleId);
		}

		// check if we have native module
		if (requiredNativeModuleExists(js_context, moduleId)) {
			return requireNativeModule(js_context, moduleId);
		}

		// FIXME We should be able to ask for the parent module's filename property to determine the dirname!
		// Parent here seems to _always_ be the global object.
		std::string dirname = currentDir__;

		auto module_path = requestResolveModule(parent, moduleId, dirname);
		if (module_path.empty() && moduleId.find(".") != 0 && moduleId.find("/") != 0) {
			// Fall back to assuming equivalent of "/" + moduleId (legacy behavior) if we don't already have a leading . or /
			module_path = requestResolveModule(parent, "/" + moduleId);
		}
		if (module_path.empty()) {
			detail::ThrowRuntimeError("require", "Could not load module " + moduleId);
		}

		// check if we have already loaded the module
		if (!reload && module_cache__.find(module_path) != module_cache__.end()) {
			return module_cache__.at(module_path);
		}

		// FIXME We need to create a new Module object where we set the filename and parent on it after resolving the filename
		// This should be what we use for parent object in sub-require() calls
		// Node basically hangs a special require and resolve function off each module object
		// the require call has references to the __dirname, __filename, parent, global, etc
		//JSObject module = js_context.CreateObject();

		const auto module_js = readRequiredModule(parent, module_path);

		if (module_js.empty()) {
			detail::ThrowRuntimeError("require", "Could not load module " + moduleId);
		}

		// Update currentDir__ pointer
		std::size_t found = module_path.rfind(COMMONJS_SEPARATOR__);
		if (found != std::string::npos) {
			currentDir__ = COMMONJS_SEPARATOR__ + module_path.substr(0, found); //  Take the dirname of this, not the full filename
		}
		else {
			currentDir__ = COMMONJS_SEPARATOR__;
		}
		boost::algorithm::replace_all(currentDir__, "//", COMMONJS_SEPARATOR__);
		try {
			JSValue result = js_context.CreateUndefined();
			if (boost::ends_with(module_path, ".json")){
				result = js_context.CreateValueFromJSON(module_js);
			} else {
				const std::string require_module_js = "(function(global) { var exports={},__OXP=exports,module={'exports':exports},__dirname='" + currentDir__ + "',__filename='"
					+ module_path + "';try {" + module_js + R"JS(
						if(module.exports !== __OXP){
							return module.exports;
						} else {
							return exports;
						}
					  } catch (E) {
						if (!E.fileName) {
						    E.fileName = __filename;
						}
						Titanium_RedScreenOfDeath(E);
					  }
					})(this);
					)JS";
				result = js_context.JSEvaluateScript(require_module_js, js_context.get_global_object(), module_path);
			}
			currentDir__ = dirname; // Should ensure this gets reset on _EVERY_ code branch possible here. Would be nice if C++/CX had finally blocks
			if (!result.IsObject()) {
				TITANIUM_LOG_WARN("GlobalObject::require: module '", moduleId, "' replaced 'exports' with a non-object: ", to_string(result));
			}
			// cache it so that we can reuse it
			module_cache__.insert({module_path, result});
			return result;
		} catch (const std::exception& exception) {
			currentDir__ = dirname; // Should ensure this gets reset on _EVERY_ code branch possible here. Would be nice if C++/CX had finally blocks
			detail::ThrowRuntimeError("require", "Error while require(" + moduleId + ") "+static_cast<std::string>(exception.what()));
		} catch (...) {
			currentDir__ = dirname; // Should ensure this gets reset on _EVERY_ code branch possible here. Would be nice if C++/CX had finally blocks
			detail::ThrowRuntimeError("require", "Unknown error while require(" + moduleId + ")");
		}
		return js_context.CreateUndefined();
	}

	unsigned GlobalObject::setTimeout(JSObject&& function, const std::chrono::milliseconds& delay) TITANIUM_NOEXCEPT
	{
		const auto timerId = timer_id_generator__++;

		RegisterCallback(std::move(function), timerId);

		Callback_t callback = [this, timerId]() mutable {
			InvokeCallback(timerId);
			clearTimeout(timerId);
		};

		StartTimer(std::move(callback), timerId, delay);

		return timerId;
	}

	void GlobalObject::clearTimeout(const unsigned& timerId) TITANIUM_NOEXCEPT
	{
		StopTimer(timerId);
	}

	unsigned GlobalObject::setInterval(JSObject&& function, const std::chrono::milliseconds& delay) TITANIUM_NOEXCEPT
	{
		const auto timerId = timer_id_generator__++;

		RegisterCallback(std::move(function), timerId);

		Callback_t callback = [this, timerId]() mutable {
			InvokeCallback(timerId);
		};

		StartTimer(std::move(callback), timerId, delay);

		return timerId;
	}

	void GlobalObject::clearInterval(const unsigned& timerId) TITANIUM_NOEXCEPT
	{
		StopTimer(timerId);
	}

	void GlobalObject::RegisterCallback(JSObject&& function, const unsigned& timerId) TITANIUM_NOEXCEPT
	{
		TITANIUM_ASSERT(function.IsFunction());
		TITANIUM_ASSERT(timer_callback_map__.find(timerId) == timer_callback_map__.end());
		timer_callback_map__.emplace(timerId, function);
	}

	void GlobalObject::UnregisterCallback(const unsigned& timerId) TITANIUM_NOEXCEPT
	{
		TITANIUM_ASSERT(timer_callback_map__.find(timerId) != timer_callback_map__.end());
		timer_callback_map__.erase(timerId);
	}

	void GlobalObject::InvokeCallback(const unsigned& timerId) TITANIUM_NOEXCEPT
	{
		TITANIUM_EXCEPTION_CATCH_START{
			const auto found = timer_callback_map__.find(timerId);
			TITANIUM_ASSERT(found != timer_callback_map__.end());
			auto callback = found->second;
			TITANIUM_ASSERT(callback.IsFunction());
			callback(get_context().get_global_object());
		} TITANIUM_EXCEPTION_CATCH_END
	}

	void GlobalObject::StartTimer(Callback_t&& callback, const unsigned& timerId, const std::chrono::milliseconds& delay) TITANIUM_NOEXCEPT
	{
		auto timer = CreateTimer(callback, delay);
		const auto timer_insert_result = timer_map__.emplace(timerId, timer);
		const bool timer_inserted = timer_insert_result.second;
		TITANIUM_ASSERT(timer_inserted);
		TITANIUM_LOG_DEBUG("GlobalObject::setInterval: starting timer for timerId ", timerId);
		timer->Start();
	}

	void GlobalObject::StopTimer(const unsigned& timerId) TITANIUM_NOEXCEPT
	{
		const auto timer_position = timer_map__.find(timerId);
		const bool timer_found = timer_position != timer_map__.end();
		if (timer_found) {
			TITANIUM_LOG_DEBUG("GlobalObject::clearInterval: timerId ", timerId, " cleared");
			auto timer_ptr = timer_position->second;
			timer_ptr->Stop();
			const auto number_of_elements_removed = timer_map__.erase(timerId);
			TITANIUM_ASSERT(number_of_elements_removed == 1);

			TITANIUM_ASSERT(timer_callback_map__.find(timerId) != timer_callback_map__.end());
			timer_callback_map__.erase(timerId);
		} else {
			TITANIUM_LOG_WARN("GlobalObject::clearTimeout: timerId ", timerId, " is not registered");
		}
	}

	bool GlobalObject::requiredModuleExists(const std::string& path) const TITANIUM_NOEXCEPT
	{
		TITANIUM_LOG_ERROR("GlobalObject::requiredModuleExists: Unimplemented");
		return false;
	}

	std::string GlobalObject::readRequiredModule(const JSObject& parent, const std::string& path) const
	{
		TITANIUM_LOG_ERROR("GlobalObject::readRequiredModule: Unimplemented");
		return "";
	}

	GlobalObject::Timer::Timer(Callback_t callback, const std::chrono::milliseconds& interval)
	    : interval__(interval)
	{
		TITANIUM_LOG_DEBUG("GlobalObject::Timer: interval = ", interval.count(), "ms");
	}

	std::chrono::milliseconds GlobalObject::Timer::get_interval() const TITANIUM_NOEXCEPT
	{
		return interval__;
	}

	class UnimplementedTimer final : public GlobalObject::Timer
	{
	public:
		UnimplementedTimer(GlobalObject::Callback_t callback, const std::chrono::milliseconds& interval)
		    : Timer(callback, interval)
		{
			TITANIUM_LOG_ERROR("GlobalObject::Timer: Unimplemented");
		}

		virtual ~UnimplementedTimer()
		{
		}

		virtual void Start() TITANIUM_NOEXCEPT override final
		{
			TITANIUM_LOG_ERROR("GlobalObject::Timer::Start: Unimplemented");
		}

		virtual void Stop() TITANIUM_NOEXCEPT override final
		{
			TITANIUM_LOG_ERROR("GlobalObject::Timer::Stop: Unimplemented");
		}
	};

	std::shared_ptr<GlobalObject::Timer> GlobalObject::CreateTimer(Callback_t callback, const std::chrono::milliseconds& interval) const TITANIUM_NOEXCEPT
	{
		TITANIUM_LOG_ERROR("GlobalObject::CreateTimer: Unimplemented");
		return std::make_shared<UnimplementedTimer>(callback, interval);
	}

	void GlobalObject::JSExportInitialize()
	{
		TITANIUM_LOG_DEBUG("GlobalObject::JSExportInitialize");
		JSExport<GlobalObject>::SetClassVersion(1);
		JSExport<GlobalObject>::SetParent(JSExport<JSExportObject>::Class());

		TITANIUM_ADD_PROPERTY_READONLY(GlobalObject, global);
		TITANIUM_ADD_FUNCTION(GlobalObject, require);
		TITANIUM_ADD_FUNCTION(GlobalObject, setTimeout);
		TITANIUM_ADD_FUNCTION(GlobalObject, clearTimeout);
		TITANIUM_ADD_FUNCTION(GlobalObject, setInterval);
		TITANIUM_ADD_FUNCTION(GlobalObject, clearInterval);
	}

	TITANIUM_PROPERTY_GETTER(GlobalObject, global)
	{
		return get_context().get_global_object();
	}

	TITANIUM_FUNCTION(GlobalObject, require)
	{
		ENSURE_STRING_AT_INDEX(moduleId, 0);

		const auto global_object = this_object.get_context().get_global_object();
		const auto global_ptr = global_object.GetPrivate<GlobalObject>();
		TITANIUM_ASSERT(global_ptr);

		require_nested_count__++;
		const auto module = global_ptr->requireModule(this_object, moduleId);
		require_nested_count__--;
		return module;
	}

	TITANIUM_FUNCTION(GlobalObject, setTimeout)
	{
		ENSURE_OBJECT_AT_INDEX(function, 0);
		ENSURE_OPTIONAL_UINT_AT_INDEX(delay, 1, 0);

		const auto chrono_delay = std::chrono::milliseconds(static_cast<std::chrono::milliseconds::rep>(delay));

		const auto global_object = this_object.get_context().get_global_object();
		const auto global_ptr = global_object.GetPrivate<GlobalObject>();
		TITANIUM_ASSERT(global_ptr);

		return this_object.get_context().CreateNumber(global_ptr->setTimeout(std::move(function), chrono_delay));
	}

	TITANIUM_FUNCTION(GlobalObject, clearTimeout)
	{
		ENSURE_UINT_AT_INDEX(timerId, 0);

		const auto global_object = this_object.get_context().get_global_object();
		const auto global_ptr = global_object.GetPrivate<GlobalObject>();
		TITANIUM_ASSERT(global_ptr);

		global_ptr->clearTimeout(timerId);

		return this_object.get_context().CreateUndefined();
	}

	TITANIUM_FUNCTION(GlobalObject, setInterval)
	{
		ENSURE_ARGUMENT_INDEX(1);
		ENSURE_OBJECT_AT_INDEX(function, 0);
		ENSURE_UINT_AT_INDEX(delay, 1);

		TITANIUM_ASSERT(function.IsFunction());
		const auto chrono_delay = std::chrono::milliseconds(static_cast<std::chrono::milliseconds::rep>(delay));

		const auto global_object = this_object.get_context().get_global_object();
		const auto global_ptr = global_object.GetPrivate<GlobalObject>();
		TITANIUM_ASSERT(global_ptr);

		return this_object.get_context().CreateNumber(global_ptr->setInterval(std::move(function), chrono_delay));
	}

	TITANIUM_FUNCTION(GlobalObject, clearInterval)
	{
		ENSURE_UINT_AT_INDEX(timerId, 0);

		const auto global_object = this_object.get_context().get_global_object();
		const auto global_ptr = global_object.GetPrivate<GlobalObject>();
		TITANIUM_ASSERT(global_ptr);

		global_ptr->clearInterval(timerId);

		return this_object.get_context().CreateUndefined();
	}

}  // namespace Titanium

ext: cpp