diff --git a/include/grpc/support/slice_buffer.h b/include/grpc/support/slice_buffer.h
index f3f14a6aa0b6903bdd4adbfdc7edfb483c670bdd..e4f296c32a300f845655d18d6c81b42bf03b12fa 100644
--- a/include/grpc/support/slice_buffer.h
+++ b/include/grpc/support/slice_buffer.h
@@ -87,7 +87,8 @@ void gpr_slice_buffer_swap(gpr_slice_buffer *a, gpr_slice_buffer *b);
/* move all of the elements of src into dst */
void gpr_slice_buffer_move_into(gpr_slice_buffer *src, gpr_slice_buffer *dst);
/* remove n bytes from the end of a slice buffer */
-void gpr_slice_buffer_trim_end(gpr_slice_buffer *src, size_t n, gpr_slice_buffer *garbage);
+void gpr_slice_buffer_trim_end(gpr_slice_buffer *src, size_t n,
+ gpr_slice_buffer *garbage);
#ifdef __cplusplus
}
diff --git a/src/core/iomgr/iomgr.c b/src/core/iomgr/iomgr.c
index a10399311fc8ea0d67209e23ec5ccf98286a2339..e61fc32925fd9f20cebda6e35a2c7873fca85ebb 100644
--- a/src/core/iomgr/iomgr.c
+++ b/src/core/iomgr/iomgr.c
@@ -51,13 +51,6 @@ static gpr_cv g_rcv;
static int g_shutdown;
static grpc_iomgr_object g_root_object;
-void grpc_kick_poller(void) {
- /* Empty. The background callback executor polls periodically. The activity
- * the kicker is trying to draw the executor's attention to will be picked up
- * either by one of the periodic wakeups or by one of the polling application
- * threads. */
-}
-
void grpc_iomgr_init(void) {
g_shutdown = 0;
gpr_mu_init(&g_mu);
diff --git a/src/core/iomgr/pollset_multipoller_with_epoll.c b/src/core/iomgr/pollset_multipoller_with_epoll.c
index b22eaa62889201ec0f2b48083634480f1b8ab295..d26e60f665007f075544e44111f27b8f8d2f006b 100644
--- a/src/core/iomgr/pollset_multipoller_with_epoll.c
+++ b/src/core/iomgr/pollset_multipoller_with_epoll.c
@@ -213,11 +213,15 @@ static void multipoll_with_epoll_pollset_maybe_work_and_unlock(
int cancel = ep_ev[i].events & (EPOLLERR | EPOLLHUP);
int read_ev = ep_ev[i].events & (EPOLLIN | EPOLLPRI);
int write_ev = ep_ev[i].events & EPOLLOUT;
- if (read_ev || cancel) {
- grpc_fd_become_readable(exec_ctx, fd);
- }
- if (write_ev || cancel) {
- grpc_fd_become_writable(exec_ctx, fd);
+ if (fd == NULL) {
+ grpc_wakeup_fd_consume_wakeup(&grpc_global_wakeup_fd);
+ } else {
+ if (read_ev || cancel) {
+ grpc_fd_become_readable(exec_ctx, fd);
+ }
+ if (write_ev || cancel) {
+ grpc_fd_become_writable(exec_ctx, fd);
+ }
}
}
}
@@ -246,6 +250,8 @@ static void epoll_become_multipoller(grpc_exec_ctx *exec_ctx,
size_t nfds) {
size_t i;
pollset_hdr *h = gpr_malloc(sizeof(pollset_hdr));
+ struct epoll_event ev;
+ int err;
pollset->vtable = &multipoll_with_epoll_pollset;
pollset->data.ptr = h;
@@ -255,6 +261,15 @@ static void epoll_become_multipoller(grpc_exec_ctx *exec_ctx,
gpr_log(GPR_ERROR, "epoll_create1 failed: %s", strerror(errno));
abort();
}
+
+ ev.events = (uint32_t)(EPOLLIN | EPOLLET);
+ ev.data.ptr = NULL;
+ err = epoll_ctl(h->epoll_fd, EPOLL_CTL_ADD, GRPC_WAKEUP_FD_GET_READ_FD(&grpc_global_wakeup_fd), &ev);
+ if (err < 0) {
+ gpr_log(GPR_ERROR, "epoll_ctl add for %d failed: %s", GRPC_WAKEUP_FD_GET_READ_FD(&grpc_global_wakeup_fd),
+ strerror(errno));
+ }
+
for (i = 0; i < nfds; i++) {
multipoll_with_epoll_pollset_add_fd(exec_ctx, pollset, fds[i], 0);
}
diff --git a/src/core/iomgr/pollset_multipoller_with_poll_posix.c b/src/core/iomgr/pollset_multipoller_with_poll_posix.c
index 63e0b9edb9ac4686247046fa1a42a2c91a75b418..1356ebe7a0659a40b3204a225b97b36e8c1d4d44 100644
--- a/src/core/iomgr/pollset_multipoller_with_poll_posix.c
+++ b/src/core/iomgr/pollset_multipoller_with_poll_posix.c
@@ -114,13 +114,16 @@ static void multipoll_with_poll_pollset_maybe_work_and_unlock(
h = pollset->data.ptr;
timeout = grpc_poll_deadline_to_millis_timeout(deadline, now);
/* TODO(ctiller): perform just one malloc here if we exceed the inline case */
- pfds = gpr_malloc(sizeof(*pfds) * (h->fd_count + 1));
- watchers = gpr_malloc(sizeof(*watchers) * (h->fd_count + 1));
+ pfds = gpr_malloc(sizeof(*pfds) * (h->fd_count + 2));
+ watchers = gpr_malloc(sizeof(*watchers) * (h->fd_count + 2));
fd_count = 0;
- pfd_count = 1;
- pfds[0].fd = GRPC_WAKEUP_FD_GET_READ_FD(&worker->wakeup_fd);
+ pfd_count = 2;
+ pfds[0].fd = GRPC_WAKEUP_FD_GET_READ_FD(&grpc_global_wakeup_fd);
pfds[0].events = POLLIN;
- pfds[0].revents = POLLOUT;
+ pfds[0].revents = 0;
+ pfds[1].fd = GRPC_WAKEUP_FD_GET_READ_FD(&worker->wakeup_fd);
+ pfds[1].events = POLLIN;
+ pfds[1].revents = 0;
for (i = 0; i < h->fd_count; i++) {
int remove = grpc_fd_is_orphaned(h->fds[i]);
for (j = 0; !remove && j < h->del_count; j++) {
@@ -143,7 +146,7 @@ static void multipoll_with_poll_pollset_maybe_work_and_unlock(
h->fd_count = fd_count;
gpr_mu_unlock(&pollset->mu);
- for (i = 1; i < pfd_count; i++) {
+ for (i = 2; i < pfd_count; i++) {
pfds[i].events = (short)grpc_fd_begin_poll(watchers[i].fd, pollset, POLLIN,
POLLOUT, &watchers[i]);
}
@@ -154,7 +157,7 @@ static void multipoll_with_poll_pollset_maybe_work_and_unlock(
r = grpc_poll_function(pfds, pfd_count, timeout);
GRPC_SCHEDULING_END_BLOCKING_REGION;
- for (i = 1; i < pfd_count; i++) {
+ for (i = 2; i < pfd_count; i++) {
grpc_fd_end_poll(exec_ctx, &watchers[i], pfds[i].revents & POLLIN,
pfds[i].revents & POLLOUT);
}
@@ -167,9 +170,12 @@ static void multipoll_with_poll_pollset_maybe_work_and_unlock(
/* do nothing */
} else {
if (pfds[0].revents & POLLIN) {
+ grpc_wakeup_fd_consume_wakeup(&grpc_global_wakeup_fd);
+ }
+ if (pfds[1].revents & POLLIN) {
grpc_wakeup_fd_consume_wakeup(&worker->wakeup_fd);
}
- for (i = 1; i < pfd_count; i++) {
+ for (i = 2; i < pfd_count; i++) {
if (watchers[i].fd == NULL) {
continue;
}
diff --git a/src/core/iomgr/pollset_posix.c b/src/core/iomgr/pollset_posix.c
index f9d6aad651fba769b71d4d6185ecab30bafcf52c..82a82cc06447c0c52f5e94d9d848466fb08f6f99 100644
--- a/src/core/iomgr/pollset_posix.c
+++ b/src/core/iomgr/pollset_posix.c
@@ -57,8 +57,16 @@
GPR_TLS_DECL(g_current_thread_poller);
GPR_TLS_DECL(g_current_thread_worker);
+/** Default poll() function - a pointer so that it can be overridden by some
+ * tests */
grpc_poll_function_type grpc_poll_function = poll;
+/** The alarm system needs to be able to wakeup 'some poller' sometimes
+ * (specifically when a new alarm needs to be triggered earlier than the next
+ * alarm 'epoch').
+ * This wakeup_fd gives us something to alert on when such a case occurs. */
+grpc_wakeup_fd grpc_global_wakeup_fd;
+
static void remove_worker(grpc_pollset *p, grpc_pollset_worker *worker) {
worker->prev->next = worker->next;
worker->next->prev = worker->prev;
@@ -121,12 +129,18 @@ void grpc_pollset_global_init(void) {
gpr_tls_init(&g_current_thread_poller);
gpr_tls_init(&g_current_thread_worker);
grpc_wakeup_fd_global_init();
+ grpc_wakeup_fd_init(&grpc_global_wakeup_fd);
}
void grpc_pollset_global_shutdown(void) {
+ grpc_wakeup_fd_destroy(&grpc_global_wakeup_fd);
+ grpc_wakeup_fd_global_destroy();
gpr_tls_destroy(&g_current_thread_poller);
gpr_tls_destroy(&g_current_thread_worker);
- grpc_wakeup_fd_global_destroy();
+}
+
+void grpc_kick_poller(void) {
+ grpc_wakeup_fd_wakeup(&grpc_global_wakeup_fd);
}
/* main interface */
@@ -193,6 +207,8 @@ void grpc_pollset_work(grpc_exec_ctx *exec_ctx, grpc_pollset *pollset,
goto done;
}
if (grpc_alarm_check(exec_ctx, now, &deadline)) {
+ gpr_mu_unlock(&pollset->mu);
+ locked = 0;
goto done;
}
if (pollset->shutting_down) {
@@ -294,7 +310,7 @@ int grpc_poll_deadline_to_millis_timeout(gpr_timespec deadline,
}
timeout = gpr_time_sub(deadline, now);
return gpr_time_to_millis(gpr_time_add(
- timeout, gpr_time_from_nanos(GPR_NS_PER_SEC - 1, GPR_TIMESPAN)));
+ timeout, gpr_time_from_nanos(GPR_NS_PER_MS - 1, GPR_TIMESPAN)));
}
/*
@@ -439,7 +455,7 @@ static void basic_pollset_maybe_work_and_unlock(grpc_exec_ctx *exec_ctx,
grpc_pollset_worker *worker,
gpr_timespec deadline,
gpr_timespec now) {
- struct pollfd pfd[2];
+ struct pollfd pfd[3];
grpc_fd *fd;
grpc_fd_watcher fd_watcher;
int timeout;
@@ -452,17 +468,20 @@ static void basic_pollset_maybe_work_and_unlock(grpc_exec_ctx *exec_ctx,
fd = pollset->data.ptr = NULL;
}
timeout = grpc_poll_deadline_to_millis_timeout(deadline, now);
- pfd[0].fd = GRPC_WAKEUP_FD_GET_READ_FD(&worker->wakeup_fd);
+ pfd[0].fd = GRPC_WAKEUP_FD_GET_READ_FD(&grpc_global_wakeup_fd);
pfd[0].events = POLLIN;
pfd[0].revents = 0;
- nfds = 1;
+ pfd[1].fd = GRPC_WAKEUP_FD_GET_READ_FD(&worker->wakeup_fd);
+ pfd[1].events = POLLIN;
+ pfd[1].revents = 0;
+ nfds = 2;
if (fd) {
- pfd[1].fd = fd->fd;
- pfd[1].revents = 0;
+ pfd[2].fd = fd->fd;
+ pfd[2].revents = 0;
gpr_mu_unlock(&pollset->mu);
- pfd[1].events =
+ pfd[2].events =
(short)grpc_fd_begin_poll(fd, pollset, POLLIN, POLLOUT, &fd_watcher);
- if (pfd[1].events != 0) {
+ if (pfd[2].events != 0) {
nfds++;
}
} else {
@@ -479,8 +498,8 @@ static void basic_pollset_maybe_work_and_unlock(grpc_exec_ctx *exec_ctx,
GRPC_TIMER_MARK(GRPC_PTAG_POLL_FINISHED, r);
if (fd) {
- grpc_fd_end_poll(exec_ctx, &fd_watcher, pfd[1].revents & POLLIN,
- pfd[1].revents & POLLOUT);
+ grpc_fd_end_poll(exec_ctx, &fd_watcher, pfd[2].revents & POLLIN,
+ pfd[2].revents & POLLOUT);
}
if (r < 0) {
@@ -491,13 +510,16 @@ static void basic_pollset_maybe_work_and_unlock(grpc_exec_ctx *exec_ctx,
/* do nothing */
} else {
if (pfd[0].revents & POLLIN) {
+ grpc_wakeup_fd_consume_wakeup(&grpc_global_wakeup_fd);
+ }
+ if (pfd[1].revents & POLLIN) {
grpc_wakeup_fd_consume_wakeup(&worker->wakeup_fd);
}
- if (nfds > 1) {
- if (pfd[1].revents & (POLLIN | POLLHUP | POLLERR)) {
+ if (nfds > 2) {
+ if (pfd[2].revents & (POLLIN | POLLHUP | POLLERR)) {
grpc_fd_become_readable(exec_ctx, fd);
}
- if (pfd[1].revents & (POLLOUT | POLLHUP | POLLERR)) {
+ if (pfd[2].revents & (POLLOUT | POLLHUP | POLLERR)) {
grpc_fd_become_writable(exec_ctx, fd);
}
}
diff --git a/src/core/iomgr/pollset_posix.h b/src/core/iomgr/pollset_posix.h
index f996dd1edfb0aa5a44e41afb74193f75c641f53d..83c525853905b41dd0f06ace430362fa3dd7be48 100644
--- a/src/core/iomgr/pollset_posix.h
+++ b/src/core/iomgr/pollset_posix.h
@@ -129,5 +129,6 @@ int grpc_pollset_has_workers(grpc_pollset *pollset);
/* 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_INTERNAL_CORE_IOMGR_POLLSET_POSIX_H */
diff --git a/src/core/iomgr/pollset_windows.c b/src/core/iomgr/pollset_windows.c
index 798b6376358940cd70cd73d8d5ce61082308c033..1b4935a8416f31a6f7f890acff3f498eb3c81b23 100644
--- a/src/core/iomgr/pollset_windows.c
+++ b/src/core/iomgr/pollset_windows.c
@@ -230,4 +230,8 @@ void grpc_pollset_kick(grpc_pollset *p, grpc_pollset_worker *specific_worker) {
}
}
+void grpc_kick_poller(void) {
+ grpc_iocp_kick();
+}
+
#endif /* GPR_WINSOCK_SOCKET */
diff --git a/test/core/end2end/tests/channel_connectivity.c b/test/core/end2end/tests/channel_connectivity.c
index 0b7a8a664b146a42536edf14a4296f60b81b3f68..be9b7fde1dd1534ed2efc25097ea978f1b112614 100644
--- a/test/core/end2end/tests/channel_connectivity.c
+++ b/test/core/end2end/tests/channel_connectivity.c
@@ -34,19 +34,49 @@
#include "test/core/end2end/end2end_tests.h"
#include <grpc/support/log.h>
+#include <grpc/support/sync.h>
+#include <grpc/support/thd.h>
#include <grpc/support/time.h>
#include "test/core/end2end/cq_verifier.h"
static void *tag(gpr_intptr t) { return (void *)t; }
+typedef struct {
+ gpr_event started;
+ grpc_channel *channel;
+ grpc_completion_queue *cq;
+} child_events;
+
+static void child_thread(void *arg) {
+ child_events *ce = arg;
+ grpc_event ev;
+ gpr_event_set(&ce->started, (void*)1);
+ gpr_log(GPR_DEBUG, "verifying");
+ ev = grpc_completion_queue_next(ce->cq, gpr_inf_future(GPR_CLOCK_MONOTONIC), NULL);
+ GPR_ASSERT(ev.type == GRPC_OP_COMPLETE);
+ GPR_ASSERT(ev.tag == tag(1));
+ GPR_ASSERT(ev.success == 0);
+}
+
static void test_connectivity(grpc_end2end_test_config config) {
grpc_end2end_test_fixture f = config.create_fixture(NULL, NULL);
grpc_connectivity_state state;
cq_verifier *cqv = cq_verifier_create(f.cq);
+ child_events ce;
+ gpr_thd_options thdopt = gpr_thd_options_default();
+ gpr_thd_id thdid;
config.init_client(&f, NULL);
+ ce.channel = f.client;
+ ce.cq = f.cq;
+ gpr_event_init(&ce.started);
+ gpr_thd_options_set_joinable(&thdopt);
+ GPR_ASSERT(gpr_thd_new(&thdid, child_thread, &ce, &thdopt));
+
+ gpr_event_wait(&ce.started, gpr_inf_future(GPR_CLOCK_MONOTONIC));
+
/* channels should start life in IDLE, and stay there */
GPR_ASSERT(grpc_channel_check_connectivity_state(f.client, 0) ==
GRPC_CHANNEL_IDLE);
@@ -55,18 +85,24 @@ static void test_connectivity(grpc_end2end_test_config config) {
GRPC_CHANNEL_IDLE);
/* start watching for a change */
+ gpr_log(GPR_DEBUG, "watching");
grpc_channel_watch_connectivity_state(f.client, GRPC_CHANNEL_IDLE,
- GRPC_TIMEOUT_SECONDS_TO_DEADLINE(3),
+ gpr_now(GPR_CLOCK_MONOTONIC),
f.cq, tag(1));
- /* nothing should happen */
- cq_verify_empty(cqv);
+
+ /* eventually the child thread completion should trigger */
+ gpr_thd_join(thdid);
/* check that we're still in idle, and start connecting */
GPR_ASSERT(grpc_channel_check_connectivity_state(f.client, 1) ==
GRPC_CHANNEL_IDLE);
+ /* start watching for a change */
+ grpc_channel_watch_connectivity_state(f.client, GRPC_CHANNEL_IDLE,
+ GRPC_TIMEOUT_SECONDS_TO_DEADLINE(3),
+ f.cq, tag(2));
/* and now the watch should trigger */
- cq_expect_completion(cqv, tag(1), 1);
+ cq_expect_completion(cqv, tag(2), 1);
cq_verify(cqv);
state = grpc_channel_check_connectivity_state(f.client, 0);
GPR_ASSERT(state == GRPC_CHANNEL_TRANSIENT_FAILURE ||
@@ -75,8 +111,8 @@ static void test_connectivity(grpc_end2end_test_config config) {
/* quickly followed by a transition to TRANSIENT_FAILURE */
grpc_channel_watch_connectivity_state(f.client, GRPC_CHANNEL_CONNECTING,
GRPC_TIMEOUT_SECONDS_TO_DEADLINE(3),
- f.cq, tag(2));
- cq_expect_completion(cqv, tag(2), 1);
+ f.cq, tag(3));
+ cq_expect_completion(cqv, tag(3), 1);
cq_verify(cqv);
state = grpc_channel_check_connectivity_state(f.client, 0);
GPR_ASSERT(state == GRPC_CHANNEL_TRANSIENT_FAILURE ||
@@ -93,8 +129,8 @@ static void test_connectivity(grpc_end2end_test_config config) {
READY is reached */
while (state != GRPC_CHANNEL_READY) {
grpc_channel_watch_connectivity_state(
- f.client, state, GRPC_TIMEOUT_SECONDS_TO_DEADLINE(3), f.cq, tag(3));
- cq_expect_completion(cqv, tag(3), 1);
+ f.client, state, GRPC_TIMEOUT_SECONDS_TO_DEADLINE(3), f.cq, tag(4));
+ cq_expect_completion(cqv, tag(4), 1);
cq_verify(cqv);
state = grpc_channel_check_connectivity_state(f.client, 0);
GPR_ASSERT(state == GRPC_CHANNEL_READY ||
@@ -108,11 +144,11 @@ static void test_connectivity(grpc_end2end_test_config config) {
grpc_channel_watch_connectivity_state(f.client, GRPC_CHANNEL_READY,
GRPC_TIMEOUT_SECONDS_TO_DEADLINE(3),
- f.cq, tag(4));
+ f.cq, tag(5));
grpc_server_shutdown_and_notify(f.server, f.cq, tag(0xdead));
- cq_expect_completion(cqv, tag(4), 1);
+ cq_expect_completion(cqv, tag(5), 1);
cq_expect_completion(cqv, tag(0xdead), 1);
cq_verify(cqv);
state = grpc_channel_check_connectivity_state(f.client, 0);