diff options
author | 2016-11-28 14:34:41 -0800 | |
---|---|---|
committer | 2016-11-28 14:34:41 -0800 | |
commit | d50a068a6dfcf0dd464313a348f4ed9e4f5a06f5 (patch) | |
tree | 51a25d9463471d1a5a51326f53f7cb0c2021e074 /src/core/lib/iomgr | |
parent | 4fb81553eceed83c1b441ad4ec4c1347dd61bd70 (diff) | |
parent | ca75ca65e0c6acba0a5ed456577cd4bcefb92e03 (diff) |
Merge branch 'master' into pss_pi
Diffstat (limited to 'src/core/lib/iomgr')
-rw-r--r-- | src/core/lib/iomgr/ev_epoll_linux.c | 31 | ||||
-rw-r--r-- | src/core/lib/iomgr/ev_poll_posix.c | 16 | ||||
-rw-r--r-- | src/core/lib/iomgr/ev_posix.h | 1 | ||||
-rw-r--r-- | src/core/lib/iomgr/tcp_windows.c | 2 |
4 files changed, 24 insertions, 26 deletions
diff --git a/src/core/lib/iomgr/ev_epoll_linux.c b/src/core/lib/iomgr/ev_epoll_linux.c index 16a22b7b9b..5d54ebc026 100644 --- a/src/core/lib/iomgr/ev_epoll_linux.c +++ b/src/core/lib/iomgr/ev_epoll_linux.c @@ -72,6 +72,11 @@ static int grpc_polling_trace = 0; /* Disabled by default */ static int grpc_wakeup_signal = -1; static bool is_grpc_wakeup_signal_initialized = false; +/* TODO: sreek: Right now, this wakes up all pollers. In future we should make + * sure to wake up one polling thread (which can wake up other threads if + * needed) */ +static grpc_wakeup_fd global_wakeup_fd; + /* Implements the function defined in grpc_posix.h. This function might be * called before even calling grpc_init() to set either a different signal to * use. If signum == -1, then the use of signals is disabled */ @@ -444,9 +449,8 @@ static void polling_island_add_wakeup_fd_locked(polling_island *pi, gpr_asprintf(&err_msg, "epoll_ctl (epoll_fd: %d) add wakeup fd: %d failed with " "error: %d (%s)", - pi->epoll_fd, - GRPC_WAKEUP_FD_GET_READ_FD(&grpc_global_wakeup_fd), errno, - strerror(errno)); + pi->epoll_fd, GRPC_WAKEUP_FD_GET_READ_FD(&global_wakeup_fd), + errno, strerror(errno)); append_error(error, GRPC_OS_ERROR(errno, err_msg), err_desc); gpr_free(err_msg); } @@ -548,7 +552,7 @@ static polling_island *polling_island_create(grpc_exec_ctx *exec_ctx, goto done; } - polling_island_add_wakeup_fd_locked(pi, &grpc_global_wakeup_fd, error); + polling_island_add_wakeup_fd_locked(pi, &global_wakeup_fd, error); polling_island_add_wakeup_fd_locked(pi, &pi->workqueue_wakeup_fd, error); if (initial_fd != NULL) { @@ -850,11 +854,6 @@ static void polling_island_global_shutdown() { * alarm 'epoch'). This wakeup_fd gives us something to alert on when such a * case occurs. */ -/* TODO: sreek: Right now, this wakes up all pollers. In future we should make - * sure to wake up one polling thread (which can wake up other threads if - * needed) */ -grpc_wakeup_fd grpc_global_wakeup_fd; - static grpc_fd *fd_freelist = NULL; static gpr_mu fd_freelist_mu; @@ -1169,11 +1168,11 @@ static grpc_error *pollset_global_init(void) { gpr_tls_init(&g_current_thread_pollset); gpr_tls_init(&g_current_thread_worker); poller_kick_init(); - return grpc_wakeup_fd_init(&grpc_global_wakeup_fd); + return grpc_wakeup_fd_init(&global_wakeup_fd); } static void pollset_global_shutdown(void) { - grpc_wakeup_fd_destroy(&grpc_global_wakeup_fd); + grpc_wakeup_fd_destroy(&global_wakeup_fd); gpr_tls_destroy(&g_current_thread_pollset); gpr_tls_destroy(&g_current_thread_worker); } @@ -1280,7 +1279,7 @@ static grpc_error *pollset_kick(grpc_pollset *p, } static grpc_error *kick_poller(void) { - return grpc_wakeup_fd_wakeup(&grpc_global_wakeup_fd); + return grpc_wakeup_fd_wakeup(&global_wakeup_fd); } static void pollset_init(grpc_pollset *pollset, gpr_mu **mu) { @@ -1509,13 +1508,11 @@ static void pollset_work_and_unlock(grpc_exec_ctx *exec_ctx, for (int i = 0; i < ep_rv; ++i) { void *data_ptr = ep_ev[i].data.ptr; - if (data_ptr == &grpc_global_wakeup_fd) { - append_error(error, - grpc_wakeup_fd_consume_wakeup(&grpc_global_wakeup_fd), + if (data_ptr == &global_wakeup_fd) { + append_error(error, grpc_wakeup_fd_consume_wakeup(&global_wakeup_fd), err_desc); } else if (data_ptr == &pi->workqueue_wakeup_fd) { - append_error(error, - grpc_wakeup_fd_consume_wakeup(&grpc_global_wakeup_fd), + append_error(error, grpc_wakeup_fd_consume_wakeup(&global_wakeup_fd), err_desc); maybe_do_workqueue_work(exec_ctx, pi); } else if (data_ptr == &polling_island_wakeup_fd) { diff --git a/src/core/lib/iomgr/ev_poll_posix.c b/src/core/lib/iomgr/ev_poll_posix.c index 27083ca6d0..21b28e5554 100644 --- a/src/core/lib/iomgr/ev_poll_posix.c +++ b/src/core/lib/iomgr/ev_poll_posix.c @@ -120,6 +120,8 @@ struct grpc_fd { grpc_pollset *read_notifier_pollset; }; +static grpc_wakeup_fd global_wakeup_fd; + /* Begin polling on an fd. Registers that the given pollset is interested in this fd - so that if read or writability interest changes, the pollset can be kicked to pick up that @@ -769,17 +771,17 @@ static grpc_error *pollset_kick(grpc_pollset *p, static grpc_error *pollset_global_init(void) { gpr_tls_init(&g_current_thread_poller); gpr_tls_init(&g_current_thread_worker); - return grpc_wakeup_fd_init(&grpc_global_wakeup_fd); + return grpc_wakeup_fd_init(&global_wakeup_fd); } static void pollset_global_shutdown(void) { - grpc_wakeup_fd_destroy(&grpc_global_wakeup_fd); + grpc_wakeup_fd_destroy(&global_wakeup_fd); gpr_tls_destroy(&g_current_thread_poller); gpr_tls_destroy(&g_current_thread_worker); } static grpc_error *kick_poller(void) { - return grpc_wakeup_fd_wakeup(&grpc_global_wakeup_fd); + return grpc_wakeup_fd_wakeup(&global_wakeup_fd); } /* main interface */ @@ -947,7 +949,7 @@ static grpc_error *pollset_work(grpc_exec_ctx *exec_ctx, grpc_pollset *pollset, fd_count = 0; pfd_count = 2; - pfds[0].fd = GRPC_WAKEUP_FD_GET_READ_FD(&grpc_global_wakeup_fd); + pfds[0].fd = GRPC_WAKEUP_FD_GET_READ_FD(&global_wakeup_fd); pfds[0].events = POLLIN; pfds[0].revents = 0; pfds[1].fd = GRPC_WAKEUP_FD_GET_READ_FD(&worker.wakeup_fd->fd); @@ -1001,8 +1003,8 @@ static grpc_error *pollset_work(grpc_exec_ctx *exec_ctx, grpc_pollset *pollset, } } else { if (pfds[0].revents & POLLIN_CHECK) { - work_combine_error( - &error, grpc_wakeup_fd_consume_wakeup(&grpc_global_wakeup_fd)); + work_combine_error(&error, + grpc_wakeup_fd_consume_wakeup(&global_wakeup_fd)); } if (pfds[1].revents & POLLIN_CHECK) { work_combine_error( @@ -1361,7 +1363,7 @@ static int cvfd_poll(struct pollfd *fds, nfds_t nfds, int timeout) { g_cvfds.cvfds[idx].cvs = cvn; // Don't bother polling if a wakeup fd is ready if (g_cvfds.cvfds[idx].is_set) { - skip_poll=1; + skip_poll = 1; } } else if (fds[i].fd >= 0) { nsockfds++; diff --git a/src/core/lib/iomgr/ev_posix.h b/src/core/lib/iomgr/ev_posix.h index 2fdef06838..cb5832539d 100644 --- a/src/core/lib/iomgr/ev_posix.h +++ b/src/core/lib/iomgr/ev_posix.h @@ -183,6 +183,5 @@ void grpc_pollset_set_del_fd(grpc_exec_ctx *exec_ctx, /* override to allow tests to hook poll() usage */ typedef int (*grpc_poll_function_type)(struct pollfd *, nfds_t, int); extern grpc_poll_function_type grpc_poll_function; -extern grpc_wakeup_fd grpc_global_wakeup_fd; #endif /* GRPC_CORE_LIB_IOMGR_EV_POSIX_H */ diff --git a/src/core/lib/iomgr/tcp_windows.c b/src/core/lib/iomgr/tcp_windows.c index 62afbcef51..d4613b674e 100644 --- a/src/core/lib/iomgr/tcp_windows.c +++ b/src/core/lib/iomgr/tcp_windows.c @@ -423,7 +423,7 @@ grpc_endpoint *grpc_tcp_create(grpc_winsocket *socket, tcp->base.vtable = &vtable; tcp->socket = socket; gpr_mu_init(&tcp->mu); - gpr_ref_init(&tcp->refcount, 2); + gpr_ref_init(&tcp->refcount, 1); grpc_closure_init(&tcp->on_read, on_read, tcp); grpc_closure_init(&tcp->on_write, on_write, tcp); tcp->peer_string = gpr_strdup(peer_string); |