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.
pull/1114/head
Nicolas Noble 10 years ago committed by Nicolas "Pixel" Noble
parent a42bd2705a
commit 8703f4d0c2
  1. 34
      src/core/iomgr/iocp_windows.c
  2. 1
      src/core/iomgr/iocp_windows.h
  3. 11
      src/core/iomgr/iomgr.c
  4. 7
      src/core/iomgr/socket_windows.c
  5. 6
      src/core/iomgr/socket_windows.h

@ -52,10 +52,11 @@ static OVERLAPPED g_iocp_custom_overlap;
static gpr_event g_shutdown_iocp; static gpr_event g_shutdown_iocp;
static gpr_event g_iocp_done; static gpr_event g_iocp_done;
static gpr_atm g_orphans = 0;
static HANDLE g_iocp; static HANDLE g_iocp;
static int do_iocp_work() { static void do_iocp_work() {
BOOL success; BOOL success;
DWORD bytes = 0; DWORD bytes = 0;
DWORD flags = 0; DWORD flags = 0;
@ -71,14 +72,14 @@ static int do_iocp_work() {
gpr_time_to_millis(wait_time)); gpr_time_to_millis(wait_time));
if (!success && !overlapped) { if (!success && !overlapped) {
/* The deadline got attained. */ /* The deadline got attained. */
return 0; return;
} }
GPR_ASSERT(completion_key && overlapped); GPR_ASSERT(completion_key && overlapped);
if (overlapped == &g_iocp_custom_overlap) { if (overlapped == &g_iocp_custom_overlap) {
if (completion_key == (ULONG_PTR) &g_iocp_kick_token) { if (completion_key == (ULONG_PTR) &g_iocp_kick_token) {
/* We were awoken from a kick. */ /* We were awoken from a kick. */
gpr_log(GPR_DEBUG, "do_iocp_work - got a kick"); gpr_log(GPR_DEBUG, "do_iocp_work - got a kick");
return 1; return;
} }
gpr_log(GPR_ERROR, "Unknown custom completion key."); gpr_log(GPR_ERROR, "Unknown custom completion key.");
abort(); abort();
@ -97,8 +98,13 @@ static int do_iocp_work() {
} }
success = WSAGetOverlappedResult(socket->socket, &info->overlapped, &bytes, success = WSAGetOverlappedResult(socket->socket, &info->overlapped, &bytes,
FALSE, &flags); FALSE, &flags);
gpr_log(GPR_DEBUG, "bytes: %u, flags: %u - op %s", bytes, flags, gpr_log(GPR_DEBUG, "bytes: %u, flags: %u - op %s %s", bytes, flags,
success ? "succeeded" : "failed"); 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->bytes_transfered = bytes;
info->wsa_error = success ? 0 : WSAGetLastError(); info->wsa_error = success ? 0 : WSAGetLastError();
GPR_ASSERT(overlapped == &info->overlapped); GPR_ASSERT(overlapped == &info->overlapped);
@ -113,12 +119,10 @@ static int do_iocp_work() {
} }
gpr_mu_unlock(&socket->state_mu); gpr_mu_unlock(&socket->state_mu);
if (f) f(opaque, 1); if (f) f(opaque, 1);
return 1;
} }
static void iocp_loop(void *p) { 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); grpc_maybe_call_delayed_callbacks(NULL, 1);
do_iocp_work(); do_iocp_work();
} }
@ -138,13 +142,19 @@ void grpc_iocp_init(void) {
gpr_thd_new(&id, iocp_loop, NULL, NULL); gpr_thd_new(&id, iocp_loop, NULL, NULL);
} }
void grpc_iocp_shutdown(void) { void grpc_iocp_kick(void) {
BOOL success; BOOL success;
gpr_event_set(&g_shutdown_iocp, (void *)1);
success = PostQueuedCompletionStatus(g_iocp, 0, success = PostQueuedCompletionStatus(g_iocp, 0,
(ULONG_PTR) &g_iocp_kick_token, (ULONG_PTR) &g_iocp_kick_token,
&g_iocp_custom_overlap); &g_iocp_custom_overlap);
GPR_ASSERT(success); 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); gpr_event_wait(&g_iocp_done, gpr_inf_future);
success = CloseHandle(g_iocp); success = CloseHandle(g_iocp);
GPR_ASSERT(success); GPR_ASSERT(success);
@ -166,6 +176,10 @@ void grpc_iocp_add_socket(grpc_winsocket *socket) {
GPR_ASSERT(ret == g_iocp); 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, static void socket_notify_on_iocp(grpc_winsocket *socket,
void(*cb)(void *, int), void *opaque, void(*cb)(void *, int), void *opaque,
grpc_winsocket_callback_info *info) { grpc_winsocket_callback_info *info) {

@ -42,6 +42,7 @@
void grpc_iocp_init(void); void grpc_iocp_init(void);
void grpc_iocp_shutdown(void); void grpc_iocp_shutdown(void);
void grpc_iocp_add_socket(grpc_winsocket *); 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 grpc_socket_notify_on_write(grpc_winsocket *, void(*cb)(void *, int success),
void *opaque); void *opaque);

@ -117,7 +117,16 @@ void grpc_iomgr_shutdown(void) {
gpr_mu_lock(&g_mu); gpr_mu_lock(&g_mu);
} }
if (g_refs) { 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, gpr_log(GPR_DEBUG,
"Failed to free %d iomgr objects before shutdown deadline: " "Failed to free %d iomgr objects before shutdown deadline: "
"memory leaks are likely", "memory leaks are likely",

@ -55,7 +55,7 @@ grpc_winsocket *grpc_winsocket_create(SOCKET socket) {
return r; return r;
} }
void shutdown_op(grpc_winsocket_callback_info *info) { static void shutdown_op(grpc_winsocket_callback_info *info) {
if (!info->cb) return; if (!info->cb) return;
grpc_iomgr_add_delayed_callback(info->cb, info->opaque, 0); 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) { void grpc_winsocket_orphan(grpc_winsocket *socket) {
gpr_log(GPR_DEBUG, "grpc_winsocket_orphan"); gpr_log(GPR_DEBUG, "grpc_winsocket_orphan");
grpc_iocp_socket_orphan(socket);
socket->orphan = 1;
grpc_iomgr_unref(); grpc_iomgr_unref();
closesocket(socket->socket); closesocket(socket->socket);
}
void grpc_winsocket_destroy(grpc_winsocket *socket) {
gpr_mu_destroy(&socket->state_mu); gpr_mu_destroy(&socket->state_mu);
gpr_free(socket); gpr_free(socket);
} }

@ -57,12 +57,13 @@ typedef struct grpc_winsocket_callback_info {
typedef struct grpc_winsocket { typedef struct grpc_winsocket {
SOCKET socket; SOCKET socket;
int added_to_iocp;
grpc_winsocket_callback_info write_info; grpc_winsocket_callback_info write_info;
grpc_winsocket_callback_info read_info; grpc_winsocket_callback_info read_info;
gpr_mu state_mu; gpr_mu state_mu;
int added_to_iocp;
int orphan;
} grpc_winsocket; } grpc_winsocket;
/* Create a wrapped windows handle. /* 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_shutdown(grpc_winsocket *socket);
void grpc_winsocket_orphan(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 */ #endif /* GRPC_INTERNAL_CORE_IOMGR_SOCKET_WINDOWS_H */

Loading…
Cancel
Save