diff options
Diffstat (limited to 'src/mongo/s')
-rw-r--r-- | src/mongo/s/query/blocking_results_merger.cpp | 49 | ||||
-rw-r--r-- | src/mongo/s/query/blocking_results_merger.h | 15 | ||||
-rw-r--r-- | src/mongo/s/query/blocking_results_merger_test.cpp | 156 | ||||
-rw-r--r-- | src/mongo/s/query/cluster_aggregate.cpp | 34 | ||||
-rw-r--r-- | src/mongo/s/query/document_source_merge_cursors.cpp | 6 | ||||
-rw-r--r-- | src/mongo/s/query/document_source_merge_cursors.h | 4 | ||||
-rw-r--r-- | src/mongo/s/query/router_stage_merge.h | 2 |
7 files changed, 237 insertions, 29 deletions
diff --git a/src/mongo/s/query/blocking_results_merger.cpp b/src/mongo/s/query/blocking_results_merger.cpp index b572a51705e..2431901340d 100644 --- a/src/mongo/s/query/blocking_results_merger.cpp +++ b/src/mongo/s/query/blocking_results_merger.cpp @@ -31,16 +31,51 @@ #include "mongo/platform/basic.h" #include "mongo/db/query/find_common.h" +#include "mongo/db/session_catalog_mongod.h" +#include "mongo/db/transaction_participant.h" #include "mongo/s/query/blocking_results_merger.h" namespace mongo { BlockingResultsMerger::BlockingResultsMerger(OperationContext* opCtx, AsyncResultsMergerParams&& armParams, - executor::TaskExecutor* executor) + executor::TaskExecutor* executor, + std::unique_ptr<ResourceYielder> resourceYielder) : _tailableMode(armParams.getTailableMode().value_or(TailableModeEnum::kNormal)), _executor(executor), - _arm(opCtx, executor, std::move(armParams)) {} + _arm(opCtx, executor, std::move(armParams)), + _resourceYielder(std::move(resourceYielder)) {} + +StatusWith<stdx::cv_status> BlockingResultsMerger::doWaiting( + OperationContext* opCtx, const std::function<StatusWith<stdx::cv_status>()>& waitFn) noexcept { + + if (_resourceYielder) { + try { + // The BRM interface returns Statuses. Be sure we respect that here. + _resourceYielder->yield(opCtx); + } catch (const DBException& e) { + return e.toStatus(); + } + } + + boost::optional<StatusWith<stdx::cv_status>> result; + try { + // This shouldn't throw, but we cannot enforce that. + result = waitFn(); + } catch (const DBException&) { + MONGO_UNREACHABLE; + } + + if (_resourceYielder) { + try { + _resourceYielder->unyield(opCtx); + } catch (const DBException& e) { + return e.toStatus(); + } + } + + return *result; +} StatusWith<ClusterQueryResult> BlockingResultsMerger::awaitNextWithTimeout( OperationContext* opCtx, RouterExecStage::ExecContext execCtx) { @@ -54,9 +89,10 @@ StatusWith<ClusterQueryResult> BlockingResultsMerger::awaitNextWithTimeout( } auto event = nextEventStatus.getValue(); - // Block until there are further results to return, or our time limit is exceeded. - auto waitStatus = - _executor->waitForEvent(opCtx, event, awaitDataState(opCtx).waitForInsertsDeadline); + const auto waitStatus = doWaiting(opCtx, [this, opCtx, &event]() { + return _executor->waitForEvent( + opCtx, event, awaitDataState(opCtx).waitForInsertsDeadline); + }); if (!waitStatus.isOK()) { return waitStatus.getStatus(); @@ -84,7 +120,8 @@ StatusWith<ClusterQueryResult> BlockingResultsMerger::blockUntilNext(OperationCo auto event = nextEventStatus.getValue(); // Block until there are further results to return. - auto status = _executor->waitForEvent(opCtx, event); + auto status = doWaiting( + opCtx, [this, opCtx, &event]() { return _executor->waitForEvent(opCtx, event); }); if (!status.isOK()) { return status.getStatus(); diff --git a/src/mongo/s/query/blocking_results_merger.h b/src/mongo/s/query/blocking_results_merger.h index f35ce904709..7d82cdff49c 100644 --- a/src/mongo/s/query/blocking_results_merger.h +++ b/src/mongo/s/query/blocking_results_merger.h @@ -43,7 +43,8 @@ class BlockingResultsMerger { public: BlockingResultsMerger(OperationContext* opCtx, AsyncResultsMergerParams&& arm, - executor::TaskExecutor*); + executor::TaskExecutor*, + std::unique_ptr<ResourceYielder> resourceYielder); /** * Blocks until the next result is available or an error is detected. @@ -100,6 +101,14 @@ private: */ StatusWith<executor::TaskExecutor::EventHandle> getNextEvent(); + /** + * Call the waitFn and return the result, yielding resources while waiting if necessary. + * 'waitFn' may not throw. + */ + StatusWith<stdx::cv_status> doWaiting( + OperationContext* opCtx, + const std::function<StatusWith<stdx::cv_status>()>& waitFn) noexcept; + TailableModeEnum _tailableMode; executor::TaskExecutor* _executor; @@ -110,6 +119,10 @@ private: // and pick back up waiting for it on the next call to 'next()'. executor::TaskExecutor::EventHandle _leftoverEventFromLastTimeout; AsyncResultsMerger _arm; + + // Provides interface for yielding and "unyielding" resources while waiting for results from + // the network. A value of nullptr implies that no such yielding or unyielding is necessary. + std::unique_ptr<ResourceYielder> _resourceYielder; }; } // namespace mongo diff --git a/src/mongo/s/query/blocking_results_merger_test.cpp b/src/mongo/s/query/blocking_results_merger_test.cpp index 961b3b5912c..2f76d576e3d 100644 --- a/src/mongo/s/query/blocking_results_merger_test.cpp +++ b/src/mongo/s/query/blocking_results_merger_test.cpp @@ -32,6 +32,7 @@ #include "mongo/s/query/blocking_results_merger.h" #include "mongo/s/query/results_merger_test_fixture.h" +#include "mongo/unittest/unittest.h" namespace mongo { @@ -43,9 +44,33 @@ TEST_F(ResultsMergerTestFixture, ShouldBeAbleToBlockUntilKilled) { std::vector<RemoteCursor> cursors; cursors.emplace_back( makeRemoteCursor(kTestShardIds[0], kTestShardHosts[0], CursorResponse(kTestNss, 1, {}))); + BlockingResultsMerger blockingMerger(operationContext(), + makeARMParamsFromExistingCursors(std::move(cursors)), + executor(), + nullptr); + + blockingMerger.kill(operationContext()); +} + +TEST_F(ResultsMergerTestFixture, ShouldBeAbleToBlockUntilDeadlineExpires) { + std::vector<RemoteCursor> cursors; + cursors.emplace_back( + makeRemoteCursor(kTestShardIds[0], kTestShardHosts[0], CursorResponse(kTestNss, 1, {}))); + auto params = makeARMParamsFromExistingCursors(std::move(cursors)); + params.setTailableMode(TailableModeEnum::kTailableAndAwaitData); BlockingResultsMerger blockingMerger( - operationContext(), makeARMParamsFromExistingCursors(std::move(cursors)), executor()); + operationContext(), std::move(params), executor(), nullptr); + // Issue a blocking wait for the next result asynchronously on a different thread. + auto future = launchAsync([&]() { + auto next = unittest::assertGet( + blockingMerger.next(operationContext(), RouterExecStage::ExecContext::kInitialFind)); + + // The timeout should hit, and return an empty object. + ASSERT_TRUE(next.isEOF()); + }); + + future.timed_get(kFutureTimeout); blockingMerger.kill(operationContext()); } @@ -53,8 +78,10 @@ TEST_F(ResultsMergerTestFixture, ShouldBeAbleToBlockUntilNextResultIsReady) { std::vector<RemoteCursor> cursors; cursors.emplace_back( makeRemoteCursor(kTestShardIds[0], kTestShardHosts[0], CursorResponse(kTestNss, 1, {}))); - BlockingResultsMerger blockingMerger( - operationContext(), makeARMParamsFromExistingCursors(std::move(cursors)), executor()); + BlockingResultsMerger blockingMerger(operationContext(), + makeARMParamsFromExistingCursors(std::move(cursors)), + executor(), + nullptr); // Issue a blocking wait for the next result asynchronously on a different thread. auto future = launchAsync([&]() { @@ -78,12 +105,57 @@ TEST_F(ResultsMergerTestFixture, ShouldBeAbleToBlockUntilNextResultIsReady) { future.timed_get(kFutureTimeout); } +TEST_F(ResultsMergerTestFixture, ShouldBeAbleToBlockUntilNextResultIsReadyWithDeadline) { + std::vector<RemoteCursor> cursors; + cursors.emplace_back( + makeRemoteCursor(kTestShardIds[0], kTestShardHosts[0], CursorResponse(kTestNss, 1, {}))); + auto params = makeARMParamsFromExistingCursors(std::move(cursors)); + params.setTailableMode(TailableModeEnum::kTailableAndAwaitData); + BlockingResultsMerger blockingMerger( + operationContext(), std::move(params), executor(), nullptr); + + // Used for synchronizing the background thread with this thread. + stdx::mutex mutex; + stdx::unique_lock<stdx::mutex> lk(mutex); + + // Issue a blocking wait for the next result asynchronously on a different thread. + auto future = launchAsync([&]() { + // Will schedule a getMore. No one will send a response, so will return EOF. + auto next = unittest::assertGet(blockingMerger.next( + operationContext(), RouterExecStage::ExecContext::kGetMoreNoResultsYet)); + ASSERT_TRUE(next.isEOF()); + + // Block until the main thread has responded to the getMore. + stdx::unique_lock<stdx::mutex> lk(mutex); + + next = unittest::assertGet(blockingMerger.next( + operationContext(), RouterExecStage::ExecContext::kGetMoreNoResultsYet)); + ASSERT_FALSE(next.isEOF()); + ASSERT_BSONOBJ_EQ(*next.getResult(), BSON("x" << 1)); + + }); + + // Schedule the response to the getMore which will return the next result and mark the cursor as + // exhausted. + onCommand([&](const auto& request) { + ASSERT(request.cmdObj["getMore"]); + return CursorResponse(kTestNss, 0LL, {BSON("x" << 1)}) + .toBSON(CursorResponse::ResponseType::SubsequentResponse); + }); + + // Unblock the other thread, allowing it to call next() on the BlockingResultsMerger. + lk.unlock(); + + future.timed_get(kFutureTimeout); +} + TEST_F(ResultsMergerTestFixture, ShouldBeInterruptableDuringBlockingNext) { std::vector<RemoteCursor> cursors; cursors.emplace_back( makeRemoteCursor(kTestShardIds[0], kTestShardHosts[0], CursorResponse(kTestNss, 1, {}))); auto params = makeARMParamsFromExistingCursors(std::move(cursors)); - BlockingResultsMerger blockingMerger(operationContext(), std::move(params), executor()); + BlockingResultsMerger blockingMerger( + operationContext(), std::move(params), executor(), nullptr); // Issue a blocking wait for the next result asynchronously on a different thread. auto future = launchAsync([&]() { @@ -116,5 +188,81 @@ TEST_F(ResultsMergerTestFixture, ShouldBeInterruptableDuringBlockingNext) { future.timed_get(kFutureTimeout); } +TEST_F(ResultsMergerTestFixture, ShouldBeAbleToHandleExceptionWhenYielding) { + class ThrowyResourceYielder : public ResourceYielder { + public: + void yield(OperationContext*) { + uasserted(ErrorCodes::BadValue, "Simulated error"); + } + + void unyield(OperationContext*) {} + }; + + std::vector<RemoteCursor> cursors; + cursors.emplace_back( + makeRemoteCursor(kTestShardIds[0], kTestShardHosts[0], CursorResponse(kTestNss, 1, {}))); + BlockingResultsMerger blockingMerger(operationContext(), + makeARMParamsFromExistingCursors(std::move(cursors)), + executor(), + std::make_unique<ThrowyResourceYielder>()); + + // Issue a blocking wait for the next result asynchronously on a different thread. + auto future = launchAsync([&]() { + // Make sure that the next() call throws correctly. + const auto status = + blockingMerger.next(operationContext(), RouterExecStage::ExecContext::kInitialFind) + .getStatus(); + ASSERT_EQ(status, ErrorCodes::BadValue); + }); + + // Schedule the response to the getMore which will return the next result and mark the cursor as + // exhausted. + onCommand([&](const auto& request) { + ASSERT(request.cmdObj["getMore"]); + return CursorResponse(kTestNss, 0LL, {BSON("x" << 1)}) + .toBSON(CursorResponse::ResponseType::SubsequentResponse); + }); + + future.timed_get(kFutureTimeout); +} + +TEST_F(ResultsMergerTestFixture, ShouldBeAbleToHandleExceptionWhenUnyielding) { + class ThrowyResourceYielder : public ResourceYielder { + public: + void yield(OperationContext*) {} + + void unyield(OperationContext*) { + uasserted(ErrorCodes::BadValue, "Simulated error"); + } + }; + + std::vector<RemoteCursor> cursors; + cursors.emplace_back( + makeRemoteCursor(kTestShardIds[0], kTestShardHosts[0], CursorResponse(kTestNss, 1, {}))); + BlockingResultsMerger blockingMerger(operationContext(), + makeARMParamsFromExistingCursors(std::move(cursors)), + executor(), + std::make_unique<ThrowyResourceYielder>()); + + // Issue a blocking wait for the next result asynchronously on a different thread. + auto future = launchAsync([&]() { + // Make sure that the next() call throws correctly. + const auto status = + blockingMerger.next(operationContext(), RouterExecStage::ExecContext::kInitialFind) + .getStatus(); + ASSERT_EQ(status, ErrorCodes::BadValue); + }); + + // Schedule the response to the getMore which will return the next result and mark the cursor as + // exhausted. + onCommand([&](const auto& request) { + ASSERT(request.cmdObj["getMore"]); + return CursorResponse(kTestNss, 0LL, {BSON("x" << 1)}) + .toBSON(CursorResponse::ResponseType::SubsequentResponse); + }); + + future.timed_get(kFutureTimeout); +} + } // namespace } // namespace mongo diff --git a/src/mongo/s/query/cluster_aggregate.cpp b/src/mongo/s/query/cluster_aggregate.cpp index 12270fc1c99..3ead06bc1f1 100644 --- a/src/mongo/s/query/cluster_aggregate.cpp +++ b/src/mongo/s/query/cluster_aggregate.cpp @@ -105,6 +105,7 @@ Status appendCursorResponseToCommandResult(const ShardId& shardId, BSONObj createCommandForMergingShard(const AggregationRequest& request, const boost::intrusive_ptr<ExpressionContext>& mergeCtx, const ShardId& shardId, + bool mergingShardContributesData, const Pipeline* pipelineForMerging) { MutableDocument mergeCmd(request.serializeToCommandObj()); @@ -119,9 +120,19 @@ BSONObj createCommandForMergingShard(const AggregationRequest& request, : Value(Document{CollationSpec::kSimpleSpec}); } + const auto txnRouter = TransactionRouter::get(mergeCtx->opCtx); + if (txnRouter && mergingShardContributesData) { + // Don't include a readConcern since we can only include read concerns on the _first_ + // command sent to a participant per transaction. Assuming the merging shard is a + // participant, it will already have received another 'aggregate' command earlier which + // contained a readConcern. + + mergeCmd.remove("readConcern"); + } + auto aggCmd = mergeCmd.freeze().toBson(); - if (auto txnRouter = TransactionRouter::get(mergeCtx->opCtx)) { + if (txnRouter) { aggCmd = txnRouter->attachTxnFieldsIfNeeded(shardId, aggCmd); } @@ -609,19 +620,16 @@ Status dispatchMergingPipeline(const boost::intrusive_ptr<ExpressionContext>& ex // therefore must have a valid routing table. invariant(routingInfo); - // TODO SERVER-33683 allowing an aggregation within a transaction can lead to a deadlock in the - // SessionCatalog when a pipeline with a $mergeCursors sends a getMore to itself. - uassert(ErrorCodes::OperationNotSupportedInTransaction, - "Cannot specify a transaction number in combination with an aggregation on mongos when " - "merging on a shard", - !opCtx->getTxnNumber()); - - ShardId mergingShardId = pickMergingShard(opCtx, - shardDispatchResults.needsPrimaryShardMerge, - targetedShards, - routingInfo->db().primaryId()); + const ShardId mergingShardId = pickMergingShard(opCtx, + shardDispatchResults.needsPrimaryShardMerge, + targetedShards, + routingInfo->db().primaryId()); + const bool mergingShardContributesData = + std::find(targetedShards.begin(), targetedShards.end(), mergingShardId) != + targetedShards.end(); - auto mergeCmdObj = createCommandForMergingShard(request, expCtx, mergingShardId, mergePipeline); + auto mergeCmdObj = createCommandForMergingShard( + request, expCtx, mergingShardId, mergingShardContributesData, mergePipeline); // Dispatch $mergeCursors to the chosen shard, store the resulting cursor, and return. auto mergeResponse = establishMergingShardCursor( diff --git a/src/mongo/s/query/document_source_merge_cursors.cpp b/src/mongo/s/query/document_source_merge_cursors.cpp index fe1303684da..fbe93ea1b53 100644 --- a/src/mongo/s/query/document_source_merge_cursors.cpp +++ b/src/mongo/s/query/document_source_merge_cursors.cpp @@ -74,7 +74,11 @@ bool DocumentSourceMergeCursors::remotesExhausted() const { void DocumentSourceMergeCursors::populateMerger() { invariant(!_blockingResultsMerger); invariant(_armParams); - _blockingResultsMerger.emplace(pExpCtx->opCtx, std::move(*_armParams), _executor); + + _blockingResultsMerger.emplace(pExpCtx->opCtx, + std::move(*_armParams), + _executor, + pExpCtx->mongoProcessInterface->getResourceYielder()); _armParams = boost::none; // '_blockingResultsMerger' now owns the cursors. _ownCursors = false; diff --git a/src/mongo/s/query/document_source_merge_cursors.h b/src/mongo/s/query/document_source_merge_cursors.h index a3d65bcab27..5249da51a8e 100644 --- a/src/mongo/s/query/document_source_merge_cursors.h +++ b/src/mongo/s/query/document_source_merge_cursors.h @@ -89,9 +89,7 @@ public: HostTypeRequirement::kNone, DiskUseRequirement::kNoDiskUse, FacetRequirement::kNotAllowed, - // TODO SERVER-33683: Permit $mergeCursors with readConcern - // level "snapshot". - TransactionRequirement::kNotAllowed); + TransactionRequirement::kAllowed); constraints.requiresInputDocSource = false; return constraints; diff --git a/src/mongo/s/query/router_stage_merge.h b/src/mongo/s/query/router_stage_merge.h index 7833c6550c0..6c8e389eded 100644 --- a/src/mongo/s/query/router_stage_merge.h +++ b/src/mongo/s/query/router_stage_merge.h @@ -47,7 +47,7 @@ public: RouterStageMerge(OperationContext* opCtx, executor::TaskExecutor* executor, AsyncResultsMergerParams&& armParams) - : RouterExecStage(opCtx), _resultsMerger(opCtx, std::move(armParams), executor) {} + : RouterExecStage(opCtx), _resultsMerger(opCtx, std::move(armParams), executor, nullptr) {} StatusWith<ClusterQueryResult> next(ExecContext execCtx) final { return _resultsMerger.next(getOpCtx(), execCtx); |