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 <grpc/support/log.h>
24 #include <grpc/support/sync.h>
25
26 #include "src/core/lib/gprpp/thd.h"
27 #include "src/core/lib/iomgr/combiner.h"
28 #include "src/core/lib/profiling/timers.h"
29
exec_ctx_run(grpc_closure * closure,grpc_error_handle error)30 static void exec_ctx_run(grpc_closure* closure, grpc_error_handle error) {
31 #ifndef NDEBUG
32 closure->scheduled = false;
33 if (grpc_trace_closure.enabled()) {
34 gpr_log(GPR_DEBUG, "running closure %p: created [%s:%d]: %s [%s:%d]",
35 closure, closure->file_created, closure->line_created,
36 closure->run ? "run" : "scheduled", closure->file_initiated,
37 closure->line_initiated);
38 }
39 #endif
40 closure->cb(closure->cb_arg, error);
41 #ifndef NDEBUG
42 if (grpc_trace_closure.enabled()) {
43 gpr_log(GPR_DEBUG, "closure %p finished", closure);
44 }
45 #endif
46 GRPC_ERROR_UNREF(error);
47 }
48
exec_ctx_sched(grpc_closure * closure,grpc_error_handle error)49 static void exec_ctx_sched(grpc_closure* closure, grpc_error_handle error) {
50 grpc_closure_list_append(grpc_core::ExecCtx::Get()->closure_list(), closure,
51 error);
52 }
53
54 static gpr_timespec g_start_time;
55 static gpr_cycle_counter g_start_cycle;
56
timespan_to_millis_round_down(gpr_timespec ts)57 static grpc_millis timespan_to_millis_round_down(gpr_timespec ts) {
58 double x = GPR_MS_PER_SEC * static_cast<double>(ts.tv_sec) +
59 static_cast<double>(ts.tv_nsec) / GPR_NS_PER_MS;
60 if (x < 0) return 0;
61 if (x > static_cast<double>(GRPC_MILLIS_INF_FUTURE)) {
62 return GRPC_MILLIS_INF_FUTURE;
63 }
64 return static_cast<grpc_millis>(x);
65 }
66
timespec_to_millis_round_down(gpr_timespec ts)67 static grpc_millis timespec_to_millis_round_down(gpr_timespec ts) {
68 return timespan_to_millis_round_down(gpr_time_sub(ts, g_start_time));
69 }
70
timespan_to_millis_round_up(gpr_timespec ts)71 static grpc_millis timespan_to_millis_round_up(gpr_timespec ts) {
72 double x = GPR_MS_PER_SEC * static_cast<double>(ts.tv_sec) +
73 static_cast<double>(ts.tv_nsec) / GPR_NS_PER_MS +
74 static_cast<double>(GPR_NS_PER_SEC - 1) /
75 static_cast<double>(GPR_NS_PER_SEC);
76 if (x < 0) return 0;
77 if (x > static_cast<double>(GRPC_MILLIS_INF_FUTURE)) {
78 return GRPC_MILLIS_INF_FUTURE;
79 }
80 return static_cast<grpc_millis>(x);
81 }
82
timespec_to_millis_round_up(gpr_timespec ts)83 static grpc_millis timespec_to_millis_round_up(gpr_timespec ts) {
84 return timespan_to_millis_round_up(gpr_time_sub(ts, g_start_time));
85 }
86
grpc_millis_to_timespec(grpc_millis millis,gpr_clock_type clock_type)87 gpr_timespec grpc_millis_to_timespec(grpc_millis millis,
88 gpr_clock_type clock_type) {
89 // special-case infinities as grpc_millis can be 32bit on some platforms
90 // while gpr_time_from_millis always takes an int64_t.
91 if (millis == GRPC_MILLIS_INF_FUTURE) {
92 return gpr_inf_future(clock_type);
93 }
94 if (millis == GRPC_MILLIS_INF_PAST) {
95 return gpr_inf_past(clock_type);
96 }
97
98 if (clock_type == GPR_TIMESPAN) {
99 return gpr_time_from_millis(millis, GPR_TIMESPAN);
100 }
101 return gpr_time_add(gpr_convert_clock_type(g_start_time, clock_type),
102 gpr_time_from_millis(millis, GPR_TIMESPAN));
103 }
104
grpc_timespec_to_millis_round_down(gpr_timespec ts)105 grpc_millis grpc_timespec_to_millis_round_down(gpr_timespec ts) {
106 return timespec_to_millis_round_down(
107 gpr_convert_clock_type(ts, g_start_time.clock_type));
108 }
109
grpc_timespec_to_millis_round_up(gpr_timespec ts)110 grpc_millis grpc_timespec_to_millis_round_up(gpr_timespec ts) {
111 return timespec_to_millis_round_up(
112 gpr_convert_clock_type(ts, g_start_time.clock_type));
113 }
114
grpc_cycle_counter_to_millis_round_down(gpr_cycle_counter cycles)115 grpc_millis grpc_cycle_counter_to_millis_round_down(gpr_cycle_counter cycles) {
116 return timespan_to_millis_round_down(
117 gpr_cycle_counter_sub(cycles, g_start_cycle));
118 }
119
grpc_cycle_counter_to_millis_round_up(gpr_cycle_counter cycles)120 grpc_millis grpc_cycle_counter_to_millis_round_up(gpr_cycle_counter cycles) {
121 return timespan_to_millis_round_up(
122 gpr_cycle_counter_sub(cycles, g_start_cycle));
123 }
124
125 namespace grpc_core {
126 GPR_TLS_CLASS_DEF(ExecCtx::exec_ctx_);
127 GPR_TLS_CLASS_DEF(ApplicationCallbackExecCtx::callback_exec_ctx_);
128
129 // WARNING: for testing purposes only!
TestOnlyGlobalInit(gpr_timespec new_val)130 void ExecCtx::TestOnlyGlobalInit(gpr_timespec new_val) {
131 g_start_time = new_val;
132 gpr_tls_init(&exec_ctx_);
133 }
134
GlobalInit(void)135 void ExecCtx::GlobalInit(void) {
136 // gpr_now(GPR_CLOCK_MONOTONIC) incurs a syscall. We don't actually know the
137 // exact cycle the time was captured, so we use the average of cycles before
138 // and after the syscall as the starting cycle.
139 const gpr_cycle_counter cycle_before = gpr_get_cycle_counter();
140 g_start_time = gpr_now(GPR_CLOCK_MONOTONIC);
141 const gpr_cycle_counter cycle_after = gpr_get_cycle_counter();
142 g_start_cycle = (cycle_before + cycle_after) / 2;
143 gpr_tls_init(&exec_ctx_);
144 }
145
Flush()146 bool ExecCtx::Flush() {
147 bool did_something = false;
148 GPR_TIMER_SCOPE("grpc_exec_ctx_flush", 0);
149 for (;;) {
150 if (!grpc_closure_list_empty(closure_list_)) {
151 grpc_closure* c = closure_list_.head;
152 closure_list_.head = closure_list_.tail = nullptr;
153 while (c != nullptr) {
154 grpc_closure* next = c->next_data.next;
155 grpc_error_handle error = c->error_data.error;
156 did_something = true;
157 exec_ctx_run(c, error);
158 c = next;
159 }
160 } else if (!grpc_combiner_continue_exec_ctx()) {
161 break;
162 }
163 }
164 GPR_ASSERT(combiner_data_.active_combiner == nullptr);
165 return did_something;
166 }
167
Now()168 grpc_millis ExecCtx::Now() {
169 if (!now_is_valid_) {
170 now_ = timespec_to_millis_round_down(gpr_now(GPR_CLOCK_MONOTONIC));
171 now_is_valid_ = true;
172 }
173 return now_;
174 }
175
Run(const DebugLocation & location,grpc_closure * closure,grpc_error_handle error)176 void ExecCtx::Run(const DebugLocation& location, grpc_closure* closure,
177 grpc_error_handle error) {
178 (void)location;
179 if (closure == nullptr) {
180 GRPC_ERROR_UNREF(error);
181 return;
182 }
183 #ifndef NDEBUG
184 if (closure->scheduled) {
185 gpr_log(GPR_ERROR,
186 "Closure already scheduled. (closure: %p, created: [%s:%d], "
187 "previously scheduled at: [%s: %d], newly scheduled at [%s: %d]",
188 closure, closure->file_created, closure->line_created,
189 closure->file_initiated, closure->line_initiated, location.file(),
190 location.line());
191 abort();
192 }
193 closure->scheduled = true;
194 closure->file_initiated = location.file();
195 closure->line_initiated = location.line();
196 closure->run = false;
197 GPR_ASSERT(closure->cb != nullptr);
198 #endif
199 exec_ctx_sched(closure, error);
200 }
201
RunList(const DebugLocation & location,grpc_closure_list * list)202 void ExecCtx::RunList(const DebugLocation& location, grpc_closure_list* list) {
203 (void)location;
204 grpc_closure* c = list->head;
205 while (c != nullptr) {
206 grpc_closure* next = c->next_data.next;
207 #ifndef NDEBUG
208 if (c->scheduled) {
209 gpr_log(GPR_ERROR,
210 "Closure already scheduled. (closure: %p, created: [%s:%d], "
211 "previously scheduled at: [%s: %d], newly scheduled at [%s:%d]",
212 c, c->file_created, c->line_created, c->file_initiated,
213 c->line_initiated, location.file(), location.line());
214 abort();
215 }
216 c->scheduled = true;
217 c->file_initiated = location.file();
218 c->line_initiated = location.line();
219 c->run = false;
220 GPR_ASSERT(c->cb != nullptr);
221 #endif
222 exec_ctx_sched(c, c->error_data.error);
223 c = next;
224 }
225 list->head = list->tail = nullptr;
226 }
227
228 } // namespace grpc_core
229