diff --git a/include/grpc++/server_builder.h b/include/grpc++/server_builder.h index a0ba4fc8d28..a56f81dc4b3 100644 --- a/include/grpc++/server_builder.h +++ b/include/grpc++/server_builder.h @@ -197,8 +197,8 @@ class ServerBuilder { SyncServerSettings() : num_cqs(gpr_cpu_num_cores()), min_pollers(1), - max_pollers(INT_MAX), - cq_timeout_msec(1000) {} + max_pollers(2), + cq_timeout_msec(10000) {} // Number of server completion queues to create to listen to incoming RPCs. int num_cqs; diff --git a/src/cpp/server/server_cc.cc b/src/cpp/server/server_cc.cc index 62ded0d239f..423f347acd2 100644 --- a/src/cpp/server/server_cc.cc +++ b/src/cpp/server/server_cc.cc @@ -328,14 +328,18 @@ class Server::SyncRequestThreadManager : public ThreadManager { } } - void ShutdownAndDrainCompletionQueue() { + void Shutdown() override { server_cq_->Shutdown(); + ThreadManager::Shutdown(); + } + void Wait() override { + ThreadManager::Wait(); // Drain any pending items from the queue void* tag; bool ok; while (server_cq_->Next(&tag, &ok)) { - // Nothing to be done here + // Do nothing } } @@ -415,7 +419,7 @@ Server::~Server() { } else if (!started_) { // Shutdown the completion queues for (auto it = sync_req_mgrs_.begin(); it != sync_req_mgrs_.end(); it++) { - (*it)->ShutdownAndDrainCompletionQueue(); + (*it)->Shutdown(); } } } @@ -579,7 +583,6 @@ void Server::ShutdownInternal(gpr_timespec deadline) { // Wait for threads in all ThreadManagers to terminate for (auto it = sync_req_mgrs_.begin(); it != sync_req_mgrs_.end(); it++) { (*it)->Wait(); - (*it)->ShutdownAndDrainCompletionQueue(); } // Drain the shutdown queue (if the previous call to AsyncNext() timed out diff --git a/src/cpp/thread_manager/thread_manager.cc b/src/cpp/thread_manager/thread_manager.cc index 1450d009e4f..ebcc4dd3787 100644 --- a/src/cpp/thread_manager/thread_manager.cc +++ b/src/cpp/thread_manager/thread_manager.cc @@ -98,80 +98,78 @@ void ThreadManager::MarkAsCompleted(WorkerThread* thd) { } void ThreadManager::CleanupCompletedThreads() { - std::unique_lock lock(list_mu_); - for (auto thd = completed_threads_.begin(); thd != completed_threads_.end(); - thd = completed_threads_.erase(thd)) { - delete *thd; + std::list completed_threads; + { + // swap out the completed threads list: allows other threads to clean up + // more quickly + std::unique_lock lock(list_mu_); + completed_threads.swap(completed_threads_); } + for (auto thd : completed_threads) delete thd; } void ThreadManager::Initialize() { - for (int i = 0; i < min_pollers_; i++) { - MaybeCreatePoller(); - } -} - -// If the number of pollers (i.e threads currently blocked in PollForWork()) is -// less than max threshold (i.e max_pollers_) and the total number of threads is -// below the maximum threshold, we can let the current thread continue as poller -bool ThreadManager::MaybeContinueAsPoller() { - std::unique_lock lock(mu_); - if (shutdown_ || num_pollers_ > max_pollers_) { - return false; + { + std::unique_lock lock(mu_); + num_pollers_ = min_pollers_; + num_threads_ = min_pollers_; } - num_pollers_++; - return true; -} - -// Create a new poller if the current number of pollers i.e num_pollers_ (i.e -// threads currently blocked in PollForWork()) is below the threshold (i.e -// min_pollers_) and the total number of threads is below the maximum threshold -void ThreadManager::MaybeCreatePoller() { - std::unique_lock lock(mu_); - if (!shutdown_ && num_pollers_ < min_pollers_) { - num_pollers_++; - num_threads_++; - + for (int i = 0; i < min_pollers_; i++) { // Create a new thread (which ends up calling the MainWorkLoop() function new WorkerThread(this); } } void ThreadManager::MainWorkLoop() { - void* tag; - bool ok; - - /* - 1. Poll for work (i.e PollForWork()) - 2. After returning from PollForWork, reduce the number of pollers by 1. If - PollForWork() returned a TIMEOUT, then it may indicate that we have more - polling threads than needed. Check if the number of pollers is greater - than min_pollers and if so, terminate the thread. - 3. Since we are short of one poller now, see if a new poller has to be - created (i.e see MaybeCreatePoller() for more details) - 4. Do the actual work (DoWork()) - 5. After doing the work, see it this thread can resume polling work (i.e - see MaybeContinueAsPoller() for more details) */ - do { + while (true) { + void* tag; + bool ok; WorkStatus work_status = PollForWork(&tag, &ok); - { - std::unique_lock lock(mu_); - num_pollers_--; - - if (work_status == TIMEOUT && num_pollers_ > min_pollers_) { + std::unique_lock lock(mu_); + // Reduce the number of pollers by 1 and check what happened with the poll + num_pollers_--; + bool done = false; + switch (work_status) { + case TIMEOUT: + // If we timed out and we have more pollers than we need (or we are + // shutdown), finish this thread + if (shutdown_ || num_pollers_ > max_pollers_) done = true; + break; + case SHUTDOWN: + // If the thread manager is shutdown, finish this thread + done = true; + break; + case WORK_FOUND: + // If we got work and there are now insufficient pollers, start a new + // one + if (!shutdown_ && num_pollers_ < min_pollers_) { + num_pollers_++; + num_threads_++; + // Drop lock before spawning thread to avoid contention + lock.unlock(); + new WorkerThread(this); + } else { + // Drop lock for consistency with above branch + lock.unlock(); + } + // Lock is always released at this point - do the application work + DoWork(tag, ok); + // Take the lock again to check post conditions + lock.lock(); + // If we're shutdown, we should finish at this point. + // If not, there's a chance that we'll exceed the max poller count: that + // is explicitly ok - we'll decrease after one poll timeout, and prevent + // some thrashing starting up and shutting down threads + if (shutdown_) done = true; break; - } - } - - // Note that MaybeCreatePoller does check for shutdown and creates a new - // thread only if ThreadManager is not shutdown - if (work_status == WORK_FOUND) { - MaybeCreatePoller(); - DoWork(tag, ok); } - } while (MaybeContinueAsPoller()); + // If we decided to finish the thread, break out of the while loop + if (done) break; + // ... otherwise increase poller count and continue + num_pollers_++; + }; CleanupCompletedThreads(); diff --git a/src/cpp/thread_manager/thread_manager.h b/src/cpp/thread_manager/thread_manager.h index 9c0569c62c1..d1050f6dede 100644 --- a/src/cpp/thread_manager/thread_manager.h +++ b/src/cpp/thread_manager/thread_manager.h @@ -89,14 +89,14 @@ class ThreadManager { // Mark the ThreadManager as shutdown and begin draining the work. This is a // non-blocking call and the caller should call Wait(), a blocking call which // returns only once the shutdown is complete - void Shutdown(); + virtual void Shutdown(); // Has Shutdown() been called bool IsShutdown(); // A blocking call that returns only after the ThreadManager has shutdown and // all the threads have drained all the outstanding work - void Wait(); + virtual void Wait(); private: // Helper wrapper class around std::thread. This takes a ThreadManager object @@ -122,14 +122,6 @@ class ThreadManager { // The main funtion in ThreadManager void MainWorkLoop(); - // Create a new poller if the number of current pollers is less than the - // minimum number of pollers needed (i.e min_pollers). - void MaybeCreatePoller(); - - // Returns true if the current thread can resume as a poller. i.e if the - // current number of pollers is less than the max_pollers. - bool MaybeContinueAsPoller(); - void MarkAsCompleted(WorkerThread* thd); void CleanupCompletedThreads();