diff --git a/src/core/ext/transport/chttp2/transport/chttp2_transport.cc b/src/core/ext/transport/chttp2/transport/chttp2_transport.cc index f6f9242caf5..fdbeced4456 100644 --- a/src/core/ext/transport/chttp2/transport/chttp2_transport.cc +++ b/src/core/ext/transport/chttp2/transport/chttp2_transport.cc @@ -54,7 +54,6 @@ #include "src/core/lib/transport/transport.h" #include "src/core/lib/transport/transport_impl.h" -#define DEFAULT_WINDOW 65535 #define DEFAULT_CONNECTION_WINDOW_TARGET (1024 * 1024) #define MAX_WINDOW 0x7fffffffu #define MAX_WRITE_BUFFER_SIZE (64 * 1024 * 1024) @@ -222,7 +221,7 @@ static void destruct_transport(grpc_exec_ctx *exec_ctx, t->write_cb_pool = next; } - t->flow_control.bdp_estimator.Destroy(); + t->flow_control.Destroy(); gpr_free(t->ping_acks); gpr_free(t->peer_string); @@ -281,10 +280,7 @@ static void init_transport(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t, t->endpoint_reading = 1; t->next_stream_id = is_client ? 1 : 2; t->is_client = is_client; - t->flow_control.remote_window = DEFAULT_WINDOW; - t->flow_control.announced_window = DEFAULT_WINDOW; - t->flow_control.target_initial_window_size = DEFAULT_WINDOW; - t->flow_control.t = t; + t->flow_control.Init(); t->deframe_state = is_client ? GRPC_DTS_FH_0 : GRPC_DTS_CLIENT_PREFIX_0; t->is_first_frame = true; grpc_connectivity_state_init( diff --git a/src/core/ext/transport/chttp2/transport/flow_control.cc b/src/core/ext/transport/chttp2/transport/flow_control.cc index 716cd71490f..3898e533370 100644 --- a/src/core/ext/transport/chttp2/transport/flow_control.cc +++ b/src/core/ext/transport/chttp2/transport/flow_control.cc @@ -16,7 +16,7 @@ * */ -#include "src/core/ext/transport/chttp2/transport/internal.h" +#include "src/core/ext/transport/chttp2/transport/flow_control.h" #include #include @@ -28,38 +28,15 @@ #include #include +#include "src/core/ext/transport/chttp2/transport/internal.h" #include "src/core/lib/support/string.h" -static uint32_t grpc_chttp2_target_announced_window( - const grpc_chttp2_transport_flowctl* tfc); - -#ifndef NDEBUG - -typedef struct { - int64_t remote_window; - int64_t target_window; - int64_t announced_window; - int64_t remote_window_delta; - int64_t local_window_delta; - int64_t announced_window_delta; - uint32_t local_init_window; - uint32_t local_max_frame; -} shadow_flow_control; - -static void pretrace(shadow_flow_control* shadow_fc, - grpc_chttp2_transport_flowctl* tfc, - grpc_chttp2_stream_flowctl* sfc) { - shadow_fc->remote_window = tfc->remote_window; - shadow_fc->target_window = grpc_chttp2_target_announced_window(tfc); - shadow_fc->announced_window = tfc->announced_window; - if (sfc != NULL) { - shadow_fc->remote_window_delta = sfc->remote_window_delta; - shadow_fc->local_window_delta = sfc->local_window_delta; - shadow_fc->announced_window_delta = sfc->announced_window_delta; - } -} +namespace grpc_core { +namespace chttp2 { + +namespace { -#define TRACE_PADDING 30 +static constexpr const int kTracePadding = 30; static char* fmt_int64_diff_str(int64_t old_val, int64_t new_val) { char* str; @@ -68,7 +45,7 @@ static char* fmt_int64_diff_str(int64_t old_val, int64_t new_val) { } else { gpr_asprintf(&str, "%" PRId64 "", old_val); } - char* str_lp = gpr_leftpad(str, ' ', TRACE_PADDING); + char* str_lp = gpr_leftpad(str, ' ', kTracePadding); gpr_free(str); return str_lp; } @@ -80,47 +57,57 @@ static char* fmt_uint32_diff_str(uint32_t old_val, uint32_t new_val) { } else { gpr_asprintf(&str, "%" PRIu32 "", old_val); } - char* str_lp = gpr_leftpad(str, ' ', TRACE_PADDING); + char* str_lp = gpr_leftpad(str, ' ', kTracePadding); gpr_free(str); return str_lp; } +} // namespace + +void FlowControlTrace::Init(const char* reason, TransportFlowControl* tfc, + StreamFlowControl* sfc) { + tfc_ = tfc; + sfc_ = sfc; + remote_window_ = tfc->remote_window(); + target_window_ = tfc->target_window(); + announced_window_ = tfc->announced_window(); + if (sfc != nullptr) { + remote_window_delta_ = sfc->remote_window_delta(); + local_window_delta_ = sfc->local_window_delta(); + announced_window_delta_ = sfc->announced_window_delta(); + } +} -static void posttrace(shadow_flow_control* shadow_fc, - grpc_chttp2_transport_flowctl* tfc, - grpc_chttp2_stream_flowctl* sfc, const char* reason) { +void FlowControlTrace::Finish() { uint32_t acked_local_window = - tfc->t->settings[GRPC_SENT_SETTINGS] - [GRPC_CHTTP2_SETTINGS_INITIAL_WINDOW_SIZE]; + tfc_->transport()->settings[GRPC_SENT_SETTINGS] + [GRPC_CHTTP2_SETTINGS_INITIAL_WINDOW_SIZE]; uint32_t remote_window = - tfc->t->settings[GRPC_PEER_SETTINGS] - [GRPC_CHTTP2_SETTINGS_INITIAL_WINDOW_SIZE]; - char* trw_str = - fmt_int64_diff_str(shadow_fc->remote_window, tfc->remote_window); - char* tlw_str = fmt_int64_diff_str(shadow_fc->target_window, - grpc_chttp2_target_announced_window(tfc)); + tfc_->transport()->settings[GRPC_PEER_SETTINGS] + [GRPC_CHTTP2_SETTINGS_INITIAL_WINDOW_SIZE]; + char* trw_str = fmt_int64_diff_str(remote_window_, tfc_->remote_window()); + char* tlw_str = fmt_int64_diff_str(target_window_, tfc_->target_window()); char* taw_str = - fmt_int64_diff_str(shadow_fc->announced_window, tfc->announced_window); + fmt_int64_diff_str(announced_window_, tfc_->announced_window()); char* srw_str; char* slw_str; char* saw_str; - if (sfc != NULL) { - srw_str = fmt_int64_diff_str(shadow_fc->remote_window_delta + remote_window, - sfc->remote_window_delta + remote_window); - slw_str = - fmt_int64_diff_str(shadow_fc->local_window_delta + acked_local_window, - sfc->local_window_delta + acked_local_window); - saw_str = fmt_int64_diff_str( - shadow_fc->announced_window_delta + acked_local_window, - sfc->announced_window_delta + acked_local_window); + if (sfc_ != nullptr) { + srw_str = fmt_int64_diff_str(remote_window_delta_ + remote_window, + sfc_->remote_window_delta() + remote_window); + slw_str = fmt_int64_diff_str(local_window_delta_ + acked_local_window, + local_window_delta_ + acked_local_window); + saw_str = fmt_int64_diff_str(announced_window_delta_ + acked_local_window, + announced_window_delta_ + acked_local_window); } else { - srw_str = gpr_leftpad("", ' ', TRACE_PADDING); - slw_str = gpr_leftpad("", ' ', TRACE_PADDING); - saw_str = gpr_leftpad("", ' ', TRACE_PADDING); + srw_str = gpr_leftpad("", ' ', kTracePadding); + slw_str = gpr_leftpad("", ' ', kTracePadding); + saw_str = gpr_leftpad("", ' ', kTracePadding); } gpr_log(GPR_DEBUG, "%p[%u][%s] | %s | trw:%s, ttw:%s, taw:%s, srw:%s, slw:%s, saw:%s", - tfc, sfc != NULL ? sfc->s->id : 0, tfc->t->is_client ? "cli" : "svr", - reason, trw_str, tlw_str, taw_str, srw_str, slw_str, saw_str); + tfc_, sfc_ != nullptr ? sfc_->stream()->id : 0, + tfc_->transport()->is_client ? "cli" : "svr", reason_, trw_str, + tlw_str, taw_str, srw_str, slw_str, saw_str); gpr_free(trw_str); gpr_free(tlw_str); gpr_free(taw_str); @@ -129,13 +116,13 @@ static void posttrace(shadow_flow_control* shadow_fc, gpr_free(saw_str); } -static const char* urgency_to_string(grpc_chttp2_flowctl_urgency urgency) { - switch (urgency) { - case GRPC_CHTTP2_FLOWCTL_NO_ACTION_NEEDED: +const char* FlowControlAction::UrgencyString(Urgency u) { + switch (u) { + case Urgency::NO_ACTION_NEEDED: return "no action"; - case GRPC_CHTTP2_FLOWCTL_UPDATE_IMMEDIATELY: + case Urgency::UPDATE_IMMEDIATELY: return "update immediately"; - case GRPC_CHTTP2_FLOWCTL_QUEUE_UPDATE: + case Urgency::QUEUE_UPDATE: return "queue update"; default: GPR_UNREACHABLE_CODE(return "unknown"); @@ -143,209 +130,110 @@ static const char* urgency_to_string(grpc_chttp2_flowctl_urgency urgency) { GPR_UNREACHABLE_CODE(return "unknown"); } -static void trace_action(grpc_chttp2_transport_flowctl* tfc, - grpc_chttp2_flowctl_action action) { +void FlowControlAction::Trace(grpc_chttp2_transport* t) const { char* iw_str = fmt_uint32_diff_str( - tfc->t->settings[GRPC_SENT_SETTINGS] - [GRPC_CHTTP2_SETTINGS_INITIAL_WINDOW_SIZE], - action.initial_window_size); + t->settings[GRPC_SENT_SETTINGS][GRPC_CHTTP2_SETTINGS_INITIAL_WINDOW_SIZE], + initial_window_size_); char* mf_str = fmt_uint32_diff_str( - tfc->t->settings[GRPC_SENT_SETTINGS][GRPC_CHTTP2_SETTINGS_MAX_FRAME_SIZE], - action.max_frame_size); + t->settings[GRPC_SENT_SETTINGS][GRPC_CHTTP2_SETTINGS_MAX_FRAME_SIZE], + max_frame_size_); gpr_log(GPR_DEBUG, "t[%s], s[%s], settings[%s] iw:%s mf:%s", - urgency_to_string(action.send_transport_update), - urgency_to_string(action.send_stream_update), - urgency_to_string(action.send_setting_update), iw_str, mf_str); + UrgencyString(send_transport_update_), + UrgencyString(send_stream_update_), + UrgencyString(send_setting_update_), iw_str, mf_str); gpr_free(iw_str); gpr_free(mf_str); } -#define PRETRACE(tfc, sfc) \ - shadow_flow_control shadow_fc; \ - GRPC_FLOW_CONTROL_IF_TRACING(pretrace(&shadow_fc, tfc, sfc)) -#define POSTTRACE(tfc, sfc, reason) \ - GRPC_FLOW_CONTROL_IF_TRACING(posttrace(&shadow_fc, tfc, sfc, reason)) -#define TRACEACTION(tfc, action) \ - GRPC_FLOW_CONTROL_IF_TRACING(trace_action(tfc, action)) -#else -#define PRETRACE(tfc, sfc) -#define POSTTRACE(tfc, sfc, reason) -#define TRACEACTION(tfc, action) -#endif - -/* How many bytes of incoming flow control would we like to advertise */ -static uint32_t grpc_chttp2_target_announced_window( - const grpc_chttp2_transport_flowctl* tfc) { - return (uint32_t)GPR_MIN((int64_t)((1u << 31) - 1), - tfc->announced_stream_total_over_incoming_window + - tfc->target_initial_window_size); -} - -// we have sent data on the wire, we must track this in our bookkeeping for the -// remote peer's flow control. -void grpc_chttp2_flowctl_sent_data(grpc_chttp2_transport_flowctl* tfc, - grpc_chttp2_stream_flowctl* sfc, - int64_t size) { - PRETRACE(tfc, sfc); - tfc->remote_window -= size; - sfc->remote_window_delta -= size; - POSTTRACE(tfc, sfc, " data sent"); -} - -static void announced_window_delta_preupdate(grpc_chttp2_transport_flowctl* tfc, - grpc_chttp2_stream_flowctl* sfc) { - if (sfc->announced_window_delta > 0) { - tfc->announced_stream_total_over_incoming_window -= - sfc->announced_window_delta; - } else { - tfc->announced_stream_total_under_incoming_window += - -sfc->announced_window_delta; - } -} - -static void announced_window_delta_postupdate( - grpc_chttp2_transport_flowctl* tfc, grpc_chttp2_stream_flowctl* sfc) { - if (sfc->announced_window_delta > 0) { - tfc->announced_stream_total_over_incoming_window += - sfc->announced_window_delta; - } else { - tfc->announced_stream_total_under_incoming_window -= - -sfc->announced_window_delta; +uint32_t TransportFlowControl::MaybeSendUpdate(bool writing_anyway) { + FlowControlTrace trace("t updt sent", this, nullptr); + const uint32_t target_announced_window = target_window(); + if ((writing_anyway || announced_window_ <= target_announced_window / 2) && + announced_window_ != target_announced_window) { + const uint32_t announce = (uint32_t)GPR_CLAMP( + target_announced_window - announced_window_, 0, UINT32_MAX); + announced_window_ += announce; + return announce; } + return 0; } -// We have received data from the wire. We must track this in our own flow -// control bookkeeping. -// Returns an error if the incoming frame violates our flow control. -grpc_error* grpc_chttp2_flowctl_recv_data(grpc_chttp2_transport_flowctl* tfc, - grpc_chttp2_stream_flowctl* sfc, - int64_t incoming_frame_size) { - uint32_t sent_init_window = - tfc->t->settings[GRPC_SENT_SETTINGS] - [GRPC_CHTTP2_SETTINGS_INITIAL_WINDOW_SIZE]; - uint32_t acked_init_window = - tfc->t->settings[GRPC_ACKED_SETTINGS] - [GRPC_CHTTP2_SETTINGS_INITIAL_WINDOW_SIZE]; - PRETRACE(tfc, sfc); - if (incoming_frame_size > tfc->announced_window) { +grpc_error* TransportFlowControl::ValidateRecvData( + int64_t incoming_frame_size) { + if (incoming_frame_size > announced_window_) { char* msg; gpr_asprintf(&msg, "frame of size %" PRId64 " overflows local window of %" PRId64, - incoming_frame_size, tfc->announced_window); + incoming_frame_size, announced_window_); grpc_error* err = GRPC_ERROR_CREATE_FROM_COPIED_STRING(msg); gpr_free(msg); return err; } - - if (sfc != NULL) { - int64_t acked_stream_window = - sfc->announced_window_delta + acked_init_window; - int64_t sent_stream_window = sfc->announced_window_delta + sent_init_window; - if (incoming_frame_size > acked_stream_window) { - if (incoming_frame_size <= sent_stream_window) { - gpr_log( - GPR_ERROR, - "Incoming frame of size %" PRId64 - " exceeds local window size of %" PRId64 - ".\n" - "The (un-acked, future) window size would be %" PRId64 - " which is not exceeded.\n" - "This would usually cause a disconnection, but allowing it due to" - "broken HTTP2 implementations in the wild.\n" - "See (for example) https://github.com/netty/netty/issues/6520.", - incoming_frame_size, acked_stream_window, sent_stream_window); - } else { - char* msg; - gpr_asprintf(&msg, "frame of size %" PRId64 - " overflows local window of %" PRId64, - incoming_frame_size, acked_stream_window); - grpc_error* err = GRPC_ERROR_CREATE_FROM_COPIED_STRING(msg); - gpr_free(msg); - return err; - } - } - - announced_window_delta_preupdate(tfc, sfc); - sfc->announced_window_delta -= incoming_frame_size; - announced_window_delta_postupdate(tfc, sfc); - sfc->local_window_delta -= incoming_frame_size; - } - - tfc->announced_window -= incoming_frame_size; - - POSTTRACE(tfc, sfc, " data recv"); return GRPC_ERROR_NONE; } -// Returns a non zero announce integer if we should send a transport window -// update -uint32_t grpc_chttp2_flowctl_maybe_send_transport_update( - grpc_chttp2_transport_flowctl* tfc, bool writing_anyway) { - PRETRACE(tfc, NULL); - uint32_t target_announced_window = grpc_chttp2_target_announced_window(tfc); - uint32_t threshold_to_send_transport_window_update = - tfc->t->outbuf.count > 0 ? 3 * target_announced_window / 4 - : target_announced_window / 2; - if ((writing_anyway || - tfc->announced_window <= threshold_to_send_transport_window_update) && - tfc->announced_window != target_announced_window) { - uint32_t announce = (uint32_t)GPR_CLAMP( - target_announced_window - tfc->announced_window, 0, UINT32_MAX); - tfc->announced_window += announce; - POSTTRACE(tfc, NULL, "t updt sent"); - return announce; +grpc_error* StreamFlowControl::RecvData(int64_t incoming_frame_size) { + FlowControlTrace trace(" data recv", tfc_, this); + + grpc_error* error = GRPC_ERROR_NONE; + error = tfc_->ValidateRecvData(incoming_frame_size); + if (error != GRPC_ERROR_NONE) return error; + + uint32_t sent_init_window = + tfc_->transport()->settings[GRPC_SENT_SETTINGS] + [GRPC_CHTTP2_SETTINGS_INITIAL_WINDOW_SIZE]; + uint32_t acked_init_window = + tfc_->transport()->settings[GRPC_ACKED_SETTINGS] + [GRPC_CHTTP2_SETTINGS_INITIAL_WINDOW_SIZE]; + + int64_t acked_stream_window = announced_window_delta_ + acked_init_window; + int64_t sent_stream_window = announced_window_delta_ + sent_init_window; + if (incoming_frame_size > acked_stream_window) { + if (incoming_frame_size <= sent_stream_window) { + gpr_log(GPR_ERROR, + "Incoming frame of size %" PRId64 + " exceeds local window size of %" PRId64 + ".\n" + "The (un-acked, future) window size would be %" PRId64 + " which is not exceeded.\n" + "This would usually cause a disconnection, but allowing it due to" + "broken HTTP2 implementations in the wild.\n" + "See (for example) https://github.com/netty/netty/issues/6520.", + incoming_frame_size, acked_stream_window, sent_stream_window); + } else { + char* msg; + gpr_asprintf(&msg, "frame of size %" PRId64 + " overflows local window of %" PRId64, + incoming_frame_size, acked_stream_window); + grpc_error* err = GRPC_ERROR_CREATE_FROM_COPIED_STRING(msg); + gpr_free(msg); + return err; + } } - GRPC_FLOW_CONTROL_IF_TRACING( - gpr_log(GPR_DEBUG, "%p[0][%s] will not send transport update", tfc, - tfc->t->is_client ? "cli" : "svr")); - return 0; + + UpdateAnnouncedWindowDelta(tfc_, -incoming_frame_size); + local_window_delta_ -= incoming_frame_size; + tfc_->CommitRecvData(incoming_frame_size); } -// Returns a non zero announce integer if we should send a stream window update -uint32_t grpc_chttp2_flowctl_maybe_send_stream_update( - grpc_chttp2_transport_flowctl* tfc, grpc_chttp2_stream_flowctl* sfc) { - PRETRACE(tfc, sfc); - if (sfc->local_window_delta > sfc->announced_window_delta) { +uint32_t StreamFlowControl::MaybeSendUpdate() { + FlowControlTrace trace("s updt sent", tfc_, this); + if (local_window_delta_ > announced_window_delta_) { uint32_t announce = (uint32_t)GPR_CLAMP( - sfc->local_window_delta - sfc->announced_window_delta, 0, UINT32_MAX); - announced_window_delta_preupdate(tfc, sfc); - sfc->announced_window_delta += announce; - announced_window_delta_postupdate(tfc, sfc); - POSTTRACE(tfc, sfc, "s updt sent"); + local_window_delta_ - announced_window_delta_, 0, UINT32_MAX); + UpdateAnnouncedWindowDelta(tfc_, announce); return announce; } - GRPC_FLOW_CONTROL_IF_TRACING( - gpr_log(GPR_DEBUG, "%p[%u][%s] will not send stream update", tfc, - sfc->s->id, tfc->t->is_client ? "cli" : "svr")); return 0; } -// we have received a WINDOW_UPDATE frame for a transport -void grpc_chttp2_flowctl_recv_transport_update( - grpc_chttp2_transport_flowctl* tfc, uint32_t size) { - PRETRACE(tfc, NULL); - tfc->remote_window += size; - POSTTRACE(tfc, NULL, "t updt recv"); -} - -// we have received a WINDOW_UPDATE frame for a stream -void grpc_chttp2_flowctl_recv_stream_update(grpc_chttp2_transport_flowctl* tfc, - grpc_chttp2_stream_flowctl* sfc, - uint32_t size) { - PRETRACE(tfc, sfc); - sfc->remote_window_delta += size; - POSTTRACE(tfc, sfc, "s updt recv"); -} - -void grpc_chttp2_flowctl_incoming_bs_update(grpc_chttp2_transport_flowctl* tfc, - grpc_chttp2_stream_flowctl* sfc, - size_t max_size_hint, - size_t have_already) { - PRETRACE(tfc, sfc); +void StreamFlowControl::IncomingByteStreamUpdate(size_t max_size_hint, + size_t have_already) { + FlowControlTrace trace("app st recv", tfc_, this); uint32_t max_recv_bytes; uint32_t sent_init_window = - tfc->t->settings[GRPC_SENT_SETTINGS] - [GRPC_CHTTP2_SETTINGS_INITIAL_WINDOW_SIZE]; + tfc_->transport()->settings[GRPC_SENT_SETTINGS] + [GRPC_CHTTP2_SETTINGS_INITIAL_WINDOW_SIZE]; /* clamp max recv hint to an allowable size */ if (max_size_hint >= UINT32_MAX - sent_init_window) { @@ -363,65 +251,18 @@ void grpc_chttp2_flowctl_incoming_bs_update(grpc_chttp2_transport_flowctl* tfc, /* add some small lookahead to keep pipelines flowing */ GPR_ASSERT(max_recv_bytes <= UINT32_MAX - sent_init_window); - if (sfc->local_window_delta < max_recv_bytes) { + if (local_window_delta_ < max_recv_bytes) { uint32_t add_max_recv_bytes = - (uint32_t)(max_recv_bytes - sfc->local_window_delta); - sfc->local_window_delta += add_max_recv_bytes; - } - POSTTRACE(tfc, sfc, "app st recv"); -} - -void grpc_chttp2_flowctl_destroy_stream(grpc_chttp2_transport_flowctl* tfc, - grpc_chttp2_stream_flowctl* sfc) { - announced_window_delta_preupdate(tfc, sfc); -} - -// Returns an urgency with which to make an update -static grpc_chttp2_flowctl_urgency delta_is_significant( - const grpc_chttp2_transport_flowctl* tfc, int32_t value, - grpc_chttp2_setting_id setting_id) { - int64_t delta = (int64_t)value - - (int64_t)tfc->t->settings[GRPC_LOCAL_SETTINGS][setting_id]; - // TODO(ncteisen): tune this - if (delta != 0 && (delta <= -value / 5 || delta >= value / 5)) { - return GRPC_CHTTP2_FLOWCTL_QUEUE_UPDATE; - } else { - return GRPC_CHTTP2_FLOWCTL_NO_ACTION_NEEDED; + (uint32_t)(max_recv_bytes - local_window_delta_); + local_window_delta_ += add_max_recv_bytes; } } -// Takes in a target and uses the pid controller to return a stabilized -// guess at the new bdp. -static double get_pid_controller_guess(grpc_exec_ctx* exec_ctx, - grpc_chttp2_transport_flowctl* tfc, - double target) { - grpc_millis now = grpc_exec_ctx_now(exec_ctx); - if (!tfc->pid_controller_initialized) { - tfc->last_pid_update = now; - tfc->pid_controller_initialized = true; - tfc->pid_controller.Init(grpc_core::PidController::Args() - .set_gain_p(4) - .set_gain_i(8) - .set_gain_d(0) - .set_initial_control_value(target) - .set_min_control_value(-1) - .set_max_control_value(25) - .set_integral_range(10)); - return pow(2, target); - } - double bdp_error = target - tfc->pid_controller->last_control_value(); - double dt = (double)(now - tfc->last_pid_update) * 1e-3; - double log2_bdp_guess = tfc->pid_controller->Update(bdp_error, dt); - tfc->last_pid_update = now; - return pow(2, log2_bdp_guess); -} - // Take in a target and modifies it based on the memory pressure of the system -static double get_target_under_memory_pressure( - grpc_chttp2_transport_flowctl* tfc, double target) { +static double AdjustForMemoryPressure(grpc_resource_quota* quota, + double target) { // do not increase window under heavy memory pressure. - double memory_pressure = grpc_resource_quota_get_memory_pressure( - grpc_resource_user_quota(grpc_endpoint_get_resource_user(tfc->t->ep))); + double memory_pressure = grpc_resource_quota_get_memory_pressure(quota); static const double kLowMemPressure = 0.1; static const double kZeroTarget = 22; static const double kHighMemPressure = 0.8; @@ -436,35 +277,19 @@ static double get_target_under_memory_pressure( return target; } -grpc_chttp2_flowctl_action grpc_chttp2_flowctl_get_action( - grpc_exec_ctx* exec_ctx, grpc_chttp2_transport_flowctl* tfc, - grpc_chttp2_stream_flowctl* sfc) { - grpc_chttp2_flowctl_action action; - memset(&action, 0, sizeof(action)); - // TODO(ncteisen): tune this - if (sfc != NULL && !sfc->s->read_closed) { - uint32_t sent_init_window = - tfc->t->settings[GRPC_SENT_SETTINGS] - [GRPC_CHTTP2_SETTINGS_INITIAL_WINDOW_SIZE]; - if ((int64_t)sfc->local_window_delta > - (int64_t)sfc->announced_window_delta && - (int64_t)sfc->announced_window_delta + sent_init_window <= - sent_init_window / 2) { - action.send_stream_update = GRPC_CHTTP2_FLOWCTL_UPDATE_IMMEDIATELY; - } else if (sfc->local_window_delta > sfc->announced_window_delta) { - action.send_stream_update = GRPC_CHTTP2_FLOWCTL_QUEUE_UPDATE; - } - } - if (tfc->enable_bdp_probe) { +FlowControlAction TransportFlowControl::UpdateForBdp(FlowControlAction action) { + if (enable_bdp_probe_) { // get bdp estimate and update initial_window accordingly. int64_t estimate = -1; - if (tfc->bdp_estimator->EstimateBdp(&estimate)) { + if (bdp_estimator_.EstimateBdp(&estimate)) { double target = 1 + log2((double)estimate); // target might change based on how much memory pressure we are under // TODO(ncteisen): experiment with setting target to be huge under low // memory pressure. - target = get_target_under_memory_pressure(tfc, target); + target = AdjustForMemoryPressure( + grpc_resource_user_quota(grpc_endpoint_get_resource_user(t_->ep)), + target); // run our target through the pid controller to stabilize change. // TODO(ncteisen): experiment with other controllers here. @@ -501,6 +326,80 @@ grpc_chttp2_flowctl_action grpc_chttp2_flowctl_get_action( } } } +} + +FlowControlAction TransportFlowControl::MakeAction(grpc_exec_ctx* exec_ctx) { + FlowControlAction action; + action = UpdateForBdp(exec_ctx, action); + return action; +} + +FlowControlAction StreamFlowControl::UpdateAction(FlowControlAction action) { + // TODO(ncteisen): tune this + if (!s_->read_closed) { + uint32_t sent_init_window = + tfc_->transport()->settings[GRPC_SENT_SETTINGS] + [GRPC_CHTTP2_SETTINGS_INITIAL_WINDOW_SIZE]; + if (local_window_delta_ > announced_window_delta_ && + announced_window_delta_ + sent_init_window <= sent_init_window / 2) { + action.set_send_stream_update( + FlowControlAction::Urgency::UPDATE_IMMEDIATELY); + } else if (local_window_delta_ > announced_window_delta_) { + action.set_send_stream_update(FlowControlAction::Urgency::QUEUE_UPDATE); + } + } + + return action; +} + +} // namespace chttp2 +} // namespace grpc_core + +// Returns an urgency with which to make an update +static grpc_chttp2_flowctl_urgency delta_is_significant( + const grpc_chttp2_transport_flowctl* tfc, int32_t value, + grpc_chttp2_setting_id setting_id) { + int64_t delta = (int64_t)value - + (int64_t)tfc->t->settings[GRPC_LOCAL_SETTINGS][setting_id]; + // TODO(ncteisen): tune this + if (delta != 0 && (delta <= -value / 5 || delta >= value / 5)) { + return GRPC_CHTTP2_FLOWCTL_QUEUE_UPDATE; + } else { + return GRPC_CHTTP2_FLOWCTL_NO_ACTION_NEEDED; + } +} + +// Takes in a target and uses the pid controller to return a stabilized +// guess at the new bdp. +static double get_pid_controller_guess(grpc_exec_ctx* exec_ctx, + grpc_chttp2_transport_flowctl* tfc, + double target) { + grpc_millis now = grpc_exec_ctx_now(exec_ctx); + if (!tfc->pid_controller_initialized) { + tfc->last_pid_update = now; + tfc->pid_controller_initialized = true; + tfc->pid_controller.Init(grpc_core::PidController::Args() + .set_gain_p(4) + .set_gain_i(8) + .set_gain_d(0) + .set_initial_control_value(target) + .set_min_control_value(-1) + .set_max_control_value(25) + .set_integral_range(10)); + return pow(2, target); + } + double bdp_error = target - tfc->pid_controller->last_control_value(); + double dt = (double)(now - tfc->last_pid_update) * 1e-3; + double log2_bdp_guess = tfc->pid_controller->Update(bdp_error, dt); + tfc->last_pid_update = now; + return pow(2, log2_bdp_guess); +} + +grpc_chttp2_flowctl_action grpc_chttp2_flowctl_get_action( + grpc_exec_ctx* exec_ctx, grpc_chttp2_transport_flowctl* tfc, + grpc_chttp2_stream_flowctl* sfc) { + grpc_chttp2_flowctl_action action; + memset(&action, 0, sizeof(action)); uint32_t target_announced_window = grpc_chttp2_target_announced_window(tfc); if (tfc->announced_window < target_announced_window / 2) { action.send_transport_update = GRPC_CHTTP2_FLOWCTL_UPDATE_IMMEDIATELY; diff --git a/src/core/ext/transport/chttp2/transport/flow_control.h b/src/core/ext/transport/chttp2/transport/flow_control.h new file mode 100644 index 00000000000..be81d3a97bd --- /dev/null +++ b/src/core/ext/transport/chttp2/transport/flow_control.h @@ -0,0 +1,294 @@ +/* + * + * Copyright 2017 gRPC authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * + */ + +#ifndef GRPC_CORE_EXT_TRANSPORT_CHTTP2_TRANSPORT_FLOW_CONTROL_H +#define GRPC_CORE_EXT_TRANSPORT_CHTTP2_TRANSPORT_FLOW_CONTROL_H + +#include + +#include +#include "src/core/lib/support/manual_constructor.h" +#include "src/core/lib/transport/bdp_estimator.h" +#include "src/core/lib/transport/pid_controller.h" + +struct grpc_chttp2_transport; +struct grpc_chttp2_stream; + +extern "C" grpc_tracer_flag grpc_flowctl_trace; + +namespace grpc_core { +namespace chttp2 { + +static constexpr uint32_t kDefaultWindow = 65535; + +class TransportFlowControl; +class StreamFlowControl; + +class FlowControlAction { + public: + enum class Urgency : uint8_t { + // Nothing to be done. + NO_ACTION_NEEDED = 0, + // Initiate a write to update the initial window immediately. + UPDATE_IMMEDIATELY, + // Push the flow control update into a send buffer, to be sent + // out the next time a write is initiated. + QUEUE_UPDATE, + }; + + Urgency send_stream_update() const { return send_stream_update_; } + Urgency send_transport_update() const { return send_transport_update_; } + Urgency send_setting_update() const { return send_setting_update_; } + uint32_t initial_window_size() const { return initial_window_size_; } + uint32_t max_frame_size() const { return max_frame_size_; } + + FlowControlAction& set_send_stream_update(Urgency u) { + send_stream_update_ = u; + return *this; + } + FlowControlAction& send_transport_update(Urgency u) { + send_transport_update_ = u; + return *this; + } + FlowControlAction& send_setting_update(Urgency u) { + send_setting_update_ = u; + return *this; + } + + static const char* UrgencyString(Urgency u); + void Trace(grpc_chttp2_transport* t) const; + + private: + Urgency send_stream_update_ = Urgency::NO_ACTION_NEEDED; + Urgency send_transport_update_ = Urgency::NO_ACTION_NEEDED; + Urgency send_setting_update_ = Urgency::NO_ACTION_NEEDED; + uint32_t initial_window_size_ = 0; + uint32_t max_frame_size_ = 0; +}; + +class FlowControlTrace { + public: + FlowControlTrace(const char* reason, TransportFlowControl* tfc, + StreamFlowControl* sfc) { + if (enabled_) Init(reason, tfc, sfc); + } + + ~FlowControlTrace() { + if (enabled_) Finish(); + } + + private: + void Init(const char* reason, TransportFlowControl* tfc, + StreamFlowControl* sfc); + void Finish(); + + const bool enabled_ = GRPC_TRACER_ON(grpc_flowctl_trace); + + TransportFlowControl* tfc_; + StreamFlowControl* sfc_; + const char* reason_; + int64_t remote_window_; + int64_t target_window_; + int64_t announced_window_; + int64_t remote_window_delta_; + int64_t local_window_delta_; + int64_t announced_window_delta_; + uint32_t local_init_window_; + uint32_t local_max_frame_; +}; + +class TransportFlowControl { + public: + ~TransportFlowControl() { + if (pid_controller_initialized_) { + pid_controller_.Destroy(); + } + } + + // returns an announce if we should send a transport update to our peer, + // else returns zero; writing_anyway indicates if a write would happen + // regardless of the send - if it is false and this function returns non-zero, + // this announce will cause a write to occur + uint32_t MaybeSendUpdate(bool writing_anyway); + + // Reads the flow control data and returns and actionable struct that will + // tell chttp2 exactly what it needs to do + FlowControlAction MakeAction(grpc_exec_ctx* exec_ctx); + + void StreamSentData(int64_t size) { remote_window_ -= size; } + + grpc_error* ValidateRecvData(int64_t incoming_frame_size); + void CommitRecvData(int64_t incoming_frame_size) { + announced_window_ -= incoming_frame_size; + } + + grpc_error* RecvData(int64_t incoming_frame_size) { + FlowControlTrace trace(" data recv", this, nullptr); + grpc_error* error = ValidateRecvData(incoming_frame_size); + if (error != GRPC_ERROR_NONE) return error; + CommitRecvData(incoming_frame_size); + return GRPC_ERROR_NONE; + } + + // we have received a WINDOW_UPDATE frame for a transport + void RecvUpdate(uint32_t size) { + FlowControlTrace trace("t updt recv", this, nullptr); + remote_window_ += size; + } + + int64_t remote_window() const { return remote_window_; } + int64_t target_window() const { + return (uint32_t)GPR_MIN((int64_t)((1u << 31) - 1), + announced_stream_total_over_incoming_window_ + + target_initial_window_size_); + } + int64_t announced_window() const { return announced_window_; } + + const grpc_chttp2_transport* transport() const { return t_; } + + void PreUpdateAnnouncedWindowOverIncomingWindow(int64_t delta) { + if (delta > 0) { + announced_stream_total_over_incoming_window_ -= delta; + } else { + announced_stream_total_under_incoming_window_ += -delta; + } + } + + void PostUpdateAnnouncedWindowOverIncomingWindow(int64_t delta) { + if (delta > 0) { + announced_stream_total_over_incoming_window_ += delta; + } else { + announced_stream_total_under_incoming_window_ -= -delta; + } + } + + private: + FlowControlAction UpdateForBdp(grpc_exec_ctx* exec_ctx, + FlowControlAction action); + double SmoothBdp(grpc_exec_ctx* exec_ctx, double value); + + const grpc_chttp2_transport* const t_; + + /** initial window change. This is tracked as we parse settings frames from + * the remote peer. If there is a positive delta, then we will make all + * streams readable since they may have become unstalled */ + int64_t initial_window_update_ = 0; + + /** Our bookkeeping for the remote peer's available window */ + int64_t remote_window_ = kDefaultWindow; + + /** calculating what we should give for local window: + we track the total amount of flow control over initial window size + across all streams: this is data that we want to receive right now (it + has an outstanding read) + and the total amount of flow control under initial window size across all + streams: this is data we've read early + we want to adjust incoming_window such that: + incoming_window = total_over - max(bdp - total_under, 0) */ + int64_t announced_stream_total_over_incoming_window_ = 0; + int64_t announced_stream_total_under_incoming_window_ = 0; + + /** This is out window according to what we have sent to our remote peer. The + * difference between this and target window is what we use to decide when + * to send WINDOW_UPDATE frames. */ + int64_t announced_window_ = kDefaultWindow; + + int32_t target_initial_window_size_ = kDefaultWindow; + + /** should we probe bdp? */ + bool enable_bdp_probe_; + + /* bdp estimation */ + grpc_core::BdpEstimator bdp_estimator_; + + /* pid controller */ + bool pid_controller_initialized_; + grpc_core::ManualConstructor pid_controller_; + grpc_millis last_pid_update_; +}; + +class StreamFlowControl { + public: + StreamFlowControl(TransportFlowControl* tfc, grpc_chttp2_stream* s); + ~StreamFlowControl() { + tfc_->PreUpdateAnnouncedWindowOverIncomingWindow(announced_window_delta_); + } + + FlowControlAction UpdateAction(FlowControlAction action); + FlowControlAction MakeAction() { return UpdateAction(FlowControlAction()); } + + // we have sent data on the wire, we must track this in our bookkeeping for + // the remote peer's flow control. + void SentData(int64_t outgoing_frame_size) { + FlowControlTrace tracer(" data sent", tfc_, this); + tfc_->StreamSentData(outgoing_frame_size); + remote_window_delta_ -= outgoing_frame_size; + } + + // we have received data from the wire + grpc_error* RecvData(int64_t incoming_frame_size); + + // returns an announce if we should send a stream update to our peer, else + // returns zero + uint32_t MaybeSendUpdate(); + + // we have received a WINDOW_UPDATE frame for a stream + void RecvUpdate(uint32_t size) { + FlowControlTrace trace("s updt recv", tfc_, this); + remote_window_delta_ += size; + } + + // the application is asking for a certain amount of bytes + void IncomingByteStreamUpdate(size_t max_size_hint, size_t have_already); + + int64_t remote_window_delta() const { return remote_window_delta_; } + int64_t local_window_delta() const { return local_window_delta_; } + int64_t announced_window_delta() const { return announced_window_delta_; } + + const grpc_chttp2_stream* stream() const { return s_; } + + private: + TransportFlowControl* const tfc_; + const grpc_chttp2_stream* const s_; + + void UpdateAnnouncedWindowDelta(TransportFlowControl* tfc, int64_t change) { + tfc->PreUpdateAnnouncedWindowOverIncomingWindow(announced_window_delta_); + announced_window_delta_ += change; + tfc->PostUpdateAnnouncedWindowOverIncomingWindow(announced_window_delta_); + } + + /** window available for us to send to peer, over or under the initial + * window + * size of the transport... ie: + * remote_window = remote_window_delta + transport.initial_window_size */ + int64_t remote_window_delta_; + + /** window available for peer to send to us (as a delta on + * transport.initial_window_size) + * local_window = local_window_delta + transport.initial_window_size */ + int64_t local_window_delta_; + + /** window available for peer to send to us over this stream that we have + * announced to the peer */ + int64_t announced_window_delta_; +}; + +} // namespace chttp2 +} // namespace grpc_core + +#endif diff --git a/src/core/ext/transport/chttp2/transport/internal.h b/src/core/ext/transport/chttp2/transport/internal.h index f0a75dfb458..57c4cfa5f3a 100644 --- a/src/core/ext/transport/chttp2/transport/internal.h +++ b/src/core/ext/transport/chttp2/transport/internal.h @@ -22,6 +22,7 @@ #include #include +#include "src/core/ext/transport/chttp2/transport/flow_control.h" #include "src/core/ext/transport/chttp2/transport/frame.h" #include "src/core/ext/transport/chttp2/transport/frame_data.h" #include "src/core/ext/transport/chttp2/transport/frame_goaway.h" @@ -38,9 +39,7 @@ #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" #include "src/core/lib/transport/transport_impl.h" #ifdef __cplusplus @@ -238,48 +237,6 @@ typedef enum { GRPC_CHTTP2_KEEPALIVE_STATE_DISABLED, } grpc_chttp2_keepalive_state; -typedef struct { - /** initial window change. This is tracked as we parse settings frames from - * the remote peer. If there is a positive delta, then we will make all - * streams readable since they may have become unstalled */ - int64_t initial_window_update; - - /** Our bookkeeping for the remote peer's available window */ - int64_t remote_window; - - /** calculating what we should give for local window: - we track the total amount of flow control over initial window size - across all streams: this is data that we want to receive right now (it - has an outstanding read) - and the total amount of flow control under initial window size across all - streams: this is data we've read early - we want to adjust incoming_window such that: - incoming_window = total_over - max(bdp - total_under, 0) */ - int64_t announced_stream_total_over_incoming_window; - int64_t announced_stream_total_under_incoming_window; - - /** This is out window according to what we have sent to our remote peer. The - * difference between this and target window is what we use to decide when - * to send WINDOW_UPDATE frames. */ - int64_t announced_window; - - int32_t target_initial_window_size; - - /** should we probe bdp? */ - bool enable_bdp_probe; - - /* bdp estimation */ - grpc_core::ManualConstructor bdp_estimator; - - /* pid controller */ - bool pid_controller_initialized; - grpc_core::ManualConstructor pid_controller; - grpc_millis last_pid_update; - - // pointer back to transport for tracing - const grpc_chttp2_transport *t; -} grpc_chttp2_transport_flowctl; - struct grpc_chttp2_transport { grpc_transport base; /* must be first */ gpr_refcount refs; @@ -395,7 +352,8 @@ struct grpc_chttp2_transport { /** parser for goaway frames */ grpc_chttp2_goaway_parser goaway_parser; - grpc_chttp2_transport_flowctl flow_control; + grpc_core::ManualConstructor + flow_control; /* deframing */ grpc_chttp2_deframe_transport_state deframe_state; @@ -477,25 +435,6 @@ typedef enum { GPRC_METADATA_PUBLISHED_AT_CLOSE } grpc_published_metadata_method; -typedef struct { - /** window available for us to send to peer, over or under the initial window - * size of the transport... ie: - * remote_window = remote_window_delta + transport.initial_window_size */ - int64_t remote_window_delta; - - /** window available for peer to send to us (as a delta on - * transport.initial_window_size) - * local_window = local_window_delta + transport.initial_window_size */ - int64_t local_window_delta; - - /** window available for peer to send to us over this stream that we have - * announced to the peer */ - int64_t announced_window_delta; - - // read only pointer back to stream for data - const grpc_chttp2_stream *s; -} grpc_chttp2_stream_flowctl; - struct grpc_chttp2_stream { grpc_chttp2_transport *t; grpc_stream_refcount *refcount; @@ -589,7 +528,8 @@ struct grpc_chttp2_stream { bool sent_initial_metadata; bool sent_trailing_metadata; - grpc_chttp2_stream_flowctl flow_control; + grpc_core::ManualConstructor + flow_control; grpc_slice_buffer flow_controlled_buffer; @@ -700,73 +640,10 @@ bool grpc_chttp2_list_remove_stalled_by_stream(grpc_chttp2_transport *t, /********* Flow Control ***************/ -// we have sent data on the wire -void grpc_chttp2_flowctl_sent_data(grpc_chttp2_transport_flowctl *tfc, - grpc_chttp2_stream_flowctl *sfc, - int64_t size); - -// we have received data from the wire -grpc_error *grpc_chttp2_flowctl_recv_data(grpc_chttp2_transport_flowctl *tfc, - grpc_chttp2_stream_flowctl *sfc, - int64_t incoming_frame_size); - -// returns an announce if we should send a transport update to our peer, -// else returns zero -uint32_t grpc_chttp2_flowctl_maybe_send_transport_update( - grpc_chttp2_transport_flowctl *tfc, bool writing_anyway); - -// returns an announce if we should send a stream update to our peer, else -// returns zero -uint32_t grpc_chttp2_flowctl_maybe_send_stream_update( - grpc_chttp2_transport_flowctl *tfc, grpc_chttp2_stream_flowctl *sfc); - -// we have received a WINDOW_UPDATE frame for a transport -void grpc_chttp2_flowctl_recv_transport_update( - grpc_chttp2_transport_flowctl *tfc, uint32_t size); - -// we have received a WINDOW_UPDATE frame for a stream -void grpc_chttp2_flowctl_recv_stream_update(grpc_chttp2_transport_flowctl *tfc, - grpc_chttp2_stream_flowctl *sfc, - uint32_t size); - -// the application is asking for a certain amount of bytes -void grpc_chttp2_flowctl_incoming_bs_update(grpc_chttp2_transport_flowctl *tfc, - grpc_chttp2_stream_flowctl *sfc, - size_t max_size_hint, - size_t have_already); - -void grpc_chttp2_flowctl_destroy_stream(grpc_chttp2_transport_flowctl *tfc, - grpc_chttp2_stream_flowctl *sfc); - -typedef enum { - // Nothing to be done. - GRPC_CHTTP2_FLOWCTL_NO_ACTION_NEEDED = 0, - // Initiate a write to update the initial window immediately. - GRPC_CHTTP2_FLOWCTL_UPDATE_IMMEDIATELY, - // Push the flow control update into a send buffer, to be sent - // out the next time a write is initiated. - GRPC_CHTTP2_FLOWCTL_QUEUE_UPDATE, -} grpc_chttp2_flowctl_urgency; - -typedef struct { - grpc_chttp2_flowctl_urgency send_stream_update; - grpc_chttp2_flowctl_urgency send_transport_update; - grpc_chttp2_flowctl_urgency send_setting_update; - uint32_t initial_window_size; - uint32_t max_frame_size; -} grpc_chttp2_flowctl_action; - -// Reads the flow control data and returns and actionable struct that will tell -// chttp2 exactly what it needs to do -grpc_chttp2_flowctl_action grpc_chttp2_flowctl_get_action( - grpc_exec_ctx *exec_ctx, grpc_chttp2_transport_flowctl *tfc, - grpc_chttp2_stream_flowctl *sfc); - // Takes in a flow control action and performs all the needed operations. -void grpc_chttp2_act_on_flowctl_action(grpc_exec_ctx *exec_ctx, - grpc_chttp2_flowctl_action action, - grpc_chttp2_transport *t, - grpc_chttp2_stream *s); +void grpc_chttp2_act_on_flowctl_action( + grpc_exec_ctx *exec_ctx, const grpc_core::chttp2::FlowControlAction &action, + grpc_chttp2_transport *t, grpc_chttp2_stream *s); /********* End of Flow Control ***************/