Revert "SERVER-49371 Revived original recursive ICE and applied to stack overflow tests"

This reverts commit 72d104ca45461075fc7a4deb11d3522ff24c62f6.
This reverts commit 77ce2860da66ab9924c0668c76fae564eb7c7dfc.

6 files changed, 34 insertions, 95 deletions

diff --git a/src/mongo/executor/connection_pool_test.cpp b/src/mongo/executor/connection_pool_test.cpp
index b08f8d3900a..6d70ecc54ba 100644
--- a/src/mongo/executor/connection_pool_test.cpp
+++ b/src/mongo/executor/connection_pool_test.cpp
@@ -105,7 +105,7 @@ protected:
     void dropConnectionsTest(std::shared_ptr<ConnectionPool> const& pool, Ptr t);
 
 private:
-    std::shared_ptr<OutOfLineExecutor> _executor = InlineQueuedCountingExecutor::make();
+    std::shared_ptr<OutOfLineExecutor> _executor = InlineCountingExecutor::make();
     std::shared_ptr<ConnectionPool> _pool;
 };
 
diff --git a/src/mongo/util/executor_test_util.h b/src/mongo/util/executor_test_util.h
index 665651f53dd..b6fa216e446 100644
--- a/src/mongo/util/executor_test_util.h
+++ b/src/mongo/util/executor_test_util.h
@@ -35,10 +35,9 @@ namespace mongo {
 /**
  * An "OutOfLineExecutor" that actually runs on the same thread of execution
  * This executor is not thread-safe, and accessing it by multiple threads is prohibited.
- * Multi-threaded accesses to instances of "InlineQueuedCountingExecutor" result in undefined
- * behavior.
+ * Multi-threaded accesses to instances of "InlineCountingExecutor" result in undefined behavior.
  */
-class InlineQueuedCountingExecutor : public OutOfLineExecutor {
+class InlineCountingExecutor : public OutOfLineExecutor {
 public:
     void schedule(Task task) override {
         // Add the task to our queue
@@ -69,7 +68,7 @@ public:
     }
 
     static auto make() {
-        return std::make_shared<InlineQueuedCountingExecutor>();
+        return std::make_shared<InlineCountingExecutor>();
     }
 
     bool inSchedule;
@@ -78,22 +77,6 @@ public:
     std::atomic<uint32_t> tasksRun{0};  // NOLINT
 };
 
-class InlineRecursiveCountingExecutor final : public OutOfLineExecutor {
-public:
-    void schedule(Task task) noexcept override {
-        // Relaxed to avoid adding synchronization where there otherwise wouldn't be. That would
-        // cause a false negative from TSAN.
-        tasksRun.fetch_add(1, std::memory_order_relaxed);
-        task(Status::OK());
-    }
-
-    static auto make() {
-        return std::make_shared<InlineRecursiveCountingExecutor>();
-    }
-
-    std::atomic<int32_t> tasksRun{0};  // NOLINT
-};
-
 class RejectingExecutor final : public OutOfLineExecutor {
 public:
     void schedule(Task task) noexcept override {
diff --git a/src/mongo/util/future_test_executor_future.cpp b/src/mongo/util/future_test_executor_future.cpp
index c4d2b43a520..1c6dc09224c 100644
--- a/src/mongo/util/future_test_executor_future.cpp
+++ b/src/mongo/util/future_test_executor_future.cpp
@@ -39,7 +39,7 @@ namespace {
 TEST(Executor_Future, Success_getAsync) {
     FUTURE_SUCCESS_TEST([] {},
                         [](/*Future<void>*/ auto&& fut) {
-                            auto exec = InlineQueuedCountingExecutor::make();
+                            auto exec = InlineCountingExecutor::make();
                             auto pf = makePromiseFuture<void>();
                             ExecutorFuture<void>(exec).thenRunOn(exec).getAsync(
                                 [outside = std::move(pf.promise)](Status status) mutable {
@@ -70,7 +70,7 @@ TEST(Executor_Future, Reject_getAsync) {
 TEST(Executor_Future, Success_then) {
     FUTURE_SUCCESS_TEST([] {},
                         [](/*Future<void>*/ auto&& fut) {
-                            auto exec = InlineQueuedCountingExecutor::make();
+                            auto exec = InlineCountingExecutor::make();
                             ASSERT_EQ(std::move(fut).thenRunOn(exec).then([]() { return 3; }).get(),
                                       3);
                             ASSERT_EQ(exec->tasksRun.load(), 1);
@@ -93,7 +93,7 @@ TEST(Executor_Future, Reject_then) {
 
 TEST(Executor_Future, Fail_then) {
     FUTURE_FAIL_TEST<void>([](/*Future<void>*/ auto&& fut) {
-        auto exec = InlineQueuedCountingExecutor::make();
+        auto exec = InlineCountingExecutor::make();
         ASSERT_EQ(std::move(fut)
                       .thenRunOn(exec)
                       .then([]() {
@@ -109,7 +109,7 @@ TEST(Executor_Future, Fail_then) {
 TEST(Executor_Future, Success_onError) {
     FUTURE_SUCCESS_TEST([] { return 3; },
                         [](/*Future<int>*/ auto&& fut) {
-                            auto exec = InlineQueuedCountingExecutor::make();
+                            auto exec = InlineCountingExecutor::make();
                             ASSERT_EQ(std::move(fut)
                                           .thenRunOn(exec)
                                           .onError([](Status&&) {
@@ -124,7 +124,7 @@ TEST(Executor_Future, Success_onError) {
 
 TEST(Executor_Future, Fail_onErrorSimple) {
     FUTURE_FAIL_TEST<int>([](/*Future<int>*/ auto&& fut) {
-        auto exec = InlineQueuedCountingExecutor::make();
+        auto exec = InlineCountingExecutor::make();
         ASSERT_EQ(std::move(fut)
                       .thenRunOn(exec)
                       .onError([](Status s) {
@@ -139,7 +139,7 @@ TEST(Executor_Future, Fail_onErrorSimple) {
 
 TEST(Executor_Future, Fail_onErrorCode_OtherCode) {
     FUTURE_FAIL_TEST<void>([](/*Future<void>*/ auto&& fut) {
-        auto exec = InlineQueuedCountingExecutor::make();
+        auto exec = InlineCountingExecutor::make();
         ASSERT_EQ(
             std::move(fut)
                 .thenRunOn(exec)
@@ -153,7 +153,7 @@ TEST(Executor_Future, Fail_onErrorCode_OtherCode) {
 TEST(Executor_Future, Success_then_onError_onError_then) {
     FUTURE_SUCCESS_TEST([] {},
                         [](/*Future<void>*/ auto&& fut) {
-                            auto exec = InlineQueuedCountingExecutor::make();
+                            auto exec = InlineCountingExecutor::make();
                             ASSERT_EQ(
                                 std::move(fut)
                                     .thenRunOn(exec)
@@ -174,7 +174,7 @@ TEST(Executor_Future, Success_reject_recoverToFallback) {
     FUTURE_SUCCESS_TEST([] {},
                         [](/*Future<void>*/ auto&& fut) {
                             auto rejecter = RejectingExecutor::make();
-                            auto accepter = InlineQueuedCountingExecutor::make();
+                            auto accepter = InlineCountingExecutor::make();
 
                             auto res = std::move(fut)
                                            .thenRunOn(rejecter)
diff --git a/src/mongo/util/future_test_shared_future.cpp b/src/mongo/util/future_test_shared_future.cpp
index 33882f2c1ad..e2740860d2e 100644
--- a/src/mongo/util/future_test_shared_future.cpp
+++ b/src/mongo/util/future_test_shared_future.cpp
@@ -116,7 +116,7 @@ TEST(SharedFuture_Void, isReady_share_TSAN_OK) {
 TEST(SharedFuture, ModificationsArePrivate) {
     FUTURE_SUCCESS_TEST([] { return 1; },
                         [](/*Future<int>*/ auto&& fut) {
-                            const auto exec = InlineQueuedCountingExecutor::make();
+                            const auto exec = InlineCountingExecutor::make();
                             const auto shared = std::move(fut).share();
 
                             const auto checkFunc = [](int&& i) {
@@ -150,7 +150,7 @@ MONGO_COMPILER_NOINLINE void useALotOfStackSpace() {
 TEST(SharedFuture, NoStackOverflow_Call) {
     FUTURE_SUCCESS_TEST([] {},
                         [](/*Future<void>*/ auto&& fut) {
-                            const auto exec = InlineRecursiveCountingExecutor::make();
+                            const auto exec = InlineCountingExecutor::make();
                             const auto shared = std::move(fut).share();
 
                             std::vector<SemiFuture<void>> collector;
@@ -170,7 +170,7 @@ TEST(SharedFuture, NoStackOverflow_Call) {
 TEST(SharedFuture, NoStackOverflow_Destruction) {
     FUTURE_SUCCESS_TEST([] {},
                         [](/*Future<void>*/ auto&& fut) {
-                            const auto exec = InlineRecursiveCountingExecutor::make();
+                            const auto exec = InlineCountingExecutor::make();
                             const auto shared = std::move(fut).share();
 
                             std::vector<SemiFuture<void>> collector;
@@ -197,7 +197,7 @@ TEST(SharedFuture, ThenChaining_Sync) {
     FUTURE_SUCCESS_TEST(
         [] {},
         [](/*Future<void>*/ auto&& fut) {
-            const auto exec = InlineQueuedCountingExecutor::make();
+            const auto exec = InlineCountingExecutor::make();
 
             auto res = std::move(fut).then([] { return SharedSemiFuture(1); });
 
@@ -214,7 +214,7 @@ TEST(SharedFuture, ThenChaining_Async) {
     FUTURE_SUCCESS_TEST(
         [] {},
         [](/*Future<void>*/ auto&& fut) {
-            const auto exec = InlineQueuedCountingExecutor::make();
+            const auto exec = InlineCountingExecutor::make();
 
             auto res = std::move(fut).then([] { return async([] { return 1; }).share(); });
 
@@ -230,7 +230,7 @@ TEST(SharedFuture, ThenChaining_Async) {
 TEST(SharedFuture, ThenChaining_Async_DoubleShare) {
     FUTURE_SUCCESS_TEST([] {},
                         [](/*Future<void>*/ auto&& fut) {
-                            const auto exec = InlineQueuedCountingExecutor::make();
+                            const auto exec = InlineCountingExecutor::make();
 
                             auto res = std::move(fut).share().thenRunOn(exec).then(
                                 [] { return async([] { return 1; }).share(); });
@@ -242,7 +242,7 @@ TEST(SharedFuture, ThenChaining_Async_DoubleShare) {
 TEST(SharedFuture, AddChild_ThenRunOn_Get) {
     FUTURE_SUCCESS_TEST([] {},
                         [](/*Future<void>*/ auto&& fut) {
-                            const auto exec = InlineQueuedCountingExecutor::make();
+                            const auto exec = InlineCountingExecutor::make();
                             auto shared = std::move(fut).share();
                             auto fut2 = shared.thenRunOn(exec).then([] {});
                             shared.get();
@@ -263,7 +263,7 @@ TEST(SharedFuture, AddChild_Split_Get) {
 TEST(SharedFuture, InterruptedGet_AddChild_Get) {
     FUTURE_SUCCESS_TEST([] {},
                         [](/*Future<void>*/ auto&& fut) {
-                            const auto exec = InlineQueuedCountingExecutor::make();
+                            const auto exec = InlineCountingExecutor::make();
                             DummyInterruptable dummyInterruptable;
 
                             auto shared = std::move(fut).share();
@@ -291,7 +291,7 @@ TEST(SharedFuture, ConcurrentTest_OneSharedFuture) {
 
         for (int i = 0; i < nThreads; i++) {
             threads[i] = stdx::thread([i, &shared] {
-                auto exec = InlineQueuedCountingExecutor::make();
+                auto exec = InlineCountingExecutor::make();
                 if (i % 5 == 0) {
                     // just wait directly on shared.
                     shared.get();
@@ -335,7 +335,7 @@ TEST(SharedFuture, ConcurrentTest_ManySharedFutures) {
         for (int i = 0; i < nThreads; i++) {
             threads[i] = stdx::thread([i, &promise] {
                 auto shared = promise.getFuture();
-                auto exec = InlineQueuedCountingExecutor::make();
+                auto exec = InlineCountingExecutor::make();
 
                 if (i % 5 == 0) {
                     // just wait directly on shared.
diff --git a/src/mongo/util/future_test_utils.h b/src/mongo/util/future_test_utils.h
index 374f00f7a7f..32aa149a088 100644
--- a/src/mongo/util/future_test_utils.h
+++ b/src/mongo/util/future_test_utils.h
@@ -149,7 +149,7 @@ void FUTURE_SUCCESS_TEST(const CompletionFunc& completion, const TestFunc& test)
     }
 
     if constexpr (doExecutorFuture) {  // immediate executor future
-        auto exec = InlineQueuedCountingExecutor::make();
+        auto exec = InlineCountingExecutor::make();
         test(Future<CompletionType>::makeReady(completion()).thenRunOn(exec));
     }
 }
@@ -178,7 +178,7 @@ void FUTURE_SUCCESS_TEST(const CompletionFunc& completion, const TestFunc& test)
 
     if constexpr (doExecutorFuture) {  // immediate executor future
         completion();
-        auto exec = InlineQueuedCountingExecutor::make();
+        auto exec = InlineCountingExecutor::make();
         test(Future<CompletionType>::makeReady().thenRunOn(exec));
     }
 }
@@ -203,7 +203,7 @@ void FUTURE_FAIL_TEST(const TestFunc& test) {
         }));
     }
     if constexpr (doExecutorFuture) {  // immediate executor future
-        auto exec = InlineQueuedCountingExecutor::make();
+        auto exec = InlineCountingExecutor::make();
         test(Future<CompletionType>::makeReady(failStatus()).thenRunOn(exec));
     }
 }
diff --git a/src/mongo/util/out_of_line_executor_test.cpp b/src/mongo/util/out_of_line_executor_test.cpp
index e7d66eb505b..29fa1892720 100644
--- a/src/mongo/util/out_of_line_executor_test.cpp
+++ b/src/mongo/util/out_of_line_executor_test.cpp
@@ -51,10 +51,10 @@ TEST(ExecutorTest, RejectingExecutor) {
     }
 }
 
-TEST(ExecutorTest, InlineQueuedCountingExecutor) {
+TEST(ExecutorTest, InlineCountingExecutor) {
     // Verify that the executor accepts every time and keeps an accurate count.
-    const auto execA = InlineQueuedCountingExecutor::make();
-    const auto execB = InlineQueuedCountingExecutor::make();
+    const auto execA = InlineCountingExecutor::make();
+    const auto execB = InlineCountingExecutor::make();
 
     // Using prime numbers so there is no chance of multiple traps
     static constexpr size_t kCountA = 1013;
@@ -95,50 +95,6 @@ TEST(ExecutorTest, InlineQueuedCountingExecutor) {
     ASSERT_EQ(execB->tasksRun.load(), kCountB);
 }
 
-TEST(ExecutorTest, InlineRecursiveCountingExecutor) {
-    // Verify that the executor accepts every time and keeps an accurate count.
-    const auto execA = InlineRecursiveCountingExecutor::make();
-    const auto execB = InlineRecursiveCountingExecutor::make();
-
-    // Using prime numbers so there is no chance of multiple traps
-    static constexpr size_t kCountA = 1013;
-    static constexpr size_t kCountB = 1511;
-
-    // Schedule kCountA tasks one at a time.
-    for (size_t i = 0; i < kCountA; ++i) {
-        execA->schedule([&](Status status) {
-            ASSERT_OK(status);
-            ASSERT_EQ(execA->tasksRun.load(), (i + 1));
-        });
-    }
-
-    {
-        // Schedule kCountB tasks recursively.
-        size_t i = 0;
-        std::function<void(Status)> recurseExec;
-        bool inTask = false;
-
-        recurseExec = [&](Status status) {
-            ASSERT(!std::exchange(inTask, true));
-            ASSERT_OK(status);
-
-            auto tasksRun = size_t(execB->tasksRun.load());
-            ASSERT_EQ(tasksRun, ++i);
-            if (tasksRun < kCountB) {
-                execB->schedule(recurseExec);
-            }
-
-            ASSERT(std::exchange(inTask, false));
-        };
-
-        execB->schedule(recurseExec);
-    }
-
-    // Verify that running executors together didn't change the expected counts.
-    ASSERT_EQ(execA->tasksRun.load(), kCountA);
-    ASSERT_EQ(execB->tasksRun.load(), kCountB);
-}
-
 DEATH_TEST(ExecutorTest,
            GuaranteedExecutor_MainInvalid_FallbackInvalid,
            GuaranteedExecutor::kNoExecutorStr) {
@@ -156,7 +112,7 @@ DEATH_TEST(ExecutorTest,
 
 TEST(ExecutorTest, GuaranteedExecutor_MainInvalid_FallbackAccepts) {
     // If we only have a fallback, then everything runs on it.
-    const auto countExec = InlineQueuedCountingExecutor::make();
+    const auto countExec = InlineCountingExecutor::make();
     const auto gwarExec = makeGuaranteedExecutor({}, countExec);
 
     static constexpr size_t kCount = 1000;
@@ -187,7 +143,7 @@ DEATH_TEST(ExecutorTest,
 TEST(ExecutorTest, GuaranteedExecutor_MainRejects_FallbackAccepts) {
     // If the main rejects and the fallback accepts, then run on the fallback.
     const auto rejectExec = RejectingExecutor::make();
-    const auto countExec = InlineQueuedCountingExecutor::make();
+    const auto countExec = InlineCountingExecutor::make();
     const auto gwarExec = makeGuaranteedExecutor(rejectExec, countExec);
 
     static constexpr size_t kCount = 1000;
@@ -200,7 +156,7 @@ TEST(ExecutorTest, GuaranteedExecutor_MainRejects_FallbackAccepts) {
 
 TEST(ExecutorTest, GuaranteedExecutor_MainAccepts_FallbackInvalid) {
     // If the main accepts and we don't have a fallback, then run on the main.
-    const auto countExec = InlineQueuedCountingExecutor::make();
+    const auto countExec = InlineCountingExecutor::make();
     const auto gwarExec = makeGuaranteedExecutor(countExec, {});
 
     static constexpr size_t kCount = 1000;
@@ -213,7 +169,7 @@ TEST(ExecutorTest, GuaranteedExecutor_MainAccepts_FallbackInvalid) {
 
 TEST(ExecutorTest, GuaranteedExecutor_MainAccepts_FallbackRejects) {
     // If the main accepts and the fallback would reject, then run on the main.
-    const auto countExec = InlineQueuedCountingExecutor::make();
+    const auto countExec = InlineCountingExecutor::make();
     const auto rejectExec = RejectingExecutor::make();
     const auto gwarExec = makeGuaranteedExecutor(countExec, rejectExec);
 
@@ -228,8 +184,8 @@ TEST(ExecutorTest, GuaranteedExecutor_MainAccepts_FallbackRejects) {
 
 TEST(ExecutorTest, GuaranteedExecutor_MainAccepts_FallbackAccepts) {
     // If both executor accepts, then run on the main.
-    const auto countExecA = InlineQueuedCountingExecutor::make();
-    const auto countExecB = InlineQueuedCountingExecutor::make();
+    const auto countExecA = InlineCountingExecutor::make();
+    const auto countExecB = InlineCountingExecutor::make();
     const auto gwarExec = makeGuaranteedExecutor(countExecA, countExecB);
 
     static constexpr size_t kCount = 1000;