From 9565fd717586d01b092dc6eb70dd9485d3c633a9 Mon Sep 17 00:00:00 2001 From: Craig Tiller Date: Wed, 18 May 2016 23:17:09 -0700 Subject: Fix compile --- src/core/ext/transport/chttp2/server/secure/server_secure_chttp2.c | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) (limited to 'src/core/ext') diff --git a/src/core/ext/transport/chttp2/server/secure/server_secure_chttp2.c b/src/core/ext/transport/chttp2/server/secure/server_secure_chttp2.c index 9898bf05ae..3286b220d4 100644 --- a/src/core/ext/transport/chttp2/server/secure/server_secure_chttp2.c +++ b/src/core/ext/transport/chttp2/server/secure/server_secure_chttp2.c @@ -144,7 +144,7 @@ static void start(grpc_exec_ctx *exec_ctx, grpc_server *server, void *statep, static void destroy_done(grpc_exec_ctx *exec_ctx, void *statep, grpc_error *error) { - grpc_server_secure_state *state = statep; + server_secure_state *state = statep; if (state->destroy_callback != NULL) { state->destroy_callback->cb(exec_ctx, state->destroy_callback->cb_arg, GRPC_ERROR_REF(error)); -- cgit v1.2.3 From 13d455e21b7cd0085c657380e527da468d9bde52 Mon Sep 17 00:00:00 2001 From: Craig Tiller Date: Fri, 27 May 2016 12:06:34 -0700 Subject: Initial pass of delayed writing --- .../transport/chttp2/transport/chttp2_transport.c | 144 +++++++++++++++------ src/core/ext/transport/chttp2/transport/internal.h | 31 +++-- src/core/ext/transport/chttp2/transport/parsing.c | 5 +- .../ext/transport/chttp2/transport/stream_lists.c | 3 +- src/core/lib/iomgr/workqueue.h | 4 + src/core/lib/iomgr/workqueue_posix.c | 6 + 6 files changed, 137 insertions(+), 56 deletions(-) (limited to 'src/core/ext') diff --git a/src/core/ext/transport/chttp2/transport/chttp2_transport.c b/src/core/ext/transport/chttp2/transport/chttp2_transport.c index be517154e6..b94a112b4a 100644 --- a/src/core/ext/transport/chttp2/transport/chttp2_transport.c +++ b/src/core/ext/transport/chttp2/transport/chttp2_transport.c @@ -47,6 +47,7 @@ #include "src/core/ext/transport/chttp2/transport/internal.h" #include "src/core/ext/transport/chttp2/transport/status_conversion.h" #include "src/core/ext/transport/chttp2/transport/timeout_encoding.h" +#include "src/core/lib/iomgr/workqueue.h" #include "src/core/lib/profiling/timers.h" #include "src/core/lib/support/string.h" #include "src/core/lib/transport/static_metadata.h" @@ -87,10 +88,12 @@ static const grpc_transport_vtable vtable; static void writing_action(grpc_exec_ctx *exec_ctx, void *t, grpc_error *error); static void reading_action(grpc_exec_ctx *exec_ctx, void *t, grpc_error *error); static void parsing_action(grpc_exec_ctx *exec_ctx, void *t, grpc_error *error); +static void initiate_writing(grpc_exec_ctx *exec_ctx, void *t, + grpc_error *error); /** Set a transport level setting, and push it to our peer */ -static void push_setting(grpc_chttp2_transport *t, grpc_chttp2_setting_id id, - uint32_t value); +static void push_setting(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t, + grpc_chttp2_setting_id id, uint32_t value); /** Start disconnection chain */ static void drop_connection(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t, @@ -137,7 +140,7 @@ static void check_read_ops(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport_global *transport_global); static void incoming_byte_stream_update_flow_control( - grpc_chttp2_transport_global *transport_global, + grpc_exec_ctx *exec_ctx, grpc_chttp2_transport_global *transport_global, grpc_chttp2_stream_global *stream_global, size_t max_size_hint, size_t have_already); static void incoming_byte_stream_destroy_locked(grpc_exec_ctx *exec_ctx, @@ -231,7 +234,7 @@ static void ref_transport(grpc_chttp2_transport *t) { gpr_ref(&t->refs); } static void init_transport(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t, const grpc_channel_args *channel_args, - grpc_endpoint *ep, uint8_t is_client) { + grpc_endpoint *ep, bool is_client) { size_t i; int j; @@ -247,6 +250,9 @@ static void init_transport(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t, /* ref is dropped at transport close() */ gpr_ref_init(&t->shutdown_ep_refs, 1); gpr_mu_init(&t->executor.mu); + GPR_ASSERT(GRPC_LOG_IF_ERROR( + "workqueue_create", + grpc_workqueue_create(exec_ctx, &t->executor.workqueue))); t->peer_string = grpc_endpoint_get_peer(ep); t->endpoint_reading = 1; t->global.next_stream_id = is_client ? 1 : 2; @@ -272,6 +278,7 @@ static void init_transport(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t, grpc_closure_init(&t->writing_action, writing_action, t); grpc_closure_init(&t->reading_action, reading_action, t); grpc_closure_init(&t->parsing_action, parsing_action, t); + grpc_closure_init(&t->initiate_writing, initiate_writing, t); gpr_slice_buffer_init(&t->parsing.qbuf); grpc_chttp2_goaway_parser_init(&t->parsing.goaway_parser); @@ -285,6 +292,7 @@ static void init_transport(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t, gpr_slice_buffer_add( &t->global.qbuf, gpr_slice_from_copied_string(GRPC_CHTTP2_CLIENT_CONNECT_STRING)); + grpc_chttp2_initiate_write(exec_ctx, &t->global); } /* 8 is a random stab in the dark as to a good initial size: it's small enough that it shouldn't waste memory for infrequently used connections, yet @@ -310,11 +318,12 @@ static void init_transport(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t, /* configure http2 the way we like it */ if (is_client) { - push_setting(t, GRPC_CHTTP2_SETTINGS_ENABLE_PUSH, 0); - push_setting(t, GRPC_CHTTP2_SETTINGS_MAX_CONCURRENT_STREAMS, 0); + push_setting(exec_ctx, t, GRPC_CHTTP2_SETTINGS_ENABLE_PUSH, 0); + push_setting(exec_ctx, t, GRPC_CHTTP2_SETTINGS_MAX_CONCURRENT_STREAMS, 0); } - push_setting(t, GRPC_CHTTP2_SETTINGS_INITIAL_WINDOW_SIZE, DEFAULT_WINDOW); - push_setting(t, GRPC_CHTTP2_SETTINGS_MAX_HEADER_LIST_SIZE, + push_setting(exec_ctx, t, GRPC_CHTTP2_SETTINGS_INITIAL_WINDOW_SIZE, + DEFAULT_WINDOW); + push_setting(exec_ctx, t, GRPC_CHTTP2_SETTINGS_MAX_HEADER_LIST_SIZE, DEFAULT_MAX_HEADER_LIST_SIZE); if (channel_args) { @@ -328,7 +337,7 @@ static void init_transport(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t, gpr_log(GPR_ERROR, "%s: must be an integer", GRPC_ARG_MAX_CONCURRENT_STREAMS); } else { - push_setting(t, GRPC_CHTTP2_SETTINGS_MAX_CONCURRENT_STREAMS, + push_setting(exec_ctx, t, GRPC_CHTTP2_SETTINGS_MAX_CONCURRENT_STREAMS, (uint32_t)channel_args->args[i].value.integer); } } else if (0 == strcmp(channel_args->args[i].key, @@ -367,7 +376,7 @@ static void init_transport(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t, gpr_log(GPR_ERROR, "%s: must be non-negative", GRPC_ARG_HTTP2_HPACK_TABLE_SIZE_DECODER); } else { - push_setting(t, GRPC_CHTTP2_SETTINGS_HEADER_TABLE_SIZE, + push_setting(exec_ctx, t, GRPC_CHTTP2_SETTINGS_HEADER_TABLE_SIZE, (uint32_t)channel_args->args[i].value.integer); } } else if (0 == strcmp(channel_args->args[i].key, @@ -392,7 +401,7 @@ static void init_transport(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t, gpr_log(GPR_ERROR, "%s: must be non-negative", GRPC_ARG_MAX_METADATA_SIZE); } else { - push_setting(t, GRPC_CHTTP2_SETTINGS_MAX_HEADER_LIST_SIZE, + push_setting(exec_ctx, t, GRPC_CHTTP2_SETTINGS_MAX_HEADER_LIST_SIZE, (uint32_t)channel_args->args[i].value.integer); } } @@ -637,14 +646,6 @@ static void finish_global_actions(grpc_exec_ctx *exec_ctx, grpc_chttp2_executor_action_header *next; for (;;) { - if (!t->executor.writing_active && !t->closed && - grpc_chttp2_unlocking_check_writes(exec_ctx, &t->global, &t->writing, - t->executor.parsing_active)) { - t->executor.writing_active = 1; - REF_TRANSPORT(t, "writing"); - prevent_endpoint_shutdown(t); - grpc_exec_ctx_sched(exec_ctx, &t->writing_action, GRPC_ERROR_NONE, NULL); - } check_read_ops(exec_ctx, &t->global); gpr_mu_lock(&t->executor.mu); @@ -727,16 +728,52 @@ void grpc_chttp2_run_with_global_lock(grpc_exec_ctx *exec_ctx, * OUTPUT PROCESSING */ -void grpc_chttp2_become_writable(grpc_chttp2_transport_global *transport_global, +void grpc_chttp2_initiate_write( + grpc_exec_ctx *exec_ctx, grpc_chttp2_transport_global *transport_global) { + grpc_chttp2_transport *t = TRANSPORT_FROM_GLOBAL(transport_global); + t->executor.writing_needed = true; + if (!t->executor.writing_initiated) { + t->executor.writing_initiated = true; + grpc_workqueue_enqueue(exec_ctx, t->executor.workqueue, + &t->initiate_writing, GRPC_ERROR_NONE); + } +} + +static void initiate_writing_locked(grpc_exec_ctx *exec_ctx, + grpc_chttp2_transport *t, + grpc_chttp2_stream *s_unused, + void *arg_ignored) { + GPR_ASSERT(t->executor.writing_needed); + GPR_ASSERT(!t->executor.writing_active); + if (!t->closed && + grpc_chttp2_unlocking_check_writes(exec_ctx, &t->global, &t->writing, + t->executor.parsing_active)) { + t->executor.writing_needed = false; + t->executor.writing_active = true; + REF_TRANSPORT(t, "writing"); + prevent_endpoint_shutdown(t); + grpc_exec_ctx_sched(exec_ctx, &t->writing_action, GRPC_ERROR_NONE, NULL); + } +} + +static void initiate_writing(grpc_exec_ctx *exec_ctx, void *arg, + grpc_error *error) { + grpc_chttp2_run_with_global_lock(exec_ctx, arg, NULL, initiate_writing_locked, + NULL, 0); +} + +void grpc_chttp2_become_writable(grpc_exec_ctx *exec_ctx, + grpc_chttp2_transport_global *transport_global, grpc_chttp2_stream_global *stream_global) { if (!TRANSPORT_FROM_GLOBAL(transport_global)->closed && grpc_chttp2_list_add_writable_stream(transport_global, stream_global)) { GRPC_CHTTP2_STREAM_REF(stream_global, "chttp2_writing"); + grpc_chttp2_initiate_write(exec_ctx, transport_global); } } -static void push_setting(grpc_chttp2_transport *t, grpc_chttp2_setting_id id, - uint32_t value) { +static void push_setting(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t, + grpc_chttp2_setting_id id, uint32_t value) { const grpc_chttp2_setting_parameters *sp = &grpc_chttp2_settings_parameters[id]; uint32_t use_value = GPR_CLAMP(value, sp->min_value, sp->max_value); @@ -747,6 +784,7 @@ static void push_setting(grpc_chttp2_transport *t, grpc_chttp2_setting_id id, if (use_value != t->global.settings[GRPC_LOCAL_SETTINGS][id]) { t->global.settings[GRPC_LOCAL_SETTINGS][id] = use_value; t->global.dirtied_local_settings = 1; + grpc_chttp2_initiate_write(exec_ctx, &t->global); } } @@ -772,10 +810,12 @@ static void terminate_writing_with_lock(grpc_exec_ctx *exec_ctx, GRPC_CHTTP2_STREAM_UNREF(exec_ctx, stream_global, "finish_writes"); } - /* leave the writing flag up on shutdown to prevent further writes in - unlock() - from starting */ - t->executor.writing_active = 0; + t->executor.writing_active = false; + if (t->executor.writing_needed) { + initiate_writing_locked(exec_ctx, t, NULL, NULL); + } else { + t->executor.writing_initiated = false; + } if (t->ep && !t->endpoint_reading) { destroy_endpoint(exec_ctx, t); } @@ -862,7 +902,7 @@ static void maybe_start_some_streams( stream_global->id, STREAM_FROM_GLOBAL(stream_global)); stream_global->in_stream_map = true; transport_global->concurrent_stream_count++; - grpc_chttp2_become_writable(transport_global, stream_global); + grpc_chttp2_become_writable(exec_ctx, transport_global, stream_global); } /* cancel out streams that will never be started */ while (transport_global->next_stream_id >= MAX_CLIENT_STREAM_ID && @@ -991,7 +1031,8 @@ static void perform_stream_op_locked(grpc_exec_ctx *exec_ctx, maybe_start_some_streams(exec_ctx, transport_global); } else { GPR_ASSERT(stream_global->id != 0); - grpc_chttp2_become_writable(transport_global, stream_global); + grpc_chttp2_become_writable(exec_ctx, transport_global, + stream_global); } } else { grpc_chttp2_complete_closure_step( @@ -1015,7 +1056,7 @@ static void perform_stream_op_locked(grpc_exec_ctx *exec_ctx, } else { stream_global->send_message = op->send_message; if (stream_global->id != 0) { - grpc_chttp2_become_writable(transport_global, stream_global); + grpc_chttp2_become_writable(exec_ctx, transport_global, stream_global); } } } @@ -1054,7 +1095,7 @@ static void perform_stream_op_locked(grpc_exec_ctx *exec_ctx, } else if (stream_global->id != 0) { /* TODO(ctiller): check if there's flow control for any outstanding bytes before going writable */ - grpc_chttp2_become_writable(transport_global, stream_global); + grpc_chttp2_become_writable(exec_ctx, transport_global, stream_global); } } } @@ -1075,8 +1116,8 @@ static void perform_stream_op_locked(grpc_exec_ctx *exec_ctx, (stream_global->incoming_frames.head == NULL || stream_global->incoming_frames.head->is_tail)) { incoming_byte_stream_update_flow_control( - transport_global, stream_global, transport_global->stream_lookahead, - 0); + exec_ctx, transport_global, stream_global, + transport_global->stream_lookahead, 0); } grpc_chttp2_list_add_check_read_ops(transport_global, stream_global); } @@ -1103,7 +1144,8 @@ static void perform_stream_op(grpc_exec_ctx *exec_ctx, grpc_transport *gt, sizeof(*op)); } -static void send_ping_locked(grpc_chttp2_transport *t, grpc_closure *on_recv) { +static void send_ping_locked(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t, + grpc_closure *on_recv) { grpc_chttp2_outstanding_ping *p = gpr_malloc(sizeof(*p)); p->next = &t->global.pings; p->prev = p->next->prev; @@ -1118,6 +1160,7 @@ static void send_ping_locked(grpc_chttp2_transport *t, grpc_closure *on_recv) { p->id[7] = (uint8_t)(t->global.ping_counter & 0xff); p->on_recv = on_recv; gpr_slice_buffer_add(&t->global.qbuf, grpc_chttp2_ping_create(0, p->id)); + grpc_chttp2_initiate_write(exec_ctx, &t->global); } static void ack_ping_locked(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t, @@ -1177,6 +1220,7 @@ static void perform_transport_op_locked(grpc_exec_ctx *exec_ctx, close_transport = grpc_chttp2_has_streams(t) ? GRPC_ERROR_NONE : GRPC_ERROR_CREATE("GOAWAY sent"); + grpc_chttp2_initiate_write(exec_ctx, &t->global); } if (op->set_accept_stream) { @@ -1194,7 +1238,7 @@ static void perform_transport_op_locked(grpc_exec_ctx *exec_ctx, } if (op->send_ping) { - send_ping_locked(t, op->send_ping); + send_ping_locked(exec_ctx, t, op->send_ping); } if (close_transport != GRPC_ERROR_NONE) { @@ -1343,6 +1387,7 @@ static void cancel_from_api(grpc_exec_ctx *exec_ctx, stream_global->id, (uint32_t)grpc_chttp2_grpc_status_to_http2_error(status), &stream_global->stats.outgoing)); + grpc_chttp2_initiate_write(exec_ctx, transport_global); } grpc_chttp2_fake_status(exec_ctx, transport_global, stream_global, status, NULL); @@ -1564,6 +1609,7 @@ static void close_from_api(grpc_exec_ctx *exec_ctx, } grpc_chttp2_mark_stream_closed(exec_ctx, transport_global, stream_global, 1, 1, err); + grpc_chttp2_initiate_write(exec_ctx, transport_global); } static void cancel_stream_cb(grpc_chttp2_transport_global *transport_global, @@ -1585,8 +1631,14 @@ static void drop_connection(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t, } /** update window from a settings change */ +typedef struct { + grpc_chttp2_transport *t; + grpc_exec_ctx *exec_ctx; +} update_global_window_args; + static void update_global_window(void *args, uint32_t id, void *stream) { - grpc_chttp2_transport *t = args; + update_global_window_args *a = args; + grpc_chttp2_transport *t = a->t; grpc_chttp2_stream *s = stream; grpc_chttp2_transport_global *transport_global = &t->global; grpc_chttp2_stream_global *stream_global = &s->global; @@ -1600,7 +1652,7 @@ static void update_global_window(void *args, uint32_t id, void *stream) { is_zero = stream_global->outgoing_window <= 0; if (was_zero && !is_zero) { - grpc_chttp2_become_writable(transport_global, stream_global); + grpc_chttp2_become_writable(a->exec_ctx, transport_global, stream_global); } } @@ -1677,14 +1729,18 @@ static void post_parse_locked(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t, grpc_chttp2_transport_global *transport_global = &t->global; grpc_chttp2_transport_parsing *transport_parsing = &t->parsing; /* copy parsing qbuf to global qbuf */ - gpr_slice_buffer_move_into(&t->parsing.qbuf, &t->global.qbuf); + if (t->parsing.qbuf.count > 0) { + gpr_slice_buffer_move_into(&t->parsing.qbuf, &t->global.qbuf); + grpc_chttp2_initiate_write(exec_ctx, transport_global); + } /* merge stream lists */ grpc_chttp2_stream_map_move_into(&t->new_stream_map, &t->parsing_stream_map); transport_global->concurrent_stream_count = (uint32_t)grpc_chttp2_stream_map_size(&t->parsing_stream_map); if (transport_parsing->initial_window_update != 0) { + update_global_window_args args = {t, exec_ctx}; grpc_chttp2_stream_map_for_each(&t->parsing_stream_map, - update_global_window, t); + update_global_window, &args); transport_parsing->initial_window_update = 0; } /* handle higher level things */ @@ -1774,6 +1830,7 @@ static void add_to_pollset_locked(grpc_exec_ctx *exec_ctx, grpc_chttp2_stream *s_unused, void *pollset) { if (t->ep) { grpc_endpoint_add_to_pollset(exec_ctx, t->ep, pollset); + grpc_workqueue_add_to_pollset(exec_ctx, t->executor.workqueue, pollset); } } @@ -1783,6 +1840,8 @@ static void add_to_pollset_set_locked(grpc_exec_ctx *exec_ctx, void *pollset_set) { if (t->ep) { grpc_endpoint_add_to_pollset_set(exec_ctx, t->ep, pollset_set); + grpc_workqueue_add_to_pollset_set(exec_ctx, t->executor.workqueue, + pollset_set); } } @@ -1808,7 +1867,7 @@ static void incoming_byte_stream_unref(grpc_exec_ctx *exec_ctx, } static void incoming_byte_stream_update_flow_control( - grpc_chttp2_transport_global *transport_global, + grpc_exec_ctx *exec_ctx, grpc_chttp2_transport_global *transport_global, grpc_chttp2_stream_global *stream_global, size_t max_size_hint, size_t have_already) { uint32_t max_recv_bytes; @@ -1843,7 +1902,7 @@ static void incoming_byte_stream_update_flow_control( add_max_recv_bytes); grpc_chttp2_list_add_unannounced_incoming_window_available(transport_global, stream_global); - grpc_chttp2_become_writable(transport_global, stream_global); + grpc_chttp2_become_writable(exec_ctx, transport_global, stream_global); } } @@ -1865,8 +1924,9 @@ static void incoming_byte_stream_next_locked(grpc_exec_ctx *exec_ctx, grpc_chttp2_stream_global *stream_global = &bs->stream->global; if (bs->is_tail) { - incoming_byte_stream_update_flow_control( - transport_global, stream_global, arg->max_size_hint, bs->slices.length); + incoming_byte_stream_update_flow_control(exec_ctx, transport_global, + stream_global, arg->max_size_hint, + bs->slices.length); } if (bs->slices.count > 0) { *arg->slice = gpr_slice_buffer_take_first(&bs->slices); diff --git a/src/core/ext/transport/chttp2/transport/internal.h b/src/core/ext/transport/chttp2/transport/internal.h index 7e5d58380e..bdc4a69f4f 100644 --- a/src/core/ext/transport/chttp2/transport/internal.h +++ b/src/core/ext/transport/chttp2/transport/internal.h @@ -315,11 +315,16 @@ struct grpc_chttp2_transport { struct { gpr_mu mu; + grpc_workqueue *workqueue; /** is a thread currently in the global lock */ bool global_active; - /** is a thread currently writing */ + /** is a write currently initiated */ + bool writing_initiated; + /** is a write actually going on right now */ bool writing_active; + /** is a write needed */ + bool writing_needed; /** is a thread currently parsing */ bool parsing_active; @@ -362,6 +367,8 @@ struct grpc_chttp2_transport { grpc_closure reading_action; /** closure to actually do parsing */ grpc_closure parsing_action; + /** closure to initiate writing */ + grpc_closure initiate_writing; /** incoming read bytes */ gpr_slice_buffer read_buffer; @@ -507,15 +514,16 @@ struct grpc_chttp2_stream { }; /** Transport writing call flow: - chttp2_transport.c calls grpc_chttp2_unlocking_check_writes to see if writes - are required; - if they are, chttp2_transport.c calls grpc_chttp2_perform_writes to do the - writes. - Once writes have been completed (meaning another write could potentially be - started), - grpc_chttp2_terminate_writing is called. This will call - grpc_chttp2_cleanup_writing, at which - point the write phase is complete. */ + grpc_chttp2_initiate_write() is called anywhere that we know bytes need to + go out on the wire. + If no other write has been started, a task is enqueued onto our workqueue. + When that task executes, it obtains the global lock, and gathers the data + to write. + The global lock is dropped and we do the syscall to write. + After writing, a follow-up check is made to see if another round of writing + should be performed. */ +void grpc_chttp2_initiate_write(grpc_exec_ctx *exec_ctx, + grpc_chttp2_transport_global *transport_global); /** Someone is unlocking the transport mutex: check to see if writes are required, and schedule them if so */ @@ -816,7 +824,8 @@ void grpc_chttp2_ack_ping(grpc_exec_ctx *exec_ctx, /** add a ref to the stream and add it to the writable list; ref will be dropped in writing.c */ -void grpc_chttp2_become_writable(grpc_chttp2_transport_global *transport_global, +void grpc_chttp2_become_writable(grpc_exec_ctx *exec_ctx, + grpc_chttp2_transport_global *transport_global, grpc_chttp2_stream_global *stream_global); #endif /* GRPC_CORE_EXT_TRANSPORT_CHTTP2_TRANSPORT_INTERNAL_H */ diff --git a/src/core/ext/transport/chttp2/transport/parsing.c b/src/core/ext/transport/chttp2/transport/parsing.c index 72b3131d7b..3d3abccd20 100644 --- a/src/core/ext/transport/chttp2/transport/parsing.c +++ b/src/core/ext/transport/chttp2/transport/parsing.c @@ -156,7 +156,7 @@ void grpc_chttp2_publish_reads( if (was_zero && !is_zero) { while (grpc_chttp2_list_pop_stalled_by_transport(transport_global, &stream_global)) { - grpc_chttp2_become_writable(transport_global, stream_global); + grpc_chttp2_become_writable(exec_ctx, transport_global, stream_global); } } @@ -168,6 +168,7 @@ void grpc_chttp2_publish_reads( announce_incoming_window, announce_bytes); GRPC_CHTTP2_FLOW_CREDIT_TRANSPORT("parsed", transport_parsing, incoming_window, announce_bytes); + grpc_chttp2_initiate_write(exec_ctx, transport_global); } /* for each stream that saw an update, fixup global state */ @@ -190,7 +191,7 @@ void grpc_chttp2_publish_reads( outgoing_window); is_zero = stream_global->outgoing_window <= 0; if (was_zero && !is_zero) { - grpc_chttp2_become_writable(transport_global, stream_global); + grpc_chttp2_become_writable(exec_ctx, transport_global, stream_global); } stream_global->max_recv_bytes -= (uint32_t)GPR_MIN( diff --git a/src/core/ext/transport/chttp2/transport/stream_lists.c b/src/core/ext/transport/chttp2/transport/stream_lists.c index 8f3ab00e6d..f3690015ea 100644 --- a/src/core/ext/transport/chttp2/transport/stream_lists.c +++ b/src/core/ext/transport/chttp2/transport/stream_lists.c @@ -344,7 +344,8 @@ void grpc_chttp2_list_flush_writing_stalled_by_transport( while (stream_list_pop(transport, &stream, GRPC_CHTTP2_LIST_WRITING_STALLED_BY_TRANSPORT)) { if (is_window_available) { - grpc_chttp2_become_writable(&transport->global, &stream->global); + grpc_chttp2_become_writable(exec_ctx, &transport->global, + &stream->global); } else { grpc_chttp2_list_add_stalled_by_transport(transport_writing, &stream->writing); diff --git a/src/core/lib/iomgr/workqueue.h b/src/core/lib/iomgr/workqueue.h index 7b6402e3a4..88e4211c47 100644 --- a/src/core/lib/iomgr/workqueue.h +++ b/src/core/lib/iomgr/workqueue.h @@ -38,6 +38,7 @@ #include "src/core/lib/iomgr/exec_ctx.h" #include "src/core/lib/iomgr/iomgr.h" #include "src/core/lib/iomgr/pollset.h" +#include "src/core/lib/iomgr/pollset_set.h" #ifdef GPR_POSIX_SOCKET #include "src/core/lib/iomgr/workqueue_posix.h" @@ -76,6 +77,9 @@ void grpc_workqueue_unref(grpc_exec_ctx *exec_ctx, grpc_workqueue *workqueue); void grpc_workqueue_add_to_pollset(grpc_exec_ctx *exec_ctx, grpc_workqueue *workqueue, grpc_pollset *pollset); +void grpc_workqueue_add_to_pollset_set(grpc_exec_ctx *exec_ctx, + grpc_workqueue *workqueue, + grpc_pollset_set *pollset_set); /** Add a work item to a workqueue */ void grpc_workqueue_enqueue(grpc_exec_ctx *exec_ctx, grpc_workqueue *workqueue, diff --git a/src/core/lib/iomgr/workqueue_posix.c b/src/core/lib/iomgr/workqueue_posix.c index 45e0f6063b..c6323e0594 100644 --- a/src/core/lib/iomgr/workqueue_posix.c +++ b/src/core/lib/iomgr/workqueue_posix.c @@ -106,6 +106,12 @@ void grpc_workqueue_add_to_pollset(grpc_exec_ctx *exec_ctx, grpc_pollset_add_fd(exec_ctx, pollset, workqueue->wakeup_read_fd); } +void grpc_workqueue_add_to_pollset_set(grpc_exec_ctx *exec_ctx, + grpc_workqueue *workqueue, + grpc_pollset_set *pollset_set) { + grpc_pollset_set_add_fd(exec_ctx, pollset_set, workqueue->wakeup_read_fd); +} + void grpc_workqueue_flush(grpc_exec_ctx *exec_ctx, grpc_workqueue *workqueue) { gpr_mu_lock(&workqueue->mu); grpc_exec_ctx_enqueue_list(exec_ctx, &workqueue->closure_list, NULL); -- cgit v1.2.3 From 248fcbf06d9b34f67a8aa2dcc552b6180894d093 Mon Sep 17 00:00:00 2001 From: Craig Tiller Date: Tue, 31 May 2016 13:56:10 -0700 Subject: Fixes --- src/core/ext/transport/chttp2/transport/chttp2_transport.c | 5 +++++ src/core/lib/iomgr/workqueue_posix.c | 2 +- 2 files changed, 6 insertions(+), 1 deletion(-) (limited to 'src/core/ext') diff --git a/src/core/ext/transport/chttp2/transport/chttp2_transport.c b/src/core/ext/transport/chttp2/transport/chttp2_transport.c index b94a112b4a..03e02d2522 100644 --- a/src/core/ext/transport/chttp2/transport/chttp2_transport.c +++ b/src/core/ext/transport/chttp2/transport/chttp2_transport.c @@ -200,6 +200,8 @@ static void destruct_transport(grpc_exec_ctx *exec_ctx, gpr_free(ping); } + GRPC_WORKQUEUE_UNREF(exec_ctx, t->executor.workqueue, "transport"); + gpr_free(t->peer_string); gpr_free(t); } @@ -734,6 +736,7 @@ void grpc_chttp2_initiate_write( t->executor.writing_needed = true; if (!t->executor.writing_initiated) { t->executor.writing_initiated = true; + REF_TRANSPORT(t, "initiate_writing"); grpc_workqueue_enqueue(exec_ctx, t->executor.workqueue, &t->initiate_writing, GRPC_ERROR_NONE); } @@ -754,6 +757,7 @@ static void initiate_writing_locked(grpc_exec_ctx *exec_ctx, prevent_endpoint_shutdown(t); grpc_exec_ctx_sched(exec_ctx, &t->writing_action, GRPC_ERROR_NONE, NULL); } + UNREF_TRANSPORT(exec_ctx, t, "initiate_writing"); } static void initiate_writing(grpc_exec_ctx *exec_ctx, void *arg, @@ -812,6 +816,7 @@ static void terminate_writing_with_lock(grpc_exec_ctx *exec_ctx, t->executor.writing_active = false; if (t->executor.writing_needed) { + REF_TRANSPORT(t, "initiate_writing"); initiate_writing_locked(exec_ctx, t, NULL, NULL); } else { t->executor.writing_initiated = false; diff --git a/src/core/lib/iomgr/workqueue_posix.c b/src/core/lib/iomgr/workqueue_posix.c index c6323e0594..bcbc2699be 100644 --- a/src/core/lib/iomgr/workqueue_posix.c +++ b/src/core/lib/iomgr/workqueue_posix.c @@ -70,7 +70,7 @@ grpc_error *grpc_workqueue_create(grpc_exec_ctx *exec_ctx, static void workqueue_destroy(grpc_exec_ctx *exec_ctx, grpc_workqueue *workqueue) { - GPR_ASSERT(grpc_closure_list_empty(workqueue->closure_list)); + grpc_exec_ctx_enqueue_list(exec_ctx, &workqueue->closure_list, NULL); grpc_fd_shutdown(exec_ctx, workqueue->wakeup_read_fd); } -- cgit v1.2.3 From d7906f5d0eaa684175eabc586d1d04b49de53696 Mon Sep 17 00:00:00 2001 From: Craig Tiller Date: Tue, 31 May 2016 16:07:27 -0700 Subject: Add state machine for writes --- .../ext/transport/chttp2/transport/chttp2_plugin.c | 3 + .../transport/chttp2/transport/chttp2_transport.c | 189 ++++++++++++++++----- src/core/ext/transport/chttp2/transport/internal.h | 30 +++- src/core/ext/transport/chttp2/transport/parsing.c | 8 +- .../ext/transport/chttp2/transport/stream_lists.c | 4 +- 5 files changed, 180 insertions(+), 54 deletions(-) (limited to 'src/core/ext') diff --git a/src/core/ext/transport/chttp2/transport/chttp2_plugin.c b/src/core/ext/transport/chttp2/transport/chttp2_plugin.c index bd87253ed3..7d5279b9da 100644 --- a/src/core/ext/transport/chttp2/transport/chttp2_plugin.c +++ b/src/core/ext/transport/chttp2/transport/chttp2_plugin.c @@ -36,11 +36,14 @@ #include "src/core/lib/debug/trace.h" #include "src/core/lib/transport/metadata.h" +extern int grpc_http_write_state_trace; + void grpc_chttp2_plugin_init(void) { grpc_chttp2_base64_encode_and_huffman_compress = grpc_chttp2_base64_encode_and_huffman_compress_impl; grpc_register_tracer("http", &grpc_http_trace); grpc_register_tracer("flowctl", &grpc_flowctl_trace); + grpc_register_tracer("http_write_state", &grpc_http_write_state_trace); } void grpc_chttp2_plugin_shutdown(void) {} diff --git a/src/core/ext/transport/chttp2/transport/chttp2_transport.c b/src/core/ext/transport/chttp2/transport/chttp2_transport.c index 03e02d2522..752591a6db 100644 --- a/src/core/ext/transport/chttp2/transport/chttp2_transport.c +++ b/src/core/ext/transport/chttp2/transport/chttp2_transport.c @@ -63,6 +63,7 @@ int grpc_http_trace = 0; int grpc_flowctl_trace = 0; +int grpc_http_write_state_trace = 0; #define TRANSPORT_FROM_WRITING(tw) \ ((grpc_chttp2_transport *)((char *)(tw)-offsetof(grpc_chttp2_transport, \ @@ -91,6 +92,8 @@ static void parsing_action(grpc_exec_ctx *exec_ctx, void *t, grpc_error *error); static void initiate_writing(grpc_exec_ctx *exec_ctx, void *t, grpc_error *error); +static void start_writing(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t); + /** Set a transport level setting, and push it to our peer */ static void push_setting(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t, grpc_chttp2_setting_id id, uint32_t value); @@ -294,7 +297,7 @@ static void init_transport(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t, gpr_slice_buffer_add( &t->global.qbuf, gpr_slice_from_copied_string(GRPC_CHTTP2_CLIENT_CONNECT_STRING)); - grpc_chttp2_initiate_write(exec_ctx, &t->global); + grpc_chttp2_initiate_write(exec_ctx, &t->global, false); } /* 8 is a random stab in the dark as to a good initial size: it's small enough that it shouldn't waste memory for infrequently used connections, yet @@ -642,6 +645,36 @@ grpc_chttp2_stream_parsing *grpc_chttp2_parsing_accept_stream( * LOCK MANAGEMENT */ +static const char *write_state_name(grpc_chttp2_write_state state) { + switch (state) { + case GRPC_CHTTP2_WRITING_INACTIVE: + return "INACTIVE"; + case GRPC_CHTTP2_WRITE_REQUESTED_NO_POLLER: + return "REQUESTED[p=0]"; + case GRPC_CHTTP2_WRITE_REQUESTED_WITH_POLLER: + return "REQUESTED[p=1]"; + case GRPC_CHTTP2_WRITE_SCHEDULED: + return "SCHEDULED"; + case GRPC_CHTTP2_WRITING: + return "WRITING"; + case GRPC_CHTTP2_WRITING_STALE_WITH_POLLER: + return "WRITING[p=1]"; + case GRPC_CHTTP2_WRITING_STALE_NO_POLLER: + return "WRITING[p=0]"; + } + GPR_UNREACHABLE_CODE(return "UNKNOWN"); +} + +static void set_write_state(grpc_chttp2_transport *t, + grpc_chttp2_write_state state, const char *reason) { + if (grpc_http_write_state_trace) { + gpr_log(GPR_DEBUG, "W:%p %s -> %s because %s", t, + write_state_name(t->executor.write_state), write_state_name(state), + reason); + } + t->executor.write_state = state; +} + static void finish_global_actions(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t) { grpc_chttp2_executor_action_header *hdr; @@ -666,8 +699,27 @@ static void finish_global_actions(grpc_exec_ctx *exec_ctx, continue; } else { t->executor.global_active = false; + switch (t->executor.write_state) { + case GRPC_CHTTP2_WRITE_REQUESTED_WITH_POLLER: + set_write_state(t, GRPC_CHTTP2_WRITE_SCHEDULED, "unlocking"); + REF_TRANSPORT(t, "initiate_writing"); + gpr_mu_unlock(&t->executor.mu); + grpc_workqueue_enqueue(exec_ctx, t->executor.workqueue, + &t->initiate_writing, GRPC_ERROR_NONE); + break; + case GRPC_CHTTP2_WRITE_REQUESTED_NO_POLLER: + start_writing(exec_ctx, t); + gpr_mu_unlock(&t->executor.mu); + break; + case GRPC_CHTTP2_WRITING_INACTIVE: + case GRPC_CHTTP2_WRITING: + case GRPC_CHTTP2_WRITING_STALE_WITH_POLLER: + case GRPC_CHTTP2_WRITING_STALE_NO_POLLER: + case GRPC_CHTTP2_WRITE_SCHEDULED: + gpr_mu_unlock(&t->executor.mu); + break; + } } - gpr_mu_unlock(&t->executor.mu); break; } } @@ -730,33 +782,69 @@ void grpc_chttp2_run_with_global_lock(grpc_exec_ctx *exec_ctx, * OUTPUT PROCESSING */ -void grpc_chttp2_initiate_write( - grpc_exec_ctx *exec_ctx, grpc_chttp2_transport_global *transport_global) { +void grpc_chttp2_initiate_write(grpc_exec_ctx *exec_ctx, + grpc_chttp2_transport_global *transport_global, + bool covered_by_poller) { grpc_chttp2_transport *t = TRANSPORT_FROM_GLOBAL(transport_global); - t->executor.writing_needed = true; - if (!t->executor.writing_initiated) { - t->executor.writing_initiated = true; - REF_TRANSPORT(t, "initiate_writing"); - grpc_workqueue_enqueue(exec_ctx, t->executor.workqueue, - &t->initiate_writing, GRPC_ERROR_NONE); + switch (t->executor.write_state) { + case GRPC_CHTTP2_WRITING_INACTIVE: + set_write_state(t, covered_by_poller + ? GRPC_CHTTP2_WRITE_REQUESTED_WITH_POLLER + : GRPC_CHTTP2_WRITE_REQUESTED_NO_POLLER, + "initiate_write"); + break; + case GRPC_CHTTP2_WRITE_REQUESTED_WITH_POLLER: + /* nothing to do: write already requested */ + break; + case GRPC_CHTTP2_WRITE_REQUESTED_NO_POLLER: + if (covered_by_poller) { + /* upgrade to note poller is available to cover the write */ + set_write_state(t, GRPC_CHTTP2_WRITE_REQUESTED_WITH_POLLER, + "initiate_write"); + } + break; + case GRPC_CHTTP2_WRITE_SCHEDULED: + /* nothing to do: write already scheduled */ + break; + case GRPC_CHTTP2_WRITING: + set_write_state(t, + covered_by_poller ? GRPC_CHTTP2_WRITING_STALE_WITH_POLLER + : GRPC_CHTTP2_WRITING_STALE_NO_POLLER, + "initiate_write"); + break; + case GRPC_CHTTP2_WRITING_STALE_WITH_POLLER: + /* nothing to do: write already requested */ + break; + case GRPC_CHTTP2_WRITING_STALE_NO_POLLER: + if (covered_by_poller) { + /* upgrade to note poller is available to cover the write */ + set_write_state(t, GRPC_CHTTP2_WRITING_STALE_WITH_POLLER, + "initiate_write"); + } + break; } } -static void initiate_writing_locked(grpc_exec_ctx *exec_ctx, - grpc_chttp2_transport *t, - grpc_chttp2_stream *s_unused, - void *arg_ignored) { - GPR_ASSERT(t->executor.writing_needed); - GPR_ASSERT(!t->executor.writing_active); +static void start_writing(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t) { + GPR_ASSERT(t->executor.write_state == GRPC_CHTTP2_WRITE_SCHEDULED || + t->executor.write_state == GRPC_CHTTP2_WRITE_REQUESTED_NO_POLLER); if (!t->closed && grpc_chttp2_unlocking_check_writes(exec_ctx, &t->global, &t->writing, t->executor.parsing_active)) { - t->executor.writing_needed = false; - t->executor.writing_active = true; + set_write_state(t, GRPC_CHTTP2_WRITING, "start_writing"); REF_TRANSPORT(t, "writing"); prevent_endpoint_shutdown(t); grpc_exec_ctx_sched(exec_ctx, &t->writing_action, GRPC_ERROR_NONE, NULL); + } else { + set_write_state(t, GRPC_CHTTP2_WRITING_INACTIVE, "start_writing"); } +} + +static void initiate_writing_locked(grpc_exec_ctx *exec_ctx, + grpc_chttp2_transport *t, + grpc_chttp2_stream *s_unused, + void *arg_ignored) { + start_writing(exec_ctx, t); UNREF_TRANSPORT(exec_ctx, t, "initiate_writing"); } @@ -768,11 +856,12 @@ static void initiate_writing(grpc_exec_ctx *exec_ctx, void *arg, void grpc_chttp2_become_writable(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport_global *transport_global, - grpc_chttp2_stream_global *stream_global) { + grpc_chttp2_stream_global *stream_global, + bool covered_by_poller) { if (!TRANSPORT_FROM_GLOBAL(transport_global)->closed && grpc_chttp2_list_add_writable_stream(transport_global, stream_global)) { GRPC_CHTTP2_STREAM_REF(stream_global, "chttp2_writing"); - grpc_chttp2_initiate_write(exec_ctx, transport_global); + grpc_chttp2_initiate_write(exec_ctx, transport_global, covered_by_poller); } } @@ -788,7 +877,7 @@ static void push_setting(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t, if (use_value != t->global.settings[GRPC_LOCAL_SETTINGS][id]) { t->global.settings[GRPC_LOCAL_SETTINGS][id] = use_value; t->global.dirtied_local_settings = 1; - grpc_chttp2_initiate_write(exec_ctx, &t->global); + grpc_chttp2_initiate_write(exec_ctx, &t->global, false); } } @@ -814,13 +903,25 @@ static void terminate_writing_with_lock(grpc_exec_ctx *exec_ctx, GRPC_CHTTP2_STREAM_UNREF(exec_ctx, stream_global, "finish_writes"); } - t->executor.writing_active = false; - if (t->executor.writing_needed) { - REF_TRANSPORT(t, "initiate_writing"); - initiate_writing_locked(exec_ctx, t, NULL, NULL); - } else { - t->executor.writing_initiated = false; + switch (t->executor.write_state) { + case GRPC_CHTTP2_WRITING_INACTIVE: + case GRPC_CHTTP2_WRITE_REQUESTED_WITH_POLLER: + case GRPC_CHTTP2_WRITE_REQUESTED_NO_POLLER: + case GRPC_CHTTP2_WRITE_SCHEDULED: + GPR_UNREACHABLE_CODE(break); + case GRPC_CHTTP2_WRITING: + set_write_state(t, GRPC_CHTTP2_WRITING_INACTIVE, "terminate_writing"); + break; + case GRPC_CHTTP2_WRITING_STALE_WITH_POLLER: + set_write_state(t, GRPC_CHTTP2_WRITE_REQUESTED_WITH_POLLER, + "terminate_writing"); + break; + case GRPC_CHTTP2_WRITING_STALE_NO_POLLER: + set_write_state(t, GRPC_CHTTP2_WRITE_REQUESTED_NO_POLLER, + "terminate_writing"); + break; } + if (t->ep && !t->endpoint_reading) { destroy_endpoint(exec_ctx, t); } @@ -907,7 +1008,8 @@ static void maybe_start_some_streams( stream_global->id, STREAM_FROM_GLOBAL(stream_global)); stream_global->in_stream_map = true; transport_global->concurrent_stream_count++; - grpc_chttp2_become_writable(exec_ctx, transport_global, stream_global); + grpc_chttp2_become_writable(exec_ctx, transport_global, stream_global, + true); } /* cancel out streams that will never be started */ while (transport_global->next_stream_id >= MAX_CLIENT_STREAM_ID && @@ -1036,8 +1138,8 @@ static void perform_stream_op_locked(grpc_exec_ctx *exec_ctx, maybe_start_some_streams(exec_ctx, transport_global); } else { GPR_ASSERT(stream_global->id != 0); - grpc_chttp2_become_writable(exec_ctx, transport_global, - stream_global); + grpc_chttp2_become_writable(exec_ctx, transport_global, stream_global, + true); } } else { grpc_chttp2_complete_closure_step( @@ -1061,7 +1163,8 @@ static void perform_stream_op_locked(grpc_exec_ctx *exec_ctx, } else { stream_global->send_message = op->send_message; if (stream_global->id != 0) { - grpc_chttp2_become_writable(exec_ctx, transport_global, stream_global); + grpc_chttp2_become_writable(exec_ctx, transport_global, stream_global, + true); } } } @@ -1100,7 +1203,8 @@ static void perform_stream_op_locked(grpc_exec_ctx *exec_ctx, } else if (stream_global->id != 0) { /* TODO(ctiller): check if there's flow control for any outstanding bytes before going writable */ - grpc_chttp2_become_writable(exec_ctx, transport_global, stream_global); + grpc_chttp2_become_writable(exec_ctx, transport_global, stream_global, + true); } } } @@ -1165,7 +1269,7 @@ static void send_ping_locked(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t, p->id[7] = (uint8_t)(t->global.ping_counter & 0xff); p->on_recv = on_recv; gpr_slice_buffer_add(&t->global.qbuf, grpc_chttp2_ping_create(0, p->id)); - grpc_chttp2_initiate_write(exec_ctx, &t->global); + grpc_chttp2_initiate_write(exec_ctx, &t->global, true); } static void ack_ping_locked(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t, @@ -1225,7 +1329,7 @@ static void perform_transport_op_locked(grpc_exec_ctx *exec_ctx, close_transport = grpc_chttp2_has_streams(t) ? GRPC_ERROR_NONE : GRPC_ERROR_CREATE("GOAWAY sent"); - grpc_chttp2_initiate_write(exec_ctx, &t->global); + grpc_chttp2_initiate_write(exec_ctx, &t->global, false); } if (op->set_accept_stream) { @@ -1392,7 +1496,7 @@ static void cancel_from_api(grpc_exec_ctx *exec_ctx, stream_global->id, (uint32_t)grpc_chttp2_grpc_status_to_http2_error(status), &stream_global->stats.outgoing)); - grpc_chttp2_initiate_write(exec_ctx, transport_global); + grpc_chttp2_initiate_write(exec_ctx, transport_global, false); } grpc_chttp2_fake_status(exec_ctx, transport_global, stream_global, status, NULL); @@ -1477,7 +1581,8 @@ void grpc_chttp2_mark_stream_closed( } if (close_writes && !stream_global->write_closed) { stream_global->write_closed = true; - if (TRANSPORT_FROM_GLOBAL(transport_global)->executor.writing_active) { + if (TRANSPORT_FROM_GLOBAL(transport_global)->executor.write_state != + GRPC_CHTTP2_WRITING_INACTIVE) { GRPC_CHTTP2_STREAM_REF(stream_global, "finish_writes"); grpc_chttp2_list_add_closed_waiting_for_writing(transport_global, stream_global); @@ -1614,7 +1719,7 @@ static void close_from_api(grpc_exec_ctx *exec_ctx, } grpc_chttp2_mark_stream_closed(exec_ctx, transport_global, stream_global, 1, 1, err); - grpc_chttp2_initiate_write(exec_ctx, transport_global); + grpc_chttp2_initiate_write(exec_ctx, transport_global, false); } static void cancel_stream_cb(grpc_chttp2_transport_global *transport_global, @@ -1657,7 +1762,8 @@ static void update_global_window(void *args, uint32_t id, void *stream) { is_zero = stream_global->outgoing_window <= 0; if (was_zero && !is_zero) { - grpc_chttp2_become_writable(a->exec_ctx, transport_global, stream_global); + grpc_chttp2_become_writable(a->exec_ctx, transport_global, stream_global, + true); } } @@ -1736,7 +1842,7 @@ static void post_parse_locked(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t, /* copy parsing qbuf to global qbuf */ if (t->parsing.qbuf.count > 0) { gpr_slice_buffer_move_into(&t->parsing.qbuf, &t->global.qbuf); - grpc_chttp2_initiate_write(exec_ctx, transport_global); + grpc_chttp2_initiate_write(exec_ctx, transport_global, false); } /* merge stream lists */ grpc_chttp2_stream_map_move_into(&t->new_stream_map, &t->parsing_stream_map); @@ -1787,7 +1893,7 @@ static void post_reading_action_locked(grpc_exec_ctx *exec_ctx, if (error != GRPC_ERROR_NONE) { drop_connection(exec_ctx, t, GRPC_ERROR_REF(error)); t->endpoint_reading = 0; - if (!t->executor.writing_active && t->ep) { + if (t->executor.write_state == GRPC_CHTTP2_WRITING_INACTIVE && t->ep) { grpc_endpoint_destroy(exec_ctx, t->ep); t->ep = NULL; /* safe as we still have a ref for read */ @@ -1907,7 +2013,8 @@ static void incoming_byte_stream_update_flow_control( add_max_recv_bytes); grpc_chttp2_list_add_unannounced_incoming_window_available(transport_global, stream_global); - grpc_chttp2_become_writable(exec_ctx, transport_global, stream_global); + grpc_chttp2_become_writable(exec_ctx, transport_global, stream_global, + false); } } diff --git a/src/core/ext/transport/chttp2/transport/internal.h b/src/core/ext/transport/chttp2/transport/internal.h index bdc4a69f4f..988e6b6b00 100644 --- a/src/core/ext/transport/chttp2/transport/internal.h +++ b/src/core/ext/transport/chttp2/transport/internal.h @@ -304,6 +304,22 @@ typedef struct grpc_chttp2_executor_action_header { void *arg; } grpc_chttp2_executor_action_header; +typedef enum { + /** no writing activity */ + GRPC_CHTTP2_WRITING_INACTIVE, + /** write has been requested, but not scheduled yet */ + GRPC_CHTTP2_WRITE_REQUESTED_WITH_POLLER, + GRPC_CHTTP2_WRITE_REQUESTED_NO_POLLER, + /** write has been requested and scheduled against the workqueue */ + GRPC_CHTTP2_WRITE_SCHEDULED, + /** write has been initiated after being reaped from the workqueue */ + GRPC_CHTTP2_WRITING, + /** write has been initiated, AND another write needs to be started once it's + done */ + GRPC_CHTTP2_WRITING_STALE_WITH_POLLER, + GRPC_CHTTP2_WRITING_STALE_NO_POLLER, +} grpc_chttp2_write_state; + struct grpc_chttp2_transport { grpc_transport base; /* must be first */ gpr_refcount refs; @@ -319,14 +335,10 @@ struct grpc_chttp2_transport { /** is a thread currently in the global lock */ bool global_active; - /** is a write currently initiated */ - bool writing_initiated; - /** is a write actually going on right now */ - bool writing_active; - /** is a write needed */ - bool writing_needed; /** is a thread currently parsing */ bool parsing_active; + /** write execution state of the transport */ + grpc_chttp2_write_state write_state; grpc_chttp2_executor_action_header *pending_actions_head; grpc_chttp2_executor_action_header *pending_actions_tail; @@ -523,7 +535,8 @@ struct grpc_chttp2_stream { After writing, a follow-up check is made to see if another round of writing should be performed. */ void grpc_chttp2_initiate_write(grpc_exec_ctx *exec_ctx, - grpc_chttp2_transport_global *transport_global); + grpc_chttp2_transport_global *transport_global, + bool covered_by_poller); /** Someone is unlocking the transport mutex: check to see if writes are required, and schedule them if so */ @@ -826,6 +839,7 @@ void grpc_chttp2_ack_ping(grpc_exec_ctx *exec_ctx, ref will be dropped in writing.c */ void grpc_chttp2_become_writable(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport_global *transport_global, - grpc_chttp2_stream_global *stream_global); + grpc_chttp2_stream_global *stream_global, + bool covered_by_poller); #endif /* GRPC_CORE_EXT_TRANSPORT_CHTTP2_TRANSPORT_INTERNAL_H */ diff --git a/src/core/ext/transport/chttp2/transport/parsing.c b/src/core/ext/transport/chttp2/transport/parsing.c index 3d3abccd20..8f80cb61b2 100644 --- a/src/core/ext/transport/chttp2/transport/parsing.c +++ b/src/core/ext/transport/chttp2/transport/parsing.c @@ -156,7 +156,8 @@ void grpc_chttp2_publish_reads( if (was_zero && !is_zero) { while (grpc_chttp2_list_pop_stalled_by_transport(transport_global, &stream_global)) { - grpc_chttp2_become_writable(exec_ctx, transport_global, stream_global); + grpc_chttp2_become_writable(exec_ctx, transport_global, stream_global, + false); } } @@ -168,7 +169,7 @@ void grpc_chttp2_publish_reads( announce_incoming_window, announce_bytes); GRPC_CHTTP2_FLOW_CREDIT_TRANSPORT("parsed", transport_parsing, incoming_window, announce_bytes); - grpc_chttp2_initiate_write(exec_ctx, transport_global); + grpc_chttp2_initiate_write(exec_ctx, transport_global, false); } /* for each stream that saw an update, fixup global state */ @@ -191,7 +192,8 @@ void grpc_chttp2_publish_reads( outgoing_window); is_zero = stream_global->outgoing_window <= 0; if (was_zero && !is_zero) { - grpc_chttp2_become_writable(exec_ctx, transport_global, stream_global); + grpc_chttp2_become_writable(exec_ctx, transport_global, stream_global, + false); } stream_global->max_recv_bytes -= (uint32_t)GPR_MIN( diff --git a/src/core/ext/transport/chttp2/transport/stream_lists.c b/src/core/ext/transport/chttp2/transport/stream_lists.c index f3690015ea..f227a0af98 100644 --- a/src/core/ext/transport/chttp2/transport/stream_lists.c +++ b/src/core/ext/transport/chttp2/transport/stream_lists.c @@ -344,8 +344,8 @@ void grpc_chttp2_list_flush_writing_stalled_by_transport( while (stream_list_pop(transport, &stream, GRPC_CHTTP2_LIST_WRITING_STALLED_BY_TRANSPORT)) { if (is_window_available) { - grpc_chttp2_become_writable(exec_ctx, &transport->global, - &stream->global); + grpc_chttp2_become_writable(exec_ctx, &transport->global, &stream->global, + true); } else { grpc_chttp2_list_add_stalled_by_transport(transport_writing, &stream->writing); -- cgit v1.2.3 From 2cdf0a8b912db8882c5f2a7c93ecde9e5a172073 Mon Sep 17 00:00:00 2001 From: Craig Tiller Date: Tue, 31 May 2016 21:47:19 -0700 Subject: Fix ref counting --- .../transport/chttp2/transport/chttp2_transport.c | 48 ++++++++++++++++------ src/core/lib/iomgr/workqueue.h | 2 +- 2 files changed, 36 insertions(+), 14 deletions(-) (limited to 'src/core/ext') diff --git a/src/core/ext/transport/chttp2/transport/chttp2_transport.c b/src/core/ext/transport/chttp2/transport/chttp2_transport.c index 752591a6db..fba1f0e314 100644 --- a/src/core/ext/transport/chttp2/transport/chttp2_transport.c +++ b/src/core/ext/transport/chttp2/transport/chttp2_transport.c @@ -93,6 +93,8 @@ static void initiate_writing(grpc_exec_ctx *exec_ctx, void *t, grpc_error *error); static void start_writing(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t); +static void end_waiting_for_write(grpc_exec_ctx *exec_ctx, + grpc_chttp2_transport *t, grpc_error *error); /** Set a transport level setting, and push it to our peer */ static void push_setting(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t, @@ -209,6 +211,7 @@ static void destruct_transport(grpc_exec_ctx *exec_ctx, gpr_free(t); } +/*#define REFCOUNTING_DEBUG 1*/ #ifdef REFCOUNTING_DEBUG #define REF_TRANSPORT(t, r) ref_transport(t, r, __FILE__, __LINE__) #define UNREF_TRANSPORT(cl, t, r) unref_transport(cl, t, r, __FILE__, __LINE__) @@ -457,6 +460,9 @@ static void close_transport_locked(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t, grpc_error *error) { if (!t->closed) { + if (grpc_http_write_state_trace) { + gpr_log(GPR_DEBUG, "W:%p close transport", t); + } t->closed = 1; connectivity_state_set(exec_ctx, &t->global, GRPC_CHANNEL_FATAL_FAILURE, GRPC_ERROR_REF(error), "close_transport"); @@ -836,7 +842,17 @@ static void start_writing(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t) { prevent_endpoint_shutdown(t); grpc_exec_ctx_sched(exec_ctx, &t->writing_action, GRPC_ERROR_NONE, NULL); } else { - set_write_state(t, GRPC_CHTTP2_WRITING_INACTIVE, "start_writing"); + if (t->closed) { + set_write_state(t, GRPC_CHTTP2_WRITING_INACTIVE, + "start_writing:transport_closed"); + } else { + set_write_state(t, GRPC_CHTTP2_WRITING_INACTIVE, + "start_writing:nothing_to_write"); + } + end_waiting_for_write(exec_ctx, t, GRPC_ERROR_CREATE("Nothing to write")); + if (!t->endpoint_reading) { + destroy_endpoint(exec_ctx, t); + } } } @@ -881,6 +897,18 @@ static void push_setting(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t, } } +static void end_waiting_for_write(grpc_exec_ctx *exec_ctx, + grpc_chttp2_transport *t, grpc_error *error) { + grpc_chttp2_stream_global *stream_global; + while (grpc_chttp2_list_pop_closed_waiting_for_writing(&t->global, + &stream_global)) { + fail_pending_writes(exec_ctx, &t->global, stream_global, + GRPC_ERROR_REF(error)); + GRPC_CHTTP2_STREAM_UNREF(exec_ctx, stream_global, "finish_writes"); + } + GRPC_ERROR_UNREF(error); +} + static void terminate_writing_with_lock(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t, grpc_chttp2_stream *s_ignored, @@ -895,13 +923,7 @@ static void terminate_writing_with_lock(grpc_exec_ctx *exec_ctx, grpc_chttp2_cleanup_writing(exec_ctx, &t->global, &t->writing); - grpc_chttp2_stream_global *stream_global; - while (grpc_chttp2_list_pop_closed_waiting_for_writing(&t->global, - &stream_global)) { - fail_pending_writes(exec_ctx, &t->global, stream_global, - GRPC_ERROR_REF(error)); - GRPC_CHTTP2_STREAM_UNREF(exec_ctx, stream_global, "finish_writes"); - } + end_waiting_for_write(exec_ctx, t, error); switch (t->executor.write_state) { case GRPC_CHTTP2_WRITING_INACTIVE: @@ -927,7 +949,6 @@ static void terminate_writing_with_lock(grpc_exec_ctx *exec_ctx, } UNREF_TRANSPORT(exec_ctx, t, "writing"); - GRPC_ERROR_UNREF(error); } void grpc_chttp2_terminate_writing(grpc_exec_ctx *exec_ctx, @@ -1893,11 +1914,12 @@ static void post_reading_action_locked(grpc_exec_ctx *exec_ctx, if (error != GRPC_ERROR_NONE) { drop_connection(exec_ctx, t, GRPC_ERROR_REF(error)); t->endpoint_reading = 0; + if (grpc_http_write_state_trace) { + gpr_log(GPR_DEBUG, "R:%p -> 0 ws=%s", t, + write_state_name(t->executor.write_state)); + } if (t->executor.write_state == GRPC_CHTTP2_WRITING_INACTIVE && t->ep) { - grpc_endpoint_destroy(exec_ctx, t->ep); - t->ep = NULL; - /* safe as we still have a ref for read */ - UNREF_TRANSPORT(exec_ctx, t, "disconnect"); + destroy_endpoint(exec_ctx, t); } } else if (!t->closed) { keep_reading = true; diff --git a/src/core/lib/iomgr/workqueue.h b/src/core/lib/iomgr/workqueue.h index 88e4211c47..f4a2cfa4fd 100644 --- a/src/core/lib/iomgr/workqueue.h +++ b/src/core/lib/iomgr/workqueue.h @@ -56,7 +56,7 @@ grpc_error *grpc_workqueue_create(grpc_exec_ctx *exec_ctx, void grpc_workqueue_flush(grpc_exec_ctx *exec_ctx, grpc_workqueue *workqueue); -#define GRPC_WORKQUEUE_REFCOUNT_DEBUG +/*#define GRPC_WORKQUEUE_REFCOUNT_DEBUG*/ #ifdef GRPC_WORKQUEUE_REFCOUNT_DEBUG #define GRPC_WORKQUEUE_REF(p, r) \ grpc_workqueue_ref((p), __FILE__, __LINE__, (r)) -- cgit v1.2.3 From caad18024d93f3c12d3e7f4c3f903e7df5a4c5c8 Mon Sep 17 00:00:00 2001 From: Craig Tiller Date: Tue, 31 May 2016 22:05:00 -0700 Subject: Fix crash --- src/core/ext/transport/chttp2/transport/chttp2_transport.c | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) (limited to 'src/core/ext') diff --git a/src/core/ext/transport/chttp2/transport/chttp2_transport.c b/src/core/ext/transport/chttp2/transport/chttp2_transport.c index fba1f0e314..827d65f610 100644 --- a/src/core/ext/transport/chttp2/transport/chttp2_transport.c +++ b/src/core/ext/transport/chttp2/transport/chttp2_transport.c @@ -850,7 +850,7 @@ static void start_writing(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t) { "start_writing:nothing_to_write"); } end_waiting_for_write(exec_ctx, t, GRPC_ERROR_CREATE("Nothing to write")); - if (!t->endpoint_reading) { + if (t->ep && !t->endpoint_reading) { destroy_endpoint(exec_ctx, t); } } -- cgit v1.2.3 From 09c464a1424067fc3b5e5f0b1280a8f8c3444342 Mon Sep 17 00:00:00 2001 From: Craig Tiller Date: Wed, 1 Jun 2016 08:16:29 -0700 Subject: Fix too much metadata --- src/core/ext/transport/chttp2/transport/chttp2_transport.c | 13 +++++++++---- src/core/lib/iomgr/error.c | 2 ++ src/core/lib/iomgr/error.h | 1 + 3 files changed, 12 insertions(+), 4 deletions(-) (limited to 'src/core/ext') diff --git a/src/core/ext/transport/chttp2/transport/chttp2_transport.c b/src/core/ext/transport/chttp2/transport/chttp2_transport.c index 827d65f610..31b43fb026 100644 --- a/src/core/ext/transport/chttp2/transport/chttp2_transport.c +++ b/src/core/ext/transport/chttp2/transport/chttp2_transport.c @@ -1140,10 +1140,15 @@ static void perform_stream_op_locked(grpc_exec_ctx *exec_ctx, transport_global->settings[GRPC_PEER_SETTINGS] [GRPC_CHTTP2_SETTINGS_MAX_HEADER_LIST_SIZE]; if (metadata_size > metadata_peer_limit) { - gpr_log(GPR_DEBUG, - "to-be-sent initial metadata size exceeds peer limit " - "(%lu vs. %lu)", - metadata_size, metadata_peer_limit); + grpc_chttp2_complete_closure_step( + exec_ctx, transport_global, stream_global, + &stream_global->send_initial_metadata_finished, + grpc_error_set_int( + grpc_error_set_int( + GRPC_ERROR_CREATE( + "to-be-sent initial metadata size exceeds peer limit"), + GRPC_ERROR_INT_SIZE, (intptr_t)metadata_size), + GRPC_ERROR_INT_LIMIT, (intptr_t)metadata_peer_limit)); cancel_from_api(exec_ctx, transport_global, stream_global, GRPC_STATUS_RESOURCE_EXHAUSTED); } else { diff --git a/src/core/lib/iomgr/error.c b/src/core/lib/iomgr/error.c index b38584100d..35154268f3 100644 --- a/src/core/lib/iomgr/error.c +++ b/src/core/lib/iomgr/error.c @@ -102,6 +102,8 @@ static const char *error_int_name(grpc_error_ints key) { return "grpc_status"; case GRPC_ERROR_INT_OFFSET: return "offset"; + case GRPC_ERROR_INT_LIMIT: + return "limit"; case GRPC_ERROR_INT_INDEX: return "index"; case GRPC_ERROR_INT_SIZE: diff --git a/src/core/lib/iomgr/error.h b/src/core/lib/iomgr/error.h index 695724c0be..ede159526e 100644 --- a/src/core/lib/iomgr/error.h +++ b/src/core/lib/iomgr/error.h @@ -62,6 +62,7 @@ typedef enum { GRPC_ERROR_INT_OFFSET, GRPC_ERROR_INT_INDEX, GRPC_ERROR_INT_SIZE, + GRPC_ERROR_INT_LIMIT, GRPC_ERROR_INT_HTTP2_ERROR, GRPC_ERROR_INT_TSI_CODE, GRPC_ERROR_INT_SECURITY_STATUS, -- cgit v1.2.3 From efe7a6e3380c9ddaa7d70b516fda281a441a7e7f Mon Sep 17 00:00:00 2001 From: Craig Tiller Date: Thu, 30 Jun 2016 14:34:04 -0700 Subject: Use endpoint workqueue via execution context --- src/core/ext/transport/chttp2/transport/chttp2_transport.c | 12 ++---------- src/core/ext/transport/chttp2/transport/internal.h | 1 - 2 files changed, 2 insertions(+), 11 deletions(-) (limited to 'src/core/ext') diff --git a/src/core/ext/transport/chttp2/transport/chttp2_transport.c b/src/core/ext/transport/chttp2/transport/chttp2_transport.c index 197cee7e0c..92722b50f4 100644 --- a/src/core/ext/transport/chttp2/transport/chttp2_transport.c +++ b/src/core/ext/transport/chttp2/transport/chttp2_transport.c @@ -204,8 +204,6 @@ static void destruct_transport(grpc_exec_ctx *exec_ctx, gpr_free(ping); } - GRPC_WORKQUEUE_UNREF(exec_ctx, t->executor.workqueue, "transport"); - gpr_free(t->peer_string); gpr_free(t); } @@ -257,9 +255,6 @@ static void init_transport(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t, /* ref is dropped at transport close() */ gpr_ref_init(&t->shutdown_ep_refs, 1); gpr_mu_init(&t->executor.mu); - GPR_ASSERT(GRPC_LOG_IF_ERROR( - "workqueue_create", - grpc_workqueue_create(exec_ctx, &t->executor.workqueue))); t->peer_string = grpc_endpoint_get_peer(ep); t->endpoint_reading = 1; t->global.next_stream_id = is_client ? 1 : 2; @@ -715,8 +710,8 @@ static void finish_global_actions(grpc_exec_ctx *exec_ctx, set_write_state(t, GRPC_CHTTP2_WRITE_SCHEDULED, "unlocking"); REF_TRANSPORT(t, "initiate_writing"); gpr_mu_unlock(&t->executor.mu); - grpc_workqueue_enqueue(exec_ctx, t->executor.workqueue, - &t->initiate_writing, GRPC_ERROR_NONE); + grpc_exec_ctx_sched(exec_ctx, &t->initiate_writing, GRPC_ERROR_NONE, + grpc_endpoint_get_workqueue(t->ep)); break; case GRPC_CHTTP2_WRITE_REQUESTED_NO_POLLER: start_writing(exec_ctx, t); @@ -2082,7 +2077,6 @@ static void add_to_pollset_locked(grpc_exec_ctx *exec_ctx, grpc_chttp2_stream *s_unused, void *pollset) { if (t->ep) { grpc_endpoint_add_to_pollset(exec_ctx, t->ep, pollset); - grpc_workqueue_add_to_pollset(exec_ctx, t->executor.workqueue, pollset); } } @@ -2092,8 +2086,6 @@ static void add_to_pollset_set_locked(grpc_exec_ctx *exec_ctx, void *pollset_set) { if (t->ep) { grpc_endpoint_add_to_pollset_set(exec_ctx, t->ep, pollset_set); - grpc_workqueue_add_to_pollset_set(exec_ctx, t->executor.workqueue, - pollset_set); } } diff --git a/src/core/ext/transport/chttp2/transport/internal.h b/src/core/ext/transport/chttp2/transport/internal.h index 2ef1c9df7a..4e0c31c111 100644 --- a/src/core/ext/transport/chttp2/transport/internal.h +++ b/src/core/ext/transport/chttp2/transport/internal.h @@ -332,7 +332,6 @@ struct grpc_chttp2_transport { struct { gpr_mu mu; - grpc_workqueue *workqueue; /** is a thread currently in the global lock */ bool global_active; -- cgit v1.2.3 From 571f3e55b4a304cb50b211dd3492ce1664d5bf0c Mon Sep 17 00:00:00 2001 From: Craig Tiller Date: Wed, 6 Jul 2016 14:00:24 -0700 Subject: Prevent a use-after-free --- src/core/ext/transport/chttp2/transport/chttp2_transport.c | 2 ++ 1 file changed, 2 insertions(+) (limited to 'src/core/ext') diff --git a/src/core/ext/transport/chttp2/transport/chttp2_transport.c b/src/core/ext/transport/chttp2/transport/chttp2_transport.c index 92722b50f4..b58d93787d 100644 --- a/src/core/ext/transport/chttp2/transport/chttp2_transport.c +++ b/src/core/ext/transport/chttp2/transport/chttp2_transport.c @@ -1178,6 +1178,7 @@ static void perform_stream_op_locked(grpc_exec_ctx *exec_ctx, true); } } else { + stream_global->send_initial_metadata = NULL; grpc_chttp2_complete_closure_step( exec_ctx, transport_global, stream_global, &stream_global->send_initial_metadata_finished, @@ -1233,6 +1234,7 @@ static void perform_stream_op_locked(grpc_exec_ctx *exec_ctx, grpc_chttp2_list_add_check_read_ops(transport_global, stream_global); } if (stream_global->write_closed) { + stream_global->send_trailing_metadata = NULL; grpc_chttp2_complete_closure_step( exec_ctx, transport_global, stream_global, &stream_global->send_trailing_metadata_finished, -- cgit v1.2.3 From 4ebfe9056063dc27c8e2cfced886b9d0b05bd8e2 Mon Sep 17 00:00:00 2001 From: Craig Tiller Date: Wed, 6 Jul 2016 16:24:45 -0700 Subject: Properly report large metadata errors --- .../transport/chttp2/transport/chttp2_transport.c | 41 +++++++++++++++++++--- src/core/ext/transport/chttp2/transport/internal.h | 6 ++-- src/core/lib/iomgr/workqueue.h | 2 +- 3 files changed, 41 insertions(+), 8 deletions(-) (limited to 'src/core/ext') diff --git a/src/core/ext/transport/chttp2/transport/chttp2_transport.c b/src/core/ext/transport/chttp2/transport/chttp2_transport.c index b58d93787d..8b7f91d3af 100644 --- a/src/core/ext/transport/chttp2/transport/chttp2_transport.c +++ b/src/core/ext/transport/chttp2/transport/chttp2_transport.c @@ -603,7 +603,8 @@ static void destroy_stream_locked(grpc_exec_ctx *exec_ctx, grpc_chttp2_incoming_metadata_buffer_destroy( &s->global.received_trailing_metadata); gpr_slice_buffer_destroy(&s->writing.flow_controlled_buffer); - GRPC_ERROR_UNREF(s->global.removal_error); + GRPC_ERROR_UNREF(s->global.read_closed_error); + GRPC_ERROR_UNREF(s->global.write_closed_error); UNREF_TRANSPORT(exec_ctx, t, "stream"); @@ -1178,7 +1179,7 @@ static void perform_stream_op_locked(grpc_exec_ctx *exec_ctx, true); } } else { - stream_global->send_initial_metadata = NULL; + stream_global->send_trailing_metadata = NULL; grpc_chttp2_complete_closure_step( exec_ctx, transport_global, stream_global, &stream_global->send_initial_metadata_finished, @@ -1637,10 +1638,38 @@ void grpc_chttp2_fake_status(grpc_exec_ctx *exec_ctx, } } +static void add_error(grpc_error *error, grpc_error **refs, size_t *nrefs) { + if (error == GRPC_ERROR_NONE) return; + for (size_t i = 0; i < *nrefs; i++) { + if (error == refs[i]) { + return; + } + } + refs[*nrefs] = error; + ++*nrefs; +} + +static grpc_error *removal_error(grpc_error *extra_error, + grpc_chttp2_stream_global *stream_global) { + grpc_error *refs[3]; + size_t nrefs = 0; + add_error(stream_global->read_closed_error, refs, &nrefs); + add_error(stream_global->write_closed_error, refs, &nrefs); + add_error(extra_error, refs, &nrefs); + grpc_error *error = GRPC_ERROR_NONE; + if (nrefs > 0) { + error = GRPC_ERROR_CREATE_REFERENCING("Failed due to stream removal", refs, + nrefs); + } + GRPC_ERROR_UNREF(extra_error); + return error; +} + static void fail_pending_writes(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport_global *transport_global, grpc_chttp2_stream_global *stream_global, grpc_error *error) { + error = removal_error(error, stream_global); grpc_chttp2_complete_closure_step( exec_ctx, transport_global, stream_global, &stream_global->send_initial_metadata_finished, GRPC_ERROR_REF(error)); @@ -1663,12 +1692,14 @@ void grpc_chttp2_mark_stream_closed( } grpc_chttp2_list_add_check_read_ops(transport_global, stream_global); if (close_reads && !stream_global->read_closed) { + stream_global->read_closed_error = GRPC_ERROR_REF(error); stream_global->read_closed = true; stream_global->published_initial_metadata = true; stream_global->published_trailing_metadata = true; decrement_active_streams_locked(exec_ctx, transport_global, stream_global); } if (close_writes && !stream_global->write_closed) { + stream_global->write_closed_error = GRPC_ERROR_REF(error); stream_global->write_closed = true; if (TRANSPORT_FROM_GLOBAL(transport_global)->executor.write_state != GRPC_CHTTP2_WRITING_INACTIVE) { @@ -1681,7 +1712,6 @@ void grpc_chttp2_mark_stream_closed( } } if (stream_global->read_closed && stream_global->write_closed) { - stream_global->removal_error = GRPC_ERROR_REF(error); if (stream_global->id != 0 && TRANSPORT_FROM_GLOBAL(transport_global)->executor.parsing_active) { grpc_chttp2_list_add_closed_waiting_for_parsing(transport_global, @@ -1689,7 +1719,8 @@ void grpc_chttp2_mark_stream_closed( } else { if (stream_global->id != 0) { remove_stream(exec_ctx, TRANSPORT_FROM_GLOBAL(transport_global), - stream_global->id, GRPC_ERROR_REF(error)); + stream_global->id, + removal_error(GRPC_ERROR_REF(error), stream_global)); } GRPC_CHTTP2_STREAM_UNREF(exec_ctx, stream_global, "chttp2"); } @@ -2008,7 +2039,7 @@ static void post_parse_locked(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t, GPR_ASSERT(stream_global->write_closed); GPR_ASSERT(stream_global->read_closed); remove_stream(exec_ctx, t, stream_global->id, - GRPC_ERROR_REF(stream_global->removal_error)); + removal_error(GRPC_ERROR_NONE, stream_global)); GRPC_CHTTP2_STREAM_UNREF(exec_ctx, stream_global, "chttp2"); } diff --git a/src/core/ext/transport/chttp2/transport/internal.h b/src/core/ext/transport/chttp2/transport/internal.h index 4e0c31c111..776017bee2 100644 --- a/src/core/ext/transport/chttp2/transport/internal.h +++ b/src/core/ext/transport/chttp2/transport/internal.h @@ -454,8 +454,10 @@ typedef struct { bool seen_error; bool exceeded_metadata_size; - /** the error that resulted in this stream being removed */ - grpc_error *removal_error; + /** the error that resulted in this stream being read-closed */ + grpc_error *read_closed_error; + /** the error that resulted in this stream being write-closed */ + grpc_error *write_closed_error; bool published_initial_metadata; bool published_trailing_metadata; diff --git a/src/core/lib/iomgr/workqueue.h b/src/core/lib/iomgr/workqueue.h index b4bd5742dd..9f7219ebf1 100644 --- a/src/core/lib/iomgr/workqueue.h +++ b/src/core/lib/iomgr/workqueue.h @@ -52,7 +52,7 @@ void grpc_workqueue_flush(grpc_exec_ctx *exec_ctx, grpc_workqueue *workqueue); -/*#define GRPC_WORKQUEUE_REFCOUNT_DEBUG*/ +//#define GRPC_WORKQUEUE_REFCOUNT_DEBUG #ifdef GRPC_WORKQUEUE_REFCOUNT_DEBUG #define GRPC_WORKQUEUE_REF(p, r) \ (grpc_workqueue_ref((p), __FILE__, __LINE__, (r)), (p)) -- cgit v1.2.3 From c0e73da8c22550070f328a6fd168abced87342d7 Mon Sep 17 00:00:00 2001 From: Craig Tiller Date: Thu, 7 Jul 2016 09:43:24 -0700 Subject: Fix flow control issue, make debugging in the future easier --- .../transport/chttp2/transport/chttp2_transport.c | 47 +++++++++++----------- src/core/ext/transport/chttp2/transport/internal.h | 7 ++-- src/core/ext/transport/chttp2/transport/parsing.c | 6 +-- .../ext/transport/chttp2/transport/stream_lists.c | 12 ++---- src/core/ext/transport/chttp2/transport/writing.c | 6 +-- 5 files changed, 36 insertions(+), 42 deletions(-) (limited to 'src/core/ext') diff --git a/src/core/ext/transport/chttp2/transport/chttp2_transport.c b/src/core/ext/transport/chttp2/transport/chttp2_transport.c index 8b7f91d3af..aeedf417b4 100644 --- a/src/core/ext/transport/chttp2/transport/chttp2_transport.c +++ b/src/core/ext/transport/chttp2/transport/chttp2_transport.c @@ -295,7 +295,7 @@ static void init_transport(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t, gpr_slice_buffer_add( &t->global.qbuf, gpr_slice_from_copied_string(GRPC_CHTTP2_CLIENT_CONNECT_STRING)); - grpc_chttp2_initiate_write(exec_ctx, &t->global, false); + grpc_chttp2_initiate_write(exec_ctx, &t->global, false, "initial_write"); } /* 8 is a random stab in the dark as to a good initial size: it's small enough that it shouldn't waste memory for infrequently used connections, yet @@ -799,14 +799,14 @@ void grpc_chttp2_run_with_global_lock(grpc_exec_ctx *exec_ctx, void grpc_chttp2_initiate_write(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport_global *transport_global, - bool covered_by_poller) { + bool covered_by_poller, const char *reason) { grpc_chttp2_transport *t = TRANSPORT_FROM_GLOBAL(transport_global); switch (t->executor.write_state) { case GRPC_CHTTP2_WRITING_INACTIVE: set_write_state(t, covered_by_poller ? GRPC_CHTTP2_WRITE_REQUESTED_WITH_POLLER : GRPC_CHTTP2_WRITE_REQUESTED_NO_POLLER, - "initiate_write"); + reason); break; case GRPC_CHTTP2_WRITE_REQUESTED_WITH_POLLER: /* nothing to do: write already requested */ @@ -815,7 +815,7 @@ void grpc_chttp2_initiate_write(grpc_exec_ctx *exec_ctx, if (covered_by_poller) { /* upgrade to note poller is available to cover the write */ set_write_state(t, GRPC_CHTTP2_WRITE_REQUESTED_WITH_POLLER, - "initiate_write"); + reason); } break; case GRPC_CHTTP2_WRITE_SCHEDULED: @@ -825,7 +825,7 @@ void grpc_chttp2_initiate_write(grpc_exec_ctx *exec_ctx, set_write_state(t, covered_by_poller ? GRPC_CHTTP2_WRITING_STALE_WITH_POLLER : GRPC_CHTTP2_WRITING_STALE_NO_POLLER, - "initiate_write"); + reason); break; case GRPC_CHTTP2_WRITING_STALE_WITH_POLLER: /* nothing to do: write already requested */ @@ -834,7 +834,7 @@ void grpc_chttp2_initiate_write(grpc_exec_ctx *exec_ctx, if (covered_by_poller) { /* upgrade to note poller is available to cover the write */ set_write_state(t, GRPC_CHTTP2_WRITING_STALE_WITH_POLLER, - "initiate_write"); + reason); } break; } @@ -881,11 +881,11 @@ static void initiate_writing(grpc_exec_ctx *exec_ctx, void *arg, void grpc_chttp2_become_writable(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport_global *transport_global, grpc_chttp2_stream_global *stream_global, - bool covered_by_poller) { + bool covered_by_poller, const char *reason) { if (!TRANSPORT_FROM_GLOBAL(transport_global)->closed && grpc_chttp2_list_add_writable_stream(transport_global, stream_global)) { GRPC_CHTTP2_STREAM_REF(stream_global, "chttp2_writing"); - grpc_chttp2_initiate_write(exec_ctx, transport_global, covered_by_poller); + grpc_chttp2_initiate_write(exec_ctx, transport_global, covered_by_poller, reason); } } @@ -901,7 +901,7 @@ static void push_setting(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t, if (use_value != t->global.settings[GRPC_LOCAL_SETTINGS][id]) { t->global.settings[GRPC_LOCAL_SETTINGS][id] = use_value; t->global.dirtied_local_settings = 1; - grpc_chttp2_initiate_write(exec_ctx, &t->global, false); + grpc_chttp2_initiate_write(exec_ctx, &t->global, false, "push_setting"); } } @@ -1040,7 +1040,7 @@ static void maybe_start_some_streams( stream_global->in_stream_map = true; transport_global->concurrent_stream_count++; grpc_chttp2_become_writable(exec_ctx, transport_global, stream_global, - true); + true, "new_stream"); } /* cancel out streams that will never be started */ while (transport_global->next_stream_id >= MAX_CLIENT_STREAM_ID && @@ -1176,7 +1176,7 @@ static void perform_stream_op_locked(grpc_exec_ctx *exec_ctx, } else { GPR_ASSERT(stream_global->id != 0); grpc_chttp2_become_writable(exec_ctx, transport_global, stream_global, - true); + true, "op.send_initial_metadata"); } } else { stream_global->send_trailing_metadata = NULL; @@ -1202,7 +1202,7 @@ static void perform_stream_op_locked(grpc_exec_ctx *exec_ctx, stream_global->send_message = op->send_message; if (stream_global->id != 0) { grpc_chttp2_become_writable(exec_ctx, transport_global, stream_global, - true); + true, "op.send_message"); } } } @@ -1247,7 +1247,7 @@ static void perform_stream_op_locked(grpc_exec_ctx *exec_ctx, /* TODO(ctiller): check if there's flow control for any outstanding bytes before going writable */ grpc_chttp2_become_writable(exec_ctx, transport_global, stream_global, - true); + true, "op.send_trailing_metadata"); } } } @@ -1313,7 +1313,7 @@ static void send_ping_locked(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t, p->id[7] = (uint8_t)(t->global.ping_counter & 0xff); p->on_recv = on_recv; gpr_slice_buffer_add(&t->global.qbuf, grpc_chttp2_ping_create(0, p->id)); - grpc_chttp2_initiate_write(exec_ctx, &t->global, true); + grpc_chttp2_initiate_write(exec_ctx, &t->global, true, "send_ping"); } static void ack_ping_locked(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t, @@ -1373,7 +1373,7 @@ static void perform_transport_op_locked(grpc_exec_ctx *exec_ctx, close_transport = grpc_chttp2_has_streams(t) ? GRPC_ERROR_NONE : GRPC_ERROR_CREATE("GOAWAY sent"); - grpc_chttp2_initiate_write(exec_ctx, &t->global, false); + grpc_chttp2_initiate_write(exec_ctx, &t->global, false, "goaway_sent"); } if (op->set_accept_stream) { @@ -1578,7 +1578,7 @@ static void cancel_from_api(grpc_exec_ctx *exec_ctx, &transport_global->qbuf, grpc_chttp2_rst_stream_create(stream_global->id, (uint32_t)http_error, &stream_global->stats.outgoing)); - grpc_chttp2_initiate_write(exec_ctx, transport_global, false); + grpc_chttp2_initiate_write(exec_ctx, transport_global, false, "rst_stream"); } const char *msg = @@ -1844,7 +1844,7 @@ static void close_from_api(grpc_exec_ctx *exec_ctx, grpc_chttp2_mark_stream_closed(exec_ctx, transport_global, stream_global, 1, 1, error); - grpc_chttp2_initiate_write(exec_ctx, transport_global, false); + grpc_chttp2_initiate_write(exec_ctx, transport_global, false, "close_from_api"); } typedef struct { @@ -1896,7 +1896,7 @@ static void update_global_window(void *args, uint32_t id, void *stream) { if (was_zero && !is_zero) { grpc_chttp2_become_writable(a->exec_ctx, transport_global, stream_global, - true); + true, "update_global_window"); } } @@ -2007,7 +2007,7 @@ static void post_parse_locked(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t, /* copy parsing qbuf to global qbuf */ if (t->parsing.qbuf.count > 0) { gpr_slice_buffer_move_into(&t->parsing.qbuf, &t->global.qbuf); - grpc_chttp2_initiate_write(exec_ctx, transport_global, false); + grpc_chttp2_initiate_write(exec_ctx, transport_global, false, "parsing_qbuf"); } /* merge stream lists */ grpc_chttp2_stream_map_move_into(&t->new_stream_map, &t->parsing_stream_map); @@ -2176,6 +2176,7 @@ static void incoming_byte_stream_update_flow_control( if (stream_global->max_recv_bytes < max_recv_bytes) { uint32_t add_max_recv_bytes = max_recv_bytes - stream_global->max_recv_bytes; + gpr_log(GPR_DEBUG, "add_max_recv_bytes:%d", add_max_recv_bytes); GRPC_CHTTP2_FLOW_CREDIT_STREAM("op", transport_global, stream_global, max_recv_bytes, add_max_recv_bytes); GRPC_CHTTP2_FLOW_CREDIT_STREAM("op", transport_global, stream_global, @@ -2187,7 +2188,7 @@ static void incoming_byte_stream_update_flow_control( grpc_chttp2_list_add_unannounced_incoming_window_available(transport_global, stream_global); grpc_chttp2_become_writable(exec_ctx, transport_global, stream_global, - false); + false, "read_incoming_stream"); } } @@ -2395,7 +2396,7 @@ static char *format_flowctl_context_var(const char *context, const char *var, if (context == NULL) { *scope = NULL; gpr_asprintf(&buf, "%s(%" PRId64 ")", var, val); - result = gpr_leftpad(buf, ' ', 40); + result = gpr_leftpad(buf, ' ', 60); gpr_free(buf); return result; } @@ -2408,7 +2409,7 @@ static char *format_flowctl_context_var(const char *context, const char *var, gpr_free(tmp); } gpr_asprintf(&buf, "%s.%s(%" PRId64 ")", underscore_pos + 1, var, val); - result = gpr_leftpad(buf, ' ', 40); + result = gpr_leftpad(buf, ' ', 60); gpr_free(buf); return result; } @@ -2441,7 +2442,7 @@ void grpc_chttp2_flowctl_trace(const char *file, int line, const char *phase, tmp_phase = gpr_leftpad(phase, ' ', 8); tmp_scope1 = gpr_leftpad(scope1, ' ', 11); - gpr_asprintf(&prefix, "FLOW %s: %s %s ", phase, clisvr, scope1); + gpr_asprintf(&prefix, "FLOW %s: %s %s ", tmp_phase, clisvr, scope1); gpr_free(tmp_phase); gpr_free(tmp_scope1); diff --git a/src/core/ext/transport/chttp2/transport/internal.h b/src/core/ext/transport/chttp2/transport/internal.h index 776017bee2..2a12afad6c 100644 --- a/src/core/ext/transport/chttp2/transport/internal.h +++ b/src/core/ext/transport/chttp2/transport/internal.h @@ -542,7 +542,7 @@ struct grpc_chttp2_stream { should be performed. */ void grpc_chttp2_initiate_write(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport_global *transport_global, - bool covered_by_poller); + bool covered_by_poller, const char *reason); /** Someone is unlocking the transport mutex: check to see if writes are required, and schedule them if so */ @@ -631,8 +631,7 @@ void grpc_chttp2_list_add_writing_stalled_by_transport( grpc_chttp2_transport_writing *transport_writing, grpc_chttp2_stream_writing *stream_writing); void grpc_chttp2_list_flush_writing_stalled_by_transport( - grpc_exec_ctx *exec_ctx, grpc_chttp2_transport_writing *transport_writing, - bool is_window_available); + grpc_exec_ctx *exec_ctx, grpc_chttp2_transport_writing *transport_writing); void grpc_chttp2_list_add_stalled_by_transport( grpc_chttp2_transport_writing *transport_writing, @@ -845,6 +844,6 @@ void grpc_chttp2_ack_ping(grpc_exec_ctx *exec_ctx, void grpc_chttp2_become_writable(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport_global *transport_global, grpc_chttp2_stream_global *stream_global, - bool covered_by_poller); + bool covered_by_poller, const char *reason); #endif /* GRPC_CORE_EXT_TRANSPORT_CHTTP2_TRANSPORT_INTERNAL_H */ diff --git a/src/core/ext/transport/chttp2/transport/parsing.c b/src/core/ext/transport/chttp2/transport/parsing.c index b42d98b3b0..efc27775f0 100644 --- a/src/core/ext/transport/chttp2/transport/parsing.c +++ b/src/core/ext/transport/chttp2/transport/parsing.c @@ -157,7 +157,7 @@ void grpc_chttp2_publish_reads( while (grpc_chttp2_list_pop_stalled_by_transport(transport_global, &stream_global)) { grpc_chttp2_become_writable(exec_ctx, transport_global, stream_global, - false); + false, "transport.read_flow_control"); } } @@ -169,7 +169,7 @@ void grpc_chttp2_publish_reads( announce_incoming_window, announce_bytes); GRPC_CHTTP2_FLOW_CREDIT_TRANSPORT("parsed", transport_parsing, incoming_window, announce_bytes); - grpc_chttp2_initiate_write(exec_ctx, transport_global, false); + grpc_chttp2_initiate_write(exec_ctx, transport_global, false, "global incoming window"); } /* for each stream that saw an update, fixup global state */ @@ -193,7 +193,7 @@ void grpc_chttp2_publish_reads( is_zero = stream_global->outgoing_window <= 0; if (was_zero && !is_zero) { grpc_chttp2_become_writable(exec_ctx, transport_global, stream_global, - false); + false, "stream.read_flow_control"); } stream_global->max_recv_bytes -= (uint32_t)GPR_MIN( diff --git a/src/core/ext/transport/chttp2/transport/stream_lists.c b/src/core/ext/transport/chttp2/transport/stream_lists.c index f227a0af98..aaa4768c7b 100644 --- a/src/core/ext/transport/chttp2/transport/stream_lists.c +++ b/src/core/ext/transport/chttp2/transport/stream_lists.c @@ -337,19 +337,13 @@ void grpc_chttp2_list_add_writing_stalled_by_transport( } void grpc_chttp2_list_flush_writing_stalled_by_transport( - grpc_exec_ctx *exec_ctx, grpc_chttp2_transport_writing *transport_writing, - bool is_window_available) { + grpc_exec_ctx *exec_ctx, grpc_chttp2_transport_writing *transport_writing) { grpc_chttp2_stream *stream; grpc_chttp2_transport *transport = TRANSPORT_FROM_WRITING(transport_writing); while (stream_list_pop(transport, &stream, GRPC_CHTTP2_LIST_WRITING_STALLED_BY_TRANSPORT)) { - if (is_window_available) { - grpc_chttp2_become_writable(exec_ctx, &transport->global, &stream->global, - true); - } else { - grpc_chttp2_list_add_stalled_by_transport(transport_writing, - &stream->writing); - } + grpc_chttp2_list_add_stalled_by_transport(transport_writing, + &stream->writing); GRPC_CHTTP2_STREAM_UNREF(exec_ctx, &stream->global, "chttp2_writing_stalled"); } diff --git a/src/core/ext/transport/chttp2/transport/writing.c b/src/core/ext/transport/chttp2/transport/writing.c index b19f5f068d..cbc57d10ad 100644 --- a/src/core/ext/transport/chttp2/transport/writing.c +++ b/src/core/ext/transport/chttp2/transport/writing.c @@ -75,9 +75,6 @@ int grpc_chttp2_unlocking_check_writes( GRPC_CHTTP2_FLOW_MOVE_TRANSPORT("write", transport_writing, outgoing_window, transport_global, outgoing_window); - bool is_window_available = transport_writing->outgoing_window > 0; - grpc_chttp2_list_flush_writing_stalled_by_transport( - exec_ctx, transport_writing, is_window_available); /* for each grpc_chttp2_stream that's become writable, frame it's data (according to available window sizes) and add to the output buffer */ @@ -331,6 +328,9 @@ void grpc_chttp2_cleanup_writing( grpc_chttp2_stream_writing *stream_writing; grpc_chttp2_stream_global *stream_global; + grpc_chttp2_list_flush_writing_stalled_by_transport(exec_ctx, + transport_writing); + while (grpc_chttp2_list_pop_written_stream( transport_global, transport_writing, &stream_global, &stream_writing)) { if (stream_writing->sent_initial_metadata) { -- cgit v1.2.3 From e940d30f4cd99616329164cfe58ee80d4dcb824b Mon Sep 17 00:00:00 2001 From: Craig Tiller Date: Thu, 7 Jul 2016 09:45:49 -0700 Subject: Remove spam --- .../transport/chttp2/transport/chttp2_transport.c | 23 +++++++++++----------- 1 file changed, 12 insertions(+), 11 deletions(-) (limited to 'src/core/ext') diff --git a/src/core/ext/transport/chttp2/transport/chttp2_transport.c b/src/core/ext/transport/chttp2/transport/chttp2_transport.c index aeedf417b4..41506094de 100644 --- a/src/core/ext/transport/chttp2/transport/chttp2_transport.c +++ b/src/core/ext/transport/chttp2/transport/chttp2_transport.c @@ -814,8 +814,7 @@ void grpc_chttp2_initiate_write(grpc_exec_ctx *exec_ctx, case GRPC_CHTTP2_WRITE_REQUESTED_NO_POLLER: if (covered_by_poller) { /* upgrade to note poller is available to cover the write */ - set_write_state(t, GRPC_CHTTP2_WRITE_REQUESTED_WITH_POLLER, - reason); + set_write_state(t, GRPC_CHTTP2_WRITE_REQUESTED_WITH_POLLER, reason); } break; case GRPC_CHTTP2_WRITE_SCHEDULED: @@ -833,8 +832,7 @@ void grpc_chttp2_initiate_write(grpc_exec_ctx *exec_ctx, case GRPC_CHTTP2_WRITING_STALE_NO_POLLER: if (covered_by_poller) { /* upgrade to note poller is available to cover the write */ - set_write_state(t, GRPC_CHTTP2_WRITING_STALE_WITH_POLLER, - reason); + set_write_state(t, GRPC_CHTTP2_WRITING_STALE_WITH_POLLER, reason); } break; } @@ -885,7 +883,8 @@ void grpc_chttp2_become_writable(grpc_exec_ctx *exec_ctx, if (!TRANSPORT_FROM_GLOBAL(transport_global)->closed && grpc_chttp2_list_add_writable_stream(transport_global, stream_global)) { GRPC_CHTTP2_STREAM_REF(stream_global, "chttp2_writing"); - grpc_chttp2_initiate_write(exec_ctx, transport_global, covered_by_poller, reason); + grpc_chttp2_initiate_write(exec_ctx, transport_global, covered_by_poller, + reason); } } @@ -1039,8 +1038,8 @@ static void maybe_start_some_streams( stream_global->id, STREAM_FROM_GLOBAL(stream_global)); stream_global->in_stream_map = true; transport_global->concurrent_stream_count++; - grpc_chttp2_become_writable(exec_ctx, transport_global, stream_global, - true, "new_stream"); + grpc_chttp2_become_writable(exec_ctx, transport_global, stream_global, true, + "new_stream"); } /* cancel out streams that will never be started */ while (transport_global->next_stream_id >= MAX_CLIENT_STREAM_ID && @@ -1578,7 +1577,8 @@ static void cancel_from_api(grpc_exec_ctx *exec_ctx, &transport_global->qbuf, grpc_chttp2_rst_stream_create(stream_global->id, (uint32_t)http_error, &stream_global->stats.outgoing)); - grpc_chttp2_initiate_write(exec_ctx, transport_global, false, "rst_stream"); + grpc_chttp2_initiate_write(exec_ctx, transport_global, false, + "rst_stream"); } const char *msg = @@ -1844,7 +1844,8 @@ static void close_from_api(grpc_exec_ctx *exec_ctx, grpc_chttp2_mark_stream_closed(exec_ctx, transport_global, stream_global, 1, 1, error); - grpc_chttp2_initiate_write(exec_ctx, transport_global, false, "close_from_api"); + grpc_chttp2_initiate_write(exec_ctx, transport_global, false, + "close_from_api"); } typedef struct { @@ -2007,7 +2008,8 @@ static void post_parse_locked(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t, /* copy parsing qbuf to global qbuf */ if (t->parsing.qbuf.count > 0) { gpr_slice_buffer_move_into(&t->parsing.qbuf, &t->global.qbuf); - grpc_chttp2_initiate_write(exec_ctx, transport_global, false, "parsing_qbuf"); + grpc_chttp2_initiate_write(exec_ctx, transport_global, false, + "parsing_qbuf"); } /* merge stream lists */ grpc_chttp2_stream_map_move_into(&t->new_stream_map, &t->parsing_stream_map); @@ -2176,7 +2178,6 @@ static void incoming_byte_stream_update_flow_control( if (stream_global->max_recv_bytes < max_recv_bytes) { uint32_t add_max_recv_bytes = max_recv_bytes - stream_global->max_recv_bytes; - gpr_log(GPR_DEBUG, "add_max_recv_bytes:%d", add_max_recv_bytes); GRPC_CHTTP2_FLOW_CREDIT_STREAM("op", transport_global, stream_global, max_recv_bytes, add_max_recv_bytes); GRPC_CHTTP2_FLOW_CREDIT_STREAM("op", transport_global, stream_global, -- cgit v1.2.3 From 6c8619bbe7a0eb8ca65782886e8253ebbec87b54 Mon Sep 17 00:00:00 2001 From: Craig Tiller Date: Thu, 7 Jul 2016 10:41:48 -0700 Subject: Better fix for flow control bug --- src/core/ext/transport/chttp2/transport/internal.h | 2 +- src/core/ext/transport/chttp2/transport/parsing.c | 14 ++++++-------- src/core/ext/transport/chttp2/transport/stream_lists.c | 9 ++++++++- src/core/ext/transport/chttp2/transport/writing.c | 14 ++++++++++++-- test/cpp/end2end/end2end_test.cc | 3 +++ 5 files changed, 30 insertions(+), 12 deletions(-) (limited to 'src/core/ext') diff --git a/src/core/ext/transport/chttp2/transport/internal.h b/src/core/ext/transport/chttp2/transport/internal.h index 2a12afad6c..6b47d702ae 100644 --- a/src/core/ext/transport/chttp2/transport/internal.h +++ b/src/core/ext/transport/chttp2/transport/internal.h @@ -630,7 +630,7 @@ int grpc_chttp2_list_pop_check_read_ops( void grpc_chttp2_list_add_writing_stalled_by_transport( grpc_chttp2_transport_writing *transport_writing, grpc_chttp2_stream_writing *stream_writing); -void grpc_chttp2_list_flush_writing_stalled_by_transport( +bool grpc_chttp2_list_flush_writing_stalled_by_transport( grpc_exec_ctx *exec_ctx, grpc_chttp2_transport_writing *transport_writing); void grpc_chttp2_list_add_stalled_by_transport( diff --git a/src/core/ext/transport/chttp2/transport/parsing.c b/src/core/ext/transport/chttp2/transport/parsing.c index efc27775f0..a8ce1db847 100644 --- a/src/core/ext/transport/chttp2/transport/parsing.c +++ b/src/core/ext/transport/chttp2/transport/parsing.c @@ -87,8 +87,8 @@ void grpc_chttp2_prepare_to_read( transport_global->settings[GRPC_SENT_SETTINGS], sizeof(transport_parsing->last_sent_settings)); transport_parsing->max_frame_size = - transport_global->settings[GRPC_ACKED_SETTINGS] - [GRPC_CHTTP2_SETTINGS_MAX_FRAME_SIZE]; + transport_global + ->settings[GRPC_ACKED_SETTINGS][GRPC_CHTTP2_SETTINGS_MAX_FRAME_SIZE]; /* update the parsing view of incoming window */ while (grpc_chttp2_list_pop_unannounced_incoming_window_available( @@ -154,11 +154,8 @@ void grpc_chttp2_publish_reads( transport_parsing, outgoing_window); is_zero = transport_global->outgoing_window <= 0; if (was_zero && !is_zero) { - while (grpc_chttp2_list_pop_stalled_by_transport(transport_global, - &stream_global)) { - grpc_chttp2_become_writable(exec_ctx, transport_global, stream_global, - false, "transport.read_flow_control"); - } + grpc_chttp2_initiate_write(exec_ctx, transport_global, false, + "new_global_flow_control"); } if (transport_parsing->incoming_window < @@ -169,7 +166,8 @@ void grpc_chttp2_publish_reads( announce_incoming_window, announce_bytes); GRPC_CHTTP2_FLOW_CREDIT_TRANSPORT("parsed", transport_parsing, incoming_window, announce_bytes); - grpc_chttp2_initiate_write(exec_ctx, transport_global, false, "global incoming window"); + grpc_chttp2_initiate_write(exec_ctx, transport_global, false, + "global incoming window"); } /* for each stream that saw an update, fixup global state */ diff --git a/src/core/ext/transport/chttp2/transport/stream_lists.c b/src/core/ext/transport/chttp2/transport/stream_lists.c index aaa4768c7b..2eb5f5f632 100644 --- a/src/core/ext/transport/chttp2/transport/stream_lists.c +++ b/src/core/ext/transport/chttp2/transport/stream_lists.c @@ -329,6 +329,7 @@ void grpc_chttp2_list_add_writing_stalled_by_transport( grpc_chttp2_transport_writing *transport_writing, grpc_chttp2_stream_writing *stream_writing) { grpc_chttp2_stream *stream = STREAM_FROM_WRITING(stream_writing); + gpr_log(GPR_DEBUG, "writing stalled %d", stream->global.id); if (!stream->included[GRPC_CHTTP2_LIST_WRITING_STALLED_BY_TRANSPORT]) { GRPC_CHTTP2_STREAM_REF(&stream->global, "chttp2_writing_stalled"); } @@ -336,22 +337,28 @@ void grpc_chttp2_list_add_writing_stalled_by_transport( GRPC_CHTTP2_LIST_WRITING_STALLED_BY_TRANSPORT); } -void grpc_chttp2_list_flush_writing_stalled_by_transport( +bool grpc_chttp2_list_flush_writing_stalled_by_transport( grpc_exec_ctx *exec_ctx, grpc_chttp2_transport_writing *transport_writing) { grpc_chttp2_stream *stream; + bool out = false; grpc_chttp2_transport *transport = TRANSPORT_FROM_WRITING(transport_writing); while (stream_list_pop(transport, &stream, GRPC_CHTTP2_LIST_WRITING_STALLED_BY_TRANSPORT)) { + gpr_log(GPR_DEBUG, "move %d from writing stalled to just stalled", + stream->global.id); grpc_chttp2_list_add_stalled_by_transport(transport_writing, &stream->writing); GRPC_CHTTP2_STREAM_UNREF(exec_ctx, &stream->global, "chttp2_writing_stalled"); + out = true; } + return out; } void grpc_chttp2_list_add_stalled_by_transport( grpc_chttp2_transport_writing *transport_writing, grpc_chttp2_stream_writing *stream_writing) { + gpr_log(GPR_DEBUG, "stalled %d", stream_writing->id); stream_list_add(TRANSPORT_FROM_WRITING(transport_writing), STREAM_FROM_WRITING(stream_writing), GRPC_CHTTP2_LIST_STALLED_BY_TRANSPORT); diff --git a/src/core/ext/transport/chttp2/transport/writing.c b/src/core/ext/transport/chttp2/transport/writing.c index cbc57d10ad..e0d87725e9 100644 --- a/src/core/ext/transport/chttp2/transport/writing.c +++ b/src/core/ext/transport/chttp2/transport/writing.c @@ -75,6 +75,13 @@ int grpc_chttp2_unlocking_check_writes( GRPC_CHTTP2_FLOW_MOVE_TRANSPORT("write", transport_writing, outgoing_window, transport_global, outgoing_window); + if (transport_writing->outgoing_window > 0) { + while (grpc_chttp2_list_pop_stalled_by_transport(transport_global, + &stream_global)) { + grpc_chttp2_become_writable(exec_ctx, transport_global, stream_global, + false, "transport.read_flow_control"); + } + } /* for each grpc_chttp2_stream that's become writable, frame it's data (according to available window sizes) and add to the output buffer */ @@ -328,8 +335,11 @@ void grpc_chttp2_cleanup_writing( grpc_chttp2_stream_writing *stream_writing; grpc_chttp2_stream_global *stream_global; - grpc_chttp2_list_flush_writing_stalled_by_transport(exec_ctx, - transport_writing); + if (grpc_chttp2_list_flush_writing_stalled_by_transport(exec_ctx, + transport_writing)) { + grpc_chttp2_initiate_write(exec_ctx, transport_global, false, + "resume_stalled_stream"); + } while (grpc_chttp2_list_pop_written_stream( transport_global, transport_writing, &stream_global, &stream_writing)) { diff --git a/test/cpp/end2end/end2end_test.cc b/test/cpp/end2end/end2end_test.cc index 354a59cedd..0f87ae3e44 100644 --- a/test/cpp/end2end/end2end_test.cc +++ b/test/cpp/end2end/end2end_test.cc @@ -1166,6 +1166,9 @@ TEST_P(ProxyEnd2endTest, HugeResponse) { request.mutable_param()->set_response_message_length(kResponseSize); ClientContext context; + std::chrono::system_clock::time_point deadline = + std::chrono::system_clock::now() + std::chrono::seconds(20); + context.set_deadline(deadline); Status s = stub_->Echo(&context, request, &response); EXPECT_EQ(kResponseSize, response.message().size()); EXPECT_TRUE(s.ok()); -- cgit v1.2.3 From 8e8027bad6c4b4720a27e29178a1431fc069f86a Mon Sep 17 00:00:00 2001 From: Craig Tiller Date: Thu, 7 Jul 2016 10:42:09 -0700 Subject: clang-format --- src/core/ext/transport/chttp2/transport/parsing.c | 4 ++-- src/core/lib/iomgr/ev_epoll_linux.c | 2 +- 2 files changed, 3 insertions(+), 3 deletions(-) (limited to 'src/core/ext') diff --git a/src/core/ext/transport/chttp2/transport/parsing.c b/src/core/ext/transport/chttp2/transport/parsing.c index a8ce1db847..fbb44ec54a 100644 --- a/src/core/ext/transport/chttp2/transport/parsing.c +++ b/src/core/ext/transport/chttp2/transport/parsing.c @@ -87,8 +87,8 @@ void grpc_chttp2_prepare_to_read( transport_global->settings[GRPC_SENT_SETTINGS], sizeof(transport_parsing->last_sent_settings)); transport_parsing->max_frame_size = - transport_global - ->settings[GRPC_ACKED_SETTINGS][GRPC_CHTTP2_SETTINGS_MAX_FRAME_SIZE]; + transport_global->settings[GRPC_ACKED_SETTINGS] + [GRPC_CHTTP2_SETTINGS_MAX_FRAME_SIZE]; /* update the parsing view of incoming window */ while (grpc_chttp2_list_pop_unannounced_incoming_window_available( diff --git a/src/core/lib/iomgr/ev_epoll_linux.c b/src/core/lib/iomgr/ev_epoll_linux.c index 0e6cba7e4f..4282d01a2b 100644 --- a/src/core/lib/iomgr/ev_epoll_linux.c +++ b/src/core/lib/iomgr/ev_epoll_linux.c @@ -1548,7 +1548,7 @@ retry: * polling_island fields in both fd and pollset to point to the merged * polling island. */ - + if (fd->orphaned) { gpr_mu_unlock(&fd->mu); gpr_mu_unlock(&pollset->mu); -- cgit v1.2.3 From c22e31fb053d7229aed69e43972cf0c94817a841 Mon Sep 17 00:00:00 2001 From: Craig Tiller Date: Fri, 8 Jul 2016 13:22:18 -0700 Subject: Make it more likely to correctly report deadline exceeded --- .../transport/chttp2/transport/chttp2_transport.c | 18 ++++-- src/core/ext/transport/chttp2/transport/internal.h | 2 + src/core/ext/transport/chttp2/transport/parsing.c | 11 ++-- .../transport/chttp2/transport/status_conversion.c | 10 +++- .../transport/chttp2/transport/status_conversion.h | 2 +- .../core/transport/chttp2/status_conversion_test.c | 68 ++++++++++++++++------ 6 files changed, 81 insertions(+), 30 deletions(-) (limited to 'src/core/ext') diff --git a/src/core/ext/transport/chttp2/transport/chttp2_transport.c b/src/core/ext/transport/chttp2/transport/chttp2_transport.c index 38e782b9b4..5aae753c07 100644 --- a/src/core/ext/transport/chttp2/transport/chttp2_transport.c +++ b/src/core/ext/transport/chttp2/transport/chttp2_transport.c @@ -513,6 +513,7 @@ static int init_stream(grpc_exec_ctx *exec_ctx, grpc_transport *gt, &s->global.received_trailing_metadata); grpc_chttp2_data_parser_init(&s->parsing.data_parser); gpr_slice_buffer_init(&s->writing.flow_controlled_buffer); + s->global.deadline = gpr_inf_future(GPR_CLOCK_MONOTONIC); REF_TRANSPORT(t, "stream"); @@ -988,6 +989,11 @@ static void perform_stream_op_locked(grpc_exec_ctx *exec_ctx, const size_t metadata_peer_limit = transport_global->settings[GRPC_PEER_SETTINGS] [GRPC_CHTTP2_SETTINGS_MAX_HEADER_LIST_SIZE]; + if (transport_global->is_client) { + stream_global->deadline = + gpr_time_min(stream_global->deadline, + stream_global->send_initial_metadata->deadline); + } if (metadata_size > metadata_peer_limit) { cancel_from_api( exec_ctx, transport_global, stream_global, @@ -1366,7 +1372,7 @@ static void remove_stream(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t, GRPC_ERROR_UNREF(error); } -static void status_codes_from_error(grpc_error *error, +static void status_codes_from_error(grpc_error *error, gpr_timespec deadline, grpc_chttp2_error_code *http2_error, grpc_status_code *grpc_status) { intptr_t ip_http; @@ -1386,8 +1392,8 @@ static void status_codes_from_error(grpc_error *error, if (have_grpc) { *grpc_status = (grpc_status_code)ip_grpc; } else if (have_http) { - *grpc_status = - grpc_chttp2_http2_error_to_grpc_status((grpc_chttp2_error_code)ip_http); + *grpc_status = grpc_chttp2_http2_error_to_grpc_status( + (grpc_chttp2_error_code)ip_http, deadline); } else { *grpc_status = GRPC_STATUS_INTERNAL; } @@ -1400,7 +1406,8 @@ static void cancel_from_api(grpc_exec_ctx *exec_ctx, if (!stream_global->read_closed || !stream_global->write_closed) { grpc_status_code grpc_status; grpc_chttp2_error_code http_error; - status_codes_from_error(due_to_error, &http_error, &grpc_status); + status_codes_from_error(due_to_error, stream_global->deadline, &http_error, + &grpc_status); if (stream_global->id != 0) { gpr_slice_buffer_add( @@ -1536,7 +1543,8 @@ static void close_from_api(grpc_exec_ctx *exec_ctx, uint32_t len = 0; grpc_status_code grpc_status; grpc_chttp2_error_code http_error; - status_codes_from_error(error, &http_error, &grpc_status); + status_codes_from_error(error, stream_global->deadline, &http_error, + &grpc_status); GPR_ASSERT(grpc_status >= 0 && (int)grpc_status < 100); diff --git a/src/core/ext/transport/chttp2/transport/internal.h b/src/core/ext/transport/chttp2/transport/internal.h index b5180c6fc8..8d79e93ceb 100644 --- a/src/core/ext/transport/chttp2/transport/internal.h +++ b/src/core/ext/transport/chttp2/transport/internal.h @@ -447,6 +447,8 @@ typedef struct { grpc_chttp2_incoming_metadata_buffer received_trailing_metadata; grpc_chttp2_incoming_frame_queue incoming_frames; + + gpr_timespec deadline; } grpc_chttp2_stream_global; typedef struct { diff --git a/src/core/ext/transport/chttp2/transport/parsing.c b/src/core/ext/transport/chttp2/transport/parsing.c index 991d7729af..c5240ce38a 100644 --- a/src/core/ext/transport/chttp2/transport/parsing.c +++ b/src/core/ext/transport/chttp2/transport/parsing.c @@ -87,8 +87,8 @@ void grpc_chttp2_prepare_to_read( transport_global->settings[GRPC_SENT_SETTINGS], sizeof(transport_parsing->last_sent_settings)); transport_parsing->max_frame_size = - transport_global->settings[GRPC_ACKED_SETTINGS] - [GRPC_CHTTP2_SETTINGS_MAX_FRAME_SIZE]; + transport_global + ->settings[GRPC_ACKED_SETTINGS][GRPC_CHTTP2_SETTINGS_MAX_FRAME_SIZE]; /* update the parsing view of incoming window */ while (grpc_chttp2_list_pop_unannounced_incoming_window_available( @@ -236,9 +236,10 @@ void grpc_chttp2_publish_reads( GRPC_ERROR_INT_HTTP2_ERROR, &reason); if (has_reason && reason != GRPC_CHTTP2_NO_ERROR) { grpc_status_code status_code = - has_reason ? grpc_chttp2_http2_error_to_grpc_status( - (grpc_chttp2_error_code)reason) - : GRPC_STATUS_INTERNAL; + has_reason + ? grpc_chttp2_http2_error_to_grpc_status( + (grpc_chttp2_error_code)reason, stream_global->deadline) + : GRPC_STATUS_INTERNAL; const char *status_details = grpc_error_string(stream_parsing->forced_close_error); gpr_slice slice_details = gpr_slice_from_copied_string(status_details); diff --git a/src/core/ext/transport/chttp2/transport/status_conversion.c b/src/core/ext/transport/chttp2/transport/status_conversion.c index c42fb9b3a1..5dce2f2d0c 100644 --- a/src/core/ext/transport/chttp2/transport/status_conversion.c +++ b/src/core/ext/transport/chttp2/transport/status_conversion.c @@ -39,6 +39,8 @@ int grpc_chttp2_grpc_status_to_http2_error(grpc_status_code status) { return GRPC_CHTTP2_NO_ERROR; case GRPC_STATUS_CANCELLED: return GRPC_CHTTP2_CANCEL; + case GRPC_STATUS_DEADLINE_EXCEEDED: + return GRPC_CHTTP2_CANCEL; case GRPC_STATUS_RESOURCE_EXHAUSTED: return GRPC_CHTTP2_ENHANCE_YOUR_CALM; case GRPC_STATUS_PERMISSION_DENIED: @@ -51,13 +53,17 @@ int grpc_chttp2_grpc_status_to_http2_error(grpc_status_code status) { } grpc_status_code grpc_chttp2_http2_error_to_grpc_status( - grpc_chttp2_error_code error) { + grpc_chttp2_error_code error, gpr_timespec deadline) { switch (error) { case GRPC_CHTTP2_NO_ERROR: /* should never be received */ return GRPC_STATUS_INTERNAL; case GRPC_CHTTP2_CANCEL: - return GRPC_STATUS_CANCELLED; + /* http2 cancel translates to STATUS_CANCELLED iff deadline hasn't been + * exceeded */ + return gpr_time_cmp(gpr_now(deadline.clock_type), deadline) >= 0 + ? GRPC_STATUS_DEADLINE_EXCEEDED + : GRPC_STATUS_CANCELLED; case GRPC_CHTTP2_ENHANCE_YOUR_CALM: return GRPC_STATUS_RESOURCE_EXHAUSTED; case GRPC_CHTTP2_INADEQUATE_SECURITY: diff --git a/src/core/ext/transport/chttp2/transport/status_conversion.h b/src/core/ext/transport/chttp2/transport/status_conversion.h index e7285e6fd5..953bc9f1e1 100644 --- a/src/core/ext/transport/chttp2/transport/status_conversion.h +++ b/src/core/ext/transport/chttp2/transport/status_conversion.h @@ -41,7 +41,7 @@ grpc_chttp2_error_code grpc_chttp2_grpc_status_to_http2_error( grpc_status_code status); grpc_status_code grpc_chttp2_http2_error_to_grpc_status( - grpc_chttp2_error_code error); + grpc_chttp2_error_code error, gpr_timespec deadline); /* Conversion of HTTP status codes (:status) to grpc status codes */ grpc_status_code grpc_chttp2_http2_status_to_grpc_status(int status); diff --git a/test/core/transport/chttp2/status_conversion_test.c b/test/core/transport/chttp2/status_conversion_test.c index e6fc785728..f5a5cd1395 100644 --- a/test/core/transport/chttp2/status_conversion_test.c +++ b/test/core/transport/chttp2/status_conversion_test.c @@ -37,8 +37,8 @@ #define GRPC_STATUS_TO_HTTP2_ERROR(a, b) \ GPR_ASSERT(grpc_chttp2_grpc_status_to_http2_error(a) == (b)) -#define HTTP2_ERROR_TO_GRPC_STATUS(a, b) \ - GPR_ASSERT(grpc_chttp2_http2_error_to_grpc_status(a) == (b)) +#define HTTP2_ERROR_TO_GRPC_STATUS(a, deadline, b) \ + GPR_ASSERT(grpc_chttp2_http2_error_to_grpc_status(a, deadline) == (b)) #define GRPC_STATUS_TO_HTTP2_STATUS(a, b) \ GPR_ASSERT(grpc_chttp2_grpc_status_to_http2_status(a) == (b)) #define HTTP2_STATUS_TO_GRPC_STATUS(a, b) \ @@ -54,8 +54,7 @@ int main(int argc, char **argv) { GRPC_STATUS_TO_HTTP2_ERROR(GRPC_STATUS_UNKNOWN, GRPC_CHTTP2_INTERNAL_ERROR); GRPC_STATUS_TO_HTTP2_ERROR(GRPC_STATUS_INVALID_ARGUMENT, GRPC_CHTTP2_INTERNAL_ERROR); - GRPC_STATUS_TO_HTTP2_ERROR(GRPC_STATUS_DEADLINE_EXCEEDED, - GRPC_CHTTP2_INTERNAL_ERROR); + GRPC_STATUS_TO_HTTP2_ERROR(GRPC_STATUS_DEADLINE_EXCEEDED, GRPC_CHTTP2_CANCEL); GRPC_STATUS_TO_HTTP2_ERROR(GRPC_STATUS_NOT_FOUND, GRPC_CHTTP2_INTERNAL_ERROR); GRPC_STATUS_TO_HTTP2_ERROR(GRPC_STATUS_ALREADY_EXISTS, GRPC_CHTTP2_INTERNAL_ERROR); @@ -95,25 +94,60 @@ int main(int argc, char **argv) { GRPC_STATUS_TO_HTTP2_STATUS(GRPC_STATUS_UNAVAILABLE, 200); GRPC_STATUS_TO_HTTP2_STATUS(GRPC_STATUS_DATA_LOSS, 200); - HTTP2_ERROR_TO_GRPC_STATUS(GRPC_CHTTP2_NO_ERROR, GRPC_STATUS_INTERNAL); - HTTP2_ERROR_TO_GRPC_STATUS(GRPC_CHTTP2_PROTOCOL_ERROR, GRPC_STATUS_INTERNAL); - HTTP2_ERROR_TO_GRPC_STATUS(GRPC_CHTTP2_INTERNAL_ERROR, GRPC_STATUS_INTERNAL); - HTTP2_ERROR_TO_GRPC_STATUS(GRPC_CHTTP2_FLOW_CONTROL_ERROR, + const gpr_timespec before_deadline = gpr_inf_future(GPR_CLOCK_MONOTONIC); + HTTP2_ERROR_TO_GRPC_STATUS(GRPC_CHTTP2_NO_ERROR, before_deadline, + GRPC_STATUS_INTERNAL); + HTTP2_ERROR_TO_GRPC_STATUS(GRPC_CHTTP2_PROTOCOL_ERROR, before_deadline, + GRPC_STATUS_INTERNAL); + HTTP2_ERROR_TO_GRPC_STATUS(GRPC_CHTTP2_INTERNAL_ERROR, before_deadline, + GRPC_STATUS_INTERNAL); + HTTP2_ERROR_TO_GRPC_STATUS(GRPC_CHTTP2_FLOW_CONTROL_ERROR, before_deadline, + GRPC_STATUS_INTERNAL); + HTTP2_ERROR_TO_GRPC_STATUS(GRPC_CHTTP2_SETTINGS_TIMEOUT, before_deadline, GRPC_STATUS_INTERNAL); - HTTP2_ERROR_TO_GRPC_STATUS(GRPC_CHTTP2_SETTINGS_TIMEOUT, + HTTP2_ERROR_TO_GRPC_STATUS(GRPC_CHTTP2_STREAM_CLOSED, before_deadline, GRPC_STATUS_INTERNAL); - HTTP2_ERROR_TO_GRPC_STATUS(GRPC_CHTTP2_STREAM_CLOSED, GRPC_STATUS_INTERNAL); - HTTP2_ERROR_TO_GRPC_STATUS(GRPC_CHTTP2_FRAME_SIZE_ERROR, + HTTP2_ERROR_TO_GRPC_STATUS(GRPC_CHTTP2_FRAME_SIZE_ERROR, before_deadline, GRPC_STATUS_INTERNAL); - HTTP2_ERROR_TO_GRPC_STATUS(GRPC_CHTTP2_REFUSED_STREAM, + HTTP2_ERROR_TO_GRPC_STATUS(GRPC_CHTTP2_REFUSED_STREAM, before_deadline, GRPC_STATUS_UNAVAILABLE); - HTTP2_ERROR_TO_GRPC_STATUS(GRPC_CHTTP2_CANCEL, GRPC_STATUS_CANCELLED); - HTTP2_ERROR_TO_GRPC_STATUS(GRPC_CHTTP2_COMPRESSION_ERROR, + HTTP2_ERROR_TO_GRPC_STATUS(GRPC_CHTTP2_CANCEL, before_deadline, + GRPC_STATUS_CANCELLED); + HTTP2_ERROR_TO_GRPC_STATUS(GRPC_CHTTP2_COMPRESSION_ERROR, before_deadline, + GRPC_STATUS_INTERNAL); + HTTP2_ERROR_TO_GRPC_STATUS(GRPC_CHTTP2_CONNECT_ERROR, before_deadline, + GRPC_STATUS_INTERNAL); + HTTP2_ERROR_TO_GRPC_STATUS(GRPC_CHTTP2_ENHANCE_YOUR_CALM, before_deadline, + GRPC_STATUS_RESOURCE_EXHAUSTED); + HTTP2_ERROR_TO_GRPC_STATUS(GRPC_CHTTP2_INADEQUATE_SECURITY, before_deadline, + GRPC_STATUS_PERMISSION_DENIED); + + const gpr_timespec after_deadline = gpr_inf_past(GPR_CLOCK_MONOTONIC); + HTTP2_ERROR_TO_GRPC_STATUS(GRPC_CHTTP2_NO_ERROR, after_deadline, + GRPC_STATUS_INTERNAL); + HTTP2_ERROR_TO_GRPC_STATUS(GRPC_CHTTP2_PROTOCOL_ERROR, after_deadline, + GRPC_STATUS_INTERNAL); + HTTP2_ERROR_TO_GRPC_STATUS(GRPC_CHTTP2_INTERNAL_ERROR, after_deadline, + GRPC_STATUS_INTERNAL); + HTTP2_ERROR_TO_GRPC_STATUS(GRPC_CHTTP2_FLOW_CONTROL_ERROR, after_deadline, + GRPC_STATUS_INTERNAL); + HTTP2_ERROR_TO_GRPC_STATUS(GRPC_CHTTP2_SETTINGS_TIMEOUT, after_deadline, + GRPC_STATUS_INTERNAL); + HTTP2_ERROR_TO_GRPC_STATUS(GRPC_CHTTP2_STREAM_CLOSED, after_deadline, + GRPC_STATUS_INTERNAL); + HTTP2_ERROR_TO_GRPC_STATUS(GRPC_CHTTP2_FRAME_SIZE_ERROR, after_deadline, + GRPC_STATUS_INTERNAL); + HTTP2_ERROR_TO_GRPC_STATUS(GRPC_CHTTP2_REFUSED_STREAM, after_deadline, + GRPC_STATUS_UNAVAILABLE); + HTTP2_ERROR_TO_GRPC_STATUS(GRPC_CHTTP2_CANCEL, after_deadline, + GRPC_STATUS_DEADLINE_EXCEEDED); + HTTP2_ERROR_TO_GRPC_STATUS(GRPC_CHTTP2_COMPRESSION_ERROR, after_deadline, + GRPC_STATUS_INTERNAL); + HTTP2_ERROR_TO_GRPC_STATUS(GRPC_CHTTP2_CONNECT_ERROR, after_deadline, GRPC_STATUS_INTERNAL); - HTTP2_ERROR_TO_GRPC_STATUS(GRPC_CHTTP2_CONNECT_ERROR, GRPC_STATUS_INTERNAL); - HTTP2_ERROR_TO_GRPC_STATUS(GRPC_CHTTP2_ENHANCE_YOUR_CALM, + HTTP2_ERROR_TO_GRPC_STATUS(GRPC_CHTTP2_ENHANCE_YOUR_CALM, after_deadline, GRPC_STATUS_RESOURCE_EXHAUSTED); - HTTP2_ERROR_TO_GRPC_STATUS(GRPC_CHTTP2_INADEQUATE_SECURITY, + HTTP2_ERROR_TO_GRPC_STATUS(GRPC_CHTTP2_INADEQUATE_SECURITY, after_deadline, GRPC_STATUS_PERMISSION_DENIED); HTTP2_STATUS_TO_GRPC_STATUS(200, GRPC_STATUS_OK); -- cgit v1.2.3 From 9a5b4348f843c567185d9fa36d189a1df1ab617b Mon Sep 17 00:00:00 2001 From: Craig Tiller Date: Mon, 11 Jul 2016 15:56:44 -0700 Subject: clang-format --- src/core/ext/transport/chttp2/transport/parsing.c | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) (limited to 'src/core/ext') diff --git a/src/core/ext/transport/chttp2/transport/parsing.c b/src/core/ext/transport/chttp2/transport/parsing.c index c5240ce38a..84eb5752f1 100644 --- a/src/core/ext/transport/chttp2/transport/parsing.c +++ b/src/core/ext/transport/chttp2/transport/parsing.c @@ -87,8 +87,8 @@ void grpc_chttp2_prepare_to_read( transport_global->settings[GRPC_SENT_SETTINGS], sizeof(transport_parsing->last_sent_settings)); transport_parsing->max_frame_size = - transport_global - ->settings[GRPC_ACKED_SETTINGS][GRPC_CHTTP2_SETTINGS_MAX_FRAME_SIZE]; + transport_global->settings[GRPC_ACKED_SETTINGS] + [GRPC_CHTTP2_SETTINGS_MAX_FRAME_SIZE]; /* update the parsing view of incoming window */ while (grpc_chttp2_list_pop_unannounced_incoming_window_available( -- cgit v1.2.3 From 609dc2d360a15e0fd5fb9d73f21054b53abb2a27 Mon Sep 17 00:00:00 2001 From: Craig Tiller Date: Tue, 12 Jul 2016 09:33:34 -0700 Subject: Fix rare flake --- src/core/ext/transport/chttp2/transport/chttp2_transport.c | 1 + 1 file changed, 1 insertion(+) (limited to 'src/core/ext') diff --git a/src/core/ext/transport/chttp2/transport/chttp2_transport.c b/src/core/ext/transport/chttp2/transport/chttp2_transport.c index 41506094de..0efc7ddb2f 100644 --- a/src/core/ext/transport/chttp2/transport/chttp2_transport.c +++ b/src/core/ext/transport/chttp2/transport/chttp2_transport.c @@ -1670,6 +1670,7 @@ static void fail_pending_writes(grpc_exec_ctx *exec_ctx, grpc_chttp2_stream_global *stream_global, grpc_error *error) { error = removal_error(error, stream_global); + stream_global->send_message = NULL; grpc_chttp2_complete_closure_step( exec_ctx, transport_global, stream_global, &stream_global->send_initial_metadata_finished, GRPC_ERROR_REF(error)); -- cgit v1.2.3 From 6ca5437b61fe32cbf2636ca436a8585ff66d574c Mon Sep 17 00:00:00 2001 From: Craig Tiller Date: Wed, 13 Jul 2016 10:41:49 -0700 Subject: Fix potential null ptr deref --- src/core/ext/transport/chttp2/transport/chttp2_transport.c | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) (limited to 'src/core/ext') diff --git a/src/core/ext/transport/chttp2/transport/chttp2_transport.c b/src/core/ext/transport/chttp2/transport/chttp2_transport.c index d5695fe49c..554a0f56dc 100644 --- a/src/core/ext/transport/chttp2/transport/chttp2_transport.c +++ b/src/core/ext/transport/chttp2/transport/chttp2_transport.c @@ -712,8 +712,9 @@ static void finish_global_actions(grpc_exec_ctx *exec_ctx, set_write_state(t, GRPC_CHTTP2_WRITE_SCHEDULED, "unlocking"); REF_TRANSPORT(t, "initiate_writing"); gpr_mu_unlock(&t->executor.mu); - grpc_exec_ctx_sched(exec_ctx, &t->initiate_writing, GRPC_ERROR_NONE, - grpc_endpoint_get_workqueue(t->ep)); + grpc_exec_ctx_sched( + exec_ctx, &t->initiate_writing, GRPC_ERROR_NONE, + t->ep != NULL ? grpc_endpoint_get_workqueue(t->ep) : NULL); break; case GRPC_CHTTP2_WRITE_REQUESTED_NO_POLLER: start_writing(exec_ctx, t); -- cgit v1.2.3 From ad1bd4602856e57c49f6f539456903dfdd021631 Mon Sep 17 00:00:00 2001 From: Craig Tiller Date: Wed, 13 Jul 2016 14:00:47 -0700 Subject: Failure reporting fix --- src/core/ext/transport/chttp2/client/secure/secure_channel_create.c | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) (limited to 'src/core/ext') diff --git a/src/core/ext/transport/chttp2/client/secure/secure_channel_create.c b/src/core/ext/transport/chttp2/client/secure/secure_channel_create.c index 721ba82d8f..9acacbd92d 100644 --- a/src/core/ext/transport/chttp2/client/secure/secure_channel_create.c +++ b/src/core/ext/transport/chttp2/client/secure/secure_channel_create.c @@ -91,11 +91,13 @@ static void on_secure_handshake_done(grpc_exec_ctx *exec_ctx, void *arg, connector *c = arg; grpc_closure *notify; gpr_mu_lock(&c->mu); + grpc_error *error = GRPC_ERROR_NONE; if (c->connecting_endpoint == NULL) { memset(c->result, 0, sizeof(*c->result)); gpr_mu_unlock(&c->mu); } else if (status != GRPC_SECURITY_OK) { - gpr_log(GPR_ERROR, "Secure handshake failed with error %d.", status); + error = grpc_error_set_int(GRPC_ERROR_CREATE("Secure handshake failed"), + GRPC_ERROR_INT_SECURITY_STATUS, status); memset(c->result, 0, sizeof(*c->result)); c->connecting_endpoint = NULL; gpr_mu_unlock(&c->mu); @@ -113,7 +115,7 @@ static void on_secure_handshake_done(grpc_exec_ctx *exec_ctx, void *arg, } notify = c->notify; c->notify = NULL; - grpc_exec_ctx_sched(exec_ctx, notify, GRPC_ERROR_NONE, NULL); + grpc_exec_ctx_sched(exec_ctx, notify, error, NULL); } static void on_initial_connect_string_sent(grpc_exec_ctx *exec_ctx, void *arg, -- cgit v1.2.3 From 6f59a86ae3894349b57ca535f786fa86177e4f3f Mon Sep 17 00:00:00 2001 From: Craig Tiller Date: Wed, 13 Jul 2016 20:14:55 -0700 Subject: Update comment --- src/core/ext/transport/chttp2/transport/internal.h | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) (limited to 'src/core/ext') diff --git a/src/core/ext/transport/chttp2/transport/internal.h b/src/core/ext/transport/chttp2/transport/internal.h index c06d94a0ba..71517c22f4 100644 --- a/src/core/ext/transport/chttp2/transport/internal.h +++ b/src/core/ext/transport/chttp2/transport/internal.h @@ -358,7 +358,8 @@ struct grpc_chttp2_transport { /** global state for reading/writing */ grpc_chttp2_transport_global global; /** state only accessible by the chain of execution that - set writing_active=1 */ + set writing_state >= GRPC_WRITING, and only by the writing closure + chain. */ grpc_chttp2_transport_writing writing; /** state only accessible by the chain of execution that set parsing_active=1 */ -- cgit v1.2.3 From 2ea0104fc48fc91159ea85c8dc50ec3c9227195b Mon Sep 17 00:00:00 2001 From: Craig Tiller Date: Wed, 13 Jul 2016 20:25:13 -0700 Subject: Add a quick note on the write chain execution\nthis is changed already in a future commit, so this should be enough for now --- .../ext/transport/chttp2/transport/chttp2_transport.c | 19 +++++++++++++++++++ src/core/ext/transport/chttp2/transport/internal.h | 5 ++++- 2 files changed, 23 insertions(+), 1 deletion(-) (limited to 'src/core/ext') diff --git a/src/core/ext/transport/chttp2/transport/chttp2_transport.c b/src/core/ext/transport/chttp2/transport/chttp2_transport.c index 554a0f56dc..be8a8f8498 100644 --- a/src/core/ext/transport/chttp2/transport/chttp2_transport.c +++ b/src/core/ext/transport/chttp2/transport/chttp2_transport.c @@ -802,6 +802,25 @@ void grpc_chttp2_run_with_global_lock(grpc_exec_ctx *exec_ctx, void grpc_chttp2_initiate_write(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport_global *transport_global, bool covered_by_poller, const char *reason) { + /* Perform state checks, and transition to a scheduled state if appropriate. + Each time we finish the global lock execution, we check if we need to + write. If we do: + - (if there is a poller surrounding the write) schedule + initiate_writing, which locks and calls initiate_writing_locked to... + - call start_writing, which verifies (under the global lock) that there + are things that need to be written by calling + grpc_chttp2_unlocking_check_writes, and if so schedules writing_action + against the current exec_ctx, to be executed OUTSIDE of the global lock + - eventually writing_action results in grpc_chttp2_terminate_writing being + called, which re-takes the global lock, updates state, checks if we need + to do *another* write immediately, and if so loops back to + start_writing. + + Current problems: + - too much lock entry/exiting + - the writing thread can become stuck indefinitely (punt through the + workqueue periodically to fix) */ + grpc_chttp2_transport *t = TRANSPORT_FROM_GLOBAL(transport_global); switch (t->executor.write_state) { case GRPC_CHTTP2_WRITING_INACTIVE: diff --git a/src/core/ext/transport/chttp2/transport/internal.h b/src/core/ext/transport/chttp2/transport/internal.h index 71517c22f4..e1dcf5262a 100644 --- a/src/core/ext/transport/chttp2/transport/internal.h +++ b/src/core/ext/transport/chttp2/transport/internal.h @@ -542,7 +542,10 @@ struct grpc_chttp2_stream { to write. The global lock is dropped and we do the syscall to write. After writing, a follow-up check is made to see if another round of writing - should be performed. */ + should be performed. + + The actual call chain is documented in the implementation of this function. + */ void grpc_chttp2_initiate_write(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport_global *transport_global, bool covered_by_poller, const char *reason); -- cgit v1.2.3