Merge remote-tracking branch 'upstream/master' into lazy-deframe

4 files changed, 193 insertions, 44 deletions

diff --git a/test/core/iomgr/ev_epoll_linux_test.c b/test/core/iomgr/ev_epoll_linux_test.c
index d69f9a9d15..5f8124aeda 100644
--- a/test/core/iomgr/ev_epoll_linux_test.c
+++ b/test/core/iomgr/ev_epoll_linux_test.c
@@ -43,6 +43,8 @@
 
 #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"
@@ -139,23 +141,25 @@ static void increment(grpc_exec_ctx *exec_ctx, void *arg, grpc_error *error) {
  * 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;
-  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];
+  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);
+  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++) {
+  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);
   }
@@ -173,7 +177,7 @@ static void test_pollset_queue_merge_items() {
   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) {
+  for (i = 0; i < NUM_CLOSURES; ++i) {
     grpc_closure_sched(&exec_ctx, closures + i, GRPC_ERROR_NONE);
   }
 
@@ -186,7 +190,7 @@ static void test_pollset_queue_merge_items() {
    * the merged polling island.
    */
   grpc_pollset_worker *worker = NULL;
-  for (i = 0; i < num_closures; ++i) {
+  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);
@@ -196,13 +200,17 @@ static void test_pollset_queue_merge_items() {
                           gpr_now(GPR_CLOCK_MONOTONIC), deadline));
     gpr_mu_unlock(pollsets[1].mu);
   }
-  GPR_ASSERT(result == num_closures);
+  GPR_ASSERT(result == NUM_CLOSURES);
 
-  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);
 }
 
