diff --git a/src/core/channel/client_channel.c b/src/core/channel/client_channel.c index 108a6dfdf10..ec6ca428896 100644 --- a/src/core/channel/client_channel.c +++ b/src/core/channel/client_channel.c @@ -460,7 +460,7 @@ static void cc_on_config_changed(void *arg, int iomgr_success) { while (wakeup_closures) { grpc_iomgr_closure *next = wakeup_closures->next; - grpc_iomgr_add_callback(wakeup_closures); + wakeup_closures->cb(wakeup_closures->cb_arg, 1); wakeup_closures = next; } diff --git a/src/core/iomgr/iomgr.c b/src/core/iomgr/iomgr.c index a18c176b305..aa4bc6e20df 100644 --- a/src/core/iomgr/iomgr.c +++ b/src/core/iomgr/iomgr.c @@ -88,6 +88,7 @@ void grpc_kick_poller(void) { void grpc_iomgr_init(void) { gpr_thd_id id; + g_shutdown = 0; gpr_mu_init(&g_mu); gpr_cv_init(&g_rcv); grpc_alarm_list_init(gpr_now(GPR_CLOCK_MONOTONIC)); diff --git a/src/core/iomgr/tcp_posix.c b/src/core/iomgr/tcp_posix.c index 1e8432d463a..63a8a2720e6 100644 --- a/src/core/iomgr/tcp_posix.c +++ b/src/core/iomgr/tcp_posix.c @@ -319,7 +319,7 @@ static void call_read_cb(grpc_tcp *tcp, gpr_slice *slices, size_t nslices, gpr_log(GPR_DEBUG, "read: status=%d", status); for (i = 0; i < nslices; i++) { char *dump = gpr_dump_slice(slices[i], GPR_DUMP_HEX | GPR_DUMP_ASCII); - gpr_log(GPR_DEBUG, "READ: %s", dump); + gpr_log(GPR_DEBUG, "READ %p: %s", tcp, dump); gpr_free(dump); } } @@ -448,7 +448,7 @@ static void grpc_tcp_notify_on_read(grpc_endpoint *ep, grpc_endpoint_read_cb cb, grpc_fd_notify_on_read(tcp->em_fd, &tcp->read_closure); } else { tcp->handle_read_closure.cb_arg = tcp; - grpc_iomgr_add_callback(&tcp->handle_read_closure); + grpc_iomgr_add_delayed_callback(&tcp->handle_read_closure, 1); } } diff --git a/src/core/transport/chttp2/internal.h b/src/core/transport/chttp2/internal.h index e7901da510e..f0eeb6de505 100644 --- a/src/core/transport/chttp2/internal.h +++ b/src/core/transport/chttp2/internal.h @@ -60,7 +60,6 @@ typedef enum { GRPC_CHTTP2_LIST_WRITABLE, GRPC_CHTTP2_LIST_WRITING, GRPC_CHTTP2_LIST_WRITTEN, - GRPC_CHTTP2_LIST_WRITABLE_WINDOW_UPDATE, GRPC_CHTTP2_LIST_PARSING_SEEN, GRPC_CHTTP2_LIST_CLOSED_WAITING_FOR_PARSING, GRPC_CHTTP2_LIST_CANCELLED_WAITING_FOR_WRITING, @@ -383,6 +382,8 @@ typedef struct { gpr_uint8 published_cancelled; /** is this stream in the stream map? (boolean) */ gpr_uint8 in_stream_map; + /** is this stream actively being written? */ + gpr_uint8 writing_now; /** stream state already published to the upper layer */ grpc_stream_state published_state; @@ -475,11 +476,17 @@ void grpc_chttp2_publish_reads(grpc_chttp2_transport_global *global, void grpc_chttp2_list_add_writable_stream( grpc_chttp2_transport_global *transport_global, grpc_chttp2_stream_global *stream_global); +void grpc_chttp2_list_add_first_writable_stream( + grpc_chttp2_transport_global *transport_global, + grpc_chttp2_stream_global *stream_global); int grpc_chttp2_list_pop_writable_stream( grpc_chttp2_transport_global *transport_global, grpc_chttp2_transport_writing *transport_writing, grpc_chttp2_stream_global **stream_global, grpc_chttp2_stream_writing **stream_writing); +void grpc_chttp2_list_remove_writable_stream( + grpc_chttp2_transport_global *transport_global, + grpc_chttp2_stream_global *stream_global); void grpc_chttp2_list_add_incoming_window_updated( grpc_chttp2_transport_global *transport_global, @@ -511,18 +518,6 @@ int grpc_chttp2_list_pop_written_stream( grpc_chttp2_stream_global **stream_global, grpc_chttp2_stream_writing **stream_writing); -void grpc_chttp2_list_add_writable_window_update_stream( - grpc_chttp2_transport_global *transport_global, - grpc_chttp2_stream_global *stream_global); -int grpc_chttp2_list_pop_writable_window_update_stream( - grpc_chttp2_transport_global *transport_global, - grpc_chttp2_transport_writing *transport_writing, - grpc_chttp2_stream_global **stream_global, - grpc_chttp2_stream_writing **stream_writing); -void grpc_chttp2_list_remove_writable_window_update_stream( - grpc_chttp2_transport_global *transport_global, - grpc_chttp2_stream_global *stream_global); - void grpc_chttp2_list_add_parsing_seen_stream( grpc_chttp2_transport_parsing *transport_parsing, grpc_chttp2_stream_parsing *stream_parsing); diff --git a/src/core/transport/chttp2/parsing.c b/src/core/transport/chttp2/parsing.c index 50a2f752f62..d84960009bd 100644 --- a/src/core/transport/chttp2/parsing.c +++ b/src/core/transport/chttp2/parsing.c @@ -182,8 +182,7 @@ void grpc_chttp2_publish_reads( stream_global->max_recv_bytes -= stream_parsing->incoming_window_delta; stream_parsing->incoming_window_delta = 0; - grpc_chttp2_list_add_writable_window_update_stream(transport_global, - stream_global); + grpc_chttp2_list_add_writable_stream(transport_global, stream_global); } /* update outgoing flow control window */ diff --git a/src/core/transport/chttp2/stream_lists.c b/src/core/transport/chttp2/stream_lists.c index 590f6abfbc3..9e68c1e146f 100644 --- a/src/core/transport/chttp2/stream_lists.c +++ b/src/core/transport/chttp2/stream_lists.c @@ -108,6 +108,23 @@ static void stream_list_maybe_remove(grpc_chttp2_transport *t, } } +static void stream_list_add_head(grpc_chttp2_transport *t, + grpc_chttp2_stream *s, + grpc_chttp2_stream_list_id id) { + grpc_chttp2_stream *old_head; + GPR_ASSERT(!s->included[id]); + old_head = t->lists[id].head; + s->links[id].next = old_head; + s->links[id].prev = NULL; + if (old_head) { + old_head->links[id].prev = s; + } else { + t->lists[id].tail = s; + } + t->lists[id].head = s; + s->included[id] = 1; +} + static void stream_list_add_tail(grpc_chttp2_transport *t, grpc_chttp2_stream *s, grpc_chttp2_stream_list_id id) { @@ -119,7 +136,6 @@ static void stream_list_add_tail(grpc_chttp2_transport *t, if (old_tail) { old_tail->links[id].next = s; } else { - s->links[id].prev = NULL; t->lists[id].head = s; } t->lists[id].tail = s; @@ -144,6 +160,18 @@ void grpc_chttp2_list_add_writable_stream( STREAM_FROM_GLOBAL(stream_global), GRPC_CHTTP2_LIST_WRITABLE); } +void grpc_chttp2_list_add_first_writable_stream( + grpc_chttp2_transport_global *transport_global, + grpc_chttp2_stream_global *stream_global) { + GPR_ASSERT(stream_global->id != 0); + gpr_log(GPR_DEBUG, "add:%d:%d:%d:%d", stream_global->id, + stream_global->write_state, stream_global->in_stream_map, + stream_global->read_closed); + stream_list_add_head(TRANSPORT_FROM_GLOBAL(transport_global), + STREAM_FROM_GLOBAL(stream_global), + GRPC_CHTTP2_LIST_WRITABLE); +} + int grpc_chttp2_list_pop_writable_stream( grpc_chttp2_transport_global *transport_global, grpc_chttp2_transport_writing *transport_writing, @@ -157,6 +185,14 @@ int grpc_chttp2_list_pop_writable_stream( return r; } +void grpc_chttp2_list_remove_writable_stream( + grpc_chttp2_transport_global *transport_global, + grpc_chttp2_stream_global *stream_global) { + stream_list_maybe_remove(TRANSPORT_FROM_GLOBAL(transport_global), + STREAM_FROM_GLOBAL(stream_global), + GRPC_CHTTP2_LIST_WRITABLE); +} + void grpc_chttp2_list_add_writing_stream( grpc_chttp2_transport_writing *transport_writing, grpc_chttp2_stream_writing *stream_writing) { @@ -202,36 +238,6 @@ int grpc_chttp2_list_pop_written_stream( return r; } -void grpc_chttp2_list_add_writable_window_update_stream( - grpc_chttp2_transport_global *transport_global, - grpc_chttp2_stream_global *stream_global) { - GPR_ASSERT(stream_global->id != 0); - stream_list_add(TRANSPORT_FROM_GLOBAL(transport_global), - STREAM_FROM_GLOBAL(stream_global), - GRPC_CHTTP2_LIST_WRITABLE_WINDOW_UPDATE); -} - -int grpc_chttp2_list_pop_writable_window_update_stream( - grpc_chttp2_transport_global *transport_global, - grpc_chttp2_transport_writing *transport_writing, - grpc_chttp2_stream_global **stream_global, - grpc_chttp2_stream_writing **stream_writing) { - grpc_chttp2_stream *stream; - int r = stream_list_pop(TRANSPORT_FROM_GLOBAL(transport_global), &stream, - GRPC_CHTTP2_LIST_WRITABLE_WINDOW_UPDATE); - *stream_global = &stream->global; - *stream_writing = &stream->writing; - return r; -} - -void grpc_chttp2_list_remove_writable_window_update_stream( - grpc_chttp2_transport_global *transport_global, - grpc_chttp2_stream_global *stream_global) { - stream_list_maybe_remove(TRANSPORT_FROM_GLOBAL(transport_global), - STREAM_FROM_GLOBAL(stream_global), - GRPC_CHTTP2_LIST_WRITABLE_WINDOW_UPDATE); -} - void grpc_chttp2_list_add_parsing_seen_stream( grpc_chttp2_transport_parsing *transport_parsing, grpc_chttp2_stream_parsing *stream_parsing) { diff --git a/src/core/transport/chttp2/writing.c b/src/core/transport/chttp2/writing.c index d8ec117aa5d..d39b0c42f7f 100644 --- a/src/core/transport/chttp2/writing.c +++ b/src/core/transport/chttp2/writing.c @@ -44,6 +44,7 @@ int grpc_chttp2_unlocking_check_writes( grpc_chttp2_transport_writing *transport_writing) { grpc_chttp2_stream_global *stream_global; grpc_chttp2_stream_writing *stream_writing; + grpc_chttp2_stream_global *first_reinserted_stream = NULL; gpr_uint32 window_delta; /* simple writes are queued to qbuf, and flushed here */ @@ -64,50 +65,54 @@ int grpc_chttp2_unlocking_check_writes( } /* for each grpc_chttp2_stream that's become writable, frame it's data - (according to - available window sizes) and add to the output buffer */ - while (grpc_chttp2_list_pop_writable_stream(transport_global, - transport_writing, &stream_global, - &stream_writing)) { + (according to available window sizes) and add to the output buffer */ + while (grpc_chttp2_list_pop_writable_stream( + transport_global, transport_writing, &stream_global, &stream_writing)) { + if (stream_global == first_reinserted_stream) { + /* prevent infinite loop */ + grpc_chttp2_list_add_first_writable_stream(transport_global, + stream_global); + break; + } + stream_writing->id = stream_global->id; - window_delta = grpc_chttp2_preencode( - stream_global->outgoing_sopb->ops, &stream_global->outgoing_sopb->nops, - GPR_MIN(transport_global->outgoing_window, - stream_global->outgoing_window), - &stream_writing->sopb); - GRPC_CHTTP2_FLOWCTL_TRACE_TRANSPORT( - "write", transport_global, outgoing_window, -(gpr_int64)window_delta); - GRPC_CHTTP2_FLOWCTL_TRACE_STREAM("write", transport_global, stream_global, - outgoing_window, -(gpr_int64)window_delta); - transport_global->outgoing_window -= window_delta; - stream_global->outgoing_window -= window_delta; - - if (stream_global->write_state == GRPC_WRITE_STATE_QUEUED_CLOSE && - stream_global->outgoing_sopb->nops == 0) { - if (!transport_global->is_client && !stream_global->read_closed) { - stream_writing->send_closed = GRPC_SEND_CLOSED_WITH_RST_STREAM; - } else { - stream_writing->send_closed = GRPC_SEND_CLOSED; + stream_writing->send_closed = GRPC_DONT_SEND_CLOSED; + GPR_ASSERT(!stream_global->writing_now); + + if (stream_global->outgoing_sopb) { + window_delta = + grpc_chttp2_preencode(stream_global->outgoing_sopb->ops, + &stream_global->outgoing_sopb->nops, + GPR_MIN(transport_global->outgoing_window, + stream_global->outgoing_window), + &stream_writing->sopb); + GRPC_CHTTP2_FLOWCTL_TRACE_TRANSPORT( + "write", transport_global, outgoing_window, -(gpr_int64)window_delta); + GRPC_CHTTP2_FLOWCTL_TRACE_STREAM("write", transport_global, stream_global, + outgoing_window, + -(gpr_int64)window_delta); + transport_global->outgoing_window -= window_delta; + stream_global->outgoing_window -= window_delta; + + if (stream_global->write_state == GRPC_WRITE_STATE_QUEUED_CLOSE && + stream_global->outgoing_sopb->nops == 0) { + if (!transport_global->is_client && !stream_global->read_closed) { + stream_writing->send_closed = GRPC_SEND_CLOSED_WITH_RST_STREAM; + } else { + stream_writing->send_closed = GRPC_SEND_CLOSED; + } } - } - if (stream_writing->sopb.nops > 0 || - stream_writing->send_closed != GRPC_DONT_SEND_CLOSED) { - grpc_chttp2_list_add_writing_stream(transport_writing, stream_writing); - } - if (stream_global->outgoing_window > 0 && - stream_global->outgoing_sopb->nops != 0) { - grpc_chttp2_list_add_writable_stream(transport_global, stream_global); + if (stream_global->outgoing_window > 0 && + stream_global->outgoing_sopb->nops != 0) { + grpc_chttp2_list_add_writable_stream(transport_global, stream_global); + if (first_reinserted_stream == NULL && + transport_global->outgoing_window == 0) { + first_reinserted_stream = stream_global; + } + } } - } - /* for each grpc_chttp2_stream that wants to update its window, add that - * window here */ - while (grpc_chttp2_list_pop_writable_window_update_stream(transport_global, - transport_writing, - &stream_global, - &stream_writing)) { - stream_writing->id = stream_global->id; if (!stream_global->read_closed && stream_global->unannounced_incoming_window > 0) { stream_writing->announce_window = stream_global->unannounced_incoming_window; GRPC_CHTTP2_FLOWCTL_TRACE_STREAM("write", transport_global, stream_global, @@ -118,6 +123,11 @@ int grpc_chttp2_unlocking_check_writes( stream_global->unannounced_incoming_window = 0; grpc_chttp2_list_add_incoming_window_updated(transport_global, stream_global); + stream_global->writing_now = 1; + grpc_chttp2_list_add_writing_stream(transport_writing, stream_writing); + } else if (stream_writing->sopb.nops > 0 || + stream_writing->send_closed != GRPC_DONT_SEND_CLOSED) { + stream_global->writing_now = 1; grpc_chttp2_list_add_writing_stream(transport_writing, stream_writing); } } @@ -205,6 +215,8 @@ void grpc_chttp2_cleanup_writing( while (grpc_chttp2_list_pop_written_stream( transport_global, transport_writing, &stream_global, &stream_writing)) { + GPR_ASSERT(stream_global->writing_now); + stream_global->writing_now = 0; if (stream_global->outgoing_sopb != NULL && stream_global->outgoing_sopb->nops == 0) { stream_global->outgoing_sopb = NULL; @@ -216,9 +228,9 @@ void grpc_chttp2_cleanup_writing( if (!transport_global->is_client) { stream_global->read_closed = 1; } - grpc_chttp2_list_add_read_write_state_changed(transport_global, - stream_global); } + grpc_chttp2_list_add_read_write_state_changed(transport_global, + stream_global); } transport_writing->outbuf.count = 0; transport_writing->outbuf.length = 0; diff --git a/src/core/transport/chttp2_transport.c b/src/core/transport/chttp2_transport.c index eb435a2ee8f..5f49b2ddd60 100644 --- a/src/core/transport/chttp2_transport.c +++ b/src/core/transport/chttp2_transport.c @@ -395,12 +395,16 @@ static void destroy_stream(grpc_transport *gt, grpc_stream *gs) { } grpc_chttp2_list_remove_incoming_window_updated(&t->global, &s->global); - grpc_chttp2_list_remove_writable_window_update_stream(&t->global, &s->global); + grpc_chttp2_list_remove_writable_stream(&t->global, &s->global); gpr_mu_unlock(&t->mu); for (i = 0; i < STREAM_LIST_COUNT; i++) { - GPR_ASSERT(!s->included[i]); + if (s->included[i]) { + gpr_log(GPR_ERROR, "%s stream %d still included in list %d", + t->global.is_client ? "client" : "server", s->global.id, i); + abort(); + } } GPR_ASSERT(s->global.outgoing_sopb == NULL); @@ -576,8 +580,6 @@ static void maybe_start_some_streams( grpc_chttp2_list_add_incoming_window_updated(transport_global, stream_global); grpc_chttp2_list_add_writable_stream(transport_global, stream_global); - grpc_chttp2_list_add_writable_window_update_stream(transport_global, - stream_global); } /* cancel out streams that will never be started */ @@ -643,8 +645,7 @@ static void perform_stream_op_locked( if (stream_global->id != 0) { grpc_chttp2_list_add_read_write_state_changed(transport_global, stream_global); - grpc_chttp2_list_add_writable_window_update_stream(transport_global, - stream_global); + grpc_chttp2_list_add_writable_stream(transport_global, stream_global); } } @@ -752,6 +753,7 @@ static void remove_stream(grpc_chttp2_transport *t, gpr_uint32 id) { if (!s) { s = grpc_chttp2_stream_map_delete(&t->new_stream_map, id); } + grpc_chttp2_list_remove_writable_stream(&t->global, &s->global); GPR_ASSERT(s); s->global.in_stream_map = 0; if (t->parsing.incoming_stream == &s->parsing) { @@ -833,6 +835,9 @@ static void unlock_check_read_write_state(grpc_chttp2_transport *t) { if (!stream_global->publish_sopb) { continue; } + if (stream_global->writing_now) { + continue; + } /* FIXME(ctiller): we include in_stream_map in our computation of whether the stream is write-closed. This is completely bogus, but has the effect of delaying stream-closed until the stream diff --git a/src/core/transport/transport_op_string.c b/src/core/transport/transport_op_string.c index 10d796fc158..f62c340e97f 100644 --- a/src/core/transport/transport_op_string.c +++ b/src/core/transport/transport_op_string.c @@ -116,10 +116,9 @@ char *grpc_transport_stream_op_string(grpc_transport_stream_op *op) { if (op->send_ops) { if (!first) gpr_strvec_add(&b, gpr_strdup(" ")); first = 0; - gpr_strvec_add(&b, gpr_strdup("SEND")); - if (op->is_last_send) { - gpr_strvec_add(&b, gpr_strdup("_LAST")); - } + gpr_asprintf(&tmp, "SEND%s:%p", op->is_last_send ? "_LAST" : "", + op->on_done_send); + gpr_strvec_add(&b, tmp); gpr_strvec_add(&b, gpr_strdup("[")); gpr_strvec_add(&b, grpc_sopb_string(op->send_ops)); gpr_strvec_add(&b, gpr_strdup("]")); @@ -128,7 +127,8 @@ char *grpc_transport_stream_op_string(grpc_transport_stream_op *op) { if (op->recv_ops) { if (!first) gpr_strvec_add(&b, gpr_strdup(" ")); first = 0; - gpr_asprintf(&tmp, "RECV:max_recv_bytes=%d", op->max_recv_bytes); + gpr_asprintf(&tmp, "RECV:%p:max_recv_bytes=%d", op->on_done_recv, + op->max_recv_bytes); gpr_strvec_add(&b, tmp); } diff --git a/test/cpp/end2end/end2end_test.cc b/test/cpp/end2end/end2end_test.cc index 20e4c4ed55b..c433b789482 100644 --- a/test/cpp/end2end/end2end_test.cc +++ b/test/cpp/end2end/end2end_test.cc @@ -144,6 +144,11 @@ class TestServiceImpl : public ::grpc::cpp::test::util::TestService::Service { if (request->has_param() && request->param().check_auth_context()) { CheckAuthContext(context); } + if (request->has_param() && + request->param().response_message_length() > 0) { + response->set_message( + grpc::string(request->param().response_message_length(), '\0')); + } return Status::OK; } @@ -786,6 +791,21 @@ TEST_F(End2endTest, ClientAuthContext) { CheckAuthContext(&context); } +// Make the response larger than the flow control window. +TEST_F(End2endTest, HugeResponse) { + ResetStub(); + EchoRequest request; + EchoResponse response; + request.set_message("huge response"); + const int kResponseSize = 1024 * (1024 + 10); + request.mutable_param()->set_response_message_length(kResponseSize); + + ClientContext context; + Status s = stub_->Echo(&context, request, &response); + EXPECT_EQ(kResponseSize, response.message().size()); + EXPECT_TRUE(s.ok()); +} + } // namespace testing } // namespace grpc diff --git a/test/cpp/util/messages.proto b/test/cpp/util/messages.proto index 3708972b905..2fad8b42a21 100644 --- a/test/cpp/util/messages.proto +++ b/test/cpp/util/messages.proto @@ -38,6 +38,7 @@ message RequestParams { optional int32 server_cancel_after_us = 3; optional bool echo_metadata = 4; optional bool check_auth_context = 5; + optional int32 response_message_length = 6; } message EchoRequest {