Revert "Revert "Revert "[event_engine] Thread pool that can handle deletion in a callback"" (#30973)" (#30995)

This reverts commit fed749d100.

BUILD

@@ -2481,7 +2481,6 @@ grpc_cc_library(
     external_deps = [
         "absl/base:core_headers",
         "absl/functional:any_invocable",
-        "absl/time",
     ],
     deps = [
         "forkable",

CMakeLists.txt

@@ -1200,7 +1200,6 @@ if(gRPC_BUILD_TESTS)
   add_dependencies(buildtests_cxx test_cpp_util_time_test)
   add_dependencies(buildtests_cxx thd_test)
   add_dependencies(buildtests_cxx thread_manager_test)
-  add_dependencies(buildtests_cxx thread_pool_test)
   add_dependencies(buildtests_cxx thread_quota_test)
   if(_gRPC_PLATFORM_LINUX OR _gRPC_PLATFORM_MAC OR _gRPC_PLATFORM_POSIX)
     add_dependencies(buildtests_cxx thread_stress_test)
@@ -18301,45 +18300,6 @@ target_link_libraries(thread_manager_test
 )
 
 
-endif()
-if(gRPC_BUILD_TESTS)
-
-add_executable(thread_pool_test
-  src/core/lib/event_engine/forkable.cc
-  src/core/lib/event_engine/thread_pool.cc
-  test/core/event_engine/thread_pool_test.cc
-  third_party/googletest/googletest/src/gtest-all.cc
-  third_party/googletest/googlemock/src/gmock-all.cc
-)
-
-target_include_directories(thread_pool_test
-  PRIVATE
-    ${CMAKE_CURRENT_SOURCE_DIR}
-    ${CMAKE_CURRENT_SOURCE_DIR}/include
-    ${_gRPC_ADDRESS_SORTING_INCLUDE_DIR}
-    ${_gRPC_RE2_INCLUDE_DIR}
-    ${_gRPC_SSL_INCLUDE_DIR}
-    ${_gRPC_UPB_GENERATED_DIR}
-    ${_gRPC_UPB_GRPC_GENERATED_DIR}
-    ${_gRPC_UPB_INCLUDE_DIR}
-    ${_gRPC_XXHASH_INCLUDE_DIR}
-    ${_gRPC_ZLIB_INCLUDE_DIR}
-    third_party/googletest/googletest/include
-    third_party/googletest/googletest
-    third_party/googletest/googlemock/include
-    third_party/googletest/googlemock
-    ${_gRPC_PROTO_GENS_DIR}
-)
-
-target_link_libraries(thread_pool_test
-  ${_gRPC_PROTOBUF_LIBRARIES}
-  ${_gRPC_ALLTARGETS_LIBRARIES}
-  absl::flat_hash_set
-  absl::any_invocable
-  gpr
-)
-
-
 endif()
 if(gRPC_BUILD_TESTS)
 

build_autogenerated.yaml

@@ -9987,21 +9987,6 @@ targets:
   deps:
   - grpc++_test_config
   - grpc++_test_util
-- name: thread_pool_test
-  gtest: true
-  build: test
-  language: c++
-  headers:
-  - src/core/lib/event_engine/forkable.h
-  - src/core/lib/event_engine/thread_pool.h
-  src:
-  - src/core/lib/event_engine/forkable.cc
-  - src/core/lib/event_engine/thread_pool.cc
-  - test/core/event_engine/thread_pool_test.cc
-  deps:
-  - absl/container:flat_hash_set
-  - absl/functional:any_invocable
-  - gpr
 - name: thread_quota_test
   gtest: true
   build: test

src/core/lib/event_engine/thread_pool.cc

@@ -20,148 +20,138 @@
 
 #include "src/core/lib/event_engine/thread_pool.h"
 
-#include <memory>
 #include <utility>
 
-#include "absl/time/clock.h"
-#include "absl/time/time.h"
-
-#include <grpc/support/log.h>
-
 #include "src/core/lib/gprpp/thd.h"
 
 namespace grpc_event_engine {
 namespace experimental {
 
-void ThreadPool::StartThread(StatePtr state) {
+ThreadPool::Thread::Thread(ThreadPool* pool)
-  state->thread_count.Add();
+    : pool_(pool),
-  grpc_core::Thread(
+      thd_(
-      "event_engine",
+          "posix_eventengine_pool",
-      [](void* arg) {
+          [](void* th) { static_cast<ThreadPool::Thread*>(th)->ThreadFunc(); },
-        ThreadFunc(*std::unique_ptr<StatePtr>(static_cast<StatePtr*>(arg)));
+          this, nullptr, grpc_core::Thread::Options().set_tracked(false)) {
-      },
+  thd_.Start();
-      new StatePtr(state), nullptr,
-      grpc_core::Thread::Options().set_tracked(false).set_joinable(false))
-      .Start();
 }
-
+ThreadPool::Thread::~Thread() { thd_.Join(); }
-void ThreadPool::ThreadFunc(StatePtr state) {
+
-  while (state->queue.Step()) {
+void ThreadPool::Thread::ThreadFunc() {
+  pool_->ThreadFunc();
+  // Now that we have killed ourselves, we should reduce the thread count
+  grpc_core::MutexLock lock(&pool_->mu_);
+  pool_->nthreads_--;
+  // Move ourselves to dead list
+  pool_->dead_threads_.push_back(this);
+
+  if (pool_->nthreads_ == 0) {
+    if (pool_->forking_) pool_->fork_cv_.Signal();
+    if (pool_->shutdown_) pool_->shutdown_cv_.Signal();
   }
-  state->thread_count.Remove();
 }
 
-bool ThreadPool::Queue::Step() {
+void ThreadPool::ThreadFunc() {
-  grpc_core::ReleasableMutexLock lock(&mu_);
+  for (;;) {
-  // Wait until work is available or we are shutting down.
+    // Wait until work is available or we are shutting down.
-  while (state_ == State::kRunning && callbacks_.empty()) {
+    grpc_core::ReleasableMutexLock lock(&mu_);
-    // If there are too many threads waiting, then quit this thread.
+    if (!forking_ && !shutdown_ && callbacks_.empty()) {
-    // TODO(ctiller): wait some time in this case to be sure.
+      // If there are too many threads waiting, then quit this thread
-    if (threads_waiting_ >= reserve_threads_) return false;
+      if (threads_waiting_ >= reserve_threads_) {
-    threads_waiting_++;
+        break;
-    cv_.Wait(&mu_);
+      }
-    threads_waiting_--;
+      threads_waiting_++;
-  }
+      cv_.Wait(&mu_);
-  switch (state_) {
+      threads_waiting_--;
-    case State::kRunning:
+    }
+    // a fork could be initiated while the thread was waiting
+    if (forking_) return;
+    // Drain callbacks before considering shutdown to ensure all work
+    // gets completed.
+    if (!callbacks_.empty()) {
+      auto cb = std::move(callbacks_.front());
+      callbacks_.pop();
+      lock.Release();
+      cb();
+    } else if (shutdown_) {
       break;
-    case State::kShutdown:
+    }
-    case State::kForking:
-      if (!callbacks_.empty()) break;
-      return false;
   }
-  GPR_ASSERT(!callbacks_.empty());
-  auto callback = std::move(callbacks_.front());
-  callbacks_.pop();
-  lock.Release();
-  callback();
-  return true;
 }
 
 ThreadPool::ThreadPool(int reserve_threads)
-    : reserve_threads_(reserve_threads) {
+    : shutdown_(false),
-  for (int i = 0; i < reserve_threads; i++) {
+      reserve_threads_(reserve_threads),
-    StartThread(state_);
+      nthreads_(0),
-  }
+      threads_waiting_(0),
+      forking_(false) {
+  grpc_core::MutexLock lock(&mu_);
+  StartNThreadsLocked(reserve_threads_);
 }
 
-ThreadPool::~ThreadPool() { state_->queue.SetShutdown(); }
+void ThreadPool::StartNThreadsLocked(int n) {
-
+  for (int i = 0; i < n; i++) {
-void ThreadPool::Add(absl::AnyInvocable<void()> callback) {
+    nthreads_++;
-  if (state_->queue.Add(std::move(callback))) {
+    new Thread(this);
-    StartThread(state_);
   }
 }
 
-bool ThreadPool::Queue::Add(absl::AnyInvocable<void()> callback) {
+void ThreadPool::ReapThreads(std::vector<Thread*>* tlist) {
-  grpc_core::MutexLock lock(&mu_);
+  for (auto* t : *tlist) delete t;
-  // Add works to the callbacks list
+  tlist->clear();
-  callbacks_.push(std::move(callback));
-  cv_.Signal();
-  switch (state_) {
-    case State::kRunning:
-    case State::kShutdown:
-      return threads_waiting_ == 0;
-    case State::kForking:
-      return false;
-  }
-  GPR_UNREACHABLE_CODE(return false);
 }
 
-void ThreadPool::Queue::SetState(State state) {
+ThreadPool::~ThreadPool() {
   grpc_core::MutexLock lock(&mu_);
-  if (state == State::kRunning) {
+  shutdown_ = true;
-    GPR_ASSERT(state_ != State::kRunning);
-  } else {
-    GPR_ASSERT(state_ == State::kRunning);
-  }
-  state_ = state;
   cv_.SignalAll();
+  while (nthreads_ != 0) {
+    shutdown_cv_.Wait(&mu_);
+  }
+  ReapThreads(&dead_threads_);
 }
 
-void ThreadPool::ThreadCount::Add() {
+void ThreadPool::Add(absl::AnyInvocable<void()> callback) {
-  grpc_core::MutexLock lock(&mu_);
-  ++threads_;
-}
-
-void ThreadPool::ThreadCount::Remove() {
   grpc_core::MutexLock lock(&mu_);
-  --threads_;
+  // Add works to the callbacks list
-  if (threads_ == 0) {
+  callbacks_.push(std::move(callback));
+  // Store the callback for later if we are forking.
+  // TODO(hork): should we block instead?
+  if (forking_) return;
+  // Increase pool size or notify as needed
+  if (threads_waiting_ == 0) {
+    // Kick off a new thread
+    nthreads_++;
+    new Thread(this);
+  } else {
     cv_.Signal();
   }
+  // Also use this chance to harvest dead threads
+  if (!dead_threads_.empty()) {
+    ReapThreads(&dead_threads_);
+  }
 }
 
-void ThreadPool::ThreadCount::Quiesce() {
+void ThreadPool::PrepareFork() {
   grpc_core::MutexLock lock(&mu_);
-  auto last_log = absl::Now();
+  forking_ = true;
-  while (threads_ > 0) {
+  cv_.SignalAll();
-    // Wait for all threads to exit.
+  while (nthreads_ != 0) {
-    // At least once every three seconds (but no faster than once per second in
+    fork_cv_.Wait(&mu_);
-    // the event of spurious wakeups) log a message indicating we're waiting to
-    // fork.
-    cv_.WaitWithTimeout(&mu_, absl::Seconds(3));
-    if (threads_ > 0 && absl::Now() - last_log > absl::Seconds(1)) {
-      gpr_log(GPR_ERROR, "Waiting for thread pool to idle before forking");
-      last_log = absl::Now();
-    }
   }
+  ReapThreads(&dead_threads_);
 }
 
-void ThreadPool::PrepareFork() {
+void ThreadPool::PostforkParent() {
-  state_->queue.SetForking();
+  grpc_core::MutexLock lock(&mu_);
-  state_->thread_count.Quiesce();
+  forking_ = false;
+  StartNThreadsLocked(reserve_threads_);
 }
 
-void ThreadPool::PostforkParent() { Postfork(); }
+void ThreadPool::PostforkChild() {
-
+  grpc_core::MutexLock lock(&mu_);
-void ThreadPool::PostforkChild() { Postfork(); }
+  forking_ = false;
-
+  StartNThreadsLocked(reserve_threads_);
-void ThreadPool::Postfork() {
-  state_->queue.Reset();
-  for (int i = 0; i < reserve_threads_; i++) {
-    StartThread(state_);
-  }
 }
 
 }  // namespace experimental

src/core/lib/event_engine/thread_pool.h

@@ -21,14 +21,15 @@
 
 #include <grpc/support/port_platform.h>
 
-#include <memory>
 #include <queue>
+#include <vector>
 
 #include "absl/base/thread_annotations.h"
 #include "absl/functional/any_invocable.h"
 
 #include "src/core/lib/event_engine/forkable.h"
 #include "src/core/lib/gprpp/sync.h"
+#include "src/core/lib/gprpp/thd.h"
 
 namespace grpc_event_engine {
 namespace experimental {
@@ -46,57 +47,32 @@ class ThreadPool final : public grpc_event_engine::experimental::Forkable {
   void PostforkChild() override;
 
  private:
-  class Queue {
+  class Thread {
    public:
-    explicit Queue(int reserve_threads) : reserve_threads_(reserve_threads) {}
+    explicit Thread(ThreadPool* pool);
-    bool Step();
+    ~Thread();
-    void SetShutdown() { SetState(State::kShutdown); }
-    void SetForking() { SetState(State::kForking); }
-    // Add a callback to the queue.
-    // Return true if we should also spin up a new thread.
-    bool Add(absl::AnyInvocable<void()> callback);
-    void Reset() { SetState(State::kRunning); }
 
    private:
-    enum class State { kRunning, kShutdown, kForking };
+    ThreadPool* pool_;
-
+    grpc_core::Thread thd_;
-    void SetState(State state);
+    void ThreadFunc();
-
-    grpc_core::Mutex mu_;
-    grpc_core::CondVar cv_;
-    std::queue<absl::AnyInvocable<void()>> callbacks_ ABSL_GUARDED_BY(mu_);
-    int threads_waiting_ ABSL_GUARDED_BY(mu_) = 0;
-    const int reserve_threads_;
-    State state_ ABSL_GUARDED_BY(mu_) = State::kRunning;
   };
 
-  class ThreadCount {
+  void ThreadFunc();
-   public:
+  void StartNThreadsLocked(int n) ABSL_EXCLUSIVE_LOCKS_REQUIRED(&mu_);
-    void Add();
+  static void ReapThreads(std::vector<Thread*>* tlist);
-    void Remove();
+
-    // Block until all threads have stopped.
+  grpc_core::Mutex mu_;
-    void Quiesce();
+  grpc_core::CondVar cv_;
-
+  grpc_core::CondVar shutdown_cv_;
-   private:
+  grpc_core::CondVar fork_cv_;
-    grpc_core::Mutex mu_;
+  bool shutdown_;
-    grpc_core::CondVar cv_;
+  std::queue<absl::AnyInvocable<void()>> callbacks_;
-    int threads_ ABSL_GUARDED_BY(mu_) = 0;
+  int reserve_threads_;
-  };
+  int nthreads_;
-
+  int threads_waiting_;
-  struct State {
+  std::vector<Thread*> dead_threads_;
-    explicit State(int reserve_threads) : queue(reserve_threads) {}
+  bool forking_;
-    Queue queue;
-    ThreadCount thread_count;
-  };
-
-  using StatePtr = std::shared_ptr<State>;
-
-  static void ThreadFunc(StatePtr state);
-  static void StartThread(StatePtr state);
-  void Postfork();
-
-  const int reserve_threads_;
-  const StatePtr state_ = std::make_shared<State>(reserve_threads_);
 };
 
 }  // namespace experimental

test/core/event_engine/BUILD

@@ -48,19 +48,6 @@ grpc_cc_test(
     ],
 )
 
-grpc_cc_test(
-    name = "thread_pool_test",
-    srcs = ["thread_pool_test.cc"],
-    external_deps = [
-        "absl/synchronization",
-        "gtest",
-    ],
-    deps = [
-        "//:event_engine_thread_pool",
-        "//:gpr",
-    ],
-)
-
 grpc_cc_test(
     name = "endpoint_config_test",
     srcs = ["endpoint_config_test.cc"],

test/core/event_engine/thread_pool_test.cc

@@ -1,98 +0,0 @@
-// Copyright 2022 The gRPC Authors
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//     http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-#include "src/core/lib/event_engine/thread_pool.h"
-
-#include <atomic>
-#include <chrono>
-#include <thread>
-
-#include <gtest/gtest.h>
-
-#include "absl/synchronization/notification.h"
-#include "gtest/gtest.h"
-
-#include <grpc/support/log.h>
-
-namespace grpc_event_engine {
-namespace experimental {
-
-TEST(ThreadPoolTest, CanRunClosure) {
-  ThreadPool p(1);
-  absl::Notification n;
-  p.Add([&n] { n.Notify(); });
-  n.WaitForNotification();
-}
-
-TEST(ThreadPoolTest, CanDestroyInsideClosure) {
-  auto p = std::make_shared<ThreadPool>(1);
-  p->Add([p]() { std::this_thread::sleep_for(std::chrono::seconds(1)); });
-}
-
-TEST(ThreadPoolTest, CanSurviveFork) {
-  ThreadPool p(1);
-  absl::Notification n;
-  gpr_log(GPR_INFO, "add callback 1");
-  p.Add([&n, &p] {
-    std::this_thread::sleep_for(std::chrono::seconds(1));
-    gpr_log(GPR_INFO, "add callback 2");
-    p.Add([&n] {
-      std::this_thread::sleep_for(std::chrono::seconds(1));
-      gpr_log(GPR_INFO, "notify");
-      n.Notify();
-    });
-  });
-  gpr_log(GPR_INFO, "prepare fork");
-  p.PrepareFork();
-  gpr_log(GPR_INFO, "wait for notification");
-  n.WaitForNotification();
-  gpr_log(GPR_INFO, "postfork child");
-  p.PostforkChild();
-  absl::Notification n2;
-  gpr_log(GPR_INFO, "add callback 3");
-  p.Add([&n2] {
-    gpr_log(GPR_INFO, "notify");
-    n2.Notify();
-  });
-  gpr_log(GPR_INFO, "wait for notification");
-  n2.WaitForNotification();
-}
-
-void ScheduleSelf(ThreadPool* p) {
-  p->Add([p] { ScheduleSelf(p); });
-}
-
-TEST(ThreadPoolDeathTest, CanDetectStucknessAtFork) {
-  ASSERT_DEATH_IF_SUPPORTED(
-      [] {
-        gpr_set_log_verbosity(GPR_LOG_SEVERITY_ERROR);
-        ThreadPool p(1);
-        ScheduleSelf(&p);
-        std::thread terminator([] {
-          std::this_thread::sleep_for(std::chrono::seconds(10));
-          abort();
-        });
-        p.PrepareFork();
-      }(),
-      "Waiting for thread pool to idle before forking");
-}
-
-}  // namespace experimental
-}  // namespace grpc_event_engine
-
-int main(int argc, char** argv) {
-  gpr_log_verbosity_init();
-  ::testing::InitGoogleTest(&argc, argv);
-  return RUN_ALL_TESTS();
-}

tools/run_tests/generated/tests.json

@@ -7205,30 +7205,6 @@
     ],
     "uses_polling": true
   },
-  {
-    "args": [],
-    "benchmark": false,
-    "ci_platforms": [
-      "linux",
-      "mac",
-      "posix",
-      "windows"
-    ],
-    "cpu_cost": 1.0,
-    "exclude_configs": [],
-    "exclude_iomgrs": [],
-    "flaky": false,
-    "gtest": true,
-    "language": "c++",
-    "name": "thread_pool_test",
-    "platforms": [
-      "linux",
-      "mac",
-      "posix",
-      "windows"
-    ],
-    "uses_polling": true
-  },
   {
     "args": [],
     "benchmark": false,