From aec96aa2232b3ab7f86bc08a99f82684406edf0b Mon Sep 17 00:00:00 2001 From: Craig Tiller Date: Tue, 7 Apr 2015 14:32:15 -0700 Subject: Fix server shutdown A previous fix to make close() occur later can cause socket reuse by servers to fail as previous sockets are left asynchronously open. This change: - adds a callback to TCP server shutdown to signal that the server is completely shutdown - wait for that callback before destroying listeners in the server (and before destroying the server) - handles fallout --- src/core/surface/completion_queue.c | 7 +++++++ 1 file changed, 7 insertions(+) (limited to 'src/core/surface/completion_queue.c') diff --git a/src/core/surface/completion_queue.c b/src/core/surface/completion_queue.c index 6a1d83ce5d..b08bd93693 100644 --- a/src/core/surface/completion_queue.c +++ b/src/core/surface/completion_queue.c @@ -432,3 +432,10 @@ void grpc_cq_dump_pending_ops(grpc_completion_queue *cc) { grpc_pollset *grpc_cq_pollset(grpc_completion_queue *cc) { return &cc->pollset; } + +void grpc_cq_hack_spin_pollset(grpc_completion_queue *cc) { + gpr_mu_lock(GRPC_POLLSET_MU(&cc->pollset)); + grpc_pollset_kick(&cc->pollset); + grpc_pollset_work(&cc->pollset, gpr_time_add(gpr_now(), gpr_time_from_millis(100))); + gpr_mu_unlock(GRPC_POLLSET_MU(&cc->pollset)); +} -- cgit v1.2.3 From c02c1d883a08a2b6a646e7d736028c48baac3026 Mon Sep 17 00:00:00 2001 From: Craig Tiller Date: Tue, 7 Apr 2015 16:21:55 -0700 Subject: clang-format --- src/core/iomgr/tcp_server.h | 6 ++--- src/core/iomgr/tcp_server_posix.c | 10 ++++---- src/core/iomgr/tcp_server_windows.c | 34 +++++++++++++-------------- src/core/security/server_secure_chttp2.c | 17 ++++++++------ src/core/surface/completion_queue.c | 3 ++- src/core/surface/completion_queue.h | 2 +- src/core/surface/server.c | 13 +++++----- src/core/surface/server.h | 5 ++-- src/core/surface/server_chttp2.c | 3 ++- test/core/end2end/tests/cancel_after_invoke.c | 3 ++- test/core/end2end/tests/cancel_test_helpers.h | 6 +++-- 11 files changed, 56 insertions(+), 46 deletions(-) (limited to 'src/core/surface/completion_queue.c') diff --git a/src/core/iomgr/tcp_server.h b/src/core/iomgr/tcp_server.h index 1e58901a7a..66bb3ef701 100644 --- a/src/core/iomgr/tcp_server.h +++ b/src/core/iomgr/tcp_server.h @@ -71,8 +71,8 @@ int grpc_tcp_server_add_port(grpc_tcp_server *s, const void *addr, up when grpc_tcp_server_destroy is called. */ int grpc_tcp_server_get_fd(grpc_tcp_server *s, unsigned index); -void grpc_tcp_server_destroy(grpc_tcp_server *server, - void (*shutdown_done)(void *shutdown_done_arg), +void grpc_tcp_server_destroy(grpc_tcp_server *server, + void (*shutdown_done)(void *shutdown_done_arg), void *shutdown_done_arg); -#endif /* GRPC_INTERNAL_CORE_IOMGR_TCP_SERVER_H */ +#endif /* GRPC_INTERNAL_CORE_IOMGR_TCP_SERVER_H */ diff --git a/src/core/iomgr/tcp_server_posix.c b/src/core/iomgr/tcp_server_posix.c index 482166e2eb..895f85fc68 100644 --- a/src/core/iomgr/tcp_server_posix.c +++ b/src/core/iomgr/tcp_server_posix.c @@ -154,13 +154,15 @@ static void destroyed_port(void *server, int success) { static void dont_care_about_shutdown_completion(void *ignored) {} -void grpc_tcp_server_destroy(grpc_tcp_server *s, - void (*shutdown_complete)(void *shutdown_complete_arg), - void *shutdown_complete_arg) { +void grpc_tcp_server_destroy( + grpc_tcp_server *s, void (*shutdown_complete)(void *shutdown_complete_arg), + void *shutdown_complete_arg) { size_t i; gpr_mu_lock(&s->mu); - s->shutdown_complete = shutdown_complete ? shutdown_complete : dont_care_about_shutdown_completion; + s->shutdown_complete = shutdown_complete + ? shutdown_complete + : dont_care_about_shutdown_completion; s->shutdown_complete_arg = shutdown_complete_arg; /* shutdown all fd's */ diff --git a/src/core/iomgr/tcp_server_windows.c b/src/core/iomgr/tcp_server_windows.c index 896c9e5d08..a43d5670a4 100644 --- a/src/core/iomgr/tcp_server_windows.c +++ b/src/core/iomgr/tcp_server_windows.c @@ -93,8 +93,8 @@ grpc_tcp_server *grpc_tcp_server_create(void) { } void grpc_tcp_server_destroy(grpc_tcp_server *s, - void(*shutdown_done)(void *shutdown_done_arg), - void *shutdown_done_arg) { + void (*shutdown_done)(void *shutdown_done_arg), + void *shutdown_done_arg) { size_t i; gpr_mu_lock(&s->mu); /* shutdown all fd's */ @@ -116,13 +116,13 @@ void grpc_tcp_server_destroy(grpc_tcp_server *s, gpr_free(s); if (shutdown_done) { - shutdown_done(shutdown_done_arg); + shutdown_done(shutdown_done_arg); } } /* Prepare a recently-created socket for listening. */ -static int prepare_socket(SOCKET sock, - const struct sockaddr *addr, int addr_len) { +static int prepare_socket(SOCKET sock, const struct sockaddr *addr, + int addr_len) { struct sockaddr_storage sockname_temp; socklen_t sockname_len; @@ -153,15 +153,15 @@ static int prepare_socket(SOCKET sock, } sockname_len = sizeof(sockname_temp); - if (getsockname(sock, (struct sockaddr *) &sockname_temp, &sockname_len) - == SOCKET_ERROR) { + if (getsockname(sock, (struct sockaddr *)&sockname_temp, &sockname_len) == + SOCKET_ERROR) { char *utf8_message = gpr_format_message(WSAGetLastError()); gpr_log(GPR_ERROR, "getsockname: %s", utf8_message); gpr_free(utf8_message); goto error; } - return grpc_sockaddr_get_port((struct sockaddr *) &sockname_temp); + return grpc_sockaddr_get_port((struct sockaddr *)&sockname_temp); error: if (sock != INVALID_SOCKET) closesocket(sock); @@ -227,8 +227,7 @@ static void on_accept(void *arg, int success) { DWORD transfered_bytes = 0; DWORD flags; BOOL wsa_success = WSAGetOverlappedResult(sock, &info->overlapped, - &transfered_bytes, FALSE, - &flags); + &transfered_bytes, FALSE, &flags); if (!wsa_success) { char *utf8_message = gpr_format_message(WSAGetLastError()); gpr_log(GPR_ERROR, "on_accept error: %s", utf8_message); @@ -263,9 +262,9 @@ static int add_socket_to_server(grpc_tcp_server *s, SOCKET sock, if (sock == INVALID_SOCKET) return -1; - status = WSAIoctl(sock, SIO_GET_EXTENSION_FUNCTION_POINTER, - &guid, sizeof(guid), &AcceptEx, sizeof(AcceptEx), - &ioctl_num_bytes, NULL, NULL); + status = + WSAIoctl(sock, SIO_GET_EXTENSION_FUNCTION_POINTER, &guid, sizeof(guid), + &AcceptEx, sizeof(AcceptEx), &ioctl_num_bytes, NULL, NULL); if (status != 0) { char *utf8_message = gpr_format_message(WSAGetLastError()); @@ -313,9 +312,8 @@ int grpc_tcp_server_add_port(grpc_tcp_server *s, const void *addr, for (i = 0; i < s->nports; i++) { sockname_len = sizeof(sockname_temp); if (0 == getsockname(s->ports[i].socket->socket, - (struct sockaddr *) &sockname_temp, - &sockname_len)) { - port = grpc_sockaddr_get_port((struct sockaddr *) &sockname_temp); + (struct sockaddr *)&sockname_temp, &sockname_len)) { + port = grpc_sockaddr_get_port((struct sockaddr *)&sockname_temp); if (port > 0) { allocated_addr = malloc(addr_len); memcpy(allocated_addr, addr, addr_len); @@ -336,7 +334,7 @@ int grpc_tcp_server_add_port(grpc_tcp_server *s, const void *addr, if (grpc_sockaddr_is_wildcard(addr, &port)) { grpc_sockaddr_make_wildcard6(port, &wildcard); - addr = (struct sockaddr *) &wildcard; + addr = (struct sockaddr *)&wildcard; addr_len = sizeof(wildcard); } @@ -375,4 +373,4 @@ void grpc_tcp_server_start(grpc_tcp_server *s, grpc_pollset **pollset, gpr_mu_unlock(&s->mu); } -#endif /* GPR_WINSOCK_SOCKET */ +#endif /* GPR_WINSOCK_SOCKET */ diff --git a/src/core/security/server_secure_chttp2.c b/src/core/security/server_secure_chttp2.c index d5b3a82b87..081272724c 100644 --- a/src/core/security/server_secure_chttp2.c +++ b/src/core/security/server_secure_chttp2.c @@ -85,10 +85,10 @@ static void on_secure_transport_setup_done(void *statep, if (status == GRPC_SECURITY_OK) { gpr_mu_lock(&state->mu); if (!state->is_shutdown) { - grpc_create_chttp2_transport( - setup_transport, state->server, - grpc_server_get_channel_args(state->server), - secure_endpoint, NULL, 0, grpc_mdctx_create(), 0); + grpc_create_chttp2_transport(setup_transport, state->server, + grpc_server_get_channel_args(state->server), + secure_endpoint, NULL, 0, + grpc_mdctx_create(), 0); } else { /* We need to consume this here, because the server may already have gone * away. */ @@ -104,7 +104,8 @@ static void on_secure_transport_setup_done(void *statep, static void on_accept(void *statep, grpc_endpoint *tcp) { grpc_server_secure_state *state = statep; state_ref(state); - grpc_setup_secure_transport(state->ctx, tcp, on_secure_transport_setup_done, state); + grpc_setup_secure_transport(state->ctx, tcp, on_secure_transport_setup_done, + state); } /* Server callback: start listening on our ports */ @@ -120,12 +121,14 @@ static void destroy(grpc_server *server, void *statep) { grpc_server_secure_state *state = statep; gpr_mu_lock(&state->mu); state->is_shutdown = 1; - grpc_tcp_server_destroy(state->tcp, grpc_server_listener_destroy_done, server); + grpc_tcp_server_destroy(state->tcp, grpc_server_listener_destroy_done, + server); gpr_mu_unlock(&state->mu); state_unref(state); } -int grpc_server_add_secure_http2_port(grpc_server *server, const char *addr, grpc_server_credentials *creds) { +int grpc_server_add_secure_http2_port(grpc_server *server, const char *addr, + grpc_server_credentials *creds) { grpc_resolved_addresses *resolved = NULL; grpc_tcp_server *tcp = NULL; grpc_server_secure_state *state = NULL; diff --git a/src/core/surface/completion_queue.c b/src/core/surface/completion_queue.c index b08bd93693..24f4a05071 100644 --- a/src/core/surface/completion_queue.c +++ b/src/core/surface/completion_queue.c @@ -436,6 +436,7 @@ grpc_pollset *grpc_cq_pollset(grpc_completion_queue *cc) { void grpc_cq_hack_spin_pollset(grpc_completion_queue *cc) { gpr_mu_lock(GRPC_POLLSET_MU(&cc->pollset)); grpc_pollset_kick(&cc->pollset); - grpc_pollset_work(&cc->pollset, gpr_time_add(gpr_now(), gpr_time_from_millis(100))); + grpc_pollset_work(&cc->pollset, + gpr_time_add(gpr_now(), gpr_time_from_millis(100))); gpr_mu_unlock(GRPC_POLLSET_MU(&cc->pollset)); } diff --git a/src/core/surface/completion_queue.h b/src/core/surface/completion_queue.h index 3e9e2d186e..3a7cc99dda 100644 --- a/src/core/surface/completion_queue.h +++ b/src/core/surface/completion_queue.h @@ -116,4 +116,4 @@ grpc_pollset *grpc_cq_pollset(grpc_completion_queue *cc); void grpc_cq_hack_spin_pollset(grpc_completion_queue *cc); -#endif /* GRPC_INTERNAL_CORE_SURFACE_COMPLETION_QUEUE_H */ +#endif /* GRPC_INTERNAL_CORE_SURFACE_COMPLETION_QUEUE_H */ diff --git a/src/core/surface/server.c b/src/core/surface/server.c index 2e013ea742..17cba9a505 100644 --- a/src/core/surface/server.c +++ b/src/core/surface/server.c @@ -592,9 +592,8 @@ static void destroy_channel_elem(grpc_channel_element *elem) { } static const grpc_channel_filter server_surface_filter = { - call_op, channel_op, sizeof(call_data), - init_call_elem, destroy_call_elem, sizeof(channel_data), - init_channel_elem, destroy_channel_elem, "server", + call_op, channel_op, sizeof(call_data), init_call_elem, destroy_call_elem, + sizeof(channel_data), init_channel_elem, destroy_channel_elem, "server", }; static void addcq(grpc_server *server, grpc_completion_queue *cq) { @@ -737,7 +736,8 @@ grpc_transport_setup_result grpc_server_setup_transport( channel = grpc_channel_create_from_filters(filters, num_filters, s->channel_args, mdctx, 0); chand = (channel_data *)grpc_channel_stack_element( - grpc_channel_get_channel_stack(channel), 0)->channel_data; + grpc_channel_get_channel_stack(channel), 0) + ->channel_data; chand->server = s; server_ref(s); chand->channel = channel; @@ -758,7 +758,7 @@ grpc_transport_setup_result grpc_server_setup_transport( method = grpc_mdstr_from_string(mdctx, rm->method); hash = GRPC_MDSTR_KV_HASH(host ? host->hash : 0, method->hash); for (probes = 0; chand->registered_methods[(hash + probes) % slots] - .server_registered_method != NULL; + .server_registered_method != NULL; probes++) ; if (probes > max_probes) max_probes = probes; @@ -927,7 +927,8 @@ void grpc_server_destroy(grpc_server *server) { gpr_mu_lock(&server->mu); } - gpr_cv_wait(&server->cv, &server->mu, gpr_time_add(gpr_now(), gpr_time_from_millis(100))); + gpr_cv_wait(&server->cv, &server->mu, + gpr_time_add(gpr_now(), gpr_time_from_millis(100))); } while (server->listeners) { diff --git a/src/core/surface/server.h b/src/core/surface/server.h index 548a16c6c9..2cfa38fa43 100644 --- a/src/core/surface/server.h +++ b/src/core/surface/server.h @@ -48,7 +48,8 @@ grpc_server *grpc_server_create_from_filters(grpc_completion_queue *cq, and when it shuts down, it will call destroy */ void grpc_server_add_listener(grpc_server *server, void *listener, void (*start)(grpc_server *server, void *arg, - grpc_pollset **pollsets, size_t npollsets), + grpc_pollset **pollsets, + size_t npollsets), void (*destroy)(grpc_server *server, void *arg)); void grpc_server_listener_destroy_done(void *server); @@ -62,4 +63,4 @@ grpc_transport_setup_result grpc_server_setup_transport( const grpc_channel_args *grpc_server_get_channel_args(grpc_server *server); -#endif /* GRPC_INTERNAL_CORE_SURFACE_SERVER_H */ +#endif /* GRPC_INTERNAL_CORE_SURFACE_SERVER_H */ diff --git a/src/core/surface/server_chttp2.c b/src/core/surface/server_chttp2.c index 9a23125752..f3b9219f8b 100644 --- a/src/core/surface/server_chttp2.c +++ b/src/core/surface/server_chttp2.c @@ -66,7 +66,8 @@ static void new_transport(void *server, grpc_endpoint *tcp) { } /* Server callback: start listening on our ports */ -static void start(grpc_server *server, void *tcpp, grpc_pollset **pollsets, size_t pollset_count) { +static void start(grpc_server *server, void *tcpp, grpc_pollset **pollsets, + size_t pollset_count) { grpc_tcp_server *tcp = tcpp; grpc_tcp_server_start(tcp, pollsets, pollset_count, new_transport, server); } diff --git a/test/core/end2end/tests/cancel_after_invoke.c b/test/core/end2end/tests/cancel_after_invoke.c index 326321f4e2..592dfd415f 100644 --- a/test/core/end2end/tests/cancel_after_invoke.c +++ b/test/core/end2end/tests/cancel_after_invoke.c @@ -110,7 +110,8 @@ static void test_cancel_after_invoke(grpc_end2end_test_config config, grpc_op ops[6]; grpc_op *op; grpc_call *c; - grpc_end2end_test_fixture f = begin_test(config, __FUNCTION__, mode, NULL, NULL); + grpc_end2end_test_fixture f = + begin_test(config, __FUNCTION__, mode, NULL, NULL); gpr_timespec deadline = five_seconds_time(); cq_verifier *v_client = cq_verifier_create(f.client_cq); grpc_metadata_array initial_metadata_recv; diff --git a/test/core/end2end/tests/cancel_test_helpers.h b/test/core/end2end/tests/cancel_test_helpers.h index 3d92b64ae4..0d680fcfe1 100644 --- a/test/core/end2end/tests/cancel_test_helpers.h +++ b/test/core/end2end/tests/cancel_test_helpers.h @@ -47,6 +47,8 @@ static grpc_call_error wait_for_deadline(grpc_call *call) { static const cancellation_mode cancellation_modes[] = { {"cancel", grpc_call_cancel, GRPC_STATUS_CANCELLED, ""}, - {"deadline", wait_for_deadline, GRPC_STATUS_DEADLINE_EXCEEDED, "Deadline Exceeded"}, }; + {"deadline", wait_for_deadline, GRPC_STATUS_DEADLINE_EXCEEDED, + "Deadline Exceeded"}, +}; -#endif /* GRPC_TEST_CORE_END2END_TESTS_CANCEL_TEST_HELPERS_H */ +#endif /* GRPC_TEST_CORE_END2END_TESTS_CANCEL_TEST_HELPERS_H */ -- cgit v1.2.3