aboutsummaryrefslogtreecommitdiffhomepage
path: root/src/core/surface/completion_queue.c
diff options
context:
space:
mode:
Diffstat (limited to 'src/core/surface/completion_queue.c')
-rw-r--r--src/core/surface/completion_queue.c61
1 files changed, 32 insertions, 29 deletions
diff --git a/src/core/surface/completion_queue.c b/src/core/surface/completion_queue.c
index b48fbace31..bd0fabf9da 100644
--- a/src/core/surface/completion_queue.c
+++ b/src/core/surface/completion_queue.c
@@ -59,9 +59,6 @@ typedef struct event {
/* Completion queue structure */
struct grpc_completion_queue {
- /* TODO(ctiller): see if this can be removed */
- int allow_polling;
-
/* When refs drops to zero, we are in shutdown mode, and will be destroyable
once all queued events are drained */
gpr_refcount refs;
@@ -76,6 +73,7 @@ struct grpc_completion_queue {
event *queue;
/* Fixed size chained hash table of events for pluck() */
event *buckets[NUM_TAG_BUCKETS];
+ int is_server_cq;
};
grpc_completion_queue *grpc_completion_queue_create(void) {
@@ -86,20 +84,31 @@ grpc_completion_queue *grpc_completion_queue_create(void) {
/* One for destroy(), one for pollset_shutdown */
gpr_ref_init(&cc->owning_refs, 2);
grpc_pollset_init(&cc->pollset);
- cc->allow_polling = 1;
return cc;
}
+#ifdef GRPC_CQ_REF_COUNT_DEBUG
+void grpc_cq_internal_ref(grpc_completion_queue *cc, const char *reason) {
+ gpr_log(GPR_DEBUG, "CQ:%p ref %d -> %d %s", cc, (int)cc->owning_refs.count,
+ (int)cc->owning_refs.count + 1, reason);
+#else
void grpc_cq_internal_ref(grpc_completion_queue *cc) {
+#endif
gpr_ref(&cc->owning_refs);
}
static void on_pollset_destroy_done(void *arg) {
grpc_completion_queue *cc = arg;
- grpc_cq_internal_unref(cc);
+ GRPC_CQ_INTERNAL_UNREF(cc, "pollset_destroy");
}
+#ifdef GRPC_CQ_REF_COUNT_DEBUG
+void grpc_cq_internal_unref(grpc_completion_queue *cc, const char *reason) {
+ gpr_log(GPR_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) {
+#endif
if (gpr_unref(&cc->owning_refs)) {
GPR_ASSERT(cc->queue == NULL);
grpc_pollset_destroy(&cc->pollset);
@@ -107,10 +116,6 @@ void grpc_cq_internal_unref(grpc_completion_queue *cc) {
}
}
-void grpc_completion_queue_dont_poll_test_only(grpc_completion_queue *cc) {
- cc->allow_polling = 0;
-}
-
/* Create and append an event to the queue. Returns the event so that its data
members can be filled in.
Requires GRPC_POLLSET_MU(&cc->pollset) locked. */
@@ -134,7 +139,6 @@ static event *add_locked(grpc_completion_queue *cc, grpc_completion_type type,
ev->bucket_prev = cc->buckets[bucket]->bucket_prev;
ev->bucket_next->bucket_prev = ev->bucket_prev->bucket_next = ev;
}
- gpr_cv_broadcast(GRPC_POLLSET_CV(&cc->pollset));
grpc_pollset_kick(&cc->pollset);
return ev;
}
@@ -157,7 +161,6 @@ void grpc_cq_end_op(grpc_completion_queue *cc, void *tag, grpc_call *call,
GPR_ASSERT(!cc->shutdown);
GPR_ASSERT(cc->shutdown_called);
cc->shutdown = 1;
- gpr_cv_broadcast(GRPC_POLLSET_CV(&cc->pollset));
shutdown = 1;
}
gpr_mu_unlock(GRPC_POLLSET_MU(&cc->pollset));
@@ -180,7 +183,7 @@ grpc_event grpc_completion_queue_next(grpc_completion_queue *cc,
event *ev = NULL;
grpc_event ret;
- grpc_cq_internal_ref(cc);
+ GRPC_CQ_INTERNAL_REF(cc, "next");
gpr_mu_lock(GRPC_POLLSET_MU(&cc->pollset));
for (;;) {
if (cc->queue != NULL) {
@@ -207,16 +210,12 @@ grpc_event grpc_completion_queue_next(grpc_completion_queue *cc,
ev = create_shutdown_event();
break;
}
- if (cc->allow_polling && grpc_pollset_work(&cc->pollset, deadline)) {
- continue;
- }
- if (gpr_cv_wait(GRPC_POLLSET_CV(&cc->pollset),
- GRPC_POLLSET_MU(&cc->pollset), deadline)) {
+ if (!grpc_pollset_work(&cc->pollset, deadline)) {
gpr_mu_unlock(GRPC_POLLSET_MU(&cc->pollset));
memset(&ret, 0, sizeof(ret));
ret.type = GRPC_QUEUE_TIMEOUT;
GRPC_SURFACE_TRACE_RETURNED_EVENT(cc, &ret);
- grpc_cq_internal_unref(cc);
+ GRPC_CQ_INTERNAL_UNREF(cc, "next");
return ret;
}
}
@@ -224,7 +223,7 @@ grpc_event grpc_completion_queue_next(grpc_completion_queue *cc,
ret = ev->base;
gpr_free(ev);
GRPC_SURFACE_TRACE_RETURNED_EVENT(cc, &ret);
- grpc_cq_internal_unref(cc);
+ GRPC_CQ_INTERNAL_UNREF(cc, "next");
return ret;
}
@@ -262,7 +261,7 @@ grpc_event grpc_completion_queue_pluck(grpc_completion_queue *cc, void *tag,
event *ev = NULL;
grpc_event ret;
- grpc_cq_internal_ref(cc);
+ GRPC_CQ_INTERNAL_REF(cc, "pluck");
gpr_mu_lock(GRPC_POLLSET_MU(&cc->pollset));
for (;;) {
if ((ev = pluck_event(cc, tag))) {
@@ -272,16 +271,12 @@ grpc_event grpc_completion_queue_pluck(grpc_completion_queue *cc, void *tag,
ev = create_shutdown_event();
break;
}
- if (cc->allow_polling && grpc_pollset_work(&cc->pollset, deadline)) {
- continue;
- }
- if (gpr_cv_wait(GRPC_POLLSET_CV(&cc->pollset),
- GRPC_POLLSET_MU(&cc->pollset), deadline)) {
+ if (!grpc_pollset_work(&cc->pollset, deadline)) {
gpr_mu_unlock(GRPC_POLLSET_MU(&cc->pollset));
memset(&ret, 0, sizeof(ret));
ret.type = GRPC_QUEUE_TIMEOUT;
GRPC_SURFACE_TRACE_RETURNED_EVENT(cc, &ret);
- grpc_cq_internal_unref(cc);
+ GRPC_CQ_INTERNAL_UNREF(cc, "pluck");
return ret;
}
}
@@ -289,7 +284,7 @@ grpc_event grpc_completion_queue_pluck(grpc_completion_queue *cc, void *tag,
ret = ev->base;
gpr_free(ev);
GRPC_SURFACE_TRACE_RETURNED_EVENT(cc, &ret);
- grpc_cq_internal_unref(cc);
+ GRPC_CQ_INTERNAL_UNREF(cc, "pluck");
return ret;
}
@@ -297,6 +292,10 @@ grpc_event grpc_completion_queue_pluck(grpc_completion_queue *cc, void *tag,
to zero here, then enter shutdown mode and wake up any waiters */
void grpc_completion_queue_shutdown(grpc_completion_queue *cc) {
gpr_mu_lock(GRPC_POLLSET_MU(&cc->pollset));
+ if (cc->shutdown_called) {
+ gpr_mu_unlock(GRPC_POLLSET_MU(&cc->pollset));
+ return;
+ }
cc->shutdown_called = 1;
gpr_mu_unlock(GRPC_POLLSET_MU(&cc->pollset));
@@ -304,14 +303,14 @@ void grpc_completion_queue_shutdown(grpc_completion_queue *cc) {
gpr_mu_lock(GRPC_POLLSET_MU(&cc->pollset));
GPR_ASSERT(!cc->shutdown);
cc->shutdown = 1;
- gpr_cv_broadcast(GRPC_POLLSET_CV(&cc->pollset));
gpr_mu_unlock(GRPC_POLLSET_MU(&cc->pollset));
grpc_pollset_shutdown(&cc->pollset, on_pollset_destroy_done, cc);
}
}
void grpc_completion_queue_destroy(grpc_completion_queue *cc) {
- grpc_cq_internal_unref(cc);
+ grpc_completion_queue_shutdown(cc);
+ GRPC_CQ_INTERNAL_UNREF(cc, "destroy");
}
grpc_pollset *grpc_cq_pollset(grpc_completion_queue *cc) {
@@ -325,3 +324,7 @@ void grpc_cq_hack_spin_pollset(grpc_completion_queue *cc) {
gpr_time_add(gpr_now(), gpr_time_from_millis(100)));
gpr_mu_unlock(GRPC_POLLSET_MU(&cc->pollset));
}
+
+void grpc_cq_mark_server_cq(grpc_completion_queue *cc) { cc->is_server_cq = 1; }
+
+int grpc_cq_is_server_cq(grpc_completion_queue *cc) { return cc->is_server_cq; }
generated by cgit v1.2.3 (git 2.39.1) at 2025-01-08 22:59:57 +0000