1 //
2 //
3 // Copyright 2016 gRPC authors.
4 //
5 // Licensed under the Apache License, Version 2.0 (the "License");
6 // you may not use this file except in compliance with the License.
7 // You may obtain a copy of the License at
8 //
9 //     http://www.apache.org/licenses/LICENSE-2.0
10 //
11 // Unless required by applicable law or agreed to in writing, software
12 // distributed under the License is distributed on an "AS IS" BASIS,
13 // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14 // See the License for the specific language governing permissions and
15 // limitations under the License.
16 //
17 //
18 
19 #include "src/cpp/thread_manager/thread_manager.h"
20 
21 #include <climits>
22 #include <initializer_list>
23 
24 #include "absl/strings/str_format.h"
25 
26 #include <grpc/support/log.h>
27 
28 #include "src/core/lib/gprpp/crash.h"
29 #include "src/core/lib/gprpp/ref_counted_ptr.h"
30 #include "src/core/lib/gprpp/thd.h"
31 #include "src/core/lib/resource_quota/resource_quota.h"
32 
33 namespace grpc {
34 
WorkerThread(ThreadManager * thd_mgr)35 ThreadManager::WorkerThread::WorkerThread(ThreadManager* thd_mgr)
36     : thd_mgr_(thd_mgr) {
37   // Make thread creation exclusive with respect to its join happening in
38   // ~WorkerThread().
39   thd_ = grpc_core::Thread(
40       "grpcpp_sync_server",
41       [](void* th) { static_cast<ThreadManager::WorkerThread*>(th)->Run(); },
42       this, &created_);
43   if (!created_) {
44     gpr_log(GPR_ERROR, "Could not create grpc_sync_server worker-thread");
45   }
46 }
47 
Run()48 void ThreadManager::WorkerThread::Run() {
49   thd_mgr_->MainWorkLoop();
50   thd_mgr_->MarkAsCompleted(this);
51 }
52 
~WorkerThread()53 ThreadManager::WorkerThread::~WorkerThread() {
54   // Don't join until the thread is fully constructed.
55   thd_.Join();
56 }
57 
ThreadManager(const char *,grpc_resource_quota * resource_quota,int min_pollers,int max_pollers)58 ThreadManager::ThreadManager(const char*, grpc_resource_quota* resource_quota,
59                              int min_pollers, int max_pollers)
60     : shutdown_(false),
61       thread_quota_(
62           grpc_core::ResourceQuota::FromC(resource_quota)->thread_quota()),
63       num_pollers_(0),
64       min_pollers_(min_pollers),
65       max_pollers_(max_pollers == -1 ? INT_MAX : max_pollers),
66       num_threads_(0),
67       max_active_threads_sofar_(0) {}
68 
~ThreadManager()69 ThreadManager::~ThreadManager() {
70   {
71     grpc_core::MutexLock lock(&mu_);
72     GPR_ASSERT(num_threads_ == 0);
73   }
74 
75   CleanupCompletedThreads();
76 }
77 
Wait()78 void ThreadManager::Wait() {
79   grpc_core::MutexLock lock(&mu_);
80   while (num_threads_ != 0) {
81     shutdown_cv_.Wait(&mu_);
82   }
83 }
84 
Shutdown()85 void ThreadManager::Shutdown() {
86   grpc_core::MutexLock lock(&mu_);
87   shutdown_ = true;
88 }
89 
IsShutdown()90 bool ThreadManager::IsShutdown() {
91   grpc_core::MutexLock lock(&mu_);
92   return shutdown_;
93 }
94 
GetMaxActiveThreadsSoFar()95 int ThreadManager::GetMaxActiveThreadsSoFar() {
96   grpc_core::MutexLock list_lock(&list_mu_);
97   return max_active_threads_sofar_;
98 }
99 
MarkAsCompleted(WorkerThread * thd)100 void ThreadManager::MarkAsCompleted(WorkerThread* thd) {
101   {
102     grpc_core::MutexLock list_lock(&list_mu_);
103     completed_threads_.push_back(thd);
104   }
105 
106   {
107     grpc_core::MutexLock lock(&mu_);
108     num_threads_--;
109     if (num_threads_ == 0) {
110       shutdown_cv_.Signal();
111     }
112   }
113 
114   // Give a thread back to the resource quota
115   thread_quota_->Release(1);
116 }
117 
CleanupCompletedThreads()118 void ThreadManager::CleanupCompletedThreads() {
119   std::list<WorkerThread*> completed_threads;
120   {
121     // swap out the completed threads list: allows other threads to clean up
122     // more quickly
123     grpc_core::MutexLock lock(&list_mu_);
124     completed_threads.swap(completed_threads_);
125   }
126   for (auto thd : completed_threads) delete thd;
127 }
128 
Initialize()129 void ThreadManager::Initialize() {
130   if (!thread_quota_->Reserve(min_pollers_)) {
131     grpc_core::Crash(absl::StrFormat(
132         "No thread quota available to even create the minimum required "
133         "polling threads (i.e %d). Unable to start the thread manager",
134         min_pollers_));
135   }
136 
137   {
138     grpc_core::MutexLock lock(&mu_);
139     num_pollers_ = min_pollers_;
140     num_threads_ = min_pollers_;
141     max_active_threads_sofar_ = min_pollers_;
142   }
143 
144   for (int i = 0; i < min_pollers_; i++) {
145     WorkerThread* worker = new WorkerThread(this);
146     GPR_ASSERT(worker->created());  // Must be able to create the minimum
147     worker->Start();
148   }
149 }
150 
MainWorkLoop()151 void ThreadManager::MainWorkLoop() {
152   while (true) {
153     void* tag;
154     bool ok;
155     WorkStatus work_status = PollForWork(&tag, &ok);
156 
157     grpc_core::LockableAndReleasableMutexLock lock(&mu_);
158     // Reduce the number of pollers by 1 and check what happened with the poll
159     num_pollers_--;
160     bool done = false;
161     switch (work_status) {
162       case TIMEOUT:
163         // If we timed out and we have more pollers than we need (or we are
164         // shutdown), finish this thread
165         if (shutdown_ || num_pollers_ > max_pollers_) done = true;
166         break;
167       case SHUTDOWN:
168         // If the thread manager is shutdown, finish this thread
169         done = true;
170         break;
171       case WORK_FOUND:
172         // If we got work and there are now insufficient pollers and there is
173         // quota available to create a new thread, start a new poller thread
174         bool resource_exhausted = false;
175         if (!shutdown_ && num_pollers_ < min_pollers_) {
176           if (thread_quota_->Reserve(1)) {
177             // We can allocate a new poller thread
178             num_pollers_++;
179             num_threads_++;
180             if (num_threads_ > max_active_threads_sofar_) {
181               max_active_threads_sofar_ = num_threads_;
182             }
183             // Drop lock before spawning thread to avoid contention
184             lock.Release();
185             WorkerThread* worker = new WorkerThread(this);
186             if (worker->created()) {
187               worker->Start();
188             } else {
189               // Get lock again to undo changes to poller/thread counters.
190               grpc_core::MutexLock failure_lock(&mu_);
191               num_pollers_--;
192               num_threads_--;
193               resource_exhausted = true;
194               delete worker;
195             }
196           } else if (num_pollers_ > 0) {
197             // There is still at least some thread polling, so we can go on
198             // even though we are below the number of pollers that we would
199             // like to have (min_pollers_)
200             lock.Release();
201           } else {
202             // There are no pollers to spare and we couldn't allocate
203             // a new thread, so resources are exhausted!
204             lock.Release();
205             resource_exhausted = true;
206           }
207         } else {
208           // There are a sufficient number of pollers available so we can do
209           // the work and continue polling with our existing poller threads
210           lock.Release();
211         }
212         // Lock is always released at this point - do the application work
213         // or return resource exhausted if there is new work but we couldn't
214         // get a thread in which to do it.
215         DoWork(tag, ok, !resource_exhausted);
216         // Take the lock again to check post conditions
217         lock.Lock();
218         // If we're shutdown, we should finish at this point.
219         if (shutdown_) done = true;
220         break;
221     }
222     // If we decided to finish the thread, break out of the while loop
223     if (done) break;
224 
225     // Otherwise go back to polling as long as it doesn't exceed max_pollers_
226     //
227     // **WARNING**:
228     // There is a possibility of threads thrashing here (i.e excessive thread
229     // shutdowns and creations than the ideal case). This happens if max_poller_
230     // count is small and the rate of incoming requests is also small. In such
231     // scenarios we can possibly configure max_pollers_ to a higher value and/or
232     // increase the cq timeout.
233     //
234     // However, not doing this check here and unconditionally incrementing
235     // num_pollers (and hoping that the system will eventually settle down) has
236     // far worse consequences i.e huge number of threads getting created to the
237     // point of thread-exhaustion. For example: if the incoming request rate is
238     // very high, all the polling threads will return very quickly from
239     // PollForWork() with WORK_FOUND. They all briefly decrement num_pollers_
240     // counter thereby possibly - and briefly - making it go below min_pollers;
241     // This will most likely result in the creation of a new poller since
242     // num_pollers_ dipped below min_pollers_.
243     //
244     // Now, If we didn't do the max_poller_ check here, all these threads will
245     // go back to doing PollForWork() and the whole cycle repeats (with a new
246     // thread being added in each cycle). Once the total number of threads in
247     // the system crosses a certain threshold (around ~1500), there is heavy
248     // contention on mutexes (the mu_ here or the mutexes in gRPC core like the
249     // pollset mutex) that makes DoWork() take longer to finish thereby causing
250     // new poller threads to be created even faster. This results in a thread
251     // avalanche.
252     if (num_pollers_ < max_pollers_) {
253       num_pollers_++;
254     } else {
255       break;
256     }
257   };
258 
259   // This thread is exiting. Do some cleanup work i.e delete already completed
260   // worker threads
261   CleanupCompletedThreads();
262 
263   // If we are here, either ThreadManager is shutting down or it already has
264   // enough threads.
265 }
266 
267 }  // namespace grpc
268