diff options
author | 2017-11-14 14:29:22 -0800 | |
---|---|---|
committer | 2017-11-14 14:35:18 -0800 | |
commit | 19c34b60bee64788063ee1bbfca5f7374ccf5b77 (patch) | |
tree | b233dff9b75c3459a4324481c717515b2a6d52b8 /src/core/lib/iomgr/timer_generic.cc | |
parent | e092ff879ea2f554956c0bc2bc063a85f7fa6866 (diff) |
Reviewer comments
Diffstat (limited to 'src/core/lib/iomgr/timer_generic.cc')
-rw-r--r-- | src/core/lib/iomgr/timer_generic.cc | 97 |
1 files changed, 55 insertions, 42 deletions
diff --git a/src/core/lib/iomgr/timer_generic.cc b/src/core/lib/iomgr/timer_generic.cc index d5c4875eab..a4bfbcb342 100644 --- a/src/core/lib/iomgr/timer_generic.cc +++ b/src/core/lib/iomgr/timer_generic.cc @@ -25,6 +25,7 @@ #include "src/core/lib/iomgr/timer.h" #include <grpc/support/alloc.h> +#include <grpc/support/cpu.h> #include <grpc/support/log.h> #include <grpc/support/string_util.h> #include <grpc/support/sync.h> @@ -37,14 +38,15 @@ #define INVALID_HEAP_INDEX 0xffffffffu -#define LOG2_NUM_SHARDS 5 -#define NUM_SHARDS (1 << LOG2_NUM_SHARDS) #define ADD_DEADLINE_SCALE 0.33 #define MIN_QUEUE_WINDOW_DURATION 0.01 #define MAX_QUEUE_WINDOW_DURATION 1 -grpc_core::TraceFlag grpc_timer_trace(false, "timer"); -grpc_core::TraceFlag grpc_timer_check_trace(false, "timer_check"); +extern "C" { +grpc_tracer_flag grpc_timer_trace = GRPC_TRACER_INITIALIZER(false, "timer"); +grpc_tracer_flag grpc_timer_check_trace = + GRPC_TRACER_INITIALIZER(false, "timer_check"); +} /* A "timer shard". Contains a 'heap' and a 'list' of timers. All timers with * deadlines earlier than 'queue_deadline" cap are maintained in the heap and @@ -71,14 +73,16 @@ typedef struct { grpc_timer list; } timer_shard; +static size_t g_num_shards; + /* Array of timer shards. Whenever a timer (grpc_timer *) is added, its address * is hashed to select the timer shard to add the timer to */ -static timer_shard g_shards[NUM_SHARDS]; +static timer_shard* g_shards; /* Maintains a sorted list of timer shards (sorted by their min_deadline, i.e * the deadline of the next timer in each shard). * Access to this is protected by g_shared_mutables.mu */ -static timer_shard* g_shard_queue[NUM_SHARDS]; +static timer_shard** g_shard_queue; #ifndef NDEBUG @@ -87,7 +91,7 @@ static timer_shard* g_shard_queue[NUM_SHARDS]; #define NUM_HASH_BUCKETS 1009 /* Prime number close to 1000 */ static gpr_mu g_hash_mu[NUM_HASH_BUCKETS]; /* One mutex per bucket */ -static grpc_timer* g_timer_ht[NUM_HASH_BUCKETS] = {NULL}; +static grpc_timer* g_timer_ht[NUM_HASH_BUCKETS] = {nullptr}; static void init_timer_ht() { for (int i = 0; i < NUM_HASH_BUCKETS; i++) { @@ -100,7 +104,7 @@ static bool is_in_ht(grpc_timer* t) { gpr_mu_lock(&g_hash_mu[i]); grpc_timer* p = g_timer_ht[i]; - while (p != NULL && p != t) { + while (p != nullptr && p != t) { p = p->hash_table_next; } gpr_mu_unlock(&g_hash_mu[i]); @@ -114,7 +118,7 @@ static void add_to_ht(grpc_timer* t) { gpr_mu_lock(&g_hash_mu[i]); grpc_timer* p = g_timer_ht[i]; - while (p != NULL && p != t) { + while (p != nullptr && p != t) { p = p->hash_table_next; } @@ -142,9 +146,9 @@ static void remove_from_ht(grpc_timer* t) { if (g_timer_ht[i] == t) { g_timer_ht[i] = g_timer_ht[i]->hash_table_next; removed = true; - } else if (g_timer_ht[i] != NULL) { + } else if (g_timer_ht[i] != nullptr) { grpc_timer* p = g_timer_ht[i]; - while (p->hash_table_next != NULL && p->hash_table_next != t) { + while (p->hash_table_next != nullptr && p->hash_table_next != t) { p = p->hash_table_next; } @@ -165,7 +169,7 @@ static void remove_from_ht(grpc_timer* t) { abort(); } - t->hash_table_next = NULL; + t->hash_table_next = nullptr; } /* If a timer is added to a timer shard (either heap or a list), it cannot @@ -238,14 +242,21 @@ static gpr_atm compute_min_deadline(timer_shard* shard) { void grpc_timer_list_init(grpc_exec_ctx* exec_ctx) { uint32_t i; + g_num_shards = GPR_MIN(1, 2 * gpr_cpu_num_cores()); + g_shards = (timer_shard*)gpr_zalloc(g_num_shards * sizeof(*g_shards)); + g_shard_queue = + (timer_shard**)gpr_zalloc(g_num_shards * sizeof(*g_shard_queue)); + g_shared_mutables.initialized = true; g_shared_mutables.checker_mu = GPR_SPINLOCK_INITIALIZER; gpr_mu_init(&g_shared_mutables.mu); g_shared_mutables.min_timer = grpc_exec_ctx_now(exec_ctx); gpr_tls_init(&g_last_seen_min_timer); gpr_tls_set(&g_last_seen_min_timer, 0); + grpc_register_tracer(&grpc_timer_trace); + grpc_register_tracer(&grpc_timer_check_trace); - for (i = 0; i < NUM_SHARDS; i++) { + for (i = 0; i < g_num_shards; i++) { timer_shard* shard = &g_shards[i]; gpr_mu_init(&shard->mu); grpc_time_averaged_stats_init(&shard->stats, 1.0 / ADD_DEADLINE_SCALE, 0.1, @@ -262,17 +273,19 @@ void grpc_timer_list_init(grpc_exec_ctx* exec_ctx) { } void grpc_timer_list_shutdown(grpc_exec_ctx* exec_ctx) { - int i; + size_t i; run_some_expired_timers( - exec_ctx, GPR_ATM_MAX, NULL, + exec_ctx, GPR_ATM_MAX, nullptr, GRPC_ERROR_CREATE_FROM_STATIC_STRING("Timer list shutdown")); - for (i = 0; i < NUM_SHARDS; i++) { + for (i = 0; i < g_num_shards; i++) { timer_shard* shard = &g_shards[i]; gpr_mu_destroy(&shard->mu); grpc_timer_heap_destroy(&shard->heap); } gpr_mu_destroy(&g_shared_mutables.mu); gpr_tls_destroy(&g_last_seen_min_timer); + gpr_free(g_shards); + gpr_free(g_shard_queue); g_shared_mutables.initialized = false; } @@ -306,7 +319,7 @@ static void note_deadline_change(timer_shard* shard) { g_shard_queue[shard->shard_queue_index - 1]->min_deadline) { swap_adjacent_shards_in_queue(shard->shard_queue_index - 1); } - while (shard->shard_queue_index < NUM_SHARDS - 1 && + while (shard->shard_queue_index < g_num_shards - 1 && shard->min_deadline > g_shard_queue[shard->shard_queue_index + 1]->min_deadline) { swap_adjacent_shards_in_queue(shard->shard_queue_index); @@ -318,15 +331,15 @@ void grpc_timer_init_unset(grpc_timer* timer) { timer->pending = false; } void grpc_timer_init(grpc_exec_ctx* exec_ctx, grpc_timer* timer, grpc_millis deadline, grpc_closure* closure) { int is_first_timer = 0; - timer_shard* shard = &g_shards[GPR_HASH_POINTER(timer, NUM_SHARDS)]; + timer_shard* shard = &g_shards[GPR_HASH_POINTER(timer, g_num_shards)]; timer->closure = closure; timer->deadline = deadline; #ifndef NDEBUG - timer->hash_table_next = NULL; + timer->hash_table_next = nullptr; #endif - if (grpc_timer_trace.enabled()) { + if (GRPC_TRACER_ON(grpc_timer_trace)) { gpr_log(GPR_DEBUG, "TIMER %p: SET %" PRIdPTR " now %" PRIdPTR " call %p[%p]", timer, deadline, grpc_exec_ctx_now(exec_ctx), closure, closure->cb); @@ -362,7 +375,7 @@ void grpc_timer_init(grpc_exec_ctx* exec_ctx, grpc_timer* timer, timer->heap_index = INVALID_HEAP_INDEX; list_join(&shard->list, timer); } - if (grpc_timer_trace.enabled()) { + if (GRPC_TRACER_ON(grpc_timer_trace)) { gpr_log(GPR_DEBUG, " .. add to shard %d with queue_deadline_cap=%" PRIdPTR " => is_first_timer=%s", @@ -384,7 +397,7 @@ void grpc_timer_init(grpc_exec_ctx* exec_ctx, grpc_timer* timer, grpc_timer_check. */ if (is_first_timer) { gpr_mu_lock(&g_shared_mutables.mu); - if (grpc_timer_trace.enabled()) { + if (GRPC_TRACER_ON(grpc_timer_trace)) { gpr_log(GPR_DEBUG, " .. old shard min_deadline=%" PRIdPTR, shard->min_deadline); } @@ -412,9 +425,9 @@ void grpc_timer_cancel(grpc_exec_ctx* exec_ctx, grpc_timer* timer) { return; } - timer_shard* shard = &g_shards[GPR_HASH_POINTER(timer, NUM_SHARDS)]; + timer_shard* shard = &g_shards[GPR_HASH_POINTER(timer, g_num_shards)]; gpr_mu_lock(&shard->mu); - if (grpc_timer_trace.enabled()) { + if (GRPC_TRACER_ON(grpc_timer_trace)) { gpr_log(GPR_DEBUG, "TIMER %p: CANCEL pending=%s", timer, timer->pending ? "true" : "false"); } @@ -455,7 +468,7 @@ static int refill_heap(timer_shard* shard, gpr_atm now) { saturating_add(GPR_MAX(now, shard->queue_deadline_cap), (gpr_atm)(deadline_delta * 1000.0)); - if (grpc_timer_check_trace.enabled()) { + if (GRPC_TRACER_ON(grpc_timer_check_trace)) { gpr_log(GPR_DEBUG, " .. shard[%d]->queue_deadline_cap --> %" PRIdPTR, (int)(shard - g_shards), shard->queue_deadline_cap); } @@ -463,7 +476,7 @@ static int refill_heap(timer_shard* shard, gpr_atm now) { next = timer->next; if (timer->deadline < shard->queue_deadline_cap) { - if (grpc_timer_check_trace.enabled()) { + if (GRPC_TRACER_ON(grpc_timer_check_trace)) { gpr_log(GPR_DEBUG, " .. add timer with deadline %" PRIdPTR " to heap", timer->deadline); } @@ -480,23 +493,23 @@ static int refill_heap(timer_shard* shard, gpr_atm now) { static grpc_timer* pop_one(timer_shard* shard, gpr_atm now) { grpc_timer* timer; for (;;) { - if (grpc_timer_check_trace.enabled()) { + if (GRPC_TRACER_ON(grpc_timer_check_trace)) { gpr_log(GPR_DEBUG, " .. shard[%d]: heap_empty=%s", (int)(shard - g_shards), grpc_timer_heap_is_empty(&shard->heap) ? "true" : "false"); } if (grpc_timer_heap_is_empty(&shard->heap)) { - if (now < shard->queue_deadline_cap) return NULL; - if (!refill_heap(shard, now)) return NULL; + if (now < shard->queue_deadline_cap) return nullptr; + if (!refill_heap(shard, now)) return nullptr; } timer = grpc_timer_heap_top(&shard->heap); - if (grpc_timer_check_trace.enabled()) { + if (GRPC_TRACER_ON(grpc_timer_check_trace)) { gpr_log(GPR_DEBUG, " .. check top timer deadline=%" PRIdPTR " now=%" PRIdPTR, timer->deadline, now); } - if (timer->deadline > now) return NULL; - if (grpc_timer_trace.enabled()) { + if (timer->deadline > now) return nullptr; + if (GRPC_TRACER_ON(grpc_timer_trace)) { gpr_log(GPR_DEBUG, "TIMER %p: FIRE %" PRIdPTR "ms late via %s scheduler", timer, now - timer->deadline, timer->closure->scheduler->vtable->name); @@ -521,7 +534,7 @@ static size_t pop_timers(grpc_exec_ctx* exec_ctx, timer_shard* shard, } *new_min_deadline = compute_min_deadline(shard); gpr_mu_unlock(&shard->mu); - if (grpc_timer_check_trace.enabled()) { + if (GRPC_TRACER_ON(grpc_timer_check_trace)) { gpr_log(GPR_DEBUG, " .. shard[%d] popped %" PRIdPTR, (int)(shard - g_shards), n); } @@ -537,7 +550,7 @@ static grpc_timer_check_result run_some_expired_timers(grpc_exec_ctx* exec_ctx, gpr_atm min_timer = gpr_atm_no_barrier_load(&g_shared_mutables.min_timer); gpr_tls_set(&g_last_seen_min_timer, min_timer); if (now < min_timer) { - if (next != NULL) *next = GPR_MIN(*next, min_timer); + if (next != nullptr) *next = GPR_MIN(*next, min_timer); return GRPC_TIMERS_CHECKED_AND_EMPTY; } @@ -545,7 +558,7 @@ static grpc_timer_check_result run_some_expired_timers(grpc_exec_ctx* exec_ctx, gpr_mu_lock(&g_shared_mutables.mu); result = GRPC_TIMERS_CHECKED_AND_EMPTY; - if (grpc_timer_check_trace.enabled()) { + if (GRPC_TRACER_ON(grpc_timer_check_trace)) { gpr_log(GPR_DEBUG, " .. shard[%d]->min_deadline = %" PRIdPTR, (int)(g_shard_queue[0] - g_shards), g_shard_queue[0]->min_deadline); @@ -563,7 +576,7 @@ static grpc_timer_check_result run_some_expired_timers(grpc_exec_ctx* exec_ctx, result = GRPC_TIMERS_FIRED; } - if (grpc_timer_check_trace.enabled()) { + if (GRPC_TRACER_ON(grpc_timer_check_trace)) { gpr_log(GPR_DEBUG, " .. result --> %d" ", shard[%d]->min_deadline %" PRIdPTR " --> %" PRIdPTR @@ -605,10 +618,10 @@ grpc_timer_check_result grpc_timer_check(grpc_exec_ctx* exec_ctx, mutable cacheline in the common case */ grpc_millis min_timer = gpr_tls_get(&g_last_seen_min_timer); if (now < min_timer) { - if (next != NULL) { + if (next != nullptr) { *next = GPR_MIN(*next, min_timer); } - if (grpc_timer_check_trace.enabled()) { + if (GRPC_TRACER_ON(grpc_timer_check_trace)) { gpr_log(GPR_DEBUG, "TIMER CHECK SKIP: now=%" PRIdPTR " min_timer=%" PRIdPTR, now, min_timer); @@ -622,9 +635,9 @@ grpc_timer_check_result grpc_timer_check(grpc_exec_ctx* exec_ctx, : GRPC_ERROR_CREATE_FROM_STATIC_STRING("Shutting down timer system"); // tracing - if (grpc_timer_check_trace.enabled()) { + if (GRPC_TRACER_ON(grpc_timer_check_trace)) { char* next_str; - if (next == NULL) { + if (next == nullptr) { next_str = gpr_strdup("NULL"); } else { gpr_asprintf(&next_str, "%" PRIdPTR, *next); @@ -640,9 +653,9 @@ grpc_timer_check_result grpc_timer_check(grpc_exec_ctx* exec_ctx, grpc_timer_check_result r = run_some_expired_timers(exec_ctx, now, next, shutdown_error); // tracing - if (grpc_timer_check_trace.enabled()) { + if (GRPC_TRACER_ON(grpc_timer_check_trace)) { char* next_str; - if (next == NULL) { + if (next == nullptr) { next_str = gpr_strdup("NULL"); } else { gpr_asprintf(&next_str, "%" PRIdPTR, *next); |