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