From 8703f4d0c2a9f1cc45f7ebc4f01bd4cafb8fc865 Mon Sep 17 00:00:00 2001 From: Nicolas Noble Date: Mon, 23 Mar 2015 13:52:18 -0700 Subject: [PATCH] Various Windows fixes. -) Introduce a wait-loop with a 100ms delay on iomgr shutdown, so that background threads have the chance to place last minute callbacks without having to stall for the whole 10 seconds deadline. -) io completion ports will get notifications on socket shutdowns, so we need to delay their deletions for after we get a notification for them. -) we need to keep some sense of how many orphan sockets are in, so we can properly collect them - let's not shutdown the iocp loop until after all orphans have been collected. --- src/core/iomgr/iocp_windows.c | 34 +++++++++++++++++++++++---------- src/core/iomgr/iocp_windows.h | 1 + src/core/iomgr/iomgr.c | 11 ++++++++++- src/core/iomgr/socket_windows.c | 7 ++++++- src/core/iomgr/socket_windows.h | 6 ++++-- 5 files changed, 45 insertions(+), 14 deletions(-) diff --git a/src/core/iomgr/iocp_windows.c b/src/core/iomgr/iocp_windows.c index 8b019e80496..aec626509aa 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 33133193a10..fa3f5eee10e 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 058685b295d..d0e6706fbd7 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 99f38b0e032..22dad41783f 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 d4776ab10f8..cbae91692cb 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 */