AutoGetOrCreateDb::AutoGetOrCreateDb(OperationContext* opCtx, StringData ns, LockMode mode)
    : _dbLock(opCtx, ns, mode), _db(dbHolder().get(opCtx, ns)) {
    invariant(mode == MODE_IX || mode == MODE_X);
    _justCreated = false;
    // If the database didn't exist, relock in MODE_X
    if (_db == NULL) {
        if (mode != MODE_X) {
            _dbLock.relockWithMode(MODE_X);
        }
        _db = dbHolder().openDb(opCtx, ns);
        _justCreated = true;
    }
}

void profile(OperationContext* txn, const Client& c, int op, CurOp& currentOp) {
    // initialize with 1kb to start, to avoid realloc later
    // doing this outside the dblock to improve performance
    BufBuilder profileBufBuilder(1024);

    try {
        // NOTE: It's kind of weird that we lock the op's namespace, but have to for now since
        // we're sometimes inside the lock already
        Lock::DBWrite lk( currentOp.getNS() );
        if (dbHolder()._isLoaded(nsToDatabase(currentOp.getNS()), storageGlobalParams.dbpath)) {
            Client::Context cx(currentOp.getNS(), storageGlobalParams.dbpath, false);
            _profile(txn, c, cx.db(),
                     currentOp, profileBufBuilder);
        }
        else {
            mongo::log() << "note: not profiling because db went away - probably a close on: "
                         << currentOp.getNS() << endl;
        }
    }
    catch (const AssertionException& assertionEx) {
        warning() << "Caught Assertion while trying to profile " << opToString(op)
                  << " against " << currentOp.getNS()
                  << ": " << assertionEx.toString() << endl;
    }
}

    /*static*/
    string Database::duplicateUncasedName(const string &name, set< string > *duplicates) {
        if ( duplicates ) {
            duplicates->clear();
        }

        vector<string> others;
        StorageEngine* storageEngine = getGlobalEnvironment()->getGlobalStorageEngine();
        storageEngine->listDatabases(&others);

        set<string> allShortNames;
        dbHolder().getAllShortNames(allShortNames);

        others.insert( others.end(), allShortNames.begin(), allShortNames.end() );

        for ( unsigned i=0; i<others.size(); i++ ) {

            if ( strcasecmp( others[i].c_str() , name.c_str() ) )
                continue;

            if ( strcmp( others[i].c_str() , name.c_str() ) == 0 )
                continue;

            if ( duplicates ) {
                duplicates->insert( others[i] );
            } else {
                return others[i];
            }
        }
        if ( duplicates ) {
            return duplicates->empty() ? "" : *duplicates->begin();
        }
        return "";
    }

