blob: dac9104e348b5dd184d829782eb6cf926e2b466e [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
42#include "src/core/lib/iomgr/ev_posix.h"
43#include "src/core/lib/iomgr/iomgr_internal.h"
44#include "src/core/lib/iomgr/lockfree_event.h"
Craig Tillerc67cc992017-04-27 10:15:51 -070045#include "src/core/lib/iomgr/wakeup_fd_posix.h"
Craig Tillerc67cc992017-04-27 10:15:51 -070046#include "src/core/lib/profiling/timers.h"
47#include "src/core/lib/support/block_annotate.h"
Craig Tillerb89bac02017-05-26 15:20:32 +000048#include "src/core/lib/support/string.h"
Craig Tillerc67cc992017-04-27 10:15:51 -070049
Craig Tillerc67cc992017-04-27 10:15:51 -070050static grpc_wakeup_fd global_wakeup_fd;
Sree Kuchibhotla5efc9132017-08-17 14:10:38 -070051
52/*******************************************************************************
53 * Singleton epoll set related fields
54 */
55
56#define MAX_EPOLL_EVENTS 100
Sree Kuchibhotla19614522017-08-25 17:10:10 -070057#define MAX_EPOLL_EVENTS_HANDLED_PER_ITERATION 1
Sree Kuchibhotla5efc9132017-08-17 14:10:38 -070058
Sree Kuchibhotlaa92a9cc2017-08-27 14:02:15 -070059 /* NOTE ON SYNCHRONIZATION:
60 - Fields in this struct are only modified by the designated poller. Hence
61 there is no need for any locks to protect the struct.
62
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 */
Sree Kuchibhotla5efc9132017-08-17 14:10:38 -070067typedef struct epoll_set {
68 int epfd;
69
70 /* The epoll_events after the last call to epoll_wait() */
71 struct epoll_event events[MAX_EPOLL_EVENTS];
72
73 /* The number of epoll_events after the last call to epoll_wait() */
Sree Kuchibhotlaa92a9cc2017-08-27 14:02:15 -070074 gpr_atm num_events;
Sree Kuchibhotla5efc9132017-08-17 14:10:38 -070075
76 /* Index of the first event in epoll_events that has to be processed. This
77 * field is only valid if num_events > 0 */
Sree Kuchibhotlaa92a9cc2017-08-27 14:02:15 -070078 gpr_atm cursor;
Sree Kuchibhotla5efc9132017-08-17 14:10:38 -070079} epoll_set;
80
81/* The global singleton epoll set */
82static epoll_set g_epoll_set;
Sree Kuchibhotlaa92a9cc2017-08-27 14:02:15 -070083static gpr_atm g_cs = 0;
Sree Kuchibhotla5efc9132017-08-17 14:10:38 -070084
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
505static grpc_error *pollset_kick_all(grpc_pollset *pollset) {
yang-gdf92a642017-08-21 22:38:45 -0700506 GPR_TIMER_BEGIN("pollset_kick_all", 0);
Craig Tiller4509c472017-04-27 19:05:13 +0000507 grpc_error *error = GRPC_ERROR_NONE;
508 if (pollset->root_worker != NULL) {
509 grpc_pollset_worker *worker = pollset->root_worker;
510 do {
Craig Tiller55624a32017-05-26 08:14:44 -0700511 switch (worker->kick_state) {
512 case KICKED:
513 break;
514 case UNKICKED:
515 SET_KICK_STATE(worker, KICKED);
516 if (worker->initialized_cv) {
517 gpr_cv_signal(&worker->cv);
518 }
519 break;
520 case DESIGNATED_POLLER:
521 SET_KICK_STATE(worker, KICKED);
522 append_error(&error, grpc_wakeup_fd_wakeup(&global_wakeup_fd),
Sree Kuchibhotla0d8431a2017-07-18 16:21:54 -0700523 "pollset_kick_all");
Craig Tiller55624a32017-05-26 08:14:44 -0700524 break;
Craig Tiller4509c472017-04-27 19:05:13 +0000525 }
526
Craig Tiller32f90ee2017-04-28 12:46:41 -0700527 worker = worker->next;
Craig Tiller4509c472017-04-27 19:05:13 +0000528 } while (worker != pollset->root_worker);
529 }
Sree Kuchibhotla0d8431a2017-07-18 16:21:54 -0700530 // TODO: sreek. Check if we need to set 'kicked_without_poller' to true here
531 // in the else case
yang-gdf92a642017-08-21 22:38:45 -0700532 GPR_TIMER_END("pollset_kick_all", 0);
Craig Tiller4509c472017-04-27 19:05:13 +0000533 return error;
534}
535
536static void pollset_maybe_finish_shutdown(grpc_exec_ctx *exec_ctx,
537 grpc_pollset *pollset) {
Craig Tillerba550da2017-05-01 14:26:31 +0000538 if (pollset->shutdown_closure != NULL && pollset->root_worker == NULL &&
539 pollset->begin_refs == 0) {
yang-gdf92a642017-08-21 22:38:45 -0700540 GPR_TIMER_MARK("pollset_finish_shutdown", 0);
ncteisen274bbbe2017-06-08 14:57:11 -0700541 GRPC_CLOSURE_SCHED(exec_ctx, pollset->shutdown_closure, GRPC_ERROR_NONE);
Craig Tiller4509c472017-04-27 19:05:13 +0000542 pollset->shutdown_closure = NULL;
543 }
544}
545
546static void pollset_shutdown(grpc_exec_ctx *exec_ctx, grpc_pollset *pollset,
547 grpc_closure *closure) {
yang-gdf92a642017-08-21 22:38:45 -0700548 GPR_TIMER_BEGIN("pollset_shutdown", 0);
Craig Tiller4509c472017-04-27 19:05:13 +0000549 GPR_ASSERT(pollset->shutdown_closure == NULL);
Craig Tillerc81512a2017-05-26 09:53:58 -0700550 GPR_ASSERT(!pollset->shutting_down);
Craig Tiller4509c472017-04-27 19:05:13 +0000551 pollset->shutdown_closure = closure;
Craig Tillerc81512a2017-05-26 09:53:58 -0700552 pollset->shutting_down = true;
Craig Tiller4509c472017-04-27 19:05:13 +0000553 GRPC_LOG_IF_ERROR("pollset_shutdown", pollset_kick_all(pollset));
554 pollset_maybe_finish_shutdown(exec_ctx, pollset);
yang-gdf92a642017-08-21 22:38:45 -0700555 GPR_TIMER_END("pollset_shutdown", 0);
Craig Tiller4509c472017-04-27 19:05:13 +0000556}
557
Craig Tiller4509c472017-04-27 19:05:13 +0000558static int poll_deadline_to_millis_timeout(gpr_timespec deadline,
559 gpr_timespec now) {
560 gpr_timespec timeout;
561 if (gpr_time_cmp(deadline, gpr_inf_future(deadline.clock_type)) == 0) {
562 return -1;
563 }
564
565 if (gpr_time_cmp(deadline, now) <= 0) {
566 return 0;
567 }
568
569 static const gpr_timespec round_up = {
570 .clock_type = GPR_TIMESPAN, .tv_sec = 0, .tv_nsec = GPR_NS_PER_MS - 1};
571 timeout = gpr_time_sub(deadline, now);
572 int millis = gpr_time_to_millis(gpr_time_add(timeout, round_up));
573 return millis >= 1 ? millis : 1;
574}
575
Sree Kuchibhotla5efc9132017-08-17 14:10:38 -0700576/* Process the epoll events found by do_epoll_wait() function.
577 - g_epoll_set.cursor points to the index of the first event to be processed
578 - This function then processes up-to MAX_EPOLL_EVENTS_PER_ITERATION and
579 updates the g_epoll_set.cursor
Craig Tiller4509c472017-04-27 19:05:13 +0000580
Sree Kuchibhotla5efc9132017-08-17 14:10:38 -0700581 NOTE ON SYNCRHONIZATION: Similar to do_epoll_wait(), this function is only
582 called by g_active_poller thread. So there is no need for synchronization
583 when accessing fields in g_epoll_set */
584static grpc_error *process_epoll_events(grpc_exec_ctx *exec_ctx,
585 grpc_pollset *pollset) {
586 static const char *err_desc = "process_events";
Craig Tiller4509c472017-04-27 19:05:13 +0000587 grpc_error *error = GRPC_ERROR_NONE;
Sree Kuchibhotla5efc9132017-08-17 14:10:38 -0700588
Sree Kuchibhotla3d609f12017-08-25 10:00:18 -0700589 GPR_TIMER_BEGIN("process_epoll_events", 0);
Sree Kuchibhotlaa92a9cc2017-08-27 14:02:15 -0700590 long num_events = gpr_atm_acq_load(&g_epoll_set.num_events);
591 long cursor = gpr_atm_acq_load(&g_epoll_set.cursor);
592 for (int idx = 0;
593 (idx < MAX_EPOLL_EVENTS_HANDLED_PER_ITERATION) && cursor != num_events;
Sree Kuchibhotla5efc9132017-08-17 14:10:38 -0700594 idx++) {
Sree Kuchibhotlaa92a9cc2017-08-27 14:02:15 -0700595 long c = cursor++;
Sree Kuchibhotla5efc9132017-08-17 14:10:38 -0700596 struct epoll_event *ev = &g_epoll_set.events[c];
597 void *data_ptr = ev->data.ptr;
598
Craig Tiller4509c472017-04-27 19:05:13 +0000599 if (data_ptr == &global_wakeup_fd) {
Craig Tiller4509c472017-04-27 19:05:13 +0000600 append_error(&error, grpc_wakeup_fd_consume_wakeup(&global_wakeup_fd),
601 err_desc);
602 } else {
603 grpc_fd *fd = (grpc_fd *)(data_ptr);
Sree Kuchibhotla5efc9132017-08-17 14:10:38 -0700604 bool cancel = (ev->events & (EPOLLERR | EPOLLHUP)) != 0;
605 bool read_ev = (ev->events & (EPOLLIN | EPOLLPRI)) != 0;
606 bool write_ev = (ev->events & EPOLLOUT) != 0;
607
Craig Tiller4509c472017-04-27 19:05:13 +0000608 if (read_ev || cancel) {
609 fd_become_readable(exec_ctx, fd, pollset);
610 }
Sree Kuchibhotla5efc9132017-08-17 14:10:38 -0700611
Craig Tiller4509c472017-04-27 19:05:13 +0000612 if (write_ev || cancel) {
613 fd_become_writable(exec_ctx, fd);
614 }
615 }
616 }
Sree Kuchibhotlaa92a9cc2017-08-27 14:02:15 -0700617 gpr_atm_rel_store(&g_epoll_set.cursor, cursor);
Sree Kuchibhotla3d609f12017-08-25 10:00:18 -0700618 GPR_TIMER_END("process_epoll_events", 0);
Craig Tiller4509c472017-04-27 19:05:13 +0000619 return error;
620}
621
Sree Kuchibhotla5efc9132017-08-17 14:10:38 -0700622/* Do epoll_wait and store the events in g_epoll_set.events field. This does not
623 "process" any of the events yet; that is done in process_epoll_events().
624 *See process_epoll_events() function for more details.
625
626 NOTE ON SYNCHRONIZATION: At any point of time, only the g_active_poller
627 (i.e the designated poller thread) will be calling this function. So there is
628 no need for any synchronization when accesing fields in g_epoll_set */
629static grpc_error *do_epoll_wait(grpc_exec_ctx *exec_ctx, grpc_pollset *ps,
630 gpr_timespec now, gpr_timespec deadline) {
Sree Kuchibhotla3d609f12017-08-25 10:00:18 -0700631 GPR_TIMER_BEGIN("do_epoll_wait", 0);
632
Sree Kuchibhotla5efc9132017-08-17 14:10:38 -0700633 int r;
634 int timeout = poll_deadline_to_millis_timeout(deadline, now);
635 if (timeout != 0) {
636 GRPC_SCHEDULING_START_BLOCKING_REGION;
637 }
638 do {
639 r = epoll_wait(g_epoll_set.epfd, g_epoll_set.events, MAX_EPOLL_EVENTS,
640 timeout);
641 } while (r < 0 && errno == EINTR);
642 if (timeout != 0) {
643 GRPC_SCHEDULING_END_BLOCKING_REGION;
644 }
645
646 if (r < 0) return GRPC_OS_ERROR(errno, "epoll_wait");
647
648 if (GRPC_TRACER_ON(grpc_polling_trace)) {
649 gpr_log(GPR_DEBUG, "ps: %p poll got %d events", ps, r);
650 }
651
Sree Kuchibhotlaa92a9cc2017-08-27 14:02:15 -0700652 gpr_atm_rel_store(&g_epoll_set.num_events, r);
653 gpr_atm_rel_store(&g_epoll_set.cursor, 0);
Sree Kuchibhotla5efc9132017-08-17 14:10:38 -0700654
Sree Kuchibhotla3d609f12017-08-25 10:00:18 -0700655 GPR_TIMER_END("do_epoll_wait", 0);
Sree Kuchibhotla5efc9132017-08-17 14:10:38 -0700656 return GRPC_ERROR_NONE;
657}
658
Craig Tiller4509c472017-04-27 19:05:13 +0000659static bool begin_worker(grpc_pollset *pollset, grpc_pollset_worker *worker,
660 grpc_pollset_worker **worker_hdl, gpr_timespec *now,
661 gpr_timespec deadline) {
yang-gdf92a642017-08-21 22:38:45 -0700662 GPR_TIMER_BEGIN("begin_worker", 0);
Craig Tiller4509c472017-04-27 19:05:13 +0000663 if (worker_hdl != NULL) *worker_hdl = worker;
664 worker->initialized_cv = false;
Craig Tiller55624a32017-05-26 08:14:44 -0700665 SET_KICK_STATE(worker, UNKICKED);
Craig Tiller50da5ec2017-05-01 13:51:14 -0700666 worker->schedule_on_end_work = (grpc_closure_list)GRPC_CLOSURE_LIST_INIT;
Craig Tillerba550da2017-05-01 14:26:31 +0000667 pollset->begin_refs++;
Craig Tiller4509c472017-04-27 19:05:13 +0000668
Craig Tiller830e82a2017-05-31 16:26:27 -0700669 if (GRPC_TRACER_ON(grpc_polling_trace)) {
670 gpr_log(GPR_ERROR, "PS:%p BEGIN_STARTS:%p", pollset, worker);
671 }
672
Craig Tiller32f90ee2017-04-28 12:46:41 -0700673 if (pollset->seen_inactive) {
674 // pollset has been observed to be inactive, we need to move back to the
675 // active list
Craig Tillere00d7332017-05-01 15:43:51 +0000676 bool is_reassigning = false;
677 if (!pollset->reassigning_neighbourhood) {
678 is_reassigning = true;
679 pollset->reassigning_neighbourhood = true;
680 pollset->neighbourhood = &g_neighbourhoods[choose_neighbourhood()];
681 }
682 pollset_neighbourhood *neighbourhood = pollset->neighbourhood;
Craig Tiller32f90ee2017-04-28 12:46:41 -0700683 gpr_mu_unlock(&pollset->mu);
Craig Tillerba550da2017-05-01 14:26:31 +0000684 // pollset unlocked: state may change (even worker->kick_state)
685 retry_lock_neighbourhood:
Craig Tiller32f90ee2017-04-28 12:46:41 -0700686 gpr_mu_lock(&neighbourhood->mu);
687 gpr_mu_lock(&pollset->mu);
Craig Tiller830e82a2017-05-31 16:26:27 -0700688 if (GRPC_TRACER_ON(grpc_polling_trace)) {
689 gpr_log(GPR_ERROR, "PS:%p BEGIN_REORG:%p kick_state=%s is_reassigning=%d",
690 pollset, worker, kick_state_string(worker->kick_state),
691 is_reassigning);
692 }
Craig Tiller32f90ee2017-04-28 12:46:41 -0700693 if (pollset->seen_inactive) {
Craig Tiller2acab6e2017-04-30 23:06:33 +0000694 if (neighbourhood != pollset->neighbourhood) {
695 gpr_mu_unlock(&neighbourhood->mu);
696 neighbourhood = pollset->neighbourhood;
697 gpr_mu_unlock(&pollset->mu);
698 goto retry_lock_neighbourhood;
699 }
Craig Tiller32f90ee2017-04-28 12:46:41 -0700700 pollset->seen_inactive = false;
Craig Tiller2acab6e2017-04-30 23:06:33 +0000701 if (neighbourhood->active_root == NULL) {
702 neighbourhood->active_root = pollset->next = pollset->prev = pollset;
Sree Kuchibhotla0d8431a2017-07-18 16:21:54 -0700703 /* TODO: sreek. Why would this worker state be other than UNKICKED
704 * here ? (since the worker isn't added to the pollset yet, there is no
705 * way it can be "found" by other threads to get kicked). */
706
707 /* If there is no designated poller, make this the designated poller */
Craig Tiller55624a32017-05-26 08:14:44 -0700708 if (worker->kick_state == UNKICKED &&
709 gpr_atm_no_barrier_cas(&g_active_poller, 0, (gpr_atm)worker)) {
710 SET_KICK_STATE(worker, DESIGNATED_POLLER);
Craig Tiller32f90ee2017-04-28 12:46:41 -0700711 }
Craig Tiller2acab6e2017-04-30 23:06:33 +0000712 } else {
713 pollset->next = neighbourhood->active_root;
714 pollset->prev = pollset->next->prev;
715 pollset->next->prev = pollset->prev->next = pollset;
Craig Tiller4509c472017-04-27 19:05:13 +0000716 }
717 }
Craig Tillere00d7332017-05-01 15:43:51 +0000718 if (is_reassigning) {
719 GPR_ASSERT(pollset->reassigning_neighbourhood);
720 pollset->reassigning_neighbourhood = false;
721 }
Craig Tiller32f90ee2017-04-28 12:46:41 -0700722 gpr_mu_unlock(&neighbourhood->mu);
723 }
Sree Kuchibhotlae6506bc2017-07-18 21:43:45 -0700724
Craig Tiller32f90ee2017-04-28 12:46:41 -0700725 worker_insert(pollset, worker);
Craig Tillerba550da2017-05-01 14:26:31 +0000726 pollset->begin_refs--;
Sree Kuchibhotla949d0752017-07-20 23:49:15 -0700727 if (worker->kick_state == UNKICKED && !pollset->kicked_without_poller) {
Craig Tillera4b8eb02017-04-29 00:13:52 +0000728 GPR_ASSERT(gpr_atm_no_barrier_load(&g_active_poller) != (gpr_atm)worker);
Craig Tiller32f90ee2017-04-28 12:46:41 -0700729 worker->initialized_cv = true;
730 gpr_cv_init(&worker->cv);
Craig Tillerc81512a2017-05-26 09:53:58 -0700731 while (worker->kick_state == UNKICKED && !pollset->shutting_down) {
Craig Tiller830e82a2017-05-31 16:26:27 -0700732 if (GRPC_TRACER_ON(grpc_polling_trace)) {
733 gpr_log(GPR_ERROR, "PS:%p BEGIN_WAIT:%p kick_state=%s shutdown=%d",
734 pollset, worker, kick_state_string(worker->kick_state),
735 pollset->shutting_down);
736 }
Sree Kuchibhotla0d8431a2017-07-18 16:21:54 -0700737
Craig Tiller32f90ee2017-04-28 12:46:41 -0700738 if (gpr_cv_wait(&worker->cv, &pollset->mu, deadline) &&
739 worker->kick_state == UNKICKED) {
Sree Kuchibhotla0d8431a2017-07-18 16:21:54 -0700740 /* If gpr_cv_wait returns true (i.e a timeout), pretend that the worker
741 received a kick */
Craig Tiller55624a32017-05-26 08:14:44 -0700742 SET_KICK_STATE(worker, KICKED);
Craig Tiller32f90ee2017-04-28 12:46:41 -0700743 }
Craig Tillerba550da2017-05-01 14:26:31 +0000744 }
Craig Tiller4509c472017-04-27 19:05:13 +0000745 *now = gpr_now(now->clock_type);
746 }
Sree Kuchibhotla949d0752017-07-20 23:49:15 -0700747
Craig Tiller830e82a2017-05-31 16:26:27 -0700748 if (GRPC_TRACER_ON(grpc_polling_trace)) {
Sree Kuchibhotla949d0752017-07-20 23:49:15 -0700749 gpr_log(GPR_ERROR,
750 "PS:%p BEGIN_DONE:%p kick_state=%s shutdown=%d "
751 "kicked_without_poller: %d",
752 pollset, worker, kick_state_string(worker->kick_state),
753 pollset->shutting_down, pollset->kicked_without_poller);
Craig Tiller830e82a2017-05-31 16:26:27 -0700754 }
Craig Tiller4509c472017-04-27 19:05:13 +0000755
Sree Kuchibhotlae6506bc2017-07-18 21:43:45 -0700756 /* We release pollset lock in this function at a couple of places:
Sree Kuchibhotlaa0616ef2017-07-18 23:49:49 -0700757 * 1. Briefly when assigning pollset to a neighbourhood
Sree Kuchibhotlae6506bc2017-07-18 21:43:45 -0700758 * 2. When doing gpr_cv_wait()
759 * It is possible that 'kicked_without_poller' was set to true during (1) and
760 * 'shutting_down' is set to true during (1) or (2). If either of them is
761 * true, this worker cannot do polling */
Sree Kuchibhotlae6506bc2017-07-18 21:43:45 -0700762 /* TODO(sreek): Perhaps there is a better way to handle kicked_without_poller
763 * case; especially when the worker is the DESIGNATED_POLLER */
764
Sree Kuchibhotlaa0616ef2017-07-18 23:49:49 -0700765 if (pollset->kicked_without_poller) {
766 pollset->kicked_without_poller = false;
yang-gdf92a642017-08-21 22:38:45 -0700767 GPR_TIMER_END("begin_worker", 0);
Sree Kuchibhotlaa0616ef2017-07-18 23:49:49 -0700768 return false;
769 }
770
yang-gdf92a642017-08-21 22:38:45 -0700771 GPR_TIMER_END("begin_worker", 0);
Sree Kuchibhotlaa0616ef2017-07-18 23:49:49 -0700772 return worker->kick_state == DESIGNATED_POLLER && !pollset->shutting_down;
Craig Tiller4509c472017-04-27 19:05:13 +0000773}
774
Craig Tillerbbf4c7a2017-04-28 15:12:10 -0700775static bool check_neighbourhood_for_available_poller(
Craig Tillera4b8eb02017-04-29 00:13:52 +0000776 pollset_neighbourhood *neighbourhood) {
yang-gdf92a642017-08-21 22:38:45 -0700777 GPR_TIMER_BEGIN("check_neighbourhood_for_available_poller", 0);
Craig Tillerbbf4c7a2017-04-28 15:12:10 -0700778 bool found_worker = false;
779 do {
780 grpc_pollset *inspect = neighbourhood->active_root;
781 if (inspect == NULL) {
782 break;
783 }
784 gpr_mu_lock(&inspect->mu);
785 GPR_ASSERT(!inspect->seen_inactive);
786 grpc_pollset_worker *inspect_worker = inspect->root_worker;
Craig Tillerbbf4c7a2017-04-28 15:12:10 -0700787 if (inspect_worker != NULL) {
Craig Tillera4b8eb02017-04-29 00:13:52 +0000788 do {
Craig Tillerba550da2017-05-01 14:26:31 +0000789 switch (inspect_worker->kick_state) {
790 case UNKICKED:
791 if (gpr_atm_no_barrier_cas(&g_active_poller, 0,
792 (gpr_atm)inspect_worker)) {
Craig Tiller830e82a2017-05-31 16:26:27 -0700793 if (GRPC_TRACER_ON(grpc_polling_trace)) {
794 gpr_log(GPR_DEBUG, " .. choose next poller to be %p",
795 inspect_worker);
796 }
Craig Tiller55624a32017-05-26 08:14:44 -0700797 SET_KICK_STATE(inspect_worker, DESIGNATED_POLLER);
Craig Tillerba550da2017-05-01 14:26:31 +0000798 if (inspect_worker->initialized_cv) {
yang-gdf92a642017-08-21 22:38:45 -0700799 GPR_TIMER_MARK("signal worker", 0);
Craig Tillerba550da2017-05-01 14:26:31 +0000800 gpr_cv_signal(&inspect_worker->cv);
801 }
Craig Tiller830e82a2017-05-31 16:26:27 -0700802 } else {
803 if (GRPC_TRACER_ON(grpc_polling_trace)) {
804 gpr_log(GPR_DEBUG, " .. beaten to choose next poller");
805 }
Craig Tillera4b8eb02017-04-29 00:13:52 +0000806 }
Craig Tillerba550da2017-05-01 14:26:31 +0000807 // even if we didn't win the cas, there's a worker, we can stop
808 found_worker = true;
809 break;
810 case KICKED:
811 break;
812 case DESIGNATED_POLLER:
813 found_worker = true; // ok, so someone else found the worker, but
814 // we'll accept that
815 break;
Craig Tillerbbf4c7a2017-04-28 15:12:10 -0700816 }
Craig Tillera4b8eb02017-04-29 00:13:52 +0000817 inspect_worker = inspect_worker->next;
Craig Tiller830e82a2017-05-31 16:26:27 -0700818 } while (!found_worker && inspect_worker != inspect->root_worker);
Craig Tillera4b8eb02017-04-29 00:13:52 +0000819 }
820 if (!found_worker) {
Craig Tiller830e82a2017-05-31 16:26:27 -0700821 if (GRPC_TRACER_ON(grpc_polling_trace)) {
822 gpr_log(GPR_DEBUG, " .. mark pollset %p inactive", inspect);
823 }
Craig Tillerbbf4c7a2017-04-28 15:12:10 -0700824 inspect->seen_inactive = true;
Craig Tiller2acab6e2017-04-30 23:06:33 +0000825 if (inspect == neighbourhood->active_root) {
Craig Tillera95bacf2017-05-01 12:51:24 -0700826 neighbourhood->active_root =
827 inspect->next == inspect ? NULL : inspect->next;
Craig Tiller2acab6e2017-04-30 23:06:33 +0000828 }
829 inspect->next->prev = inspect->prev;
830 inspect->prev->next = inspect->next;
Craig Tillere00d7332017-05-01 15:43:51 +0000831 inspect->next = inspect->prev = NULL;
Craig Tillerbbf4c7a2017-04-28 15:12:10 -0700832 }
833 gpr_mu_unlock(&inspect->mu);
834 } while (!found_worker);
yang-gdf92a642017-08-21 22:38:45 -0700835 GPR_TIMER_END("check_neighbourhood_for_available_poller", 0);
Craig Tillerbbf4c7a2017-04-28 15:12:10 -0700836 return found_worker;
837}
838
Craig Tiller4509c472017-04-27 19:05:13 +0000839static void end_worker(grpc_exec_ctx *exec_ctx, grpc_pollset *pollset,
840 grpc_pollset_worker *worker,
841 grpc_pollset_worker **worker_hdl) {
yang-gdf92a642017-08-21 22:38:45 -0700842 GPR_TIMER_BEGIN("end_worker", 0);
Craig Tiller830e82a2017-05-31 16:26:27 -0700843 if (GRPC_TRACER_ON(grpc_polling_trace)) {
844 gpr_log(GPR_DEBUG, "PS:%p END_WORKER:%p", pollset, worker);
845 }
Craig Tiller8502ecb2017-04-28 14:22:01 -0700846 if (worker_hdl != NULL) *worker_hdl = NULL;
Craig Tiller830e82a2017-05-31 16:26:27 -0700847 /* Make sure we appear kicked */
Craig Tiller55624a32017-05-26 08:14:44 -0700848 SET_KICK_STATE(worker, KICKED);
Craig Tiller50da5ec2017-05-01 13:51:14 -0700849 grpc_closure_list_move(&worker->schedule_on_end_work,
850 &exec_ctx->closure_list);
Craig Tiller8502ecb2017-04-28 14:22:01 -0700851 if (gpr_atm_no_barrier_load(&g_active_poller) == (gpr_atm)worker) {
Craig Tillera4b8eb02017-04-29 00:13:52 +0000852 if (worker->next != worker && worker->next->kick_state == UNKICKED) {
Craig Tiller830e82a2017-05-31 16:26:27 -0700853 if (GRPC_TRACER_ON(grpc_polling_trace)) {
854 gpr_log(GPR_DEBUG, " .. choose next poller to be peer %p", worker);
855 }
Craig Tiller2acab6e2017-04-30 23:06:33 +0000856 GPR_ASSERT(worker->next->initialized_cv);
Craig Tiller32f90ee2017-04-28 12:46:41 -0700857 gpr_atm_no_barrier_store(&g_active_poller, (gpr_atm)worker->next);
Craig Tiller55624a32017-05-26 08:14:44 -0700858 SET_KICK_STATE(worker->next, DESIGNATED_POLLER);
Craig Tiller32f90ee2017-04-28 12:46:41 -0700859 gpr_cv_signal(&worker->next->cv);
Craig Tiller8502ecb2017-04-28 14:22:01 -0700860 if (grpc_exec_ctx_has_work(exec_ctx)) {
861 gpr_mu_unlock(&pollset->mu);
862 grpc_exec_ctx_flush(exec_ctx);
863 gpr_mu_lock(&pollset->mu);
864 }
Craig Tiller32f90ee2017-04-28 12:46:41 -0700865 } else {
866 gpr_atm_no_barrier_store(&g_active_poller, 0);
Craig Tillerbbf4c7a2017-04-28 15:12:10 -0700867 size_t poller_neighbourhood_idx =
868 (size_t)(pollset->neighbourhood - g_neighbourhoods);
Craig Tillerbb742672017-05-17 22:19:05 +0000869 gpr_mu_unlock(&pollset->mu);
Craig Tiller32f90ee2017-04-28 12:46:41 -0700870 bool found_worker = false;
Craig Tillerba550da2017-05-01 14:26:31 +0000871 bool scan_state[MAX_NEIGHBOURHOODS];
Craig Tillerbbf4c7a2017-04-28 15:12:10 -0700872 for (size_t i = 0; !found_worker && i < g_num_neighbourhoods; i++) {
873 pollset_neighbourhood *neighbourhood =
874 &g_neighbourhoods[(poller_neighbourhood_idx + i) %
875 g_num_neighbourhoods];
876 if (gpr_mu_trylock(&neighbourhood->mu)) {
877 found_worker =
Craig Tillera4b8eb02017-04-29 00:13:52 +0000878 check_neighbourhood_for_available_poller(neighbourhood);
Craig Tillerbbf4c7a2017-04-28 15:12:10 -0700879 gpr_mu_unlock(&neighbourhood->mu);
Craig Tillerba550da2017-05-01 14:26:31 +0000880 scan_state[i] = true;
Craig Tillerbbf4c7a2017-04-28 15:12:10 -0700881 } else {
Craig Tillerba550da2017-05-01 14:26:31 +0000882 scan_state[i] = false;
Craig Tiller32f90ee2017-04-28 12:46:41 -0700883 }
Craig Tillerbbf4c7a2017-04-28 15:12:10 -0700884 }
Craig Tiller2acab6e2017-04-30 23:06:33 +0000885 for (size_t i = 0; !found_worker && i < g_num_neighbourhoods; i++) {
Craig Tillerba550da2017-05-01 14:26:31 +0000886 if (scan_state[i]) continue;
Craig Tiller2acab6e2017-04-30 23:06:33 +0000887 pollset_neighbourhood *neighbourhood =
888 &g_neighbourhoods[(poller_neighbourhood_idx + i) %
889 g_num_neighbourhoods];
890 gpr_mu_lock(&neighbourhood->mu);
Craig Tillerba550da2017-05-01 14:26:31 +0000891 found_worker = check_neighbourhood_for_available_poller(neighbourhood);
Craig Tiller2acab6e2017-04-30 23:06:33 +0000892 gpr_mu_unlock(&neighbourhood->mu);
Craig Tillerbbf4c7a2017-04-28 15:12:10 -0700893 }
Craig Tiller8502ecb2017-04-28 14:22:01 -0700894 grpc_exec_ctx_flush(exec_ctx);
Craig Tiller32f90ee2017-04-28 12:46:41 -0700895 gpr_mu_lock(&pollset->mu);
896 }
Craig Tiller50da5ec2017-05-01 13:51:14 -0700897 } else if (grpc_exec_ctx_has_work(exec_ctx)) {
898 gpr_mu_unlock(&pollset->mu);
899 grpc_exec_ctx_flush(exec_ctx);
900 gpr_mu_lock(&pollset->mu);
Craig Tiller4509c472017-04-27 19:05:13 +0000901 }
902 if (worker->initialized_cv) {
903 gpr_cv_destroy(&worker->cv);
904 }
Craig Tiller830e82a2017-05-31 16:26:27 -0700905 if (GRPC_TRACER_ON(grpc_polling_trace)) {
906 gpr_log(GPR_DEBUG, " .. remove worker");
907 }
Craig Tiller32f90ee2017-04-28 12:46:41 -0700908 if (EMPTIED == worker_remove(pollset, worker)) {
Craig Tiller4509c472017-04-27 19:05:13 +0000909 pollset_maybe_finish_shutdown(exec_ctx, pollset);
910 }
Craig Tillera4b8eb02017-04-29 00:13:52 +0000911 GPR_ASSERT(gpr_atm_no_barrier_load(&g_active_poller) != (gpr_atm)worker);
yang-gdf92a642017-08-21 22:38:45 -0700912 GPR_TIMER_END("end_worker", 0);
Craig Tiller4509c472017-04-27 19:05:13 +0000913}
914
915/* pollset->po.mu lock must be held by the caller before calling this.
916 The function pollset_work() may temporarily release the lock (pollset->po.mu)
917 during the course of its execution but it will always re-acquire the lock and
918 ensure that it is held by the time the function returns */
Sree Kuchibhotla5efc9132017-08-17 14:10:38 -0700919static grpc_error *pollset_work(grpc_exec_ctx *exec_ctx, grpc_pollset *ps,
Craig Tiller4509c472017-04-27 19:05:13 +0000920 grpc_pollset_worker **worker_hdl,
921 gpr_timespec now, gpr_timespec deadline) {
922 grpc_pollset_worker worker;
923 grpc_error *error = GRPC_ERROR_NONE;
924 static const char *err_desc = "pollset_work";
yang-gdf92a642017-08-21 22:38:45 -0700925 GPR_TIMER_BEGIN("pollset_work", 0);
Sree Kuchibhotlab154cd12017-08-25 10:33:41 -0700926 if (ps->kicked_without_poller) {
927 ps->kicked_without_poller = false;
yang-gdf92a642017-08-21 22:38:45 -0700928 GPR_TIMER_END("pollset_work", 0);
Craig Tiller4509c472017-04-27 19:05:13 +0000929 return GRPC_ERROR_NONE;
930 }
Sree Kuchibhotla5efc9132017-08-17 14:10:38 -0700931
932 if (begin_worker(ps, &worker, worker_hdl, &now, deadline)) {
Sree Kuchibhotlaa92a9cc2017-08-27 14:02:15 -0700933 GPR_ASSERT(gpr_atm_no_barrier_cas(&g_cs, 0, 1));
Sree Kuchibhotla5efc9132017-08-17 14:10:38 -0700934 gpr_tls_set(&g_current_thread_pollset, (intptr_t)ps);
Craig Tiller4509c472017-04-27 19:05:13 +0000935 gpr_tls_set(&g_current_thread_worker, (intptr_t)&worker);
Sree Kuchibhotla5efc9132017-08-17 14:10:38 -0700936 GPR_ASSERT(!ps->shutting_down);
937 GPR_ASSERT(!ps->seen_inactive);
938
939 gpr_mu_unlock(&ps->mu); /* unlock */
Sree Kuchibhotla5efc9132017-08-17 14:10:38 -0700940 /* This is the designated polling thread at this point and should ideally do
941 polling. However, if there are unprocessed events left from a previous
942 call to do_epoll_wait(), skip calling epoll_wait() in this iteration and
943 process the pending epoll events.
944
945 The reason for decoupling do_epoll_wait and process_epoll_events is to
946 better distrubute the work (i.e handling epoll events) across multiple
947 threads
948
949 process_epoll_events() returns very quickly: It just queues the work on
950 exec_ctx but does not execute it (the actual exectution or more
951 accurately grpc_exec_ctx_flush() happens in end_worker() AFTER selecting
952 a designated poller). So we are not waiting long periods without a
953 designated poller */
Sree Kuchibhotlaa92a9cc2017-08-27 14:02:15 -0700954 if (gpr_atm_acq_load(&g_epoll_set.cursor) ==
955 gpr_atm_acq_load(&g_epoll_set.num_events)) {
Sree Kuchibhotla5efc9132017-08-17 14:10:38 -0700956 append_error(&error, do_epoll_wait(exec_ctx, ps, now, deadline),
957 err_desc);
958 }
959 append_error(&error, process_epoll_events(exec_ctx, ps), err_desc);
960
961 gpr_mu_lock(&ps->mu); /* lock */
962
Craig Tiller4509c472017-04-27 19:05:13 +0000963 gpr_tls_set(&g_current_thread_worker, 0);
Sree Kuchibhotlaa92a9cc2017-08-27 14:02:15 -0700964 GPR_ASSERT(gpr_atm_no_barrier_cas(&g_cs, 1, 0));
Craig Tiller830e82a2017-05-31 16:26:27 -0700965 } else {
Sree Kuchibhotla5efc9132017-08-17 14:10:38 -0700966 gpr_tls_set(&g_current_thread_pollset, (intptr_t)ps);
Craig Tiller4509c472017-04-27 19:05:13 +0000967 }
Sree Kuchibhotla5efc9132017-08-17 14:10:38 -0700968 end_worker(exec_ctx, ps, &worker, worker_hdl);
969
Craig Tiller8502ecb2017-04-28 14:22:01 -0700970 gpr_tls_set(&g_current_thread_pollset, 0);
yang-gdf92a642017-08-21 22:38:45 -0700971 GPR_TIMER_END("pollset_work", 0);
Craig Tiller4509c472017-04-27 19:05:13 +0000972 return error;
973}
974
975static grpc_error *pollset_kick(grpc_pollset *pollset,
976 grpc_pollset_worker *specific_worker) {
yang-gdf92a642017-08-21 22:38:45 -0700977 GPR_TIMER_BEGIN("pollset_kick", 0);
978 grpc_error *ret_err = GRPC_ERROR_NONE;
Craig Tillerb89bac02017-05-26 15:20:32 +0000979 if (GRPC_TRACER_ON(grpc_polling_trace)) {
980 gpr_strvec log;
981 gpr_strvec_init(&log);
982 char *tmp;
Craig Tiller75aef7f2017-05-26 08:26:08 -0700983 gpr_asprintf(
984 &tmp, "PS:%p KICK:%p curps=%p curworker=%p root=%p", pollset,
985 specific_worker, (void *)gpr_tls_get(&g_current_thread_pollset),
986 (void *)gpr_tls_get(&g_current_thread_worker), pollset->root_worker);
Craig Tillerb89bac02017-05-26 15:20:32 +0000987 gpr_strvec_add(&log, tmp);
988 if (pollset->root_worker != NULL) {
Craig Tiller830e82a2017-05-31 16:26:27 -0700989 gpr_asprintf(&tmp, " {kick_state=%s next=%p {kick_state=%s}}",
990 kick_state_string(pollset->root_worker->kick_state),
991 pollset->root_worker->next,
992 kick_state_string(pollset->root_worker->next->kick_state));
Craig Tillerb89bac02017-05-26 15:20:32 +0000993 gpr_strvec_add(&log, tmp);
994 }
995 if (specific_worker != NULL) {
Craig Tiller830e82a2017-05-31 16:26:27 -0700996 gpr_asprintf(&tmp, " worker_kick_state=%s",
997 kick_state_string(specific_worker->kick_state));
Craig Tillerb89bac02017-05-26 15:20:32 +0000998 gpr_strvec_add(&log, tmp);
999 }
1000 tmp = gpr_strvec_flatten(&log, NULL);
1001 gpr_strvec_destroy(&log);
Craig Tiller830e82a2017-05-31 16:26:27 -07001002 gpr_log(GPR_ERROR, "%s", tmp);
Craig Tillerb89bac02017-05-26 15:20:32 +00001003 gpr_free(tmp);
1004 }
Craig Tiller4509c472017-04-27 19:05:13 +00001005 if (specific_worker == NULL) {
1006 if (gpr_tls_get(&g_current_thread_pollset) != (intptr_t)pollset) {
Craig Tiller375eb252017-04-27 23:29:12 +00001007 grpc_pollset_worker *root_worker = pollset->root_worker;
1008 if (root_worker == NULL) {
Craig Tiller4509c472017-04-27 19:05:13 +00001009 pollset->kicked_without_poller = true;
Craig Tiller75aef7f2017-05-26 08:26:08 -07001010 if (GRPC_TRACER_ON(grpc_polling_trace)) {
Craig Tiller830e82a2017-05-31 16:26:27 -07001011 gpr_log(GPR_ERROR, " .. kicked_without_poller");
Craig Tiller75aef7f2017-05-26 08:26:08 -07001012 }
yang-gdf92a642017-08-21 22:38:45 -07001013 goto done;
Craig Tiller375eb252017-04-27 23:29:12 +00001014 }
Craig Tiller32f90ee2017-04-28 12:46:41 -07001015 grpc_pollset_worker *next_worker = root_worker->next;
Craig Tiller830e82a2017-05-31 16:26:27 -07001016 if (root_worker->kick_state == KICKED) {
Craig Tiller75aef7f2017-05-26 08:26:08 -07001017 if (GRPC_TRACER_ON(grpc_polling_trace)) {
Craig Tiller830e82a2017-05-31 16:26:27 -07001018 gpr_log(GPR_ERROR, " .. already kicked %p", root_worker);
1019 }
1020 SET_KICK_STATE(root_worker, KICKED);
yang-gdf92a642017-08-21 22:38:45 -07001021 goto done;
Craig Tiller830e82a2017-05-31 16:26:27 -07001022 } else if (next_worker->kick_state == KICKED) {
1023 if (GRPC_TRACER_ON(grpc_polling_trace)) {
1024 gpr_log(GPR_ERROR, " .. already kicked %p", next_worker);
1025 }
1026 SET_KICK_STATE(next_worker, KICKED);
yang-gdf92a642017-08-21 22:38:45 -07001027 goto done;
Craig Tiller830e82a2017-05-31 16:26:27 -07001028 } else if (root_worker ==
1029 next_worker && // only try and wake up a poller if
1030 // there is no next worker
1031 root_worker == (grpc_pollset_worker *)gpr_atm_no_barrier_load(
1032 &g_active_poller)) {
1033 if (GRPC_TRACER_ON(grpc_polling_trace)) {
1034 gpr_log(GPR_ERROR, " .. kicked %p", root_worker);
Craig Tiller75aef7f2017-05-26 08:26:08 -07001035 }
Craig Tiller55624a32017-05-26 08:14:44 -07001036 SET_KICK_STATE(root_worker, KICKED);
yang-gdf92a642017-08-21 22:38:45 -07001037 ret_err = grpc_wakeup_fd_wakeup(&global_wakeup_fd);
1038 goto done;
Craig Tiller8502ecb2017-04-28 14:22:01 -07001039 } else if (next_worker->kick_state == UNKICKED) {
Craig Tiller75aef7f2017-05-26 08:26:08 -07001040 if (GRPC_TRACER_ON(grpc_polling_trace)) {
Craig Tiller830e82a2017-05-31 16:26:27 -07001041 gpr_log(GPR_ERROR, " .. kicked %p", next_worker);
Craig Tiller75aef7f2017-05-26 08:26:08 -07001042 }
Craig Tiller8502ecb2017-04-28 14:22:01 -07001043 GPR_ASSERT(next_worker->initialized_cv);
Craig Tiller55624a32017-05-26 08:14:44 -07001044 SET_KICK_STATE(next_worker, KICKED);
Craig Tiller375eb252017-04-27 23:29:12 +00001045 gpr_cv_signal(&next_worker->cv);
yang-gdf92a642017-08-21 22:38:45 -07001046 goto done;
Craig Tiller55624a32017-05-26 08:14:44 -07001047 } else if (next_worker->kick_state == DESIGNATED_POLLER) {
1048 if (root_worker->kick_state != DESIGNATED_POLLER) {
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(
1051 GPR_ERROR,
1052 " .. kicked root non-poller %p (initialized_cv=%d) (poller=%p)",
1053 root_worker, root_worker->initialized_cv, next_worker);
Craig Tiller75aef7f2017-05-26 08:26:08 -07001054 }
Craig Tiller55624a32017-05-26 08:14:44 -07001055 SET_KICK_STATE(root_worker, KICKED);
1056 if (root_worker->initialized_cv) {
1057 gpr_cv_signal(&root_worker->cv);
1058 }
yang-gdf92a642017-08-21 22:38:45 -07001059 goto done;
Craig Tiller55624a32017-05-26 08:14:44 -07001060 } else {
Craig Tiller75aef7f2017-05-26 08:26:08 -07001061 if (GRPC_TRACER_ON(grpc_polling_trace)) {
Craig Tiller830e82a2017-05-31 16:26:27 -07001062 gpr_log(GPR_ERROR, " .. non-root poller %p (root=%p)", next_worker,
Craig Tiller75aef7f2017-05-26 08:26:08 -07001063 root_worker);
1064 }
Craig Tiller55624a32017-05-26 08:14:44 -07001065 SET_KICK_STATE(next_worker, KICKED);
yang-gdf92a642017-08-21 22:38:45 -07001066 ret_err = grpc_wakeup_fd_wakeup(&global_wakeup_fd);
1067 goto done;
Craig Tiller55624a32017-05-26 08:14:44 -07001068 }
Craig Tiller8502ecb2017-04-28 14:22:01 -07001069 } else {
Craig Tiller55624a32017-05-26 08:14:44 -07001070 GPR_ASSERT(next_worker->kick_state == KICKED);
1071 SET_KICK_STATE(next_worker, KICKED);
yang-gdf92a642017-08-21 22:38:45 -07001072 goto done;
Craig Tiller4509c472017-04-27 19:05:13 +00001073 }
1074 } else {
Craig Tiller830e82a2017-05-31 16:26:27 -07001075 if (GRPC_TRACER_ON(grpc_polling_trace)) {
1076 gpr_log(GPR_ERROR, " .. kicked while waking up");
1077 }
yang-gdf92a642017-08-21 22:38:45 -07001078 goto done;
Craig Tiller4509c472017-04-27 19:05:13 +00001079 }
Craig Tiller43bf2592017-04-28 23:21:01 +00001080 } else if (specific_worker->kick_state == KICKED) {
Craig Tiller75aef7f2017-05-26 08:26:08 -07001081 if (GRPC_TRACER_ON(grpc_polling_trace)) {
Craig Tiller830e82a2017-05-31 16:26:27 -07001082 gpr_log(GPR_ERROR, " .. specific worker already kicked");
Craig Tiller75aef7f2017-05-26 08:26:08 -07001083 }
yang-gdf92a642017-08-21 22:38:45 -07001084 goto done;
Craig Tiller4509c472017-04-27 19:05:13 +00001085 } else if (gpr_tls_get(&g_current_thread_worker) ==
1086 (intptr_t)specific_worker) {
Craig Tiller75aef7f2017-05-26 08:26:08 -07001087 if (GRPC_TRACER_ON(grpc_polling_trace)) {
Craig Tiller830e82a2017-05-31 16:26:27 -07001088 gpr_log(GPR_ERROR, " .. mark %p kicked", specific_worker);
Craig Tiller75aef7f2017-05-26 08:26:08 -07001089 }
Craig Tiller55624a32017-05-26 08:14:44 -07001090 SET_KICK_STATE(specific_worker, KICKED);
yang-gdf92a642017-08-21 22:38:45 -07001091 goto done;
Craig Tiller32f90ee2017-04-28 12:46:41 -07001092 } else if (specific_worker ==
1093 (grpc_pollset_worker *)gpr_atm_no_barrier_load(&g_active_poller)) {
Craig Tiller75aef7f2017-05-26 08:26:08 -07001094 if (GRPC_TRACER_ON(grpc_polling_trace)) {
Craig Tiller830e82a2017-05-31 16:26:27 -07001095 gpr_log(GPR_ERROR, " .. kick active poller");
Craig Tiller75aef7f2017-05-26 08:26:08 -07001096 }
Craig Tiller55624a32017-05-26 08:14:44 -07001097 SET_KICK_STATE(specific_worker, KICKED);
yang-gdf92a642017-08-21 22:38:45 -07001098 ret_err = grpc_wakeup_fd_wakeup(&global_wakeup_fd);
1099 goto done;
Craig Tiller8502ecb2017-04-28 14:22:01 -07001100 } else if (specific_worker->initialized_cv) {
Craig Tiller75aef7f2017-05-26 08:26:08 -07001101 if (GRPC_TRACER_ON(grpc_polling_trace)) {
Craig Tiller830e82a2017-05-31 16:26:27 -07001102 gpr_log(GPR_ERROR, " .. kick waiting worker");
Craig Tiller75aef7f2017-05-26 08:26:08 -07001103 }
Craig Tiller55624a32017-05-26 08:14:44 -07001104 SET_KICK_STATE(specific_worker, KICKED);
Craig Tiller4509c472017-04-27 19:05:13 +00001105 gpr_cv_signal(&specific_worker->cv);
yang-gdf92a642017-08-21 22:38:45 -07001106 goto done;
Craig Tiller8502ecb2017-04-28 14:22:01 -07001107 } else {
Craig Tiller75aef7f2017-05-26 08:26:08 -07001108 if (GRPC_TRACER_ON(grpc_polling_trace)) {
Craig Tiller830e82a2017-05-31 16:26:27 -07001109 gpr_log(GPR_ERROR, " .. kick non-waiting worker");
Craig Tiller75aef7f2017-05-26 08:26:08 -07001110 }
Craig Tiller55624a32017-05-26 08:14:44 -07001111 SET_KICK_STATE(specific_worker, KICKED);
yang-gdf92a642017-08-21 22:38:45 -07001112 goto done;
Craig Tiller4509c472017-04-27 19:05:13 +00001113 }
yang-gdf92a642017-08-21 22:38:45 -07001114done:
1115 GPR_TIMER_END("pollset_kick", 0);
1116 return ret_err;
Craig Tiller4509c472017-04-27 19:05:13 +00001117}
1118
1119static void pollset_add_fd(grpc_exec_ctx *exec_ctx, grpc_pollset *pollset,
1120 grpc_fd *fd) {}
1121
Craig Tiller4509c472017-04-27 19:05:13 +00001122/*******************************************************************************
Craig Tillerc67cc992017-04-27 10:15:51 -07001123 * Pollset-set Definitions
1124 */
1125
1126static grpc_pollset_set *pollset_set_create(void) {
1127 return (grpc_pollset_set *)((intptr_t)0xdeafbeef);
1128}
1129
1130static void pollset_set_destroy(grpc_exec_ctx *exec_ctx,
1131 grpc_pollset_set *pss) {}
1132
1133static void pollset_set_add_fd(grpc_exec_ctx *exec_ctx, grpc_pollset_set *pss,
1134 grpc_fd *fd) {}
1135
1136static void pollset_set_del_fd(grpc_exec_ctx *exec_ctx, grpc_pollset_set *pss,
1137 grpc_fd *fd) {}
1138
1139static void pollset_set_add_pollset(grpc_exec_ctx *exec_ctx,
1140 grpc_pollset_set *pss, grpc_pollset *ps) {}
1141
1142static void pollset_set_del_pollset(grpc_exec_ctx *exec_ctx,
1143 grpc_pollset_set *pss, grpc_pollset *ps) {}
1144
1145static void pollset_set_add_pollset_set(grpc_exec_ctx *exec_ctx,
1146 grpc_pollset_set *bag,
1147 grpc_pollset_set *item) {}
1148
1149static void pollset_set_del_pollset_set(grpc_exec_ctx *exec_ctx,
1150 grpc_pollset_set *bag,
1151 grpc_pollset_set *item) {}
1152
1153/*******************************************************************************
1154 * Event engine binding
1155 */
1156
1157static void shutdown_engine(void) {
1158 fd_global_shutdown();
1159 pollset_global_shutdown();
Sree Kuchibhotla5efc9132017-08-17 14:10:38 -07001160 epoll_set_shutdown();
Craig Tillerc67cc992017-04-27 10:15:51 -07001161}
1162
1163static const grpc_event_engine_vtable vtable = {
1164 .pollset_size = sizeof(grpc_pollset),
1165
1166 .fd_create = fd_create,
1167 .fd_wrapped_fd = fd_wrapped_fd,
1168 .fd_orphan = fd_orphan,
1169 .fd_shutdown = fd_shutdown,
1170 .fd_is_shutdown = fd_is_shutdown,
1171 .fd_notify_on_read = fd_notify_on_read,
1172 .fd_notify_on_write = fd_notify_on_write,
1173 .fd_get_read_notifier_pollset = fd_get_read_notifier_pollset,
Craig Tillerc67cc992017-04-27 10:15:51 -07001174
1175 .pollset_init = pollset_init,
1176 .pollset_shutdown = pollset_shutdown,
1177 .pollset_destroy = pollset_destroy,
1178 .pollset_work = pollset_work,
1179 .pollset_kick = pollset_kick,
1180 .pollset_add_fd = pollset_add_fd,
1181
1182 .pollset_set_create = pollset_set_create,
1183 .pollset_set_destroy = pollset_set_destroy,
1184 .pollset_set_add_pollset = pollset_set_add_pollset,
1185 .pollset_set_del_pollset = pollset_set_del_pollset,
1186 .pollset_set_add_pollset_set = pollset_set_add_pollset_set,
1187 .pollset_set_del_pollset_set = pollset_set_del_pollset_set,
1188 .pollset_set_add_fd = pollset_set_add_fd,
1189 .pollset_set_del_fd = pollset_set_del_fd,
1190
Craig Tillerc67cc992017-04-27 10:15:51 -07001191 .shutdown_engine = shutdown_engine,
1192};
1193
1194/* It is possible that GLIBC has epoll but the underlying kernel doesn't.
Sree Kuchibhotla5efc9132017-08-17 14:10:38 -07001195 * Create epoll_fd (epoll_set_init() takes care of that) to make sure epoll
1196 * support is available */
Craig Tiller6f0af492017-04-27 19:26:16 +00001197const grpc_event_engine_vtable *grpc_init_epoll1_linux(bool explicit_request) {
Sree Kuchibhotla107840f2017-07-24 18:45:09 -07001198 if (!explicit_request) {
1199 return NULL;
1200 }
1201
Craig Tillerc67cc992017-04-27 10:15:51 -07001202 if (!grpc_has_wakeup_fd()) {
1203 return NULL;
1204 }
1205
Sree Kuchibhotla5efc9132017-08-17 14:10:38 -07001206 if (!epoll_set_init()) {
Craig Tillerc67cc992017-04-27 10:15:51 -07001207 return NULL;
1208 }
1209
Craig Tillerc67cc992017-04-27 10:15:51 -07001210 fd_global_init();
1211
1212 if (!GRPC_LOG_IF_ERROR("pollset_global_init", pollset_global_init())) {
Craig Tiller4509c472017-04-27 19:05:13 +00001213 fd_global_shutdown();
Sree Kuchibhotla5efc9132017-08-17 14:10:38 -07001214 epoll_set_shutdown();
Craig Tillerc67cc992017-04-27 10:15:51 -07001215 return NULL;
1216 }
1217
1218 return &vtable;
1219}
1220
1221#else /* defined(GRPC_LINUX_EPOLL) */
1222#if defined(GRPC_POSIX_SOCKET)
1223#include "src/core/lib/iomgr/ev_posix.h"
1224/* If GRPC_LINUX_EPOLL is not defined, it means epoll is not available. Return
1225 * NULL */
Craig Tiller9ddb3152017-04-27 21:32:56 +00001226const grpc_event_engine_vtable *grpc_init_epoll1_linux(bool explicit_request) {
1227 return NULL;
1228}
Craig Tillerc67cc992017-04-27 10:15:51 -07001229#endif /* defined(GRPC_POSIX_SOCKET) */
1230#endif /* !defined(GRPC_LINUX_EPOLL) */