|
|
|
|
@ -18,6 +18,7 @@ |
|
|
|
|
|
|
|
|
|
#include "src/core/ext/transport/chttp2/transport/internal.h" |
|
|
|
|
|
|
|
|
|
#include <math.h> |
|
|
|
|
#include <string.h> |
|
|
|
|
|
|
|
|
|
#include <grpc/support/alloc.h> |
|
|
|
|
@ -39,6 +40,8 @@ typedef struct { |
|
|
|
|
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, |
|
|
|
|
@ -54,14 +57,28 @@ static void pretrace(shadow_flow_control* shadow_fc, |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
static char* fmt_str(int64_t old, int64_t new) { |
|
|
|
|
#define TRACE_PADDING 30 |
|
|
|
|
|
|
|
|
|
static char* fmt_int64_diff_str(int64_t old, int64_t new) { |
|
|
|
|
char* str; |
|
|
|
|
if (old != new) { |
|
|
|
|
gpr_asprintf(&str, "%" PRId64 " -> %" PRId64 "", old, new); |
|
|
|
|
} else { |
|
|
|
|
gpr_asprintf(&str, "%" PRId64 "", old); |
|
|
|
|
} |
|
|
|
|
char* str_lp = gpr_leftpad(str, ' ', 30); |
|
|
|
|
char* str_lp = gpr_leftpad(str, ' ', TRACE_PADDING); |
|
|
|
|
gpr_free(str); |
|
|
|
|
return str_lp; |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
static char* fmt_uint32_diff_str(uint32_t old, uint32_t new) { |
|
|
|
|
char* str; |
|
|
|
|
if (new > 0 && old != new) { |
|
|
|
|
gpr_asprintf(&str, "%" PRIu32 " -> %" PRIu32 "", old, new); |
|
|
|
|
} else { |
|
|
|
|
gpr_asprintf(&str, "%" PRIu32 "", old); |
|
|
|
|
} |
|
|
|
|
char* str_lp = gpr_leftpad(str, ' ', TRACE_PADDING); |
|
|
|
|
gpr_free(str); |
|
|
|
|
return str_lp; |
|
|
|
|
} |
|
|
|
|
@ -75,24 +92,28 @@ static void posttrace(shadow_flow_control* shadow_fc, |
|
|
|
|
uint32_t remote_window = |
|
|
|
|
tfc->t->settings[GRPC_PEER_SETTINGS] |
|
|
|
|
[GRPC_CHTTP2_SETTINGS_INITIAL_WINDOW_SIZE]; |
|
|
|
|
char* trw_str = fmt_str(shadow_fc->remote_window, tfc->remote_window); |
|
|
|
|
char* tlw_str = fmt_str(shadow_fc->target_window, |
|
|
|
|
grpc_chttp2_target_announced_window(tfc)); |
|
|
|
|
char* taw_str = fmt_str(shadow_fc->announced_window, tfc->announced_window); |
|
|
|
|
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)); |
|
|
|
|
char* taw_str = |
|
|
|
|
fmt_int64_diff_str(shadow_fc->announced_window, tfc->announced_window); |
|
|
|
|
char* srw_str; |
|
|
|
|
char* slw_str; |
|
|
|
|
char* saw_str; |
|
|
|
|
if (sfc != NULL) { |
|
|
|
|
srw_str = fmt_str(shadow_fc->remote_window_delta + remote_window, |
|
|
|
|
sfc->remote_window_delta + remote_window); |
|
|
|
|
slw_str = fmt_str(shadow_fc->local_window_delta + acked_local_window, |
|
|
|
|
sfc->local_window_delta + acked_local_window); |
|
|
|
|
saw_str = fmt_str(shadow_fc->announced_window_delta + acked_local_window, |
|
|
|
|
sfc->announced_window_delta + acked_local_window); |
|
|
|
|
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); |
|
|
|
|
} else { |
|
|
|
|
srw_str = gpr_leftpad("", ' ', 30); |
|
|
|
|
slw_str = gpr_leftpad("", ' ', 30); |
|
|
|
|
saw_str = gpr_leftpad("", ' ', 30); |
|
|
|
|
srw_str = gpr_leftpad("", ' ', TRACE_PADDING); |
|
|
|
|
slw_str = gpr_leftpad("", ' ', TRACE_PADDING); |
|
|
|
|
saw_str = gpr_leftpad("", ' ', TRACE_PADDING); |
|
|
|
|
} |
|
|
|
|
gpr_log(GPR_DEBUG, |
|
|
|
|
"%p[%u][%s] | %s | trw:%s, ttw:%s, taw:%s, srw:%s, slw:%s, saw:%s", |
|
|
|
|
@ -120,10 +141,21 @@ static char* urgency_to_string(grpc_chttp2_flowctl_urgency urgency) { |
|
|
|
|
GPR_UNREACHABLE_CODE(return "unknown"); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
static void trace_action(grpc_chttp2_flowctl_action action) { |
|
|
|
|
gpr_log(GPR_DEBUG, "transport: %s, stream: %s", |
|
|
|
|
static void trace_action(grpc_chttp2_transport_flowctl* tfc, |
|
|
|
|
grpc_chttp2_flowctl_action action) { |
|
|
|
|
char* iw_str = fmt_uint32_diff_str( |
|
|
|
|
tfc->t->settings[GRPC_SENT_SETTINGS] |
|
|
|
|
[GRPC_CHTTP2_SETTINGS_INITIAL_WINDOW_SIZE], |
|
|
|
|
action.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); |
|
|
|
|
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_stream_update), |
|
|
|
|
urgency_to_string(action.send_setting_update), iw_str, mf_str); |
|
|
|
|
gpr_free(iw_str); |
|
|
|
|
gpr_free(mf_str); |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
#define PRETRACE(tfc, sfc) \ |
|
|
|
|
@ -131,11 +163,12 @@ static void trace_action(grpc_chttp2_flowctl_action action) { |
|
|
|
|
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(action) GRPC_FLOW_CONTROL_IF_TRACING(trace_action(action)) |
|
|
|
|
#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(action) |
|
|
|
|
#define TRACEACTION(tfc, action) |
|
|
|
|
#endif |
|
|
|
|
|
|
|
|
|
/* How many bytes of incoming flow control would we like to advertise */ |
|
|
|
|
@ -342,15 +375,58 @@ void grpc_chttp2_flowctl_destroy_stream(grpc_chttp2_transport_flowctl* tfc, |
|
|
|
|
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; |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
// 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_chttp2_transport_flowctl* tfc, |
|
|
|
|
double target) { |
|
|
|
|
double bdp_error = target - grpc_pid_controller_last(&tfc->pid_controller); |
|
|
|
|
gpr_timespec now = gpr_now(GPR_CLOCK_MONOTONIC); |
|
|
|
|
gpr_timespec dt_timespec = gpr_time_sub(now, tfc->last_pid_update); |
|
|
|
|
double dt = (double)dt_timespec.tv_sec + dt_timespec.tv_nsec * 1e-9; |
|
|
|
|
if (dt > 0.1) { |
|
|
|
|
dt = 0.1; |
|
|
|
|
} |
|
|
|
|
double log2_bdp_guess = |
|
|
|
|
grpc_pid_controller_update(&tfc->pid_controller, 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) { |
|
|
|
|
// 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))); |
|
|
|
|
if (memory_pressure > 0.8) { |
|
|
|
|
target *= 1 - GPR_MIN(1, (memory_pressure - 0.8) / 0.1); |
|
|
|
|
} |
|
|
|
|
return target; |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
grpc_chttp2_flowctl_action grpc_chttp2_flowctl_get_action( |
|
|
|
|
const grpc_chttp2_transport_flowctl* tfc, |
|
|
|
|
const grpc_chttp2_stream_flowctl* sfc) { |
|
|
|
|
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; |
|
|
|
|
} |
|
|
|
|
// TODO(ncteisen): tune this
|
|
|
|
|
if (sfc != NULL && !sfc->s->read_closed) { |
|
|
|
|
uint32_t sent_init_window = |
|
|
|
|
tfc->t->settings[GRPC_SENT_SETTINGS] |
|
|
|
|
@ -364,6 +440,61 @@ grpc_chttp2_flowctl_action grpc_chttp2_flowctl_get_action( |
|
|
|
|
action.send_stream_update = GRPC_CHTTP2_FLOWCTL_QUEUE_UPDATE; |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
TRACEACTION(action); |
|
|
|
|
TRACEACTION(tfc, action); |
|
|
|
|
return action; |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
grpc_chttp2_flowctl_action grpc_chttp2_flowctl_get_bdp_action( |
|
|
|
|
grpc_chttp2_transport_flowctl* tfc) { |
|
|
|
|
grpc_chttp2_flowctl_action action; |
|
|
|
|
memset(&action, 0, sizeof(action)); |
|
|
|
|
if (tfc->enable_bdp_probe) { |
|
|
|
|
action.need_ping = grpc_bdp_estimator_need_ping(&tfc->bdp_estimator); |
|
|
|
|
|
|
|
|
|
// get bdp estimate and update initial_window accordingly.
|
|
|
|
|
int64_t estimate = -1; |
|
|
|
|
int32_t bdp = -1; |
|
|
|
|
if (grpc_bdp_estimator_get_estimate(&tfc->bdp_estimator, &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); |
|
|
|
|
|
|
|
|
|
// run our target through the pid controller to stabilize change.
|
|
|
|
|
// TODO(ncteisen): experiment with other controllers here.
|
|
|
|
|
double bdp_guess = get_pid_controller_guess(tfc, target); |
|
|
|
|
|
|
|
|
|
// Though initial window 'could' drop to 0, we keep the floor at 128
|
|
|
|
|
bdp = GPR_MAX((int32_t)bdp_guess, 128); |
|
|
|
|
|
|
|
|
|
grpc_chttp2_flowctl_urgency init_window_update_urgency = |
|
|
|
|
delta_is_significant(tfc, bdp, |
|
|
|
|
GRPC_CHTTP2_SETTINGS_INITIAL_WINDOW_SIZE); |
|
|
|
|
if (init_window_update_urgency != GRPC_CHTTP2_FLOWCTL_NO_ACTION_NEEDED) { |
|
|
|
|
action.send_setting_update = init_window_update_urgency; |
|
|
|
|
action.initial_window_size = (uint32_t)bdp; |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
// get bandwidth estimate and update max_frame accordingly.
|
|
|
|
|
double bw_dbl = -1; |
|
|
|
|
if (grpc_bdp_estimator_get_bw(&tfc->bdp_estimator, &bw_dbl)) { |
|
|
|
|
// we target the max of BDP or bandwidth in microseconds.
|
|
|
|
|
int32_t frame_size = |
|
|
|
|
GPR_CLAMP(GPR_MAX((int32_t)bw_dbl / 1000, bdp), 16384, 16777215); |
|
|
|
|
grpc_chttp2_flowctl_urgency frame_size_urgency = delta_is_significant( |
|
|
|
|
tfc, frame_size, GRPC_CHTTP2_SETTINGS_MAX_FRAME_SIZE); |
|
|
|
|
if (frame_size_urgency != GRPC_CHTTP2_FLOWCTL_NO_ACTION_NEEDED) { |
|
|
|
|
if (frame_size_urgency > action.send_setting_update) { |
|
|
|
|
action.send_setting_update = frame_size_urgency; |
|
|
|
|
} |
|
|
|
|
action.max_frame_size = (uint32_t)frame_size; |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
TRACEACTION(tfc, action); |
|
|
|
|
return action; |
|
|
|
|
} |
|
|
|
|
|
Noah Eisen
7 years ago
committed by
GitHub