diff options
Diffstat (limited to 'src/core')
26 files changed, 470 insertions, 99 deletions
diff --git a/src/core/channel/http_server_filter.c b/src/core/channel/http_server_filter.c index f565cbf3ae..9da8b333ca 100644 --- a/src/core/channel/http_server_filter.c +++ b/src/core/channel/http_server_filter.c @@ -189,7 +189,8 @@ static void call_op(grpc_call_element *elem, grpc_call_element *from_elem, /* translate host to :authority since :authority may be omitted */ grpc_mdelem *authority = grpc_mdelem_from_metadata_strings( - channeld->mdctx, channeld->authority_key, op->data.metadata->value); + channeld->mdctx, grpc_mdstr_ref(channeld->authority_key), + grpc_mdstr_ref(op->data.metadata->value)); grpc_mdelem_unref(op->data.metadata); op->data.metadata = authority; /* pass the event up */ diff --git a/src/core/httpcli/parser.c b/src/core/httpcli/parser.c index f4decda98a..7b2a62060c 100644 --- a/src/core/httpcli/parser.c +++ b/src/core/httpcli/parser.c @@ -177,6 +177,8 @@ static int addbyte(grpc_httpcli_parser *parser, gpr_uint8 byte) { } gpr_log(GPR_ERROR, "should never reach here"); abort(); + + return 0; } void grpc_httpcli_parser_init(grpc_httpcli_parser *parser) { diff --git a/src/core/iomgr/iocp_windows.c b/src/core/iomgr/iocp_windows.c index 8b019e8049..aec626509a 100644 --- a/src/core/iomgr/iocp_windows.c +++ b/src/core/iomgr/iocp_windows.c @@ -52,10 +52,11 @@ static OVERLAPPED g_iocp_custom_overlap; static gpr_event g_shutdown_iocp; static gpr_event g_iocp_done; +static gpr_atm g_orphans = 0; static HANDLE g_iocp; -static int do_iocp_work() { +static void do_iocp_work() { BOOL success; DWORD bytes = 0; DWORD flags = 0; @@ -71,14 +72,14 @@ static int do_iocp_work() { gpr_time_to_millis(wait_time)); if (!success && !overlapped) { /* The deadline got attained. */ - return 0; + return; } GPR_ASSERT(completion_key && overlapped); if (overlapped == &g_iocp_custom_overlap) { if (completion_key == (ULONG_PTR) &g_iocp_kick_token) { /* We were awoken from a kick. */ gpr_log(GPR_DEBUG, "do_iocp_work - got a kick"); - return 1; + return; } gpr_log(GPR_ERROR, "Unknown custom completion key."); abort(); @@ -97,8 +98,13 @@ static int do_iocp_work() { } success = WSAGetOverlappedResult(socket->socket, &info->overlapped, &bytes, FALSE, &flags); - gpr_log(GPR_DEBUG, "bytes: %u, flags: %u - op %s", bytes, flags, - success ? "succeeded" : "failed"); + gpr_log(GPR_DEBUG, "bytes: %u, flags: %u - op %s %s", bytes, flags, + success ? "succeeded" : "failed", socket->orphan ? "orphan" : ""); + if (socket->orphan) { + grpc_winsocket_destroy(socket); + gpr_atm_full_fetch_add(&g_orphans, -1); + return; + } info->bytes_transfered = bytes; info->wsa_error = success ? 0 : WSAGetLastError(); GPR_ASSERT(overlapped == &info->overlapped); @@ -113,12 +119,10 @@ static int do_iocp_work() { } gpr_mu_unlock(&socket->state_mu); if (f) f(opaque, 1); - - return 1; } static void iocp_loop(void *p) { - while (!gpr_event_get(&g_shutdown_iocp)) { + while (gpr_atm_acq_load(&g_orphans) || !gpr_event_get(&g_shutdown_iocp)) { grpc_maybe_call_delayed_callbacks(NULL, 1); do_iocp_work(); } @@ -138,13 +142,19 @@ void grpc_iocp_init(void) { gpr_thd_new(&id, iocp_loop, NULL, NULL); } -void grpc_iocp_shutdown(void) { +void grpc_iocp_kick(void) { BOOL success; - gpr_event_set(&g_shutdown_iocp, (void *)1); + success = PostQueuedCompletionStatus(g_iocp, 0, (ULONG_PTR) &g_iocp_kick_token, &g_iocp_custom_overlap); GPR_ASSERT(success); +} + +void grpc_iocp_shutdown(void) { + BOOL success; + gpr_event_set(&g_shutdown_iocp, (void *)1); + grpc_iocp_kick(); gpr_event_wait(&g_iocp_done, gpr_inf_future); success = CloseHandle(g_iocp); GPR_ASSERT(success); @@ -166,6 +176,10 @@ void grpc_iocp_add_socket(grpc_winsocket *socket) { GPR_ASSERT(ret == g_iocp); } +void grpc_iocp_socket_orphan(grpc_winsocket *socket) { + gpr_atm_full_fetch_add(&g_orphans, 1); +} + static void socket_notify_on_iocp(grpc_winsocket *socket, void(*cb)(void *, int), void *opaque, grpc_winsocket_callback_info *info) { diff --git a/src/core/iomgr/iocp_windows.h b/src/core/iomgr/iocp_windows.h index 33133193a1..fa3f5eee10 100644 --- a/src/core/iomgr/iocp_windows.h +++ b/src/core/iomgr/iocp_windows.h @@ -42,6 +42,7 @@ void grpc_iocp_init(void); void grpc_iocp_shutdown(void); void grpc_iocp_add_socket(grpc_winsocket *); +void grpc_iocp_socket_orphan(grpc_winsocket *); void grpc_socket_notify_on_write(grpc_winsocket *, void(*cb)(void *, int success), void *opaque); diff --git a/src/core/iomgr/iomgr.c b/src/core/iomgr/iomgr.c index 058685b295..d0e6706fbd 100644 --- a/src/core/iomgr/iomgr.c +++ b/src/core/iomgr/iomgr.c @@ -117,7 +117,16 @@ void grpc_iomgr_shutdown(void) { gpr_mu_lock(&g_mu); } if (g_refs) { - if (gpr_cv_wait(&g_rcv, &g_mu, shutdown_deadline) && g_cbs_head == NULL) { + int timeout = 0; + gpr_timespec short_deadline = gpr_time_add(gpr_now(), + gpr_time_from_millis(100)); + while (gpr_cv_wait(&g_rcv, &g_mu, short_deadline) && g_cbs_head == NULL) { + if (gpr_time_cmp(gpr_now(), shutdown_deadline) > 0) { + timeout = 1; + break; + } + } + if (timeout) { gpr_log(GPR_DEBUG, "Failed to free %d iomgr objects before shutdown deadline: " "memory leaks are likely", diff --git a/src/core/iomgr/socket_windows.c b/src/core/iomgr/socket_windows.c index 99f38b0e03..22dad41783 100644 --- a/src/core/iomgr/socket_windows.c +++ b/src/core/iomgr/socket_windows.c @@ -55,7 +55,7 @@ grpc_winsocket *grpc_winsocket_create(SOCKET socket) { return r; } -void shutdown_op(grpc_winsocket_callback_info *info) { +static void shutdown_op(grpc_winsocket_callback_info *info) { if (!info->cb) return; grpc_iomgr_add_delayed_callback(info->cb, info->opaque, 0); } @@ -68,8 +68,13 @@ void grpc_winsocket_shutdown(grpc_winsocket *socket) { void grpc_winsocket_orphan(grpc_winsocket *socket) { gpr_log(GPR_DEBUG, "grpc_winsocket_orphan"); + grpc_iocp_socket_orphan(socket); + socket->orphan = 1; grpc_iomgr_unref(); closesocket(socket->socket); +} + +void grpc_winsocket_destroy(grpc_winsocket *socket) { gpr_mu_destroy(&socket->state_mu); gpr_free(socket); } diff --git a/src/core/iomgr/socket_windows.h b/src/core/iomgr/socket_windows.h index d4776ab10f..cbae91692c 100644 --- a/src/core/iomgr/socket_windows.h +++ b/src/core/iomgr/socket_windows.h @@ -57,12 +57,13 @@ typedef struct grpc_winsocket_callback_info { typedef struct grpc_winsocket { SOCKET socket; - int added_to_iocp; - grpc_winsocket_callback_info write_info; grpc_winsocket_callback_info read_info; gpr_mu state_mu; + + int added_to_iocp; + int orphan; } grpc_winsocket; /* Create a wrapped windows handle. @@ -71,5 +72,6 @@ grpc_winsocket *grpc_winsocket_create(SOCKET socket); void grpc_winsocket_shutdown(grpc_winsocket *socket); void grpc_winsocket_orphan(grpc_winsocket *socket); +void grpc_winsocket_destroy(grpc_winsocket *socket); #endif /* GRPC_INTERNAL_CORE_IOMGR_SOCKET_WINDOWS_H */ diff --git a/src/core/iomgr/tcp_server.h b/src/core/iomgr/tcp_server.h index 68ee85c5a7..66bb3ef701 100644 --- a/src/core/iomgr/tcp_server.h +++ b/src/core/iomgr/tcp_server.h @@ -71,6 +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 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 90b7eb451d..895f85fc68 100644 --- a/src/core/iomgr/tcp_server_posix.c +++ b/src/core/iomgr/tcp_server_posix.c @@ -102,12 +102,18 @@ struct grpc_tcp_server { gpr_cv cv; /* active port count: how many ports are actually still listening */ - int active_ports; + size_t active_ports; + /* destroyed port count: how many ports are completely destroyed */ + size_t destroyed_ports; /* all listening ports */ server_port *ports; size_t nports; size_t port_capacity; + + /* shutdown callback */ + void (*shutdown_complete)(void *); + void *shutdown_complete_arg; }; grpc_tcp_server *grpc_tcp_server_create(void) { @@ -115,6 +121,7 @@ grpc_tcp_server *grpc_tcp_server_create(void) { gpr_mu_init(&s->mu); gpr_cv_init(&s->cv); s->active_ports = 0; + s->destroyed_ports = 0; s->cb = NULL; s->cb_arg = NULL; s->ports = gpr_malloc(sizeof(server_port) * INIT_PORT_CAP); @@ -123,29 +130,64 @@ grpc_tcp_server *grpc_tcp_server_create(void) { return s; } -void grpc_tcp_server_destroy(grpc_tcp_server *s) { +static void finish_shutdown(grpc_tcp_server *s) { + s->shutdown_complete(s->shutdown_complete_arg); + + gpr_mu_destroy(&s->mu); + gpr_cv_destroy(&s->cv); + + gpr_free(s->ports); + gpr_free(s); +} + +static void destroyed_port(void *server, int success) { + grpc_tcp_server *s = server; + gpr_mu_lock(&s->mu); + s->destroyed_ports++; + if (s->destroyed_ports == s->nports) { + gpr_mu_unlock(&s->mu); + finish_shutdown(s); + } else { + gpr_mu_unlock(&s->mu); + } +} + +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) { size_t i; gpr_mu_lock(&s->mu); + + s->shutdown_complete = shutdown_complete + ? shutdown_complete + : dont_care_about_shutdown_completion; + s->shutdown_complete_arg = shutdown_complete_arg; + /* shutdown all fd's */ for (i = 0; i < s->nports; i++) { grpc_fd_shutdown(s->ports[i].emfd); } /* wait while that happens */ + /* TODO(ctiller): make this asynchronous also */ while (s->active_ports) { gpr_cv_wait(&s->cv, &s->mu, gpr_inf_future); } gpr_mu_unlock(&s->mu); /* delete ALL the things */ - for (i = 0; i < s->nports; i++) { - server_port *sp = &s->ports[i]; - if (sp->addr.sockaddr.sa_family == AF_UNIX) { - unlink_if_unix_domain_socket(&sp->addr.un); + if (s->nports) { + for (i = 0; i < s->nports; i++) { + server_port *sp = &s->ports[i]; + if (sp->addr.sockaddr.sa_family == AF_UNIX) { + unlink_if_unix_domain_socket(&sp->addr.un); + } + grpc_fd_orphan(sp->emfd, destroyed_port, s); } - grpc_fd_orphan(sp->emfd, NULL, NULL); + } else { + finish_shutdown(s); } - gpr_free(s->ports); - gpr_free(s); } /* get max listen queue size on linux */ diff --git a/src/core/iomgr/tcp_server_windows.c b/src/core/iomgr/tcp_server_windows.c index 0c3ab1dc91..a43d5670a4 100644 --- a/src/core/iomgr/tcp_server_windows.c +++ b/src/core/iomgr/tcp_server_windows.c @@ -92,7 +92,9 @@ grpc_tcp_server *grpc_tcp_server_create(void) { return s; } -void grpc_tcp_server_destroy(grpc_tcp_server *s) { +void grpc_tcp_server_destroy(grpc_tcp_server *s, + void (*shutdown_done)(void *shutdown_done_arg), + void *shutdown_done_arg) { size_t i; gpr_mu_lock(&s->mu); /* shutdown all fd's */ @@ -112,11 +114,15 @@ void grpc_tcp_server_destroy(grpc_tcp_server *s) { } gpr_free(s->ports); gpr_free(s); + + if (shutdown_done) { + 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; @@ -147,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); @@ -221,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); @@ -257,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()); @@ -307,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); @@ -330,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); } @@ -369,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 c155b80b7e..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_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; @@ -213,7 +216,7 @@ error: grpc_resolved_addresses_destroy(resolved); } if (tcp) { - grpc_tcp_server_destroy(tcp); + grpc_tcp_server_destroy(tcp, NULL, NULL); } if (state) { gpr_free(state); diff --git a/src/core/support/thd.c b/src/core/support/thd.c new file mode 100644 index 0000000000..ec308f3119 --- /dev/null +++ b/src/core/support/thd.c @@ -0,0 +1,66 @@ +/* + * + * Copyright 2015, Google Inc. + * All rights reserved. + * + * Redistribution and use in source and binary forms, with or without + * modification, are permitted provided that the following conditions are + * met: + * + * * Redistributions of source code must retain the above copyright + * notice, this list of conditions and the following disclaimer. + * * Redistributions in binary form must reproduce the above + * copyright notice, this list of conditions and the following disclaimer + * in the documentation and/or other materials provided with the + * distribution. + * * Neither the name of Google Inc. nor the names of its + * contributors may be used to endorse or promote products derived from + * this software without specific prior written permission. + * + * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS + * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT + * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR + * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT + * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, + * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT + * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, + * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY + * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT + * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE + * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. + * + */ + +/* Posix implementation for gpr threads. */ + +#include <memory.h> + +#include <grpc/support/thd.h> + +enum { + GPR_THD_JOINABLE = 1 +}; + +gpr_thd_options gpr_thd_options_default(void) { + gpr_thd_options options; + memset(&options, 0, sizeof(options)); + return options; +} + +void gpr_thd_options_set_detached(gpr_thd_options *options) { + options->flags &= ~GPR_THD_JOINABLE; +} + +void gpr_thd_options_set_joinable(gpr_thd_options *options) { + options->flags |= GPR_THD_JOINABLE; +} + +int gpr_thd_options_is_detached(const gpr_thd_options *options) { + if (!options) return 1; + return (options->flags & GPR_THD_JOINABLE) == 0; +} + +int gpr_thd_options_is_joinable(const gpr_thd_options *options) { + if (!options) return 0; + return (options->flags & GPR_THD_JOINABLE) == GPR_THD_JOINABLE; +} diff --git a/src/core/support/thd_posix.c b/src/core/support/thd_posix.c index f50ea58335..fa4eb50556 100644 --- a/src/core/support/thd_posix.c +++ b/src/core/support/thd_posix.c @@ -68,7 +68,11 @@ int gpr_thd_new(gpr_thd_id *t, void (*thd_body)(void *arg), void *arg, a->arg = arg; GPR_ASSERT(pthread_attr_init(&attr) == 0); - GPR_ASSERT(pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED) == 0); + if (gpr_thd_options_is_detached(options)) { + GPR_ASSERT(pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED) == 0); + } else { + GPR_ASSERT(pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_JOINABLE) == 0); + } thread_started = (pthread_create(&p, &attr, &thread_body, a) == 0); GPR_ASSERT(pthread_attr_destroy(&attr) == 0); if (!thread_started) { @@ -78,14 +82,12 @@ int gpr_thd_new(gpr_thd_id *t, void (*thd_body)(void *arg), void *arg, return thread_started; } -gpr_thd_options gpr_thd_options_default(void) { - gpr_thd_options options; - memset(&options, 0, sizeof(options)); - return options; -} - gpr_thd_id gpr_thd_currentid(void) { return (gpr_thd_id)pthread_self(); } +void gpr_thd_join(gpr_thd_id t) { + pthread_join((pthread_t)t, NULL); +} + #endif /* GPR_POSIX_SYNC */ diff --git a/src/core/support/thd_win32.c b/src/core/support/thd_win32.c index 347cad57e3..3cc798293a 100644 --- a/src/core/support/thd_win32.c +++ b/src/core/support/thd_win32.c @@ -31,7 +31,7 @@ * */ -/* Posix implementation for gpr threads. */ +/* Windows implementation for gpr threads. */ #include <grpc/support/port_platform.h> @@ -40,47 +40,81 @@ #include <windows.h> #include <string.h> #include <grpc/support/alloc.h> +#include <grpc/support/log.h> #include <grpc/support/thd.h> -struct thd_arg { +#if defined(_MSC_VER) +#define thread_local __declspec(thread) +#elif defined(__GNUC__) +#define thread_local __thread +#else +#error "Unknown compiler - please file a bug report" +#endif + +struct thd_info { void (*body)(void *arg); /* body of a thread */ void *arg; /* argument to a thread */ + HANDLE join_event; /* if joinable, the join event */ + int joinable; /* true if not detached */ }; +static thread_local struct thd_info *g_thd_info; + +/* Destroys a thread info */ +static void destroy_thread(struct thd_info *t) { + if (t->joinable) CloseHandle(t->join_event); + gpr_free(t); +} + /* Body of every thread started via gpr_thd_new. */ static DWORD WINAPI thread_body(void *v) { - struct thd_arg a = *(struct thd_arg *)v; - gpr_free(v); - (*a.body)(a.arg); + g_thd_info = (struct thd_info *)v; + g_thd_info->body(g_thd_info->arg); + if (g_thd_info->joinable) { + BOOL ret = SetEvent(g_thd_info->join_event); + GPR_ASSERT(ret); + } else { + destroy_thread(g_thd_info); + } return 0; } int gpr_thd_new(gpr_thd_id *t, void (*thd_body)(void *arg), void *arg, const gpr_thd_options *options) { HANDLE handle; - DWORD thread_id; - struct thd_arg *a = gpr_malloc(sizeof(*a)); - a->body = thd_body; - a->arg = arg; + struct thd_info *info = gpr_malloc(sizeof(*info)); + info->body = thd_body; + info->arg = arg; *t = 0; - handle = CreateThread(NULL, 64 * 1024, thread_body, a, 0, &thread_id); + if (gpr_thd_options_is_joinable(options)) { + info->joinable = 1; + info->join_event = CreateEvent(NULL, FALSE, FALSE, NULL); + if (info->join_event == NULL) { + gpr_free(info); + return 0; + } + } else { + info->joinable = 0; + } + handle = CreateThread(NULL, 64 * 1024, thread_body, info, 0, NULL); if (handle == NULL) { - gpr_free(a); + destroy_thread(info); } else { - CloseHandle(handle); /* threads are "detached" */ + *t = (gpr_thd_id)info; + CloseHandle(handle); } - *t = (gpr_thd_id)thread_id; return handle != NULL; } -gpr_thd_options gpr_thd_options_default(void) { - gpr_thd_options options; - memset(&options, 0, sizeof(options)); - return options; +gpr_thd_id gpr_thd_currentid(void) { + return (gpr_thd_id)g_thd_info; } -gpr_thd_id gpr_thd_currentid(void) { - return (gpr_thd_id)GetCurrentThreadId(); +void gpr_thd_join(gpr_thd_id t) { + struct thd_info *info = (struct thd_info *)t; + DWORD ret = WaitForSingleObject(info->join_event, INFINITE); + GPR_ASSERT(ret == WAIT_OBJECT_0); + destroy_thread(info); } #endif /* GPR_WIN32 */ diff --git a/src/core/surface/call.c b/src/core/surface/call.c index cfce943794..dba63058b8 100644 --- a/src/core/surface/call.c +++ b/src/core/surface/call.c @@ -1006,6 +1006,8 @@ grpc_call_error grpc_call_start_batch(grpc_call *call, const grpc_op *ops, const grpc_op *op; grpc_ioreq *req; + GRPC_CALL_LOG_BATCH(GPR_INFO, call, ops, nops, tag); + if (nops == 0) { grpc_cq_begin_op(call->cq, call, GRPC_OP_COMPLETE); grpc_cq_end_op_complete(call->cq, tag, call, do_nothing, NULL, GRPC_OP_OK); diff --git a/src/core/surface/call.h b/src/core/surface/call.h index cb81cb52c2..06434f87ac 100644 --- a/src/core/surface/call.h +++ b/src/core/surface/call.h @@ -119,4 +119,13 @@ grpc_call_stack *grpc_call_get_call_stack(grpc_call *call); /* Given the top call_element, get the call object. */ grpc_call *grpc_call_from_top_element(grpc_call_element *surface_element); +extern int grpc_trace_batch; + +void grpc_call_log_batch(char *file, int line, gpr_log_severity severity, + grpc_call *call, const grpc_op *ops, size_t nops, + void *tag); + +#define GRPC_CALL_LOG_BATCH(sev, call, ops, nops, tag) \ + if (grpc_trace_batch) grpc_call_log_batch(sev, call, ops, nops, tag) + #endif /* GRPC_INTERNAL_CORE_SURFACE_CALL_H */ diff --git a/src/core/surface/call_log_batch.c b/src/core/surface/call_log_batch.c new file mode 100644 index 0000000000..a33583a12d --- /dev/null +++ b/src/core/surface/call_log_batch.c @@ -0,0 +1,121 @@ +/* + * + * Copyright 2015, Google Inc. + * All rights reserved. + * + * Redistribution and use in source and binary forms, with or without + * modification, are permitted provided that the following conditions are + * met: + * + * * Redistributions of source code must retain the above copyright + * notice, this list of conditions and the following disclaimer. + * * Redistributions in binary form must reproduce the above + * copyright notice, this list of conditions and the following disclaimer + * in the documentation and/or other materials provided with the + * distribution. + * * Neither the name of Google Inc. nor the names of its + * contributors may be used to endorse or promote products derived from + * this software without specific prior written permission. + * + * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS + * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT + * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR + * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT + * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, + * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT + * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, + * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY + * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT + * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE + * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. + * + */ + +#include "src/core/surface/call.h" + +#include "src/core/support/string.h" +#include <grpc/support/alloc.h> + +int grpc_trace_batch = 0; + +static void add_metadata(gpr_strvec *b, const grpc_metadata *md, size_t count) { + size_t i; + for(i = 0; i < count; i++) { + gpr_strvec_add(b, gpr_strdup("\nkey=")); + gpr_strvec_add(b, gpr_strdup(md[i].key)); + + gpr_strvec_add(b, gpr_strdup(" value=")); + gpr_strvec_add(b, gpr_hexdump(md[i].value, md[i].value_length, + GPR_HEXDUMP_PLAINTEXT)); + } +} + +char *grpc_op_string(const grpc_op *op) { + char *tmp; + char *out; + + gpr_strvec b; + gpr_strvec_init(&b); + + switch (op->op) { + case GRPC_OP_SEND_INITIAL_METADATA: + gpr_strvec_add(&b, gpr_strdup("SEND_INITIAL_METADATA")); + add_metadata(&b, op->data.send_initial_metadata.metadata, + op->data.send_initial_metadata.count); + break; + case GRPC_OP_SEND_MESSAGE: + gpr_asprintf(&tmp, "SEND_MESSAGE ptr=%p", op->data.send_message); + gpr_strvec_add(&b, tmp); + break; + case GRPC_OP_SEND_CLOSE_FROM_CLIENT: + gpr_strvec_add(&b, gpr_strdup("SEND_CLOSE_FROM_CLIENT")); + break; + case GRPC_OP_SEND_STATUS_FROM_SERVER: + gpr_asprintf(&tmp, "SEND_STATUS_FROM_SERVER status=%d details=%s", + op->data.send_status_from_server.status, + op->data.send_status_from_server.status_details); + gpr_strvec_add(&b, tmp); + add_metadata(&b, op->data.send_status_from_server.trailing_metadata, + op->data.send_status_from_server.trailing_metadata_count); + break; + case GRPC_OP_RECV_INITIAL_METADATA: + gpr_asprintf(&tmp, "RECV_INITIAL_METADATA ptr=%p", + op->data.recv_initial_metadata); + gpr_strvec_add(&b, tmp); + break; + case GRPC_OP_RECV_MESSAGE: + gpr_asprintf(&tmp, "RECV_MESSAGE ptr=%p", op->data.recv_message); + gpr_strvec_add(&b, tmp); + break; + case GRPC_OP_RECV_STATUS_ON_CLIENT: + gpr_asprintf(&tmp, + "RECV_STATUS_ON_CLIENT metadata=%p status=%p details=%p", + op->data.recv_status_on_client.trailing_metadata, + op->data.recv_status_on_client.status, + op->data.recv_status_on_client.status_details); + gpr_strvec_add(&b, tmp); + break; + case GRPC_OP_RECV_CLOSE_ON_SERVER: + gpr_asprintf(&tmp, "RECV_CLOSE_ON_SERVER cancelled=%p", + op->data.recv_close_on_server.cancelled); + gpr_strvec_add(&b, tmp); + } + out = gpr_strvec_flatten(&b, NULL); + gpr_strvec_destroy(&b); + + return out; +} + +void grpc_call_log_batch(char *file, int line, gpr_log_severity severity, + grpc_call *call, const grpc_op *ops, size_t nops, + void *tag) { + char *tmp; + size_t i; + gpr_log(file, line, severity, + "grpc_call_start_batch(%p, %p, %d, 0x%x)", call, ops, nops, tag); + for(i = 0; i < nops; i++) { + tmp = grpc_op_string(&ops[i]); + gpr_log(file, line, severity, "ops[%d]: %s", i, tmp); + gpr_free(tmp); + } +} diff --git a/src/core/surface/completion_queue.c b/src/core/surface/completion_queue.c index 6a1d83ce5d..24f4a05071 100644 --- a/src/core/surface/completion_queue.c +++ b/src/core/surface/completion_queue.c @@ -432,3 +432,11 @@ 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)); +} diff --git a/src/core/surface/completion_queue.h b/src/core/surface/completion_queue.h index 3054264cad..3a7cc99dda 100644 --- a/src/core/surface/completion_queue.h +++ b/src/core/surface/completion_queue.h @@ -114,4 +114,6 @@ void grpc_cq_dump_pending_ops(grpc_completion_queue *cc); grpc_pollset *grpc_cq_pollset(grpc_completion_queue *cc); -#endif /* GRPC_INTERNAL_CORE_SURFACE_COMPLETION_QUEUE_H */ +void grpc_cq_hack_spin_pollset(grpc_completion_queue *cc); + +#endif /* GRPC_INTERNAL_CORE_SURFACE_COMPLETION_QUEUE_H */ diff --git a/src/core/surface/init.c b/src/core/surface/init.c index e48c4202e5..d4f0eb40e8 100644 --- a/src/core/surface/init.c +++ b/src/core/surface/init.c @@ -36,6 +36,7 @@ #include "src/core/debug/trace.h" #include "src/core/statistics/census_interface.h" #include "src/core/channel/channel_stack.h" +#include "src/core/surface/call.h" #include "src/core/surface/init.h" #include "src/core/surface/surface_trace.h" #include "src/core/transport/chttp2_transport.h" @@ -57,6 +58,7 @@ void grpc_init(void) { grpc_register_tracer("channel", &grpc_trace_channel); grpc_register_tracer("surface", &grpc_surface_trace); grpc_register_tracer("http", &grpc_http_trace); + grpc_register_tracer("batch", &grpc_trace_batch); grpc_security_pre_init(); grpc_tracer_init("GRPC_TRACE"); grpc_iomgr_init(); @@ -82,4 +84,3 @@ int grpc_is_initialized(void) { gpr_mu_unlock(&g_init_mu); return r; } - diff --git a/src/core/surface/server.c b/src/core/surface/server.c index 424734c54c..17cba9a505 100644 --- a/src/core/surface/server.c +++ b/src/core/surface/server.c @@ -137,6 +137,7 @@ struct grpc_server { size_t cq_count; gpr_mu mu; + gpr_cv cv; registered_method *registered_methods; requested_call_array requested_calls; @@ -149,6 +150,7 @@ struct grpc_server { channel_data root_channel_data; listener *listeners; + int listeners_destroyed; gpr_refcount internal_refcount; }; @@ -263,6 +265,7 @@ 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) { @@ -589,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) { @@ -620,6 +622,7 @@ grpc_server *grpc_server_create_from_filters(grpc_completion_queue *cq, if (cq) addcq(server, cq); gpr_mu_init(&server->mu); + gpr_cv_init(&server->cv); server->unregistered_cq = cq; /* decremented by grpc_server_destroy */ @@ -733,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; @@ -754,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; @@ -781,6 +785,15 @@ 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) { listener *l; @@ -878,11 +891,6 @@ static void shutdown_internal(grpc_server *server, gpr_uint8 have_shutdown_tag, for (l = server->listeners; l; l = l->next) { l->destroy(server, l->arg); } - while (server->listeners) { - l = server->listeners; - server->listeners = l->next; - gpr_free(l); - } } void grpc_server_shutdown(grpc_server *server) { @@ -893,8 +901,18 @@ 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); + gpr_mu_unlock(&server->mu); +} + void grpc_server_destroy(grpc_server *server) { channel_data *c; + listener *l; + size_t i; gpr_mu_lock(&server->mu); if (!server->shutdown) { gpr_mu_unlock(&server->mu); @@ -902,6 +920,23 @@ void grpc_server_destroy(grpc_server *server) { gpr_mu_lock(&server->mu); } + 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); + } + + gpr_cv_wait(&server->cv, &server->mu, + gpr_time_add(gpr_now(), gpr_time_from_millis(100))); + } + + while (server->listeners) { + l = server->listeners; + server->listeners = l->next; + gpr_free(l); + } + for (c = server->root_channel_data.next; c != &server->root_channel_data; c = c->next) { shutdown_channel(c); diff --git a/src/core/surface/server.h b/src/core/surface/server.h index e33f69b8c7..2cfa38fa43 100644 --- a/src/core/surface/server.h +++ b/src/core/surface/server.h @@ -48,9 +48,12 @@ 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); + /* Setup a transport - creates a channel stack, binds the transport to the server */ grpc_transport_setup_result grpc_server_setup_transport( @@ -60,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 27434b39e2..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); } @@ -75,7 +76,7 @@ static void start(grpc_server *server, void *tcpp, grpc_pollset **pollsets, size callbacks) */ static void destroy(grpc_server *server, void *tcpp) { grpc_tcp_server *tcp = tcpp; - grpc_tcp_server_destroy(tcp); + grpc_tcp_server_destroy(tcp, grpc_server_listener_destroy_done, server); } int grpc_server_add_http2_port(grpc_server *server, const char *addr) { @@ -131,7 +132,7 @@ error: grpc_resolved_addresses_destroy(resolved); } if (tcp) { - grpc_tcp_server_destroy(tcp); + grpc_tcp_server_destroy(tcp, NULL, NULL); } return 0; } diff --git a/src/core/transport/chttp2_transport.c b/src/core/transport/chttp2_transport.c index 2b15b2a812..4c0394d46f 100644 --- a/src/core/transport/chttp2_transport.c +++ b/src/core/transport/chttp2_transport.c @@ -1710,6 +1710,8 @@ static int process_read(transport *t, gpr_slice slice) { gpr_log(GPR_ERROR, "should never reach here"); abort(); + + return 0; } /* tcp read callback */ diff --git a/src/core/transport/metadata.c b/src/core/transport/metadata.c index 1c15716fad..066cc263a1 100644 --- a/src/core/transport/metadata.c +++ b/src/core/transport/metadata.c @@ -97,7 +97,7 @@ static void internal_string_ref(internal_string *s); static void internal_string_unref(internal_string *s); static void discard_metadata(grpc_mdctx *ctx); static void gc_mdtab(grpc_mdctx *ctx); -static void metadata_context_destroy(grpc_mdctx *ctx); +static void metadata_context_destroy_locked(grpc_mdctx *ctx); static void lock(grpc_mdctx *ctx) { gpr_mu_lock(&ctx->mu); } @@ -122,8 +122,7 @@ static void unlock(grpc_mdctx *ctx) { discard_metadata(ctx); } if (ctx->strtab_count == 0) { - gpr_mu_unlock(&ctx->mu); - metadata_context_destroy(ctx); + metadata_context_destroy_locked(ctx); return; } } @@ -185,8 +184,7 @@ static void discard_metadata(grpc_mdctx *ctx) { } } -static void metadata_context_destroy(grpc_mdctx *ctx) { - gpr_mu_lock(&ctx->mu); +static void metadata_context_destroy_locked(grpc_mdctx *ctx) { GPR_ASSERT(ctx->strtab_count == 0); GPR_ASSERT(ctx->mdtab_count == 0); GPR_ASSERT(ctx->mdtab_free == 0); diff --git a/src/core/tsi/ssl_transport_security.c b/src/core/tsi/ssl_transport_security.c index 33645ca8b8..018ddc4456 100644 --- a/src/core/tsi/ssl_transport_security.c +++ b/src/core/tsi/ssl_transport_security.c @@ -567,7 +567,8 @@ static tsi_result populate_ssl_context( EC_KEY* ecdh = EC_KEY_new_by_curve_name(NID_X9_62_prime256v1); if (!SSL_CTX_set_tmp_ecdh(context, ecdh)) { gpr_log(GPR_ERROR, "Could not set ephemeral ECDH key."); - result = TSI_INTERNAL_ERROR; + EC_KEY_free(ecdh); + return TSI_INTERNAL_ERROR; } SSL_CTX_set_options(context, SSL_OP_SINGLE_ECDH_USE); EC_KEY_free(ecdh); @@ -604,6 +605,7 @@ static tsi_result build_alpn_protocol_name_list( unsigned char* current; *protocol_name_list = NULL; *protocol_name_list_length = 0; + if (num_alpn_protocols == 0) return TSI_INVALID_ARGUMENT; for (i = 0; i < num_alpn_protocols; i++) { if (alpn_protocols_lengths[i] == 0) { gpr_log(GPR_ERROR, "Invalid 0-length protocol name."); |