diff options
Diffstat (limited to 'src/cpp/server/dynamic_thread_pool.cc')
-rw-r--r-- | src/cpp/server/dynamic_thread_pool.cc | 131 |
1 files changed, 131 insertions, 0 deletions
diff --git a/src/cpp/server/dynamic_thread_pool.cc b/src/cpp/server/dynamic_thread_pool.cc new file mode 100644 index 0000000000..f58d0420df --- /dev/null +++ b/src/cpp/server/dynamic_thread_pool.cc @@ -0,0 +1,131 @@ +/* + * + * Copyright 2015, Google Inc. + * All rights reserved. + * + * Redistribution and use in source and binary forms, with or without + * modification, are permitted provided that the following conditions are + * met: + * + * * Redistributions of source code must retain the above copyright + * notice, this list of conditions and the following disclaimer. + * * Redistributions in binary form must reproduce the above + * copyright notice, this list of conditions and the following disclaimer + * in the documentation and/or other materials provided with the + * distribution. + * * Neither the name of Google Inc. nor the names of its + * contributors may be used to endorse or promote products derived from + * this software without specific prior written permission. + * + * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS + * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT + * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR + * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT + * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, + * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT + * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, + * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY + * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT + * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE + * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. + * + */ + +#include <grpc++/impl/sync.h> +#include <grpc++/impl/thd.h> +#include <grpc++/dynamic_thread_pool.h> + +namespace grpc { +DynamicThreadPool::DynamicThread::DynamicThread(DynamicThreadPool *pool): + pool_(pool), + thd_(new grpc::thread(&DynamicThreadPool::DynamicThread::ThreadFunc, this)) { +} +DynamicThreadPool::DynamicThread::~DynamicThread() { + thd_->join(); + thd_.reset(); +} + +void DynamicThreadPool::DynamicThread::ThreadFunc() { + pool_->ThreadFunc(); + // Now that we have killed ourselves, we should reduce the thread count + grpc::unique_lock<grpc::mutex> lock(pool_->mu_); + pool_->nthreads_--; + // Move ourselves to dead list + pool_->dead_threads_.push_back(this); + + if ((pool_->shutdown_) && (pool_->nthreads_ == 0)) { + pool_->shutdown_cv_.notify_one(); + } +} + +void DynamicThreadPool::ThreadFunc() { + for (;;) { + // Wait until work is available or we are shutting down. + grpc::unique_lock<grpc::mutex> lock(mu_); + if (!shutdown_ && callbacks_.empty()) { + // If there are too many threads waiting, then quit this thread + if (threads_waiting_ >= reserve_threads_) { + break; + } + threads_waiting_++; + cv_.wait(lock); + threads_waiting_--; + } + // 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_) { + break; + } + } +} + +DynamicThreadPool::DynamicThreadPool(int reserve_threads) : + shutdown_(false), reserve_threads_(reserve_threads), nthreads_(0), + threads_waiting_(0) { + for (int i = 0; i < reserve_threads_; i++) { + grpc::lock_guard<grpc::mutex> lock(mu_); + nthreads_++; + new DynamicThread(this); + } +} + +void DynamicThreadPool::ReapThreads(std::list<DynamicThread*>* tlist) { + for (auto t = tlist->begin(); t != tlist->end(); t = tlist->erase(t)) { + delete *t; + } +} + +DynamicThreadPool::~DynamicThreadPool() { + grpc::unique_lock<grpc::mutex> lock(mu_); + shutdown_ = true; + cv_.notify_all(); + while (nthreads_ != 0) { + shutdown_cv_.wait(lock); + } + ReapThreads(&dead_threads_); +} + +void DynamicThreadPool::Add(const std::function<void()>& callback) { + grpc::lock_guard<grpc::mutex> lock(mu_); + // Add works to the callbacks list + callbacks_.push(callback); + // Increase pool size or notify as needed + if (threads_waiting_ == 0) { + // Kick off a new thread + nthreads_++; + new DynamicThread(this); + } else { + cv_.notify_one(); + } + // Also use this chance to harvest dead threads + if (!dead_threads_.empty()) { + ReapThreads(&dead_threads_); + } +} + +} // namespace grpc |