+#undef NUM_FDS
+#undef NUM_POLLSETS
+#undef NUM_CLOSURES
+
 /*
  * Cases to test:
  *  case 1) Polling islands of both fd and pollset are NULL
@@ -213,18 +221,20 @@ static void test_pollset_queue_merge_items() {
  *     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;
-  const int num_fds = 8;
-  const int num_pollsets = 4;
-  test_fd tfds[num_fds];
-  int fds[num_fds];
-  test_pollset pollsets[num_pollsets];
+  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);
+  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
@@ -285,22 +295,107 @@ 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);
 }
 
+#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;
+  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_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);
@@ -310,6 +405,7 @@ int main(int argc, char **argv) {
   if (poll_strategy != NULL && strcmp(poll_strategy, "epoll") == 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 'epoll' "
diff --git a/test/core/iomgr/resource_quota_test.c b/test/core/iomgr/resource_quota_test.c
index a5b28f210d..ebce8b9da6 100644
--- a/test/core/iomgr/resource_quota_test.c
+++ b/test/core/iomgr/resource_quota_test.c
@@ -682,6 +682,56 @@ static void test_one_slice_deleted_late(void) {
   }
 }
 
+static void test_resize_to_zero(void) {
+  gpr_log(GPR_INFO, "** test_resize_to_zero **");
+  grpc_resource_quota *q = grpc_resource_quota_create("test_resize_to_zero");
+  grpc_resource_quota_resize(q, 0);
+  grpc_resource_quota_unref(q);
+}
+
+static void test_negative_rq_free_pool(void) {
+  gpr_log(GPR_INFO, "** test_negative_rq_free_pool **");
+  grpc_resource_quota *q =
+      grpc_resource_quota_create("test_negative_rq_free_pool");
+  grpc_resource_quota_resize(q, 1024);
+
+  grpc_resource_user *usr = grpc_resource_user_create(q, "usr");
+
+  grpc_resource_user_slice_allocator alloc;
+  int num_allocs = 0;
+  grpc_resource_user_slice_allocator_init(&alloc, usr, inc_int_cb, &num_allocs);
+
+  grpc_slice_buffer buffer;
+  grpc_slice_buffer_init(&buffer);
+
+  {
+    const int start_allocs = num_allocs;
+    grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT;
+    grpc_resource_user_alloc_slices(&exec_ctx, &alloc, 1024, 1, &buffer);
+    grpc_exec_ctx_finish(&exec_ctx);
+    GPR_ASSERT(num_allocs == start_allocs + 1);
+  }
+
+  grpc_resource_quota_resize(q, 512);
+
+  double eps = 0.0001;
+  GPR_ASSERT(grpc_resource_quota_get_memory_pressure(q) < 1 + eps);
+  GPR_ASSERT(grpc_resource_quota_get_memory_pressure(q) > 1 - eps);
+
+  {
+    grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT;
+    grpc_resource_user_unref(&exec_ctx, usr);
+    grpc_exec_ctx_finish(&exec_ctx);
+  }
+
+  grpc_resource_quota_unref(q);
+  {
+    grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT;
+    grpc_slice_buffer_destroy_internal(&exec_ctx, &buffer);
+    grpc_exec_ctx_finish(&exec_ctx);
+  }
+}
+
 int main(int argc, char **argv) {
   grpc_test_init(argc, argv);
   grpc_init();
@@ -705,6 +755,8 @@ int main(int argc, char **argv) {
   test_reclaimers_can_be_posted_repeatedly();
   test_one_slice();
   test_one_slice_deleted_late();
+  test_resize_to_zero();
+  test_negative_rq_free_pool();
   grpc_shutdown();
   return 0;
 }
diff --git a/test/core/iomgr/timer_heap_test.c b/test/core/iomgr/timer_heap_test.c
index 410d972313..153304fa7b 100644
--- a/test/core/iomgr/timer_heap_test.c
+++ b/test/core/iomgr/timer_heap_test.c
@@ -47,13 +47,7 @@
 
 #include "test/core/util/test_config.h"
 
-static gpr_timespec random_deadline(void) {
-  gpr_timespec ts;
-  ts.tv_sec = rand();
-  ts.tv_nsec = rand();
-  ts.clock_type = GPR_CLOCK_REALTIME;
-  return ts;
-}
+static gpr_atm random_deadline(void) { return rand(); }
 
 static grpc_timer *create_test_elements(size_t num_elements) {
   grpc_timer *elems = gpr_malloc(num_elements * sizeof(grpc_timer));
@@ -78,12 +72,10 @@ static void check_valid(grpc_timer_heap *pq) {
     size_t left_child = 1u + 2u * i;
     size_t right_child = left_child + 1u;
     if (left_child < pq->timer_count) {
-      GPR_ASSERT(gpr_time_cmp(pq->timers[i]->deadline,
-                              pq->timers[left_child]->deadline) <= 0);
+      GPR_ASSERT(pq->timers[i]->deadline <= pq->timers[left_child]->deadline);
     }
     if (right_child < pq->timer_count) {
-      GPR_ASSERT(gpr_time_cmp(pq->timers[i]->deadline,
-                              pq->timers[right_child]->deadline) <= 0);
+      GPR_ASSERT(pq->timers[i]->deadline <= pq->timers[right_child]->deadline);
     }
   }
 }
@@ -227,20 +219,19 @@ static void test2(void) {
     }
 
     if (num_inserted) {
-      gpr_timespec *min_deadline = NULL;
+      gpr_atm *min_deadline = NULL;
       for (size_t i = 0; i < elems_size; i++) {
         if (elems[i].inserted) {
           if (min_deadline == NULL) {
             min_deadline = &elems[i].elem.deadline;
           } else {
-            if (gpr_time_cmp(elems[i].elem.deadline, *min_deadline) < 0) {
+            if (elems[i].elem.deadline < *min_deadline) {
               min_deadline = &elems[i].elem.deadline;
             }
           }
         }
       }
-      GPR_ASSERT(
-          0 == gpr_time_cmp(grpc_timer_heap_top(&pq)->deadline, *min_deadline));
+      GPR_ASSERT(grpc_timer_heap_top(&pq)->deadline == *min_deadline);
     }
   }
 
diff --git a/test/core/iomgr/timer_list_test.c b/test/core/iomgr/timer_list_test.c
index 85ad5277cc..46e41dd449 100644
--- a/test/core/iomgr/timer_list_test.c
+++ b/test/core/iomgr/timer_list_test.c
@@ -45,6 +45,9 @@
 
 #define MAX_CB 30
 
+extern int grpc_timer_trace;
+extern int grpc_timer_check_trace;
+
 static int cb_called[MAX_CB][2];
 
 static void cb(grpc_exec_ctx *exec_ctx, void *arg, grpc_error *error) {
@@ -57,7 +60,11 @@ static void add_test(void) {
   grpc_timer timers[20];
   grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT;
 
+  gpr_log(GPR_INFO, "add_test");
+
   grpc_timer_list_init(start);
+  grpc_timer_trace = 1;
+  grpc_timer_check_trace = 1;
   memset(cb_called, 0, sizeof(cb_called));
 
   /* 10 ms timers.  will expire in the current epoch */
@@ -120,9 +127,7 @@ static void add_test(void) {
 }
 
 static gpr_timespec tfm(int m) {
-  gpr_timespec t = gpr_time_from_millis(m, GPR_TIMESPAN);
-  t.clock_type = GPR_CLOCK_REALTIME;
-  return t;
+  return gpr_time_from_millis(m, GPR_CLOCK_REALTIME);
 }
 
 /* Cleaning up a list with pending timers. */
@@ -130,7 +135,11 @@ void destruction_test(void) {
   grpc_timer timers[5];
   grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT;
 
+  gpr_log(GPR_INFO, "destruction_test");
+
   grpc_timer_list_init(gpr_time_0(GPR_CLOCK_REALTIME));
+  grpc_timer_trace = 1;
+  grpc_timer_check_trace = 1;
   memset(cb_called, 0, sizeof(cb_called));
 
   grpc_timer_init(
@@ -170,6 +179,7 @@ void destruction_test(void) {
 
 int main(int argc, char **argv) {
   grpc_test_init(argc, argv);
+  gpr_set_log_verbosity(GPR_LOG_SEVERITY_DEBUG);
   add_test();
   destruction_test();
   return 0;