1 /*
2 *
3 * Copyright 2015 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 <grpc/support/port_platform.h>
20
21 #include "src/core/lib/iomgr/executor.h"
22
23 #include <string.h>
24
25 #include <grpc/support/alloc.h>
26 #include <grpc/support/cpu.h>
27 #include <grpc/support/log.h>
28 #include <grpc/support/sync.h>
29
30 #include "src/core/lib/gpr/tls.h"
31 #include "src/core/lib/gpr/useful.h"
32 #include "src/core/lib/gprpp/memory.h"
33 #include "src/core/lib/iomgr/exec_ctx.h"
34 #include "src/core/lib/iomgr/iomgr_internal.h"
35
36 #define MAX_DEPTH 2
37
38 #define EXECUTOR_TRACE(format, ...) \
39 do { \
40 if (GRPC_TRACE_FLAG_ENABLED(executor_trace)) { \
41 gpr_log(GPR_INFO, "EXECUTOR " format, __VA_ARGS__); \
42 } \
43 } while (0)
44
45 #define EXECUTOR_TRACE0(str) \
46 do { \
47 if (GRPC_TRACE_FLAG_ENABLED(executor_trace)) { \
48 gpr_log(GPR_INFO, "EXECUTOR " str); \
49 } \
50 } while (0)
51
52 namespace grpc_core {
53 namespace {
54
55 static GPR_THREAD_LOCAL(ThreadState*) g_this_thread_state;
56
57 Executor* executors[static_cast<size_t>(ExecutorType::NUM_EXECUTORS)];
58
default_enqueue_short(grpc_closure * closure,grpc_error_handle error)59 void default_enqueue_short(grpc_closure* closure, grpc_error_handle error) {
60 executors[static_cast<size_t>(ExecutorType::DEFAULT)]->Enqueue(
61 closure, error, true /* is_short */);
62 }
63
default_enqueue_long(grpc_closure * closure,grpc_error_handle error)64 void default_enqueue_long(grpc_closure* closure, grpc_error_handle error) {
65 executors[static_cast<size_t>(ExecutorType::DEFAULT)]->Enqueue(
66 closure, error, false /* is_short */);
67 }
68
resolver_enqueue_short(grpc_closure * closure,grpc_error_handle error)69 void resolver_enqueue_short(grpc_closure* closure, grpc_error_handle error) {
70 executors[static_cast<size_t>(ExecutorType::RESOLVER)]->Enqueue(
71 closure, error, true /* is_short */);
72 }
73
resolver_enqueue_long(grpc_closure * closure,grpc_error_handle error)74 void resolver_enqueue_long(grpc_closure* closure, grpc_error_handle error) {
75 executors[static_cast<size_t>(ExecutorType::RESOLVER)]->Enqueue(
76 closure, error, false /* is_short */);
77 }
78
79 using EnqueueFunc = void (*)(grpc_closure* closure, grpc_error_handle error);
80
81 const EnqueueFunc
82 executor_enqueue_fns_[static_cast<size_t>(ExecutorType::NUM_EXECUTORS)]
83 [static_cast<size_t>(ExecutorJobType::NUM_JOB_TYPES)] =
84 {{default_enqueue_short, default_enqueue_long},
85 {resolver_enqueue_short, resolver_enqueue_long}};
86
87 } // namespace
88
89 TraceFlag executor_trace(false, "executor");
90
Executor(const char * name)91 Executor::Executor(const char* name) : name_(name) {
92 adding_thread_lock_ = GPR_SPINLOCK_STATIC_INITIALIZER;
93 gpr_atm_rel_store(&num_threads_, 0);
94 max_threads_ = std::max(1u, 2 * gpr_cpu_num_cores());
95 }
96
Init()97 void Executor::Init() { SetThreading(true); }
98
RunClosures(const char * executor_name,grpc_closure_list list)99 size_t Executor::RunClosures(const char* executor_name,
100 grpc_closure_list list) {
101 size_t n = 0;
102
103 // In the executor, the ExecCtx for the thread is declared in the executor
104 // thread itself, but this is the point where we could start seeing
105 // application-level callbacks. No need to create a new ExecCtx, though,
106 // since there already is one and it is flushed (but not destructed) in this
107 // function itself. The ApplicationCallbackExecCtx will have its callbacks
108 // invoked on its destruction, which will be after completing any closures in
109 // the executor's closure list (which were explicitly scheduled onto the
110 // executor).
111 grpc_core::ApplicationCallbackExecCtx callback_exec_ctx(
112 GRPC_APP_CALLBACK_EXEC_CTX_FLAG_IS_INTERNAL_THREAD);
113
114 grpc_closure* c = list.head;
115 while (c != nullptr) {
116 grpc_closure* next = c->next_data.next;
117 grpc_error_handle error = c->error_data.error;
118 #ifndef NDEBUG
119 EXECUTOR_TRACE("(%s) run %p [created by %s:%d]", executor_name, c,
120 c->file_created, c->line_created);
121 c->scheduled = false;
122 #else
123 EXECUTOR_TRACE("(%s) run %p", executor_name, c);
124 #endif
125 c->cb(c->cb_arg, error);
126 GRPC_ERROR_UNREF(error);
127 c = next;
128 n++;
129 grpc_core::ExecCtx::Get()->Flush();
130 }
131
132 return n;
133 }
134
IsThreaded() const135 bool Executor::IsThreaded() const {
136 return gpr_atm_acq_load(&num_threads_) > 0;
137 }
138
SetThreading(bool threading)139 void Executor::SetThreading(bool threading) {
140 gpr_atm curr_num_threads = gpr_atm_acq_load(&num_threads_);
141 EXECUTOR_TRACE("(%s) SetThreading(%d) begin", name_, threading);
142
143 if (threading) {
144 if (curr_num_threads > 0) {
145 EXECUTOR_TRACE("(%s) SetThreading(true). curr_num_threads > 0", name_);
146 return;
147 }
148
149 GPR_ASSERT(num_threads_ == 0);
150 gpr_atm_rel_store(&num_threads_, 1);
151 thd_state_ = static_cast<ThreadState*>(
152 gpr_zalloc(sizeof(ThreadState) * max_threads_));
153
154 for (size_t i = 0; i < max_threads_; i++) {
155 gpr_mu_init(&thd_state_[i].mu);
156 gpr_cv_init(&thd_state_[i].cv);
157 thd_state_[i].id = i;
158 thd_state_[i].name = name_;
159 thd_state_[i].thd = grpc_core::Thread();
160 thd_state_[i].elems = GRPC_CLOSURE_LIST_INIT;
161 }
162
163 thd_state_[0].thd =
164 grpc_core::Thread(name_, &Executor::ThreadMain, &thd_state_[0]);
165 thd_state_[0].thd.Start();
166 } else { // !threading
167 if (curr_num_threads == 0) {
168 EXECUTOR_TRACE("(%s) SetThreading(false). curr_num_threads == 0", name_);
169 return;
170 }
171
172 for (size_t i = 0; i < max_threads_; i++) {
173 gpr_mu_lock(&thd_state_[i].mu);
174 thd_state_[i].shutdown = true;
175 gpr_cv_signal(&thd_state_[i].cv);
176 gpr_mu_unlock(&thd_state_[i].mu);
177 }
178
179 /* Ensure no thread is adding a new thread. Once this is past, then no
180 * thread will try to add a new one either (since shutdown is true) */
181 gpr_spinlock_lock(&adding_thread_lock_);
182 gpr_spinlock_unlock(&adding_thread_lock_);
183
184 curr_num_threads = gpr_atm_no_barrier_load(&num_threads_);
185 for (gpr_atm i = 0; i < curr_num_threads; i++) {
186 thd_state_[i].thd.Join();
187 EXECUTOR_TRACE("(%s) Thread %" PRIdPTR " of %" PRIdPTR " joined", name_,
188 i + 1, curr_num_threads);
189 }
190
191 gpr_atm_rel_store(&num_threads_, 0);
192 for (size_t i = 0; i < max_threads_; i++) {
193 gpr_mu_destroy(&thd_state_[i].mu);
194 gpr_cv_destroy(&thd_state_[i].cv);
195 RunClosures(thd_state_[i].name, thd_state_[i].elems);
196 }
197
198 gpr_free(thd_state_);
199
200 // grpc_iomgr_shutdown_background_closure() will close all the registered
201 // fds in the background poller, and wait for all pending closures to
202 // finish. Thus, never call Executor::SetThreading(false) in the middle of
203 // an application.
204 // TODO(guantaol): create another method to finish all the pending closures
205 // registered in the background poller by grpc_core::Executor.
206 grpc_iomgr_platform_shutdown_background_closure();
207 }
208
209 EXECUTOR_TRACE("(%s) SetThreading(%d) done", name_, threading);
210 }
211
Shutdown()212 void Executor::Shutdown() { SetThreading(false); }
213
ThreadMain(void * arg)214 void Executor::ThreadMain(void* arg) {
215 ThreadState* ts = static_cast<ThreadState*>(arg);
216 g_this_thread_state = ts;
217
218 grpc_core::ExecCtx exec_ctx(GRPC_EXEC_CTX_FLAG_IS_INTERNAL_THREAD);
219
220 size_t subtract_depth = 0;
221 for (;;) {
222 EXECUTOR_TRACE("(%s) [%" PRIdPTR "]: step (sub_depth=%" PRIdPTR ")",
223 ts->name, ts->id, subtract_depth);
224
225 gpr_mu_lock(&ts->mu);
226 ts->depth -= subtract_depth;
227 // Wait for closures to be enqueued or for the executor to be shutdown
228 while (grpc_closure_list_empty(ts->elems) && !ts->shutdown) {
229 ts->queued_long_job = false;
230 gpr_cv_wait(&ts->cv, &ts->mu, gpr_inf_future(GPR_CLOCK_MONOTONIC));
231 }
232
233 if (ts->shutdown) {
234 EXECUTOR_TRACE("(%s) [%" PRIdPTR "]: shutdown", ts->name, ts->id);
235 gpr_mu_unlock(&ts->mu);
236 break;
237 }
238
239 grpc_closure_list closures = ts->elems;
240 ts->elems = GRPC_CLOSURE_LIST_INIT;
241 gpr_mu_unlock(&ts->mu);
242
243 EXECUTOR_TRACE("(%s) [%" PRIdPTR "]: execute", ts->name, ts->id);
244
245 grpc_core::ExecCtx::Get()->InvalidateNow();
246 subtract_depth = RunClosures(ts->name, closures);
247 }
248
249 g_this_thread_state = nullptr;
250 }
251
Enqueue(grpc_closure * closure,grpc_error_handle error,bool is_short)252 void Executor::Enqueue(grpc_closure* closure, grpc_error_handle error,
253 bool is_short) {
254 bool retry_push;
255
256 do {
257 retry_push = false;
258 size_t cur_thread_count =
259 static_cast<size_t>(gpr_atm_acq_load(&num_threads_));
260
261 // If the number of threads is zero(i.e either the executor is not threaded
262 // or already shutdown), then queue the closure on the exec context itself
263 if (cur_thread_count == 0) {
264 #ifndef NDEBUG
265 EXECUTOR_TRACE("(%s) schedule %p (created %s:%d) inline", name_, closure,
266 closure->file_created, closure->line_created);
267 #else
268 EXECUTOR_TRACE("(%s) schedule %p inline", name_, closure);
269 #endif
270 grpc_closure_list_append(grpc_core::ExecCtx::Get()->closure_list(),
271 closure, error);
272 return;
273 }
274
275 if (grpc_iomgr_platform_add_closure_to_background_poller(closure, error)) {
276 return;
277 }
278
279 ThreadState* ts = g_this_thread_state;
280 if (ts == nullptr) {
281 ts = &thd_state_[grpc_core::HashPointer(grpc_core::ExecCtx::Get(),
282 cur_thread_count)];
283 }
284
285 ThreadState* orig_ts = ts;
286 bool try_new_thread = false;
287
288 for (;;) {
289 #ifndef NDEBUG
290 EXECUTOR_TRACE(
291 "(%s) try to schedule %p (%s) (created %s:%d) to thread "
292 "%" PRIdPTR,
293 name_, closure, is_short ? "short" : "long", closure->file_created,
294 closure->line_created, ts->id);
295 #else
296 EXECUTOR_TRACE("(%s) try to schedule %p (%s) to thread %" PRIdPTR, name_,
297 closure, is_short ? "short" : "long", ts->id);
298 #endif
299
300 gpr_mu_lock(&ts->mu);
301 if (ts->queued_long_job) {
302 // if there's a long job queued, we never queue anything else to this
303 // queue (since long jobs can take 'infinite' time and we need to
304 // guarantee no starvation). Spin through queues and try again
305 gpr_mu_unlock(&ts->mu);
306 size_t idx = ts->id;
307 ts = &thd_state_[(idx + 1) % cur_thread_count];
308 if (ts == orig_ts) {
309 // We cycled through all the threads. Retry enqueue again by creating
310 // a new thread
311 //
312 // TODO (sreek): There is a potential issue here. We are
313 // unconditionally setting try_new_thread to true here. What if the
314 // executor is shutdown OR if cur_thread_count is already equal to
315 // max_threads ?
316 // (Fortunately, this is not an issue yet (as of july 2018) because
317 // there is only one instance of long job in gRPC and hence we will
318 // not hit this code path)
319 retry_push = true;
320 try_new_thread = true;
321 break;
322 }
323
324 continue; // Try the next thread-state
325 }
326
327 // == Found the thread state (i.e thread) to enqueue this closure! ==
328
329 // Also, if this thread has been waiting for closures, wake it up.
330 // - If grpc_closure_list_empty() is true and the Executor is not
331 // shutdown, it means that the thread must be waiting in ThreadMain()
332 // - Note that gpr_cv_signal() won't immediately wakeup the thread. That
333 // happens after we release the mutex &ts->mu a few lines below
334 if (grpc_closure_list_empty(ts->elems) && !ts->shutdown) {
335 gpr_cv_signal(&ts->cv);
336 }
337
338 grpc_closure_list_append(&ts->elems, closure, error);
339
340 // If we already queued more than MAX_DEPTH number of closures on this
341 // thread, use this as a hint to create more threads
342 ts->depth++;
343 try_new_thread = ts->depth > MAX_DEPTH &&
344 cur_thread_count < max_threads_ && !ts->shutdown;
345
346 ts->queued_long_job = !is_short;
347
348 gpr_mu_unlock(&ts->mu);
349 break;
350 }
351
352 if (try_new_thread && gpr_spinlock_trylock(&adding_thread_lock_)) {
353 cur_thread_count = static_cast<size_t>(gpr_atm_acq_load(&num_threads_));
354 if (cur_thread_count < max_threads_) {
355 // Increment num_threads (safe to do a store instead of a cas because we
356 // always increment num_threads under the 'adding_thread_lock')
357 gpr_atm_rel_store(&num_threads_, cur_thread_count + 1);
358
359 thd_state_[cur_thread_count].thd = grpc_core::Thread(
360 name_, &Executor::ThreadMain, &thd_state_[cur_thread_count]);
361 thd_state_[cur_thread_count].thd.Start();
362 }
363 gpr_spinlock_unlock(&adding_thread_lock_);
364 }
365 } while (retry_push);
366 }
367
368 // Executor::InitAll() and Executor::ShutdownAll() functions are called in the
369 // the grpc_init() and grpc_shutdown() code paths which are protected by a
370 // global mutex. So it is okay to assume that these functions are thread-safe
InitAll()371 void Executor::InitAll() {
372 EXECUTOR_TRACE0("Executor::InitAll() enter");
373
374 // Return if Executor::InitAll() is already called earlier
375 if (executors[static_cast<size_t>(ExecutorType::DEFAULT)] != nullptr) {
376 GPR_ASSERT(executors[static_cast<size_t>(ExecutorType::RESOLVER)] !=
377 nullptr);
378 return;
379 }
380
381 executors[static_cast<size_t>(ExecutorType::DEFAULT)] =
382 new Executor("default-executor");
383 executors[static_cast<size_t>(ExecutorType::RESOLVER)] =
384 new Executor("resolver-executor");
385
386 executors[static_cast<size_t>(ExecutorType::DEFAULT)]->Init();
387 executors[static_cast<size_t>(ExecutorType::RESOLVER)]->Init();
388
389 EXECUTOR_TRACE0("Executor::InitAll() done");
390 }
391
Run(grpc_closure * closure,grpc_error_handle error,ExecutorType executor_type,ExecutorJobType job_type)392 void Executor::Run(grpc_closure* closure, grpc_error_handle error,
393 ExecutorType executor_type, ExecutorJobType job_type) {
394 executor_enqueue_fns_[static_cast<size_t>(executor_type)]
395 [static_cast<size_t>(job_type)](closure, error);
396 }
397
ShutdownAll()398 void Executor::ShutdownAll() {
399 EXECUTOR_TRACE0("Executor::ShutdownAll() enter");
400
401 // Return if Executor:SshutdownAll() is already called earlier
402 if (executors[static_cast<size_t>(ExecutorType::DEFAULT)] == nullptr) {
403 GPR_ASSERT(executors[static_cast<size_t>(ExecutorType::RESOLVER)] ==
404 nullptr);
405 return;
406 }
407
408 executors[static_cast<size_t>(ExecutorType::DEFAULT)]->Shutdown();
409 executors[static_cast<size_t>(ExecutorType::RESOLVER)]->Shutdown();
410
411 // Delete the executor objects.
412 //
413 // NOTE: It is important to call Shutdown() on all executors first before
414 // calling delete because it is possible for one executor (that is not
415 // shutdown yet) to call Enqueue() on a different executor which is already
416 // shutdown. This is legal and in such cases, the Enqueue() operation
417 // effectively "fails" and enqueues that closure on the calling thread's
418 // exec_ctx.
419 //
420 // By ensuring that all executors are shutdown first, we are also ensuring
421 // that no thread is active across all executors.
422
423 delete executors[static_cast<size_t>(ExecutorType::DEFAULT)];
424 delete executors[static_cast<size_t>(ExecutorType::RESOLVER)];
425 executors[static_cast<size_t>(ExecutorType::DEFAULT)] = nullptr;
426 executors[static_cast<size_t>(ExecutorType::RESOLVER)] = nullptr;
427
428 EXECUTOR_TRACE0("Executor::ShutdownAll() done");
429 }
430
IsThreaded(ExecutorType executor_type)431 bool Executor::IsThreaded(ExecutorType executor_type) {
432 GPR_ASSERT(executor_type < ExecutorType::NUM_EXECUTORS);
433 return executors[static_cast<size_t>(executor_type)]->IsThreaded();
434 }
435
IsThreadedDefault()436 bool Executor::IsThreadedDefault() {
437 return Executor::IsThreaded(ExecutorType::DEFAULT);
438 }
439
SetThreadingAll(bool enable)440 void Executor::SetThreadingAll(bool enable) {
441 EXECUTOR_TRACE("Executor::SetThreadingAll(%d) called", enable);
442 for (size_t i = 0; i < static_cast<size_t>(ExecutorType::NUM_EXECUTORS);
443 i++) {
444 executors[i]->SetThreading(enable);
445 }
446 }
447
SetThreadingDefault(bool enable)448 void Executor::SetThreadingDefault(bool enable) {
449 EXECUTOR_TRACE("Executor::SetThreadingDefault(%d) called", enable);
450 executors[static_cast<size_t>(ExecutorType::DEFAULT)]->SetThreading(enable);
451 }
452
grpc_executor_global_init()453 void grpc_executor_global_init() {}
454
455 } // namespace grpc_core
456