mongodb 数据块迁移的源码分析

1. 简介

上一篇我们聊到了mongodb数据块的基本概念，和数据块迁移的主要流程，这篇文章我们聊聊源码实现部分。
2. 迁移序列图

数据块迁移的请求是从配置服务器（config server）发给（donor，捐献方），再有捐献方发起迁移请求给目标节点（recipient，接收方），后续迁移由捐献方和接收方配合完成。
数据迁移结束时，捐献方再提交迁移结果给配置服务器，三方交互序列图如下：

 
可以看到，序列图中的5个步骤，是对应前面文章的迁移流程中的5个步骤，其中接收方的流程控制代码在migration_destination_manager.cpp中的_migrateDriver方法中，捐献方的流程控制代码在donor的move_chunk_command.cpp中的_runImpl方法中完成，代码如下：

1. static void _runImpl(OperationContext* opCtx, const MoveChunkRequest& moveChunkRequest) {
2.         const auto writeConcernForRangeDeleter =
3.             uassertStatusOK(ChunkMoveWriteConcernOptions::getEffectiveWriteConcern(
4.                 opCtx, moveChunkRequest.getSecondaryThrottle()));
6.         // Resolve the donor and recipient shards and their connection string
7.         auto const shardRegistry = Grid::get(opCtx)->shardRegistry();
8.         // 准备donor和recipient的连接
9.         const auto donorConnStr =
10.             uassertStatusOK(shardRegistry->getShard(opCtx, moveChunkRequest.getFromShardId()))
11.                 ->getConnString();
12.         const auto recipientHost = uassertStatusOK([&] {
13.             auto recipientShard =
14.                 uassertStatusOK(shardRegistry->getShard(opCtx, moveChunkRequest.getToShardId()));
16.             return recipientShard->getTargeter()->findHost(
17.                 opCtx, ReadPreferenceSetting{ReadPreference::PrimaryOnly});
18.         }());
20.         std::string unusedErrMsg;
21.         // 用于统计每一步的耗时情况
22.         MoveTimingHelper moveTimingHelper(opCtx,
23.                                           "from",
24.                                           moveChunkRequest.getNss().ns(),
25.                                           moveChunkRequest.getMinKey(),
26.                                           moveChunkRequest.getMaxKey(),
27.                                           6,  // Total number of steps
28.                                           &unusedErrMsg,
29.                                           moveChunkRequest.getToShardId(),
30.                                           moveChunkRequest.getFromShardId());
32.         moveTimingHelper.done(1);
33.         moveChunkHangAtStep1.pauseWhileSet();
35.         if (moveChunkRequest.getFromShardId() == moveChunkRequest.getToShardId()) {
36.             // TODO: SERVER-46669 handle wait for delete.
37.             return;
38.         }
39.         // 构建迁移任务管理器
40.         MigrationSourceManager migrationSourceManager(
41.             opCtx, moveChunkRequest, donorConnStr, recipientHost);
43.         moveTimingHelper.done(2);
44.         moveChunkHangAtStep2.pauseWhileSet();
46.         // 向接收方发送迁移命令
47.         uassertStatusOKWithWarning(migrationSourceManager.startClone());
48.         moveTimingHelper.done(3);
49.         moveChunkHangAtStep3.pauseWhileSet();
51.         // 等待块数据和变更数据都拷贝完成
52.         uassertStatusOKWithWarning(migrationSourceManager.awaitToCatchUp());
53.         moveTimingHelper.done(4);
54.         moveChunkHangAtStep4.pauseWhileSet();
56.         // 进入临界区
57.         uassertStatusOKWithWarning(migrationSourceManager.enterCriticalSection());
59.         // 通知接收方
60.         uassertStatusOKWithWarning(migrationSourceManager.commitChunkOnRecipient());
61.         moveTimingHelper.done(5);
62.         moveChunkHangAtStep5.pauseWhileSet();
64.         // 在配置服务器提交分块元数据信息
65.         uassertStatusOKWithWarning(migrationSourceManager.commitChunkMetadataOnConfig());
66.         moveTimingHelper.done(6);
67.         moveChunkHangAtStep6.pauseWhileSet();
68.     }

复制代码

下面对每一个步骤的代码做分析。
3. 各步骤源码分析

3.1 启动迁移（ _recvChunkStart）

 
在启动阶段，捐献方主要做了三件事：
1. 参数检查，在MigrationSourceManager 构造函数中完成，不再赘述。
2. 注册监听器，用于记录在迁移期间该数据块内发生的变更数据，代码如下：
3. 向接收方发送迁移命令_recvChunkStart。
步骤2和3的代码实现在一个方法中，如下：

