Merge pull request #4376 from ctiller/wqtest

Cover more of workqueue

src/core/iomgr/workqueue_posix.c

@@ -103,6 +103,9 @@ void grpc_workqueue_add_to_pollset(grpc_exec_ctx *exec_ctx,
 
 void grpc_workqueue_flush(grpc_exec_ctx *exec_ctx, grpc_workqueue *workqueue) {
   gpr_mu_lock(&workqueue->mu);
+  if (grpc_closure_list_empty(workqueue->closure_list)) {
+    grpc_wakeup_fd_wakeup(&workqueue->wakeup_fd);
+  }
   grpc_closure_list_move(&exec_ctx->closure_list, &workqueue->closure_list);
   gpr_mu_unlock(&workqueue->mu);
 }

test/core/iomgr/workqueue_test.c

@@ -48,6 +48,15 @@ static void must_succeed(grpc_exec_ctx *exec_ctx, void *p, int success) {
   gpr_mu_unlock(GRPC_POLLSET_MU(&g_pollset));
 }
 
+static void test_ref_unref(void) {
+  grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT;
+  grpc_workqueue *wq = grpc_workqueue_create(&exec_ctx);
+  GRPC_WORKQUEUE_REF(wq, "test");
+  GRPC_WORKQUEUE_UNREF(&exec_ctx, wq, "test");
+  GRPC_WORKQUEUE_UNREF(&exec_ctx, wq, "destroy");
+  grpc_exec_ctx_finish(&exec_ctx);
+}
+
 static void test_add_closure(void) {
   grpc_closure c;
   int done = 0;
@@ -72,6 +81,31 @@ static void test_add_closure(void) {
   grpc_exec_ctx_finish(&exec_ctx);
 }
 
+static void test_flush(void) {
+  grpc_closure c;
+  int done = 0;
+  grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT;
+  grpc_workqueue *wq = grpc_workqueue_create(&exec_ctx);
+  gpr_timespec deadline = GRPC_TIMEOUT_SECONDS_TO_DEADLINE(5);
+  grpc_pollset_worker worker;
+  grpc_closure_init(&c, must_succeed, &done);
+
+  grpc_exec_ctx_enqueue(&exec_ctx, &c, 1);
+  grpc_workqueue_flush(&exec_ctx, wq);
+  grpc_workqueue_add_to_pollset(&exec_ctx, wq, &g_pollset);
+
+  gpr_mu_lock(GRPC_POLLSET_MU(&g_pollset));
+  GPR_ASSERT(!done);
+  grpc_pollset_work(&exec_ctx, &g_pollset, &worker,
+                    gpr_now(deadline.clock_type), deadline);
+  gpr_mu_unlock(GRPC_POLLSET_MU(&g_pollset));
+  grpc_exec_ctx_finish(&exec_ctx);
+  GPR_ASSERT(done);
+
+  GRPC_WORKQUEUE_UNREF(&exec_ctx, wq, "destroy");
+  grpc_exec_ctx_finish(&exec_ctx);
+}
+
 static void destroy_pollset(grpc_exec_ctx *exec_ctx, void *p, int success) {
   grpc_pollset_destroy(p);
 }
@@ -83,7 +117,9 @@ int main(int argc, char **argv) {
   grpc_init();
   grpc_pollset_init(&g_pollset);
 
+  test_ref_unref();
   test_add_closure();
+  test_flush();
 
   grpc_closure_init(&destroyed, destroy_pollset, &g_pollset);
   grpc_pollset_shutdown(&exec_ctx, &g_pollset, &destroyed);