third_party/grpc/src/cpp/thread_manager/thread_manager.cc - bazel - Git at Google

 /*
  *
  * Copyright 2016 gRPC authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
  * You may obtain a copy of the License at
  *
  *     http://www.apache.org/licenses/LICENSE-2.0
  *
  * Unless required by applicable law or agreed to in writing, software
  * distributed under the License is distributed on an "AS IS" BASIS,
  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  * See the License for the specific language governing permissions and
  * limitations under the License.
  *
  */

 #include "src/cpp/thread_manager/thread_manager.h"

 #include <climits>
 #include <mutex>

 #include <grpc/support/log.h>
 #include "src/core/lib/gprpp/thd.h"
 #include "src/core/lib/iomgr/exec_ctx.h"

 namespace grpc {

 ThreadManager::WorkerThread::WorkerThread(ThreadManager* thd_mgr)
     : thd_mgr_(thd_mgr) {
   // Make thread creation exclusive with respect to its join happening in
   // ~WorkerThread().
   thd_ = grpc_core::Thread(
       "grpcpp_sync_server",
       [](void* th) { static_cast<ThreadManager::WorkerThread*>(th)->Run(); },
       this);
   thd_.Start();
 }

 void ThreadManager::WorkerThread::Run() {
   thd_mgr_->MainWorkLoop();
   thd_mgr_->MarkAsCompleted(this);
 }

 ThreadManager::WorkerThread::~WorkerThread() {
   // Don't join until the thread is fully constructed.
   thd_.Join();
 }

 ThreadManager::ThreadManager(const char* name,
                              grpc_resource_quota* resource_quota,
                              int min_pollers, int max_pollers)
     : shutdown_(false),
       num_pollers_(0),
       min_pollers_(min_pollers),
       max_pollers_(max_pollers == -1 ? INT_MAX : max_pollers),
       num_threads_(0),
       max_active_threads_sofar_(0) {
   resource_user_ = grpc_resource_user_create(resource_quota, name);
 }

 ThreadManager::~ThreadManager() {
   {
     std::lock_guard<std::mutex> lock(mu_);
     GPR_ASSERT(num_threads_ == 0);
   }

   grpc_core::ExecCtx exec_ctx;  // grpc_resource_user_unref needs an exec_ctx
   grpc_resource_user_unref(resource_user_);
   CleanupCompletedThreads();
 }

 void ThreadManager::Wait() {
   std::unique_lock<std::mutex> lock(mu_);
   while (num_threads_ != 0) {
     shutdown_cv_.wait(lock);
   }
 }

 void ThreadManager::Shutdown() {
   std::lock_guard<std::mutex> lock(mu_);
   shutdown_ = true;
 }

 bool ThreadManager::IsShutdown() {
   std::lock_guard<std::mutex> lock(mu_);
   return shutdown_;
 }

 int ThreadManager::GetMaxActiveThreadsSoFar() {
   std::lock_guard<std::mutex> list_lock(list_mu_);
   return max_active_threads_sofar_;
 }

 void ThreadManager::MarkAsCompleted(WorkerThread* thd) {
   {
     std::lock_guard<std::mutex> list_lock(list_mu_);
     completed_threads_.push_back(thd);
   }

   {
     std::lock_guard<std::mutex> lock(mu_);
     num_threads_--;
     if (num_threads_ == 0) {
       shutdown_cv_.notify_one();
     }
   }

   // Give a thread back to the resource quota
   grpc_resource_user_free_threads(resource_user_, 1);
 }

 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_);
     completed_threads.swap(completed_threads_);
   }
   for (auto thd : completed_threads) delete thd;
 }

 void ThreadManager::Initialize() {
   if (!grpc_resource_user_allocate_threads(resource_user_, min_pollers_)) {
     gpr_log(GPR_ERROR,
             "No thread quota available to even create the minimum required "
             "polling threads (i.e %d). Unable to start the thread manager",
             min_pollers_);
     abort();
   }

   {
     std::unique_lock<std::mutex> lock(mu_);
     num_pollers_ = min_pollers_;
     num_threads_ = min_pollers_;
     max_active_threads_sofar_ = min_pollers_;
   }

   for (int i = 0; i < min_pollers_; i++) {
     new WorkerThread(this);
   }
 }

 void ThreadManager::MainWorkLoop() {
   while (true) {
     void* tag;
     bool ok;
     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_--;
     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 and there is
         // quota available to create a new thread, start a new poller thread
         bool resource_exhausted = false;
         if (!shutdown_ && num_pollers_ < min_pollers_) {
           if (grpc_resource_user_allocate_threads(resource_user_, 1)) {
             // We can allocate a new poller thread
             num_pollers_++;
             num_threads_++;
             if (num_threads_ > max_active_threads_sofar_) {
               max_active_threads_sofar_ = num_threads_;
             }
             // Drop lock before spawning thread to avoid contention
             lock.unlock();
             new WorkerThread(this);
           } else if (num_pollers_ > 0) {
             // There is still at least some thread polling, so we can go on
             // even though we are below the number of pollers that we would
             // like to have (min_pollers_)
             lock.unlock();
           } else {
             // There are no pollers to spare and we couldn't allocate
             // a new thread, so resources are exhausted!
             lock.unlock();
             resource_exhausted = true;
           }
         } else {
           // There are a sufficient number of pollers available so we can do
           // the work and continue polling with our existing poller threads
           lock.unlock();
         }
         // Lock is always released at this point - do the application work
         // or return resource exhausted if there is new work but we couldn't
         // get a thread in which to do it.
         DoWork(tag, ok, !resource_exhausted);
         // 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;
     }
     // If we decided to finish the thread, break out of the while loop
     if (done) break;

     // Otherwise go back to polling as long as it doesn't exceed max_pollers_
     //
     // **WARNING**:
     // There is a possibility of threads thrashing here (i.e excessive thread
     // shutdowns and creations than the ideal case). This happens if max_poller_
     // count is small and the rate of incoming requests is also small. In such
     // scenarios we can possibly configure max_pollers_ to a higher value and/or
     // increase the cq timeout.
     //
     // However, not doing this check here and unconditionally incrementing
     // num_pollers (and hoping that the system will eventually settle down) has
     // far worse consequences i.e huge number of threads getting created to the
     // point of thread-exhaustion. For example: if the incoming request rate is
     // very high, all the polling threads will return very quickly from
     // PollForWork() with WORK_FOUND. They all briefly decrement num_pollers_
     // counter thereby possibly - and briefly - making it go below min_pollers;
     // This will most likely result in the creation of a new poller since
     // num_pollers_ dipped below min_pollers_.
     //
     // Now, If we didn't do the max_poller_ check here, all these threads will
     // go back to doing PollForWork() and the whole cycle repeats (with a new
     // thread being added in each cycle). Once the total number of threads in
     // the system crosses a certain threshold (around ~1500), there is heavy
     // contention on mutexes (the mu_ here or the mutexes in gRPC core like the
     // pollset mutex) that makes DoWork() take longer to finish thereby causing
     // new poller threads to be created even faster. This results in a thread
     // avalanche.
     if (num_pollers_ < max_pollers_) {
       num_pollers_++;
     } else {
       break;
     }
   };

   // This thread is exiting. Do some cleanup work i.e delete already completed
   // worker threads
   CleanupCompletedThreads();

   // If we are here, either ThreadManager is shutting down or it already has
   // enough threads.
 }

 }  // namespace grpc
	/*
	*
	* Copyright 2016 gRPC authors.
	*
	* Licensed under the Apache License, Version 2.0 (the "License");
	* you may not use this file except in compliance with the License.
	* You may obtain a copy of the License at
	*
	* http://www.apache.org/licenses/LICENSE-2.0
	*
	* Unless required by applicable law or agreed to in writing, software
	* distributed under the License is distributed on an "AS IS" BASIS,
	* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
	* See the License for the specific language governing permissions and
	* limitations under the License.
	*
	*/

	#include "src/cpp/thread_manager/thread_manager.h"

	#include <climits>
	#include <mutex>

	#include <grpc/support/log.h>
	#include "src/core/lib/gprpp/thd.h"
	#include "src/core/lib/iomgr/exec_ctx.h"

	namespace grpc {

	ThreadManager::WorkerThread::WorkerThread(ThreadManager* thd_mgr)
	: thd_mgr_(thd_mgr) {
	// Make thread creation exclusive with respect to its join happening in
	// ~WorkerThread().
	thd_ = grpc_core::Thread(
	"grpcpp_sync_server",
	[](void* th) { static_cast<ThreadManager::WorkerThread*>(th)->Run(); },
	this);
	thd_.Start();
	}

	void ThreadManager::WorkerThread::Run() {
	thd_mgr_->MainWorkLoop();
	thd_mgr_->MarkAsCompleted(this);
	}

	ThreadManager::WorkerThread::~WorkerThread() {
	// Don't join until the thread is fully constructed.
	thd_.Join();
	}

	ThreadManager::ThreadManager(const char* name,
	grpc_resource_quota* resource_quota,
	int min_pollers, int max_pollers)
	: shutdown_(false),
	num_pollers_(0),
	min_pollers_(min_pollers),
	max_pollers_(max_pollers == -1 ? INT_MAX : max_pollers),
	num_threads_(0),
	max_active_threads_sofar_(0) {
	resource_user_ = grpc_resource_user_create(resource_quota, name);
	}

	ThreadManager::~ThreadManager() {
	{
	std::lock_guard<std::mutex> lock(mu_);
	GPR_ASSERT(num_threads_ == 0);
	}

	grpc_core::ExecCtx exec_ctx; // grpc_resource_user_unref needs an exec_ctx
	grpc_resource_user_unref(resource_user_);
	CleanupCompletedThreads();
	}

	void ThreadManager::Wait() {
	std::unique_lock<std::mutex> lock(mu_);
	while (num_threads_ != 0) {
	shutdown_cv_.wait(lock);
	}
	}

	void ThreadManager::Shutdown() {
	std::lock_guard<std::mutex> lock(mu_);
	shutdown_ = true;
	}

	bool ThreadManager::IsShutdown() {
	std::lock_guard<std::mutex> lock(mu_);
	return shutdown_;
	}

	int ThreadManager::GetMaxActiveThreadsSoFar() {
	std::lock_guard<std::mutex> list_lock(list_mu_);
	return max_active_threads_sofar_;
	}

	void ThreadManager::MarkAsCompleted(WorkerThread* thd) {
	{
	std::lock_guard<std::mutex> list_lock(list_mu_);
	completed_threads_.push_back(thd);
	}

	{
	std::lock_guard<std::mutex> lock(mu_);
	num_threads_--;
	if (num_threads_ == 0) {
	shutdown_cv_.notify_one();
	}
	}

	// Give a thread back to the resource quota
	grpc_resource_user_free_threads(resource_user_, 1);
	}

	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_);
	completed_threads.swap(completed_threads_);
	}
	for (auto thd : completed_threads) delete thd;
	}

	void ThreadManager::Initialize() {
	if (!grpc_resource_user_allocate_threads(resource_user_, min_pollers_)) {
	gpr_log(GPR_ERROR,
	"No thread quota available to even create the minimum required "
	"polling threads (i.e %d). Unable to start the thread manager",
	min_pollers_);
	abort();
	}

	{
	std::unique_lock<std::mutex> lock(mu_);
	num_pollers_ = min_pollers_;
	num_threads_ = min_pollers_;
	max_active_threads_sofar_ = min_pollers_;
	}

	for (int i = 0; i < min_pollers_; i++) {
	new WorkerThread(this);
	}
	}

	void ThreadManager::MainWorkLoop() {
	while (true) {
	void* tag;
	bool ok;
	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_--;
	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 and there is
	// quota available to create a new thread, start a new poller thread
	bool resource_exhausted = false;
	if (!shutdown_ && num_pollers_ < min_pollers_) {
	if (grpc_resource_user_allocate_threads(resource_user_, 1)) {
	// We can allocate a new poller thread
	num_pollers_++;
	num_threads_++;
	if (num_threads_ > max_active_threads_sofar_) {
	max_active_threads_sofar_ = num_threads_;
	}
	// Drop lock before spawning thread to avoid contention
	lock.unlock();
	new WorkerThread(this);
	} else if (num_pollers_ > 0) {
	// There is still at least some thread polling, so we can go on
	// even though we are below the number of pollers that we would
	// like to have (min_pollers_)
	lock.unlock();
	} else {
	// There are no pollers to spare and we couldn't allocate
	// a new thread, so resources are exhausted!
	lock.unlock();
	resource_exhausted = true;
	}
	} else {
	// There are a sufficient number of pollers available so we can do
	// the work and continue polling with our existing poller threads
	lock.unlock();
	}
	// Lock is always released at this point - do the application work
	// or return resource exhausted if there is new work but we couldn't
	// get a thread in which to do it.
	DoWork(tag, ok, !resource_exhausted);
	// 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;
	}
	// If we decided to finish the thread, break out of the while loop
	if (done) break;

	// Otherwise go back to polling as long as it doesn't exceed max_pollers_
	//
	// WARNING:
	// There is a possibility of threads thrashing here (i.e excessive thread
	// shutdowns and creations than the ideal case). This happens if max_poller_
	// count is small and the rate of incoming requests is also small. In such
	// scenarios we can possibly configure max_pollers_ to a higher value and/or
	// increase the cq timeout.
	//
	// However, not doing this check here and unconditionally incrementing
	// num_pollers (and hoping that the system will eventually settle down) has
	// far worse consequences i.e huge number of threads getting created to the
	// point of thread-exhaustion. For example: if the incoming request rate is
	// very high, all the polling threads will return very quickly from
	// PollForWork() with WORK_FOUND. They all briefly decrement num_pollers_
	// counter thereby possibly - and briefly - making it go below min_pollers;
	// This will most likely result in the creation of a new poller since
	// num_pollers_ dipped below min_pollers_.
	//
	// Now, If we didn't do the max_poller_ check here, all these threads will
	// go back to doing PollForWork() and the whole cycle repeats (with a new
	// thread being added in each cycle). Once the total number of threads in
	// the system crosses a certain threshold (around ~1500), there is heavy
	// contention on mutexes (the mu_ here or the mutexes in gRPC core like the
	// pollset mutex) that makes DoWork() take longer to finish thereby causing
	// new poller threads to be created even faster. This results in a thread
	// avalanche.
	if (num_pollers_ < max_pollers_) {
	num_pollers_++;
	} else {
	break;
	}
	};

	// This thread is exiting. Do some cleanup work i.e delete already completed
	// worker threads
	CleanupCompletedThreads();

	// If we are here, either ThreadManager is shutting down or it already has
	// enough threads.
	}

	} // namespace grpc