1. Status MigrationSourceManager::startClone() {
2.     ...// 省略了部分代码
4.     _cloneAndCommitTimer.reset();
6.     auto replCoord = repl::ReplicationCoordinator::get(_opCtx);
7.     auto replEnabled = replCoord->isReplEnabled();
9.     {
10.         const auto metadata = _getCurrentMetadataAndCheckEpoch();
12.         // Having the metadata manager registered on the collection sharding state is what indicates
13.         // that a chunk on that collection is being migrated. With an active migration, write
14.         // operations require the cloner to be present in order to track changes to the chunk which
15.         // needs to be transmitted to the recipient.
16.         // 注册监听器，_cloneDriver除了迁移数据外，还会用于记录在迁移过程中该数据块增量变化的数据（比如新增的数据）
17.         _cloneDriver = std::make_unique<MigrationChunkClonerSourceLegacy>(
18.             _args, metadata.getKeyPattern(), _donorConnStr, _recipientHost);
20.         AutoGetCollection autoColl(_opCtx,
21.                                    getNss(),
22.                                    replEnabled ? MODE_IX : MODE_X,
23.                                    AutoGetCollectionViewMode::kViewsForbidden,
24.                                    _opCtx->getServiceContext()->getPreciseClockSource()->now() +
25.                                        Milliseconds(migrationLockAcquisitionMaxWaitMS.load()));
27.         auto csr = CollectionShardingRuntime::get(_opCtx, getNss());
28.         auto lockedCsr = CollectionShardingRuntime::CSRLock::lockExclusive(_opCtx, csr);
29.         invariant(nullptr == std::exchange(msmForCsr(csr), this));
31.         _coordinator = std::make_unique<migrationutil::MigrationCoordinator>(
32.             _cloneDriver->getSessionId(),
33.             _args.getFromShardId(),
34.             _args.getToShardId(),
35.             getNss(),
36.             *_collectionUUID,
37.             ChunkRange(_args.getMinKey(), _args.getMaxKey()),
38.             _chunkVersion,
39.             _args.getWaitForDelete());
41.         _state = kCloning;
42.     }
44.     if (replEnabled) {
45.         auto const readConcernArgs = repl::ReadConcernArgs(
46.             replCoord->getMyLastAppliedOpTime(), repl::ReadConcernLevel::kLocalReadConcern);
48.         // 检查当前节点状态是否满足repl::ReadConcernLevel::kLocalReadConcern
49.         auto waitForReadConcernStatus =
50.             waitForReadConcern(_opCtx, readConcernArgs, StringData(), false);
51.         if (!waitForReadConcernStatus.isOK()) {
52.             return waitForReadConcernStatus;
53.         }
54.         setPrepareConflictBehaviorForReadConcern(
55.             _opCtx, readConcernArgs, PrepareConflictBehavior::kEnforce);
56.     }
58.     _coordinator->startMigration(_opCtx);
60.     // 向接收方发送开始拷贝数据的命令（_recvChunkStart）
61.     Status startCloneStatus = _cloneDriver->startClone(_opCtx,
62.                                                        _coordinator->getMigrationId(),
63.                                                        _coordinator->getLsid(),
64.                                                        _coordinator->getTxnNumber());
65.     if (!startCloneStatus.isOK()) {
66.         return startCloneStatus;
67.     }
69.     scopedGuard.dismiss();
70.     return Status::OK();
71. }

复制代码

 
接收方在收到迁移请求后，会先检查本地是否有该表，如果没有的话，会先建表会创建表的索引：

