Chiebot-Mirror
/
grpc
mirror of https://github.com/grpc/grpc.git
8
0
Fork 0
Code Issues Projects Releases Wiki Activity

Merge pull request #1522 from nicolasnoble/better-win32-orphans

Browse Source 
Better win32 orphans and event support.
pull/1533/head^2
Jan Tattermusch 10 years ago
parent 72a96463f6 7f2e98c6ce
commit d781835542
7 changed files with 68 additions and 108 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Unified View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 2
      src/core/iomgr/endpoint_pair_windows.c
  2. 35
      src/core/iomgr/iocp_windows.c
  3. 7
      src/core/iomgr/socket_windows.c
  4. 13
      src/core/iomgr/socket_windows.h
  5. 24
      src/core/iomgr/tcp_client_windows.c
  6. 3
      src/core/iomgr/tcp_server_windows.c
  7. 92
      src/core/iomgr/tcp_windows.c

2
src/core/iomgr/endpoint_pair_windows.c
Unescape View File

@ -68,6 +68,8 @@ static void create_sockets(SOCKET sv[2]) {
GPR_ASSERT(svr_sock != INVALID_SOCKET); GPR_ASSERT(svr_sock != INVALID_SOCKET);
closesocket(lst_sock); closesocket(lst_sock);
grpc_tcp_prepare_socket(cli_sock);
grpc_tcp_prepare_socket(svr_sock);
sv[1] = cli_sock; sv[1] = cli_sock;
sv[0] = svr_sock; sv[0] = svr_sock;

35
src/core/iomgr/iocp_windows.c
Unescape View File

@ -53,6 +53,7 @@ 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 gpr_atm g_orphans = 0;
static gpr_atm g_custom_events = 0;
static HANDLE g_iocp; static HANDLE g_iocp;
@ -62,20 +63,19 @@ static void do_iocp_work() {
DWORD flags = 0; DWORD flags = 0;
ULONG_PTR completion_key; ULONG_PTR completion_key;
LPOVERLAPPED overlapped; LPOVERLAPPED overlapped;
gpr_timespec wait_time = gpr_inf_future;
grpc_winsocket *socket; grpc_winsocket *socket;
grpc_winsocket_callback_info *info; grpc_winsocket_callback_info *info;
void(*f)(void *, int) = NULL; void(*f)(void *, int) = NULL;
void *opaque = NULL; void *opaque = NULL;
success = GetQueuedCompletionStatus(g_iocp, &bytes, success = GetQueuedCompletionStatus(g_iocp, &bytes,
&completion_key, &overlapped, &completion_key, &overlapped,
gpr_time_to_millis(wait_time)); INFINITE);
if (!success && !overlapped) { /* success = 0 and overlapped = NULL means the deadline got attained.
/* The deadline got attained. */ Which is impossible. since our wait time is +inf */
return; GPR_ASSERT(success || overlapped);
}
GPR_ASSERT(completion_key && overlapped); GPR_ASSERT(completion_key && overlapped);
if (overlapped == &g_iocp_custom_overlap) { if (overlapped == &g_iocp_custom_overlap) {
gpr_atm_full_fetch_add(&g_custom_events, -1);
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. */
return; return;
@ -93,13 +93,17 @@ static void do_iocp_work() {
gpr_log(GPR_ERROR, "Unknown IOCP operation"); gpr_log(GPR_ERROR, "Unknown IOCP operation");
abort(); abort();
} }
success = WSAGetOverlappedResult(socket->socket, &info->overlapped, &bytes, GPR_ASSERT(info->outstanding);
FALSE, &flags);
if (socket->orphan) { if (socket->orphan) {
grpc_winsocket_destroy(socket); info->outstanding = 0;
gpr_atm_full_fetch_add(&g_orphans, -1); if (!socket->read_info.outstanding && !socket->write_info.outstanding) {
grpc_winsocket_destroy(socket);
gpr_atm_full_fetch_add(&g_orphans, -1);
}
return; return;
} }
success = WSAGetOverlappedResult(socket->socket, &info->overlapped, &bytes,
FALSE, &flags);
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);
@ -117,10 +121,14 @@ static void do_iocp_work() {
} }
static void iocp_loop(void *p) { static void iocp_loop(void *p) {
while (gpr_atm_acq_load(&g_orphans) || !gpr_event_get(&g_shutdown_iocp)) { void * eventshutdown = NULL;
while (gpr_atm_acq_load(&g_orphans) ||
gpr_atm_acq_load(&g_custom_events) ||
!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();
} }
gpr_log(GPR_DEBUG, "iocp_loop is done");
gpr_event_set(&g_iocp_done, (void *)1); gpr_event_set(&g_iocp_done, (void *)1);
} }
@ -128,8 +136,8 @@ static void iocp_loop(void *p) {
void grpc_iocp_init(void) { void grpc_iocp_init(void) {
gpr_thd_id id; gpr_thd_id id;
g_iocp = CreateIoCompletionPort(INVALID_HANDLE_VALUE, NULL, g_iocp = CreateIoCompletionPort(INVALID_HANDLE_VALUE,
(ULONG_PTR)NULL, 0); NULL, (ULONG_PTR)NULL, 0);
GPR_ASSERT(g_iocp); GPR_ASSERT(g_iocp);
gpr_event_init(&g_iocp_done); gpr_event_init(&g_iocp_done);
@ -140,6 +148,7 @@ void grpc_iocp_init(void) {
void grpc_iocp_kick(void) { void grpc_iocp_kick(void) {
BOOL success; BOOL success;
gpr_atm_full_fetch_add(&g_custom_events, 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);

7
src/core/iomgr/socket_windows.c
Unescape View File

@ -75,15 +75,14 @@ void grpc_winsocket_shutdown(grpc_winsocket *socket) {
/* Abandons a socket. Either we're going to queue it up for garbage collecting /* 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 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 mechanisms assumes that we're either always waiting for an operation, or we
explicitely know that we don't. If there is a future case where we can have 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 an "idle" socket which is neither trying to read or write, we'd start leaking
both memory and sockets. */ both memory and sockets. */
void grpc_winsocket_orphan(grpc_winsocket *winsocket) { void grpc_winsocket_orphan(grpc_winsocket *winsocket) {
SOCKET socket = winsocket->socket; SOCKET socket = winsocket->socket;
if (!winsocket->closed_early) { if (winsocket->read_info.outstanding || winsocket->write_info.outstanding) {
grpc_iocp_socket_orphan(winsocket); grpc_iocp_socket_orphan(winsocket);
} } else {
if (winsocket->closed_early) {
grpc_winsocket_destroy(winsocket); grpc_winsocket_destroy(winsocket);
} }
closesocket(socket); closesocket(socket);

13
src/core/iomgr/socket_windows.h
Unescape View File

@ -65,12 +65,14 @@ typedef struct grpc_winsocket_callback_info {
/* The results of the overlapped operation. */ /* The results of the overlapped operation. */
DWORD bytes_transfered; DWORD bytes_transfered;
int wsa_error; int wsa_error;
/* A boolean indicating that we started an operation. */
int outstanding;
} grpc_winsocket_callback_info; } grpc_winsocket_callback_info;
/* This is a wrapper to a Windows socket. A socket can have one outstanding /* This is a wrapper to a Windows socket. A socket can have one outstanding
read, and one outstanding write. Doing an asynchronous accept means waiting read, and one outstanding write. Doing an asynchronous accept means waiting
for a read operation. Doing an asynchronous connect means waiting for a for a read operation. Doing an asynchronous connect means waiting for a
write operation. These are completely abitrary ties between the operation write operation. These are completely arbitrary ties between the operation
and the kind of event, because we can have one overlapped per pending and the kind of event, because we can have one overlapped per pending
operation, whichever its nature is. So we could have more dedicated pending operation, whichever its nature is. So we could have more dedicated pending
operation callbacks for connect and listen. But given the scope of listen operation callbacks for connect and listen. But given the scope of listen
@ -87,17 +89,10 @@ typedef struct grpc_winsocket {
/* You can't add the same socket twice to the same IO Completion Port. /* You can't add the same socket twice to the same IO Completion Port.
This prevents that. */ This prevents that. */
int added_to_iocp; int added_to_iocp;
/* A boolean to indicate that the caller has abandonned that socket, but /* 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 there is a pending operation that the IO Completion Port will have to
wait for. The socket will be collected at that time. */ wait for. The socket will be collected at that time. */
int orphan; int orphan;
/* A boolean to indicate that the socket was already closed somehow, and
that no operation is going to be pending. Trying to abandon a socket in
that state won't result in an orphan, but will instead be destroyed
without further delay. We could avoid that boolean by adding one into
grpc_winsocket_callback_info describing that the operation is pending,
but that 1) waste memory more and 2) obfuscate the intent a bit more. */
int closed_early;
} grpc_winsocket; } grpc_winsocket;
/* Create a wrapped windows handle. This takes ownership of it, meaning that /* Create a wrapped windows handle. This takes ownership of it, meaning that

24
src/core/iomgr/tcp_client_windows.c
Unescape View File

@ -74,7 +74,7 @@ static void async_connect_cleanup(async_connect *ac) {
static void on_alarm(void *acp, int occured) { static void on_alarm(void *acp, int occured) {
async_connect *ac = acp; async_connect *ac = acp;
gpr_mu_lock(&ac->mu); gpr_mu_lock(&ac->mu);
/* If the alarm didn't occor, it got cancelled. */ /* If the alarm didn't occur, it got cancelled. */
if (ac->socket != NULL && occured) { if (ac->socket != NULL && occured) {
grpc_winsocket_shutdown(ac->socket); grpc_winsocket_shutdown(ac->socket);
} }
@ -98,6 +98,7 @@ static void on_connect(void *acp, int from_iocp) {
if (from_iocp) { if (from_iocp) {
DWORD transfered_bytes = 0; DWORD transfered_bytes = 0;
DWORD flags; DWORD flags;
info->outstanding = 0;
BOOL wsa_success = WSAGetOverlappedResult(sock, &info->overlapped, BOOL wsa_success = WSAGetOverlappedResult(sock, &info->overlapped,
&transfered_bytes, FALSE, &transfered_bytes, FALSE,
&flags); &flags);
@ -106,10 +107,8 @@ static void on_connect(void *acp, int from_iocp) {
char *utf8_message = gpr_format_message(WSAGetLastError()); char *utf8_message = gpr_format_message(WSAGetLastError());
gpr_log(GPR_ERROR, "on_connect error: %s", utf8_message); gpr_log(GPR_ERROR, "on_connect error: %s", utf8_message);
gpr_free(utf8_message); gpr_free(utf8_message);
goto finish; } else if (!aborted) {
} else {
ep = grpc_tcp_create(ac->socket); ep = grpc_tcp_create(ac->socket);
goto finish;
} }
} else { } else {
gpr_log(GPR_ERROR, "on_connect is shutting down"); gpr_log(GPR_ERROR, "on_connect is shutting down");
@ -125,20 +124,12 @@ static void on_connect(void *acp, int from_iocp) {
return; return;
} }
abort(); ac->socket->write_info.outstanding = 0;
finish:
/* If we don't have an endpoint, it means the connection failed, /* 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 so it doesn't matter if it aborted or failed. We need to orphan
that socket. */ that socket. */
if (!ep || aborted) { if (!ep || aborted) grpc_winsocket_orphan(ac->socket);
/* If the connection failed, it means we won't get an IOCP notification,
so let's flag it as already closed. But if the connection was aborted,
while we still got an endpoint, we have to wait for the IOCP to collect
that socket. So let's properly flag that. */
ac->socket->closed_early = !ep;
grpc_winsocket_orphan(ac->socket);
}
async_connect_cleanup(ac); async_connect_cleanup(ac);
/* If the connection was aborted, the callback was already called when /* If the connection was aborted, the callback was already called when
the deadline was met. */ the deadline was met. */
@ -189,7 +180,7 @@ void grpc_tcp_client_connect(void(*cb)(void *arg, grpc_endpoint *tcp),
&ioctl_num_bytes, NULL, NULL); &ioctl_num_bytes, NULL, NULL);
if (status != 0) { if (status != 0) {
message = "Unable to retreive ConnectEx pointer: %s"; message = "Unable to retrieve ConnectEx pointer: %s";
goto failure; goto failure;
} }
@ -204,6 +195,7 @@ void grpc_tcp_client_connect(void(*cb)(void *arg, grpc_endpoint *tcp),
socket = grpc_winsocket_create(sock); socket = grpc_winsocket_create(sock);
info = &socket->write_info; info = &socket->write_info;
info->outstanding = 1;
success = ConnectEx(sock, addr, addr_len, NULL, 0, NULL, &info->overlapped); 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 /* It wouldn't be unusual to get a success immediately. But we'll still get
@ -225,6 +217,7 @@ void grpc_tcp_client_connect(void(*cb)(void *arg, grpc_endpoint *tcp),
ac->aborted = 0; ac->aborted = 0;
grpc_alarm_init(&ac->alarm, deadline, on_alarm, ac, gpr_now()); grpc_alarm_init(&ac->alarm, deadline, on_alarm, ac, gpr_now());
socket->write_info.outstanding = 1;
grpc_socket_notify_on_write(socket, on_connect, ac); grpc_socket_notify_on_write(socket, on_connect, ac);
return; return;
@ -233,7 +226,6 @@ failure:
gpr_log(GPR_ERROR, message, utf8_message); gpr_log(GPR_ERROR, message, utf8_message);
gpr_free(utf8_message); gpr_free(utf8_message);
if (socket) { if (socket) {
socket->closed_early = 1;
grpc_winsocket_orphan(socket); grpc_winsocket_orphan(socket);
} else if (sock != INVALID_SOCKET) { } else if (sock != INVALID_SOCKET) {
closesocket(sock); closesocket(sock);

3
src/core/iomgr/tcp_server_windows.c
Unescape View File

@ -123,7 +123,6 @@ void grpc_tcp_server_destroy(grpc_tcp_server *s,
closed by the system. */ closed by the system. */
for (i = 0; i < s->nports; i++) { for (i = 0; i < s->nports; i++) {
server_port *sp = &s->ports[i]; server_port *sp = &s->ports[i];
sp->socket->closed_early = 1;
grpc_winsocket_orphan(sp->socket); grpc_winsocket_orphan(sp->socket);
} }
gpr_free(s->ports); gpr_free(s->ports);
@ -249,6 +248,7 @@ static void on_accept(void *arg, int from_iocp) {
if (sp->shutting_down) { if (sp->shutting_down) {
GPR_ASSERT(from_iocp); GPR_ASSERT(from_iocp);
sp->shutting_down = 0; sp->shutting_down = 0;
sp->socket->read_info.outstanding = 0;
gpr_mu_lock(&sp->server->mu); gpr_mu_lock(&sp->server->mu);
if (0 == --sp->server->active_ports) { if (0 == --sp->server->active_ports) {
gpr_cv_broadcast(&sp->server->cv); gpr_cv_broadcast(&sp->server->cv);
@ -420,6 +420,7 @@ void grpc_tcp_server_start(grpc_tcp_server *s, grpc_pollset **pollset,
s->cb = cb; s->cb = cb;
s->cb_arg = cb_arg; s->cb_arg = cb_arg;
for (i = 0; i < s->nports; i++) { for (i = 0; i < s->nports; i++) {
s->ports[i].socket->read_info.outstanding = 1;
start_accept(s->ports + i); start_accept(s->ports + i);
s->active_ports++; s->active_ports++;
} }

92
src/core/iomgr/tcp_windows.c
Unescape View File

@ -86,12 +86,10 @@ typedef struct grpc_tcp {
grpc_endpoint_read_cb read_cb; grpc_endpoint_read_cb read_cb;
void *read_user_data; void *read_user_data;
gpr_slice read_slice; gpr_slice read_slice;
int outstanding_read;
grpc_endpoint_write_cb write_cb; grpc_endpoint_write_cb write_cb;
void *write_user_data; void *write_user_data;
gpr_slice_buffer write_slices; gpr_slice_buffer write_slices;
int outstanding_write;
/* The IO Completion Port runs from another thread. We need some mechanism /* The IO Completion Port runs from another thread. We need some mechanism
to protect ourselves when requesting a shutdown. */ to protect ourselves when requesting a shutdown. */
@ -141,14 +139,13 @@ static void on_read(void *tcpp, int from_iocp) {
return; return;
} }
GPR_ASSERT(tcp->outstanding_read); GPR_ASSERT(tcp->socket->read_info.outstanding);
if (socket->read_info.wsa_error != 0) { if (socket->read_info.wsa_error != 0) {
char *utf8_message = gpr_format_message(info->wsa_error); char *utf8_message = gpr_format_message(info->wsa_error);
gpr_log(GPR_ERROR, "ReadFile overlapped error: %s", utf8_message); gpr_log(GPR_ERROR, "ReadFile overlapped error: %s", utf8_message);
gpr_free(utf8_message); gpr_free(utf8_message);
status = GRPC_ENDPOINT_CB_ERROR; status = GRPC_ENDPOINT_CB_ERROR;
socket->closed_early = 1;
} else { } else {
if (info->bytes_transfered != 0) { if (info->bytes_transfered != 0) {
sub = gpr_slice_sub(tcp->read_slice, 0, info->bytes_transfered); sub = gpr_slice_sub(tcp->read_slice, 0, info->bytes_transfered);
@ -161,7 +158,7 @@ static void on_read(void *tcpp, int from_iocp) {
} }
} }
tcp->outstanding_read = 0; tcp->socket->read_info.outstanding = 0;
tcp_unref(tcp); tcp_unref(tcp);
cb(opaque, slice, nslices, status); cb(opaque, slice, nslices, status);
@ -175,13 +172,15 @@ static void win_notify_on_read(grpc_endpoint *ep,
int status; int status;
DWORD bytes_read = 0; DWORD bytes_read = 0;
DWORD flags = 0; DWORD flags = 0;
int error;
WSABUF buffer; WSABUF buffer;
GPR_ASSERT(!tcp->outstanding_read); GPR_ASSERT(!tcp->socket->read_info.outstanding);
GPR_ASSERT(!tcp->shutting_down); if (tcp->shutting_down) {
cb(arg, NULL, 0, GRPC_ENDPOINT_CB_SHUTDOWN);
return;
}
tcp_ref(tcp); tcp_ref(tcp);
tcp->outstanding_read = 1; tcp->socket->read_info.outstanding = 1;
tcp->read_cb = cb; tcp->read_cb = cb;
tcp->read_user_data = arg; tcp->read_user_data = arg;
@ -208,34 +207,13 @@ static void win_notify_on_read(grpc_endpoint *ep,
status = WSARecv(tcp->socket->socket, &buffer, 1, &bytes_read, &flags, status = WSARecv(tcp->socket->socket, &buffer, 1, &bytes_read, &flags,
&info->overlapped, NULL); &info->overlapped, NULL);
if (status == 0) { if (status != 0) {
grpc_socket_notify_on_read(tcp->socket, on_read, tcp); int wsa_error = WSAGetLastError();
return; if (wsa_error != WSA_IO_PENDING) {
} info->wsa_error = wsa_error;
on_read(tcp, 1);
error = WSAGetLastError(); return;
}
if (error != WSA_IO_PENDING) {
char *utf8_message = gpr_format_message(WSAGetLastError());
gpr_log(GPR_ERROR, "WSARecv error: %s - this means we're going to leak.",
utf8_message);
gpr_free(utf8_message);
/* I'm pretty sure this is a very bad situation there. Hence the log.
What will happen now is that the socket will neither wait for read
or write, unless the caller retry, which is unlikely, but I am not
sure if that's guaranteed. And there might also be a write pending.
This means that the future orphanage of that socket will be in limbo,
and we're going to leak it. I have no idea what could cause this
specific case however, aside from a parameter error from our call.
Normal read errors would actually happen during the overlapped
operation, which is the supported way to go for that. */
tcp->outstanding_read = 0;
tcp_unref(tcp);
cb(arg, NULL, 0, GRPC_ENDPOINT_CB_ERROR);
/* Per the comment above, I'm going to treat that case as a hard failure
for now, and leave the option to catch that and debug. */
__debugbreak();
return;
} }
grpc_socket_notify_on_read(tcp->socket, on_read, tcp); grpc_socket_notify_on_read(tcp->socket, on_read, tcp);
@ -260,7 +238,7 @@ static void on_write(void *tcpp, int from_iocp) {
} }
gpr_mu_unlock(&tcp->mu); gpr_mu_unlock(&tcp->mu);
GPR_ASSERT(tcp->outstanding_write); GPR_ASSERT(tcp->socket->write_info.outstanding);
if (do_abort) { if (do_abort) {
if (from_iocp) gpr_slice_buffer_reset_and_unref(&tcp->write_slices); if (from_iocp) gpr_slice_buffer_reset_and_unref(&tcp->write_slices);
@ -274,13 +252,12 @@ static void on_write(void *tcpp, int from_iocp) {
gpr_log(GPR_ERROR, "WSASend overlapped error: %s", utf8_message); gpr_log(GPR_ERROR, "WSASend overlapped error: %s", utf8_message);
gpr_free(utf8_message); gpr_free(utf8_message);
status = GRPC_ENDPOINT_CB_ERROR; status = GRPC_ENDPOINT_CB_ERROR;
tcp->socket->closed_early = 1;
} else { } else {
GPR_ASSERT(info->bytes_transfered == tcp->write_slices.length); GPR_ASSERT(info->bytes_transfered == tcp->write_slices.length);
} }
gpr_slice_buffer_reset_and_unref(&tcp->write_slices); gpr_slice_buffer_reset_and_unref(&tcp->write_slices);
tcp->outstanding_write = 0; tcp->socket->write_info.outstanding = 0;
tcp_unref(tcp); tcp_unref(tcp);
cb(opaque, status); cb(opaque, status);
@ -301,11 +278,13 @@ static grpc_endpoint_write_status win_write(grpc_endpoint *ep,
WSABUF *allocated = NULL; WSABUF *allocated = NULL;
WSABUF *buffers = local_buffers; WSABUF *buffers = local_buffers;
GPR_ASSERT(!tcp->outstanding_write); GPR_ASSERT(!tcp->socket->write_info.outstanding);
GPR_ASSERT(!tcp->shutting_down); if (tcp->shutting_down) {
return GRPC_ENDPOINT_WRITE_ERROR;
}
tcp_ref(tcp); tcp_ref(tcp);
tcp->outstanding_write = 1; tcp->socket->write_info.outstanding = 1;
tcp->write_cb = cb; tcp->write_cb = cb;
tcp->write_user_data = arg; tcp->write_user_data = arg;
@ -341,7 +320,7 @@ static grpc_endpoint_write_status win_write(grpc_endpoint *ep,
} }
if (allocated) gpr_free(allocated); if (allocated) gpr_free(allocated);
gpr_slice_buffer_reset_and_unref(&tcp->write_slices); gpr_slice_buffer_reset_and_unref(&tcp->write_slices);
tcp->outstanding_write = 0; tcp->socket->write_info.outstanding = 0;
tcp_unref(tcp); tcp_unref(tcp);
return ret; return ret;
} }
@ -353,29 +332,12 @@ static grpc_endpoint_write_status win_write(grpc_endpoint *ep,
&bytes_sent, 0, &socket->write_info.overlapped, NULL); &bytes_sent, 0, &socket->write_info.overlapped, NULL);
if (allocated) gpr_free(allocated); if (allocated) gpr_free(allocated);
/* It is possible the operation completed then. But we'd still get an IOCP
notification. So let's ignore it and wait for the IOCP. */
if (status != 0) { if (status != 0) {
int error = WSAGetLastError(); int wsa_error = WSAGetLastError();
if (error != WSA_IO_PENDING) { if (wsa_error != WSA_IO_PENDING) {
char *utf8_message = gpr_format_message(WSAGetLastError()); gpr_slice_buffer_reset_and_unref(&tcp->write_slices);
gpr_log(GPR_ERROR, "WSASend error: %s - this means we're going to leak.", tcp->socket->write_info.outstanding = 0;
utf8_message);
gpr_free(utf8_message);
/* I'm pretty sure this is a very bad situation there. Hence the log.
What will happen now is that the socket will neither wait for read
or write, unless the caller retry, which is unlikely, but I am not
sure if that's guaranteed. And there might also be a read pending.
This means that the future orphanage of that socket will be in limbo,
and we're going to leak it. I have no idea what could cause this
specific case however, aside from a parameter error from our call.
Normal read errors would actually happen during the overlapped
operation, which is the supported way to go for that. */
tcp->outstanding_write = 0;
tcp_unref(tcp); tcp_unref(tcp);
/* Per the comment above, I'm going to treat that case as a hard failure
for now, and leave the option to catch that and debug. */
__debugbreak();
return GRPC_ENDPOINT_WRITE_ERROR; return GRPC_ENDPOINT_WRITE_ERROR;
} }
} }

Write Preview
Loading…
Cancel
Save