aboutsummaryrefslogtreecommitdiffhomepage
path: root/src/core
diff options
context:
space:
mode:
authorGravatar Nicolas Noble <nnoble@google.com>2015-05-11 17:40:26 -0700
committerGravatar Nicolas "Pixel" Noble <pixel@nobis-crew.org>2015-05-12 23:24:15 +0200
commite1445368526e047d1f50a81c0194c1ca902b4d70 (patch)
tree5e0c33057f7b0feeda8537deca01bfd219d4512b /src/core
parent903810191e19ef0d2bb593815b0b5277b3eb5487 (diff)
Various Win32 fixes and improvements.
-) Properly setting up the endpoint pair. -) Beancounting on socket shutdown to properly add references. -) Only proceed to clear out data when called from the IOCP thread. -) Enabling ALL the tests. -) Fixing run_tests.py to properly invoke them.
Diffstat (limited to 'src/core')
-rw-r--r--src/core/iomgr/endpoint_pair_windows.c2
-rw-r--r--src/core/iomgr/iocp_windows.h1
-rw-r--r--src/core/iomgr/socket_windows.c6
-rw-r--r--src/core/iomgr/socket_windows.h4
-rw-r--r--src/core/iomgr/tcp_windows.c30
5 files changed, 31 insertions, 12 deletions
diff --git a/src/core/iomgr/endpoint_pair_windows.c b/src/core/iomgr/endpoint_pair_windows.c
index 7c945ebad4..988d622d01 100644
--- a/src/core/iomgr/endpoint_pair_windows.c
+++ b/src/core/iomgr/endpoint_pair_windows.c
@@ -56,6 +56,8 @@ static void create_sockets(SOCKET sv[2]) {
GPR_ASSERT(lst_sock != INVALID_SOCKET);
memset(&addr, 0, sizeof(addr));
+ addr.sin_addr.s_addr = htonl(INADDR_LOOPBACK);
+ addr.sin_family = AF_INET;
GPR_ASSERT(bind(lst_sock, (struct sockaddr*)&addr, sizeof(addr)) != SOCKET_ERROR);
GPR_ASSERT(listen(lst_sock, SOMAXCONN) != SOCKET_ERROR);
GPR_ASSERT(getsockname(lst_sock, (struct sockaddr*)&addr, &addr_len) != SOCKET_ERROR);
diff --git a/src/core/iomgr/iocp_windows.h b/src/core/iomgr/iocp_windows.h
index fa3f5eee10..4503fdd55e 100644
--- a/src/core/iomgr/iocp_windows.h
+++ b/src/core/iomgr/iocp_windows.h
@@ -40,6 +40,7 @@
#include "src/core/iomgr/socket_windows.h"
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 *);
diff --git a/src/core/iomgr/socket_windows.c b/src/core/iomgr/socket_windows.c
index 35dbfa1587..ee5150a696 100644
--- a/src/core/iomgr/socket_windows.c
+++ b/src/core/iomgr/socket_windows.c
@@ -59,17 +59,21 @@ grpc_winsocket *grpc_winsocket_create(SOCKET socket) {
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. */
-void grpc_winsocket_shutdown(grpc_winsocket *socket) {
+int grpc_winsocket_shutdown(grpc_winsocket *socket) {
+ int callbacks_set = 0;
gpr_mu_lock(&socket->state_mu);
if (socket->read_info.cb) {
+ callbacks_set++;
grpc_iomgr_add_delayed_callback(socket->read_info.cb,
socket->read_info.opaque, 0);
}
if (socket->write_info.cb) {
+ callbacks_set++;
grpc_iomgr_add_delayed_callback(socket->write_info.cb,
socket->write_info.opaque, 0);
}
gpr_mu_unlock(&socket->state_mu);
+ return callbacks_set;
}
/* Abandons a socket. Either we're going to queue it up for garbage collecting
diff --git a/src/core/iomgr/socket_windows.h b/src/core/iomgr/socket_windows.h
index 8898def854..b27eb14219 100644
--- a/src/core/iomgr/socket_windows.h
+++ b/src/core/iomgr/socket_windows.h
@@ -100,8 +100,8 @@ typedef struct grpc_winsocket {
grpc_winsocket *grpc_winsocket_create(SOCKET socket);
/* Initiate an asynchronous shutdown of the socket. Will call off any pending
- operation to cancel them. */
-void grpc_winsocket_shutdown(grpc_winsocket *socket);
+ operation to cancel them. Returns the number of callbacks that got setup. */
+int grpc_winsocket_shutdown(grpc_winsocket *socket);
/* Abandon a socket. */
void grpc_winsocket_orphan(grpc_winsocket *socket);
diff --git a/src/core/iomgr/tcp_windows.c b/src/core/iomgr/tcp_windows.c
index f16b4c1268..3341f558a3 100644
--- a/src/core/iomgr/tcp_windows.c
+++ b/src/core/iomgr/tcp_windows.c
@@ -118,12 +118,14 @@ static void on_read(void *tcpp, int from_iocp) {
gpr_slice *slice = NULL;
size_t nslices = 0;
grpc_endpoint_cb_status status;
- grpc_endpoint_read_cb cb = tcp->read_cb;
+ grpc_endpoint_read_cb cb;
grpc_winsocket_callback_info *info = &socket->read_info;
void *opaque = tcp->read_user_data;
int do_abort = 0;
gpr_mu_lock(&tcp->mu);
+ cb = tcp->read_cb;
+ tcp->read_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
@@ -133,9 +135,12 @@ static void on_read(void *tcpp, int from_iocp) {
gpr_mu_unlock(&tcp->mu);
if (do_abort) {
- if (from_iocp) gpr_slice_unref(tcp->read_slice);
+ if (from_iocp) {
+ tcp->socket->read_info.outstanding = 0;
+ gpr_slice_unref(tcp->read_slice);
+ }
tcp_unref(tcp);
- cb(opaque, NULL, 0, GRPC_ENDPOINT_CB_SHUTDOWN);
+ if (cb) cb(opaque, NULL, 0, GRPC_ENDPOINT_CB_SHUTDOWN);
return;
}
@@ -225,11 +230,13 @@ static void on_write(void *tcpp, int from_iocp) {
grpc_winsocket *handle = tcp->socket;
grpc_winsocket_callback_info *info = &handle->write_info;
grpc_endpoint_cb_status status = GRPC_ENDPOINT_CB_OK;
- grpc_endpoint_write_cb cb = tcp->write_cb;
+ grpc_endpoint_write_cb cb;
void *opaque = tcp->write_user_data;
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
@@ -238,15 +245,18 @@ static void on_write(void *tcpp, int from_iocp) {
}
gpr_mu_unlock(&tcp->mu);
- GPR_ASSERT(tcp->socket->write_info.outstanding);
-
if (do_abort) {
- if (from_iocp) gpr_slice_buffer_reset_and_unref(&tcp->write_slices);
+ if (from_iocp) {
+ tcp->socket->write_info.outstanding = 0;
+ gpr_slice_buffer_reset_and_unref(&tcp->write_slices);
+ }
tcp_unref(tcp);
- cb(opaque, GRPC_ENDPOINT_CB_SHUTDOWN);
+ if (cb) cb(opaque, GRPC_ENDPOINT_CB_SHUTDOWN);
return;
}
+ GPR_ASSERT(tcp->socket->write_info.outstanding);
+
if (info->wsa_error != 0) {
char *utf8_message = gpr_format_message(info->wsa_error);
gpr_log(GPR_ERROR, "WSASend overlapped error: %s", utf8_message);
@@ -361,11 +371,13 @@ static void win_add_to_pollset(grpc_endpoint *ep, grpc_pollset *pollset) {
concurrent access of the data structure in that regard. */
static void win_shutdown(grpc_endpoint *ep) {
grpc_tcp *tcp = (grpc_tcp *) ep;
+ int extra_refs = 0;
gpr_mu_lock(&tcp->mu);
/* At that point, what may happen is that we're already inside the IOCP
callback. See the comments in on_read and on_write. */
tcp->shutting_down = 1;
- grpc_winsocket_shutdown(tcp->socket);
+ extra_refs = grpc_winsocket_shutdown(tcp->socket);
+ while (extra_refs--) tcp_ref(tcp);
gpr_mu_unlock(&tcp->mu);
}