tsan failures

src/cpp/rpcmanager/grpc_rpc_manager.cc

@@ -63,8 +63,10 @@ GrpcRpcManager::GrpcRpcManager(int min_pollers, int max_pollers)
       num_threads_(0) {}
 
 GrpcRpcManager::~GrpcRpcManager() {
+  {
     std::unique_lock<grpc::mutex> lock(mu_);
     GPR_ASSERT(num_threads_ == 0);
+  }
 
   CleanupCompletedThreads();
 }
@@ -87,10 +89,18 @@ bool GrpcRpcManager::IsShutdown() {
 }
 
 void GrpcRpcManager::MarkAsCompleted(GrpcRpcManagerThread* thd) {
-  std::unique_lock<grpc::mutex> lock(list_mu_);
+  {
+    std::unique_lock<grpc::mutex> list_lock(list_mu_);
     completed_threads_.push_back(thd);
   }
 
+  grpc::unique_lock<grpc::mutex> lock(mu_);
+  num_threads_--;
+  if (num_threads_ == 0) {
+    shutdown_cv_.notify_one();
+  }
+}
+
 void GrpcRpcManager::CleanupCompletedThreads() {
   std::unique_lock<grpc::mutex> lock(list_mu_);
   for (auto thd = completed_threads_.begin(); thd != completed_threads_.end();
@@ -169,17 +179,10 @@ void GrpcRpcManager::MainWorkLoop() {
     }
   } while (MaybeContinueAsPoller());
 
-  // If we are here, either GrpcRpcManager is shutting down or it already has
-  // enough threads. In both cases, current thread can be terminated
-  {
-    grpc::unique_lock<grpc::mutex> lock(mu_);
-    num_threads_--;
-    if (num_threads_ == 0) {
-      shutdown_cv_.notify_one();
-    }
-  }
-
   CleanupCompletedThreads();
+
+  // If we are here, either GrpcRpcManager is shutting down or it already has
+  // enough threads.
 }
 
 }  // namespace grpc