: shutdown_(false),

num_pollers_(0),

min_pollers_(min_pollers),

max_pollers_(max_pollers == -1 ? INT_MAX : max_pollers),

std::unique_lock<std::mutex> lock(mu_);

while (num_threads_ != 0) {

shutdown_cv_.wait(lock);

}

{

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::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;

{

std::unique_lock<std::mutex> lock(mu_);

num_pollers_ = min_pollers_;

num_threads_ = min_pollers_;

}

for (int i = 0; i < min_pollers_; i++) {

// Create a new thread (which ends up calling the MainWorkLoop() function

new WorkerThread(this);

}

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, start a new

// one

if (!shutdown_ && num_pollers_ < min_pollers_) {

num_pollers_++;

num_threads_++;

// Drop lock before spawning thread to avoid contention

lock.unlock();

new WorkerThread(this);

} else {

// Drop lock for consistency with above branch

lock.unlock();

}

// Lock is always released at this point - do the application work

DoWork(tag, ok);

// 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;

}

};

CleanupCompletedThreads();

// If we are here, either ThreadManager is shutting down or it already has

// enough threads.