aboutsummaryrefslogtreecommitdiffhomepage
path: root/src
diff options
context:
space:
mode:
Diffstat (limited to 'src')
-rw-r--r--src/core/iomgr/pollset_posix.c31
-rw-r--r--src/core/support/sync.c4
-rw-r--r--src/core/surface/completion_queue.c38
-rw-r--r--src/core/surface/server.c198
-rw-r--r--src/cpp/server/server.cc13
-rw-r--r--src/csharp/Grpc.Core/Internal/ServerSafeHandle.cs21
-rw-r--r--src/csharp/Grpc.Core/Server.cs24
-rw-r--r--src/csharp/ext/grpc_csharp_ext.c11
-rw-r--r--src/node/ext/server.cc41
-rw-r--r--src/node/ext/server.h3
-rw-r--r--src/php/ext/grpc/server.c3
11 files changed, 248 insertions, 139 deletions
diff --git a/src/core/iomgr/pollset_posix.c b/src/core/iomgr/pollset_posix.c
index a8e6069002..d2f615271e 100644
--- a/src/core/iomgr/pollset_posix.c
+++ b/src/core/iomgr/pollset_posix.c
@@ -174,8 +174,6 @@ void grpc_pollset_del_fd(grpc_pollset *pollset, grpc_fd *fd) {
int grpc_pollset_work(grpc_pollset *pollset, gpr_timespec deadline) {
/* pollset->mu already held */
gpr_timespec now = gpr_now();
- /* FIXME(ctiller): see below */
- gpr_timespec maximum_deadline = gpr_time_add(now, gpr_time_from_seconds(1));
int r;
if (gpr_time_cmp(now, deadline) > 0) {
return 0;
@@ -186,14 +184,25 @@ int grpc_pollset_work(grpc_pollset *pollset, gpr_timespec deadline) {
if (grpc_alarm_check(&pollset->mu, now, &deadline)) {
return 1;
}
- /* FIXME(ctiller): we should not clamp deadline, however we have some
- stuck at shutdown bugs that this resolves */
- if (gpr_time_cmp(deadline, maximum_deadline) > 0) {
- deadline = maximum_deadline;
+ if (pollset->shutting_down) {
+ return 1;
}
gpr_tls_set(&g_current_thread_poller, (gpr_intptr)pollset);
r = pollset->vtable->maybe_work(pollset, deadline, now, 1);
gpr_tls_set(&g_current_thread_poller, 0);
+ if (pollset->shutting_down) {
+ if (pollset->counter > 0) {
+ grpc_pollset_kick(pollset);
+ } else if (pollset->in_flight_cbs == 0) {
+ gpr_mu_unlock(&pollset->mu);
+ pollset->shutdown_done_cb(pollset->shutdown_done_arg);
+ /* Continuing to access pollset here is safe -- it is the caller's
+ * responsibility to not destroy when it has outstanding calls to
+ * grpc_pollset_work.
+ * TODO(dklempner): Can we refactor the shutdown logic to avoid this? */
+ gpr_mu_lock(&pollset->mu);
+ }
+ }
return r;
}
@@ -201,13 +210,19 @@ void grpc_pollset_shutdown(grpc_pollset *pollset,
void (*shutdown_done)(void *arg),
void *shutdown_done_arg) {
int in_flight_cbs;
+ int counter;
gpr_mu_lock(&pollset->mu);
pollset->shutting_down = 1;
in_flight_cbs = pollset->in_flight_cbs;
+ counter = pollset->counter;
pollset->shutdown_done_cb = shutdown_done;
pollset->shutdown_done_arg = shutdown_done_arg;
+ if (counter > 0) {
+ grpc_pollset_kick(pollset);
+ }
gpr_mu_unlock(&pollset->mu);
- if (in_flight_cbs == 0) {
+
+ if (in_flight_cbs == 0 && counter == 0) {
shutdown_done(shutdown_done_arg);
}
}
@@ -294,7 +309,7 @@ static void unary_poll_do_promote(void *args, int success) {
pollset->in_flight_cbs--;
if (pollset->shutting_down) {
/* We don't care about this pollset anymore. */
- if (pollset->in_flight_cbs == 0) {
+ if (pollset->in_flight_cbs == 0 && pollset->counter == 0) {
do_shutdown_cb = 1;
}
} else if (grpc_fd_is_orphaned(fd)) {
diff --git a/src/core/support/sync.c b/src/core/support/sync.c
index ccfe1e25f4..856b5adb86 100644
--- a/src/core/support/sync.c
+++ b/src/core/support/sync.c
@@ -118,7 +118,9 @@ void gpr_refn(gpr_refcount *r, int n) {
}
int gpr_unref(gpr_refcount *r) {
- return gpr_atm_full_fetch_add(&r->count, -1) == 1;
+ gpr_atm prior = gpr_atm_full_fetch_add(&r->count, -1);
+ GPR_ASSERT(prior > 0);
+ return prior == 1;
}
void gpr_stats_init(gpr_stats_counter *c, gpr_intptr n) {
diff --git a/src/core/surface/completion_queue.c b/src/core/surface/completion_queue.c
index 8c9ca48a05..b48fbace31 100644
--- a/src/core/surface/completion_queue.c
+++ b/src/core/surface/completion_queue.c
@@ -83,7 +83,8 @@ grpc_completion_queue *grpc_completion_queue_create(void) {
memset(cc, 0, sizeof(*cc));
/* Initial ref is dropped by grpc_completion_queue_shutdown */
gpr_ref_init(&cc->refs, 1);
- gpr_ref_init(&cc->owning_refs, 1);
+ /* 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;
@@ -95,14 +96,14 @@ void grpc_cq_internal_ref(grpc_completion_queue *cc) {
static void on_pollset_destroy_done(void *arg) {
grpc_completion_queue *cc = arg;
- grpc_pollset_destroy(&cc->pollset);
- gpr_free(cc);
+ grpc_cq_internal_unref(cc);
}
void grpc_cq_internal_unref(grpc_completion_queue *cc) {
if (gpr_unref(&cc->owning_refs)) {
GPR_ASSERT(cc->queue == NULL);
- grpc_pollset_shutdown(&cc->pollset, on_pollset_destroy_done, cc);
+ grpc_pollset_destroy(&cc->pollset);
+ gpr_free(cc);
}
}
@@ -145,25 +146,25 @@ void grpc_cq_begin_op(grpc_completion_queue *cc, grpc_call *call) {
/* Signal the end of an operation - if this is the last waiting-to-be-queued
event, then enter shutdown mode */
-static void end_op_locked(grpc_completion_queue *cc,
- grpc_completion_type type) {
- if (gpr_unref(&cc->refs)) {
- GPR_ASSERT(!cc->shutdown);
- GPR_ASSERT(cc->shutdown_called);
- cc->shutdown = 1;
- gpr_cv_broadcast(GRPC_POLLSET_CV(&cc->pollset));
- }
-}
-
void grpc_cq_end_op(grpc_completion_queue *cc, void *tag, grpc_call *call,
int success) {
event *ev;
+ int shutdown = 0;
gpr_mu_lock(GRPC_POLLSET_MU(&cc->pollset));
ev = add_locked(cc, GRPC_OP_COMPLETE, tag, call);
ev->base.success = success;
- end_op_locked(cc, GRPC_OP_COMPLETE);
+ if (gpr_unref(&cc->refs)) {
+ 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));
if (call) GRPC_CALL_INTERNAL_UNREF(call, "cq", 0);
+ if (shutdown) {
+ grpc_pollset_shutdown(&cc->pollset, on_pollset_destroy_done, cc);
+ }
}
/* Create a GRPC_QUEUE_SHUTDOWN event without queuing it anywhere */
@@ -179,6 +180,7 @@ grpc_event grpc_completion_queue_next(grpc_completion_queue *cc,
event *ev = NULL;
grpc_event ret;
+ grpc_cq_internal_ref(cc);
gpr_mu_lock(GRPC_POLLSET_MU(&cc->pollset));
for (;;) {
if (cc->queue != NULL) {
@@ -214,6 +216,7 @@ grpc_event grpc_completion_queue_next(grpc_completion_queue *cc,
memset(&ret, 0, sizeof(ret));
ret.type = GRPC_QUEUE_TIMEOUT;
GRPC_SURFACE_TRACE_RETURNED_EVENT(cc, &ret);
+ grpc_cq_internal_unref(cc);
return ret;
}
}
@@ -221,6 +224,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);
return ret;
}
@@ -258,6 +262,7 @@ grpc_event grpc_completion_queue_pluck(grpc_completion_queue *cc, void *tag,
event *ev = NULL;
grpc_event ret;
+ grpc_cq_internal_ref(cc);
gpr_mu_lock(GRPC_POLLSET_MU(&cc->pollset));
for (;;) {
if ((ev = pluck_event(cc, tag))) {
@@ -276,6 +281,7 @@ grpc_event grpc_completion_queue_pluck(grpc_completion_queue *cc, void *tag,
memset(&ret, 0, sizeof(ret));
ret.type = GRPC_QUEUE_TIMEOUT;
GRPC_SURFACE_TRACE_RETURNED_EVENT(cc, &ret);
+ grpc_cq_internal_unref(cc);
return ret;
}
}
@@ -283,6 +289,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);
return ret;
}
@@ -299,6 +306,7 @@ void grpc_completion_queue_shutdown(grpc_completion_queue *cc) {
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);
}
}
diff --git a/src/core/surface/server.c b/src/core/surface/server.c
index 733f0e8a11..f9172d9850 100644
--- a/src/core/surface/server.c
+++ b/src/core/surface/server.c
@@ -127,6 +127,11 @@ struct channel_data {
grpc_iomgr_closure finish_destroy_channel_closure;
};
+typedef struct shutdown_tag {
+ void *tag;
+ grpc_completion_queue *cq;
+} shutdown_tag;
+
struct grpc_server {
size_t channel_filter_count;
const grpc_channel_filter **channel_filters;
@@ -137,14 +142,14 @@ struct grpc_server {
size_t cq_count;
gpr_mu mu;
- gpr_cv cv;
registered_method *registered_methods;
requested_call_array requested_calls;
gpr_uint8 shutdown;
+ gpr_uint8 shutdown_published;
size_t num_shutdown_tags;
- void **shutdown_tags;
+ shutdown_tag *shutdown_tags;
call_data *lists[CALL_LIST_COUNT];
channel_data root_channel_data;
@@ -261,29 +266,32 @@ static void server_ref(grpc_server *server) {
gpr_ref(&server->internal_refcount);
}
-static void server_unref(grpc_server *server) {
+static void server_delete(grpc_server *server) {
registered_method *rm;
size_t i;
+ grpc_channel_args_destroy(server->channel_args);
+ gpr_mu_destroy(&server->mu);
+ gpr_free(server->channel_filters);
+ requested_call_array_destroy(&server->requested_calls);
+ while ((rm = server->registered_methods) != NULL) {
+ server->registered_methods = rm->next;
+ gpr_free(rm->method);
+ gpr_free(rm->host);
+ requested_call_array_destroy(&rm->requested);
+ gpr_free(rm);
+ }
+ for (i = 0; i < server->cq_count; i++) {
+ grpc_cq_internal_unref(server->cqs[i]);
+ }
+ gpr_free(server->cqs);
+ gpr_free(server->pollsets);
+ gpr_free(server->shutdown_tags);
+ gpr_free(server);
+}
+
+static void server_unref(grpc_server *server) {
if (gpr_unref(&server->internal_refcount)) {
- grpc_channel_args_destroy(server->channel_args);
- gpr_mu_destroy(&server->mu);
- gpr_cv_destroy(&server->cv);
- gpr_free(server->channel_filters);
- requested_call_array_destroy(&server->requested_calls);
- while ((rm = server->registered_methods) != NULL) {
- server->registered_methods = rm->next;
- gpr_free(rm->method);
- gpr_free(rm->host);
- requested_call_array_destroy(&rm->requested);
- gpr_free(rm);
- }
- for (i = 0; i < server->cq_count; i++) {
- grpc_cq_internal_unref(server->cqs[i]);
- }
- gpr_free(server->cqs);
- gpr_free(server->pollsets);
- gpr_free(server->shutdown_tags);
- gpr_free(server);
+ server_delete(server);
}
}
@@ -378,6 +386,26 @@ static void kill_zombie(void *elem, int success) {
grpc_call_destroy(grpc_call_from_top_element(elem));
}
+static int num_listeners(grpc_server *server) {
+ listener *l;
+ int n = 0;
+ for (l = server->listeners; l; l = l->next) {
+ n++;
+ }
+ return n;
+}
+
+static void maybe_finish_shutdown(grpc_server *server) {
+ size_t i;
+ if (server->shutdown && !server->shutdown_published && server->lists[ALL_CALLS] == NULL && server->listeners_destroyed == num_listeners(server)) {
+ server->shutdown_published = 1;
+ for (i = 0; i < server->num_shutdown_tags; i++) {
+ grpc_cq_end_op(server->shutdown_tags[i].cq, server->shutdown_tags[i].tag,
+ NULL, 1);
+ }
+ }
+}
+
static grpc_mdelem *server_filter(void *user_data, grpc_mdelem *md) {
grpc_call_element *elem = user_data;
channel_data *chand = elem->channel_data;
@@ -441,6 +469,9 @@ static void server_on_recv(void *ptr, int success) {
grpc_iomgr_add_callback(&calld->kill_zombie_closure);
}
+ if (call_list_remove(calld, ALL_CALLS)) {
+ maybe_finish_shutdown(chand->server);
+ }
gpr_mu_unlock(&chand->server->mu);
break;
}
@@ -539,19 +570,15 @@ static void init_call_elem(grpc_call_element *elem,
static void destroy_call_elem(grpc_call_element *elem) {
channel_data *chand = elem->channel_data;
call_data *calld = elem->call_data;
- size_t i, j;
+ int removed[CALL_LIST_COUNT];
+ size_t i;
gpr_mu_lock(&chand->server->mu);
for (i = 0; i < CALL_LIST_COUNT; i++) {
- call_list_remove(elem->call_data, i);
+ removed[i] = call_list_remove(elem->call_data, i);
}
- if (chand->server->shutdown && chand->server->lists[ALL_CALLS] == NULL) {
- for (i = 0; i < chand->server->num_shutdown_tags; i++) {
- for (j = 0; j < chand->server->cq_count; j++) {
- grpc_cq_end_op(chand->server->cqs[j], chand->server->shutdown_tags[i],
- NULL, 1);
- }
- }
+ if (removed[ALL_CALLS]) {
+ maybe_finish_shutdown(chand->server);
}
gpr_mu_unlock(&chand->server->mu);
@@ -646,7 +673,6 @@ grpc_server *grpc_server_create_from_filters(grpc_channel_filter **filters,
memset(server, 0, sizeof(grpc_server));
gpr_mu_init(&server->mu);
- gpr_cv_init(&server->cv);
/* decremented by grpc_server_destroy */
gpr_ref_init(&server->internal_refcount, 1);
@@ -806,38 +832,28 @@ grpc_transport_setup_result grpc_server_setup_transport(
return result;
}
-static int num_listeners(grpc_server *server) {
- listener *l;
- int n = 0;
- for (l = server->listeners; l; l = l->next) {
- n++;
- }
- return n;
-}
-
-static void shutdown_internal(grpc_server *server, gpr_uint8 have_shutdown_tag,
- void *shutdown_tag) {
+void grpc_server_shutdown_and_notify(grpc_server *server,
+ grpc_completion_queue *cq, void *tag) {
listener *l;
requested_call_array requested_calls;
channel_data **channels;
channel_data *c;
size_t nchannels;
- size_t i, j;
+ size_t i;
grpc_channel_op op;
grpc_channel_element *elem;
registered_method *rm;
+ shutdown_tag *sdt;
/* lock, and gather up some stuff to do */
gpr_mu_lock(&server->mu);
- if (have_shutdown_tag) {
- for (i = 0; i < server->cq_count; i++) {
- grpc_cq_begin_op(server->cqs[i], NULL);
- }
- server->shutdown_tags =
- gpr_realloc(server->shutdown_tags,
- sizeof(void *) * (server->num_shutdown_tags + 1));
- server->shutdown_tags[server->num_shutdown_tags++] = shutdown_tag;
- }
+ grpc_cq_begin_op(cq, NULL);
+ server->shutdown_tags =
+ gpr_realloc(server->shutdown_tags,
+ sizeof(shutdown_tag) * (server->num_shutdown_tags + 1));
+ sdt = &server->shutdown_tags[server->num_shutdown_tags++];
+ sdt->tag = tag;
+ sdt->cq = cq;
if (server->shutdown) {
gpr_mu_unlock(&server->mu);
return;
@@ -878,13 +894,7 @@ static void shutdown_internal(grpc_server *server, gpr_uint8 have_shutdown_tag,
}
server->shutdown = 1;
- if (server->lists[ALL_CALLS] == NULL) {
- for (i = 0; i < server->num_shutdown_tags; i++) {
- for (j = 0; j < server->cq_count; j++) {
- grpc_cq_end_op(server->cqs[j], server->shutdown_tags[i], NULL, 1);
- }
- }
- }
+ maybe_finish_shutdown(server);
gpr_mu_unlock(&server->mu);
for (i = 0; i < nchannels; i++) {
@@ -914,46 +924,64 @@ static void shutdown_internal(grpc_server *server, gpr_uint8 have_shutdown_tag,
}
}
-void grpc_server_shutdown(grpc_server *server) {
- shutdown_internal(server, 0, NULL);
-}
-
-void grpc_server_shutdown_and_notify(grpc_server *server, void *tag) {
- shutdown_internal(server, 1, tag);
-}
-
void grpc_server_listener_destroy_done(void *s) {
grpc_server *server = s;
gpr_mu_lock(&server->mu);
server->listeners_destroyed++;
- gpr_cv_signal(&server->cv);
+ maybe_finish_shutdown(server);
gpr_mu_unlock(&server->mu);
}
-void grpc_server_destroy(grpc_server *server) {
- channel_data *c;
- listener *l;
- size_t i;
+void grpc_server_cancel_all_calls(grpc_server *server) {
call_data *calld;
+ grpc_call **calls;
+ size_t call_count;
+ size_t call_capacity;
+ int is_first = 1;
+ size_t i;
gpr_mu_lock(&server->mu);
- if (!server->shutdown) {
+
+ GPR_ASSERT(server->shutdown);
+
+ if (!server->lists[ALL_CALLS]) {
gpr_mu_unlock(&server->mu);
- grpc_server_shutdown(server);
- gpr_mu_lock(&server->mu);
+ return;
}
- while (server->listeners_destroyed != num_listeners(server)) {
- for (i = 0; i < server->cq_count; i++) {
- gpr_mu_unlock(&server->mu);
- grpc_cq_hack_spin_pollset(server->cqs[i]);
- gpr_mu_lock(&server->mu);
+ call_capacity = 8;
+ call_count = 0;
+ calls = gpr_malloc(sizeof(grpc_call *) * call_capacity);
+
+ for (calld = server->lists[ALL_CALLS]; calld != server->lists[ALL_CALLS] || is_first; calld = calld->links[ALL_CALLS].next) {
+ if (call_count == call_capacity) {
+ call_capacity *= 2;
+ calls = gpr_realloc(calls, sizeof(grpc_call *) * call_capacity);
}
+ calls[call_count++] = calld->call;
+ GRPC_CALL_INTERNAL_REF(calld->call, "cancel_all");
+ is_first = 0;
+ }
+
+ gpr_mu_unlock(&server->mu);
- gpr_cv_wait(&server->cv, &server->mu,
- gpr_time_add(gpr_now(), gpr_time_from_millis(100)));
+ for (i = 0; i < call_count; i++) {
+ grpc_call_cancel_with_status(calls[i], GRPC_STATUS_UNAVAILABLE, "Unavailable");
+ GRPC_CALL_INTERNAL_UNREF(calls[i], "cancel_all", 1);
}
+ gpr_free(calls);
+}
+
+void grpc_server_destroy(grpc_server *server) {
+ channel_data *c;
+ listener *l;
+ call_data *calld;
+
+ gpr_mu_lock(&server->mu);
+ GPR_ASSERT(server->shutdown);
+ GPR_ASSERT(server->listeners_destroyed == num_listeners(server));
+
while (server->listeners) {
l = server->listeners;
server->listeners = l->next;
@@ -962,10 +990,6 @@ void grpc_server_destroy(grpc_server *server) {
while ((calld = call_list_remove_head(&server->lists[PENDING_START],
PENDING_START)) != NULL) {
- /* TODO(dgq): If we knew the size of the call list (or an upper bound), we
- * could allocate all the memory for the closures in advance in a single
- * chunk */
- gpr_log(GPR_DEBUG, "server destroys call %p", calld->call);
calld->state = ZOMBIED;
grpc_iomgr_closure_init(
&calld->kill_zombie_closure, kill_zombie,
diff --git a/src/cpp/server/server.cc b/src/cpp/server/server.cc
index 80eb488b41..dbd88c5b8c 100644
--- a/src/cpp/server/server.cc
+++ b/src/cpp/server/server.cc
@@ -52,6 +52,14 @@
namespace grpc {
+class Server::ShutdownRequest GRPC_FINAL : public CompletionQueueTag {
+ public:
+ bool FinalizeResult(void** tag, bool* status) {
+ delete this;
+ return false;
+ }
+};
+
class Server::SyncRequest GRPC_FINAL : public CompletionQueueTag {
public:
SyncRequest(RpcServiceMethod* method, void* tag)
@@ -217,6 +225,9 @@ Server::~Server() {
Shutdown();
}
}
+ void* got_tag;
+ bool ok;
+ GPR_ASSERT(!cq_.Next(&got_tag, &ok));
grpc_server_destroy(server_);
if (thread_pool_owned_) {
delete thread_pool_;
@@ -290,7 +301,7 @@ void Server::Shutdown() {
grpc::unique_lock<grpc::mutex> lock(mu_);
if (started_ && !shutdown_) {
shutdown_ = true;
- grpc_server_shutdown(server_);
+ grpc_server_shutdown_and_notify(server_, cq_.cq(), new ShutdownRequest());
cq_.Shutdown();
// Wait for running callbacks to finish.
diff --git a/src/csharp/Grpc.Core/Internal/ServerSafeHandle.cs b/src/csharp/Grpc.Core/Internal/ServerSafeHandle.cs
index edd9c490ff..eda9afcb87 100644
--- a/src/csharp/Grpc.Core/Internal/ServerSafeHandle.cs
+++ b/src/csharp/Grpc.Core/Internal/ServerSafeHandle.cs
@@ -60,10 +60,10 @@ namespace Grpc.Core.Internal
static extern GRPCCallError grpcsharp_server_request_call(ServerSafeHandle server, CompletionQueueSafeHandle cq, BatchContextSafeHandle ctx);
[DllImport("grpc_csharp_ext.dll")]
- static extern void grpcsharp_server_shutdown(ServerSafeHandle server);
+ static extern void grpcsharp_server_cancel_all_calls(ServerSafeHandle server);
[DllImport("grpc_csharp_ext.dll")]
- static extern void grpcsharp_server_shutdown_and_notify_callback(ServerSafeHandle server, BatchContextSafeHandle ctx);
+ static extern void grpcsharp_server_shutdown_and_notify_callback(ServerSafeHandle server, CompletionQueueSafeHandle cq, BatchContextSafeHandle ctx);
[DllImport("grpc_csharp_ext.dll")]
static extern void grpcsharp_server_destroy(IntPtr server);
@@ -91,17 +91,12 @@ namespace Grpc.Core.Internal
{
grpcsharp_server_start(this);
}
-
- public void Shutdown()
- {
- grpcsharp_server_shutdown(this);
- }
-
- public void ShutdownAndNotify(BatchCompletionDelegate callback)
+
+ public void ShutdownAndNotify(CompletionQueueSafeHandle cq, BatchCompletionDelegate callback)
{
var ctx = BatchContextSafeHandle.Create();
GrpcEnvironment.CompletionRegistry.RegisterBatchCompletion(ctx, callback);
- grpcsharp_server_shutdown_and_notify_callback(this, ctx);
+ grpcsharp_server_shutdown_and_notify_callback(this, cq, ctx);
}
public void RequestCall(CompletionQueueSafeHandle cq, BatchCompletionDelegate callback)
@@ -116,5 +111,11 @@ namespace Grpc.Core.Internal
grpcsharp_server_destroy(handle);
return true;
}
+
+ // Only to be called after ShutdownAndNotify.
+ public void CancelAllCalls()
+ {
+ grpcsharp_server_cancel_all_calls(this);
+ }
}
}
diff --git a/src/csharp/Grpc.Core/Server.cs b/src/csharp/Grpc.Core/Server.cs
index da59fc7232..3352fc93a1 100644
--- a/src/csharp/Grpc.Core/Server.cs
+++ b/src/csharp/Grpc.Core/Server.cs
@@ -52,9 +52,6 @@ namespace Grpc.Core
/// </summary>
public const int PickUnusedPort = 0;
- //readonly OpCompletionDelegate serverShutdownHandler;
- //readonly OpCompletionDelegate newServerRpcHandler;
-
readonly ServerSafeHandle handle;
readonly object myLock = new object();
@@ -67,8 +64,6 @@ namespace Grpc.Core
public Server()
{
this.handle = ServerSafeHandle.NewServer(GetCompletionQueue(), IntPtr.Zero);
- //this.newServerRpcHandler = HandleNewServerRpc;
- //this.serverShutdownHandler = HandleServerShutdown;
}
/// <summary>
@@ -142,8 +137,7 @@ namespace Grpc.Core
shutdownRequested = true;
}
- var ctx = BatchContextSafeHandle.Create();
- handle.ShutdownAndNotify(HandleServerShutdown);
+ handle.ShutdownAndNotify(GetCompletionQueue(), HandleServerShutdown);
await shutdownTcs.Task;
handle.Dispose();
}
@@ -159,8 +153,22 @@ namespace Grpc.Core
}
}
- public void Kill()
+ /// <summary>
+ /// Requests server shutdown while cancelling all the in-progress calls.
+ /// The returned task finishes when shutdown procedure is complete.
+ /// </summary>
+ public async Task KillAsync()
{
+ lock (myLock)
+ {
+ Preconditions.CheckState(startRequested);
+ Preconditions.CheckState(!shutdownRequested);
+ shutdownRequested = true;
+ }
+
+ handle.ShutdownAndNotify(GetCompletionQueue(), HandleServerShutdown);
+ handle.CancelAllCalls();
+ await shutdownTcs.Task;
handle.Dispose();
}
diff --git a/src/csharp/ext/grpc_csharp_ext.c b/src/csharp/ext/grpc_csharp_ext.c
index 57be445331..c2a0b729d4 100644
--- a/src/csharp/ext/grpc_csharp_ext.c
+++ b/src/csharp/ext/grpc_csharp_ext.c
@@ -616,14 +616,15 @@ GPR_EXPORT void GPR_CALLTYPE grpcsharp_server_start(grpc_server *server) {
grpc_server_start(server);
}
-GPR_EXPORT void GPR_CALLTYPE grpcsharp_server_shutdown(grpc_server *server) {
- grpc_server_shutdown(server);
-}
-
GPR_EXPORT void GPR_CALLTYPE
grpcsharp_server_shutdown_and_notify_callback(grpc_server *server,
+ grpc_completion_queue *cq,
grpcsharp_batch_context *ctx) {
- grpc_server_shutdown_and_notify(server, ctx);
+ grpc_server_shutdown_and_notify(server, cq, ctx);
+}
+
+GPR_EXPORT void GPR_CALLTYPE grpcsharp_server_cancel_all_calls(grpc_server *server) {
+ grpc_server_cancel_all_calls(server);
}
GPR_EXPORT void GPR_CALLTYPE grpcsharp_server_destroy(grpc_server *server) {
diff --git a/src/node/ext/server.cc b/src/node/ext/server.cc
index eb97f7348b..51c55ba965 100644
--- a/src/node/ext/server.cc
+++ b/src/node/ext/server.cc
@@ -112,9 +112,17 @@ class NewCallOp : public Op {
}
};
-Server::Server(grpc_server *server) : wrapped_server(server) {}
+Server::Server(grpc_server *server) : wrapped_server(server) {
+ shutdown_queue = grpc_completion_queue_create();
+ grpc_server_register_completion_queue(server, shutdown_queue);
+}
-Server::~Server() { grpc_server_destroy(wrapped_server); }
+Server::~Server() {
+ this->ShutdownServer();
+ grpc_completion_queue_shutdown(this->shutdown_queue);
+ grpc_server_destroy(wrapped_server);
+ grpc_completion_queue_destroy(this->shutdown_queue);
+}
void Server::Init(Handle<Object> exports) {
NanScope();
@@ -148,6 +156,16 @@ bool Server::HasInstance(Handle<Value> val) {
return NanHasInstance(fun_tpl, val);
}
+void Server::ShutdownServer() {
+ if (this->wrapped_server != NULL) {
+ grpc_server_shutdown_and_notify(this->wrapped_server,
+ this->shutdown_queue,
+ NULL);
+ grpc_completion_queue_pluck(this->shutdown_queue, NULL, gpr_inf_future);
+ this->wrapped_server = NULL;
+ }
+}
+
NAN_METHOD(Server::New) {
NanScope();
@@ -207,6 +225,9 @@ NAN_METHOD(Server::RequestCall) {
return NanThrowTypeError("requestCall can only be called on a Server");
}
Server *server = ObjectWrap::Unwrap<Server>(args.This());
+ if (server->wrapped_server == NULL) {
+ return NanThrowError("requestCall cannot be called on a shut down Server");
+ }
NewCallOp *op = new NewCallOp();
unique_ptr<OpVec> ops(new OpVec());
ops->push_back(unique_ptr<Op>(op));
@@ -232,6 +253,9 @@ NAN_METHOD(Server::AddHttp2Port) {
return NanThrowTypeError("addHttp2Port's argument must be a String");
}
Server *server = ObjectWrap::Unwrap<Server>(args.This());
+ if (server->wrapped_server == NULL) {
+ return NanThrowError("addHttp2Port cannot be called on a shut down Server");
+ }
NanReturnValue(NanNew<Number>(grpc_server_add_http2_port(
server->wrapped_server, *NanUtf8String(args[0]))));
}
@@ -251,6 +275,10 @@ NAN_METHOD(Server::AddSecureHttp2Port) {
"addSecureHttp2Port's second argument must be ServerCredentials");
}
Server *server = ObjectWrap::Unwrap<Server>(args.This());
+ if (server->wrapped_server == NULL) {
+ return NanThrowError(
+ "addSecureHttp2Port cannot be called on a shut down Server");
+ }
ServerCredentials *creds = ObjectWrap::Unwrap<ServerCredentials>(
args[1]->ToObject());
NanReturnValue(NanNew<Number>(grpc_server_add_secure_http2_port(
@@ -264,17 +292,24 @@ NAN_METHOD(Server::Start) {
return NanThrowTypeError("start can only be called on a Server");
}
Server *server = ObjectWrap::Unwrap<Server>(args.This());
+ if (server->wrapped_server == NULL) {
+ return NanThrowError("start cannot be called on a shut down Server");
+ }
grpc_server_start(server->wrapped_server);
NanReturnUndefined();
}
+NAN_METHOD(ShutdownCallback) {
+ NanReturnUndefined();
+}
+
NAN_METHOD(Server::Shutdown) {
NanScope();
if (!HasInstance(args.This())) {
return NanThrowTypeError("shutdown can only be called on a Server");
}
Server *server = ObjectWrap::Unwrap<Server>(args.This());
- grpc_server_shutdown(server->wrapped_server);
+ server->ShutdownServer();
NanReturnUndefined();
}
diff --git a/src/node/ext/server.h b/src/node/ext/server.h
index 641d5ccb3e..5b4b18a0e0 100644
--- a/src/node/ext/server.h
+++ b/src/node/ext/server.h
@@ -61,6 +61,8 @@ class Server : public ::node::ObjectWrap {
Server(const Server &);
Server &operator=(const Server &);
+ void ShutdownServer();
+
static NAN_METHOD(New);
static NAN_METHOD(RequestCall);
static NAN_METHOD(AddHttp2Port);
@@ -71,6 +73,7 @@ class Server : public ::node::ObjectWrap {
static v8::Persistent<v8::FunctionTemplate> fun_tpl;
grpc_server *wrapped_server;
+ grpc_completion_queue *shutdown_queue;
};
} // namespace node
diff --git a/src/php/ext/grpc/server.c b/src/php/ext/grpc/server.c
index b7995b6b8b..02c886c715 100644
--- a/src/php/ext/grpc/server.c
+++ b/src/php/ext/grpc/server.c
@@ -63,7 +63,8 @@ zend_class_entry *grpc_ce_server;
void free_wrapped_grpc_server(void *object TSRMLS_DC) {
wrapped_grpc_server *server = (wrapped_grpc_server *)object;
if (server->wrapped != NULL) {
- grpc_server_shutdown(server->wrapped);
+ grpc_server_shutdown_and_notify(server->wrapped, completion_queue, NULL);
+ grpc_completion_queue_pluck(completion_queue, NULL, gpr_inf_future);
grpc_server_destroy(server->wrapped);
}
efree(server);
generated by cgit v1.2.3 (git 2.39.1) at 2024-12-14 09:16:02 +0000