1. void MigrationDestinationManager::cloneCollectionIndexesAndOptions(
2.     OperationContext* opCtx,
3.     const NamespaceString& nss,
4.     const CollectionOptionsAndIndexes& collectionOptionsAndIndexes) {
5.     {
6.         // 1. Create the collection (if it doesn't already exist) and create any indexes we are
7.         // missing (auto-heal indexes).
9.         ...// 省略部分代码
11.         {
12.             AutoGetCollection collection(opCtx, nss, MODE_IS);
13.             // 如果存在表，且不缺索引，则退出
14.             if (collection) {
15.                 checkUUIDsMatch(collection.getCollection());
16.                 auto indexSpecs =
17.                     checkEmptyOrGetMissingIndexesFromDonor(collection.getCollection());
18.                 if (indexSpecs.empty()) {
19.                     return;
20.                 }
21.             }
22.         }
24.         // Take the exclusive database lock if the collection does not exist or indexes are missing
25.         // (needs auto-heal).
26.         // 建表时，需要对数据库加锁
27.         AutoGetDb autoDb(opCtx, nss.db(), MODE_X);
28.         auto db = autoDb.ensureDbExists();
30.         auto collection = CollectionCatalog::get(opCtx)->lookupCollectionByNamespace(opCtx, nss);
31.         if (collection) {
32.             checkUUIDsMatch(collection);
33.         } else {
34.             ...// 省略部分代码// We do not have a collection by this name. Create the collection with the donor's
35.             // options.
36.             // 建表
37.             OperationShardingState::ScopedAllowImplicitCollectionCreate_UNSAFE
38.                 unsafeCreateCollection(opCtx);
39.             WriteUnitOfWork wuow(opCtx);
40.             CollectionOptions collectionOptions = uassertStatusOK(
41.                 CollectionOptions::parse(collectionOptionsAndIndexes.options,
42.                                          CollectionOptions::ParseKind::parseForStorage));
43.             const bool createDefaultIndexes = true;
44.             uassertStatusOK(db->userCreateNS(opCtx,
45.                                              nss,
46.                                              collectionOptions,
47.                                              createDefaultIndexes,
48.                                              collectionOptionsAndIndexes.idIndexSpec));
49.             wuow.commit();
50.             collection = CollectionCatalog::get(opCtx)->lookupCollectionByNamespace(opCtx, nss);
51.         }
52.         // 创建对应的索引
53.         auto indexSpecs = checkEmptyOrGetMissingIndexesFromDonor(collection);
54.         if (!indexSpecs.empty()) {
55.             WriteUnitOfWork wunit(opCtx);
56.             auto fromMigrate = true;
57.             CollectionWriter collWriter(opCtx, collection->uuid());
58.             IndexBuildsCoordinator::get(opCtx)->createIndexesOnEmptyCollection(
59.                 opCtx, collWriter, indexSpecs, fromMigrate);
60.             wunit.commit();
61.         }
62.     }
63. }

复制代码

 
3.2 接收方拉取存量数据（ _migrateClone）

接收方的拉取存量数据时，做了六件事情：
1. 定义了一个批量插入记录的方法。
2. 定义了一个批量拉取数据的方法。
3. 定义生产者和消费队列。
4. 启动数据写入线程，该线程会消费队列中的数据，并调用批量插入记录的方法把记录保存到本地。
5. 循环向捐献方发起拉取数据请求（步骤2的方法），并写入步骤3的队列中。
6. 数据拉取结束后（写入空记录到队列中，触发步骤5结束），则同步等待步骤5的线程也结束。
详细代码如下：

