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/exec_ctx.h"
22
23 #include "absl/strings/str_format.h"
24
25 #include <grpc/support/log.h>
26 #include <grpc/support/sync.h>
27
28 #include "src/core/lib/gprpp/crash.h"
29 #include "src/core/lib/iomgr/combiner.h"
30 #include "src/core/lib/iomgr/error.h"
31
exec_ctx_run(grpc_closure * closure)32 static void exec_ctx_run(grpc_closure* closure) {
33 #ifndef NDEBUG
34 closure->scheduled = false;
35 if (grpc_trace_closure.enabled()) {
36 gpr_log(GPR_DEBUG, "running closure %p: created [%s:%d]: %s [%s:%d]",
37 closure, closure->file_created, closure->line_created,
38 closure->run ? "run" : "scheduled", closure->file_initiated,
39 closure->line_initiated);
40 }
41 #endif
42 grpc_error_handle error =
43 grpc_core::internal::StatusMoveFromHeapPtr(closure->error_data.error);
44 closure->error_data.error = 0;
45 closure->cb(closure->cb_arg, std::move(error));
46 #ifndef NDEBUG
47 if (grpc_trace_closure.enabled()) {
48 gpr_log(GPR_DEBUG, "closure %p finished", closure);
49 }
50 #endif
51 }
52
exec_ctx_sched(grpc_closure * closure)53 static void exec_ctx_sched(grpc_closure* closure) {
54 grpc_closure_list_append(grpc_core::ExecCtx::Get()->closure_list(), closure);
55 }
56
57 namespace grpc_core {
58
59 #if !defined(_WIN32) || !defined(_DLL)
60 thread_local ExecCtx* ExecCtx::exec_ctx_;
61 thread_local ApplicationCallbackExecCtx*
62 ApplicationCallbackExecCtx::callback_exec_ctx_;
63 #else // _WIN32
64 ExecCtx*& ExecCtx::exec_ctx() {
65 static thread_local ExecCtx* exec_ctx;
66 return exec_ctx;
67 }
68
69 ApplicationCallbackExecCtx*& ApplicationCallbackExecCtx::callback_exec_ctx() {
70 static thread_local ApplicationCallbackExecCtx* callback_exec_ctx;
71 return callback_exec_ctx;
72 }
73 #endif // _WIN32
74
Flush()75 bool ExecCtx::Flush() {
76 bool did_something = false;
77 for (;;) {
78 if (!grpc_closure_list_empty(closure_list_)) {
79 grpc_closure* c = closure_list_.head;
80 closure_list_.head = closure_list_.tail = nullptr;
81 while (c != nullptr) {
82 grpc_closure* next = c->next_data.next;
83 did_something = true;
84 exec_ctx_run(c);
85 c = next;
86 }
87 } else if (!grpc_combiner_continue_exec_ctx()) {
88 break;
89 }
90 }
91 GPR_ASSERT(combiner_data_.active_combiner == nullptr);
92 return did_something;
93 }
94
Run(const DebugLocation & location,grpc_closure * closure,grpc_error_handle error)95 void ExecCtx::Run(const DebugLocation& location, grpc_closure* closure,
96 grpc_error_handle error) {
97 (void)location;
98 if (closure == nullptr) {
99 return;
100 }
101 #ifndef NDEBUG
102 if (closure->scheduled) {
103 Crash(absl::StrFormat(
104 "Closure already scheduled. (closure: %p, created: [%s:%d], "
105 "previously scheduled at: [%s: %d], newly scheduled at [%s: %d]",
106 closure, closure->file_created, closure->line_created,
107 closure->file_initiated, closure->line_initiated, location.file(),
108 location.line()));
109 }
110 closure->scheduled = true;
111 closure->file_initiated = location.file();
112 closure->line_initiated = location.line();
113 closure->run = false;
114 GPR_ASSERT(closure->cb != nullptr);
115 #endif
116 closure->error_data.error = internal::StatusAllocHeapPtr(error);
117 exec_ctx_sched(closure);
118 }
119
RunList(const DebugLocation & location,grpc_closure_list * list)120 void ExecCtx::RunList(const DebugLocation& location, grpc_closure_list* list) {
121 (void)location;
122 grpc_closure* c = list->head;
123 while (c != nullptr) {
124 grpc_closure* next = c->next_data.next;
125 #ifndef NDEBUG
126 if (c->scheduled) {
127 Crash(absl::StrFormat(
128 "Closure already scheduled. (closure: %p, created: [%s:%d], "
129 "previously scheduled at: [%s: %d], newly scheduled at [%s:%d]",
130 c, c->file_created, c->line_created, c->file_initiated,
131 c->line_initiated, location.file(), location.line()));
132 }
133 c->scheduled = true;
134 c->file_initiated = location.file();
135 c->line_initiated = location.line();
136 c->run = false;
137 GPR_ASSERT(c->cb != nullptr);
138 #endif
139 exec_ctx_sched(c);
140 c = next;
141 }
142 list->head = list->tail = nullptr;
143 }
144
145 } // namespace grpc_core
146