summaryrefslogtreecommitdiff
path: root/src
diff options
context:
space:
mode:
Diffstat (limited to 'src')
-rw-r--r--src/mongo/db/s/resharding/resharding_collection_cloner.cpp87
-rw-r--r--src/mongo/db/s/resharding/resharding_collection_cloner.h11
2 files changed, 58 insertions, 40 deletions
diff --git a/src/mongo/db/s/resharding/resharding_collection_cloner.cpp b/src/mongo/db/s/resharding/resharding_collection_cloner.cpp
index b3e89a0a59f..ff482eb55ef 100644
--- a/src/mongo/db/s/resharding/resharding_collection_cloner.cpp
+++ b/src/mongo/db/s/resharding/resharding_collection_cloner.cpp
@@ -104,6 +104,8 @@
MONGO_FAIL_POINT_DEFINE(reshardingCollectionClonerAbort);
MONGO_FAIL_POINT_DEFINE(reshardingCollectionClonerPauseBeforeAttempt);
+MONGO_FAIL_POINT_DEFINE(reshardingCollectionClonerShouldFailWithStaleConfig);
+MONGO_FAIL_POINT_DEFINE(reshardingCollectionClonerPauseBeforeWriteNaturalOrder);
namespace mongo {
namespace {
@@ -304,21 +306,21 @@ public:
ReshardingCloneFetcher(std::shared_ptr<executor::TaskExecutor> executor,
std::shared_ptr<executor::TaskExecutor> cleanupExecutor,
CancellationToken cancelToken,
- sharded_agg_helpers::DispatchShardPipelineResults dispatchResults,
+ std::vector<OwnedRemoteCursor> remoteCursors,
int batchSizeLimitBytes,
int numWriteThreads)
: _executor(std::move(executor)),
_cleanupExecutor(std::move(cleanupExecutor)),
_cancelSource(cancelToken),
_factory(_cancelSource.token(), _executor),
- _dispatchResults(std::move(dispatchResults)),
+ _remoteCursors(std::move(remoteCursors)),
_numWriteThreads(numWriteThreads),
_queues(_numWriteThreads),
_activeCursors(0),
_openConsumers(0) {
constexpr int kQueueDepthPerDonor = 2;
MultiProducerSingleConsumerQueue<QueueData>::Options qOptions;
- qOptions.maxQueueDepth = _dispatchResults.remoteCursors.size() * kQueueDepthPerDonor;
+ qOptions.maxQueueDepth = _remoteCursors.size() * kQueueDepthPerDonor;
for (auto& queue : _queues) {
queue.emplace(qOptions);
}
@@ -431,16 +433,15 @@ public:
}
void setupReaderThreads(OperationContext* opCtx) {
- auto& remoteCursors = _dispatchResults.remoteCursors;
// Network commands can start immediately, so reserve here to avoid the
// vector being resized while setting up.
- _shardIds.reserve(remoteCursors.size());
- for (int i = 0; i < int(remoteCursors.size()); i++) {
+ _shardIds.reserve(_remoteCursors.size());
+ for (int i = 0; i < int(_remoteCursors.size()); i++) {
{
std::lock_guard lk(_mutex);
_activeCursors++;
}
- auto& cursor = _dispatchResults.remoteCursors[i];
+ auto& cursor = _remoteCursors[i];
GetMoreCommandRequest getMoreRequest(
cursor->getCursorResponse().getCursorId(),
cursor->getCursorResponse().getNSS().coll().toString());
@@ -540,7 +541,7 @@ private:
std::shared_ptr<executor::TaskExecutor> _cleanupExecutor;
CancellationSource _cancelSource;
CancelableOperationContextFactory _factory;
- sharded_agg_helpers::DispatchShardPipelineResults _dispatchResults;
+ std::vector<OwnedRemoteCursor> _remoteCursors;
int _numWriteThreads;
std::vector<ExecutorFuture<executor::TaskExecutor::ResponseStatus>> _cmdFutures;
std::vector<ExecutorFuture<void>> _writerFutures;
@@ -658,20 +659,6 @@ ReshardingCollectionCloner::_queryOnceWithNaturalOrder(
std::move(resumeTokenMap),
std::move(shardsToSkip));
- return dispatchResults;
-}
-
-void ReshardingCollectionCloner::_writeOnceWithNaturalOrder(
- OperationContext* opCtx,
- std::shared_ptr<executor::TaskExecutor> executor,
- std::shared_ptr<executor::TaskExecutor> cleanupExecutor,
- CancellationToken cancelToken,
- sharded_agg_helpers::DispatchShardPipelineResults& dispatchResults) {
- // If we don't establish any cursors, there is no work to do. Return.
- if (dispatchResults.remoteCursors.empty()) {
- return;
- }
-
bool hasSplitPipeline = !!dispatchResults.splitPipeline;
std::string shardsPipelineStr;
std::string mergePipelineStr;
@@ -700,13 +687,34 @@ void ReshardingCollectionCloner::_writeOnceWithNaturalOrder(
"numProducers"_attr = dispatchResults.numProducers,
"hasExchangeSpec"_attr = dispatchResults.exchangeSpec != boost::none);
+ return dispatchResults;
+}
+
+void ReshardingCollectionCloner::_writeOnceWithNaturalOrder(
+ OperationContext* opCtx,
+ std::shared_ptr<executor::TaskExecutor> executor,
+ std::shared_ptr<executor::TaskExecutor> cleanupExecutor,
+ CancellationToken cancelToken,
+ std::vector<OwnedRemoteCursor> remoteCursors) {
ReshardingCloneFetcher reshardingCloneFetcher(
std::move(executor),
std::move(cleanupExecutor),
cancelToken,
- std::move(dispatchResults),
+ std::move(remoteCursors),
resharding::gReshardingCollectionClonerBatchSizeInBytes.load(),
resharding::gReshardingCollectionClonerWriteThreadCount);
+
+ if (reshardingCollectionClonerShouldFailWithStaleConfig.shouldFail()) {
+ uassert(StaleConfigInfo(_sourceNss,
+ ShardVersionFactory::make(ChunkVersion::IGNORED(),
+ boost::optional<CollectionIndexes>(
+ boost::none)) /* receivedVersion */,
+ boost::none /* wantedVersion */,
+ ShardId{"0"}),
+ str::stream() << "Throwing staleConfig for reshardingCollectionCloner failpoint.",
+ false);
+ }
+
reshardingCloneFetcher
.run(opCtx,
[this](OperationContext* opCtx,
@@ -742,15 +750,29 @@ void ReshardingCollectionCloner::_runOnceWithNaturalOrder(
std::shared_ptr<executor::TaskExecutor> executor,
std::shared_ptr<executor::TaskExecutor> cleanupExecutor,
CancellationToken cancelToken) {
- auto dispatchResults = shardVersionRetry(
- opCtx,
- Grid::get(opCtx)->catalogCache(),
- _sourceNss,
- "resharding collection cloner fetching with natural order (query stage)"_sd,
- [&] { return _queryOnceWithNaturalOrder(opCtx, mongoProcessInterface); });
-
+ // We can run into StaleConfig errors when cloning collections. To make it
+ // safe during retry, we retry the whole cloning process and rely on the
+ // resume token to be correct. Note that the remote cursors need to be reestablished during
+ // retry since _writeOnceWithNaturalOrder can partially or completely consume them.
resharding::data_copy::withOneStaleConfigRetry(opCtx, [&] {
- _writeOnceWithNaturalOrder(opCtx, executor, cleanupExecutor, cancelToken, dispatchResults);
+ auto dispatchResults = shardVersionRetry(
+ opCtx,
+ Grid::get(opCtx)->catalogCache(),
+ _sourceNss,
+ "resharding collection cloner fetching with natural order (query stage)"_sd,
+ [&] { return _queryOnceWithNaturalOrder(opCtx, mongoProcessInterface); });
+
+ // If we don't establish any cursors, there is no work to do. Return.
+ if (dispatchResults.remoteCursors.empty()) {
+ return;
+ }
+
+ reshardingCollectionClonerPauseBeforeWriteNaturalOrder.pauseWhileSet();
+ _writeOnceWithNaturalOrder(opCtx,
+ executor,
+ cleanupExecutor,
+ cancelToken,
+ std::move(dispatchResults.remoteCursors));
});
}
@@ -890,9 +912,6 @@ SemiFuture<void> ReshardingCollectionCloner::run(
reshardingCollectionClonerPauseBeforeAttempt.pauseWhileSet();
if (reshardingImprovementsEnabled) {
auto opCtx = factory.makeOperationContext(&cc());
- // We can run into StaleConfig errors when cloning collections. To make it
- // safer during retry, we retry the whole cloning process and rely on the
- // resume token to be correct.
_runOnceWithNaturalOrder(opCtx.get(),
MongoProcessInterface::create(opCtx.get()),
executor,
diff --git a/src/mongo/db/s/resharding/resharding_collection_cloner.h b/src/mongo/db/s/resharding/resharding_collection_cloner.h
index 117a52c52ec..55bc256bf4d 100644
--- a/src/mongo/db/s/resharding/resharding_collection_cloner.h
+++ b/src/mongo/db/s/resharding/resharding_collection_cloner.h
@@ -133,12 +133,11 @@ private:
sharded_agg_helpers::DispatchShardPipelineResults _queryOnceWithNaturalOrder(
OperationContext* opCtx, std::shared_ptr<MongoProcessInterface> mongoProcessInterface);
- void _writeOnceWithNaturalOrder(
- OperationContext* opCtx,
- std::shared_ptr<executor::TaskExecutor> executor,
- std::shared_ptr<executor::TaskExecutor> cleanupExecutor,
- CancellationToken cancelToken,
- sharded_agg_helpers::DispatchShardPipelineResults& dispatchResults);
+ void _writeOnceWithNaturalOrder(OperationContext* opCtx,
+ std::shared_ptr<executor::TaskExecutor> executor,
+ std::shared_ptr<executor::TaskExecutor> cleanupExecutor,
+ CancellationToken cancelToken,
+ std::vector<OwnedRemoteCursor> remoteCursors);
void _runOnceWithNaturalOrder(OperationContext* opCtx,
std::shared_ptr<MongoProcessInterface> mongoProcessInterface,
generated by cgit v1.2.3 (git 2.39.1) at 2026-09-11 11:57:31 +0000