Add debugging
diff --git a/src/core/iomgr/pollset_posix.c b/src/core/iomgr/pollset_posix.c
index 71d6dc0..8b91086 100644
--- a/src/core/iomgr/pollset_posix.c
+++ b/src/core/iomgr/pollset_posix.c
@@ -134,6 +134,7 @@
r = pollset->vtable->maybe_work(pollset, deadline, now, 1);
gpr_tls_set(&g_current_thread_poller, 0);
if (pollset->shutting_down) {
+ gpr_log(GPR_DEBUG, "shutting_down': cbs=%d ctr=%d", pollset->in_flight_cbs, pollset->counter);
if (pollset->counter > 0) {
grpc_pollset_kick(pollset);
} else if (pollset->in_flight_cbs == 0) {
@@ -155,9 +156,11 @@
int in_flight_cbs;
int counter;
gpr_mu_lock(&pollset->mu);
+ GPR_ASSERT(!pollset->shutting_down);
pollset->shutting_down = 1;
in_flight_cbs = pollset->in_flight_cbs;
counter = pollset->counter;
+ gpr_log(GPR_DEBUG, "shutting_down: cbs=%d ctr=%d", in_flight_cbs, counter);
pollset->shutdown_done_cb = shutdown_done;
pollset->shutdown_done_arg = shutdown_done_arg;
if (counter > 0) {
@@ -207,7 +210,7 @@
gpr_mu_lock(&pollset->mu);
/* First we need to ensure that nobody is polling concurrently */
- while (pollset->counter != 0) {
+ if (pollset->counter != 0) {
grpc_pollset_kick(pollset);
grpc_iomgr_add_callback(basic_do_promote, up_args);
gpr_mu_unlock(&pollset->mu);
@@ -344,10 +347,10 @@
pfd[0].events = POLLIN;
pfd[0].revents = 0;
nfds = 1;
+ pollset->counter++;
if (fd) {
pfd[1].fd = fd->fd;
pfd[1].revents = 0;
- pollset->counter++;
gpr_mu_unlock(&pollset->mu);
pfd[1].events =
grpc_fd_begin_poll(fd, pollset, POLLIN, POLLOUT, &fd_watcher);
diff --git a/src/core/surface/completion_queue.c b/src/core/surface/completion_queue.c
index 9e4c636..3ab2dfc 100644
--- a/src/core/surface/completion_queue.c
+++ b/src/core/surface/completion_queue.c
@@ -86,16 +86,38 @@
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);
@@ -170,7 +192,7 @@
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) {
@@ -202,7 +224,7 @@
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;
}
}
@@ -210,7 +232,7 @@
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;
}
@@ -248,7 +270,7 @@
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))) {
@@ -263,7 +285,7 @@
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;
}
}
@@ -271,7 +293,7 @@
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;
}
@@ -292,7 +314,7 @@
}
void grpc_completion_queue_destroy(grpc_completion_queue *cc) {
- grpc_cq_internal_unref(cc);
+ GRPC_CQ_INTERNAL_UNREF(cc, "destroy");
}
grpc_pollset *grpc_cq_pollset(grpc_completion_queue *cc) {
diff --git a/src/core/surface/completion_queue.h b/src/core/surface/completion_queue.h
index 28a4874..2249d0e 100644
--- a/src/core/surface/completion_queue.h
+++ b/src/core/surface/completion_queue.h
@@ -39,8 +39,17 @@
#include "src/core/iomgr/pollset.h"
#include <grpc/grpc.h>
+#ifdef GRPC_CQ_REF_COUNT_DEBUG
+void grpc_cq_internal_ref(grpc_completion_queue *cc, const char *reason);
+void grpc_cq_internal_unref(grpc_completion_queue *cc, const char *reason);
+#define GRPC_CQ_INTERNAL_REF(cc, reason) grpc_cq_internal_ref(cc, reason)
+#define GRPC_CQ_INTERNAL_UNREF(cc, reason) grpc_cq_internal_unref(cc, reason)
+#else
void grpc_cq_internal_ref(grpc_completion_queue *cc);
void grpc_cq_internal_unref(grpc_completion_queue *cc);
+#define GRPC_CQ_INTERNAL_REF(cc, reason) grpc_cq_internal_ref(cc)
+#define GRPC_CQ_INTERNAL_UNREF(cc, reason) grpc_cq_internal_unref(cc)
+#endif
/* Flag that an operation is beginning: the completion channel will not finish
shutdown until a corrensponding grpc_cq_end_* call is made */
diff --git a/src/core/surface/server.c b/src/core/surface/server.c
index b1e3ec9..b72ea54 100644
--- a/src/core/surface/server.c
+++ b/src/core/surface/server.c
@@ -276,7 +276,7 @@
gpr_free(rm);
}
for (i = 0; i < server->cq_count; i++) {
- grpc_cq_internal_unref(server->cqs[i]);
+ GRPC_CQ_INTERNAL_UNREF(server->cqs[i], "server");
}
gpr_free(server->cqs);
gpr_free(server->pollsets);
@@ -632,7 +632,7 @@
for (i = 0; i < server->cq_count; i++) {
if (server->cqs[i] == cq) return;
}
- grpc_cq_internal_ref(cq);
+ GRPC_CQ_INTERNAL_REF(cq, "server");
n = server->cq_count++;
server->cqs = gpr_realloc(server->cqs,
server->cq_count * sizeof(grpc_completion_queue *));