Merge branch 'hiertune' into epex4+epoll1

1 files changed, 428 insertions, 0 deletions

diff --git a/test/core/iomgr/ev_epollsig_linux_test.c b/test/core/iomgr/ev_epollsig_linux_test.c
new file mode 100644
index 0000000000..efc8ed36be
--- /dev/null
+++ b/test/core/iomgr/ev_epollsig_linux_test.c
@@ -0,0 +1,428 @@
+/*
+ *
+ * Copyright 2015, Google Inc.
+ * All rights reserved.
+ *
+ * Redistribution and use in source and binary forms, with or without
+ * modification, are permitted provided that the following conditions are
+ * met:
+ *
+ *     * Redistributions of source code must retain the above copyright
+ * notice, this list of conditions and the following disclaimer.
+ *     * Redistributions in binary form must reproduce the above
+ * copyright notice, this list of conditions and the following disclaimer
+ * in the documentation and/or other materials provided with the
+ * distribution.
+ *     * Neither the name of Google Inc. nor the names of its
+ * contributors may be used to endorse or promote products derived from
+ * this software without specific prior written permission.
+ *
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+ * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+ * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+ * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+ * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+ * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+ * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+ * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+ * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+ * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+ * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ *
+ */
+#include "src/core/lib/iomgr/port.h"
+
+/* This test only relevant on linux systems where epoll() is available */
+#ifdef GRPC_LINUX_EPOLL
+#include "src/core/lib/iomgr/ev_epollsig_linux.h"
+#include "src/core/lib/iomgr/ev_posix.h"
+
+#include <errno.h>
+#include <string.h>
+#include <unistd.h>
+
+#include <grpc/support/alloc.h>
+#include <grpc/support/log.h>
+#include <grpc/support/thd.h>
+#include <grpc/support/useful.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 {
+  grpc_pollset *pollset;
+  gpr_mu *mu;
+} test_pollset;
+
+typedef struct test_fd {
+  int inner_fd;
+  grpc_fd *fd;
+} 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");
+  }
+}
+
+static void test_fd_cleanup(grpc_exec_ctx *exec_ctx, test_fd *tfds,
+                            int num_fds) {
+  int release_fd;
+  int i;
+
+  for (i = 0; i < num_fds; i++) {
+    grpc_fd_shutdown(exec_ctx, tfds[i].fd,
+                     GRPC_ERROR_CREATE_FROM_STATIC_STRING("test_fd_cleanup"));
+    grpc_exec_ctx_flush(exec_ctx);
+
+    grpc_fd_orphan(exec_ctx, tfds[i].fd, NULL, &release_fd, "test_fd_cleanup");
+    grpc_exec_ctx_flush(exec_ctx);
+
+    GPR_ASSERT(release_fd == tfds[i].inner_fd);
+    close(tfds[i].inner_fd);
+  }
+}
+
+static void test_pollset_init(test_pollset *pollsets, int num_pollsets) {
+  int i;
+  for (i = 0; i < num_pollsets; i++) {
+    pollsets[i].pollset = gpr_zalloc(grpc_pollset_size());
+    grpc_pollset_init(pollsets[i].pollset, &pollsets[i].mu);
+  }
+}
+
+static void destroy_pollset(grpc_exec_ctx *exec_ctx, void *p,
+                            grpc_error *error) {
+  grpc_pollset_destroy(exec_ctx, p);
+}
+
+static void test_pollset_cleanup(grpc_exec_ctx *exec_ctx,
+                                 test_pollset *pollsets, int num_pollsets) {
+  grpc_closure destroyed;
+  int i;
+
+  for (i = 0; i < num_pollsets; i++) {
+    grpc_closure_init(&destroyed, destroy_pollset, pollsets[i].pollset,
+                      grpc_schedule_on_exec_ctx);
+    grpc_pollset_shutdown(exec_ctx, pollsets[i].pollset, &destroyed);
+
+    grpc_exec_ctx_flush(exec_ctx);
+    gpr_free(pollsets[i].pollset);
+  }
+}
+
+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.
+ */
+
+#define NUM_FDS 2
+#define NUM_POLLSETS 2
+#define NUM_CLOSURES 4
+
+static void test_pollset_queue_merge_items() {
+  grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT;
+  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);
+}
+
+#undef NUM_FDS
+#undef NUM_POLLSETS
+#undef NUM_CLOSURES
+
+/*
+ * Cases to test:
+ *  case 1) Polling islands of both fd and pollset are NULL
+ *  case 2) Polling island of fd is NULL but that of pollset is not-NULL
+ *  case 3) Polling island of fd is not-NULL but that of pollset is NULL
+ *  case 4) Polling islands of both fd and pollset are not-NULL and:
+ *     case 4.1) Polling islands of fd and pollset are equal
+ *     case 4.2) Polling islands of fd and pollset are NOT-equal (This results
+ *     in a merge)
+ * */
+
+#define NUM_FDS 8
+#define NUM_POLLSETS 4
+
+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];
+  void *expected_pi = NULL;
+  int i;
+
+  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
+   * the following configuration:
+   *   polling island 0 = { fds:0,1,2, pollsets:0}
+   *   polling island 1 = { fds:3,4,   pollsets:1}
+   *   polling island 2 = { fds:5,6,7  pollsets:2}
+   *
+   *Step 2.
+   * Add pollset 3 to polling island 0 (by adding fds 0 and 1 to pollset 3)
+   * (This will exercise test cases 3 and 4.1). The configuration becomes:
+   *   polling island 0 = { fds:0,1,2, pollsets:0,3} <<< pollset 3 added here
+   *   polling island 1 = { fds:3,4,   pollsets:1}
+   *   polling island 2 = { fds:5,6,7  pollsets:2}
+   *
+   *Step 3.
+   * Merge polling islands 0 and 1 by adding fd 0 to pollset 1 (This will
+   * exercise test case 4.2). The configuration becomes:
+   *   polling island (merged) = {fds: 0,1,2,3,4, pollsets: 0,1,3}
+   *   polling island 2 = {fds: 5,6,7 pollsets: 2}
+   *
+   *Step 4.
+   * Finally do one more merge by adding fd 3 to pollset 2.
+   *   polling island (merged) = {fds: 0,1,2,3,4,5,6,7, pollsets: 0,1,2,3}
+   */
+
+  /* == Step 1 == */
+  for (i = 0; i <= 2; i++) {
+    grpc_pollset_add_fd(&exec_ctx, pollsets[0].pollset, tfds[i].fd);
+    grpc_exec_ctx_flush(&exec_ctx);
+  }
+
+  for (i = 3; i <= 4; i++) {
+    grpc_pollset_add_fd(&exec_ctx, pollsets[1].pollset, tfds[i].fd);
+    grpc_exec_ctx_flush(&exec_ctx);
+  }
+
+  for (i = 5; i <= 7; i++) {
+    grpc_pollset_add_fd(&exec_ctx, pollsets[2].pollset, tfds[i].fd);
+    grpc_exec_ctx_flush(&exec_ctx);
+  }
+
+  /* == Step 2 == */
+  for (i = 0; i <= 1; i++) {
+    grpc_pollset_add_fd(&exec_ctx, pollsets[3].pollset, tfds[i].fd);
+    grpc_exec_ctx_flush(&exec_ctx);
+  }
+
+  /* == Step 3 == */
+  grpc_pollset_add_fd(&exec_ctx, pollsets[1].pollset, tfds[0].fd);
+  grpc_exec_ctx_flush(&exec_ctx);
+
+  /* == Step 4 == */
+  grpc_pollset_add_fd(&exec_ctx, pollsets[2].pollset, tfds[3].fd);
+  grpc_exec_ctx_flush(&exec_ctx);
+
+  /* All polling islands are merged at this point */
+
+  /* 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++) {
+    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++) {
+    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);
+  grpc_exec_ctx_finish(&exec_ctx);
+}
+
+#undef NUM_FDS
+#undef NUM_POLLSETS
+
+typedef struct threading_shared {
+  gpr_mu *mu;
+  grpc_pollset *pollset;
+  grpc_wakeup_fd *wakeup_fd;
+  grpc_fd *wakeup_desc;
+  grpc_closure on_wakeup;
+  int wakeups;
+} threading_shared;
+
+static __thread int thread_wakeups = 0;
+
+static void test_threading_loop(void *arg) {
+  threading_shared *shared = arg;
+  while (thread_wakeups < 1000000) {
+    grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT;
+    grpc_pollset_worker *worker;
+    gpr_mu_lock(shared->mu);
+    GPR_ASSERT(GRPC_LOG_IF_ERROR(
+        "pollset_work",
+        grpc_pollset_work(&exec_ctx, shared->pollset, &worker,
+                          gpr_now(GPR_CLOCK_MONOTONIC),
+                          gpr_inf_future(GPR_CLOCK_MONOTONIC))));
+    gpr_mu_unlock(shared->mu);
+    grpc_exec_ctx_finish(&exec_ctx);
+  }
+}
+
+static void test_threading_wakeup(grpc_exec_ctx *exec_ctx, void *arg,
+                                  grpc_error *error) {
+  threading_shared *shared = arg;
+  ++shared->wakeups;
+  ++thread_wakeups;
+  if (error == GRPC_ERROR_NONE) {
+    GPR_ASSERT(GRPC_LOG_IF_ERROR(
+        "consume_wakeup", grpc_wakeup_fd_consume_wakeup(shared->wakeup_fd)));
+    grpc_fd_notify_on_read(exec_ctx, shared->wakeup_desc, &shared->on_wakeup);
+    GPR_ASSERT(GRPC_LOG_IF_ERROR("wakeup_next",
+                                 grpc_wakeup_fd_wakeup(shared->wakeup_fd)));
+  }
+}
+
+static void test_threading(void) {
+  threading_shared shared;
+  shared.pollset = gpr_zalloc(grpc_pollset_size());
+  grpc_pollset_init(shared.pollset, &shared.mu);
+
+  gpr_thd_id thds[10];
+  for (size_t i = 0; i < GPR_ARRAY_SIZE(thds); i++) {
+    gpr_thd_options opt = gpr_thd_options_default();
+    gpr_thd_options_set_joinable(&opt);
+    gpr_thd_new(&thds[i], test_threading_loop, &shared, &opt);
+  }
+  grpc_wakeup_fd fd;
+  GPR_ASSERT(GRPC_LOG_IF_ERROR("wakeup_fd_init", grpc_wakeup_fd_init(&fd)));
+  shared.wakeup_fd = &fd;
+  shared.wakeup_desc = grpc_fd_create(fd.read_fd, "wakeup");
+  shared.wakeups = 0;
+  {
+    grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT;
+    grpc_pollset_add_fd(&exec_ctx, shared.pollset, shared.wakeup_desc);
+    grpc_fd_notify_on_read(
+        &exec_ctx, shared.wakeup_desc,
+        grpc_closure_init(&shared.on_wakeup, test_threading_wakeup, &shared,
+                          grpc_schedule_on_exec_ctx));
+    grpc_exec_ctx_finish(&exec_ctx);
+  }
+  GPR_ASSERT(GRPC_LOG_IF_ERROR("wakeup_first",
+                               grpc_wakeup_fd_wakeup(shared.wakeup_fd)));
+  for (size_t i = 0; i < GPR_ARRAY_SIZE(thds); i++) {
+    gpr_thd_join(thds[i]);
+  }
+  fd.read_fd = 0;
+  grpc_wakeup_fd_destroy(&fd);
+  {
+    grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT;
+    grpc_fd_shutdown(&exec_ctx, shared.wakeup_desc, GRPC_ERROR_CANCELLED);
+    grpc_fd_orphan(&exec_ctx, shared.wakeup_desc, NULL, NULL, "done");
+    grpc_pollset_shutdown(&exec_ctx, shared.pollset,
+                          grpc_closure_create(destroy_pollset, shared.pollset,
+                                              grpc_schedule_on_exec_ctx));
+    grpc_exec_ctx_finish(&exec_ctx);
+  }
+  gpr_free(shared.pollset);
+}
+
+int main(int argc, char **argv) {
+  const char *poll_strategy = NULL;
+  grpc_test_init(argc, argv);
+  grpc_iomgr_init();
+
+  poll_strategy = grpc_get_poll_strategy_name();
+  if (poll_strategy != NULL && strcmp(poll_strategy, "epollsig") == 0) {
+    test_add_fd_to_pollset();
+    test_pollset_queue_merge_items();
+    test_threading();
+  } else {
+    gpr_log(GPR_INFO,
+            "Skipping the test. The test is only relevant for 'epollsig' "
+            "strategy. and the current strategy is: '%s'",
+            poll_strategy);
+  }
+
+  {
+    grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT;
+    grpc_iomgr_shutdown(&exec_ctx);
+    grpc_exec_ctx_finish(&exec_ctx);
+  }
+  return 0;
+}
+#else /* defined(GRPC_LINUX_EPOLL) */
+int main(int argc, char **argv) { return 0; }
+#endif /* !defined(GRPC_LINUX_EPOLL) */