diff options
Diffstat (limited to 'src/core/lib')
84 files changed, 1027 insertions, 991 deletions
diff --git a/src/core/lib/channel/channel_stack.cc b/src/core/lib/channel/channel_stack.cc index 45d6fceb03..195fe0b195 100644 --- a/src/core/lib/channel/channel_stack.cc +++ b/src/core/lib/channel/channel_stack.cc @@ -23,7 +23,7 @@ #include <stdlib.h> #include <string.h> -grpc_tracer_flag grpc_trace_channel = GRPC_TRACER_INITIALIZER(false, "channel"); +grpc_core::TraceFlag grpc_trace_channel(false, "channel"); /* Memory layouts. diff --git a/src/core/lib/channel/channel_stack.h b/src/core/lib/channel/channel_stack.h index 4496b864dc..63382c05b2 100644 --- a/src/core/lib/channel/channel_stack.h +++ b/src/core/lib/channel/channel_stack.h @@ -84,6 +84,7 @@ typedef struct { typedef struct { grpc_call_stats stats; grpc_status_code final_status; + const char** error_string; } grpc_call_final_info; /* Channel filters specify: @@ -273,10 +274,10 @@ void grpc_call_log_op(const char* file, int line, gpr_log_severity severity, grpc_call_element* elem, grpc_transport_stream_op_batch* op); -extern grpc_tracer_flag grpc_trace_channel; +extern grpc_core::TraceFlag grpc_trace_channel; #define GRPC_CALL_LOG_OP(sev, elem, op) \ - if (GRPC_TRACER_ON(grpc_trace_channel)) grpc_call_log_op(sev, elem, op) + if (grpc_trace_channel.enabled()) grpc_call_log_op(sev, elem, op) #ifdef __cplusplus } diff --git a/src/core/lib/channel/channel_stack_builder.cc b/src/core/lib/channel/channel_stack_builder.cc index 9cef3501bc..fcba826644 100644 --- a/src/core/lib/channel/channel_stack_builder.cc +++ b/src/core/lib/channel/channel_stack_builder.cc @@ -23,8 +23,8 @@ #include <grpc/support/alloc.h> #include <grpc/support/string_util.h> -grpc_tracer_flag grpc_trace_channel_stack_builder = - GRPC_TRACER_INITIALIZER(false, "channel_stack_builder"); +grpc_core::TraceFlag grpc_trace_channel_stack_builder(false, + "channel_stack_builder"); typedef struct filter_node { struct filter_node* next; diff --git a/src/core/lib/channel/channel_stack_builder.h b/src/core/lib/channel/channel_stack_builder.h index f4e4a200d9..302083e8d4 100644 --- a/src/core/lib/channel/channel_stack_builder.h +++ b/src/core/lib/channel/channel_stack_builder.h @@ -157,7 +157,7 @@ grpc_error* grpc_channel_stack_builder_finish( /// Destroy the builder without creating a channel stack void grpc_channel_stack_builder_destroy(grpc_channel_stack_builder* builder); -extern grpc_tracer_flag grpc_trace_channel_stack_builder; +extern grpc_core::TraceFlag grpc_trace_channel_stack_builder; #ifdef __cplusplus } diff --git a/src/core/lib/debug/trace.cc b/src/core/lib/debug/trace.cc index 9c75ef124a..4c63983bdc 100644 --- a/src/core/lib/debug/trace.cc +++ b/src/core/lib/debug/trace.cc @@ -27,26 +27,61 @@ int grpc_tracer_set_enabled(const char* name, int enabled); -typedef struct tracer { - grpc_tracer_flag* flag; - struct tracer* next; -} tracer; -static tracer* tracers; - -#ifdef GRPC_THREADSAFE_TRACER -#define TRACER_SET(flag, on) gpr_atm_no_barrier_store(&(flag).value, (on)) -#else -#define TRACER_SET(flag, on) (flag).value = (on) -#endif - -void grpc_register_tracer(grpc_tracer_flag* flag) { - tracer* t = (tracer*)gpr_malloc(sizeof(*t)); - t->flag = flag; - t->next = tracers; - TRACER_SET(*flag, false); - tracers = t; +namespace grpc_core { + +TraceFlag* TraceFlagList::root_tracer_ = nullptr; + +bool TraceFlagList::Set(const char* name, bool enabled) { + TraceFlag* t; + if (0 == strcmp(name, "all")) { + for (t = root_tracer_; t; t = t->next_tracer_) { + t->set_enabled(enabled); + } + } else if (0 == strcmp(name, "list_tracers")) { + LogAllTracers(); + } else if (0 == strcmp(name, "refcount")) { + for (t = root_tracer_; t; t = t->next_tracer_) { + if (strstr(t->name_, "refcount") != nullptr) { + t->set_enabled(enabled); + } + } + } else { + bool found = false; + for (t = root_tracer_; t; t = t->next_tracer_) { + if (0 == strcmp(name, t->name_)) { + t->set_enabled(enabled); + found = true; + } + } + if (!found) { + gpr_log(GPR_ERROR, "Unknown trace var: '%s'", name); + return false; /* early return */ + } + } + return true; +} + +void TraceFlagList::Add(TraceFlag* flag) { + flag->next_tracer_ = root_tracer_; + root_tracer_ = flag; } +void TraceFlagList::LogAllTracers() { + gpr_log(GPR_DEBUG, "available tracers:"); + TraceFlag* t; + for (t = root_tracer_; t != nullptr; t = t->next_tracer_) { + gpr_log(GPR_DEBUG, "\t%s", t->name_); + } +} + +// Flags register themselves on the list during construction +TraceFlag::TraceFlag(bool default_enabled, const char* name) + : name_(name), value_(default_enabled) { + TraceFlagList::Add(this); +} + +} // namespace grpc_core + static void add(const char* beg, const char* end, char*** ss, size_t* ns) { size_t n = *ns; size_t np = n + 1; @@ -80,9 +115,9 @@ static void parse(const char* s) { for (i = 0; i < nstrings; i++) { if (strings[i][0] == '-') { - grpc_tracer_set_enabled(strings[i] + 1, 0); + grpc_core::TraceFlagList::Set(strings[i] + 1, false); } else { - grpc_tracer_set_enabled(strings[i], 1); + grpc_core::TraceFlagList::Set(strings[i], true); } } @@ -92,14 +127,6 @@ static void parse(const char* s) { gpr_free(strings); } -static void list_tracers() { - gpr_log(GPR_DEBUG, "available tracers:"); - tracer* t; - for (t = tracers; t; t = t->next) { - gpr_log(GPR_DEBUG, "\t%s", t->flag->name); - } -} - void grpc_tracer_init(const char* env_var) { char* e = gpr_getenv(env_var); if (e != nullptr) { @@ -108,40 +135,8 @@ void grpc_tracer_init(const char* env_var) { } } -void grpc_tracer_shutdown(void) { - while (tracers) { - tracer* t = tracers; - tracers = t->next; - gpr_free(t); - } -} +void grpc_tracer_shutdown(void) {} int grpc_tracer_set_enabled(const char* name, int enabled) { - tracer* t; - if (0 == strcmp(name, "all")) { - for (t = tracers; t; t = t->next) { - TRACER_SET(*t->flag, enabled); - } - } else if (0 == strcmp(name, "list_tracers")) { - list_tracers(); - } else if (0 == strcmp(name, "refcount")) { - for (t = tracers; t; t = t->next) { - if (strstr(t->flag->name, "refcount") != nullptr) { - TRACER_SET(*t->flag, enabled); - } - } - } else { - int found = 0; - for (t = tracers; t; t = t->next) { - if (0 == strcmp(name, t->flag->name)) { - TRACER_SET(*t->flag, enabled); - found = 1; - } - } - if (!found) { - gpr_log(GPR_ERROR, "Unknown trace var: '%s'", name); - return 0; /* early return */ - } - } - return 1; + return grpc_core::TraceFlagList::Set(name, enabled != 0); } diff --git a/src/core/lib/debug/trace.h b/src/core/lib/debug/trace.h index 7447d5d94a..b58c16f3ca 100644 --- a/src/core/lib/debug/trace.h +++ b/src/core/lib/debug/trace.h @@ -27,37 +27,89 @@ extern "C" { #endif +void grpc_tracer_init(const char* env_var_name); +void grpc_tracer_shutdown(void); + +#ifdef __cplusplus +} +#endif + #if defined(__has_feature) #if __has_feature(thread_sanitizer) #define GRPC_THREADSAFE_TRACER #endif #endif -typedef struct { +#ifdef __cplusplus + +namespace grpc_core { + +class TraceFlag; +class TraceFlagList { + public: + static bool Set(const char* name, bool enabled); + static void Add(TraceFlag* flag); + + private: + static void LogAllTracers(); + static TraceFlag* root_tracer_; +}; + +namespace testing { +void grpc_tracer_enable_flag(grpc_core::TraceFlag* flag); +} + +class TraceFlag { + public: + TraceFlag(bool default_enabled, const char* name); + ~TraceFlag() {} + + const char* name() const { return name_; } + + bool enabled() { +#ifdef GRPC_THREADSAFE_TRACER + return gpr_atm_no_barrier_load(&value_) != 0; +#else + return value_; +#endif + } + + private: + friend void grpc_core::testing::grpc_tracer_enable_flag(TraceFlag* flag); + friend class TraceFlagList; + + void set_enabled(bool enabled) { #ifdef GRPC_THREADSAFE_TRACER - gpr_atm value; + gpr_atm_no_barrier_store(&value_, enabled); #else - bool value; + value_ = enabled; #endif - const char* name; -} grpc_tracer_flag; + } + TraceFlag* next_tracer_; + const char* const name_; #ifdef GRPC_THREADSAFE_TRACER -#define GRPC_TRACER_ON(flag) (gpr_atm_no_barrier_load(&(flag).value) != 0) -#define GRPC_TRACER_INITIALIZER(on, name) \ - { (gpr_atm)(on), (name) } + gpr_atm value_; #else -#define GRPC_TRACER_ON(flag) ((flag).value) -#define GRPC_TRACER_INITIALIZER(on, name) \ - { (on), (name) } + bool value_; #endif +}; -void grpc_register_tracer(grpc_tracer_flag* flag); -void grpc_tracer_init(const char* env_var_name); -void grpc_tracer_shutdown(void); +#ifndef NDEBUG +typedef TraceFlag DebugOnlyTraceFlag; +#else +class DebugOnlyTraceFlag { + public: + DebugOnlyTraceFlag(bool default_enabled, const char* name) {} + bool enabled() { return false; } -#ifdef __cplusplus -} + private: + void set_enabled(bool enabled) {} +}; #endif +} // namespace grpc_core + +#endif // __cplusplus + #endif /* GRPC_CORE_LIB_DEBUG_TRACE_H */ diff --git a/src/core/lib/http/parser.cc b/src/core/lib/http/parser.cc index 9134b0ced2..fb4eb234a7 100644 --- a/src/core/lib/http/parser.cc +++ b/src/core/lib/http/parser.cc @@ -25,7 +25,7 @@ #include <grpc/support/log.h> #include <grpc/support/useful.h> -grpc_tracer_flag grpc_http1_trace = GRPC_TRACER_INITIALIZER(false, "http1"); +grpc_core::TraceFlag grpc_http1_trace(false, "http1"); static char* buf2str(void* buffer, size_t length) { char* out = (char*)gpr_malloc(length + 1); @@ -294,7 +294,7 @@ static grpc_error* addbyte(grpc_http_parser* parser, uint8_t byte, case GRPC_HTTP_FIRST_LINE: case GRPC_HTTP_HEADERS: if (parser->cur_line_length >= GRPC_HTTP_PARSER_MAX_HEADER_LENGTH) { - if (GRPC_TRACER_ON(grpc_http1_trace)) + if (grpc_http1_trace.enabled()) gpr_log(GPR_ERROR, "HTTP header max line length (%d) exceeded", GRPC_HTTP_PARSER_MAX_HEADER_LENGTH); return GRPC_ERROR_CREATE_FROM_STATIC_STRING( diff --git a/src/core/lib/http/parser.h b/src/core/lib/http/parser.h index 3d28481c4c..391bd350de 100644 --- a/src/core/lib/http/parser.h +++ b/src/core/lib/http/parser.h @@ -111,7 +111,7 @@ grpc_error* grpc_http_parser_eof(grpc_http_parser* parser); void grpc_http_request_destroy(grpc_http_request* request); void grpc_http_response_destroy(grpc_http_response* response); -extern grpc_tracer_flag grpc_http1_trace; +extern grpc_core::TraceFlag grpc_http1_trace; #ifdef __cplusplus } diff --git a/src/core/lib/iomgr/call_combiner.cc b/src/core/lib/iomgr/call_combiner.cc index 32b4244748..a9f48fb3c2 100644 --- a/src/core/lib/iomgr/call_combiner.cc +++ b/src/core/lib/iomgr/call_combiner.cc @@ -24,8 +24,7 @@ #include "src/core/lib/debug/stats.h" #include "src/core/lib/profiling/timers.h" -grpc_tracer_flag grpc_call_combiner_trace = - GRPC_TRACER_INITIALIZER(false, "call_combiner"); +grpc_core::TraceFlag grpc_call_combiner_trace(false, "call_combiner"); static grpc_error* decode_cancel_state_error(gpr_atm cancel_state) { if (cancel_state & 1) { @@ -62,7 +61,7 @@ void grpc_call_combiner_start(grpc_call_combiner* call_combiner, grpc_error* error DEBUG_ARGS, const char* reason) { GPR_TIMER_BEGIN("call_combiner_start", 0); - if (GRPC_TRACER_ON(grpc_call_combiner_trace)) { + if (grpc_call_combiner_trace.enabled()) { gpr_log(GPR_DEBUG, "==> grpc_call_combiner_start() [%p] closure=%p [" DEBUG_FMT_STR "%s] error=%s", @@ -71,7 +70,7 @@ void grpc_call_combiner_start(grpc_call_combiner* call_combiner, } size_t prev_size = (size_t)gpr_atm_full_fetch_add(&call_combiner->size, (gpr_atm)1); - if (GRPC_TRACER_ON(grpc_call_combiner_trace)) { + if (grpc_call_combiner_trace.enabled()) { gpr_log(GPR_DEBUG, " size: %" PRIdPTR " -> %" PRIdPTR, prev_size, prev_size + 1); } @@ -80,13 +79,13 @@ void grpc_call_combiner_start(grpc_call_combiner* call_combiner, GRPC_STATS_INC_CALL_COMBINER_LOCKS_INITIATED(); GPR_TIMER_MARK("call_combiner_initiate", 0); - if (GRPC_TRACER_ON(grpc_call_combiner_trace)) { + if (grpc_call_combiner_trace.enabled()) { gpr_log(GPR_DEBUG, " EXECUTING IMMEDIATELY"); } // Queue was empty, so execute this closure immediately. GRPC_CLOSURE_SCHED(closure, error); } else { - if (GRPC_TRACER_ON(grpc_call_combiner_trace)) { + if (grpc_call_combiner_trace.enabled()) { gpr_log(GPR_INFO, " QUEUING"); } // Queue was not empty, so add closure to queue. @@ -99,21 +98,21 @@ void grpc_call_combiner_start(grpc_call_combiner* call_combiner, void grpc_call_combiner_stop(grpc_call_combiner* call_combiner DEBUG_ARGS, const char* reason) { GPR_TIMER_BEGIN("call_combiner_stop", 0); - if (GRPC_TRACER_ON(grpc_call_combiner_trace)) { + if (grpc_call_combiner_trace.enabled()) { gpr_log(GPR_DEBUG, "==> grpc_call_combiner_stop() [%p] [" DEBUG_FMT_STR "%s]", call_combiner DEBUG_FMT_ARGS, reason); } size_t prev_size = (size_t)gpr_atm_full_fetch_add(&call_combiner->size, (gpr_atm)-1); - if (GRPC_TRACER_ON(grpc_call_combiner_trace)) { + if (grpc_call_combiner_trace.enabled()) { gpr_log(GPR_DEBUG, " size: %" PRIdPTR " -> %" PRIdPTR, prev_size, prev_size - 1); } GPR_ASSERT(prev_size >= 1); if (prev_size > 1) { while (true) { - if (GRPC_TRACER_ON(grpc_call_combiner_trace)) { + if (grpc_call_combiner_trace.enabled()) { gpr_log(GPR_DEBUG, " checking queue"); } bool empty; @@ -122,19 +121,19 @@ void grpc_call_combiner_stop(grpc_call_combiner* call_combiner DEBUG_ARGS, if (closure == nullptr) { // This can happen either due to a race condition within the mpscq // code or because of a race with grpc_call_combiner_start(). - if (GRPC_TRACER_ON(grpc_call_combiner_trace)) { + if (grpc_call_combiner_trace.enabled()) { gpr_log(GPR_DEBUG, " queue returned no result; checking again"); } continue; } - if (GRPC_TRACER_ON(grpc_call_combiner_trace)) { + if (grpc_call_combiner_trace.enabled()) { gpr_log(GPR_DEBUG, " EXECUTING FROM QUEUE: closure=%p error=%s", closure, grpc_error_string(closure->error_data.error)); } GRPC_CLOSURE_SCHED(closure, closure->error_data.error); break; } - } else if (GRPC_TRACER_ON(grpc_call_combiner_trace)) { + } else if (grpc_call_combiner_trace.enabled()) { gpr_log(GPR_DEBUG, " queue empty"); } GPR_TIMER_END("call_combiner_stop", 0); @@ -150,7 +149,7 @@ void grpc_call_combiner_set_notify_on_cancel(grpc_call_combiner* call_combiner, // If error is set, invoke the cancellation closure immediately. // Otherwise, store the new closure. if (original_error != GRPC_ERROR_NONE) { - if (GRPC_TRACER_ON(grpc_call_combiner_trace)) { + if (grpc_call_combiner_trace.enabled()) { gpr_log(GPR_DEBUG, "call_combiner=%p: scheduling notify_on_cancel callback=%p " "for pre-existing cancellation", @@ -161,7 +160,7 @@ void grpc_call_combiner_set_notify_on_cancel(grpc_call_combiner* call_combiner, } else { if (gpr_atm_full_cas(&call_combiner->cancel_state, original_state, (gpr_atm)closure)) { - if (GRPC_TRACER_ON(grpc_call_combiner_trace)) { + if (grpc_call_combiner_trace.enabled()) { gpr_log(GPR_DEBUG, "call_combiner=%p: setting notify_on_cancel=%p", call_combiner, closure); } @@ -170,7 +169,7 @@ void grpc_call_combiner_set_notify_on_cancel(grpc_call_combiner* call_combiner, // up any resources they may be holding for the callback. if (original_state != 0) { closure = (grpc_closure*)original_state; - if (GRPC_TRACER_ON(grpc_call_combiner_trace)) { + if (grpc_call_combiner_trace.enabled()) { gpr_log(GPR_DEBUG, "call_combiner=%p: scheduling old cancel callback=%p", call_combiner, closure); @@ -198,7 +197,7 @@ void grpc_call_combiner_cancel(grpc_call_combiner* call_combiner, encode_cancel_state_error(error))) { if (original_state != 0) { grpc_closure* notify_on_cancel = (grpc_closure*)original_state; - if (GRPC_TRACER_ON(grpc_call_combiner_trace)) { + if (grpc_call_combiner_trace.enabled()) { gpr_log(GPR_DEBUG, "call_combiner=%p: scheduling notify_on_cancel callback=%p", call_combiner, notify_on_cancel); diff --git a/src/core/lib/iomgr/call_combiner.h b/src/core/lib/iomgr/call_combiner.h index 364f77a167..cf928a13f4 100644 --- a/src/core/lib/iomgr/call_combiner.h +++ b/src/core/lib/iomgr/call_combiner.h @@ -40,7 +40,7 @@ extern "C" { // when it is done with the action that was kicked off by the original // callback. -extern grpc_tracer_flag grpc_call_combiner_trace; +extern grpc_core::TraceFlag grpc_call_combiner_trace; typedef struct { gpr_atm size; // size_t, num closures in queue or currently executing diff --git a/src/core/lib/iomgr/closure.cc b/src/core/lib/iomgr/closure.cc deleted file mode 100644 index 32766110cc..0000000000 --- a/src/core/lib/iomgr/closure.cc +++ /dev/null @@ -1,216 +0,0 @@ -/* - * - * Copyright 2015 gRPC authors. - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - * - */ - -#include "src/core/lib/iomgr/closure.h" - -#include <assert.h> -#include <grpc/support/alloc.h> -#include <grpc/support/log.h> - -#include "src/core/lib/profiling/timers.h" - -#ifndef NDEBUG -grpc_tracer_flag grpc_trace_closure = GRPC_TRACER_INITIALIZER(false, "closure"); -#endif - -#ifndef NDEBUG -grpc_closure* grpc_closure_init(const char* file, int line, - grpc_closure* closure, grpc_iomgr_cb_func cb, - void* cb_arg, - grpc_closure_scheduler* scheduler) { -#else -grpc_closure* grpc_closure_init(grpc_closure* closure, grpc_iomgr_cb_func cb, - void* cb_arg, - grpc_closure_scheduler* scheduler) { -#endif - closure->cb = cb; - closure->cb_arg = cb_arg; - closure->scheduler = scheduler; -#ifndef NDEBUG - closure->scheduled = false; - closure->file_initiated = nullptr; - closure->line_initiated = 0; - closure->run = false; - closure->file_created = file; - closure->line_created = line; -#endif - return closure; -} - -void grpc_closure_list_init(grpc_closure_list* closure_list) { - closure_list->head = closure_list->tail = nullptr; -} - -bool grpc_closure_list_append(grpc_closure_list* closure_list, - grpc_closure* closure, grpc_error* error) { - if (closure == nullptr) { - GRPC_ERROR_UNREF(error); - return false; - } - closure->error_data.error = error; - closure->next_data.next = nullptr; - bool was_empty = (closure_list->head == nullptr); - if (was_empty) { - closure_list->head = closure; - } else { - closure_list->tail->next_data.next = closure; - } - closure_list->tail = closure; - return was_empty; -} - -void grpc_closure_list_fail_all(grpc_closure_list* list, - grpc_error* forced_failure) { - for (grpc_closure* c = list->head; c != nullptr; c = c->next_data.next) { - if (c->error_data.error == GRPC_ERROR_NONE) { - c->error_data.error = GRPC_ERROR_REF(forced_failure); - } - } - GRPC_ERROR_UNREF(forced_failure); -} - -bool grpc_closure_list_empty(grpc_closure_list closure_list) { - return closure_list.head == nullptr; -} - -void grpc_closure_list_move(grpc_closure_list* src, grpc_closure_list* dst) { - if (src->head == nullptr) { - return; - } - if (dst->head == nullptr) { - *dst = *src; - } else { - dst->tail->next_data.next = src->head; - dst->tail = src->tail; - } - src->head = src->tail = nullptr; -} - -typedef struct { - grpc_iomgr_cb_func cb; - void* cb_arg; - grpc_closure wrapper; -} wrapped_closure; - -static void closure_wrapper(void* arg, grpc_error* error) { - wrapped_closure* wc = (wrapped_closure*)arg; - grpc_iomgr_cb_func cb = wc->cb; - void* cb_arg = wc->cb_arg; - gpr_free(wc); - cb(cb_arg, error); -} - -#ifndef NDEBUG -grpc_closure* grpc_closure_create(const char* file, int line, - grpc_iomgr_cb_func cb, void* cb_arg, - grpc_closure_scheduler* scheduler) { -#else -grpc_closure* grpc_closure_create(grpc_iomgr_cb_func cb, void* cb_arg, - grpc_closure_scheduler* scheduler) { -#endif - wrapped_closure* wc = (wrapped_closure*)gpr_malloc(sizeof(*wc)); - wc->cb = cb; - wc->cb_arg = cb_arg; -#ifndef NDEBUG - grpc_closure_init(file, line, &wc->wrapper, closure_wrapper, wc, scheduler); -#else - grpc_closure_init(&wc->wrapper, closure_wrapper, wc, scheduler); -#endif - return &wc->wrapper; -} - -#ifndef NDEBUG -void grpc_closure_run(const char* file, int line, grpc_closure* c, - grpc_error* error) { -#else -void grpc_closure_run(grpc_closure* c, grpc_error* error) { -#endif - GPR_TIMER_BEGIN("grpc_closure_run", 0); - if (c != nullptr) { -#ifndef NDEBUG - c->file_initiated = file; - c->line_initiated = line; - c->run = true; -#endif - assert(c->cb); - c->scheduler->vtable->run(c, error); - } else { - GRPC_ERROR_UNREF(error); - } - GPR_TIMER_END("grpc_closure_run", 0); -} - -#ifndef NDEBUG -void grpc_closure_sched(const char* file, int line, grpc_closure* c, - grpc_error* error) { -#else -void grpc_closure_sched(grpc_closure* c, grpc_error* error) { -#endif - GPR_TIMER_BEGIN("grpc_closure_sched", 0); - if (c != nullptr) { -#ifndef NDEBUG - if (c->scheduled) { - gpr_log(GPR_ERROR, - "Closure already scheduled. (closure: %p, created: [%s:%d], " - "previously scheduled at: [%s: %d] run?: %s", - c, c->file_created, c->line_created, c->file_initiated, - c->line_initiated, c->run ? "true" : "false"); - abort(); - } - c->scheduled = true; - c->file_initiated = file; - c->line_initiated = line; - c->run = false; -#endif - assert(c->cb); - c->scheduler->vtable->sched(c, error); - } else { - GRPC_ERROR_UNREF(error); - } - GPR_TIMER_END("grpc_closure_sched", 0); -} - -#ifndef NDEBUG -void grpc_closure_list_sched(const char* file, int line, - grpc_closure_list* list) { -#else -void grpc_closure_list_sched(grpc_closure_list* list) { -#endif - grpc_closure* c = list->head; - while (c != nullptr) { - grpc_closure* next = c->next_data.next; -#ifndef NDEBUG - if (c->scheduled) { - gpr_log(GPR_ERROR, - "Closure already scheduled. (closure: %p, created: [%s:%d], " - "previously scheduled at: [%s: %d] run?: %s", - c, c->file_created, c->line_created, c->file_initiated, - c->line_initiated, c->run ? "true" : "false"); - abort(); - } - c->scheduled = true; - c->file_initiated = file; - c->line_initiated = line; - c->run = false; -#endif - assert(c->cb); - c->scheduler->vtable->sched(c, c->error_data.error); - c = next; - } - list->head = list->tail = nullptr; -} diff --git a/src/core/lib/iomgr/closure.h b/src/core/lib/iomgr/closure.h index 63843ad76d..88af76006a 100644 --- a/src/core/lib/iomgr/closure.h +++ b/src/core/lib/iomgr/closure.h @@ -21,21 +21,19 @@ #include <grpc/support/port_platform.h> +#include <assert.h> #include <grpc/impl/codegen/exec_ctx_fwd.h> +#include <grpc/support/alloc.h> +#include <grpc/support/log.h> #include <stdbool.h> #include "src/core/lib/iomgr/error.h" +#include "src/core/lib/profiling/timers.h" #include "src/core/lib/support/mpscq.h" -#ifdef __cplusplus -extern "C" { -#endif - struct grpc_closure; typedef struct grpc_closure grpc_closure; -#ifndef NDEBUG -extern grpc_tracer_flag grpc_trace_closure; -#endif +extern grpc_core::DebugOnlyTraceFlag grpc_trace_closure; typedef struct grpc_closure_list { grpc_closure* head; @@ -82,8 +80,8 @@ struct grpc_closure { /** Arguments to be passed to "cb". */ void* cb_arg; - /** Scheduler to schedule against: NULL to schedule against current execution - context */ + /** Scheduler to schedule against: nullptr to schedule against current + execution context */ grpc_closure_scheduler* scheduler; /** Once queued, the result of the closure. Before then: scratch space */ @@ -104,96 +102,253 @@ struct grpc_closure { #endif }; +#ifndef NDEBUG +inline grpc_closure* grpc_closure_init(const char* file, int line, + grpc_closure* closure, + grpc_iomgr_cb_func cb, void* cb_arg, + grpc_closure_scheduler* scheduler) { +#else +inline grpc_closure* grpc_closure_init(grpc_closure* closure, + grpc_iomgr_cb_func cb, void* cb_arg, + grpc_closure_scheduler* scheduler) { +#endif + closure->cb = cb; + closure->cb_arg = cb_arg; + closure->scheduler = scheduler; +#ifndef NDEBUG + closure->scheduled = false; + closure->file_initiated = nullptr; + closure->line_initiated = 0; + closure->run = false; + closure->file_created = file; + closure->line_created = line; +#endif + return closure; +} + /** Initializes \a closure with \a cb and \a cb_arg. Returns \a closure. */ #ifndef NDEBUG -grpc_closure* grpc_closure_init(const char* file, int line, - grpc_closure* closure, grpc_iomgr_cb_func cb, - void* cb_arg, - grpc_closure_scheduler* scheduler); #define GRPC_CLOSURE_INIT(closure, cb, cb_arg, scheduler) \ grpc_closure_init(__FILE__, __LINE__, closure, cb, cb_arg, scheduler) #else -grpc_closure* grpc_closure_init(grpc_closure* closure, grpc_iomgr_cb_func cb, - void* cb_arg, - grpc_closure_scheduler* scheduler); #define GRPC_CLOSURE_INIT(closure, cb, cb_arg, scheduler) \ grpc_closure_init(closure, cb, cb_arg, scheduler) #endif +namespace closure_impl { + +typedef struct { + grpc_iomgr_cb_func cb; + void* cb_arg; + grpc_closure wrapper; +} wrapped_closure; + +inline void closure_wrapper(void* arg, grpc_error* error) { + wrapped_closure* wc = (wrapped_closure*)arg; + grpc_iomgr_cb_func cb = wc->cb; + void* cb_arg = wc->cb_arg; + gpr_free(wc); + cb(cb_arg, error); +} + +} // namespace closure_impl + +#ifndef NDEBUG +inline grpc_closure* grpc_closure_create(const char* file, int line, + grpc_iomgr_cb_func cb, void* cb_arg, + grpc_closure_scheduler* scheduler) { +#else +inline grpc_closure* grpc_closure_create(grpc_iomgr_cb_func cb, void* cb_arg, + grpc_closure_scheduler* scheduler) { +#endif + closure_impl::wrapped_closure* wc = + (closure_impl::wrapped_closure*)gpr_malloc(sizeof(*wc)); + wc->cb = cb; + wc->cb_arg = cb_arg; +#ifndef NDEBUG + grpc_closure_init(file, line, &wc->wrapper, closure_impl::closure_wrapper, wc, + scheduler); +#else + grpc_closure_init(&wc->wrapper, closure_impl::closure_wrapper, wc, scheduler); +#endif + return &wc->wrapper; +} + /* Create a heap allocated closure: try to avoid except for very rare events */ #ifndef NDEBUG -grpc_closure* grpc_closure_create(const char* file, int line, - grpc_iomgr_cb_func cb, void* cb_arg, - grpc_closure_scheduler* scheduler); #define GRPC_CLOSURE_CREATE(cb, cb_arg, scheduler) \ grpc_closure_create(__FILE__, __LINE__, cb, cb_arg, scheduler) #else -grpc_closure* grpc_closure_create(grpc_iomgr_cb_func cb, void* cb_arg, - grpc_closure_scheduler* scheduler); #define GRPC_CLOSURE_CREATE(cb, cb_arg, scheduler) \ grpc_closure_create(cb, cb_arg, scheduler) #endif #define GRPC_CLOSURE_LIST_INIT \ - { NULL, NULL } + { nullptr, nullptr } -void grpc_closure_list_init(grpc_closure_list* list); +inline void grpc_closure_list_init(grpc_closure_list* closure_list) { + closure_list->head = closure_list->tail = nullptr; +} /** add \a closure to the end of \a list and set \a closure's result to \a error Returns true if \a list becomes non-empty */ -bool grpc_closure_list_append(grpc_closure_list* list, grpc_closure* closure, - grpc_error* error); +inline bool grpc_closure_list_append(grpc_closure_list* closure_list, + grpc_closure* closure, grpc_error* error) { + if (closure == nullptr) { + GRPC_ERROR_UNREF(error); + return false; + } + closure->error_data.error = error; + closure->next_data.next = nullptr; + bool was_empty = (closure_list->head == nullptr); + if (was_empty) { + closure_list->head = closure; + } else { + closure_list->tail->next_data.next = closure; + } + closure_list->tail = closure; + return was_empty; +} /** force all success bits in \a list to false */ -void grpc_closure_list_fail_all(grpc_closure_list* list, - grpc_error* forced_failure); +inline void grpc_closure_list_fail_all(grpc_closure_list* list, + grpc_error* forced_failure) { + for (grpc_closure* c = list->head; c != nullptr; c = c->next_data.next) { + if (c->error_data.error == GRPC_ERROR_NONE) { + c->error_data.error = GRPC_ERROR_REF(forced_failure); + } + } + GRPC_ERROR_UNREF(forced_failure); +} /** append all closures from \a src to \a dst and empty \a src. */ -void grpc_closure_list_move(grpc_closure_list* src, grpc_closure_list* dst); +inline void grpc_closure_list_move(grpc_closure_list* src, + grpc_closure_list* dst) { + if (src->head == nullptr) { + return; + } + if (dst->head == nullptr) { + *dst = *src; + } else { + dst->tail->next_data.next = src->head; + dst->tail = src->tail; + } + src->head = src->tail = nullptr; +} /** return whether \a list is empty. */ -bool grpc_closure_list_empty(grpc_closure_list list); +inline bool grpc_closure_list_empty(grpc_closure_list closure_list) { + return closure_list.head == nullptr; +} + +#ifndef NDEBUG +inline void grpc_closure_run(const char* file, int line, grpc_closure* c, + grpc_error* error) { +#else +inline void grpc_closure_run(grpc_closure* c, grpc_error* error) { +#endif + GPR_TIMER_BEGIN("grpc_closure_run", 0); + if (c != nullptr) { +#ifndef NDEBUG + c->file_initiated = file; + c->line_initiated = line; + c->run = true; +#endif + assert(c->cb); + c->scheduler->vtable->run(c, error); + } else { + GRPC_ERROR_UNREF(error); + } + GPR_TIMER_END("grpc_closure_run", 0); +} /** Run a closure directly. Caller ensures that no locks are being held above. * Note that calling this at the end of a closure callback function itself is * by definition safe. */ #ifndef NDEBUG -void grpc_closure_run(const char* file, int line, grpc_closure* closure, - grpc_error* error); #define GRPC_CLOSURE_RUN(closure, error) \ grpc_closure_run(__FILE__, __LINE__, closure, error) #else -void grpc_closure_run(grpc_closure* closure, grpc_error* error); #define GRPC_CLOSURE_RUN(closure, error) grpc_closure_run(closure, error) #endif +#ifndef NDEBUG +inline void grpc_closure_sched(const char* file, int line, grpc_closure* c, + grpc_error* error) { +#else +inline void grpc_closure_sched(grpc_closure* c, grpc_error* error) { +#endif + GPR_TIMER_BEGIN("grpc_closure_sched", 0); + if (c != nullptr) { +#ifndef NDEBUG + if (c->scheduled) { + gpr_log(GPR_ERROR, + "Closure already scheduled. (closure: %p, created: [%s:%d], " + "previously scheduled at: [%s: %d] run?: %s", + c, c->file_created, c->line_created, c->file_initiated, + c->line_initiated, c->run ? "true" : "false"); + abort(); + } + c->scheduled = true; + c->file_initiated = file; + c->line_initiated = line; + c->run = false; +#endif + assert(c->cb); + c->scheduler->vtable->sched(c, error); + } else { + GRPC_ERROR_UNREF(error); + } + GPR_TIMER_END("grpc_closure_sched", 0); +} + /** Schedule a closure to be run. Does not need to be run from a safe point. */ #ifndef NDEBUG -void grpc_closure_sched(const char* file, int line, grpc_closure* closure, - grpc_error* error); #define GRPC_CLOSURE_SCHED(closure, error) \ grpc_closure_sched(__FILE__, __LINE__, closure, error) #else -void grpc_closure_sched(grpc_closure* closure, grpc_error* error); #define GRPC_CLOSURE_SCHED(closure, error) grpc_closure_sched(closure, error) #endif +#ifndef NDEBUG +inline void grpc_closure_list_sched(const char* file, int line, + grpc_closure_list* list) { +#else +inline void grpc_closure_list_sched(grpc_closure_list* list) { +#endif + grpc_closure* c = list->head; + while (c != nullptr) { + grpc_closure* next = c->next_data.next; +#ifndef NDEBUG + if (c->scheduled) { + gpr_log(GPR_ERROR, + "Closure already scheduled. (closure: %p, created: [%s:%d], " + "previously scheduled at: [%s: %d] run?: %s", + c, c->file_created, c->line_created, c->file_initiated, + c->line_initiated, c->run ? "true" : "false"); + abort(); + } + c->scheduled = true; + c->file_initiated = file; + c->line_initiated = line; + c->run = false; +#endif + assert(c->cb); + c->scheduler->vtable->sched(c, c->error_data.error); + c = next; + } + list->head = list->tail = nullptr; +} + /** Schedule all closures in a list to be run. Does not need to be run from a * safe point. */ #ifndef NDEBUG -void grpc_closure_list_sched(const char* file, int line, - grpc_closure_list* closure_list); #define GRPC_CLOSURE_LIST_SCHED(closure_list) \ grpc_closure_list_sched(__FILE__, __LINE__, closure_list) #else -void grpc_closure_list_sched(grpc_closure_list* closure_list); #define GRPC_CLOSURE_LIST_SCHED(closure_list) \ grpc_closure_list_sched(closure_list) #endif -#ifdef __cplusplus -} -#endif - #endif /* GRPC_CORE_LIB_IOMGR_CLOSURE_H */ diff --git a/src/core/lib/iomgr/combiner.cc b/src/core/lib/iomgr/combiner.cc index d2b7cf4b10..e4d7a6abd8 100644 --- a/src/core/lib/iomgr/combiner.cc +++ b/src/core/lib/iomgr/combiner.cc @@ -29,14 +29,13 @@ #include "src/core/lib/iomgr/executor.h" #include "src/core/lib/profiling/timers.h" -grpc_tracer_flag grpc_combiner_trace = - GRPC_TRACER_INITIALIZER(false, "combiner"); - -#define GRPC_COMBINER_TRACE(fn) \ - do { \ - if (GRPC_TRACER_ON(grpc_combiner_trace)) { \ - fn; \ - } \ +grpc_core::TraceFlag grpc_combiner_trace(false, "combiner"); + +#define GRPC_COMBINER_TRACE(fn) \ + do { \ + if (grpc_combiner_trace.enabled()) { \ + fn; \ + } \ } while (0) #define STATE_UNORPHANED 1 @@ -104,7 +103,7 @@ static void start_destroy(grpc_combiner* lock) { #ifndef NDEBUG #define GRPC_COMBINER_DEBUG_SPAM(op, delta) \ - if (GRPC_TRACER_ON(grpc_combiner_trace)) { \ + if (grpc_combiner_trace.enabled()) { \ gpr_log(file, line, GPR_LOG_SEVERITY_DEBUG, \ "C:%p %s %" PRIdPTR " --> %" PRIdPTR " %s", lock, (op), \ gpr_atm_no_barrier_load(&lock->refs.count), \ diff --git a/src/core/lib/iomgr/combiner.h b/src/core/lib/iomgr/combiner.h index bb450db2fc..8555fa979f 100644 --- a/src/core/lib/iomgr/combiner.h +++ b/src/core/lib/iomgr/combiner.h @@ -63,7 +63,7 @@ grpc_closure_scheduler* grpc_combiner_finally_scheduler(grpc_combiner* lock); bool grpc_combiner_continue_exec_ctx(); -extern grpc_tracer_flag grpc_combiner_trace; +extern grpc_core::TraceFlag grpc_combiner_trace; #ifdef __cplusplus } diff --git a/src/core/lib/iomgr/error.cc b/src/core/lib/iomgr/error.cc index 2364b1a369..42cd7c455d 100644 --- a/src/core/lib/iomgr/error.cc +++ b/src/core/lib/iomgr/error.cc @@ -37,10 +37,9 @@ #include "src/core/lib/profiling/timers.h" #include "src/core/lib/slice/slice_internal.h" -#ifndef NDEBUG -grpc_tracer_flag grpc_trace_error_refcount = - GRPC_TRACER_INITIALIZER(false, "error_refcount"); -#endif +grpc_core::DebugOnlyTraceFlag grpc_trace_error_refcount(false, + "error_refcount"); +grpc_core::DebugOnlyTraceFlag grpc_trace_closure(false, "closure"); static const char* error_int_name(grpc_error_ints key) { switch (key) { @@ -130,7 +129,7 @@ bool grpc_error_is_special(grpc_error* err) { #ifndef NDEBUG grpc_error* grpc_error_ref(grpc_error* err, const char* file, int line) { if (grpc_error_is_special(err)) return err; - if (GRPC_TRACER_ON(grpc_trace_error_refcount)) { + if (grpc_trace_error_refcount.enabled()) { gpr_log(GPR_DEBUG, "%p: %" PRIdPTR " -> %" PRIdPTR " [%s:%d]", err, gpr_atm_no_barrier_load(&err->atomics.refs.count), gpr_atm_no_barrier_load(&err->atomics.refs.count) + 1, file, line); @@ -179,7 +178,7 @@ static void error_destroy(grpc_error* err) { #ifndef NDEBUG void grpc_error_unref(grpc_error* err, const char* file, int line) { if (grpc_error_is_special(err)) return; - if (GRPC_TRACER_ON(grpc_trace_error_refcount)) { + if (grpc_trace_error_refcount.enabled()) { gpr_log(GPR_DEBUG, "%p: %" PRIdPTR " -> %" PRIdPTR " [%s:%d]", err, gpr_atm_no_barrier_load(&err->atomics.refs.count), gpr_atm_no_barrier_load(&err->atomics.refs.count) - 1, file, line); @@ -212,7 +211,7 @@ static uint8_t get_placement(grpc_error** err, size_t size) { *err = (grpc_error*)gpr_realloc( *err, sizeof(grpc_error) + (*err)->arena_capacity * sizeof(intptr_t)); #ifndef NDEBUG - if (GRPC_TRACER_ON(grpc_trace_error_refcount)) { + if (grpc_trace_error_refcount.enabled()) { if (*err != orig) { gpr_log(GPR_DEBUG, "realloc %p -> %p", orig, *err); } @@ -325,7 +324,7 @@ grpc_error* grpc_error_create(const char* file, int line, grpc_slice desc, return GRPC_ERROR_OOM; } #ifndef NDEBUG - if (GRPC_TRACER_ON(grpc_trace_error_refcount)) { + if (grpc_trace_error_refcount.enabled()) { gpr_log(GPR_DEBUG, "%p create [%s:%d]", err, file, line); } #endif @@ -407,7 +406,7 @@ static grpc_error* copy_error_and_unref(grpc_error* in) { out = (grpc_error*)gpr_malloc(sizeof(*in) + new_arena_capacity * sizeof(intptr_t)); #ifndef NDEBUG - if (GRPC_TRACER_ON(grpc_trace_error_refcount)) { + if (grpc_trace_error_refcount.enabled()) { gpr_log(GPR_DEBUG, "%p create copying %p", out, in); } #endif diff --git a/src/core/lib/iomgr/error.h b/src/core/lib/iomgr/error.h index 8d7aea4872..d10bf0b359 100644 --- a/src/core/lib/iomgr/error.h +++ b/src/core/lib/iomgr/error.h @@ -39,9 +39,7 @@ extern "C" { typedef struct grpc_error grpc_error; -#ifndef NDEBUG -extern grpc_tracer_flag grpc_trace_error_refcount; -#endif +extern grpc_core::DebugOnlyTraceFlag grpc_trace_error_refcount; typedef enum { /// 'errno' from the operating system diff --git a/src/core/lib/iomgr/ev_epoll1_linux.cc b/src/core/lib/iomgr/ev_epoll1_linux.cc index 1598657afe..d9e8a30f5e 100644 --- a/src/core/lib/iomgr/ev_epoll1_linux.cc +++ b/src/core/lib/iomgr/ev_epoll1_linux.cc @@ -263,12 +263,14 @@ static grpc_fd* fd_create(int fd, const char* name) { if (new_fd == nullptr) { new_fd = (grpc_fd*)gpr_malloc(sizeof(grpc_fd)); + new_fd->read_closure.Init(); + new_fd->write_closure.Init(); } new_fd->fd = fd; - new_fd->read_closure.Init(); - new_fd->write_closure.Init(); - gpr_atm_no_barrier_store(&new_fd->read_notifier_pollset, (gpr_atm) nullptr); + new_fd->read_closure->InitEvent(); + new_fd->write_closure->InitEvent(); + gpr_atm_no_barrier_store(&new_fd->read_notifier_pollset, (gpr_atm)NULL); new_fd->freelist_next = nullptr; @@ -276,7 +278,7 @@ static grpc_fd* fd_create(int fd, const char* name) { gpr_asprintf(&fd_name, "%s fd=%d", name, fd); grpc_iomgr_register_object(&new_fd->iomgr_object, fd_name); #ifndef NDEBUG - if (GRPC_TRACER_ON(grpc_trace_fd_refcount)) { + if (grpc_trace_fd_refcount.enabled()) { gpr_log(GPR_DEBUG, "FD %d %p create %s", fd, new_fd, fd_name); } #endif @@ -334,8 +336,8 @@ static void fd_orphan(grpc_fd* fd, grpc_closure* on_done, int* release_fd, GRPC_CLOSURE_SCHED(on_done, GRPC_ERROR_REF(error)); grpc_iomgr_unregister_object(&fd->iomgr_object); - fd->read_closure.Destroy(); - fd->write_closure.Destroy(); + fd->read_closure->DestroyEvent(); + fd->write_closure->DestroyEvent(); gpr_mu_lock(&fd_freelist_mu); fd->freelist_next = fd_freelist; @@ -637,7 +639,7 @@ static grpc_error* do_epoll_wait(grpc_pollset* ps, grpc_millis deadline) { GRPC_STATS_INC_POLL_EVENTS_RETURNED(r); - if (GRPC_TRACER_ON(grpc_polling_trace)) { + if (grpc_polling_trace.enabled()) { gpr_log(GPR_DEBUG, "ps: %p poll got %d events", ps, r); } @@ -658,7 +660,7 @@ static bool begin_worker(grpc_pollset* pollset, grpc_pollset_worker* worker, worker->schedule_on_end_work = (grpc_closure_list)GRPC_CLOSURE_LIST_INIT; pollset->begin_refs++; - if (GRPC_TRACER_ON(grpc_polling_trace)) { + if (grpc_polling_trace.enabled()) { gpr_log(GPR_ERROR, "PS:%p BEGIN_STARTS:%p", pollset, worker); } @@ -677,7 +679,7 @@ static bool begin_worker(grpc_pollset* pollset, grpc_pollset_worker* worker, retry_lock_neighborhood: gpr_mu_lock(&neighborhood->mu); gpr_mu_lock(&pollset->mu); - if (GRPC_TRACER_ON(grpc_polling_trace)) { + if (grpc_polling_trace.enabled()) { gpr_log(GPR_ERROR, "PS:%p BEGIN_REORG:%p kick_state=%s is_reassigning=%d", pollset, worker, kick_state_string(worker->state), is_reassigning); @@ -729,7 +731,7 @@ static bool begin_worker(grpc_pollset* pollset, grpc_pollset_worker* worker, worker->initialized_cv = true; gpr_cv_init(&worker->cv); while (worker->state == UNKICKED && !pollset->shutting_down) { - if (GRPC_TRACER_ON(grpc_polling_trace)) { + if (grpc_polling_trace.enabled()) { gpr_log(GPR_ERROR, "PS:%p BEGIN_WAIT:%p kick_state=%s shutdown=%d", pollset, worker, kick_state_string(worker->state), pollset->shutting_down); @@ -746,7 +748,7 @@ static bool begin_worker(grpc_pollset* pollset, grpc_pollset_worker* worker, grpc_core::ExecCtx::Get()->InvalidateNow(); } - if (GRPC_TRACER_ON(grpc_polling_trace)) { + if (grpc_polling_trace.enabled()) { gpr_log(GPR_ERROR, "PS:%p BEGIN_DONE:%p kick_state=%s shutdown=%d " "kicked_without_poller: %d", @@ -791,7 +793,7 @@ static bool check_neighborhood_for_available_poller( case UNKICKED: if (gpr_atm_no_barrier_cas(&g_active_poller, 0, (gpr_atm)inspect_worker)) { - if (GRPC_TRACER_ON(grpc_polling_trace)) { + if (grpc_polling_trace.enabled()) { gpr_log(GPR_DEBUG, " .. choose next poller to be %p", inspect_worker); } @@ -802,7 +804,7 @@ static bool check_neighborhood_for_available_poller( gpr_cv_signal(&inspect_worker->cv); } } else { - if (GRPC_TRACER_ON(grpc_polling_trace)) { + if (grpc_polling_trace.enabled()) { gpr_log(GPR_DEBUG, " .. beaten to choose next poller"); } } @@ -820,7 +822,7 @@ static bool check_neighborhood_for_available_poller( } while (!found_worker && inspect_worker != inspect->root_worker); } if (!found_worker) { - if (GRPC_TRACER_ON(grpc_polling_trace)) { + if (grpc_polling_trace.enabled()) { gpr_log(GPR_DEBUG, " .. mark pollset %p inactive", inspect); } inspect->seen_inactive = true; @@ -841,7 +843,7 @@ static bool check_neighborhood_for_available_poller( static void end_worker(grpc_pollset* pollset, grpc_pollset_worker* worker, grpc_pollset_worker** worker_hdl) { GPR_TIMER_BEGIN("end_worker", 0); - if (GRPC_TRACER_ON(grpc_polling_trace)) { + if (grpc_polling_trace.enabled()) { gpr_log(GPR_DEBUG, "PS:%p END_WORKER:%p", pollset, worker); } if (worker_hdl != nullptr) *worker_hdl = nullptr; @@ -851,7 +853,7 @@ static void end_worker(grpc_pollset* pollset, grpc_pollset_worker* worker, grpc_core::ExecCtx::Get()->closure_list()); if (gpr_atm_no_barrier_load(&g_active_poller) == (gpr_atm)worker) { if (worker->next != worker && worker->next->state == UNKICKED) { - if (GRPC_TRACER_ON(grpc_polling_trace)) { + if (grpc_polling_trace.enabled()) { gpr_log(GPR_DEBUG, " .. choose next poller to be peer %p", worker); } GPR_ASSERT(worker->next->initialized_cv); @@ -903,7 +905,7 @@ static void end_worker(grpc_pollset* pollset, grpc_pollset_worker* worker, if (worker->initialized_cv) { gpr_cv_destroy(&worker->cv); } - if (GRPC_TRACER_ON(grpc_polling_trace)) { + if (grpc_polling_trace.enabled()) { gpr_log(GPR_DEBUG, " .. remove worker"); } if (EMPTIED == worker_remove(pollset, worker)) { @@ -975,7 +977,7 @@ static grpc_error* pollset_kick(grpc_pollset* pollset, GPR_TIMER_BEGIN("pollset_kick", 0); GRPC_STATS_INC_POLLSET_KICK(); grpc_error* ret_err = GRPC_ERROR_NONE; - if (GRPC_TRACER_ON(grpc_polling_trace)) { + if (grpc_polling_trace.enabled()) { gpr_strvec log; gpr_strvec_init(&log); char* tmp; @@ -1008,7 +1010,7 @@ static grpc_error* pollset_kick(grpc_pollset* pollset, if (root_worker == nullptr) { GRPC_STATS_INC_POLLSET_KICKED_WITHOUT_POLLER(); pollset->kicked_without_poller = true; - if (GRPC_TRACER_ON(grpc_polling_trace)) { + if (grpc_polling_trace.enabled()) { gpr_log(GPR_ERROR, " .. kicked_without_poller"); } goto done; @@ -1016,14 +1018,14 @@ static grpc_error* pollset_kick(grpc_pollset* pollset, grpc_pollset_worker* next_worker = root_worker->next; if (root_worker->state == KICKED) { GRPC_STATS_INC_POLLSET_KICKED_AGAIN(); - if (GRPC_TRACER_ON(grpc_polling_trace)) { + if (grpc_polling_trace.enabled()) { gpr_log(GPR_ERROR, " .. already kicked %p", root_worker); } SET_KICK_STATE(root_worker, KICKED); goto done; } else if (next_worker->state == KICKED) { GRPC_STATS_INC_POLLSET_KICKED_AGAIN(); - if (GRPC_TRACER_ON(grpc_polling_trace)) { + if (grpc_polling_trace.enabled()) { gpr_log(GPR_ERROR, " .. already kicked %p", next_worker); } SET_KICK_STATE(next_worker, KICKED); @@ -1034,7 +1036,7 @@ static grpc_error* pollset_kick(grpc_pollset* pollset, root_worker == (grpc_pollset_worker*)gpr_atm_no_barrier_load( &g_active_poller)) { GRPC_STATS_INC_POLLSET_KICK_WAKEUP_FD(); - if (GRPC_TRACER_ON(grpc_polling_trace)) { + if (grpc_polling_trace.enabled()) { gpr_log(GPR_ERROR, " .. kicked %p", root_worker); } SET_KICK_STATE(root_worker, KICKED); @@ -1042,7 +1044,7 @@ static grpc_error* pollset_kick(grpc_pollset* pollset, goto done; } else if (next_worker->state == UNKICKED) { GRPC_STATS_INC_POLLSET_KICK_WAKEUP_CV(); - if (GRPC_TRACER_ON(grpc_polling_trace)) { + if (grpc_polling_trace.enabled()) { gpr_log(GPR_ERROR, " .. kicked %p", next_worker); } GPR_ASSERT(next_worker->initialized_cv); @@ -1051,7 +1053,7 @@ static grpc_error* pollset_kick(grpc_pollset* pollset, goto done; } else if (next_worker->state == DESIGNATED_POLLER) { if (root_worker->state != DESIGNATED_POLLER) { - if (GRPC_TRACER_ON(grpc_polling_trace)) { + if (grpc_polling_trace.enabled()) { gpr_log( GPR_ERROR, " .. kicked root non-poller %p (initialized_cv=%d) (poller=%p)", @@ -1065,7 +1067,7 @@ static grpc_error* pollset_kick(grpc_pollset* pollset, goto done; } else { GRPC_STATS_INC_POLLSET_KICK_WAKEUP_FD(); - if (GRPC_TRACER_ON(grpc_polling_trace)) { + if (grpc_polling_trace.enabled()) { gpr_log(GPR_ERROR, " .. non-root poller %p (root=%p)", next_worker, root_worker); } @@ -1081,7 +1083,7 @@ static grpc_error* pollset_kick(grpc_pollset* pollset, } } else { GRPC_STATS_INC_POLLSET_KICK_OWN_THREAD(); - if (GRPC_TRACER_ON(grpc_polling_trace)) { + if (grpc_polling_trace.enabled()) { gpr_log(GPR_ERROR, " .. kicked while waking up"); } goto done; @@ -1091,14 +1093,14 @@ static grpc_error* pollset_kick(grpc_pollset* pollset, } if (specific_worker->state == KICKED) { - if (GRPC_TRACER_ON(grpc_polling_trace)) { + if (grpc_polling_trace.enabled()) { gpr_log(GPR_ERROR, " .. specific worker already kicked"); } goto done; } else if (gpr_tls_get(&g_current_thread_worker) == (intptr_t)specific_worker) { GRPC_STATS_INC_POLLSET_KICK_OWN_THREAD(); - if (GRPC_TRACER_ON(grpc_polling_trace)) { + if (grpc_polling_trace.enabled()) { gpr_log(GPR_ERROR, " .. mark %p kicked", specific_worker); } SET_KICK_STATE(specific_worker, KICKED); @@ -1106,7 +1108,7 @@ static grpc_error* pollset_kick(grpc_pollset* pollset, } else if (specific_worker == (grpc_pollset_worker*)gpr_atm_no_barrier_load(&g_active_poller)) { GRPC_STATS_INC_POLLSET_KICK_WAKEUP_FD(); - if (GRPC_TRACER_ON(grpc_polling_trace)) { + if (grpc_polling_trace.enabled()) { gpr_log(GPR_ERROR, " .. kick active poller"); } SET_KICK_STATE(specific_worker, KICKED); @@ -1114,7 +1116,7 @@ static grpc_error* pollset_kick(grpc_pollset* pollset, goto done; } else if (specific_worker->initialized_cv) { GRPC_STATS_INC_POLLSET_KICK_WAKEUP_CV(); - if (GRPC_TRACER_ON(grpc_polling_trace)) { + if (grpc_polling_trace.enabled()) { gpr_log(GPR_ERROR, " .. kick waiting worker"); } SET_KICK_STATE(specific_worker, KICKED); @@ -1122,7 +1124,7 @@ static grpc_error* pollset_kick(grpc_pollset* pollset, goto done; } else { GRPC_STATS_INC_POLLSET_KICKED_AGAIN(); - if (GRPC_TRACER_ON(grpc_polling_trace)) { + if (grpc_polling_trace.enabled()) { gpr_log(GPR_ERROR, " .. kick non-waiting worker"); } SET_KICK_STATE(specific_worker, KICKED); diff --git a/src/core/lib/iomgr/ev_epollex_linux.cc b/src/core/lib/iomgr/ev_epollex_linux.cc index d83fe8f80a..b2817156a8 100644 --- a/src/core/lib/iomgr/ev_epollex_linux.cc +++ b/src/core/lib/iomgr/ev_epollex_linux.cc @@ -59,10 +59,8 @@ #define MAX_EPOLL_EVENTS 100 #define MAX_EPOLL_EVENTS_HANDLED_EACH_POLL_CALL 5 -#ifndef NDEBUG -grpc_tracer_flag grpc_trace_pollable_refcount = - GRPC_TRACER_INITIALIZER(false, "pollable_refcount"); -#endif +grpc_core::DebugOnlyTraceFlag grpc_trace_pollable_refcount(false, + "pollable_refcount"); /******************************************************************************* * pollable Declarations @@ -262,7 +260,7 @@ static gpr_mu fd_freelist_mu; #define UNREF_BY(fd, n, reason) unref_by(fd, n, reason, __FILE__, __LINE__) static void ref_by(grpc_fd* fd, int n, const char* reason, const char* file, int line) { - if (GRPC_TRACER_ON(grpc_trace_fd_refcount)) { + if (grpc_trace_fd_refcount.enabled()) { gpr_log(GPR_DEBUG, "FD %d %p ref %d %" PRIdPTR " -> %" PRIdPTR " [%s; %s:%d]", fd->fd, fd, n, gpr_atm_no_barrier_load(&fd->refst), @@ -287,8 +285,8 @@ static void fd_destroy(void* arg, grpc_error* error) { fd->freelist_next = fd_freelist; fd_freelist = fd; - fd->read_closure.Destroy(); - fd->write_closure.Destroy(); + fd->read_closure->DestroyEvent(); + fd->write_closure->DestroyEvent(); gpr_mu_unlock(&fd_freelist_mu); } @@ -296,7 +294,7 @@ static void fd_destroy(void* arg, grpc_error* error) { #ifndef NDEBUG static void unref_by(grpc_fd* fd, int n, const char* reason, const char* file, int line) { - if (GRPC_TRACER_ON(grpc_trace_fd_refcount)) { + if (grpc_trace_fd_refcount.enabled()) { gpr_log(GPR_DEBUG, "FD %d %p unref %d %" PRIdPTR " -> %" PRIdPTR " [%s; %s:%d]", fd->fd, fd, n, gpr_atm_no_barrier_load(&fd->refst), @@ -340,6 +338,8 @@ static grpc_fd* fd_create(int fd, const char* name) { if (new_fd == nullptr) { new_fd = (grpc_fd*)gpr_malloc(sizeof(grpc_fd)); + new_fd->read_closure.Init(); + new_fd->write_closure.Init(); } gpr_mu_init(&new_fd->pollable_mu); @@ -347,9 +347,9 @@ static grpc_fd* fd_create(int fd, const char* name) { new_fd->pollable_obj = nullptr; gpr_atm_rel_store(&new_fd->refst, (gpr_atm)1); new_fd->fd = fd; - new_fd->read_closure.Init(); - new_fd->write_closure.Init(); - gpr_atm_no_barrier_store(&new_fd->read_notifier_pollset, (gpr_atm) nullptr); + new_fd->read_closure->InitEvent(); + new_fd->write_closure->InitEvent(); + gpr_atm_no_barrier_store(&new_fd->read_notifier_pollset, (gpr_atm)NULL); new_fd->freelist_next = nullptr; new_fd->on_done_closure = nullptr; @@ -358,7 +358,7 @@ static grpc_fd* fd_create(int fd, const char* name) { gpr_asprintf(&fd_name, "%s fd=%d", name, fd); grpc_iomgr_register_object(&new_fd->iomgr_object, fd_name); #ifndef NDEBUG - if (GRPC_TRACER_ON(grpc_trace_fd_refcount)) { + if (grpc_trace_fd_refcount.enabled()) { gpr_log(GPR_DEBUG, "FD %d %p create %s", fd, new_fd, fd_name); } #endif @@ -477,7 +477,7 @@ static grpc_error* pollable_create(pollable_type type, pollable** p) { static pollable* pollable_ref(pollable* p) { #else static pollable* pollable_ref(pollable* p, int line, const char* reason) { - if (GRPC_TRACER_ON(grpc_trace_pollable_refcount)) { + if (grpc_trace_pollable_refcount.enabled()) { int r = (int)gpr_atm_no_barrier_load(&p->refs.count); gpr_log(__FILE__, line, GPR_LOG_SEVERITY_DEBUG, "POLLABLE:%p ref %d->%d %s", p, r, r + 1, reason); @@ -492,7 +492,7 @@ static void pollable_unref(pollable* p) { #else static void pollable_unref(pollable* p, int line, const char* reason) { if (p == nullptr) return; - if (GRPC_TRACER_ON(grpc_trace_pollable_refcount)) { + if (grpc_trace_pollable_refcount.enabled()) { int r = (int)gpr_atm_no_barrier_load(&p->refs.count); gpr_log(__FILE__, line, GPR_LOG_SEVERITY_DEBUG, "POLLABLE:%p unref %d->%d %s", p, r, r - 1, reason); @@ -510,7 +510,7 @@ static grpc_error* pollable_add_fd(pollable* p, grpc_fd* fd) { static const char* err_desc = "pollable_add_fd"; const int epfd = p->epfd; - if (GRPC_TRACER_ON(grpc_polling_trace)) { + if (grpc_polling_trace.enabled()) { gpr_log(GPR_DEBUG, "add fd %p (%d) to pollable %p", fd, fd->fd, p); } @@ -551,7 +551,7 @@ static void pollset_global_shutdown(void) { /* pollset->mu must be held while calling this function */ static void pollset_maybe_finish_shutdown(grpc_pollset* pollset) { - if (GRPC_TRACER_ON(grpc_polling_trace)) { + if (grpc_polling_trace.enabled()) { gpr_log(GPR_DEBUG, "PS:%p (pollable:%p) maybe_finish_shutdown sc=%p (target:!NULL) " "rw=%p (target:NULL) cpsc=%d (target:0)", @@ -573,14 +573,14 @@ static grpc_error* kick_one_worker(grpc_pollset_worker* specific_worker) { grpc_core::mu_guard lock(&p->mu); GPR_ASSERT(specific_worker != nullptr); if (specific_worker->kicked) { - if (GRPC_TRACER_ON(grpc_polling_trace)) { + if (grpc_polling_trace.enabled()) { gpr_log(GPR_DEBUG, "PS:%p kicked_specific_but_already_kicked", p); } GRPC_STATS_INC_POLLSET_KICKED_AGAIN(); return GRPC_ERROR_NONE; } if (gpr_tls_get(&g_current_thread_worker) == (intptr_t)specific_worker) { - if (GRPC_TRACER_ON(grpc_polling_trace)) { + if (grpc_polling_trace.enabled()) { gpr_log(GPR_DEBUG, "PS:%p kicked_specific_but_awake", p); } GRPC_STATS_INC_POLLSET_KICK_OWN_THREAD(); @@ -589,7 +589,7 @@ static grpc_error* kick_one_worker(grpc_pollset_worker* specific_worker) { } if (specific_worker == p->root_worker) { GRPC_STATS_INC_POLLSET_KICK_WAKEUP_FD(); - if (GRPC_TRACER_ON(grpc_polling_trace)) { + if (grpc_polling_trace.enabled()) { gpr_log(GPR_DEBUG, "PS:%p kicked_specific_via_wakeup_fd", p); } specific_worker->kicked = true; @@ -598,7 +598,7 @@ static grpc_error* kick_one_worker(grpc_pollset_worker* specific_worker) { } if (specific_worker->initialized_cv) { GRPC_STATS_INC_POLLSET_KICK_WAKEUP_CV(); - if (GRPC_TRACER_ON(grpc_polling_trace)) { + if (grpc_polling_trace.enabled()) { gpr_log(GPR_DEBUG, "PS:%p kicked_specific_via_cv", p); } specific_worker->kicked = true; @@ -613,7 +613,7 @@ static grpc_error* kick_one_worker(grpc_pollset_worker* specific_worker) { static grpc_error* pollset_kick(grpc_pollset* pollset, grpc_pollset_worker* specific_worker) { GRPC_STATS_INC_POLLSET_KICK(); - if (GRPC_TRACER_ON(grpc_polling_trace)) { + if (grpc_polling_trace.enabled()) { gpr_log(GPR_DEBUG, "PS:%p kick %p tls_pollset=%p tls_worker=%p pollset.root_worker=%p", pollset, specific_worker, @@ -623,7 +623,7 @@ static grpc_error* pollset_kick(grpc_pollset* pollset, if (specific_worker == nullptr) { if (gpr_tls_get(&g_current_thread_pollset) != (intptr_t)pollset) { if (pollset->root_worker == nullptr) { - if (GRPC_TRACER_ON(grpc_polling_trace)) { + if (grpc_polling_trace.enabled()) { gpr_log(GPR_DEBUG, "PS:%p kicked_any_without_poller", pollset); } GRPC_STATS_INC_POLLSET_KICKED_WITHOUT_POLLER(); @@ -649,7 +649,7 @@ static grpc_error* pollset_kick(grpc_pollset* pollset, pollset->root_worker->links[PWLINK_POLLSET].next); } } else { - if (GRPC_TRACER_ON(grpc_polling_trace)) { + if (grpc_polling_trace.enabled()) { gpr_log(GPR_DEBUG, "PS:%p kicked_any_but_awake", pollset); } GRPC_STATS_INC_POLLSET_KICK_OWN_THREAD(); @@ -750,7 +750,7 @@ static grpc_error* pollable_process_events(grpc_pollset* pollset, struct epoll_event* ev = &pollable_obj->events[n]; void* data_ptr = ev->data.ptr; if (1 & (intptr_t)data_ptr) { - if (GRPC_TRACER_ON(grpc_polling_trace)) { + if (grpc_polling_trace.enabled()) { gpr_log(GPR_DEBUG, "PS:%p got pollset_wakeup %p", pollset, data_ptr); } append_error(&error, @@ -762,7 +762,7 @@ static grpc_error* pollable_process_events(grpc_pollset* pollset, bool cancel = (ev->events & (EPOLLERR | EPOLLHUP)) != 0; bool read_ev = (ev->events & (EPOLLIN | EPOLLPRI)) != 0; bool write_ev = (ev->events & EPOLLOUT) != 0; - if (GRPC_TRACER_ON(grpc_polling_trace)) { + if (grpc_polling_trace.enabled()) { gpr_log(GPR_DEBUG, "PS:%p got fd %p: cancel=%d read=%d " "write=%d", @@ -789,7 +789,7 @@ static void pollset_destroy(grpc_pollset* pollset) { static grpc_error* pollable_epoll(pollable* p, grpc_millis deadline) { int timeout = poll_deadline_to_millis_timeout(deadline); - if (GRPC_TRACER_ON(grpc_polling_trace)) { + if (grpc_polling_trace.enabled()) { char* desc = pollable_desc(p); gpr_log(GPR_DEBUG, "POLLABLE:%p[%s] poll for %dms", p, desc, timeout); gpr_free(desc); @@ -809,7 +809,7 @@ static grpc_error* pollable_epoll(pollable* p, grpc_millis deadline) { if (r < 0) return GRPC_OS_ERROR(errno, "epoll_wait"); - if (GRPC_TRACER_ON(grpc_polling_trace)) { + if (grpc_polling_trace.enabled()) { gpr_log(GPR_DEBUG, "POLLABLE:%p got %d events", p, r); } @@ -876,7 +876,7 @@ static bool begin_worker(grpc_pollset* pollset, grpc_pollset_worker* worker, worker->initialized_cv = true; gpr_cv_init(&worker->cv); gpr_mu_unlock(&pollset->mu); - if (GRPC_TRACER_ON(grpc_polling_trace) && + if (grpc_polling_trace.enabled() && worker->pollable_obj->root_worker != worker) { gpr_log(GPR_DEBUG, "PS:%p wait %p w=%p for %dms", pollset, worker->pollable_obj, worker, @@ -885,18 +885,18 @@ static bool begin_worker(grpc_pollset* pollset, grpc_pollset_worker* worker, while (do_poll && worker->pollable_obj->root_worker != worker) { if (gpr_cv_wait(&worker->cv, &worker->pollable_obj->mu, grpc_millis_to_timespec(deadline, GPR_CLOCK_REALTIME))) { - if (GRPC_TRACER_ON(grpc_polling_trace)) { + if (grpc_polling_trace.enabled()) { gpr_log(GPR_DEBUG, "PS:%p timeout_wait %p w=%p", pollset, worker->pollable_obj, worker); } do_poll = false; } else if (worker->kicked) { - if (GRPC_TRACER_ON(grpc_polling_trace)) { + if (grpc_polling_trace.enabled()) { gpr_log(GPR_DEBUG, "PS:%p wakeup %p w=%p", pollset, worker->pollable_obj, worker); } do_poll = false; - } else if (GRPC_TRACER_ON(grpc_polling_trace) && + } else if (grpc_polling_trace.enabled() && worker->pollable_obj->root_worker != worker) { gpr_log(GPR_DEBUG, "PS:%p spurious_wakeup %p w=%p", pollset, worker->pollable_obj, worker); @@ -966,7 +966,7 @@ static grpc_error* pollset_work(grpc_pollset* pollset, #ifndef NDEBUG WORKER_PTR->originator = gettid(); #endif - if (GRPC_TRACER_ON(grpc_polling_trace)) { + if (grpc_polling_trace.enabled()) { gpr_log(GPR_DEBUG, "PS:%p work hdl=%p worker=%p now=%" PRIdPTR " deadline=%" PRIdPTR " kwp=%d pollable=%p", @@ -1007,7 +1007,7 @@ static grpc_error* pollset_transition_pollable_from_empty_to_fd_locked( grpc_pollset* pollset, grpc_fd* fd) { static const char* err_desc = "pollset_transition_pollable_from_empty_to_fd"; grpc_error* error = GRPC_ERROR_NONE; - if (GRPC_TRACER_ON(grpc_polling_trace)) { + if (grpc_polling_trace.enabled()) { gpr_log(GPR_DEBUG, "PS:%p add fd %p (%d); transition pollable from empty to fd", pollset, fd, fd->fd); @@ -1023,7 +1023,7 @@ static grpc_error* pollset_transition_pollable_from_fd_to_multi_locked( grpc_pollset* pollset, grpc_fd* and_add_fd) { static const char* err_desc = "pollset_transition_pollable_from_fd_to_multi"; grpc_error* error = GRPC_ERROR_NONE; - if (GRPC_TRACER_ON(grpc_polling_trace)) { + if (grpc_polling_trace.enabled()) { gpr_log( GPR_DEBUG, "PS:%p add fd %p (%d); transition pollable from fd %p to multipoller", @@ -1168,7 +1168,7 @@ static void pollset_set_unref(grpc_pollset_set* pss) { } static void pollset_set_add_fd(grpc_pollset_set* pss, grpc_fd* fd) { - if (GRPC_TRACER_ON(grpc_polling_trace)) { + if (grpc_polling_trace.enabled()) { gpr_log(GPR_DEBUG, "PSS:%p: add fd %p (%d)", pss, fd, fd->fd); } grpc_error* error = GRPC_ERROR_NONE; @@ -1191,7 +1191,7 @@ static void pollset_set_add_fd(grpc_pollset_set* pss, grpc_fd* fd) { } static void pollset_set_del_fd(grpc_pollset_set* pss, grpc_fd* fd) { - if (GRPC_TRACER_ON(grpc_polling_trace)) { + if (grpc_polling_trace.enabled()) { gpr_log(GPR_DEBUG, "PSS:%p: del fd %p", pss, fd); } pss = pss_lock_adam(pss); @@ -1211,7 +1211,7 @@ static void pollset_set_del_fd(grpc_pollset_set* pss, grpc_fd* fd) { } static void pollset_set_del_pollset(grpc_pollset_set* pss, grpc_pollset* ps) { - if (GRPC_TRACER_ON(grpc_polling_trace)) { + if (grpc_polling_trace.enabled()) { gpr_log(GPR_DEBUG, "PSS:%p: del pollset %p", pss, ps); } pss = pss_lock_adam(pss); @@ -1261,7 +1261,7 @@ static grpc_error* add_fds_to_pollsets(grpc_fd** fds, size_t fd_count, } static void pollset_set_add_pollset(grpc_pollset_set* pss, grpc_pollset* ps) { - if (GRPC_TRACER_ON(grpc_polling_trace)) { + if (grpc_polling_trace.enabled()) { gpr_log(GPR_DEBUG, "PSS:%p: add pollset %p", pss, ps); } grpc_error* error = GRPC_ERROR_NONE; @@ -1297,7 +1297,7 @@ static void pollset_set_add_pollset(grpc_pollset_set* pss, grpc_pollset* ps) { static void pollset_set_add_pollset_set(grpc_pollset_set* a, grpc_pollset_set* b) { - if (GRPC_TRACER_ON(grpc_polling_trace)) { + if (grpc_polling_trace.enabled()) { gpr_log(GPR_DEBUG, "PSS: merge (%p, %p)", a, b); } grpc_error* error = GRPC_ERROR_NONE; @@ -1331,7 +1331,7 @@ static void pollset_set_add_pollset_set(grpc_pollset_set* a, if (b_size > a_size) { GPR_SWAP(grpc_pollset_set*, a, b); } - if (GRPC_TRACER_ON(grpc_polling_trace)) { + if (grpc_polling_trace.enabled()) { gpr_log(GPR_DEBUG, "PSS: parent %p to %p", b, a); } gpr_ref(&a->refs); @@ -1431,10 +1431,6 @@ const grpc_event_engine_vtable* grpc_init_epollex_linux( return nullptr; } -#ifndef NDEBUG - grpc_register_tracer(&grpc_trace_pollable_refcount); -#endif - fd_global_init(); if (!GRPC_LOG_IF_ERROR("pollset_global_init", pollset_global_init())) { diff --git a/src/core/lib/iomgr/ev_epollsig_linux.cc b/src/core/lib/iomgr/ev_epollsig_linux.cc index cbea6e4680..7a8962f4a8 100644 --- a/src/core/lib/iomgr/ev_epollsig_linux.cc +++ b/src/core/lib/iomgr/ev_epollsig_linux.cc @@ -54,9 +54,9 @@ #define GRPC_POLLSET_KICK_BROADCAST ((grpc_pollset_worker*)1) -#define GRPC_POLLING_TRACE(...) \ - if (GRPC_TRACER_ON(grpc_polling_trace)) { \ - gpr_log(GPR_INFO, __VA_ARGS__); \ +#define GRPC_POLLING_TRACE(...) \ + if (grpc_polling_trace.enabled()) { \ + gpr_log(GPR_INFO, __VA_ARGS__); \ } static int grpc_wakeup_signal = -1; @@ -288,7 +288,7 @@ static void pi_unref(polling_island* pi); #ifndef NDEBUG static void pi_add_ref_dbg(polling_island* pi, const char* reason, const char* file, int line) { - if (GRPC_TRACER_ON(grpc_polling_trace)) { + if (grpc_polling_trace.enabled()) { gpr_atm old_cnt = gpr_atm_acq_load(&pi->ref_count); gpr_log(GPR_DEBUG, "Add ref pi: %p, old:%" PRIdPTR " -> new:%" PRIdPTR @@ -300,7 +300,7 @@ static void pi_add_ref_dbg(polling_island* pi, const char* reason, static void pi_unref_dbg(polling_island* pi, const char* reason, const char* file, int line) { - if (GRPC_TRACER_ON(grpc_polling_trace)) { + if (grpc_polling_trace.enabled()) { gpr_atm old_cnt = gpr_atm_acq_load(&pi->ref_count); gpr_log(GPR_DEBUG, "Unref pi: %p, old:%" PRIdPTR " -> new:%" PRIdPTR @@ -731,7 +731,7 @@ static gpr_mu fd_freelist_mu; #define UNREF_BY(fd, n, reason) unref_by(fd, n, reason, __FILE__, __LINE__) static void ref_by(grpc_fd* fd, int n, const char* reason, const char* file, int line) { - if (GRPC_TRACER_ON(grpc_trace_fd_refcount)) { + if (grpc_trace_fd_refcount.enabled()) { gpr_log(GPR_DEBUG, "FD %d %p ref %d %" PRIdPTR " -> %" PRIdPTR " [%s; %s:%d]", fd->fd, fd, n, gpr_atm_no_barrier_load(&fd->refst), @@ -748,7 +748,7 @@ static void ref_by(grpc_fd* fd, int n) { #ifndef NDEBUG static void unref_by(grpc_fd* fd, int n, const char* reason, const char* file, int line) { - if (GRPC_TRACER_ON(grpc_trace_fd_refcount)) { + if (grpc_trace_fd_refcount.enabled()) { gpr_log(GPR_DEBUG, "FD %d %p unref %d %" PRIdPTR " -> %" PRIdPTR " [%s; %s:%d]", fd->fd, fd, n, gpr_atm_no_barrier_load(&fd->refst), @@ -765,8 +765,8 @@ static void unref_by(grpc_fd* fd, int n) { fd_freelist = fd; grpc_iomgr_unregister_object(&fd->iomgr_object); - fd->read_closure.Destroy(); - fd->write_closure.Destroy(); + fd->read_closure->DestroyEvent(); + fd->write_closure->DestroyEvent(); gpr_mu_unlock(&fd_freelist_mu); } else { @@ -817,6 +817,8 @@ static grpc_fd* fd_create(int fd, const char* name) { if (new_fd == nullptr) { new_fd = (grpc_fd*)gpr_malloc(sizeof(grpc_fd)); gpr_mu_init(&new_fd->po.mu); + new_fd->read_closure.Init(); + new_fd->write_closure.Init(); } /* Note: It is not really needed to get the new_fd->po.mu lock here. If this @@ -831,9 +833,9 @@ static grpc_fd* fd_create(int fd, const char* name) { gpr_atm_rel_store(&new_fd->refst, (gpr_atm)1); new_fd->fd = fd; new_fd->orphaned = false; - new_fd->read_closure.Init(); - new_fd->write_closure.Init(); - gpr_atm_no_barrier_store(&new_fd->read_notifier_pollset, (gpr_atm) nullptr); + new_fd->read_closure->InitEvent(); + new_fd->write_closure->InitEvent(); + gpr_atm_no_barrier_store(&new_fd->read_notifier_pollset, (gpr_atm)NULL); new_fd->freelist_next = nullptr; new_fd->on_done_closure = nullptr; diff --git a/src/core/lib/iomgr/ev_poll_posix.cc b/src/core/lib/iomgr/ev_poll_posix.cc index 09ebe894fa..1246b382d5 100644 --- a/src/core/lib/iomgr/ev_poll_posix.cc +++ b/src/core/lib/iomgr/ev_poll_posix.cc @@ -284,7 +284,7 @@ cv_fd_table g_cvfds; #define UNREF_BY(fd, n, reason) unref_by(fd, n, reason, __FILE__, __LINE__) static void ref_by(grpc_fd* fd, int n, const char* reason, const char* file, int line) { - if (GRPC_TRACER_ON(grpc_trace_fd_refcount)) { + if (grpc_trace_fd_refcount.enabled()) { gpr_log(GPR_DEBUG, "FD %d %p ref %d %" PRIdPTR " -> %" PRIdPTR " [%s; %s:%d]", fd->fd, fd, n, gpr_atm_no_barrier_load(&fd->refst), @@ -301,7 +301,7 @@ static void ref_by(grpc_fd* fd, int n) { #ifndef NDEBUG static void unref_by(grpc_fd* fd, int n, const char* reason, const char* file, int line) { - if (GRPC_TRACER_ON(grpc_trace_fd_refcount)) { + if (grpc_trace_fd_refcount.enabled()) { gpr_log(GPR_DEBUG, "FD %d %p unref %d %" PRIdPTR " -> %" PRIdPTR " [%s; %s:%d]", fd->fd, fd, n, gpr_atm_no_barrier_load(&fd->refst), @@ -978,7 +978,7 @@ static grpc_error* pollset_work(grpc_pollset* pollset, r = grpc_poll_function(pfds, pfd_count, timeout); GRPC_SCHEDULING_END_BLOCKING_REGION; - if (GRPC_TRACER_ON(grpc_polling_trace)) { + if (grpc_polling_trace.enabled()) { gpr_log(GPR_DEBUG, "%p poll=%d", pollset, r); } @@ -1002,7 +1002,7 @@ static grpc_error* pollset_work(grpc_pollset* pollset, } } else { if (pfds[0].revents & POLLIN_CHECK) { - if (GRPC_TRACER_ON(grpc_polling_trace)) { + if (grpc_polling_trace.enabled()) { gpr_log(GPR_DEBUG, "%p: got_wakeup", pollset); } work_combine_error( @@ -1012,7 +1012,7 @@ static grpc_error* pollset_work(grpc_pollset* pollset, if (watchers[i].fd == nullptr) { fd_end_poll(&watchers[i], 0, 0, nullptr); } else { - if (GRPC_TRACER_ON(grpc_polling_trace)) { + if (grpc_polling_trace.enabled()) { gpr_log(GPR_DEBUG, "%p got_event: %d r:%d w:%d [%d]", pollset, pfds[i].fd, (pfds[i].revents & POLLIN_CHECK) != 0, (pfds[i].revents & POLLOUT_CHECK) != 0, pfds[i].revents); diff --git a/src/core/lib/iomgr/ev_posix.cc b/src/core/lib/iomgr/ev_posix.cc index 985469942f..6dd7dcea30 100644 --- a/src/core/lib/iomgr/ev_posix.cc +++ b/src/core/lib/iomgr/ev_posix.cc @@ -36,13 +36,9 @@ #include "src/core/lib/iomgr/ev_poll_posix.h" #include "src/core/lib/support/env.h" -grpc_tracer_flag grpc_polling_trace = - GRPC_TRACER_INITIALIZER(false, "polling"); /* Disabled by default */ - -#ifndef NDEBUG -grpc_tracer_flag grpc_trace_fd_refcount = - GRPC_TRACER_INITIALIZER(false, "fd_refcount"); -#endif +grpc_core::TraceFlag grpc_polling_trace(false, + "polling"); /* Disabled by default */ +grpc_core::DebugOnlyTraceFlag grpc_trace_fd_refcount(false, "fd_refcount"); /** Default poll() function - a pointer so that it can be overridden by some * tests */ @@ -153,8 +149,6 @@ const grpc_event_engine_vtable* grpc_get_event_engine_test_only() { const char* grpc_get_poll_strategy_name() { return g_poll_strategy_name; } void grpc_event_engine_init(void) { - grpc_register_tracer(&grpc_polling_trace); - char* s = gpr_getenv("GRPC_POLL_STRATEGY"); if (s == nullptr) { s = gpr_strdup("all"); diff --git a/src/core/lib/iomgr/ev_posix.h b/src/core/lib/iomgr/ev_posix.h index 93d8618d4d..39017a9fdd 100644 --- a/src/core/lib/iomgr/ev_posix.h +++ b/src/core/lib/iomgr/ev_posix.h @@ -31,7 +31,7 @@ extern "C" { #endif -extern grpc_tracer_flag grpc_polling_trace; /* Disabled by default */ +extern grpc_core::TraceFlag grpc_polling_trace; /* Disabled by default */ typedef struct grpc_fd grpc_fd; diff --git a/src/core/lib/iomgr/ev_windows.cc b/src/core/lib/iomgr/ev_windows.cc index c24dfaeaf7..697697d0b0 100644 --- a/src/core/lib/iomgr/ev_windows.cc +++ b/src/core/lib/iomgr/ev_windows.cc @@ -22,7 +22,7 @@ #include "src/core/lib/debug/trace.h" -grpc_tracer_flag grpc_polling_trace = - GRPC_TRACER_INITIALIZER(false, "polling"); /* Disabled by default */ +grpc_core::TraceFlag grpc_polling_trace(false, + "polling"); /* Disabled by default */ #endif // GRPC_WINSOCK_SOCKET diff --git a/src/core/lib/iomgr/exec_ctx.cc b/src/core/lib/iomgr/exec_ctx.cc index a5d65f3d2c..4bd92a4141 100644 --- a/src/core/lib/iomgr/exec_ctx.cc +++ b/src/core/lib/iomgr/exec_ctx.cc @@ -25,13 +25,10 @@ #include "src/core/lib/iomgr/combiner.h" #include "src/core/lib/profiling/timers.h" -#define GRPC_START_TIME_UPDATE_INTERVAL 10000 -extern "C" grpc_tracer_flag grpc_timer_check_trace; - static void exec_ctx_run(grpc_closure* closure, grpc_error* error) { #ifndef NDEBUG closure->scheduled = false; - if (GRPC_TRACER_ON(grpc_trace_closure)) { + if (grpc_trace_closure.enabled()) { gpr_log(GPR_DEBUG, "running closure %p: created [%s:%d]: %s [%s:%d]", closure, closure->file_created, closure->line_created, closure->run ? "run" : "scheduled", closure->file_initiated, @@ -40,7 +37,7 @@ static void exec_ctx_run(grpc_closure* closure, grpc_error* error) { #endif closure->cb(closure->cb_arg, error); #ifndef NDEBUG - if (GRPC_TRACER_ON(grpc_trace_closure)) { + if (grpc_trace_closure.enabled()) { gpr_log(GPR_DEBUG, "closure %p finished", closure); } #endif @@ -52,39 +49,10 @@ static void exec_ctx_sched(grpc_closure* closure, grpc_error* error) { error); } -/* This time pair is not entirely thread-safe as store/load of tv_sec and - * tv_nsec are performed separately. However g_start_time do not need to have - * sub-second precision, so it is ok if the value of tv_nsec is off in this - * case. */ -typedef struct time_atm_pair { - gpr_atm tv_sec; - gpr_atm tv_nsec; -} time_atm_pair; - -static time_atm_pair - g_start_time[GPR_TIMESPAN + 1]; // assumes GPR_TIMESPAN is the - // last enum value in - // gpr_clock_type -static grpc_millis g_last_start_time_update; - -static gpr_timespec timespec_from_time_atm_pair(const time_atm_pair* src, - gpr_clock_type clock_type) { - gpr_timespec time; - time.tv_nsec = (int32_t)gpr_atm_no_barrier_load(&src->tv_nsec); - time.tv_sec = (int64_t)gpr_atm_no_barrier_load(&src->tv_sec); - time.clock_type = clock_type; - return time; -} - -static void time_atm_pair_store(time_atm_pair* dst, const gpr_timespec src) { - gpr_atm_no_barrier_store(&dst->tv_sec, src.tv_sec); - gpr_atm_no_barrier_store(&dst->tv_nsec, src.tv_nsec); -} +static gpr_timespec g_start_time; static gpr_atm timespec_to_atm_round_down(gpr_timespec ts) { - gpr_timespec start_time = - timespec_from_time_atm_pair(&g_start_time[ts.clock_type], ts.clock_type); - ts = gpr_time_sub(ts, start_time); + ts = gpr_time_sub(ts, g_start_time); double x = GPR_MS_PER_SEC * (double)ts.tv_sec + (double)ts.tv_nsec / GPR_NS_PER_MS; if (x < 0) return 0; @@ -93,9 +61,7 @@ static gpr_atm timespec_to_atm_round_down(gpr_timespec ts) { } static gpr_atm timespec_to_atm_round_up(gpr_timespec ts) { - gpr_timespec start_time = - timespec_from_time_atm_pair(&g_start_time[ts.clock_type], ts.clock_type); - ts = gpr_time_sub(ts, start_time); + ts = gpr_time_sub(ts, g_start_time); double x = GPR_MS_PER_SEC * (double)ts.tv_sec + (double)ts.tv_nsec / GPR_NS_PER_MS + (double)(GPR_NS_PER_SEC - 1) / (double)GPR_NS_PER_SEC; @@ -118,41 +84,18 @@ gpr_timespec grpc_millis_to_timespec(grpc_millis millis, if (clock_type == GPR_TIMESPAN) { return gpr_time_from_millis(millis, GPR_TIMESPAN); } - gpr_timespec start_time = - timespec_from_time_atm_pair(&g_start_time[clock_type], clock_type); - return gpr_time_add(start_time, gpr_time_from_millis(millis, GPR_TIMESPAN)); + return gpr_time_add(gpr_convert_clock_type(g_start_time, clock_type), + gpr_time_from_millis(millis, GPR_TIMESPAN)); } grpc_millis grpc_timespec_to_millis_round_down(gpr_timespec ts) { - return timespec_to_atm_round_down(ts); + return timespec_to_atm_round_down( + gpr_convert_clock_type(ts, g_start_time.clock_type)); } grpc_millis grpc_timespec_to_millis_round_up(gpr_timespec ts) { - return timespec_to_atm_round_up(ts); -} - -void grpc_exec_ctx_maybe_update_start_time() { - grpc_millis now = grpc_core::ExecCtx::Get()->Now(); - grpc_millis last_start_time_update = - gpr_atm_no_barrier_load(&g_last_start_time_update); - - if (now > last_start_time_update && - now - last_start_time_update > GRPC_START_TIME_UPDATE_INTERVAL) { - /* Get the current system time and subtract \a now from it, where \a now is - * the relative time from grpc_init() from monotonic clock. This calibrates - * the time when grpc_exec_ctx_global_init was called based on current - * system clock. */ - gpr_atm_no_barrier_store(&g_last_start_time_update, now); - gpr_timespec real_now = gpr_now(GPR_CLOCK_REALTIME); - gpr_timespec real_start_time = - gpr_time_sub(real_now, gpr_time_from_millis(now, GPR_TIMESPAN)); - time_atm_pair_store(&g_start_time[GPR_CLOCK_REALTIME], real_start_time); - - if (GRPC_TRACER_ON(grpc_timer_check_trace)) { - gpr_log(GPR_DEBUG, "Update realtime clock start time: %" PRId64 "s %dns", - real_start_time.tv_sec, real_start_time.tv_nsec); - } - } + return timespec_to_atm_round_up( + gpr_convert_clock_type(ts, g_start_time.clock_type)); } static const grpc_closure_scheduler_vtable exec_ctx_scheduler_vtable = { @@ -189,11 +132,7 @@ bool ExecCtx::Flush() { } void ExecCtx::GlobalInit(void) { - for (int i = 0; i < GPR_TIMESPAN; i++) { - time_atm_pair_store(&g_start_time[i], gpr_now((gpr_clock_type)i)); - } - // allows uniform treatment in conversion functions - time_atm_pair_store(&g_start_time[GPR_TIMESPAN], gpr_time_0(GPR_TIMESPAN)); + g_start_time = gpr_now(GPR_CLOCK_MONOTONIC); #ifdef GPR_PTHREAD_TLS gpr_tls_init(&exec_ctx_); #endif diff --git a/src/core/lib/iomgr/executor.cc b/src/core/lib/iomgr/executor.cc index 12f336cb37..d2fb772765 100644 --- a/src/core/lib/iomgr/executor.cc +++ b/src/core/lib/iomgr/executor.cc @@ -51,8 +51,7 @@ static gpr_spinlock g_adding_thread_lock = GPR_SPINLOCK_STATIC_INITIALIZER; GPR_TLS_DECL(g_this_thread_state); -static grpc_tracer_flag executor_trace = - GRPC_TRACER_INITIALIZER(false, "executor"); +grpc_core::TraceFlag executor_trace(false, "executor"); static void executor_thread(void* arg); @@ -63,7 +62,7 @@ static size_t run_closures(grpc_closure_list list) { while (c != nullptr) { grpc_closure* next = c->next_data.next; grpc_error* error = c->error_data.error; - if (GRPC_TRACER_ON(executor_trace)) { + if (executor_trace.enabled()) { #ifndef NDEBUG gpr_log(GPR_DEBUG, "EXECUTOR: run %p [created by %s:%d]", c, c->file_created, c->line_created); @@ -134,7 +133,6 @@ void grpc_executor_set_threading(bool threading) { } void grpc_executor_init() { - grpc_register_tracer(&executor_trace); gpr_atm_no_barrier_store(&g_cur_threads, 0); grpc_executor_set_threading(true); } @@ -149,7 +147,7 @@ static void executor_thread(void* arg) { size_t subtract_depth = 0; for (;;) { - if (GRPC_TRACER_ON(executor_trace)) { + if (executor_trace.enabled()) { gpr_log(GPR_DEBUG, "EXECUTOR[%d]: step (sub_depth=%" PRIdPTR ")", (int)(ts - g_thread_state), subtract_depth); } @@ -160,7 +158,7 @@ static void executor_thread(void* arg) { gpr_cv_wait(&ts->cv, &ts->mu, gpr_inf_future(GPR_CLOCK_REALTIME)); } if (ts->shutdown) { - if (GRPC_TRACER_ON(executor_trace)) { + if (executor_trace.enabled()) { gpr_log(GPR_DEBUG, "EXECUTOR[%d]: shutdown", (int)(ts - g_thread_state)); } @@ -171,7 +169,7 @@ static void executor_thread(void* arg) { grpc_closure_list exec = ts->elems; ts->elems = GRPC_CLOSURE_LIST_INIT; gpr_mu_unlock(&ts->mu); - if (GRPC_TRACER_ON(executor_trace)) { + if (executor_trace.enabled()) { gpr_log(GPR_DEBUG, "EXECUTOR[%d]: execute", (int)(ts - g_thread_state)); } @@ -192,7 +190,7 @@ static void executor_push(grpc_closure* closure, grpc_error* error, retry_push = false; size_t cur_thread_count = (size_t)gpr_atm_no_barrier_load(&g_cur_threads); if (cur_thread_count == 0) { - if (GRPC_TRACER_ON(executor_trace)) { + if (executor_trace.enabled()) { #ifndef NDEBUG gpr_log(GPR_DEBUG, "EXECUTOR: schedule %p (created %s:%d) inline", closure, closure->file_created, closure->line_created); @@ -215,7 +213,7 @@ static void executor_push(grpc_closure* closure, grpc_error* error, bool try_new_thread; for (;;) { - if (GRPC_TRACER_ON(executor_trace)) { + if (executor_trace.enabled()) { #ifndef NDEBUG gpr_log( GPR_DEBUG, diff --git a/src/core/lib/iomgr/iomgr_posix.cc b/src/core/lib/iomgr/iomgr_posix.cc index f5875a247e..f8f6fe2353 100644 --- a/src/core/lib/iomgr/iomgr_posix.cc +++ b/src/core/lib/iomgr/iomgr_posix.cc @@ -28,7 +28,6 @@ void grpc_iomgr_platform_init(void) { grpc_wakeup_fd_global_init(); grpc_event_engine_init(); - grpc_register_tracer(&grpc_tcp_trace); } void grpc_iomgr_platform_flush(void) {} diff --git a/src/core/lib/iomgr/iomgr_uv.cc b/src/core/lib/iomgr/iomgr_uv.cc index 5823bb7ec2..95b06e2ca5 100644 --- a/src/core/lib/iomgr/iomgr_uv.cc +++ b/src/core/lib/iomgr/iomgr_uv.cc @@ -31,7 +31,7 @@ gpr_thd_id g_init_thread; void grpc_iomgr_platform_init(void) { grpc_core::ExecCtx _local_exec_ctx; grpc_pollset_global_init(); - grpc_register_tracer(&grpc_tcp_trace); + grpc_executor_set_threading(false); g_init_thread = gpr_thd_currentid(); } diff --git a/src/core/lib/iomgr/load_file.cc b/src/core/lib/iomgr/load_file.cc index 03ed98f3ba..1afe6ba96d 100644 --- a/src/core/lib/iomgr/load_file.cc +++ b/src/core/lib/iomgr/load_file.cc @@ -30,6 +30,7 @@ grpc_error* grpc_load_file(const char* filename, int add_null_terminator, grpc_slice* output) { + grpc_core::ExecCtx exec_ctx; unsigned char* contents = nullptr; size_t contents_size = 0; grpc_slice result = grpc_empty_slice(); diff --git a/src/core/lib/iomgr/lockfree_event.cc b/src/core/lib/iomgr/lockfree_event.cc index 288e597c29..7b194e3db5 100644 --- a/src/core/lib/iomgr/lockfree_event.cc +++ b/src/core/lib/iomgr/lockfree_event.cc @@ -22,7 +22,7 @@ #include "src/core/lib/debug/trace.h" -extern grpc_tracer_flag grpc_polling_trace; +extern grpc_core::TraceFlag grpc_polling_trace; /* 'state' holds the to call when the fd is readable or writable respectively. It can contain one of the following values: @@ -57,7 +57,9 @@ extern grpc_tracer_flag grpc_polling_trace; namespace grpc_core { -LockfreeEvent::LockfreeEvent() { +LockfreeEvent::LockfreeEvent() { InitEvent(); } + +void LockfreeEvent::InitEvent() { /* Perform an atomic store to start the state machine. Note carefully that LockfreeEvent *MAY* be used whilst in a destroyed @@ -67,7 +69,7 @@ LockfreeEvent::LockfreeEvent() { gpr_atm_no_barrier_store(&state_, kClosureNotReady); } -LockfreeEvent::~LockfreeEvent() { +void LockfreeEvent::DestroyEvent() { gpr_atm curr; do { curr = gpr_atm_no_barrier_load(&state_); @@ -86,7 +88,7 @@ LockfreeEvent::~LockfreeEvent() { void LockfreeEvent::NotifyOn(grpc_closure* closure) { while (true) { gpr_atm curr = gpr_atm_no_barrier_load(&state_); - if (GRPC_TRACER_ON(grpc_polling_trace)) { + if (grpc_polling_trace.enabled()) { gpr_log(GPR_ERROR, "LockfreeEvent::NotifyOn: %p curr=%p closure=%p", this, (void*)curr, closure); } @@ -152,7 +154,7 @@ bool LockfreeEvent::SetShutdown(grpc_error* shutdown_err) { while (true) { gpr_atm curr = gpr_atm_no_barrier_load(&state_); - if (GRPC_TRACER_ON(grpc_polling_trace)) { + if (grpc_polling_trace.enabled()) { gpr_log(GPR_ERROR, "LockfreeEvent::SetShutdown: %p curr=%p err=%s", &state_, (void*)curr, grpc_error_string(shutdown_err)); } @@ -201,7 +203,7 @@ void LockfreeEvent::SetReady() { while (true) { gpr_atm curr = gpr_atm_no_barrier_load(&state_); - if (GRPC_TRACER_ON(grpc_polling_trace)) { + if (grpc_polling_trace.enabled()) { gpr_log(GPR_ERROR, "LockfreeEvent::SetReady: %p curr=%p", &state_, (void*)curr); } diff --git a/src/core/lib/iomgr/lockfree_event.h b/src/core/lib/iomgr/lockfree_event.h index 70fa1ac89d..3bd3fd72f1 100644 --- a/src/core/lib/iomgr/lockfree_event.h +++ b/src/core/lib/iomgr/lockfree_event.h @@ -30,11 +30,16 @@ namespace grpc_core { class LockfreeEvent { public: LockfreeEvent(); - ~LockfreeEvent(); LockfreeEvent(const LockfreeEvent&) = delete; LockfreeEvent& operator=(const LockfreeEvent&) = delete; + // These methods are used to initialize and destroy the internal state. These + // cannot be done in constructor and destructor because SetReady may be called + // when the event is destroyed and put in a freelist. + void InitEvent(); + void DestroyEvent(); + bool IsShutdown() const { return (gpr_atm_no_barrier_load(&state_) & kShutdownBit) != 0; } diff --git a/src/core/lib/iomgr/pollset.h b/src/core/lib/iomgr/pollset.h index 0645c7d919..c32c086971 100644 --- a/src/core/lib/iomgr/pollset.h +++ b/src/core/lib/iomgr/pollset.h @@ -29,9 +29,7 @@ extern "C" { #endif -#ifndef NDEBUG -extern grpc_tracer_flag grpc_trace_fd_refcount; -#endif +extern grpc_core::DebugOnlyTraceFlag grpc_trace_fd_refcount; /* A grpc_pollset is a set of file descriptors that a higher level item is interested in. For example: diff --git a/src/core/lib/iomgr/pollset_uv.cc b/src/core/lib/iomgr/pollset_uv.cc index 89d28a0f9d..d040c0cccb 100644 --- a/src/core/lib/iomgr/pollset_uv.cc +++ b/src/core/lib/iomgr/pollset_uv.cc @@ -34,10 +34,7 @@ #include "src/core/lib/debug/trace.h" -#ifndef NDEBUG -grpc_tracer_flag grpc_trace_fd_refcount = - GRPC_TRACER_INITIALIZER(false, "fd_refcount"); -#endif +grpc_core::DebugOnlyTraceFlag grpc_trace_fd_refcount(false, "fd_refcount"); struct grpc_pollset { uv_timer_t* timer; diff --git a/src/core/lib/iomgr/pollset_windows.cc b/src/core/lib/iomgr/pollset_windows.cc index 81e1d009ca..162037a8da 100644 --- a/src/core/lib/iomgr/pollset_windows.cc +++ b/src/core/lib/iomgr/pollset_windows.cc @@ -30,10 +30,7 @@ #define GRPC_POLLSET_KICK_BROADCAST ((grpc_pollset_worker*)1) -#ifndef NDEBUG -grpc_tracer_flag grpc_trace_fd_refcount = - GRPC_TRACER_INITIALIZER(false, "fd_refcount"); -#endif +grpc_core::DebugOnlyTraceFlag grpc_trace_fd_refcount(false, "fd_refcount"); gpr_mu grpc_polling_mu; static grpc_pollset_worker* g_active_poller; diff --git a/src/core/lib/iomgr/resource_quota.cc b/src/core/lib/iomgr/resource_quota.cc index 819d997ce9..55d559c466 100644 --- a/src/core/lib/iomgr/resource_quota.cc +++ b/src/core/lib/iomgr/resource_quota.cc @@ -31,8 +31,7 @@ #include "src/core/lib/iomgr/combiner.h" -grpc_tracer_flag grpc_resource_quota_trace = - GRPC_TRACER_INITIALIZER(false, "resource_quota"); +grpc_core::TraceFlag grpc_resource_quota_trace(false, "resource_quota"); #define MEMORY_USAGE_ESTIMATION_MAX 65536 @@ -287,7 +286,7 @@ static bool rq_alloc(grpc_resource_quota* resource_quota) { while ((resource_user = rulist_pop_head(resource_quota, GRPC_RULIST_AWAITING_ALLOCATION))) { gpr_mu_lock(&resource_user->mu); - if (GRPC_TRACER_ON(grpc_resource_quota_trace)) { + if (grpc_resource_quota_trace.enabled()) { gpr_log(GPR_DEBUG, "RQ: check allocation for user %p shutdown=%" PRIdPTR " free_pool=%" PRId64, @@ -313,14 +312,14 @@ static bool rq_alloc(grpc_resource_quota* resource_quota) { resource_user->free_pool = 0; resource_quota->free_pool -= amt; rq_update_estimate(resource_quota); - if (GRPC_TRACER_ON(grpc_resource_quota_trace)) { + if (grpc_resource_quota_trace.enabled()) { 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_TRACER_ON(grpc_resource_quota_trace) && + } else if (grpc_resource_quota_trace.enabled() && resource_user->free_pool >= 0) { gpr_log(GPR_DEBUG, "RQ %s %s: discard already satisfied alloc request", resource_quota->name, resource_user->name); @@ -351,7 +350,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_TRACER_ON(grpc_resource_quota_trace)) { + if (grpc_resource_quota_trace.enabled()) { gpr_log(GPR_DEBUG, "RQ %s %s: reclaim_from_per_user_free_pool %" PRId64 " bytes; rq_free_pool -> %" PRId64, @@ -374,7 +373,7 @@ static bool rq_reclaim(grpc_resource_quota* resource_quota, bool destructive) { : GRPC_RULIST_RECLAIMER_BENIGN; grpc_resource_user* resource_user = rulist_pop_head(resource_quota, list); if (resource_user == nullptr) return false; - if (GRPC_TRACER_ON(grpc_resource_quota_trace)) { + if (grpc_resource_quota_trace.enabled()) { gpr_log(GPR_DEBUG, "RQ %s %s: initiate %s reclamation", resource_quota->name, resource_user->name, destructive ? "destructive" : "benign"); @@ -504,7 +503,7 @@ static void ru_post_destructive_reclaimer(void* ru, grpc_error* error) { } static void ru_shutdown(void* ru, grpc_error* error) { - if (GRPC_TRACER_ON(grpc_resource_quota_trace)) { + if (grpc_resource_quota_trace.enabled()) { gpr_log(GPR_DEBUG, "RU shutdown %p", ru); } grpc_resource_user* resource_user = (grpc_resource_user*)ru; @@ -787,7 +786,7 @@ void grpc_resource_user_alloc(grpc_resource_user* resource_user, size_t size, ru_ref_by(resource_user, (gpr_atm)size); resource_user->free_pool -= (int64_t)size; resource_user->outstanding_allocations += (int64_t)size; - if (GRPC_TRACER_ON(grpc_resource_quota_trace)) { + if (grpc_resource_quota_trace.enabled()) { 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 +809,7 @@ void grpc_resource_user_free(grpc_resource_user* resource_user, size_t size) { 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_TRACER_ON(grpc_resource_quota_trace)) { + if (grpc_resource_quota_trace.enabled()) { 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); @@ -836,7 +835,7 @@ void grpc_resource_user_post_reclaimer(grpc_resource_user* resource_user, } void grpc_resource_user_finish_reclamation(grpc_resource_user* resource_user) { - if (GRPC_TRACER_ON(grpc_resource_quota_trace)) { + if (grpc_resource_quota_trace.enabled()) { gpr_log(GPR_DEBUG, "RQ %s %s: reclamation complete", resource_user->resource_quota->name, resource_user->name); } @@ -864,9 +863,3 @@ void grpc_resource_user_alloc_slices( grpc_resource_user_alloc(slice_allocator->resource_user, count * length, &slice_allocator->on_allocated); } - -grpc_slice grpc_resource_user_slice_malloc(grpc_resource_user* resource_user, - size_t size) { - grpc_resource_user_alloc(resource_user, size, nullptr); - return ru_slice_create(resource_user, size); -} diff --git a/src/core/lib/iomgr/resource_quota.h b/src/core/lib/iomgr/resource_quota.h index b5ccdc76ca..ae7f487b5d 100644 --- a/src/core/lib/iomgr/resource_quota.h +++ b/src/core/lib/iomgr/resource_quota.h @@ -65,7 +65,7 @@ extern "C" { maintain lists of users (which users arrange to leave before they are destroyed) */ -extern grpc_tracer_flag grpc_resource_quota_trace; +extern grpc_core::TraceFlag 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.cc b/src/core/lib/iomgr/tcp_client_posix.cc index 6e0fbf2a3c..84c5a674d4 100644 --- a/src/core/lib/iomgr/tcp_client_posix.cc +++ b/src/core/lib/iomgr/tcp_client_posix.cc @@ -43,7 +43,7 @@ #include "src/core/lib/iomgr/unix_sockets_posix.h" #include "src/core/lib/support/string.h" -extern grpc_tracer_flag grpc_tcp_trace; +extern grpc_core::TraceFlag grpc_tcp_trace; typedef struct { gpr_mu mu; @@ -99,7 +99,7 @@ done: static void tc_on_alarm(void* acp, grpc_error* error) { int done; async_connect* ac = (async_connect*)acp; - if (GRPC_TRACER_ON(grpc_tcp_trace)) { + if (grpc_tcp_trace.enabled()) { const char* str = grpc_error_string(error); gpr_log(GPR_DEBUG, "CLIENT_CONNECT: %s: on_alarm: error=%s", ac->addr_str, str); @@ -136,7 +136,7 @@ static void on_writable(void* acp, grpc_error* error) { GRPC_ERROR_REF(error); - if (GRPC_TRACER_ON(grpc_tcp_trace)) { + if (grpc_tcp_trace.enabled()) { const char* str = grpc_error_string(error); gpr_log(GPR_DEBUG, "CLIENT_CONNECT: %s: on_writable: error=%s", ac->addr_str, str); @@ -312,7 +312,7 @@ static void tcp_client_connect_impl(grpc_closure* closure, grpc_endpoint** ep, grpc_schedule_on_exec_ctx); ac->channel_args = grpc_channel_args_copy(channel_args); - if (GRPC_TRACER_ON(grpc_tcp_trace)) { + if (grpc_tcp_trace.enabled()) { gpr_log(GPR_DEBUG, "CLIENT_CONNECT: %s: asynchronously connecting fd %p", ac->addr_str, fdobj); } diff --git a/src/core/lib/iomgr/tcp_client_uv.cc b/src/core/lib/iomgr/tcp_client_uv.cc index 0bf3a043df..0ba074500b 100644 --- a/src/core/lib/iomgr/tcp_client_uv.cc +++ b/src/core/lib/iomgr/tcp_client_uv.cc @@ -32,7 +32,7 @@ #include "src/core/lib/iomgr/tcp_uv.h" #include "src/core/lib/iomgr/timer.h" -extern grpc_tracer_flag grpc_tcp_trace; +extern grpc_core::TraceFlag grpc_tcp_trace; typedef struct grpc_uv_tcp_connect { uv_connect_t connect_req; @@ -57,7 +57,7 @@ static void tcp_close_callback(uv_handle_t* handle) { gpr_free(handle); } static void uv_tc_on_alarm(void* acp, grpc_error* error) { int done; grpc_uv_tcp_connect* connect = (grpc_uv_tcp_connect*)acp; - if (GRPC_TRACER_ON(grpc_tcp_trace)) { + if (grpc_tcp_trace.enabled()) { const char* str = grpc_error_string(error); gpr_log(GPR_DEBUG, "CLIENT_CONNECT: %s: on_alarm: error=%s", connect->addr_name, str); @@ -143,7 +143,7 @@ static void tcp_client_connect_impl(grpc_closure* closure, grpc_endpoint** ep, connect->connect_req.data = connect; connect->refs = 2; // One for the connect operation, one for the timer. - if (GRPC_TRACER_ON(grpc_tcp_trace)) { + if (grpc_tcp_trace.enabled()) { gpr_log(GPR_DEBUG, "CLIENT_CONNECT: %s: asynchronously connecting", connect->addr_name); } diff --git a/src/core/lib/iomgr/tcp_posix.cc b/src/core/lib/iomgr/tcp_posix.cc index c1305bef8e..155329d2e8 100644 --- a/src/core/lib/iomgr/tcp_posix.cc +++ b/src/core/lib/iomgr/tcp_posix.cc @@ -61,7 +61,7 @@ typedef GRPC_MSG_IOVLEN_TYPE msg_iovlen_type; typedef size_t msg_iovlen_type; #endif -grpc_tracer_flag grpc_tcp_trace = GRPC_TRACER_INITIALIZER(false, "tcp"); +grpc_core::TraceFlag grpc_tcp_trace(false, "tcp"); typedef struct { grpc_endpoint base; @@ -81,9 +81,7 @@ typedef struct { grpc_slice_buffer* incoming_buffer; grpc_slice_buffer* outgoing_buffer; - /** slice within outgoing_buffer to write next */ - size_t outgoing_slice_idx; - /** byte within outgoing_buffer->slices[outgoing_slice_idx] to write next */ + /** byte within outgoing_buffer->slices[0] to write next */ size_t outgoing_byte_idx; grpc_closure* read_cb; @@ -117,7 +115,7 @@ static void tcp_drop_uncovered_then_handle_write(void* arg /* grpc_tcp */, static void done_poller(void* bp, grpc_error* error_ignored) { backup_poller* p = (backup_poller*)bp; - if (GRPC_TRACER_ON(grpc_tcp_trace)) { + if (grpc_tcp_trace.enabled()) { gpr_log(GPR_DEBUG, "BACKUP_POLLER:%p destroy", p); } grpc_pollset_destroy(BACKUP_POLLER_POLLSET(p)); @@ -126,7 +124,7 @@ static void done_poller(void* bp, grpc_error* error_ignored) { static void run_poller(void* bp, grpc_error* error_ignored) { backup_poller* p = (backup_poller*)bp; - if (GRPC_TRACER_ON(grpc_tcp_trace)) { + if (grpc_tcp_trace.enabled()) { gpr_log(GPR_DEBUG, "BACKUP_POLLER:%p run", p); } gpr_mu_lock(p->pollset_mu); @@ -142,18 +140,18 @@ static void run_poller(void* bp, grpc_error* error_ignored) { gpr_atm_full_cas(&g_uncovered_notifications_pending, 1, 0)) { gpr_mu_lock(p->pollset_mu); bool cas_ok = gpr_atm_full_cas(&g_backup_poller, (gpr_atm)p, 0); - if (GRPC_TRACER_ON(grpc_tcp_trace)) { + if (grpc_tcp_trace.enabled()) { gpr_log(GPR_DEBUG, "BACKUP_POLLER:%p done cas_ok=%d", p, cas_ok); } gpr_mu_unlock(p->pollset_mu); - if (GRPC_TRACER_ON(grpc_tcp_trace)) { + if (grpc_tcp_trace.enabled()) { gpr_log(GPR_DEBUG, "BACKUP_POLLER:%p shutdown", p); } grpc_pollset_shutdown(BACKUP_POLLER_POLLSET(p), GRPC_CLOSURE_INIT(&p->run_poller, done_poller, p, grpc_schedule_on_exec_ctx)); } else { - if (GRPC_TRACER_ON(grpc_tcp_trace)) { + if (grpc_tcp_trace.enabled()) { gpr_log(GPR_DEBUG, "BACKUP_POLLER:%p reschedule", p); } GRPC_CLOSURE_SCHED(&p->run_poller, GRPC_ERROR_NONE); @@ -164,7 +162,7 @@ static void drop_uncovered(grpc_tcp* tcp) { backup_poller* p = (backup_poller*)gpr_atm_acq_load(&g_backup_poller); gpr_atm old_count = gpr_atm_no_barrier_fetch_add(&g_uncovered_notifications_pending, -1); - if (GRPC_TRACER_ON(grpc_tcp_trace)) { + if (grpc_tcp_trace.enabled()) { gpr_log(GPR_DEBUG, "BACKUP_POLLER:%p uncover cnt %d->%d", p, (int)old_count, (int)old_count - 1); } @@ -175,14 +173,14 @@ static void cover_self(grpc_tcp* tcp) { backup_poller* p; gpr_atm old_count = gpr_atm_no_barrier_fetch_add(&g_uncovered_notifications_pending, 2); - if (GRPC_TRACER_ON(grpc_tcp_trace)) { + if (grpc_tcp_trace.enabled()) { gpr_log(GPR_DEBUG, "BACKUP_POLLER: cover cnt %d->%d", (int)old_count, 2 + (int)old_count); } if (old_count == 0) { GRPC_STATS_INC_TCP_BACKUP_POLLERS_CREATED(); p = (backup_poller*)gpr_zalloc(sizeof(*p) + grpc_pollset_size()); - if (GRPC_TRACER_ON(grpc_tcp_trace)) { + if (grpc_tcp_trace.enabled()) { gpr_log(GPR_DEBUG, "BACKUP_POLLER:%p create", p); } grpc_pollset_init(BACKUP_POLLER_POLLSET(p), &p->pollset_mu); @@ -197,7 +195,7 @@ static void cover_self(grpc_tcp* tcp) { // spin waiting for backup poller } } - if (GRPC_TRACER_ON(grpc_tcp_trace)) { + if (grpc_tcp_trace.enabled()) { gpr_log(GPR_DEBUG, "BACKUP_POLLER:%p add %p", p, tcp); } grpc_pollset_add_fd(BACKUP_POLLER_POLLSET(p), tcp->em_fd); @@ -207,7 +205,7 @@ static void cover_self(grpc_tcp* tcp) { } static void notify_on_read(grpc_tcp* tcp) { - if (GRPC_TRACER_ON(grpc_tcp_trace)) { + if (grpc_tcp_trace.enabled()) { gpr_log(GPR_DEBUG, "TCP:%p notify_on_read", tcp); } GRPC_CLOSURE_INIT(&tcp->read_done_closure, tcp_handle_read, tcp, @@ -216,7 +214,7 @@ static void notify_on_read(grpc_tcp* tcp) { } static void notify_on_write(grpc_tcp* tcp) { - if (GRPC_TRACER_ON(grpc_tcp_trace)) { + if (grpc_tcp_trace.enabled()) { gpr_log(GPR_DEBUG, "TCP:%p notify_on_write", tcp); } cover_self(tcp); @@ -227,7 +225,7 @@ static void notify_on_write(grpc_tcp* tcp) { } static void tcp_drop_uncovered_then_handle_write(void* arg, grpc_error* error) { - if (GRPC_TRACER_ON(grpc_tcp_trace)) { + if (grpc_tcp_trace.enabled()) { gpr_log(GPR_DEBUG, "TCP:%p got_write: %s", arg, grpc_error_string(error)); } drop_uncovered((grpc_tcp*)arg); @@ -300,7 +298,7 @@ static void tcp_free(grpc_tcp* tcp) { #define TCP_REF(tcp, reason) tcp_ref((tcp), (reason), __FILE__, __LINE__) static void tcp_unref(grpc_tcp* tcp, const char* reason, const char* file, int line) { - if (GRPC_TRACER_ON(grpc_tcp_trace)) { + if (grpc_tcp_trace.enabled()) { gpr_atm val = gpr_atm_no_barrier_load(&tcp->refcount.count); gpr_log(file, line, GPR_LOG_SEVERITY_DEBUG, "TCP unref %p : %s %" PRIdPTR " -> %" PRIdPTR, tcp, reason, val, @@ -313,7 +311,7 @@ static void tcp_unref(grpc_tcp* tcp, const char* reason, const char* file, static void tcp_ref(grpc_tcp* tcp, const char* reason, const char* file, int line) { - if (GRPC_TRACER_ON(grpc_tcp_trace)) { + if (grpc_tcp_trace.enabled()) { gpr_atm val = gpr_atm_no_barrier_load(&tcp->refcount.count); gpr_log(file, line, GPR_LOG_SEVERITY_DEBUG, "TCP ref %p : %s %" PRIdPTR " -> %" PRIdPTR, tcp, reason, val, @@ -343,7 +341,7 @@ static void tcp_destroy(grpc_endpoint* ep) { static void call_read_cb(grpc_tcp* tcp, grpc_error* error) { grpc_closure* cb = tcp->read_cb; - if (GRPC_TRACER_ON(grpc_tcp_trace)) { + if (grpc_tcp_trace.enabled()) { gpr_log(GPR_DEBUG, "TCP:%p call_cb %p %p:%p", tcp, cb, cb->cb, cb->cb_arg); size_t i; const char* str = grpc_error_string(error); @@ -436,7 +434,7 @@ static void tcp_do_read(grpc_tcp* tcp) { static void tcp_read_allocation_done(void* tcpp, grpc_error* error) { grpc_tcp* tcp = (grpc_tcp*)tcpp; - if (GRPC_TRACER_ON(grpc_tcp_trace)) { + if (grpc_tcp_trace.enabled()) { gpr_log(GPR_DEBUG, "TCP:%p read_allocation_done: %s", tcp, grpc_error_string(error)); } @@ -454,13 +452,13 @@ static void tcp_continue_read(grpc_tcp* tcp) { size_t target_read_size = get_target_read_size(tcp); if (tcp->incoming_buffer->length < target_read_size && tcp->incoming_buffer->count < MAX_READ_IOVEC) { - if (GRPC_TRACER_ON(grpc_tcp_trace)) { + if (grpc_tcp_trace.enabled()) { gpr_log(GPR_DEBUG, "TCP:%p alloc_slices", tcp); } grpc_resource_user_alloc_slices(&tcp->slice_allocator, target_read_size, 1, tcp->incoming_buffer); } else { - if (GRPC_TRACER_ON(grpc_tcp_trace)) { + if (grpc_tcp_trace.enabled()) { gpr_log(GPR_DEBUG, "TCP:%p do_read", tcp); } tcp_do_read(tcp); @@ -470,7 +468,7 @@ static void tcp_continue_read(grpc_tcp* tcp) { static void tcp_handle_read(void* arg /* grpc_tcp */, grpc_error* error) { grpc_tcp* tcp = (grpc_tcp*)arg; GPR_ASSERT(!tcp->finished_edge); - if (GRPC_TRACER_ON(grpc_tcp_trace)) { + if (grpc_tcp_trace.enabled()) { gpr_log(GPR_DEBUG, "TCP:%p got_read: %s", tcp, grpc_error_string(error)); } @@ -513,23 +511,26 @@ static bool tcp_flush(grpc_tcp* tcp, grpc_error** error) { size_t unwind_slice_idx; size_t unwind_byte_idx; + // We always start at zero, because we eagerly unref and trim the slice + // buffer as we write + size_t outgoing_slice_idx = 0; + for (;;) { sending_length = 0; - unwind_slice_idx = tcp->outgoing_slice_idx; + unwind_slice_idx = outgoing_slice_idx; unwind_byte_idx = tcp->outgoing_byte_idx; - for (iov_size = 0; tcp->outgoing_slice_idx != tcp->outgoing_buffer->count && + for (iov_size = 0; outgoing_slice_idx != tcp->outgoing_buffer->count && iov_size != MAX_WRITE_IOVEC; iov_size++) { iov[iov_size].iov_base = GRPC_SLICE_START_PTR( - tcp->outgoing_buffer->slices[tcp->outgoing_slice_idx]) + + tcp->outgoing_buffer->slices[outgoing_slice_idx]) + tcp->outgoing_byte_idx; iov[iov_size].iov_len = - GRPC_SLICE_LENGTH( - tcp->outgoing_buffer->slices[tcp->outgoing_slice_idx]) - + GRPC_SLICE_LENGTH(tcp->outgoing_buffer->slices[outgoing_slice_idx]) - tcp->outgoing_byte_idx; sending_length += iov[iov_size].iov_len; - tcp->outgoing_slice_idx++; + outgoing_slice_idx++; tcp->outgoing_byte_idx = 0; } GPR_ASSERT(iov_size > 0); @@ -555,16 +556,23 @@ static bool tcp_flush(grpc_tcp* tcp, grpc_error** error) { if (sent_length < 0) { if (errno == EAGAIN) { - tcp->outgoing_slice_idx = unwind_slice_idx; tcp->outgoing_byte_idx = unwind_byte_idx; + // unref all and forget about all slices that have been written to this + // point + for (size_t idx = 0; idx < unwind_slice_idx; ++idx) { + grpc_slice_unref_internal( + grpc_slice_buffer_take_first(tcp->outgoing_buffer)); + } return false; } else if (errno == EPIPE) { *error = grpc_error_set_int(GRPC_OS_ERROR(errno, "sendmsg"), GRPC_ERROR_INT_GRPC_STATUS, GRPC_STATUS_UNAVAILABLE); + grpc_slice_buffer_reset_and_unref_internal(tcp->outgoing_buffer); return true; } else { *error = tcp_annotate_error(GRPC_OS_ERROR(errno, "sendmsg"), tcp); + grpc_slice_buffer_reset_and_unref_internal(tcp->outgoing_buffer); return true; } } @@ -574,9 +582,9 @@ static bool tcp_flush(grpc_tcp* tcp, grpc_error** error) { while (trailing > 0) { size_t slice_length; - tcp->outgoing_slice_idx--; - slice_length = GRPC_SLICE_LENGTH( - tcp->outgoing_buffer->slices[tcp->outgoing_slice_idx]); + outgoing_slice_idx--; + slice_length = + GRPC_SLICE_LENGTH(tcp->outgoing_buffer->slices[outgoing_slice_idx]); if (slice_length > trailing) { tcp->outgoing_byte_idx = slice_length - trailing; break; @@ -585,11 +593,12 @@ static bool tcp_flush(grpc_tcp* tcp, grpc_error** error) { } } - if (tcp->outgoing_slice_idx == tcp->outgoing_buffer->count) { + if (outgoing_slice_idx == tcp->outgoing_buffer->count) { *error = GRPC_ERROR_NONE; + grpc_slice_buffer_reset_and_unref_internal(tcp->outgoing_buffer); return true; } - }; + } } static void tcp_handle_write(void* arg /* grpc_tcp */, grpc_error* error) { @@ -605,14 +614,14 @@ static void tcp_handle_write(void* arg /* grpc_tcp */, grpc_error* error) { } if (!tcp_flush(tcp, &error)) { - if (GRPC_TRACER_ON(grpc_tcp_trace)) { + if (grpc_tcp_trace.enabled()) { gpr_log(GPR_DEBUG, "write: delayed"); } notify_on_write(tcp); } else { cb = tcp->write_cb; tcp->write_cb = nullptr; - if (GRPC_TRACER_ON(grpc_tcp_trace)) { + if (grpc_tcp_trace.enabled()) { const char* str = grpc_error_string(error); gpr_log(GPR_DEBUG, "write: %s", str); } @@ -627,7 +636,7 @@ static void tcp_write(grpc_endpoint* ep, grpc_slice_buffer* buf, grpc_tcp* tcp = (grpc_tcp*)ep; grpc_error* error = GRPC_ERROR_NONE; - if (GRPC_TRACER_ON(grpc_tcp_trace)) { + if (grpc_tcp_trace.enabled()) { size_t i; for (i = 0; i < buf->count; i++) { @@ -651,18 +660,17 @@ static void tcp_write(grpc_endpoint* ep, grpc_slice_buffer* buf, return; } tcp->outgoing_buffer = buf; - tcp->outgoing_slice_idx = 0; tcp->outgoing_byte_idx = 0; if (!tcp_flush(tcp, &error)) { TCP_REF(tcp, "write"); tcp->write_cb = cb; - if (GRPC_TRACER_ON(grpc_tcp_trace)) { + if (grpc_tcp_trace.enabled()) { gpr_log(GPR_DEBUG, "write: delayed"); } notify_on_write(tcp); } else { - if (GRPC_TRACER_ON(grpc_tcp_trace)) { + if (grpc_tcp_trace.enabled()) { 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 a6320306fd..2fa1256ec7 100644 --- a/src/core/lib/iomgr/tcp_posix.h +++ b/src/core/lib/iomgr/tcp_posix.h @@ -37,7 +37,7 @@ extern "C" { #endif -extern grpc_tracer_flag grpc_tcp_trace; +extern grpc_core::TraceFlag 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.cc b/src/core/lib/iomgr/tcp_server_posix.cc index 475e9fef79..99e1c6cd06 100644 --- a/src/core/lib/iomgr/tcp_server_posix.cc +++ b/src/core/lib/iomgr/tcp_server_posix.cc @@ -240,7 +240,7 @@ static void on_read(void* arg, grpc_error* err) { addr_str = grpc_sockaddr_to_uri(&addr); gpr_asprintf(&name, "tcp-server-connection:%s", addr_str); - if (GRPC_TRACER_ON(grpc_tcp_trace)) { + if (grpc_tcp_trace.enabled()) { gpr_log(GPR_DEBUG, "SERVER_CONNECT: incoming connection: %s", addr_str); } diff --git a/src/core/lib/iomgr/tcp_server_uv.cc b/src/core/lib/iomgr/tcp_server_uv.cc index daa7afe95f..2181b12b23 100644 --- a/src/core/lib/iomgr/tcp_server_uv.cc +++ b/src/core/lib/iomgr/tcp_server_uv.cc @@ -205,7 +205,7 @@ static void finish_accept(grpc_tcp_listener* sp) { } else { gpr_log(GPR_INFO, "uv_tcp_getpeername error: %s", uv_strerror(err)); } - if (GRPC_TRACER_ON(grpc_tcp_trace)) { + if (grpc_tcp_trace.enabled()) { if (peer_name_string) { gpr_log(GPR_DEBUG, "SERVER_CONNECT: %p accepted connection: %s", sp->server, peer_name_string); @@ -238,7 +238,7 @@ static void on_connect(uv_stream_t* server, int status) { GPR_ASSERT(!sp->has_pending_connection); - if (GRPC_TRACER_ON(grpc_tcp_trace)) { + if (grpc_tcp_trace.enabled()) { gpr_log(GPR_DEBUG, "SERVER_CONNECT: %p incoming connection", sp->server); } @@ -393,7 +393,7 @@ grpc_error* grpc_tcp_server_add_port(grpc_tcp_server* s, gpr_free(allocated_addr); - if (GRPC_TRACER_ON(grpc_tcp_trace)) { + if (grpc_tcp_trace.enabled()) { char* port_string; grpc_sockaddr_to_string(&port_string, addr, 0); const char* str = grpc_error_string(error); @@ -425,7 +425,7 @@ void grpc_tcp_server_start(grpc_tcp_server* server, grpc_pollset** pollsets, (void)pollsets; (void)pollset_count; GRPC_UV_ASSERT_SAME_THREAD(); - if (GRPC_TRACER_ON(grpc_tcp_trace)) { + if (grpc_tcp_trace.enabled()) { gpr_log(GPR_DEBUG, "SERVER_START %p", server); } GPR_ASSERT(on_accept_cb); diff --git a/src/core/lib/iomgr/tcp_uv.cc b/src/core/lib/iomgr/tcp_uv.cc index 742ab9a754..d1e91357ea 100644 --- a/src/core/lib/iomgr/tcp_uv.cc +++ b/src/core/lib/iomgr/tcp_uv.cc @@ -38,7 +38,7 @@ #include "src/core/lib/slice/slice_string_helpers.h" #include "src/core/lib/support/string.h" -grpc_tracer_flag grpc_tcp_trace = GRPC_TRACER_INITIALIZER(false, "tcp"); +grpc_core::TraceFlag grpc_tcp_trace(false, "tcp"); typedef struct { grpc_endpoint base; @@ -52,12 +52,12 @@ typedef struct { grpc_closure* read_cb; grpc_closure* write_cb; - grpc_slice read_slice; grpc_slice_buffer* read_slices; grpc_slice_buffer* write_slices; uv_buf_t* write_buffers; grpc_resource_user* resource_user; + grpc_resource_user_slice_allocator slice_allocator; bool shutting_down; @@ -65,8 +65,7 @@ typedef struct { grpc_pollset* pollset; } grpc_tcp; -static void tcp_free(grpc_tcp* tcp) { - grpc_slice_unref_internal(tcp->read_slice); +static void tcp_free(grpc_exec_ctx* exec_ctx, grpc_tcp* tcp) { grpc_resource_user_unref(tcp->resource_user); gpr_free(tcp->handle); gpr_free(tcp->peer_string); @@ -78,7 +77,7 @@ static void tcp_free(grpc_tcp* tcp) { #define TCP_REF(tcp, reason) tcp_ref((tcp), (reason), __FILE__, __LINE__) static void tcp_unref(grpc_tcp* tcp, const char* reason, const char* file, int line) { - if (GRPC_TRACER_ON(grpc_tcp_trace)) { + if (grpc_tcp_trace.enabled()) { gpr_atm val = gpr_atm_no_barrier_load(&tcp->refcount.count); gpr_log(file, line, GPR_LOG_SEVERITY_DEBUG, "TCP unref %p : %s %" PRIdPTR " -> %" PRIdPTR, tcp, reason, val, @@ -91,7 +90,7 @@ static void tcp_unref(grpc_tcp* tcp, const char* reason, const char* file, static void tcp_ref(grpc_tcp* tcp, const char* reason, const char* file, int line) { - if (GRPC_TRACER_ON(grpc_tcp_trace)) { + if (grpc_tcp_trace.enabled()) { gpr_atm val = gpr_atm_no_barrier_load(&tcp->refcount.count); gpr_log(file, line, GPR_LOG_SEVERITY_DEBUG, "TCP ref %p : %s %" PRIdPTR " -> %" PRIdPTR, tcp, reason, val, @@ -117,88 +116,116 @@ static void uv_close_callback(uv_handle_t* handle) { TCP_UNREF(tcp, "destroy"); } -static grpc_slice alloc_read_slice(grpc_resource_user* resource_user) { - return grpc_resource_user_slice_malloc(resource_user, - GRPC_TCP_DEFAULT_READ_SLICE_SIZE); -} - static void alloc_uv_buf(uv_handle_t* handle, size_t suggested_size, uv_buf_t* buf) { grpc_core::ExecCtx _local_exec_ctx; grpc_tcp* tcp = (grpc_tcp*)handle->data; (void)suggested_size; - buf->base = (char*)GRPC_SLICE_START_PTR(tcp->read_slice); - buf->len = GRPC_SLICE_LENGTH(tcp->read_slice); + /* Before calling uv_read_start, we allocate a buffer with exactly one slice + * to tcp->read_slices and wait for the callback indicating that the + * allocation was successful. So slices[0] should always exist here */ + buf->base = (char*)GRPC_SLICE_START_PTR(tcp->read_slices->slices[0]); + buf->len = GRPC_SLICE_LENGTH(tcp->read_slices->slices[0]); + grpc_exec_ctx_finish(); +} + +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.enabled()) { + gpr_log(GPR_DEBUG, "TCP:%p call_cb %p %p:%p", tcp, cb, cb->cb, cb->cb_arg); + size_t i; + const char* str = grpc_error_string(error); + gpr_log(GPR_DEBUG, "read: error=%s", str); + + for (i = 0; i < tcp->read_slices->count; i++) { + char* dump = grpc_dump_slice(tcp->read_slices->slices[i], + GPR_DUMP_HEX | GPR_DUMP_ASCII); + gpr_log(GPR_DEBUG, "READ %p (peer=%s): %s", tcp, tcp->peer_string, dump); + gpr_free(dump); + } + } + tcp->read_slices = NULL; + tcp->read_cb = NULL; + GRPC_CLOSURE_RUN(exec_ctx, cb, error); } static void read_callback(uv_stream_t* stream, ssize_t nread, const uv_buf_t* buf) { - grpc_slice sub; grpc_error* error; grpc_core::ExecCtx _local_exec_ctx; grpc_tcp* tcp = (grpc_tcp*)stream->data; - grpc_closure* cb = tcp->read_cb; + grpc_slice_buffer garbage; if (nread == 0) { // Nothing happened. Wait for the next callback return; } TCP_UNREF(tcp, "read"); - tcp->read_cb = NULL; // TODO(murgatroid99): figure out what the return value here means uv_read_stop(stream); if (nread == UV_EOF) { error = GRPC_ERROR_CREATE_FROM_STATIC_STRING("EOF"); + grpc_slice_buffer_reset_and_unref_internal(&exec_ctx, tcp->read_slices); } else if (nread > 0) { // Successful read - sub = grpc_slice_sub_no_ref(tcp->read_slice, 0, (size_t)nread); - grpc_slice_buffer_add(tcp->read_slices, sub); - tcp->read_slice = alloc_read_slice(tcp->resource_user); error = GRPC_ERROR_NONE; - 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); - - for (i = 0; i < tcp->read_slices->count; i++) { - char* dump = grpc_dump_slice(tcp->read_slices->slices[i], - GPR_DUMP_HEX | GPR_DUMP_ASCII); - gpr_log(GPR_DEBUG, "READ %p (peer=%s): %s", tcp, tcp->peer_string, - dump); - gpr_free(dump); - } + if ((size_t)nread < tcp->read_slices->length) { + /* TODO(murgatroid99): Instead of discarding the unused part of the read + * buffer, reuse it as the next read buffer. */ + grpc_slice_buffer_init(&garbage); + grpc_slice_buffer_trim_end( + tcp->read_slices, tcp->read_slices->length - (size_t)nread, &garbage); + grpc_slice_buffer_reset_and_unref_internal(&exec_ctx, &garbage); } } else { // nread < 0: Error error = GRPC_ERROR_CREATE_FROM_STATIC_STRING("TCP Read failed"); + grpc_slice_buffer_reset_and_unref_internal(&exec_ctx, tcp->read_slices); + } + call_read_cb(tcp, error); + grpc_exec_ctx_finish(); +} + +static void tcp_read_allocation_done(void* tcpp, grpc_error* error) { + int status; + grpc_tcp* tcp = (grpc_tcp*)tcpp; + if (grpc_tcp_trace.enabled()) { + gpr_log(GPR_DEBUG, "TCP:%p read_allocation_done: %s", tcp, + grpc_error_string(error)); + } + if (error == GRPC_ERROR_NONE) { + status = + uv_read_start((uv_stream_t*)tcp->handle, alloc_uv_buf, read_callback); + if (status != 0) { + error = GRPC_ERROR_CREATE_FROM_STATIC_STRING("TCP Read failed at start"); + error = grpc_error_set_str( + error, GRPC_ERROR_STR_OS_ERROR, + grpc_slice_from_static_string(uv_strerror(status))); + } + } + if (error != GRPC_ERROR_NONE) { + grpc_slice_buffer_reset_and_unref_internal(tcp->read_slices); + call_read_cb(tcp, GRPC_ERROR_REF(error)); + TCP_UNREF(tcp, "read"); + } + if (grpc_tcp_trace.enabled()) { + const char* str = grpc_error_string(error); + gpr_log(GPR_DEBUG, "Initiating read on %p: error=%s", tcp, str); } - GRPC_CLOSURE_SCHED(cb, error); } static void uv_endpoint_read(grpc_endpoint* ep, grpc_slice_buffer* read_slices, grpc_closure* cb) { grpc_tcp* tcp = (grpc_tcp*)ep; - int status; - grpc_error* error = GRPC_ERROR_NONE; GRPC_UV_ASSERT_SAME_THREAD(); GPR_ASSERT(tcp->read_cb == NULL); tcp->read_cb = cb; tcp->read_slices = read_slices; grpc_slice_buffer_reset_and_unref_internal(read_slices); TCP_REF(tcp, "read"); - // TODO(murgatroid99): figure out what the return value here means - status = - uv_read_start((uv_stream_t*)tcp->handle, alloc_uv_buf, read_callback); - if (status != 0) { - error = GRPC_ERROR_CREATE_FROM_STATIC_STRING("TCP Read failed at start"); - error = - grpc_error_set_str(error, GRPC_ERROR_STR_OS_ERROR, - grpc_slice_from_static_string(uv_strerror(status))); - GRPC_CLOSURE_SCHED(cb, error); - } - if (GRPC_TRACER_ON(grpc_tcp_trace)) { - const char* str = grpc_error_string(error); - gpr_log(GPR_DEBUG, "Initiating read on %p: error=%s", tcp, str); - } + grpc_resource_user_alloc_slices(&tcp->slice_allocator, + GRPC_TCP_DEFAULT_READ_SLICE_SIZE, 1, + tcp->read_slices); } static void write_callback(uv_write_t* req, int status) { @@ -213,14 +240,13 @@ static void write_callback(uv_write_t* req, int status) { } else { error = GRPC_ERROR_CREATE_FROM_STATIC_STRING("TCP Write failed"); } - if (GRPC_TRACER_ON(grpc_tcp_trace)) { + if (grpc_tcp_trace.enabled()) { const char* str = grpc_error_string(error); gpr_log(GPR_DEBUG, "write complete on %p: error=%s", tcp, str); } gpr_free(tcp->write_buffers); - grpc_resource_user_free(tcp->resource_user, - sizeof(uv_buf_t) * tcp->write_slices->count); GRPC_CLOSURE_SCHED(cb, error); + grpc_exec_ctx_finish(); } static void uv_endpoint_write(grpc_endpoint* ep, @@ -234,7 +260,7 @@ static void uv_endpoint_write(grpc_endpoint* ep, uv_write_t* write_req; GRPC_UV_ASSERT_SAME_THREAD(); - if (GRPC_TRACER_ON(grpc_tcp_trace)) { + if (grpc_tcp_trace.enabled()) { size_t j; for (j = 0; j < write_slices->count; j++) { @@ -264,8 +290,6 @@ static void uv_endpoint_write(grpc_endpoint* ep, tcp->write_cb = cb; buffer_count = (unsigned int)tcp->write_slices->count; buffers = (uv_buf_t*)gpr_malloc(sizeof(uv_buf_t) * buffer_count); - grpc_resource_user_alloc(tcp->resource_user, sizeof(uv_buf_t) * buffer_count, - NULL); for (i = 0; i < buffer_count; i++) { slice = &tcp->write_slices->slices[i]; buffers[i].base = (char*)GRPC_SLICE_START_PTR(*slice); @@ -310,7 +334,7 @@ static void shutdown_callback(uv_shutdown_t* req, int status) {} static void uv_endpoint_shutdown(grpc_endpoint* ep, grpc_error* why) { grpc_tcp* tcp = (grpc_tcp*)ep; if (!tcp->shutting_down) { - if (GRPC_TRACER_ON(grpc_tcp_trace)) { + if (grpc_tcp_trace.enabled()) { const char* str = grpc_error_string(why); gpr_log(GPR_DEBUG, "TCP %p shutdown why=%s", tcp->handle, str); } @@ -357,7 +381,7 @@ grpc_endpoint* grpc_tcp_create(uv_tcp_t* handle, grpc_tcp* tcp = (grpc_tcp*)gpr_malloc(sizeof(grpc_tcp)); grpc_core::ExecCtx _local_exec_ctx; - if (GRPC_TRACER_ON(grpc_tcp_trace)) { + if (grpc_tcp_trace.enabled()) { gpr_log(GPR_DEBUG, "Creating TCP endpoint %p", tcp); } @@ -371,8 +395,10 @@ grpc_endpoint* grpc_tcp_create(uv_tcp_t* handle, gpr_ref_init(&tcp->refcount, 1); tcp->peer_string = gpr_strdup(peer_string); tcp->shutting_down = false; + tcp->read_slices = NULL; tcp->resource_user = grpc_resource_user_create(resource_quota, peer_string); - tcp->read_slice = alloc_read_slice(tcp->resource_user); + grpc_resource_user_slice_allocator_init( + &tcp->slice_allocator, tcp->resource_user, tcp_read_allocation_done, tcp); /* Tell network status tracking code about the new endpoint */ grpc_network_status_register_endpoint(&tcp->base); diff --git a/src/core/lib/iomgr/tcp_uv.h b/src/core/lib/iomgr/tcp_uv.h index 708e8469e6..4b4da3608f 100644 --- a/src/core/lib/iomgr/tcp_uv.h +++ b/src/core/lib/iomgr/tcp_uv.h @@ -38,7 +38,7 @@ #include <uv.h> -extern grpc_tracer_flag grpc_tcp_trace; +extern grpc_core::TraceFlag grpc_tcp_trace; #define GRPC_TCP_DEFAULT_READ_SLICE_SIZE 8192 diff --git a/src/core/lib/iomgr/tcp_windows.cc b/src/core/lib/iomgr/tcp_windows.cc index ab9a8a5c1e..6d091b77bb 100644 --- a/src/core/lib/iomgr/tcp_windows.cc +++ b/src/core/lib/iomgr/tcp_windows.cc @@ -49,7 +49,7 @@ #define GRPC_FIONBIO FIONBIO #endif -grpc_tracer_flag grpc_tcp_trace = GRPC_TRACER_INITIALIZER(false, "tcp"); +grpc_core::TraceFlag grpc_tcp_trace(false, "tcp"); static grpc_error* set_non_block(SOCKET sock) { int status; @@ -123,7 +123,7 @@ static void tcp_free(grpc_tcp* tcp) { #define TCP_REF(tcp, reason) tcp_ref((tcp), (reason), __FILE__, __LINE__) static void tcp_unref(grpc_tcp* tcp, const char* reason, const char* file, int line) { - if (GRPC_TRACER_ON(grpc_tcp_trace)) { + if (grpc_tcp_trace.enabled()) { gpr_atm val = gpr_atm_no_barrier_load(&tcp->refcount.count); gpr_log(file, line, GPR_LOG_SEVERITY_DEBUG, "TCP unref %p : %s %" PRIdPTR " -> %" PRIdPTR, tcp, reason, val, @@ -136,7 +136,7 @@ static void tcp_unref(grpc_tcp* tcp, const char* reason, const char* file, static void tcp_ref(grpc_tcp* tcp, const char* reason, const char* file, int line) { - if (GRPC_TRACER_ON(grpc_tcp_trace)) { + if (grpc_tcp_trace.enabled()) { gpr_atm val = gpr_atm_no_barrier_load(&tcp->refcount.count); gpr_log(file, line, GPR_LOG_SEVERITY_DEBUG, "TCP ref %p : %s %" PRIdPTR " -> %" PRIdPTR, tcp, reason, val, diff --git a/src/core/lib/iomgr/timer_generic.cc b/src/core/lib/iomgr/timer_generic.cc index 861a85577b..103144eb3b 100644 --- a/src/core/lib/iomgr/timer_generic.cc +++ b/src/core/lib/iomgr/timer_generic.cc @@ -42,11 +42,8 @@ #define MIN_QUEUE_WINDOW_DURATION 0.01 #define MAX_QUEUE_WINDOW_DURATION 1 -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"); -} +grpc_core::TraceFlag grpc_timer_trace(false, "timer"); +grpc_core::TraceFlag grpc_timer_check_trace(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 @@ -252,8 +249,6 @@ void grpc_timer_list_init() { g_shared_mutables.min_timer = grpc_core::ExecCtx::Get()->Now(); 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 < g_num_shards; i++) { timer_shard* shard = &g_shards[i]; @@ -338,7 +333,7 @@ void grpc_timer_init(grpc_timer* timer, grpc_millis deadline, timer->hash_table_next = nullptr; #endif - if (GRPC_TRACER_ON(grpc_timer_trace)) { + if (grpc_timer_trace.enabled()) { gpr_log(GPR_DEBUG, "TIMER %p: SET %" PRIdPTR " now %" PRIdPTR " call %p[%p]", timer, deadline, grpc_core::ExecCtx::Get()->Now(), closure, closure->cb); @@ -374,7 +369,7 @@ void grpc_timer_init(grpc_timer* timer, grpc_millis deadline, timer->heap_index = INVALID_HEAP_INDEX; list_join(&shard->list, timer); } - if (GRPC_TRACER_ON(grpc_timer_trace)) { + if (grpc_timer_trace.enabled()) { gpr_log(GPR_DEBUG, " .. add to shard %d with queue_deadline_cap=%" PRIdPTR " => is_first_timer=%s", @@ -396,7 +391,7 @@ void grpc_timer_init(grpc_timer* timer, grpc_millis deadline, grpc_timer_check. */ if (is_first_timer) { gpr_mu_lock(&g_shared_mutables.mu); - if (GRPC_TRACER_ON(grpc_timer_trace)) { + if (grpc_timer_trace.enabled()) { gpr_log(GPR_DEBUG, " .. old shard min_deadline=%" PRIdPTR, shard->min_deadline); } @@ -426,7 +421,7 @@ void grpc_timer_cancel(grpc_timer* timer) { timer_shard* shard = &g_shards[GPR_HASH_POINTER(timer, g_num_shards)]; gpr_mu_lock(&shard->mu); - if (GRPC_TRACER_ON(grpc_timer_trace)) { + if (grpc_timer_trace.enabled()) { gpr_log(GPR_DEBUG, "TIMER %p: CANCEL pending=%s", timer, timer->pending ? "true" : "false"); } @@ -467,7 +462,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_TRACER_ON(grpc_timer_check_trace)) { + if (grpc_timer_check_trace.enabled()) { gpr_log(GPR_DEBUG, " .. shard[%d]->queue_deadline_cap --> %" PRIdPTR, (int)(shard - g_shards), shard->queue_deadline_cap); } @@ -475,7 +470,7 @@ static int refill_heap(timer_shard* shard, gpr_atm now) { next = timer->next; if (timer->deadline < shard->queue_deadline_cap) { - if (GRPC_TRACER_ON(grpc_timer_check_trace)) { + if (grpc_timer_check_trace.enabled()) { gpr_log(GPR_DEBUG, " .. add timer with deadline %" PRIdPTR " to heap", timer->deadline); } @@ -492,7 +487,7 @@ 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_TRACER_ON(grpc_timer_check_trace)) { + if (grpc_timer_check_trace.enabled()) { gpr_log(GPR_DEBUG, " .. shard[%d]: heap_empty=%s", (int)(shard - g_shards), grpc_timer_heap_is_empty(&shard->heap) ? "true" : "false"); @@ -502,13 +497,13 @@ static grpc_timer* pop_one(timer_shard* shard, gpr_atm now) { if (!refill_heap(shard, now)) return nullptr; } timer = grpc_timer_heap_top(&shard->heap); - if (GRPC_TRACER_ON(grpc_timer_check_trace)) { + if (grpc_timer_check_trace.enabled()) { gpr_log(GPR_DEBUG, " .. check top timer deadline=%" PRIdPTR " now=%" PRIdPTR, timer->deadline, now); } if (timer->deadline > now) return nullptr; - if (GRPC_TRACER_ON(grpc_timer_trace)) { + if (grpc_timer_trace.enabled()) { gpr_log(GPR_DEBUG, "TIMER %p: FIRE %" PRIdPTR "ms late via %s scheduler", timer, now - timer->deadline, timer->closure->scheduler->vtable->name); @@ -532,7 +527,7 @@ static size_t pop_timers(timer_shard* shard, gpr_atm now, } *new_min_deadline = compute_min_deadline(shard); gpr_mu_unlock(&shard->mu); - if (GRPC_TRACER_ON(grpc_timer_check_trace)) { + if (grpc_timer_check_trace.enabled()) { gpr_log(GPR_DEBUG, " .. shard[%d] popped %" PRIdPTR, (int)(shard - g_shards), n); } @@ -555,7 +550,7 @@ static grpc_timer_check_result run_some_expired_timers(gpr_atm now, gpr_mu_lock(&g_shared_mutables.mu); result = GRPC_TIMERS_CHECKED_AND_EMPTY; - if (GRPC_TRACER_ON(grpc_timer_check_trace)) { + if (grpc_timer_check_trace.enabled()) { gpr_log(GPR_DEBUG, " .. shard[%d]->min_deadline = %" PRIdPTR, (int)(g_shard_queue[0] - g_shards), g_shard_queue[0]->min_deadline); @@ -572,7 +567,7 @@ static grpc_timer_check_result run_some_expired_timers(gpr_atm now, result = GRPC_TIMERS_FIRED; } - if (GRPC_TRACER_ON(grpc_timer_check_trace)) { + if (grpc_timer_check_trace.enabled()) { gpr_log(GPR_DEBUG, " .. result --> %d" ", shard[%d]->min_deadline %" PRIdPTR " --> %" PRIdPTR @@ -616,7 +611,7 @@ grpc_timer_check_result grpc_timer_check(grpc_millis* next) { if (next != nullptr) { *next = GPR_MIN(*next, min_timer); } - if (GRPC_TRACER_ON(grpc_timer_check_trace)) { + if (grpc_timer_check_trace.enabled()) { gpr_log(GPR_DEBUG, "TIMER CHECK SKIP: now=%" PRIdPTR " min_timer=%" PRIdPTR, now, min_timer); @@ -630,7 +625,7 @@ grpc_timer_check_result grpc_timer_check(grpc_millis* next) { : GRPC_ERROR_CREATE_FROM_STATIC_STRING("Shutting down timer system"); // tracing - if (GRPC_TRACER_ON(grpc_timer_check_trace)) { + if (grpc_timer_check_trace.enabled()) { char* next_str; if (next == nullptr) { next_str = gpr_strdup("NULL"); @@ -648,7 +643,7 @@ grpc_timer_check_result grpc_timer_check(grpc_millis* next) { grpc_timer_check_result r = run_some_expired_timers(now, next, shutdown_error); // tracing - if (GRPC_TRACER_ON(grpc_timer_check_trace)) { + if (grpc_timer_check_trace.enabled()) { char* next_str; if (next == nullptr) { next_str = gpr_strdup("NULL"); diff --git a/src/core/lib/iomgr/timer_manager.cc b/src/core/lib/iomgr/timer_manager.cc index dfa381a1f1..0e1fa0c049 100644 --- a/src/core/lib/iomgr/timer_manager.cc +++ b/src/core/lib/iomgr/timer_manager.cc @@ -33,7 +33,7 @@ typedef struct completed_thread { struct completed_thread* next; } completed_thread; -extern "C" grpc_tracer_flag grpc_timer_check_trace; +extern grpc_core::TraceFlag grpc_timer_check_trace; // global mutex static gpr_mu g_mu; @@ -81,7 +81,7 @@ static void start_timer_thread_and_unlock(void) { ++g_waiter_count; ++g_thread_count; gpr_mu_unlock(&g_mu); - if (GRPC_TRACER_ON(grpc_timer_check_trace)) { + if (grpc_timer_check_trace.enabled()) { gpr_log(GPR_DEBUG, "Spawn timer thread"); } gpr_thd_options opt = gpr_thd_options_default(); @@ -114,7 +114,7 @@ static void run_some_timers() { // 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) { - if (GRPC_TRACER_ON(grpc_timer_check_trace)) { + if (grpc_timer_check_trace.enabled()) { gpr_log(GPR_DEBUG, "kick untimed waiter"); } gpr_cv_signal(&g_cv_wait); @@ -122,7 +122,7 @@ static void run_some_timers() { gpr_mu_unlock(&g_mu); } // without our lock, flush the exec_ctx - if (GRPC_TRACER_ON(grpc_timer_check_trace)) { + if (grpc_timer_check_trace.enabled()) { gpr_log(GPR_DEBUG, "flush exec_ctx"); } grpc_core::ExecCtx::Get()->Flush(); @@ -177,7 +177,7 @@ static bool wait_until(grpc_millis next) { g_has_timed_waiter = true; g_timed_waiter_deadline = next; - if (GRPC_TRACER_ON(grpc_timer_check_trace)) { + if (grpc_timer_check_trace.enabled()) { grpc_millis wait_time = next - grpc_core::ExecCtx::Get()->Now(); gpr_log(GPR_DEBUG, "sleep for a %" PRIdPTR " milliseconds", wait_time); @@ -187,15 +187,14 @@ static bool wait_until(grpc_millis next) { } } - if (GRPC_TRACER_ON(grpc_timer_check_trace) && - next == GRPC_MILLIS_INF_FUTURE) { + if (grpc_timer_check_trace.enabled() && next == GRPC_MILLIS_INF_FUTURE) { gpr_log(GPR_DEBUG, "sleep until kicked"); } gpr_cv_wait(&g_cv_wait, &g_mu, grpc_millis_to_timespec(next, GPR_CLOCK_REALTIME)); - if (GRPC_TRACER_ON(grpc_timer_check_trace)) { + if (grpc_timer_check_trace.enabled()) { gpr_log(GPR_DEBUG, "wait ended: was_timed:%d kicked:%d", my_timed_waiter_generation == g_timed_waiter_generation, g_kicked); @@ -224,9 +223,6 @@ static void timer_main_loop() { for (;;) { grpc_millis next = GRPC_MILLIS_INF_FUTURE; grpc_core::ExecCtx::Get()->InvalidateNow(); - /* Calibrate g_start_time in exec_ctx.cc with a regular interval in case the - * system clock has changed */ - grpc_exec_ctx_maybe_update_start_time(); // check timer state, updates next to the next time to run a check switch (grpc_timer_check(&next)) { @@ -243,7 +239,7 @@ static void timer_main_loop() { Consequently, we can just sleep forever here and be happy at some saved wakeup cycles. */ - if (GRPC_TRACER_ON(grpc_timer_check_trace)) { + if (grpc_timer_check_trace.enabled()) { gpr_log(GPR_DEBUG, "timers not checked: expect another thread to"); } next = GRPC_MILLIS_INF_FUTURE; @@ -269,7 +265,7 @@ static void timer_thread_cleanup(completed_thread* ct) { ct->next = g_completed_threads; g_completed_threads = ct; gpr_mu_unlock(&g_mu); - if (GRPC_TRACER_ON(grpc_timer_check_trace)) { + if (grpc_timer_check_trace.enabled()) { gpr_log(GPR_DEBUG, "End timer thread"); } } @@ -311,18 +307,18 @@ void grpc_timer_manager_init(void) { static void stop_threads(void) { gpr_mu_lock(&g_mu); - if (GRPC_TRACER_ON(grpc_timer_check_trace)) { + if (grpc_timer_check_trace.enabled()) { gpr_log(GPR_DEBUG, "stop timer threads: threaded=%d", g_threaded); } if (g_threaded) { g_threaded = false; gpr_cv_broadcast(&g_cv_wait); - if (GRPC_TRACER_ON(grpc_timer_check_trace)) { + if (grpc_timer_check_trace.enabled()) { gpr_log(GPR_DEBUG, "num timer threads: %d", g_thread_count); } while (g_thread_count > 0) { gpr_cv_wait(&g_cv_shutdown, &g_mu, gpr_inf_future(GPR_CLOCK_REALTIME)); - if (GRPC_TRACER_ON(grpc_timer_check_trace)) { + if (grpc_timer_check_trace.enabled()) { gpr_log(GPR_DEBUG, "num timer threads: %d", g_thread_count); } gc_completed_threads(); diff --git a/src/core/lib/iomgr/timer_uv.cc b/src/core/lib/iomgr/timer_uv.cc index 1432eba51e..879890603c 100644 --- a/src/core/lib/iomgr/timer_uv.cc +++ b/src/core/lib/iomgr/timer_uv.cc @@ -29,11 +29,8 @@ #include <uv.h> -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"); -} +grpc_core::TraceFlag grpc_timer_trace(false, "timer"); +grpc_core::TraceFlag grpc_timer_check_trace(false, "timer_check"); static void timer_close_callback(uv_handle_t* handle) { gpr_free(handle); } diff --git a/src/core/lib/security/context/security_context.cc b/src/core/lib/security/context/security_context.cc index 7d9b4fbcb2..d4efdac583 100644 --- a/src/core/lib/security/context/security_context.cc +++ b/src/core/lib/security/context/security_context.cc @@ -29,10 +29,8 @@ #include <grpc/support/log.h> #include <grpc/support/string_util.h> -#ifndef NDEBUG -grpc_tracer_flag grpc_trace_auth_context_refcount = - GRPC_TRACER_INITIALIZER(false, "auth_context_refcount"); -#endif +grpc_core::DebugOnlyTraceFlag grpc_trace_auth_context_refcount( + false, "auth_context_refcount"); /* --- grpc_call --- */ @@ -134,7 +132,7 @@ grpc_auth_context* grpc_auth_context_ref(grpc_auth_context* ctx, const char* file, int line, const char* reason) { if (ctx == nullptr) return nullptr; - if (GRPC_TRACER_ON(grpc_trace_auth_context_refcount)) { + if (grpc_trace_auth_context_refcount.enabled()) { gpr_atm val = gpr_atm_no_barrier_load(&ctx->refcount.count); gpr_log(file, line, GPR_LOG_SEVERITY_DEBUG, "AUTH_CONTEXT:%p ref %" PRIdPTR " -> %" PRIdPTR " %s", ctx, val, @@ -152,7 +150,7 @@ grpc_auth_context* grpc_auth_context_ref(grpc_auth_context* ctx) { void grpc_auth_context_unref(grpc_auth_context* ctx, const char* file, int line, const char* reason) { if (ctx == nullptr) return; - if (GRPC_TRACER_ON(grpc_trace_auth_context_refcount)) { + if (grpc_trace_auth_context_refcount.enabled()) { gpr_atm val = gpr_atm_no_barrier_load(&ctx->refcount.count); gpr_log(file, line, GPR_LOG_SEVERITY_DEBUG, "AUTH_CONTEXT:%p unref %" PRIdPTR " -> %" PRIdPTR " %s", ctx, val, diff --git a/src/core/lib/security/context/security_context.h b/src/core/lib/security/context/security_context.h index 4f049c4a3b..5b27d1a42e 100644 --- a/src/core/lib/security/context/security_context.h +++ b/src/core/lib/security/context/security_context.h @@ -22,9 +22,7 @@ #include "src/core/lib/iomgr/pollset.h" #include "src/core/lib/security/credentials/credentials.h" -#ifndef NDEBUG -extern grpc_tracer_flag grpc_trace_auth_context_refcount; -#endif +extern grpc_core::DebugOnlyTraceFlag grpc_trace_auth_context_refcount; #ifdef __cplusplus extern "C" { diff --git a/src/core/lib/security/credentials/jwt/jwt_credentials.cc b/src/core/lib/security/credentials/jwt/jwt_credentials.cc index df26b4e2c0..3eaedeea4e 100644 --- a/src/core/lib/security/credentials/jwt/jwt_credentials.cc +++ b/src/core/lib/security/credentials/jwt/jwt_credentials.cc @@ -167,7 +167,7 @@ static char* redact_private_key(const char* json_key) { grpc_call_credentials* grpc_service_account_jwt_access_credentials_create( const char* json_key, gpr_timespec token_lifetime, void* reserved) { - if (GRPC_TRACER_ON(grpc_api_trace)) { + if (grpc_api_trace.enabled()) { char* clean_json = redact_private_key(json_key); gpr_log(GPR_INFO, "grpc_service_account_jwt_access_credentials_create(" diff --git a/src/core/lib/security/credentials/oauth2/oauth2_credentials.cc b/src/core/lib/security/credentials/oauth2/oauth2_credentials.cc index f95df0f021..0fe368d3f2 100644 --- a/src/core/lib/security/credentials/oauth2/oauth2_credentials.cc +++ b/src/core/lib/security/credentials/oauth2/oauth2_credentials.cc @@ -461,7 +461,7 @@ grpc_call_credentials* grpc_google_refresh_token_credentials_create( const char* json_refresh_token, void* reserved) { grpc_auth_refresh_token token = grpc_auth_refresh_token_create_from_string(json_refresh_token); - if (GRPC_TRACER_ON(grpc_api_trace)) { + if (grpc_api_trace.enabled()) { char* loggable_token = create_loggable_refresh_token(&token); gpr_log(GPR_INFO, "grpc_refresh_token_credentials_create(json_refresh_token=%s, " diff --git a/src/core/lib/security/credentials/plugin/plugin_credentials.cc b/src/core/lib/security/credentials/plugin/plugin_credentials.cc index 27a762d9a3..a75ae033b9 100644 --- a/src/core/lib/security/credentials/plugin/plugin_credentials.cc +++ b/src/core/lib/security/credentials/plugin/plugin_credentials.cc @@ -31,8 +31,7 @@ #include "src/core/lib/surface/api_trace.h" #include "src/core/lib/surface/validate_metadata.h" -grpc_tracer_flag grpc_plugin_credentials_trace = - GRPC_TRACER_INITIALIZER(false, "plugin_credentials"); +grpc_core::TraceFlag grpc_plugin_credentials_trace(false, "plugin_credentials"); static void plugin_destruct(grpc_call_credentials* creds) { grpc_plugin_credentials* c = (grpc_plugin_credentials*)creds; @@ -120,7 +119,7 @@ static void plugin_md_request_metadata_ready(void* request, GRPC_EXEC_CTX_FLAG_THREAD_RESOURCE_LOOP); grpc_plugin_credentials_pending_request* r = (grpc_plugin_credentials_pending_request*)request; - if (GRPC_TRACER_ON(grpc_plugin_credentials_trace)) { + if (grpc_plugin_credentials_trace.enabled()) { gpr_log(GPR_INFO, "plugin_credentials[%p]: request %p: plugin returned " "asynchronously", @@ -133,7 +132,7 @@ static void plugin_md_request_metadata_ready(void* request, grpc_error* error = process_plugin_result(r, md, num_md, status, error_details); GRPC_CLOSURE_SCHED(r->on_request_metadata, error); - } else if (GRPC_TRACER_ON(grpc_plugin_credentials_trace)) { + } else if (grpc_plugin_credentials_trace.enabled()) { gpr_log(GPR_INFO, "plugin_credentials[%p]: request %p: plugin was previously " "cancelled", @@ -167,7 +166,7 @@ static bool plugin_get_request_metadata(grpc_call_credentials* creds, c->pending_requests = pending_request; gpr_mu_unlock(&c->mu); // Invoke the plugin. The callback holds a ref to us. - if (GRPC_TRACER_ON(grpc_plugin_credentials_trace)) { + if (grpc_plugin_credentials_trace.enabled()) { gpr_log(GPR_INFO, "plugin_credentials[%p]: request %p: invoking plugin", c, pending_request); } @@ -180,7 +179,7 @@ static bool plugin_get_request_metadata(grpc_call_credentials* creds, plugin_md_request_metadata_ready, pending_request, creds_md, &num_creds_md, &status, &error_details)) { - if (GRPC_TRACER_ON(grpc_plugin_credentials_trace)) { + if (grpc_plugin_credentials_trace.enabled()) { gpr_log(GPR_INFO, "plugin_credentials[%p]: request %p: plugin will return " "asynchronously", @@ -195,7 +194,7 @@ static bool plugin_get_request_metadata(grpc_call_credentials* creds, // asynchronously by plugin_cancel_get_request_metadata(), so return // false. Otherwise, process the result. if (pending_request->cancelled) { - if (GRPC_TRACER_ON(grpc_plugin_credentials_trace)) { + if (grpc_plugin_credentials_trace.enabled()) { gpr_log(GPR_INFO, "plugin_credentials[%p]: request %p was cancelled, error " "will be returned asynchronously", @@ -203,7 +202,7 @@ static bool plugin_get_request_metadata(grpc_call_credentials* creds, } retval = false; } else { - if (GRPC_TRACER_ON(grpc_plugin_credentials_trace)) { + if (grpc_plugin_credentials_trace.enabled()) { gpr_log(GPR_INFO, "plugin_credentials[%p]: request %p: plugin returned " "synchronously", @@ -232,7 +231,7 @@ static void plugin_cancel_get_request_metadata( c->pending_requests; pending_request != nullptr; pending_request = pending_request->next) { if (pending_request->md_array == md_array) { - if (GRPC_TRACER_ON(grpc_plugin_credentials_trace)) { + if (grpc_plugin_credentials_trace.enabled()) { gpr_log(GPR_INFO, "plugin_credentials[%p]: cancelling request %p", c, pending_request); } diff --git a/src/core/lib/security/credentials/plugin/plugin_credentials.h b/src/core/lib/security/credentials/plugin/plugin_credentials.h index fc0955c695..e1467b0824 100644 --- a/src/core/lib/security/credentials/plugin/plugin_credentials.h +++ b/src/core/lib/security/credentials/plugin/plugin_credentials.h @@ -21,7 +21,7 @@ #include "src/core/lib/security/credentials/credentials.h" -extern grpc_tracer_flag grpc_plugin_credentials_trace; +extern grpc_core::TraceFlag grpc_plugin_credentials_trace; struct grpc_plugin_credentials; diff --git a/src/core/lib/security/transport/secure_endpoint.cc b/src/core/lib/security/transport/secure_endpoint.cc index 4ea40e7fa1..e5c089de9c 100644 --- a/src/core/lib/security/transport/secure_endpoint.cc +++ b/src/core/lib/security/transport/secure_endpoint.cc @@ -61,8 +61,7 @@ typedef struct { gpr_refcount ref; } secure_endpoint; -grpc_tracer_flag grpc_trace_secure_endpoint = - GRPC_TRACER_INITIALIZER(false, "secure_endpoint"); +grpc_core::TraceFlag grpc_trace_secure_endpoint(false, "secure_endpoint"); static void destroy(secure_endpoint* secure_ep) { secure_endpoint* ep = secure_ep; @@ -85,7 +84,7 @@ static void destroy(secure_endpoint* secure_ep) { secure_endpoint_ref((ep), (reason), __FILE__, __LINE__) static void secure_endpoint_unref(secure_endpoint* ep, const char* reason, const char* file, int line) { - if (GRPC_TRACER_ON(grpc_trace_secure_endpoint)) { + if (grpc_trace_secure_endpoint.enabled()) { gpr_atm val = gpr_atm_no_barrier_load(&ep->ref.count); gpr_log(file, line, GPR_LOG_SEVERITY_DEBUG, "SECENDP unref %p : %s %" PRIdPTR " -> %" PRIdPTR, ep, reason, val, @@ -98,7 +97,7 @@ static void secure_endpoint_unref(secure_endpoint* ep, const char* reason, static void secure_endpoint_ref(secure_endpoint* ep, const char* reason, const char* file, int line) { - if (GRPC_TRACER_ON(grpc_trace_secure_endpoint)) { + if (grpc_trace_secure_endpoint.enabled()) { gpr_atm val = gpr_atm_no_barrier_load(&ep->ref.count); gpr_log(file, line, GPR_LOG_SEVERITY_DEBUG, "SECENDP ref %p : %s %" PRIdPTR " -> %" PRIdPTR, ep, reason, val, @@ -127,7 +126,7 @@ static void flush_read_staging_buffer(secure_endpoint* ep, uint8_t** cur, } static void call_read_cb(secure_endpoint* ep, grpc_error* error) { - if (GRPC_TRACER_ON(grpc_trace_secure_endpoint)) { + if (grpc_trace_secure_endpoint.enabled()) { size_t i; for (i = 0; i < ep->read_buffer->count; i++) { char* data = grpc_dump_slice(ep->read_buffer->slices[i], @@ -263,7 +262,7 @@ static void endpoint_write(grpc_endpoint* secure_ep, grpc_slice_buffer* slices, grpc_slice_buffer_reset_and_unref_internal(&ep->output_buffer); - if (GRPC_TRACER_ON(grpc_trace_secure_endpoint)) { + if (grpc_trace_secure_endpoint.enabled()) { for (i = 0; i < slices->count; i++) { char* data = grpc_dump_slice(slices->slices[i], GPR_DUMP_HEX | GPR_DUMP_ASCII); diff --git a/src/core/lib/security/transport/secure_endpoint.h b/src/core/lib/security/transport/secure_endpoint.h index db8233f6e6..92c457421b 100644 --- a/src/core/lib/security/transport/secure_endpoint.h +++ b/src/core/lib/security/transport/secure_endpoint.h @@ -29,7 +29,7 @@ extern "C" { struct tsi_frame_protector; struct tsi_zero_copy_grpc_protector; -extern grpc_tracer_flag grpc_trace_secure_endpoint; +extern grpc_core::TraceFlag grpc_trace_secure_endpoint; /* Takes ownership of protector, zero_copy_protector, and to_wrap, and refs * leftover_slices. If zero_copy_protector is not NULL, protector will never be diff --git a/src/core/lib/security/transport/security_connector.cc b/src/core/lib/security/transport/security_connector.cc index 0c99268578..fd139714da 100644 --- a/src/core/lib/security/transport/security_connector.cc +++ b/src/core/lib/security/transport/security_connector.cc @@ -44,10 +44,8 @@ #include "src/core/tsi/ssl_transport_security.h" #include "src/core/tsi/transport_security_adapter.h" -#ifndef NDEBUG -grpc_tracer_flag grpc_trace_security_connector_refcount = - GRPC_TRACER_INITIALIZER(false, "security_connector_refcount"); -#endif +grpc_core::DebugOnlyTraceFlag grpc_trace_security_connector_refcount( + false, "security_connector_refcount"); /* -- Constants. -- */ @@ -197,7 +195,7 @@ grpc_security_connector* grpc_security_connector_ref( grpc_security_connector* sc, const char* file, int line, const char* reason) { if (sc == nullptr) return nullptr; - if (GRPC_TRACER_ON(grpc_trace_security_connector_refcount)) { + if (grpc_trace_security_connector_refcount.enabled()) { gpr_atm val = gpr_atm_no_barrier_load(&sc->refcount.count); gpr_log(file, line, GPR_LOG_SEVERITY_DEBUG, "SECURITY_CONNECTOR:%p ref %" PRIdPTR " -> %" PRIdPTR " %s", sc, @@ -217,7 +215,7 @@ void grpc_security_connector_unref(grpc_security_connector* sc, const char* file, int line, const char* reason) { if (sc == nullptr) return; - if (GRPC_TRACER_ON(grpc_trace_security_connector_refcount)) { + if (grpc_trace_security_connector_refcount.enabled()) { gpr_atm val = gpr_atm_no_barrier_load(&sc->refcount.count); gpr_log(file, line, GPR_LOG_SEVERITY_DEBUG, "SECURITY_CONNECTOR:%p unref %" PRIdPTR " -> %" PRIdPTR " %s", sc, diff --git a/src/core/lib/security/transport/security_connector.h b/src/core/lib/security/transport/security_connector.h index bb66cae422..78c9656a8b 100644 --- a/src/core/lib/security/transport/security_connector.h +++ b/src/core/lib/security/transport/security_connector.h @@ -33,9 +33,7 @@ extern "C" { #endif -#ifndef NDEBUG -extern grpc_tracer_flag grpc_trace_security_connector_refcount; -#endif +extern grpc_core::DebugOnlyTraceFlag grpc_trace_security_connector_refcount; /* --- status enum. --- */ diff --git a/src/core/lib/slice/slice_internal.h b/src/core/lib/slice/slice_internal.h index be66b8ab00..a3205290cb 100644 --- a/src/core/lib/slice/slice_internal.h +++ b/src/core/lib/slice/slice_internal.h @@ -31,6 +31,8 @@ extern "C" { grpc_slice grpc_slice_ref_internal(grpc_slice slice); void grpc_slice_unref_internal(grpc_slice slice); void grpc_slice_buffer_reset_and_unref_internal(grpc_slice_buffer* sb); +void grpc_slice_buffer_partial_unref_internal(grpc_slice_buffer* sb, + size_t idx); void grpc_slice_buffer_destroy_internal(grpc_slice_buffer* sb); /* Check if a slice is interned */ diff --git a/src/core/lib/support/abstract.h b/src/core/lib/support/abstract.h new file mode 100644 index 0000000000..5498769a7d --- /dev/null +++ b/src/core/lib/support/abstract.h @@ -0,0 +1,29 @@ +/* + * + * Copyright 2017 gRPC authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * + */ + +#ifndef GRPC_CORE_LIB_SUPPORT_ABSTRACT_H +#define GRPC_CORE_LIB_SUPPORT_ABSTRACT_H + +// This is needed to support abstract base classes in the c core. Since gRPC +// doesn't have a c++ runtime, it will hit a linker error on delete unless +// we define a virtual operator delete. See this blog for more info: +// https://eli.thegreenplace.net/2015/c-deleting-destructors-and-virtual-operator-delete/ +#define GRPC_ABSTRACT_BASE_CLASS \ + static void operator delete(void* p) { abort(); } + +#endif /* GRPC_CORE_LIB_SUPPORT_ABSTRACT_H */ diff --git a/src/core/lib/support/cpu_posix.cc b/src/core/lib/support/cpu_posix.cc index 503a96b4c8..bca14a0c12 100644 --- a/src/core/lib/support/cpu_posix.cc +++ b/src/core/lib/support/cpu_posix.cc @@ -18,21 +18,23 @@ #include <grpc/support/port_platform.h> -#ifdef GPR_CPU_POSIX +#if defined(GPR_CPU_POSIX) #include <errno.h> +#include <pthread.h> #include <string.h> #include <unistd.h> +#include <grpc/support/alloc.h> #include <grpc/support/cpu.h> #include <grpc/support/log.h> #include <grpc/support/sync.h> #include <grpc/support/useful.h> -static __thread char magic_thread_local; - static long ncpus = 0; +static pthread_key_t thread_id_key; + static void init_ncpus() { ncpus = sysconf(_SC_NPROCESSORS_ONLN); if (ncpus < 1 || ncpus > INT32_MAX) { @@ -47,12 +49,32 @@ unsigned gpr_cpu_num_cores(void) { return (unsigned)ncpus; } +static void delete_thread_id(void* value) { + if (value) { + gpr_free(value); + } +} + +static void init_thread_id_key(void) { + pthread_key_create(&thread_id_key, delete_thread_id); +} + unsigned gpr_cpu_current_cpu(void) { /* NOTE: there's no way I know to return the actual cpu index portably... most code that's using this is using it to shard across work queues though, so here we use thread identity instead to achieve a similar though not identical effect */ - return (unsigned)GPR_HASH_POINTER(&magic_thread_local, gpr_cpu_num_cores()); + static gpr_once once = GPR_ONCE_INIT; + gpr_once_init(&once, init_thread_id_key); + + unsigned int* thread_id = + static_cast<unsigned int*>(pthread_getspecific(thread_id_key)); + if (thread_id == nullptr) { + thread_id = static_cast<unsigned int*>(gpr_malloc(sizeof(unsigned int))); + pthread_setspecific(thread_id_key, thread_id); + } + + return (unsigned)GPR_HASH_POINTER(thread_id, gpr_cpu_num_cores()); } #endif /* GPR_CPU_POSIX */ diff --git a/src/core/lib/support/manual_constructor.h b/src/core/lib/support/manual_constructor.h index d753cf98a0..fda7653dbc 100644 --- a/src/core/lib/support/manual_constructor.h +++ b/src/core/lib/support/manual_constructor.h @@ -22,12 +22,147 @@ // manually construct a region of memory with some type #include <stddef.h> +#include <stdlib.h> #include <new> #include <type_traits> #include <utility> +#include <grpc/support/log.h> + namespace grpc_core { +// this contains templated helpers needed to implement the ManualConstructors +// in this file. +namespace manual_ctor_impl { + +// is_one_of returns true it a class, Member, is present in a variadic list of +// classes, List. +template <class Member, class... List> +class is_one_of; + +template <class Member, class... List> +class is_one_of<Member, Member, List...> { + public: + static constexpr const bool value = true; +}; + +template <class Member, class A, class... List> +class is_one_of<Member, A, List...> { + public: + static constexpr const bool value = is_one_of<Member, List...>::value; +}; + +template <class Member> +class is_one_of<Member> { + public: + static constexpr const bool value = false; +}; + +// max_size_of returns sizeof(Type) for the largest type in the variadic list +// of classes, Types. +template <class... Types> +class max_size_of; + +template <class A> +class max_size_of<A> { + public: + static constexpr const size_t value = sizeof(A); +}; + +template <class A, class... B> +class max_size_of<A, B...> { + public: + static constexpr const size_t value = sizeof(A) > max_size_of<B...>::value + ? sizeof(A) + : max_size_of<B...>::value; +}; + +// max_size_of returns alignof(Type) for the largest type in the variadic list +// of classes, Types. +template <class... Types> +class max_align_of; + +template <class A> +class max_align_of<A> { + public: + static constexpr const size_t value = alignof(A); +}; + +template <class A, class... B> +class max_align_of<A, B...> { + public: + static constexpr const size_t value = alignof(A) > max_align_of<B...>::value + ? alignof(A) + : max_align_of<B...>::value; +}; + +} // namespace manual_ctor_impl + +template <class BaseType, class... DerivedTypes> +class PolymorphicManualConstructor { + public: + // No constructor or destructor because one of the most useful uses of + // this class is as part of a union, and members of a union could not have + // constructors or destructors till C++11. And, anyway, the whole point of + // this class is to bypass constructor and destructor. + + BaseType* get() { return reinterpret_cast<BaseType*>(&space_); } + const BaseType* get() const { + return reinterpret_cast<const BaseType*>(&space_); + } + + BaseType* operator->() { return get(); } + const BaseType* operator->() const { return get(); } + + BaseType& operator*() { return *get(); } + const BaseType& operator*() const { return *get(); } + + template <class DerivedType> + void Init() { + FinishInit(new (&space_) DerivedType); + } + + // Init() constructs the Type instance using the given arguments + // (which are forwarded to Type's constructor). + // + // Note that Init() with no arguments performs default-initialization, + // not zero-initialization (i.e it behaves the same as "new Type;", not + // "new Type();"), so it will leave non-class types uninitialized. + template <class DerivedType, typename... Ts> + void Init(Ts&&... args) { + FinishInit(new (&space_) DerivedType(std::forward<Ts>(args)...)); + } + + // Init() that is equivalent to copy and move construction. + // Enables usage like this: + // ManualConstructor<std::vector<int>> v; + // v.Init({1, 2, 3}); + template <class DerivedType> + void Init(const DerivedType& x) { + FinishInit(new (&space_) DerivedType(x)); + } + template <class DerivedType> + void Init(DerivedType&& x) { + FinishInit(new (&space_) DerivedType(std::move(x))); + } + + void Destroy() { get()->~BaseType(); } + + private: + template <class DerivedType> + void FinishInit(DerivedType* p) { + static_assert( + manual_ctor_impl::is_one_of<DerivedType, DerivedTypes...>::value, + "DerivedType must be one of the predeclared DerivedTypes"); + GPR_ASSERT(reinterpret_cast<BaseType*>(static_cast<DerivedType*>(p)) == p); + } + + typename std::aligned_storage< + grpc_core::manual_ctor_impl::max_size_of<DerivedTypes...>::value, + grpc_core::manual_ctor_impl::max_align_of<DerivedTypes...>::value>::type + space_; +}; + template <typename Type> class ManualConstructor { public: diff --git a/src/core/lib/surface/alarm.cc b/src/core/lib/surface/alarm.cc index 00d609b80d..9a13a7b323 100644 --- a/src/core/lib/surface/alarm.cc +++ b/src/core/lib/surface/alarm.cc @@ -27,10 +27,8 @@ #include "src/core/lib/iomgr/timer.h" #include "src/core/lib/surface/completion_queue.h" -#ifndef NDEBUG -grpc_tracer_flag grpc_trace_alarm_refcount = - GRPC_TRACER_INITIALIZER(false, "alarm_refcount"); -#endif +grpc_core::DebugOnlyTraceFlag grpc_trace_alarm_refcount(false, + "alarm_refcount"); struct grpc_alarm { gpr_refcount refs; @@ -59,7 +57,7 @@ static void alarm_unref(grpc_alarm* alarm) { #ifndef NDEBUG static void alarm_ref_dbg(grpc_alarm* alarm, const char* reason, const char* file, int line) { - if (GRPC_TRACER_ON(grpc_trace_alarm_refcount)) { + if (grpc_trace_alarm_refcount.enabled()) { gpr_atm val = gpr_atm_no_barrier_load(&alarm->refs.count); gpr_log(file, line, GPR_LOG_SEVERITY_DEBUG, "Alarm:%p ref %" PRIdPTR " -> %" PRIdPTR " %s", alarm, val, @@ -71,7 +69,7 @@ static void alarm_ref_dbg(grpc_alarm* alarm, const char* reason, static void alarm_unref_dbg(grpc_alarm* alarm, const char* reason, const char* file, int line) { - if (GRPC_TRACER_ON(grpc_trace_alarm_refcount)) { + if (grpc_trace_alarm_refcount.enabled()) { gpr_atm val = gpr_atm_no_barrier_load(&alarm->refs.count); gpr_log(file, line, GPR_LOG_SEVERITY_DEBUG, "Alarm:%p Unref %" PRIdPTR " -> %" PRIdPTR " %s", alarm, val, @@ -102,7 +100,7 @@ grpc_alarm* grpc_alarm_create(void* reserved) { grpc_alarm* alarm = (grpc_alarm*)gpr_malloc(sizeof(grpc_alarm)); #ifndef NDEBUG - if (GRPC_TRACER_ON(grpc_trace_alarm_refcount)) { + if (grpc_trace_alarm_refcount.enabled()) { gpr_log(GPR_DEBUG, "Alarm:%p created (ref: 1)", alarm); } #endif diff --git a/src/core/lib/surface/alarm_internal.h b/src/core/lib/surface/alarm_internal.h index 136b60547f..2ee3a311f7 100644 --- a/src/core/lib/surface/alarm_internal.h +++ b/src/core/lib/surface/alarm_internal.h @@ -22,14 +22,14 @@ #include <grpc/support/log.h> #include "src/core/lib/debug/trace.h" +extern grpc_core::DebugOnlyTraceFlag grpc_trace_alarm_refcount; + #ifdef __cplusplus extern "C" { #endif #ifndef NDEBUG -extern grpc_tracer_flag grpc_trace_alarm_refcount; - #define GRPC_ALARM_REF(a, reason) alarm_ref_dbg(a, reason, __FILE__, __LINE__) #define GRPC_ALARM_UNREF(a, reason) \ alarm_unref_dbg(a, reason, __FILE__, __LINE__) diff --git a/src/core/lib/surface/api_trace.cc b/src/core/lib/surface/api_trace.cc index 56973303da..7ab836a9ba 100644 --- a/src/core/lib/surface/api_trace.cc +++ b/src/core/lib/surface/api_trace.cc @@ -19,4 +19,4 @@ #include "src/core/lib/surface/api_trace.h" #include "src/core/lib/debug/trace.h" -grpc_tracer_flag grpc_api_trace = GRPC_TRACER_INITIALIZER(false, "api"); +grpc_core::TraceFlag grpc_api_trace(false, "api"); diff --git a/src/core/lib/surface/api_trace.h b/src/core/lib/surface/api_trace.h index 105abdf629..a4e11ce154 100644 --- a/src/core/lib/surface/api_trace.h +++ b/src/core/lib/surface/api_trace.h @@ -22,11 +22,7 @@ #include <grpc/support/log.h> #include "src/core/lib/debug/trace.h" -#ifdef __cplusplus -extern "C" { -#endif - -extern grpc_tracer_flag grpc_api_trace; +extern grpc_core::TraceFlag grpc_api_trace; /* Provide unwrapping macros because we're in C89 and variadic macros weren't introduced until C99... */ @@ -47,12 +43,8 @@ extern grpc_tracer_flag grpc_api_trace; /* Due to the limitations of C89's preprocessor, the arity of the var-arg list 'nargs' must be specified. */ #define GRPC_API_TRACE(fmt, nargs, args) \ - if (GRPC_TRACER_ON(grpc_api_trace)) { \ + if (grpc_api_trace.enabled()) { \ gpr_log(GPR_INFO, fmt GRPC_API_TRACE_UNWRAP##nargs args); \ } -#ifdef __cplusplus -} -#endif - #endif /* GRPC_CORE_LIB_SURFACE_API_TRACE_H */ diff --git a/src/core/lib/surface/call.cc b/src/core/lib/surface/call.cc index 1caa717195..e4c8f01505 100644 --- a/src/core/lib/surface/call.cc +++ b/src/core/lib/surface/call.cc @@ -234,6 +234,7 @@ struct grpc_call { struct { grpc_status_code* status; grpc_slice* status_details; + const char** error_string; } client; struct { int* cancelled; @@ -259,10 +260,8 @@ struct grpc_call { gpr_atm recv_state; }; -grpc_tracer_flag grpc_call_error_trace = - GRPC_TRACER_INITIALIZER(false, "call_error"); -grpc_tracer_flag grpc_compression_trace = - GRPC_TRACER_INITIALIZER(false, "compression"); +grpc_core::TraceFlag grpc_call_error_trace(false, "call_error"); +grpc_core::TraceFlag grpc_compression_trace(false, "compression"); #define CALL_STACK_FROM_CALL(call) ((grpc_call_stack*)((call) + 1)) #define CALL_FROM_CALL_STACK(call_stack) (((grpc_call*)(call_stack)) - 1) @@ -280,10 +279,9 @@ static void cancel_with_error(grpc_call* c, status_source source, grpc_error* error); static void destroy_call(void* call_stack, grpc_error* error); static void receiving_slice_ready(void* bctlp, grpc_error* error); -static void get_final_status(grpc_call* call, - void (*set_value)(grpc_status_code code, - void* user_data), - void* set_value_user_data, grpc_slice* details); +static void get_final_status( + grpc_call* call, void (*set_value)(grpc_status_code code, void* user_data), + void* set_value_user_data, grpc_slice* details, const char** error_string); static void set_status_value_directly(grpc_status_code status, void* dest); static void set_status_from_error(grpc_call* call, status_source source, grpc_error* error); @@ -541,7 +539,7 @@ static void destroy_call(void* call, grpc_error* error) { } get_final_status(c, set_status_value_directly, &c->final_info.final_status, - nullptr); + nullptr, c->final_info.error_string); c->final_info.stats.latency = gpr_time_sub(gpr_now(GPR_CLOCK_MONOTONIC), c->start_time); @@ -724,10 +722,11 @@ static void cancel_with_status(grpc_call* c, status_source source, static bool get_final_status_from( grpc_call* call, grpc_error* error, bool allow_ok_status, void (*set_value)(grpc_status_code code, void* user_data), - void* set_value_user_data, grpc_slice* details) { + void* set_value_user_data, grpc_slice* details, const char** error_string) { grpc_status_code code; grpc_slice slice = grpc_empty_slice(); - grpc_error_get_status(error, call->send_deadline, &code, &slice, nullptr); + grpc_error_get_status(error, call->send_deadline, &code, &slice, nullptr, + error_string); if (code == GRPC_STATUS_OK && !allow_ok_status) { return false; } @@ -739,16 +738,15 @@ static bool get_final_status_from( return true; } -static void get_final_status(grpc_call* call, - void (*set_value)(grpc_status_code code, - void* user_data), - void* set_value_user_data, grpc_slice* details) { +static void get_final_status( + grpc_call* call, void (*set_value)(grpc_status_code code, void* user_data), + void* set_value_user_data, grpc_slice* details, const char** error_string) { int i; received_status status[STATUS_SOURCE_COUNT]; for (i = 0; i < STATUS_SOURCE_COUNT; i++) { status[i] = unpack_received_status(gpr_atm_acq_load(&call->status[i])); } - if (GRPC_TRACER_ON(grpc_call_error_trace)) { + if (grpc_call_error_trace.enabled()) { gpr_log(GPR_DEBUG, "get_final_status %s", call->is_client ? "CLI" : "SVR"); for (i = 0; i < STATUS_SOURCE_COUNT; i++) { if (status[i].is_set) { @@ -765,7 +763,8 @@ static void get_final_status(grpc_call* call, if (status[i].is_set && grpc_error_has_clear_grpc_status(status[i].error)) { if (get_final_status_from(call, status[i].error, allow_ok_status != 0, - set_value, set_value_user_data, details)) { + set_value, set_value_user_data, details, + error_string)) { return; } } @@ -774,7 +773,8 @@ static void get_final_status(grpc_call* call, for (i = 0; i < STATUS_SOURCE_COUNT; i++) { if (status[i].is_set) { if (get_final_status_from(call, status[i].error, allow_ok_status != 0, - set_value, set_value_user_data, details)) { + set_value, set_value_user_data, details, + error_string)) { return; } } @@ -1309,10 +1309,11 @@ static void post_batch_completion(batch_control* bctl) { if (call->is_client) { get_final_status(call, set_status_value_directly, call->final_op.client.status, - call->final_op.client.status_details); + call->final_op.client.status_details, + call->final_op.client.error_string); } else { get_final_status(call, set_cancelled_value, - call->final_op.server.cancelled, nullptr); + call->final_op.server.cancelled, nullptr, nullptr); } GRPC_ERROR_UNREF(error); @@ -1400,7 +1401,7 @@ static void receiving_slice_ready(void* bctlp, grpc_error* error) { } if (error != GRPC_ERROR_NONE) { - if (GRPC_TRACER_ON(grpc_trace_operation_failures)) { + if (grpc_trace_operation_failures.enabled()) { GRPC_LOG_IF_ERROR("receiving_slice_ready", GRPC_ERROR_REF(error)); } grpc_byte_stream_destroy(call->receiving_stream); @@ -1499,7 +1500,7 @@ static void validate_filtered_metadata(batch_control* bctl) { GPR_ASSERT(call->stream_encodings_accepted_by_peer != 0); if (!GPR_BITGET(call->stream_encodings_accepted_by_peer, call->incoming_stream_compression_algorithm)) { - if (GRPC_TRACER_ON(grpc_compression_trace)) { + if (grpc_compression_trace.enabled()) { const char* algo_name = nullptr; grpc_stream_compression_algorithm_name( call->incoming_stream_compression_algorithm, &algo_name); @@ -1542,7 +1543,7 @@ static void validate_filtered_metadata(batch_control* bctl) { GPR_ASSERT(call->encodings_accepted_by_peer != 0); if (!GPR_BITGET(call->encodings_accepted_by_peer, call->incoming_compression_algorithm)) { - if (GRPC_TRACER_ON(grpc_compression_trace)) { + if (grpc_compression_trace.enabled()) { const char* algo_name = nullptr; grpc_compression_algorithm_name(call->incoming_compression_algorithm, &algo_name); @@ -1954,6 +1955,8 @@ static grpc_call_error call_start_batch(grpc_call* call, const grpc_op* ops, call->final_op.client.status = op->data.recv_status_on_client.status; call->final_op.client.status_details = op->data.recv_status_on_client.status_details; + call->final_op.client.error_string = + op->data.recv_status_on_client.error_string; stream_op->recv_trailing_metadata = true; stream_op->collect_stats = true; stream_op_payload->recv_trailing_metadata.recv_trailing_metadata = diff --git a/src/core/lib/surface/call.h b/src/core/lib/surface/call.h index 61cc6520da..49e391864f 100644 --- a/src/core/lib/surface/call.h +++ b/src/core/lib/surface/call.h @@ -96,8 +96,7 @@ void grpc_call_context_set(grpc_call* call, grpc_context_index elem, void* grpc_call_context_get(grpc_call* call, grpc_context_index elem); #define GRPC_CALL_LOG_BATCH(sev, call, ops, nops, tag) \ - if (GRPC_TRACER_ON(grpc_api_trace)) \ - grpc_call_log_batch(sev, call, ops, nops, tag) + if (grpc_api_trace.enabled()) grpc_call_log_batch(sev, call, ops, nops, tag) uint8_t grpc_call_is_client(grpc_call* call); @@ -106,8 +105,8 @@ uint8_t grpc_call_is_client(grpc_call* call); grpc_compression_algorithm grpc_call_compression_for_level( grpc_call* call, grpc_compression_level level); -extern grpc_tracer_flag grpc_call_error_trace; -extern grpc_tracer_flag grpc_compression_trace; +extern grpc_core::TraceFlag grpc_call_error_trace; +extern grpc_core::TraceFlag grpc_compression_trace; #ifdef __cplusplus } diff --git a/src/core/lib/surface/completion_queue.cc b/src/core/lib/surface/completion_queue.cc index c84f35c508..958318f814 100644 --- a/src/core/lib/surface/completion_queue.cc +++ b/src/core/lib/surface/completion_queue.cc @@ -40,14 +40,9 @@ #include "src/core/lib/surface/call.h" #include "src/core/lib/surface/event_string.h" -grpc_tracer_flag grpc_trace_operation_failures = - GRPC_TRACER_INITIALIZER(false, "op_failure"); -#ifndef NDEBUG -grpc_tracer_flag grpc_trace_pending_tags = - GRPC_TRACER_INITIALIZER(false, "pending_tags"); -grpc_tracer_flag grpc_trace_cq_refcount = - GRPC_TRACER_INITIALIZER(false, "cq_refcount"); -#endif +grpc_core::TraceFlag grpc_trace_operation_failures(false, "op_failure"); +grpc_core::DebugOnlyTraceFlag grpc_trace_pending_tags(false, "pending_tags"); +grpc_core::DebugOnlyTraceFlag grpc_trace_cq_refcount(false, "cq_refcount"); // Specifies a cq thread local cache. // The first event that occurs on a thread @@ -325,18 +320,15 @@ static const cq_vtable g_cq_vtable[] = { #define POLLSET_FROM_CQ(cq) \ ((grpc_pollset*)(cq->vtable->data_size + (char*)DATA_FROM_CQ(cq))) -grpc_tracer_flag grpc_cq_pluck_trace = - GRPC_TRACER_INITIALIZER(true, "queue_pluck"); -grpc_tracer_flag grpc_cq_event_timeout_trace = - GRPC_TRACER_INITIALIZER(true, "queue_timeout"); - -#define GRPC_SURFACE_TRACE_RETURNED_EVENT(cq, event) \ - if (GRPC_TRACER_ON(grpc_api_trace) && \ - (GRPC_TRACER_ON(grpc_cq_pluck_trace) || \ - (event)->type != GRPC_QUEUE_TIMEOUT)) { \ - char* _ev = grpc_event_string(event); \ - gpr_log(GPR_INFO, "RETURN_EVENT[%p]: %s", cq, _ev); \ - gpr_free(_ev); \ +grpc_core::TraceFlag grpc_cq_pluck_trace(true, "queue_pluck"); +grpc_core::TraceFlag grpc_cq_event_timeout_trace(true, "queue_timeout"); + +#define GRPC_SURFACE_TRACE_RETURNED_EVENT(cq, event) \ + if (grpc_api_trace.enabled() && (grpc_cq_pluck_trace.enabled() || \ + (event)->type != GRPC_QUEUE_TIMEOUT)) { \ + char* _ev = grpc_event_string(event); \ + gpr_log(GPR_INFO, "RETURN_EVENT[%p]: %s", cq, _ev); \ + gpr_free(_ev); \ } static void on_pollset_shutdown_done(void* cq, grpc_error* error); @@ -513,7 +505,7 @@ int grpc_get_cq_poll_num(grpc_completion_queue* cq) { #ifndef NDEBUG void grpc_cq_internal_ref(grpc_completion_queue* cq, const char* reason, const char* file, int line) { - if (GRPC_TRACER_ON(grpc_trace_cq_refcount)) { + if (grpc_trace_cq_refcount.enabled()) { gpr_atm val = gpr_atm_no_barrier_load(&cq->owning_refs.count); gpr_log(file, line, GPR_LOG_SEVERITY_DEBUG, "CQ:%p ref %" PRIdPTR " -> %" PRIdPTR " %s", cq, val, val + 1, @@ -533,7 +525,7 @@ static void on_pollset_shutdown_done(void* arg, grpc_error* error) { #ifndef NDEBUG void grpc_cq_internal_unref(grpc_completion_queue* cq, const char* reason, const char* file, int line) { - if (GRPC_TRACER_ON(grpc_trace_cq_refcount)) { + if (grpc_trace_cq_refcount.enabled()) { gpr_atm val = gpr_atm_no_barrier_load(&cq->owning_refs.count); gpr_log(file, line, GPR_LOG_SEVERITY_DEBUG, "CQ:%p unref %" PRIdPTR " -> %" PRIdPTR " %s", cq, val, val - 1, @@ -632,16 +624,14 @@ static void cq_end_op_for_next(grpc_completion_queue* cq, void* tag, void* done_arg, grpc_cq_completion* storage) { GPR_TIMER_BEGIN("cq_end_op_for_next", 0); - if (GRPC_TRACER_ON(grpc_api_trace) || - (GRPC_TRACER_ON(grpc_trace_operation_failures) && - error != GRPC_ERROR_NONE)) { + if (grpc_api_trace.enabled() || + (grpc_trace_operation_failures.enabled() && error != GRPC_ERROR_NONE)) { const char* errmsg = grpc_error_string(error); GRPC_API_TRACE( "cq_end_op_for_next(cq=%p, tag=%p, error=%s, " "done=%p, done_arg=%p, storage=%p)", 6, (cq, tag, errmsg, done, done_arg, storage)); - if (GRPC_TRACER_ON(grpc_trace_operation_failures) && - error != GRPC_ERROR_NONE) { + if (grpc_trace_operation_failures.enabled() && error != GRPC_ERROR_NONE) { gpr_log(GPR_ERROR, "Operation failed: tag=%p, error=%s", tag, errmsg); } } @@ -719,16 +709,14 @@ static void cq_end_op_for_pluck(grpc_completion_queue* cq, void* tag, GPR_TIMER_BEGIN("cq_end_op_for_pluck", 0); - if (GRPC_TRACER_ON(grpc_api_trace) || - (GRPC_TRACER_ON(grpc_trace_operation_failures) && - error != GRPC_ERROR_NONE)) { + if (grpc_api_trace.enabled() || + (grpc_trace_operation_failures.enabled() && error != GRPC_ERROR_NONE)) { const char* errmsg = grpc_error_string(error); GRPC_API_TRACE( "cq_end_op_for_pluck(cq=%p, tag=%p, error=%s, " "done=%p, done_arg=%p, storage=%p)", 6, (cq, tag, errmsg, done, done_arg, storage)); - if (GRPC_TRACER_ON(grpc_trace_operation_failures) && - error != GRPC_ERROR_NONE) { + if (grpc_trace_operation_failures.enabled() && error != GRPC_ERROR_NONE) { gpr_log(GPR_ERROR, "Operation failed: tag=%p, error=%s", tag, errmsg); } } @@ -829,7 +817,7 @@ class ExecCtxNext : public grpc_core::ExecCtx { #ifndef NDEBUG static void dump_pending_tags(grpc_completion_queue* cq) { - if (!GRPC_TRACER_ON(grpc_trace_pending_tags)) return; + if (!grpc_trace_pending_tags.enabled()) return; gpr_strvec v; gpr_strvec_init(&v); @@ -1098,7 +1086,7 @@ static grpc_event cq_pluck(grpc_completion_queue* cq, void* tag, GPR_TIMER_BEGIN("grpc_completion_queue_pluck", 0); - if (GRPC_TRACER_ON(grpc_cq_pluck_trace)) { + if (grpc_cq_pluck_trace.enabled()) { GRPC_API_TRACE( "grpc_completion_queue_pluck(" "cq=%p, tag=%p, " diff --git a/src/core/lib/surface/completion_queue.h b/src/core/lib/surface/completion_queue.h index 5ddafd9eb5..3a96ab59b0 100644 --- a/src/core/lib/surface/completion_queue.h +++ b/src/core/lib/surface/completion_queue.h @@ -27,14 +27,11 @@ /* These trace flags default to 1. The corresponding lines are only traced if grpc_api_trace is also truthy */ -extern grpc_tracer_flag grpc_cq_pluck_trace; -extern grpc_tracer_flag grpc_cq_event_timeout_trace; -extern grpc_tracer_flag grpc_trace_operation_failures; - -#ifndef NDEBUG -extern grpc_tracer_flag grpc_trace_pending_tags; -extern grpc_tracer_flag grpc_trace_cq_refcount; -#endif +extern grpc_core::TraceFlag grpc_cq_pluck_trace; +extern grpc_core::TraceFlag grpc_cq_event_timeout_trace; +extern grpc_core::TraceFlag grpc_trace_operation_failures; +extern grpc_core::DebugOnlyTraceFlag grpc_trace_pending_tags; +extern grpc_core::DebugOnlyTraceFlag grpc_trace_cq_refcount; #ifdef __cplusplus extern "C" { diff --git a/src/core/lib/surface/init.cc b/src/core/lib/surface/init.cc index b64e862b40..678f837cb7 100644 --- a/src/core/lib/surface/init.cc +++ b/src/core/lib/surface/init.cc @@ -124,30 +124,6 @@ void grpc_init(void) { grpc_slice_intern_init(); grpc_mdctx_global_init(); grpc_channel_init_init(); - grpc_register_tracer(&grpc_api_trace); - grpc_register_tracer(&grpc_trace_channel); - grpc_register_tracer(&grpc_connectivity_state_trace); - grpc_register_tracer(&grpc_trace_channel_stack_builder); - grpc_register_tracer(&grpc_http1_trace); - grpc_register_tracer(&grpc_cq_pluck_trace); // default on - grpc_register_tracer(&grpc_call_combiner_trace); - grpc_register_tracer(&grpc_combiner_trace); - grpc_register_tracer(&grpc_server_channel_trace); - grpc_register_tracer(&grpc_bdp_estimator_trace); - grpc_register_tracer(&grpc_cq_event_timeout_trace); // default on - grpc_register_tracer(&grpc_trace_operation_failures); - grpc_register_tracer(&grpc_resource_quota_trace); - grpc_register_tracer(&grpc_call_error_trace); -#ifndef NDEBUG - grpc_register_tracer(&grpc_trace_pending_tags); - grpc_register_tracer(&grpc_trace_alarm_refcount); - grpc_register_tracer(&grpc_trace_cq_refcount); - grpc_register_tracer(&grpc_trace_closure); - grpc_register_tracer(&grpc_trace_error_refcount); - grpc_register_tracer(&grpc_trace_stream_refcount); - grpc_register_tracer(&grpc_trace_fd_refcount); - grpc_register_tracer(&grpc_trace_metadata); -#endif grpc_security_pre_init(); grpc_iomgr_init(); gpr_timers_global_init(); diff --git a/src/core/lib/surface/init_secure.cc b/src/core/lib/surface/init_secure.cc index c8935c767d..75ed9faef0 100644 --- a/src/core/lib/surface/init_secure.cc +++ b/src/core/lib/surface/init_secure.cc @@ -24,6 +24,7 @@ #include <string.h> #include "src/core/lib/debug/trace.h" +#include "src/core/lib/security/context/security_context.h" #include "src/core/lib/security/credentials/credentials.h" #include "src/core/lib/security/credentials/plugin/plugin_credentials.h" #include "src/core/lib/security/transport/auth_filters.h" @@ -33,18 +34,7 @@ #include "src/core/lib/surface/channel_init.h" #include "src/core/tsi/transport_security_interface.h" -#ifndef NDEBUG -#include "src/core/lib/security/context/security_context.h" -#endif - -void grpc_security_pre_init(void) { - grpc_register_tracer(&grpc_trace_secure_endpoint); - grpc_register_tracer(&tsi_tracing_enabled); -#ifndef NDEBUG - grpc_register_tracer(&grpc_trace_auth_context_refcount); - grpc_register_tracer(&grpc_trace_security_connector_refcount); -#endif -} +void grpc_security_pre_init(void) {} static bool maybe_prepend_client_auth_filter( grpc_channel_stack_builder* builder, void* arg) { @@ -85,7 +75,4 @@ void grpc_register_security_filters(void) { maybe_prepend_server_auth_filter, nullptr); } -void grpc_security_init() { - grpc_security_register_handshaker_factories(); - grpc_register_tracer(&grpc_plugin_credentials_trace); -} +void grpc_security_init() { grpc_security_register_handshaker_factories(); } diff --git a/src/core/lib/surface/lame_client.cc b/src/core/lib/surface/lame_client.cc index 67188b46fc..d79f9a9ab2 100644 --- a/src/core/lib/surface/lame_client.cc +++ b/src/core/lib/surface/lame_client.cc @@ -25,7 +25,6 @@ #include "src/core/lib/support/atomic.h" -extern "C" { #include "src/core/lib/channel/channel_stack.h" #include "src/core/lib/support/string.h" #include "src/core/lib/surface/api_trace.h" @@ -33,7 +32,6 @@ extern "C" { #include "src/core/lib/surface/channel.h" #include "src/core/lib/surface/lame_client.h" #include "src/core/lib/transport/static_metadata.h" -} namespace grpc_core { diff --git a/src/core/lib/surface/server.cc b/src/core/lib/surface/server.cc index 49f453569c..e003b5410b 100644 --- a/src/core/lib/surface/server.cc +++ b/src/core/lib/surface/server.cc @@ -59,8 +59,7 @@ typedef struct registered_method registered_method; typedef enum { BATCH_CALL, REGISTERED_CALL } requested_call_type; -grpc_tracer_flag grpc_server_channel_trace = - GRPC_TRACER_INITIALIZER(false, "server_channel"); +grpc_core::TraceFlag grpc_server_channel_trace(false, "server_channel"); typedef struct requested_call { gpr_mpscq_node request_link; /* must be first */ @@ -343,14 +342,8 @@ static void request_matcher_kill_requests(grpc_server* server, grpc_error* error) { requested_call* rc; for (size_t i = 0; i < server->cq_count; i++) { - /* Here we know: - 1. no requests are being added (since the server is shut down) - 2. no other threads are pulling (since the shut down process is single - threaded) - So, we can ignore the queue lock and just pop, with the guarantee that a - NULL returned here truly means that the queue is empty */ - while ((rc = (requested_call*)gpr_mpscq_pop( - &rm->requests_per_cq[i].queue)) != nullptr) { + while ((rc = (requested_call*)gpr_locked_mpscq_pop( + &rm->requests_per_cq[i])) != nullptr) { fail_call(server, i, rc, GRPC_ERROR_REF(error)); } } @@ -425,7 +418,7 @@ static void destroy_channel(channel_data* chand, grpc_error* error) { GRPC_CLOSURE_INIT(&chand->finish_destroy_channel_closure, finish_destroy_channel, chand, grpc_schedule_on_exec_ctx); - if (GRPC_TRACER_ON(grpc_server_channel_trace) && error != GRPC_ERROR_NONE) { + if (grpc_server_channel_trace.enabled() && error != GRPC_ERROR_NONE) { const char* msg = grpc_error_string(error); gpr_log(GPR_INFO, "Disconnected client: %s", msg); } diff --git a/src/core/lib/surface/server.h b/src/core/lib/surface/server.h index dfd777e8a5..3f53870075 100644 --- a/src/core/lib/surface/server.h +++ b/src/core/lib/surface/server.h @@ -31,7 +31,7 @@ extern "C" { extern const grpc_channel_filter grpc_server_top_filter; /** Lightweight tracing of server channel state */ -extern grpc_tracer_flag grpc_server_channel_trace; +extern grpc_core::TraceFlag grpc_server_channel_trace; /* Add a listener to the server: when the server starts, it will call start, and when it shuts down, it will call destroy */ diff --git a/src/core/lib/transport/bdp_estimator.cc b/src/core/lib/transport/bdp_estimator.cc index d2b6e5db25..5fcc62ec43 100644 --- a/src/core/lib/transport/bdp_estimator.cc +++ b/src/core/lib/transport/bdp_estimator.cc @@ -23,8 +23,7 @@ #include <grpc/support/useful.h> -grpc_tracer_flag grpc_bdp_estimator_trace = - GRPC_TRACER_INITIALIZER(false, "bdp_estimator"); +grpc_core::TraceFlag grpc_bdp_estimator_trace(false, "bdp_estimator"); namespace grpc_core { @@ -44,7 +43,7 @@ grpc_millis BdpEstimator::CompletePing() { double dt = (double)dt_ts.tv_sec + 1e-9 * (double)dt_ts.tv_nsec; double bw = dt > 0 ? ((double)accumulator_ / dt) : 0; int start_inter_ping_delay = inter_ping_delay_; - if (GRPC_TRACER_ON(grpc_bdp_estimator_trace)) { + if (grpc_bdp_estimator_trace.enabled()) { gpr_log(GPR_DEBUG, "bdp[%s]:complete acc=%" PRId64 " est=%" PRId64 " dt=%lf bw=%lfMbs bw_est=%lfMbs", @@ -55,7 +54,7 @@ grpc_millis BdpEstimator::CompletePing() { if (accumulator_ > 2 * estimate_ / 3 && bw > bw_est_) { estimate_ = GPR_MAX(accumulator_, estimate_ * 2); bw_est_ = bw; - if (GRPC_TRACER_ON(grpc_bdp_estimator_trace)) { + if (grpc_bdp_estimator_trace.enabled()) { gpr_log(GPR_DEBUG, "bdp[%s]: estimate increased to %" PRId64, name_, estimate_); } @@ -72,7 +71,7 @@ grpc_millis BdpEstimator::CompletePing() { } if (start_inter_ping_delay != inter_ping_delay_) { stable_estimate_count_ = 0; - if (GRPC_TRACER_ON(grpc_bdp_estimator_trace)) { + if (grpc_bdp_estimator_trace.enabled()) { gpr_log(GPR_DEBUG, "bdp[%s]:update_inter_time to %dms", name_, inter_ping_delay_); } diff --git a/src/core/lib/transport/bdp_estimator.h b/src/core/lib/transport/bdp_estimator.h index 431d4b845c..e703af121c 100644 --- a/src/core/lib/transport/bdp_estimator.h +++ b/src/core/lib/transport/bdp_estimator.h @@ -31,7 +31,7 @@ #include "src/core/lib/debug/trace.h" #include "src/core/lib/iomgr/exec_ctx.h" -extern grpc_tracer_flag grpc_bdp_estimator_trace; +extern grpc_core::TraceFlag grpc_bdp_estimator_trace; namespace grpc_core { @@ -49,7 +49,7 @@ class BdpEstimator { // grpc_bdp_estimator_add_incoming_bytes once a ping has been scheduled by a // transport (but not necessarily started) void SchedulePing() { - if (GRPC_TRACER_ON(grpc_bdp_estimator_trace)) { + if (grpc_bdp_estimator_trace.enabled()) { gpr_log(GPR_DEBUG, "bdp[%s]:sched acc=%" PRId64 " est=%" PRId64, name_, accumulator_, estimate_); } @@ -62,7 +62,7 @@ class BdpEstimator { // once // the ping is on the wire void StartPing() { - if (GRPC_TRACER_ON(grpc_bdp_estimator_trace)) { + if (grpc_bdp_estimator_trace.enabled()) { gpr_log(GPR_DEBUG, "bdp[%s]:start acc=%" PRId64 " est=%" PRId64, name_, accumulator_, estimate_); } diff --git a/src/core/lib/transport/connectivity_state.cc b/src/core/lib/transport/connectivity_state.cc index 51ee20d7b0..c42cc9c8d0 100644 --- a/src/core/lib/transport/connectivity_state.cc +++ b/src/core/lib/transport/connectivity_state.cc @@ -24,8 +24,7 @@ #include <grpc/support/log.h> #include <grpc/support/string_util.h> -grpc_tracer_flag grpc_connectivity_state_trace = - GRPC_TRACER_INITIALIZER(false, "connectivity_state"); +grpc_core::TraceFlag grpc_connectivity_state_trace(false, "connectivity_state"); const char* grpc_connectivity_state_name(grpc_connectivity_state state) { switch (state) { @@ -77,7 +76,7 @@ grpc_connectivity_state grpc_connectivity_state_check( grpc_connectivity_state cur = (grpc_connectivity_state)gpr_atm_no_barrier_load( &tracker->current_state_atm); - if (GRPC_TRACER_ON(grpc_connectivity_state_trace)) { + if (grpc_connectivity_state_trace.enabled()) { gpr_log(GPR_DEBUG, "CONWATCH: %p %s: get %s", tracker, tracker->name, grpc_connectivity_state_name(cur)); } @@ -89,7 +88,7 @@ grpc_connectivity_state grpc_connectivity_state_get( grpc_connectivity_state cur = (grpc_connectivity_state)gpr_atm_no_barrier_load( &tracker->current_state_atm); - if (GRPC_TRACER_ON(grpc_connectivity_state_trace)) { + if (grpc_connectivity_state_trace.enabled()) { gpr_log(GPR_DEBUG, "CONWATCH: %p %s: get %s", tracker, tracker->name, grpc_connectivity_state_name(cur)); } @@ -110,7 +109,7 @@ bool grpc_connectivity_state_notify_on_state_change( grpc_connectivity_state cur = (grpc_connectivity_state)gpr_atm_no_barrier_load( &tracker->current_state_atm); - if (GRPC_TRACER_ON(grpc_connectivity_state_trace)) { + if (grpc_connectivity_state_trace.enabled()) { if (current == nullptr) { gpr_log(GPR_DEBUG, "CONWATCH: %p %s: unsubscribe notify=%p", tracker, tracker->name, notify); @@ -162,7 +161,7 @@ void grpc_connectivity_state_set(grpc_connectivity_state_tracker* tracker, (grpc_connectivity_state)gpr_atm_no_barrier_load( &tracker->current_state_atm); grpc_connectivity_state_watcher* w; - if (GRPC_TRACER_ON(grpc_connectivity_state_trace)) { + if (grpc_connectivity_state_trace.enabled()) { const char* error_string = grpc_error_string(error); gpr_log(GPR_DEBUG, "SET: %p %s: %s --> %s [%s] error=%p %s", tracker, tracker->name, grpc_connectivity_state_name(cur), @@ -189,7 +188,7 @@ void grpc_connectivity_state_set(grpc_connectivity_state_tracker* tracker, while ((w = tracker->watchers) != nullptr) { *w->current = state; tracker->watchers = w->next; - if (GRPC_TRACER_ON(grpc_connectivity_state_trace)) { + if (grpc_connectivity_state_trace.enabled()) { gpr_log(GPR_DEBUG, "NOTIFY: %p %s: %p", tracker, tracker->name, w->notify); } diff --git a/src/core/lib/transport/connectivity_state.h b/src/core/lib/transport/connectivity_state.h index 83ce99e6d6..440bd7a9fa 100644 --- a/src/core/lib/transport/connectivity_state.h +++ b/src/core/lib/transport/connectivity_state.h @@ -47,7 +47,7 @@ typedef struct { char* name; } grpc_connectivity_state_tracker; -extern grpc_tracer_flag grpc_connectivity_state_trace; +extern grpc_core::TraceFlag grpc_connectivity_state_trace; /** enum --> string conversion */ const char* grpc_connectivity_state_name(grpc_connectivity_state state); diff --git a/src/core/lib/transport/error_utils.cc b/src/core/lib/transport/error_utils.cc index 2db87b5d29..ffaf327081 100644 --- a/src/core/lib/transport/error_utils.cc +++ b/src/core/lib/transport/error_utils.cc @@ -18,6 +18,7 @@ #include "src/core/lib/transport/error_utils.h" +#include <grpc/support/string_util.h> #include "src/core/lib/iomgr/error_internal.h" #include "src/core/lib/transport/status_conversion.h" @@ -41,7 +42,8 @@ static grpc_error* recursively_find_error_with_field(grpc_error* error, void grpc_error_get_status(grpc_error* error, grpc_millis deadline, grpc_status_code* code, grpc_slice* slice, - grpc_http2_error_code* http_error) { + grpc_http2_error_code* http_error, + const char** error_string) { // Start with the parent error and recurse through the tree of children // until we find the first one that has a status code. grpc_error* found_error = @@ -68,6 +70,10 @@ void grpc_error_get_status(grpc_error* error, grpc_millis deadline, } if (code != nullptr) *code = status; + if (error_string != NULL && status != GRPC_STATUS_OK) { + *error_string = gpr_strdup(grpc_error_string(error)); + } + if (http_error != nullptr) { if (grpc_error_get_int(found_error, GRPC_ERROR_INT_HTTP2_ERROR, &integer)) { *http_error = (grpc_http2_error_code)integer; diff --git a/src/core/lib/transport/error_utils.h b/src/core/lib/transport/error_utils.h index 73ca7cf3ab..7d56282372 100644 --- a/src/core/lib/transport/error_utils.h +++ b/src/core/lib/transport/error_utils.h @@ -30,12 +30,14 @@ extern "C" { /// A utility function to get the status code and message to be returned /// to the application. If not set in the top-level message, looks /// through child errors until it finds the first one with these attributes. -/// All attributes are pulled from the same child error. If any of the -/// attributes (code, msg, http_status) are unneeded, they can be passed as +/// All attributes are pulled from the same child error. error_string will +/// be populated with the entire error string. If any of the attributes (code, +/// msg, http_status, error_string) are unneeded, they can be passed as /// NULL. void grpc_error_get_status(grpc_error* error, grpc_millis deadline, grpc_status_code* code, grpc_slice* slice, - grpc_http2_error_code* http_status); + grpc_http2_error_code* http_status, + const char** error_string); /// A utility function to check whether there is a clear status code that /// doesn't need to be guessed in \a error. This means that \a error or some diff --git a/src/core/lib/transport/metadata.cc b/src/core/lib/transport/metadata.cc index 7596afa59b..5f0673e014 100644 --- a/src/core/lib/transport/metadata.cc +++ b/src/core/lib/transport/metadata.cc @@ -48,9 +48,9 @@ * used to determine which kind of element a pointer refers to. */ +grpc_core::DebugOnlyTraceFlag grpc_trace_metadata(false, "metadata"); + #ifndef NDEBUG -grpc_tracer_flag grpc_trace_metadata = - GRPC_TRACER_INITIALIZER(false, "metadata"); #define DEBUG_ARGS , const char *file, int line #define FWD_DEBUG_ARGS , file, line #define REF_MD_LOCKED(shard, s) ref_md_locked((shard), (s), __FILE__, __LINE__) @@ -149,7 +149,7 @@ static int is_mdelem_static(grpc_mdelem e) { static void ref_md_locked(mdtab_shard* shard, interned_metadata* md DEBUG_ARGS) { #ifndef NDEBUG - if (GRPC_TRACER_ON(grpc_trace_metadata)) { + if (grpc_trace_metadata.enabled()) { char* key_str = grpc_slice_to_c_string(md->key); char* value_str = grpc_slice_to_c_string(md->value); gpr_log(file, line, GPR_LOG_SEVERITY_DEBUG, @@ -252,7 +252,7 @@ grpc_mdelem grpc_mdelem_create( allocated->value = grpc_slice_ref_internal(value); gpr_atm_rel_store(&allocated->refcnt, 1); #ifndef NDEBUG - if (GRPC_TRACER_ON(grpc_trace_metadata)) { + if (grpc_trace_metadata.enabled()) { char* key_str = grpc_slice_to_c_string(allocated->key); char* value_str = grpc_slice_to_c_string(allocated->value); gpr_log(GPR_DEBUG, "ELM ALLOC:%p:%" PRIdPTR ": '%s' = '%s'", @@ -306,7 +306,7 @@ grpc_mdelem grpc_mdelem_create( shard->elems[idx] = md; gpr_mu_init(&md->mu_user_data); #ifndef NDEBUG - if (GRPC_TRACER_ON(grpc_trace_metadata)) { + if (grpc_trace_metadata.enabled()) { char* key_str = grpc_slice_to_c_string(md->key); char* value_str = grpc_slice_to_c_string(md->value); gpr_log(GPR_DEBUG, "ELM NEW:%p:%" PRIdPTR ": '%s' = '%s'", (void*)md, @@ -371,7 +371,7 @@ grpc_mdelem grpc_mdelem_ref(grpc_mdelem gmd DEBUG_ARGS) { case GRPC_MDELEM_STORAGE_INTERNED: { interned_metadata* md = (interned_metadata*)GRPC_MDELEM_DATA(gmd); #ifndef NDEBUG - if (GRPC_TRACER_ON(grpc_trace_metadata)) { + if (grpc_trace_metadata.enabled()) { char* key_str = grpc_slice_to_c_string(md->key); char* value_str = grpc_slice_to_c_string(md->value); gpr_log(file, line, GPR_LOG_SEVERITY_DEBUG, @@ -393,7 +393,7 @@ grpc_mdelem grpc_mdelem_ref(grpc_mdelem gmd DEBUG_ARGS) { case GRPC_MDELEM_STORAGE_ALLOCATED: { allocated_metadata* md = (allocated_metadata*)GRPC_MDELEM_DATA(gmd); #ifndef NDEBUG - if (GRPC_TRACER_ON(grpc_trace_metadata)) { + if (grpc_trace_metadata.enabled()) { char* key_str = grpc_slice_to_c_string(md->key); char* value_str = grpc_slice_to_c_string(md->value); gpr_log(file, line, GPR_LOG_SEVERITY_DEBUG, @@ -423,7 +423,7 @@ void grpc_mdelem_unref(grpc_mdelem gmd DEBUG_ARGS) { case GRPC_MDELEM_STORAGE_INTERNED: { interned_metadata* md = (interned_metadata*)GRPC_MDELEM_DATA(gmd); #ifndef NDEBUG - if (GRPC_TRACER_ON(grpc_trace_metadata)) { + if (grpc_trace_metadata.enabled()) { char* key_str = grpc_slice_to_c_string(md->key); char* value_str = grpc_slice_to_c_string(md->value); gpr_log(file, line, GPR_LOG_SEVERITY_DEBUG, @@ -449,7 +449,7 @@ void grpc_mdelem_unref(grpc_mdelem gmd DEBUG_ARGS) { case GRPC_MDELEM_STORAGE_ALLOCATED: { allocated_metadata* md = (allocated_metadata*)GRPC_MDELEM_DATA(gmd); #ifndef NDEBUG - if (GRPC_TRACER_ON(grpc_trace_metadata)) { + if (grpc_trace_metadata.enabled()) { char* key_str = grpc_slice_to_c_string(md->key); char* value_str = grpc_slice_to_c_string(md->value); gpr_log(file, line, GPR_LOG_SEVERITY_DEBUG, diff --git a/src/core/lib/transport/metadata.h b/src/core/lib/transport/metadata.h index 552eabf9b7..cf7fd0d43a 100644 --- a/src/core/lib/transport/metadata.h +++ b/src/core/lib/transport/metadata.h @@ -25,9 +25,7 @@ #include "src/core/lib/iomgr/exec_ctx.h" -#ifndef NDEBUG -extern grpc_tracer_flag grpc_trace_metadata; -#endif +extern grpc_core::DebugOnlyTraceFlag grpc_trace_metadata; #ifdef __cplusplus extern "C" { diff --git a/src/core/lib/transport/transport.cc b/src/core/lib/transport/transport.cc index d867123d6f..2e878b7f10 100644 --- a/src/core/lib/transport/transport.cc +++ b/src/core/lib/transport/transport.cc @@ -31,14 +31,12 @@ #include "src/core/lib/support/string.h" #include "src/core/lib/transport/transport_impl.h" -#ifndef NDEBUG -grpc_tracer_flag grpc_trace_stream_refcount = - GRPC_TRACER_INITIALIZER(false, "stream_refcount"); -#endif +grpc_core::DebugOnlyTraceFlag grpc_trace_stream_refcount(false, + "stream_refcount"); #ifndef NDEBUG void grpc_stream_ref(grpc_stream_refcount* refcount, const char* reason) { - if (GRPC_TRACER_ON(grpc_trace_stream_refcount)) { + if (grpc_trace_stream_refcount.enabled()) { gpr_atm val = gpr_atm_no_barrier_load(&refcount->refs.count); gpr_log(GPR_DEBUG, "%s %p:%p REF %" PRIdPTR "->%" PRIdPTR " %s", refcount->object_type, refcount, refcount->destroy.cb_arg, val, @@ -52,7 +50,7 @@ void grpc_stream_ref(grpc_stream_refcount* refcount) { #ifndef NDEBUG void grpc_stream_unref(grpc_stream_refcount* refcount, const char* reason) { - if (GRPC_TRACER_ON(grpc_trace_stream_refcount)) { + if (grpc_trace_stream_refcount.enabled()) { gpr_atm val = gpr_atm_no_barrier_load(&refcount->refs.count); gpr_log(GPR_DEBUG, "%s %p:%p UNREF %" PRIdPTR "->%" PRIdPTR " %s", refcount->object_type, refcount, refcount->destroy.cb_arg, val, diff --git a/src/core/lib/transport/transport.h b/src/core/lib/transport/transport.h index a6692ba1c2..c842e28a58 100644 --- a/src/core/lib/transport/transport.h +++ b/src/core/lib/transport/transport.h @@ -43,9 +43,7 @@ typedef struct grpc_transport grpc_transport; for a stream. */ typedef struct grpc_stream grpc_stream; -#ifndef NDEBUG -extern grpc_tracer_flag grpc_trace_stream_refcount; -#endif +extern grpc_core::DebugOnlyTraceFlag grpc_trace_stream_refcount; typedef struct grpc_stream_refcount { gpr_refcount refs; |