aboutsummaryrefslogtreecommitdiffhomepage
path: root/src/core/lib/surface/completion_queue.c
diff options
context:
space:
mode:
Diffstat (limited to 'src/core/lib/surface/completion_queue.c')
-rw-r--r--src/core/lib/surface/completion_queue.c124
1 files changed, 53 insertions, 71 deletions
diff --git a/src/core/lib/surface/completion_queue.c b/src/core/lib/surface/completion_queue.c
index eae3f103b1..df5b70205c 100644
--- a/src/core/lib/surface/completion_queue.c
+++ b/src/core/lib/surface/completion_queue.c
@@ -50,9 +50,9 @@
#include "src/core/lib/surface/call.h"
#include "src/core/lib/surface/event_string.h"
-int grpc_trace_operation_failures;
+grpc_tracer_flag grpc_trace_operation_failures = GRPC_TRACER_INITIALIZER(false);
#ifndef NDEBUG
-int grpc_trace_pending_tags;
+grpc_tracer_flag grpc_trace_pending_tags = GRPC_TRACER_INITIALIZER(false);
#endif
typedef struct {
@@ -72,7 +72,7 @@ typedef struct {
gpr_timespec deadline);
void (*shutdown)(grpc_exec_ctx *exec_ctx, grpc_pollset *pollset,
grpc_closure *closure);
- void (*destroy)(grpc_pollset *pollset);
+ void (*destroy)(grpc_exec_ctx *exec_ctx, grpc_pollset *pollset);
} cq_poller_vtable;
typedef struct non_polling_worker {
@@ -98,7 +98,8 @@ static void non_polling_poller_init(grpc_pollset *pollset, gpr_mu **mu) {
*mu = &npp->mu;
}
-static void non_polling_poller_destroy(grpc_pollset *pollset) {
+static void non_polling_poller_destroy(grpc_exec_ctx *exec_ctx,
+ grpc_pollset *pollset) {
non_polling_poller *npp = (non_polling_poller *)pollset;
gpr_mu_destroy(&npp->mu);
}
@@ -242,15 +243,16 @@ struct grpc_completion_queue {
#define POLLSET_FROM_CQ(cq) ((grpc_pollset *)(cq + 1))
#define CQ_FROM_POLLSET(ps) (((grpc_completion_queue *)ps) - 1)
-int grpc_cq_pluck_trace;
-int grpc_cq_event_timeout_trace;
+grpc_tracer_flag grpc_cq_pluck_trace = GRPC_TRACER_INITIALIZER(true);
+grpc_tracer_flag grpc_cq_event_timeout_trace = GRPC_TRACER_INITIALIZER(true);
-#define GRPC_SURFACE_TRACE_RETURNED_EVENT(cq, event) \
- if (grpc_api_trace && \
- (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); \
+#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); \
}
static void on_pollset_shutdown_done(grpc_exec_ctx *exec_ctx, void *cc,
@@ -322,20 +324,21 @@ void grpc_cq_internal_ref(grpc_completion_queue *cc) {
static void on_pollset_shutdown_done(grpc_exec_ctx *exec_ctx, void *arg,
grpc_error *error) {
grpc_completion_queue *cc = arg;
- GRPC_CQ_INTERNAL_UNREF(cc, "pollset_destroy");
+ GRPC_CQ_INTERNAL_UNREF(exec_ctx, cc, "pollset_destroy");
}
#ifdef GRPC_CQ_REF_COUNT_DEBUG
-void grpc_cq_internal_unref(grpc_completion_queue *cc, const char *reason,
- const char *file, int line) {
+void grpc_cq_internal_unref(grpc_exec_ctx *exec_ctx, grpc_completion_queue *cc,
+ const char *reason, const char *file, int line) {
gpr_log(file, line, GPR_LOG_SEVERITY_DEBUG, "CQ:%p unref %d -> %d %s", cc,
(int)cc->owning_refs.count, (int)cc->owning_refs.count - 1, reason);
#else
-void grpc_cq_internal_unref(grpc_completion_queue *cc) {
+void grpc_cq_internal_unref(grpc_exec_ctx *exec_ctx,
+ grpc_completion_queue *cc) {
#endif
if (gpr_unref(&cc->owning_refs)) {
GPR_ASSERT(cc->completed_head.next == (uintptr_t)&cc->completed_head);
- cc->poller_vtable->destroy(POLLSET_FROM_CQ(cc));
+ cc->poller_vtable->destroy(exec_ctx, POLLSET_FROM_CQ(cc));
#ifndef NDEBUG
gpr_free(cc->outstanding_tags);
#endif
@@ -375,14 +378,16 @@ void grpc_cq_end_op(grpc_exec_ctx *exec_ctx, grpc_completion_queue *cc,
#endif
GPR_TIMER_BEGIN("grpc_cq_end_op", 0);
- if (grpc_api_trace ||
- (grpc_trace_operation_failures && error != GRPC_ERROR_NONE)) {
+ if (GRPC_TRACER_ON(grpc_api_trace) ||
+ (GRPC_TRACER_ON(grpc_trace_operation_failures) &&
+ error != GRPC_ERROR_NONE)) {
const char *errmsg = grpc_error_string(error);
GRPC_API_TRACE(
"grpc_cq_end_op(exec_ctx=%p, cc=%p, tag=%p, error=%s, done=%p, "
"done_arg=%p, storage=%p)",
7, (exec_ctx, cc, tag, errmsg, done, done_arg, storage));
- if (grpc_trace_operation_failures && error != GRPC_ERROR_NONE) {
+ if (GRPC_TRACER_ON(grpc_trace_operation_failures) &&
+ error != GRPC_ERROR_NONE) {
gpr_log(GPR_ERROR, "Operation failed: tag=%p, error=%s", tag, errmsg);
}
}
@@ -481,7 +486,7 @@ static bool cq_is_next_finished(grpc_exec_ctx *exec_ctx, void *arg) {
#ifndef NDEBUG
static void dump_pending_tags(grpc_completion_queue *cc) {
- if (!grpc_trace_pending_tags) return;
+ if (!GRPC_TRACER_ON(grpc_trace_pending_tags)) return;
gpr_strvec v;
gpr_strvec_init(&v);
@@ -580,36 +585,23 @@ grpc_event grpc_completion_queue_next(grpc_completion_queue *cc,
dump_pending_tags(cc);
break;
}
- /* Check alarms - these are a global resource so we just ping
- each time through on every pollset.
- May update deadline to ensure timely wakeups.
- TODO(ctiller): can this work be localized? */
- gpr_timespec iteration_deadline = deadline;
- if (grpc_timer_check(&exec_ctx, now, &iteration_deadline)) {
- GPR_TIMER_MARK("alarm_triggered", 0);
+ grpc_error *err = cc->poller_vtable->work(&exec_ctx, POLLSET_FROM_CQ(cc),
+ NULL, now, deadline);
+ if (err != GRPC_ERROR_NONE) {
gpr_mu_unlock(cc->mu);
- grpc_exec_ctx_flush(&exec_ctx);
- gpr_mu_lock(cc->mu);
- continue;
- } else {
- grpc_error *err = cc->poller_vtable->work(&exec_ctx, POLLSET_FROM_CQ(cc),
- NULL, now, iteration_deadline);
- if (err != GRPC_ERROR_NONE) {
- gpr_mu_unlock(cc->mu);
- const char *msg = grpc_error_string(err);
- gpr_log(GPR_ERROR, "Completion queue next failed: %s", msg);
+ const char *msg = grpc_error_string(err);
+ gpr_log(GPR_ERROR, "Completion queue next failed: %s", msg);
- GRPC_ERROR_UNREF(err);
- memset(&ret, 0, sizeof(ret));
- ret.type = GRPC_QUEUE_TIMEOUT;
- dump_pending_tags(cc);
- break;
- }
+ GRPC_ERROR_UNREF(err);
+ memset(&ret, 0, sizeof(ret));
+ ret.type = GRPC_QUEUE_TIMEOUT;
+ dump_pending_tags(cc);
+ break;
}
is_finished_arg.first_loop = false;
}
GRPC_SURFACE_TRACE_RETURNED_EVENT(cc, &ret);
- GRPC_CQ_INTERNAL_UNREF(cc, "next");
+ GRPC_CQ_INTERNAL_UNREF(&exec_ctx, cc, "next");
grpc_exec_ctx_finish(&exec_ctx);
GPR_ASSERT(is_finished_arg.stolen_completion == NULL);
@@ -690,7 +682,7 @@ grpc_event grpc_completion_queue_pluck(grpc_completion_queue *cc, void *tag,
abort();
}
- if (grpc_cq_pluck_trace) {
+ if (GRPC_TRACER_ON(grpc_cq_pluck_trace)) {
GRPC_API_TRACE(
"grpc_completion_queue_pluck("
"cc=%p, tag=%p, "
@@ -773,38 +765,26 @@ grpc_event grpc_completion_queue_pluck(grpc_completion_queue *cc, void *tag,
dump_pending_tags(cc);
break;
}
- /* Check alarms - these are a global resource so we just ping
- each time through on every pollset.
- May update deadline to ensure timely wakeups.
- TODO(ctiller): can this work be localized? */
- gpr_timespec iteration_deadline = deadline;
- if (grpc_timer_check(&exec_ctx, now, &iteration_deadline)) {
- GPR_TIMER_MARK("alarm_triggered", 0);
+ grpc_error *err = cc->poller_vtable->work(&exec_ctx, POLLSET_FROM_CQ(cc),
+ &worker, now, deadline);
+ if (err != GRPC_ERROR_NONE) {
+ del_plucker(cc, tag, &worker);
gpr_mu_unlock(cc->mu);
- grpc_exec_ctx_flush(&exec_ctx);
- gpr_mu_lock(cc->mu);
- } else {
- grpc_error *err = cc->poller_vtable->work(
- &exec_ctx, POLLSET_FROM_CQ(cc), &worker, now, iteration_deadline);
- if (err != GRPC_ERROR_NONE) {
- del_plucker(cc, tag, &worker);
- gpr_mu_unlock(cc->mu);
- const char *msg = grpc_error_string(err);
- gpr_log(GPR_ERROR, "Completion queue next failed: %s", msg);
+ const char *msg = grpc_error_string(err);
+ gpr_log(GPR_ERROR, "Completion queue next failed: %s", msg);
- GRPC_ERROR_UNREF(err);
- memset(&ret, 0, sizeof(ret));
- ret.type = GRPC_QUEUE_TIMEOUT;
- dump_pending_tags(cc);
- break;
- }
+ GRPC_ERROR_UNREF(err);
+ memset(&ret, 0, sizeof(ret));
+ ret.type = GRPC_QUEUE_TIMEOUT;
+ dump_pending_tags(cc);
+ break;
}
is_finished_arg.first_loop = false;
del_plucker(cc, tag, &worker);
}
done:
GRPC_SURFACE_TRACE_RETURNED_EVENT(cc, &ret);
- GRPC_CQ_INTERNAL_UNREF(cc, "pluck");
+ GRPC_CQ_INTERNAL_UNREF(&exec_ctx, cc, "pluck");
grpc_exec_ctx_finish(&exec_ctx);
GPR_ASSERT(is_finished_arg.stolen_completion == NULL);
@@ -841,7 +821,9 @@ void grpc_completion_queue_destroy(grpc_completion_queue *cc) {
GRPC_API_TRACE("grpc_completion_queue_destroy(cc=%p)", 1, (cc));
GPR_TIMER_BEGIN("grpc_completion_queue_destroy", 0);
grpc_completion_queue_shutdown(cc);
- GRPC_CQ_INTERNAL_UNREF(cc, "destroy");
+ grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT;
+ GRPC_CQ_INTERNAL_UNREF(&exec_ctx, cc, "destroy");
+ grpc_exec_ctx_finish(&exec_ctx);
GPR_TIMER_END("grpc_completion_queue_destroy", 0);
}
generated by cgit v1.2.3 (git 2.39.1) at 2025-01-06 13:16:16 +0000