diff options
Diffstat (limited to 'src/mongo/s/async_requests_sender.cpp')
| -rw-r--r-- | src/mongo/s/async_requests_sender.cpp | 113 |
1 files changed, 64 insertions, 49 deletions
diff --git a/src/mongo/s/async_requests_sender.cpp b/src/mongo/s/async_requests_sender.cpp index 1fceefc428e..a5100472f69 100644 --- a/src/mongo/s/async_requests_sender.cpp +++ b/src/mongo/s/async_requests_sender.cpp @@ -37,6 +37,7 @@ #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" @@ -87,6 +88,8 @@ 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 { @@ -117,9 +120,22 @@ 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(); @@ -170,9 +186,10 @@ AsyncRequestsSender::RemoteData::RemoteData(AsyncRequestsSender* ars, BSONObj cmdObj) : _ars(ars), _shardId(std::move(shardId)), _cmdObj(std::move(cmdObj)) {} -std::shared_ptr<Shard> AsyncRequestsSender::RemoteData::getShard() { - // TODO: Pass down an OperationContext* to use here. - return Grid::get(getGlobalServiceContext())->shardRegistry()->getShardNoReload(_shardId); +SemiFuture<std::shared_ptr<Shard>> AsyncRequestsSender::RemoteData::getShard() noexcept { + return Grid::get(getGlobalServiceContext()) + ->shardRegistry() + ->getShard(*_ars->_subBaton, _shardId); } void AsyncRequestsSender::RemoteData::executeRequest() { @@ -192,7 +209,12 @@ void AsyncRequestsSender::RemoteData::executeRequest() { auto AsyncRequestsSender::RemoteData::scheduleRequest() -> SemiFuture<RemoteCommandOnAnyCallbackArgs> { - return resolveShardIdToHostAndPorts(_ars->_readPreference) + return getShard() + .thenRunOn(*_ars->_subBaton) + .then([this](auto&& shard) { + return shard->getTargeter()->findHosts(_ars->_readPreference, + CancellationToken::uncancelable()); + }) .thenRunOn(*_ars->_subBaton) .then([this](auto&& hostAndPorts) { _shardHostAndPort.emplace(hostAndPorts.front()); @@ -202,17 +224,6 @@ 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( @@ -278,43 +289,47 @@ auto AsyncRequestsSender::RemoteData::handleResponse(RemoteCommandOnAnyCallbackA } // There was an error with either the response or the command. - 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; - } + 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; - 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(); - } - } + if (rcr.response.target) { + failedTargets = {*rcr.response.target}; + } else { + failedTargets = rcr.request.target; + } - // Status' in the response.status field that aren't retried get converted to top level errors - uassertStatusOK(rcr.response.status); + 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); - // We're not okay (on the remote), but still not going to retry - return std::move(rcr); + // We're not okay (on the remote), but still not going to retry + return Future<RemoteCommandOnAnyCallbackArgs>::makeReady(std::move(rcr)).semi(); + }) + .semi(); }; } // namespace mongo |