Chiebot-Mirror
/
grpc
mirror of https://github.com/grpc/grpc.git
8
0
Fork 0
Code Issues Projects Releases Wiki Activity

Add shutdown_starting callbacks to tcp_server.

Browse Source 
tcp_server_posix_test illustrates how this can be used to implement a
weak referencing mechanism.
pull/4745/head
Dan Born 9 years ago
parent b13a69da41
commit 9c12bc252e
4 changed files with 93 additions and 6 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Split View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 7
      src/core/iomgr/tcp_server.h
  2. 20
      src/core/iomgr/tcp_server_posix.c
  3. 20
      src/core/iomgr/tcp_server_windows.c
  4. 52
      test/core/iomgr/tcp_server_posix_test.c

7
src/core/iomgr/tcp_server.h
Unescape View File

@ -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);

20
src/core/iomgr/tcp_server_posix.c
Unescape View File

@ -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);
}
}

20
src/core/iomgr/tcp_server_windows.c
Unescape View File

@ -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);
}
}

52
test/core/iomgr/tcp_server_posix_test.c
Unescape View File

@ -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);
}

Write Preview
Loading…
Cancel
Save