diff options
| author |  ncteisen <ncteisen@gmail.com> | 2017-07-06 16:23:14 -0700 |
|---|---|---|
| committer | ncteisen <ncteisen@gmail.com> | 2017-07-18 17:06:57 -0700 |
| commit | 268a82398c0b71a0ed9cb4f5168098f1de8945da (patch) | |
| tree | 6abbea6f3842f82668ac12272162d00f2c79cc78 | |
| parent | 10bed24e8b6a93d9017927371e607f85619fb59b (diff) | |
Pull flow control into one module
27 files changed, 571 insertions, 428 deletions
@@ -1212,6 +1212,7 @@ grpc_cc_library( "src/core/ext/transport/chttp2/transport/bin_encoder.c", "src/core/ext/transport/chttp2/transport/chttp2_plugin.c", "src/core/ext/transport/chttp2/transport/chttp2_transport.c", + "src/core/ext/transport/chttp2/transport/flow_control.c", "src/core/ext/transport/chttp2/transport/frame_data.c", "src/core/ext/transport/chttp2/transport/frame_goaway.c", "src/core/ext/transport/chttp2/transport/frame_ping.c", diff --git a/CMakeLists.txt b/CMakeLists.txt index 0f87beb02b..f8d9fd11cd 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -1078,6 +1078,7 @@ add_library(grpc src/core/ext/transport/chttp2/transport/bin_encoder.c src/core/ext/transport/chttp2/transport/chttp2_plugin.c src/core/ext/transport/chttp2/transport/chttp2_transport.c + src/core/ext/transport/chttp2/transport/flow_control.c src/core/ext/transport/chttp2/transport/frame_data.c src/core/ext/transport/chttp2/transport/frame_goaway.c src/core/ext/transport/chttp2/transport/frame_ping.c @@ -1424,6 +1425,7 @@ add_library(grpc_cronet src/core/ext/transport/chttp2/transport/bin_encoder.c src/core/ext/transport/chttp2/transport/chttp2_plugin.c src/core/ext/transport/chttp2/transport/chttp2_transport.c + src/core/ext/transport/chttp2/transport/flow_control.c src/core/ext/transport/chttp2/transport/frame_data.c src/core/ext/transport/chttp2/transport/frame_goaway.c src/core/ext/transport/chttp2/transport/frame_ping.c @@ -2006,6 +2008,7 @@ add_library(grpc_unsecure src/core/ext/transport/chttp2/transport/bin_encoder.c src/core/ext/transport/chttp2/transport/chttp2_plugin.c src/core/ext/transport/chttp2/transport/chttp2_transport.c + src/core/ext/transport/chttp2/transport/flow_control.c src/core/ext/transport/chttp2/transport/frame_data.c src/core/ext/transport/chttp2/transport/frame_goaway.c src/core/ext/transport/chttp2/transport/frame_ping.c @@ -2516,6 +2519,7 @@ add_library(grpc++_cronet src/core/ext/transport/chttp2/transport/bin_encoder.c src/core/ext/transport/chttp2/transport/chttp2_plugin.c src/core/ext/transport/chttp2/transport/chttp2_transport.c + src/core/ext/transport/chttp2/transport/flow_control.c src/core/ext/transport/chttp2/transport/frame_data.c src/core/ext/transport/chttp2/transport/frame_goaway.c src/core/ext/transport/chttp2/transport/frame_ping.c @@ -3022,6 +3022,7 @@ LIBGRPC_SRC = \ src/core/ext/transport/chttp2/transport/bin_encoder.c \ src/core/ext/transport/chttp2/transport/chttp2_plugin.c \ src/core/ext/transport/chttp2/transport/chttp2_transport.c \ + src/core/ext/transport/chttp2/transport/flow_control.c \ src/core/ext/transport/chttp2/transport/frame_data.c \ src/core/ext/transport/chttp2/transport/frame_goaway.c \ src/core/ext/transport/chttp2/transport/frame_ping.c \ @@ -3366,6 +3367,7 @@ LIBGRPC_CRONET_SRC = \ src/core/ext/transport/chttp2/transport/bin_encoder.c \ src/core/ext/transport/chttp2/transport/chttp2_plugin.c \ src/core/ext/transport/chttp2/transport/chttp2_transport.c \ + src/core/ext/transport/chttp2/transport/flow_control.c \ src/core/ext/transport/chttp2/transport/frame_data.c \ src/core/ext/transport/chttp2/transport/frame_goaway.c \ src/core/ext/transport/chttp2/transport/frame_ping.c \ @@ -3917,6 +3919,7 @@ LIBGRPC_UNSECURE_SRC = \ src/core/ext/transport/chttp2/transport/bin_encoder.c \ src/core/ext/transport/chttp2/transport/chttp2_plugin.c \ src/core/ext/transport/chttp2/transport/chttp2_transport.c \ + src/core/ext/transport/chttp2/transport/flow_control.c \ src/core/ext/transport/chttp2/transport/frame_data.c \ src/core/ext/transport/chttp2/transport/frame_goaway.c \ src/core/ext/transport/chttp2/transport/frame_ping.c \ @@ -4411,6 +4414,7 @@ LIBGRPC++_CRONET_SRC = \ src/core/ext/transport/chttp2/transport/bin_encoder.c \ src/core/ext/transport/chttp2/transport/chttp2_plugin.c \ src/core/ext/transport/chttp2/transport/chttp2_transport.c \ + src/core/ext/transport/chttp2/transport/flow_control.c \ src/core/ext/transport/chttp2/transport/frame_data.c \ src/core/ext/transport/chttp2/transport/frame_goaway.c \ src/core/ext/transport/chttp2/transport/frame_ping.c \ diff --git a/binding.gyp b/binding.gyp index d11a60a68a..fae887d9f7 100644 --- a/binding.gyp +++ b/binding.gyp @@ -767,6 +767,7 @@ 'src/core/ext/transport/chttp2/transport/bin_encoder.c', 'src/core/ext/transport/chttp2/transport/chttp2_plugin.c', 'src/core/ext/transport/chttp2/transport/chttp2_transport.c', + 'src/core/ext/transport/chttp2/transport/flow_control.c', 'src/core/ext/transport/chttp2/transport/frame_data.c', 'src/core/ext/transport/chttp2/transport/frame_goaway.c', 'src/core/ext/transport/chttp2/transport/frame_ping.c', diff --git a/build.yaml b/build.yaml index 5dac770614..4a40ee6173 100644 --- a/build.yaml +++ b/build.yaml @@ -749,6 +749,7 @@ filegroups: - src/core/ext/transport/chttp2/transport/bin_encoder.c - src/core/ext/transport/chttp2/transport/chttp2_plugin.c - src/core/ext/transport/chttp2/transport/chttp2_transport.c + - src/core/ext/transport/chttp2/transport/flow_control.c - src/core/ext/transport/chttp2/transport/frame_data.c - src/core/ext/transport/chttp2/transport/frame_goaway.c - src/core/ext/transport/chttp2/transport/frame_ping.c @@ -217,6 +217,7 @@ if test "$PHP_GRPC" != "no"; then src/core/ext/transport/chttp2/transport/bin_encoder.c \ src/core/ext/transport/chttp2/transport/chttp2_plugin.c \ src/core/ext/transport/chttp2/transport/chttp2_transport.c \ + src/core/ext/transport/chttp2/transport/flow_control.c \ src/core/ext/transport/chttp2/transport/frame_data.c \ src/core/ext/transport/chttp2/transport/frame_goaway.c \ src/core/ext/transport/chttp2/transport/frame_ping.c \ diff --git a/config.w32 b/config.w32 index c00f3f953d..8529bd549f 100644 --- a/config.w32 +++ b/config.w32 @@ -194,6 +194,7 @@ if (PHP_GRPC != "no") { "src\\core\\ext\\transport\\chttp2\\transport\\bin_encoder.c " + "src\\core\\ext\\transport\\chttp2\\transport\\chttp2_plugin.c " + "src\\core\\ext\\transport\\chttp2\\transport\\chttp2_transport.c " + + "src\\core\\ext\\transport\\chttp2\\transport\\flow_control.c " + "src\\core\\ext\\transport\\chttp2\\transport\\frame_data.c " + "src\\core\\ext\\transport\\chttp2\\transport\\frame_goaway.c " + "src\\core\\ext\\transport\\chttp2\\transport\\frame_ping.c " + diff --git a/gRPC-Core.podspec b/gRPC-Core.podspec index 57b6f92d1e..541e6f304d 100644 --- a/gRPC-Core.podspec +++ b/gRPC-Core.podspec @@ -591,6 +591,7 @@ Pod::Spec.new do |s| 'src/core/ext/transport/chttp2/transport/bin_encoder.c', 'src/core/ext/transport/chttp2/transport/chttp2_plugin.c', 'src/core/ext/transport/chttp2/transport/chttp2_transport.c', + 'src/core/ext/transport/chttp2/transport/flow_control.c', 'src/core/ext/transport/chttp2/transport/frame_data.c', 'src/core/ext/transport/chttp2/transport/frame_goaway.c', 'src/core/ext/transport/chttp2/transport/frame_ping.c', diff --git a/grpc.gemspec b/grpc.gemspec index dedee020b7..04e3b5e32b 100755 --- a/grpc.gemspec +++ b/grpc.gemspec @@ -527,6 +527,7 @@ Gem::Specification.new do |s| s.files += %w( src/core/ext/transport/chttp2/transport/bin_encoder.c ) s.files += %w( src/core/ext/transport/chttp2/transport/chttp2_plugin.c ) s.files += %w( src/core/ext/transport/chttp2/transport/chttp2_transport.c ) + s.files += %w( src/core/ext/transport/chttp2/transport/flow_control.c ) s.files += %w( src/core/ext/transport/chttp2/transport/frame_data.c ) s.files += %w( src/core/ext/transport/chttp2/transport/frame_goaway.c ) s.files += %w( src/core/ext/transport/chttp2/transport/frame_ping.c ) diff --git a/package.xml b/package.xml index 38f51424e6..25ea40a72a 100644 --- a/package.xml +++ b/package.xml @@ -541,6 +541,7 @@ <file baseinstalldir="/" name="src/core/ext/transport/chttp2/transport/bin_encoder.c" role="src" /> <file baseinstalldir="/" name="src/core/ext/transport/chttp2/transport/chttp2_plugin.c" role="src" /> <file baseinstalldir="/" name="src/core/ext/transport/chttp2/transport/chttp2_transport.c" role="src" /> + <file baseinstalldir="/" name="src/core/ext/transport/chttp2/transport/flow_control.c" role="src" /> <file baseinstalldir="/" name="src/core/ext/transport/chttp2/transport/frame_data.c" role="src" /> <file baseinstalldir="/" name="src/core/ext/transport/chttp2/transport/frame_goaway.c" role="src" /> <file baseinstalldir="/" name="src/core/ext/transport/chttp2/transport/frame_ping.c" role="src" /> diff --git a/src/core/ext/transport/chttp2/transport/chttp2_transport.c b/src/core/ext/transport/chttp2/transport/chttp2_transport.c index 731ebf400f..437162bdbd 100644 --- a/src/core/ext/transport/chttp2/transport/chttp2_transport.c +++ b/src/core/ext/transport/chttp2/transport/chttp2_transport.c @@ -114,11 +114,6 @@ static void connectivity_state_set(grpc_exec_ctx *exec_ctx, grpc_connectivity_state state, grpc_error *error, const char *reason); -static void incoming_byte_stream_update_flow_control(grpc_exec_ctx *exec_ctx, - grpc_chttp2_transport *t, - grpc_chttp2_stream *s, - size_t max_size_hint, - size_t have_already); static void incoming_byte_stream_destroy_locked(grpc_exec_ctx *exec_ctx, void *byte_stream, grpc_error *error_ignored); @@ -270,8 +265,9 @@ 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->outgoing_window = DEFAULT_WINDOW; - t->incoming_window = DEFAULT_WINDOW; + t->local_window = DEFAULT_WINDOW; + t->remote_window = DEFAULT_WINDOW; + t->announced_window = DEFAULT_WINDOW; t->deframe_state = is_client ? GRPC_DTS_FH_0 : GRPC_DTS_CLIENT_PREFIX_0; t->is_first_frame = true; grpc_connectivity_state_init( @@ -758,13 +754,7 @@ static void destroy_stream_locked(grpc_exec_ctx *exec_ctx, void *sp, GRPC_ERROR_UNREF(s->write_closed_error); GRPC_ERROR_UNREF(s->byte_stream_error); - if (s->incoming_window_delta > 0) { - GRPC_CHTTP2_FLOW_DEBIT_STREAM_INCOMING_WINDOW_DELTA( - "destroy", t, s, s->incoming_window_delta); - } else if (s->incoming_window_delta < 0) { - GRPC_CHTTP2_FLOW_CREDIT_STREAM_INCOMING_WINDOW_DELTA( - "destroy", t, s, -s->incoming_window_delta); - } + grpc_chttp2_flowctl_destroy_stream(s); GRPC_CHTTP2_UNREF_TRANSPORT(exec_ctx, t, "stream"); @@ -1460,7 +1450,7 @@ static void perform_stream_op_locked(grpc_exec_ctx *exec_ctx, void *stream_op, already_received = s->frame_storage.length + s->unprocessed_incoming_frames_buffer.length; } - incoming_byte_stream_update_flow_control(exec_ctx, t, s, 5, + grpc_chttp2_flowctl_incoming_bs_update(exec_ctx, t, s, 5, already_received); } grpc_chttp2_maybe_complete_recv_message(exec_ctx, t, s); @@ -2159,9 +2149,10 @@ static void update_bdp(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t, if (delta == 0 || (delta > -bdp / 10 && delta < bdp / 10)) { return; } - if (GRPC_TRACER_ON(grpc_bdp_estimator_trace)) { - gpr_log(GPR_DEBUG, "%s: update initial window size to %d", t->peer_string, - (int)bdp); + if (GRPC_TRACER_ON(grpc_bdp_estimator_trace) || + GRPC_TRACER_ON(grpc_flowctl_trace)) { + gpr_log(GPR_DEBUG, "%s | %p[%s] | update initial window size to %d", + t->peer_string, t, t->is_client ? "cli" : "svr", (int)bdp); } queue_setting_update(exec_ctx, t, GRPC_CHTTP2_SETTINGS_INITIAL_WINDOW_SIZE, (uint32_t)bdp); @@ -2544,54 +2535,6 @@ static void incoming_byte_stream_unref(grpc_exec_ctx *exec_ctx, } } -static void incoming_byte_stream_update_flow_control(grpc_exec_ctx *exec_ctx, - grpc_chttp2_transport *t, - grpc_chttp2_stream *s, - size_t max_size_hint, - size_t have_already) { - uint32_t max_recv_bytes; - uint32_t initial_window_size = - t->settings[GRPC_SENT_SETTINGS][GRPC_CHTTP2_SETTINGS_INITIAL_WINDOW_SIZE]; - - /* clamp max recv hint to an allowable size */ - if (max_size_hint >= UINT32_MAX - initial_window_size) { - max_recv_bytes = UINT32_MAX - initial_window_size; - } else { - max_recv_bytes = (uint32_t)max_size_hint; - } - - /* account for bytes already received but unknown to higher layers */ - if (max_recv_bytes >= have_already) { - max_recv_bytes -= (uint32_t)have_already; - } else { - max_recv_bytes = 0; - } - - /* add some small lookahead to keep pipelines flowing */ - GPR_ASSERT(max_recv_bytes <= UINT32_MAX - initial_window_size); - if (s->incoming_window_delta < max_recv_bytes && !s->read_closed) { - uint32_t add_max_recv_bytes = - (uint32_t)(max_recv_bytes - s->incoming_window_delta); - grpc_chttp2_stream_write_type write_type = - GRPC_CHTTP2_STREAM_WRITE_INITIATE_UNCOVERED; - if (s->incoming_window_delta + initial_window_size < - (int64_t)have_already) { - write_type = GRPC_CHTTP2_STREAM_WRITE_INITIATE_COVERED; - } - GRPC_CHTTP2_FLOW_CREDIT_STREAM_INCOMING_WINDOW_DELTA("op", t, s, - add_max_recv_bytes); - GRPC_CHTTP2_FLOW_CREDIT_STREAM("op", t, s, announce_window, - add_max_recv_bytes); - if ((int64_t)s->incoming_window_delta + (int64_t)initial_window_size - - (int64_t)s->announce_window > - (int64_t)initial_window_size / 2) { - write_type = GRPC_CHTTP2_STREAM_WRITE_PIGGYBACK; - } - grpc_chttp2_become_writable(exec_ctx, t, s, write_type, - "read_incoming_stream"); - } -} - static void incoming_byte_stream_next_locked(grpc_exec_ctx *exec_ctx, void *argp, grpc_error *error_ignored) { @@ -2600,7 +2543,7 @@ static void incoming_byte_stream_next_locked(grpc_exec_ctx *exec_ctx, grpc_chttp2_stream *s = bs->stream; size_t cur_length = s->frame_storage.length; - incoming_byte_stream_update_flow_control( + grpc_chttp2_flowctl_incoming_bs_update( exec_ctx, t, s, bs->next_action.max_size_hint, cur_length); GPR_ASSERT(s->unprocessed_incoming_frames_buffer.length == 0); @@ -2875,83 +2818,6 @@ static void destructive_reclaimer_locked(grpc_exec_ctx *exec_ctx, void *arg, } /******************************************************************************* - * TRACING - */ - -static char *format_flowctl_context_var(const char *context, const char *var, - int64_t val, uint32_t id) { - char *name; - if (context == NULL) { - name = gpr_strdup(var); - } else if (0 == strcmp(context, "t")) { - GPR_ASSERT(id == 0); - gpr_asprintf(&name, "TRANSPORT:%s", var); - } else if (0 == strcmp(context, "s")) { - GPR_ASSERT(id != 0); - gpr_asprintf(&name, "STREAM[%d]:%s", id, var); - } else { - gpr_asprintf(&name, "BAD_CONTEXT[%s][%d]:%s", context, id, var); - } - char *name_fld = gpr_leftpad(name, ' ', 64); - char *value; - gpr_asprintf(&value, "%" PRId64, val); - char *value_fld = gpr_leftpad(value, ' ', 8); - char *result; - gpr_asprintf(&result, "%s %s", name_fld, value_fld); - gpr_free(name); - gpr_free(name_fld); - gpr_free(value); - gpr_free(value_fld); - return result; -} - -void grpc_chttp2_flowctl_trace(const char *file, int line, const char *phase, - grpc_chttp2_flowctl_op op, const char *context1, - const char *var1, const char *context2, - const char *var2, int is_client, - uint32_t stream_id, int64_t val1, int64_t val2) { - char *tmp_phase; - char *label1 = format_flowctl_context_var(context1, var1, val1, stream_id); - char *label2 = format_flowctl_context_var(context2, var2, val2, stream_id); - char *clisvr = is_client ? "client" : "server"; - char *prefix; - - tmp_phase = gpr_leftpad(phase, ' ', 8); - gpr_asprintf(&prefix, "FLOW %s: %s ", tmp_phase, clisvr); - gpr_free(tmp_phase); - - switch (op) { - case GRPC_CHTTP2_FLOWCTL_MOVE: - if (val2 != 0) { - gpr_log(file, line, GPR_LOG_SEVERITY_DEBUG, - "%sMOVE %s <- %s giving %" PRId64, prefix, label1, label2, - val1 + val2); - } - break; - case GRPC_CHTTP2_FLOWCTL_CREDIT: - GPR_ASSERT(val2 >= 0); - if (val2 != 0) { - gpr_log(file, line, GPR_LOG_SEVERITY_DEBUG, - "%sCREDIT %s by %s giving %" PRId64, prefix, label1, label2, - val1 + val2); - } - break; - case GRPC_CHTTP2_FLOWCTL_DEBIT: - GPR_ASSERT(val2 >= 0); - if (val2 != 0) { - gpr_log(file, line, GPR_LOG_SEVERITY_DEBUG, - "%sDEBIT %s by %s giving %" PRId64, prefix, label1, label2, - val1 - val2); - } - break; - } - - gpr_free(label1); - gpr_free(label2); - gpr_free(prefix); -} - -/******************************************************************************* * INTEGRATION GLUE */ diff --git a/src/core/ext/transport/chttp2/transport/flow_control.c b/src/core/ext/transport/chttp2/transport/flow_control.c new file mode 100644 index 0000000000..3baa6b8a14 --- /dev/null +++ b/src/core/ext/transport/chttp2/transport/flow_control.c @@ -0,0 +1,364 @@ +/* + * + * 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. + * + */ + +#include "src/core/ext/transport/chttp2/transport/internal.h" + +#include <string.h> + +#include <grpc/support/alloc.h> +#include <grpc/support/log.h> +#include <grpc/support/string_util.h> +#include <grpc/support/useful.h> + +#include "src/core/lib/support/string.h" + +#ifndef NDEBUG + +typedef struct { + int64_t remote_window; + int64_t local_window; + int64_t announced_window; + int64_t remote_window_delta; + int64_t local_window_delta; + int64_t announced_window_delta; +} shadow_flow_control; + +static void pretrace(shadow_flow_control* sfc, grpc_chttp2_transport* t, + grpc_chttp2_stream* s) { + sfc->remote_window = t->remote_window; + sfc->local_window = t->local_window; + sfc->announced_window = t->announced_window; + if (s != NULL) { + sfc->remote_window_delta = s->remote_window_delta; + sfc->local_window_delta = s->local_window_delta; + sfc->announced_window_delta = s->announced_window_delta; + } +} + +static char* fmt_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); + gpr_free(str); + return str_lp; +} + +static void posttrace(shadow_flow_control* sfc, grpc_chttp2_transport* t, + grpc_chttp2_stream* s, char* reason) { + uint32_t acked_local_window = + t->settings[GRPC_SENT_SETTINGS][GRPC_CHTTP2_SETTINGS_INITIAL_WINDOW_SIZE]; + uint32_t remote_window = + t->settings[GRPC_PEER_SETTINGS][GRPC_CHTTP2_SETTINGS_INITIAL_WINDOW_SIZE]; + char* trw_str = fmt_str(sfc->remote_window, t->remote_window); + char* tlw_str = fmt_str(sfc->local_window, t->local_window); + char* taw_str = fmt_str(sfc->announced_window, t->announced_window); + char* srw_str; + char* slw_str; + char* saw_str; + if (s != NULL) { + srw_str = fmt_str(sfc->remote_window_delta + remote_window, + s->remote_window_delta + remote_window); + slw_str = fmt_str(sfc->local_window_delta + acked_local_window, + s->local_window_delta + acked_local_window); + saw_str = fmt_str(sfc->announced_window_delta + acked_local_window, + s->announced_window_delta + acked_local_window); + } else { + srw_str = gpr_leftpad("", ' ', 30); + slw_str = gpr_leftpad("", ' ', 30); + saw_str = gpr_leftpad("", ' ', 30); + } + gpr_log(GPR_DEBUG, + "%p[%u][%s] | %s | trw:%s, tlw:%s, taw:%s, srw:%s, slw:%s, saw:%s", t, + s != NULL ? s->id : 0, t->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); + gpr_free(srw_str); + gpr_free(slw_str); + gpr_free(saw_str); +} + +#define PRETRACE(t, s) \ + shadow_flow_control sfc; \ + GRPC_FLOW_CONTROL_IF_TRACING(pretrace(&sfc, t, s)) +#define POSTTRACE(t, s, reason) \ + GRPC_FLOW_CONTROL_IF_TRACING(posttrace(&sfc, t, s, reason)) +#else +#define PRETRACE(t, s) +#define POSTTRACE(t, s, reason) ; +#endif + +/* How many bytes of incoming flow control would we like to advertise */ +static uint32_t grpc_chttp2_target_announced_window(grpc_chttp2_transport* t) { + return (uint32_t)GPR_MIN( + (int64_t)((1u << 31) - 1), + t->announced_stream_total_over_incoming_window + + t->settings[GRPC_ACKED_SETTINGS] + [GRPC_CHTTP2_SETTINGS_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* t, + grpc_chttp2_stream* s, int64_t size) { + PRETRACE(t, s); + t->remote_window -= size; + s->remote_window_delta -= size; + POSTTRACE(t, s, " data sent"); +} + +static void announced_window_delta_preupdate(grpc_chttp2_transport* t, + grpc_chttp2_stream* s) { + if (s->announced_window_delta > 0) { + t->announced_stream_total_over_incoming_window -= s->announced_window_delta; + } else { + t->announced_stream_total_under_incoming_window += + -s->announced_window_delta; + } +} + +static void announced_window_delta_postupdate(grpc_chttp2_transport* t, + grpc_chttp2_stream* s) { + if (s->announced_window_delta > 0) { + t->announced_stream_total_over_incoming_window += s->announced_window_delta; + } else { + t->announced_stream_total_under_incoming_window -= + -s->announced_window_delta; + } +} + +// 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_exec_ctx* exec_ctx, + grpc_chttp2_transport* t, + grpc_chttp2_stream* s, + int64_t incoming_frame_size) { + PRETRACE(t, s); + if (incoming_frame_size > t->local_window) { + char* msg; + gpr_asprintf(&msg, "frame of size %d overflows local window of %" PRId64, + t->incoming_frame_size, t->local_window); + grpc_error* err = GRPC_ERROR_CREATE_FROM_COPIED_STRING(msg); + gpr_free(msg); + return err; + } + + // TODO(ncteisen): can this ever be null? ANSWER: only when incoming frame + // size is zero? + if (s != NULL) { + int64_t acked_stream_window = + s->announced_window_delta + + t->settings[GRPC_ACKED_SETTINGS] + [GRPC_CHTTP2_SETTINGS_INITIAL_WINDOW_SIZE]; + int64_t sent_stream_window = + s->announced_window_delta + + t->settings[GRPC_SENT_SETTINGS] + [GRPC_CHTTP2_SETTINGS_INITIAL_WINDOW_SIZE]; + if (incoming_frame_size > acked_stream_window) { + if (incoming_frame_size <= sent_stream_window) { + gpr_log( + GPR_ERROR, + "Incoming frame of size %d 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.", + t->incoming_frame_size, acked_stream_window, sent_stream_window); + } else { + char* msg; + gpr_asprintf(&msg, + "frame of size %d overflows local window of %" PRId64, + t->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(t, s); + s->announced_window_delta -= incoming_frame_size; + announced_window_delta_postupdate(t, s); + s->local_window_delta -= incoming_frame_size; + s->received_bytes += incoming_frame_size; + + if (s->announced_window_delta > 0) { + t->announced_stream_total_over_incoming_window += + s->announced_window_delta; + } else { + t->announced_stream_total_under_incoming_window -= + -s->announced_window_delta; + } + + // TODO(control bit) + if ((int64_t)s->local_window_delta > (int64_t)s->announced_window_delta && (int64_t)s->announced_window_delta <= + (int64_t)t->settings[GRPC_SENT_SETTINGS] + [GRPC_CHTTP2_SETTINGS_INITIAL_WINDOW_SIZE] / + 2) { + grpc_chttp2_become_writable(exec_ctx, t, s, + GRPC_CHTTP2_STREAM_WRITE_INITIATE_UNCOVERED, + "window-update-required"); + } + } + + t->announced_window -= incoming_frame_size; + t->local_window -= incoming_frame_size; + + // TODO(control bit) + uint32_t target_announced_window = grpc_chttp2_target_announced_window(t); + if (t->announced_window <= target_announced_window / 2) { + grpc_chttp2_initiate_write(exec_ctx, t, "flow_control"); + } + + POSTTRACE(t, s, " 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* t) { + PRETRACE(t, NULL); + uint32_t target_announced_window = grpc_chttp2_target_announced_window(t); + uint32_t threshold_to_send_transport_window_update = + t->outbuf.count > 0 ? 3 * target_announced_window / 4 + : target_announced_window / 2; + if (t->announced_window < t->local_window && + t->announced_window <= threshold_to_send_transport_window_update && + t->announced_window != target_announced_window) { + uint32_t announce = (uint32_t)GPR_CLAMP( + target_announced_window - t->announced_window, 0, UINT32_MAX); + t->announced_window += announce; + t->local_window = + t->announced_window; // announced should never be higher than local. + POSTTRACE(t, NULL, "t updt sent"); + return announce; + } + + // uint32_t announce = 0; + // if (t->local_window > t->announced_window) { + // announce = (uint32_t)GPR_CLAMP( + // t->local_window - t->announced_window, 0, UINT32_MAX); + // t->announced_window += announce; + // POSTTRACE(t, NULL, "t updt sent"); + // } + GRPC_FLOW_CONTROL_IF_TRACING( + gpr_log(GPR_DEBUG, "%p[0][%s] will not to send transport update", t, + t->is_client ? "cli" : "svr")); + return 0; +} + +// 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_stream* s) { + PRETRACE(s->t, s); + if (s->local_window_delta > s->announced_window_delta) { + uint32_t announce = (uint32_t)GPR_CLAMP( + s->local_window_delta - s->announced_window_delta, 0, UINT32_MAX); + announced_window_delta_preupdate(s->t, s); + s->announced_window_delta += announce; + announced_window_delta_postupdate(s->t, s); + POSTTRACE(s->t, s, "s updt sent"); + return announce; + } + GRPC_FLOW_CONTROL_IF_TRACING( + gpr_log(GPR_DEBUG, "%p[%u][%s] will not to send stream update", s->t, + s->id, s->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* t, + uint32_t size) { + PRETRACE(t, NULL); + t->remote_window += size; + POSTTRACE(t, NULL, "t updt recv"); +} + +// we have received a WINDOW_UPDATE frame for a stream +void grpc_chttp2_flowctl_recv_stream_update(grpc_chttp2_stream* s, + uint32_t size) { + PRETRACE(s->t, s); + s->remote_window_delta += size; + POSTTRACE(s->t, s, "s updt recv"); +} + +void grpc_chttp2_flowctl_incoming_bs_update(grpc_exec_ctx *exec_ctx, + grpc_chttp2_transport *t, + grpc_chttp2_stream *s, + size_t max_size_hint, + size_t have_already) { + PRETRACE(t, s); + uint32_t max_recv_bytes; + uint32_t initial_window_size = + t->settings[GRPC_SENT_SETTINGS][GRPC_CHTTP2_SETTINGS_INITIAL_WINDOW_SIZE]; + + /* clamp max recv hint to an allowable size */ + if (max_size_hint >= UINT32_MAX - initial_window_size) { + max_recv_bytes = UINT32_MAX - initial_window_size; + } else { + max_recv_bytes = (uint32_t)max_size_hint; + } + + /* account for bytes already received but unknown to higher layers */ + if (max_recv_bytes >= have_already) { + max_recv_bytes -= (uint32_t)have_already; + } else { + max_recv_bytes = 0; + } + + /* add some small lookahead to keep pipelines flowing */ + GPR_ASSERT(max_recv_bytes <= UINT32_MAX - initial_window_size); + if (s->local_window_delta < max_recv_bytes && !s->read_closed) { + uint32_t add_max_recv_bytes = + (uint32_t)(max_recv_bytes - s->local_window_delta); + grpc_chttp2_stream_write_type write_type = + GRPC_CHTTP2_STREAM_WRITE_INITIATE_UNCOVERED; + s->local_window_delta += add_max_recv_bytes; + s->t->local_window += add_max_recv_bytes; + // TODO(control bits) + if ((int64_t)initial_window_size + (int64_t)s->announced_window_delta > + (int64_t)initial_window_size / 2 && + t->announced_window > (int64_t)initial_window_size / 2) { + write_type = GRPC_CHTTP2_STREAM_WRITE_PIGGYBACK; // TODO(contol bits) + } + GRPC_FLOW_CONTROL_IF_TRACING(gpr_log( + GPR_DEBUG, "%p[%u][%s] becoming writable, %sinitiating read", t, s->id, + t->is_client ? "cli" : "svr", + write_type == GRPC_CHTTP2_STREAM_WRITE_PIGGYBACK ? "not " : "")); + grpc_chttp2_become_writable(exec_ctx, t, s, write_type, + "read_incoming_stream"); + } + POSTTRACE(t, s, "app st recv"); +} + +void grpc_chttp2_flowctl_destroy_stream(grpc_chttp2_stream* s) { + if (s->announced_window_delta > 0) { + s->t->announced_stream_total_over_incoming_window -= + s->announced_window_delta; + } else { + s->t->announced_stream_total_under_incoming_window += + -s->announced_window_delta; + } +} diff --git a/src/core/ext/transport/chttp2/transport/frame_settings.c b/src/core/ext/transport/chttp2/transport/frame_settings.c index e3e432a94a..032f2ac426 100644 --- a/src/core/ext/transport/chttp2/transport/frame_settings.c +++ b/src/core/ext/transport/chttp2/transport/frame_settings.c @@ -203,8 +203,10 @@ grpc_error *grpc_chttp2_settings_parser_parse(grpc_exec_ctx *exec_ctx, void *p, parser->incoming_settings[id] != parser->value) { t->initial_window_update += (int64_t)parser->value - parser->incoming_settings[id]; - if (GRPC_TRACER_ON(grpc_http_trace)) { - gpr_log(GPR_DEBUG, "adding %d for initial_window change", + if (GRPC_TRACER_ON(grpc_http_trace) || + GRPC_TRACER_ON(grpc_flowctl_trace)) { + gpr_log(GPR_DEBUG, "%p[%s] adding %d for initial_window change", + t, t->is_client ? "cli" : "svr", (int)t->initial_window_update); } } diff --git a/src/core/ext/transport/chttp2/transport/frame_window_update.c b/src/core/ext/transport/chttp2/transport/frame_window_update.c index 682be2c89b..9cbadf4cfa 100644 --- a/src/core/ext/transport/chttp2/transport/frame_window_update.c +++ b/src/core/ext/transport/chttp2/transport/frame_window_update.c @@ -95,8 +95,8 @@ grpc_error *grpc_chttp2_window_update_parser_parse( if (t->incoming_stream_id != 0) { if (s != NULL) { - GRPC_CHTTP2_FLOW_CREDIT_STREAM("parse", t, s, outgoing_window_delta, - received_update); + grpc_chttp2_flowctl_recv_stream_update(s, received_update); + // TODO(control bits) if (grpc_chttp2_list_remove_stalled_by_stream(t, s)) { grpc_chttp2_become_writable( exec_ctx, t, s, GRPC_CHTTP2_STREAM_WRITE_INITIATE_UNCOVERED, @@ -104,10 +104,10 @@ grpc_error *grpc_chttp2_window_update_parser_parse( } } } else { - bool was_zero = t->outgoing_window <= 0; - GRPC_CHTTP2_FLOW_CREDIT_TRANSPORT("parse", t, outgoing_window, - received_update); - bool is_zero = t->outgoing_window <= 0; + bool was_zero = t->remote_window <= 0; + grpc_chttp2_flowctl_recv_transport_update(t, received_update); + // TODO(control bits) + bool is_zero = t->remote_window <= 0; if (was_zero && !is_zero) { grpc_chttp2_initiate_write(exec_ctx, t, "new_global_flow_control"); } diff --git a/src/core/ext/transport/chttp2/transport/internal.h b/src/core/ext/transport/chttp2/transport/internal.h index 4563b78e75..ee19090270 100644 --- a/src/core/ext/transport/chttp2/transport/internal.h +++ b/src/core/ext/transport/chttp2/transport/internal.h @@ -271,7 +271,6 @@ struct grpc_chttp2_transport { grpc_slice_buffer outbuf; /** hpack encoding */ grpc_chttp2_hpack_compressor hpack_compressor; - int64_t outgoing_window; /** is this a client? */ uint8_t is_client; @@ -328,11 +327,26 @@ struct grpc_chttp2_transport { /** parser for goaway frames */ grpc_chttp2_goaway_parser goaway_parser; - /** initial window change */ + /*********** Flow Control **************/ + + /** 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; - /** window available for peer to send to us */ - int64_t incoming_window; + /** Our bookkeeping for the remote peer's available window */ + int64_t remote_window; + + /** Our bookkeeping for our window. Essentially this tracks available buffer + * space to hold data that peer sends to us. This is our local view of the + * window. It does not reflect how the remote peer sees it. */ + int64_t local_window; + + /** This is out window according to what we have sent to our remote peer. The + * difference between this and local_window is what we use to decide when + * to send WINDOW_UPDATE frames. */ + int64_t announced_window; + /** calculating what we should give for incoming 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 @@ -341,8 +355,17 @@ struct grpc_chttp2_transport { 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 stream_total_over_incoming_window; - int64_t stream_total_under_incoming_window; + int64_t announced_stream_total_over_incoming_window; + int64_t announced_stream_total_under_incoming_window; + + /* bdp estimation */ + grpc_bdp_estimator bdp_estimator; + + /* pid controller */ + grpc_pid_controller pid_controller; + gpr_timespec last_pid_update; + + /*********** End of Flow Control **************/ /* deframing */ grpc_chttp2_deframe_transport_state deframe_state; @@ -369,11 +392,8 @@ struct grpc_chttp2_transport { grpc_chttp2_write_cb *write_cb_pool; /* bdp estimator */ - grpc_bdp_estimator bdp_estimator; - grpc_pid_controller pid_controller; grpc_closure start_bdp_ping_locked; grpc_closure finish_bdp_ping_locked; - gpr_timespec last_pid_update; /* if non-NULL, close the transport with this error when writes are finished */ @@ -435,10 +455,6 @@ struct grpc_chttp2_stream { /** HTTP2 stream id for this stream, or zero if one has not been assigned */ uint32_t id; - /** window available for us to send to peer, over or under the initial window - * size of the transport... ie: - * outgoing_window = outgoing_window_delta + transport.initial_window_size */ - int64_t outgoing_window_delta; /** things the upper layers would like to send */ grpc_metadata_batch *send_initial_metadata; grpc_closure *send_initial_metadata_finished; @@ -505,10 +521,6 @@ struct grpc_chttp2_stream { grpc_error *forced_close_error; /** how many header frames have we received? */ uint8_t header_frames_received; - /** window available for peer to send to us (as a delta on - * transport.initial_window_size) - * incoming_window = incoming_window_delta + transport.initial_window_size */ - int64_t incoming_window_delta; /** parsing state for data frames */ /* Accessed only by transport thread when stream->pending_byte_stream == false * Accessed only by application thread when stream->pending_byte_stream == @@ -519,8 +531,25 @@ struct grpc_chttp2_stream { bool sent_initial_metadata; bool sent_trailing_metadata; - /** how much window should we announce? */ - uint32_t announce_window; + + /*********** Flow Control ***********/ + + /** 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; + + /*********** End of Flow Control ***********/ + grpc_slice_buffer flow_controlled_buffer; grpc_chttp2_write_cb *on_write_finished_cbs; @@ -601,6 +630,42 @@ bool grpc_chttp2_list_pop_stalled_by_stream(grpc_chttp2_transport *t, bool grpc_chttp2_list_remove_stalled_by_stream(grpc_chttp2_transport *t, grpc_chttp2_stream *s); +/********* Flow Control ***************/ + +// we have sent data on the wire +void grpc_chttp2_flowctl_sent_data(grpc_chttp2_transport *t, + grpc_chttp2_stream *s, int64_t size); + +// we have received data from the wire +grpc_error *grpc_chttp2_flowctl_recv_data(grpc_exec_ctx *exec_ctx, + grpc_chttp2_transport *t, + grpc_chttp2_stream *s, + int64_t incoming_frame_size); + +uint32_t grpc_chttp2_flowctl_maybe_send_transport_update( + grpc_chttp2_transport *t); + +uint32_t grpc_chttp2_flowctl_maybe_send_stream_update(grpc_chttp2_stream *s); + +// we have received a WINDOW_UPDATE frame for a transport +void grpc_chttp2_flowctl_recv_transport_update(grpc_chttp2_transport *t, + uint32_t size); + +// we have received a WINDOW_UPDATE frame for a stream +void grpc_chttp2_flowctl_recv_stream_update(grpc_chttp2_stream *s, + uint32_t size); + +// the application is asking for a certain amount of bytes +void grpc_chttp2_flowctl_incoming_bs_update(grpc_exec_ctx *exec_ctx, + grpc_chttp2_transport *t, + grpc_chttp2_stream *s, + size_t max_size_hint, + size_t have_already); + +void grpc_chttp2_flowctl_destroy_stream(grpc_chttp2_stream *s); + +/********* End of Flow Control ***************/ + grpc_chttp2_stream *grpc_chttp2_parsing_lookup_stream(grpc_chttp2_transport *t, uint32_t id); grpc_chttp2_stream *grpc_chttp2_parsing_accept_stream(grpc_exec_ctx *exec_ctx, @@ -628,126 +693,22 @@ void grpc_chttp2_complete_closure_step(grpc_exec_ctx *exec_ctx, extern grpc_tracer_flag grpc_http_trace; extern grpc_tracer_flag grpc_flowctl_trace; +#ifndef NDEBUG +#define GRPC_FLOW_CONTROL_IF_TRACING(stmt) \ + if (!(GRPC_TRACER_ON(grpc_flowctl_trace))) \ + ; \ + else \ + stmt +#else +#define GRPC_FLOW_CONTROL_IF_TRACING(stmt) +#endif + #define GRPC_CHTTP2_IF_TRACING(stmt) \ if (!(GRPC_TRACER_ON(grpc_http_trace))) \ ; \ else \ stmt -typedef enum { - GRPC_CHTTP2_FLOWCTL_MOVE, - GRPC_CHTTP2_FLOWCTL_CREDIT, - GRPC_CHTTP2_FLOWCTL_DEBIT -} grpc_chttp2_flowctl_op; - -#define GRPC_CHTTP2_FLOW_MOVE_COMMON(phase, transport, id1, id2, dst_context, \ - dst_var, src_context, src_var) \ - do { \ - assert(id1 == id2); \ - if (GRPC_TRACER_ON(grpc_flowctl_trace)) { \ - grpc_chttp2_flowctl_trace( \ - __FILE__, __LINE__, phase, GRPC_CHTTP2_FLOWCTL_MOVE, #dst_context, \ - #dst_var, #src_context, #src_var, transport->is_client, id1, \ - dst_context->dst_var, src_context->src_var); \ - } \ - dst_context->dst_var += src_context->src_var; \ - src_context->src_var = 0; \ - } while (0) - -#define GRPC_CHTTP2_FLOW_MOVE_STREAM(phase, transport, dst_context, dst_var, \ - src_context, src_var) \ - GRPC_CHTTP2_FLOW_MOVE_COMMON(phase, transport, dst_context->id, \ - src_context->id, dst_context, dst_var, \ - src_context, src_var) -#define GRPC_CHTTP2_FLOW_MOVE_TRANSPORT(phase, dst_context, dst_var, \ - src_context, src_var) \ - GRPC_CHTTP2_FLOW_MOVE_COMMON(phase, dst_context, 0, 0, dst_context, dst_var, \ - src_context, src_var) - -#define GRPC_CHTTP2_FLOW_CREDIT_COMMON(phase, transport, id, dst_context, \ - dst_var, amount) \ - do { \ - if (GRPC_TRACER_ON(grpc_flowctl_trace)) { \ - grpc_chttp2_flowctl_trace(__FILE__, __LINE__, phase, \ - GRPC_CHTTP2_FLOWCTL_CREDIT, #dst_context, \ - #dst_var, NULL, #amount, transport->is_client, \ - id, dst_context->dst_var, amount); \ - } \ - dst_context->dst_var += amount; \ - } while (0) - -#define GRPC_CHTTP2_FLOW_CREDIT_STREAM(phase, transport, dst_context, dst_var, \ - amount) \ - GRPC_CHTTP2_FLOW_CREDIT_COMMON(phase, transport, dst_context->id, \ - dst_context, dst_var, amount) -#define GRPC_CHTTP2_FLOW_CREDIT_TRANSPORT(phase, dst_context, dst_var, amount) \ - GRPC_CHTTP2_FLOW_CREDIT_COMMON(phase, dst_context, 0, dst_context, dst_var, \ - amount) - -#define GRPC_CHTTP2_FLOW_STREAM_INCOMING_WINDOW_DELTA_PREUPDATE( \ - phase, transport, dst_context) \ - if (dst_context->incoming_window_delta < 0) { \ - transport->stream_total_under_incoming_window += \ - dst_context->incoming_window_delta; \ - } else if (dst_context->incoming_window_delta > 0) { \ - transport->stream_total_over_incoming_window -= \ - dst_context->incoming_window_delta; \ - } - -#define GRPC_CHTTP2_FLOW_STREAM_INCOMING_WINDOW_DELTA_POSTUPDATE( \ - phase, transport, dst_context) \ - if (dst_context->incoming_window_delta < 0) { \ - transport->stream_total_under_incoming_window -= \ - dst_context->incoming_window_delta; \ - } else if (dst_context->incoming_window_delta > 0) { \ - transport->stream_total_over_incoming_window += \ - dst_context->incoming_window_delta; \ - } - -#define GRPC_CHTTP2_FLOW_DEBIT_STREAM_INCOMING_WINDOW_DELTA( \ - phase, transport, dst_context, amount) \ - GRPC_CHTTP2_FLOW_STREAM_INCOMING_WINDOW_DELTA_PREUPDATE(phase, transport, \ - dst_context); \ - GRPC_CHTTP2_FLOW_DEBIT_STREAM(phase, transport, dst_context, \ - incoming_window_delta, amount); \ - GRPC_CHTTP2_FLOW_STREAM_INCOMING_WINDOW_DELTA_POSTUPDATE(phase, transport, \ - dst_context); - -#define GRPC_CHTTP2_FLOW_CREDIT_STREAM_INCOMING_WINDOW_DELTA( \ - phase, transport, dst_context, amount) \ - GRPC_CHTTP2_FLOW_STREAM_INCOMING_WINDOW_DELTA_PREUPDATE(phase, transport, \ - dst_context); \ - GRPC_CHTTP2_FLOW_CREDIT_STREAM(phase, transport, dst_context, \ - incoming_window_delta, amount); \ - GRPC_CHTTP2_FLOW_STREAM_INCOMING_WINDOW_DELTA_POSTUPDATE(phase, transport, \ - dst_context); - -#define GRPC_CHTTP2_FLOW_DEBIT_COMMON(phase, transport, id, dst_context, \ - dst_var, amount) \ - do { \ - if (GRPC_TRACER_ON(grpc_flowctl_trace)) { \ - grpc_chttp2_flowctl_trace(__FILE__, __LINE__, phase, \ - GRPC_CHTTP2_FLOWCTL_DEBIT, #dst_context, \ - #dst_var, NULL, #amount, transport->is_client, \ - id, dst_context->dst_var, amount); \ - } \ - dst_context->dst_var -= amount; \ - } while (0) - -#define GRPC_CHTTP2_FLOW_DEBIT_STREAM(phase, transport, dst_context, dst_var, \ - amount) \ - GRPC_CHTTP2_FLOW_DEBIT_COMMON(phase, transport, dst_context->id, \ - dst_context, dst_var, amount) -#define GRPC_CHTTP2_FLOW_DEBIT_TRANSPORT(phase, dst_context, dst_var, amount) \ - GRPC_CHTTP2_FLOW_DEBIT_COMMON(phase, dst_context, 0, dst_context, dst_var, \ - amount) - -void grpc_chttp2_flowctl_trace(const char *file, int line, const char *phase, - grpc_chttp2_flowctl_op op, const char *context1, - const char *var1, const char *context2, - const char *var2, int is_client, - uint32_t stream_id, int64_t val1, int64_t val2); - void grpc_chttp2_fake_status(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t, grpc_chttp2_stream *stream, grpc_error *error); void grpc_chttp2_mark_stream_closed(grpc_exec_ctx *exec_ctx, @@ -849,8 +810,6 @@ void grpc_chttp2_fail_pending_writes(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t, grpc_chttp2_stream *s, grpc_error *error); -uint32_t grpc_chttp2_target_incoming_window(grpc_chttp2_transport *t); - /** Set the default keepalive configurations, must only be called at initialization */ void grpc_chttp2_config_default_keepalive_args(grpc_channel_args *args, diff --git a/src/core/ext/transport/chttp2/transport/parsing.c b/src/core/ext/transport/chttp2/transport/parsing.c index 3c8b470b4f..2e5d93da3f 100644 --- a/src/core/ext/transport/chttp2/transport/parsing.c +++ b/src/core/ext/transport/chttp2/transport/parsing.c @@ -349,87 +349,12 @@ void grpc_chttp2_parsing_become_skip_parser(grpc_exec_ctx *exec_ctx, t->parser == grpc_chttp2_header_parser_parse); } -static grpc_error *update_incoming_window(grpc_exec_ctx *exec_ctx, - grpc_chttp2_transport *t, - grpc_chttp2_stream *s) { - uint32_t incoming_frame_size = t->incoming_frame_size; - if (incoming_frame_size > t->incoming_window) { - char *msg; - gpr_asprintf(&msg, "frame of size %d overflows incoming window of %" PRId64, - t->incoming_frame_size, t->incoming_window); - grpc_error *err = GRPC_ERROR_CREATE_FROM_COPIED_STRING(msg); - gpr_free(msg); - return err; - } - - if (s != NULL) { - if (incoming_frame_size > - s->incoming_window_delta + - t->settings[GRPC_ACKED_SETTINGS] - [GRPC_CHTTP2_SETTINGS_INITIAL_WINDOW_SIZE]) { - if (incoming_frame_size <= - s->incoming_window_delta + - t->settings[GRPC_SENT_SETTINGS] - [GRPC_CHTTP2_SETTINGS_INITIAL_WINDOW_SIZE]) { - gpr_log( - GPR_ERROR, - "Incoming frame of size %d exceeds incoming 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.", - t->incoming_frame_size, - s->incoming_window_delta + - t->settings[GRPC_ACKED_SETTINGS] - [GRPC_CHTTP2_SETTINGS_INITIAL_WINDOW_SIZE], - s->incoming_window_delta + - t->settings[GRPC_SENT_SETTINGS] - [GRPC_CHTTP2_SETTINGS_INITIAL_WINDOW_SIZE]); - } else { - char *msg; - gpr_asprintf(&msg, - "frame of size %d overflows incoming window of %" PRId64, - t->incoming_frame_size, - s->incoming_window_delta + - t->settings[GRPC_ACKED_SETTINGS] - [GRPC_CHTTP2_SETTINGS_INITIAL_WINDOW_SIZE]); - grpc_error *err = GRPC_ERROR_CREATE_FROM_COPIED_STRING(msg); - gpr_free(msg); - return err; - } - } - - GRPC_CHTTP2_FLOW_DEBIT_STREAM_INCOMING_WINDOW_DELTA("parse", t, s, - incoming_frame_size); - if ((int64_t)s->incoming_window_delta - (int64_t)s->announce_window <= - -(int64_t)t->settings[GRPC_SENT_SETTINGS] - [GRPC_CHTTP2_SETTINGS_INITIAL_WINDOW_SIZE] / - 2) { - grpc_chttp2_become_writable(exec_ctx, t, s, - GRPC_CHTTP2_STREAM_WRITE_INITIATE_UNCOVERED, - "window-update-required"); - } - s->received_bytes += incoming_frame_size; - } - - uint32_t target_incoming_window = grpc_chttp2_target_incoming_window(t); - GRPC_CHTTP2_FLOW_DEBIT_TRANSPORT("parse", t, incoming_window, - incoming_frame_size); - if (t->incoming_window <= target_incoming_window / 2) { - grpc_chttp2_initiate_write(exec_ctx, t, "flow_control"); - } - - return GRPC_ERROR_NONE; -} - static grpc_error *init_data_frame_parser(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t) { grpc_chttp2_stream *s = grpc_chttp2_parsing_lookup_stream(t, t->incoming_stream_id); grpc_error *err = GRPC_ERROR_NONE; - err = update_incoming_window(exec_ctx, t, s); + err = grpc_chttp2_flowctl_recv_data(exec_ctx, t, s, t->incoming_frame_size); if (err != GRPC_ERROR_NONE) { goto error_handler; } diff --git a/src/core/ext/transport/chttp2/transport/stream_lists.c b/src/core/ext/transport/chttp2/transport/stream_lists.c index 1bf5b34510..7cc85dea9c 100644 --- a/src/core/ext/transport/chttp2/transport/stream_lists.c +++ b/src/core/ext/transport/chttp2/transport/stream_lists.c @@ -150,12 +150,17 @@ void grpc_chttp2_list_remove_waiting_for_concurrency(grpc_chttp2_transport *t, void grpc_chttp2_list_add_stalled_by_transport(grpc_chttp2_transport *t, grpc_chttp2_stream *s) { + GRPC_FLOW_CONTROL_IF_TRACING( + gpr_log(GPR_DEBUG, "stream %u stalled by transport", s->id)); stream_list_add(t, s, GRPC_CHTTP2_LIST_STALLED_BY_TRANSPORT); } bool grpc_chttp2_list_pop_stalled_by_transport(grpc_chttp2_transport *t, grpc_chttp2_stream **s) { - return stream_list_pop(t, s, GRPC_CHTTP2_LIST_STALLED_BY_TRANSPORT); + bool ret = stream_list_pop(t, s, GRPC_CHTTP2_LIST_STALLED_BY_TRANSPORT); + GRPC_FLOW_CONTROL_IF_TRACING(if (ret) gpr_log( + GPR_DEBUG, "stream %u un-stalled by transport", (*s)->id)); + return ret; } void grpc_chttp2_list_remove_stalled_by_transport(grpc_chttp2_transport *t, @@ -165,15 +170,23 @@ void grpc_chttp2_list_remove_stalled_by_transport(grpc_chttp2_transport *t, void grpc_chttp2_list_add_stalled_by_stream(grpc_chttp2_transport *t, grpc_chttp2_stream *s) { + GRPC_FLOW_CONTROL_IF_TRACING( + gpr_log(GPR_DEBUG, "stream %u stalled by stream", s->id)); stream_list_add(t, s, GRPC_CHTTP2_LIST_STALLED_BY_STREAM); } bool grpc_chttp2_list_pop_stalled_by_stream(grpc_chttp2_transport *t, grpc_chttp2_stream **s) { - return stream_list_pop(t, s, GRPC_CHTTP2_LIST_STALLED_BY_STREAM); + bool ret = stream_list_pop(t, s, GRPC_CHTTP2_LIST_STALLED_BY_STREAM); + GRPC_FLOW_CONTROL_IF_TRACING( + if (ret) gpr_log(GPR_DEBUG, "stream %u un-stalled by stream", (*s)->id)); + return ret; } bool grpc_chttp2_list_remove_stalled_by_stream(grpc_chttp2_transport *t, grpc_chttp2_stream *s) { - return stream_list_maybe_remove(t, s, GRPC_CHTTP2_LIST_STALLED_BY_STREAM); + bool ret = stream_list_maybe_remove(t, s, GRPC_CHTTP2_LIST_STALLED_BY_STREAM); + GRPC_FLOW_CONTROL_IF_TRACING( + if (ret) gpr_log(GPR_DEBUG, "stream %u un-stalled by stream", s->id)); + return ret; } diff --git a/src/core/ext/transport/chttp2/transport/writing.c b/src/core/ext/transport/chttp2/transport/writing.c index 315f2a67a2..9f4c286515 100644 --- a/src/core/ext/transport/chttp2/transport/writing.c +++ b/src/core/ext/transport/chttp2/transport/writing.c @@ -148,15 +148,6 @@ static bool stream_ref_if_not_destroyed(gpr_refcount *r) { return true; } -/* How many bytes of incoming flow control would we like to advertise */ -uint32_t grpc_chttp2_target_incoming_window(grpc_chttp2_transport *t) { - return (uint32_t)GPR_MIN( - (int64_t)((1u << 31) - 1), - t->stream_total_over_incoming_window + - t->settings[GRPC_SENT_SETTINGS] - [GRPC_CHTTP2_SETTINGS_INITIAL_WINDOW_SIZE]); -} - /* How many bytes would we like to put on the wire during a single syscall */ static uint32_t target_write_size(grpc_chttp2_transport *t) { return 1024 * 1024; @@ -201,7 +192,7 @@ grpc_chttp2_begin_write_result grpc_chttp2_begin_write( &t->hpack_compressor, t->settings[GRPC_PEER_SETTINGS][GRPC_CHTTP2_SETTINGS_HEADER_TABLE_SIZE]); - if (t->outgoing_window > 0) { + if (t->remote_window > 0) { while (grpc_chttp2_list_pop_stalled_by_transport(t, &s)) { if (!t->closed && grpc_chttp2_list_add_writable_stream(t, s) && stream_ref_if_not_destroyed(&s->refcount->refs)) { @@ -227,10 +218,11 @@ grpc_chttp2_begin_write_result grpc_chttp2_begin_write( bool sent_initial_metadata = s->sent_initial_metadata; bool now_writing = false; - GRPC_CHTTP2_IF_TRACING(gpr_log( - GPR_DEBUG, "W:%p %s[%d] im-(sent,send)=(%d,%d) announce=%d", t, - t->is_client ? "CLIENT" : "SERVER", s->id, sent_initial_metadata, - s->send_initial_metadata != NULL, s->announce_window)); + GRPC_CHTTP2_IF_TRACING( + gpr_log(GPR_DEBUG, "W:%p %s[%d] im-(sent,send)=(%d,%d) announce=%d", t, + t->is_client ? "CLIENT" : "SERVER", s->id, + sent_initial_metadata, s->send_initial_metadata != NULL, + (int)(s->local_window_delta - s->announced_window_delta))); grpc_mdelem *extra_headers_for_trailing_metadata[2]; size_t num_extra_headers_for_trailing_metadata = 0; @@ -287,11 +279,11 @@ grpc_chttp2_begin_write_result grpc_chttp2_begin_write( sent_initial_metadata = true; } /* send any window updates */ - if (s->announce_window > 0) { - uint32_t announce = s->announce_window; - grpc_slice_buffer_add(&t->outbuf, - grpc_chttp2_window_update_create( - s->id, s->announce_window, &s->stats.outgoing)); + uint32_t stream_announce = grpc_chttp2_flowctl_maybe_send_stream_update(s); + if (stream_announce > 0) { + grpc_slice_buffer_add( + &t->outbuf, grpc_chttp2_window_update_create(s->id, stream_announce, + &s->stats.outgoing)); t->ping_state.pings_before_data_required = t->ping_policy.max_pings_without_data; if (!t->is_client) { @@ -299,20 +291,19 @@ grpc_chttp2_begin_write_result grpc_chttp2_begin_write( gpr_inf_past(GPR_CLOCK_MONOTONIC); t->ping_recv_state.ping_strikes = 0; } - GRPC_CHTTP2_FLOW_DEBIT_STREAM("write", t, s, announce_window, announce); } if (sent_initial_metadata) { /* send any body bytes, if allowed by flow control */ if (s->flow_controlled_buffer.length > 0) { - uint32_t stream_outgoing_window = (uint32_t)GPR_MAX( + uint32_t stream_remote_window = (uint32_t)GPR_MAX( 0, - s->outgoing_window_delta + + s->remote_window_delta + (int64_t)t->settings[GRPC_PEER_SETTINGS] [GRPC_CHTTP2_SETTINGS_INITIAL_WINDOW_SIZE]); - uint32_t max_outgoing = (uint32_t)GPR_MIN( - t->settings[GRPC_PEER_SETTINGS] - [GRPC_CHTTP2_SETTINGS_MAX_FRAME_SIZE], - GPR_MIN(stream_outgoing_window, t->outgoing_window)); + uint32_t max_outgoing = + (uint32_t)GPR_MIN(t->settings[GRPC_PEER_SETTINGS] + [GRPC_CHTTP2_SETTINGS_MAX_FRAME_SIZE], + GPR_MIN(stream_remote_window, t->remote_window)); if (max_outgoing > 0) { uint32_t send_bytes = (uint32_t)GPR_MIN(max_outgoing, s->flow_controlled_buffer.length); @@ -325,10 +316,7 @@ grpc_chttp2_begin_write_result grpc_chttp2_begin_write( grpc_chttp2_encode_data(s->id, &s->flow_controlled_buffer, send_bytes, is_last_frame, &s->stats.outgoing, &t->outbuf); - GRPC_CHTTP2_FLOW_DEBIT_STREAM("write", t, s, outgoing_window_delta, - send_bytes); - GRPC_CHTTP2_FLOW_DEBIT_TRANSPORT("write", t, outgoing_window, - send_bytes); + grpc_chttp2_flowctl_sent_data(t, s, send_bytes); t->ping_state.pings_before_data_required = t->ping_policy.max_pings_without_data; if (!t->is_client) { @@ -351,10 +339,10 @@ grpc_chttp2_begin_write_result grpc_chttp2_begin_write( GRPC_CHTTP2_STREAM_REF(s, "chttp2_writing:fork"); grpc_chttp2_list_add_writable_stream(t, s); } - } else if (t->outgoing_window == 0) { + } else if (t->remote_window == 0) { grpc_chttp2_list_add_stalled_by_transport(t, s); now_writing = true; - } else if (stream_outgoing_window == 0) { + } else if (stream_remote_window == 0) { grpc_chttp2_list_add_stalled_by_stream(t, s); now_writing = true; } @@ -406,22 +394,15 @@ grpc_chttp2_begin_write_result grpc_chttp2_begin_write( } } - /* if the grpc_chttp2_transport is ready to send a window update, do so here - also; 3/4 is a magic number that will likely get tuned soon */ - uint32_t target_incoming_window = grpc_chttp2_target_incoming_window(t); - uint32_t threshold_to_send_transport_window_update = - t->outbuf.count > 0 ? 3 * target_incoming_window / 4 - : target_incoming_window / 2; - if (t->incoming_window <= threshold_to_send_transport_window_update && - t->incoming_window != target_incoming_window) { + uint32_t transport_announce = + grpc_chttp2_flowctl_maybe_send_transport_update(t); + if (transport_announce) { maybe_initiate_ping(exec_ctx, t, GRPC_CHTTP2_PING_BEFORE_TRANSPORT_WINDOW_UPDATE); - uint32_t announced = (uint32_t)GPR_CLAMP( - target_incoming_window - t->incoming_window, 0, UINT32_MAX); - GRPC_CHTTP2_FLOW_CREDIT_TRANSPORT("write", t, incoming_window, announced); grpc_transport_one_way_stats throwaway_stats; - grpc_slice_buffer_add(&t->outbuf, grpc_chttp2_window_update_create( - 0, announced, &throwaway_stats)); + grpc_slice_buffer_add( + &t->outbuf, grpc_chttp2_window_update_create(0, transport_announce, + &throwaway_stats)); t->ping_state.pings_before_data_required = t->ping_policy.max_pings_without_data; if (!t->is_client) { diff --git a/src/python/grpcio/grpc_core_dependencies.py b/src/python/grpcio/grpc_core_dependencies.py index 605044b65e..aa074df849 100644 --- a/src/python/grpcio/grpc_core_dependencies.py +++ b/src/python/grpcio/grpc_core_dependencies.py @@ -193,6 +193,7 @@ CORE_SOURCE_FILES = [ 'src/core/ext/transport/chttp2/transport/bin_encoder.c', 'src/core/ext/transport/chttp2/transport/chttp2_plugin.c', 'src/core/ext/transport/chttp2/transport/chttp2_transport.c', + 'src/core/ext/transport/chttp2/transport/flow_control.c', 'src/core/ext/transport/chttp2/transport/frame_data.c', 'src/core/ext/transport/chttp2/transport/frame_goaway.c', 'src/core/ext/transport/chttp2/transport/frame_ping.c', diff --git a/test/cpp/microbenchmarks/bm_chttp2_transport.cc b/test/cpp/microbenchmarks/bm_chttp2_transport.cc index 567ef1cf24..3b42c82de2 100644 --- a/test/cpp/microbenchmarks/bm_chttp2_transport.cc +++ b/test/cpp/microbenchmarks/bm_chttp2_transport.cc @@ -391,8 +391,8 @@ static void BM_TransportStreamSend(benchmark::State &state) { MakeClosure([&](grpc_exec_ctx *exec_ctx, grpc_error *error) { if (!state.KeepRunning()) return; // force outgoing window to be yuge - s.chttp2_stream()->outgoing_window_delta = 1024 * 1024 * 1024; - f.chttp2_transport()->outgoing_window = 1024 * 1024 * 1024; + s.chttp2_stream()->remote_window_delta = 1024 * 1024 * 1024; + f.chttp2_transport()->remote_window = 1024 * 1024 * 1024; grpc_slice_buffer_stream_init(&send_stream, &send_buffer, 0); reset_op(); op.on_complete = c.get(); @@ -521,8 +521,10 @@ static void BM_TransportStreamRecv(benchmark::State &state) { MakeClosure([&](grpc_exec_ctx *exec_ctx, grpc_error *error) { if (!state.KeepRunning()) return; // force outgoing window to be yuge - s.chttp2_stream()->incoming_window_delta = 1024 * 1024 * 1024; - f.chttp2_transport()->incoming_window = 1024 * 1024 * 1024; + s.chttp2_stream()->local_window_delta = 1024 * 1024 * 1024; + f.chttp2_transport()->local_window = 1024 * 1024 * 1024; + s.chttp2_stream()->announced_window_delta = 1024 * 1024 * 1024; + f.chttp2_transport()->announced_window = 1024 * 1024 * 1024; received = 0; reset_op(); op.on_complete = do_nothing.get(); diff --git a/test/cpp/microbenchmarks/bm_fullstack_trickle.cc b/test/cpp/microbenchmarks/bm_fullstack_trickle.cc index aeec7d831b..3464999bf0 100644 --- a/test/cpp/microbenchmarks/bm_fullstack_trickle.cc +++ b/test/cpp/microbenchmarks/bm_fullstack_trickle.cc @@ -73,11 +73,12 @@ class TrickledCHTTP2 : public EndpointPairFixture { log_.reset(new std::ofstream(fn.str().c_str())); write_csv(log_.get(), "t", "iteration", "client_backlog", "server_backlog", "client_t_stall", "client_s_stall", - "server_t_stall", "server_s_stall", "client_t_outgoing", - "server_t_outgoing", "client_t_incoming", "server_t_incoming", - "client_s_outgoing_delta", "server_s_outgoing_delta", - "client_s_incoming_delta", "server_s_incoming_delta", - "client_s_announce_window", "server_s_announce_window", + "server_t_stall", "server_s_stall", "client_t_remote", + "server_t_remote", "client_t_local", "server_t_local", + "client_t_announced", "server_t_announced", + "client_s_remote_delta", "server_s_remote_delta", + "client_s_local_delta", "server_s_local_delta", + "client_s_announced_delta", "server_s_announced_delta", "client_peer_iws", "client_local_iws", "client_sent_iws", "client_acked_iws", "server_peer_iws", "server_local_iws", "server_sent_iws", "server_acked_iws", "client_queued_bytes", @@ -127,14 +128,15 @@ class TrickledCHTTP2 : public EndpointPairFixture { client->lists[GRPC_CHTTP2_LIST_STALLED_BY_STREAM].head != nullptr, server->lists[GRPC_CHTTP2_LIST_STALLED_BY_TRANSPORT].head != nullptr, server->lists[GRPC_CHTTP2_LIST_STALLED_BY_STREAM].head != nullptr, - client->outgoing_window, server->outgoing_window, - client->incoming_window, server->incoming_window, - client_stream ? client_stream->outgoing_window_delta : -1, - server_stream ? server_stream->outgoing_window_delta : -1, - client_stream ? client_stream->incoming_window_delta : -1, - server_stream ? server_stream->incoming_window_delta : -1, - client_stream ? client_stream->announce_window : -1, - server_stream ? server_stream->announce_window : -1, + client->remote_window, server->remote_window, client->local_window, + server->local_window, client->announced_window, + server->announced_window, + client_stream ? client_stream->remote_window_delta : -1, + server_stream ? server_stream->remote_window_delta : -1, + client_stream ? client_stream->local_window_delta : -1, + server_stream ? server_stream->local_window_delta : -1, + client_stream ? client_stream->announced_window_delta : -1, + server_stream ? server_stream->announced_window_delta : -1, client->settings[GRPC_PEER_SETTINGS] [GRPC_CHTTP2_SETTINGS_INITIAL_WINDOW_SIZE], client->settings[GRPC_LOCAL_SETTINGS] diff --git a/tools/doxygen/Doxyfile.core.internal b/tools/doxygen/Doxyfile.core.internal index 4ad8a0ddb4..e69c8d411f 100644 --- a/tools/doxygen/Doxyfile.core.internal +++ b/tools/doxygen/Doxyfile.core.internal @@ -1014,6 +1014,7 @@ src/core/ext/transport/chttp2/transport/bin_encoder.h \ src/core/ext/transport/chttp2/transport/chttp2_plugin.c \ src/core/ext/transport/chttp2/transport/chttp2_transport.c \ src/core/ext/transport/chttp2/transport/chttp2_transport.h \ +src/core/ext/transport/chttp2/transport/flow_control.c \ src/core/ext/transport/chttp2/transport/frame.h \ src/core/ext/transport/chttp2/transport/frame_data.c \ src/core/ext/transport/chttp2/transport/frame_data.h \ diff --git a/tools/run_tests/generated/sources_and_headers.json b/tools/run_tests/generated/sources_and_headers.json index 8993e9340a..4178ede737 100644 --- a/tools/run_tests/generated/sources_and_headers.json +++ b/tools/run_tests/generated/sources_and_headers.json @@ -8667,6 +8667,7 @@ "src/core/ext/transport/chttp2/transport/chttp2_plugin.c", "src/core/ext/transport/chttp2/transport/chttp2_transport.c", "src/core/ext/transport/chttp2/transport/chttp2_transport.h", + "src/core/ext/transport/chttp2/transport/flow_control.c", "src/core/ext/transport/chttp2/transport/frame.h", "src/core/ext/transport/chttp2/transport/frame_data.c", "src/core/ext/transport/chttp2/transport/frame_data.h", diff --git a/vsprojects/vcxproj/grpc/grpc.vcxproj b/vsprojects/vcxproj/grpc/grpc.vcxproj index 41fe1b222b..7a4810ce9c 100644 --- a/vsprojects/vcxproj/grpc/grpc.vcxproj +++ b/vsprojects/vcxproj/grpc/grpc.vcxproj @@ -786,6 +786,8 @@ </ClCompile> <ClCompile Include="$(SolutionDir)\..\src\core\ext\transport\chttp2\transport\chttp2_transport.c"> </ClCompile> + <ClCompile Include="$(SolutionDir)\..\src\core\ext\transport\chttp2\transport\flow_control.c"> + </ClCompile> <ClCompile Include="$(SolutionDir)\..\src\core\ext\transport\chttp2\transport\frame_data.c"> </ClCompile> <ClCompile Include="$(SolutionDir)\..\src\core\ext\transport\chttp2\transport\frame_goaway.c"> diff --git a/vsprojects/vcxproj/grpc/grpc.vcxproj.filters b/vsprojects/vcxproj/grpc/grpc.vcxproj.filters index 00b8705826..5b4b6d9609 100644 --- a/vsprojects/vcxproj/grpc/grpc.vcxproj.filters +++ b/vsprojects/vcxproj/grpc/grpc.vcxproj.filters @@ -397,6 +397,9 @@ <ClCompile Include="$(SolutionDir)\..\src\core\ext\transport\chttp2\transport\chttp2_transport.c"> <Filter>src\core\ext\transport\chttp2\transport</Filter> </ClCompile> + <ClCompile Include="$(SolutionDir)\..\src\core\ext\transport\chttp2\transport\flow_control.c"> + <Filter>src\core\ext\transport\chttp2\transport</Filter> + </ClCompile> <ClCompile Include="$(SolutionDir)\..\src\core\ext\transport\chttp2\transport\frame_data.c"> <Filter>src\core\ext\transport\chttp2\transport</Filter> </ClCompile> diff --git a/vsprojects/vcxproj/grpc_unsecure/grpc_unsecure.vcxproj b/vsprojects/vcxproj/grpc_unsecure/grpc_unsecure.vcxproj index ac98dc0540..828596ae42 100644 --- a/vsprojects/vcxproj/grpc_unsecure/grpc_unsecure.vcxproj +++ b/vsprojects/vcxproj/grpc_unsecure/grpc_unsecure.vcxproj @@ -754,6 +754,8 @@ </ClCompile> <ClCompile Include="$(SolutionDir)\..\src\core\ext\transport\chttp2\transport\chttp2_transport.c"> </ClCompile> + <ClCompile Include="$(SolutionDir)\..\src\core\ext\transport\chttp2\transport\flow_control.c"> + </ClCompile> <ClCompile Include="$(SolutionDir)\..\src\core\ext\transport\chttp2\transport\frame_data.c"> </ClCompile> <ClCompile Include="$(SolutionDir)\..\src\core\ext\transport\chttp2\transport\frame_goaway.c"> diff --git a/vsprojects/vcxproj/grpc_unsecure/grpc_unsecure.vcxproj.filters b/vsprojects/vcxproj/grpc_unsecure/grpc_unsecure.vcxproj.filters index fea68c51dc..71f5518d0e 100644 --- a/vsprojects/vcxproj/grpc_unsecure/grpc_unsecure.vcxproj.filters +++ b/vsprojects/vcxproj/grpc_unsecure/grpc_unsecure.vcxproj.filters @@ -403,6 +403,9 @@ <ClCompile Include="$(SolutionDir)\..\src\core\ext\transport\chttp2\transport\chttp2_transport.c"> <Filter>src\core\ext\transport\chttp2\transport</Filter> </ClCompile> + <ClCompile Include="$(SolutionDir)\..\src\core\ext\transport\chttp2\transport\flow_control.c"> + <Filter>src\core\ext\transport\chttp2\transport</Filter> + </ClCompile> <ClCompile Include="$(SolutionDir)\..\src\core\ext\transport\chttp2\transport\frame_data.c"> <Filter>src\core\ext\transport\chttp2\transport</Filter> </ClCompile> |