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/port.h"
22
23 #ifdef GRPC_POSIX_SOCKET_EV
24
25 #include <string.h>
26
27 #include <grpc/support/alloc.h>
28 #include <grpc/support/log.h>
29 #include <grpc/support/string_util.h>
30
31 #include "src/core/lib/debug/trace.h"
32 #include "src/core/lib/gpr/useful.h"
33 #include "src/core/lib/gprpp/global_config.h"
34 #include "src/core/lib/iomgr/ev_epoll1_linux.h"
35 #include "src/core/lib/iomgr/ev_epollex_linux.h"
36 #include "src/core/lib/iomgr/ev_poll_posix.h"
37 #include "src/core/lib/iomgr/ev_posix.h"
38 #include "src/core/lib/iomgr/internal_errqueue.h"
39
40 GPR_GLOBAL_CONFIG_DEFINE_STRING(
41 grpc_poll_strategy, "all",
42 "Declares which polling engines to try when starting gRPC. "
43 "This is a comma-separated list of engines, which are tried in priority "
44 "order first -> last.")
45
46 grpc_core::DebugOnlyTraceFlag grpc_polling_trace(
47 false, "polling"); /* Disabled by default */
48
49 /* Traces fd create/close operations */
50 grpc_core::DebugOnlyTraceFlag grpc_fd_trace(false, "fd_trace");
51 grpc_core::DebugOnlyTraceFlag grpc_trace_fd_refcount(false, "fd_refcount");
52 grpc_core::DebugOnlyTraceFlag grpc_polling_api_trace(false, "polling_api");
53
54 // Polling API trace only enabled in debug builds
55 #ifndef NDEBUG
56 #define GRPC_POLLING_API_TRACE(format, ...) \
57 if (GRPC_TRACE_FLAG_ENABLED(grpc_polling_api_trace)) { \
58 gpr_log(GPR_INFO, "(polling-api) " format, __VA_ARGS__); \
59 }
60 #else
61 #define GRPC_POLLING_API_TRACE(...)
62 #endif // NDEBUG
63
64 /** Default poll() function - a pointer so that it can be overridden by some
65 * tests */
66 #ifndef GPR_AIX
67 grpc_poll_function_type grpc_poll_function = poll;
68 #else
aix_poll(struct pollfd fds[],nfds_t nfds,int timeout)69 int aix_poll(struct pollfd fds[], nfds_t nfds, int timeout) {
70 return poll(fds, nfds, timeout);
71 }
72 grpc_poll_function_type grpc_poll_function = aix_poll;
73 #endif // GPR_AIX
74
75 grpc_wakeup_fd grpc_global_wakeup_fd;
76
77 static const grpc_event_engine_vtable* g_event_engine = nullptr;
78 static const char* g_poll_strategy_name = nullptr;
79
80 typedef const grpc_event_engine_vtable* (*event_engine_factory_fn)(
81 bool explicit_request);
82
83 struct event_engine_factory {
84 const char* name;
85 event_engine_factory_fn factory;
86 };
87 namespace {
88
89 grpc_poll_function_type real_poll_function;
90
phony_poll(struct pollfd fds[],nfds_t nfds,int timeout)91 int phony_poll(struct pollfd fds[], nfds_t nfds, int timeout) {
92 if (timeout == 0) {
93 return real_poll_function(fds, nfds, 0);
94 } else {
95 gpr_log(GPR_ERROR, "Attempted a blocking poll when declared non-polling.");
96 GPR_ASSERT(false);
97 return -1;
98 }
99 }
100
init_non_polling(bool explicit_request)101 const grpc_event_engine_vtable* init_non_polling(bool explicit_request) {
102 if (!explicit_request) {
103 return nullptr;
104 }
105 // return the simplest engine as a phony but also override the poller
106 auto ret = grpc_init_poll_posix(explicit_request);
107 real_poll_function = grpc_poll_function;
108 grpc_poll_function = phony_poll;
109
110 return ret;
111 }
112 } // namespace
113
114 #define ENGINE_HEAD_CUSTOM "head_custom"
115 #define ENGINE_TAIL_CUSTOM "tail_custom"
116
117 // The global array of event-engine factories. Each entry is a pair with a name
118 // and an event-engine generator function (nullptr if there is no generator
119 // registered for this name). The middle entries are the engines predefined by
120 // open-source gRPC. The head entries represent an opportunity for specific
121 // high-priority custom pollers to be added by the initializer plugins of
122 // custom-built gRPC libraries. The tail entries represent the same, but for
123 // low-priority custom pollers. The actual poller selected is either the first
124 // available one in the list if no specific poller is requested, or the first
125 // specific poller that is requested by name in the GRPC_POLL_STRATEGY
126 // environment variable if that variable is set (which should be a
127 // comma-separated list of one or more event engine names)
128 static event_engine_factory g_factories[] = {
129 {ENGINE_HEAD_CUSTOM, nullptr}, {ENGINE_HEAD_CUSTOM, nullptr},
130 {ENGINE_HEAD_CUSTOM, nullptr}, {ENGINE_HEAD_CUSTOM, nullptr},
131 {"epollex", grpc_init_epollex_linux}, {"epoll1", grpc_init_epoll1_linux},
132 {"poll", grpc_init_poll_posix}, {"none", init_non_polling},
133 {ENGINE_TAIL_CUSTOM, nullptr}, {ENGINE_TAIL_CUSTOM, nullptr},
134 {ENGINE_TAIL_CUSTOM, nullptr}, {ENGINE_TAIL_CUSTOM, nullptr},
135 };
136
add(const char * beg,const char * end,char *** ss,size_t * ns)137 static void add(const char* beg, const char* end, char*** ss, size_t* ns) {
138 size_t n = *ns;
139 size_t np = n + 1;
140 char* s;
141 size_t len;
142 GPR_ASSERT(end >= beg);
143 len = static_cast<size_t>(end - beg);
144 s = static_cast<char*>(gpr_malloc(len + 1));
145 memcpy(s, beg, len);
146 s[len] = 0;
147 *ss = static_cast<char**>(gpr_realloc(*ss, sizeof(char**) * np));
148 (*ss)[n] = s;
149 *ns = np;
150 }
151
split(const char * s,char *** ss,size_t * ns)152 static void split(const char* s, char*** ss, size_t* ns) {
153 const char* c = strchr(s, ',');
154 if (c == nullptr) {
155 add(s, s + strlen(s), ss, ns);
156 } else {
157 add(s, c, ss, ns);
158 split(c + 1, ss, ns);
159 }
160 }
161
is(const char * want,const char * have)162 static bool is(const char* want, const char* have) {
163 return 0 == strcmp(want, "all") || 0 == strcmp(want, have);
164 }
165
try_engine(const char * engine)166 static void try_engine(const char* engine) {
167 for (size_t i = 0; i < GPR_ARRAY_SIZE(g_factories); i++) {
168 if (g_factories[i].factory != nullptr && is(engine, g_factories[i].name)) {
169 if ((g_event_engine = g_factories[i].factory(
170 0 == strcmp(engine, g_factories[i].name)))) {
171 g_poll_strategy_name = g_factories[i].name;
172 gpr_log(GPR_DEBUG, "Using polling engine: %s", g_factories[i].name);
173 return;
174 }
175 }
176 }
177 }
178
179 /* Call this before calling grpc_event_engine_init() */
grpc_register_event_engine_factory(const char * name,event_engine_factory_fn factory,bool add_at_head)180 void grpc_register_event_engine_factory(const char* name,
181 event_engine_factory_fn factory,
182 bool add_at_head) {
183 const char* custom_match =
184 add_at_head ? ENGINE_HEAD_CUSTOM : ENGINE_TAIL_CUSTOM;
185
186 // Overwrite an existing registration if already registered
187 for (size_t i = 0; i < GPR_ARRAY_SIZE(g_factories); i++) {
188 if (0 == strcmp(name, g_factories[i].name)) {
189 g_factories[i].factory = factory;
190 return;
191 }
192 }
193
194 // Otherwise fill in an available custom slot
195 for (size_t i = 0; i < GPR_ARRAY_SIZE(g_factories); i++) {
196 if (0 == strcmp(g_factories[i].name, custom_match)) {
197 g_factories[i].name = name;
198 g_factories[i].factory = factory;
199 return;
200 }
201 }
202
203 // Otherwise fail
204 GPR_ASSERT(false);
205 }
206
207 /*If grpc_event_engine_init() has been called, returns the poll_strategy_name.
208 * Otherwise, returns nullptr. */
grpc_get_poll_strategy_name()209 const char* grpc_get_poll_strategy_name() { return g_poll_strategy_name; }
210
grpc_event_engine_init(void)211 void grpc_event_engine_init(void) {
212 grpc_core::UniquePtr<char> value = GPR_GLOBAL_CONFIG_GET(grpc_poll_strategy);
213
214 char** strings = nullptr;
215 size_t nstrings = 0;
216 split(value.get(), &strings, &nstrings);
217
218 for (size_t i = 0; g_event_engine == nullptr && i < nstrings; i++) {
219 try_engine(strings[i]);
220 }
221
222 for (size_t i = 0; i < nstrings; i++) {
223 gpr_free(strings[i]);
224 }
225 gpr_free(strings);
226
227 if (g_event_engine == nullptr) {
228 gpr_log(GPR_ERROR, "No event engine could be initialized from %s",
229 value.get());
230 abort();
231 }
232 }
233
grpc_event_engine_shutdown(void)234 void grpc_event_engine_shutdown(void) {
235 g_event_engine->shutdown_engine();
236 g_event_engine = nullptr;
237 }
238
grpc_event_engine_can_track_errors(void)239 bool grpc_event_engine_can_track_errors(void) {
240 /* Only track errors if platform supports errqueue. */
241 if (grpc_core::kernel_supports_errqueue()) {
242 return g_event_engine->can_track_err;
243 }
244 return false;
245 }
246
grpc_event_engine_run_in_background(void)247 bool grpc_event_engine_run_in_background(void) {
248 // g_event_engine is nullptr when using a custom iomgr.
249 return g_event_engine != nullptr && g_event_engine->run_in_background;
250 }
251
grpc_fd_create(int fd,const char * name,bool track_err)252 grpc_fd* grpc_fd_create(int fd, const char* name, bool track_err) {
253 GRPC_POLLING_API_TRACE("fd_create(%d, %s, %d)", fd, name, track_err);
254 GRPC_FD_TRACE("fd_create(%d, %s, %d)", fd, name, track_err);
255 return g_event_engine->fd_create(
256 fd, name, track_err && grpc_event_engine_can_track_errors());
257 }
258
grpc_fd_wrapped_fd(grpc_fd * fd)259 int grpc_fd_wrapped_fd(grpc_fd* fd) {
260 return g_event_engine->fd_wrapped_fd(fd);
261 }
262
grpc_fd_orphan(grpc_fd * fd,grpc_closure * on_done,int * release_fd,const char * reason)263 void grpc_fd_orphan(grpc_fd* fd, grpc_closure* on_done, int* release_fd,
264 const char* reason) {
265 GRPC_POLLING_API_TRACE("fd_orphan(%d, %p, %p, %s)", grpc_fd_wrapped_fd(fd),
266 on_done, release_fd, reason);
267 GRPC_FD_TRACE("grpc_fd_orphan, fd:%d closed", grpc_fd_wrapped_fd(fd));
268
269 g_event_engine->fd_orphan(fd, on_done, release_fd, reason);
270 }
271
grpc_fd_shutdown(grpc_fd * fd,grpc_error_handle why)272 void grpc_fd_shutdown(grpc_fd* fd, grpc_error_handle why) {
273 GRPC_POLLING_API_TRACE("fd_shutdown(%d)", grpc_fd_wrapped_fd(fd));
274 GRPC_FD_TRACE("fd_shutdown(%d)", grpc_fd_wrapped_fd(fd));
275 g_event_engine->fd_shutdown(fd, why);
276 }
277
grpc_fd_is_shutdown(grpc_fd * fd)278 bool grpc_fd_is_shutdown(grpc_fd* fd) {
279 return g_event_engine->fd_is_shutdown(fd);
280 }
281
grpc_fd_notify_on_read(grpc_fd * fd,grpc_closure * closure)282 void grpc_fd_notify_on_read(grpc_fd* fd, grpc_closure* closure) {
283 g_event_engine->fd_notify_on_read(fd, closure);
284 }
285
grpc_fd_notify_on_write(grpc_fd * fd,grpc_closure * closure)286 void grpc_fd_notify_on_write(grpc_fd* fd, grpc_closure* closure) {
287 g_event_engine->fd_notify_on_write(fd, closure);
288 }
289
grpc_fd_notify_on_error(grpc_fd * fd,grpc_closure * closure)290 void grpc_fd_notify_on_error(grpc_fd* fd, grpc_closure* closure) {
291 g_event_engine->fd_notify_on_error(fd, closure);
292 }
293
grpc_fd_set_readable(grpc_fd * fd)294 void grpc_fd_set_readable(grpc_fd* fd) { g_event_engine->fd_set_readable(fd); }
295
grpc_fd_set_writable(grpc_fd * fd)296 void grpc_fd_set_writable(grpc_fd* fd) { g_event_engine->fd_set_writable(fd); }
297
grpc_fd_set_error(grpc_fd * fd)298 void grpc_fd_set_error(grpc_fd* fd) { g_event_engine->fd_set_error(fd); }
299
pollset_size(void)300 static size_t pollset_size(void) { return g_event_engine->pollset_size; }
301
pollset_init(grpc_pollset * pollset,gpr_mu ** mu)302 static void pollset_init(grpc_pollset* pollset, gpr_mu** mu) {
303 GRPC_POLLING_API_TRACE("pollset_init(%p)", pollset);
304 g_event_engine->pollset_init(pollset, mu);
305 }
306
pollset_shutdown(grpc_pollset * pollset,grpc_closure * closure)307 static void pollset_shutdown(grpc_pollset* pollset, grpc_closure* closure) {
308 GRPC_POLLING_API_TRACE("pollset_shutdown(%p)", pollset);
309 g_event_engine->pollset_shutdown(pollset, closure);
310 }
311
pollset_destroy(grpc_pollset * pollset)312 static void pollset_destroy(grpc_pollset* pollset) {
313 GRPC_POLLING_API_TRACE("pollset_destroy(%p)", pollset);
314 g_event_engine->pollset_destroy(pollset);
315 }
316
pollset_work(grpc_pollset * pollset,grpc_pollset_worker ** worker,grpc_millis deadline)317 static grpc_error_handle pollset_work(grpc_pollset* pollset,
318 grpc_pollset_worker** worker,
319 grpc_millis deadline) {
320 GRPC_POLLING_API_TRACE("pollset_work(%p, %" PRId64 ") begin", pollset,
321 deadline);
322 grpc_error_handle err =
323 g_event_engine->pollset_work(pollset, worker, deadline);
324 GRPC_POLLING_API_TRACE("pollset_work(%p, %" PRId64 ") end", pollset,
325 deadline);
326 return err;
327 }
328
pollset_kick(grpc_pollset * pollset,grpc_pollset_worker * specific_worker)329 static grpc_error_handle pollset_kick(grpc_pollset* pollset,
330 grpc_pollset_worker* specific_worker) {
331 GRPC_POLLING_API_TRACE("pollset_kick(%p, %p)", pollset, specific_worker);
332 return g_event_engine->pollset_kick(pollset, specific_worker);
333 }
334
grpc_pollset_add_fd(grpc_pollset * pollset,struct grpc_fd * fd)335 void grpc_pollset_add_fd(grpc_pollset* pollset, struct grpc_fd* fd) {
336 GRPC_POLLING_API_TRACE("pollset_add_fd(%p, %d)", pollset,
337 grpc_fd_wrapped_fd(fd));
338 g_event_engine->pollset_add_fd(pollset, fd);
339 }
340
pollset_global_init()341 void pollset_global_init() {}
pollset_global_shutdown()342 void pollset_global_shutdown() {}
343
344 grpc_pollset_vtable grpc_posix_pollset_vtable = {
345 pollset_global_init, pollset_global_shutdown,
346 pollset_init, pollset_shutdown,
347 pollset_destroy, pollset_work,
348 pollset_kick, pollset_size};
349
pollset_set_create(void)350 static grpc_pollset_set* pollset_set_create(void) {
351 grpc_pollset_set* pss = g_event_engine->pollset_set_create();
352 GRPC_POLLING_API_TRACE("pollset_set_create(%p)", pss);
353 return pss;
354 }
355
pollset_set_destroy(grpc_pollset_set * pollset_set)356 static void pollset_set_destroy(grpc_pollset_set* pollset_set) {
357 GRPC_POLLING_API_TRACE("pollset_set_destroy(%p)", pollset_set);
358 g_event_engine->pollset_set_destroy(pollset_set);
359 }
360
pollset_set_add_pollset(grpc_pollset_set * pollset_set,grpc_pollset * pollset)361 static void pollset_set_add_pollset(grpc_pollset_set* pollset_set,
362 grpc_pollset* pollset) {
363 GRPC_POLLING_API_TRACE("pollset_set_add_pollset(%p, %p)", pollset_set,
364 pollset);
365 g_event_engine->pollset_set_add_pollset(pollset_set, pollset);
366 }
367
pollset_set_del_pollset(grpc_pollset_set * pollset_set,grpc_pollset * pollset)368 static void pollset_set_del_pollset(grpc_pollset_set* pollset_set,
369 grpc_pollset* pollset) {
370 GRPC_POLLING_API_TRACE("pollset_set_del_pollset(%p, %p)", pollset_set,
371 pollset);
372 g_event_engine->pollset_set_del_pollset(pollset_set, pollset);
373 }
374
pollset_set_add_pollset_set(grpc_pollset_set * bag,grpc_pollset_set * item)375 static void pollset_set_add_pollset_set(grpc_pollset_set* bag,
376 grpc_pollset_set* item) {
377 GRPC_POLLING_API_TRACE("pollset_set_add_pollset_set(%p, %p)", bag, item);
378 g_event_engine->pollset_set_add_pollset_set(bag, item);
379 }
380
pollset_set_del_pollset_set(grpc_pollset_set * bag,grpc_pollset_set * item)381 static void pollset_set_del_pollset_set(grpc_pollset_set* bag,
382 grpc_pollset_set* item) {
383 GRPC_POLLING_API_TRACE("pollset_set_del_pollset_set(%p, %p)", bag, item);
384 g_event_engine->pollset_set_del_pollset_set(bag, item);
385 }
386
387 grpc_pollset_set_vtable grpc_posix_pollset_set_vtable = {
388 pollset_set_create, pollset_set_destroy,
389 pollset_set_add_pollset, pollset_set_del_pollset,
390 pollset_set_add_pollset_set, pollset_set_del_pollset_set};
391
grpc_pollset_set_add_fd(grpc_pollset_set * pollset_set,grpc_fd * fd)392 void grpc_pollset_set_add_fd(grpc_pollset_set* pollset_set, grpc_fd* fd) {
393 GRPC_POLLING_API_TRACE("pollset_set_add_fd(%p, %d)", pollset_set,
394 grpc_fd_wrapped_fd(fd));
395 g_event_engine->pollset_set_add_fd(pollset_set, fd);
396 }
397
grpc_pollset_set_del_fd(grpc_pollset_set * pollset_set,grpc_fd * fd)398 void grpc_pollset_set_del_fd(grpc_pollset_set* pollset_set, grpc_fd* fd) {
399 GRPC_POLLING_API_TRACE("pollset_set_del_fd(%p, %d)", pollset_set,
400 grpc_fd_wrapped_fd(fd));
401 g_event_engine->pollset_set_del_fd(pollset_set, fd);
402 }
403
grpc_is_any_background_poller_thread(void)404 bool grpc_is_any_background_poller_thread(void) {
405 return g_event_engine->is_any_background_poller_thread();
406 }
407
grpc_add_closure_to_background_poller(grpc_closure * closure,grpc_error_handle error)408 bool grpc_add_closure_to_background_poller(grpc_closure* closure,
409 grpc_error_handle error) {
410 return g_event_engine->add_closure_to_background_poller(closure, error);
411 }
412
grpc_shutdown_background_closure(void)413 void grpc_shutdown_background_closure(void) {
414 g_event_engine->shutdown_background_closure();
415 }
416
417 #endif // GRPC_POSIX_SOCKET_EV
418