1. // 1. 定义批量写入函数
2.         auto insertBatchFn = [&](OperationContext* opCtx, BSONObj arr) {
3.             auto it = arr.begin();
4.             while (it != arr.end()) {
5.                 int batchNumCloned = 0;
6.                 int batchClonedBytes = 0;
7.                 const int batchMaxCloned = migrateCloneInsertionBatchSize.load();
9.                 assertNotAborted(opCtx);
11.                 write_ops::InsertCommandRequest insertOp(_nss);
12.                 insertOp.getWriteCommandRequestBase().setOrdered(true);
13.                 insertOp.setDocuments([&] {
14.                     std::vector<BSONObj> toInsert;
15.                     while (it != arr.end() &&
16.                            (batchMaxCloned <= 0 || batchNumCloned < batchMaxCloned)) {
17.                         const auto& doc = *it;
18.                         BSONObj docToClone = doc.Obj();
19.                         toInsert.push_back(docToClone);
20.                         batchNumCloned++;
21.                         batchClonedBytes += docToClone.objsize();
22.                         ++it;
23.                     }
24.                     return toInsert;
25.                 }());
27.                 const auto reply =
28.                     write_ops_exec::performInserts(opCtx, insertOp, OperationSource::kFromMigrate);
30.                 for (unsigned long i = 0; i < reply.results.size(); ++i) {
31.                     uassertStatusOKWithContext(
32.                         reply.results[i],
33.                         str::stream() << "Insert of " << insertOp.getDocuments()[i] << " failed.");
34.                 }
36.                 {
37.                     stdx::lock_guard<Latch> statsLock(_mutex);
38.                     _numCloned += batchNumCloned;
39.                     ShardingStatistics::get(opCtx).countDocsClonedOnRecipient.addAndFetch(
40.                         batchNumCloned);
41.                     _clonedBytes += batchClonedBytes;
42.                 }
43.                 if (_writeConcern.needToWaitForOtherNodes()) {
44.                     runWithoutSession(outerOpCtx, [&] {
45.                         repl::ReplicationCoordinator::StatusAndDuration replStatus =
46.                             repl::ReplicationCoordinator::get(opCtx)->awaitReplication(
47.                                 opCtx,
48.                                 repl::ReplClientInfo::forClient(opCtx->getClient()).getLastOp(),
49.                                 _writeConcern);
50.                         if (replStatus.status.code() == ErrorCodes::WriteConcernFailed) {
51.                             LOGV2_WARNING(
52.                                 22011,
53.                                 "secondaryThrottle on, but doc insert timed out; continuing",
54.                                 "migrationId"_attr = _migrationId->toBSON());
55.                         } else {
56.                             uassertStatusOK(replStatus.status);
57.                         }
58.                     });
59.                 }
61.                 sleepmillis(migrateCloneInsertionBatchDelayMS.load());
62.             }
63.         };
64.         // 2. 定义批量拉取函数
65.         auto fetchBatchFn = [&](OperationContext* opCtx) {
66.             auto res = uassertStatusOKWithContext(
67.                 fromShard->runCommand(opCtx,
68.                                       ReadPreferenceSetting(ReadPreference::PrimaryOnly),
69.                                       "admin",
70.                                       migrateCloneRequest,
71.                                       Shard::RetryPolicy::kNoRetry),
72.                 "_migrateClone failed: ");
74.             uassertStatusOKWithContext(Shard::CommandResponse::getEffectiveStatus(res),
75.                                        "_migrateClone failed: ");
77.             return res.response;
78.         };
80. SingleProducerSingleConsumerQueue<BSONObj>::Options options;
81.     options.maxQueueDepth = 1;
82.     // 3. 使用生产者和消费者队列来把同步的数据写入到本地
83.     SingleProducerSingleConsumerQueue<BSONObj> batches(options);
84.     repl::OpTime lastOpApplied;
86.     // 4. 定义写数据线程，该线程会读取队列中的数据并写入本地节点，直到无需要同步的数据时线程退出
87.     stdx::thread inserterThread{[&] {
88.         Client::initThread("chunkInserter", opCtx->getServiceContext(), nullptr);
89.         auto client = Client::getCurrent();
90.         {
91.             stdx::lock_guard lk(*client);
92.             client->setSystemOperationKillableByStepdown(lk);
93.         }
94.         auto executor =
95.             Grid::get(opCtx->getServiceContext())->getExecutorPool()->getFixedExecutor();
96.         auto inserterOpCtx = CancelableOperationContext(
97.             cc().makeOperationContext(), opCtx->getCancellationToken(), executor);
99.         auto consumerGuard = makeGuard([&] {
100.             batches.closeConsumerEnd();
101.             lastOpApplied = repl::ReplClientInfo::forClient(inserterOpCtx->getClient()).getLastOp();
102.         });
104.         try {
105.             while (true) {
106.                 auto nextBatch = batches.pop(inserterOpCtx.get());
107.                 auto arr = nextBatch["objects"].Obj();
108.                 if (arr.isEmpty()) {
109.                     return;
110.                 }
111.                 insertBatchFn(inserterOpCtx.get(), arr);
112.             }
113.         } catch (...) {
114.             stdx::lock_guard<Client> lk(*opCtx->getClient());
115.             opCtx->getServiceContext()->killOperation(lk, opCtx, ErrorCodes::Error(51008));
116.             LOGV2(21999,
117.                   "Batch insertion failed: {error}",
118.                   "Batch insertion failed",
119.                   "error"_attr = redact(exceptionToStatus()));
120.         }
121.     }};
124.     {
125.         //6.  makeGuard的作用是延迟执行inserterThread.join()
126.         auto inserterThreadJoinGuard = makeGuard([&] {
127.             batches.closeProducerEnd();
128.             inserterThread.join();
129.         });
130.         // 5. 向捐献方发起拉取请求，并把数据写入队列中
131.         while (true) {
132.             auto res = fetchBatchFn(opCtx);
133.             try {
134.                 batches.push(res.getOwned(), opCtx);
135.                 auto arr = res["objects"].Obj();
136.                 if (arr.isEmpty()) {
137.                     break;
138.                 }
139.             } catch (const ExceptionFor<ErrorCodes::ProducerConsumerQueueEndClosed>&) {
140.                 break;
141.             }
142.         }
143.     }  // This scope ensures that the guard is destroyed

