diff options
Diffstat (limited to 'src/core/ext/transport/chttp2')
4 files changed, 104 insertions, 44 deletions
diff --git a/src/core/ext/transport/chttp2/transport/chttp2_transport.cc b/src/core/ext/transport/chttp2/transport/chttp2_transport.cc index e4b19a2c4a..9462d1085e 100644 --- a/src/core/ext/transport/chttp2/transport/chttp2_transport.cc +++ b/src/core/ext/transport/chttp2/transport/chttp2_transport.cc @@ -152,10 +152,14 @@ static void close_transport_locked(grpc_exec_ctx *exec_ctx, static void end_all_the_calls(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t, grpc_error *error); +static void schedule_bdp_ping_locked(grpc_exec_ctx *exec_ctx, + grpc_chttp2_transport *t); static void start_bdp_ping_locked(grpc_exec_ctx *exec_ctx, void *tp, grpc_error *error); static void finish_bdp_ping_locked(grpc_exec_ctx *exec_ctx, void *tp, grpc_error *error); +static void next_bdp_ping_timer_expired_locked(grpc_exec_ctx *exec_ctx, + void *tp, grpc_error *error); static void cancel_pings(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t, grpc_error *error); @@ -218,6 +222,9 @@ static void destruct_transport(grpc_exec_ctx *exec_ctx, t->write_cb_pool = next; } + t->flow_control.bdp_estimator.Destroy(); + + GRPC_ERROR_UNREF(t->closed_with_error); gpr_free(t->ping_acks); gpr_free(t->peer_string); gpr_free(t); @@ -303,6 +310,9 @@ static void init_transport(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t, grpc_combiner_scheduler(t->combiner)); GRPC_CLOSURE_INIT(&t->finish_bdp_ping_locked, finish_bdp_ping_locked, t, grpc_combiner_scheduler(t->combiner)); + GRPC_CLOSURE_INIT(&t->next_bdp_ping_timer_expired_locked, + next_bdp_ping_timer_expired_locked, t, + grpc_combiner_scheduler(t->combiner)); GRPC_CLOSURE_INIT(&t->init_keepalive_ping_locked, init_keepalive_ping_locked, t, grpc_combiner_scheduler(t->combiner)); GRPC_CLOSURE_INIT(&t->start_keepalive_ping_locked, @@ -315,7 +325,7 @@ static void init_transport(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t, keepalive_watchdog_fired_locked, t, grpc_combiner_scheduler(t->combiner)); - grpc_bdp_estimator_init(&t->flow_control.bdp_estimator, t->peer_string); + t->flow_control.bdp_estimator.Init(t->peer_string); grpc_chttp2_goaway_parser_init(&t->goaway_parser); grpc_chttp2_hpack_parser_init(exec_ctx, &t->hpack_parser); @@ -562,6 +572,11 @@ static void init_transport(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t, t->keepalive_state = GRPC_CHTTP2_KEEPALIVE_STATE_DISABLED; } + if (t->flow_control.enable_bdp_probe) { + GRPC_CHTTP2_REF_TRANSPORT(t, "bdp_ping"); + schedule_bdp_ping_locked(exec_ctx, t); + } + grpc_chttp2_act_on_flowctl_action( exec_ctx, grpc_chttp2_flowctl_get_action(exec_ctx, &t->flow_control, NULL), t, @@ -595,7 +610,9 @@ static void destroy_transport(grpc_exec_ctx *exec_ctx, grpc_transport *gt) { static void close_transport_locked(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t, grpc_error *error) { - if (!t->closed) { + end_all_the_calls(exec_ctx, t, GRPC_ERROR_REF(error)); + cancel_pings(exec_ctx, t, GRPC_ERROR_REF(error)); + if (t->closed_with_error == GRPC_ERROR_NONE) { if (!grpc_error_has_clear_grpc_status(error)) { error = grpc_error_set_int(error, GRPC_ERROR_INT_GRPC_STATUS, GRPC_STATUS_UNAVAILABLE); @@ -610,13 +627,16 @@ static void close_transport_locked(grpc_exec_ctx *exec_ctx, grpc_error_add_child(t->close_transport_on_writes_finished, error); return; } - t->closed = 1; + GPR_ASSERT(error != GRPC_ERROR_NONE); + t->closed_with_error = GRPC_ERROR_REF(error); connectivity_state_set(exec_ctx, t, GRPC_CHANNEL_SHUTDOWN, GRPC_ERROR_REF(error), "close_transport"); - grpc_endpoint_shutdown(exec_ctx, t->ep, GRPC_ERROR_REF(error)); if (t->ping_state.is_delayed_ping_timer_set) { grpc_timer_cancel(exec_ctx, &t->ping_state.delayed_ping_timer); } + if (t->have_next_bdp_ping_timer) { + grpc_timer_cancel(exec_ctx, &t->next_bdp_ping_timer); + } switch (t->keepalive_state) { case GRPC_CHTTP2_KEEPALIVE_STATE_WAITING: grpc_timer_cancel(exec_ctx, &t->keepalive_ping_timer); @@ -636,8 +656,8 @@ static void close_transport_locked(grpc_exec_ctx *exec_ctx, while (grpc_chttp2_list_pop_writable_stream(t, &s)) { GRPC_CHTTP2_STREAM_UNREF(exec_ctx, s, "chttp2_writing:close"); } - end_all_the_calls(exec_ctx, t, GRPC_ERROR_REF(error)); - cancel_pings(exec_ctx, t, GRPC_ERROR_REF(error)); + GPR_ASSERT(t->write_state == GRPC_CHTTP2_WRITE_STATE_IDLE); + grpc_endpoint_shutdown(exec_ctx, t->ep, GRPC_ERROR_REF(error)); } GRPC_ERROR_UNREF(error); } @@ -940,7 +960,8 @@ void grpc_chttp2_initiate_write(grpc_exec_ctx *exec_ctx, void grpc_chttp2_mark_stream_writable(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t, grpc_chttp2_stream *s) { - if (!t->closed && grpc_chttp2_list_add_writable_stream(t, s)) { + if (t->closed_with_error == GRPC_ERROR_NONE && + grpc_chttp2_list_add_writable_stream(t, s)) { GRPC_CHTTP2_STREAM_REF(s, "chttp2_writing:become"); } } @@ -993,7 +1014,7 @@ static void write_action_begin_locked(grpc_exec_ctx *exec_ctx, void *gt, grpc_chttp2_transport *t = (grpc_chttp2_transport *)gt; GPR_ASSERT(t->write_state != GRPC_CHTTP2_WRITE_STATE_IDLE); grpc_chttp2_begin_write_result r; - if (t->closed) { + if (t->closed_with_error != GRPC_ERROR_NONE) { r.writing = false; } else { r = grpc_chttp2_begin_write(exec_ctx, t); @@ -1456,7 +1477,7 @@ static void perform_stream_op_locked(grpc_exec_ctx *exec_ctx, void *stream_op, } if (!s->write_closed) { if (t->is_client) { - if (!t->closed) { + if (t->closed_with_error == GRPC_ERROR_NONE) { GPR_ASSERT(s->id == 0); grpc_chttp2_list_add_waiting_for_concurrency(t, s); maybe_start_some_streams(exec_ctx, t); @@ -1464,7 +1485,8 @@ static void perform_stream_op_locked(grpc_exec_ctx *exec_ctx, void *stream_op, grpc_chttp2_cancel_stream( exec_ctx, t, s, grpc_error_set_int( - GRPC_ERROR_CREATE_FROM_STATIC_STRING("Transport closed"), + GRPC_ERROR_CREATE_REFERENCING_FROM_STATIC_STRING( + "Transport closed", &t->closed_with_error, 1), GRPC_ERROR_INT_GRPC_STATUS, GRPC_STATUS_UNAVAILABLE)); } } else { @@ -1686,6 +1708,7 @@ static void cancel_pings(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t, /* callback remaining pings: they're not allowed to call into the transpot, and maybe they hold resources that need to be freed */ grpc_chttp2_ping_queue *pq = &t->ping_queue; + GPR_ASSERT(error != GRPC_ERROR_NONE); for (size_t j = 0; j < GRPC_CHTTP2_PCL_COUNT; j++) { grpc_closure_list_fail_all(&pq->lists[j], GRPC_ERROR_REF(error)); GRPC_CLOSURE_LIST_SCHED(exec_ctx, &pq->lists[j]); @@ -1695,6 +1718,12 @@ static void cancel_pings(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t, static void send_ping_locked(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t, grpc_closure *on_initiate, grpc_closure *on_ack) { + if (t->closed_with_error != GRPC_ERROR_NONE) { + GRPC_CLOSURE_SCHED(exec_ctx, on_initiate, + GRPC_ERROR_REF(t->closed_with_error)); + GRPC_CLOSURE_SCHED(exec_ctx, on_ack, GRPC_ERROR_REF(t->closed_with_error)); + return; + } grpc_chttp2_ping_queue *pq = &t->ping_queue; grpc_closure_list_append(&pq->lists[GRPC_CHTTP2_PCL_INITIATE], on_initiate, GRPC_ERROR_NONE); @@ -1753,7 +1782,9 @@ void grpc_chttp2_add_ping_strike(grpc_exec_ctx *exec_ctx, GRPC_ERROR_INT_HTTP2_ERROR, GRPC_HTTP2_ENHANCE_YOUR_CALM)); /*The transport will be closed after the write is done */ close_transport_locked( - exec_ctx, t, GRPC_ERROR_CREATE_FROM_STATIC_STRING("Too many pings")); + exec_ctx, t, grpc_error_set_int( + GRPC_ERROR_CREATE_FROM_STATIC_STRING("Too many pings"), + GRPC_ERROR_INT_GRPC_STATUS, GRPC_STATUS_UNAVAILABLE)); } } @@ -2432,12 +2463,6 @@ void grpc_chttp2_act_on_flowctl_action(grpc_exec_ctx *exec_ctx, GRPC_CHTTP2_INITIATE_WRITE_SEND_SETTINGS); } } - if (action.need_ping) { - GRPC_CHTTP2_REF_TRANSPORT(t, "bdp_ping"); - grpc_bdp_estimator_schedule_ping(&t->flow_control.bdp_estimator); - send_ping_locked(exec_ctx, t, &t->start_bdp_ping_locked, - &t->finish_bdp_ping_locked); - } } static grpc_error *try_http_parsing(grpc_exec_ctx *exec_ctx, @@ -2487,14 +2512,13 @@ static void read_action_locked(grpc_exec_ctx *exec_ctx, void *tp, } GPR_SWAP(grpc_error *, err, error); GRPC_ERROR_UNREF(err); - if (!t->closed) { + if (t->closed_with_error == GRPC_ERROR_NONE) { GPR_TIMER_BEGIN("reading_action.parse", 0); size_t i = 0; grpc_error *errors[3] = {GRPC_ERROR_REF(error), GRPC_ERROR_NONE, GRPC_ERROR_NONE}; for (; i < t->read_buffer.count && errors[1] == GRPC_ERROR_NONE; i++) { - grpc_bdp_estimator_add_incoming_bytes( - &t->flow_control.bdp_estimator, + t->flow_control.bdp_estimator->AddIncomingBytes( (int64_t)GRPC_SLICE_LENGTH(t->read_buffer.slices[i])); errors[1] = grpc_chttp2_perform_read(exec_ctx, t, t->read_buffer.slices[i]); @@ -2528,13 +2552,14 @@ static void read_action_locked(grpc_exec_ctx *exec_ctx, void *tp, GPR_TIMER_BEGIN("post_reading_action_locked", 0); bool keep_reading = false; - if (error == GRPC_ERROR_NONE && t->closed) { - error = GRPC_ERROR_CREATE_FROM_STATIC_STRING("Transport closed"); + if (error == GRPC_ERROR_NONE && t->closed_with_error != GRPC_ERROR_NONE) { + error = GRPC_ERROR_CREATE_REFERENCING_FROM_STATIC_STRING( + "Transport closed", &t->closed_with_error, 1); } if (error != GRPC_ERROR_NONE) { close_transport_locked(exec_ctx, t, GRPC_ERROR_REF(error)); t->endpoint_reading = 0; - } else if (!t->closed) { + } else if (t->closed_with_error == GRPC_ERROR_NONE) { keep_reading = true; GRPC_CHTTP2_REF_TRANSPORT(t, "keep_reading"); } @@ -2559,28 +2584,57 @@ static void read_action_locked(grpc_exec_ctx *exec_ctx, void *tp, GPR_TIMER_END("reading_action_locked", 0); } +// t is reffed prior to calling the first time, and once the callback chain +// that kicks off finishes, it's unreffed +static void schedule_bdp_ping_locked(grpc_exec_ctx *exec_ctx, + grpc_chttp2_transport *t) { + t->flow_control.bdp_estimator->SchedulePing(); + send_ping_locked(exec_ctx, t, &t->start_bdp_ping_locked, + &t->finish_bdp_ping_locked); +} + static void start_bdp_ping_locked(grpc_exec_ctx *exec_ctx, void *tp, grpc_error *error) { grpc_chttp2_transport *t = (grpc_chttp2_transport *)tp; if (GRPC_TRACER_ON(grpc_http_trace)) { - gpr_log(GPR_DEBUG, "%s: Start BDP ping", t->peer_string); + gpr_log(GPR_DEBUG, "%s: Start BDP ping err=%s", t->peer_string, + grpc_error_string(error)); } /* Reset the keepalive ping timer */ if (t->keepalive_state == GRPC_CHTTP2_KEEPALIVE_STATE_WAITING) { grpc_timer_cancel(exec_ctx, &t->keepalive_ping_timer); } - grpc_bdp_estimator_start_ping(&t->flow_control.bdp_estimator); + t->flow_control.bdp_estimator->StartPing(); } static void finish_bdp_ping_locked(grpc_exec_ctx *exec_ctx, void *tp, grpc_error *error) { grpc_chttp2_transport *t = (grpc_chttp2_transport *)tp; if (GRPC_TRACER_ON(grpc_http_trace)) { - gpr_log(GPR_DEBUG, "%s: Complete BDP ping", t->peer_string); + gpr_log(GPR_DEBUG, "%s: Complete BDP ping err=%s", t->peer_string, + grpc_error_string(error)); + } + if (error != GRPC_ERROR_NONE) { + GRPC_CHTTP2_UNREF_TRANSPORT(exec_ctx, t, "bdp_ping"); + return; } - grpc_bdp_estimator_complete_ping(exec_ctx, &t->flow_control.bdp_estimator); + grpc_millis next_ping = t->flow_control.bdp_estimator->CompletePing(exec_ctx); + GPR_ASSERT(!t->have_next_bdp_ping_timer); + t->have_next_bdp_ping_timer = true; + grpc_timer_init(exec_ctx, &t->next_bdp_ping_timer, next_ping, + &t->next_bdp_ping_timer_expired_locked); +} - GRPC_CHTTP2_UNREF_TRANSPORT(exec_ctx, t, "bdp_ping"); +static void next_bdp_ping_timer_expired_locked(grpc_exec_ctx *exec_ctx, + void *tp, grpc_error *error) { + grpc_chttp2_transport *t = (grpc_chttp2_transport *)tp; + GPR_ASSERT(t->have_next_bdp_ping_timer); + t->have_next_bdp_ping_timer = false; + if (error != GRPC_ERROR_NONE) { + GRPC_CHTTP2_UNREF_TRANSPORT(exec_ctx, t, "bdp_ping"); + return; + } + schedule_bdp_ping_locked(exec_ctx, t); } void grpc_chttp2_config_default_keepalive_args(grpc_channel_args *args, @@ -2645,7 +2699,7 @@ static void init_keepalive_ping_locked(grpc_exec_ctx *exec_ctx, void *arg, grpc_error *error) { grpc_chttp2_transport *t = (grpc_chttp2_transport *)arg; GPR_ASSERT(t->keepalive_state == GRPC_CHTTP2_KEEPALIVE_STATE_WAITING); - if (t->destroying || t->closed) { + if (t->destroying || t->closed_with_error != GRPC_ERROR_NONE) { t->keepalive_state = GRPC_CHTTP2_KEEPALIVE_STATE_DYING; } else if (error == GRPC_ERROR_NONE) { if (t->keepalive_permit_without_calls || @@ -2703,8 +2757,11 @@ static void keepalive_watchdog_fired_locked(grpc_exec_ctx *exec_ctx, void *arg, if (t->keepalive_state == GRPC_CHTTP2_KEEPALIVE_STATE_PINGING) { if (error == GRPC_ERROR_NONE) { t->keepalive_state = GRPC_CHTTP2_KEEPALIVE_STATE_DYING; - close_transport_locked(exec_ctx, t, GRPC_ERROR_CREATE_FROM_STATIC_STRING( - "keepalive watchdog timeout")); + close_transport_locked( + exec_ctx, t, + grpc_error_set_int(GRPC_ERROR_CREATE_FROM_STATIC_STRING( + "keepalive watchdog timeout"), + GRPC_ERROR_INT_GRPC_STATUS, GRPC_STATUS_INTERNAL)); } } else { /* The watchdog timer should have been cancelled by diff --git a/src/core/ext/transport/chttp2/transport/flow_control.cc b/src/core/ext/transport/chttp2/transport/flow_control.cc index 2428e2526d..d0e80c4bd5 100644 --- a/src/core/ext/transport/chttp2/transport/flow_control.cc +++ b/src/core/ext/transport/chttp2/transport/flow_control.cc @@ -459,12 +459,9 @@ grpc_chttp2_flowctl_action grpc_chttp2_flowctl_get_action( } } if (tfc->enable_bdp_probe) { - action.need_ping = - grpc_bdp_estimator_need_ping(exec_ctx, &tfc->bdp_estimator); - // get bdp estimate and update initial_window accordingly. int64_t estimate = -1; - if (grpc_bdp_estimator_get_estimate(&tfc->bdp_estimator, &estimate)) { + if (tfc->bdp_estimator->EstimateBdp(&estimate)) { double target = 1 + log2((double)estimate); // target might change based on how much memory pressure we are under @@ -491,7 +488,7 @@ grpc_chttp2_flowctl_action grpc_chttp2_flowctl_get_action( // get bandwidth estimate and update max_frame accordingly. double bw_dbl = -1; - if (grpc_bdp_estimator_get_bw(&tfc->bdp_estimator, &bw_dbl)) { + if (tfc->bdp_estimator->EstimateBandwidth(&bw_dbl)) { // we target the max of BDP or bandwidth in microseconds. int32_t frame_size = (int32_t)GPR_CLAMP( GPR_MAX((int32_t)GPR_CLAMP(bw_dbl, 0, INT_MAX) / 1000, diff --git a/src/core/ext/transport/chttp2/transport/internal.h b/src/core/ext/transport/chttp2/transport/internal.h index 5d27bb8d67..703f3ba348 100644 --- a/src/core/ext/transport/chttp2/transport/internal.h +++ b/src/core/ext/transport/chttp2/transport/internal.h @@ -37,6 +37,7 @@ #include "src/core/lib/iomgr/combiner.h" #include "src/core/lib/iomgr/endpoint.h" #include "src/core/lib/iomgr/timer.h" +#include "src/core/lib/support/manual_constructor.h" #include "src/core/lib/transport/bdp_estimator.h" #include "src/core/lib/transport/connectivity_state.h" #include "src/core/lib/transport/pid_controller.h" @@ -268,7 +269,7 @@ typedef struct { bool enable_bdp_probe; /* bdp estimation */ - grpc_bdp_estimator bdp_estimator; + grpc_core::ManualConstructor<grpc_core::BdpEstimator> bdp_estimator; /* pid controller */ bool pid_controller_initialized; @@ -297,7 +298,7 @@ struct grpc_chttp2_transport { /** is the transport destroying itself? */ uint8_t destroying; /** has the upper layer closed the transport? */ - uint8_t closed; + grpc_error *closed_with_error; /** is there a read request to the endpoint outstanding? */ uint8_t endpoint_reading; @@ -339,7 +340,7 @@ struct grpc_chttp2_transport { /** hpack encoding */ grpc_chttp2_hpack_compressor hpack_compressor; /** is this a client? */ - uint8_t is_client; + bool is_client; /** data to write next write */ grpc_slice_buffer qbuf; @@ -349,14 +350,14 @@ struct grpc_chttp2_transport { uint32_t write_buffer_size; /** have we seen a goaway */ - uint8_t seen_goaway; + bool seen_goaway; /** have we sent a goaway */ grpc_chttp2_sent_goaway_state sent_goaway_state; /** are the local settings dirty and need to be sent? */ - uint8_t dirtied_local_settings; + bool dirtied_local_settings; /** have local settings been sent? */ - uint8_t sent_local_settings; + bool sent_local_settings; /** bitmask of setting indexes to send out */ uint32_t force_send_settings; /** settings values */ @@ -421,6 +422,7 @@ struct grpc_chttp2_transport { grpc_chttp2_write_cb *write_cb_pool; /* bdp estimator */ + grpc_closure next_bdp_ping_timer_expired_locked; grpc_closure start_bdp_ping_locked; grpc_closure finish_bdp_ping_locked; @@ -441,6 +443,10 @@ struct grpc_chttp2_transport { /** destructive cleanup closure */ grpc_closure destructive_reclaimer_locked; + /* next bdp ping timer */ + bool have_next_bdp_ping_timer; + grpc_timer next_bdp_ping_timer; + /* keep-alive ping support */ /** Closure to initialize a keepalive ping */ grpc_closure init_keepalive_ping_locked; @@ -748,7 +754,6 @@ typedef struct { grpc_chttp2_flowctl_urgency send_setting_update; uint32_t initial_window_size; uint32_t max_frame_size; - bool need_ping; } grpc_chttp2_flowctl_action; // Reads the flow control data and returns and actionable struct that will tell diff --git a/src/core/ext/transport/chttp2/transport/writing.cc b/src/core/ext/transport/chttp2/transport/writing.cc index 25c1a5ef05..c6fecf2ee9 100644 --- a/src/core/ext/transport/chttp2/transport/writing.cc +++ b/src/core/ext/transport/chttp2/transport/writing.cc @@ -245,7 +245,8 @@ class WriteContext { void UpdateStreamsNoLongerStalled() { grpc_chttp2_stream *s; while (grpc_chttp2_list_pop_stalled_by_transport(t_, &s)) { - if (!t_->closed && grpc_chttp2_list_add_writable_stream(t_, s)) { + if (t_->closed_with_error == GRPC_ERROR_NONE && + grpc_chttp2_list_add_writable_stream(t_, s)) { if (!stream_ref_if_not_destroyed(&s->refcount->refs)) { grpc_chttp2_list_remove_writable_stream(t_, s); } |