void dropAllDatabasesExceptLocal(OperationContext* txn) {
    ScopedTransaction transaction(txn, MODE_X);
    Lock::GlobalWrite lk(txn->lockState());

    vector<string> n;
    StorageEngine* storageEngine = getGlobalServiceContext()->getGlobalStorageEngine();
    storageEngine->listDatabases(&n);

    if (n.size() == 0)
        return;
    log() << "dropAllDatabasesExceptLocal " << n.size();

    repl::getGlobalReplicationCoordinator()->dropAllSnapshots();
    for (vector<string>::iterator i = n.begin(); i != n.end(); i++) {
        if (*i != "local") {
            MONGO_WRITE_CONFLICT_RETRY_LOOP_BEGIN {
                Database* db = dbHolder().get(txn, *i);
                // This is needed since dropDatabase can't be rolled back.
                // This is safe be replaced by "invariant(db);dropDatabase(txn, db);" once fixed
                if (db == nullptr) {
                    log() << "database disappeared after listDatabases but before drop: " << *i;
                } else {
                    Database::dropDatabase(txn, db);
                }
            }
            MONGO_WRITE_CONFLICT_RETRY_LOOP_END(txn, "dropAllDatabasesExceptLocal", *i);
        }
    }

/*static*/
string Database::duplicateUncasedName( bool inholderlock, const string &name, const string &path, set< string > *duplicates ) {
    Lock::assertAtLeastReadLocked(name);

    if ( duplicates ) {
        duplicates->clear();
    }

    vector<string> others;
    getDatabaseNames( others , path );

    set<string> allShortNames;
    dbHolder().getAllShortNames( allShortNames );

    others.insert( others.end(), allShortNames.begin(), allShortNames.end() );

    for ( unsigned i=0; i<others.size(); i++ ) {

        if ( strcasecmp( others[i].c_str() , name.c_str() ) )
            continue;

        if ( strcmp( others[i].c_str() , name.c_str() ) == 0 )
            continue;

        if ( duplicates ) {
            duplicates->insert( others[i] );
        } else {
            return others[i];
        }
    }
    if ( duplicates ) {
        return duplicates->empty() ? "" : *duplicates->begin();
    }
    return "";
}

    void IndexBuilder::run() {
        Client::initThread(name().c_str());
        LOG(2) << "IndexBuilder building index " << _index;

        OperationContextImpl txn;

        Lock::ParallelBatchWriterMode::iAmABatchParticipant(txn.lockState());

        txn.getClient()->getAuthorizationSession()->grantInternalAuthorization();

        txn.getCurOp()->reset(HostAndPort(), dbInsert);
        NamespaceString ns(_index["ns"].String());

        ScopedTransaction transaction(&txn, MODE_IX);
        Lock::DBLock dlk(txn.lockState(), ns.db(), MODE_X);
        Client::Context ctx(&txn, ns.getSystemIndexesCollection());

        Database* db = dbHolder().get(&txn, ns.db().toString());

        Status status = _build(&txn, db, true, &dlk);
        if ( !status.isOK() ) {
            error() << "IndexBuilder could not build index: " << status.toString();
            fassert(28555, ErrorCodes::isInterruption(status.code()));
        }

        txn.getClient()->shutdown();
    }

    void AutoGetCollectionForRead::_init() {
        massert(28535, "need a non-empty collection name", !_nss.coll().empty());

        // TODO: Client::Context legacy, needs to be removed
        _txn->getCurOp()->ensureStarted();
        _txn->getCurOp()->setNS(_nss.toString());

        // Lock both the DB and the collection (DB is locked in the constructor), because this is
        // necessary in order to to shard version checking.
        const ResourceId resId(RESOURCE_COLLECTION, _nss);
        const LockMode collLockMode = supportsDocLocking() ? MODE_IS : MODE_S;

        invariant(LOCK_OK == _txn->lockState()->lock(resId, collLockMode));

        // Shard version check needs to be performed under the collection lock
        ensureShardVersionOKOrThrow(_nss);

        // At this point, we are locked in shared mode for the database by the DB lock in the
        // constructor, so it is safe to load the DB pointer.
        _db = dbHolder().get(_txn, _nss.db());
        if (_db != NULL) {
            // TODO: Client::Context legacy, needs to be removed
            _txn->getCurOp()->enter(_nss.toString().c_str(), _db->getProfilingLevel());

            _coll = _db->getCollection(_txn, _nss);
        }
    }

    void profile(OperationContext* txn, const Client& c, int op, CurOp& currentOp) {
        // initialize with 1kb to start, to avoid realloc later
        // doing this outside the dblock to improve performance
        BufBuilder profileBufBuilder(1024);

        try {
            // NOTE: It's kind of weird that we lock the op's namespace, but have to for now since
            // we're sometimes inside the lock already
            Lock::DBWrite lk(txn->lockState(), currentOp.getNS() );
            if (dbHolder().get(txn, nsToDatabase(currentOp.getNS())) != NULL) {
                // We are ok with the profiling happening in a different WUOW from the actual op.
                WriteUnitOfWork wunit(txn->recoveryUnit());
                Client::Context cx(txn, currentOp.getNS(), false);
                _profile(txn, c, cx.db(), currentOp, profileBufBuilder);
                wunit.commit();
            }
            else {
                mongo::log() << "note: not profiling because db went away - probably a close on: "
                             << currentOp.getNS() << endl;
            }
        }
        catch (const AssertionException& assertionEx) {
            warning() << "Caught Assertion while trying to profile " << opToString(op)
                      << " against " << currentOp.getNS()
                      << ": " << assertionEx.toString() << endl;
        }
    }

/*static*/
string Database::duplicateUncasedName(const string& name, set<string>* duplicates) {
    if (duplicates) {
        duplicates->clear();
    }

    set<string> allShortNames;
    dbHolder().getAllShortNames(allShortNames);

    for (const auto& dbname : allShortNames) {
        if (strcasecmp(dbname.c_str(), name.c_str()))
            continue;

        if (strcmp(dbname.c_str(), name.c_str()) == 0)
            continue;

        if (duplicates) {
            duplicates->insert(dbname);
        } else {
            return dbname;
        }
    }
    if (duplicates) {
        return duplicates->empty() ? "" : *duplicates->begin();
    }
    return "";
}

        bool run(OperationContext* txn,
                 const string& dbname,
                 BSONObj& jsobj,
                 int,
                 string& errmsg,
                 BSONObjBuilder& result,
                 bool /*fromRepl*/) {

            Lock::DBRead lk( txn->lockState(), dbname );

            const Database* d = dbHolder().get( txn, dbname );
            const DatabaseCatalogEntry* dbEntry = NULL;

            list<string> names;
            if ( d ) {
                dbEntry = d->getDatabaseCatalogEntry();
                dbEntry->getCollectionNamespaces( &names );
                names.sort();
            }

            scoped_ptr<MatchExpression> matcher;
            if ( jsobj["filter"].isABSONObj() ) {
                StatusWithMatchExpression parsed =
                    MatchExpressionParser::parse( jsobj["filter"].Obj() );
                if ( !parsed.isOK() ) {
                    return appendCommandStatus( result, parsed.getStatus() );
                }
                matcher.reset( parsed.getValue() );
            }

            BSONArrayBuilder arr;

            for ( list<string>::const_iterator i = names.begin(); i != names.end(); ++i ) {
                string ns = *i;

                StringData collection = nsToCollectionSubstring( ns );
                if ( collection == "system.namespaces" ) {
                    continue;
                }

                BSONObjBuilder b;
                b.append( "name", collection );

                CollectionOptions options =
                    dbEntry->getCollectionCatalogEntry( txn, ns )->getCollectionOptions(txn);
                b.append( "options", options.toBSON() );

                BSONObj maybe = b.obj();
                if ( matcher && !matcher->matchesBSON( maybe ) ) {
                    continue;
                }

                arr.append( maybe );
            }

            result.append( "collections", arr.arr() );

            return true;
        }

    void Client::Context::_finishInit() {
        _db = dbHolder().openDb(_txn, _ns, &_justCreated);
        invariant(_db);

        if( _doVersion ) checkNotStale();

        _client->_curOp->enter(_ns.c_str(), _db->getProfilingLevel());
    }

    void Client::Context::_finishInit() {
        _db = dbHolder().getOrCreate(_txn, _ns, _justCreated);
        invariant(_db);

        if( _doVersion ) checkNotStale();

        _client->_curOp->enter( this );
    }

    void TxnCompleteHooksImpl::noteTxnAbortedFileOps(const set<string> &namespaces, const set<string> &dbs) {
        for (set<string>::const_iterator i = namespaces.begin(); i != namespaces.end(); i++) {
            const char *ns = i->c_str();

            // We cannot be holding a read lock at this point, since we're in one of two situations:
            // - Single-statement txn is aborting. If it did fileops, it had to hold a write lock,
            //   and therefore it still is.
            // - Multi-statement txn is aborting. The only way to do this is through a command that
            //   takes no lock, therefore we're not read locked.
            verify(!Lock::isReadLocked());

            // If something is already write locked we must be in the single-statement case, so
            // assert that the write locked namespace is this one.
            if (Lock::somethingWriteLocked()) {
                verify(Lock::isWriteLocked(ns));
            }

            // The ydb requires that a txn closes any dictionaries it created beforeaborting.
            // Hold a write lock while trying to close the namespace in the nsindex.
            Lock::DBWrite lk(ns);
            if (dbHolder().__isLoaded(ns, dbpath)) {
                scoped_ptr<Client::Context> ctx(cc().getContext() == NULL ?
                                                new Client::Context(ns) : NULL);
                // Pass aborting = true to close_ns(), which hints to the implementation
                // that the calling transaction is about to abort.
                (void) nsindex(ns)->close_ns(ns, true);
            }
        }

        for (set<string>::const_iterator it = dbs.begin(); it != dbs.end(); ++it) {
            const string &db = *it;

            // The same locking rules above apply here.
            verify(!Lock::isReadLocked());
            if (Lock::somethingWriteLocked()) {
                verify(Lock::isWriteLocked(db));
            }

            Lock::DBWrite lk(db);
            if (dbHolder().__isLoaded(db, dbpath)) {
                scoped_ptr<Client::Context> ctx(cc().getContext() == NULL ?
                                                new Client::Context(db) : NULL);
                nsindex(db.c_str())->rollbackCreate();
            }
        }
    }

 // Profile the current op in an alternate transaction
 void lockedDoProfile(const Client& c, int op, CurOp& currentOp) {
     if ( dbHolder().__isLoaded( nsToDatabase( currentOp.getNS() ) , dbpath ) ) {
         Client::Context ctx(currentOp.getNS(), dbpath);
         Client::AlternateTransactionStack altStack;
         Client::Transaction txn(DB_SERIALIZABLE);
         profile(c, op, currentOp);
         txn.commit();
     }
 }

    void Client::Context::_finishInit() {
        OperationContextImpl txn; // TODO get rid of this once reads require transactions
        _db = dbHolder().getOrCreate(&txn, _ns, _justCreated);
        invariant(_db);

        if( _doVersion ) checkNotStale();

        _client->_curOp->enter( this );
    }

TEST_F(SyncTailTest, SyncApplyInsertDocumentCollectionMissing) {
    {
        Lock::GlobalWrite globalLock(_txn->lockState());
        bool justCreated = false;
        Database* db = dbHolder().openDb(_txn.get(), "test", &justCreated);
        ASSERT_TRUE(db);
        ASSERT_TRUE(justCreated);
    }
    _testSyncApplyInsertDocument(MODE_X);
}

        bool run(OperationContext* txn, const string& dbname , BSONObj& jsobj, int, string& errmsg, BSONObjBuilder& result, bool /*fromRepl*/) {
            vector< string > dbNames;
            globalStorageEngine->listDatabases( &dbNames );

            vector< BSONObj > dbInfos;

            set<string> seen;
            intmax_t totalSize = 0;
            for ( vector< string >::iterator i = dbNames.begin(); i != dbNames.end(); ++i ) {
                BSONObjBuilder b;
                b.append( "name", *i );

                intmax_t size = dbSize( i->c_str() );
                b.append( "sizeOnDisk", (double) size );
                totalSize += size;

                {
                    Client::ReadContext rc(txn, *i );
                    b.appendBool( "empty", rc.ctx().db()->getDatabaseCatalogEntry()->isEmpty() );
                }

                dbInfos.push_back( b.obj() );

                seen.insert( i->c_str() );
            }

            set<string> allShortNames;
            {
                Lock::GlobalRead lk(txn->lockState());
                dbHolder().getAllShortNames(allShortNames);
            }

            for ( set<string>::iterator i = allShortNames.begin(); i != allShortNames.end(); i++ ) {
                string name = *i;

                if ( seen.count( name ) )
                    continue;

                BSONObjBuilder b;
                b.append( "name" , name );
                b.append( "sizeOnDisk" , (double)1.0 );

                {
                    Client::ReadContext ctx(txn, name);
                    b.appendBool( "empty", ctx.ctx().db()->getDatabaseCatalogEntry()->isEmpty() );
                }

                dbInfos.push_back( b.obj() );
            }

            result.append( "databases", dbInfos );
            result.append( "totalSize", double( totalSize ) );
            return true;
        }

/**
 * Due to SERVER-23274, versions 3.2.0 through 3.2.4 of MongoDB incorrectly mark the final output
 * collections of aggregations with $out stages as temporary on most replica set secondaries. Rather
 * than risk deleting collections that the user did not intend to be temporary when newer nodes
 * start up or get promoted to be replica set primaries, newer nodes clear the temp flags left by
 * these versions.
 */
bool isSubjectToSERVER23299(OperationContext* txn) {
    // We are already called under global X lock as part of the startup sequence
    invariant(txn->lockState()->isW());

    if (storageGlobalParams.readOnly) {
        return false;
    }

    // Ensure that the local database is open since we are still early in the server startup
    // sequence
    dbHolder().openDb(txn, startupLogCollectionName.db());

    // Only used as a shortcut to obtain a reference to the startup log collection
    AutoGetCollection autoColl(txn, startupLogCollectionName, MODE_IS);

    // No startup log or an empty one means either that the user was not running an affected
    // version, or that they manually deleted the startup collection since they last started an
    // affected version.
    LOG(1) << "Checking node for SERVER-23299 eligibility";
    if (!autoColl.getCollection()) {
        LOG(1) << "Didn't find " << startupLogCollectionName;
        return false;
    }
    LOG(1) << "Checking node for SERVER-23299 applicability - reading startup log";
    BSONObj lastStartupLogDoc;
    if (!Helpers::getLast(txn, startupLogCollectionName.ns().c_str(), lastStartupLogDoc)) {
        return false;
    }
    std::vector<int> versionComponents;
    try {
        for (auto elem : lastStartupLogDoc["buildinfo"]["versionArray"].Obj()) {
            versionComponents.push_back(elem.Int());
        }
        uassert(40050,
                str::stream() << "Expected three elements in buildinfo.versionArray; found "
                              << versionComponents.size(),
                versionComponents.size() >= 3);
    } catch (const DBException& ex) {
        log() << "Last entry of " << startupLogCollectionName
              << " has no well-formed  buildinfo.versionArray field; ignoring " << causedBy(ex);
        return false;
    }
    LOG(1)
        << "Checking node for SERVER-23299 applicability - checking version 3.2.x for x in [0, 4]";
    if (versionComponents[0] != 3)
        return false;
    if (versionComponents[1] != 2)
        return false;
    if (versionComponents[2] > 4)
        return false;
    LOG(1) << "Node eligible for SERVER-23299";
    return true;
}

    void dropCollection() {
        ScopedTransaction transaction(&_txn, MODE_X);
        Lock::DBLock dbLock(_txn.lockState(), nss.db(), MODE_X);
        Database* database = dbHolder().get(&_txn, nss.db());
        if (!database) {
            return;
        }

        WriteUnitOfWork wuow(&_txn);
        database->dropCollection(&_txn, nss.ns());
        wuow.commit();
    }

    TEST(DBHelperTests, FindDiskLocs) {

        DBDirectClient client;
        OperationContextImpl txn;

        // Some unique tag we can use to make sure we're pulling back the right data
        OID tag = OID::gen();
        client.remove( ns, BSONObj() );

        int numDocsInserted = 10;
        for ( int i = 0; i < numDocsInserted; ++i ) {
            client.insert( ns, BSON( "_id" << i << "tag" << tag ) );
        }

        long long maxSizeBytes = 1024 * 1024 * 1024;

        set<DiskLoc> locs;
        long long numDocsFound;
        long long estSizeBytes;
        {
            // search _id range (0, 10)
            Lock::DBRead lk(txn.lockState(), ns);

            KeyRange range( ns,
                            BSON( "_id" << 0 ),
                            BSON( "_id" << numDocsInserted ),
                            BSON( "_id" << 1 ) );

            Status result = Helpers::getLocsInRange( &txn,
                                                     range,
                                                     maxSizeBytes,
                                                     &locs,
                                                     &numDocsFound,
                                                     &estSizeBytes );

            ASSERT_EQUALS( result, Status::OK() );
            ASSERT_EQUALS( numDocsFound, numDocsInserted );
            ASSERT_NOT_EQUALS( estSizeBytes, 0 );
            ASSERT_LESS_THAN( estSizeBytes, maxSizeBytes );

            Database* db = dbHolder().get(
                                    &txn, nsToDatabase(range.ns), storageGlobalParams.dbpath);
            const Collection* collection = db->getCollection(&txn, ns);

            // Make sure all the disklocs actually correspond to the right info
            for ( set<DiskLoc>::const_iterator it = locs.begin(); it != locs.end(); ++it ) {
                const BSONObj obj = collection->docFor(*it);
                ASSERT_EQUALS(obj["tag"].OID(), tag);
            }
        }
    }