blob: 6a3c2654a8c691f48c6853f24f73e83d78d27da5 [file] [log] [blame]
Craig Tillerc67cc992017-04-27 10:15:51 -07001/*
2 *
Jan Tattermusch7897ae92017-06-07 22:57:36 +02003 * Copyright 2017 gRPC authors.
Craig Tillerc67cc992017-04-27 10:15:51 -07004 *
Jan Tattermusch7897ae92017-06-07 22:57:36 +02005 * 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
Craig Tillerc67cc992017-04-27 10:15:51 -07008 *
Jan Tattermusch7897ae92017-06-07 22:57:36 +02009 * http://www.apache.org/licenses/LICENSE-2.0
Craig Tillerc67cc992017-04-27 10:15:51 -070010 *
Jan Tattermusch7897ae92017-06-07 22:57:36 +020011 * 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.
Craig Tillerc67cc992017-04-27 10:15:51 -070016 *
17 */
18
19#include "src/core/lib/iomgr/port.h"
20
21/* This polling engine is only relevant on linux kernels supporting epoll() */
22#ifdef GRPC_LINUX_EPOLL
23
Craig Tiller4509c472017-04-27 19:05:13 +000024#include "src/core/lib/iomgr/ev_epoll1_linux.h"
Craig Tillerc67cc992017-04-27 10:15:51 -070025
26#include <assert.h>
27#include <errno.h>
28#include <poll.h>
29#include <pthread.h>
30#include <string.h>
31#include <sys/epoll.h>
32#include <sys/socket.h>
33#include <unistd.h>
34
35#include <grpc/support/alloc.h>
Craig Tiller6de05932017-04-28 09:17:38 -070036#include <grpc/support/cpu.h>
Craig Tillerc67cc992017-04-27 10:15:51 -070037#include <grpc/support/log.h>
38#include <grpc/support/string_util.h>
39#include <grpc/support/tls.h>
40#include <grpc/support/useful.h>
41
Craig Tillerb4bb1cd2017-07-20 14:18:17 -070042#include "src/core/lib/debug/stats.h"
Craig Tillerc67cc992017-04-27 10:15:51 -070043#include "src/core/lib/iomgr/ev_posix.h"
44#include "src/core/lib/iomgr/iomgr_internal.h"
45#include "src/core/lib/iomgr/lockfree_event.h"
Craig Tillerc67cc992017-04-27 10:15:51 -070046#include "src/core/lib/iomgr/wakeup_fd_posix.h"
Craig Tillerc67cc992017-04-27 10:15:51 -070047#include "src/core/lib/profiling/timers.h"
48#include "src/core/lib/support/block_annotate.h"
Craig Tillerb89bac02017-05-26 15:20:32 +000049#include "src/core/lib/support/string.h"
Craig Tillerc67cc992017-04-27 10:15:51 -070050
Craig Tillerc67cc992017-04-27 10:15:51 -070051static grpc_wakeup_fd global_wakeup_fd;
Sree Kuchibhotla5efc9132017-08-17 14:10:38 -070052
53/*******************************************************************************
54 * Singleton epoll set related fields
55 */
56
57#define MAX_EPOLL_EVENTS 100
Sree Kuchibhotla19614522017-08-25 17:10:10 -070058#define MAX_EPOLL_EVENTS_HANDLED_PER_ITERATION 1
Sree Kuchibhotla5efc9132017-08-17 14:10:38 -070059
Sree Kuchibhotlae01940f2017-08-27 18:10:12 -070060/* NOTE ON SYNCHRONIZATION:
61 * - Fields in this struct are only modified by the designated poller. Hence
62 * there is no need for any locks to protect the struct.
63 * - num_events and cursor fields have to be of atomic type to provide memory
64 * visibility guarantees only. i.e In case of multiple pollers, the designated
65 * polling thread keeps changing; the thread that wrote these values may be
66 * different from the thread reading the values
67 */
Sree Kuchibhotla5efc9132017-08-17 14:10:38 -070068typedef struct epoll_set {
69 int epfd;
70
71 /* The epoll_events after the last call to epoll_wait() */
72 struct epoll_event events[MAX_EPOLL_EVENTS];
73
74 /* The number of epoll_events after the last call to epoll_wait() */
Sree Kuchibhotlaa92a9cc2017-08-27 14:02:15 -070075 gpr_atm num_events;
Sree Kuchibhotla5efc9132017-08-17 14:10:38 -070076
77 /* Index of the first event in epoll_events that has to be processed. This
78 * field is only valid if num_events > 0 */
Sree Kuchibhotlaa92a9cc2017-08-27 14:02:15 -070079 gpr_atm cursor;
Sree Kuchibhotla5efc9132017-08-17 14:10:38 -070080} epoll_set;
81
82/* The global singleton epoll set */
83static epoll_set g_epoll_set;
84
85/* Must be called *only* once */
86static bool epoll_set_init() {
87 g_epoll_set.epfd = epoll_create1(EPOLL_CLOEXEC);
88 if (g_epoll_set.epfd < 0) {
89 gpr_log(GPR_ERROR, "epoll unavailable");
90 return false;
91 }
92
Sree Kuchibhotlaa92a9cc2017-08-27 14:02:15 -070093 gpr_log(GPR_INFO, "grpc epoll fd: %d", g_epoll_set.epfd);
94 gpr_atm_no_barrier_store(&g_epoll_set.num_events, 0);
95 gpr_atm_no_barrier_store(&g_epoll_set.cursor, 0);
Sree Kuchibhotla5efc9132017-08-17 14:10:38 -070096 return true;
97}
98
99/* epoll_set_init() MUST be called before calling this. */
100static void epoll_set_shutdown() {
101 if (g_epoll_set.epfd >= 0) {
102 close(g_epoll_set.epfd);
103 g_epoll_set.epfd = -1;
104 }
105}
Craig Tillerc67cc992017-04-27 10:15:51 -0700106
107/*******************************************************************************
108 * Fd Declarations
109 */
110
111struct grpc_fd {
112 int fd;
113
Craig Tillerc67cc992017-04-27 10:15:51 -0700114 gpr_atm read_closure;
115 gpr_atm write_closure;
116
117 struct grpc_fd *freelist_next;
Craig Tillerc67cc992017-04-27 10:15:51 -0700118
119 /* The pollset that last noticed that the fd is readable. The actual type
120 * stored in this is (grpc_pollset *) */
121 gpr_atm read_notifier_pollset;
122
123 grpc_iomgr_object iomgr_object;
124};
125
126static void fd_global_init(void);
127static void fd_global_shutdown(void);
128
129/*******************************************************************************
130 * Pollset Declarations
131 */
132
Craig Tiller43bf2592017-04-28 23:21:01 +0000133typedef enum { UNKICKED, KICKED, DESIGNATED_POLLER } kick_state;
Craig Tillerc67cc992017-04-27 10:15:51 -0700134
Craig Tiller830e82a2017-05-31 16:26:27 -0700135static const char *kick_state_string(kick_state st) {
136 switch (st) {
137 case UNKICKED:
138 return "UNKICKED";
139 case KICKED:
140 return "KICKED";
141 case DESIGNATED_POLLER:
142 return "DESIGNATED_POLLER";
143 }
144 GPR_UNREACHABLE_CODE(return "UNKNOWN");
145}
146
Craig Tillerc67cc992017-04-27 10:15:51 -0700147struct grpc_pollset_worker {
Craig Tiller32f90ee2017-04-28 12:46:41 -0700148 kick_state kick_state;
Craig Tiller55624a32017-05-26 08:14:44 -0700149 int kick_state_mutator; // which line of code last changed kick state
Craig Tillerc67cc992017-04-27 10:15:51 -0700150 bool initialized_cv;
Craig Tiller32f90ee2017-04-28 12:46:41 -0700151 grpc_pollset_worker *next;
152 grpc_pollset_worker *prev;
Craig Tillerc67cc992017-04-27 10:15:51 -0700153 gpr_cv cv;
Craig Tiller50da5ec2017-05-01 13:51:14 -0700154 grpc_closure_list schedule_on_end_work;
Craig Tillerc67cc992017-04-27 10:15:51 -0700155};
156
Craig Tiller55624a32017-05-26 08:14:44 -0700157#define SET_KICK_STATE(worker, state) \
158 do { \
159 (worker)->kick_state = (state); \
160 (worker)->kick_state_mutator = __LINE__; \
161 } while (false)
162
Craig Tillerba550da2017-05-01 14:26:31 +0000163#define MAX_NEIGHBOURHOODS 1024
164
Craig Tiller6de05932017-04-28 09:17:38 -0700165typedef struct pollset_neighbourhood {
166 gpr_mu mu;
167 grpc_pollset *active_root;
Craig Tiller6de05932017-04-28 09:17:38 -0700168 char pad[GPR_CACHELINE_SIZE];
169} pollset_neighbourhood;
170
Craig Tillerc67cc992017-04-27 10:15:51 -0700171struct grpc_pollset {
Craig Tiller6de05932017-04-28 09:17:38 -0700172 gpr_mu mu;
173 pollset_neighbourhood *neighbourhood;
Craig Tillere00d7332017-05-01 15:43:51 +0000174 bool reassigning_neighbourhood;
Craig Tiller4509c472017-04-27 19:05:13 +0000175 grpc_pollset_worker *root_worker;
176 bool kicked_without_poller;
Sree Kuchibhotla0d8431a2017-07-18 16:21:54 -0700177
178 /* Set to true if the pollset is observed to have no workers available to
Sree Kuchibhotla5efc9132017-08-17 14:10:38 -0700179 poll */
Craig Tiller6de05932017-04-28 09:17:38 -0700180 bool seen_inactive;
Sree Kuchibhotla0d8431a2017-07-18 16:21:54 -0700181 bool shutting_down; /* Is the pollset shutting down ? */
Craig Tiller4509c472017-04-27 19:05:13 +0000182 grpc_closure *shutdown_closure; /* Called after after shutdown is complete */
Sree Kuchibhotla0d8431a2017-07-18 16:21:54 -0700183
184 /* Number of workers who are *about-to* attach themselves to the pollset
185 * worker list */
Craig Tillerba550da2017-05-01 14:26:31 +0000186 int begin_refs;
Craig Tiller6de05932017-04-28 09:17:38 -0700187
188 grpc_pollset *next;
189 grpc_pollset *prev;
Craig Tillerc67cc992017-04-27 10:15:51 -0700190};
191
192/*******************************************************************************
193 * Pollset-set Declarations
194 */
Craig Tiller6de05932017-04-28 09:17:38 -0700195
Craig Tiller61f96c12017-05-12 13:36:39 -0700196struct grpc_pollset_set {
197 char unused;
198};
Craig Tillerc67cc992017-04-27 10:15:51 -0700199
200/*******************************************************************************
201 * Common helpers
202 */
203
204static bool append_error(grpc_error **composite, grpc_error *error,
205 const char *desc) {
206 if (error == GRPC_ERROR_NONE) return true;
207 if (*composite == GRPC_ERROR_NONE) {
208 *composite = GRPC_ERROR_CREATE_FROM_COPIED_STRING(desc);
209 }
210 *composite = grpc_error_add_child(*composite, error);
211 return false;
212}
213
214/*******************************************************************************
215 * Fd Definitions
216 */
217
218/* We need to keep a freelist not because of any concerns of malloc performance
219 * but instead so that implementations with multiple threads in (for example)
220 * epoll_wait deal with the race between pollset removal and incoming poll
221 * notifications.
222 *
223 * The problem is that the poller ultimately holds a reference to this
224 * object, so it is very difficult to know when is safe to free it, at least
225 * without some expensive synchronization.
226 *
227 * If we keep the object freelisted, in the worst case losing this race just
228 * becomes a spurious read notification on a reused fd.
229 */
230
231/* The alarm system needs to be able to wakeup 'some poller' sometimes
232 * (specifically when a new alarm needs to be triggered earlier than the next
233 * alarm 'epoch'). This wakeup_fd gives us something to alert on when such a
234 * case occurs. */
235
236static grpc_fd *fd_freelist = NULL;
237static gpr_mu fd_freelist_mu;
238
Craig Tillerc67cc992017-04-27 10:15:51 -0700239static void fd_global_init(void) { gpr_mu_init(&fd_freelist_mu); }
240
241static void fd_global_shutdown(void) {
242 gpr_mu_lock(&fd_freelist_mu);
243 gpr_mu_unlock(&fd_freelist_mu);
244 while (fd_freelist != NULL) {
245 grpc_fd *fd = fd_freelist;
246 fd_freelist = fd_freelist->freelist_next;
Craig Tillerc67cc992017-04-27 10:15:51 -0700247 gpr_free(fd);
248 }
249 gpr_mu_destroy(&fd_freelist_mu);
250}
251
252static grpc_fd *fd_create(int fd, const char *name) {
253 grpc_fd *new_fd = NULL;
254
255 gpr_mu_lock(&fd_freelist_mu);
256 if (fd_freelist != NULL) {
257 new_fd = fd_freelist;
258 fd_freelist = fd_freelist->freelist_next;
259 }
260 gpr_mu_unlock(&fd_freelist_mu);
261
262 if (new_fd == NULL) {
263 new_fd = gpr_malloc(sizeof(grpc_fd));
Craig Tillerc67cc992017-04-27 10:15:51 -0700264 }
265
Craig Tillerc67cc992017-04-27 10:15:51 -0700266 new_fd->fd = fd;
Craig Tillerc67cc992017-04-27 10:15:51 -0700267 grpc_lfev_init(&new_fd->read_closure);
268 grpc_lfev_init(&new_fd->write_closure);
269 gpr_atm_no_barrier_store(&new_fd->read_notifier_pollset, (gpr_atm)NULL);
270
271 new_fd->freelist_next = NULL;
Craig Tillerc67cc992017-04-27 10:15:51 -0700272
273 char *fd_name;
274 gpr_asprintf(&fd_name, "%s fd=%d", name, fd);
275 grpc_iomgr_register_object(&new_fd->iomgr_object, fd_name);
Noah Eisen264879f2017-06-20 17:14:47 -0700276#ifndef NDEBUG
277 if (GRPC_TRACER_ON(grpc_trace_fd_refcount)) {
278 gpr_log(GPR_DEBUG, "FD %d %p create %s", fd, new_fd, fd_name);
279 }
Craig Tillerc67cc992017-04-27 10:15:51 -0700280#endif
281 gpr_free(fd_name);
Craig Tiller9ddb3152017-04-27 21:32:56 +0000282
283 struct epoll_event ev = {.events = (uint32_t)(EPOLLIN | EPOLLOUT | EPOLLET),
284 .data.ptr = new_fd};
Sree Kuchibhotla5efc9132017-08-17 14:10:38 -0700285 if (epoll_ctl(g_epoll_set.epfd, EPOLL_CTL_ADD, fd, &ev) != 0) {
Craig Tiller9ddb3152017-04-27 21:32:56 +0000286 gpr_log(GPR_ERROR, "epoll_ctl failed: %s", strerror(errno));
287 }
288
Craig Tillerc67cc992017-04-27 10:15:51 -0700289 return new_fd;
290}
291
Craig Tiller4509c472017-04-27 19:05:13 +0000292static int fd_wrapped_fd(grpc_fd *fd) { return fd->fd; }
Craig Tillerc67cc992017-04-27 10:15:51 -0700293
Sree Kuchibhotlaf2641472017-08-02 23:46:40 -0700294/* if 'releasing_fd' is true, it means that we are going to detach the internal
295 * fd from grpc_fd structure (i.e which means we should not be calling
296 * shutdown() syscall on that fd) */
297static void fd_shutdown_internal(grpc_exec_ctx *exec_ctx, grpc_fd *fd,
298 grpc_error *why, bool releasing_fd) {
Craig Tiller9ddb3152017-04-27 21:32:56 +0000299 if (grpc_lfev_set_shutdown(exec_ctx, &fd->read_closure,
300 GRPC_ERROR_REF(why))) {
Sree Kuchibhotlaf2641472017-08-02 23:46:40 -0700301 if (!releasing_fd) {
302 shutdown(fd->fd, SHUT_RDWR);
303 }
Craig Tiller9ddb3152017-04-27 21:32:56 +0000304 grpc_lfev_set_shutdown(exec_ctx, &fd->write_closure, GRPC_ERROR_REF(why));
305 }
306 GRPC_ERROR_UNREF(why);
307}
308
Sree Kuchibhotlaf2641472017-08-02 23:46:40 -0700309/* Might be called multiple times */
310static void fd_shutdown(grpc_exec_ctx *exec_ctx, grpc_fd *fd, grpc_error *why) {
311 fd_shutdown_internal(exec_ctx, fd, why, false);
312}
313
Craig Tillerc67cc992017-04-27 10:15:51 -0700314static void fd_orphan(grpc_exec_ctx *exec_ctx, grpc_fd *fd,
315 grpc_closure *on_done, int *release_fd,
Yuchen Zengd40a7ae2017-07-12 15:59:56 -0700316 bool already_closed, const char *reason) {
Craig Tillerc67cc992017-04-27 10:15:51 -0700317 grpc_error *error = GRPC_ERROR_NONE;
Sree Kuchibhotlaf2641472017-08-02 23:46:40 -0700318 bool is_release_fd = (release_fd != NULL);
Craig Tillerc67cc992017-04-27 10:15:51 -0700319
Craig Tiller9ddb3152017-04-27 21:32:56 +0000320 if (!grpc_lfev_is_shutdown(&fd->read_closure)) {
Sree Kuchibhotlaf2641472017-08-02 23:46:40 -0700321 fd_shutdown_internal(exec_ctx, fd,
322 GRPC_ERROR_CREATE_FROM_COPIED_STRING(reason),
323 is_release_fd);
Craig Tiller9ddb3152017-04-27 21:32:56 +0000324 }
325
Craig Tillerc67cc992017-04-27 10:15:51 -0700326 /* If release_fd is not NULL, we should be relinquishing control of the file
327 descriptor fd->fd (but we still own the grpc_fd structure). */
Sree Kuchibhotlaf2641472017-08-02 23:46:40 -0700328 if (is_release_fd) {
Craig Tillerc67cc992017-04-27 10:15:51 -0700329 *release_fd = fd->fd;
Yuchen Zengd40a7ae2017-07-12 15:59:56 -0700330 } else if (!already_closed) {
Craig Tillerc67cc992017-04-27 10:15:51 -0700331 close(fd->fd);
Craig Tillerc67cc992017-04-27 10:15:51 -0700332 }
333
ncteisen274bbbe2017-06-08 14:57:11 -0700334 GRPC_CLOSURE_SCHED(exec_ctx, on_done, GRPC_ERROR_REF(error));
Craig Tillerc67cc992017-04-27 10:15:51 -0700335
Craig Tiller4509c472017-04-27 19:05:13 +0000336 grpc_iomgr_unregister_object(&fd->iomgr_object);
337 grpc_lfev_destroy(&fd->read_closure);
338 grpc_lfev_destroy(&fd->write_closure);
Craig Tillerc67cc992017-04-27 10:15:51 -0700339
Craig Tiller4509c472017-04-27 19:05:13 +0000340 gpr_mu_lock(&fd_freelist_mu);
341 fd->freelist_next = fd_freelist;
342 fd_freelist = fd;
343 gpr_mu_unlock(&fd_freelist_mu);
Craig Tillerc67cc992017-04-27 10:15:51 -0700344}
345
346static grpc_pollset *fd_get_read_notifier_pollset(grpc_exec_ctx *exec_ctx,
347 grpc_fd *fd) {
348 gpr_atm notifier = gpr_atm_acq_load(&fd->read_notifier_pollset);
349 return (grpc_pollset *)notifier;
350}
351
352static bool fd_is_shutdown(grpc_fd *fd) {
353 return grpc_lfev_is_shutdown(&fd->read_closure);
354}
355
Craig Tillerc67cc992017-04-27 10:15:51 -0700356static void fd_notify_on_read(grpc_exec_ctx *exec_ctx, grpc_fd *fd,
357 grpc_closure *closure) {
Craig Tiller830e82a2017-05-31 16:26:27 -0700358 grpc_lfev_notify_on(exec_ctx, &fd->read_closure, closure, "read");
Craig Tillerc67cc992017-04-27 10:15:51 -0700359}
360
361static void fd_notify_on_write(grpc_exec_ctx *exec_ctx, grpc_fd *fd,
362 grpc_closure *closure) {
Craig Tiller830e82a2017-05-31 16:26:27 -0700363 grpc_lfev_notify_on(exec_ctx, &fd->write_closure, closure, "write");
Craig Tillerc67cc992017-04-27 10:15:51 -0700364}
365
Craig Tiller4509c472017-04-27 19:05:13 +0000366static void fd_become_readable(grpc_exec_ctx *exec_ctx, grpc_fd *fd,
367 grpc_pollset *notifier) {
Craig Tiller830e82a2017-05-31 16:26:27 -0700368 grpc_lfev_set_ready(exec_ctx, &fd->read_closure, "read");
Craig Tiller4509c472017-04-27 19:05:13 +0000369 /* Use release store to match with acquire load in fd_get_read_notifier */
370 gpr_atm_rel_store(&fd->read_notifier_pollset, (gpr_atm)notifier);
371}
372
373static void fd_become_writable(grpc_exec_ctx *exec_ctx, grpc_fd *fd) {
Craig Tiller830e82a2017-05-31 16:26:27 -0700374 grpc_lfev_set_ready(exec_ctx, &fd->write_closure, "write");
Craig Tillerc67cc992017-04-27 10:15:51 -0700375}
376
377/*******************************************************************************
378 * Pollset Definitions
379 */
380
Craig Tiller6de05932017-04-28 09:17:38 -0700381GPR_TLS_DECL(g_current_thread_pollset);
382GPR_TLS_DECL(g_current_thread_worker);
Sree Kuchibhotla5efc9132017-08-17 14:10:38 -0700383
384/* The designated poller */
Craig Tiller6de05932017-04-28 09:17:38 -0700385static gpr_atm g_active_poller;
Sree Kuchibhotla5efc9132017-08-17 14:10:38 -0700386
Craig Tiller6de05932017-04-28 09:17:38 -0700387static pollset_neighbourhood *g_neighbourhoods;
Craig Tiller32f90ee2017-04-28 12:46:41 -0700388static size_t g_num_neighbourhoods;
Craig Tiller6de05932017-04-28 09:17:38 -0700389
Craig Tillerc67cc992017-04-27 10:15:51 -0700390/* Return true if first in list */
Craig Tiller32f90ee2017-04-28 12:46:41 -0700391static bool worker_insert(grpc_pollset *pollset, grpc_pollset_worker *worker) {
392 if (pollset->root_worker == NULL) {
393 pollset->root_worker = worker;
394 worker->next = worker->prev = worker;
Craig Tillerc67cc992017-04-27 10:15:51 -0700395 return true;
396 } else {
Craig Tiller32f90ee2017-04-28 12:46:41 -0700397 worker->next = pollset->root_worker;
398 worker->prev = worker->next->prev;
399 worker->next->prev = worker;
400 worker->prev->next = worker;
Craig Tillerc67cc992017-04-27 10:15:51 -0700401 return false;
402 }
403}
404
405/* Return true if last in list */
406typedef enum { EMPTIED, NEW_ROOT, REMOVED } worker_remove_result;
407
Craig Tiller32f90ee2017-04-28 12:46:41 -0700408static worker_remove_result worker_remove(grpc_pollset *pollset,
Craig Tillerc67cc992017-04-27 10:15:51 -0700409 grpc_pollset_worker *worker) {
Craig Tiller32f90ee2017-04-28 12:46:41 -0700410 if (worker == pollset->root_worker) {
411 if (worker == worker->next) {
412 pollset->root_worker = NULL;
Craig Tillerc67cc992017-04-27 10:15:51 -0700413 return EMPTIED;
414 } else {
Craig Tiller32f90ee2017-04-28 12:46:41 -0700415 pollset->root_worker = worker->next;
416 worker->prev->next = worker->next;
417 worker->next->prev = worker->prev;
Craig Tillerc67cc992017-04-27 10:15:51 -0700418 return NEW_ROOT;
419 }
420 } else {
Craig Tiller32f90ee2017-04-28 12:46:41 -0700421 worker->prev->next = worker->next;
422 worker->next->prev = worker->prev;
Craig Tillerc67cc992017-04-27 10:15:51 -0700423 return REMOVED;
424 }
425}
426
Craig Tillerba550da2017-05-01 14:26:31 +0000427static size_t choose_neighbourhood(void) {
428 return (size_t)gpr_cpu_current_cpu() % g_num_neighbourhoods;
429}
430
Craig Tiller4509c472017-04-27 19:05:13 +0000431static grpc_error *pollset_global_init(void) {
Craig Tiller4509c472017-04-27 19:05:13 +0000432 gpr_tls_init(&g_current_thread_pollset);
433 gpr_tls_init(&g_current_thread_worker);
Craig Tiller6de05932017-04-28 09:17:38 -0700434 gpr_atm_no_barrier_store(&g_active_poller, 0);
Craig Tiller375eb252017-04-27 23:29:12 +0000435 global_wakeup_fd.read_fd = -1;
436 grpc_error *err = grpc_wakeup_fd_init(&global_wakeup_fd);
437 if (err != GRPC_ERROR_NONE) return err;
Craig Tiller4509c472017-04-27 19:05:13 +0000438 struct epoll_event ev = {.events = (uint32_t)(EPOLLIN | EPOLLET),
439 .data.ptr = &global_wakeup_fd};
Sree Kuchibhotla5efc9132017-08-17 14:10:38 -0700440 if (epoll_ctl(g_epoll_set.epfd, EPOLL_CTL_ADD, global_wakeup_fd.read_fd,
441 &ev) != 0) {
Craig Tiller4509c472017-04-27 19:05:13 +0000442 return GRPC_OS_ERROR(errno, "epoll_ctl");
443 }
Craig Tillerba550da2017-05-01 14:26:31 +0000444 g_num_neighbourhoods = GPR_CLAMP(gpr_cpu_num_cores(), 1, MAX_NEIGHBOURHOODS);
Craig Tiller32f90ee2017-04-28 12:46:41 -0700445 g_neighbourhoods =
446 gpr_zalloc(sizeof(*g_neighbourhoods) * g_num_neighbourhoods);
447 for (size_t i = 0; i < g_num_neighbourhoods; i++) {
448 gpr_mu_init(&g_neighbourhoods[i].mu);
Craig Tiller32f90ee2017-04-28 12:46:41 -0700449 }
Craig Tiller4509c472017-04-27 19:05:13 +0000450 return GRPC_ERROR_NONE;
451}
452
453static void pollset_global_shutdown(void) {
Craig Tiller4509c472017-04-27 19:05:13 +0000454 gpr_tls_destroy(&g_current_thread_pollset);
455 gpr_tls_destroy(&g_current_thread_worker);
Craig Tiller375eb252017-04-27 23:29:12 +0000456 if (global_wakeup_fd.read_fd != -1) grpc_wakeup_fd_destroy(&global_wakeup_fd);
Craig Tiller32f90ee2017-04-28 12:46:41 -0700457 for (size_t i = 0; i < g_num_neighbourhoods; i++) {
458 gpr_mu_destroy(&g_neighbourhoods[i].mu);
459 }
460 gpr_free(g_neighbourhoods);
Craig Tiller4509c472017-04-27 19:05:13 +0000461}
462
463static void pollset_init(grpc_pollset *pollset, gpr_mu **mu) {
Craig Tiller6de05932017-04-28 09:17:38 -0700464 gpr_mu_init(&pollset->mu);
465 *mu = &pollset->mu;
Craig Tillerba550da2017-05-01 14:26:31 +0000466 pollset->neighbourhood = &g_neighbourhoods[choose_neighbourhood()];
Sree Kuchibhotla30882302017-08-16 13:46:52 -0700467 pollset->reassigning_neighbourhood = false;
468 pollset->root_worker = NULL;
469 pollset->kicked_without_poller = false;
Craig Tiller6de05932017-04-28 09:17:38 -0700470 pollset->seen_inactive = true;
Sree Kuchibhotla30882302017-08-16 13:46:52 -0700471 pollset->shutting_down = false;
472 pollset->shutdown_closure = NULL;
473 pollset->begin_refs = 0;
474 pollset->next = pollset->prev = NULL;
Craig Tiller6de05932017-04-28 09:17:38 -0700475}
476
Craig Tillerc6109852017-05-01 14:26:49 -0700477static void pollset_destroy(grpc_exec_ctx *exec_ctx, grpc_pollset *pollset) {
Craig Tillere00d7332017-05-01 15:43:51 +0000478 gpr_mu_lock(&pollset->mu);
Craig Tillerba550da2017-05-01 14:26:31 +0000479 if (!pollset->seen_inactive) {
Craig Tillere00d7332017-05-01 15:43:51 +0000480 pollset_neighbourhood *neighbourhood = pollset->neighbourhood;
481 gpr_mu_unlock(&pollset->mu);
Craig Tillera95bacf2017-05-01 12:51:24 -0700482 retry_lock_neighbourhood:
Craig Tillere00d7332017-05-01 15:43:51 +0000483 gpr_mu_lock(&neighbourhood->mu);
484 gpr_mu_lock(&pollset->mu);
485 if (!pollset->seen_inactive) {
486 if (pollset->neighbourhood != neighbourhood) {
487 gpr_mu_unlock(&neighbourhood->mu);
488 neighbourhood = pollset->neighbourhood;
489 gpr_mu_unlock(&pollset->mu);
490 goto retry_lock_neighbourhood;
491 }
492 pollset->prev->next = pollset->next;
493 pollset->next->prev = pollset->prev;
494 if (pollset == pollset->neighbourhood->active_root) {
495 pollset->neighbourhood->active_root =
496 pollset->next == pollset ? NULL : pollset->next;
497 }
Craig Tillerba550da2017-05-01 14:26:31 +0000498 }
499 gpr_mu_unlock(&pollset->neighbourhood->mu);
Craig Tiller6de05932017-04-28 09:17:38 -0700500 }
Craig Tillere00d7332017-05-01 15:43:51 +0000501 gpr_mu_unlock(&pollset->mu);
Craig Tiller32f90ee2017-04-28 12:46:41 -0700502 gpr_mu_destroy(&pollset->mu);
Craig Tiller4509c472017-04-27 19:05:13 +0000503}
504
Craig Tiller0ff222a2017-09-01 09:41:43 -0700505static grpc_error *pollset_kick_all(grpc_exec_ctx *exec_ctx,
506 grpc_pollset *pollset) {
yang-gdf92a642017-08-21 22:38:45 -0700507 GPR_TIMER_BEGIN("pollset_kick_all", 0);
Craig Tiller4509c472017-04-27 19:05:13 +0000508 grpc_error *error = GRPC_ERROR_NONE;
509 if (pollset->root_worker != NULL) {
510 grpc_pollset_worker *worker = pollset->root_worker;
511 do {
Craig Tiller0ff222a2017-09-01 09:41:43 -0700512 GRPC_STATS_INC_POLLSET_KICK(exec_ctx);
Craig Tiller55624a32017-05-26 08:14:44 -0700513 switch (worker->kick_state) {
514 case KICKED:
515 break;
516 case UNKICKED:
517 SET_KICK_STATE(worker, KICKED);
518 if (worker->initialized_cv) {
519 gpr_cv_signal(&worker->cv);
520 }
521 break;
522 case DESIGNATED_POLLER:
523 SET_KICK_STATE(worker, KICKED);
524 append_error(&error, grpc_wakeup_fd_wakeup(&global_wakeup_fd),
Sree Kuchibhotla0d8431a2017-07-18 16:21:54 -0700525 "pollset_kick_all");
Craig Tiller55624a32017-05-26 08:14:44 -0700526 break;
Craig Tiller4509c472017-04-27 19:05:13 +0000527 }
528
Craig Tiller32f90ee2017-04-28 12:46:41 -0700529 worker = worker->next;
Craig Tiller4509c472017-04-27 19:05:13 +0000530 } while (worker != pollset->root_worker);
531 }
Sree Kuchibhotla0d8431a2017-07-18 16:21:54 -0700532 // TODO: sreek. Check if we need to set 'kicked_without_poller' to true here
533 // in the else case
yang-gdf92a642017-08-21 22:38:45 -0700534 GPR_TIMER_END("pollset_kick_all", 0);
Craig Tiller4509c472017-04-27 19:05:13 +0000535 return error;
536}
537
538static void pollset_maybe_finish_shutdown(grpc_exec_ctx *exec_ctx,
539 grpc_pollset *pollset) {
Craig Tillerba550da2017-05-01 14:26:31 +0000540 if (pollset->shutdown_closure != NULL && pollset->root_worker == NULL &&
541 pollset->begin_refs == 0) {
yang-gdf92a642017-08-21 22:38:45 -0700542 GPR_TIMER_MARK("pollset_finish_shutdown", 0);
ncteisen274bbbe2017-06-08 14:57:11 -0700543 GRPC_CLOSURE_SCHED(exec_ctx, pollset->shutdown_closure, GRPC_ERROR_NONE);
Craig Tiller4509c472017-04-27 19:05:13 +0000544 pollset->shutdown_closure = NULL;
545 }
546}
547
548static void pollset_shutdown(grpc_exec_ctx *exec_ctx, grpc_pollset *pollset,
549 grpc_closure *closure) {
yang-gdf92a642017-08-21 22:38:45 -0700550 GPR_TIMER_BEGIN("pollset_shutdown", 0);
Craig Tiller4509c472017-04-27 19:05:13 +0000551 GPR_ASSERT(pollset->shutdown_closure == NULL);
Craig Tillerc81512a2017-05-26 09:53:58 -0700552 GPR_ASSERT(!pollset->shutting_down);
Craig Tiller4509c472017-04-27 19:05:13 +0000553 pollset->shutdown_closure = closure;
Craig Tillerc81512a2017-05-26 09:53:58 -0700554 pollset->shutting_down = true;
Craig Tiller0ff222a2017-09-01 09:41:43 -0700555 GRPC_LOG_IF_ERROR("pollset_shutdown", pollset_kick_all(exec_ctx, pollset));
Craig Tiller4509c472017-04-27 19:05:13 +0000556 pollset_maybe_finish_shutdown(exec_ctx, pollset);
yang-gdf92a642017-08-21 22:38:45 -0700557 GPR_TIMER_END("pollset_shutdown", 0);
Craig Tiller4509c472017-04-27 19:05:13 +0000558}
559
Craig Tiller4509c472017-04-27 19:05:13 +0000560static int poll_deadline_to_millis_timeout(gpr_timespec deadline,
561 gpr_timespec now) {
562 gpr_timespec timeout;
563 if (gpr_time_cmp(deadline, gpr_inf_future(deadline.clock_type)) == 0) {
564 return -1;
565 }
566
567 if (gpr_time_cmp(deadline, now) <= 0) {
568 return 0;
569 }
570
571 static const gpr_timespec round_up = {
572 .clock_type = GPR_TIMESPAN, .tv_sec = 0, .tv_nsec = GPR_NS_PER_MS - 1};
573 timeout = gpr_time_sub(deadline, now);
574 int millis = gpr_time_to_millis(gpr_time_add(timeout, round_up));
575 return millis >= 1 ? millis : 1;
576}
577
Sree Kuchibhotla5efc9132017-08-17 14:10:38 -0700578/* Process the epoll events found by do_epoll_wait() function.
579 - g_epoll_set.cursor points to the index of the first event to be processed
580 - This function then processes up-to MAX_EPOLL_EVENTS_PER_ITERATION and
581 updates the g_epoll_set.cursor
Craig Tiller4509c472017-04-27 19:05:13 +0000582
Sree Kuchibhotla5efc9132017-08-17 14:10:38 -0700583 NOTE ON SYNCRHONIZATION: Similar to do_epoll_wait(), this function is only
584 called by g_active_poller thread. So there is no need for synchronization
585 when accessing fields in g_epoll_set */
586static grpc_error *process_epoll_events(grpc_exec_ctx *exec_ctx,
587 grpc_pollset *pollset) {
588 static const char *err_desc = "process_events";
Craig Tiller4509c472017-04-27 19:05:13 +0000589 grpc_error *error = GRPC_ERROR_NONE;
Sree Kuchibhotla5efc9132017-08-17 14:10:38 -0700590
Sree Kuchibhotla3d609f12017-08-25 10:00:18 -0700591 GPR_TIMER_BEGIN("process_epoll_events", 0);
Sree Kuchibhotlaa92a9cc2017-08-27 14:02:15 -0700592 long num_events = gpr_atm_acq_load(&g_epoll_set.num_events);
593 long cursor = gpr_atm_acq_load(&g_epoll_set.cursor);
594 for (int idx = 0;
595 (idx < MAX_EPOLL_EVENTS_HANDLED_PER_ITERATION) && cursor != num_events;
Sree Kuchibhotla5efc9132017-08-17 14:10:38 -0700596 idx++) {
Sree Kuchibhotlaa92a9cc2017-08-27 14:02:15 -0700597 long c = cursor++;
Sree Kuchibhotla5efc9132017-08-17 14:10:38 -0700598 struct epoll_event *ev = &g_epoll_set.events[c];
599 void *data_ptr = ev->data.ptr;
600
Craig Tiller4509c472017-04-27 19:05:13 +0000601 if (data_ptr == &global_wakeup_fd) {
Craig Tiller4509c472017-04-27 19:05:13 +0000602 append_error(&error, grpc_wakeup_fd_consume_wakeup(&global_wakeup_fd),
603 err_desc);
604 } else {
605 grpc_fd *fd = (grpc_fd *)(data_ptr);
Sree Kuchibhotla5efc9132017-08-17 14:10:38 -0700606 bool cancel = (ev->events & (EPOLLERR | EPOLLHUP)) != 0;
607 bool read_ev = (ev->events & (EPOLLIN | EPOLLPRI)) != 0;
608 bool write_ev = (ev->events & EPOLLOUT) != 0;
609
Craig Tiller4509c472017-04-27 19:05:13 +0000610 if (read_ev || cancel) {
611 fd_become_readable(exec_ctx, fd, pollset);
612 }
Sree Kuchibhotla5efc9132017-08-17 14:10:38 -0700613
Craig Tiller4509c472017-04-27 19:05:13 +0000614 if (write_ev || cancel) {
615 fd_become_writable(exec_ctx, fd);
616 }
617 }
618 }
Sree Kuchibhotlaa92a9cc2017-08-27 14:02:15 -0700619 gpr_atm_rel_store(&g_epoll_set.cursor, cursor);
Sree Kuchibhotla3d609f12017-08-25 10:00:18 -0700620 GPR_TIMER_END("process_epoll_events", 0);
Craig Tiller4509c472017-04-27 19:05:13 +0000621 return error;
622}
623
Sree Kuchibhotla5efc9132017-08-17 14:10:38 -0700624/* Do epoll_wait and store the events in g_epoll_set.events field. This does not
625 "process" any of the events yet; that is done in process_epoll_events().
626 *See process_epoll_events() function for more details.
627
628 NOTE ON SYNCHRONIZATION: At any point of time, only the g_active_poller
629 (i.e the designated poller thread) will be calling this function. So there is
630 no need for any synchronization when accesing fields in g_epoll_set */
631static grpc_error *do_epoll_wait(grpc_exec_ctx *exec_ctx, grpc_pollset *ps,
632 gpr_timespec now, gpr_timespec deadline) {
Sree Kuchibhotla3d609f12017-08-25 10:00:18 -0700633 GPR_TIMER_BEGIN("do_epoll_wait", 0);
634
Sree Kuchibhotla5efc9132017-08-17 14:10:38 -0700635 int r;
636 int timeout = poll_deadline_to_millis_timeout(deadline, now);
637 if (timeout != 0) {
638 GRPC_SCHEDULING_START_BLOCKING_REGION;
639 }
640 do {
Craig Tillerb4bb1cd2017-07-20 14:18:17 -0700641 GRPC_STATS_INC_SYSCALL_POLL(exec_ctx);
Sree Kuchibhotla5efc9132017-08-17 14:10:38 -0700642 r = epoll_wait(g_epoll_set.epfd, g_epoll_set.events, MAX_EPOLL_EVENTS,
643 timeout);
644 } while (r < 0 && errno == EINTR);
645 if (timeout != 0) {
646 GRPC_SCHEDULING_END_BLOCKING_REGION;
647 }
648
649 if (r < 0) return GRPC_OS_ERROR(errno, "epoll_wait");
650
Craig Tiller0ff222a2017-09-01 09:41:43 -0700651 GRPC_STATS_INC_POLL_EVENTS_RETURNED(exec_ctx, r);
652
Sree Kuchibhotla5efc9132017-08-17 14:10:38 -0700653 if (GRPC_TRACER_ON(grpc_polling_trace)) {
654 gpr_log(GPR_DEBUG, "ps: %p poll got %d events", ps, r);
655 }
656
Sree Kuchibhotlaa92a9cc2017-08-27 14:02:15 -0700657 gpr_atm_rel_store(&g_epoll_set.num_events, r);
658 gpr_atm_rel_store(&g_epoll_set.cursor, 0);
Sree Kuchibhotla5efc9132017-08-17 14:10:38 -0700659
Sree Kuchibhotla3d609f12017-08-25 10:00:18 -0700660 GPR_TIMER_END("do_epoll_wait", 0);
Sree Kuchibhotla5efc9132017-08-17 14:10:38 -0700661 return GRPC_ERROR_NONE;
662}
663
Craig Tiller4509c472017-04-27 19:05:13 +0000664static bool begin_worker(grpc_pollset *pollset, grpc_pollset_worker *worker,
665 grpc_pollset_worker **worker_hdl, gpr_timespec *now,
666 gpr_timespec deadline) {
yang-gdf92a642017-08-21 22:38:45 -0700667 GPR_TIMER_BEGIN("begin_worker", 0);
Craig Tiller4509c472017-04-27 19:05:13 +0000668 if (worker_hdl != NULL) *worker_hdl = worker;
669 worker->initialized_cv = false;
Craig Tiller55624a32017-05-26 08:14:44 -0700670 SET_KICK_STATE(worker, UNKICKED);
Craig Tiller50da5ec2017-05-01 13:51:14 -0700671 worker->schedule_on_end_work = (grpc_closure_list)GRPC_CLOSURE_LIST_INIT;
Craig Tillerba550da2017-05-01 14:26:31 +0000672 pollset->begin_refs++;
Craig Tiller4509c472017-04-27 19:05:13 +0000673
Craig Tiller830e82a2017-05-31 16:26:27 -0700674 if (GRPC_TRACER_ON(grpc_polling_trace)) {
675 gpr_log(GPR_ERROR, "PS:%p BEGIN_STARTS:%p", pollset, worker);
676 }
677
Craig Tiller32f90ee2017-04-28 12:46:41 -0700678 if (pollset->seen_inactive) {
679 // pollset has been observed to be inactive, we need to move back to the
680 // active list
Craig Tillere00d7332017-05-01 15:43:51 +0000681 bool is_reassigning = false;
682 if (!pollset->reassigning_neighbourhood) {
683 is_reassigning = true;
684 pollset->reassigning_neighbourhood = true;
685 pollset->neighbourhood = &g_neighbourhoods[choose_neighbourhood()];
686 }
687 pollset_neighbourhood *neighbourhood = pollset->neighbourhood;
Craig Tiller32f90ee2017-04-28 12:46:41 -0700688 gpr_mu_unlock(&pollset->mu);
Craig Tillerba550da2017-05-01 14:26:31 +0000689 // pollset unlocked: state may change (even worker->kick_state)
690 retry_lock_neighbourhood:
Craig Tiller32f90ee2017-04-28 12:46:41 -0700691 gpr_mu_lock(&neighbourhood->mu);
692 gpr_mu_lock(&pollset->mu);
Craig Tiller830e82a2017-05-31 16:26:27 -0700693 if (GRPC_TRACER_ON(grpc_polling_trace)) {
694 gpr_log(GPR_ERROR, "PS:%p BEGIN_REORG:%p kick_state=%s is_reassigning=%d",
695 pollset, worker, kick_state_string(worker->kick_state),
696 is_reassigning);
697 }
Craig Tiller32f90ee2017-04-28 12:46:41 -0700698 if (pollset->seen_inactive) {
Craig Tiller2acab6e2017-04-30 23:06:33 +0000699 if (neighbourhood != pollset->neighbourhood) {
700 gpr_mu_unlock(&neighbourhood->mu);
701 neighbourhood = pollset->neighbourhood;
702 gpr_mu_unlock(&pollset->mu);
703 goto retry_lock_neighbourhood;
704 }
Craig Tiller32f90ee2017-04-28 12:46:41 -0700705 pollset->seen_inactive = false;
Craig Tiller2acab6e2017-04-30 23:06:33 +0000706 if (neighbourhood->active_root == NULL) {
707 neighbourhood->active_root = pollset->next = pollset->prev = pollset;
Sree Kuchibhotla0d8431a2017-07-18 16:21:54 -0700708 /* TODO: sreek. Why would this worker state be other than UNKICKED
709 * here ? (since the worker isn't added to the pollset yet, there is no
710 * way it can be "found" by other threads to get kicked). */
711
712 /* If there is no designated poller, make this the designated poller */
Craig Tiller55624a32017-05-26 08:14:44 -0700713 if (worker->kick_state == UNKICKED &&
714 gpr_atm_no_barrier_cas(&g_active_poller, 0, (gpr_atm)worker)) {
715 SET_KICK_STATE(worker, DESIGNATED_POLLER);
Craig Tiller32f90ee2017-04-28 12:46:41 -0700716 }
Craig Tiller2acab6e2017-04-30 23:06:33 +0000717 } else {
718 pollset->next = neighbourhood->active_root;
719 pollset->prev = pollset->next->prev;
720 pollset->next->prev = pollset->prev->next = pollset;
Craig Tiller4509c472017-04-27 19:05:13 +0000721 }
722 }
Craig Tillere00d7332017-05-01 15:43:51 +0000723 if (is_reassigning) {
724 GPR_ASSERT(pollset->reassigning_neighbourhood);
725 pollset->reassigning_neighbourhood = false;
726 }
Craig Tiller32f90ee2017-04-28 12:46:41 -0700727 gpr_mu_unlock(&neighbourhood->mu);
728 }
Sree Kuchibhotlae6506bc2017-07-18 21:43:45 -0700729
Craig Tiller32f90ee2017-04-28 12:46:41 -0700730 worker_insert(pollset, worker);
Craig Tillerba550da2017-05-01 14:26:31 +0000731 pollset->begin_refs--;
Sree Kuchibhotla949d0752017-07-20 23:49:15 -0700732 if (worker->kick_state == UNKICKED && !pollset->kicked_without_poller) {
Craig Tillera4b8eb02017-04-29 00:13:52 +0000733 GPR_ASSERT(gpr_atm_no_barrier_load(&g_active_poller) != (gpr_atm)worker);
Craig Tiller32f90ee2017-04-28 12:46:41 -0700734 worker->initialized_cv = true;
735 gpr_cv_init(&worker->cv);
Craig Tillerc81512a2017-05-26 09:53:58 -0700736 while (worker->kick_state == UNKICKED && !pollset->shutting_down) {
Craig Tiller830e82a2017-05-31 16:26:27 -0700737 if (GRPC_TRACER_ON(grpc_polling_trace)) {
738 gpr_log(GPR_ERROR, "PS:%p BEGIN_WAIT:%p kick_state=%s shutdown=%d",
739 pollset, worker, kick_state_string(worker->kick_state),
740 pollset->shutting_down);
741 }
Sree Kuchibhotla0d8431a2017-07-18 16:21:54 -0700742
Craig Tiller32f90ee2017-04-28 12:46:41 -0700743 if (gpr_cv_wait(&worker->cv, &pollset->mu, deadline) &&
744 worker->kick_state == UNKICKED) {
Sree Kuchibhotla0d8431a2017-07-18 16:21:54 -0700745 /* If gpr_cv_wait returns true (i.e a timeout), pretend that the worker
746 received a kick */
Craig Tiller55624a32017-05-26 08:14:44 -0700747 SET_KICK_STATE(worker, KICKED);
Craig Tiller32f90ee2017-04-28 12:46:41 -0700748 }
Craig Tillerba550da2017-05-01 14:26:31 +0000749 }
Craig Tiller4509c472017-04-27 19:05:13 +0000750 *now = gpr_now(now->clock_type);
751 }
Sree Kuchibhotla949d0752017-07-20 23:49:15 -0700752
Craig Tiller830e82a2017-05-31 16:26:27 -0700753 if (GRPC_TRACER_ON(grpc_polling_trace)) {
Sree Kuchibhotla949d0752017-07-20 23:49:15 -0700754 gpr_log(GPR_ERROR,
755 "PS:%p BEGIN_DONE:%p kick_state=%s shutdown=%d "
756 "kicked_without_poller: %d",
757 pollset, worker, kick_state_string(worker->kick_state),
758 pollset->shutting_down, pollset->kicked_without_poller);
Craig Tiller830e82a2017-05-31 16:26:27 -0700759 }
Craig Tiller4509c472017-04-27 19:05:13 +0000760
Sree Kuchibhotlae6506bc2017-07-18 21:43:45 -0700761 /* We release pollset lock in this function at a couple of places:
Sree Kuchibhotlaa0616ef2017-07-18 23:49:49 -0700762 * 1. Briefly when assigning pollset to a neighbourhood
Sree Kuchibhotlae6506bc2017-07-18 21:43:45 -0700763 * 2. When doing gpr_cv_wait()
764 * It is possible that 'kicked_without_poller' was set to true during (1) and
765 * 'shutting_down' is set to true during (1) or (2). If either of them is
766 * true, this worker cannot do polling */
Sree Kuchibhotlae6506bc2017-07-18 21:43:45 -0700767 /* TODO(sreek): Perhaps there is a better way to handle kicked_without_poller
768 * case; especially when the worker is the DESIGNATED_POLLER */
769
Sree Kuchibhotlaa0616ef2017-07-18 23:49:49 -0700770 if (pollset->kicked_without_poller) {
771 pollset->kicked_without_poller = false;
yang-gdf92a642017-08-21 22:38:45 -0700772 GPR_TIMER_END("begin_worker", 0);
Sree Kuchibhotlaa0616ef2017-07-18 23:49:49 -0700773 return false;
774 }
775
yang-gdf92a642017-08-21 22:38:45 -0700776 GPR_TIMER_END("begin_worker", 0);
Sree Kuchibhotlaa0616ef2017-07-18 23:49:49 -0700777 return worker->kick_state == DESIGNATED_POLLER && !pollset->shutting_down;
Craig Tiller4509c472017-04-27 19:05:13 +0000778}
779
Craig Tillerbbf4c7a2017-04-28 15:12:10 -0700780static bool check_neighbourhood_for_available_poller(
Craig Tillera4b8eb02017-04-29 00:13:52 +0000781 pollset_neighbourhood *neighbourhood) {
yang-gdf92a642017-08-21 22:38:45 -0700782 GPR_TIMER_BEGIN("check_neighbourhood_for_available_poller", 0);
Craig Tillerbbf4c7a2017-04-28 15:12:10 -0700783 bool found_worker = false;
784 do {
785 grpc_pollset *inspect = neighbourhood->active_root;
786 if (inspect == NULL) {
787 break;
788 }
789 gpr_mu_lock(&inspect->mu);
790 GPR_ASSERT(!inspect->seen_inactive);
791 grpc_pollset_worker *inspect_worker = inspect->root_worker;
Craig Tillerbbf4c7a2017-04-28 15:12:10 -0700792 if (inspect_worker != NULL) {
Craig Tillera4b8eb02017-04-29 00:13:52 +0000793 do {
Craig Tillerba550da2017-05-01 14:26:31 +0000794 switch (inspect_worker->kick_state) {
795 case UNKICKED:
796 if (gpr_atm_no_barrier_cas(&g_active_poller, 0,
797 (gpr_atm)inspect_worker)) {
Craig Tiller830e82a2017-05-31 16:26:27 -0700798 if (GRPC_TRACER_ON(grpc_polling_trace)) {
799 gpr_log(GPR_DEBUG, " .. choose next poller to be %p",
800 inspect_worker);
801 }
Craig Tiller55624a32017-05-26 08:14:44 -0700802 SET_KICK_STATE(inspect_worker, DESIGNATED_POLLER);
Craig Tillerba550da2017-05-01 14:26:31 +0000803 if (inspect_worker->initialized_cv) {
yang-gdf92a642017-08-21 22:38:45 -0700804 GPR_TIMER_MARK("signal worker", 0);
Craig Tillerba550da2017-05-01 14:26:31 +0000805 gpr_cv_signal(&inspect_worker->cv);
806 }
Craig Tiller830e82a2017-05-31 16:26:27 -0700807 } else {
808 if (GRPC_TRACER_ON(grpc_polling_trace)) {
809 gpr_log(GPR_DEBUG, " .. beaten to choose next poller");
810 }
Craig Tillera4b8eb02017-04-29 00:13:52 +0000811 }
Craig Tillerba550da2017-05-01 14:26:31 +0000812 // even if we didn't win the cas, there's a worker, we can stop
813 found_worker = true;
814 break;
815 case KICKED:
816 break;
817 case DESIGNATED_POLLER:
818 found_worker = true; // ok, so someone else found the worker, but
819 // we'll accept that
820 break;
Craig Tillerbbf4c7a2017-04-28 15:12:10 -0700821 }
Craig Tillera4b8eb02017-04-29 00:13:52 +0000822 inspect_worker = inspect_worker->next;
Craig Tiller830e82a2017-05-31 16:26:27 -0700823 } while (!found_worker && inspect_worker != inspect->root_worker);
Craig Tillera4b8eb02017-04-29 00:13:52 +0000824 }
825 if (!found_worker) {
Craig Tiller830e82a2017-05-31 16:26:27 -0700826 if (GRPC_TRACER_ON(grpc_polling_trace)) {
827 gpr_log(GPR_DEBUG, " .. mark pollset %p inactive", inspect);
828 }
Craig Tillerbbf4c7a2017-04-28 15:12:10 -0700829 inspect->seen_inactive = true;
Craig Tiller2acab6e2017-04-30 23:06:33 +0000830 if (inspect == neighbourhood->active_root) {
Craig Tillera95bacf2017-05-01 12:51:24 -0700831 neighbourhood->active_root =
832 inspect->next == inspect ? NULL : inspect->next;
Craig Tiller2acab6e2017-04-30 23:06:33 +0000833 }
834 inspect->next->prev = inspect->prev;
835 inspect->prev->next = inspect->next;
Craig Tillere00d7332017-05-01 15:43:51 +0000836 inspect->next = inspect->prev = NULL;
Craig Tillerbbf4c7a2017-04-28 15:12:10 -0700837 }
838 gpr_mu_unlock(&inspect->mu);
839 } while (!found_worker);
yang-gdf92a642017-08-21 22:38:45 -0700840 GPR_TIMER_END("check_neighbourhood_for_available_poller", 0);
Craig Tillerbbf4c7a2017-04-28 15:12:10 -0700841 return found_worker;
842}
843
Craig Tiller4509c472017-04-27 19:05:13 +0000844static void end_worker(grpc_exec_ctx *exec_ctx, grpc_pollset *pollset,
845 grpc_pollset_worker *worker,
846 grpc_pollset_worker **worker_hdl) {
yang-gdf92a642017-08-21 22:38:45 -0700847 GPR_TIMER_BEGIN("end_worker", 0);
Craig Tiller830e82a2017-05-31 16:26:27 -0700848 if (GRPC_TRACER_ON(grpc_polling_trace)) {
849 gpr_log(GPR_DEBUG, "PS:%p END_WORKER:%p", pollset, worker);
850 }
Craig Tiller8502ecb2017-04-28 14:22:01 -0700851 if (worker_hdl != NULL) *worker_hdl = NULL;
Craig Tiller830e82a2017-05-31 16:26:27 -0700852 /* Make sure we appear kicked */
Craig Tiller55624a32017-05-26 08:14:44 -0700853 SET_KICK_STATE(worker, KICKED);
Craig Tiller50da5ec2017-05-01 13:51:14 -0700854 grpc_closure_list_move(&worker->schedule_on_end_work,
855 &exec_ctx->closure_list);
Craig Tiller8502ecb2017-04-28 14:22:01 -0700856 if (gpr_atm_no_barrier_load(&g_active_poller) == (gpr_atm)worker) {
Craig Tillera4b8eb02017-04-29 00:13:52 +0000857 if (worker->next != worker && worker->next->kick_state == UNKICKED) {
Craig Tiller830e82a2017-05-31 16:26:27 -0700858 if (GRPC_TRACER_ON(grpc_polling_trace)) {
859 gpr_log(GPR_DEBUG, " .. choose next poller to be peer %p", worker);
860 }
Craig Tiller2acab6e2017-04-30 23:06:33 +0000861 GPR_ASSERT(worker->next->initialized_cv);
Craig Tiller32f90ee2017-04-28 12:46:41 -0700862 gpr_atm_no_barrier_store(&g_active_poller, (gpr_atm)worker->next);
Craig Tiller55624a32017-05-26 08:14:44 -0700863 SET_KICK_STATE(worker->next, DESIGNATED_POLLER);
Craig Tiller32f90ee2017-04-28 12:46:41 -0700864 gpr_cv_signal(&worker->next->cv);
Craig Tiller8502ecb2017-04-28 14:22:01 -0700865 if (grpc_exec_ctx_has_work(exec_ctx)) {
866 gpr_mu_unlock(&pollset->mu);
867 grpc_exec_ctx_flush(exec_ctx);
868 gpr_mu_lock(&pollset->mu);
869 }
Craig Tiller32f90ee2017-04-28 12:46:41 -0700870 } else {
871 gpr_atm_no_barrier_store(&g_active_poller, 0);
Craig Tillerbbf4c7a2017-04-28 15:12:10 -0700872 size_t poller_neighbourhood_idx =
873 (size_t)(pollset->neighbourhood - g_neighbourhoods);
Craig Tillerbb742672017-05-17 22:19:05 +0000874 gpr_mu_unlock(&pollset->mu);
Craig Tiller32f90ee2017-04-28 12:46:41 -0700875 bool found_worker = false;
Craig Tillerba550da2017-05-01 14:26:31 +0000876 bool scan_state[MAX_NEIGHBOURHOODS];
Craig Tillerbbf4c7a2017-04-28 15:12:10 -0700877 for (size_t i = 0; !found_worker && i < g_num_neighbourhoods; i++) {
878 pollset_neighbourhood *neighbourhood =
879 &g_neighbourhoods[(poller_neighbourhood_idx + i) %
880 g_num_neighbourhoods];
881 if (gpr_mu_trylock(&neighbourhood->mu)) {
882 found_worker =
Craig Tillera4b8eb02017-04-29 00:13:52 +0000883 check_neighbourhood_for_available_poller(neighbourhood);
Craig Tillerbbf4c7a2017-04-28 15:12:10 -0700884 gpr_mu_unlock(&neighbourhood->mu);
Craig Tillerba550da2017-05-01 14:26:31 +0000885 scan_state[i] = true;
Craig Tillerbbf4c7a2017-04-28 15:12:10 -0700886 } else {
Craig Tillerba550da2017-05-01 14:26:31 +0000887 scan_state[i] = false;
Craig Tiller32f90ee2017-04-28 12:46:41 -0700888 }
Craig Tillerbbf4c7a2017-04-28 15:12:10 -0700889 }
Craig Tiller2acab6e2017-04-30 23:06:33 +0000890 for (size_t i = 0; !found_worker && i < g_num_neighbourhoods; i++) {
Craig Tillerba550da2017-05-01 14:26:31 +0000891 if (scan_state[i]) continue;
Craig Tiller2acab6e2017-04-30 23:06:33 +0000892 pollset_neighbourhood *neighbourhood =
893 &g_neighbourhoods[(poller_neighbourhood_idx + i) %
894 g_num_neighbourhoods];
895 gpr_mu_lock(&neighbourhood->mu);
Craig Tillerba550da2017-05-01 14:26:31 +0000896 found_worker = check_neighbourhood_for_available_poller(neighbourhood);
Craig Tiller2acab6e2017-04-30 23:06:33 +0000897 gpr_mu_unlock(&neighbourhood->mu);
Craig Tillerbbf4c7a2017-04-28 15:12:10 -0700898 }
Craig Tiller8502ecb2017-04-28 14:22:01 -0700899 grpc_exec_ctx_flush(exec_ctx);
Craig Tiller32f90ee2017-04-28 12:46:41 -0700900 gpr_mu_lock(&pollset->mu);
901 }
Craig Tiller50da5ec2017-05-01 13:51:14 -0700902 } else if (grpc_exec_ctx_has_work(exec_ctx)) {
903 gpr_mu_unlock(&pollset->mu);
904 grpc_exec_ctx_flush(exec_ctx);
905 gpr_mu_lock(&pollset->mu);
Craig Tiller4509c472017-04-27 19:05:13 +0000906 }
907 if (worker->initialized_cv) {
908 gpr_cv_destroy(&worker->cv);
909 }
Craig Tiller830e82a2017-05-31 16:26:27 -0700910 if (GRPC_TRACER_ON(grpc_polling_trace)) {
911 gpr_log(GPR_DEBUG, " .. remove worker");
912 }
Craig Tiller32f90ee2017-04-28 12:46:41 -0700913 if (EMPTIED == worker_remove(pollset, worker)) {
Craig Tiller4509c472017-04-27 19:05:13 +0000914 pollset_maybe_finish_shutdown(exec_ctx, pollset);
915 }
Craig Tillera4b8eb02017-04-29 00:13:52 +0000916 GPR_ASSERT(gpr_atm_no_barrier_load(&g_active_poller) != (gpr_atm)worker);
yang-gdf92a642017-08-21 22:38:45 -0700917 GPR_TIMER_END("end_worker", 0);
Craig Tiller4509c472017-04-27 19:05:13 +0000918}
919
920/* pollset->po.mu lock must be held by the caller before calling this.
921 The function pollset_work() may temporarily release the lock (pollset->po.mu)
922 during the course of its execution but it will always re-acquire the lock and
923 ensure that it is held by the time the function returns */
Sree Kuchibhotla5efc9132017-08-17 14:10:38 -0700924static grpc_error *pollset_work(grpc_exec_ctx *exec_ctx, grpc_pollset *ps,
Craig Tiller4509c472017-04-27 19:05:13 +0000925 grpc_pollset_worker **worker_hdl,
926 gpr_timespec now, gpr_timespec deadline) {
927 grpc_pollset_worker worker;
928 grpc_error *error = GRPC_ERROR_NONE;
929 static const char *err_desc = "pollset_work";
yang-gdf92a642017-08-21 22:38:45 -0700930 GPR_TIMER_BEGIN("pollset_work", 0);
Sree Kuchibhotlab154cd12017-08-25 10:33:41 -0700931 if (ps->kicked_without_poller) {
932 ps->kicked_without_poller = false;
yang-gdf92a642017-08-21 22:38:45 -0700933 GPR_TIMER_END("pollset_work", 0);
Craig Tiller4509c472017-04-27 19:05:13 +0000934 return GRPC_ERROR_NONE;
935 }
Sree Kuchibhotla5efc9132017-08-17 14:10:38 -0700936
937 if (begin_worker(ps, &worker, worker_hdl, &now, deadline)) {
938 gpr_tls_set(&g_current_thread_pollset, (intptr_t)ps);
Craig Tiller4509c472017-04-27 19:05:13 +0000939 gpr_tls_set(&g_current_thread_worker, (intptr_t)&worker);
Sree Kuchibhotla5efc9132017-08-17 14:10:38 -0700940 GPR_ASSERT(!ps->shutting_down);
941 GPR_ASSERT(!ps->seen_inactive);
942
943 gpr_mu_unlock(&ps->mu); /* unlock */
Sree Kuchibhotla5efc9132017-08-17 14:10:38 -0700944 /* This is the designated polling thread at this point and should ideally do
945 polling. However, if there are unprocessed events left from a previous
946 call to do_epoll_wait(), skip calling epoll_wait() in this iteration and
947 process the pending epoll events.
948
949 The reason for decoupling do_epoll_wait and process_epoll_events is to
950 better distrubute the work (i.e handling epoll events) across multiple
951 threads
952
953 process_epoll_events() returns very quickly: It just queues the work on
954 exec_ctx but does not execute it (the actual exectution or more
955 accurately grpc_exec_ctx_flush() happens in end_worker() AFTER selecting
956 a designated poller). So we are not waiting long periods without a
957 designated poller */
Sree Kuchibhotlaa92a9cc2017-08-27 14:02:15 -0700958 if (gpr_atm_acq_load(&g_epoll_set.cursor) ==
959 gpr_atm_acq_load(&g_epoll_set.num_events)) {
Sree Kuchibhotla5efc9132017-08-17 14:10:38 -0700960 append_error(&error, do_epoll_wait(exec_ctx, ps, now, deadline),
961 err_desc);
962 }
963 append_error(&error, process_epoll_events(exec_ctx, ps), err_desc);
964
965 gpr_mu_lock(&ps->mu); /* lock */
966
Craig Tiller4509c472017-04-27 19:05:13 +0000967 gpr_tls_set(&g_current_thread_worker, 0);
Craig Tiller830e82a2017-05-31 16:26:27 -0700968 } else {
Sree Kuchibhotla5efc9132017-08-17 14:10:38 -0700969 gpr_tls_set(&g_current_thread_pollset, (intptr_t)ps);
Craig Tiller4509c472017-04-27 19:05:13 +0000970 }
Sree Kuchibhotla5efc9132017-08-17 14:10:38 -0700971 end_worker(exec_ctx, ps, &worker, worker_hdl);
972
Craig Tiller8502ecb2017-04-28 14:22:01 -0700973 gpr_tls_set(&g_current_thread_pollset, 0);
yang-gdf92a642017-08-21 22:38:45 -0700974 GPR_TIMER_END("pollset_work", 0);
Craig Tiller4509c472017-04-27 19:05:13 +0000975 return error;
976}
977
Craig Tiller0ff222a2017-09-01 09:41:43 -0700978static grpc_error *pollset_kick(grpc_exec_ctx *exec_ctx, grpc_pollset *pollset,
Craig Tiller4509c472017-04-27 19:05:13 +0000979 grpc_pollset_worker *specific_worker) {
yang-gdf92a642017-08-21 22:38:45 -0700980 GPR_TIMER_BEGIN("pollset_kick", 0);
Craig Tiller0ff222a2017-09-01 09:41:43 -0700981 GRPC_STATS_INC_POLLSET_KICK(exec_ctx);
yang-gdf92a642017-08-21 22:38:45 -0700982 grpc_error *ret_err = GRPC_ERROR_NONE;
Craig Tillerb89bac02017-05-26 15:20:32 +0000983 if (GRPC_TRACER_ON(grpc_polling_trace)) {
984 gpr_strvec log;
985 gpr_strvec_init(&log);
986 char *tmp;
Craig Tiller75aef7f2017-05-26 08:26:08 -0700987 gpr_asprintf(
988 &tmp, "PS:%p KICK:%p curps=%p curworker=%p root=%p", pollset,
989 specific_worker, (void *)gpr_tls_get(&g_current_thread_pollset),
990 (void *)gpr_tls_get(&g_current_thread_worker), pollset->root_worker);
Craig Tillerb89bac02017-05-26 15:20:32 +0000991 gpr_strvec_add(&log, tmp);
992 if (pollset->root_worker != NULL) {
Craig Tiller830e82a2017-05-31 16:26:27 -0700993 gpr_asprintf(&tmp, " {kick_state=%s next=%p {kick_state=%s}}",
994 kick_state_string(pollset->root_worker->kick_state),
995 pollset->root_worker->next,
996 kick_state_string(pollset->root_worker->next->kick_state));
Craig Tillerb89bac02017-05-26 15:20:32 +0000997 gpr_strvec_add(&log, tmp);
998 }
999 if (specific_worker != NULL) {
Craig Tiller830e82a2017-05-31 16:26:27 -07001000 gpr_asprintf(&tmp, " worker_kick_state=%s",
1001 kick_state_string(specific_worker->kick_state));
Craig Tillerb89bac02017-05-26 15:20:32 +00001002 gpr_strvec_add(&log, tmp);
1003 }
1004 tmp = gpr_strvec_flatten(&log, NULL);
1005 gpr_strvec_destroy(&log);
Craig Tiller830e82a2017-05-31 16:26:27 -07001006 gpr_log(GPR_ERROR, "%s", tmp);
Craig Tillerb89bac02017-05-26 15:20:32 +00001007 gpr_free(tmp);
1008 }
Craig Tiller4509c472017-04-27 19:05:13 +00001009 if (specific_worker == NULL) {
1010 if (gpr_tls_get(&g_current_thread_pollset) != (intptr_t)pollset) {
Craig Tiller375eb252017-04-27 23:29:12 +00001011 grpc_pollset_worker *root_worker = pollset->root_worker;
1012 if (root_worker == NULL) {
Craig Tiller0ff222a2017-09-01 09:41:43 -07001013 GRPC_STATS_INC_POLLSET_KICKED_WITHOUT_POLLER(exec_ctx);
Craig Tiller4509c472017-04-27 19:05:13 +00001014 pollset->kicked_without_poller = true;
Craig Tiller75aef7f2017-05-26 08:26:08 -07001015 if (GRPC_TRACER_ON(grpc_polling_trace)) {
Craig Tiller830e82a2017-05-31 16:26:27 -07001016 gpr_log(GPR_ERROR, " .. kicked_without_poller");
Craig Tiller75aef7f2017-05-26 08:26:08 -07001017 }
yang-gdf92a642017-08-21 22:38:45 -07001018 goto done;
Craig Tiller375eb252017-04-27 23:29:12 +00001019 }
Craig Tiller32f90ee2017-04-28 12:46:41 -07001020 grpc_pollset_worker *next_worker = root_worker->next;
Craig Tiller830e82a2017-05-31 16:26:27 -07001021 if (root_worker->kick_state == KICKED) {
Craig Tiller0ff222a2017-09-01 09:41:43 -07001022 GRPC_STATS_INC_POLLSET_KICKED_AGAIN(exec_ctx);
Craig Tiller75aef7f2017-05-26 08:26:08 -07001023 if (GRPC_TRACER_ON(grpc_polling_trace)) {
Craig Tiller830e82a2017-05-31 16:26:27 -07001024 gpr_log(GPR_ERROR, " .. already kicked %p", root_worker);
1025 }
1026 SET_KICK_STATE(root_worker, KICKED);
yang-gdf92a642017-08-21 22:38:45 -07001027 goto done;
Craig Tiller830e82a2017-05-31 16:26:27 -07001028 } else if (next_worker->kick_state == KICKED) {
Craig Tiller0ff222a2017-09-01 09:41:43 -07001029 GRPC_STATS_INC_POLLSET_KICKED_AGAIN(exec_ctx);
Craig Tiller830e82a2017-05-31 16:26:27 -07001030 if (GRPC_TRACER_ON(grpc_polling_trace)) {
1031 gpr_log(GPR_ERROR, " .. already kicked %p", next_worker);
1032 }
1033 SET_KICK_STATE(next_worker, KICKED);
yang-gdf92a642017-08-21 22:38:45 -07001034 goto done;
Craig Tiller830e82a2017-05-31 16:26:27 -07001035 } else if (root_worker ==
1036 next_worker && // only try and wake up a poller if
1037 // there is no next worker
1038 root_worker == (grpc_pollset_worker *)gpr_atm_no_barrier_load(
1039 &g_active_poller)) {
Craig Tiller0ff222a2017-09-01 09:41:43 -07001040 GRPC_STATS_INC_POLLSET_KICK_WAKEUP_FD(exec_ctx);
Craig Tiller830e82a2017-05-31 16:26:27 -07001041 if (GRPC_TRACER_ON(grpc_polling_trace)) {
1042 gpr_log(GPR_ERROR, " .. kicked %p", root_worker);
Craig Tiller75aef7f2017-05-26 08:26:08 -07001043 }
Craig Tiller55624a32017-05-26 08:14:44 -07001044 SET_KICK_STATE(root_worker, KICKED);
yang-gdf92a642017-08-21 22:38:45 -07001045 ret_err = grpc_wakeup_fd_wakeup(&global_wakeup_fd);
1046 goto done;
Craig Tiller8502ecb2017-04-28 14:22:01 -07001047 } else if (next_worker->kick_state == UNKICKED) {
Craig Tiller0ff222a2017-09-01 09:41:43 -07001048 GRPC_STATS_INC_POLLSET_KICK_WAKEUP_CV(exec_ctx);
Craig Tiller75aef7f2017-05-26 08:26:08 -07001049 if (GRPC_TRACER_ON(grpc_polling_trace)) {
Craig Tiller830e82a2017-05-31 16:26:27 -07001050 gpr_log(GPR_ERROR, " .. kicked %p", next_worker);
Craig Tiller75aef7f2017-05-26 08:26:08 -07001051 }
Craig Tiller8502ecb2017-04-28 14:22:01 -07001052 GPR_ASSERT(next_worker->initialized_cv);
Craig Tiller55624a32017-05-26 08:14:44 -07001053 SET_KICK_STATE(next_worker, KICKED);
Craig Tiller375eb252017-04-27 23:29:12 +00001054 gpr_cv_signal(&next_worker->cv);
yang-gdf92a642017-08-21 22:38:45 -07001055 goto done;
Craig Tiller55624a32017-05-26 08:14:44 -07001056 } else if (next_worker->kick_state == DESIGNATED_POLLER) {
1057 if (root_worker->kick_state != DESIGNATED_POLLER) {
Craig Tiller75aef7f2017-05-26 08:26:08 -07001058 if (GRPC_TRACER_ON(grpc_polling_trace)) {
Craig Tiller830e82a2017-05-31 16:26:27 -07001059 gpr_log(
1060 GPR_ERROR,
1061 " .. kicked root non-poller %p (initialized_cv=%d) (poller=%p)",
1062 root_worker, root_worker->initialized_cv, next_worker);
Craig Tiller75aef7f2017-05-26 08:26:08 -07001063 }
Craig Tiller55624a32017-05-26 08:14:44 -07001064 SET_KICK_STATE(root_worker, KICKED);
1065 if (root_worker->initialized_cv) {
Craig Tiller0ff222a2017-09-01 09:41:43 -07001066 GRPC_STATS_INC_POLLSET_KICK_WAKEUP_CV(exec_ctx);
Craig Tiller55624a32017-05-26 08:14:44 -07001067 gpr_cv_signal(&root_worker->cv);
1068 }
yang-gdf92a642017-08-21 22:38:45 -07001069 goto done;
Craig Tiller55624a32017-05-26 08:14:44 -07001070 } else {
Craig Tiller0ff222a2017-09-01 09:41:43 -07001071 GRPC_STATS_INC_POLLSET_KICK_WAKEUP_FD(exec_ctx);
Craig Tiller75aef7f2017-05-26 08:26:08 -07001072 if (GRPC_TRACER_ON(grpc_polling_trace)) {
Craig Tiller830e82a2017-05-31 16:26:27 -07001073 gpr_log(GPR_ERROR, " .. non-root poller %p (root=%p)", next_worker,
Craig Tiller75aef7f2017-05-26 08:26:08 -07001074 root_worker);
1075 }
Craig Tiller55624a32017-05-26 08:14:44 -07001076 SET_KICK_STATE(next_worker, KICKED);
yang-gdf92a642017-08-21 22:38:45 -07001077 ret_err = grpc_wakeup_fd_wakeup(&global_wakeup_fd);
1078 goto done;
Craig Tiller55624a32017-05-26 08:14:44 -07001079 }
Craig Tiller8502ecb2017-04-28 14:22:01 -07001080 } else {
Craig Tiller0ff222a2017-09-01 09:41:43 -07001081 GRPC_STATS_INC_POLLSET_KICKED_AGAIN(exec_ctx);
Craig Tiller55624a32017-05-26 08:14:44 -07001082 GPR_ASSERT(next_worker->kick_state == KICKED);
1083 SET_KICK_STATE(next_worker, KICKED);
yang-gdf92a642017-08-21 22:38:45 -07001084 goto done;
Craig Tiller4509c472017-04-27 19:05:13 +00001085 }
1086 } else {
Craig Tiller0ff222a2017-09-01 09:41:43 -07001087 GRPC_STATS_INC_POLLSET_KICKED_AGAIN(exec_ctx);
Craig Tiller830e82a2017-05-31 16:26:27 -07001088 if (GRPC_TRACER_ON(grpc_polling_trace)) {
1089 gpr_log(GPR_ERROR, " .. kicked while waking up");
1090 }
yang-gdf92a642017-08-21 22:38:45 -07001091 goto done;
Craig Tiller4509c472017-04-27 19:05:13 +00001092 }
Craig Tiller43bf2592017-04-28 23:21:01 +00001093 } else if (specific_worker->kick_state == KICKED) {
Craig Tiller0ff222a2017-09-01 09:41:43 -07001094 GRPC_STATS_INC_POLLSET_KICKED_AGAIN(exec_ctx);
Craig Tiller75aef7f2017-05-26 08:26:08 -07001095 if (GRPC_TRACER_ON(grpc_polling_trace)) {
Craig Tiller830e82a2017-05-31 16:26:27 -07001096 gpr_log(GPR_ERROR, " .. specific worker already kicked");
Craig Tiller75aef7f2017-05-26 08:26:08 -07001097 }
yang-gdf92a642017-08-21 22:38:45 -07001098 goto done;
Craig Tiller4509c472017-04-27 19:05:13 +00001099 } else if (gpr_tls_get(&g_current_thread_worker) ==
1100 (intptr_t)specific_worker) {
Craig Tiller0ff222a2017-09-01 09:41:43 -07001101 GRPC_STATS_INC_POLLSET_KICK_OWN_THREAD(exec_ctx);
Craig Tiller75aef7f2017-05-26 08:26:08 -07001102 if (GRPC_TRACER_ON(grpc_polling_trace)) {
Craig Tiller830e82a2017-05-31 16:26:27 -07001103 gpr_log(GPR_ERROR, " .. mark %p kicked", specific_worker);
Craig Tiller75aef7f2017-05-26 08:26:08 -07001104 }
Craig Tiller55624a32017-05-26 08:14:44 -07001105 SET_KICK_STATE(specific_worker, KICKED);
yang-gdf92a642017-08-21 22:38:45 -07001106 goto done;
Craig Tiller32f90ee2017-04-28 12:46:41 -07001107 } else if (specific_worker ==
1108 (grpc_pollset_worker *)gpr_atm_no_barrier_load(&g_active_poller)) {
Craig Tiller0ff222a2017-09-01 09:41:43 -07001109 GRPC_STATS_INC_POLLSET_KICK_WAKEUP_FD(exec_ctx);
Craig Tiller75aef7f2017-05-26 08:26:08 -07001110 if (GRPC_TRACER_ON(grpc_polling_trace)) {
Craig Tiller830e82a2017-05-31 16:26:27 -07001111 gpr_log(GPR_ERROR, " .. kick active poller");
Craig Tiller75aef7f2017-05-26 08:26:08 -07001112 }
Craig Tiller55624a32017-05-26 08:14:44 -07001113 SET_KICK_STATE(specific_worker, KICKED);
yang-gdf92a642017-08-21 22:38:45 -07001114 ret_err = grpc_wakeup_fd_wakeup(&global_wakeup_fd);
1115 goto done;
Craig Tiller8502ecb2017-04-28 14:22:01 -07001116 } else if (specific_worker->initialized_cv) {
Craig Tiller0ff222a2017-09-01 09:41:43 -07001117 GRPC_STATS_INC_POLLSET_KICK_WAKEUP_CV(exec_ctx);
Craig Tiller75aef7f2017-05-26 08:26:08 -07001118 if (GRPC_TRACER_ON(grpc_polling_trace)) {
Craig Tiller830e82a2017-05-31 16:26:27 -07001119 gpr_log(GPR_ERROR, " .. kick waiting worker");
Craig Tiller75aef7f2017-05-26 08:26:08 -07001120 }
Craig Tiller55624a32017-05-26 08:14:44 -07001121 SET_KICK_STATE(specific_worker, KICKED);
Craig Tiller4509c472017-04-27 19:05:13 +00001122 gpr_cv_signal(&specific_worker->cv);
yang-gdf92a642017-08-21 22:38:45 -07001123 goto done;
Craig Tiller8502ecb2017-04-28 14:22:01 -07001124 } else {
Craig Tiller0ff222a2017-09-01 09:41:43 -07001125 GRPC_STATS_INC_POLLSET_KICKED_AGAIN(exec_ctx);
Craig Tiller75aef7f2017-05-26 08:26:08 -07001126 if (GRPC_TRACER_ON(grpc_polling_trace)) {
Craig Tiller830e82a2017-05-31 16:26:27 -07001127 gpr_log(GPR_ERROR, " .. kick non-waiting worker");
Craig Tiller75aef7f2017-05-26 08:26:08 -07001128 }
Craig Tiller55624a32017-05-26 08:14:44 -07001129 SET_KICK_STATE(specific_worker, KICKED);
yang-gdf92a642017-08-21 22:38:45 -07001130 goto done;
Craig Tiller4509c472017-04-27 19:05:13 +00001131 }
yang-gdf92a642017-08-21 22:38:45 -07001132done:
1133 GPR_TIMER_END("pollset_kick", 0);
1134 return ret_err;
Craig Tiller4509c472017-04-27 19:05:13 +00001135}
1136
1137static void pollset_add_fd(grpc_exec_ctx *exec_ctx, grpc_pollset *pollset,
1138 grpc_fd *fd) {}
1139
Craig Tiller4509c472017-04-27 19:05:13 +00001140/*******************************************************************************
Craig Tillerc67cc992017-04-27 10:15:51 -07001141 * Pollset-set Definitions
1142 */
1143
1144static grpc_pollset_set *pollset_set_create(void) {
1145 return (grpc_pollset_set *)((intptr_t)0xdeafbeef);
1146}
1147
1148static void pollset_set_destroy(grpc_exec_ctx *exec_ctx,
1149 grpc_pollset_set *pss) {}
1150
1151static void pollset_set_add_fd(grpc_exec_ctx *exec_ctx, grpc_pollset_set *pss,
1152 grpc_fd *fd) {}
1153
1154static void pollset_set_del_fd(grpc_exec_ctx *exec_ctx, grpc_pollset_set *pss,
1155 grpc_fd *fd) {}
1156
1157static void pollset_set_add_pollset(grpc_exec_ctx *exec_ctx,
1158 grpc_pollset_set *pss, grpc_pollset *ps) {}
1159
1160static void pollset_set_del_pollset(grpc_exec_ctx *exec_ctx,
1161 grpc_pollset_set *pss, grpc_pollset *ps) {}
1162
1163static void pollset_set_add_pollset_set(grpc_exec_ctx *exec_ctx,
1164 grpc_pollset_set *bag,
1165 grpc_pollset_set *item) {}
1166
1167static void pollset_set_del_pollset_set(grpc_exec_ctx *exec_ctx,
1168 grpc_pollset_set *bag,
1169 grpc_pollset_set *item) {}
1170
1171/*******************************************************************************
1172 * Event engine binding
1173 */
1174
1175static void shutdown_engine(void) {
1176 fd_global_shutdown();
1177 pollset_global_shutdown();
Sree Kuchibhotla5efc9132017-08-17 14:10:38 -07001178 epoll_set_shutdown();
Craig Tillerc67cc992017-04-27 10:15:51 -07001179}
1180
1181static const grpc_event_engine_vtable vtable = {
1182 .pollset_size = sizeof(grpc_pollset),
1183
1184 .fd_create = fd_create,
1185 .fd_wrapped_fd = fd_wrapped_fd,
1186 .fd_orphan = fd_orphan,
1187 .fd_shutdown = fd_shutdown,
1188 .fd_is_shutdown = fd_is_shutdown,
1189 .fd_notify_on_read = fd_notify_on_read,
1190 .fd_notify_on_write = fd_notify_on_write,
1191 .fd_get_read_notifier_pollset = fd_get_read_notifier_pollset,
Craig Tillerc67cc992017-04-27 10:15:51 -07001192
1193 .pollset_init = pollset_init,
1194 .pollset_shutdown = pollset_shutdown,
1195 .pollset_destroy = pollset_destroy,
1196 .pollset_work = pollset_work,
1197 .pollset_kick = pollset_kick,
1198 .pollset_add_fd = pollset_add_fd,
1199
1200 .pollset_set_create = pollset_set_create,
1201 .pollset_set_destroy = pollset_set_destroy,
1202 .pollset_set_add_pollset = pollset_set_add_pollset,
1203 .pollset_set_del_pollset = pollset_set_del_pollset,
1204 .pollset_set_add_pollset_set = pollset_set_add_pollset_set,
1205 .pollset_set_del_pollset_set = pollset_set_del_pollset_set,
1206 .pollset_set_add_fd = pollset_set_add_fd,
1207 .pollset_set_del_fd = pollset_set_del_fd,
1208
Craig Tillerc67cc992017-04-27 10:15:51 -07001209 .shutdown_engine = shutdown_engine,
1210};
1211
1212/* It is possible that GLIBC has epoll but the underlying kernel doesn't.
Sree Kuchibhotla5efc9132017-08-17 14:10:38 -07001213 * Create epoll_fd (epoll_set_init() takes care of that) to make sure epoll
1214 * support is available */
Craig Tiller6f0af492017-04-27 19:26:16 +00001215const grpc_event_engine_vtable *grpc_init_epoll1_linux(bool explicit_request) {
Craig Tillerc67cc992017-04-27 10:15:51 -07001216 if (!grpc_has_wakeup_fd()) {
1217 return NULL;
1218 }
1219
Sree Kuchibhotla5efc9132017-08-17 14:10:38 -07001220 if (!epoll_set_init()) {
Craig Tillerc67cc992017-04-27 10:15:51 -07001221 return NULL;
1222 }
1223
Craig Tillerc67cc992017-04-27 10:15:51 -07001224 fd_global_init();
1225
1226 if (!GRPC_LOG_IF_ERROR("pollset_global_init", pollset_global_init())) {
Craig Tiller4509c472017-04-27 19:05:13 +00001227 fd_global_shutdown();
Sree Kuchibhotla5efc9132017-08-17 14:10:38 -07001228 epoll_set_shutdown();
Craig Tillerc67cc992017-04-27 10:15:51 -07001229 return NULL;
1230 }
1231
1232 return &vtable;
1233}
1234
1235#else /* defined(GRPC_LINUX_EPOLL) */
1236#if defined(GRPC_POSIX_SOCKET)
1237#include "src/core/lib/iomgr/ev_posix.h"
1238/* If GRPC_LINUX_EPOLL is not defined, it means epoll is not available. Return
1239 * NULL */
Craig Tiller9ddb3152017-04-27 21:32:56 +00001240const grpc_event_engine_vtable *grpc_init_epoll1_linux(bool explicit_request) {
1241 return NULL;
1242}
Craig Tillerc67cc992017-04-27 10:15:51 -07001243#endif /* defined(GRPC_POSIX_SOCKET) */
1244#endif /* !defined(GRPC_LINUX_EPOLL) */