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 <grpc/support/port_platform.h>
20
21 #include "src/core/lib/iomgr/combiner.h"
22
23 #include <assert.h>
24 #include <inttypes.h>
25 #include <string.h>
26
27 #include <grpc/support/alloc.h>
28 #include <grpc/support/log.h>
29
30 #include "src/core/lib/debug/stats.h"
31 #include "src/core/lib/gprpp/mpscq.h"
32 #include "src/core/lib/iomgr/executor.h"
33 #include "src/core/lib/iomgr/iomgr.h"
34 #include "src/core/lib/profiling/timers.h"
35
36 grpc_core::DebugOnlyTraceFlag grpc_combiner_trace(false, "combiner");
37
38 #define GRPC_COMBINER_TRACE(fn) \
39 do { \
40 if (grpc_combiner_trace.enabled()) { \
41 fn; \
42 } \
43 } while (0)
44
45 #define STATE_UNORPHANED 1
46 #define STATE_ELEM_COUNT_LOW_BIT 2
47
48 static void combiner_exec(grpc_core::Combiner* lock, grpc_closure* closure,
49 grpc_error* error);
50 static void combiner_finally_exec(grpc_core::Combiner* lock,
51 grpc_closure* closure, grpc_error* error);
52
53 static void offload(void* arg, grpc_error* error);
54
grpc_combiner_create(void)55 grpc_core::Combiner* grpc_combiner_create(void) {
56 grpc_core::Combiner* lock = new grpc_core::Combiner();
57 gpr_ref_init(&lock->refs, 1);
58 gpr_atm_no_barrier_store(&lock->state, STATE_UNORPHANED);
59 grpc_closure_list_init(&lock->final_list);
60 GRPC_CLOSURE_INIT(&lock->offload, offload, lock, nullptr);
61 GRPC_COMBINER_TRACE(gpr_log(GPR_INFO, "C:%p create", lock));
62 return lock;
63 }
64
really_destroy(grpc_core::Combiner * lock)65 static void really_destroy(grpc_core::Combiner* lock) {
66 GRPC_COMBINER_TRACE(gpr_log(GPR_INFO, "C:%p really_destroy", lock));
67 GPR_ASSERT(gpr_atm_no_barrier_load(&lock->state) == 0);
68 delete lock;
69 }
70
start_destroy(grpc_core::Combiner * lock)71 static void start_destroy(grpc_core::Combiner* lock) {
72 gpr_atm old_state = gpr_atm_full_fetch_add(&lock->state, -STATE_UNORPHANED);
73 GRPC_COMBINER_TRACE(gpr_log(
74 GPR_INFO, "C:%p really_destroy old_state=%" PRIdPTR, lock, old_state));
75 if (old_state == 1) {
76 really_destroy(lock);
77 }
78 }
79
80 #ifndef NDEBUG
81 #define GRPC_COMBINER_DEBUG_SPAM(op, delta) \
82 if (grpc_combiner_trace.enabled()) { \
83 gpr_log(file, line, GPR_LOG_SEVERITY_DEBUG, \
84 "C:%p %s %" PRIdPTR " --> %" PRIdPTR " %s", lock, (op), \
85 gpr_atm_no_barrier_load(&lock->refs.count), \
86 gpr_atm_no_barrier_load(&lock->refs.count) + (delta), reason); \
87 }
88 #else
89 #define GRPC_COMBINER_DEBUG_SPAM(op, delta)
90 #endif
91
grpc_combiner_unref(grpc_core::Combiner * lock GRPC_COMBINER_DEBUG_ARGS)92 void grpc_combiner_unref(grpc_core::Combiner* lock GRPC_COMBINER_DEBUG_ARGS) {
93 GRPC_COMBINER_DEBUG_SPAM("UNREF", -1);
94 if (gpr_unref(&lock->refs)) {
95 start_destroy(lock);
96 }
97 }
98
grpc_combiner_ref(grpc_core::Combiner * lock GRPC_COMBINER_DEBUG_ARGS)99 grpc_core::Combiner* grpc_combiner_ref(
100 grpc_core::Combiner* lock GRPC_COMBINER_DEBUG_ARGS) {
101 GRPC_COMBINER_DEBUG_SPAM(" REF", 1);
102 gpr_ref(&lock->refs);
103 return lock;
104 }
105
push_last_on_exec_ctx(grpc_core::Combiner * lock)106 static void push_last_on_exec_ctx(grpc_core::Combiner* lock) {
107 lock->next_combiner_on_this_exec_ctx = nullptr;
108 if (grpc_core::ExecCtx::Get()->combiner_data()->active_combiner == nullptr) {
109 grpc_core::ExecCtx::Get()->combiner_data()->active_combiner =
110 grpc_core::ExecCtx::Get()->combiner_data()->last_combiner = lock;
111 } else {
112 grpc_core::ExecCtx::Get()
113 ->combiner_data()
114 ->last_combiner->next_combiner_on_this_exec_ctx = lock;
115 grpc_core::ExecCtx::Get()->combiner_data()->last_combiner = lock;
116 }
117 }
118
push_first_on_exec_ctx(grpc_core::Combiner * lock)119 static void push_first_on_exec_ctx(grpc_core::Combiner* lock) {
120 lock->next_combiner_on_this_exec_ctx =
121 grpc_core::ExecCtx::Get()->combiner_data()->active_combiner;
122 grpc_core::ExecCtx::Get()->combiner_data()->active_combiner = lock;
123 if (lock->next_combiner_on_this_exec_ctx == nullptr) {
124 grpc_core::ExecCtx::Get()->combiner_data()->last_combiner = lock;
125 }
126 }
127
combiner_exec(grpc_core::Combiner * lock,grpc_closure * cl,grpc_error * error)128 static void combiner_exec(grpc_core::Combiner* lock, grpc_closure* cl,
129 grpc_error* error) {
130 GPR_TIMER_SCOPE("combiner.execute", 0);
131 GRPC_STATS_INC_COMBINER_LOCKS_SCHEDULED_ITEMS();
132 gpr_atm last = gpr_atm_full_fetch_add(&lock->state, STATE_ELEM_COUNT_LOW_BIT);
133 GRPC_COMBINER_TRACE(gpr_log(GPR_INFO,
134 "C:%p grpc_combiner_execute c=%p last=%" PRIdPTR,
135 lock, cl, last));
136 if (last == 1) {
137 GRPC_STATS_INC_COMBINER_LOCKS_INITIATED();
138 GPR_TIMER_MARK("combiner.initiated", 0);
139 gpr_atm_no_barrier_store(&lock->initiating_exec_ctx_or_null,
140 (gpr_atm)grpc_core::ExecCtx::Get());
141 // first element on this list: add it to the list of combiner locks
142 // executing within this exec_ctx
143 push_last_on_exec_ctx(lock);
144 } else {
145 // there may be a race with setting here: if that happens, we may delay
146 // offload for one or two actions, and that's fine
147 gpr_atm initiator =
148 gpr_atm_no_barrier_load(&lock->initiating_exec_ctx_or_null);
149 if (initiator != 0 &&
150 initiator != reinterpret_cast<gpr_atm>(grpc_core::ExecCtx::Get())) {
151 gpr_atm_no_barrier_store(&lock->initiating_exec_ctx_or_null, 0);
152 }
153 }
154 GPR_ASSERT(last & STATE_UNORPHANED); // ensure lock has not been destroyed
155 assert(cl->cb);
156 cl->error_data.error = error;
157 lock->queue.Push(cl->next_data.mpscq_node.get());
158 }
159
move_next()160 static void move_next() {
161 grpc_core::ExecCtx::Get()->combiner_data()->active_combiner =
162 grpc_core::ExecCtx::Get()
163 ->combiner_data()
164 ->active_combiner->next_combiner_on_this_exec_ctx;
165 if (grpc_core::ExecCtx::Get()->combiner_data()->active_combiner == nullptr) {
166 grpc_core::ExecCtx::Get()->combiner_data()->last_combiner = nullptr;
167 }
168 }
169
offload(void * arg,grpc_error *)170 static void offload(void* arg, grpc_error* /*error*/) {
171 grpc_core::Combiner* lock = static_cast<grpc_core::Combiner*>(arg);
172 push_last_on_exec_ctx(lock);
173 }
174
queue_offload(grpc_core::Combiner * lock)175 static void queue_offload(grpc_core::Combiner* lock) {
176 GRPC_STATS_INC_COMBINER_LOCKS_OFFLOADED();
177 move_next();
178 GRPC_COMBINER_TRACE(gpr_log(GPR_INFO, "C:%p queue_offload", lock));
179 grpc_core::Executor::Run(&lock->offload, GRPC_ERROR_NONE);
180 }
181
grpc_combiner_continue_exec_ctx()182 bool grpc_combiner_continue_exec_ctx() {
183 GPR_TIMER_SCOPE("combiner.continue_exec_ctx", 0);
184 if (grpc_core::ExecCtx::Get() == nullptr) {
185 return false;
186 }
187 grpc_core::Combiner* lock =
188 grpc_core::ExecCtx::Get()->combiner_data()->active_combiner;
189 if (lock == nullptr) {
190 return false;
191 }
192
193 bool contended =
194 gpr_atm_no_barrier_load(&lock->initiating_exec_ctx_or_null) == 0;
195
196 GRPC_COMBINER_TRACE(gpr_log(GPR_INFO,
197 "C:%p grpc_combiner_continue_exec_ctx "
198 "contended=%d "
199 "exec_ctx_ready_to_finish=%d "
200 "time_to_execute_final_list=%d",
201 lock, contended,
202 grpc_core::ExecCtx::Get()->IsReadyToFinish(),
203 lock->time_to_execute_final_list));
204
205 // offload only if all the following conditions are true:
206 // 1. the combiner is contended and has more than one closure to execute
207 // 2. the current execution context needs to finish as soon as possible
208 // 3. the current thread is not a worker for any background poller
209 // 4. the DEFAULT executor is threaded
210 if (contended && grpc_core::ExecCtx::Get()->IsReadyToFinish() &&
211 !grpc_iomgr_is_any_background_poller_thread() &&
212 grpc_core::Executor::IsThreadedDefault()) {
213 GPR_TIMER_MARK("offload_from_finished_exec_ctx", 0);
214 // this execution context wants to move on: schedule remaining work to be
215 // picked up on the executor
216 queue_offload(lock);
217 return true;
218 }
219
220 if (!lock->time_to_execute_final_list ||
221 // peek to see if something new has shown up, and execute that with
222 // priority
223 (gpr_atm_acq_load(&lock->state) >> 1) > 1) {
224 grpc_core::MultiProducerSingleConsumerQueue::Node* n = lock->queue.Pop();
225 GRPC_COMBINER_TRACE(
226 gpr_log(GPR_INFO, "C:%p maybe_finish_one n=%p", lock, n));
227 if (n == nullptr) {
228 // queue is in an inconsistent state: use this as a cue that we should
229 // go off and do something else for a while (and come back later)
230 GPR_TIMER_MARK("delay_busy", 0);
231 queue_offload(lock);
232 return true;
233 }
234 GPR_TIMER_SCOPE("combiner.exec1", 0);
235 grpc_closure* cl = reinterpret_cast<grpc_closure*>(n);
236 grpc_error* cl_err = cl->error_data.error;
237 #ifndef NDEBUG
238 cl->scheduled = false;
239 #endif
240 cl->cb(cl->cb_arg, cl_err);
241 GRPC_ERROR_UNREF(cl_err);
242 } else {
243 grpc_closure* c = lock->final_list.head;
244 GPR_ASSERT(c != nullptr);
245 grpc_closure_list_init(&lock->final_list);
246 int loops = 0;
247 while (c != nullptr) {
248 GPR_TIMER_SCOPE("combiner.exec_1final", 0);
249 GRPC_COMBINER_TRACE(
250 gpr_log(GPR_INFO, "C:%p execute_final[%d] c=%p", lock, loops, c));
251 grpc_closure* next = c->next_data.next;
252 grpc_error* error = c->error_data.error;
253 #ifndef NDEBUG
254 c->scheduled = false;
255 #endif
256 c->cb(c->cb_arg, error);
257 GRPC_ERROR_UNREF(error);
258 c = next;
259 }
260 }
261
262 GPR_TIMER_MARK("unref", 0);
263 move_next();
264 lock->time_to_execute_final_list = false;
265 gpr_atm old_state =
266 gpr_atm_full_fetch_add(&lock->state, -STATE_ELEM_COUNT_LOW_BIT);
267 GRPC_COMBINER_TRACE(
268 gpr_log(GPR_INFO, "C:%p finish old_state=%" PRIdPTR, lock, old_state));
269 // Define a macro to ease readability of the following switch statement.
270 #define OLD_STATE_WAS(orphaned, elem_count) \
271 (((orphaned) ? 0 : STATE_UNORPHANED) | \
272 ((elem_count)*STATE_ELEM_COUNT_LOW_BIT))
273 // Depending on what the previous state was, we need to perform different
274 // actions.
275 switch (old_state) {
276 default:
277 // we have multiple queued work items: just continue executing them
278 break;
279 case OLD_STATE_WAS(false, 2):
280 case OLD_STATE_WAS(true, 2):
281 // we're down to one queued item: if it's the final list we should do that
282 if (!grpc_closure_list_empty(lock->final_list)) {
283 lock->time_to_execute_final_list = true;
284 }
285 break;
286 case OLD_STATE_WAS(false, 1):
287 // had one count, one unorphaned --> unlocked unorphaned
288 return true;
289 case OLD_STATE_WAS(true, 1):
290 // and one count, one orphaned --> unlocked and orphaned
291 really_destroy(lock);
292 return true;
293 case OLD_STATE_WAS(false, 0):
294 case OLD_STATE_WAS(true, 0):
295 // these values are illegal - representing an already unlocked or
296 // deleted lock
297 GPR_UNREACHABLE_CODE(return true);
298 }
299 push_first_on_exec_ctx(lock);
300 return true;
301 }
302
303 static void enqueue_finally(void* closure, grpc_error* error);
304
combiner_finally_exec(grpc_core::Combiner * lock,grpc_closure * closure,grpc_error * error)305 static void combiner_finally_exec(grpc_core::Combiner* lock,
306 grpc_closure* closure, grpc_error* error) {
307 GPR_ASSERT(lock != nullptr);
308 GPR_TIMER_SCOPE("combiner.execute_finally", 0);
309 GRPC_STATS_INC_COMBINER_LOCKS_SCHEDULED_FINAL_ITEMS();
310 GRPC_COMBINER_TRACE(gpr_log(
311 GPR_INFO, "C:%p grpc_combiner_execute_finally c=%p; ac=%p", lock, closure,
312 grpc_core::ExecCtx::Get()->combiner_data()->active_combiner));
313 if (grpc_core::ExecCtx::Get()->combiner_data()->active_combiner != lock) {
314 GPR_TIMER_MARK("slowpath", 0);
315 // Using error_data.scratch to store the combiner so that it can be accessed
316 // in enqueue_finally.
317 closure->error_data.scratch = reinterpret_cast<uintptr_t>(lock);
318 lock->Run(GRPC_CLOSURE_CREATE(enqueue_finally, closure, nullptr), error);
319 return;
320 }
321
322 if (grpc_closure_list_empty(lock->final_list)) {
323 gpr_atm_full_fetch_add(&lock->state, STATE_ELEM_COUNT_LOW_BIT);
324 }
325 grpc_closure_list_append(&lock->final_list, closure, error);
326 }
327
enqueue_finally(void * closure,grpc_error * error)328 static void enqueue_finally(void* closure, grpc_error* error) {
329 grpc_closure* cl = static_cast<grpc_closure*>(closure);
330 combiner_finally_exec(
331 reinterpret_cast<grpc_core::Combiner*>(cl->error_data.scratch), cl,
332 GRPC_ERROR_REF(error));
333 }
334
335 namespace grpc_core {
Run(grpc_closure * closure,grpc_error * error)336 void Combiner::Run(grpc_closure* closure, grpc_error* error) {
337 combiner_exec(this, closure, error);
338 }
339
FinallyRun(grpc_closure * closure,grpc_error * error)340 void Combiner::FinallyRun(grpc_closure* closure, grpc_error* error) {
341 combiner_finally_exec(this, closure, error);
342 }
343 } // namespace grpc_core
344