Merge pull request #3555 from ctiller/pfft

Fix alarms

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
 }

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);

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);
   }

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;
       }

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);
       }
     }

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 */

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 */

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);