diff options
Diffstat (limited to 'src/mongo')
30 files changed, 328 insertions, 217 deletions
diff --git a/src/mongo/db/SConscript b/src/mongo/db/SConscript index 204dcfe02fc..4128794d1c1 100644 --- a/src/mongo/db/SConscript +++ b/src/mongo/db/SConscript @@ -594,12 +594,14 @@ env.Library( 'curop_metrics', 'diag_log', 'lasterror', - 'write_ops', 'ops/write_ops_parsers', 'rw_concern_d', 's/sharding', 'storage/storage_options', ], + LIBDEPS_PRIVATE=[ + 'ops/write_ops_exec', + ], ) env.Library( @@ -735,6 +737,9 @@ env.Library( '$BUILD_DIR/mongo/db/catalog/database', "$BUILD_DIR/mongo/db/storage/mmap_v1/storage_mmapv1", ], + LIBDEPS_PRIVATE=[ + "$BUILD_DIR/mongo/db/repl/oplog", + ], ) env.Library( @@ -1505,7 +1510,6 @@ env.Library( 'ops/update.cpp', 'ops/update_lifecycle_impl.cpp', 'ops/update_result.cpp', - 'ops/write_ops_exec.cpp', 'ops/write_ops_retryability.cpp', 'session.cpp', 'session_catalog.cpp', diff --git a/src/mongo/db/catalog/SConscript b/src/mongo/db/catalog/SConscript index 1a06fe1ba80..6480a091972 100644 --- a/src/mongo/db/catalog/SConscript +++ b/src/mongo/db/catalog/SConscript @@ -229,22 +229,23 @@ env.Library( 'index_create', 'index_key_validate', '$BUILD_DIR/mongo/base', + '$BUILD_DIR/mongo/db/background', '$BUILD_DIR/mongo/db/clientcursor', + '$BUILD_DIR/mongo/db/collection_index_usage_tracker', '$BUILD_DIR/mongo/db/concurrency/lock_manager', '$BUILD_DIR/mongo/db/curop', + '$BUILD_DIR/mongo/db/db_raii', + '$BUILD_DIR/mongo/db/index/index_access_methods', '$BUILD_DIR/mongo/db/query/query', '$BUILD_DIR/mongo/db/repl/drop_pending_collection_reaper', + '$BUILD_DIR/mongo/db/repl/oplog', '$BUILD_DIR/mongo/db/repl/serveronly', + '$BUILD_DIR/mongo/db/s/balancer', '$BUILD_DIR/mongo/db/service_context', - '$BUILD_DIR/mongo/db/storage/mmap_v1/storage_mmapv1', '$BUILD_DIR/mongo/db/storage/key_string', + '$BUILD_DIR/mongo/db/storage/mmap_v1/storage_mmapv1', '$BUILD_DIR/mongo/db/system_index', '$BUILD_DIR/mongo/db/ttl_collection_cache', - '$BUILD_DIR/mongo/db/collection_index_usage_tracker', - '$BUILD_DIR/mongo/db/background', - '$BUILD_DIR/mongo/db/db_raii', - '$BUILD_DIR/mongo/db/index/index_access_methods', - '$BUILD_DIR/mongo/db/s/balancer', '$BUILD_DIR/mongo/db/views/views_mongod', ], ) diff --git a/src/mongo/db/catalog/collection.h b/src/mongo/db/catalog/collection.h index df30097f6ca..fd0f9b3b1b4 100644 --- a/src/mongo/db/catalog/collection.h +++ b/src/mongo/db/catalog/collection.h @@ -48,6 +48,7 @@ #include "mongo/db/op_observer.h" #include "mongo/db/query/collation/collator_interface.h" #include "mongo/db/record_id.h" +#include "mongo/db/repl/oplog.h" #include "mongo/db/storage/capped_callback.h" #include "mongo/db/storage/record_store.h" #include "mongo/db/storage/snapshot.h" @@ -96,21 +97,6 @@ struct CompactStats { long long corruptDocuments = 0; }; -struct InsertStatement { -public: - InsertStatement() = default; - explicit InsertStatement(BSONObj toInsert) : doc(toInsert) {} - - InsertStatement(StmtId statementId, BSONObj toInsert) : stmtId(statementId), doc(toInsert) {} - InsertStatement(StmtId statementId, BSONObj toInsert, SnapshotName ts) - : stmtId(statementId), timestamp(ts), doc(toInsert) {} - InsertStatement(BSONObj toInsert, SnapshotName ts) : timestamp(ts), doc(toInsert) {} - - StmtId stmtId = kUninitializedStmtId; - SnapshotName timestamp = SnapshotName(); - BSONObj doc; -}; - /** * Queries with the awaitData option use this notifier object to wait for more data to be * inserted into the capped collection. diff --git a/src/mongo/db/catalog/collection_impl.cpp b/src/mongo/db/catalog/collection_impl.cpp index 5896b3b9f46..d78c3e6f43d 100644 --- a/src/mongo/db/catalog/collection_impl.cpp +++ b/src/mongo/db/catalog/collection_impl.cpp @@ -170,7 +170,6 @@ CollectionImpl::CollectionImpl(Collection* _this_init, _cursorManager(_ns), _cappedNotifier(_recordStore->isCapped() ? stdx::make_unique<CappedInsertNotifier>() : nullptr), - _mustTakeCappedLockOnInsert(isCapped() && !_ns.isSystemDotProfile() && !_ns.isOplog()), _this(_this_init) {} void CollectionImpl::init(OperationContext* opCtx) { @@ -305,7 +304,6 @@ Status CollectionImpl::insertDocumentsForOplog(OperationContext* opCtx, // because it would defeat the purpose of using DocWriter. invariant(!_validator); invariant(!_indexCatalog.haveAnyIndexes()); - invariant(!_mustTakeCappedLockOnInsert); Status status = _recordStore->insertRecordsWithDocWriter(opCtx, docs, timestamps, nDocs); if (!status.isOK()) @@ -355,9 +353,6 @@ Status CollectionImpl::insertDocuments(OperationContext* opCtx, const SnapshotId sid = opCtx->recoveryUnit()->getSnapshotId(); - if (_mustTakeCappedLockOnInsert) - synchronizeOnCappedInFlightResource(opCtx->lockState(), _ns); - Status status = _insertDocuments(opCtx, begin, end, enforceQuota, opDebug); if (!status.isOK()) return status; @@ -407,8 +402,6 @@ Status CollectionImpl::insertDocument(OperationContext* opCtx, dassert(opCtx->lockState()->isCollectionLockedForMode(ns().toString(), MODE_IX)); - if (_mustTakeCappedLockOnInsert) - synchronizeOnCappedInFlightResource(opCtx->lockState(), _ns); // TODO SERVER-30638: using timestamp 0 for these inserts, which are non-oplog so we don't yet // care about their correct timestamps. StatusWith<RecordId> loc = _recordStore->insertRecord( @@ -425,7 +418,14 @@ Status CollectionImpl::insertDocument(OperationContext* opCtx, } vector<InsertStatement> inserts; - inserts.emplace_back(doc); + OplogSlot slot; + // Fetch a new optime now, if necessary. + auto replCoord = repl::ReplicationCoordinator::get(opCtx); + if (!replCoord->isOplogDisabledFor(opCtx, _ns)) { + // Populate 'slot' with a new optime. + repl::getNextOpTimes(opCtx, 1, &slot); + } + inserts.emplace_back(kUninitializedStmtId, doc, slot); getGlobalServiceContext()->getOpObserver()->onInserts( opCtx, ns(), uuid(), inserts.begin(), inserts.end(), false); @@ -468,7 +468,7 @@ Status CollectionImpl::_insertDocuments(OperationContext* opCtx, for (auto it = begin; it != end; it++) { Record record = {RecordId(), RecordData(it->doc.objdata(), it->doc.objsize())}; records.push_back(record); - Timestamp timestamp = Timestamp(it->timestamp.asU64()); + Timestamp timestamp = Timestamp(it->oplogSlot.opTime.getTimestamp()); timestamps.push_back(timestamp); } Status status = diff --git a/src/mongo/db/catalog/collection_impl.h b/src/mongo/db/catalog/collection_impl.h index 4a66cb1a8bc..f8db017330e 100644 --- a/src/mongo/db/catalog/collection_impl.h +++ b/src/mongo/db/catalog/collection_impl.h @@ -464,8 +464,6 @@ private: // This is non-null if and only if the collection is a capped collection. const std::shared_ptr<CappedInsertNotifier> _cappedNotifier; - const bool _mustTakeCappedLockOnInsert; - // The earliest snapshot that is allowed to use this collection. boost::optional<SnapshotName> _minVisibleSnapshot; diff --git a/src/mongo/db/catalog/database_impl.cpp b/src/mongo/db/catalog/database_impl.cpp index d57e4352b90..6096e81106f 100644 --- a/src/mongo/db/catalog/database_impl.cpp +++ b/src/mongo/db/catalog/database_impl.cpp @@ -161,7 +161,7 @@ void DatabaseImpl::close(OperationContext* opCtx, const std::string& reason) { // XXX? - Do we need to close database under global lock or just DB-lock is sufficient ? invariant(opCtx->lockState()->isW()); - // oplog caches some things, dirty its caches + // Clear cache of oplog Collection pointer. repl::oplogCheckCloseDatabase(opCtx, this->_this); if (BackgroundOperation::inProgForDb(_name)) { diff --git a/src/mongo/db/commands/SConscript b/src/mongo/db/commands/SConscript index 527ae13670c..3efcaa19f17 100644 --- a/src/mongo/db/commands/SConscript +++ b/src/mongo/db/commands/SConscript @@ -216,6 +216,7 @@ env.Library( '$BUILD_DIR/mongo/db/ops/write_ops_parsers', '$BUILD_DIR/mongo/db/pipeline/serveronly', '$BUILD_DIR/mongo/db/repair_database', + '$BUILD_DIR/mongo/db/repl/dbcheck', '$BUILD_DIR/mongo/db/repl/oplog', '$BUILD_DIR/mongo/db/repl/isself', '$BUILD_DIR/mongo/db/repl/repl_coordinator_impl', @@ -232,6 +233,9 @@ env.Library( 'killcursors_common', 'write_commands_common', ], + LIBDEPS_PRIVATE=[ + '$BUILD_DIR/mongo/db/ops/write_ops_exec', + ], ) env.Library( diff --git a/src/mongo/db/concurrency/d_concurrency.cpp b/src/mongo/db/concurrency/d_concurrency.cpp index d5d9fe1684c..0876f5f4f8e 100644 --- a/src/mongo/db/concurrency/d_concurrency.cpp +++ b/src/mongo/db/concurrency/d_concurrency.cpp @@ -311,20 +311,9 @@ void Lock::ResourceLock::unlock() { } } -void synchronizeOnCappedInFlightResource(Locker* lockState, const NamespaceString& cappedNs) { +void synchronizeOnOplogInFlightResource(Locker* lockState) { dassert(lockState->inAWriteUnitOfWork()); - const ResourceId resource = cappedNs.db() == "local" ? resourceCappedInFlightForLocalDb - : resourceCappedInFlightForOtherDb; - - // It is illegal to acquire the capped in-flight lock for non-local dbs while holding the - // capped in-flight lock for the local db. (Unless we already hold the otherDb lock since - // reacquiring a lock in the same mode never blocks.) - if (resource == resourceCappedInFlightForOtherDb) { - dassert(!lockState->isLockHeldForMode(resourceCappedInFlightForLocalDb, MODE_IX) || - lockState->isLockHeldForMode(resourceCappedInFlightForOtherDb, MODE_IX)); - } - - Lock::ResourceLock heldUntilEndOfWUOW{lockState, resource, MODE_IX}; + Lock::ResourceLock heldUntilEndOfWUOW{lockState, resourceInFlightForOplog, MODE_IX}; } } // namespace mongo diff --git a/src/mongo/db/concurrency/d_concurrency.h b/src/mongo/db/concurrency/d_concurrency.h index 44be439d539..6b0cf4e9547 100644 --- a/src/mongo/db/concurrency/d_concurrency.h +++ b/src/mongo/db/concurrency/d_concurrency.h @@ -384,8 +384,8 @@ public: /** * Takes a lock on resourceCappedInFlight in MODE_IX which will be held until the end of your - * WUOW. This ensures that a MODE_X lock on this resource will wait for all in-flight capped + * WUOW. This ensures that a MODE_X lock on this resource will wait for all in-flight oplog * inserts to either commit or rollback and block new ones from starting. */ -void synchronizeOnCappedInFlightResource(Locker* opCtx, const NamespaceString& cappedNs); +void synchronizeOnOplogInFlightResource(Locker* opCtx); } diff --git a/src/mongo/db/concurrency/lock_manager_defs.h b/src/mongo/db/concurrency/lock_manager_defs.h index e1b72f7b1df..5951955eeb1 100644 --- a/src/mongo/db/concurrency/lock_manager_defs.h +++ b/src/mongo/db/concurrency/lock_manager_defs.h @@ -188,8 +188,7 @@ public: SINGLETON_PARALLEL_BATCH_WRITER_MODE, SINGLETON_GLOBAL, SINGLETON_MMAPV1_FLUSH, - SINGLETON_CAPPED_IN_FLIGHT_OTHER_DB, - SINGLETON_CAPPED_IN_FLIGHT_LOCAL_DB, + SINGLETON_IN_FLIGHT_OPLOG, }; ResourceId() : _fullHash(0) {} @@ -265,17 +264,12 @@ extern const ResourceId resourceIdAdminDB; // TODO: Merge this with resourceIdGlobal extern const ResourceId resourceIdParallelBatchWriterMode; -// Everywhere that starts in-flight capped inserts which allocate capped collection RecordIds in -// a way that could trigger hiding of newer records takes this lock in MODE_IX and holds it -// until the end of their WriteUnitOfWork. The localDb resource is for capped collections in the -// local database (including the oplog). The otherDb resource is for capped collections in any other -// database. +// Every place that starts oplog inserts takes this lock in MODE_IX and holds it +// until the end of their WriteUnitOfWork. // -// Threads that need a consistent view of the world can lock both of these in MODE_X to prevent -// concurrent in-flight capped inserts. The otherDb resource must be acquired before the localDb -// resource. -extern const ResourceId resourceCappedInFlightForLocalDb; -extern const ResourceId resourceCappedInFlightForOtherDb; +// Threads that need a consistent view of the world can lock this in MODE_X to prevent +// concurrent in-flight oplog inserts. +extern const ResourceId resourceInFlightForOplog; /** * Interface on which granted lock requests will be notified. See the contract for the notify diff --git a/src/mongo/db/concurrency/lock_state.cpp b/src/mongo/db/concurrency/lock_state.cpp index 721067e3aad..5409c1190f6 100644 --- a/src/mongo/db/concurrency/lock_state.cpp +++ b/src/mongo/db/concurrency/lock_state.cpp @@ -964,9 +964,7 @@ const ResourceId resourceIdOplog = ResourceId(RESOURCE_COLLECTION, StringData("l const ResourceId resourceIdAdminDB = ResourceId(RESOURCE_DATABASE, StringData("admin")); const ResourceId resourceIdParallelBatchWriterMode = ResourceId(RESOURCE_GLOBAL, ResourceId::SINGLETON_PARALLEL_BATCH_WRITER_MODE); -const ResourceId resourceCappedInFlightForLocalDb = - ResourceId(RESOURCE_METADATA, ResourceId::SINGLETON_CAPPED_IN_FLIGHT_LOCAL_DB); -const ResourceId resourceCappedInFlightForOtherDb = - ResourceId(RESOURCE_METADATA, ResourceId::SINGLETON_CAPPED_IN_FLIGHT_OTHER_DB); +const ResourceId resourceInFlightForOplog = + ResourceId(RESOURCE_METADATA, ResourceId::SINGLETON_IN_FLIGHT_OPLOG); } // namespace mongo diff --git a/src/mongo/db/keys_collection_manager_sharding_test.cpp b/src/mongo/db/keys_collection_manager_sharding_test.cpp index d7af13a09fc..d7c88d097f3 100644 --- a/src/mongo/db/keys_collection_manager_sharding_test.cpp +++ b/src/mongo/db/keys_collection_manager_sharding_test.cpp @@ -63,6 +63,10 @@ protected: serverGlobalParams.featureCompatibility.validateFeaturesAsMaster.store(true); auto clockSource = stdx::make_unique<ClockSourceMock>(); + // Timestamps of "0 seconds" are not allowed, so we must advance our clock mock to the first + // real second. + clockSource->advance(Seconds(1)); + operationContext()->getServiceContext()->setFastClockSource(std::move(clockSource)); auto catalogClient = Grid::get(operationContext())->catalogClient(); _keyManager = diff --git a/src/mongo/db/logical_clock_test.cpp b/src/mongo/db/logical_clock_test.cpp index bf3b78a3062..4dd64809a86 100644 --- a/src/mongo/db/logical_clock_test.cpp +++ b/src/mongo/db/logical_clock_test.cpp @@ -133,7 +133,7 @@ TEST_F(LogicalClockTest, InitFromTrustedSourceCanAcceptVeryOldLogicalTime) { // Verify writes to the oplog advance cluster time. TEST_F(LogicalClockTest, WritesToOplogAdvanceClusterTime) { - Timestamp tX(1); + Timestamp tX(1, 0); auto initialTime = LogicalTime(tX); getClock()->setClusterTimeFromTrustedSource(initialTime); diff --git a/src/mongo/db/ops/SConscript b/src/mongo/db/ops/SConscript index 2254606f570..beadfc36549 100644 --- a/src/mongo/db/ops/SConscript +++ b/src/mongo/db/ops/SConscript @@ -5,6 +5,24 @@ Import("env") env = env.Clone() env.Library( + target='write_ops_exec', + source=[ + 'write_ops_exec.cpp', + ], + LIBDEPS_PRIVATE=[ + '$BUILD_DIR/mongo/base', + '$BUILD_DIR/mongo/db/repl/oplog', + '$BUILD_DIR/mongo/db/repl/repl_coordinator_interface', + '$BUILD_DIR/mongo/util/fail_point', + '$BUILD_DIR/mongo/db/write_ops', + '$BUILD_DIR/mongo/db/curop', + '$BUILD_DIR/mongo/db/db_raii', + ], +) + + + +env.Library( target='update', source=[ 'modifier_add_to_set.cpp', diff --git a/src/mongo/db/ops/write_ops_exec.cpp b/src/mongo/db/ops/write_ops_exec.cpp index 666ca4efcd4..6db044b5d6f 100644 --- a/src/mongo/db/ops/write_ops_exec.cpp +++ b/src/mongo/db/ops/write_ops_exec.cpp @@ -298,11 +298,32 @@ WriteResult performCreateIndexes(OperationContext* opCtx, const write_ops::Inser void insertDocuments(OperationContext* opCtx, Collection* collection, - std::vector<InsertStatement>::const_iterator begin, - std::vector<InsertStatement>::const_iterator end) { + std::vector<InsertStatement>::iterator begin, + std::vector<InsertStatement>::iterator end) { // Intentionally not using writeConflictRetry. That is handled by the caller so it can react to // oversized batches. WriteUnitOfWork wuow(opCtx); + + // Acquire optimes and fill them in for each item in the batch. + // This must only be done for doc-locking storage engines, which are allowed to insert oplog + // documents out-of-timestamp-order. For other storage engines, the oplog entries must be + // physically written in timestamp order, so we defer optime assignment until the oplog is about + // to be written. + auto batchSize = std::distance(begin, end); + std::unique_ptr<OplogSlot[]> slots(new OplogSlot[batchSize]); + if (supportsDocLocking()) { + auto replCoord = repl::ReplicationCoordinator::get(opCtx); + if (!replCoord->isOplogDisabledFor(opCtx, collection->ns())) { + // Populate 'slots' with new optimes for each insert. + // This also notifies the storage engine of each new timestamp. + repl::getNextOpTimes(opCtx, batchSize, slots.get()); + OplogSlot* slot = slots.get(); + for (auto it = begin; it != end; it++) { + it->oplogSlot = *slot++; + } + } + } + uassertStatusOK(collection->insertDocuments( opCtx, begin, end, &CurOp::get(opCtx)->debug(), /*enforceQuota*/ true)); wuow.commit(); @@ -313,7 +334,7 @@ void insertDocuments(OperationContext* opCtx, */ bool insertBatchAndHandleErrors(OperationContext* opCtx, const write_ops::Insert& wholeOp, - const std::vector<InsertStatement>& batch, + std::vector<InsertStatement>& batch, LastOpFixer* lastOpFixer, WriteResult* out) { if (batch.empty()) diff --git a/src/mongo/db/repair_database.cpp b/src/mongo/db/repair_database.cpp index 552f4577f04..b8feba841c1 100644 --- a/src/mongo/db/repair_database.cpp +++ b/src/mongo/db/repair_database.cpp @@ -249,8 +249,11 @@ Status repairDatabase(OperationContext* opCtx, if (engine->isMmapV1()) { // MMAPv1 is a layering violation so it implements its own repairDatabase. - return static_cast<MMAPV1Engine*>(engine)->repairDatabase( + auto status = static_cast<MMAPV1Engine*>(engine)->repairDatabase( opCtx, dbName, preserveClonedFilesOnFailure, backupOriginalFiles); + // Restore oplog Collection pointer cache. + repl::acquireOplogCollectionForLogging(opCtx); + return status; } // These are MMAPv1 specific @@ -278,6 +281,9 @@ Status repairDatabase(OperationContext* opCtx, for (auto&& collection : *db) { collection->setMinimumVisibleSnapshot(snapshotName); } + + // Restore oplog Collection pointer cache. + repl::acquireOplogCollectionForLogging(opCtx); } catch (...) { severe() << "Unexpected exception encountered while reopening database after repair."; std::terminate(); // Logs additional info about the specific error. diff --git a/src/mongo/db/repl/SConscript b/src/mongo/db/repl/SConscript index 4cf37c9f16e..0b1cf62ad90 100644 --- a/src/mongo/db/repl/SConscript +++ b/src/mongo/db/repl/SConscript @@ -9,25 +9,31 @@ env.Library( source=[ 'apply_ops.cpp', 'oplog.cpp', - 'oplogreader.cpp', ], - LIBDEPS=[ - 'dbcheck', - 'repl_coordinator_interface', - 'repl_coordinator_global', + LIBDEPS_PRIVATE=[ '$BUILD_DIR/mongo/base', '$BUILD_DIR/mongo/db/background', - '$BUILD_DIR/mongo/db/catalog/database', - '$BUILD_DIR/mongo/db/catalog/database_holder', - '$BUILD_DIR/mongo/db/concurrency/lock_manager', + '$BUILD_DIR/mongo/db/catalog/catalog_helpers', '$BUILD_DIR/mongo/db/db_raii', '$BUILD_DIR/mongo/db/dbdirectclient', '$BUILD_DIR/mongo/db/dbhelpers', '$BUILD_DIR/mongo/db/index_d', - '$BUILD_DIR/mongo/db/index/index_descriptor', - '$BUILD_DIR/mongo/db/write_ops', - '$BUILD_DIR/mongo/db/catalog/catalog_helpers', - '$BUILD_DIR/mongo/db/namespace_string', + 'dbcheck', + 'repl_coordinator_interface', + ], +) + +env.Library( + target='oplogreader', + source=[ + 'oplogreader.cpp', + ], + LIBDEPS_PRIVATE=[ + '$BUILD_DIR/mongo/base', + '$BUILD_DIR/mongo/client/clientdriver', + '$BUILD_DIR/mongo/db/auth/authcommon', + '$BUILD_DIR/mongo/db/auth/authorization_manager_global', + '$BUILD_DIR/mongo/util/net/network', ], ) @@ -37,12 +43,12 @@ env.CppUnitTest( 'oplog_test.cpp', ], LIBDEPS=[ - 'oplog', - 'oplog_entry', - 'replmocks', '$BUILD_DIR/mongo/db/service_context_d_test_fixture', '$BUILD_DIR/mongo/unittest/concurrency', '$BUILD_DIR/mongo/util/concurrency/thread_pool', + 'oplog', + 'oplog_entry', + 'replmocks', ], ) @@ -54,6 +60,7 @@ env.Library( LIBDEPS=[ 'oplog', 'oplog_interface_remote', + 'oplogreader', 'repl_coordinator_interface', 'repl_coordinator_global', '$BUILD_DIR/mongo/base', @@ -475,6 +482,7 @@ env.Library( 'oplog', 'replication_process', 'roll_back_local_operations', + '$BUILD_DIR/mongo/db/catalog/catalog_helpers', '$BUILD_DIR/mongo/db/catalog/database_holder', '$BUILD_DIR/mongo/db/s/sharding', '$BUILD_DIR/mongo/db/write_ops', diff --git a/src/mongo/db/repl/initial_syncer.cpp b/src/mongo/db/repl/initial_syncer.cpp index 2f438ec436c..edd114366a4 100644 --- a/src/mongo/db/repl/initial_syncer.cpp +++ b/src/mongo/db/repl/initial_syncer.cpp @@ -505,7 +505,7 @@ void InitialSyncer::_chooseSyncSourceCallback( // There is no need to schedule separate task to create oplog collection since we are already in // a callback and we are certain there's no existing operation context (required for creating // collections and dropping user databases) attached to the current thread. - status = _recreateOplogAndDropReplicatedDatabases(); + status = _truncateOplogAndDropReplicatedDatabases(); if (!status.isOK()) { onCompletionGuard->setResultAndCancelRemainingWork_inlock(lock, status); return; @@ -527,9 +527,9 @@ void InitialSyncer::_chooseSyncSourceCallback( _getBaseRollbackIdHandle = scheduleResult.getValue(); } -Status InitialSyncer::_recreateOplogAndDropReplicatedDatabases() { - // drop/create oplog; drop user databases. - LOG(1) << "About to drop+create the oplog, if it exists, ns:" << _opts.localOplogNS +Status InitialSyncer::_truncateOplogAndDropReplicatedDatabases() { + // truncate oplog; drop user databases. + LOG(1) << "About to truncate the oplog, if it exists, ns:" << _opts.localOplogNS << ", and drop all user databases (so that we can clone them)."; auto opCtx = makeOpCtx(); @@ -537,23 +537,21 @@ Status InitialSyncer::_recreateOplogAndDropReplicatedDatabases() { // We are not replicating nor validating these writes. UnreplicatedWritesBlock unreplicatedWritesBlock(opCtx.get()); - // 1.) Drop the oplog. - LOG(2) << "Dropping the existing oplog: " << _opts.localOplogNS; - auto status = _storage->dropCollection(opCtx.get(), _opts.localOplogNS); + // 1.) Truncate the oplog. + LOG(2) << "Truncating the existing oplog: " << _opts.localOplogNS; + auto status = _storage->truncateCollection(opCtx.get(), _opts.localOplogNS); if (!status.isOK()) { - return status; + // 1a.) Create the oplog. + LOG(2) << "Creating the oplog: " << _opts.localOplogNS; + status = _storage->createOplog(opCtx.get(), _opts.localOplogNS); + if (!status.isOK()) { + return status; + } } // 2.) Drop user databases. - LOG(2) << "Dropping user databases"; - status = _storage->dropReplicatedDatabases(opCtx.get()); - if (!status.isOK()) { - return status; - } - - // 3.) Create the oplog. - LOG(2) << "Creating the oplog: " << _opts.localOplogNS; - return _storage->createOplog(opCtx.get(), _opts.localOplogNS); + LOG(2) << "Dropping user databases"; + return _storage->dropReplicatedDatabases(opCtx.get()); } void InitialSyncer::_rollbackCheckerResetCallback( diff --git a/src/mongo/db/repl/initial_syncer.h b/src/mongo/db/repl/initial_syncer.h index 8ee406d8876..63095486232 100644 --- a/src/mongo/db/repl/initial_syncer.h +++ b/src/mongo/db/repl/initial_syncer.h @@ -278,7 +278,7 @@ private: * | * | * V - * _recreateOplogAndDropReplicatedDatabases() + * _truncateOplogAndDropReplicatedDatabases() * | * | * V @@ -368,11 +368,10 @@ private: /** * This function does the following: - * 1.) Drop oplog. + * 1.) Truncate oplog. * 2.) Drop user databases (replicated dbs). - * 3.) Create oplog. */ - Status _recreateOplogAndDropReplicatedDatabases(); + Status _truncateOplogAndDropReplicatedDatabases(); /** * Callback for rollback checker's first replSetGetRBID command before starting data cloning. diff --git a/src/mongo/db/repl/initial_syncer_test.cpp b/src/mongo/db/repl/initial_syncer_test.cpp index 274678a10bf..8d765cefc04 100644 --- a/src/mongo/db/repl/initial_syncer_test.cpp +++ b/src/mongo/db/repl/initial_syncer_test.cpp @@ -216,6 +216,7 @@ public: protected: struct StorageInterfaceResults { bool createOplogCalled = false; + bool truncateCalled = false; bool insertedOplogEntries = false; int oplogEntriesInserted = 0; bool droppedUserDBs = false; @@ -235,6 +236,12 @@ protected: _storageInterfaceWorkDone.createOplogCalled = true; return Status::OK(); }; + _storageInterface->truncateCollFn = [this](OperationContext* opCtx, + const NamespaceString& nss) { + LockGuard lock(_storageInterfaceWorkDoneMutex); + _storageInterfaceWorkDone.truncateCalled = true; + return Status::OK(); + }; _storageInterface->insertDocumentFn = [this]( OperationContext* opCtx, const NamespaceString& nss, const TimestampedBSONObj& doc) { LockGuard lock(_storageInterfaceWorkDoneMutex); @@ -921,14 +928,13 @@ TEST_F(InitialSyncerTest, InitialSyncerResetsOnCompletionCallbackFunctionPointer ASSERT_TRUE(sharedCallbackStateDestroyed); } -TEST_F(InitialSyncerTest, InitialSyncerRecreatesOplogAndDropsReplicatedDatabases) { - // We are not interested in proceeding beyond the oplog creation stage so we inject a failure - // after setting '_storageInterfaceWorkDone.createOplogCalled' to true. - auto oldCreateOplogFn = _storageInterface->createOplogFn; - _storageInterface->createOplogFn = [oldCreateOplogFn](OperationContext* opCtx, - const NamespaceString& nss) { - oldCreateOplogFn(opCtx, nss).transitional_ignore(); - return Status(ErrorCodes::OperationFailed, "oplog creation failed"); +TEST_F(InitialSyncerTest, InitialSyncerTruncatesOplogAndDropsReplicatedDatabases) { + // We are not interested in proceeding beyond the dropUserDB stage so we inject a failure + // after setting '_storageInterfaceWorkDone.droppedUserDBs' to true. + auto oldDropUserDBsFn = _storageInterface->dropUserDBsFn; + _storageInterface->dropUserDBsFn = [oldDropUserDBsFn](OperationContext* opCtx) { + ASSERT_OK(oldDropUserDBsFn(opCtx)); + return Status(ErrorCodes::OperationFailed, "drop userdbs failed"); }; auto initialSyncer = &getInitialSyncer(); @@ -941,8 +947,8 @@ TEST_F(InitialSyncerTest, InitialSyncerRecreatesOplogAndDropsReplicatedDatabases ASSERT_EQUALS(ErrorCodes::OperationFailed, _lastApplied); LockGuard lock(_storageInterfaceWorkDoneMutex); + ASSERT_TRUE(_storageInterfaceWorkDone.truncateCalled); ASSERT_TRUE(_storageInterfaceWorkDone.droppedUserDBs); - ASSERT_TRUE(_storageInterfaceWorkDone.createOplogCalled); } TEST_F(InitialSyncerTest, InitialSyncerPassesThroughGetRollbackIdScheduleError) { @@ -973,13 +979,13 @@ TEST_F(InitialSyncerTest, InitialSyncerPassesThroughGetRollbackIdScheduleError) TEST_F( InitialSyncerTest, InitialSyncerReturnsShutdownInProgressIfSchedulingRollbackCheckerFailedDueToExecutorShutdown) { - // The rollback id request is sent immediately after oplog creation. We shut the task executor - // down before returning from createOplog() to make the scheduleRemoteCommand() call for + // The rollback id request is sent immediately after oplog truncation. We shut the task executor + // down before returning from truncate() to make the scheduleRemoteCommand() call for // replSetGetRBID fail. - auto oldCreateOplogFn = _storageInterface->createOplogFn; - _storageInterface->createOplogFn = [oldCreateOplogFn, this](OperationContext* opCtx, - const NamespaceString& nss) { - auto status = oldCreateOplogFn(opCtx, nss); + auto oldTruncateCollFn = _storageInterface->truncateCollFn; + _storageInterface->truncateCollFn = [oldTruncateCollFn, this](OperationContext* opCtx, + const NamespaceString& nss) { + auto status = oldTruncateCollFn(opCtx, nss); getExecutor().shutdown(); return status; }; @@ -994,7 +1000,7 @@ TEST_F( ASSERT_EQUALS(ErrorCodes::ShutdownInProgress, _lastApplied); LockGuard lock(_storageInterfaceWorkDoneMutex); - ASSERT_TRUE(_storageInterfaceWorkDone.createOplogCalled); + ASSERT_TRUE(_storageInterfaceWorkDone.truncateCalled); } TEST_F(InitialSyncerTest, InitialSyncerCancelsRollbackCheckerOnShutdown) { diff --git a/src/mongo/db/repl/oplog.cpp b/src/mongo/db/repl/oplog.cpp index 4508ef2621f..a3fc12b7adb 100644 --- a/src/mongo/db/repl/oplog.cpp +++ b/src/mongo/db/repl/oplog.cpp @@ -116,13 +116,17 @@ std::string masterSlaveOplogName = "local.oplog.$main"; MONGO_FP_DECLARE(disableSnapshotting); namespace { -// cached copy...so don't rename, drop, etc.!!! +/** + * The `_localOplogCollection` pointer is always valid (or null) because an + * operation must take the global exclusive lock to set the pointer to null when + * the Collection instance is destroyed. See `oplogCheckCloseDatabase`. + */ Collection* _localOplogCollection = nullptr; PseudoRandom hashGenerator(std::unique_ptr<SecureRandom>(SecureRandom::create())->nextInt64()); -// Synchronizes the section where a new Timestamp is generated and when it actually -// appears in the oplog. +// Synchronizes the section where a new Timestamp is generated and when it is registered in the +// storage engine. stdx::mutex newOpMutex; stdx::condition_variable newTimestampNotifier; // Remembers that last timestamp generated for creating new oplog entries or last timestamp of @@ -137,27 +141,16 @@ void checkOplogInsert(Status result) { massert(17322, str::stream() << "write to oplog failed: " << result.toString(), result.isOK()); } -struct OplogSlot { - OpTime opTime; - int64_t hash = 0; -}; - -/** - * Allocates an optime for a new entry in the oplog, and updates the replication coordinator to - * reflect that new optime. Returns the new optime and the correct value of the "h" field for - * the new oplog entry. - */ -void getNextOpTime(OperationContext* opCtx, - Collection* oplog, - ReplicationCoordinator* replCoord, - ReplicationCoordinator::Mode replicationMode, - unsigned count, - OplogSlot* slotsOut) { - synchronizeOnCappedInFlightResource(opCtx->lockState(), oplog->ns()); +void _getNextOpTimes(OperationContext* opCtx, + Collection* oplog, + std::size_t count, + OplogSlot* slotsOut) { + synchronizeOnOplogInFlightResource(opCtx->lockState()); + auto replCoord = ReplicationCoordinator::get(opCtx); long long term = OpTime::kUninitializedTerm; // Fetch term out of the newOpMutex. - if (replicationMode == ReplicationCoordinator::modeReplSet && + if (replCoord->getReplicationMode() == ReplicationCoordinator::modeReplSet && replCoord->isV1ElectionProtocol()) { // Current term. If we're not a replset of pv=1, it remains kOldProtocolVersionTerm. term = replCoord->getTerm(); @@ -172,8 +165,8 @@ void getNextOpTime(OperationContext* opCtx, fassert(28560, oplog->getRecordStore()->oplogDiskLocRegister(opCtx, ts)); // Set hash if we're in replset mode, otherwise it remains 0 in master/slave. - const bool needHash = (replicationMode == ReplicationCoordinator::modeReplSet); - for (unsigned i = 0; i < count; i++) { + const bool needHash = (replCoord->getReplicationMode() == ReplicationCoordinator::modeReplSet); + for (std::size_t i = 0; i < count; i++) { slotsOut[i].opTime = {Timestamp(ts.asULL() + i), term}; if (needHash) { slotsOut[i].hash = hashGenerator.nextInt64(); @@ -222,11 +215,18 @@ private: } // namespace void setOplogCollectionName() { - if (getGlobalReplicationCoordinator()->getReplicationMode() == - ReplicationCoordinator::modeReplSet) { - _oplogCollectionName = NamespaceString::kRsOplogNamespace.ns(); - } else { - _oplogCollectionName = masterSlaveOplogName; + switch (getGlobalReplicationCoordinator()->getReplicationMode()) { + case ReplicationCoordinator::modeReplSet: + _oplogCollectionName = NamespaceString::kRsOplogNamespace.ns(); + break; + case ReplicationCoordinator::modeMasterSlave: + _oplogCollectionName = masterSlaveOplogName; + break; + case ReplicationCoordinator::modeNone: + // leave empty. + break; + default: + MONGO_UNREACHABLE; } } @@ -275,21 +275,6 @@ void createIndexForApplyOps(OperationContext* opCtx, namespace { -Collection* getLocalOplogCollection(OperationContext* opCtx, - const std::string& oplogCollectionName) { - if (_localOplogCollection) - return _localOplogCollection; - - AutoGetCollection autoColl(opCtx, NamespaceString(oplogCollectionName), MODE_IX); - _localOplogCollection = autoColl.getCollection(); - massert(13347, - "the oplog collection " + oplogCollectionName + - " missing. did you drop it? if so, restart the server", - _localOplogCollection); - - return _localOplogCollection; -} - /** * Attaches the session information of a write to an oplog entry if it exists. */ @@ -392,11 +377,9 @@ void _logOpsInner(OperationContext* opCtx, Timestamp* timestamps, size_t nDocs, Collection* oplogCollection, - ReplicationCoordinator::Mode replicationMode, OpTime finalOpTime) { - ReplicationCoordinator* replCoord = getGlobalReplicationCoordinator(); - - if (nss.size() && replicationMode == ReplicationCoordinator::modeReplSet && + auto replCoord = ReplicationCoordinator::get(opCtx); + if (nss.size() && replCoord->getReplicationMode() == ReplicationCoordinator::modeReplSet && !replCoord->canAcceptWritesFor(opCtx, nss)) { severe() << "logOp() but can't accept write to collection " << nss.ns(); fassertFailed(17405); @@ -431,12 +414,11 @@ OpTime logOp(OperationContext* opCtx, Lock::DBLock lk(opCtx, NamespaceString::kLocalDb, MODE_IX); Lock::CollectionLock lock(opCtx->lockState(), _oplogCollectionName, MODE_IX); - auto const oplog = getLocalOplogCollection(opCtx, _oplogCollectionName); - auto const replMode = replCoord->getReplicationMode(); + auto const oplog = _localOplogCollection; OplogSlot slot; WriteUnitOfWork wuow(opCtx); - getNextOpTime(opCtx, oplog, replCoord, replMode, 1, &slot); + _getNextOpTimes(opCtx, oplog, 1, &slot); auto writer = _logOpWriter(opCtx, opstr, @@ -453,7 +435,7 @@ OpTime logOp(OperationContext* opCtx, oplogLink); const DocWriter* basePtr = &writer; auto timestamp = slot.opTime.getTimestamp(); - _logOpsInner(opCtx, nss, &basePtr, ×tamp, 1, oplog, replMode, slot.opTime); + _logOpsInner(opCtx, nss, &basePtr, ×tamp, 1, oplog, slot.opTime); wuow.commit(); return slot.opTime; } @@ -476,14 +458,11 @@ repl::OpTime logInsertOps(OperationContext* opCtx, const size_t count = end - begin; std::vector<OplogDocWriter> writers; writers.reserve(count); - Collection* oplog = getLocalOplogCollection(opCtx, _oplogCollectionName); + Collection* oplog = _localOplogCollection; Lock::DBLock lk(opCtx, "local", MODE_IX); Lock::CollectionLock lock(opCtx->lockState(), _oplogCollectionName, MODE_IX); - std::unique_ptr<OplogSlot[]> slots(new OplogSlot[count]); - auto replMode = replCoord->getReplicationMode(); WriteUnitOfWork wuow(opCtx); - getNextOpTime(opCtx, oplog, replCoord, replMode, count, slots.get()); auto wallTime = Date_t::now(); OplogLink oplogLink; @@ -496,39 +475,40 @@ repl::OpTime logInsertOps(OperationContext* opCtx, } auto timestamps = stdx::make_unique<Timestamp[]>(count); + OpTime lastOpTime; for (size_t i = 0; i < count; i++) { - auto insertStatement = begin[i]; + // Make a mutable copy. + auto insertStatementOplogSlot = begin[i].oplogSlot; + // Fetch optime now, if not already fetched. + if (insertStatementOplogSlot.opTime.isNull()) { + _getNextOpTimes(opCtx, oplog, 1, &insertStatementOplogSlot); + } writers.emplace_back(_logOpWriter(opCtx, "i", nss, uuid, - insertStatement.doc, + begin[i].doc, NULL, fromMigrate, - slots[i].opTime, - slots[i].hash, + insertStatementOplogSlot.opTime, + insertStatementOplogSlot.hash, wallTime, sessionInfo, - insertStatement.stmtId, + begin[i].stmtId, oplogLink)); - oplogLink.prevTs = slots[i].opTime.getTimestamp(); - timestamps[i] = slots[i].opTime.getTimestamp(); + oplogLink.prevTs = insertStatementOplogSlot.opTime.getTimestamp(); + timestamps[i] = oplogLink.prevTs; + lastOpTime = insertStatementOplogSlot.opTime; } std::unique_ptr<DocWriter const* []> basePtrs(new DocWriter const*[count]); for (size_t i = 0; i < count; i++) { basePtrs[i] = &writers[i]; } - _logOpsInner(opCtx, - nss, - basePtrs.get(), - timestamps.get(), - count, - oplog, - replMode, - slots[count - 1].opTime); + invariant(!lastOpTime.isNull()); + _logOpsInner(opCtx, nss, basePtrs.get(), timestamps.get(), count, oplog, lastOpTime); wuow.commit(); - return slots[count - 1].opTime; + return lastOpTime; } namespace { @@ -599,7 +579,7 @@ void createOplog(OperationContext* opCtx, const std::string& oplogCollectionName throw AssertionException(13257, ss.str()); } } - + acquireOplogCollectionForLogging(opCtx); if (!isReplSet) initTimestampFromOplog(opCtx, oplogCollectionName); return; @@ -619,6 +599,7 @@ void createOplog(OperationContext* opCtx, const std::string& oplogCollectionName writeConflictRetry(opCtx, "createCollection", oplogCollectionName, [&] { WriteUnitOfWork uow(opCtx); invariant(ctx.db()->createCollection(opCtx, oplogCollectionName, options)); + acquireOplogCollectionForLogging(opCtx); if (!isReplSet) getGlobalServiceContext()->getOpObserver()->onOpMessage(opCtx, BSONObj()); uow.commit(); @@ -627,6 +608,7 @@ void createOplog(OperationContext* opCtx, const std::string& oplogCollectionName /* sync here so we don't get any surprising lag later when we try to sync */ StorageEngine* storageEngine = getGlobalServiceContext()->getGlobalStorageEngine(); storageEngine->flushAllFiles(opCtx, true); + log() << "******" << endl; } @@ -636,6 +618,14 @@ void createOplog(OperationContext* opCtx) { createOplog(opCtx, _oplogCollectionName, isReplSet); } +void getNextOpTimes(OperationContext* opCtx, std::size_t count, OplogSlot* slotsOut) { + // The local oplog collection pointer must already be established by this point. + // We can't establish it here because that would require locking the local database, which would + // be a lock order violation. + invariant(_localOplogCollection); + _getNextOpTimes(opCtx, _localOplogCollection, count, slotsOut); +} + // ------------------------------------- namespace { @@ -1317,6 +1307,7 @@ void setNewTimestamp(ServiceContext* service, const Timestamp& newTime) { void initTimestampFromOplog(OperationContext* opCtx, const std::string& oplogNS) { DBDirectClient c(opCtx); + static const BSONObj reverseNaturalObj = BSON("$natural" << -1); BSONObj lastOp = c.findOne(oplogNS, Query().sort(reverseNaturalObj), NULL, QueryOption_SlaveOk); if (!lastOp.isEmpty()) { @@ -1328,8 +1319,18 @@ void initTimestampFromOplog(OperationContext* opCtx, const std::string& oplogNS) void oplogCheckCloseDatabase(OperationContext* opCtx, Database* db) { invariant(opCtx->lockState()->isW()); + if (db->name() == "local") { + _localOplogCollection = nullptr; + } +} + - _localOplogCollection = nullptr; +void acquireOplogCollectionForLogging(OperationContext* opCtx) { + if (!_oplogCollectionName.empty()) { + AutoGetCollection autoColl(opCtx, NamespaceString(_oplogCollectionName), MODE_IX); + _localOplogCollection = autoColl.getCollection(); + fassert(13347, _localOplogCollection); + } } void signalOplogWaiters() { @@ -1429,12 +1430,10 @@ void SnapshotThread::run() { SnapshotName name(0); // assigned real value in block. { - // Make sure there are no in-flight capped inserts while we create our snapshot. + // Make sure there are no in-flight oplog inserts while we create our snapshot. // This lock cannot be aquired until all writes holding the resource commit/abort. - Lock::ResourceLock cappedInsertLockForOtherDb( - opCtx->lockState(), resourceCappedInFlightForOtherDb, MODE_X); - Lock::ResourceLock cappedInsertLockForLocalDb( - opCtx->lockState(), resourceCappedInFlightForLocalDb, MODE_X); + Lock::ResourceLock cappedInsertLockForOplog( + opCtx->lockState(), resourceInFlightForOplog, MODE_X); // Reserve the name immediately before we take our snapshot. This ensures that all // names that compare lower must be from points in time visible to this named diff --git a/src/mongo/db/repl/oplog.h b/src/mongo/db/repl/oplog.h index 812efb81310..2198bb86c7e 100644 --- a/src/mongo/db/repl/oplog.h +++ b/src/mongo/db/repl/oplog.h @@ -34,10 +34,11 @@ #include "mongo/base/status.h" #include "mongo/bson/bsonobj.h" #include "mongo/bson/timestamp.h" -#include "mongo/db/catalog/collection.h" #include "mongo/db/catalog/collection_options.h" +#include "mongo/db/logical_session_id.h" #include "mongo/db/repl/optime.h" #include "mongo/db/repl/replication_coordinator.h" +#include "mongo/db/storage/snapshot_name.h" #include "mongo/stdx/functional.h" namespace mongo { @@ -48,6 +49,30 @@ class OperationContext; class OperationSessionInfo; class Session; +struct OplogSlot { + OplogSlot() {} + OplogSlot(repl::OpTime opTime, std::int64_t hash) : opTime(opTime), hash(hash) {} + repl::OpTime opTime; + std::int64_t hash = 0; +}; + +struct InsertStatement { +public: + InsertStatement() = default; + explicit InsertStatement(BSONObj toInsert) : doc(toInsert) {} + + InsertStatement(StmtId statementId, BSONObj toInsert) : stmtId(statementId), doc(toInsert) {} + InsertStatement(StmtId statementId, BSONObj toInsert, OplogSlot os) + : stmtId(statementId), oplogSlot(os), doc(toInsert) {} + InsertStatement(BSONObj toInsert, SnapshotName ts) + : oplogSlot(repl::OpTime(Timestamp(ts.asU64()), repl::OpTime::kUninitializedTerm), 0), + doc(toInsert) {} + + StmtId stmtId = kUninitializedStmtId; + OplogSlot oplogSlot; + BSONObj doc; +}; + namespace repl { class ReplSettings; @@ -118,10 +143,15 @@ OpTime logOp(OperationContext* opCtx, StmtId stmtId, const OplogLink& oplogLink); -// Flush out the cached pointers to the local database and oplog. +// Flush out the cached pointer to the oplog. // Used by the closeDatabase command to ensure we don't cache closed things. void oplogCheckCloseDatabase(OperationContext* opCtx, Database* db); +/** + * Establish the cached pointer to the local oplog. + */ +void acquireOplogCollectionForLogging(OperationContext* opCtx); + using IncrementOpsAppliedStatsFn = stdx::function<void()>; /** * Take the object field of a BSONObj, the BSONObj, and the namespace of @@ -184,5 +214,11 @@ void createIndexForApplyOps(OperationContext* opCtx, const NamespaceString& indexNss, IncrementOpsAppliedStatsFn incrementOpsAppliedStats); +/** + * Allocates optimes for new entries in the oplog. Returns an array of OplogSlots, which contain + * the new optimes along with their terms and newly calculated hash fields. + */ +void getNextOpTimes(OperationContext* opCtx, std::size_t count, OplogSlot* slotsOut); + } // namespace repl } // namespace mongo diff --git a/src/mongo/db/repl/oplogreader.cpp b/src/mongo/db/repl/oplogreader.cpp index 744045df7f3..659e061c554 100644 --- a/src/mongo/db/repl/oplogreader.cpp +++ b/src/mongo/db/repl/oplogreader.cpp @@ -50,8 +50,6 @@ using std::string; namespace repl { -const BSONObj reverseNaturalObj = BSON("$natural" << -1); - bool replAuthenticate(DBClientBase* conn) { if (isInternalAuthSet()) return conn->authenticateInternalUser(); diff --git a/src/mongo/db/repl/oplogreader.h b/src/mongo/db/repl/oplogreader.h index 7307d14e406..92e224a6498 100644 --- a/src/mongo/db/repl/oplogreader.h +++ b/src/mongo/db/repl/oplogreader.h @@ -37,9 +37,6 @@ namespace mongo { namespace repl { -// {"$natural": -1 } -extern const BSONObj reverseNaturalObj; - /** * Authenticates conn using the server's cluster-membership credentials. * diff --git a/src/mongo/db/repl/replication_coordinator_external_state_impl.cpp b/src/mongo/db/repl/replication_coordinator_external_state_impl.cpp index 06b78d50ef5..674ec4d5940 100644 --- a/src/mongo/db/repl/replication_coordinator_external_state_impl.cpp +++ b/src/mongo/db/repl/replication_coordinator_external_state_impl.cpp @@ -224,6 +224,9 @@ bool ReplicationCoordinatorExternalStateImpl::isInitialSyncFlagSet(OperationCont void ReplicationCoordinatorExternalStateImpl::startSteadyStateReplication( OperationContext* opCtx, ReplicationCoordinator* replCoord) { + // Initialize the cached pointer to the oplog collection, for writing to the oplog. + acquireOplogCollectionForLogging(opCtx); + LockGuard lk(_threadMutex); invariant(replCoord); invariant(!_bgSync); diff --git a/src/mongo/db/repl/storage_interface.h b/src/mongo/db/repl/storage_interface.h index 840b40e2c16..309b1715aa0 100644 --- a/src/mongo/db/repl/storage_interface.h +++ b/src/mongo/db/repl/storage_interface.h @@ -144,11 +144,16 @@ public: const CollectionOptions& options) = 0; /** - * Drops a collection, like the oplog. + * Drops a collection. */ virtual Status dropCollection(OperationContext* opCtx, const NamespaceString& nss) = 0; /** + * Truncates a collection. + */ + virtual Status truncateCollection(OperationContext* opCtx, const NamespaceString& nss) = 0; + + /** * Renames a collection from the "fromNS" to the "toNS". Fails if the new collection already * exists. */ diff --git a/src/mongo/db/repl/storage_interface_impl.cpp b/src/mongo/db/repl/storage_interface_impl.cpp index db090542ed6..e9f6b002cf6 100644 --- a/src/mongo/db/repl/storage_interface_impl.cpp +++ b/src/mongo/db/repl/storage_interface_impl.cpp @@ -411,10 +411,32 @@ Status StorageInterfaceImpl::dropCollection(OperationContext* opCtx, const Names } WriteUnitOfWork wunit(opCtx); const auto status = autoDB.getDb()->dropCollectionEvenIfSystem(opCtx, nss); - if (status.isOK()) { - wunit.commit(); + if (!status.isOK()) { + return status; } - return status; + wunit.commit(); + return Status::OK(); + }); +} + +Status StorageInterfaceImpl::truncateCollection(OperationContext* opCtx, + const NamespaceString& nss) { + return writeConflictRetry(opCtx, "StorageInterfaceImpl::truncateCollection", nss.ns(), [&] { + AutoGetCollection autoColl(opCtx, nss, MODE_X); + auto collectionResult = + getCollection(autoColl, nss, "The collection must exist before truncating."); + if (!collectionResult.isOK()) { + return collectionResult.getStatus(); + } + auto collection = collectionResult.getValue(); + + WriteUnitOfWork wunit(opCtx); + const auto status = collection->truncate(opCtx); + if (!status.isOK()) { + return status; + } + wunit.commit(); + return Status::OK(); }); } diff --git a/src/mongo/db/repl/storage_interface_impl.h b/src/mongo/db/repl/storage_interface_impl.h index 2b8ba24f022..dfb367f398d 100644 --- a/src/mongo/db/repl/storage_interface_impl.h +++ b/src/mongo/db/repl/storage_interface_impl.h @@ -84,6 +84,8 @@ public: Status dropCollection(OperationContext* opCtx, const NamespaceString& nss) override; + Status truncateCollection(OperationContext* opCtx, const NamespaceString& nss) override; + Status renameCollection(OperationContext* opCtx, const NamespaceString& fromNS, const NamespaceString& toNS, diff --git a/src/mongo/db/repl/storage_interface_mock.h b/src/mongo/db/repl/storage_interface_mock.h index a0cd570cf8d..18902950699 100644 --- a/src/mongo/db/repl/storage_interface_mock.h +++ b/src/mongo/db/repl/storage_interface_mock.h @@ -99,6 +99,8 @@ public: stdx::function<Status(OperationContext* opCtx, const NamespaceString& nss)>; using CreateCollectionFn = stdx::function<Status( OperationContext* opCtx, const NamespaceString& nss, const CollectionOptions& options)>; + using TruncateCollectionFn = + stdx::function<Status(OperationContext* opCtx, const NamespaceString& nss)>; using DropCollectionFn = stdx::function<Status(OperationContext* opCtx, const NamespaceString& nss)>; using FindDocumentsFn = @@ -168,6 +170,10 @@ public: return dropCollFn(opCtx, nss); }; + Status truncateCollection(OperationContext* opCtx, const NamespaceString& nss) override { + return truncateCollFn(opCtx, nss); + } + Status renameCollection(OperationContext* opCtx, const NamespaceString& fromNS, const NamespaceString& toNS, @@ -292,6 +298,9 @@ public: [](OperationContext* opCtx, const NamespaceString& nss, const CollectionOptions& options) { return Status{ErrorCodes::IllegalOperation, "CreateCollectionFn not implemented."}; }; + TruncateCollectionFn truncateCollFn = [](OperationContext* opCtx, const NamespaceString& nss) { + return Status{ErrorCodes::IllegalOperation, "TruncateCollectionFn not implemented."}; + }; DropCollectionFn dropCollFn = [](OperationContext* opCtx, const NamespaceString& nss) { return Status{ErrorCodes::IllegalOperation, "DropCollectionFn not implemented."}; }; diff --git a/src/mongo/db/storage/ephemeral_for_test/ephemeral_for_test_record_store.cpp b/src/mongo/db/storage/ephemeral_for_test/ephemeral_for_test_record_store.cpp index 807cb94716e..0c7dc580452 100644 --- a/src/mongo/db/storage/ephemeral_for_test/ephemeral_for_test_record_store.cpp +++ b/src/mongo/db/storage/ephemeral_for_test/ephemeral_for_test_record_store.cpp @@ -385,9 +385,15 @@ StatusWith<RecordId> EphemeralForTestRecordStore::extractAndCheckLocForOplog(con if (!status.isOK()) return status; - if (!_data->records.empty() && status.getValue() <= _data->records.rbegin()->first) - return StatusWith<RecordId>(ErrorCodes::BadValue, "ts not higher than highest"); - + if (!_data->records.empty() && status.getValue() <= _data->records.rbegin()->first) { + + return StatusWith<RecordId>(ErrorCodes::BadValue, + str::stream() << "attempted out-of-order oplog insert of " + << status.getValue() + << " (oplog last insert was " + << _data->records.rbegin()->first + << " )"); + } return status; } @@ -432,8 +438,8 @@ Status EphemeralForTestRecordStore::insertRecordsWithDocWriter(OperationContext* for (size_t i = 0; i < nDocs; i++) { const int len = docs[i]->documentSize(); if (_isCapped && len > _cappedMaxSize) { - // We use dataSize for capped rollover and we don't want to delete everything if we know - // this won't fit. + // We use dataSize for capped rollover and we don't want to delete everything if we + // know this won't fit. return Status(ErrorCodes::BadValue, "object to insert exceeds cappedMaxSize"); } |