#include <grpc/support/port_platform.h>
#include "src/core/lib/iomgr/port.h"
#include <inttypes.h>
#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>
#include "src/core/lib/debug/trace.h"
#include "src/core/lib/gpr/spinlock.h"
#include "src/core/lib/gpr/tls.h"
#include "src/core/lib/gpr/useful.h"
#include "src/core/lib/iomgr/exec_ctx.h"
#include "src/core/lib/iomgr/time_averaged_stats.h"
#include "src/core/lib/iomgr/timer_heap.h"
#define INVALID_HEAP_INDEX 0xffffffffu
#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");
typedef struct {
gpr_mu mu;
grpc_time_averaged_stats stats;
grpc_millis queue_deadline_cap;
grpc_millis min_deadline;
uint32_t shard_queue_index;
grpc_timer_heap heap;
grpc_timer list;
} timer_shard;
static size_t g_num_shards;
static timer_shard* g_shards;
static timer_shard** g_shard_queue;
#ifndef NDEBUG
#define NUM_HASH_BUCKETS 1009
static gpr_mu g_hash_mu[NUM_HASH_BUCKETS];
static grpc_timer* g_timer_ht[NUM_HASH_BUCKETS] = {nullptr};
static void init_timer_ht() {
for (int i = 0; i < NUM_HASH_BUCKETS; i++) {
gpr_mu_init(&g_hash_mu[i]);
}
}
static void destroy_timer_ht() {
for (int i = 0; i < NUM_HASH_BUCKETS; i++) {
gpr_mu_destroy(&g_hash_mu[i]);
}
}
static bool is_in_ht(grpc_timer* t) {
size_t i = GPR_HASH_POINTER(t, NUM_HASH_BUCKETS);
gpr_mu_lock(&g_hash_mu[i]);
grpc_timer* p = g_timer_ht[i];
while (p != nullptr && p != t) {
p = p->hash_table_next;
}
gpr_mu_unlock(&g_hash_mu[i]);
return (p == t);
}
static void add_to_ht(grpc_timer* t) {
GPR_ASSERT(!t->hash_table_next);
size_t i = GPR_HASH_POINTER(t, NUM_HASH_BUCKETS);
gpr_mu_lock(&g_hash_mu[i]);
grpc_timer* p = g_timer_ht[i];
while (p != nullptr && p != t) {
p = p->hash_table_next;
}
if (p == t) {
grpc_closure* c = t->closure;
gpr_log(GPR_ERROR,
"** Duplicate timer (%p) being added. Closure: (%p), created at: "
"(%s:%d), scheduled at: (%s:%d) **",
t, c, c->file_created, c->line_created, c->file_initiated,
c->line_initiated);
abort();
}
t->hash_table_next = g_timer_ht[i];
g_timer_ht[i] = t;
gpr_mu_unlock(&g_hash_mu[i]);
}
static void remove_from_ht(grpc_timer* t) {
size_t i = GPR_HASH_POINTER(t, NUM_HASH_BUCKETS);
bool removed = false;
gpr_mu_lock(&g_hash_mu[i]);
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] != nullptr) {
grpc_timer* p = g_timer_ht[i];
while (p->hash_table_next != nullptr && p->hash_table_next != t) {
p = p->hash_table_next;
}
if (p->hash_table_next == t) {
p->hash_table_next = t->hash_table_next;
removed = true;
}
}
gpr_mu_unlock(&g_hash_mu[i]);
if (!removed) {
grpc_closure* c = t->closure;
gpr_log(GPR_ERROR,
"** Removing timer (%p) that is not added to hash table. Closure "
"(%p), created at: (%s:%d), scheduled at: (%s:%d) **",
t, c, c->file_created, c->line_created, c->file_initiated,
c->line_initiated);
abort();
}
t->hash_table_next = nullptr;
}
static void validate_non_pending_timer(grpc_timer* t) {
if (!t->pending && is_in_ht(t)) {
grpc_closure* c = t->closure;
gpr_log(GPR_ERROR,
"** gpr_timer_cancel() called on a non-pending timer (%p) which "
"is in the hash table. Closure: (%p), created at: (%s:%d), "
"scheduled at: (%s:%d) **",
t, c, c->file_created, c->line_created, c->file_initiated,
c->line_initiated);
abort();
}
}
#define INIT_TIMER_HASH_TABLE() init_timer_ht()
#define DESTROY_TIMER_HASH_TABLE() destroy_timer_ht()
#define ADD_TO_HASH_TABLE(t) add_to_ht((t))
#define REMOVE_FROM_HASH_TABLE(t) remove_from_ht((t))
#define VALIDATE_NON_PENDING_TIMER(t) validate_non_pending_timer((t))
#else
#define INIT_TIMER_HASH_TABLE()
#define DESTROY_TIMER_HASH_TABLE()
#define ADD_TO_HASH_TABLE(t)
#define REMOVE_FROM_HASH_TABLE(t)
#define VALIDATE_NON_PENDING_TIMER(t)
#endif
#if GPR_ARCH_64
GPR_TLS_DECL(g_last_seen_min_timer);
#endif
struct shared_mutables {
grpc_millis min_timer;
gpr_spinlock checker_mu;
bool initialized;
gpr_mu mu;
} GPR_ALIGN_STRUCT(GPR_CACHELINE_SIZE);
static struct shared_mutables g_shared_mutables;
static grpc_millis saturating_add(grpc_millis a, grpc_millis b) {
if (a > GRPC_MILLIS_INF_FUTURE - b) {
return GRPC_MILLIS_INF_FUTURE;
}
return a + b;
}
static grpc_timer_check_result run_some_expired_timers(grpc_millis now,
grpc_millis* next,
grpc_error* error);
static grpc_millis compute_min_deadline(timer_shard* shard) {
return grpc_timer_heap_is_empty(&shard->heap)
? saturating_add(shard->queue_deadline_cap, 1)
: grpc_timer_heap_top(&shard->heap)->deadline;
}
static void timer_list_init() {
uint32_t i;
g_num_shards = GPR_CLAMP(2 * gpr_cpu_num_cores(), 1, 32);
g_shards =
static_cast<timer_shard*>(gpr_zalloc(g_num_shards * sizeof(*g_shards)));
g_shard_queue = static_cast<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_core::ExecCtx::Get()->Now();
#if GPR_ARCH_64
gpr_tls_init(&g_last_seen_min_timer);
gpr_tls_set(&g_last_seen_min_timer, 0);
#endif
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,
0.5);
shard->queue_deadline_cap = g_shared_mutables.min_timer;
shard->shard_queue_index = i;
grpc_timer_heap_init(&shard->heap);
shard->list.next = shard->list.prev = &shard->list;
shard->min_deadline = compute_min_deadline(shard);
g_shard_queue[i] = shard;
}
INIT_TIMER_HASH_TABLE();
}
static void timer_list_shutdown() {
size_t i;
run_some_expired_timers(
GRPC_MILLIS_INF_FUTURE, nullptr,
GRPC_ERROR_CREATE_FROM_STATIC_STRING("Timer list shutdown"));
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);
#if GPR_ARCH_64
gpr_tls_destroy(&g_last_seen_min_timer);
#endif
gpr_free(g_shards);
gpr_free(g_shard_queue);
g_shared_mutables.initialized = false;
DESTROY_TIMER_HASH_TABLE();
}
static void list_join(grpc_timer* head, grpc_timer* timer) {
timer->next = head;
timer->prev = head->prev;
timer->next->prev = timer->prev->next = timer;
}
static void list_remove(grpc_timer* timer) {
timer->next->prev = timer->prev;
timer->prev->next = timer->next;
}
static void swap_adjacent_shards_in_queue(uint32_t first_shard_queue_index) {
timer_shard* temp;
temp = g_shard_queue[first_shard_queue_index];
g_shard_queue[first_shard_queue_index] =
g_shard_queue[first_shard_queue_index + 1];
g_shard_queue[first_shard_queue_index + 1] = temp;
g_shard_queue[first_shard_queue_index]->shard_queue_index =
first_shard_queue_index;
g_shard_queue[first_shard_queue_index + 1]->shard_queue_index =
first_shard_queue_index + 1;
}
static void note_deadline_change(timer_shard* shard) {
while (shard->shard_queue_index > 0 &&
shard->min_deadline <
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 < 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);
}
}
void grpc_timer_init_unset(grpc_timer* timer) { timer->pending = false; }
static void timer_init(grpc_timer* timer, grpc_millis deadline,
grpc_closure* closure) {
int is_first_timer = 0;
timer_shard* shard = &g_shards[GPR_HASH_POINTER(timer, g_num_shards)];
timer->closure = closure;
timer->deadline = deadline;
#ifndef NDEBUG
timer->hash_table_next = nullptr;
#endif
if (grpc_timer_trace.enabled()) {
gpr_log(GPR_INFO, "TIMER %p: SET %" PRId64 " now %" PRId64 " call %p[%p]",
timer, deadline, grpc_core::ExecCtx::Get()->Now(), closure,
closure->cb);
}
if (!g_shared_mutables.initialized) {
timer->pending = false;
GRPC_CLOSURE_SCHED(timer->closure,
GRPC_ERROR_CREATE_FROM_STATIC_STRING(
"Attempt to create timer before initialization"));
return;
}
gpr_mu_lock(&shard->mu);
timer->pending = true;
grpc_millis now = grpc_core::ExecCtx::Get()->Now();
if (deadline <= now) {
timer->pending = false;
GRPC_CLOSURE_SCHED(timer->closure, GRPC_ERROR_NONE);
gpr_mu_unlock(&shard->mu);
return;
}
grpc_time_averaged_stats_add_sample(
&shard->stats, static_cast<double>(deadline - now) / 1000.0);
ADD_TO_HASH_TABLE(timer);
if (deadline < shard->queue_deadline_cap) {
is_first_timer = grpc_timer_heap_add(&shard->heap, timer);
} else {
timer->heap_index = INVALID_HEAP_INDEX;
list_join(&shard->list, timer);
}
if (grpc_timer_trace.enabled()) {
gpr_log(GPR_INFO,
" .. add to shard %d with queue_deadline_cap=%" PRId64
" => is_first_timer=%s",
static_cast<int>(shard - g_shards), shard->queue_deadline_cap,
is_first_timer ? "true" : "false");
}
gpr_mu_unlock(&shard->mu);
if (is_first_timer) {
gpr_mu_lock(&g_shared_mutables.mu);
if (grpc_timer_trace.enabled()) {
gpr_log(GPR_INFO, " .. old shard min_deadline=%" PRId64,
shard->min_deadline);
}
if (deadline < shard->min_deadline) {
grpc_millis old_min_deadline = g_shard_queue[0]->min_deadline;
shard->min_deadline = deadline;
note_deadline_change(shard);
if (shard->shard_queue_index == 0 && deadline < old_min_deadline) {
#if GPR_ARCH_64
gpr_atm_no_barrier_store((gpr_atm*)(&g_shared_mutables.min_timer),
deadline);
#else
g_shared_mutables.min_timer = deadline;
#endif
grpc_kick_poller();
}
}
gpr_mu_unlock(&g_shared_mutables.mu);
}
}
static void timer_consume_kick(void) {
#if GPR_ARCH_64
gpr_tls_set(&g_last_seen_min_timer, 0);
#endif
}
static void timer_cancel(grpc_timer* timer) {
if (!g_shared_mutables.initialized) {
return;
}
timer_shard* shard = &g_shards[GPR_HASH_POINTER(timer, g_num_shards)];
gpr_mu_lock(&shard->mu);
if (grpc_timer_trace.enabled()) {
gpr_log(GPR_INFO, "TIMER %p: CANCEL pending=%s", timer,
timer->pending ? "true" : "false");
}
if (timer->pending) {
REMOVE_FROM_HASH_TABLE(timer);
GRPC_CLOSURE_SCHED(timer->closure, GRPC_ERROR_CANCELLED);
timer->pending = false;
if (timer->heap_index == INVALID_HEAP_INDEX) {
list_remove(timer);
} else {
grpc_timer_heap_remove(&shard->heap, timer);
}
} else {
VALIDATE_NON_PENDING_TIMER(timer);
}
gpr_mu_unlock(&shard->mu);
}
static bool refill_heap(timer_shard* shard, grpc_millis now) {
double computed_deadline_delta =
grpc_time_averaged_stats_update_average(&shard->stats) *
ADD_DEADLINE_SCALE;
double deadline_delta =
GPR_CLAMP(computed_deadline_delta, MIN_QUEUE_WINDOW_DURATION,
MAX_QUEUE_WINDOW_DURATION);
grpc_timer *timer, *next;
shard->queue_deadline_cap =
saturating_add(GPR_MAX(now, shard->queue_deadline_cap),
static_cast<grpc_millis>(deadline_delta * 1000.0));
if (grpc_timer_check_trace.enabled()) {
gpr_log(GPR_INFO, " .. shard[%d]->queue_deadline_cap --> %" PRId64,
static_cast<int>(shard - g_shards), shard->queue_deadline_cap);
}
for (timer = shard->list.next; timer != &shard->list; timer = next) {
next = timer->next;
if (timer->deadline < shard->queue_deadline_cap) {
if (grpc_timer_check_trace.enabled()) {
gpr_log(GPR_INFO, " .. add timer with deadline %" PRId64 " to heap",
timer->deadline);
}
list_remove(timer);
grpc_timer_heap_add(&shard->heap, timer);
}
}
return !grpc_timer_heap_is_empty(&shard->heap);
}
static grpc_timer* pop_one(timer_shard* shard, grpc_millis now) {
grpc_timer* timer;
for (;;) {
if (grpc_timer_check_trace.enabled()) {
gpr_log(GPR_INFO, " .. shard[%d]: heap_empty=%s",
static_cast<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 nullptr;
if (!refill_heap(shard, now)) return nullptr;
}
timer = grpc_timer_heap_top(&shard->heap);
if (grpc_timer_check_trace.enabled()) {
gpr_log(GPR_INFO,
" .. check top timer deadline=%" PRId64 " now=%" PRId64,
timer->deadline, now);
}
if (timer->deadline > now) return nullptr;
if (grpc_timer_trace.enabled()) {
gpr_log(GPR_INFO, "TIMER %p: FIRE %" PRId64 "ms late via %s scheduler",
timer, now - timer->deadline,
timer->closure->scheduler->vtable->name);
}
timer->pending = false;
grpc_timer_heap_pop(&shard->heap);
return timer;
}
}
static size_t pop_timers(timer_shard* shard, grpc_millis now,
grpc_millis* new_min_deadline, grpc_error* error) {
size_t n = 0;
grpc_timer* timer;
gpr_mu_lock(&shard->mu);
while ((timer = pop_one(shard, now))) {
REMOVE_FROM_HASH_TABLE(timer);
GRPC_CLOSURE_SCHED(timer->closure, GRPC_ERROR_REF(error));
n++;
}
*new_min_deadline = compute_min_deadline(shard);
gpr_mu_unlock(&shard->mu);
if (grpc_timer_check_trace.enabled()) {
gpr_log(GPR_INFO, " .. shard[%d] popped %" PRIdPTR,
static_cast<int>(shard - g_shards), n);
}
return n;
}
static grpc_timer_check_result run_some_expired_timers(grpc_millis now,
grpc_millis* next,
grpc_error* error) {
grpc_timer_check_result result = GRPC_TIMERS_NOT_CHECKED;
#if GPR_ARCH_64
grpc_millis min_timer = static_cast<grpc_millis>(
gpr_atm_no_barrier_load((gpr_atm*)(&g_shared_mutables.min_timer)));
gpr_tls_set(&g_last_seen_min_timer, min_timer);
#else
gpr_mu_lock(&g_shared_mutables.mu);
grpc_millis min_timer = g_shared_mutables.min_timer;
gpr_mu_unlock(&g_shared_mutables.mu);
#endif
if (now < min_timer) {
if (next != nullptr) *next = GPR_MIN(*next, min_timer);
return GRPC_TIMERS_CHECKED_AND_EMPTY;
}
if (gpr_spinlock_trylock(&g_shared_mutables.checker_mu)) {
gpr_mu_lock(&g_shared_mutables.mu);
result = GRPC_TIMERS_CHECKED_AND_EMPTY;
if (grpc_timer_check_trace.enabled()) {
gpr_log(GPR_INFO, " .. shard[%d]->min_deadline = %" PRId64,
static_cast<int>(g_shard_queue[0] - g_shards),
g_shard_queue[0]->min_deadline);
}
while (g_shard_queue[0]->min_deadline < now ||
(now != GRPC_MILLIS_INF_FUTURE &&
g_shard_queue[0]->min_deadline == now)) {
grpc_millis new_min_deadline;
if (pop_timers(g_shard_queue[0], now, &new_min_deadline, error) > 0) {
result = GRPC_TIMERS_FIRED;
}
if (grpc_timer_check_trace.enabled()) {
gpr_log(GPR_INFO,
" .. result --> %d"
", shard[%d]->min_deadline %" PRId64 " --> %" PRId64
", now=%" PRId64,
result, static_cast<int>(g_shard_queue[0] - g_shards),
g_shard_queue[0]->min_deadline, new_min_deadline, now);
}
g_shard_queue[0]->min_deadline = new_min_deadline;
note_deadline_change(g_shard_queue[0]);
}
if (next) {
*next = GPR_MIN(*next, g_shard_queue[0]->min_deadline);
}
#if GPR_ARCH_64
gpr_atm_no_barrier_store((gpr_atm*)(&g_shared_mutables.min_timer),
g_shard_queue[0]->min_deadline);
#else
g_shared_mutables.min_timer = g_shard_queue[0]->min_deadline;
#endif
gpr_mu_unlock(&g_shared_mutables.mu);
gpr_spinlock_unlock(&g_shared_mutables.checker_mu);
}
GRPC_ERROR_UNREF(error);
return result;
}
static grpc_timer_check_result timer_check(grpc_millis* next) {
grpc_millis now = grpc_core::ExecCtx::Get()->Now();
#if GPR_ARCH_64
grpc_millis min_timer = gpr_tls_get(&g_last_seen_min_timer);
#else
gpr_mu_lock(&g_shared_mutables.mu);
grpc_millis min_timer = g_shared_mutables.min_timer;
gpr_mu_unlock(&g_shared_mutables.mu);
#endif
if (now < min_timer) {
if (next != nullptr) {
*next = GPR_MIN(*next, min_timer);
}
if (grpc_timer_check_trace.enabled()) {
gpr_log(GPR_INFO, "TIMER CHECK SKIP: now=%" PRId64 " min_timer=%" PRId64,
now, min_timer);
}
return GRPC_TIMERS_CHECKED_AND_EMPTY;
}
grpc_error* shutdown_error =
now != GRPC_MILLIS_INF_FUTURE
? GRPC_ERROR_NONE
: GRPC_ERROR_CREATE_FROM_STATIC_STRING("Shutting down timer system");
if (grpc_timer_check_trace.enabled()) {
char* next_str;
if (next == nullptr) {
next_str = gpr_strdup("NULL");
} else {
gpr_asprintf(&next_str, "%" PRId64, *next);
}
#if GPR_ARCH_64
gpr_log(GPR_INFO,
"TIMER CHECK BEGIN: now=%" PRId64 " next=%s tls_min=%" PRId64
" glob_min=%" PRId64,
now, next_str, min_timer,
static_cast<grpc_millis>(gpr_atm_no_barrier_load(
(gpr_atm*)(&g_shared_mutables.min_timer))));
#else
gpr_log(GPR_INFO, "TIMER CHECK BEGIN: now=%" PRId64 " next=%s min=%" PRId64,
now, next_str, min_timer);
#endif
gpr_free(next_str);
}
grpc_timer_check_result r =
run_some_expired_timers(now, next, shutdown_error);
if (grpc_timer_check_trace.enabled()) {
char* next_str;
if (next == nullptr) {
next_str = gpr_strdup("NULL");
} else {
gpr_asprintf(&next_str, "%" PRId64, *next);
}
gpr_log(GPR_INFO, "TIMER CHECK END: r=%d; next=%s", r, next_str);
gpr_free(next_str);
}
return r;
}
grpc_timer_vtable grpc_generic_timer_vtable = {
timer_init, timer_cancel, timer_check,
timer_list_init, timer_list_shutdown, timer_consume_kick};