From fe1fac1682fc9b1e36a0deb9f5c76b3e90086ed7 Mon Sep 17 00:00:00 2001 From: "auto-revert-app[bot]" <166078896+auto-revert-app[bot]@users.noreply.github.com> Date: Tue, 17 Sep 2024 10:48:06 +0000 Subject: Revert "SERVER-92565 Remove NITL implementation structures used to facilitate hedging (#27021)" (#27151) GitOrigin-RevId: b7faf4d14cb9c690918aaf5d62ac29c1ea7157c4 --- src/mongo/executor/network_interface_tl.cpp | 362 +++++++++++++++++----------- src/mongo/executor/network_interface_tl.h | 166 +++++++++---- 2 files changed, 339 insertions(+), 189 deletions(-) diff --git a/src/mongo/executor/network_interface_tl.cpp b/src/mongo/executor/network_interface_tl.cpp index 974619aa5fa..b4085be759a 100644 --- a/src/mongo/executor/network_interface_tl.cpp +++ b/src/mongo/executor/network_interface_tl.cpp @@ -428,14 +428,12 @@ NetworkInterfaceTL::CommandStateBase::CommandStateBase( RemoteCommandRequestOnAny request_, const TaskExecutor::CallbackHandle& cbHandle_) : interface(interface_), - request(RemoteCommandRequest(std::move(request_), 0)), - requestToSend(request), + requestOnAny(std::move(request_)), cbHandle(cbHandle_), timer(interface->_reactor->makeTimer()), - operationKey(request.operationKey) {} + operationKey(requestOnAny.operationKey) {} NetworkInterfaceTL::CommandStateBase::~CommandStateBase() { - invariant(!conn); interface->_unregisterCommand(cbHandle); } @@ -450,6 +448,7 @@ auto NetworkInterfaceTL::CommandState::make(NetworkInterfaceTL* interface, auto state = std::make_shared(interface, std::move(request), cbHandle); auto [promise, future] = makePromiseFuture(); state->promise = std::move(promise); + state->requestManager = std::make_unique(state.get()); interface->_registerCommand(cbHandle, state); @@ -473,8 +472,7 @@ auto NetworkInterfaceTL::CommandState::make(NetworkInterfaceTL* interface, return std::pair(state, std::move(future)); } -AsyncDBClient* NetworkInterfaceTL::CommandStateBase::getClient( - const ConnectionHandle& conn) noexcept { +AsyncDBClient* NetworkInterfaceTL::RequestState::getClient(const ConnectionHandle& conn) noexcept { if (!conn) { return nullptr; } @@ -482,42 +480,44 @@ AsyncDBClient* NetworkInterfaceTL::CommandStateBase::getClient( return checked_cast(conn.get())->client(); } -void NetworkInterfaceTL::CommandStateBase::setTimer() { +void NetworkInterfaceTL::CommandStateBase::setTimer( + const std::shared_ptr& requestState) { auto nowVal = interface->now(); triggerSendRequestNetworkTimeout.executeIf( [&](const BSONObj& data) { LOGV2(6496503, "triggerSendRequestNetworkTimeout failpoint enabled, timing out request", - "request"_attr = request.cmdObj.toString()); + "request"_attr = requestOnAny.cmdObj.toString()); // Sleep to make sure the elapsed wait time for connection timeout is > 1 millisecond. sleepmillis(100); deadline = nowVal; }, [&](const BSONObj& data) { return data["collectionNS"].valueStringData() == - request.cmdObj.firstElement().valueStringData(); + requestOnAny.cmdObj.firstElement().valueStringData(); }); - if (deadline == kNoExpirationDate || !request.enforceLocalTimeout) { + if (deadline == kNoExpirationDate || !requestOnAny.enforceLocalTimeout) { return; } const auto timeoutCode = - request.timeoutCode.get_value_or(ErrorCodes::NetworkInterfaceExceededTimeLimit); + requestOnAny.timeoutCode.get_value_or(ErrorCodes::NetworkInterfaceExceededTimeLimit); - // We don't need to capture an anchor for the CommandStateBase (i.e. this). If the request gets - // fulfilled and misses cancelling the timer (i.e. we can't lock the weak_ptr), we just want to - // return. Ideally we'd ensure that cancellation could never miss timers, but since they will - // eventually fire anyways it's not a huge deal that we don't. + // We don't need to capture an anchor for the CommandStateBase (i.e. this) since requestState + // owns a full shared_ptr to it. If the request gets fulfilled and misses cancelling the + // timer (i.e. we can't lock the weak_ptr), we just want to return. Ideally we'd ensure that + // cancellation could never miss timers, but since they will eventually fire anyways it's not a + // huge deal that we don't. timer->waitUntil(deadline, baton) - .getAsync([this, weakState = weak_from_this(), timeoutCode](Status status) { + .getAsync([this, timeoutCode, weakReq = std::weak_ptr(requestState)](Status status) { if (!status.isOK()) { return; } - auto cmdState = weakState.lock(); - if (!cmdState) { + auto requestState = weakReq.lock(); + if (!requestState) { return; } @@ -525,22 +525,23 @@ void NetworkInterfaceTL::CommandStateBase::setTimer() { return; } - const std::string message = str::stream() << "Request " << request.id << " timed out" - << ", deadline was " << deadline.toString() - << ", op was " << redact(request.toString()); + const std::string message = str::stream() + << "Request " << requestOnAny.id << " timed out" + << ", deadline was " << deadline.toString() << ", op was " + << redact(requestOnAny.toString()); LOGV2_DEBUG(22595, 2, "Request timed out", - "requestId"_attr = request.id, + "requestId"_attr = requestOnAny.id, "deadline"_attr = deadline, - "request"_attr = request); + "request"_attr = requestOnAny); fulfillFinalPromise(StatusWith(RemoteCommandOnAnyResponse( - request.target, Status(timeoutCode, message), stopwatch.elapsed()))); + requestState->host, Status(timeoutCode, message), stopwatch.elapsed()))); }); } -void NetworkInterfaceTL::CommandStateBase::returnConnection(Status status) noexcept { +void NetworkInterfaceTL::RequestState::returnConnection(Status status) noexcept { invariant(conn); auto connToReturn = std::exchange(conn, {}); @@ -558,7 +559,7 @@ void NetworkInterfaceTL::CommandStateBase::tryFinish(Status status) noexcept { invariant(promiseFulfilling.load()); LOGV2_DEBUG( - 4646302, 2, "Finished request", "requestId"_attr = request.id, "status"_attr = status); + 4646302, 2, "Finished request", "requestId"_attr = requestOnAny.id, "status"_attr = status); // The command has resolved one way or another. timer->cancel(baton); @@ -568,14 +569,17 @@ void NetworkInterfaceTL::CommandStateBase::tryFinish(Status status) noexcept { interface->_counters->recordResult(status); } - if (operationKey) { + invariant(requestManager); + if (operationKey && + !MONGO_unlikely(networkInterfaceShouldNotKillPendingRequests.shouldFail())) { // Kill operations for the request that we didn't use to fulfill the promise. - killOperation(); + requestManager->killOperationsForPendingRequests(); } if (!status.isOK()) { - // Cancel after we issue _killOperations - cancel(); + // We cancel after we issue _killOperations because, if we cancel before, existing + // RequestStates may finish and destruct to quickly. + requestManager->cancelRequests(); } networkInterfaceCommandsFailedWithErrorCode.shouldFail([&](const BSONObj& data) { @@ -584,7 +588,7 @@ void NetworkInterfaceTL::CommandStateBase::tryFinish(Status status) noexcept { return false; } - const std::string requestCmdName = request.cmdObj.firstElement().fieldName(); + const std::string requestCmdName = requestOnAny.cmdObj.firstElement().fieldName(); for (auto&& cmdName : data.getObjectField("cmdNames")) { if (cmdName.type() == String && cmdName.valueStringData() == requestCmdName) { return true; @@ -606,16 +610,18 @@ void NetworkInterfaceTL::_unregisterCommand(const TaskExecutor::CallbackHandle& } } -void NetworkInterfaceTL::CommandStateBase::cancel() noexcept { - LOGV2_DEBUG(4646301, 2, "Cancelling request", "requestId"_attr = request.id); - +void NetworkInterfaceTL::RequestState::cancel() noexcept { auto connToCancel = weakConn.lock(); if (auto clientPtr = getClient(connToCancel)) { // If we have a client, cancel it - clientPtr->cancel(baton); + clientPtr->cancel(cmdState->baton); } } +NetworkInterfaceTL::RequestState::~RequestState() { + invariant(!conn); +} + Status NetworkInterfaceTL::startCommand(const TaskExecutor::CallbackHandle& cbHandle, RemoteCommandRequestOnAny& request, RemoteCommandCompletionFn&& onFinish, @@ -636,8 +642,8 @@ Status NetworkInterfaceTL::startCommand(const TaskExecutor::CallbackHandle& cbHa auto targetNode = request.target.front(); auto [cmdState, future] = CommandState::make(this, request, cbHandle); - if (cmdState->request.timeout != cmdState->request.kNoTimeout) { - cmdState->deadline = cmdState->stopwatch.start() + cmdState->request.timeout; + if (cmdState->requestOnAny.timeout != cmdState->requestOnAny.kNoTimeout) { + cmdState->deadline = cmdState->stopwatch.start() + cmdState->requestOnAny.timeout; } cmdState->baton = baton; @@ -664,14 +670,15 @@ Status NetworkInterfaceTL::startCommand(const TaskExecutor::CallbackHandle& cbHa numConnectionNetworkTimeouts.increment(1); timeSpentWaitingBeforeConnectionTimeoutMillis.increment( durationCount(cmdState->connTimeoutWaitTime)); - auto timeoutCode = cmdState->request.timeoutCode; - if (timeoutCode && cmdState->connTimeoutWaitTime >= cmdState->request.timeout) { + auto timeoutCode = cmdState->requestOnAny.timeoutCode; + if (timeoutCode && + cmdState->connTimeoutWaitTime >= cmdState->requestOnAny.timeout) { rs.status = Status(*timeoutCode, rs.status.reason()); } if (gEnableDetailedConnectionHealthMetricLogLines.load()) { LOGV2(6496500, "Operation timed out while waiting to acquire connection", - "requestId"_attr = cmdState->request.id, + "requestId"_attr = cmdState->requestOnAny.id, "duration"_attr = cmdState->connTimeoutWaitTime); } } @@ -679,7 +686,7 @@ Status NetworkInterfaceTL::startCommand(const TaskExecutor::CallbackHandle& cbHa LOGV2_DEBUG(22597, 2, "Request finished with response", - "requestId"_attr = cmdState->request.id, + "requestId"_attr = cmdState->requestOnAny.id, "isOK"_attr = rs.isOK(), "response"_attr = redact(rs.isOK() ? rs.data.toString() : rs.status.toString())); @@ -695,11 +702,11 @@ Status NetworkInterfaceTL::startCommand(const TaskExecutor::CallbackHandle& cbHa auto connFuture = _pool->get(targetNode, request.sslMode, request.timeout); if (connFuture.isReady()) { - cmdState->trySend(std::move(connFuture).getNoThrow()); + cmdState->requestManager->trySend(std::move(connFuture).getNoThrow()); } else { // Otherwise, schedule the request. std::move(connFuture).thenRunOn(_reactor).getAsync([cmdState = cmdState](auto swConn) { - cmdState->trySend(std::move(swConn)); + cmdState->requestManager->trySend(std::move(swConn)); }); } @@ -720,17 +727,19 @@ void NetworkInterfaceTL::testEgress(const HostAndPort& hostAndPort, } } -Future NetworkInterfaceTL::CommandState::sendRequest() { - return makeReadyFutureWith([this] { - setTimer(); +Future NetworkInterfaceTL::CommandState::sendRequest( + std::shared_ptr requestState) { + return makeReadyFutureWith([this, requestState] { + setTimer(requestState); const auto connAcquiredTimer = - checked_cast(conn.get()) + checked_cast(requestState->conn.get()) ->getConnAcquiredTimer(); - return getClient(conn)->runCommandRequest( - requestToSend, baton, std::move(connAcquiredTimer)); + return RequestState::getClient(requestState->conn) + ->runCommandRequest(*requestState->request, baton, std::move(connAcquiredTimer)); }) - .then([this](RemoteCommandResponse response) { - uassertStatusOK(doMetadataHook(RemoteCommandOnAnyResponse(request.target, response))); + .then([this, requestState](RemoteCommandResponse response) { + uassertStatusOK( + doMetadataHook(RemoteCommandOnAnyResponse(requestState->host, response))); return response; }); } @@ -750,58 +759,107 @@ void NetworkInterfaceTL::CommandState::fulfillFinalPromise( promiseFulfilled.set(); } +NetworkInterfaceTL::RequestManager::RequestManager(CommandStateBase* cmdState_) + : cmdState{cmdState_} {} -void NetworkInterfaceTL::CommandStateBase::killOperation() { +void NetworkInterfaceTL::RequestManager::cancelRequests() { + std::shared_ptr requestToCancel; { stdx::lock_guard lk(mutex); - if (!conn) { + + // Once we've set isLocked to true, no more requests will be created for this manager. + // Thus, only those that have been sent already need to be cancelled. + isLocked = true; + if (!isSent) { + return; + } + requestToCancel = request.request.lock(); + if (!requestToCancel) { return; } } - if (auto status = interface->_killOperation(this); !status.isOK()) { + LOGV2_DEBUG(4646301, 2, "Cancelling request", "requestId"_attr = cmdState->requestOnAny.id); + requestToCancel->cancel(); + requestToCancel.reset(); +} + +void NetworkInterfaceTL::RequestManager::killOperationsForPendingRequests() { + // Send `_killOperation` out of band to the target with the initialized request (which + // acquired a connection), regardless of its state so long as it's not used to fulfill the + // operation. + { + stdx::lock_guard lk(mutex); + isLocked = true; + if (!isSent) { + return; + } + auto& context = request; + invariant(context.initialized); + if (auto requestState = context.request.lock(); + requestState && requestState->fulfilledPromise) { + return; + } + } + + if (auto status = cmdState->interface->_killOperation(cmdState); !status.isOK()) { LOGV2_DEBUG(4664810, 2, "Failed to send remote _killOperations", "error"_attr = status); } } -void NetworkInterfaceTL::CommandStateBase::trySend( +void NetworkInterfaceTL::RequestManager::trySend( StatusWith swConn) noexcept { forceConnectionNetworkTimeout.executeIf( [&](const BSONObj& data) { LOGV2(6496502, "forceConnectionNetworkTimeout failpoint enabled, timing out request", - "request"_attr = request.cmdObj.toString()); + "request"_attr = cmdState->requestOnAny.cmdObj.toString()); swConn = Status(ErrorCodes::PooledConnectionAcquisitionExceededTimeLimit, "PooledConnectionAcquisitionExceededTimeLimit triggered via fail point."); }, [&](const BSONObj& data) { return data["collectionNS"].valueStringData() == - request.cmdObj.firstElement().valueStringData(); + cmdState->requestOnAny.cmdObj.firstElement().valueStringData(); }); // Our connection wasn't any good if (!swConn.isOK()) { { stdx::lock_guard lk(mutex); - invariant(!conn); + + auto currentConnsResolved = ++connsResolved; + if (currentConnsResolved < cmdState->maxPossibleConns()) { + // If we still have connections outstanding, we don't need to fail the promise. + return; + } + + if (isSent) { + // If a request has been sent, we shouldn't fail the promise. + return; + } + + if (isLocked) { + // If we've finished, obviously we don't need to fail the promise. + return; + } } // We're the last one, set the promise if it hasn't already been set via cancel or timeout - if (!promiseFulfilling.swap(true)) { + if (!cmdState->promiseFulfilling.swap(true)) { if (swConn.getStatus() == ErrorCodes::PooledConnectionAcquisitionExceededTimeLimit) { - connTimeoutWaitTime = stopwatch.elapsed(); + cmdState->connTimeoutWaitTime = cmdState->stopwatch.elapsed(); } - auto& reactor = interface->_reactor; - boost::optional target = request.target; + auto& reactor = cmdState->interface->_reactor; + boost::optional target = cmdState->requestOnAny.target.front(); if (reactor->onReactorThread()) { - fulfillFinalPromise(StatusWith( + cmdState->fulfillFinalPromise(StatusWith( RemoteCommandOnAnyResponse(target, std::move(swConn.getStatus())))); } else { ExecutorFuture(reactor, swConn.getStatus()) - .getAsync([this, anchor = shared_from_this(), target](Status status) { - fulfillFinalPromise(StatusWith( + .getAsync([this, anchor = cmdState->shared_from_this(), target](Status status) { + cmdState->fulfillFinalPromise(StatusWith( RemoteCommandOnAnyResponse(target, std::move(status)))); }); } @@ -811,111 +869,137 @@ void NetworkInterfaceTL::CommandStateBase::trySend( checked_cast(swConn.getValue().get()) ->startConnAcquiredTimer(); + std::shared_ptr requestState; bool logSetMaxTimeMS = false; + RemoteCommandRequestImpl* requestImpl; { stdx::lock_guard lk(mutex); - invariant(!conn); + // Increment the number of conns we were able to resolve. + ++connsResolved; + + if (isSent || isLocked) { + // Our command has already been satisfied or we have already sent out + // the request. + swConn.getValue()->indicateSuccess(); + return; + } + + isSent = true; + + requestState = std::make_shared(this, cmdState->shared_from_this()); // Set conn/weakConn+request under the lock so they will always be observed during cancel. - conn = std::move(swConn.getValue()); - weakConn = conn; + requestState->conn = std::move(swConn.getValue()); + requestState->weakConn = requestState->conn; + + requestState->request = RemoteCommandRequest(cmdState->requestOnAny, 0); + requestState->host = requestState->request->target; - if (interface->_svcCtx && request.timeout != RemoteCommandRequest::kNoTimeout && - WireSpec::getWireSpec(interface->_svcCtx).get()->isInternalClient) { + requestImpl = &requestState->request.value(); + + if (cmdState->interface->_svcCtx && + requestImpl->timeout != RemoteCommandRequest::kNoTimeout && + WireSpec::getWireSpec(cmdState->interface->_svcCtx).get()->isInternalClient) { logSetMaxTimeMS = true; BSONObjBuilder updatedCmdBuilder; - updatedCmdBuilder.appendElements(requestToSend.cmdObj); - updatedCmdBuilder.append("maxTimeMSOpOnly", requestToSend.timeout.count()); - requestToSend.cmdObj = updatedCmdBuilder.obj(); + updatedCmdBuilder.appendElements(requestImpl->cmdObj); + updatedCmdBuilder.append("maxTimeMSOpOnly", requestImpl->timeout.count()); + requestImpl->cmdObj = updatedCmdBuilder.obj(); } + + auto& context = request; + context.initialized = true; + context.request = requestState; } LOGV2_DEBUG(4646300, 2, "Sending request", - "requestId"_attr = request.id, - "target"_attr = request.target); + "requestId"_attr = cmdState->requestOnAny.id, + "target"_attr = cmdState->requestOnAny.target.front()); if (logSetMaxTimeMS) { LOGV2_DEBUG(4924402, 2, "Set maxTimeMSOpOnly for request", - "maxTimeMSOpOnly"_attr = request.timeout, - "requestId"_attr = request.id, - "target"_attr = request.target); + "maxTimeMSOpOnly"_attr = requestImpl->timeout, + "requestId"_attr = cmdState->requestOnAny.id, + "target"_attr = cmdState->requestOnAny.target.front()); } LOGV2_DEBUG(4630601, 2, "Request acquired a connection", - "requestId"_attr = request.id, - "target"_attr = request.target); + "requestId"_attr = requestState->request->id, + "target"_attr = requestState->request->target); networkInterfaceHangCommandsAfterAcquireConn.pauseWhileSet(); // An attempt to avoid sending a request after its command has been canceled or already executed // using another connection. Just a best effort to mitigate unnecessary resource consumption if // possible, and allow deterministic cancellation of requests in testing. - if (promiseFulfilling.load()) { + if (cmdState->promiseFulfilling.load()) { LOGV2_DEBUG(5813901, 2, "Skipping request as it has already been fulfilled or canceled", - "requestId"_attr = request.id, - "target"_attr = request.target); - returnConnection(Status::OK()); + "requestId"_attr = requestState->request->id, + "target"_attr = requestState->request->target); + requestState->returnConnection(Status::OK()); return; } - if (auto counters = interface->_counters) { + if (auto counters = cmdState->interface->_counters) { counters->recordSent(); } if (waitForShutdownBeforeSendRequest.shouldFail()) { - invariant(!interface->onNetworkThread()); - promiseFulfilled.get(); + invariant(!cmdState->interface->onNetworkThread()); + cmdState->promiseFulfilled.get(); } - resolve(sendRequest()); + requestState->resolve(cmdState->sendRequest(requestState)); } -void NetworkInterfaceTL::CommandStateBase::resolve(Future future) noexcept { +void NetworkInterfaceTL::RequestState::resolve(Future future) noexcept { + auto& reactor = interface()->_reactor; + auto& baton = cmdState->baton; + // Convert the RemoteCommandResponse to a RemoteCommandOnAnyResponse and wrap any error - auto anyFuture = std::move(future) - .then([this, anchor = shared_from_this()](RemoteCommandResponse response) { - // The RCRq ran successfully, wrap the result with the host in question - return RemoteCommandOnAnyResponse(request.target, std::move(response)); - }) - .onError([this, anchor = shared_from_this()](Status error) { - // The RCRq failed, wrap the error into a RCRsp with the host and - // duration - return RemoteCommandOnAnyResponse( - request.target, std::move(error), stopwatch.elapsed()); - }); - - std::move(anyFuture) - .thenRunOn( - makeGuaranteedExecutor(baton, interface->_reactor)) // Switch to the baton/reactor. + auto anyFuture = + std::move(future) + .then([this, anchor = shared_from_this()](RemoteCommandResponse response) { + // The RCRq ran successfully, wrap the result with the host in question + return RemoteCommandOnAnyResponse(host, std::move(response)); + }) + .onError([this, anchor = shared_from_this()](Status error) { + // The RCRq failed, wrap the error into a RCRsp with the host and duration + return RemoteCommandOnAnyResponse(host, std::move(error), stopwatch.elapsed()); + }); + + std::move(anyFuture) // + .thenRunOn(makeGuaranteedExecutor(baton, reactor)) // Switch to the baton/reactor. .getAsync([this, anchor = shared_from_this()](auto swr) noexcept { auto response = uassertStatusOK(swr); auto status = response.status; returnConnection(status); - if (promiseFulfilling.swap(true)) { + if (cmdState->promiseFulfilling.swap(true)) { LOGV2_DEBUG(4754301, 2, - "Skipping the response because the operation was cancelled", - "requestId"_attr = request.id, - "target"_attr = request.target, + "Skipping the response because it was already received from other node", + "requestId"_attr = request->id, + "target"_attr = request->target, "status"_attr = response.status, "response"_attr = redact(response.data)); return; } - fulfillFinalPromise(std::move(response)); + fulfilledPromise = true; + cmdState->fulfillFinalPromise(std::move(response)); }); } @@ -935,6 +1019,7 @@ auto NetworkInterfaceTL::ExhaustCommandState::make(NetworkInterfaceTL* interface auto state = std::make_shared( interface, std::move(request), cbHandle, std::move(onReply)); auto [promise, future] = makePromiseFuture(); + state->requestManager = std::make_unique(state.get()); state->promise = std::move(promise); interface->_registerCommand(cbHandle, state); @@ -956,16 +1041,17 @@ auto NetworkInterfaceTL::ExhaustCommandState::make(NetworkInterfaceTL* interface return state; } -Future NetworkInterfaceTL::ExhaustCommandState::sendRequest() try { +Future NetworkInterfaceTL::ExhaustCommandState::sendRequest( + std::shared_ptr requestState) try { auto [promise, future] = makePromiseFuture(); finalResponsePromise = std::move(promise); - setTimer(); - getClient(conn) - ->beginExhaustCommandRequest(request, baton) - .thenRunOn(interface->_reactor) - .getAsync([this](StatusWith swResponse) mutable { - continueExhaustRequest(swResponse); + setTimer(requestState); + requestState->getClient(requestState->conn) + ->beginExhaustCommandRequest(*requestState->request, baton) + .thenRunOn(requestState->interface()->_reactor) + .getAsync([this, requestState](StatusWith swResponse) mutable { + continueExhaustRequest(std::move(requestState), swResponse); }); return std::move(future).then([this](const auto& finalResponse) { return finalResponse; }); } catch (const DBException& ex) { @@ -986,7 +1072,7 @@ void NetworkInterfaceTL::ExhaustCommandState::fulfillFinalPromise( } void NetworkInterfaceTL::ExhaustCommandState::continueExhaustRequest( - StatusWith swResponse) { + std::shared_ptr requestState, StatusWith swResponse) { RemoteCommandResponse response; if (!swResponse.isOK()) { response = RemoteCommandResponse(std::move(swResponse.getStatus())); @@ -994,12 +1080,13 @@ void NetworkInterfaceTL::ExhaustCommandState::continueExhaustRequest( response = std::move(swResponse.getValue()); } - if (interface->inShutdown() || ErrorCodes::isCancellationError(response.status)) { + if (requestState->interface()->inShutdown() || + ErrorCodes::isCancellationError(response.status)) { finalResponsePromise.emplaceValue(response); return; } - auto onAnyResponse = RemoteCommandOnAnyResponse(request.target, response); + auto onAnyResponse = RemoteCommandOnAnyResponse(requestState->host, response); if (Status metadataHookStatus = doMetadataHook(onAnyResponse); !metadataHookStatus.isOK()) { finalResponsePromise.setError(metadataHookStatus); return; @@ -1024,16 +1111,16 @@ void NetworkInterfaceTL::ExhaustCommandState::continueExhaustRequest( stopwatch.restart(); } if (deadline != kNoExpirationDate) { - deadline = stopwatch.start() + request.timeout; + deadline = stopwatch.start() + requestOnAny.timeout; } - setTimer(); + setTimer(requestState); - getClient(conn) + requestState->getClient(requestState->conn) ->awaitExhaustCommand(baton) - .thenRunOn(interface->_reactor) - .getAsync([this](StatusWith swResponse) mutable { - continueExhaustRequest(swResponse); + .thenRunOn(requestState->interface()->_reactor) + .getAsync([this, requestState](StatusWith swResponse) mutable { + continueExhaustRequest(std::move(requestState), swResponse); }); } @@ -1054,20 +1141,21 @@ Status NetworkInterfaceTL::startExhaustCommand(const TaskExecutor::CallbackHandl } auto cmdState = ExhaustCommandState::make(this, request, cbHandle, std::move(onReply), baton); - if (cmdState->request.timeout != cmdState->request.kNoTimeout) { - cmdState->deadline = cmdState->stopwatch.start() + cmdState->request.timeout; + if (cmdState->requestOnAny.timeout != cmdState->requestOnAny.kNoTimeout) { + cmdState->deadline = cmdState->stopwatch.start() + cmdState->requestOnAny.timeout; } cmdState->baton = baton; + cmdState->requestManager = std::make_unique(cmdState.get()); // Attempt to get a connection to the target host auto connFuture = _pool->get(request.target.front(), request.sslMode, request.timeout); if (connFuture.isReady()) { - cmdState->trySend(std::move(connFuture).getNoThrow()); + cmdState->requestManager->trySend(std::move(connFuture).getNoThrow()); } else { // For every connection future we didn't have immediately ready, schedule std::move(connFuture).thenRunOn(_reactor).getAsync([cmdState](auto swConn) { - cmdState->trySend(std::move(swConn)); + cmdState->requestManager->trySend(std::move(swConn)); }); } @@ -1095,18 +1183,18 @@ void NetworkInterfaceTL::cancelCommand(const TaskExecutor::CallbackHandle& cbHan LOGV2_DEBUG(22599, 2, "Canceling operation for request", - "request"_attr = redact(cmdStateToCancel->request.cmdObj)); - cmdStateToCancel->fulfillFinalPromise({ErrorCodes::CallbackCanceled, - str::stream() - << "Command canceled; original request was: " - << redact(cmdStateToCancel->request.cmdObj)}); + "request"_attr = redact(cmdStateToCancel->requestOnAny.toString())); + cmdStateToCancel->fulfillFinalPromise( + {ErrorCodes::CallbackCanceled, + str::stream() << "Command canceled; original request was: " + << redact(cmdStateToCancel->requestOnAny.toString())}); } } Status NetworkInterfaceTL::_killOperation(CommandStateBase* cmdStateToKill) try { auto [target, sslMode] = [&] { - const auto& request = cmdStateToKill->request; - return std::make_pair(request.target, request.sslMode); + const auto& request = cmdStateToKill->requestOnAny; + return std::make_pair(request.target.front(), request.sslMode); }(); auto operationKey = cmdStateToKill->operationKey.value(); @@ -1146,7 +1234,7 @@ Status NetworkInterfaceTL::_killOperation(CommandStateBase* cmdStateToKill) try std::move(connFuture) .thenRunOn(_reactor) .getAsync([this, killOpCmdState = killOpCmdState](auto swConn) { - killOpCmdState->trySend(std::move(swConn)); + killOpCmdState->requestManager->trySend(std::move(swConn)); }); return Status::OK(); } catch (const DBException& ex) { diff --git a/src/mongo/executor/network_interface_tl.h b/src/mongo/executor/network_interface_tl.h index 942c581a8fd..88d22debaa5 100644 --- a/src/mongo/executor/network_interface_tl.h +++ b/src/mongo/executor/network_interface_tl.h @@ -163,10 +163,16 @@ public: Milliseconds timeout) override; private: + struct RequestState; + struct RequestManager; + /** - * For an RPC, an instance of `CommandState` is created to capture the state of the - * remote command. As part of running a remote command, `NITL` sends out a request - * to the specified target. + * For each logical RPC, an instance of `CommandState` is created to capture the state of the + * remote command. As part of running a remote command, `NITL` sends out one or more requests + * to the specified targets, and `RequestState` represents the state of each request. + * `CommandState` owns a `RequestManager` that tracks individual requests. For each request sent + * over the wire, `RequestManager` creates a `Context` that holds a weak pointer to the + * `Request`, as well as the index of the target. */ struct CommandStateBase : public std::enable_shared_from_this { @@ -175,10 +181,16 @@ private: const TaskExecutor::CallbackHandle& cbHandle_); virtual ~CommandStateBase(); - using ConnectionHandle = std::shared_ptr; - using WeakConnectionHandle = std::weak_ptr; + /** + * Use the current RequestState to send out a command request. + */ + virtual Future sendRequest( + std::shared_ptr requestState) = 0; - virtual Future sendRequest() = 0; + /** + * Set a timer to fulfill the promise with a timeout error. + */ + void setTimer(const std::shared_ptr& requestState); /** * Fulfill the promise with the response. @@ -194,51 +206,22 @@ private: */ void tryFinish(Status status) noexcept; - /** - * Return the current connection to the pool and unset it locally. - * - * This must be called from the networking thread (i.e. the reactor). - */ - void returnConnection(Status status) noexcept; - - void trySend(StatusWith swConn) noexcept; - - void killOperation(); - - /** - * Set a timer to fulfill the promise with a timeout error. - */ - virtual void setTimer(); - - /** - * Resolve an eventual response - */ - void resolve(Future future) noexcept; - - /** - * Return the client for a given connection - */ - static AsyncDBClient* getClient(const ConnectionHandle& conn) noexcept; - - /** - * Cancel the current client operation or do nothing if there is no client. - */ - void cancel() noexcept; - /** * Run the NetworkInterface's MetadataHook on a given request if this Command isn't already * finished. */ Status doMetadataHook(const RemoteCommandOnAnyResponse& response); - NetworkInterfaceTL* interface; - - // Original request as received from the caller. - const RemoteCommandRequest request; + /** + * Return the most connections we expect to be able to acquire. + */ + size_t maxPossibleConns() const noexcept { + return requestOnAny.target.size(); + } - // Modified request to emit on the wire. - RemoteCommandRequest requestToSend; + NetworkInterfaceTL* interface; + RemoteCommandRequestOnAny requestOnAny; TaskExecutor::CallbackHandle cbHandle; Date_t deadline = kNoExpirationDate; @@ -247,6 +230,8 @@ private: BatonHandle baton; std::unique_ptr timer; + std::unique_ptr requestManager; + // The thread that sets this bit must subsequently call fulfillFinalPromise() exactly once. // Once it is set, no other thread may call fulfillFinalPromise(). Atomic promiseFulfilling{false}; @@ -258,12 +243,6 @@ private: // Total time spent waiting for connections that eventually time out. Milliseconds connTimeoutWaitTime{0}; - - ConnectionHandle conn; - WeakConnectionHandle weakConn; - - // Synchronizes requestToSend, conn, and weakConn. - Mutex mutex = MONGO_MAKE_LATCH("NetworkInterfaceTL::CommandStateBase::mutex"); }; struct CommandState final : public CommandStateBase { @@ -278,7 +257,8 @@ private: RemoteCommandRequestOnAny request, const TaskExecutor::CallbackHandle& cbHandle); - Future sendRequest() override; + Future sendRequest( + std::shared_ptr requestState) override; void fulfillFinalPromise(StatusWith response) override; @@ -300,11 +280,13 @@ private: RemoteCommandOnReplyFn&& onReply, const BatonHandle& baton); - Future sendRequest() override; + Future sendRequest( + std::shared_ptr requestState) override; void fulfillFinalPromise(StatusWith response) override; - void continueExhaustRequest(StatusWith swResponse); + void continueExhaustRequest(std::shared_ptr requestState, + StatusWith swResponse); // Protects against race between reactor thread restarting stopwatch during exhaust // request and main thread reading stopwatch elapsed time during shutdown. @@ -315,6 +297,86 @@ private: RemoteCommandOnReplyFn onReplyFn; }; + struct RequestManager { + RequestManager(CommandStateBase* cmdState); + + void trySend(StatusWith swConn) noexcept; + void cancelRequests(); + void killOperationsForPendingRequests(); + + CommandStateBase* cmdState; + + /** + * Holds context for individual requests, and is only valid if initialized. + */ + struct Context { + bool initialized = false; + std::weak_ptr request; + }; + Context request; + + Mutex mutex = MONGO_MAKE_LATCH("NetworkInterfaceTL::RequestManager::mutex"); + + // Number of connections we've resolved. + size_t connsResolved{0}; + + // Set to true after we have sent the request. + bool isSent{false}; + + // Set to true when the command finishes or is canceled to block remaining requests. + bool isLocked{false}; + }; + + struct RequestState final : public std::enable_shared_from_this { + using ConnectionHandle = std::shared_ptr; + using WeakConnectionHandle = std::weak_ptr; + RequestState(RequestManager* mgr, std::shared_ptr cmdState_) + : cmdState{std::move(cmdState_)}, requestManager(mgr) {} + + ~RequestState(); + + /** + * Return the client for a given connection + */ + static AsyncDBClient* getClient(const ConnectionHandle& conn) noexcept; + + /** + * Cancel the current client operation or do nothing if there is no client. + */ + void cancel() noexcept; + + /** + * Return the current connection to the pool and unset it locally. + * + * This must be called from the networking thread (i.e. the reactor). + */ + void returnConnection(Status status) noexcept; + + /** + * Resolve an eventual response + */ + void resolve(Future future) noexcept; + + NetworkInterfaceTL* interface() noexcept { + return cmdState->interface; + } + + std::shared_ptr cmdState; + + ClockSource::StopWatch stopwatch; + + RequestManager* const requestManager{nullptr}; + + boost::optional request; + HostAndPort host; + ConnectionHandle conn; + WeakConnectionHandle weakConn; + + // Set to true if the response to the request is used to fulfill the command's + // promise. + bool fulfilledPromise{false}; + }; + struct AlarmState { AlarmState(Date_t when_, TaskExecutor::CallbackHandle cbHandle_, -- cgit v1.2.3