Update upstream source from tag 'upstream/6.0.0'master

Update to upstream version '6.0.0'
with Debian dir 5604a80ec1c96ca76f25f40d78e6ef855abec322

1 files changed, 48 insertions, 144 deletions

diff --git a/src/mongo/executor/task_executor_cursor.cpp b/src/mongo/executor/task_executor_cursor.cpp
index 57a947ed4c6..46d1bd846b4 100644
--- a/src/mongo/executor/task_executor_cursor.cpp
+++ b/src/mongo/executor/task_executor_cursor.cpp
@@ -36,45 +36,29 @@
 #include "mongo/bson/bsonobjbuilder.h"
 #include "mongo/db/query/getmore_command_gen.h"
 #include "mongo/db/query/kill_cursors_gen.h"
-#include "mongo/executor/pinned_connection_task_executor_factory.h"
-#include "mongo/logv2/log.h"
-#include "mongo/util/assert_util.h"
+#include "mongo/util/scopeguard.h"
 #include "mongo/util/time_support.h"
 
 namespace mongo {
 namespace executor {
-namespace {
-MONGO_FAIL_POINT_DEFINE(blockBeforePinnedExecutorIsDestroyedOnUnderlying);
-}  // namespace
 
-TaskExecutorCursor::TaskExecutorCursor(std::shared_ptr<executor::TaskExecutor> executor,
+TaskExecutorCursor::TaskExecutorCursor(executor::TaskExecutor* executor,
                                        const RemoteCommandRequest& rcr,
-                                       Options options)
-    : _rcr(rcr), _options(std::move(options)), _batchIter(_batch.end()) {
+                                       Options&& options)
+    : _executor(executor), _rcr(rcr), _options(std::move(options)), _batchIter(_batch.end()) {
 
     if (rcr.opCtx) {
         _lsid = rcr.opCtx->getLogicalSessionId();
     }
-    if (_options.pinConnection) {
-        _executor = makePinnedConnectionTaskExecutor(executor);
-        _underlyingExecutor = std::move(executor);
-    } else {
-        _executor = std::move(executor);
-    }
 
     _runRemoteCommand(_createRequest(_rcr.opCtx, _rcr.cmdObj));
 }
 
-TaskExecutorCursor::TaskExecutorCursor(std::shared_ptr<executor::TaskExecutor> executor,
-                                       std::shared_ptr<executor::TaskExecutor> underlyingExec,
+TaskExecutorCursor::TaskExecutorCursor(executor::TaskExecutor* executor,
                                        CursorResponse&& response,
                                        RemoteCommandRequest& rcr,
                                        Options&& options)
-    : _executor(std::move(executor)),
-      _underlyingExecutor(std::move(underlyingExec)),
-      _rcr(rcr),
-      _options(std::move(options)),
-      _batchIter(_batch.end()) {
+    : _executor(executor), _rcr(rcr), _options(std::move(options)), _batchIter(_batch.end()) {
 
     tassert(6253101, "rcr must have an opCtx to use construct cursor from response", rcr.opCtx);
     _lsid = rcr.opCtx->getLogicalSessionId();
@@ -82,16 +66,16 @@ TaskExecutorCursor::TaskExecutorCursor(std::shared_ptr<executor::TaskExecutor> e
 }
 
 TaskExecutorCursor::TaskExecutorCursor(TaskExecutorCursor&& other)
-    : _executor(std::move(other._executor)),
-      _underlyingExecutor(std::move(other._underlyingExecutor)),
+    : _executor(other._executor),
       _rcr(other._rcr),
       _options(std::move(other._options)),
       _lsid(other._lsid),
-      _cmdState(std::move(other._cmdState)),
+      _cbHandle(std::move(other._cbHandle)),
       _cursorId(other._cursorId),
       _millisecondsWaiting(other._millisecondsWaiting),
       _ns(other._ns),
       _batchNum(other._batchNum),
+      _pipe(std::move(other._pipe)),
       _additionalCursors(std::move(other._additionalCursors)) {
     // Copy the status of the batch.
     auto batchIterIndex = other._batchIter - other._batch.begin();
@@ -107,74 +91,30 @@ TaskExecutorCursor::TaskExecutorCursor(TaskExecutorCursor&& other)
     }
     // Other is no longer responsible for this cursor id.
     other._cursorId = 0;
-
-    // Other no longer owns the state for the in progress command (if there is any).
-    other._cmdState.reset();
+    // Other should not cancel the callback on destruction.
+    other._cbHandle = boost::none;
 }
 
 TaskExecutorCursor::~TaskExecutorCursor() {
     try {
-        if (_cursorId < kMinLegalCursorId || _options.pinConnection) {
-            // The initial find to establish the cursor has to be canceled to avoid leaking cursors.
-            // Once the cursor is established, killing the cursor will interrupt any ongoing
-            // `getMore` operation.
-            // Additionally, in pinned mode, we should cancel any in-progress RPC if there is one,
-            // even at the cost of churning the connection, because it's the only way to interrupt
-            // the ongoing operation.
-            if (_cmdState) {
-                _executor->cancel(_cmdState->cbHandle);
-            }
-            if (_cursorId < kMinLegalCursorId) {
-                return;
-            }
+        if (_cbHandle) {
+            _executor->cancel(*_cbHandle);
         }
 
-        // We deliberately ignore failures to kill the cursor. This "best effort" is acceptable
-        // because some timeout mechanism on the remote host can be expected to reap it later.
-        //
-        // That timeout mechanism could be the default cursor timeout, or the logical session
-        // timeout if an lsid is used.
-        //
-        // In non-pinned mode, killing the cursor also interrupts any ongoing getMore operations on
-        // this cursor. Avoid canceling the remote command through its callback handle as that may
-        // close the underlying connection.
-        //
-        // In pinned mode, we must await completion of the killCursors to safely reuse the pinned
-        // connection. This requires allocating an executor thread (from `_underlyingExecutor`) upon
-        // completion of the killCursors command to shutdown and destroy the pinned executor. This
-        // is necessary as joining an executor from its own threads results in a deadlock.
-        TaskExecutor::RemoteCommandCallbackFn callbackToRun = [](const auto&) {};
-        if (_options.pinConnection) {
-            invariant(_underlyingExecutor,
-                      "TaskExecutorCursor in pinning mode must have an underlying executor");
-            callbackToRun = [main = _executor, underlying = _underlyingExecutor](const auto&) {
-                underlying->schedule([main = std::move(main)](const auto&) {
-                    if (MONGO_unlikely(
-                            blockBeforePinnedExecutorIsDestroyedOnUnderlying.shouldFail())) {
-                        LOGV2(7361300,
-                              "Hanging before destroying a TaskExecutorCursor's pinning executor.");
-                        blockBeforePinnedExecutorIsDestroyedOnUnderlying.pauseWhileSet();
-                    }
-                    // Returning from this callback will destroy the pinned executor on
-                    // underlying if this is the last TaskExecutorCursor using that pinned executor.
-                });
-            };
-        }
-        auto swCallback = _executor->scheduleRemoteCommand(
-            _createRequest(nullptr, KillCursorsCommandRequest(_ns, {_cursorId}).toBSON(BSONObj{})),
-            callbackToRun);
-
-        // It's possible the executor is already shutdown and rejects work. If so, run the callback
-        // inline.
-        if (!swCallback.isOK()) {
-            TaskExecutor::RemoteCommandCallbackArgs args(
-                _executor.get(), {}, {}, swCallback.getStatus());
-            callbackToRun(args);
+        if (_cursorId >= kMinLegalCursorId) {
+            // We deliberately ignore failures to kill the cursor.  This "best effort" is acceptable
+            // because some timeout mechanism on the remote host can be expected to reap it later.
+            //
+            // That timeout mechanism could be the default cursor timeout, or the logical session
+            // timeout if an lsid is used.
+            _executor
+                ->scheduleRemoteCommand(
+                    _createRequest(nullptr,
+                                   KillCursorsCommandRequest(_ns, {_cursorId}).toBSON(BSONObj{})),
+                    [](const auto&) {})
+                .isOK();
         }
-    } catch (const DBException& ex) {
-        LOGV2(6531704,
-              "Encountered an error while destroying a cursor executor",
-              "error"_attr = ex.toStatus());
+    } catch (const DBException&) {
     }
 }
 
@@ -196,7 +136,7 @@ void TaskExecutorCursor::populateCursor(OperationContext* opCtx) {
             _cursorId == kUnitializedCursorId);
     tassert(6253503,
             "populateCursors should only be called after a remote command has been run",
-            _cmdState);
+            _cbHandle);
     // We really only care about populating the cursor "first batch" fields, but at some point we'll
     // have to do all of the work done by this function anyway. This would have been called by
     // getNext() the first time it was called.
@@ -227,18 +167,21 @@ const RemoteCommandRequest& TaskExecutorCursor::_createRequest(OperationContext*
 }
 
 void TaskExecutorCursor::_runRemoteCommand(const RemoteCommandRequest& rcr) {
-    auto state = std::make_shared<CommandState>();
-    state->cbHandle = uassertStatusOK(_executor->scheduleRemoteCommand(
-        rcr, [state](const TaskExecutor::RemoteCommandCallbackArgs& args) {
-            if (args.response.isOK()) {
-                state->promise.emplaceValue(args.response.data);
-            } else {
-                state->promise.setError(args.response.status);
+    _cbHandle = uassertStatusOK(_executor->scheduleRemoteCommand(
+        rcr, [p = _pipe.producer](const TaskExecutor::RemoteCommandCallbackArgs& args) {
+            try {
+                if (args.response.isOK()) {
+                    p.push(args.response.data);
+                } else {
+                    p.push(args.response.status);
+                }
+            } catch (const DBException&) {
+                // If anything goes wrong, make sure we close the pipe to wake the caller of
+                // getNext()
+                p.close();
             }
         }));
-    _cmdState.swap(state);
 }
-
 void TaskExecutorCursor::_processResponse(OperationContext* opCtx, CursorResponse&& response) {
     // If this was our first batch.
     if (_cursorId == kUnitializedCursorId) {
@@ -253,51 +196,23 @@ void TaskExecutorCursor::_processResponse(OperationContext* opCtx, CursorRespons
     _batch = response.releaseBatch();
     _batchIter = _batch.begin();
 
-    // If the previous response contained a cursorId and pre-fetching is enabled, schedule the
-    // getMore.
-    if ((_cursorId != kClosedCursorId) && _options.preFetchNextBatch) {
-        _scheduleGetMore(opCtx);
-    }
-}
-
-void TaskExecutorCursor::_scheduleGetMore(OperationContext* opCtx) {
-    // The previous response must have returned an open cursor ID.
-    invariant(_cursorId >= kMinLegalCursorId);
-    // There cannot be an existing in-flight request.
-    invariant(!_cmdState);
-    GetMoreCommandRequest getMoreRequest(_cursorId, _ns.coll().toString());
-    getMoreRequest.setBatchSize(_options.batchSize);
-
-    if (_options.getMoreAugmentationWriter) {
-        // Prefetching must be disabled to use the augmenting functionality.
-        invariant(!_options.preFetchNextBatch);
-        BSONObjBuilder getMoreBob;
-        getMoreRequest.serialize({}, &getMoreBob);
-        _options.getMoreAugmentationWriter(getMoreBob);
-        _runRemoteCommand(_createRequest(opCtx, getMoreBob.obj()));
-    } else {
+    // If we got a cursor id back, pre-fetch the next batch
+    if (_cursorId) {
+        GetMoreCommandRequest getMoreRequest(_cursorId, _ns.coll().toString());
+        getMoreRequest.setBatchSize(_options.batchSize);
         _runRemoteCommand(_createRequest(opCtx, getMoreRequest.toBSON({})));
     }
 }
 
 void TaskExecutorCursor::_getNextBatch(OperationContext* opCtx) {
-    // If we don't have an in-flight request, schedule one. This will occur when the
-    // 'preFetchNextBatch' option is false.
-    if (!_cmdState) {
-        invariant(!_options.preFetchNextBatch);
-        _scheduleGetMore(opCtx);
-    }
-
-    // There should be an in-flight request at this point, either sent asyncronously when we
-    // processed the previous response or just scheduled.
-    invariant(_cmdState);
+    invariant(_cbHandle, "_getNextBatch() requires an async request to have already been sent.");
     invariant(_cursorId != kClosedCursorId);
 
     auto clock = opCtx->getServiceContext()->getPreciseClockSource();
     auto dateStart = clock->now();
     // pull out of the pipe before setting cursor id so we don't spoil this object if we're opCtx
     // interrupted
-    auto out = _cmdState->promise.getFuture().getNoThrow(opCtx);
+    auto out = _pipe.consumer.pop(opCtx);
     auto dateEnd = clock->now();
     _millisecondsWaiting += std::max(Milliseconds(0), dateEnd - dateStart);
     uassertStatusOK(out);
@@ -311,31 +226,20 @@ void TaskExecutorCursor::_getNextBatch(OperationContext* opCtx) {
 
     // if we've received a response from our last request (initial or getmore), our remote operation
     // is done.
-    _cmdState.reset();
+    _cbHandle.reset();
 
     // Parse into a vector in case the remote sent back multiple cursors.
     auto cursorResponses = CursorResponse::parseFromBSONMany(out.getValue());
     tassert(6253100, "Expected at least one response for cursor", cursorResponses.size() > 0);
     CursorResponse cr = uassertStatusOK(std::move(cursorResponses[0]));
     _processResponse(opCtx, std::move(cr));
-
     // If we have more responses, build them into cursors then hold them until a caller accesses
     // them. Skip the first response, we used it to populate this cursor.
-    // Ensure we update the RCR we give to each 'child cursor' with the current opCtx.
-    auto freshRcr = _createRequest(opCtx, _rcr.cmdObj);
-    auto copyOptions = [&] {
-        TaskExecutorCursor::Options options;
-        // In the case that pinConnection is true, we need to ensure that additional cursors also
-        // pin their connection to the same socket as the original cursor.
-        options.pinConnection = _options.pinConnection;
-        return options;
-    };
     for (unsigned int i = 1; i < cursorResponses.size(); ++i) {
         _additionalCursors.emplace_back(_executor,
-                                        _underlyingExecutor,
                                         uassertStatusOK(std::move(cursorResponses[i])),
-                                        freshRcr,
-                                        copyOptions());
+                                        _rcr,
+                                        TaskExecutorCursor::Options());
     }
 }