Merge pull request #10711 from ctiller/tm1

Thread manager fixes/refactoring

include/grpc++/server_builder.h

@@ -195,10 +195,7 @@ class ServerBuilder {
 
   struct SyncServerSettings {
     SyncServerSettings()
-        : num_cqs(1),
-          min_pollers(1),
-          max_pollers(INT_MAX),
-          cq_timeout_msec(1000) {}
+        : num_cqs(1), min_pollers(1), max_pollers(2), cq_timeout_msec(10000) {}
 
     // Number of server completion queues to create to listen to incoming RPCs.
     int num_cqs;

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

src/cpp/thread_manager/thread_manager.cc

@@ -98,80 +98,78 @@ void ThreadManager::MarkAsCompleted(WorkerThread* thd) {
 }
 
 void ThreadManager::CleanupCompletedThreads() {
+  std::list<WorkerThread*> completed_threads;
+  {
+    // swap out the completed threads list: allows other threads to clean up
+    // more quickly
     std::unique_lock<std::mutex> lock(list_mu_);
-  for (auto thd = completed_threads_.begin(); thd != completed_threads_.end();
-       thd = completed_threads_.erase(thd)) {
-    delete *thd;
+    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<std::mutex> lock(mu_);
-  if (shutdown_ || num_pollers_ > max_pollers_) {
-    return false;
-  }
-
-  num_pollers_++;
-  return true;
+    num_pollers_ = min_pollers_;
+    num_threads_ = min_pollers_;
   }
 
-// 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<std::mutex> 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() {
+  while (true) {
     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 {
     WorkStatus work_status = PollForWork(&tag, &ok);
 
-    {
     std::unique_lock<std::mutex> lock(mu_);
+    // Reduce the number of pollers by 1 and check what happened with the poll
     num_pollers_--;
-
-      if (work_status == TIMEOUT && num_pollers_ > min_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();
         }
-    }
-
-    // Note that MaybeCreatePoller does check for shutdown and creates a new
-    // thread only if ThreadManager is not shutdown
-    if (work_status == WORK_FOUND) {
-      MaybeCreatePoller();
+        // 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 (shutdown_) done = true;
+        break;
     }
-  } while (MaybeContinueAsPoller());
+    // If we decided to finish the thread, break out of the while loop
+    if (done) break;
+    // ... otherwise increase poller count and continue
+    // 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
+    num_pollers_++;
+  };
 
   CleanupCompletedThreads();
 

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();