aboutsummaryrefslogtreecommitdiffhomepage
path: root/src/cpp
diff options
context:
space:
mode:
Diffstat (limited to 'src/cpp')
-rw-r--r--src/cpp/common/resource_quota_cc.cc4
-rw-r--r--src/cpp/server/server_builder.cc2
-rw-r--r--src/cpp/server/server_cc.cc31
-rw-r--r--src/cpp/thread_manager/thread_manager.cc53
-rw-r--r--src/cpp/thread_manager/thread_manager.h48
5 files changed, 23 insertions, 115 deletions
diff --git a/src/cpp/common/resource_quota_cc.cc b/src/cpp/common/resource_quota_cc.cc
index 276e5f7954..daeb0ba171 100644
--- a/src/cpp/common/resource_quota_cc.cc
+++ b/src/cpp/common/resource_quota_cc.cc
@@ -33,8 +33,4 @@ ResourceQuota& ResourceQuota::Resize(size_t new_size) {
return *this;
}
-ResourceQuota& ResourceQuota::SetMaxThreads(int new_max_threads) {
- grpc_resource_quota_set_max_threads(impl_, new_max_threads);
- return *this;
-}
} // namespace grpc
diff --git a/src/cpp/server/server_builder.cc b/src/cpp/server/server_builder.cc
index 0ab3cd0e32..e0b9b7a62b 100644
--- a/src/cpp/server/server_builder.cc
+++ b/src/cpp/server/server_builder.cc
@@ -261,7 +261,7 @@ std::unique_ptr<Server> ServerBuilder::BuildAndStart() {
}
std::unique_ptr<Server> server(new Server(
- max_receive_message_size_, &args, sync_server_cqs, resource_quota_,
+ max_receive_message_size_, &args, sync_server_cqs,
sync_server_settings_.min_pollers, sync_server_settings_.max_pollers,
sync_server_settings_.cq_timeout_msec));
diff --git a/src/cpp/server/server_cc.cc b/src/cpp/server/server_cc.cc
index 472c5035fc..0d77510e29 100644
--- a/src/cpp/server/server_cc.cc
+++ b/src/cpp/server/server_cc.cc
@@ -47,12 +47,6 @@
namespace grpc {
namespace {
-// The default value for maximum number of threads that can be created in the
-// sync server. This value of 500 is empirically chosen. To increase the max
-// number of threads in a sync server, pass a custom ResourceQuota object (with
-// the desired number of max-threads set) to the server builder
-#define DEFAULT_MAX_SYNC_SERVER_THREADS 500
-
class DefaultGlobalCallbacks final : public Server::GlobalCallbacks {
public:
~DefaultGlobalCallbacks() override {}
@@ -272,9 +266,9 @@ class Server::SyncRequestThreadManager : public ThreadManager {
public:
SyncRequestThreadManager(Server* server, CompletionQueue* server_cq,
std::shared_ptr<GlobalCallbacks> global_callbacks,
- grpc_resource_quota* rq, int min_pollers,
- int max_pollers, int cq_timeout_msec)
- : ThreadManager("SyncServer", rq, min_pollers, max_pollers),
+ int min_pollers, int max_pollers,
+ int cq_timeout_msec)
+ : ThreadManager(min_pollers, max_pollers),
server_(server),
server_cq_(server_cq),
cq_timeout_msec_(cq_timeout_msec),
@@ -382,8 +376,7 @@ Server::Server(
int max_receive_message_size, ChannelArguments* args,
std::shared_ptr<std::vector<std::unique_ptr<ServerCompletionQueue>>>
sync_server_cqs,
- grpc_resource_quota* server_rq, int min_pollers, int max_pollers,
- int sync_cq_timeout_msec)
+ int min_pollers, int max_pollers, int sync_cq_timeout_msec)
: max_receive_message_size_(max_receive_message_size),
sync_server_cqs_(std::move(sync_server_cqs)),
started_(false),
@@ -399,22 +392,10 @@ Server::Server(
global_callbacks_->UpdateArguments(args);
if (sync_server_cqs_ != nullptr) {
- bool default_rq_created = false;
- if (server_rq == nullptr) {
- server_rq = grpc_resource_quota_create("SyncServer-default-rq");
- grpc_resource_quota_set_max_threads(server_rq,
- DEFAULT_MAX_SYNC_SERVER_THREADS);
- default_rq_created = true;
- }
-
for (const auto& it : *sync_server_cqs_) {
sync_req_mgrs_.emplace_back(new SyncRequestThreadManager(
- this, it.get(), global_callbacks_, server_rq, min_pollers,
- max_pollers, sync_cq_timeout_msec));
- }
-
- if (default_rq_created) {
- grpc_resource_quota_unref(server_rq);
+ this, it.get(), global_callbacks_, min_pollers, max_pollers,
+ sync_cq_timeout_msec));
}
}
diff --git a/src/cpp/thread_manager/thread_manager.cc b/src/cpp/thread_manager/thread_manager.cc
index fa9eec5f9b..02ac56a3fd 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,17 +48,12 @@ ThreadManager::WorkerThread::~WorkerThread() {
thd_.Join();
}
-ThreadManager::ThreadManager(const char* name,
- grpc_resource_quota* resource_quota,
- int min_pollers, int max_pollers)
+ThreadManager::ThreadManager(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);
-}
+ num_threads_(0) {}
ThreadManager::~ThreadManager() {
{
@@ -66,8 +61,6 @@ 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();
}
@@ -88,27 +81,17 @@ 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() {
@@ -123,22 +106,14 @@ 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);
}
}
@@ -164,15 +139,11 @@ void ThreadManager::MainWorkLoop() {
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
- if (!shutdown_ && num_pollers_ < min_pollers_ &&
- grpc_resource_user_allocate_threads(resource_user_, 1)) {
+ // If we got work and there are now insufficient pollers, start a new
+ // one
+ if (!shutdown_ && num_pollers_ < min_pollers_) {
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);
@@ -225,8 +196,6 @@ 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 01043edb31..5a40f2de47 100644
--- a/src/cpp/thread_manager/thread_manager.h
+++ b/src/cpp/thread_manager/thread_manager.h
@@ -27,14 +27,12 @@
#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(const char* name, grpc_resource_quota* resource_quota,
- int min_pollers, int max_pollers);
+ explicit ThreadManager(int min_pollers, int max_pollers);
virtual ~ThreadManager();
// Initializes and Starts the Rpc Manager threads
@@ -86,11 +84,6 @@ 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.
@@ -98,24 +91,6 @@ 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);
@@ -136,21 +111,13 @@ class ThreadManager {
void MarkAsCompleted(WorkerThread* thd);
void CleanupCompletedThreads();
- // Protects shutdown_, num_pollers_, num_threads_ and
- // max_active_threads_sofar_
+ // Protects shutdown_, num_pollers_ and num_threads_
+ // TODO: sreek - Change num_pollers and num_threads_ to atomics
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_;
@@ -158,15 +125,10 @@ class ThreadManager {
int min_pollers_;
int max_pollers_;
- // The total number of threads currently active (includes threads includes the
- // threads that are currently polling i.e num_pollers_)
+ // The total number of threads (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_;
};
generated by cgit v1.2.3 (git 2.39.1) at 2025-01-07 23:32:02 +0000