diff --git a/src/core/iomgr/pollset_posix.c b/src/core/iomgr/pollset_posix.c index 7581c3593e6..08e34c5d567 100644 --- a/src/core/iomgr/pollset_posix.c +++ b/src/core/iomgr/pollset_posix.c @@ -228,31 +228,44 @@ void grpc_pollset_work(grpc_exec_ctx *exec_ctx, grpc_pollset *pollset, int added_worker = 0; int locked = 1; int queued_work = 0; + int keep_polling = 0; /* this must happen before we (potentially) drop pollset->mu */ worker->next = worker->prev = NULL; worker->reevaluate_polling_on_wakeup = 0; /* TODO(ctiller): pool these */ grpc_wakeup_fd_init(&worker->wakeup_fd); + /* If there's work waiting for the pollset to be idle, and the + pollset is idle, then do that work */ if (!grpc_pollset_has_workers(pollset) && !grpc_closure_list_empty(pollset->idle_jobs)) { grpc_exec_ctx_enqueue_list(exec_ctx, &pollset->idle_jobs); goto done; } + /* Check alarms - these are a global resource so we just ping + each time through on every pollset. + May update deadline to ensure timely wakeups. + TODO(ctiller): can this work be localized? */ if (grpc_alarm_check(exec_ctx, now, &deadline)) { gpr_mu_unlock(&pollset->mu); locked = 0; goto done; } + /* If we're shutting down then we don't execute any extended work */ if (pollset->shutting_down) { goto done; } + /* Give do_promote priority so we don't starve it out */ if (pollset->in_flight_cbs) { - /* Give do_promote priority so we don't starve it out */ gpr_mu_unlock(&pollset->mu); locked = 0; goto done; } - for (;;) { + /* Start polling, and keep doing so while we're being asked to + re-evaluate our pollers (this allows poll() based pollers to + ensure they don't miss wakeups) */ + keep_polling = 1; + while (keep_polling) { + keep_polling = 0; if (!pollset->kicked_without_pollers) { if (!added_worker) { push_front_worker(pollset, worker); @@ -268,21 +281,29 @@ void grpc_pollset_work(grpc_exec_ctx *exec_ctx, grpc_pollset *pollset, } else { pollset->kicked_without_pollers = 0; } + /* Finished execution - start cleaning up. + Note that we may arrive here from outside the enclosing while() loop. + In that case we won't loop though as we haven't added worker to the + worker list, which means nobody could ask us to re-evaluate polling). */ done: if (!locked) { queued_work |= grpc_exec_ctx_flush(exec_ctx); gpr_mu_lock(&pollset->mu); locked = 1; } + /* If we're forced to re-evaluate polling (via grpc_pollset_kick with + GRPC_POLLSET_REEVALUATE_POLLING_ON_WAKEUP) then we land here and force + a loop */ if (worker->reevaluate_polling_on_wakeup) { worker->reevaluate_polling_on_wakeup = 0; pollset->kicked_without_pollers = 0; if (queued_work) { + /* If there's queued work on the list, then set the deadline to be + immediate so we get back out of the polling loop quickly */ deadline = gpr_inf_past(GPR_CLOCK_MONOTONIC); } - continue; + keep_polling = 1; } - break; } if (added_worker) { remove_worker(pollset, worker);