Update Windows to new endpoint API
Also solve an infinite recursion in chttp2_transport
diff --git a/src/core/iomgr/tcp_windows.c b/src/core/iomgr/tcp_windows.c
index 123f46d..a63123c 100644
--- a/src/core/iomgr/tcp_windows.c
+++ b/src/core/iomgr/tcp_windows.c
@@ -82,13 +82,11 @@
/* Refcounting how many operations are in progress. */
gpr_refcount refcount;
- grpc_endpoint_read_cb read_cb;
- void *read_user_data;
+ grpc_iomgr_closure *read_cb;
+ grpc_iomgr_closure *write_cb;
gpr_slice read_slice;
-
- grpc_endpoint_write_cb write_cb;
- void *write_user_data;
- gpr_slice_buffer write_slices;
+ gpr_slice_buffer *write_slices;
+ gpr_slice_buffer *read_slices;
/* The IO Completion Port runs from another thread. We need some mechanism
to protect ourselves when requesting a shutdown. */
@@ -102,7 +100,6 @@
static void tcp_unref(grpc_tcp *tcp) {
if (gpr_unref(&tcp->refcount)) {
- gpr_slice_buffer_destroy(&tcp->write_slices);
grpc_winsocket_orphan(tcp->socket);
gpr_mu_destroy(&tcp->mu);
gpr_free(tcp->peer_string);
@@ -111,21 +108,16 @@
}
/* Asynchronous callback from the IOCP, or the background thread. */
-static void on_read(void *tcpp, int from_iocp) {
- grpc_tcp *tcp = (grpc_tcp *)tcpp;
+static int on_read(grpc_tcp *tcp, int from_iocp) {
grpc_winsocket *socket = tcp->socket;
gpr_slice sub;
gpr_slice *slice = NULL;
size_t nslices = 0;
- grpc_endpoint_cb_status status;
- grpc_endpoint_read_cb cb;
+ int success;
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
@@ -140,8 +132,7 @@
gpr_slice_unref(tcp->read_slice);
}
tcp_unref(tcp);
- if (cb) cb(opaque, NULL, 0, GRPC_ENDPOINT_CB_SHUTDOWN);
- return;
+ return 0;
}
GPR_ASSERT(tcp->socket->read_info.outstanding);
@@ -152,27 +143,33 @@
gpr_log(GPR_ERROR, "ReadFile overlapped error: %s", utf8_message);
gpr_free(utf8_message);
}
- status = GRPC_ENDPOINT_CB_ERROR;
+ success = 0;
} else {
if (info->bytes_transfered != 0) {
sub = gpr_slice_sub_no_ref(tcp->read_slice, 0, info->bytes_transfered);
- status = GRPC_ENDPOINT_CB_OK;
- slice = ⊂
- nslices = 1;
+ gpr_slice_buffer_add(tcp->read_slices, sub);
+ success = 1;
} else {
gpr_slice_unref(tcp->read_slice);
- status = GRPC_ENDPOINT_CB_EOF;
+ success = 0;
}
}
tcp->socket->read_info.outstanding = 0;
- tcp_unref(tcp);
- cb(opaque, slice, nslices, status);
+ return success;
}
-static void win_notify_on_read(grpc_endpoint *ep, grpc_endpoint_read_cb cb,
- void *arg) {
+static void on_read_cb(void *tcpp, int from_iocp) {
+ grpc_tcp *tcp = tcpp;
+ grpc_iomgr_closure *cb = tcp->read_cb;
+ int success = on_read(tcp, from_iocp);
+ tcp->read_cb = NULL;
+ tcp_unref(tcp);
+ cb->cb(cb->cb_arg, success);
+}
+
+static grpc_endpoint_op_status win_read(grpc_endpoint *ep, gpr_slice_buffer *read_slices, grpc_iomgr_closure *cb) {
grpc_tcp *tcp = (grpc_tcp *)ep;
grpc_winsocket *handle = tcp->socket;
grpc_winsocket_callback_info *info = &handle->read_info;
@@ -183,13 +180,11 @@
GPR_ASSERT(!tcp->socket->read_info.outstanding);
if (tcp->shutting_down) {
- cb(arg, NULL, 0, GRPC_ENDPOINT_CB_SHUTDOWN);
- return;
+ return GRPC_ENDPOINT_ERROR;
}
- tcp_ref(tcp);
tcp->socket->read_info.outstanding = 1;
tcp->read_cb = cb;
- tcp->read_user_data = arg;
+ tcp->read_slices = read_slices;
tcp->read_slice = gpr_slice_malloc(8192);
@@ -204,9 +199,8 @@
/* Did we get data immediately ? Yay. */
if (info->wsa_error != WSAEWOULDBLOCK) {
info->bytes_transfered = bytes_read;
- /* This might heavily recurse. */
- on_read(tcp, 1);
- return;
+ gpr_log(GPR_DEBUG, "immread: %d bytes", bytes_read);
+ return on_read(tcp, 1) ? GRPC_ENDPOINT_DONE : GRPC_ENDPOINT_ERROR;
}
/* Otherwise, let's retry, by queuing a read. */
@@ -218,12 +212,14 @@
int wsa_error = WSAGetLastError();
if (wsa_error != WSA_IO_PENDING) {
info->wsa_error = wsa_error;
- on_read(tcp, 1);
- return;
+ gpr_log(GPR_DEBUG, "immread: err=%d", wsa_error);
+ return on_read(tcp, 1) ? GRPC_ENDPOINT_DONE : GRPC_ENDPOINT_ERROR;
}
}
- grpc_socket_notify_on_read(tcp->socket, on_read, tcp);
+ tcp_ref(tcp);
+ grpc_socket_notify_on_read(tcp->socket, on_read_cb, tcp);
+ return GRPC_ENDPOINT_PENDING;
}
/* Asynchronous callback from the IOCP, or the background thread. */
@@ -231,9 +227,8 @@
grpc_tcp *tcp = (grpc_tcp *)tcpp;
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;
- void *opaque = tcp->write_user_data;
+ grpc_iomgr_closure *cb;
+ int success;
int do_abort = 0;
gpr_mu_lock(&tcp->mu);
@@ -250,10 +245,11 @@
if (do_abort) {
if (from_iocp) {
tcp->socket->write_info.outstanding = 0;
- gpr_slice_buffer_reset_and_unref(&tcp->write_slices);
}
tcp_unref(tcp);
- if (cb) cb(opaque, GRPC_ENDPOINT_CB_SHUTDOWN);
+ if (cb) {
+ cb->cb(cb->cb_arg, 0);
+ }
return;
}
@@ -265,23 +261,22 @@
gpr_log(GPR_ERROR, "WSASend overlapped error: %s", utf8_message);
gpr_free(utf8_message);
}
- status = GRPC_ENDPOINT_CB_ERROR;
+ success = 0;
} else {
- GPR_ASSERT(info->bytes_transfered == tcp->write_slices.length);
+ GPR_ASSERT(info->bytes_transfered == tcp->write_slices->length);
+ success = 1;
}
- gpr_slice_buffer_reset_and_unref(&tcp->write_slices);
tcp->socket->write_info.outstanding = 0;
tcp_unref(tcp);
- cb(opaque, status);
+ cb->cb(cb->cb_arg, success);
}
/* Initiates a write. */
-static grpc_endpoint_write_status win_write(grpc_endpoint *ep,
- gpr_slice *slices, size_t nslices,
- grpc_endpoint_write_cb cb,
- void *arg) {
+static grpc_endpoint_op_status win_write(grpc_endpoint *ep,
+ gpr_slice_buffer *slices,
+ grpc_iomgr_closure *cb) {
grpc_tcp *tcp = (grpc_tcp *)ep;
grpc_winsocket *socket = tcp->socket;
grpc_winsocket_callback_info *info = &socket->write_info;
@@ -294,28 +289,26 @@
GPR_ASSERT(!tcp->socket->write_info.outstanding);
if (tcp->shutting_down) {
- return GRPC_ENDPOINT_WRITE_ERROR;
+ return GRPC_ENDPOINT_ERROR;
}
tcp_ref(tcp);
tcp->socket->write_info.outstanding = 1;
tcp->write_cb = cb;
- tcp->write_user_data = arg;
+ tcp->write_slices = slices;
- gpr_slice_buffer_addn(&tcp->write_slices, slices, nslices);
-
- if (tcp->write_slices.count > GPR_ARRAY_SIZE(local_buffers)) {
- buffers = (WSABUF *)gpr_malloc(sizeof(WSABUF) * tcp->write_slices.count);
+ if (tcp->write_slices->count > GPR_ARRAY_SIZE(local_buffers)) {
+ buffers = (WSABUF *)gpr_malloc(sizeof(WSABUF) * tcp->write_slices->count);
allocated = buffers;
}
- for (i = 0; i < tcp->write_slices.count; i++) {
- buffers[i].len = GPR_SLICE_LENGTH(tcp->write_slices.slices[i]);
- buffers[i].buf = (char *)GPR_SLICE_START_PTR(tcp->write_slices.slices[i]);
+ for (i = 0; i < tcp->write_slices->count; i++) {
+ buffers[i].len = GPR_SLICE_LENGTH(tcp->write_slices->slices[i]);
+ buffers[i].buf = (char *)GPR_SLICE_START_PTR(tcp->write_slices->slices[i]);
}
/* First, let's try a synchronous, non-blocking write. */
- status = WSASend(socket->socket, buffers, tcp->write_slices.count,
+ status = WSASend(socket->socket, buffers, tcp->write_slices->count,
&bytes_sent, 0, NULL, NULL);
info->wsa_error = status == 0 ? 0 : WSAGetLastError();
@@ -323,10 +316,10 @@
connection that has its send queue filled up. But if we don't, then we can
avoid doing an async write operation at all. */
if (info->wsa_error != WSAEWOULDBLOCK) {
- grpc_endpoint_write_status ret = GRPC_ENDPOINT_WRITE_ERROR;
+ grpc_endpoint_op_status ret = GRPC_ENDPOINT_ERROR;
if (status == 0) {
- ret = GRPC_ENDPOINT_WRITE_DONE;
- GPR_ASSERT(bytes_sent == tcp->write_slices.length);
+ ret = GRPC_ENDPOINT_DONE;
+ GPR_ASSERT(bytes_sent == tcp->write_slices->length);
} else {
if (socket->read_info.wsa_error != WSAECONNRESET) {
char *utf8_message = gpr_format_message(info->wsa_error);
@@ -335,7 +328,6 @@
}
}
if (allocated) gpr_free(allocated);
- gpr_slice_buffer_reset_and_unref(&tcp->write_slices);
tcp->socket->write_info.outstanding = 0;
tcp_unref(tcp);
return ret;
@@ -344,24 +336,23 @@
/* 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));
- status = WSASend(socket->socket, buffers, tcp->write_slices.count,
+ status = WSASend(socket->socket, buffers, tcp->write_slices->count,
&bytes_sent, 0, &socket->write_info.overlapped, NULL);
if (allocated) gpr_free(allocated);
if (status != 0) {
int wsa_error = WSAGetLastError();
if (wsa_error != WSA_IO_PENDING) {
- gpr_slice_buffer_reset_and_unref(&tcp->write_slices);
tcp->socket->write_info.outstanding = 0;
tcp_unref(tcp);
- return GRPC_ENDPOINT_WRITE_ERROR;
+ return GRPC_ENDPOINT_ERROR;
}
}
/* As all is now setup, we can now ask for the IOCP notification. It may
trigger the callback immediately however, but no matter. */
grpc_socket_notify_on_write(socket, on_write, tcp);
- return GRPC_ENDPOINT_WRITE_PENDING;
+ return GRPC_ENDPOINT_PENDING;
}
static void win_add_to_pollset(grpc_endpoint *ep, grpc_pollset *ps) {
@@ -407,7 +398,7 @@
}
static grpc_endpoint_vtable vtable = {
- win_notify_on_read, win_write, win_add_to_pollset, win_add_to_pollset_set,
+ win_read, win_write, win_add_to_pollset, win_add_to_pollset_set,
win_shutdown, win_destroy, win_get_peer};
grpc_endpoint *grpc_tcp_create(grpc_winsocket *socket, char *peer_string) {
@@ -416,7 +407,6 @@
tcp->base.vtable = &vtable;
tcp->socket = socket;
gpr_mu_init(&tcp->mu);
- gpr_slice_buffer_init(&tcp->write_slices);
gpr_ref_init(&tcp->refcount, 1);
tcp->peer_string = gpr_strdup(peer_string);
return &tcp->base;
diff --git a/src/core/transport/chttp2/internal.h b/src/core/transport/chttp2/internal.h
index 0e7f94b..a1b773b 100644
--- a/src/core/transport/chttp2/internal.h
+++ b/src/core/transport/chttp2/internal.h
@@ -331,8 +331,6 @@
/** closure to execute writing */
grpc_iomgr_closure writing_action;
- /** closure to start reading from the endpoint */
- grpc_iomgr_closure reading_action;
/** closure to finish reading from the endpoint */
grpc_iomgr_closure recv_data;
diff --git a/src/core/transport/chttp2_transport.c b/src/core/transport/chttp2_transport.c
index 3d3d708..46ab0a5 100644
--- a/src/core/transport/chttp2_transport.c
+++ b/src/core/transport/chttp2_transport.c
@@ -84,7 +84,6 @@
/* forward declarations of various callbacks that we'll build closures around */
static void writing_action(void *t, int iomgr_success_ignored);
-static void reading_action(void *t, int iomgr_success_ignored);
/** Set a transport level setting, and push it to our peer */
static void push_setting(grpc_chttp2_transport *t, grpc_chttp2_setting_id id,
@@ -249,7 +248,6 @@
gpr_slice_buffer_init(&t->writing.outbuf);
grpc_chttp2_hpack_compressor_init(&t->writing.hpack_compressor, mdctx);
grpc_iomgr_closure_init(&t->writing_action, writing_action, t);
- grpc_iomgr_closure_init(&t->reading_action, reading_action, t);
gpr_slice_buffer_init(&t->parsing.qbuf);
grpc_chttp2_goaway_parser_init(&t->parsing.goaway_parser);
@@ -1065,10 +1063,9 @@
}
/* tcp read callback */
-static void recv_data(void *tp, int success) {
- grpc_chttp2_transport *t = tp;
+static int recv_data_loop(grpc_chttp2_transport *t, int *success) {
size_t i;
- int unref = 0;
+ int keep_reading = 0;
lock(t);
i = 0;
@@ -1077,12 +1074,12 @@
t->parsing_active = 1;
/* merge stream lists */
grpc_chttp2_stream_map_move_into(&t->new_stream_map,
- &t->parsing_stream_map);
+ &t->parsing_stream_map);
grpc_chttp2_prepare_to_read(&t->global, &t->parsing);
gpr_mu_unlock(&t->mu);
for (; i < t->read_buffer.count &&
- grpc_chttp2_perform_read(&t->parsing, t->read_buffer.slices[i]);
- i++)
+ grpc_chttp2_perform_read(&t->parsing, t->read_buffer.slices[i]);
+ i++)
;
gpr_mu_lock(&t->mu);
if (i != t->read_buffer.count) {
@@ -1090,48 +1087,53 @@
}
/* merge stream lists */
grpc_chttp2_stream_map_move_into(&t->new_stream_map,
- &t->parsing_stream_map);
+ &t->parsing_stream_map);
t->global.concurrent_stream_count =
- grpc_chttp2_stream_map_size(&t->parsing_stream_map);
+ grpc_chttp2_stream_map_size(&t->parsing_stream_map);
if (t->parsing.initial_window_update != 0) {
grpc_chttp2_stream_map_for_each(&t->parsing_stream_map,
- update_global_window, t);
+ update_global_window, t);
t->parsing.initial_window_update = 0;
}
/* handle higher level things */
grpc_chttp2_publish_reads(&t->global, &t->parsing);
t->parsing_active = 0;
}
- if (!success) {
+ if (!*success || i != t->read_buffer.count) {
drop_connection(t);
read_error_locked(t);
- unref = 1;
- } else if (i == t->read_buffer.count) {
- grpc_chttp2_schedule_closure(&t->global, &t->reading_action, 1);
- } else {
- read_error_locked(t);
- unref = 1;
+ }
+ else {
+ keep_reading = 1;
}
gpr_slice_buffer_reset_and_unref(&t->read_buffer);
unlock(t);
- if (unref) {
- UNREF_TRANSPORT(t, "recv_data");
+ if (keep_reading) {
+ switch (grpc_endpoint_read(t->ep, &t->read_buffer, &t->recv_data)) {
+ case GRPC_ENDPOINT_DONE:
+ *success = 1;
+ return 1;
+ case GRPC_ENDPOINT_ERROR:
+ *success = 0;
+ return 1;
+ case GRPC_ENDPOINT_PENDING:
+ return 0;
+ }
}
+ else {
+ UNREF_TRANSPORT(t, "recv_data");
+ return 0;
+ }
+
+ gpr_log(GPR_ERROR, "should never reach here");
+ abort();
}
-static void reading_action(void *pt, int iomgr_success_ignored) {
- grpc_chttp2_transport *t = pt;
- switch (grpc_endpoint_read(t->ep, &t->read_buffer, &t->recv_data)) {
- case GRPC_ENDPOINT_DONE:
- recv_data(t, 1);
- break;
- case GRPC_ENDPOINT_ERROR:
- recv_data(t, 0);
- break;
- case GRPC_ENDPOINT_PENDING:
- break;
- }
+static void recv_data(void *tp, int success) {
+ grpc_chttp2_transport *t = tp;
+
+ while (recv_data_loop(t, &success));
}
/*