diff options
author | 2015-08-28 08:22:48 -0700 | |
---|---|---|
committer | 2015-08-28 08:22:48 -0700 | |
commit | 9f80fcf8e79d71e99b28f152a7b8662d815a6ee2 (patch) | |
tree | 0bce8e0136760df172a458d9826c2815cca1ddf8 /src/core/iomgr | |
parent | ae69ad1bcf0322200ebeed5476ee9978101beeb7 (diff) |
Cleaning up Windows endpoint code
- removed the need to track some state
- fixed chttp2 transport endpoint shutdown management
- made a bunch of tests pass
Diffstat (limited to 'src/core/iomgr')
-rw-r--r-- | src/core/iomgr/iocp_windows.c | 20 | ||||
-rw-r--r-- | src/core/iomgr/iocp_windows.h | 1 | ||||
-rw-r--r-- | src/core/iomgr/socket_windows.c | 41 | ||||
-rw-r--r-- | src/core/iomgr/socket_windows.h | 16 | ||||
-rw-r--r-- | src/core/iomgr/tcp_client_windows.c | 9 | ||||
-rw-r--r-- | src/core/iomgr/tcp_server_windows.c | 6 | ||||
-rw-r--r-- | src/core/iomgr/tcp_windows.c | 114 |
7 files changed, 46 insertions, 161 deletions
diff --git a/src/core/iomgr/iocp_windows.c b/src/core/iomgr/iocp_windows.c index 09a457dd9a..0799622497 100644 --- a/src/core/iomgr/iocp_windows.c +++ b/src/core/iomgr/iocp_windows.c @@ -52,7 +52,6 @@ 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 gpr_atm g_custom_events = 0; static HANDLE g_iocp; @@ -92,22 +91,13 @@ static void do_iocp_work() { gpr_log(GPR_ERROR, "Unknown IOCP operation"); abort(); } - GPR_ASSERT(info->outstanding); - if (socket->orphan) { - info->outstanding = 0; - if (!socket->read_info.outstanding && !socket->write_info.outstanding) { - grpc_winsocket_destroy(socket); - gpr_atm_full_fetch_add(&g_orphans, -1); - } - return; - } success = WSAGetOverlappedResult(socket->socket, &info->overlapped, &bytes, FALSE, &flags); info->bytes_transfered = bytes; info->wsa_error = success ? 0 : WSAGetLastError(); GPR_ASSERT(overlapped == &info->overlapped); - gpr_mu_lock(&socket->state_mu); GPR_ASSERT(!info->has_pending_iocp); + gpr_mu_lock(&socket->state_mu); if (info->cb) { f = info->cb; opaque = info->opaque; @@ -120,7 +110,7 @@ static void do_iocp_work() { } static void iocp_loop(void *p) { - while (gpr_atm_acq_load(&g_orphans) || gpr_atm_acq_load(&g_custom_events) || + while (gpr_atm_acq_load(&g_custom_events) || !gpr_event_get(&g_shutdown_iocp)) { grpc_maybe_call_delayed_callbacks(NULL, 1); do_iocp_work(); @@ -175,12 +165,6 @@ void grpc_iocp_add_socket(grpc_winsocket *socket) { GPR_ASSERT(ret == g_iocp); } -void grpc_iocp_socket_orphan(grpc_winsocket *socket) { - GPR_ASSERT(!socket->orphan); - gpr_atm_full_fetch_add(&g_orphans, 1); - socket->orphan = 1; -} - /* Calling notify_on_read or write means either of two things: -) The IOCP already completed in the background, and we need to call the callback now. diff --git a/src/core/iomgr/iocp_windows.h b/src/core/iomgr/iocp_windows.h index ee3847a229..7d2dc45176 100644 --- a/src/core/iomgr/iocp_windows.h +++ b/src/core/iomgr/iocp_windows.h @@ -42,7 +42,6 @@ void grpc_iocp_init(void); void grpc_iocp_kick(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/socket_windows.c b/src/core/iomgr/socket_windows.c index 7d8421376b..dbc64637b5 100644 --- a/src/core/iomgr/socket_windows.c +++ b/src/core/iomgr/socket_windows.c @@ -62,46 +62,13 @@ grpc_winsocket *grpc_winsocket_create(SOCKET socket, const char *name) { operations to abort them. We need to do that this way because of the various callsites of that function, which happens to be in various mutex hold states, and that'd be unsafe to call them directly. */ -int grpc_winsocket_shutdown(grpc_winsocket *winsocket) { - int callbacks_set = 0; - SOCKET socket; - gpr_mu_lock(&winsocket->state_mu); - socket = winsocket->socket; - if (winsocket->read_info.cb) { - callbacks_set++; - grpc_iomgr_closure_init(&winsocket->shutdown_closure, - winsocket->read_info.cb, - winsocket->read_info.opaque); - grpc_iomgr_add_delayed_callback(&winsocket->shutdown_closure, 0); - } - if (winsocket->write_info.cb) { - callbacks_set++; - grpc_iomgr_closure_init(&winsocket->shutdown_closure, - winsocket->write_info.cb, - winsocket->write_info.opaque); - grpc_iomgr_add_delayed_callback(&winsocket->shutdown_closure, 0); - } - gpr_mu_unlock(&winsocket->state_mu); - closesocket(socket); - return callbacks_set; -} - -/* Abandons a socket. Either we're going to queue it up for garbage collecting - from the IO Completion Port thread, or destroy it immediately. Note that this - mechanisms assumes that we're either always waiting for an operation, or we - explicitly know that we don't. If there is a future case where we can have - an "idle" socket which is neither trying to read or write, we'd start leaking - both memory and sockets. */ -void grpc_winsocket_orphan(grpc_winsocket *winsocket) { - grpc_iomgr_unregister_object(&winsocket->iomgr_object); - if (winsocket->read_info.outstanding || winsocket->write_info.outstanding) { - grpc_iocp_socket_orphan(winsocket); - } else { - grpc_winsocket_destroy(winsocket); - } +void grpc_winsocket_shutdown(grpc_winsocket *winsocket) { + shutdown(winsocket->socket, SD_BOTH); } void grpc_winsocket_destroy(grpc_winsocket *winsocket) { + closesocket(winsocket->socket); + grpc_iomgr_unregister_object(&winsocket->iomgr_object); gpr_mu_destroy(&winsocket->state_mu); gpr_free(winsocket); } diff --git a/src/core/iomgr/socket_windows.h b/src/core/iomgr/socket_windows.h index ecf2530173..498921e0fd 100644 --- a/src/core/iomgr/socket_windows.h +++ b/src/core/iomgr/socket_windows.h @@ -68,8 +68,6 @@ typedef struct grpc_winsocket_callback_info { /* The results of the overlapped operation. */ DWORD bytes_transfered; int wsa_error; - /* A boolean indicating that we started an operation. */ - int outstanding; } grpc_winsocket_callback_info; /* This is a wrapper to a Windows socket. A socket can have one outstanding @@ -92,10 +90,6 @@ typedef struct grpc_winsocket { /* You can't add the same socket twice to the same IO Completion Port. This prevents that. */ int added_to_iocp; - /* A boolean to indicate that the caller has abandoned that socket, but - there is a pending operation that the IO Completion Port will have to - wait for. The socket will be collected at that time. */ - int orphan; grpc_iomgr_closure shutdown_closure; @@ -108,14 +102,10 @@ typedef struct grpc_winsocket { grpc_winsocket *grpc_winsocket_create(SOCKET socket, const char *name); /* Initiate an asynchronous shutdown of the socket. Will call off any pending - operation to cancel them. Returns the number of callbacks that got setup. */ -int grpc_winsocket_shutdown(grpc_winsocket *socket); + operation to cancel them. */ +void grpc_winsocket_shutdown(grpc_winsocket *socket); -/* Abandon a socket. */ -void grpc_winsocket_orphan(grpc_winsocket *socket); - -/* Destroy a socket. Should only be called by the IO Completion Port thread, - or by grpc_winsocket_orphan if there's no pending operation. */ +/* Destroy a socket. Should only be called if there's no pending operation. */ void grpc_winsocket_destroy(grpc_winsocket *socket); #endif /* GRPC_INTERNAL_CORE_IOMGR_SOCKET_WINDOWS_H */ diff --git a/src/core/iomgr/tcp_client_windows.c b/src/core/iomgr/tcp_client_windows.c index 79a58fe2af..665ef2885f 100644 --- a/src/core/iomgr/tcp_client_windows.c +++ b/src/core/iomgr/tcp_client_windows.c @@ -102,7 +102,6 @@ static void on_connect(void *acp, int from_iocp) { DWORD flags; BOOL wsa_success = WSAGetOverlappedResult(sock, &info->overlapped, &transfered_bytes, FALSE, &flags); - info->outstanding = 0; GPR_ASSERT(transfered_bytes == 0); if (!wsa_success) { char *utf8_message = gpr_format_message(WSAGetLastError()); @@ -125,12 +124,10 @@ static void on_connect(void *acp, int from_iocp) { return; } - ac->socket->write_info.outstanding = 0; - /* If we don't have an endpoint, it means the connection failed, so it doesn't matter if it aborted or failed. We need to orphan that socket. */ - if (!ep || aborted) grpc_winsocket_orphan(ac->socket); + if (!ep || aborted) grpc_winsocket_destroy(ac->socket); async_connect_cleanup(ac); /* If the connection was aborted, the callback was already called when the deadline was met. */ @@ -196,7 +193,6 @@ void grpc_tcp_client_connect(void (*cb)(void *arg, grpc_endpoint *tcp), socket = grpc_winsocket_create(sock, "client"); info = &socket->write_info; - info->outstanding = 1; success = ConnectEx(sock, addr, addr_len, NULL, 0, NULL, &info->overlapped); /* It wouldn't be unusual to get a success immediately. But we'll still get @@ -220,7 +216,6 @@ void grpc_tcp_client_connect(void (*cb)(void *arg, grpc_endpoint *tcp), grpc_alarm_init(&ac->alarm, deadline, on_alarm, ac, gpr_now(GPR_CLOCK_MONOTONIC)); - socket->write_info.outstanding = 1; grpc_socket_notify_on_write(socket, on_connect, ac); return; @@ -229,7 +224,7 @@ failure: gpr_log(GPR_ERROR, message, utf8_message); gpr_free(utf8_message); if (socket) { - grpc_winsocket_orphan(socket); + grpc_winsocket_destroy(socket); } else if (sock != INVALID_SOCKET) { closesocket(sock); } diff --git a/src/core/iomgr/tcp_server_windows.c b/src/core/iomgr/tcp_server_windows.c index d0478d3604..79ea223d54 100644 --- a/src/core/iomgr/tcp_server_windows.c +++ b/src/core/iomgr/tcp_server_windows.c @@ -116,7 +116,7 @@ void grpc_tcp_server_destroy(grpc_tcp_server *s, for (i = 0; i < s->nports; i++) { server_port *sp = &s->ports[i]; sp->shutting_down = 1; - s->iomgr_callbacks_pending += grpc_winsocket_shutdown(sp->socket); + grpc_winsocket_shutdown(sp->socket); } /* This happens asynchronously. Wait while that happens. */ while (s->active_ports || s->iomgr_callbacks_pending) { @@ -129,7 +129,7 @@ void grpc_tcp_server_destroy(grpc_tcp_server *s, closed by the system. */ for (i = 0; i < s->nports; i++) { server_port *sp = &s->ports[i]; - grpc_winsocket_orphan(sp->socket); + grpc_winsocket_destroy(sp->socket); } gpr_free(s->ports); gpr_free(s); @@ -189,7 +189,6 @@ error: static void decrement_active_ports_and_notify(server_port *sp) { sp->shutting_down = 0; - sp->socket->read_info.outstanding = 0; gpr_mu_lock(&sp->server->mu); GPR_ASSERT(sp->server->active_ports > 0); if (0 == --sp->server->active_ports) { @@ -462,7 +461,6 @@ void grpc_tcp_server_start(grpc_tcp_server *s, grpc_pollset **pollset, s->cb = cb; s->cb_arg = cb_arg; for (i = 0; i < s->nports; i++) { - s->ports[i].socket->read_info.outstanding = 1; start_accept(s->ports + i); s->active_ports++; } diff --git a/src/core/iomgr/tcp_windows.c b/src/core/iomgr/tcp_windows.c index 58f9160ef9..fe3673c607 100644 --- a/src/core/iomgr/tcp_windows.c +++ b/src/core/iomgr/tcp_windows.c @@ -97,7 +97,7 @@ typedef struct grpc_tcp { } grpc_tcp; static void tcp_free(grpc_tcp *tcp) { - grpc_winsocket_orphan(tcp->socket); + grpc_winsocket_destroy(tcp->socket); gpr_mu_destroy(&tcp->mu); gpr_free(tcp->peer_string); gpr_free(tcp); @@ -135,55 +135,35 @@ static void tcp_ref(grpc_tcp *tcp) { gpr_ref(&tcp->refcount); } #endif /* Asynchronous callback from the IOCP, or the background thread. */ -static int on_read(grpc_tcp *tcp, int from_iocp) { +static int on_read(grpc_tcp *tcp, int success) { grpc_winsocket *socket = tcp->socket; gpr_slice sub; gpr_slice *slice = NULL; size_t nslices = 0; - int success; grpc_winsocket_callback_info *info = &socket->read_info; int do_abort = 0; - gpr_mu_lock(&tcp->mu); - if (!from_iocp || tcp->shutting_down) { - /* If we are here with from_iocp set to true, it means we got raced to - shutting down the endpoint. No actual abort callback will happen - though, so we're going to do it from here. */ - do_abort = 1; - } - gpr_mu_unlock(&tcp->mu); - - if (do_abort) { - if (from_iocp) { - tcp->socket->read_info.outstanding = 0; + if (success) { + if (socket->read_info.wsa_error != 0) { + if (socket->read_info.wsa_error != WSAECONNRESET) { + char *utf8_message = gpr_format_message(info->wsa_error); + gpr_log(GPR_ERROR, "ReadFile overlapped error: %s", utf8_message); + gpr_free(utf8_message); + } + success = 0; gpr_slice_unref(tcp->read_slice); - } - return 0; - } - - GPR_ASSERT(tcp->socket->read_info.outstanding); - - if (socket->read_info.wsa_error != 0) { - if (socket->read_info.wsa_error != WSAECONNRESET) { - char *utf8_message = gpr_format_message(info->wsa_error); - gpr_log(GPR_ERROR, "ReadFile overlapped error: %s", utf8_message); - gpr_free(utf8_message); - } - success = 0; - gpr_slice_unref(tcp->read_slice); - } else { - if (info->bytes_transfered != 0) { - sub = gpr_slice_sub_no_ref(tcp->read_slice, 0, info->bytes_transfered); - gpr_slice_buffer_add(tcp->read_slices, sub); - success = 1; } else { - gpr_slice_unref(tcp->read_slice); - success = 0; + if (info->bytes_transfered != 0) { + sub = gpr_slice_sub_no_ref(tcp->read_slice, 0, info->bytes_transfered); + gpr_slice_buffer_add(tcp->read_slices, sub); + success = 1; + } else { + gpr_slice_unref(tcp->read_slice); + success = 0; + } } } - tcp->socket->read_info.outstanding = 0; - return success; } @@ -209,14 +189,10 @@ static grpc_endpoint_op_status win_read(grpc_endpoint *ep, DWORD flags = 0; WSABUF buffer; - GPR_ASSERT(!tcp->socket->read_info.outstanding); if (tcp->shutting_down) { return GRPC_ENDPOINT_ERROR; } - TCP_REF(tcp, "read"); - - tcp->socket->read_info.outstanding = 1; tcp->read_cb = cb; tcp->read_slices = read_slices; gpr_slice_buffer_reset_and_unref(read_slices); @@ -236,10 +212,11 @@ static grpc_endpoint_op_status win_read(grpc_endpoint *ep, int ok; info->bytes_transfered = bytes_read; ok = on_read(tcp, 1); - TCP_UNREF(tcp, "read"); return ok ? GRPC_ENDPOINT_DONE : GRPC_ENDPOINT_ERROR; } + TCP_REF(tcp, "read"); + /* Otherwise, let's retry, by queuing a read. */ memset(&tcp->socket->read_info.overlapped, 0, sizeof(OVERLAPPED)); status = WSARecv(tcp->socket->socket, &buffer, 1, &bytes_read, &flags, @@ -260,52 +237,31 @@ static grpc_endpoint_op_status win_read(grpc_endpoint *ep, } /* Asynchronous callback from the IOCP, or the background thread. */ -static void on_write(void *tcpp, int from_iocp) { +static void on_write(void *tcpp, int success) { grpc_tcp *tcp = (grpc_tcp *)tcpp; grpc_winsocket *handle = tcp->socket; grpc_winsocket_callback_info *info = &handle->write_info; grpc_iomgr_closure *cb; - int success; int do_abort = 0; gpr_mu_lock(&tcp->mu); cb = tcp->write_cb; tcp->write_cb = NULL; - if (!from_iocp || tcp->shutting_down) { - /* If we are here with from_iocp set to true, it means we got raced to - shutting down the endpoint. No actual abort callback will happen - though, so we're going to do it from here. */ - do_abort = 1; - } gpr_mu_unlock(&tcp->mu); - if (do_abort) { - if (from_iocp) { - tcp->socket->write_info.outstanding = 0; - } - TCP_UNREF(tcp, "write"); - if (cb) { - cb->cb(cb->cb_arg, 0); - } - return; - } - - GPR_ASSERT(tcp->socket->write_info.outstanding); - - if (info->wsa_error != 0) { - if (info->wsa_error != WSAECONNRESET) { - char *utf8_message = gpr_format_message(info->wsa_error); - gpr_log(GPR_ERROR, "WSASend overlapped error: %s", utf8_message); - gpr_free(utf8_message); + if (success) { + if (info->wsa_error != 0) { + if (info->wsa_error != WSAECONNRESET) { + char *utf8_message = gpr_format_message(info->wsa_error); + gpr_log(GPR_ERROR, "WSASend overlapped error: %s", utf8_message); + gpr_free(utf8_message); + } + success = 0; + } else { + GPR_ASSERT(info->bytes_transfered == tcp->write_slices->length); } - success = 0; - } else { - GPR_ASSERT(info->bytes_transfered == tcp->write_slices->length); - success = 1; } - tcp->socket->write_info.outstanding = 0; - TCP_UNREF(tcp, "write"); cb->cb(cb->cb_arg, success); } @@ -324,13 +280,10 @@ static grpc_endpoint_op_status win_write(grpc_endpoint *ep, WSABUF *allocated = NULL; WSABUF *buffers = local_buffers; - GPR_ASSERT(!tcp->socket->write_info.outstanding); if (tcp->shutting_down) { return GRPC_ENDPOINT_ERROR; } - TCP_REF(tcp, "write"); - tcp->socket->write_info.outstanding = 1; tcp->write_cb = cb; tcp->write_slices = slices; @@ -365,11 +318,11 @@ static grpc_endpoint_op_status win_write(grpc_endpoint *ep, } } if (allocated) gpr_free(allocated); - tcp->socket->write_info.outstanding = 0; - TCP_UNREF(tcp, "write"); return ret; } + TCP_REF(tcp, "write"); + /* If we got a WSAEWOULDBLOCK earlier, then we need to re-do the same operation, this time asynchronously. */ memset(&socket->write_info.overlapped, 0, sizeof(OVERLAPPED)); @@ -380,7 +333,6 @@ static grpc_endpoint_op_status win_write(grpc_endpoint *ep, if (status != 0) { int wsa_error = WSAGetLastError(); if (wsa_error != WSA_IO_PENDING) { - tcp->socket->write_info.outstanding = 0; TCP_UNREF(tcp, "write"); return GRPC_ENDPOINT_ERROR; } |