aboutsummaryrefslogtreecommitdiffhomepage
path: root/src/cpp/thread_manager
diff options
context:
space:
mode:
Diffstat (limited to 'src/cpp/thread_manager')
-rw-r--r--src/cpp/thread_manager/thread_manager.cc81
-rw-r--r--src/cpp/thread_manager/thread_manager.h54
2 files changed, 111 insertions, 24 deletions
diff --git a/src/cpp/thread_manager/thread_manager.cc b/src/cpp/thread_manager/thread_manager.cc
index 02ac56a3fd..3e8606a76f 100644
--- a/src/cpp/thread_manager/thread_manager.cc
+++ b/src/cpp/thread_manager/thread_manager.cc
@@ -22,8 +22,8 @@
#include <mutex>
#include <grpc/support/log.h>
-
#include "src/core/lib/gprpp/thd.h"
+#include "src/core/lib/iomgr/exec_ctx.h"
namespace grpc {
@@ -48,12 +48,17 @@ ThreadManager::WorkerThread::~WorkerThread() {
thd_.Join();
}
-ThreadManager::ThreadManager(int min_pollers, int max_pollers)
+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) {}
+ num_threads_(0),
+ max_active_threads_sofar_(0) {
+ resource_user_ = grpc_resource_user_create(resource_quota, name);
+}
ThreadManager::~ThreadManager() {
{
@@ -61,6 +66,8 @@ ThreadManager::~ThreadManager() {
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();
}
@@ -81,17 +88,27 @@ bool ThreadManager::IsShutdown() {
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();
+ {
+ 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() {
@@ -106,14 +123,22 @@ void ThreadManager::CleanupCompletedThreads() {
}
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++) {
- // Create a new thread (which ends up calling the MainWorkLoop() function
new WorkerThread(this);
}
}
@@ -139,20 +164,40 @@ void ThreadManager::MainWorkLoop() {
done = true;
break;
case WORK_FOUND:
- // If we got work and there are now insufficient pollers, start a new
- // one
+ // 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_) {
- num_pollers_++;
- num_threads_++;
- // Drop lock before spawning thread to avoid contention
- lock.unlock();
- new WorkerThread(this);
+ 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 {
- // Drop lock for consistency with above branch
+ // 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
- DoWork(tag, ok);
+ // 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.
@@ -196,6 +241,8 @@ void ThreadManager::MainWorkLoop() {
}
};
+ // 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
diff --git a/src/cpp/thread_manager/thread_manager.h b/src/cpp/thread_manager/thread_manager.h
index 5a40f2de47..6f0bd17c5f 100644
--- a/src/cpp/thread_manager/thread_manager.h
+++ b/src/cpp/thread_manager/thread_manager.h
@@ -27,12 +27,14 @@
#include <grpcpp/support/config.h>
#include "src/core/lib/gprpp/thd.h"
+#include "src/core/lib/iomgr/resource_quota.h"
namespace grpc {
class ThreadManager {
public:
- explicit ThreadManager(int min_pollers, int max_pollers);
+ explicit ThreadManager(const char* name, grpc_resource_quota* resource_quota,
+ int min_pollers, int max_pollers);
virtual ~ThreadManager();
// Initializes and Starts the Rpc Manager threads
@@ -65,12 +67,14 @@ class ThreadManager {
// The implementation of DoWork() is supposed to perform the work found by
// PollForWork(). The tag and ok parameters are the same as returned by
- // PollForWork()
+ // PollForWork(). The resources parameter indicates that the call actually
+ // has the resources available for performing the RPC's work. If it doesn't,
+ // the implementation should fail it appropriately.
//
// The implementation of DoWork() should also do any setup needed to ensure
// that the next call to PollForWork() (not necessarily by the current thread)
// actually finds some work
- virtual void DoWork(void* tag, bool ok) = 0;
+ virtual void DoWork(void* tag, bool ok, bool resources) = 0;
// 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
@@ -84,6 +88,11 @@ class ThreadManager {
// all the threads have drained all the outstanding work
virtual void Wait();
+ // Max number of concurrent threads that were ever active in this thread
+ // manager so far. This is useful for debugging purposes (and in unit tests)
+ // to check if resource_quota is properly being enforced.
+ int GetMaxActiveThreadsSoFar();
+
private:
// Helper wrapper class around grpc_core::Thread. Takes a ThreadManager object
// and starts a new grpc_core::Thread to calls the Run() function.
@@ -91,6 +100,24 @@ class ThreadManager {
// The Run() function calls ThreadManager::MainWorkLoop() function and once
// that completes, it marks the WorkerThread completed by calling
// ThreadManager::MarkAsCompleted()
+ //
+ // WHY IS THIS NEEDED?:
+ // When a thread terminates, some other thread *must* call Join() on that
+ // thread so that the resources are released. Having a WorkerThread wrapper
+ // will make this easier. Once Run() completes, each thread calls the
+ // following two functions:
+ // ThreadManager::CleanupCompletedThreads()
+ // ThreadManager::MarkAsCompleted()
+ //
+ // - MarkAsCompleted() puts the WorkerThread object in the ThreadManger's
+ // completed_threads_ list
+ // - CleanupCompletedThreads() calls "Join()" on the threads that are already
+ // in the completed_threads_ list (since a thread cannot call Join() on
+ // itself, it calls CleanupCompletedThreads() *before* calling
+ // MarkAsCompleted())
+ //
+ // TODO(sreek): Consider creating the threads 'detached' so that Join() need
+ // not be called (and the need for this WorkerThread class is eliminated)
class WorkerThread {
public:
WorkerThread(ThreadManager* thd_mgr);
@@ -111,13 +138,21 @@ class ThreadManager {
void MarkAsCompleted(WorkerThread* thd);
void CleanupCompletedThreads();
- // Protects shutdown_, num_pollers_ and num_threads_
- // TODO: sreek - Change num_pollers and num_threads_ to atomics
+ // Protects shutdown_, num_pollers_, num_threads_ and
+ // max_active_threads_sofar_
std::mutex mu_;
bool shutdown_;
std::condition_variable shutdown_cv_;
+ // The resource user object to use when requesting quota to create threads
+ //
+ // Note: The user of this ThreadManager object must create grpc_resource_quota
+ // object (that contains the actual max thread quota) and a grpc_resource_user
+ // object through which quota is requested whenver new threads need to be
+ // created
+ grpc_resource_user* resource_user_;
+
// Number of threads doing polling
int num_pollers_;
@@ -125,10 +160,15 @@ class ThreadManager {
int min_pollers_;
int max_pollers_;
- // The total number of threads (includes threads includes the threads that are
- // currently polling i.e num_pollers_)
+ // The total number of threads currently active (includes threads includes the
+ // threads that are currently polling i.e num_pollers_)
int num_threads_;
+ // See GetMaxActiveThreadsSoFar()'s description.
+ // To be more specific, this variable tracks the max value num_threads_ was
+ // ever set so far
+ int max_active_threads_sofar_;
+
std::mutex list_mu_;
std::list<WorkerThread*> completed_threads_;
};