From 9c12bc252e7bb8cdfe051884862e17d4850f3ab0 Mon Sep 17 00:00:00 2001 From: Dan Born Date: Wed, 13 Jan 2016 16:52:20 -0800 Subject: [PATCH] Add shutdown_starting callbacks to tcp_server. tcp_server_posix_test illustrates how this can be used to implement a weak referencing mechanism. --- src/core/iomgr/tcp_server.h | 7 ++++ src/core/iomgr/tcp_server_posix.c | 20 +++++++++- src/core/iomgr/tcp_server_windows.c | 20 +++++++++- test/core/iomgr/tcp_server_posix_test.c | 52 +++++++++++++++++++++++-- 4 files changed, 93 insertions(+), 6 deletions(-) diff --git a/src/core/iomgr/tcp_server.h b/src/core/iomgr/tcp_server.h index 5199c25eed5..8f3184ff1e0 100644 --- a/src/core/iomgr/tcp_server.h +++ b/src/core/iomgr/tcp_server.h @@ -34,6 +34,7 @@ #ifndef GRPC_INTERNAL_CORE_IOMGR_TCP_SERVER_H #define GRPC_INTERNAL_CORE_IOMGR_TCP_SERVER_H +#include "src/core/iomgr/closure.h" #include "src/core/iomgr/endpoint.h" /* Forward decl of grpc_tcp_server */ @@ -90,6 +91,12 @@ int grpc_tcp_server_port_fd(grpc_tcp_server *s, unsigned port_index, /* Ref s and return s. */ grpc_tcp_server *grpc_tcp_server_ref(grpc_tcp_server *s); +/* shutdown_starting is called when ref count has reached zero and the server is + about to be destroyed. The server will be deleted after it returns. Calling + grpc_tcp_server_ref() from it has no effect. */ +void grpc_tcp_server_shutdown_starting_add(grpc_tcp_server *s, + grpc_closure *shutdown_starting); + /* If the recount drops to zero, delete s, and call (exec_ctx==NULL) or enqueue a call (exec_ctx!=NULL) to shutdown_complete. */ void grpc_tcp_server_unref(grpc_exec_ctx *exec_ctx, grpc_tcp_server *s); diff --git a/src/core/iomgr/tcp_server_posix.c b/src/core/iomgr/tcp_server_posix.c index 6a22c08edba..adf14aeb597 100644 --- a/src/core/iomgr/tcp_server_posix.c +++ b/src/core/iomgr/tcp_server_posix.c @@ -128,6 +128,9 @@ struct grpc_tcp_server { grpc_tcp_listener *tail; unsigned nports; + /* List of closures passed to shutdown_starting_add(). */ + grpc_closure_list shutdown_starting; + /* shutdown callback */ grpc_closure *shutdown_complete; @@ -144,6 +147,8 @@ grpc_tcp_server *grpc_tcp_server_create(grpc_closure *shutdown_complete) { s->active_ports = 0; s->destroyed_ports = 0; s->shutdown = 0; + s->shutdown_starting.head = NULL; + s->shutdown_starting.tail = NULL; s->shutdown_complete = shutdown_complete; s->on_accept_cb = NULL; s->on_accept_cb_arg = NULL; @@ -585,13 +590,26 @@ grpc_tcp_server *grpc_tcp_server_ref(grpc_tcp_server *s) { return s; } +void grpc_tcp_server_shutdown_starting_add(grpc_tcp_server *s, + grpc_closure *shutdown_starting) { + gpr_mu_lock(&s->mu); + grpc_closure_list_add(&s->shutdown_starting, shutdown_starting, 1); + gpr_mu_unlock(&s->mu); +} + void grpc_tcp_server_unref(grpc_exec_ctx *exec_ctx, grpc_tcp_server *s) { if (gpr_unref(&s->refs)) { + /* Complete shutdown_starting work before destroying. */ + grpc_exec_ctx local_exec_ctx = GRPC_EXEC_CTX_INIT; + gpr_mu_lock(&s->mu); + grpc_exec_ctx_enqueue_list(&local_exec_ctx, &s->shutdown_starting); + gpr_mu_unlock(&s->mu); if (exec_ctx == NULL) { - grpc_exec_ctx local_exec_ctx = GRPC_EXEC_CTX_INIT; + grpc_exec_ctx_flush(&local_exec_ctx); tcp_server_destroy(&local_exec_ctx, s); grpc_exec_ctx_finish(&local_exec_ctx); } else { + grpc_exec_ctx_finish(&local_exec_ctx); tcp_server_destroy(exec_ctx, s); } } diff --git a/src/core/iomgr/tcp_server_windows.c b/src/core/iomgr/tcp_server_windows.c index 46617a6c652..8ee8149f251 100644 --- a/src/core/iomgr/tcp_server_windows.c +++ b/src/core/iomgr/tcp_server_windows.c @@ -93,6 +93,9 @@ struct grpc_tcp_server { grpc_tcp_listener *head; grpc_tcp_listener *tail; + /* List of closures passed to shutdown_starting_add(). */ + grpc_closure_list shutdown_starting; + /* shutdown callback */ grpc_closure *shutdown_complete; }; @@ -108,6 +111,8 @@ grpc_tcp_server *grpc_tcp_server_create(grpc_closure *shutdown_complete) { s->on_accept_cb_arg = NULL; s->head = NULL; s->tail = NULL; + s->shutdown_starting.head = NULL; + s->shutdown_starting.tail = NULL; s->shutdown_complete = shutdown_complete; return s; } @@ -135,6 +140,13 @@ grpc_tcp_server *grpc_tcp_server_ref(grpc_tcp_server *s) { return s; } +void grpc_tcp_server_shutdown_starting_add(grpc_tcp_server *s, + grpc_closure *shutdown_starting) { + gpr_mu_lock(&s->mu); + grpc_closure_list_add(&s->shutdown_starting, shutdown_starting, 1); + gpr_mu_unlock(&s->mu); +} + static void tcp_server_destroy(grpc_exec_ctx *exec_ctx, grpc_tcp_server *s) { int immediately_done = 0; grpc_tcp_listener *sp; @@ -158,11 +170,17 @@ static void tcp_server_destroy(grpc_exec_ctx *exec_ctx, grpc_tcp_server *s) { void grpc_tcp_server_unref(grpc_exec_ctx *exec_ctx, grpc_tcp_server *s) { if (gpr_unref(&s->refs)) { + /* Complete shutdown_starting work before destroying. */ + grpc_exec_ctx local_exec_ctx = GRPC_EXEC_CTX_INIT; + gpr_mu_lock(&s->mu); + grpc_exec_ctx_enqueue_list(&local_exec_ctx, &s->shutdown_starting); + gpr_mu_unlock(&s->mu); if (exec_ctx == NULL) { - grpc_exec_ctx local_exec_ctx = GRPC_EXEC_CTX_INIT; + grpc_exec_ctx_flush(&local_exec_ctx); tcp_server_destroy(&local_exec_ctx, s); grpc_exec_ctx_finish(&local_exec_ctx); } else { + grpc_exec_ctx_finish(&local_exec_ctx); tcp_server_destroy(exec_ctx, s); } } diff --git a/test/core/iomgr/tcp_server_posix_test.c b/test/core/iomgr/tcp_server_posix_test.c index 0f92166ab26..23b368df3af 100644 --- a/test/core/iomgr/tcp_server_posix_test.c +++ b/test/core/iomgr/tcp_server_posix_test.c @@ -60,15 +60,24 @@ typedef struct on_connect_result { int server_fd; } on_connect_result; -void on_connect_result_init(on_connect_result *result) { +typedef struct server_weak_ref { + grpc_tcp_server *server; + + /* arg is this server_weak_ref. */ + grpc_closure server_shutdown; +} server_weak_ref; + +static on_connect_result g_result = {NULL, 0, 0, -1}; + +static void on_connect_result_init(on_connect_result *result) { result->server = NULL; result->port_index = 0; result->fd_index = 0; result->server_fd = -1; } -void on_connect_result_set(on_connect_result *result, - const grpc_tcp_server_acceptor *acceptor) { +static void on_connect_result_set(on_connect_result *result, + const grpc_tcp_server_acceptor *acceptor) { result->server = grpc_tcp_server_ref(acceptor->from_server); result->port_index = acceptor->port_index; result->fd_index = acceptor->fd_index; @@ -76,7 +85,29 @@ void on_connect_result_set(on_connect_result *result, result->server, acceptor->port_index, acceptor->fd_index); } -static on_connect_result g_result = {NULL, 0, 0, -1}; + +static void server_weak_ref_shutdown(grpc_exec_ctx *exec_ctx, void *arg, + int success) { + server_weak_ref *weak_ref = arg; + weak_ref->server = NULL; +} + +static void server_weak_ref_init(server_weak_ref *weak_ref) { + weak_ref->server = NULL; + grpc_closure_init(&weak_ref->server_shutdown, server_weak_ref_shutdown, + weak_ref); +} + +/* Make weak_ref->server_shutdown a shutdown_starting cb on server. + grpc_tcp_server promises that the server object will live until + weak_ref->server_shutdown has returned. A strong ref on grpc_tcp_server + should be held until server_weak_ref_set() returns to avoid a race where the + server is deleted before the shutdown_starting cb is added. */ +static void server_weak_ref_set(server_weak_ref *weak_ref, + grpc_tcp_server *server) { + grpc_tcp_server_shutdown_starting_add(server, &weak_ref->server_shutdown); + weak_ref->server = server; +} static void on_connect(grpc_exec_ctx *exec_ctx, void *arg, grpc_endpoint *tcp, grpc_tcp_server_acceptor *acceptor) { @@ -182,6 +213,8 @@ static void test_connect(unsigned n) { grpc_tcp_server *s = grpc_tcp_server_create(NULL); grpc_pollset *pollsets[1]; unsigned i; + server_weak_ref weak_ref; + server_weak_ref_init(&weak_ref); LOG_TEST("test_connect"); gpr_log(GPR_INFO, "clients=%d", n); memset(&addr, 0, sizeof(addr)); @@ -242,6 +275,9 @@ static void test_connect(unsigned n) { GPR_ASSERT(result.port_index == 0); GPR_ASSERT(result.fd_index < svr_fd_count); GPR_ASSERT(result.server == s); + if (weak_ref.server == NULL) { + server_weak_ref_set(&weak_ref, result.server); + } grpc_tcp_server_unref(&exec_ctx, result.server); on_connect_result_init(&result); @@ -256,7 +292,15 @@ static void test_connect(unsigned n) { grpc_tcp_server_unref(&exec_ctx, result.server); } + /* Weak ref to server valid until final unref. */ + GPR_ASSERT(weak_ref.server != NULL); + GPR_ASSERT(grpc_tcp_server_port_fd(s, 0, 0) >= 0); + grpc_tcp_server_unref(&exec_ctx, s); + + /* Weak ref lost. */ + GPR_ASSERT(weak_ref.server == NULL); + grpc_exec_ctx_finish(&exec_ctx); }