diff options
-rw-r--r-- | src/cpp/server/thread_pool.cc | 41 | ||||
-rw-r--r-- | src/cpp/server/thread_pool.h | 2 |
2 files changed, 23 insertions, 20 deletions
diff --git a/src/cpp/server/thread_pool.cc b/src/cpp/server/thread_pool.cc index d3013b806c..3eb172a3a6 100644 --- a/src/cpp/server/thread_pool.cc +++ b/src/cpp/server/thread_pool.cc @@ -35,28 +35,29 @@ namespace grpc { +void ThreadPool::ThreadFunc() { + for (;;) { + // Wait until work is available or we are shutting down. + std::unique_lock<std::mutex> lock(mu_); + if (!shutdown_ && callbacks_.empty()) { + cv_.wait(lock); + } + // Drain callbacks before considering shutdown to ensure all work + // gets completed. + if (!callbacks_.empty()) { + auto cb = callbacks_.front(); + callbacks_.pop(); + lock.unlock(); + cb(); + } else if (shutdown_) { + return; + } + } +} + ThreadPool::ThreadPool(int num_threads) : shutdown_(false) { for (int i = 0; i < num_threads; i++) { - threads_.push_back(std::thread([this]() { - for (;;) { - // Wait until work is available or we are shutting down. - auto have_work = [this]() { return shutdown_ || !callbacks_.empty(); }; - std::unique_lock<std::mutex> lock(mu_); - if (!have_work()) { - cv_.wait(lock, have_work); - } - // Drain callbacks before considering shutdown to ensure all work - // gets completed. - if (!callbacks_.empty()) { - auto cb = callbacks_.front(); - callbacks_.pop(); - lock.unlock(); - cb(); - } else if (shutdown_) { - return; - } - } - })); + threads_.push_back(std::thread(&ThreadPool::ThreadFunc, this)); } } diff --git a/src/cpp/server/thread_pool.h b/src/cpp/server/thread_pool.h index 6225d82a0b..41e2009ff1 100644 --- a/src/cpp/server/thread_pool.h +++ b/src/cpp/server/thread_pool.h @@ -58,6 +58,8 @@ class ThreadPool GRPC_FINAL : public ThreadPoolInterface { bool shutdown_; std::queue<std::function<void()>> callbacks_; std::vector<std::thread> threads_; + + void ThreadFunc(); }; } // namespace grpc |