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 * error)30 static void exec_ctx_run(grpc_closure* closure, grpc_error* 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 * error)49 static void exec_ctx_sched(grpc_closure* closure, grpc_error* 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
timespec_to_millis_round_down(gpr_timespec ts)56 static grpc_millis timespec_to_millis_round_down(gpr_timespec ts) {
57 ts = gpr_time_sub(ts, g_start_time);
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 > GRPC_MILLIS_INF_FUTURE) return GRPC_MILLIS_INF_FUTURE;
62 return static_cast<grpc_millis>(x);
63 }
64
timespec_to_millis_round_up(gpr_timespec ts)65 static grpc_millis timespec_to_millis_round_up(gpr_timespec ts) {
66 ts = gpr_time_sub(ts, g_start_time);
67 double x = GPR_MS_PER_SEC * static_cast<double>(ts.tv_sec) +
68 static_cast<double>(ts.tv_nsec) / GPR_NS_PER_MS +
69 static_cast<double>(GPR_NS_PER_SEC - 1) /
70 static_cast<double>(GPR_NS_PER_SEC);
71 if (x < 0) return 0;
72 if (x > GRPC_MILLIS_INF_FUTURE) return GRPC_MILLIS_INF_FUTURE;
73 return static_cast<grpc_millis>(x);
74 }
75
grpc_millis_to_timespec(grpc_millis millis,gpr_clock_type clock_type)76 gpr_timespec grpc_millis_to_timespec(grpc_millis millis,
77 gpr_clock_type clock_type) {
78 // special-case infinities as grpc_millis can be 32bit on some platforms
79 // while gpr_time_from_millis always takes an int64_t.
80 if (millis == GRPC_MILLIS_INF_FUTURE) {
81 return gpr_inf_future(clock_type);
82 }
83 if (millis == GRPC_MILLIS_INF_PAST) {
84 return gpr_inf_past(clock_type);
85 }
86
87 if (clock_type == GPR_TIMESPAN) {
88 return gpr_time_from_millis(millis, GPR_TIMESPAN);
89 }
90 return gpr_time_add(gpr_convert_clock_type(g_start_time, clock_type),
91 gpr_time_from_millis(millis, GPR_TIMESPAN));
92 }
93
grpc_timespec_to_millis_round_down(gpr_timespec ts)94 grpc_millis grpc_timespec_to_millis_round_down(gpr_timespec ts) {
95 return timespec_to_millis_round_down(
96 gpr_convert_clock_type(ts, g_start_time.clock_type));
97 }
98
grpc_timespec_to_millis_round_up(gpr_timespec ts)99 grpc_millis grpc_timespec_to_millis_round_up(gpr_timespec ts) {
100 return timespec_to_millis_round_up(
101 gpr_convert_clock_type(ts, g_start_time.clock_type));
102 }
103
104 static const grpc_closure_scheduler_vtable exec_ctx_scheduler_vtable = {
105 exec_ctx_run, exec_ctx_sched, "exec_ctx"};
106 static grpc_closure_scheduler exec_ctx_scheduler = {&exec_ctx_scheduler_vtable};
107 grpc_closure_scheduler* grpc_schedule_on_exec_ctx = &exec_ctx_scheduler;
108
109 namespace grpc_core {
110 GPR_TLS_CLASS_DEF(ExecCtx::exec_ctx_);
111
112 // WARNING: for testing purposes only!
TestOnlyGlobalInit(gpr_timespec new_val)113 void ExecCtx::TestOnlyGlobalInit(gpr_timespec new_val) {
114 g_start_time = new_val;
115 gpr_tls_init(&exec_ctx_);
116 }
117
GlobalInit(void)118 void ExecCtx::GlobalInit(void) {
119 g_start_time = gpr_now(GPR_CLOCK_MONOTONIC);
120 gpr_tls_init(&exec_ctx_);
121 }
122
Flush()123 bool ExecCtx::Flush() {
124 bool did_something = 0;
125 GPR_TIMER_SCOPE("grpc_exec_ctx_flush", 0);
126 for (;;) {
127 if (!grpc_closure_list_empty(closure_list_)) {
128 grpc_closure* c = closure_list_.head;
129 closure_list_.head = closure_list_.tail = nullptr;
130 while (c != nullptr) {
131 grpc_closure* next = c->next_data.next;
132 grpc_error* error = c->error_data.error;
133 did_something = true;
134 exec_ctx_run(c, error);
135 c = next;
136 }
137 } else if (!grpc_combiner_continue_exec_ctx()) {
138 break;
139 }
140 }
141 GPR_ASSERT(combiner_data_.active_combiner == nullptr);
142 return did_something;
143 }
144
Now()145 grpc_millis ExecCtx::Now() {
146 if (!now_is_valid_) {
147 now_ = timespec_to_millis_round_down(gpr_now(GPR_CLOCK_MONOTONIC));
148 now_is_valid_ = true;
149 }
150 return now_;
151 }
152
153 } // namespace grpc_core
154