diff --git a/src/core/iomgr/workqueue_posix.c b/src/core/iomgr/workqueue_posix.c index 2e301781313..d2a1c346120 100644 --- a/src/core/iomgr/workqueue_posix.c +++ b/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); } diff --git a/test/core/iomgr/workqueue_test.c b/test/core/iomgr/workqueue_test.c index 90f7ba7a834..d1f9dabc574 100644 --- a/test/core/iomgr/workqueue_test.c +++ b/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);