From 93b60e05a0a6d3eed569d0e2619aee8eae854c83 Mon Sep 17 00:00:00 2001 From: Sree Kuchibhotla Date: Fri, 7 Apr 2017 18:50:32 -0700 Subject: [PATCH 1/2] Fix asan bug --- .../microbenchmarks/bm_cq_multiple_threads.cc | 38 ++++++++++++++----- 1 file changed, 28 insertions(+), 10 deletions(-) diff --git a/test/cpp/microbenchmarks/bm_cq_multiple_threads.cc b/test/cpp/microbenchmarks/bm_cq_multiple_threads.cc index 967c226ac73..eccbd84d32b 100644 --- a/test/cpp/microbenchmarks/bm_cq_multiple_threads.cc +++ b/test/cpp/microbenchmarks/bm_cq_multiple_threads.cc @@ -51,11 +51,10 @@ struct grpc_pollset { namespace grpc { namespace testing { -static void* make_tag(int i) { return (void*)(intptr_t)i; } +static void* g_tag = (void*)(intptr_t)10; // Some random number static grpc_completion_queue* g_cq; static grpc_event_engine_vtable g_vtable; -static __thread int g_thread_idx; static __thread grpc_cq_completion g_cq_completion; static void pollset_shutdown(grpc_exec_ctx* exec_ctx, grpc_pollset* ps, @@ -83,8 +82,8 @@ static grpc_error* pollset_work(grpc_exec_ctx* exec_ctx, grpc_pollset* ps, grpc_pollset_worker** worker, gpr_timespec now, gpr_timespec deadline) { gpr_mu_unlock(&ps->mu); - grpc_cq_end_op(exec_ctx, g_cq, make_tag(g_thread_idx), GRPC_ERROR_NONE, - cq_done_cb, NULL, &g_cq_completion); + grpc_cq_end_op(exec_ctx, g_cq, g_tag, GRPC_ERROR_NONE, cq_done_cb, NULL, + &g_cq_completion); grpc_exec_ctx_flush(exec_ctx); gpr_mu_lock(&ps->mu); return GRPC_ERROR_NONE; @@ -109,26 +108,45 @@ static void setup() { g_cq = grpc_completion_queue_create(NULL); } +static void teardown() { + grpc_completion_queue_shutdown(g_cq); + grpc_completion_queue_destroy(g_cq); +} + +/* A few notes about Multi-theaded benchmarks: + + Setup: + The benchmark framework ensures that none of the threads proceed beyond the + state.KeepRunning() call unless all the threads have called state.keepRunning + atleast once. So it is safe to do the initialization in one of the threads + before state.KeepRunning() is called. + + Teardown: + The benchmark framework also ensures that no thread is running the benchmark + code (i.e the code between two successive calls of state.KeepRunning()) if + state.KeepRunning() returns false. So it is safe to do the teardown in one + of the threads after state.keepRunning() returns false. +*/ static void BM_Cq_Throughput(benchmark::State& state) { TrackCounters track_counters; - gpr_timespec deadline = gpr_inf_past(GPR_CLOCK_MONOTONIC); + gpr_timespec deadline = gpr_inf_future(GPR_CLOCK_MONOTONIC); if (state.thread_index == 0) { setup(); } while (state.KeepRunning()) { - g_thread_idx = state.thread_index; - void* dummy_tag = make_tag(g_thread_idx); - grpc_cq_begin_op(g_cq, dummy_tag); + grpc_cq_begin_op(g_cq, g_tag); + + /* Note that the tag dequeued by the following might have been enqueued + by another thread. */ grpc_completion_queue_next(g_cq, deadline, NULL); } state.SetItemsProcessed(state.iterations()); if (state.thread_index == 0) { - grpc_completion_queue_shutdown(g_cq); - grpc_completion_queue_destroy(g_cq); + teardown(); } track_counters.Finish(state); From 1f0c8271146b7fae637aa59a2429af137013fb8b Mon Sep 17 00:00:00 2001 From: Sree Kuchibhotla Date: Sat, 8 Apr 2017 16:25:58 -0700 Subject: [PATCH 2/2] Fix asan and tsan bugs. Simplify the code --- .../microbenchmarks/bm_cq_multiple_threads.cc | 20 +++++++++---------- 1 file changed, 10 insertions(+), 10 deletions(-) diff --git a/test/cpp/microbenchmarks/bm_cq_multiple_threads.cc b/test/cpp/microbenchmarks/bm_cq_multiple_threads.cc index eccbd84d32b..86274632042 100644 --- a/test/cpp/microbenchmarks/bm_cq_multiple_threads.cc +++ b/test/cpp/microbenchmarks/bm_cq_multiple_threads.cc @@ -36,6 +36,8 @@ #include #include +#include +#include #include "test/cpp/microbenchmarks/helpers.h" extern "C" { @@ -55,8 +57,6 @@ static void* g_tag = (void*)(intptr_t)10; // Some random number static grpc_completion_queue* g_cq; static grpc_event_engine_vtable g_vtable; -static __thread grpc_cq_completion g_cq_completion; - static void pollset_shutdown(grpc_exec_ctx* exec_ctx, grpc_pollset* ps, grpc_closure* closure) { grpc_closure_sched(exec_ctx, closure, GRPC_ERROR_NONE); @@ -75,15 +75,18 @@ static grpc_error* pollset_kick(grpc_pollset* p, grpc_pollset_worker* worker) { /* Callback when the tag is dequeued from the completion queue. Does nothing */ static void cq_done_cb(grpc_exec_ctx* exec_ctx, void* done_arg, - grpc_cq_completion* cq_completion) {} + grpc_cq_completion* cq_completion) { + gpr_free(cq_completion); +} /* Queues a completion tag. ZERO polling overhead */ static grpc_error* pollset_work(grpc_exec_ctx* exec_ctx, grpc_pollset* ps, grpc_pollset_worker** worker, gpr_timespec now, gpr_timespec deadline) { gpr_mu_unlock(&ps->mu); + grpc_cq_begin_op(g_cq, g_tag); grpc_cq_end_op(exec_ctx, g_cq, g_tag, GRPC_ERROR_NONE, cq_done_cb, NULL, - &g_cq_completion); + (grpc_cq_completion*)gpr_malloc(sizeof(grpc_cq_completion))); grpc_exec_ctx_flush(exec_ctx); gpr_mu_lock(&ps->mu); return GRPC_ERROR_NONE; @@ -113,7 +116,7 @@ static void teardown() { grpc_completion_queue_destroy(g_cq); } -/* A few notes about Multi-theaded benchmarks: +/* A few notes about Multi-threaded benchmarks: Setup: The benchmark framework ensures that none of the threads proceed beyond the @@ -136,11 +139,8 @@ static void BM_Cq_Throughput(benchmark::State& state) { } while (state.KeepRunning()) { - grpc_cq_begin_op(g_cq, g_tag); - - /* Note that the tag dequeued by the following might have been enqueued - by another thread. */ - grpc_completion_queue_next(g_cq, deadline, NULL); + GPR_ASSERT(grpc_completion_queue_next(g_cq, deadline, NULL).type == + GRPC_OP_COMPLETE); } state.SetItemsProcessed(state.iterations());