diff options
Diffstat (limited to 'src/core/iomgr')
-rw-r--r-- | src/core/iomgr/iocp_windows.c | 34 | ||||
-rw-r--r-- | src/core/iomgr/iocp_windows.h | 1 | ||||
-rw-r--r-- | src/core/iomgr/iomgr.c | 11 | ||||
-rw-r--r-- | src/core/iomgr/sockaddr_win32.h | 5 | ||||
-rw-r--r-- | src/core/iomgr/socket_windows.c | 7 | ||||
-rw-r--r-- | src/core/iomgr/socket_windows.h | 6 | ||||
-rw-r--r-- | src/core/iomgr/tcp_server_windows.c | 3 | ||||
-rw-r--r-- | src/core/iomgr/tcp_windows.c | 4 |
8 files changed, 52 insertions, 19 deletions
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/sockaddr_win32.h b/src/core/iomgr/sockaddr_win32.h index 3a5f27bb34..c0385ea614 100644 --- a/src/core/iomgr/sockaddr_win32.h +++ b/src/core/iomgr/sockaddr_win32.h @@ -38,4 +38,9 @@ #include <winsock2.h> #include <mswsock.h> +#ifdef __MINGW32__ +/* mingw seems to be missing that definition. */ +const char *inet_ntop(int af, const void *src, char *dst, socklen_t size); +#endif + #endif /* GRPC_INTERNAL_CORE_IOMGR_SOCKADDR_WIN32_H */ 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_windows.c b/src/core/iomgr/tcp_server_windows.c index 59319da26d..0c3ab1dc91 100644 --- a/src/core/iomgr/tcp_server_windows.c +++ b/src/core/iomgr/tcp_server_windows.c @@ -53,9 +53,6 @@ #define INIT_PORT_CAP 2 #define MIN_SAFE_ACCEPT_QUEUE_SIZE 100 -static gpr_once s_init_max_accept_queue_size; -static int s_max_accept_queue_size; - /* one listening port */ typedef struct server_port { gpr_uint8 addresses[sizeof(struct sockaddr_in6) * 2 + 32]; diff --git a/src/core/iomgr/tcp_windows.c b/src/core/iomgr/tcp_windows.c index 3efd69a71b..ec5496e7ee 100644 --- a/src/core/iomgr/tcp_windows.c +++ b/src/core/iomgr/tcp_windows.c @@ -172,7 +172,7 @@ static void win_notify_on_read(grpc_endpoint *ep, tcp->read_slice = gpr_slice_malloc(8192); buffer.len = GPR_SLICE_LENGTH(tcp->read_slice); - buffer.buf = GPR_SLICE_START_PTR(tcp->read_slice); + buffer.buf = (char *)GPR_SLICE_START_PTR(tcp->read_slice); gpr_log(GPR_DEBUG, "win_notify_on_read: calling WSARecv without overlap"); status = WSARecv(tcp->socket->socket, &buffer, 1, &bytes_read, &flags, @@ -284,7 +284,7 @@ static grpc_endpoint_write_status win_write(grpc_endpoint *ep, for (i = 0; i < tcp->write_slices.count; i++) { buffers[i].len = GPR_SLICE_LENGTH(tcp->write_slices.slices[i]); - buffers[i].buf = GPR_SLICE_START_PTR(tcp->write_slices.slices[i]); + buffers[i].buf = (char *)GPR_SLICE_START_PTR(tcp->write_slices.slices[i]); } gpr_log(GPR_DEBUG, "win_write: calling WSASend without overlap"); |