1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
|
/*
*
* Copyright 2015 gRPC authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*
*/
#include "src/core/lib/iomgr/exec_ctx.h"
#include <grpc/support/log.h>
#include <grpc/support/sync.h>
#include <grpc/support/thd.h>
#include "src/core/lib/iomgr/combiner.h"
#include "src/core/lib/profiling/timers.h"
bool grpc_exec_ctx_ready_to_finish(grpc_exec_ctx *exec_ctx) {
if ((exec_ctx->flags & GRPC_EXEC_CTX_FLAG_IS_FINISHED) == 0) {
if (exec_ctx->check_ready_to_finish(exec_ctx,
exec_ctx->check_ready_to_finish_arg)) {
exec_ctx->flags |= GRPC_EXEC_CTX_FLAG_IS_FINISHED;
return true;
}
return false;
} else {
return true;
}
}
bool grpc_never_ready_to_finish(grpc_exec_ctx *exec_ctx, void *arg_ignored) {
return false;
}
bool grpc_always_ready_to_finish(grpc_exec_ctx *exec_ctx, void *arg_ignored) {
return true;
}
bool grpc_exec_ctx_has_work(grpc_exec_ctx *exec_ctx) {
return exec_ctx->active_combiner != NULL ||
!grpc_closure_list_empty(exec_ctx->closure_list);
}
void grpc_exec_ctx_finish(grpc_exec_ctx *exec_ctx) {
exec_ctx->flags |= GRPC_EXEC_CTX_FLAG_IS_FINISHED;
grpc_exec_ctx_flush(exec_ctx);
}
static void exec_ctx_run(grpc_exec_ctx *exec_ctx, grpc_closure *closure,
grpc_error *error) {
#ifndef NDEBUG
closure->scheduled = false;
if (GRPC_TRACER_ON(grpc_trace_closure)) {
gpr_log(GPR_DEBUG, "running closure %p: created [%s:%d]: %s [%s:%d]",
closure, closure->file_created, closure->line_created,
closure->run ? "run" : "scheduled", closure->file_initiated,
closure->line_initiated);
}
#endif
closure->cb(exec_ctx, closure->cb_arg, error);
#ifndef NDEBUG
if (GRPC_TRACER_ON(grpc_trace_closure)) {
gpr_log(GPR_DEBUG, "closure %p finished", closure);
}
#endif
GRPC_ERROR_UNREF(error);
}
bool grpc_exec_ctx_flush(grpc_exec_ctx *exec_ctx) {
bool did_something = 0;
GPR_TIMER_BEGIN("grpc_exec_ctx_flush", 0);
for (;;) {
if (!grpc_closure_list_empty(exec_ctx->closure_list)) {
grpc_closure *c = exec_ctx->closure_list.head;
exec_ctx->closure_list.head = exec_ctx->closure_list.tail = NULL;
while (c != NULL) {
grpc_closure *next = c->next_data.next;
grpc_error *error = c->error_data.error;
did_something = true;
exec_ctx_run(exec_ctx, c, error);
c = next;
}
} else if (!grpc_combiner_continue_exec_ctx(exec_ctx)) {
break;
}
}
GPR_ASSERT(exec_ctx->active_combiner == NULL);
GPR_TIMER_END("grpc_exec_ctx_flush", 0);
return did_something;
}
static void exec_ctx_sched(grpc_exec_ctx *exec_ctx, grpc_closure *closure,
grpc_error *error) {
grpc_closure_list_append(&exec_ctx->closure_list, closure, error);
}
static gpr_timespec
g_start_time[GPR_TIMESPAN + 1]; // assumes GPR_TIMESPAN is the
// last enum value in
// gpr_clock_type
void grpc_exec_ctx_global_init(void) {
for (int i = 0; i < GPR_TIMESPAN; i++) {
g_start_time[i] = gpr_now((gpr_clock_type)i);
}
// allows uniform treatment in conversion functions
g_start_time[GPR_TIMESPAN] = gpr_time_0(GPR_TIMESPAN);
}
void grpc_exec_ctx_global_shutdown(void) {}
static gpr_atm timespec_to_atm_round_down(gpr_timespec ts) {
ts = gpr_time_sub(ts, g_start_time[ts.clock_type]);
double x =
GPR_MS_PER_SEC * (double)ts.tv_sec + (double)ts.tv_nsec / GPR_NS_PER_MS;
if (x < 0) return 0;
if (x > GPR_ATM_MAX) return GPR_ATM_MAX;
return (gpr_atm)x;
}
static gpr_atm timespec_to_atm_round_up(gpr_timespec ts) {
ts = gpr_time_sub(ts, g_start_time[ts.clock_type]);
double x = GPR_MS_PER_SEC * (double)ts.tv_sec +
(double)ts.tv_nsec / GPR_NS_PER_MS +
(double)(GPR_NS_PER_SEC - 1) / (double)GPR_NS_PER_SEC;
if (x < 0) return 0;
if (x > GPR_ATM_MAX) return GPR_ATM_MAX;
return (gpr_atm)x;
}
grpc_millis grpc_exec_ctx_now(grpc_exec_ctx *exec_ctx) {
if (!exec_ctx->now_is_valid) {
exec_ctx->now = timespec_to_atm_round_down(gpr_now(GPR_CLOCK_MONOTONIC));
exec_ctx->now_is_valid = true;
}
return exec_ctx->now;
}
void grpc_exec_ctx_invalidate_now(grpc_exec_ctx *exec_ctx) {
exec_ctx->now_is_valid = false;
}
gpr_timespec grpc_millis_to_timespec(grpc_millis millis,
gpr_clock_type clock_type) {
// special-case infinities as grpc_millis can be 32bit on some platforms
// while gpr_time_from_millis always takes an int64_t.
if (millis == GRPC_MILLIS_INF_FUTURE) {
return gpr_inf_future(clock_type);
}
if (millis == GRPC_MILLIS_INF_PAST) {
return gpr_inf_past(clock_type);
}
if (clock_type == GPR_TIMESPAN) {
return gpr_time_from_millis(millis, GPR_TIMESPAN);
}
return gpr_time_add(g_start_time[clock_type],
gpr_time_from_millis(millis, GPR_TIMESPAN));
}
grpc_millis grpc_timespec_to_millis_round_down(gpr_timespec ts) {
return timespec_to_atm_round_down(ts);
}
grpc_millis grpc_timespec_to_millis_round_up(gpr_timespec ts) {
return timespec_to_atm_round_up(ts);
}
static const grpc_closure_scheduler_vtable exec_ctx_scheduler_vtable = {
exec_ctx_run, exec_ctx_sched, "exec_ctx"};
static grpc_closure_scheduler exec_ctx_scheduler = {&exec_ctx_scheduler_vtable};
grpc_closure_scheduler *grpc_schedule_on_exec_ctx = &exec_ctx_scheduler;
|
