Chiebot-Mirror
/
grpc
mirror of https://github.com/grpc/grpc.git
8
0
Fork 0
Code Issues Projects Releases Wiki Activity

First attempt at periodic updates to flow control

Browse Source
pull/12915/head
Craig Tiller 7 years ago
parent 6d5258cdac
commit b278bc7018
4 changed files with 38 additions and 35 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Split View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 32
      src/core/ext/transport/chttp2/transport/chttp2_transport.cc
  2. 17
      src/core/ext/transport/chttp2/transport/flow_control.cc
  3. 20
      src/core/ext/transport/chttp2/transport/flow_control.h
  4. 4
      src/core/ext/transport/chttp2/transport/parsing.cc

32
src/core/ext/transport/chttp2/transport/chttp2_transport.cc
Unescape View File

@ -565,13 +565,13 @@ 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) {
if (t->flow_control->bdp_probe()) {
GRPC_CHTTP2_REF_TRANSPORT(t, "bdp_ping");
schedule_bdp_ping_locked(exec_ctx, t);
}
grpc_chttp2_act_on_flowctl_action(
exec_ctx, t->flow_control->MakeAction(exec_ctx), t, NULL);
exec_ctx, t->flow_control->PeriodicUpdate(exec_ctx), t, NULL);
grpc_chttp2_initiate_write(exec_ctx, t,
GRPC_CHTTP2_INITIATE_WRITE_INITIAL_WRITE);
@ -1636,8 +1636,8 @@ static void perform_stream_op_locked(grpc_exec_ctx *exec_ctx, void *stream_op,
already_received = s->frame_storage.length;
s->flow_control->IncomingByteStreamUpdate(GRPC_HEADER_SIZE_IN_BYTES,
already_received);
grpc_chttp2_act_on_flowctl_action(
exec_ctx, s->flow_control->MakeAction(exec_ctx), t, s);
grpc_chttp2_act_on_flowctl_action(exec_ctx,
s->flow_control->MakeAction(), t, s);
}
}
grpc_chttp2_maybe_complete_recv_message(exec_ctx, t, s);
@ -1768,10 +1768,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_set_int(
GRPC_ERROR_CREATE_FROM_STATIC_STRING("Too many pings"),
GRPC_ERROR_INT_GRPC_STATUS, GRPC_STATUS_UNAVAILABLE));
exec_ctx, t, grpc_error_set_int(
GRPC_ERROR_CREATE_FROM_STATIC_STRING("Too many pings"),
GRPC_ERROR_INT_GRPC_STATUS, GRPC_STATUS_UNAVAILABLE));
}
}
@ -2550,8 +2549,8 @@ static void read_action_locked(grpc_exec_ctx *exec_ctx, void *tp,
if (keep_reading) {
grpc_endpoint_read(exec_ctx, t->ep, &t->read_buffer,
&t->read_action_locked);
grpc_chttp2_act_on_flowctl_action(
exec_ctx, t->flow_control->MakeAction(exec_ctx), t, NULL);
grpc_chttp2_act_on_flowctl_action(exec_ctx, t->flow_control->MakeAction(),
t, NULL);
GRPC_CHTTP2_UNREF_TRANSPORT(exec_ctx, t, "keep_reading");
} else {
GRPC_CHTTP2_UNREF_TRANSPORT(exec_ctx, t, "reading_action");
@ -2598,6 +2597,8 @@ static void finish_bdp_ping_locked(grpc_exec_ctx *exec_ctx, void *tp,
}
grpc_millis next_ping =
t->flow_control->bdp_estimator()->CompletePing(exec_ctx);
grpc_chttp2_act_on_flowctl_action(
exec_ctx, t->flow_control->PeriodicUpdate(exec_ctx), t, nullptr);
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,
@ -2736,8 +2737,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_set_int(GRPC_ERROR_CREATE_FROM_STATIC_STRING(
"keepalive watchdog timeout"), GRPC_ERROR_INT_GRPC_STATUS, GRPC_STATUS_INTERNAL));
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
@ -2822,8 +2826,8 @@ static void incoming_byte_stream_next_locked(grpc_exec_ctx *exec_ctx,
if (!s->read_closed) {
s->flow_control->IncomingByteStreamUpdate(bs->next_action.max_size_hint,
cur_length);
grpc_chttp2_act_on_flowctl_action(
exec_ctx, s->flow_control->MakeAction(exec_ctx), t, s);
grpc_chttp2_act_on_flowctl_action(exec_ctx, s->flow_control->MakeAction(),
t, s);
}
GPR_ASSERT(s->unprocessed_incoming_frames_buffer.length == 0);
if (s->frame_storage.length > 0) {

17
src/core/ext/transport/chttp2/transport/flow_control.cc
Unescape View File

@ -321,8 +321,9 @@ FlowControlAction::Urgency TransportFlowControl::DeltaUrgency(
}
}
FlowControlAction TransportFlowControl::UpdateForBdp(grpc_exec_ctx* exec_ctx,
FlowControlAction action) {
FlowControlAction TransportFlowControl::PeriodicUpdate(
grpc_exec_ctx* exec_ctx) {
FlowControlAction action;
if (enable_bdp_probe_) {
// get bdp estimate and update initial_window accordingly.
int64_t estimate = -1;
@ -359,17 +360,7 @@ FlowControlAction TransportFlowControl::UpdateForBdp(grpc_exec_ctx* exec_ctx,
frame_size);
}
}
return action;
}
FlowControlAction TransportFlowControl::MakeAction(grpc_exec_ctx* exec_ctx) {
FlowControlAction action;
if (announced_window_ < target_window() / 2) {
action.set_send_transport_update(
FlowControlAction::Urgency::UPDATE_IMMEDIATELY);
}
action = UpdateForBdp(exec_ctx, action);
return action;
return UpdateAction(action);
}
FlowControlAction StreamFlowControl::UpdateAction(FlowControlAction action) {

20
src/core/ext/transport/chttp2/transport/flow_control.h
Unescape View File

@ -138,6 +138,8 @@ class TransportFlowControl {
// TODO(ctiller): make this safe to dynamically toggle
void SetBdpProbe(bool enable) { enable_bdp_probe_ = enable; }
bool bdp_probe() const { return enable_bdp_probe_; }
// 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,
@ -146,7 +148,7 @@ class TransportFlowControl {
// 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);
FlowControlAction MakeAction() { return UpdateAction(FlowControlAction()); }
void StreamSentData(int64_t size) { remote_window_ -= size; }
@ -197,13 +199,21 @@ class TransportFlowControl {
BdpEstimator* bdp_estimator() { return &bdp_estimator_; }
FlowControlAction PeriodicUpdate(grpc_exec_ctx* exec_ctx);
private:
FlowControlAction UpdateForBdp(grpc_exec_ctx* exec_ctx,
FlowControlAction action);
double SmoothLogBdp(grpc_exec_ctx* exec_ctx, double value);
FlowControlAction::Urgency DeltaUrgency(int32_t value,
grpc_chttp2_setting_id setting_id);
FlowControlAction UpdateAction(FlowControlAction action) {
if (announced_window_ < target_window() / 2) {
action.set_send_transport_update(
FlowControlAction::Urgency::UPDATE_IMMEDIATELY);
}
return action;
}
const grpc_chttp2_transport* const t_;
/** Our bookkeeping for the remote peer's available window */
@ -247,9 +257,7 @@ class StreamFlowControl {
}
FlowControlAction UpdateAction(FlowControlAction action);
FlowControlAction MakeAction(grpc_exec_ctx* exec_ctx) {
return UpdateAction(tfc_->MakeAction(exec_ctx));
}
FlowControlAction MakeAction() { return UpdateAction(tfc_->MakeAction()); }
// we have sent data on the wire, we must track this in our bookkeeping for
// the remote peer's flow control.

4
src/core/ext/transport/chttp2/transport/parsing.cc
Unescape View File

@ -358,10 +358,10 @@ static grpc_error *init_data_frame_parser(grpc_exec_ctx *exec_ctx,
grpc_core::chttp2::FlowControlAction action;
if (s == nullptr) {
err = t->flow_control->RecvData(t->incoming_frame_size);
action = t->flow_control->MakeAction(exec_ctx);
action = t->flow_control->MakeAction();
} else {
err = s->flow_control->RecvData(t->incoming_frame_size);
action = s->flow_control->MakeAction(exec_ctx);
action = s->flow_control->MakeAction();
}
grpc_chttp2_act_on_flowctl_action(exec_ctx, action, t, s);
if (err != GRPC_ERROR_NONE) {

Write Preview
Loading…
Cancel
Save