Allow adding events to cq after shutdown is called.
diff --git a/include/grpc/impl/codegen/grpc_types.h b/include/grpc/impl/codegen/grpc_types.h
index f1c457c..bb3c90e 100644
--- a/include/grpc/impl/codegen/grpc_types.h
+++ b/include/grpc/impl/codegen/grpc_types.h
@@ -333,7 +333,9 @@
/** this batch of operations leads to more operations than allowed */
GRPC_CALL_ERROR_BATCH_TOO_BIG,
/** payload type requested is not the type registered */
- GRPC_CALL_ERROR_PAYLOAD_TYPE_MISMATCH
+ GRPC_CALL_ERROR_PAYLOAD_TYPE_MISMATCH,
+ /** completion queue has been shutdown */
+ GRPC_CALL_ERROR_COMPLETION_QUEUE_SHUTDOWN
} grpc_call_error;
/** Default send/receive message size limits in bytes. -1 for unlimited. */
diff --git a/src/core/ext/filters/client_channel/channel_connectivity.c b/src/core/ext/filters/client_channel/channel_connectivity.c
index c3dca14..e30dcc1 100644
--- a/src/core/ext/filters/client_channel/channel_connectivity.c
+++ b/src/core/ext/filters/client_channel/channel_connectivity.c
@@ -208,7 +208,7 @@
7, (channel, (int)last_observed_state, deadline.tv_sec, deadline.tv_nsec,
(int)deadline.clock_type, cq, tag));
- grpc_cq_begin_op(cq, tag);
+ GPR_ASSERT(grpc_cq_begin_op(cq, tag) == 0);
gpr_mu_init(&w->mu);
GRPC_CLOSURE_INIT(&w->on_complete, watch_complete, w,
diff --git a/src/core/lib/surface/alarm.c b/src/core/lib/surface/alarm.c
index ef8405c..632cd6e 100644
--- a/src/core/lib/surface/alarm.c
+++ b/src/core/lib/surface/alarm.c
@@ -18,6 +18,7 @@
#include <grpc/grpc.h>
#include <grpc/support/alloc.h>
+#include <grpc/support/log.h>
#include "src/core/lib/iomgr/timer.h"
#include "src/core/lib/surface/completion_queue.h"
@@ -49,7 +50,7 @@
alarm->cq = cq;
alarm->tag = tag;
- grpc_cq_begin_op(cq, tag);
+ GPR_ASSERT(grpc_cq_begin_op(cq, tag) == 0);
GRPC_CLOSURE_INIT(&alarm->on_alarm, alarm_cb, alarm,
grpc_schedule_on_exec_ctx);
grpc_timer_init(&exec_ctx, &alarm->alarm,
diff --git a/src/core/lib/surface/call.c b/src/core/lib/surface/call.c
index 2365d27..a173fad 100644
--- a/src/core/lib/surface/call.c
+++ b/src/core/lib/surface/call.c
@@ -1422,7 +1422,7 @@
if (nops == 0) {
if (!is_notify_tag_closure) {
- grpc_cq_begin_op(call->cq, notify_tag);
+ GPR_ASSERT(grpc_cq_begin_op(call->cq, notify_tag) == 0);
grpc_cq_end_op(exec_ctx, call->cq, notify_tag, GRPC_ERROR_NONE,
free_no_op_completion, NULL,
gpr_malloc(sizeof(grpc_cq_completion)));
@@ -1723,7 +1723,7 @@
GRPC_CALL_INTERNAL_REF(call, "completion");
if (!is_notify_tag_closure) {
- grpc_cq_begin_op(call->cq, notify_tag);
+ GPR_ASSERT(grpc_cq_begin_op(call->cq, notify_tag) == 0);
}
gpr_ref_init(&bctl->steps_to_complete, num_completion_callbacks_needed);
@@ -1844,6 +1844,8 @@
return "GRPC_CALL_ERROR_PAYLOAD_TYPE_MISMATCH";
case GRPC_CALL_ERROR_TOO_MANY_OPERATIONS:
return "GRPC_CALL_ERROR_TOO_MANY_OPERATIONS";
+ case GRPC_CALL_ERROR_COMPLETION_QUEUE_SHUTDOWN:
+ return "GRPC_CALL_ERROR_COMPLETION_QUEUE_SHUTDOWN";
case GRPC_CALL_OK:
return "GRPC_CALL_OK";
}
diff --git a/src/core/lib/surface/channel_ping.c b/src/core/lib/surface/channel_ping.c
index 80eb80a..8b1150b 100644
--- a/src/core/lib/surface/channel_ping.c
+++ b/src/core/lib/surface/channel_ping.c
@@ -59,7 +59,7 @@
GRPC_CLOSURE_INIT(&pr->closure, ping_done, pr, grpc_schedule_on_exec_ctx);
op->send_ping = &pr->closure;
op->bind_pollset = grpc_cq_pollset(cq);
- grpc_cq_begin_op(cq, tag);
+ GPR_ASSERT(grpc_cq_begin_op(cq, tag) == 0);
top_elem->filter->start_transport_op(&exec_ctx, top_elem, op);
grpc_exec_ctx_finish(&exec_ctx);
}
diff --git a/src/core/lib/surface/completion_queue.c b/src/core/lib/surface/completion_queue.c
index 3851993..ef1dfb1 100644
--- a/src/core/lib/surface/completion_queue.c
+++ b/src/core/lib/surface/completion_queue.c
@@ -525,7 +525,16 @@
static void cq_begin_op_for_next(grpc_completion_queue *cq, void *tag) {
cq_next_data *cqd = DATA_FROM_CQ(cq);
GPR_ASSERT(!cqd->shutdown_called);
- gpr_atm_no_barrier_fetch_add(&cqd->pending_events, 1);
+ while (true) {
+ gpr_atm count = gpr_atm_no_barrier_load(&cqd->pending_events.count);
+ if (count == 0) {
+ return 1;
+ } else if (gpr_atm_no_barrier_cas(&cqd->pending_events.count, count,
+ count + 1)) {
+ break;
+ }
+ }
+ return 0;
}
static void cq_begin_op_for_pluck(grpc_completion_queue *cq, void *tag) {
diff --git a/src/core/lib/surface/completion_queue.h b/src/core/lib/surface/completion_queue.h
index af44482..7f3d398 100644
--- a/src/core/lib/surface/completion_queue.h
+++ b/src/core/lib/surface/completion_queue.h
@@ -72,8 +72,9 @@
/* Flag that an operation is beginning: the completion channel will not finish
shutdown until a corrensponding grpc_cq_end_* call is made.
- \a tag is currently used only in debug builds. */
-void grpc_cq_begin_op(grpc_completion_queue *cc, void *tag);
+ \a tag is currently used only in debug builds. Return 0 on success and 1 if
+ completion_queue has been shutdown. */
+int grpc_cq_begin_op(grpc_completion_queue *cc, void *tag);
/* Queue a GRPC_OP_COMPLETED operation; tag must correspond to the tag passed to
grpc_cq_begin_op */
diff --git a/src/core/lib/surface/server.c b/src/core/lib/surface/server.c
index fce7f8d..957593a 100644
--- a/src/core/lib/surface/server.c
+++ b/src/core/lib/surface/server.c
@@ -1259,7 +1259,7 @@
}
/* stay locked, and gather up some stuff to do */
- grpc_cq_begin_op(cq, tag);
+ GPR_ASSERT(grpc_cq_begin_op(cq, tag) == 0);
if (server->shutdown_published) {
grpc_cq_end_op(&exec_ctx, cq, tag, GRPC_ERROR_NONE, done_published_shutdown,
NULL, gpr_malloc(sizeof(grpc_cq_completion)));
@@ -1446,7 +1446,11 @@
error = GRPC_CALL_ERROR_NOT_SERVER_COMPLETION_QUEUE;
goto done;
}
- grpc_cq_begin_op(cq_for_notification, tag);
+ if (grpc_cq_begin_op(cq_for_notification, tag)) {
+ gpr_free(rc);
+ error = GRPC_CALL_ERROR_COMPLETION_QUEUE_SHUTDOWN;
+ goto done;
+ }
details->reserved = NULL;
rc->cq_idx = cq_idx;
rc->type = BATCH_CALL;
@@ -1496,7 +1500,11 @@
error = GRPC_CALL_ERROR_PAYLOAD_TYPE_MISMATCH;
goto done;
}
- grpc_cq_begin_op(cq_for_notification, tag);
+ if (grpc_cq_begin_op(cq_for_notification, tag)) {
+ gpr_free(rc);
+ error = GRPC_CALL_ERROR_COMPLETION_QUEUE_SHUTDOWN;
+ goto done;
+ }
rc->cq_idx = cq_idx;
rc->type = REGISTERED_CALL;
rc->server = server;
diff --git a/src/cpp/server/server_cc.cc b/src/cpp/server/server_cc.cc
index 92bacbe..91d9804 100644
--- a/src/cpp/server/server_cc.cc
+++ b/src/cpp/server/server_cc.cc
@@ -151,19 +151,24 @@
GPR_ASSERT(cq_ && !in_flight_);
in_flight_ = true;
if (tag_) {
- GPR_ASSERT(GRPC_CALL_OK ==
- grpc_server_request_registered_call(
- server, tag_, &call_, &deadline_, &request_metadata_,
- has_request_payload_ ? &request_payload_ : nullptr, cq_,
- notify_cq, this));
+ if (grpc_server_request_registered_call(
+ server, tag_, &call_, &deadline_, &request_metadata_,
+ has_request_payload_ ? &request_payload_ : nullptr, cq_,
+ notify_cq, this) != GRPC_CALL_OK) {
+ TeardownRequest();
+ return;
+ }
} else {
if (!call_details_) {
call_details_ = new grpc_call_details;
grpc_call_details_init(call_details_);
}
- GPR_ASSERT(GRPC_CALL_OK == grpc_server_request_call(
- server, &call_, call_details_,
- &request_metadata_, cq_, notify_cq, this));
+ if (grpc_server_request_call(server, &call_, call_details_,
+ &request_metadata_, cq_, notify_cq,
+ this) != GRPC_CALL_OK) {
+ TeardownRequest();
+ return;
+ }
}
}
@@ -286,12 +291,10 @@
if (ok) {
// Calldata takes ownership of the completion queue inside sync_req
SyncRequest::CallData cd(server_, sync_req);
- {
- // Prepare for the next request
- if (!IsShutdown()) {
- sync_req->SetupRequest(); // Create new completion queue for sync_req
- sync_req->Request(server_->c_server(), server_cq_->cq());
- }
+ // Prepare for the next request
+ if (!IsShutdown()) {
+ sync_req->SetupRequest(); // Create new completion queue for sync_req
+ sync_req->Request(server_->c_server(), server_cq_->cq());
}
GPR_TIMER_SCOPE("cd.Run()", 0);
@@ -316,8 +319,8 @@
}
void Shutdown() override {
- server_cq_->Shutdown();
ThreadManager::Shutdown();
+ server_cq_->Shutdown();
}
void Wait() override {
@@ -652,10 +655,11 @@
void ServerInterface::RegisteredAsyncRequest::IssueRequest(
void* registered_method, grpc_byte_buffer** payload,
ServerCompletionQueue* notification_cq) {
- grpc_server_request_registered_call(
- server_->server(), registered_method, &call_, &context_->deadline_,
- context_->client_metadata_.arr(), payload, call_cq_->cq(),
- notification_cq->cq(), this);
+ GPR_ASSERT(GRPC_CALL_OK == grpc_server_request_registered_call(
+ server_->server(), registered_method, &call_,
+ &context_->deadline_,
+ context_->client_metadata_.arr(), payload,
+ call_cq_->cq(), notification_cq->cq(), this));
}
ServerInterface::GenericAsyncRequest::GenericAsyncRequest(
@@ -667,9 +671,10 @@
grpc_call_details_init(&call_details_);
GPR_ASSERT(notification_cq);
GPR_ASSERT(call_cq);
- grpc_server_request_call(server->server(), &call_, &call_details_,
- context->client_metadata_.arr(), call_cq->cq(),
- notification_cq->cq(), this);
+ GPR_ASSERT(GRPC_CALL_OK == grpc_server_request_call(
+ server->server(), &call_, &call_details_,
+ context->client_metadata_.arr(), call_cq->cq(),
+ notification_cq->cq(), this));
}
bool ServerInterface::GenericAsyncRequest::FinalizeResult(void** tag,
diff --git a/test/core/end2end/fuzzers/server_fuzzer.c b/test/core/end2end/fuzzers/server_fuzzer.c
index 3ddc1ae..ef4c0a4 100644
--- a/test/core/end2end/fuzzers/server_fuzzer.c
+++ b/test/core/end2end/fuzzers/server_fuzzer.c
@@ -72,8 +72,9 @@
grpc_metadata_array_init(&request_metadata1);
int requested_calls = 0;
- grpc_server_request_call(server, &call1, &call_details1, &request_metadata1,
- cq, cq, tag(1));
+ GPR_ASSERT(GRPC_CALL_OK ==
+ grpc_server_request_call(server, &call1, &call_details1,
+ &request_metadata1, cq, cq, tag(1)));
requested_calls++;
grpc_event ev;
diff --git a/test/core/fling/server.c b/test/core/fling/server.c
index 0f0f22f..b3a7fa2 100644
--- a/test/core/fling/server.c
+++ b/test/core/fling/server.c
@@ -77,8 +77,10 @@
static void request_call(void) {
grpc_metadata_array_init(&request_metadata_recv);
- grpc_server_request_call(server, &call, &call_details, &request_metadata_recv,
- cq, cq, tag(FLING_SERVER_NEW_REQUEST));
+ GPR_ASSERT(GRPC_CALL_OK ==
+ grpc_server_request_call(server, &call, &call_details,
+ &request_metadata_recv, cq, cq,
+ tag(FLING_SERVER_NEW_REQUEST)));
}
static void handle_unary_method(void) {
diff --git a/test/core/surface/completion_queue_test.c b/test/core/surface/completion_queue_test.c
index f9d88d6..869ac96 100644
--- a/test/core/surface/completion_queue_test.c
+++ b/test/core/surface/completion_queue_test.c
@@ -144,7 +144,7 @@
cc = grpc_completion_queue_create(
grpc_completion_queue_factory_lookup(&attr), &attr, NULL);
- grpc_cq_begin_op(cc, tag);
+ GPR_ASSERT(grpc_cq_begin_op(cc, tag) == 0);
grpc_cq_end_op(&exec_ctx, cc, tag, GRPC_ERROR_NONE,
do_nothing_end_completion, NULL, &completion);
@@ -233,7 +233,7 @@
grpc_completion_queue_factory_lookup(&attr), &attr, NULL);
for (i = 0; i < GPR_ARRAY_SIZE(tags); i++) {
- grpc_cq_begin_op(cc, tags[i]);
+ GPR_ASSERT(grpc_cq_begin_op(cc, tags[i]) == 0);
grpc_cq_end_op(&exec_ctx, cc, tags[i], GRPC_ERROR_NONE,
do_nothing_end_completion, NULL, &completions[i]);
}
@@ -245,7 +245,7 @@
}
for (i = 0; i < GPR_ARRAY_SIZE(tags); i++) {
- grpc_cq_begin_op(cc, tags[i]);
+ GPR_ASSERT(grpc_cq_begin_op(cc, tags[i]) == 0);
grpc_cq_end_op(&exec_ctx, cc, tags[i], GRPC_ERROR_NONE,
do_nothing_end_completion, NULL, &completions[i]);
}
diff --git a/test/core/surface/completion_queue_threading_test.c b/test/core/surface/completion_queue_threading_test.c
index 99d0fa4..9120cbb 100644
--- a/test/core/surface/completion_queue_threading_test.c
+++ b/test/core/surface/completion_queue_threading_test.c
@@ -107,7 +107,7 @@
GPR_ASSERT(ev.type == GRPC_QUEUE_TIMEOUT);
for (i = 0; i < GPR_ARRAY_SIZE(tags); i++) {
- grpc_cq_begin_op(cc, tags[i]);
+ GPR_ASSERT(grpc_cq_begin_op(cc, tags[i]) == 0);
grpc_cq_end_op(&exec_ctx, cc, tags[i], GRPC_ERROR_NONE,
do_nothing_end_completion, NULL, &completions[i]);
}
@@ -153,7 +153,7 @@
gpr_log(GPR_INFO, "producer %d phase 1", opt->id);
for (i = 0; i < TEST_THREAD_EVENTS; i++) {
- grpc_cq_begin_op(opt->cc, (void *)(intptr_t)1);
+ GPR_ASSERT(grpc_cq_begin_op(opt->cc, (void *)(intptr_t)1) == 0);
}
gpr_log(GPR_INFO, "producer %d phase 1 done", opt->id);
diff --git a/test/cpp/microbenchmarks/bm_cq.cc b/test/cpp/microbenchmarks/bm_cq.cc
index 18308a2..104fe51 100644
--- a/test/cpp/microbenchmarks/bm_cq.cc
+++ b/test/cpp/microbenchmarks/bm_cq.cc
@@ -23,6 +23,7 @@
#include <grpc++/completion_queue.h>
#include <grpc++/impl/grpc_library.h>
#include <grpc/grpc.h>
+#include <grpc/support/log.h>
#include "test/cpp/microbenchmarks/helpers.h"
extern "C" {
@@ -82,7 +83,7 @@
grpc_cq_completion completion;
DummyTag dummy_tag;
grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT;
- grpc_cq_begin_op(c_cq, &dummy_tag);
+ GPR_ASSERT(grpc_cq_begin_op(c_cq, &dummy_tag) == 0);
grpc_cq_end_op(&exec_ctx, c_cq, &dummy_tag, GRPC_ERROR_NONE,
DoneWithCompletionOnStack, NULL, &completion);
grpc_exec_ctx_finish(&exec_ctx);
@@ -102,7 +103,7 @@
while (state.KeepRunning()) {
grpc_cq_completion completion;
grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT;
- grpc_cq_begin_op(cq, NULL);
+ GPR_ASSERT(grpc_cq_begin_op(cq, NULL) == 0);
grpc_cq_end_op(&exec_ctx, cq, NULL, GRPC_ERROR_NONE,
DoneWithCompletionOnStack, NULL, &completion);
grpc_exec_ctx_finish(&exec_ctx);
@@ -121,7 +122,7 @@
while (state.KeepRunning()) {
grpc_cq_completion completion;
grpc_exec_ctx exec_ctx = GRPC_EXEC_CTX_INIT;
- grpc_cq_begin_op(cq, NULL);
+ GPR_ASSERT(grpc_cq_begin_op(cq, NULL) == 0);
grpc_cq_end_op(&exec_ctx, cq, NULL, GRPC_ERROR_NONE,
DoneWithCompletionOnStack, NULL, &completion);
grpc_exec_ctx_finish(&exec_ctx);
diff --git a/test/cpp/microbenchmarks/bm_cq_multiple_threads.cc b/test/cpp/microbenchmarks/bm_cq_multiple_threads.cc
index f79db15..c1439ba 100644
--- a/test/cpp/microbenchmarks/bm_cq_multiple_threads.cc
+++ b/test/cpp/microbenchmarks/bm_cq_multiple_threads.cc
@@ -78,7 +78,7 @@
}
gpr_mu_unlock(&ps->mu);
- grpc_cq_begin_op(g_cq, g_tag);
+ GPR_ASSERT(grpc_cq_begin_op(g_cq, g_tag) == 0);
grpc_cq_end_op(exec_ctx, g_cq, g_tag, GRPC_ERROR_NONE, cq_done_cb, NULL,
(grpc_cq_completion*)gpr_malloc(sizeof(grpc_cq_completion)));
grpc_exec_ctx_flush(exec_ctx);