|
|
@ -228,31 +228,44 @@ void grpc_pollset_work(grpc_exec_ctx *exec_ctx, grpc_pollset *pollset, |
|
|
|
int added_worker = 0; |
|
|
|
int added_worker = 0; |
|
|
|
int locked = 1; |
|
|
|
int locked = 1; |
|
|
|
int queued_work = 0; |
|
|
|
int queued_work = 0; |
|
|
|
|
|
|
|
int keep_polling = 0; |
|
|
|
/* this must happen before we (potentially) drop pollset->mu */ |
|
|
|
/* this must happen before we (potentially) drop pollset->mu */ |
|
|
|
worker->next = worker->prev = NULL; |
|
|
|
worker->next = worker->prev = NULL; |
|
|
|
worker->reevaluate_polling_on_wakeup = 0; |
|
|
|
worker->reevaluate_polling_on_wakeup = 0; |
|
|
|
/* TODO(ctiller): pool these */ |
|
|
|
/* TODO(ctiller): pool these */ |
|
|
|
grpc_wakeup_fd_init(&worker->wakeup_fd); |
|
|
|
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) && |
|
|
|
if (!grpc_pollset_has_workers(pollset) && |
|
|
|
!grpc_closure_list_empty(pollset->idle_jobs)) { |
|
|
|
!grpc_closure_list_empty(pollset->idle_jobs)) { |
|
|
|
grpc_exec_ctx_enqueue_list(exec_ctx, &pollset->idle_jobs); |
|
|
|
grpc_exec_ctx_enqueue_list(exec_ctx, &pollset->idle_jobs); |
|
|
|
goto done; |
|
|
|
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)) { |
|
|
|
if (grpc_alarm_check(exec_ctx, now, &deadline)) { |
|
|
|
gpr_mu_unlock(&pollset->mu); |
|
|
|
gpr_mu_unlock(&pollset->mu); |
|
|
|
locked = 0; |
|
|
|
locked = 0; |
|
|
|
goto done; |
|
|
|
goto done; |
|
|
|
} |
|
|
|
} |
|
|
|
|
|
|
|
/* If we're shutting down then we don't execute any extended work */ |
|
|
|
if (pollset->shutting_down) { |
|
|
|
if (pollset->shutting_down) { |
|
|
|
goto done; |
|
|
|
goto done; |
|
|
|
} |
|
|
|
} |
|
|
|
if (pollset->in_flight_cbs) { |
|
|
|
|
|
|
|
/* Give do_promote priority so we don't starve it out */ |
|
|
|
/* Give do_promote priority so we don't starve it out */ |
|
|
|
|
|
|
|
if (pollset->in_flight_cbs) { |
|
|
|
gpr_mu_unlock(&pollset->mu); |
|
|
|
gpr_mu_unlock(&pollset->mu); |
|
|
|
locked = 0; |
|
|
|
locked = 0; |
|
|
|
goto done; |
|
|
|
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 (!pollset->kicked_without_pollers) { |
|
|
|
if (!added_worker) { |
|
|
|
if (!added_worker) { |
|
|
|
push_front_worker(pollset, worker); |
|
|
|
push_front_worker(pollset, worker); |
|
|
@ -268,21 +281,29 @@ void grpc_pollset_work(grpc_exec_ctx *exec_ctx, grpc_pollset *pollset, |
|
|
|
} else { |
|
|
|
} else { |
|
|
|
pollset->kicked_without_pollers = 0; |
|
|
|
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: |
|
|
|
done: |
|
|
|
if (!locked) { |
|
|
|
if (!locked) { |
|
|
|
queued_work |= grpc_exec_ctx_flush(exec_ctx); |
|
|
|
queued_work |= grpc_exec_ctx_flush(exec_ctx); |
|
|
|
gpr_mu_lock(&pollset->mu); |
|
|
|
gpr_mu_lock(&pollset->mu); |
|
|
|
locked = 1; |
|
|
|
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) { |
|
|
|
if (worker->reevaluate_polling_on_wakeup) { |
|
|
|
worker->reevaluate_polling_on_wakeup = 0; |
|
|
|
worker->reevaluate_polling_on_wakeup = 0; |
|
|
|
pollset->kicked_without_pollers = 0; |
|
|
|
pollset->kicked_without_pollers = 0; |
|
|
|
if (queued_work) { |
|
|
|
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); |
|
|
|
deadline = gpr_inf_past(GPR_CLOCK_MONOTONIC); |
|
|
|
} |
|
|
|
} |
|
|
|
continue; |
|
|
|
keep_polling = 1; |
|
|
|
} |
|
|
|
} |
|
|
|
break; |
|
|
|
|
|
|
|
} |
|
|
|
} |
|
|
|
if (added_worker) { |
|
|
|
if (added_worker) { |
|
|
|
remove_worker(pollset, worker); |
|
|
|
remove_worker(pollset, worker); |
|
|
|