diff options
Diffstat (limited to 'src/mongo/db/s/migration_chunk_cloner_source_legacy.cpp')
| -rw-r--r-- | src/mongo/db/s/migration_chunk_cloner_source_legacy.cpp | 414 |
1 files changed, 350 insertions, 64 deletions
diff --git a/src/mongo/db/s/migration_chunk_cloner_source_legacy.cpp b/src/mongo/db/s/migration_chunk_cloner_source_legacy.cpp index 547080ff4ea..89d8168a295 100644 --- a/src/mongo/db/s/migration_chunk_cloner_source_legacy.cpp +++ b/src/mongo/db/s/migration_chunk_cloner_source_legacy.cpp @@ -33,19 +33,26 @@ #include "mongo/db/s/migration_chunk_cloner_source_legacy.h" +#include <fmt/format.h> + #include "mongo/base/status.h" +#include "mongo/bson/bsonobj.h" #include "mongo/client/read_preference.h" #include "mongo/db/catalog/index_catalog.h" #include "mongo/db/catalog_raii.h" +#include "mongo/db/db_raii.h" +#include "mongo/db/dbdirectclient.h" #include "mongo/db/dbhelpers.h" #include "mongo/db/exec/working_set_common.h" #include "mongo/db/index/index_access_method.h" #include "mongo/db/index/index_descriptor.h" #include "mongo/db/ops/write_ops_retryability.h" +#include "mongo/db/query/get_executor.h" #include "mongo/db/repl/optime.h" #include "mongo/db/repl/replication_process.h" #include "mongo/db/s/collection_sharding_runtime.h" #include "mongo/db/s/migration_source_manager.h" +#include "mongo/db/s/operation_sharding_state.h" #include "mongo/db/s/shard_key_index_util.h" #include "mongo/db/s/sharding_runtime_d_params_gen.h" #include "mongo/db/s/sharding_statistics.h" @@ -68,6 +75,8 @@ namespace mongo { namespace { +using namespace fmt::literals; + const char kRecvChunkStatus[] = "_recvChunkStatus"; const char kRecvChunkCommit[] = "_recvChunkCommit"; const char kRecvChunkAbort[] = "_recvChunkAbort"; @@ -76,13 +85,24 @@ const int kMaxObjectPerChunk{250000}; const Hours kMaxWaitToCommitCloneForJumboChunk(6); MONGO_FAIL_POINT_DEFINE(failTooMuchMemoryUsed); +MONGO_FAIL_POINT_DEFINE(hangAfterProcessingDeferredXferMods); + +/** + * Returns true if the given BSON object in the shard key value pair format is within the given + * range. + */ +bool isShardKeyValueInRange(const BSONObj& shardKeyValue, const BSONObj& min, const BSONObj& max) { + return shardKeyValue.woCompare(min) >= 0 && shardKeyValue.woCompare(max) < 0; +} -bool isInRange(const BSONObj& obj, - const BSONObj& min, - const BSONObj& max, - const ShardKeyPattern& shardKeyPattern) { - BSONObj k = shardKeyPattern.extractShardKeyFromDoc(obj); - return k.woCompare(min) >= 0 && k.woCompare(max) < 0; +/** + * Returns true if the given BSON document is within the given chunk range. + */ +bool isDocInRange(const BSONObj& obj, + const BSONObj& min, + const BSONObj& max, + const ShardKeyPattern& shardKeyPattern) { + return isShardKeyValueInRange(shardKeyPattern.extractShardKeyFromDoc(obj), min, max); } BSONObj createRequestWithSessionId(StringData commandName, @@ -96,9 +116,8 @@ BSONObj createRequestWithSessionId(StringData commandName, return builder.obj(); } -BSONObj getDocumentKeyFromReplOperation(repl::ReplOperation replOperation, - repl::OpTypeEnum opType) { - switch (opType) { +BSONObj getDocumentKeyFromReplOperation(repl::ReplOperation replOperation) { + switch (replOperation.getOpType()) { case repl::OpTypeEnum::kInsert: case repl::OpTypeEnum::kDelete: return replOperation.getObject(); @@ -157,6 +176,30 @@ private: const repl::OpTime _opTime; }; +LogTransactionOperationsForShardingHandler::LogTransactionOperationsForShardingHandler( + LogicalSessionId lsid, + const std::vector<repl::OplogEntry>& stmts, + repl::OpTime prepareOrCommitOpTime) + : _lsid(std::move(lsid)), _prepareOrCommitOpTime(std::move(prepareOrCommitOpTime)) { + _stmts.reserve(stmts.size()); + _ownedReplBSONObj.reserve(stmts.size()); + + for (const auto& op : stmts) { + auto ownedBSON = op.getDurableReplOperation().toBSON().getOwned(); + _ownedReplBSONObj.push_back(ownedBSON); + _stmts.push_back( + repl::ReplOperation::parse({"MigrationChunkClonerSource_toReplOperation"}, ownedBSON)); + } +} + +LogTransactionOperationsForShardingHandler::LogTransactionOperationsForShardingHandler( + LogicalSessionId lsid, + const std::vector<repl::ReplOperation>& stmts, + repl::OpTime prepareOrCommitOpTime) + : _lsid(std::move(lsid)), + _stmts(stmts), + _prepareOrCommitOpTime(std::move(prepareOrCommitOpTime)) {} + void LogTransactionOperationsForShardingHandler::commit(boost::optional<Timestamp>) { std::set<NamespaceString> namespacesTouchedByTransaction; @@ -214,39 +257,45 @@ void LogTransactionOperationsForShardingHandler::commit(boost::optional<Timestam continue; } - auto documentKey = getDocumentKeyFromReplOperation(stmt, opType); + auto preImageDocKey = getDocumentKeyFromReplOperation(stmt); - auto idElement = documentKey["_id"]; + auto idElement = preImageDocKey["_id"]; if (idElement.eoo()) { LOGV2_WARNING(21994, - "Received a document without an _id field, ignoring: {documentKey}", "Received a document without an _id and will ignore that document", - "documentKey"_attr = redact(documentKey)); + "documentKey"_attr = redact(preImageDocKey)); continue; } - auto const& minKey = cloner->_args.getMin().get(); - auto const& maxKey = cloner->_args.getMax().get(); - auto const& shardKeyPattern = cloner->_shardKeyPattern; - - if (!isInRange(documentKey, minKey, maxKey, shardKeyPattern)) { - // If the preImageDoc is not in range but the postImageDoc was, we know that the - // document has changed shard keys and no longer belongs in the chunk being cloned. - // We will model the deletion of the preImage document so that the destination chunk - // does not receive an outdated version of this document. - if (opType == repl::OpTypeEnum::kUpdate && - isInRange(stmt.getPreImageDocumentKey(), minKey, maxKey, shardKeyPattern) && - !stmt.getPreImageDocumentKey()["_id"].eoo()) { - opType = repl::OpTypeEnum::kDelete; - idElement = stmt.getPreImageDocumentKey()["id"]; + if (opType == repl::OpTypeEnum::kUpdate) { + auto const& shardKeyPattern = cloner->_shardKeyPattern; + auto preImageShardKeyValues = + shardKeyPattern.extractShardKeyFromDocumentKey(preImageDocKey); + + // If prepare was performed from another term, we will not have the post image doc key + // since it is not persisted in the oplog. + auto postImageDocKey = stmt.getPostImageDocumentKey(); + if (!postImageDocKey.isEmpty()) { + if (!cloner->_processUpdateForXferMod(preImageDocKey, postImageDocKey)) { + // We don't need to add this op to session migration if neither post or pre + // image doc falls within the chunk range. + continue; + } } else { - continue; + // We can't perform reads here using the same recovery unit because the transaction + // is already committed. We instead defer performing the reads when xferMods command + // is called. Also allow this op to be added to session migration since we can't + // tell whether post image doc will fall within the chunk range. If it turns out + // both preImage and postImage doc don't fall into the chunk range, it is not wrong + // for this op to be added to session migration, but it will result in wasted work + // and unneccesary extra oplog storage on the destination. + cloner->_deferProcessingForXferMod(preImageDocKey); } + } else { + cloner->_addToTransferModsQueue(idElement.wrap(), getOpCharForCrudOpType(opType), {}); } addToSessionMigrationOptimeQueueIfNeeded(cloner, nss, _prepareOrCommitOpTime); - - cloner->_addToTransferModsQueue(idElement.wrap(), getOpCharForCrudOpType(opType), {}); } } @@ -379,7 +428,7 @@ StatusWith<BSONObj> MigrationChunkClonerSourceLegacy::commitClone(OperationConte } } else { invariant(PlanExecutor::IS_EOF == _jumboChunkCloneState->clonerState); - invariant(_cloneLocs.empty()); + invariant(!_cloneList.hasMore()); } } @@ -443,7 +492,7 @@ void MigrationChunkClonerSourceLegacy::cancelClone(OperationContext* opCtx) noex } bool MigrationChunkClonerSourceLegacy::isDocumentInMigratingChunk(const BSONObj& doc) { - return isInRange(doc, getMin(), getMax(), _shardKeyPattern); + return isDocInRange(doc, getMin(), getMax(), _shardKeyPattern); } void MigrationChunkClonerSourceLegacy::onInsertOp(OperationContext* opCtx, @@ -462,7 +511,7 @@ void MigrationChunkClonerSourceLegacy::onInsertOp(OperationContext* opCtx, return; } - if (!isInRange(insertedDoc, getMin(), getMax(), _shardKeyPattern)) { + if (!isDocInRange(insertedDoc, getMin(), getMax(), _shardKeyPattern)) { return; } @@ -497,12 +546,12 @@ void MigrationChunkClonerSourceLegacy::onUpdateOp(OperationContext* opCtx, return; } - if (!isInRange(postImageDoc, getMin(), getMax(), _shardKeyPattern)) { + if (!isDocInRange(postImageDoc, getMin(), getMax(), _shardKeyPattern)) { // If the preImageDoc is not in range but the postImageDoc was, we know that the document // has changed shard keys and no longer belongs in the chunk being cloned. We will model // the deletion of the preImage document so that the destination chunk does not receive an // outdated version of this document. - if (preImageDoc && isInRange(*preImageDoc, getMin(), getMax(), _shardKeyPattern)) { + if (preImageDoc && isDocInRange(*preImageDoc, getMin(), getMax(), _shardKeyPattern)) { onDeleteOp(opCtx, *preImageDoc, opTime, prePostImageOpTime); } return; @@ -664,6 +713,7 @@ void MigrationChunkClonerSourceLegacy::_nextCloneBatchFromIndexScan(OperationCon lk.unlock(); ShardingStatistics::get(opCtx).countDocsClonedOnDonor.addAndFetch(1); + ShardingStatistics::get(opCtx).countBytesClonedOnDonor.addAndFetch(obj.objsize()); } } catch (DBException& exception) { exception.addContext("Executor error while scanning for documents belonging to chunk"); @@ -685,38 +735,57 @@ void MigrationChunkClonerSourceLegacy::_nextCloneBatchFromCloneLocs(OperationCon internalQueryExecYieldIterations.load(), Milliseconds(internalQueryExecYieldPeriodMS.load())); - stdx::unique_lock<Latch> lk(_mutex); - auto iter = _cloneLocs.begin(); + while (true) { + int recordsNoLongerExist = 0; + auto docInFlight = _cloneList.getNextDoc(opCtx, collection, &recordsNoLongerExist); + + if (recordsNoLongerExist) { + stdx::lock_guard lk(_mutex); + _numRecordsPassedOver += recordsNoLongerExist; + } + + const auto& doc = docInFlight->getDoc(); + if (!doc) { + break; + } - for (; iter != _cloneLocs.end(); ++iter) { // We must always make progress in this method by at least one document because empty // return indicates there is no more initial clone data. if (arrBuilder->arrSize() && tracker.intervalHasElapsed()) { + _cloneList.insertOverflowDoc(*doc); break; } - auto nextRecordId = *iter; - - lk.unlock(); - - Snapshotted<BSONObj> doc; - if (collection->findDoc(opCtx, nextRecordId, &doc)) { - // Use the builder size instead of accumulating the document sizes directly so - // that we take into consideration the overhead of BSONArray indices. - if (arrBuilder->arrSize() && - (arrBuilder->len() + doc.value().objsize() + 1024) > BSONObjMaxUserSize) { - - break; + // Do not send documents that are no longer in the chunk range being moved. This can + // happen when document shard key value of the document changed after the initial + // index scan during cloning. This is needed because the destination is very + // conservative in processing xferMod deletes and won't delete docs that are not in + // the range of the chunk being migrated. + if (!isDocInRange( + doc->value(), _args.getMin().value(), _args.getMax().value(), _shardKeyPattern)) { + { + stdx::lock_guard lk(_mutex); + _numRecordsPassedOver++; } + continue; + } - arrBuilder->append(doc.value()); - ShardingStatistics::get(opCtx).countDocsClonedOnDonor.addAndFetch(1); + // Use the builder size instead of accumulating the document sizes directly so + // that we take into consideration the overhead of BSONArray indices. + if (arrBuilder->arrSize() && + (arrBuilder->len() + doc->value().objsize() + 1024) > BSONObjMaxUserSize) { + _cloneList.insertOverflowDoc(*doc); + break; } - lk.lock(); + { + stdx::lock_guard lk(_mutex); + _numRecordsCloned++; + } + arrBuilder->append(doc->value()); + ShardingStatistics::get(opCtx).countDocsClonedOnDonor.addAndFetch(1); + ShardingStatistics::get(opCtx).countBytesClonedOnDonor.addAndFetch(doc->value().objsize()); } - - _cloneLocs.erase(_cloneLocs.begin(), iter); } uint64_t MigrationChunkClonerSourceLegacy::getCloneBatchBufferAllocationSize() { @@ -725,7 +794,7 @@ uint64_t MigrationChunkClonerSourceLegacy::getCloneBatchBufferAllocationSize() { return static_cast<uint64_t>(BSONObjMaxUserSize); return std::min(static_cast<uint64_t>(BSONObjMaxUserSize), - _averageObjectSizeForCloneLocs * _cloneLocs.size()); + _averageObjectSizeForCloneLocs * _cloneList.size()); } Status MigrationChunkClonerSourceLegacy::nextCloneBatch(OperationContext* opCtx, @@ -748,18 +817,91 @@ Status MigrationChunkClonerSourceLegacy::nextCloneBatch(OperationContext* opCtx, return Status::OK(); } +bool MigrationChunkClonerSourceLegacy::_processUpdateForXferMod(const BSONObj& preImageDocKey, + const BSONObj& postImageDocKey) { + auto const& minKey = _args.getMin().value(); + auto const& maxKey = _args.getMax().value(); + + auto postShardKeyValues = _shardKeyPattern.extractShardKeyFromDocumentKey(postImageDocKey); + fassert(6836100, !postShardKeyValues.isEmpty()); + + auto opType = repl::OpTypeEnum::kUpdate; + auto idElement = preImageDocKey["_id"]; + + if (!isShardKeyValueInRange(postShardKeyValues, minKey, maxKey)) { + // If the preImageDoc is not in range but the postImageDoc was, we know that the + // document has changed shard keys and no longer belongs in the chunk being cloned. + // We will model the deletion of the preImage document so that the destination chunk + // does not receive an outdated version of this document. + + auto preImageShardKeyValues = + _shardKeyPattern.extractShardKeyFromDocumentKey(preImageDocKey); + fassert(6836101, !preImageShardKeyValues.isEmpty()); + + if (!isShardKeyValueInRange(preImageShardKeyValues, minKey, maxKey)) { + return false; + } + + opType = repl::OpTypeEnum::kDelete; + idElement = postImageDocKey["_id"]; + } + + _addToTransferModsQueue(idElement.wrap(), getOpCharForCrudOpType(opType), {}); + + return true; +} + +void MigrationChunkClonerSourceLegacy::_deferProcessingForXferMod(const BSONObj& preImageDocKey) { + stdx::lock_guard<Latch> sl(_mutex); + _deferredReloadOrDeletePreImageDocKeys.push_back(preImageDocKey.getOwned()); + _deferredUntransferredOpsCounter++; +} + +void MigrationChunkClonerSourceLegacy::_processDeferredXferMods(OperationContext* opCtx, + Database* db) { + std::vector<BSONObj> deferredReloadOrDeletePreImageDocKeys; + + { + stdx::unique_lock lk(_mutex); + deferredReloadOrDeletePreImageDocKeys.swap(_deferredReloadOrDeletePreImageDocKeys); + } + + for (const auto& preImageDocKey : deferredReloadOrDeletePreImageDocKeys) { + auto idElement = preImageDocKey["_id"]; + BSONObj newerVersionDoc; + if (!Helpers::findById(opCtx, db, nss().ns(), BSON("_id" << idElement), newerVersionDoc)) { + // If the document can no longer be found, this means that another later op must have + // deleted it. That delete would have been captured by the xferMods so nothing else to + // do here. + continue; + } + + auto postImageDocKey = + CollectionMetadata::extractDocumentKey(&_shardKeyPattern, newerVersionDoc); + static_cast<void>(_processUpdateForXferMod(preImageDocKey, postImageDocKey)); + } + + hangAfterProcessingDeferredXferMods.execute([&](const auto& data) { + if (!deferredReloadOrDeletePreImageDocKeys.empty()) { + hangAfterProcessingDeferredXferMods.pauseWhileSet(); + } + }); +} + Status MigrationChunkClonerSourceLegacy::nextModsBatch(OperationContext* opCtx, Database* db, BSONObjBuilder* builder) { dassert(opCtx->lockState()->isCollectionLockedForMode(nss(), MODE_IS)); + _processDeferredXferMods(opCtx, db); + std::list<BSONObj> deleteList; std::list<BSONObj> updateList; { // All clone data must have been drained before starting to fetch the incremental changes. stdx::unique_lock<Latch> lk(_mutex); - invariant(_cloneLocs.empty()); + invariant(!_cloneList.hasMore()); // The "snapshot" for delete and update list must be taken under a single lock. This is to // ensure that we will preserve the causal order of writes. Always consume the delete @@ -771,6 +913,11 @@ Status MigrationChunkClonerSourceLegacy::nextModsBatch(OperationContext* opCtx, updateList.splice(updateList.cbegin(), _reload); } + // It's important to abandon any open snapshots before processing updates so that we are sure + // that our snapshot is at least as new as those updates. It's possible for a stale snapshot to + // still be open from reads performed by _processDeferredXferMods(), above. + opCtx->recoveryUnit()->abandonSnapshot(); + StringData ns = nss().ns().c_str(); BSONArrayBuilder arrDel(builder->subarrayStart("deleted")); auto noopFn = [](BSONObj idDoc, BSONObj* fullDoc) { @@ -797,6 +944,7 @@ Status MigrationChunkClonerSourceLegacy::nextModsBatch(OperationContext* opCtx, _untransferredDeletesCounter = _deleted.size(); _reload.splice(_reload.cbegin(), updateList); _untransferredUpsertsCounter = _reload.size(); + _deferredUntransferredOpsCounter = _deferredReloadOrDeletePreImageDocKeys.size(); return Status::OK(); } @@ -811,6 +959,8 @@ void MigrationChunkClonerSourceLegacy::_cleanup() { _untransferredUpsertsCounter = 0; _deleted.clear(); _untransferredDeletesCounter = 0; + _deferredReloadOrDeletePreImageDocKeys.clear(); + _deferredUntransferredOpsCounter = 0; } StatusWith<BSONObj> MigrationChunkClonerSourceLegacy::_callRecipient(OperationContext* opCtx, @@ -938,6 +1088,8 @@ Status MigrationChunkClonerSourceLegacy::_storeCurrentLocs(OperationContext* opC try { BSONObj obj; RecordId recordId; + RecordIdSet recordIdSet; + while (PlanExecutor::ADVANCED == exec->getNext(&obj, &recordId)) { Status interruptStatus = opCtx->checkForInterruptNoAssert(); if (!interruptStatus.isOK()) { @@ -945,19 +1097,20 @@ Status MigrationChunkClonerSourceLegacy::_storeCurrentLocs(OperationContext* opC } if (!isLargeChunk) { - stdx::lock_guard<Latch> lk(_mutex); - _cloneLocs.insert(recordId); + recordIdSet.insert(recordId); } if (++recCount > maxRecsWhenFull) { isLargeChunk = true; if (_forceJumbo) { - _cloneLocs.clear(); + recordIdSet.clear(); break; } } } + + _cloneList.populateList(std::move(recordIdSet)); } catch (DBException& exception) { exception.addContext("Executor error while scanning for documents belonging to chunk"); throw; @@ -1055,9 +1208,9 @@ Status MigrationChunkClonerSourceLegacy::_checkRecipientCloningStatus(OperationC stdx::lock_guard<Latch> sl(_mutex); - const std::size_t cloneLocsRemaining = _cloneLocs.size(); int64_t untransferredModsSizeBytes = _untransferredDeletesCounter * _averageObjectIdSize + - _untransferredUpsertsCounter * _averageObjectSizeForCloneLocs; + (_untransferredUpsertsCounter + _deferredUntransferredOpsCounter) * + _averageObjectSizeForCloneLocs; if (_forceJumbo && _jumboChunkCloneState) { LOGV2(21992, @@ -1077,13 +1230,14 @@ Status MigrationChunkClonerSourceLegacy::_checkRecipientCloningStatus(OperationC "moveChunk data transfer progress", "response"_attr = redact(res), "memoryUsedBytes"_attr = _memoryUsed, - "docsRemainingToClone"_attr = cloneLocsRemaining, + "docsRemainingToClone"_attr = + _cloneList.size() - _numRecordsCloned - _numRecordsPassedOver, "untransferredModsSizeBytes"_attr = untransferredModsSizeBytes); } if (res["state"].String() == "steady" && sessionCatalogSourceInCatchupPhase && estimateUntransferredSessionsSize == 0) { - if (cloneLocsRemaining != 0 || + if (_cloneList.hasMore() || (_jumboChunkCloneState && _forceJumbo && PlanExecutor::IS_EOF != _jumboChunkCloneState->clonerState)) { return {ErrorCodes::OperationIncomplete, @@ -1122,6 +1276,7 @@ Status MigrationChunkClonerSourceLegacy::_checkRecipientCloningStatus(OperationC "moveChunk data transfer within threshold to allow write blocking", "_untransferredUpsertsCounter"_attr = _untransferredUpsertsCounter, "_untransferredDeletesCounter"_attr = _untransferredDeletesCounter, + "_deferredUntransferredOpsCounter"_attr = _deferredUntransferredOpsCounter, "_averageObjectSizeForCloneLocs"_attr = _averageObjectSizeForCloneLocs, "_averageObjectIdSize"_attr = _averageObjectIdSize, "untransferredModsSizeBytes"_attr = untransferredModsSizeBytes, @@ -1227,4 +1382,135 @@ MigrationChunkClonerSourceLegacy::getNotificationForNextSessionMigrationBatch() return _sessionCatalogSource->getNotificationForNewOplog(); } +MigrationChunkClonerSourceLegacy::CloneList::DocumentInFlightWithLock::DocumentInFlightWithLock( + WithLock lock, MigrationChunkClonerSourceLegacy::CloneList& clonerList) + : _inProgressReadToken( + std::make_unique<MigrationChunkClonerSourceLegacy::CloneList::InProgressReadToken>( + lock, clonerList)) {} + +void MigrationChunkClonerSourceLegacy::CloneList::DocumentInFlightWithLock::setDoc( + boost::optional<Snapshotted<BSONObj>> doc) { + _doc = std::move(doc); +} + +std::unique_ptr<MigrationChunkClonerSourceLegacy::CloneList::DocumentInFlightWhileNotInLock> +MigrationChunkClonerSourceLegacy::CloneList::DocumentInFlightWithLock::release() { + invariant(_inProgressReadToken); + + return std::make_unique< + MigrationChunkClonerSourceLegacy::CloneList::DocumentInFlightWhileNotInLock>( + std::move(_inProgressReadToken), std::move(_doc)); +} + +MigrationChunkClonerSourceLegacy::CloneList::DocumentInFlightWhileNotInLock:: + DocumentInFlightWhileNotInLock( + std::unique_ptr<CloneList::InProgressReadToken> inProgressReadToken, + boost::optional<Snapshotted<BSONObj>> doc) + : _inProgressReadToken(std::move(inProgressReadToken)), _doc(std::move(doc)) {} + +void MigrationChunkClonerSourceLegacy::CloneList::DocumentInFlightWhileNotInLock::setDoc( + boost::optional<Snapshotted<BSONObj>> doc) { + _doc = std::move(doc); +} + +const boost::optional<Snapshotted<BSONObj>>& +MigrationChunkClonerSourceLegacy::CloneList::DocumentInFlightWhileNotInLock::getDoc() { + return _doc; +} + +MigrationChunkClonerSourceLegacy::CloneList::InProgressReadToken::InProgressReadToken( + WithLock withLock, CloneList& cloneList) + : _cloneList(cloneList) { + _cloneList._startedOneInProgressRead(withLock); +} + +MigrationChunkClonerSourceLegacy::CloneList::InProgressReadToken::~InProgressReadToken() { + _cloneList._finishedOneInProgressRead(); +} + +MigrationChunkClonerSourceLegacy::CloneList::CloneList() { + _recordIdsIter = _recordIds.begin(); +} + +void MigrationChunkClonerSourceLegacy::CloneList::populateList(RecordIdSet recordIds) { + stdx::lock_guard lk(_mutex); + _recordIds = std::move(recordIds); + _recordIdsIter = _recordIds.begin(); +} + +void MigrationChunkClonerSourceLegacy::CloneList::insertOverflowDoc(Snapshotted<BSONObj> doc) { + stdx::lock_guard lk(_mutex); + invariant(_inProgressReads >= 1); + _overflowDocs.push_back(std::move(doc)); +} + +bool MigrationChunkClonerSourceLegacy::CloneList::hasMore() const { + stdx::lock_guard lk(_mutex); + return _recordIdsIter != _recordIds.cend() && _inProgressReads > 0; +} + +std::unique_ptr<MigrationChunkClonerSourceLegacy::CloneList::DocumentInFlightWhileNotInLock> +MigrationChunkClonerSourceLegacy::CloneList::getNextDoc(OperationContext* opCtx, + const CollectionPtr& collection, + int* numRecordsNoLongerExist) { + while (true) { + stdx::unique_lock lk(_mutex); + invariant(_inProgressReads >= 0); + RecordId nextRecordId; + + opCtx->waitForConditionOrInterrupt(_moreDocsCV, lk, [&]() { + return _recordIdsIter != _recordIds.end() || !_overflowDocs.empty() || + _inProgressReads == 0; + }); + + DocumentInFlightWithLock docInFlight(lk, *this); + + // One of the following must now be true (corresponding to the three if conditions): + // 1. There is a document in the overflow set + // 2. The iterator has not reached the end of the record id set + // 3. The overflow set is empty, the iterator is at the end, and + // no threads are holding a document. This condition indicates + // that there are no more docs to return for the cloning phase. + if (!_overflowDocs.empty()) { + docInFlight.setDoc(std::move(_overflowDocs.front())); + _overflowDocs.pop_front(); + return docInFlight.release(); + } else if (_recordIdsIter != _recordIds.end()) { + nextRecordId = *_recordIdsIter; + ++_recordIdsIter; + } else { + return docInFlight.release(); + } + + lk.unlock(); + + auto docInFlightWhileNotLocked = docInFlight.release(); + + Snapshotted<BSONObj> doc; + if (collection->findDoc(opCtx, nextRecordId, &doc)) { + docInFlightWhileNotLocked->setDoc(std::move(doc)); + return docInFlightWhileNotLocked; + } + + if (numRecordsNoLongerExist) { + (*numRecordsNoLongerExist)++; + } + } +} + +size_t MigrationChunkClonerSourceLegacy::CloneList::size() const { + stdx::unique_lock lk(_mutex); + return _recordIds.size(); +} + +void MigrationChunkClonerSourceLegacy::CloneList::_startedOneInProgressRead(WithLock) { + _inProgressReads++; +} + +void MigrationChunkClonerSourceLegacy::CloneList::_finishedOneInProgressRead() { + stdx::lock_guard lk(_mutex); + _inProgressReads--; + _moreDocsCV.notify_one(); +} + } // namespace mongo |