复制代码

3.3 接收方拉取变更数据（ _recvChunkStart）

在本步骤，接收方会再拉取变更数据，即在前面迁移过程中，捐献方上发生的针对该数据块的写入、更新和删除的记录，代码如下：

1. // 同步变更数据（_transferMods）
2.     const BSONObj xferModsRequest = createTransferModsRequest(_nss, *_sessionId);
4.     {
5.         // 5. Do bulk of mods
6.         // 5. 批量拉取变更数据，循环拉取，直至无变更数据
7.         _setState(CATCHUP);
9.         while (true) {
10.             auto res = uassertStatusOKWithContext(
11.                 fromShard->runCommand(opCtx,
12.                                       ReadPreferenceSetting(ReadPreference::PrimaryOnly),
13.                                       "admin",
14.                                       xferModsRequest,
15.                                       Shard::RetryPolicy::kNoRetry),
16.                 "_transferMods failed: ");
18.             uassertStatusOKWithContext(Shard::CommandResponse::getEffectiveStatus(res),
19.                                        "_transferMods failed: ");
21.             const auto& mods = res.response;
23.             if (mods["size"].number() == 0) {
24.                 // There are no more pending modifications to be applied. End the catchup phase
25.                 // 无变更数据时，停止循环
26.                 break;
27.             }
28.             // 应用拉取到的变更数据
29.             if (!_applyMigrateOp(opCtx, mods, &lastOpApplied)) {
30.                 continue;
31.             }
33.             const int maxIterations = 3600 * 50;
35.             // 等待从节点完成数据同步
36.             int i;
37.             for (i = 0; i < maxIterations; i++) {
38.                 opCtx->checkForInterrupt();
39.                 outerOpCtx->checkForInterrupt();
41.                 if (getState() == ABORT) {
42.                     LOGV2(22002,
43.                           "Migration aborted while waiting for replication at catch up stage",
44.                           "migrationId"_attr = _migrationId->toBSON());
45.                     return;
46.                 }
48.                 if (runWithoutSession(outerOpCtx, [&] {
49.                         return opReplicatedEnough(opCtx, lastOpApplied, _writeConcern);
50.                     })) {
51.                     break;
52.                 }
54.                 if (i > 100) {
55.                     LOGV2(22003,
56.                           "secondaries having hard time keeping up with migrate",
57.                           "migrationId"_attr = _migrationId->toBSON());
58.                 }
60.                 sleepmillis(20);
61.             }
63.             if (i == maxIterations) {
64.                 _setStateFail("secondary can't keep up with migrate");
65.                 return;
66.             }
67.         }
69.         timing.done(5);
70.         migrateThreadHangAtStep5.pauseWhileSet();
71.     }

复制代码

变更数据拉取结束，就进入等待捐献方进入临界区，在临界区内，捐献方会阻塞写入请求，因此在未进入临界区前，仍然需要拉取变更数据：

1.         // 6. Wait for commit
2.         // 6. 等待donor进入临界区
3.         _setState(STEADY);
5.         bool transferAfterCommit = false;
6.         while (getState() == STEADY || getState() == COMMIT_START) {
7.             opCtx->checkForInterrupt();
8.             outerOpCtx->checkForInterrupt();
10.             // Make sure we do at least one transfer after recv'ing the commit message. If we
11.             // aren't sure that at least one transfer happens *after* our state changes to
12.             // COMMIT_START, there could be mods still on the FROM shard that got logged
13.             // *after* our _transferMods but *before* the critical section.
14.             if (getState() == COMMIT_START) {
15.                 transferAfterCommit = true;
16.             }
18.             auto res = uassertStatusOKWithContext(
19.                 fromShard->runCommand(opCtx,
20.                                       ReadPreferenceSetting(ReadPreference::PrimaryOnly),
21.                                       "admin",
22.                                       xferModsRequest,
23.                                       Shard::RetryPolicy::kNoRetry),
24.                 "_transferMods failed in STEADY STATE: ");
26.             uassertStatusOKWithContext(Shard::CommandResponse::getEffectiveStatus(res),
27.                                        "_transferMods failed in STEADY STATE: ");
29.             auto mods = res.response;
31.             // 如果请求到变更数据，则应用到本地，并继续请求变更数据，直到所有变更数据都迁移结束
32.             if (mods["size"].number() > 0 && _applyMigrateOp(opCtx, mods, &lastOpApplied)) {
33.                 continue;
34.             }
36.             if (getState() == ABORT) {
37.                 LOGV2(22006,
38.                       "Migration aborted while transferring mods",
39.                       "migrationId"_attr = _migrationId->toBSON());
40.                 return;
41.             }
43.             // We know we're finished when:
44.             // 1) The from side has told us that it has locked writes (COMMIT_START)
45.             // 2) We've checked at least one more time for un-transmitted mods
46.             // 检查transferAfterCommit的原因：进入COMMIT_START(临界区)后，需要再拉取一次变更数据
47.             if (getState() == COMMIT_START && transferAfterCommit == true) {
48.                 // 检查所有数据同步到从节点后，数据迁移流程结束
49.                 if (runWithoutSession(outerOpCtx,
50.                                       [&] { return _flushPendingWrites(opCtx, lastOpApplied); })) {
51.                     break;
52.                 }
53.             }
55.             // Only sleep if we aren't committing
56.             if (getState() == STEADY)
57.                 sleepmillis(10);
58.         }

