diff options
Diffstat (limited to 'src/core/lib/iomgr')
-rw-r--r-- | src/core/lib/iomgr/combiner.c | 12 | ||||
-rw-r--r-- | src/core/lib/iomgr/combiner.h | 3 | ||||
-rw-r--r-- | src/core/lib/iomgr/iomgr.c | 3 | ||||
-rw-r--r-- | src/core/lib/iomgr/iomgr.h | 3 | ||||
-rw-r--r-- | src/core/lib/iomgr/resource_quota.c | 16 | ||||
-rw-r--r-- | src/core/lib/iomgr/resource_quota.h | 3 | ||||
-rw-r--r-- | src/core/lib/iomgr/tcp_client_posix.c | 8 | ||||
-rw-r--r-- | src/core/lib/iomgr/tcp_posix.c | 14 | ||||
-rw-r--r-- | src/core/lib/iomgr/tcp_posix.h | 3 | ||||
-rw-r--r-- | src/core/lib/iomgr/tcp_server_posix.c | 2 | ||||
-rw-r--r-- | src/core/lib/iomgr/timer_generic.c | 49 | ||||
-rw-r--r-- | src/core/lib/iomgr/timer_manager.c | 56 | ||||
-rw-r--r-- | src/core/lib/iomgr/timer_manager.h | 9 |
13 files changed, 115 insertions, 66 deletions
diff --git a/src/core/lib/iomgr/combiner.c b/src/core/lib/iomgr/combiner.c index 05cdbdad2b..fa7d576133 100644 --- a/src/core/lib/iomgr/combiner.c +++ b/src/core/lib/iomgr/combiner.c @@ -42,13 +42,13 @@ #include "src/core/lib/iomgr/workqueue.h" #include "src/core/lib/profiling/timers.h" -int grpc_combiner_trace = 0; +grpc_tracer_flag grpc_combiner_trace; -#define GRPC_COMBINER_TRACE(fn) \ - do { \ - if (grpc_combiner_trace) { \ - fn; \ - } \ +#define GRPC_COMBINER_TRACE(fn) \ + do { \ + if (GRPC_TRACER_ON(grpc_combiner_trace)) { \ + fn; \ + } \ } while (0) #define STATE_UNORPHANED 1 diff --git a/src/core/lib/iomgr/combiner.h b/src/core/lib/iomgr/combiner.h index 75dcb0b70a..6ab7a2b26b 100644 --- a/src/core/lib/iomgr/combiner.h +++ b/src/core/lib/iomgr/combiner.h @@ -37,6 +37,7 @@ #include <stddef.h> #include <grpc/support/atm.h> +#include "src/core/lib/debug/trace.h" #include "src/core/lib/iomgr/exec_ctx.h" #include "src/core/lib/support/mpscq.h" @@ -78,6 +79,6 @@ grpc_closure_scheduler *grpc_combiner_finally_scheduler(grpc_combiner *lock, bool grpc_combiner_continue_exec_ctx(grpc_exec_ctx *exec_ctx); -extern int grpc_combiner_trace; +extern grpc_tracer_flag grpc_combiner_trace; #endif /* GRPC_CORE_LIB_IOMGR_COMBINER_H */ diff --git a/src/core/lib/iomgr/iomgr.c b/src/core/lib/iomgr/iomgr.c index 445f7aa422..37804d8600 100644 --- a/src/core/lib/iomgr/iomgr.c +++ b/src/core/lib/iomgr/iomgr.c @@ -66,6 +66,9 @@ void grpc_iomgr_init(void) { g_root_object.name = "root"; grpc_network_status_init(); grpc_iomgr_platform_init(); +} + +void grpc_iomgr_start(void) { grpc_timer_manager_init(); } diff --git a/src/core/lib/iomgr/iomgr.h b/src/core/lib/iomgr/iomgr.h index 245a1e08aa..6e2e023615 100644 --- a/src/core/lib/iomgr/iomgr.h +++ b/src/core/lib/iomgr/iomgr.h @@ -40,6 +40,9 @@ /** Initializes the iomgr. */ void grpc_iomgr_init(void); +/** Starts any background threads for iomgr. */ +void grpc_iomgr_start(void); + /** Signals the intention to shutdown the iomgr. Expects to be able to flush * exec_ctx. */ void grpc_iomgr_shutdown(grpc_exec_ctx *exec_ctx); diff --git a/src/core/lib/iomgr/resource_quota.c b/src/core/lib/iomgr/resource_quota.c index c3ee878651..52ccfc0aad 100644 --- a/src/core/lib/iomgr/resource_quota.c +++ b/src/core/lib/iomgr/resource_quota.c @@ -44,7 +44,7 @@ #include "src/core/lib/iomgr/combiner.h" -int grpc_resource_quota_trace = 0; +grpc_tracer_flag grpc_resource_quota_trace; #define MEMORY_USAGE_ESTIMATION_MAX 65536 @@ -307,13 +307,13 @@ static bool rq_alloc(grpc_exec_ctx *exec_ctx, resource_user->free_pool = 0; resource_quota->free_pool -= amt; rq_update_estimate(resource_quota); - if (grpc_resource_quota_trace) { + if (GRPC_TRACER_ON(grpc_resource_quota_trace)) { gpr_log(GPR_DEBUG, "RQ %s %s: grant alloc %" PRId64 " bytes; rq_free_pool -> %" PRId64, resource_quota->name, resource_user->name, amt, resource_quota->free_pool); } - } else if (grpc_resource_quota_trace && resource_user->free_pool >= 0) { + } else if (GRPC_TRACER_ON(grpc_resource_quota_trace) && resource_user->free_pool >= 0) { gpr_log(GPR_DEBUG, "RQ %s %s: discard already satisfied alloc request", resource_quota->name, resource_user->name); } @@ -342,7 +342,7 @@ static bool rq_reclaim_from_per_user_free_pool( resource_user->free_pool = 0; resource_quota->free_pool += amt; rq_update_estimate(resource_quota); - if (grpc_resource_quota_trace) { + if (GRPC_TRACER_ON(grpc_resource_quota_trace)) { gpr_log(GPR_DEBUG, "RQ %s %s: reclaim_from_per_user_free_pool %" PRId64 " bytes; rq_free_pool -> %" PRId64, resource_quota->name, resource_user->name, amt, @@ -365,7 +365,7 @@ static bool rq_reclaim(grpc_exec_ctx *exec_ctx, : GRPC_RULIST_RECLAIMER_BENIGN; grpc_resource_user *resource_user = rulist_pop_head(resource_quota, list); if (resource_user == NULL) return false; - if (grpc_resource_quota_trace) { + if (GRPC_TRACER_ON(grpc_resource_quota_trace)) { gpr_log(GPR_DEBUG, "RQ %s %s: initiate %s reclamation", resource_quota->name, resource_user->name, destructive ? "destructive" : "benign"); @@ -786,7 +786,7 @@ void grpc_resource_user_alloc(grpc_exec_ctx *exec_ctx, gpr_mu_lock(&resource_user->mu); ru_ref_by(resource_user, (gpr_atm)size); resource_user->free_pool -= (int64_t)size; - if (grpc_resource_quota_trace) { + if (GRPC_TRACER_ON(grpc_resource_quota_trace)) { gpr_log(GPR_DEBUG, "RQ %s %s: alloc %" PRIdPTR "; free_pool -> %" PRId64, resource_user->resource_quota->name, resource_user->name, size, resource_user->free_pool); @@ -810,7 +810,7 @@ void grpc_resource_user_free(grpc_exec_ctx *exec_ctx, gpr_mu_lock(&resource_user->mu); bool was_zero_or_negative = resource_user->free_pool <= 0; resource_user->free_pool += (int64_t)size; - if (grpc_resource_quota_trace) { + if (GRPC_TRACER_ON(grpc_resource_quota_trace)) { gpr_log(GPR_DEBUG, "RQ %s %s: free %" PRIdPTR "; free_pool -> %" PRId64, resource_user->resource_quota->name, resource_user->name, size, resource_user->free_pool); @@ -839,7 +839,7 @@ void grpc_resource_user_post_reclaimer(grpc_exec_ctx *exec_ctx, void grpc_resource_user_finish_reclamation(grpc_exec_ctx *exec_ctx, grpc_resource_user *resource_user) { - if (grpc_resource_quota_trace) { + if (GRPC_TRACER_ON(grpc_resource_quota_trace)) { gpr_log(GPR_DEBUG, "RQ %s %s: reclamation complete", resource_user->resource_quota->name, resource_user->name); } diff --git a/src/core/lib/iomgr/resource_quota.h b/src/core/lib/iomgr/resource_quota.h index 6f99be0d51..51122dad01 100644 --- a/src/core/lib/iomgr/resource_quota.h +++ b/src/core/lib/iomgr/resource_quota.h @@ -36,6 +36,7 @@ #include <grpc/grpc.h> +#include "src/core/lib/debug/trace.h" #include "src/core/lib/iomgr/exec_ctx.h" /** \file Tracks resource usage against a pool. @@ -75,7 +76,7 @@ maintain lists of users (which users arrange to leave before they are destroyed) */ -extern int grpc_resource_quota_trace; +extern grpc_tracer_flag grpc_resource_quota_trace; grpc_resource_quota *grpc_resource_quota_ref_internal( grpc_resource_quota *resource_quota); diff --git a/src/core/lib/iomgr/tcp_client_posix.c b/src/core/lib/iomgr/tcp_client_posix.c index a2692707d9..ed3fd94a98 100644 --- a/src/core/lib/iomgr/tcp_client_posix.c +++ b/src/core/lib/iomgr/tcp_client_posix.c @@ -58,7 +58,7 @@ #include "src/core/lib/iomgr/unix_sockets_posix.h" #include "src/core/lib/support/string.h" -extern int grpc_tcp_trace; +extern grpc_tracer_flag grpc_tcp_trace; typedef struct { gpr_mu mu; @@ -114,7 +114,7 @@ done: static void tc_on_alarm(grpc_exec_ctx *exec_ctx, void *acp, grpc_error *error) { int done; async_connect *ac = acp; - if (grpc_tcp_trace) { + if (GRPC_TRACER_ON(grpc_tcp_trace)) { const char *str = grpc_error_string(error); gpr_log(GPR_DEBUG, "CLIENT_CONNECT: %s: on_alarm: error=%s", ac->addr_str, str); @@ -152,7 +152,7 @@ static void on_writable(grpc_exec_ctx *exec_ctx, void *acp, grpc_error *error) { GRPC_ERROR_REF(error); - if (grpc_tcp_trace) { + if (GRPC_TRACER_ON(grpc_tcp_trace)) { const char *str = grpc_error_string(error); gpr_log(GPR_DEBUG, "CLIENT_CONNECT: %s: on_writable: error=%s", ac->addr_str, str); @@ -330,7 +330,7 @@ static void tcp_client_connect_impl(grpc_exec_ctx *exec_ctx, grpc_schedule_on_exec_ctx); ac->channel_args = grpc_channel_args_copy(channel_args); - if (grpc_tcp_trace) { + if (GRPC_TRACER_ON(grpc_tcp_trace)) { gpr_log(GPR_DEBUG, "CLIENT_CONNECT: %s: asynchronously connecting", ac->addr_str); } diff --git a/src/core/lib/iomgr/tcp_posix.c b/src/core/lib/iomgr/tcp_posix.c index 5f4b38de2b..233dea9bec 100644 --- a/src/core/lib/iomgr/tcp_posix.c +++ b/src/core/lib/iomgr/tcp_posix.c @@ -74,7 +74,7 @@ typedef GRPC_MSG_IOVLEN_TYPE msg_iovlen_type; typedef size_t msg_iovlen_type; #endif -int grpc_tcp_trace = 0; +grpc_tracer_flag grpc_tcp_trace; typedef struct { grpc_endpoint base; @@ -221,7 +221,7 @@ static void call_read_cb(grpc_exec_ctx *exec_ctx, grpc_tcp *tcp, grpc_error *error) { grpc_closure *cb = tcp->read_cb; - if (grpc_tcp_trace) { + if (GRPC_TRACER_ON(grpc_tcp_trace)) { size_t i; const char *str = grpc_error_string(error); gpr_log(GPR_DEBUG, "read: error=%s", str); @@ -468,14 +468,14 @@ static void tcp_handle_write(grpc_exec_ctx *exec_ctx, void *arg /* grpc_tcp */, } if (!tcp_flush(tcp, &error)) { - if (grpc_tcp_trace) { + if (GRPC_TRACER_ON(grpc_tcp_trace)) { gpr_log(GPR_DEBUG, "write: delayed"); } grpc_fd_notify_on_write(exec_ctx, tcp->em_fd, &tcp->write_closure); } else { cb = tcp->write_cb; tcp->write_cb = NULL; - if (grpc_tcp_trace) { + if (GRPC_TRACER_ON(grpc_tcp_trace)) { const char *str = grpc_error_string(error); gpr_log(GPR_DEBUG, "write: %s", str); } @@ -490,7 +490,7 @@ static void tcp_write(grpc_exec_ctx *exec_ctx, grpc_endpoint *ep, grpc_tcp *tcp = (grpc_tcp *)ep; grpc_error *error = GRPC_ERROR_NONE; - if (grpc_tcp_trace) { + if (GRPC_TRACER_ON(grpc_tcp_trace)) { size_t i; for (i = 0; i < buf->count; i++) { @@ -521,12 +521,12 @@ static void tcp_write(grpc_exec_ctx *exec_ctx, grpc_endpoint *ep, if (!tcp_flush(tcp, &error)) { TCP_REF(tcp, "write"); tcp->write_cb = cb; - if (grpc_tcp_trace) { + if (GRPC_TRACER_ON(grpc_tcp_trace)) { gpr_log(GPR_DEBUG, "write: delayed"); } grpc_fd_notify_on_write(exec_ctx, tcp->em_fd, &tcp->write_closure); } else { - if (grpc_tcp_trace) { + if (GRPC_TRACER_ON(grpc_tcp_trace)) { const char *str = grpc_error_string(error); gpr_log(GPR_DEBUG, "write: %s", str); } diff --git a/src/core/lib/iomgr/tcp_posix.h b/src/core/lib/iomgr/tcp_posix.h index 1ad5788331..4ad60c116e 100644 --- a/src/core/lib/iomgr/tcp_posix.h +++ b/src/core/lib/iomgr/tcp_posix.h @@ -44,10 +44,11 @@ otherwise specified. */ +#include "src/core/lib/debug/trace.h" #include "src/core/lib/iomgr/endpoint.h" #include "src/core/lib/iomgr/ev_posix.h" -extern int grpc_tcp_trace; +extern grpc_tracer_flag grpc_tcp_trace; /* Create a tcp endpoint given a file desciptor and a read slice size. Takes ownership of fd. */ diff --git a/src/core/lib/iomgr/tcp_server_posix.c b/src/core/lib/iomgr/tcp_server_posix.c index e66ffc9b1c..08997b5e2b 100644 --- a/src/core/lib/iomgr/tcp_server_posix.c +++ b/src/core/lib/iomgr/tcp_server_posix.c @@ -257,7 +257,7 @@ static void on_read(grpc_exec_ctx *exec_ctx, void *arg, grpc_error *err) { addr_str = grpc_sockaddr_to_uri(&addr); gpr_asprintf(&name, "tcp-server-connection:%s", addr_str); - if (grpc_tcp_trace) { + if (GRPC_TRACER_ON(grpc_tcp_trace)) { gpr_log(GPR_DEBUG, "SERVER_CONNECT: incoming connection: %s", addr_str); } diff --git a/src/core/lib/iomgr/timer_generic.c b/src/core/lib/iomgr/timer_generic.c index d8e6068431..ec0bb169cb 100644 --- a/src/core/lib/iomgr/timer_generic.c +++ b/src/core/lib/iomgr/timer_generic.c @@ -56,8 +56,8 @@ #define MIN_QUEUE_WINDOW_DURATION 0.01 #define MAX_QUEUE_WINDOW_DURATION 1 -int grpc_timer_trace = 0; -int grpc_timer_check_trace = 0; +grpc_tracer_flag grpc_timer_trace; +grpc_tracer_flag grpc_timer_check_trace; typedef struct { gpr_mu mu; @@ -232,14 +232,13 @@ void grpc_timer_init(grpc_exec_ctx *exec_ctx, grpc_timer *timer, GPR_ASSERT(deadline.clock_type == g_clock_type); GPR_ASSERT(now.clock_type == g_clock_type); timer->closure = closure; - timer->deadline = timespec_to_atm_round_up(deadline); + gpr_atm deadline_atm = timer->deadline = timespec_to_atm_round_up(deadline); - if (grpc_timer_trace) { + if (GRPC_TRACER_ON(grpc_timer_trace)) { gpr_log(GPR_DEBUG, "TIMER %p: SET %" PRId64 ".%09d [%" PRIdPTR "] now %" PRId64 ".%09d [%" PRIdPTR "] call %p[%p]", - timer, deadline.tv_sec, deadline.tv_nsec, timer->deadline, - now.tv_sec, now.tv_nsec, timespec_to_atm_round_down(now), closure, - closure->cb); + timer, deadline.tv_sec, deadline.tv_nsec, deadline_atm, now.tv_sec, + now.tv_nsec, timespec_to_atm_round_down(now), closure, closure->cb); } if (!g_shared_mutables.initialized) { @@ -262,13 +261,13 @@ void grpc_timer_init(grpc_exec_ctx *exec_ctx, grpc_timer *timer, grpc_time_averaged_stats_add_sample(&shard->stats, ts_to_dbl(gpr_time_sub(deadline, now))); - if (timer->deadline < shard->queue_deadline_cap) { + if (deadline_atm < 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) { + if (GRPC_TRACER_ON(grpc_timer_trace)) { gpr_log(GPR_DEBUG, " .. add to shard %d with queue_deadline_cap=%" PRIdPTR " => is_first_timer=%s", (int)(shard - g_shards), shard->queue_deadline_cap, @@ -289,16 +288,16 @@ 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) { + if (GRPC_TRACER_ON(grpc_timer_trace)) { gpr_log(GPR_DEBUG, " .. old shard min_deadline=%" PRIdPTR, shard->min_deadline); } - if (timer->deadline < shard->min_deadline) { + if (deadline_atm < shard->min_deadline) { gpr_atm old_min_deadline = g_shard_queue[0]->min_deadline; - shard->min_deadline = timer->deadline; + shard->min_deadline = deadline_atm; note_deadline_change(shard); - if (shard->shard_queue_index == 0 && timer->deadline < old_min_deadline) { - gpr_atm_no_barrier_store(&g_shared_mutables.min_timer, timer->deadline); + if (shard->shard_queue_index == 0 && deadline_atm < old_min_deadline) { + gpr_atm_no_barrier_store(&g_shared_mutables.min_timer, deadline_atm); grpc_kick_poller(); } } @@ -319,7 +318,7 @@ void grpc_timer_cancel(grpc_exec_ctx *exec_ctx, grpc_timer *timer) { shard_type *shard = &g_shards[GPR_HASH_POINTER(timer, NUM_SHARDS)]; gpr_mu_lock(&shard->mu); - if (grpc_timer_trace) { + if (GRPC_TRACER_ON(grpc_timer_trace)) { gpr_log(GPR_DEBUG, "TIMER %p: CANCEL pending=%s", timer, timer->pending ? "true" : "false"); } @@ -355,7 +354,7 @@ static int refill_queue(shard_type *shard, gpr_atm now) { saturating_add(GPR_MAX(now, shard->queue_deadline_cap), (gpr_atm)(deadline_delta * 1000.0)); - if (grpc_timer_check_trace) { + 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); } @@ -363,7 +362,7 @@ static int refill_queue(shard_type *shard, gpr_atm now) { next = timer->next; if (timer->deadline < shard->queue_deadline_cap) { - if (grpc_timer_check_trace) { + if (GRPC_TRACER_ON(grpc_timer_check_trace)) { gpr_log(GPR_DEBUG, " .. add timer with deadline %" PRIdPTR " to heap", timer->deadline); } @@ -380,7 +379,7 @@ static int refill_queue(shard_type *shard, gpr_atm now) { static grpc_timer *pop_one(shard_type *shard, gpr_atm now) { grpc_timer *timer; for (;;) { - if (grpc_timer_check_trace) { + 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"); @@ -390,13 +389,13 @@ static grpc_timer *pop_one(shard_type *shard, gpr_atm now) { if (!refill_queue(shard, now)) return NULL; } timer = grpc_timer_heap_top(&shard->heap); - if (grpc_timer_check_trace) { + 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) { + if (GRPC_TRACER_ON(grpc_timer_trace)) { gpr_log(GPR_DEBUG, "TIMER %p: FIRE %" PRIdPTR "ms late", timer, now - timer->deadline); } @@ -436,7 +435,7 @@ static int run_some_expired_timers(grpc_exec_ctx *exec_ctx, gpr_atm now, if (gpr_spinlock_trylock(&g_shared_mutables.checker_mu)) { gpr_mu_lock(&g_shared_mutables.mu); - if (grpc_timer_check_trace) { + 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); @@ -452,7 +451,7 @@ static int run_some_expired_timers(grpc_exec_ctx *exec_ctx, gpr_atm now, n += pop_timers(exec_ctx, g_shard_queue[0], now, &new_min_deadline, error); - if (grpc_timer_check_trace) { + if (GRPC_TRACER_ON(grpc_timer_check_trace)) { gpr_log(GPR_DEBUG, " .. popped --> %" PRIdPTR ", shard[%d]->min_deadline %" PRIdPTR " --> %" PRIdPTR ", now=%" PRIdPTR, @@ -509,7 +508,7 @@ bool grpc_timer_check(grpc_exec_ctx *exec_ctx, gpr_timespec now, *next = atm_to_timespec(GPR_MIN(timespec_to_atm_round_up(*next), min_timer)); } - if (grpc_timer_check_trace) { + if (GRPC_TRACER_ON(grpc_timer_check_trace)) { gpr_log(GPR_DEBUG, "TIMER CHECK SKIP: now_atm=%" PRIdPTR " min_timer=%" PRIdPTR, now_atm, min_timer); @@ -523,7 +522,7 @@ bool grpc_timer_check(grpc_exec_ctx *exec_ctx, gpr_timespec now, : GRPC_ERROR_CREATE_FROM_STATIC_STRING("Shutting down timer system"); // tracing - if (grpc_timer_check_trace) { + if (GRPC_TRACER_ON(grpc_timer_check_trace)) { char *next_str; if (next == NULL) { next_str = gpr_strdup("NULL"); @@ -549,7 +548,7 @@ bool grpc_timer_check(grpc_exec_ctx *exec_ctx, gpr_timespec now, *next = atm_to_timespec(next_atm); } // tracing - if (grpc_timer_check_trace) { + if (GRPC_TRACER_ON(grpc_timer_check_trace)) { char *next_str; if (next == NULL) { next_str = gpr_strdup("NULL"); diff --git a/src/core/lib/iomgr/timer_manager.c b/src/core/lib/iomgr/timer_manager.c index 00e868de01..1d83341299 100644 --- a/src/core/lib/iomgr/timer_manager.c +++ b/src/core/lib/iomgr/timer_manager.c @@ -45,11 +45,11 @@ typedef struct completed_thread { } completed_thread; static gpr_mu g_mu; +static bool g_threaded; static gpr_cv g_cv_wait; static gpr_cv g_cv_shutdown; static int g_thread_count; static int g_waiter_count; -static bool g_shutdown; static completed_thread *g_completed_threads; static bool g_kicked; @@ -83,6 +83,14 @@ static void start_timer_thread_and_unlock(void) { gpr_thd_new(&thd, timer_thread, NULL, &opt); } +void grpc_timer_manager_tick() { + grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT; + gpr_timespec next = gpr_inf_future(GPR_CLOCK_MONOTONIC); + gpr_timespec now = gpr_now(GPR_CLOCK_MONOTONIC); + grpc_timer_check(&exec_ctx, now, &next); + grpc_exec_ctx_finish(&exec_ctx); +} + static void timer_thread(void *unused) { grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INITIALIZER(0, grpc_never_ready_to_finish, NULL); @@ -93,7 +101,7 @@ static void timer_thread(void *unused) { gpr_mu_lock(&g_mu); --g_waiter_count; bool start_thread = g_waiter_count == 0; - if (start_thread && !g_shutdown) { + if (start_thread && g_threaded) { start_timer_thread_and_unlock(); } else { gpr_mu_unlock(&g_mu); @@ -105,7 +113,7 @@ static void timer_thread(void *unused) { gpr_mu_unlock(&g_mu); } else { gpr_mu_lock(&g_mu); - if (g_shutdown) break; + if (!g_threaded) break; if (gpr_cv_wait(&g_cv_wait, &g_mu, next)) { if (g_kicked) { grpc_timer_consume_kick(); @@ -130,34 +138,58 @@ static void timer_thread(void *unused) { gpr_log(GPR_DEBUG, "End timer thread"); } +static void start_threads(void) { + gpr_mu_lock(&g_mu); + if (!g_threaded) { + g_threaded = true; + start_timer_thread_and_unlock(); + } else { + g_threaded = false; + gpr_mu_unlock(&g_mu); + } +} + void grpc_timer_manager_init(void) { gpr_mu_init(&g_mu); gpr_cv_init(&g_cv_wait); gpr_cv_init(&g_cv_shutdown); + g_threaded = false; g_thread_count = 0; g_waiter_count = 0; - g_shutdown = false; g_completed_threads = NULL; - gpr_mu_lock(&g_mu); - start_timer_thread_and_unlock(); + start_threads(); } -void grpc_timer_manager_shutdown(void) { +static void stop_threads(void) { gpr_mu_lock(&g_mu); - g_shutdown = true; - gpr_cv_broadcast(&g_cv_wait); - while (g_thread_count > 0) { - gpr_cv_wait(&g_cv_shutdown, &g_mu, gpr_inf_future(GPR_CLOCK_REALTIME)); - gc_completed_threads(); + if (g_threaded) { + g_threaded = false; + gpr_cv_broadcast(&g_cv_wait); + while (g_thread_count > 0) { + gpr_cv_wait(&g_cv_shutdown, &g_mu, gpr_inf_future(GPR_CLOCK_REALTIME)); + gc_completed_threads(); + } } gpr_mu_unlock(&g_mu); +} + +void grpc_timer_manager_shutdown(void) { + stop_threads(); gpr_mu_destroy(&g_mu); gpr_cv_destroy(&g_cv_wait); gpr_cv_destroy(&g_cv_shutdown); } +void grpc_timer_manager_set_threading(bool threaded) { + if (threaded) { + start_threads(); + } else { + stop_threads(); + } +} + void grpc_kick_poller(void) { gpr_mu_lock(&g_mu); g_kicked = true; diff --git a/src/core/lib/iomgr/timer_manager.h b/src/core/lib/iomgr/timer_manager.h index a24c9da328..0b21262b1a 100644 --- a/src/core/lib/iomgr/timer_manager.h +++ b/src/core/lib/iomgr/timer_manager.h @@ -34,10 +34,19 @@ #ifndef GRPC_CORE_IOMGR_TIMER_MANAGER_H #define GRPC_CORE_IOMGR_TIMER_MANAGER_H +#include <stdbool.h> + /* Timer Manager tries to keep one thread waiting for the next timeout at all times */ void grpc_timer_manager_init(void); void grpc_timer_manager_shutdown(void); +/* enable/disable threading - must be called after grpc_timer_manager_init and + * before grpc_timer_manager_shutdown */ +void grpc_timer_manager_set_threading(bool enabled); +/* explicitly perform one tick of the timer system - for when threading is + * disabled */ +void grpc_timer_manager_tick(void); + #endif |