Initial thread manager fixes

include/grpc++/server_builder.h

@@ -197,8 +197,8 @@ class ServerBuilder {
     SyncServerSettings()
         : num_cqs(1),
           min_pollers(1),
-          max_pollers(INT_MAX),
+          max_pollers(2),
-          cq_timeout_msec(1000) {}
+          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,15 +328,9 @@ class Server::SyncRequestThreadManager : public ThreadManager {
     }
   }
 
-  void ShutdownAndDrainCompletionQueue() {
+  void Shutdown() override {
     server_cq_->Shutdown();
-
+ThreadManager::Shutdown();
-    // Drain any pending items from the queue
-    void* tag;
-    bool ok;
-    while (server_cq_->Next(&tag, &ok)) {
-      // Nothing to be done here
-    }
   }
 
   void Start() {
@@ -415,7 +409,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 +573,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,11 +98,12 @@ void ThreadManager::MarkAsCompleted(WorkerThread* thd) {
 }
 
 void ThreadManager::CleanupCompletedThreads() {
-  std::unique_lock<std::mutex> lock(list_mu_);
+  std::list<WorkerThread*> completed_threads;
-  for (auto thd = completed_threads_.begin(); thd != completed_threads_.end();
+  {
-       thd = completed_threads_.erase(thd)) {
+    std::unique_lock<std::mutex> lock(list_mu_);
-    delete *thd;
+    completed_threads.swap(completed_threads_);
   }
+  for (auto thd : completed_threads) delete thd;
 }
 
 void ThreadManager::Initialize() {
@@ -114,9 +115,10 @@ void ThreadManager::Initialize() {
 // 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() {
+bool ThreadManager::MaybeContinueAsPoller(bool work_found) {
   std::unique_lock<std::mutex> lock(mu_);
-  if (shutdown_ || num_pollers_ > max_pollers_) {
+  gpr_log(GPR_DEBUG, "s=%d wf=%d np=%d mp=%d", shutdown_, work_found, num_pollers_, max_pollers_);
+  if (shutdown_ || (!work_found && num_pollers_ > max_pollers_)) {
     return false;
   }
 
@@ -133,6 +135,8 @@ void ThreadManager::MaybeCreatePoller() {
     num_pollers_++;
     num_threads_++;
 
+    lock.unlock();
+
     // Create a new thread (which ends up calling the MainWorkLoop() function
     new WorkerThread(this);
   }
@@ -153,25 +157,36 @@ void ThreadManager::MainWorkLoop() {
    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;
-    WorkStatus work_status = PollForWork(&tag, &ok);
+  while (true) {
+  bool done = false;
+    work_status = PollForWork(&tag, &ok);
 
-    {
+    std::unique_lock<std::mutex> lock(mu_);
-      std::unique_lock<std::mutex> lock(mu_);
+    num_pollers_--;
-      num_pollers_--;
+    gpr_log(GPR_DEBUG, "%p: work_status:%d num_pollers:%d min_pollers:%d max_pollers:%d num_threads:%d shutdown:%d", this, work_status, num_pollers_, min_pollers_, max_pollers_, num_threads_, shutdown_);
-
+    switch (work_status) {
-      if (work_status == TIMEOUT && num_pollers_ > min_pollers_) {
+     case TIMEOUT:
-        break;
+      if (shutdown_ || num_pollers_ >= max_pollers_) done = true;
+      break;
+     case SHUTDOWN: done = true; break;
+     case WORK_FOUND:
+      if (!shutdown_ && num_pollers_ < min_pollers_) {
+        num_pollers_++;
+        num_threads_++;
+        lock.unlock();
+        new WorkerThread(this);
+      } else {
+        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();
       DoWork(tag, ok);
+      lock.lock();
+      if (shutdown_) done = true;
+      break;
     }
-  } while (MaybeContinueAsPoller());
+if (done) break;
+    num_pollers_++;
+  };
 
   CleanupCompletedThreads();
 

src/cpp/thread_manager/thread_manager.h

@@ -89,7 +89,7 @@ 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();
@@ -128,7 +128,7 @@ class ThreadManager {
 
   // 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();
+  bool MaybeContinueAsPoller(bool work_found);
 
   void MarkAsCompleted(WorkerThread* thd);
   void CleanupCompletedThreads();