summaryrefslogtreecommitdiff
path: root/src/mongo/s/async_requests_sender.cpp
diff options
context:
space:
mode:
Diffstat (limited to 'src/mongo/s/async_requests_sender.cpp')
-rw-r--r--src/mongo/s/async_requests_sender.cpp113
1 files changed, 49 insertions, 64 deletions
diff --git a/src/mongo/s/async_requests_sender.cpp b/src/mongo/s/async_requests_sender.cpp
index a5100472f69..1fceefc428e 100644
--- a/src/mongo/s/async_requests_sender.cpp
+++ b/src/mongo/s/async_requests_sender.cpp
@@ -37,7 +37,6 @@
#include <memory>
#include "mongo/client/remote_command_targeter.h"
-#include "mongo/db/curop.h"
#include "mongo/executor/remote_command_request.h"
#include "mongo/logv2/log.h"
#include "mongo/rpc/get_status_from_command_result.h"
@@ -88,8 +87,6 @@ AsyncRequestsSender::AsyncRequestsSender(OperationContext* opCtx,
// Kick off requests immediately.
_remotes.emplace_back(this, request.shardId, request.cmdObj).executeRequest();
}
-
- CurOp::get(_opCtx)->ensureRecordRemoteOpWait();
}
AsyncRequestsSender::Response AsyncRequestsSender::next() noexcept {
@@ -120,22 +117,9 @@ AsyncRequestsSender::Response AsyncRequestsSender::next() noexcept {
_resourceYielder->yield(_opCtx);
}
- auto curOp = CurOp::get(_opCtx);
- // Calculating the total wait time for remote operations relies on the CurOp's timing
- // measurement facility and we can't use such facility when the current operation is marked
- // as done. Some commands such as 'analyzeShardKey' command may send remote operations using
- // AsyncRequestsSender even after marking the current operation done and so we need to check
- // whether the current operation is still in progress.
- auto curOpInProgress = !curOp->isDone();
- if (curOpInProgress) {
- curOp->startRemoteOpWaitTimer();
- }
// Only wait for the next result without popping it, so an error unyielding doesn't
// discard an already popped response.
auto waitStatus = _responseQueue.waitForNonEmptyNoThrow(_opCtx);
- if (curOpInProgress) {
- curOp->stopRemoteOpWaitTimer();
- }
auto unyieldStatus =
_resourceYielder ? _resourceYielder->unyieldNoThrow(_opCtx) : Status::OK();
@@ -186,10 +170,9 @@ AsyncRequestsSender::RemoteData::RemoteData(AsyncRequestsSender* ars,
BSONObj cmdObj)
: _ars(ars), _shardId(std::move(shardId)), _cmdObj(std::move(cmdObj)) {}
-SemiFuture<std::shared_ptr<Shard>> AsyncRequestsSender::RemoteData::getShard() noexcept {
- return Grid::get(getGlobalServiceContext())
- ->shardRegistry()
- ->getShard(*_ars->_subBaton, _shardId);
+std::shared_ptr<Shard> AsyncRequestsSender::RemoteData::getShard() {
+ // TODO: Pass down an OperationContext* to use here.
+ return Grid::get(getGlobalServiceContext())->shardRegistry()->getShardNoReload(_shardId);
}
void AsyncRequestsSender::RemoteData::executeRequest() {
@@ -209,12 +192,7 @@ void AsyncRequestsSender::RemoteData::executeRequest() {
auto AsyncRequestsSender::RemoteData::scheduleRequest()
-> SemiFuture<RemoteCommandOnAnyCallbackArgs> {
- return getShard()
- .thenRunOn(*_ars->_subBaton)
- .then([this](auto&& shard) {
- return shard->getTargeter()->findHosts(_ars->_readPreference,
- CancellationToken::uncancelable());
- })
+ return resolveShardIdToHostAndPorts(_ars->_readPreference)
.thenRunOn(*_ars->_subBaton)
.then([this](auto&& hostAndPorts) {
_shardHostAndPort.emplace(hostAndPorts.front());
@@ -224,6 +202,17 @@ auto AsyncRequestsSender::RemoteData::scheduleRequest()
.semi();
}
+SemiFuture<std::vector<HostAndPort>> AsyncRequestsSender::RemoteData::resolveShardIdToHostAndPorts(
+ const ReadPreferenceSetting& readPref) {
+ const auto shard = getShard();
+ if (!shard) {
+ return Status(ErrorCodes::ShardNotFound,
+ str::stream() << "Could not find shard " << _shardId);
+ }
+
+ return shard->getTargeter()->findHosts(readPref, CancellationToken::uncancelable());
+}
+
auto AsyncRequestsSender::RemoteData::scheduleRemoteCommand(std::vector<HostAndPort>&& hostAndPorts)
-> SemiFuture<RemoteCommandOnAnyCallbackArgs> {
hangBeforeSchedulingRemoteCommand.executeIf(
@@ -289,47 +278,43 @@ auto AsyncRequestsSender::RemoteData::handleResponse(RemoteCommandOnAnyCallbackA
}
// There was an error with either the response or the command.
- return getShard()
- .thenRunOn(*_ars->_subBaton)
- .then([this, status = std::move(status), rcr = std::move(rcr)](
- std::shared_ptr<mongo::Shard>&& shard) {
- std::vector<HostAndPort> failedTargets;
-
- if (rcr.response.target) {
- failedTargets = {*rcr.response.target};
- } else {
- failedTargets = rcr.request.target;
- }
+ auto shard = getShard();
+ if (!shard) {
+ uasserted(ErrorCodes::ShardNotFound, str::stream() << "Could not find shard " << _shardId);
+ } else {
+ std::vector<HostAndPort> failedTargets;
+
+ if (rcr.response.target) {
+ failedTargets = {*rcr.response.target};
+ } else {
+ failedTargets = rcr.request.target;
+ }
- shard->updateReplSetMonitor(failedTargets.front(), status);
- bool isStartingTransaction = _cmdObj.getField("startTransaction").booleanSafe();
- if (!_ars->_stopRetrying &&
- shard->isRetriableError(status.code(), _ars->_retryPolicy) &&
- _retryCount < kMaxNumFailedHostRetryAttempts && !isStartingTransaction) {
-
- LOGV2_DEBUG(
- 4615637,
- 1,
- "Command to remote {shardId} for hosts {hosts} failed with retryable error "
- "{error} and will be retried",
- "Command to remote shard failed with retryable error and will be retried",
- "shardId"_attr = _shardId,
- "hosts"_attr = failedTargets,
- "error"_attr = redact(status));
- ++_retryCount;
- _shardHostAndPort.reset();
- // retry through recursion
- return scheduleRequest();
- }
+ shard->updateReplSetMonitor(failedTargets.front(), status);
+ bool isStartingTransaction = _cmdObj.getField("startTransaction").booleanSafe();
+ if (!_ars->_stopRetrying && shard->isRetriableError(status.code(), _ars->_retryPolicy) &&
+ _retryCount < kMaxNumFailedHostRetryAttempts && !isStartingTransaction) {
+
+ LOGV2_DEBUG(4615637,
+ 1,
+ "Command to remote {shardId} for hosts {hosts} failed with retryable error "
+ "{error} and will be retried",
+ "Command to remote shard failed with retryable error and will be retried",
+ "shardId"_attr = _shardId,
+ "hosts"_attr = failedTargets,
+ "error"_attr = redact(status));
+ ++_retryCount;
+ _shardHostAndPort.reset();
+ // retry through recursion
+ return scheduleRequest();
+ }
+ }
- // Status' in the response.status field that aren't retried get converted to top level
- // errors
- uassertStatusOK(rcr.response.status);
+ // Status' in the response.status field that aren't retried get converted to top level errors
+ uassertStatusOK(rcr.response.status);
- // We're not okay (on the remote), but still not going to retry
- return Future<RemoteCommandOnAnyCallbackArgs>::makeReady(std::move(rcr)).semi();
- })
- .semi();
+ // We're not okay (on the remote), but still not going to retry
+ return std::move(rcr);
};
} // namespace mongo
generated by cgit v1.2.3 (git 2.39.1) at 2026-09-11 13:24:43 +0000