复制代码

 
3.4 进入临界区（ _recvChunkStatus，_recvChunkCommit）

在该步骤，捐献方主要做了三件事：
1. 等待接收方完成数据同步（_recvChunkStatus）。
2. 标记本节点进入临界区，阻塞写操作。
3. 通知接收方进入临界区（_recvChunkCommit）。
相关代码如下：

1. Status MigrationSourceManager::awaitToCatchUp() {
2.     invariant(!_opCtx->lockState()->isLocked());
3.     invariant(_state == kCloning);
4.     auto scopedGuard = makeGuard([&] { cleanupOnError(); });
5.     _stats.totalDonorChunkCloneTimeMillis.addAndFetch(_cloneAndCommitTimer.millis());
6.     _cloneAndCommitTimer.reset();
8.     // Block until the cloner deems it appropriate to enter the critical section.
9.     // 等待数据拷贝完成，这里会向接收方发送_recvChunkStatus，检查接收方的状态是否是STEADY
10.     Status catchUpStatus = _cloneDriver->awaitUntilCriticalSectionIsAppropriate(
11.         _opCtx, kMaxWaitToEnterCriticalSectionTimeout);
12.     if (!catchUpStatus.isOK()) {
13.         return catchUpStatus;
14.     }
16.     _state = kCloneCaughtUp;
17.     scopedGuard.dismiss();
18.     return Status::OK();
19. }
20. <br>// 进入临界区
21. Status MigrationSourceManager::enterCriticalSection() {
22.     ...// 省略部分代码<br>     // 标记进入临界区，后续更新类操作会被阻塞(通过ShardingMigrationCriticalSection::getSignal()检查该标记)
23.     _critSec.emplace(_opCtx, _args.getNss(), _critSecReason);
25.     _state = kCriticalSection;
27.     // Persist a signal to secondaries that we've entered the critical section. This is will cause
28.     // secondaries to refresh their routing table when next accessed, which will block behind the
29.     // critical section. This ensures causal consistency by preventing a stale mongos with a cluster
30.     // time inclusive of the migration config commit update from accessing secondary data.
31.     // Note: this write must occur after the critSec flag is set, to ensure the secondary refresh
32.     // will stall behind the flag.
33.     // 通知从节点此时主节点已进入临界区，如果有数据访问时要刷新路由信息（保证因果一致性）
34.     Status signalStatus = shardmetadatautil::updateShardCollectionsEntry(
35.         _opCtx,
36.         BSON(ShardCollectionType::kNssFieldName << getNss().ns()),
37.         BSON("$inc" << BSON(ShardCollectionType::kEnterCriticalSectionCounterFieldName << 1)),
38.         false /*upsert*/);
39.     if (!signalStatus.isOK()) {
40.         return {
41.             ErrorCodes::OperationFailed,
42.             str::stream() << "Failed to persist critical section signal for secondaries due to: "
43.                           << signalStatus.toString()};
44.     }
46.     LOGV2(22017,
47.           "Migration successfully entered critical section",
48.           "migrationId"_attr = _coordinator->getMigrationId());
50.     scopedGuard.dismiss();
51.     return Status::OK();
52. }<br><br>

复制代码

 
4. 小结

至此，mongodb的数据块迁移的源代码基本分析完毕，这里补充一下监听变更数据的代码实现。
前面有提到监听变更数据是由_cloneDriver完成的，下面看下_cloneDriver的接口定义：

1. [/code][code]

复制代码

[code][/code] 

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！