Merge pull request #8810 from htuch/pollset-fix

Fix workqueue_move_items_to_parent invocation bug in polling_island_merge

src/core/lib/iomgr/ev_epoll_linux.c

@@ -796,7 +796,7 @@ static polling_island *polling_island_merge(polling_island *p,
     gpr_atm_rel_store(&p->merged_to, (gpr_atm)q);
     PI_ADD_REF(q, "pi_merge"); /* To account for the new incoming ref from p */
 
-    workqueue_move_items_to_parent(q);
+    workqueue_move_items_to_parent(p);
   }
   /* else if p == q, nothing needs to be done */
 

test/core/iomgr/ev_epoll_linux_test.c

@@ -45,6 +45,7 @@
 #include <grpc/support/log.h>
 
 #include "src/core/lib/iomgr/iomgr.h"
+#include "src/core/lib/iomgr/workqueue.h"
 #include "test/core/util/test_config.h"
 
 typedef struct test_pollset {
@@ -60,6 +61,22 @@ typedef struct test_fd {
 /* num_fds should be an even number */
 static void test_fd_init(test_fd *tfds, int *fds, int num_fds) {
   int i;
+  int r;
+
+  /* Create some dummy file descriptors. Currently using pipe file descriptors
+   * for this test but we could use any other type of file descriptors. Also,
+   * since pipe() used in this test creates two fds in each call, num_fds should
+   * be an even number */
+  GPR_ASSERT((num_fds % 2) == 0);
+  for (i = 0; i < num_fds; i = i + 2) {
+    r = pipe(fds + i);
+    if (r != 0) {
+      gpr_log(GPR_ERROR, "Error in creating pipe. %d (%s)", errno,
+              strerror(errno));
+      return;
+    }
+  }
+
   for (i = 0; i < num_fds; i++) {
     tfds[i].inner_fd = fds[i];
     tfds[i].fd = grpc_fd_create(fds[i], "test_fd");
@@ -111,8 +128,80 @@ static void test_pollset_cleanup(grpc_exec_ctx *exec_ctx,
   }
 }
 
-#define NUM_FDS 8
-#define NUM_POLLSETS 4
+static void increment(grpc_exec_ctx *exec_ctx, void *arg, grpc_error *error) {
+  ++*(int *)arg;
+}
+
+/*
+ * Validate that merging two workqueues preserves the closures in each queue.
+ * This is a regression test for a bug in
+ * polling_island_merge()[ev_epoll_linux.c], where the parent relationship was
+ * inverted.
+ */
+static void test_pollset_queue_merge_items() {
+  grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT;
+  const int num_fds = 2;
+  const int num_pollsets = 2;
+  const int num_closures = 4;
+  test_fd tfds[num_fds];
+  int fds[num_fds];
+  test_pollset pollsets[num_pollsets];
+  grpc_closure closures[num_closures];
+  int i;
+  int result = 0;
+
+  test_fd_init(tfds, fds, num_fds);
+  test_pollset_init(pollsets, num_pollsets);
+
+  /* Two distinct polling islands, each with their own FD and pollset. */
+  for (i = 0; i < num_fds; i++) {
+    grpc_pollset_add_fd(&exec_ctx, pollsets[i].pollset, tfds[i].fd);
+    grpc_exec_ctx_flush(&exec_ctx);
+  }
+
+  /* Enqeue the closures, 3 to polling island 0 and 1 to polling island 1. */
+  grpc_closure_init(
+      closures, increment, &result,
+      grpc_workqueue_scheduler(grpc_fd_get_polling_island(tfds[0].fd)));
+  grpc_closure_init(
+      closures + 1, increment, &result,
+      grpc_workqueue_scheduler(grpc_fd_get_polling_island(tfds[0].fd)));
+  grpc_closure_init(
+      closures + 2, increment, &result,
+      grpc_workqueue_scheduler(grpc_fd_get_polling_island(tfds[0].fd)));
+  grpc_closure_init(
+      closures + 3, increment, &result,
+      grpc_workqueue_scheduler(grpc_fd_get_polling_island(tfds[1].fd)));
+  for (i = 0; i < num_closures; ++i) {
+    grpc_closure_sched(&exec_ctx, closures + i, GRPC_ERROR_NONE);
+  }
+
+  /* Merge the two polling islands. */
+  grpc_pollset_add_fd(&exec_ctx, pollsets[0].pollset, tfds[1].fd);
+  grpc_exec_ctx_flush(&exec_ctx);
+
+  /*
+   * Execute the closures, verify we see each one execute when executing work on
+   * the merged polling island.
+   */
+  grpc_pollset_worker *worker = NULL;
+  for (i = 0; i < num_closures; ++i) {
+    const gpr_timespec deadline = gpr_time_add(
+        gpr_now(GPR_CLOCK_MONOTONIC), gpr_time_from_seconds(2, GPR_TIMESPAN));
+    gpr_mu_lock(pollsets[1].mu);
+    GRPC_LOG_IF_ERROR(
+        "grpc_pollset_work",
+        grpc_pollset_work(&exec_ctx, pollsets[1].pollset, &worker,
+                          gpr_now(GPR_CLOCK_MONOTONIC), deadline));
+    gpr_mu_unlock(pollsets[1].mu);
+  }
+  GPR_ASSERT(result == num_closures);
+
+  test_fd_cleanup(&exec_ctx, tfds, num_fds);
+  test_pollset_cleanup(&exec_ctx, pollsets, num_pollsets);
+  grpc_exec_ctx_finish(&exec_ctx);
+}
+
 /*
  * Cases to test:
  *  case 1) Polling islands of both fd and pollset are NULL
@@ -125,28 +214,16 @@ static void test_pollset_cleanup(grpc_exec_ctx *exec_ctx,
  * */
 static void test_add_fd_to_pollset() {
   grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT;
-  test_fd tfds[NUM_FDS];
-  int fds[NUM_FDS];
-  test_pollset pollsets[NUM_POLLSETS];
+  const int num_fds = 8;
+  const int num_pollsets = 4;
+  test_fd tfds[num_fds];
+  int fds[num_fds];
+  test_pollset pollsets[num_pollsets];
   void *expected_pi = NULL;
   int i;
-  int r;
 
-  /* Create some dummy file descriptors. Currently using pipe file descriptors
-   * for this test but we could use any other type of file descriptors. Also,
-   * since pipe() used in this test creates two fds in each call, NUM_FDS should
-   * be an even number */
-  for (i = 0; i < NUM_FDS; i = i + 2) {
-    r = pipe(fds + i);
-    if (r != 0) {
-      gpr_log(GPR_ERROR, "Error in creating pipe. %d (%s)", errno,
-              strerror(errno));
-      return;
-    }
-  }
-
-  test_fd_init(tfds, fds, NUM_FDS);
-  test_pollset_init(pollsets, NUM_POLLSETS);
+  test_fd_init(tfds, fds, num_fds);
+  test_pollset_init(pollsets, num_pollsets);
 
   /*Step 1.
    * Create three polling islands (This will exercise test case 1 and 2) with
@@ -207,19 +284,19 @@ static void test_add_fd_to_pollset() {
 
   /* Compare Fd:0's polling island with that of all other Fds */
   expected_pi = grpc_fd_get_polling_island(tfds[0].fd);
-  for (i = 1; i < NUM_FDS; i++) {
+  for (i = 1; i < num_fds; i++) {
     GPR_ASSERT(grpc_are_polling_islands_equal(
         expected_pi, grpc_fd_get_polling_island(tfds[i].fd)));
   }
 
   /* Compare Fd:0's polling island with that of all other pollsets */
-  for (i = 0; i < NUM_POLLSETS; i++) {
+  for (i = 0; i < num_pollsets; i++) {
     GPR_ASSERT(grpc_are_polling_islands_equal(
         expected_pi, grpc_pollset_get_polling_island(pollsets[i].pollset)));
   }
 
-  test_fd_cleanup(&exec_ctx, tfds, NUM_FDS);
-  test_pollset_cleanup(&exec_ctx, pollsets, NUM_POLLSETS);
+  test_fd_cleanup(&exec_ctx, tfds, num_fds);
+  test_pollset_cleanup(&exec_ctx, pollsets, num_pollsets);
   grpc_exec_ctx_finish(&exec_ctx);
 }
 
@@ -231,6 +308,7 @@ int main(int argc, char **argv) {
   poll_strategy = grpc_get_poll_strategy_name();
   if (poll_strategy != NULL && strcmp(poll_strategy, "epoll") == 0) {
     test_add_fd_to_pollset();
+    test_pollset_queue_merge_items();
   } else {
     gpr_log(GPR_INFO,
             "Skipping the test. The test is only relevant for 'epoll' "