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