diff options
18 files changed, 247 insertions, 51 deletions
diff --git a/src/mongo/db/repl/SConscript b/src/mongo/db/repl/SConscript index 5544f897842..d83a3609938 100644 --- a/src/mongo/db/repl/SConscript +++ b/src/mongo/db/repl/SConscript @@ -881,7 +881,9 @@ env.Library( 'data_replicator_external_state_mock.cpp', ], LIBDEPS=[ + 'oplog_entry', 'optime', + '$BUILD_DIR/mongo/util/net/hostandport', ], ) diff --git a/src/mongo/db/repl/bgsync.cpp b/src/mongo/db/repl/bgsync.cpp index dfe7b287654..71c8d28dbcd 100644 --- a/src/mongo/db/repl/bgsync.cpp +++ b/src/mongo/db/repl/bgsync.cpp @@ -80,8 +80,10 @@ const Milliseconds kOplogSocketTimeout(30000); */ class DataReplicatorExternalStateBackgroundSync : public DataReplicatorExternalStateImpl { public: - DataReplicatorExternalStateBackgroundSync(ReplicationCoordinator* replicationCoordinator, - BackgroundSync* bgsync); + DataReplicatorExternalStateBackgroundSync( + ReplicationCoordinator* replicationCoordinator, + ReplicationCoordinatorExternalState* replicationCoordinatorExternalState, + BackgroundSync* bgsync); bool shouldStopFetching(const HostAndPort& source, const OpTime& sourceOpTime, bool sourceHasSyncSource) override; @@ -91,8 +93,11 @@ private: }; DataReplicatorExternalStateBackgroundSync::DataReplicatorExternalStateBackgroundSync( - ReplicationCoordinator* replicationCoordinator, BackgroundSync* bgsync) - : DataReplicatorExternalStateImpl(replicationCoordinator), _bgsync(bgsync) {} + ReplicationCoordinator* replicationCoordinator, + ReplicationCoordinatorExternalState* replicationCoordinatorExternalState, + BackgroundSync* bgsync) + : DataReplicatorExternalStateImpl(replicationCoordinator, replicationCoordinatorExternalState), + _bgsync(bgsync) {} bool DataReplicatorExternalStateBackgroundSync::shouldStopFetching(const HostAndPort& source, const OpTime& sourceOpTime, @@ -157,8 +162,6 @@ BackgroundSync::BackgroundSync() _threadPoolTaskExecutor(makeThreadPool(), executor::makeNetworkInterface("NetworkInterfaceASIO-BGSync")), _replCoord(getGlobalReplicationCoordinator()), - _dataReplicatorExternalState( - stdx::make_unique<DataReplicatorExternalStateBackgroundSync>(_replCoord, this)), _syncSourceResolver(_replCoord), _lastOpTimeFetched(Timestamp(std::numeric_limits<int>::max(), 0), std::numeric_limits<long long>::max()) {} @@ -184,7 +187,8 @@ void BackgroundSync::shutdown() { } } -void BackgroundSync::producerThread() { +void BackgroundSync::producerThread( + ReplicationCoordinatorExternalState* replicationCoordinatorExternalState) { Client::initThread("rsBackgroundSync"); AuthorizationSession::get(cc())->grantInternalAuthorization(); @@ -196,7 +200,7 @@ void BackgroundSync::producerThread() { while (!inShutdown()) { try { - _producerThread(); + _producerThread(replicationCoordinatorExternalState); } catch (const DBException& e) { std::string msg(str::stream() << "sync producer problem: " << e.toString()); error() << msg; @@ -222,7 +226,8 @@ void BackgroundSync::_signalNoNewDataForApplier() { } } -void BackgroundSync::_producerThread() { +void BackgroundSync::_producerThread( + ReplicationCoordinatorExternalState* replicationCoordinatorExternalState) { const MemberState state = _replCoord->getMemberState(); // Stop when the state changes to primary. if (_replCoord->isWaitingForApplierToDrain() || state.primary()) { @@ -256,10 +261,12 @@ void BackgroundSync::_producerThread() { start(&txn); } - _produce(&txn); + _produce(&txn, replicationCoordinatorExternalState); } -void BackgroundSync::_produce(OperationContext* txn) { +void BackgroundSync::_produce( + OperationContext* txn, + ReplicationCoordinatorExternalState* replicationCoordinatorExternalState) { // this oplog reader does not do a handshake because we don't want the server it's syncing // from to track how far it has synced { @@ -339,6 +346,8 @@ void BackgroundSync::_produce(OperationContext* txn) { // "lastFetched" not used. Already set in _enqueueDocuments. Status fetcherReturnStatus = Status::OK(); + DataReplicatorExternalStateBackgroundSync dataReplicatorExternalState( + _replCoord, replicationCoordinatorExternalState, this); OplogFetcher* oplogFetcher; try { auto config = _replCoord->getConfig(); @@ -354,7 +363,7 @@ void BackgroundSync::_produce(OperationContext* txn) { source, NamespaceString(rsOplogName), config, - _dataReplicatorExternalState.get(), + &dataReplicatorExternalState, stdx::bind(&BackgroundSync::_enqueueDocuments, this, stdx::placeholders::_1, diff --git a/src/mongo/db/repl/bgsync.h b/src/mongo/db/repl/bgsync.h index f9291c91b44..40d44ae8f93 100644 --- a/src/mongo/db/repl/bgsync.h +++ b/src/mongo/db/repl/bgsync.h @@ -52,6 +52,7 @@ namespace repl { class Member; class ReplicationCoordinator; +class ReplicationCoordinatorExternalState; // This interface exists to facilitate easier testing; // the test infrastructure implements these functions with stubs. @@ -102,8 +103,13 @@ public: virtual ~BackgroundSync() {} - // starts the producer thread - void producerThread(); + /** + * Starts the producer thread which runs until shutdown. Upon resolving the current sync source + * the producer thread uses the OplogFetcher (which requires the replication coordinator + * external state at construction) to fetch oplog entries from the source's oplog via a long + * running find query. + */ + void producerThread(ReplicationCoordinatorExternalState* replicationCoordinatorExternalState); // starts the sync target notifying thread void notifierThread(); @@ -151,8 +157,9 @@ private: BackgroundSync operator=(const BackgroundSync& s); // Production thread - void _producerThread(); - void _produce(OperationContext* txn); + void _producerThread(ReplicationCoordinatorExternalState* replicationCoordinatorExternalState); + void _produce(OperationContext* txn, + ReplicationCoordinatorExternalState* replicationCoordinatorExternalState); /** * Signals to the applier that we have no new data, @@ -211,10 +218,6 @@ private: // A pointer to the replication coordinator running the show. ReplicationCoordinator* _replCoord; - // Data replicator external state required by the oplog fetcher. - // Owned by us. - std::unique_ptr<DataReplicatorExternalState> _dataReplicatorExternalState; - // Used to determine sync source. // TODO(dannenberg) move into DataReplicator. SyncSourceResolver _syncSourceResolver; diff --git a/src/mongo/db/repl/data_replicator.cpp b/src/mongo/db/repl/data_replicator.cpp index c72427f3d40..41e15b19daf 100644 --- a/src/mongo/db/repl/data_replicator.cpp +++ b/src/mongo/db/repl/data_replicator.cpp @@ -428,7 +428,6 @@ DataReplicator::DataReplicator( _applierActive(false), _applierPaused(false), _oplogBuffer(kOplogBufferSize, &getSize) { - uassert(ErrorCodes::BadValue, "invalid applier function", _opts.applierFn); uassert(ErrorCodes::BadValue, "invalid rollback function", _opts.rollbackFn); uassert(ErrorCodes::BadValue, "invalid replSetUpdatePosition command object creation function", @@ -1147,7 +1146,6 @@ Status DataReplicator::_scheduleApplyBatch_inlock() { return status.getStatus(); } } - invariant(_opts.applierFn); invariant(!(_applier && _applier->isActive())); return _scheduleApplyBatch_inlock(ops); } @@ -1155,6 +1153,27 @@ Status DataReplicator::_scheduleApplyBatch_inlock() { } Status DataReplicator::_scheduleApplyBatch_inlock(const Operations& ops) { + MultiApplier::ApplyOperationFn applierFn; + if (_state == DataReplicatorState::Steady) { + applierFn = stdx::bind(&DataReplicatorExternalState::_multiSyncApply, + _dataReplicatorExternalState.get(), + stdx::placeholders::_1); + } else { + invariant(_state == DataReplicatorState::InitialSync); + // "_syncSource" has to be copied to stdx::bind result. + HostAndPort source = _syncSource; + applierFn = stdx::bind(&DataReplicatorExternalState::_multiInitialSyncApply, + _dataReplicatorExternalState.get(), + stdx::placeholders::_1, + source); + } + + auto multiApplyFn = stdx::bind(&DataReplicatorExternalState::_multiApply, + _dataReplicatorExternalState.get(), + stdx::placeholders::_1, + stdx::placeholders::_2, + stdx::placeholders::_3); + auto lambda = [this](const TimestampStatus& ts, const Operations& theOps) { CBHStatus status = _exec->scheduleWork(stdx::bind(&DataReplicator::_onApplyBatchFinish, this, @@ -1172,7 +1191,7 @@ Status DataReplicator::_scheduleApplyBatch_inlock(const Operations& ops) { _exec->wait(status.getValue()); }; - _applier.reset(new MultiApplier(_exec, ops, _opts.applierFn, _opts.multiApplyFn, lambda)); + _applier.reset(new MultiApplier(_exec, ops, applierFn, multiApplyFn, lambda)); return _applier->start(); } diff --git a/src/mongo/db/repl/data_replicator.h b/src/mongo/db/repl/data_replicator.h index 441cbc10aa3..e1c491ebd4d 100644 --- a/src/mongo/db/repl/data_replicator.h +++ b/src/mongo/db/repl/data_replicator.h @@ -136,8 +136,6 @@ struct DataReplicatorOptions { std::string scopeNS; BSONObj filterCriteria; - MultiApplier::ApplyOperationFn applierFn; - MultiApplier::MultiApplyFn multiApplyFn; RollbackFn rollbackFn; Reporter::PrepareReplSetUpdatePositionCommandFn prepareReplSetUpdatePositionCommandFn; GetMyLastOptimeFn getMyLastOptime; diff --git a/src/mongo/db/repl/data_replicator_external_state.h b/src/mongo/db/repl/data_replicator_external_state.h index 983290a2148..d19f46f9711 100644 --- a/src/mongo/db/repl/data_replicator_external_state.h +++ b/src/mongo/db/repl/data_replicator_external_state.h @@ -29,6 +29,7 @@ #pragma once #include "mongo/base/disallow_copying.h" +#include "mongo/db/repl/multiapplier.h" #include "mongo/db/repl/optime.h" #include "mongo/db/repl/optime_with.h" #include "mongo/rpc/metadata/repl_set_metadata.h" @@ -38,6 +39,8 @@ namespace mongo { namespace repl { +class DataReplicator; + /** * Holds current term and last committed optime necessary to populate find/getMore command requests. */ @@ -76,6 +79,37 @@ public: virtual bool shouldStopFetching(const HostAndPort& source, const OpTime& sourceOpTime, bool sourceHasSyncSource) = 0; + +private: + /** + * Applies the operations described in the oplog entries contained in "ops" using the + * "applyOperation" function. + * + * Used exclusively by the DataReplicator to construct a MultiApplier. + */ + virtual StatusWith<OpTime> _multiApply(OperationContext* txn, + const MultiApplier::Operations& ops, + MultiApplier::ApplyOperationFn applyOperation) = 0; + + /** + * Used by _multiApply() to write operations to database during steady state replication. + * + * Used exclusively by the DataReplicator to construct a MultiApplier. + */ + virtual void _multiSyncApply(const MultiApplier::Operations& ops) = 0; + + /** + * Used by _multiApply() to write operations to database during initial sync. + * Fetches missing documents from "source". + * + * Used exclusively by the DataReplicator to construct a MultiApplier. + */ + virtual void _multiInitialSyncApply(const MultiApplier::Operations& ops, + const HostAndPort& source) = 0; + + // Provides DataReplicator with access to _multiApply, _multiSyncApply and + // _multiInitialSyncApply. + friend class DataReplicator; }; } // namespace repl diff --git a/src/mongo/db/repl/data_replicator_external_state_impl.cpp b/src/mongo/db/repl/data_replicator_external_state_impl.cpp index 828a2aa51a5..558f3faf700 100644 --- a/src/mongo/db/repl/data_replicator_external_state_impl.cpp +++ b/src/mongo/db/repl/data_replicator_external_state_impl.cpp @@ -33,14 +33,17 @@ #include "mongo/db/repl/data_replicator_external_state_impl.h" #include "mongo/db/repl/replication_coordinator.h" +#include "mongo/db/repl/replication_coordinator_external_state.h" #include "mongo/util/log.h" namespace mongo { namespace repl { DataReplicatorExternalStateImpl::DataReplicatorExternalStateImpl( - ReplicationCoordinator* replicationCoordinator) - : _replicationCoordinator(replicationCoordinator) {} + ReplicationCoordinator* replicationCoordinator, + ReplicationCoordinatorExternalState* replicationCoordinatorExternalState) + : _replicationCoordinator(replicationCoordinator), + _replicationCoordinatorExternalState(replicationCoordinatorExternalState) {} OpTimeWithTerm DataReplicatorExternalStateImpl::getCurrentTermAndLastCommittedOpTime() { if (!_replicationCoordinator->isV1ElectionProtocol()) { @@ -70,9 +73,30 @@ bool DataReplicatorExternalStateImpl::shouldStopFetching(const HostAndPort& sour return false; } +StatusWith<OpTime> DataReplicatorExternalStateImpl::_multiApply( + OperationContext* txn, + const MultiApplier::Operations& ops, + MultiApplier::ApplyOperationFn applyOperation) { + return _replicationCoordinatorExternalState->multiApply(txn, ops, applyOperation); +} + +void DataReplicatorExternalStateImpl::_multiSyncApply(const MultiApplier::Operations& ops) { + _replicationCoordinatorExternalState->multiSyncApply(ops); +} + +void DataReplicatorExternalStateImpl::_multiInitialSyncApply(const MultiApplier::Operations& ops, + const HostAndPort& source) { + _replicationCoordinatorExternalState->multiInitialSyncApply(ops, source); +} + ReplicationCoordinator* DataReplicatorExternalStateImpl::getReplicationCoordinator() const { return _replicationCoordinator; } +ReplicationCoordinatorExternalState* +DataReplicatorExternalStateImpl::getReplicationCoordinatorExternalState() const { + return _replicationCoordinatorExternalState; +} + } // namespace repl } // namespace mongo diff --git a/src/mongo/db/repl/data_replicator_external_state_impl.h b/src/mongo/db/repl/data_replicator_external_state_impl.h index 8fc84ff218c..25a09e1d7db 100644 --- a/src/mongo/db/repl/data_replicator_external_state_impl.h +++ b/src/mongo/db/repl/data_replicator_external_state_impl.h @@ -34,6 +34,7 @@ namespace mongo { namespace repl { class ReplicationCoordinator; +class ReplicationCoordinatorExternalState; /** * Data replicator external state implementation using a replication coordinator. @@ -41,7 +42,9 @@ class ReplicationCoordinator; class DataReplicatorExternalStateImpl : public DataReplicatorExternalState { public: - DataReplicatorExternalStateImpl(ReplicationCoordinator* replicationCoordinator); + DataReplicatorExternalStateImpl( + ReplicationCoordinator* replicationCoordinator, + ReplicationCoordinatorExternalState* replicationCoordinatorExternalState); OpTimeWithTerm getCurrentTermAndLastCommittedOpTime() override; @@ -51,12 +54,26 @@ public: const OpTime& sourceOpTime, bool sourceHasSyncSource) override; +private: + StatusWith<OpTime> _multiApply(OperationContext* txn, + const MultiApplier::Operations& ops, + MultiApplier::ApplyOperationFn applyOperation) override; + + void _multiSyncApply(const MultiApplier::Operations& ops) override; + + void _multiInitialSyncApply(const MultiApplier::Operations& ops, + const HostAndPort& source) override; + protected: ReplicationCoordinator* getReplicationCoordinator() const; + ReplicationCoordinatorExternalState* getReplicationCoordinatorExternalState() const; private: // Not owned by us. ReplicationCoordinator* _replicationCoordinator; + + // Not owned by us. + ReplicationCoordinatorExternalState* _replicationCoordinatorExternalState; }; 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 501d9ae70c3..83a6e3fd83c 100644 --- a/src/mongo/db/repl/data_replicator_external_state_mock.cpp +++ b/src/mongo/db/repl/data_replicator_external_state_mock.cpp @@ -33,6 +33,11 @@ namespace mongo { namespace repl { +DataReplicatorExternalStateMock::DataReplicatorExternalStateMock() + : multiApplyFn([](OperationContext*, + const MultiApplier::Operations& ops, + MultiApplier::ApplyOperationFn) { return ops.back().getOpTime(); }) {} + OpTimeWithTerm DataReplicatorExternalStateMock::getCurrentTermAndLastCommittedOpTime() { return {currentTerm, lastCommittedOpTime}; } @@ -50,5 +55,17 @@ bool DataReplicatorExternalStateMock::shouldStopFetching(const HostAndPort& sour return shouldStopFetchingResult; } +StatusWith<OpTime> DataReplicatorExternalStateMock::_multiApply( + OperationContext* txn, + const MultiApplier::Operations& ops, + MultiApplier::ApplyOperationFn applyOperation) { + return multiApplyFn(txn, ops, applyOperation); +} + +void DataReplicatorExternalStateMock::_multiSyncApply(const MultiApplier::Operations& ops) {} + +void DataReplicatorExternalStateMock::_multiInitialSyncApply(const MultiApplier::Operations& ops, + const HostAndPort& source) {} + } // namespace repl } // namespace mongo diff --git a/src/mongo/db/repl/data_replicator_external_state_mock.h b/src/mongo/db/repl/data_replicator_external_state_mock.h index ef78c691157..4705fb57bd3 100644 --- a/src/mongo/db/repl/data_replicator_external_state_mock.h +++ b/src/mongo/db/repl/data_replicator_external_state_mock.h @@ -41,6 +41,8 @@ class ReplicationCoordinator; class DataReplicatorExternalStateMock : public DataReplicatorExternalState { public: + DataReplicatorExternalStateMock(); + OpTimeWithTerm getCurrentTermAndLastCommittedOpTime() override; void processMetadata(const rpc::ReplSetMetadata& metadata) override; @@ -63,6 +65,19 @@ public: // Returned by shouldStopFetching. bool shouldStopFetchingResult = false; + + // Override to change multiApply behavior. + MultiApplier::MultiApplyFn multiApplyFn; + +private: + StatusWith<OpTime> _multiApply(OperationContext* txn, + const MultiApplier::Operations& ops, + MultiApplier::ApplyOperationFn applyOperation) override; + + void _multiSyncApply(const MultiApplier::Operations& ops) override; + + void _multiInitialSyncApply(const MultiApplier::Operations& ops, + const HostAndPort& source) override; }; diff --git a/src/mongo/db/repl/data_replicator_test.cpp b/src/mongo/db/repl/data_replicator_test.cpp index 8e6352fe218..939989d8aff 100644 --- a/src/mongo/db/repl/data_replicator_test.cpp +++ b/src/mongo/db/repl/data_replicator_test.cpp @@ -106,11 +106,6 @@ public: * clear/reset state */ void reset() { - _applierFn = [](const MultiApplier::Operations&) {}; - _multiApplyFn = [](OperationContext*, - const MultiApplier::Operations& ops, - MultiApplier::ApplyOperationFn) - -> StatusWith<OpTime> { return ops.back().getOpTime(); }; _rollbackFn = [](OperationContext*, const OpTime&, const HostAndPort&) -> Status { return Status::OK(); }; _setMyLastOptime = [this](const OpTime& opTime) { _myLastOpTime = opTime; }; @@ -181,6 +176,10 @@ public: return *_dr; } + DataReplicatorExternalStateMock* getExternalState() { + return _externalState; + } + protected: void setUp() override { ReplicationExecutorTest::setUp(); @@ -194,11 +193,6 @@ protected: DataReplicatorOptions options; options.initialSyncRetryWait = Milliseconds(0); - options.applierFn = [this](const MultiApplier::Operations& ops) { return _applierFn(ops); }; - options.multiApplyFn = - [this](OperationContext* txn, - const MultiApplier::Operations& ops, - MultiApplier::ApplyOperationFn func) { return _multiApplyFn(txn, ops, func); }; options.rollbackFn = [this](OperationContext* txn, const OpTime& lastOpTimeWritten, const HostAndPort& syncSource) -> Status { @@ -231,6 +225,7 @@ protected: auto dataReplicatorExternalState = stdx::make_unique<DataReplicatorExternalStateMock>(); dataReplicatorExternalState->currentTerm = 1LL; dataReplicatorExternalState->lastCommittedOpTime = _myLastOpTime; + _externalState = dataReplicatorExternalState.get(); try { _dr.reset(new DataReplicator( @@ -246,8 +241,6 @@ protected: // Executor may still invoke callback before shutting down. } - MultiApplier::ApplyOperationFn _applierFn; - MultiApplier::MultiApplyFn _multiApplyFn; DataReplicatorOptions::RollbackFn _rollbackFn; DataReplicatorOptions::SetMyLastOptimeFn _setMyLastOptime; OpTime _myLastOpTime; @@ -255,6 +248,7 @@ protected: std::unique_ptr<SyncSourceSelector> _syncSourceSelector; private: + DataReplicatorExternalStateMock* _externalState; std::unique_ptr<DataReplicator> _dr; }; @@ -521,7 +515,7 @@ TEST_F(InitialSyncTest, Complete) { TEST_F(InitialSyncTest, MissingDocOnMultiApplyCompletes) { DataReplicatorOptions opts; int applyCounter{0}; - _multiApplyFn = + getExternalState()->multiApplyFn = [&](OperationContext*, const MultiApplier::Operations& ops, MultiApplier::ApplyOperationFn) -> StatusWith<OpTime> { if (++applyCounter == 1) { @@ -944,7 +938,7 @@ TEST_F(SteadyStateTest, PauseDataReplicator) { unittest::Barrier barrier(2U); Timestamp lastTimestampApplied; BSONObj operationApplied; - _multiApplyFn = + getExternalState()->multiApplyFn = [&](OperationContext*, const MultiApplier::Operations& ops, MultiApplier::ApplyOperationFn) -> StatusWith<OpTime> { stdx::lock_guard<stdx::mutex> lock(mutex); @@ -1036,7 +1030,7 @@ TEST_F(SteadyStateTest, ApplyOneOperation) { unittest::Barrier barrier(2U); Timestamp lastTimestampApplied; BSONObj operationApplied; - _multiApplyFn = + getExternalState()->multiApplyFn = [&](OperationContext*, const MultiApplier::Operations& ops, MultiApplier::ApplyOperationFn) -> StatusWith<OpTime> { stdx::lock_guard<stdx::mutex> lock(mutex); diff --git a/src/mongo/db/repl/replication_coordinator_external_state.h b/src/mongo/db/repl/replication_coordinator_external_state.h index 7be2d475eda..4a58018474a 100644 --- a/src/mongo/db/repl/replication_coordinator_external_state.h +++ b/src/mongo/db/repl/replication_coordinator_external_state.h @@ -33,6 +33,7 @@ #include "mongo/base/disallow_copying.h" #include "mongo/bson/timestamp.h" #include "mongo/db/repl/member_state.h" +#include "mongo/db/repl/multiapplier.h" #include "mongo/db/repl/optime.h" #include "mongo/stdx/functional.h" #include "mongo/util/time_support.h" @@ -268,6 +269,26 @@ public: * Returns true if the current storage engine supports read committed. */ virtual bool isReadCommittedSupportedByStorageEngine(OperationContext* txn) const = 0; + + /** + * Applies the operations described in the oplog entries contained in "ops" using the + * "applyOperation" function. + */ + virtual StatusWith<OpTime> multiApply(OperationContext* txn, + const MultiApplier::Operations& ops, + MultiApplier::ApplyOperationFn applyOperation) = 0; + + /** + * Used by multiApply() to writes operations to database during steady state replication. + */ + virtual void multiSyncApply(const MultiApplier::Operations& ops) = 0; + + /** + * Used by multiApply() to writes operations to database during initial sync. + * Fetches missing documents from "source". + */ + virtual void multiInitialSyncApply(const MultiApplier::Operations& ops, + const HostAndPort& source) = 0; }; } // namespace repl 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 152016b2ddc..e5042d8f24b 100644 --- a/src/mongo/db/repl/replication_coordinator_external_state_impl.cpp +++ b/src/mongo/db/repl/replication_coordinator_external_state_impl.cpp @@ -57,6 +57,7 @@ #include "mongo/db/repl/rs_initialsync.h" #include "mongo/db/repl/snapshot_thread.h" #include "mongo/db/repl/storage_interface.h" +#include "mongo/db/repl/sync_tail.h" #include "mongo/db/server_parameters.h" #include "mongo/db/service_context.h" #include "mongo/db/s/sharding_state.h" @@ -107,7 +108,7 @@ void ReplicationCoordinatorExternalStateImpl::startInitialSync(OnInitialSyncFini invariant(!(bgsync == nullptr && !inShutdownStrict())); // bgsync can be null @shutdown. invariant(!_producerThread); // The producer thread should not be init'd before this. _producerThread.reset( - new stdx::thread(stdx::bind(&BackgroundSync::producerThread, bgsync))); + new stdx::thread(stdx::bind(&BackgroundSync::producerThread, bgsync, this))); // Do initial sync. syncDoInitialSync(); finished(); @@ -119,7 +120,7 @@ void ReplicationCoordinatorExternalStateImpl::startSteadyStateReplication() { log() << "Starting replication fetcher thread"; BackgroundSync* bgsync = BackgroundSync::get(); _producerThread.reset( - new stdx::thread(stdx::bind(&BackgroundSync::producerThread, bgsync))); + new stdx::thread(stdx::bind(&BackgroundSync::producerThread, bgsync, this))); } log() << "Starting replication applier threads"; invariant(!_applierThread); @@ -501,6 +502,30 @@ bool ReplicationCoordinatorExternalStateImpl::isReadCommittedSupportedByStorageE return storageEngine->getSnapshotManager(); } +StatusWith<OpTime> ReplicationCoordinatorExternalStateImpl::multiApply( + OperationContext* txn, + const MultiApplier::Operations& ops, + MultiApplier::ApplyOperationFn applyOperation) { + return repl::multiApply(txn, ops, applyOperation); +} + +void ReplicationCoordinatorExternalStateImpl::multiSyncApply(const MultiApplier::Operations& ops) { + // SyncTail* argument is not used by repl::multiSyncApply(). + repl::multiSyncApply(ops, nullptr); +} + +void ReplicationCoordinatorExternalStateImpl::multiInitialSyncApply( + const MultiApplier::Operations& ops, const HostAndPort& source) { + // repl::multiInitialSyncApply uses SyncTail::shouldRetry() (and implicitly getMissingDoc()) + // to fetch missing documents during initial sync. Therefore, it is fine to construct SyncTail + // with invalid BackgroundSync and MultiSyncApplyFunc arguments because we will not be accessing + // any SyncTail functionality that require these constructor parameters. + SyncTail syncTail(nullptr, SyncTail::MultiSyncApplyFunc()); + syncTail.setHostname(source.toString()); + repl::multiInitialSyncApply(ops, &syncTail); +} + + JournalListener::Token ReplicationCoordinatorExternalStateImpl::getToken() { return repl::getGlobalReplicationCoordinator()->getMyLastAppliedOpTime(); } diff --git a/src/mongo/db/repl/replication_coordinator_external_state_impl.h b/src/mongo/db/repl/replication_coordinator_external_state_impl.h index e08cf1fb75e..e8b8aaa7207 100644 --- a/src/mongo/db/repl/replication_coordinator_external_state_impl.h +++ b/src/mongo/db/repl/replication_coordinator_external_state_impl.h @@ -87,6 +87,12 @@ public: virtual void notifyOplogMetadataWaiters(); virtual double getElectionTimeoutOffsetLimitFraction() const; virtual bool isReadCommittedSupportedByStorageEngine(OperationContext* txn) const; + virtual StatusWith<OpTime> multiApply(OperationContext* txn, + const MultiApplier::Operations& ops, + MultiApplier::ApplyOperationFn applyOperation) override; + virtual void multiSyncApply(const MultiApplier::Operations& ops) override; + virtual void multiInitialSyncApply(const MultiApplier::Operations& ops, + const HostAndPort& source) override; std::string getNextOpContextThreadName(); diff --git a/src/mongo/db/repl/replication_coordinator_external_state_mock.cpp b/src/mongo/db/repl/replication_coordinator_external_state_mock.cpp index 0396bb37b5f..2d7736def87 100644 --- a/src/mongo/db/repl/replication_coordinator_external_state_mock.cpp +++ b/src/mongo/db/repl/replication_coordinator_external_state_mock.cpp @@ -236,6 +236,16 @@ bool ReplicationCoordinatorExternalStateMock::isReadCommittedSupportedByStorageE return _isReadCommittedSupported; } +StatusWith<OpTime> ReplicationCoordinatorExternalStateMock::multiApply( + OperationContext*, const MultiApplier::Operations&, MultiApplier::ApplyOperationFn) { + return {ErrorCodes::InternalError, "Method not implemented"}; +} + +void ReplicationCoordinatorExternalStateMock::multiSyncApply(const MultiApplier::Operations& ops) {} + +void ReplicationCoordinatorExternalStateMock::multiInitialSyncApply( + const MultiApplier::Operations& ops, const HostAndPort& source) {} + void ReplicationCoordinatorExternalStateMock::setIsReadCommittedEnabled(bool val) { _isReadCommittedSupported = val; } diff --git a/src/mongo/db/repl/replication_coordinator_external_state_mock.h b/src/mongo/db/repl/replication_coordinator_external_state_mock.h index 77d883df480..049f39745c8 100644 --- a/src/mongo/db/repl/replication_coordinator_external_state_mock.h +++ b/src/mongo/db/repl/replication_coordinator_external_state_mock.h @@ -88,6 +88,12 @@ public: virtual void notifyOplogMetadataWaiters(); virtual double getElectionTimeoutOffsetLimitFraction() const; virtual bool isReadCommittedSupportedByStorageEngine(OperationContext* txn) const; + virtual StatusWith<OpTime> multiApply(OperationContext* txn, + const MultiApplier::Operations& ops, + MultiApplier::ApplyOperationFn applyOperation) override; + virtual void multiSyncApply(const MultiApplier::Operations& ops) override; + virtual void multiInitialSyncApply(const MultiApplier::Operations& ops, + const HostAndPort& source) override; /** * Adds "host" to the list of hosts that this mock will match when responding to "isSelf" diff --git a/src/mongo/db/repl/replication_coordinator_impl.cpp b/src/mongo/db/repl/replication_coordinator_impl.cpp index 14eba0564ed..3b5319b8cc7 100644 --- a/src/mongo/db/repl/replication_coordinator_impl.cpp +++ b/src/mongo/db/repl/replication_coordinator_impl.cpp @@ -203,10 +203,6 @@ ReplicationCoordinator::Mode getReplicationModeFromSettings(const ReplSettings& DataReplicatorOptions createDataReplicatorOptions(ReplicationCoordinator* replCoord) { DataReplicatorOptions options; - options.applierFn = [](const MultiApplier::Operations&) {}; - options.multiApplyFn = - [](OperationContext*, const MultiApplier::Operations&, MultiApplier::ApplyOperationFn) - -> OpTime { return OpTime(); }; options.rollbackFn = [](OperationContext*, const OpTime&, const HostAndPort&) -> Status { return Status::OK(); }; options.prepareReplSetUpdatePositionCommandFn = @@ -256,7 +252,7 @@ ReplicationCoordinatorImpl::ReplicationCoordinatorImpl( _canAcceptNonLocalWrites(!(settings.usingReplSets() || settings.isSlave())), _canServeNonLocalReads(0U), _dr(createDataReplicatorOptions(this), - stdx::make_unique<DataReplicatorExternalStateImpl>(this), + stdx::make_unique<DataReplicatorExternalStateImpl>(this, externalState), &_replExecutor), _isDurableStorageEngine(isDurableStorageEngineFn ? *isDurableStorageEngineFn : []() -> bool { return getGlobalServiceContext()->getGlobalStorageEngine()->isDurable(); diff --git a/src/mongo/db/repl/sync_tail.h b/src/mongo/db/repl/sync_tail.h index d2758fc478c..21459b723a6 100644 --- a/src/mongo/db/repl/sync_tail.h +++ b/src/mongo/db/repl/sync_tail.h @@ -180,7 +180,7 @@ private: }; /** - * Applies the opeartions described in the oplog entries contained in "ops" using the + * Applies the operations described in the oplog entries contained in "ops" using the * "applyOperation" function. * * Returns ErrorCode::InterruptedAtShutdown if the node enters shutdown while applying ops, |