summaryrefslogtreecommitdiff
path: root/src
diff options
context:
space:
mode:
Diffstat (limited to 'src')
-rw-r--r--src/mongo/db/repl/SConscript3
-rw-r--r--src/mongo/db/repl/data_replicator_external_state_mock.cpp12
-rw-r--r--src/mongo/db/repl/oplog_application_bm.cpp3
-rw-r--r--src/mongo/db/repl/oplog_applier.h27
-rw-r--r--src/mongo/db/repl/oplog_applier_impl.cpp183
-rw-r--r--src/mongo/db/repl/oplog_applier_impl.h13
-rw-r--r--src/mongo/db/repl/oplog_applier_test.cpp11
-rw-r--r--src/mongo/db/repl/oplog_write_bm.cpp100
-rw-r--r--src/mongo/db/repl/oplog_writer.h29
-rw-r--r--src/mongo/db/repl/oplog_writer_impl.cpp163
-rw-r--r--src/mongo/db/repl/oplog_writer_impl.h47
-rw-r--r--src/mongo/db/repl/oplog_writer_impl_test.cpp85
-rw-r--r--src/mongo/db/repl/replication_coordinator_external_state_impl.cpp39
13 files changed, 363 insertions, 352 deletions
diff --git a/src/mongo/db/repl/SConscript b/src/mongo/db/repl/SConscript
index cfb4850e7cd..58f65d54d43 100644
--- a/src/mongo/db/repl/SConscript
+++ b/src/mongo/db/repl/SConscript
@@ -694,7 +694,6 @@ env.Library(
'$BUILD_DIR/mongo/executor/task_executor_interface',
'$BUILD_DIR/mongo/util/concurrency/thread_pool',
'oplog_entry',
- 'repl_server_parameters',
],
LIBDEPS_PRIVATE=[
'$BUILD_DIR/mongo/db/admission/execution_admission_context',
@@ -705,6 +704,7 @@ env.Library(
'$BUILD_DIR/mongo/util/fail_point',
'$BUILD_DIR/mongo/util/processinfo',
'repl_coordinator_interface',
+ 'repl_server_parameters',
],
)
@@ -727,7 +727,6 @@ env.Library(
'oplog',
'oplog_application_interface',
'oplog_entry',
- 'oplog_write',
'repl_coordinator_interface',
'repl_settings',
'replication_metrics',
diff --git a/src/mongo/db/repl/data_replicator_external_state_mock.cpp b/src/mongo/db/repl/data_replicator_external_state_mock.cpp
index b873e549058..ce6c818798e 100644
--- a/src/mongo/db/repl/data_replicator_external_state_mock.cpp
+++ b/src/mongo/db/repl/data_replicator_external_state_mock.cpp
@@ -60,6 +60,12 @@ public:
_observer(observer),
_externalState(externalState) {}
+ void scheduleWritesToOplogAndChangeCollection(OperationContext* opCtx,
+ StorageInterface* storageInterface,
+ ThreadPool* writerPool,
+ const std::vector<OplogEntry>& ops,
+ bool skipWritesToOplog) override;
+
private:
void _run(OplogBuffer* oplogBuffer) final {}
StatusWith<OpTime> _applyOplogBatch(OperationContext* opCtx,
@@ -71,6 +77,12 @@ private:
DataReplicatorExternalStateMock* const _externalState;
};
+void OplogApplierMock::scheduleWritesToOplogAndChangeCollection(OperationContext* opCtx,
+ StorageInterface* storageInterface,
+ ThreadPool* writerPool,
+ const std::vector<OplogEntry>& ops,
+ bool skipWritesToOplog) {}
+
} // namespace
DataReplicatorExternalStateMock::DataReplicatorExternalStateMock()
diff --git a/src/mongo/db/repl/oplog_application_bm.cpp b/src/mongo/db/repl/oplog_application_bm.cpp
index e15bd20b818..5777069175d 100644
--- a/src/mongo/db/repl/oplog_application_bm.cpp
+++ b/src/mongo/db/repl/oplog_application_bm.cpp
@@ -222,7 +222,8 @@ public:
repl::OplogApplier::Options oplogApplierOptions(
repl::OplogApplication::Mode::kSecondary,
false /* allowNamespaceNotFoundErrorsOnCrudOps */,
- true /* skipWritesToOplog */);
+ true /* skipWritesToOplog */,
+ true /* skipWritesToChangeCollection */);
_oplogApplier = std::make_unique<repl::OplogApplierImpl>(nullptr,
_oplogBuffer.get(),
&repl::noopOplogApplierObserver,
diff --git a/src/mongo/db/repl/oplog_applier.h b/src/mongo/db/repl/oplog_applier.h
index 0ddd049cb55..dab7776ae2b 100644
--- a/src/mongo/db/repl/oplog_applier.h
+++ b/src/mongo/db/repl/oplog_applier.h
@@ -77,25 +77,27 @@ public:
allowNamespaceNotFoundErrorsOnCrudOps(inputMode ==
OplogApplication::Mode::kInitialSync ||
OplogApplication::inRecovering(inputMode)),
- skipWritesToOplog(
- (feature_flags::gReduceMajorityWriteLatency.isEnabled(
- serverGlobalParams.featureCompatibility.acquireFCVSnapshot()) &&
- inputMode == OplogApplication::Mode::kSecondary) ||
- OplogApplication::inRecovering(inputMode)) {}
+ skipWritesToOplog(OplogApplication::inRecovering(inputMode)),
+ skipWritesToChangeCollection(false) {}
- Options(OplogApplication::Mode inputMode, bool skipWritesToOplog)
+ Options(OplogApplication::Mode inputMode,
+ bool skipWritesToOplog,
+ bool skipWritesToChangeCollection)
: mode(inputMode),
allowNamespaceNotFoundErrorsOnCrudOps(inputMode ==
OplogApplication::Mode::kInitialSync ||
OplogApplication::inRecovering(inputMode)),
- skipWritesToOplog(skipWritesToOplog) {}
+ skipWritesToOplog(skipWritesToOplog),
+ skipWritesToChangeCollection(skipWritesToChangeCollection) {}
Options(OplogApplication::Mode mode,
bool allowNamespaceNotFoundErrorsOnCrudOps,
- bool skipWritesToOplog)
+ bool skipWritesToOplog,
+ bool skipWritesToChangeCollection)
: mode(mode),
allowNamespaceNotFoundErrorsOnCrudOps(allowNamespaceNotFoundErrorsOnCrudOps),
- skipWritesToOplog(skipWritesToOplog) {}
+ skipWritesToOplog(skipWritesToOplog),
+ skipWritesToChangeCollection(skipWritesToChangeCollection) {}
// Used to determine which operations should be applied. Only initial sync will set this to
// be something other than the null optime.
@@ -104,6 +106,7 @@ public:
const OplogApplication::Mode mode;
const bool allowNamespaceNotFoundErrorsOnCrudOps;
const bool skipWritesToOplog;
+ const bool skipWritesToChangeCollection;
};
// Used to report oplog application progress.
@@ -183,6 +186,12 @@ public:
*/
StatusWith<OpTime> applyOplogBatch(OperationContext* opCtx, std::vector<OplogEntry> ops);
+ virtual void scheduleWritesToOplogAndChangeCollection(OperationContext* opCtx,
+ StorageInterface* storageInterface,
+ ThreadPool* writerPool,
+ const std::vector<OplogEntry>& ops,
+ bool skipWritesToOplog) = 0;
+
/**
* Calls the OplogBatcher's getNextApplierBatch.
*/
diff --git a/src/mongo/db/repl/oplog_applier_impl.cpp b/src/mongo/db/repl/oplog_applier_impl.cpp
index 56d562b29bb..d02aba3e0c2 100644
--- a/src/mongo/db/repl/oplog_applier_impl.cpp
+++ b/src/mongo/db/repl/oplog_applier_impl.cpp
@@ -67,7 +67,6 @@
#include "mongo/db/repl/oplog_applier_utils.h"
#include "mongo/db/repl/oplog_batcher.h"
#include "mongo/db/repl/oplog_entry_gen.h"
-#include "mongo/db/repl/oplog_writer_impl.h"
#include "mongo/db/repl/replication_metrics.h"
#include "mongo/db/repl/split_prepare_session_manager.h"
#include "mongo/db/repl/transaction_oplog_application.h"
@@ -342,6 +341,58 @@ void _addOplogChainOpsToWriterVectors(OperationContext* opCtx,
opCtx, &extractedOps, writerVectors, collPropertiesCache, shouldSerialize);
}
+Status _insertDocumentsToOplogAndChangeCollections(
+ OperationContext* opCtx,
+ std::vector<InsertStatement>::const_iterator begin,
+ std::vector<InsertStatement>::const_iterator end,
+ bool skipWritesToOplog) {
+ WriteUnitOfWork wunit(opCtx);
+ boost::optional<AutoGetOplogFastPath> autoOplog;
+ boost::optional<ChangeStreamChangeCollectionManager::ChangeCollectionsWriter>
+ changeCollectionWriter;
+
+ // Acquire locks. We must acquire the locks for all collections we intend to write to before
+ // performing any writes. This avoids potential deadlocks created by waiting for locks while
+ // having generated oplog holes.
+ if (!skipWritesToOplog) {
+ autoOplog.emplace(opCtx, OplogAccessMode::kWrite);
+ }
+ const bool changeCollectionsMode =
+ change_stream_serverless_helpers::isChangeCollectionsModeActive();
+ if (changeCollectionsMode) {
+ changeCollectionWriter = boost::make_optional(
+ ChangeStreamChangeCollectionManager::get(opCtx).createChangeCollectionsWriter(
+ opCtx, begin, end, nullptr /* opDebug */));
+ changeCollectionWriter->acquireLocks();
+ }
+
+ // Write entries to the oplog.
+ if (!skipWritesToOplog) {
+ auto& oplogColl = autoOplog->getCollection();
+ if (!oplogColl) {
+ return {ErrorCodes::NamespaceNotFound, "Oplog collection does not exist"};
+ }
+ auto status = collection_internal::insertDocuments(
+ opCtx, oplogColl, begin, end, nullptr /* OpDebug */, false /* fromMigrate */);
+ if (!status.isOK()) {
+ return status;
+ }
+ }
+
+ // Write the corresponding oplog entries to tenants respective change
+ // collections in the serverless.
+ if (changeCollectionsMode) {
+ auto status = changeCollectionWriter->write();
+ if (!status.isOK()) {
+ return status;
+ }
+ }
+
+ wunit.commit();
+
+ return Status::OK();
+}
+
void _setOplogApplicationWorkerOpCtxStates(OperationContext* opCtx) {
// Do not enforce constraints.
opCtx->setEnforceConstraints(false);
@@ -476,23 +527,8 @@ OplogApplierImpl::OplogApplierImpl(executor::TaskExecutor* executor,
_replCoord(replCoord),
_writerPool(writerPool),
_storageInterface(storageInterface),
- _consistencyMarkers(consistencyMarkers) {
-
- // Change collections are always written as part of oplog application, even though
- // in steady state mode where a separate OplogWriter thread is responsible for
- // writing oplog entries. This is because the OplogWriter thread can be ahead of
- // oplog application, but DDL ops on change collections (e.g. create/drop) must be
- // first applied before we can perform writes on those collections.
- _oplogWriter = std::make_unique<OplogWriterImpl>(
- nullptr /* executor */,
- nullptr /* writeBuffer */,
- nullptr /* applyBuffer */,
- replCoord,
- storageInterface,
- consistencyMarkers,
- &noopOplogWriterObserver,
- OplogWriter::Options(options.skipWritesToOplog, false /* skipWritesToChangeColl */));
-}
+ _consistencyMarkers(consistencyMarkers),
+ _beginApplyingOpTime(options.beginApplyingOpTime) {}
void OplogApplierImpl::_run(OplogBuffer* oplogBuffer) {
// Start up a thread from the batcher to pull from the oplog buffer into the batcher's oplog
@@ -618,6 +654,86 @@ void OplogApplierImpl::_run(OplogBuffer* oplogBuffer) {
}
}
+
+// Schedules the writes to the oplog and the change collection for 'ops' into threadPool. The caller
+// must guarantee that 'ops' stays valid until all scheduled work in the thread pool completes.
+void OplogApplierImpl::scheduleWritesToOplogAndChangeCollection(OperationContext* opCtx,
+ StorageInterface* storageInterface,
+ ThreadPool* writerPool,
+ const std::vector<OplogEntry>& ops,
+ bool skipWritesToOplog) {
+ // Skip performing any writes during the startup recovery when running in the non-serverless
+ // environment.
+ if (skipWritesToOplog && !change_stream_serverless_helpers::isChangeCollectionsModeActive()) {
+ return;
+ }
+
+ auto makeOplogWriterForRange = [storageInterface, &ops, skipWritesToOplog](size_t begin,
+ size_t end) {
+ // The returned function will be run in a separate thread after this returns. Therefore
+ // all captures other than 'ops' must be by value since they will not be available. The
+ // caller guarantees that 'ops' will stay in scope until the spawned threads complete.
+ return [storageInterface, &ops, begin, end, skipWritesToOplog](auto status) {
+ invariant(status);
+ auto opCtx = cc().makeOperationContext();
+
+ // Oplog writes are crucial to the stability of the replica set. We mark the operations
+ // as having Immediate priority so that it skips waiting for ticket acquisition and flow
+ // control.
+ ScopedAdmissionPriority<ExecutionAdmissionContext> priority(
+ opCtx.get(), AdmissionContext::Priority::kExempt);
+
+ UnreplicatedWritesBlock uwb(opCtx.get());
+
+ std::vector<InsertStatement> docs;
+ docs.reserve(end - begin);
+ for (size_t i = begin; i < end; i++) {
+ docs.emplace_back(InsertStatement{ops[i].getEntry().getRaw(),
+ ops[i].getOpTime().getTimestamp(),
+ ops[i].getOpTime().getTerm()});
+ }
+
+ // The 'nsOrUUID' is used only to log the debug message when retrying inserts on the
+ // oplog and change collections. The 'writeConflictRetry' assumes operations are done on
+ // a single namespace. But the method '_insertDocumentsToOplogAndChangeCollections' can
+ // perform inserts on the oplog and multiple change collections, ie. several namespaces.
+ // As such 'writeConflictRetry' will not log the correct namespace when retrying.
+ NamespaceStringOrUUID nsOrUUID = !skipWritesToOplog
+ ? NamespaceString::kRsOplogNamespace
+ : NamespaceString::makeChangeCollectionNSS(boost::none /* tenantId */);
+
+ fassert(6663400,
+ storage_helpers::insertBatchAndHandleRetry(
+ opCtx.get(), nsOrUUID, docs, [&](auto* opCtx, auto begin, auto end) {
+ return _insertDocumentsToOplogAndChangeCollections(
+ opCtx, begin, end, skipWritesToOplog);
+ }));
+ };
+ };
+
+ // We want to be able to take advantage of bulk inserts so we don't use multiple threads if it
+ // would result too little work per thread. This also ensures that we can amortize the
+ // setup/teardown overhead across many writes.
+ const size_t kMinOplogEntriesPerThread = 16;
+ const bool enoughToMultiThread =
+ ops.size() >= kMinOplogEntriesPerThread * writerPool->getStats().options.maxThreads;
+
+ // Storage engines support parallel writes to the oplog because they are required to ensure that
+ // oplog entries are ordered correctly, even if inserted out-of-order.
+ if (!enoughToMultiThread) {
+ writerPool->schedule(makeOplogWriterForRange(0, ops.size()));
+ return;
+ }
+
+ const size_t numOplogThreads = writerPool->getStats().options.maxThreads;
+ const size_t numOpsPerThread = ops.size() / numOplogThreads;
+ for (size_t thread = 0; thread < numOplogThreads; thread++) {
+ size_t begin = thread * numOpsPerThread;
+ size_t end = (thread == numOplogThreads - 1) ? ops.size() : begin + numOpsPerThread;
+ writerPool->schedule(makeOplogWriterForRange(begin, end));
+ }
+}
+
StatusWith<OpTime> OplogApplierImpl::_applyOplogBatch(OperationContext* opCtx,
std::vector<OplogEntry> ops) {
invariant(!ops.empty());
@@ -642,19 +758,16 @@ StatusWith<OpTime> OplogApplierImpl::_applyOplogBatch(OperationContext* opCtx,
// because the spawned threads refer to objects on the stack
ON_BLOCK_EXIT([&] { _writerPool->waitForIdle(); });
- // Write ops into the oplog collection and change collections if needed.
- //
- // - In steady state mode, a separate OplogWriter is responsible for writing the oplog
- // collection, but the change collections must be written as part of oplog application
- // because DDL ops (e.g. create/drop) on change collections are explicitly replicated,
- // and so must be first applied before writes are performed to those collections.
- //
- // - In initial sync mode, the applier is responsible for writing the oplog collection
- // as well as the change collections.
- //
- // - In recovering modes, there is no need to write to the oplog collection, and the
- // applier is responsible for writing the change collections.
- _oplogWriter->writeOplogBatch(opCtx, ops, _writerPool);
+ // Write batch of ops into oplog.
+ if (!getOptions().skipWritesToOplog) {
+ _consistencyMarkers->setOplogTruncateAfterPoint(
+ opCtx, _replCoord->getMyLastAppliedOpTime().getTimestamp());
+ }
+
+ if (!getOptions().skipWritesToOplog || !getOptions().skipWritesToChangeCollection) {
+ scheduleWritesToOplogAndChangeCollection(
+ opCtx, _storageInterface, _writerPool, ops, getOptions().skipWritesToOplog);
+ }
// Holds 'pseudo operations' generated by secondaries to aid in replication.
// Keep in scope until all operations in 'ops' and 'derivedOps' have been applied.
@@ -669,6 +782,9 @@ StatusWith<OpTime> OplogApplierImpl::_applyOplogBatch(OperationContext* opCtx,
_writerPool->getStats().options.maxThreads);
_fillWriterVectors(opCtx, &ops, &writerVectors, &derivedOps);
+ // Wait for writes to finish before applying ops.
+ _writerPool->waitForIdle();
+
// Use this fail point to hang after we have written the oplog entries but before we have
// applied them.
if (MONGO_unlikely(pauseBatchApplicationAfterWritingOplogEntries.shouldFail())) {
@@ -682,6 +798,11 @@ StatusWith<OpTime> OplogApplierImpl::_applyOplogBatch(OperationContext* opCtx,
const bool isDataConsistent =
_consistencyMarkers->getMinValid(opCtx) < ops.front().getOpTime();
+ // Reset consistency markers in case the node fails while applying ops.
+ if (!getOptions().skipWritesToOplog) {
+ _consistencyMarkers->setOplogTruncateAfterPoint(opCtx, Timestamp());
+ }
+
{
std::vector<Status> statusVector(_writerPool->getStats().options.maxThreads,
Status::OK());
diff --git a/src/mongo/db/repl/oplog_applier_impl.h b/src/mongo/db/repl/oplog_applier_impl.h
index f6ab46a6e20..dae3b9b7361 100644
--- a/src/mongo/db/repl/oplog_applier_impl.h
+++ b/src/mongo/db/repl/oplog_applier_impl.h
@@ -44,7 +44,6 @@
#include "mongo/db/repl/oplog_buffer.h"
#include "mongo/db/repl/oplog_entry.h"
#include "mongo/db/repl/oplog_entry_or_grouped_inserts.h"
-#include "mongo/db/repl/oplog_writer.h"
#include "mongo/db/repl/optime.h"
#include "mongo/db/repl/replication_consistency_markers.h"
#include "mongo/db/repl/replication_coordinator.h"
@@ -92,6 +91,12 @@ public:
std::vector<std::vector<ApplierOperation>>* writerVectors,
std::vector<std::vector<OplogEntry>>* derivedOps) noexcept;
+ void scheduleWritesToOplogAndChangeCollection(OperationContext* opCtx,
+ StorageInterface* storageInterface,
+ ThreadPool* writerPool,
+ const std::vector<OplogEntry>& ops,
+ bool skipWritesToOplog) override;
+
private:
/**
* Runs oplog application in a loop until shutdown() is called.
@@ -134,11 +139,13 @@ private:
// Not owned by us.
ThreadPool* const _writerPool;
- StorageInterface* const _storageInterface;
+ StorageInterface* _storageInterface;
ReplicationConsistencyMarkers* const _consistencyMarkers;
- std::unique_ptr<OplogWriter> _oplogWriter;
+ // Used to determine which operations should be applied during initial sync. If this is null,
+ // we will apply all operations that were fetched.
+ OpTime _beginApplyingOpTime = OpTime();
protected:
// Marked as protected for use in unit tests.
diff --git a/src/mongo/db/repl/oplog_applier_test.cpp b/src/mongo/db/repl/oplog_applier_test.cpp
index 56d1796fe62..eddf69ff8f7 100644
--- a/src/mongo/db/repl/oplog_applier_test.cpp
+++ b/src/mongo/db/repl/oplog_applier_test.cpp
@@ -73,6 +73,11 @@ public:
void _run(OplogBuffer* oplogBuffer) final;
StatusWith<OpTime> _applyOplogBatch(OperationContext* opCtx, std::vector<OplogEntry> ops) final;
+ void scheduleWritesToOplogAndChangeCollection(OperationContext* opCtx,
+ StorageInterface* storageInterface,
+ ThreadPool* writerPool,
+ const std::vector<OplogEntry>& ops,
+ bool skipWritesToOplog) override;
};
OplogApplierMock::OplogApplierMock(OplogBuffer* oplogBuffer)
@@ -88,6 +93,12 @@ StatusWith<OpTime> OplogApplierMock::_applyOplogBatch(OperationContext* opCtx,
return OpTime();
}
+void OplogApplierMock::scheduleWritesToOplogAndChangeCollection(OperationContext* opCtx,
+ StorageInterface* storageInterface,
+ ThreadPool* writerPool,
+ const std::vector<OplogEntry>& ops,
+ bool skipWritesToOplog) {}
+
class OplogApplierTest : public ServiceContextTest {
public:
void setUp() override;
diff --git a/src/mongo/db/repl/oplog_write_bm.cpp b/src/mongo/db/repl/oplog_write_bm.cpp
index eecb7579835..479911fcc9f 100644
--- a/src/mongo/db/repl/oplog_write_bm.cpp
+++ b/src/mongo/db/repl/oplog_write_bm.cpp
@@ -78,7 +78,6 @@
#include "mongo/db/repl/oplog_buffer.h"
#include "mongo/db/repl/oplog_buffer_blocking_queue.h"
#include "mongo/db/repl/oplog_entry.h"
-#include "mongo/db/repl/oplog_writer_impl.h"
#include "mongo/db/repl/optime.h"
#include "mongo/db/repl/repl_settings.h"
#include "mongo/db/repl/replication_consistency_markers_mock.h"
@@ -208,21 +207,26 @@ public:
std::make_unique<MongoDSessionCatalog>(
std::make_unique<MongoDSessionCatalogTransactionInterfaceImpl>()));
+ _oplogBuffer = std::make_unique<repl::OplogBufferBlockingQueue>(kOplogBufferSize);
+
repl::replWriterThreadCount = numThreads; // Repl worker thread count
repl::replWriterMinThreadCount = numThreads;
- _oplogWriterPool = repl::makeReplWriterPool();
-
- _oplogWriter = std::make_unique<repl::OplogWriterImpl>(
- nullptr,
- nullptr,
- nullptr,
- _replCoord,
- _storageInterface,
- &_consistencyMarkers,
- &repl::noopOplogWriterObserver,
- repl::OplogWriter::Options(false /* skipWritesToOplogColl */,
- true /* skipWritesToChangeColl */));
+ _oplogApplierThreadPool = repl::makeReplWriterPool();
+
+ repl::OplogApplier::Options oplogApplierOptions(
+ repl::OplogApplication::Mode::kSecondary,
+ false /* allowNamespaceNotFoundErrorsOnCrudOps */,
+ false /* skipWritesToOplog */,
+ false /* skipWritesToChangeCollection */);
+ _oplogApplier = std::make_unique<repl::OplogApplierImpl>(nullptr,
+ _oplogBuffer.get(),
+ &repl::noopOplogApplierObserver,
+ _replCoord,
+ &_consistencyMarkers,
+ _storageInterface,
+ oplogApplierOptions,
+ _oplogApplierThreadPool.get());
_svcCtx->notifyStorageStartupRecoveryComplete();
}
@@ -289,12 +293,12 @@ public:
return _replCoord;
}
- repl::OplogWriter* getOplogWriter() {
- return _oplogWriter.get();
+ repl::OplogApplier* getOplogApplier() {
+ return _oplogApplier.get();
}
ThreadPool* getThreadPool() {
- return _oplogWriterPool.get();
+ return _oplogApplierThreadPool.get();
}
repl::StorageInterface* getStorageInterface() {
@@ -305,10 +309,13 @@ private:
ServiceContext* _svcCtx;
Client* _client;
repl::ReplicationCoordinatorMock* _replCoord;
- std::unique_ptr<repl::OplogWriter> _oplogWriter;
+ std::unique_ptr<repl::OplogApplier> _oplogApplier;
repl::StorageInterface* _storageInterface;
+
+ // This class also owns objects necessary for `_oplogApplier`.
+ std::unique_ptr<repl::OplogBufferBlockingQueue> _oplogBuffer;
repl::ReplicationConsistencyMarkersMock _consistencyMarkers;
- std::unique_ptr<ThreadPool> _oplogWriterPool;
+ std::unique_ptr<ThreadPool> _oplogApplierThreadPool;
boost::optional<unittest::TempDir> _tempDir;
};
@@ -322,7 +329,7 @@ class Fixture {
public:
Fixture(TestServiceContext* testSvcCtx) : _testSvcCtx(testSvcCtx), _foobarUUID(UUID::gen()) {}
- void generateEntries(int totalOps, int entrySize) {
+ void createBatch(int totalOps, int entrySize) {
const long long term1 = 1;
for (int idx = 0; idx < totalOps; ++idx) {
auto x1 =
@@ -375,40 +382,27 @@ public:
}
}
- std::vector<std::vector<BSONObj>> getOplogBatches(size_t numEntriesPerBatch,
- size_t numBytesPerBatch) {
- invariant(!_oplogEntries.empty());
-
- std::vector<std::vector<BSONObj>> batches;
- std::vector<BSONObj>::iterator first = _oplogEntries.begin();
-
- std::size_t batchEntries = 1;
- std::size_t batchBytes = first->objsize();
-
- for (auto last = first + 1; last != _oplogEntries.end(); ++last) {
- // Create a new batch if batch limits exceeded.
- if ((batchEntries + 1 > numEntriesPerBatch) ||
- (batchBytes + last->objsize() > numBytesPerBatch)) {
- batches.emplace_back(first, last);
- batchEntries = 1;
- batchBytes = last->objsize();
- first = last;
- continue;
- }
- // Otherwise update batch stats and move on.
- ++batchEntries;
- batchBytes += last->objsize();
- }
- batches.emplace_back(first, _oplogEntries.end());
-
- return batches;
+ void enqueueOplog(OperationContext* opCtx) {
+ _testSvcCtx->getOplogApplier()->enqueue(opCtx, _oplogEntries.begin(), _oplogEntries.end());
}
- void writeOplog(OperationContext* opCtx, const std::vector<std::vector<BSONObj>>& batches) {
- for (const auto& batch : batches) {
+ void writeOplog(OperationContext* opCtx, size_t numEntriesPerBatch, size_t numBytesPerBatch) {
+ while (!_testSvcCtx->getOplogApplier()->getBuffer()->isEmpty()) {
+ auto oplogBatch = invariantStatusOK(_testSvcCtx->getOplogApplier()->getNextApplierBatch(
+ opCtx, {numBytesPerBatch, numEntriesPerBatch}))
+ .releaseBatch();
+
AutoGetDb autoDb(opCtx, _foobarNs.dbName(), MODE_X);
- invariant(_testSvcCtx->getOplogWriter()->writeOplogBatch(
- opCtx, batch, _testSvcCtx->getThreadPool()));
+
+ _testSvcCtx->getOplogApplier()->scheduleWritesToOplogAndChangeCollection(
+ opCtx,
+ _testSvcCtx->getStorageInterface(),
+ _testSvcCtx->getThreadPool(),
+ std::move(oplogBatch),
+ false);
+
+ // Wait for writes to finish
+ _testSvcCtx->getThreadPool()->waitForIdle();
}
}
@@ -427,9 +421,9 @@ void runBMTest(TestServiceContext& testSvcCtx, Fixture& fixture, benchmark::Stat
auto opCtxRaii = testSvcCtx.getSvcCtx()->makeOperationContext(testSvcCtx.getClient());
auto opCtx = opCtxRaii.get();
repl::createOplog(opCtx);
- auto batches = fixture.getOplogBatches(state.range(3), state.range(4));
+ fixture.enqueueOplog(opCtx);
auto start = mongo::stdx::chrono::high_resolution_clock::now();
- fixture.writeOplog(opCtx, batches);
+ fixture.writeOplog(opCtx, state.range(3), state.range(4));
auto end = mongo::stdx::chrono::high_resolution_clock::now();
auto elapsed_seconds =
mongo::stdx::chrono::duration_cast<mongo::stdx::chrono::duration<double>>(end - start);
@@ -440,7 +434,7 @@ void runBMTest(TestServiceContext& testSvcCtx, Fixture& fixture, benchmark::Stat
void BM_TestWriteOps(benchmark::State& state) {
TestServiceContext testSvcCtx(state.range(0));
Fixture fixture(&testSvcCtx);
- fixture.generateEntries(state.range(1), state.range(2));
+ fixture.createBatch(state.range(1), state.range(2));
runBMTest(testSvcCtx, fixture, state);
}
diff --git a/src/mongo/db/repl/oplog_writer.h b/src/mongo/db/repl/oplog_writer.h
index 4a0681ae98a..401289261fb 100644
--- a/src/mongo/db/repl/oplog_writer.h
+++ b/src/mongo/db/repl/oplog_writer.h
@@ -33,7 +33,6 @@
#include "mongo/db/repl/oplog_buffer.h"
#include "mongo/db/repl/oplog_writer_batcher.h"
#include "mongo/executor/task_executor.h"
-#include "mongo/util/concurrency/thread_pool.h"
namespace mongo {
namespace repl {
@@ -51,13 +50,11 @@ public:
*/
class Options {
public:
- Options() = delete;
- explicit Options(bool skipWritesToOplogColl, bool skipWritesToChangeColl)
- : skipWritesToOplogColl(skipWritesToOplogColl),
- skipWritesToChangeColl(skipWritesToChangeColl) {}
+ Options() : skipWritesToOplogColl(false) {}
+ explicit Options(bool skipWritesToOplogColl)
+ : skipWritesToOplogColl(skipWritesToOplogColl) {}
const bool skipWritesToOplogColl;
- const bool skipWritesToChangeColl;
};
// Used to report oplog write progress.
@@ -106,27 +103,17 @@ public:
boost::optional<std::size_t> bytes = boost::none);
/**
- * Writes a batch of oplog entries to the oplog and/or the change collections.
+ * Writes a batch of oplog entries to the oplog and/or the change collection.
*
- * Returns false if nothing is written, true otherwise.
+ * If the batch write is successful, returns the optime of the last op written,
+ * which should be the last op in the batch.
*
* Oplog visibility and updates to replication coordinator timestamps should be
* handled by caller.
*/
- virtual bool writeOplogBatch(OperationContext* opCtx,
- const std::vector<BSONObj>& ops,
- ThreadPool* writerPool = nullptr) = 0;
+ virtual StatusWith<OpTime> writeOplogBatch(OperationContext* opCtx,
+ const std::vector<BSONObj>& ops) = 0;
- /**
- * Same as above, except for the type of the oplog entries.
- */
- virtual bool writeOplogBatch(OperationContext* opCtx,
- const std::vector<OplogEntry>& ops,
- ThreadPool* writerPool = nullptr) = 0;
-
- /**
- * Returns the options used to configure the behavior of this OplogWriter.
- */
const Options& getOptions() const;
private:
diff --git a/src/mongo/db/repl/oplog_writer_impl.cpp b/src/mongo/db/repl/oplog_writer_impl.cpp
index fb2c2b21bc3..e2595d78bff 100644
--- a/src/mongo/db/repl/oplog_writer_impl.cpp
+++ b/src/mongo/db/repl/oplog_writer_impl.cpp
@@ -46,7 +46,7 @@ namespace repl {
namespace {
-constexpr size_t kMinOpsPerThread = 16;
+const auto changeCollNss = NamespaceString::makeChangeCollectionNSS(boost::none);
auto checkFeatureFlagReduceMajorityWriteLatencyFn = [] {
return feature_flags::gReduceMajorityWriteLatency.isEnabled(
@@ -106,35 +106,6 @@ Status insertDocsToOplogAndChangeCollections(OperationContext* opCtx,
return Status::OK();
}
-std::vector<InsertStatement> createInsertStatements(const std::vector<BSONObj>& ops,
- size_t begin,
- size_t end) {
- std::vector<InsertStatement> docs;
- docs.reserve(end - begin);
-
- for (size_t i = begin; i < end; ++i) {
- auto opTime = invariantStatusOK(OpTime::parseFromOplogEntry(ops[i]));
- docs.emplace_back(ops[i], opTime.getTimestamp(), opTime.getTerm());
- }
-
- return docs;
-}
-
-std::vector<InsertStatement> createInsertStatements(const std::vector<repl::OplogEntry>& ops,
- size_t begin,
- size_t end) {
- std::vector<InsertStatement> docs;
- docs.reserve(end - begin);
-
- for (size_t i = begin; i < end; ++i) {
- docs.emplace_back(ops[i].getEntry().getRaw(),
- ops[i].getOpTime().getTimestamp(),
- ops[i].getOpTime().getTerm());
- }
-
- return docs;
-}
-
} // namespace
@@ -158,14 +129,12 @@ OplogWriterImpl::OplogWriterImpl(executor::TaskExecutor* executor,
OplogBuffer* applyBuffer,
ReplicationCoordinator* replCoord,
StorageInterface* storageInterface,
- ReplicationConsistencyMarkers* consistencyMarkers,
Observer* observer,
const OplogWriter::Options& options)
: OplogWriter(executor, writeBuffer, options),
_applyBuffer(applyBuffer),
_replCoord(replCoord),
_storageInterface(storageInterface),
- _consistencyMarkers(consistencyMarkers),
_observer(observer) {}
void OplogWriterImpl::_run() {
@@ -219,16 +188,14 @@ void OplogWriterImpl::_run() {
auto lastOpTimeAndWallTime =
invariantStatusOK(OpTimeAndWallTime::parseOpTimeAndWallTimeFromOplogEntry(ops.back()));
- {
- LOGV2_DEBUG(8352100, 2, "Oplog write batch size", "size"_attr = ops.size());
-
- // Increment the batch stats.
- oplogWriterMetric.incrementBatchSize(ops.size());
- TimerHolder timer(&oplogWriterMetric.getBatches());
-
- // Write the operations in this batch.
- invariant(writeOplogBatch(opCtx, ops));
+ // Write the operations in this batch. 'writeOplogBatch' returns the optime of
+ // the last op that was written, which should be the last optime in the batch.
+ auto swLastOpTime = writeOplogBatch(opCtx, ops);
+ if (swLastOpTime.getStatus().code() == ErrorCodes::InterruptedAtShutdown) {
+ return;
}
+ fassertNoTrace(8543103, swLastOpTime);
+ invariant(swLastOpTime.getValue() == lastOpTimeAndWallTime.opTime);
// Update various things that care about our last written optime.
finalizeOplogBatch(opCtx, lastOpTimeAndWallTime, flushJournal);
@@ -238,16 +205,36 @@ void OplogWriterImpl::_run() {
}
}
-bool OplogWriterImpl::writeOplogBatch(OperationContext* opCtx,
- const std::vector<BSONObj>& ops,
- ThreadPool* writerPool) {
- return _writeOplogBatch(opCtx, ops, writerPool);
-}
+StatusWith<OpTime> OplogWriterImpl::writeOplogBatch(OperationContext* opCtx,
+ const std::vector<BSONObj>& ops) {
+ invariant(!ops.empty());
+ LOGV2_DEBUG(8352100, 2, "Oplog write batch size", "size"_attr = ops.size());
+
+ bool writeOplogColl = !getOptions().skipWritesToOplogColl;
+ bool writeChangeColl = change_stream_serverless_helpers::isChangeCollectionsModeActive();
+
+ // This should only happen for recovery modes.
+ if (!writeOplogColl && !writeChangeColl) {
+ return OpTime();
+ }
+
+ // Increment the batch stats.
+ oplogWriterMetric.incrementBatchSize(ops.size());
+ TimerHolder timer(&oplogWriterMetric.getBatches());
-bool OplogWriterImpl::writeOplogBatch(OperationContext* opCtx,
- const std::vector<OplogEntry>& ops,
- ThreadPool* writerPool) {
- return _writeOplogBatch(opCtx, ops, writerPool);
+ // Create insert statements from the oplog entries.
+ std::vector<InsertStatement> docs;
+ docs.reserve(ops.size());
+
+ for (const auto& op : ops) {
+ auto opTime = invariantStatusOK(OpTime::parseFromOplogEntry(op));
+ docs.emplace_back(InsertStatement{op, opTime.getTimestamp(), opTime.getTerm()});
+ }
+
+ // Perform writes to oplog collection and/or change collection.
+ _writeOplogBatchImpl(opCtx, docs, writeOplogColl, writeChangeColl);
+
+ return docs.back().oplogSlot;
}
void OplogWriterImpl::finalizeOplogBatch(OperationContext* opCtx,
@@ -269,84 +256,16 @@ void OplogWriterImpl::finalizeOplogBatch(OperationContext* opCtx,
}
}
-template <typename T>
-bool OplogWriterImpl::_writeOplogBatch(OperationContext* opCtx,
- const std::vector<T>& ops,
- ThreadPool* writerPool) {
- invariant(!ops.empty());
-
- bool writeOplogColl = !getOptions().skipWritesToOplogColl;
- bool writeChangeColl = !getOptions().skipWritesToChangeColl &&
- change_stream_serverless_helpers::isChangeCollectionsModeActive();
-
- // Don't do anything if not writing to the oplog collection nor the change collections.
- if (!writeOplogColl && !writeChangeColl) {
- return false;
- }
-
- // Perform writes to oplog collection and/or change collections with the current thread.
- if (!writerPool) {
- _writeOplogBatchForRange(opCtx, ops, 0, ops.size(), writeOplogColl, writeChangeColl);
- return true;
- }
-
- // Perform writes to oplog collection and/or change collections using the thread pool.
-
- // When performing writes with multiple threads, we must set oplogTruncateAfterPoint
- // in case the server crashes before all the threads finish. In such cases the oplog
- // will be truncated after this opTime during startup recovery in order to make sure
- // there are no holes in the oplog.
- if (writeOplogColl) {
- _consistencyMarkers->setOplogTruncateAfterPoint(
- opCtx, _replCoord->getMyLastWrittenOpTime().getTimestamp());
- }
-
- auto makeOplogWriteForRange = [this, &ops, writeOplogColl, writeChangeColl](size_t begin,
- size_t end) {
- return [this, &ops, begin, end, writeOplogColl, writeChangeColl](auto status) {
- invariant(status);
- auto opCtx = cc().makeOperationContext();
- _writeOplogBatchForRange(opCtx.get(), ops, begin, end, writeOplogColl, writeChangeColl);
- };
- };
-
- const auto poolMaxThreads = writerPool->getStats().options.maxThreads;
- const auto enoughToMultiThread = ops.size() >= kMinOpsPerThread * poolMaxThreads;
- const auto numWriteThreads = enoughToMultiThread ? poolMaxThreads : 1;
- const size_t numOpsPerThread = ops.size() / numWriteThreads;
-
- for (size_t t = 0; t < numWriteThreads; ++t) {
- size_t begin = t * numOpsPerThread;
- size_t end = (t == numWriteThreads - 1) ? ops.size() : begin + numOpsPerThread;
- writerPool->schedule(makeOplogWriteForRange(begin, end));
- }
-
- // Wait for all scheduled writes to complete.
- writerPool->waitForIdle();
-
- // Reset oplogTruncateAfterPoint after writes are complete.
- if (writeOplogColl) {
- _consistencyMarkers->setOplogTruncateAfterPoint(opCtx, Timestamp());
- }
-
- return true;
-}
-
-template <typename T>
-void OplogWriterImpl::_writeOplogBatchForRange(OperationContext* opCtx,
- const std::vector<T>& ops,
- size_t begin,
- size_t end,
- bool writeOplogColl,
- bool writeChangeColl) {
+void OplogWriterImpl::_writeOplogBatchImpl(OperationContext* opCtx,
+ const std::vector<InsertStatement>& docs,
+ bool writeOplogColl,
+ bool writeChangeColl) {
// Oplog writes are crucial to the stability of the replica set. We give the operations
// Immediate priority so that it skips waiting for ticket acquisition and flow control.
ScopedAdmissionPriority<ExecutionAdmissionContext> priority(
opCtx, AdmissionContext::Priority::kExempt);
UnreplicatedWritesBlock uwb(opCtx);
- auto docs = createInsertStatements(ops, begin, end);
-
// The 'nsOrUUID' is used only to log the debug message when retrying inserts on the
// oplog and change collections. The 'writeConflictRetry' helper assumes operations
// are done on a single a single namespace. But the provided insert function can do
diff --git a/src/mongo/db/repl/oplog_writer_impl.h b/src/mongo/db/repl/oplog_writer_impl.h
index 78fe7f015db..3a4ba11de5e 100644
--- a/src/mongo/db/repl/oplog_writer_impl.h
+++ b/src/mongo/db/repl/oplog_writer_impl.h
@@ -30,9 +30,9 @@
#pragma once
#include "mongo/db/repl/oplog_writer.h"
-#include "mongo/db/repl/replication_consistency_markers.h"
#include "mongo/db/repl/storage_interface.h"
#include "mongo/db/stats/timer_stats.h"
+#include "mongo/util/concurrency/thread_pool.h"
namespace mongo {
namespace repl {
@@ -75,31 +75,24 @@ public:
OplogBuffer* applyBuffer,
ReplicationCoordinator* replCoord,
StorageInterface* storageInterface,
- ReplicationConsistencyMarkers* consistencyMarkers,
Observer* observer,
const OplogWriter::Options& options);
/**
- * Writes a batch of oplog entries to the oplog and/or the change collections.
+ * Writes a batch of oplog entries to the oplog and/or the change collection.
*
- * Returns false if nothing is written, true otherwise.
+ * The current implementation uses one thread to write to the oplog collection,
+ * and in serverless environment uses another thread to write to the serverless
+ * change collection in parallel.
*
- * If 'writerPool' is not set, the caller thread is used to perform the writes,
- * otherwise 'writerPool' is used to perform the writes with multiple threads.
+ * If the batch write is successful, returns the optime of the last op written,
+ * which should be the last op in the batch.
*
* External states such as oplog visibility, replication opTimes and journaling
* are not updated in this function.
*/
- bool writeOplogBatch(OperationContext* opCtx,
- const std::vector<BSONObj>& ops,
- ThreadPool* writerPool = nullptr) override;
-
- /**
- * Same as above, except for the type of the oplog entries.
- */
- bool writeOplogBatch(OperationContext* opCtx,
- const std::vector<OplogEntry>& ops,
- ThreadPool* writerPool = nullptr) override;
+ StatusWith<OpTime> writeOplogBatch(OperationContext* opCtx,
+ const std::vector<BSONObj>& ops) override;
/**
* Finalizes the batch after writing it to storage, which updates various external
@@ -122,30 +115,18 @@ private:
*/
void _run() override;
- template <typename T>
- bool _writeOplogBatch(OperationContext* opCtx,
- const std::vector<T>& ops,
- ThreadPool* writerPool);
-
- template <typename T>
- void _writeOplogBatchForRange(OperationContext* opCtx,
- const std::vector<T>& ops,
- size_t begin,
- size_t end,
- bool writeOplogColl,
- bool writeChangeColl);
+ void _writeOplogBatchImpl(OperationContext* opCtx,
+ const std::vector<InsertStatement>& docs,
+ bool writeOplogColl,
+ bool writeChangeColl);
- // Not owned by us.
OplogBuffer* const _applyBuffer;
// Not owned by us.
ReplicationCoordinator* const _replCoord;
// Not owned by us.
- StorageInterface* const _storageInterface;
-
- // Not owned by us.
- ReplicationConsistencyMarkers* const _consistencyMarkers;
+ StorageInterface* _storageInterface;
// Not owned by us.
Observer* const _observer;
diff --git a/src/mongo/db/repl/oplog_writer_impl_test.cpp b/src/mongo/db/repl/oplog_writer_impl_test.cpp
index 587999c34d3..7e52fed7bcd 100644
--- a/src/mongo/db/repl/oplog_writer_impl_test.cpp
+++ b/src/mongo/db/repl/oplog_writer_impl_test.cpp
@@ -33,7 +33,6 @@
#include "mongo/db/repl/oplog_batcher_test_fixture.h"
#include "mongo/db/repl/oplog_writer.h"
#include "mongo/db/repl/oplog_writer_impl.h"
-#include "mongo/db/repl/replication_consistency_markers_mock.h"
#include "mongo/db/repl/replication_coordinator_mock.h"
#include "mongo/db/repl/storage_interface.h"
#include "mongo/db/repl/storage_interface_impl.h"
@@ -105,16 +104,12 @@ protected:
OperationContext* opCtx() const;
- ThreadPool* getWriterPool() const;
ReplicationCoordinator* getReplCoord() const;
StorageInterface* getStorageInterface() const;
- ReplicationConsistencyMarkers* getConsistencyMarkers() const;
JournalListenerMock* getJournalListener() const;
ServiceContext* _serviceContext;
ServiceContext::UniqueOperationContext _opCtxHolder;
- std::unique_ptr<ThreadPool> _writerPool;
- std::unique_ptr<ReplicationConsistencyMarkers> _consistencyMarkers;
std::unique_ptr<CountOpsObserver> _observer;
};
@@ -130,8 +125,6 @@ void OplogWriterImplTest::setUp() {
StorageInterface::set(_serviceContext, std::make_unique<StorageInterfaceImpl>());
- _consistencyMarkers = std::make_unique<ReplicationConsistencyMarkersMock>();
-
MongoDSessionCatalog::set(
_serviceContext,
std::make_unique<MongoDSessionCatalog>(
@@ -139,14 +132,11 @@ void OplogWriterImplTest::setUp() {
repl::createOplog(opCtx());
- _writerPool = makeReplWriterPool();
_observer = std::make_unique<CountOpsObserver>();
}
void OplogWriterImplTest::tearDown() {
_opCtxHolder = {};
- _writerPool = {};
- _consistencyMarkers = {};
_observer = {};
StorageInterface::set(_serviceContext, {});
ServiceContextMongoDTest::tearDown();
@@ -156,10 +146,6 @@ OperationContext* OplogWriterImplTest::opCtx() const {
return _opCtxHolder.get();
}
-ThreadPool* OplogWriterImplTest::getWriterPool() const {
- return _writerPool.get();
-}
-
ReplicationCoordinator* OplogWriterImplTest::getReplCoord() const {
return ReplicationCoordinator::get(_serviceContext);
}
@@ -168,52 +154,43 @@ StorageInterface* OplogWriterImplTest::getStorageInterface() const {
return StorageInterface::get(_serviceContext);
}
-ReplicationConsistencyMarkers* OplogWriterImplTest::getConsistencyMarkers() const {
- return _consistencyMarkers.get();
-}
-
JournalListenerMock* OplogWriterImplTest::getJournalListener() const {
return static_cast<JournalListenerMock*>(_journalListener.get());
}
DEATH_TEST_F(OplogWriterImplTest, WriteEmptyBatchFails, "!ops.empty()") {
- OplogWriter::Options options(false /* skipWritesToOplogColl */,
- false /* skipWritesToChangeColl */);
-
OplogWriterImpl oplogWriter(nullptr, // executor
nullptr, // writeBuffer
nullptr, // applyBuffer
getReplCoord(),
getStorageInterface(),
- getConsistencyMarkers(),
&noopOplogWriterObserver,
- options);
+ OplogWriter::Options());
// Writing an empty batch should hit an invariant.
- oplogWriter.writeOplogBatch(opCtx(), std::vector<BSONObj>{});
+ oplogWriter.writeOplogBatch(opCtx(), {}).getStatus().ignore();
}
TEST_F(OplogWriterImplTest, WriteOplogCollectionOnly) {
- OplogWriter::Options options(false /* skipWritesToOplogColl */,
- true /* skipWritesToChangeColl */);
-
OplogWriterImpl oplogWriter(nullptr, // executor
nullptr, // writeBuffer
nullptr, // applyBuffer
getReplCoord(),
getStorageInterface(),
- getConsistencyMarkers(),
_observer.get(),
- options);
+ OplogWriter::Options());
std::vector<BSONObj> ops;
ops.push_back(makeRawInsertOplogEntry(1, kNss1));
ops.push_back(makeRawInsertOplogEntry(2, kNss2));
- auto written = oplogWriter.writeOplogBatch(opCtx(), std::move(ops));
+ auto returnOpTime = OpTime::parseFromOplogEntry(ops.back()).getValue();
+ auto statusWith = oplogWriter.writeOplogBatch(opCtx(), std::move(ops));
+
+ ASSERT_OK(statusWith);
+ ASSERT_EQ(returnOpTime, statusWith.getValue());
// Verify that the batch is only written to the oplog collection.
- ASSERT(written);
ASSERT_EQ(2, _observer->oplogCollDocsCount.load());
ASSERT_EQ(0, _observer->changeCollDocsCount.load());
}
@@ -226,26 +203,25 @@ TEST_F(OplogWriterImplTest, WriteChangeCollectionsOnly) {
ChangeStreamChangeCollectionManager::create(_serviceContext);
- OplogWriter::Options options(true /* skipWritesToOplogColl */,
- false /* skipWritesToChangeColl */);
-
OplogWriterImpl oplogWriter(nullptr, // executor
nullptr, // writeBuffer
nullptr, // applyBuffer
getReplCoord(),
getStorageInterface(),
- getConsistencyMarkers(),
_observer.get(),
- options);
+ OplogWriter::Options(true /* skipWritesToOplogColl */));
std::vector<BSONObj> ops;
ops.push_back(makeRawInsertOplogEntry(1, kNss1));
ops.push_back(makeRawInsertOplogEntry(2, kNss2));
- auto written = oplogWriter.writeOplogBatch(opCtx(), std::move(ops));
+ auto returnOpTime = OpTime::parseFromOplogEntry(ops.back()).getValue();
+ auto statusWith = oplogWriter.writeOplogBatch(opCtx(), std::move(ops));
+
+ ASSERT_OK(statusWith);
+ ASSERT_EQ(returnOpTime, statusWith.getValue());
// Verify that the batch is only written to the change collections.
- ASSERT(written);
ASSERT_EQ(0, _observer->oplogCollDocsCount.load());
ASSERT_EQ(2, _observer->changeCollDocsCount.load());
}
@@ -258,67 +234,60 @@ TEST_F(OplogWriterImplTest, WriteBothOplogAndChangeCollections) {
ChangeStreamChangeCollectionManager::create(_serviceContext);
- OplogWriter::Options options(false /* skipWritesToOplogColl */,
- false /* skipWritesToChangeColl */);
-
OplogWriterImpl oplogWriter(nullptr, // executor
nullptr, // writeBuffer
nullptr, // applyBuffer
getReplCoord(),
getStorageInterface(),
- getConsistencyMarkers(),
_observer.get(),
- options);
+ OplogWriter::Options());
std::vector<BSONObj> ops;
ops.push_back(makeRawInsertOplogEntry(1, kNss1));
ops.push_back(makeRawInsertOplogEntry(2, kNss2));
- auto written = oplogWriter.writeOplogBatch(opCtx(), std::move(ops));
+ auto returnOpTime = OpTime::parseFromOplogEntry(ops.back()).getValue();
+ auto statusWith = oplogWriter.writeOplogBatch(opCtx(), std::move(ops));
+
+ ASSERT_OK(statusWith);
+ ASSERT_EQ(returnOpTime, statusWith.getValue());
// Verify that the batch written to both the oplog and change collections.
- ASSERT(written);
ASSERT_EQ(2, _observer->oplogCollDocsCount.load());
ASSERT_EQ(2, _observer->changeCollDocsCount.load());
}
TEST_F(OplogWriterImplTest, WriteNeitherOplogNorChangeCollections) {
- OplogWriter::Options options(true /* skipWritesToOplogColl */,
- true /* skipWritesToChangeColl */);
-
OplogWriterImpl oplogWriter(nullptr, // executor
nullptr, // writeBuffer
nullptr, // applyBuffer
getReplCoord(),
getStorageInterface(),
- getConsistencyMarkers(),
_observer.get(),
- options);
+ OplogWriter::Options(true /* skipWritesToOplogColl */));
std::vector<BSONObj> ops;
ops.push_back(makeRawInsertOplogEntry(1, kNss1));
ops.push_back(makeRawInsertOplogEntry(2, kNss2));
- auto written = oplogWriter.writeOplogBatch(opCtx(), std::move(ops));
+ auto statusWith = oplogWriter.writeOplogBatch(opCtx(), std::move(ops));
+
+ ASSERT_OK(statusWith);
+ ASSERT_EQ(OpTime(), statusWith.getValue());
- // Verify that the batch is not written any collection.
- ASSERT(!written);
+ // Verify that the batch is only written to the oplog collection.
ASSERT_EQ(0, _observer->oplogCollDocsCount.load());
ASSERT_EQ(0, _observer->changeCollDocsCount.load());
}
TEST_F(OplogWriterImplTest, finalizeOplogBatchCorrectlyUpdatesOpTimes) {
- OplogWriter::Options options(false /* skipWritesToOplogColl */,
- false /* skipWritesToChangeColl */);
-
OplogWriterImpl oplogWriter(nullptr, // executor
nullptr, // writeBuffer
nullptr, // applyBuffer
getReplCoord(),
getStorageInterface(),
- getConsistencyMarkers(),
&noopOplogWriterObserver,
- options);
+ OplogWriter::Options());
auto curOpTime = OpTime(Timestamp(2, 2), 1);
auto curWallTime = Date_t::now();
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 1e1cc64f850..21d1189e0ea 100644
--- a/src/mongo/db/repl/replication_coordinator_external_state_impl.cpp
+++ b/src/mongo/db/repl/replication_coordinator_external_state_impl.cpp
@@ -297,28 +297,29 @@ void ReplicationCoordinatorExternalStateImpl::startSteadyStateReplication(
// Using noop observer for both writer and applier. During steady state replication,
// there is no need to log details on every batch we apply.
+ // TODO (SERVER-87674): use a different thread pool.
if (useOplogWriter) {
- _oplogWriter = std::make_unique<OplogWriterImpl>(
- _oplogWriterTaskExecutor.get(),
- _oplogWriteBuffer.get(),
- _oplogApplyBuffer.get(),
- replCoord,
- _storageInterface,
- _replicationProcess->getConsistencyMarkers(),
- &noopOplogWriterObserver,
- OplogWriter::Options(false /* skipWritesToOplogColl */,
- true /* skipWritesToChangeColl */));
+ _oplogWriter = std::make_unique<OplogWriterImpl>(_oplogWriterTaskExecutor.get(),
+ _oplogWriteBuffer.get(),
+ _oplogApplyBuffer.get(),
+ replCoord,
+ _storageInterface,
+ &noopOplogWriterObserver,
+ OplogWriter::Options());
}
- _oplogApplier = std::make_unique<OplogApplierImpl>(
- _oplogApplierTaskExecutor.get(),
- _oplogApplyBuffer.get(),
- &noopOplogApplierObserver,
- replCoord,
- _replicationProcess->getConsistencyMarkers(),
- _storageInterface,
- OplogApplier::Options(OplogApplication::Mode::kSecondary),
- _writerPool.get());
+ // TODO (SERVER-85697): clean up the applier options.
+ OplogApplier::Options applierOptions(OplogApplication::Mode::kSecondary,
+ useOplogWriter /* skipWritesToOplog */,
+ useOplogWriter /* skipWritesToChangeCollection */);
+ _oplogApplier = std::make_unique<OplogApplierImpl>(_oplogApplierTaskExecutor.get(),
+ _oplogApplyBuffer.get(),
+ &noopOplogApplierObserver,
+ replCoord,
+ _replicationProcess->getConsistencyMarkers(),
+ _storageInterface,
+ applierOptions,
+ _writerPool.get());
invariant(!_bgSync);
_bgSync = std::make_unique<BackgroundSync>(
generated by cgit v1.2.3 (git 2.39.1) at 2026-09-11 12:49:59 +0000