|
|
|
|
@ -44,16 +44,26 @@ typedef struct completed_thread { |
|
|
|
|
struct completed_thread *next; |
|
|
|
|
} completed_thread; |
|
|
|
|
|
|
|
|
|
// global mutex
|
|
|
|
|
static gpr_mu g_mu; |
|
|
|
|
// are we multi-threaded
|
|
|
|
|
static bool g_threaded; |
|
|
|
|
// cv to wait until a thread is needed
|
|
|
|
|
static gpr_cv g_cv_wait; |
|
|
|
|
// cv for notification when threading ends
|
|
|
|
|
static gpr_cv g_cv_shutdown; |
|
|
|
|
// number of threads in the system
|
|
|
|
|
static int g_thread_count; |
|
|
|
|
// number of threads sitting around waiting
|
|
|
|
|
static int g_waiter_count; |
|
|
|
|
// linked list of threads that have completed (and need joining)
|
|
|
|
|
static completed_thread *g_completed_threads; |
|
|
|
|
// was the manager kicked by the timer system
|
|
|
|
|
static bool g_kicked; |
|
|
|
|
|
|
|
|
|
#define MAX_WAITERS 3 |
|
|
|
|
// is there a thread waiting until the next timer should fire?
|
|
|
|
|
static bool g_has_timed_waiter; |
|
|
|
|
// generation counter to track which thread is waiting for the next timer
|
|
|
|
|
static uint64_t g_timed_waiter_generation; |
|
|
|
|
|
|
|
|
|
static void timer_thread(void *unused); |
|
|
|
|
|
|
|
|
|
@ -92,39 +102,79 @@ void grpc_timer_manager_tick() { |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
static void timer_thread(void *unused) { |
|
|
|
|
// this threads exec_ctx: we try to run things through to completion here
|
|
|
|
|
// since it's easy to spin up new threads
|
|
|
|
|
grpc_exec_ctx exec_ctx = |
|
|
|
|
GRPC_EXEC_CTX_INITIALIZER(0, grpc_never_ready_to_finish, NULL); |
|
|
|
|
const gpr_timespec inf_future = gpr_inf_future(GPR_CLOCK_MONOTONIC); |
|
|
|
|
for (;;) { |
|
|
|
|
gpr_timespec next = gpr_inf_future(GPR_CLOCK_MONOTONIC); |
|
|
|
|
gpr_timespec next = inf_future; |
|
|
|
|
gpr_timespec now = gpr_now(GPR_CLOCK_MONOTONIC); |
|
|
|
|
// check timer state, updates next to the next time to run a check
|
|
|
|
|
if (grpc_timer_check(&exec_ctx, now, &next)) { |
|
|
|
|
// if there's something to execute...
|
|
|
|
|
gpr_mu_lock(&g_mu); |
|
|
|
|
// remove a waiter from the pool, and start another thread if necessary
|
|
|
|
|
--g_waiter_count; |
|
|
|
|
bool start_thread = g_waiter_count == 0; |
|
|
|
|
if (start_thread && g_threaded) { |
|
|
|
|
if (g_waiter_count == 0 && g_threaded) { |
|
|
|
|
start_timer_thread_and_unlock(); |
|
|
|
|
} else { |
|
|
|
|
// if there's no thread waiting with a timeout, kick an existing waiter
|
|
|
|
|
// so that the next deadline is not missed
|
|
|
|
|
if (!g_has_timed_waiter) { |
|
|
|
|
gpr_log(GPR_DEBUG, "kick untimed waiter"); |
|
|
|
|
gpr_cv_signal(&g_cv_wait); |
|
|
|
|
} |
|
|
|
|
gpr_mu_unlock(&g_mu); |
|
|
|
|
} |
|
|
|
|
// without our lock, flush the exec_ctx
|
|
|
|
|
grpc_exec_ctx_flush(&exec_ctx); |
|
|
|
|
gpr_mu_lock(&g_mu); |
|
|
|
|
// garbage collect any threads hanging out that are dead
|
|
|
|
|
gc_completed_threads(); |
|
|
|
|
// get ready to wait again
|
|
|
|
|
++g_waiter_count; |
|
|
|
|
gpr_mu_unlock(&g_mu); |
|
|
|
|
} else { |
|
|
|
|
gpr_mu_lock(&g_mu); |
|
|
|
|
// if we're not threaded anymore, leave
|
|
|
|
|
if (!g_threaded) break; |
|
|
|
|
if (gpr_cv_wait(&g_cv_wait, &g_mu, next)) { |
|
|
|
|
if (g_kicked) { |
|
|
|
|
grpc_timer_consume_kick(); |
|
|
|
|
g_kicked = false; |
|
|
|
|
} else if (g_waiter_count > MAX_WAITERS) { |
|
|
|
|
break; |
|
|
|
|
} |
|
|
|
|
// if there's no timed waiter, we should become one: that waiter waits
|
|
|
|
|
// only until the next timer should expire
|
|
|
|
|
// all other timers wait forever
|
|
|
|
|
uint64_t my_timed_waiter_generation = g_timed_waiter_generation - 1; |
|
|
|
|
if (!g_has_timed_waiter) { |
|
|
|
|
g_has_timed_waiter = true; |
|
|
|
|
// we use a generation counter to track the timed waiter so we can
|
|
|
|
|
// cancel an existing one quickly (and when it actually times out it'll
|
|
|
|
|
// figure stuff out instead of incurring a wakeup)
|
|
|
|
|
my_timed_waiter_generation = ++g_timed_waiter_generation; |
|
|
|
|
gpr_log(GPR_DEBUG, "sleep for a while"); |
|
|
|
|
} else { |
|
|
|
|
next = inf_future; |
|
|
|
|
gpr_log(GPR_DEBUG, "sleep until kicked"); |
|
|
|
|
} |
|
|
|
|
gpr_cv_wait(&g_cv_wait, &g_mu, next); |
|
|
|
|
gpr_log(GPR_DEBUG, "wait ended: was_timed:%d kicked:%d", |
|
|
|
|
my_timed_waiter_generation == g_timed_waiter_generation, |
|
|
|
|
g_kicked); |
|
|
|
|
// if this was the timed waiter, then we need to check timers, and flag
|
|
|
|
|
// that there's now no timed waiter... we'll look for a replacement if
|
|
|
|
|
// there's work to do after checking timers (code above)
|
|
|
|
|
if (my_timed_waiter_generation == g_timed_waiter_generation) { |
|
|
|
|
g_has_timed_waiter = false; |
|
|
|
|
} |
|
|
|
|
// if this was a kick from the timer system, consume it (and don't stop
|
|
|
|
|
// this thread yet)
|
|
|
|
|
if (g_kicked) { |
|
|
|
|
grpc_timer_consume_kick(); |
|
|
|
|
g_kicked = false; |
|
|
|
|
} |
|
|
|
|
gpr_mu_unlock(&g_mu); |
|
|
|
|
} |
|
|
|
|
gpr_mu_unlock(&g_mu); |
|
|
|
|
} |
|
|
|
|
// terminate the thread: drop the waiter count, thread count, and let whomever
|
|
|
|
|
// stopped the threading stuff know that we're done
|
|
|
|
|
--g_waiter_count; |
|
|
|
|
--g_thread_count; |
|
|
|
|
if (0 == g_thread_count) { |
|
|
|
|
@ -135,6 +185,7 @@ static void timer_thread(void *unused) { |
|
|
|
|
ct->next = g_completed_threads; |
|
|
|
|
g_completed_threads = ct; |
|
|
|
|
gpr_mu_unlock(&g_mu); |
|
|
|
|
grpc_exec_ctx_finish(&exec_ctx); |
|
|
|
|
gpr_log(GPR_DEBUG, "End timer thread"); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
@ -193,6 +244,8 @@ void grpc_timer_manager_set_threading(bool threaded) { |
|
|
|
|
void grpc_kick_poller(void) { |
|
|
|
|
gpr_mu_lock(&g_mu); |
|
|
|
|
g_kicked = true; |
|
|
|
|
g_has_timed_waiter = false; |
|
|
|
|
++g_timed_waiter_generation; |
|
|
|
|
gpr_cv_signal(&g_cv_wait); |
|
|
|
|
gpr_mu_unlock(&g_mu); |
|
|
|
|
} |
|
|
|
|
|
Craig Tiller
8 years ago