From 3d7c60944474fd194f491355cd61ae4ad5adfaa4 Mon Sep 17 00:00:00 2001
From: Craig Tiller <ctiller@google.com>
Date: Fri, 26 Aug 2016 11:39:05 -0700
Subject: [PATCH] Start merging writes with the main transport thread
---
.../chttp2/transport/chttp2_transport.c | 1190 +++++++----------
.../ext/transport/chttp2/transport/frame.h | 4 +-
.../transport/chttp2/transport/frame_data.h | 8 +-
.../transport/chttp2/transport/frame_goaway.h | 9 +-
.../transport/chttp2/transport/frame_ping.h | 8 +-
.../chttp2/transport/frame_rst_stream.h | 9 +-
.../chttp2/transport/frame_settings.h | 9 +-
.../chttp2/transport/frame_window_update.h | 5 +-
.../transport/chttp2/transport/hpack_parser.h | 9 +-
.../ext/transport/chttp2/transport/internal.h | 452 +++----
10 files changed, 681 insertions(+), 1022 deletions(-)
diff --git a/src/core/ext/transport/chttp2/transport/chttp2_transport.c b/src/core/ext/transport/chttp2/transport/chttp2_transport.c
index 19e988670c..7a70744e5c 100644
--- a/src/core/ext/transport/chttp2/transport/chttp2_transport.c
+++ b/src/core/ext/transport/chttp2/transport/chttp2_transport.c
@@ -63,18 +63,6 @@
#define MAX_CLIENT_STREAM_ID 0x7fffffffu
int grpc_http_trace = 0;
int grpc_flowctl_trace = 0;
-int grpc_http_write_state_trace = 0;
-
-#define TRANSPORT_FROM_WRITING(tw) \
- ((grpc_chttp2_transport *)((char *)(tw)-offsetof(grpc_chttp2_transport, \
- writing)))
-
-#define TRANSPORT_FROM_GLOBAL(tg) \
- ((grpc_chttp2_transport *)((char *)(tg)-offsetof(grpc_chttp2_transport, \
- global)))
-
-#define STREAM_FROM_GLOBAL(sg) \
- ((grpc_chttp2_stream *)((char *)(sg)-offsetof(grpc_chttp2_stream, global)))
static const grpc_transport_vtable vtable;
@@ -91,8 +79,6 @@ static void terminate_writing_with_lock(grpc_exec_ctx *exec_ctx, void *t,
grpc_error *error);
static void start_writing(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t);
-static void end_waiting_for_write(grpc_exec_ctx *exec_ctx,
- grpc_chttp2_transport *t, grpc_error *error);
/** Set a transport level setting, and push it to our peer */
static void push_setting(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t,
@@ -102,37 +88,32 @@ static void push_setting(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t,
static void drop_connection(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t,
grpc_error *error);
-static void close_from_api(grpc_exec_ctx *exec_ctx,
- grpc_chttp2_transport_global *transport_global,
- grpc_chttp2_stream_global *stream_global,
- grpc_error *error);
+static void close_from_api(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t,
+ grpc_chttp2_stream *s, grpc_error *error);
/** Start new streams that have been created if we can */
-static void maybe_start_some_streams(
- grpc_exec_ctx *exec_ctx, grpc_chttp2_transport_global *transport_global);
+static void maybe_start_some_streams(grpc_exec_ctx *exec_ctx,
+ grpc_chttp2_transport *t);
-static void connectivity_state_set(
- grpc_exec_ctx *exec_ctx, grpc_chttp2_transport_global *transport_global,
- grpc_connectivity_state state, grpc_error *error, const char *reason);
+static void connectivity_state_set(grpc_exec_ctx *exec_ctx,
+ grpc_chttp2_transport *t,
+ grpc_connectivity_state state,
+ grpc_error *error, const char *reason);
-static void check_read_ops(grpc_exec_ctx *exec_ctx,
- grpc_chttp2_transport_global *transport_global);
+static void check_read_ops(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t);
-static void incoming_byte_stream_update_flow_control(
- grpc_exec_ctx *exec_ctx, grpc_chttp2_transport_global *transport_global,
- grpc_chttp2_stream_global *stream_global, size_t max_size_hint,
- size_t have_already);
+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);
static void fail_pending_writes(grpc_exec_ctx *exec_ctx,
- grpc_chttp2_transport_global *transport_global,
- grpc_chttp2_stream_global *stream_global,
+ grpc_chttp2_transport *t, grpc_chttp2_stream *s,
grpc_error *error);
-static void set_write_state(grpc_chttp2_transport *t,
- grpc_chttp2_write_state state, const char *reason);
-
/*******************************************************************************
* CONSTRUCTION/DESTRUCTION/REFCOUNTING
*/
@@ -143,14 +124,14 @@ static void destruct_transport(grpc_exec_ctx *exec_ctx,
grpc_endpoint_destroy(exec_ctx, t->ep);
- gpr_slice_buffer_destroy(&t->global.qbuf);
+ gpr_slice_buffer_destroy(&t->qbuf);
- gpr_slice_buffer_destroy(&t->writing.outbuf);
- grpc_chttp2_hpack_compressor_destroy(&t->writing.hpack_compressor);
+ gpr_slice_buffer_destroy(&t->outbuf);
+ grpc_chttp2_hpack_compressor_destroy(&t->hpack_compressor);
gpr_slice_buffer_destroy(&t->read_buffer);
- grpc_chttp2_hpack_parser_destroy(&t->global.hpack_parser);
- grpc_chttp2_goaway_parser_destroy(&t->global.goaway_parser);
+ grpc_chttp2_hpack_parser_destroy(&t->hpack_parser);
+ grpc_chttp2_goaway_parser_destroy(&t->goaway_parser);
for (i = 0; i < STREAM_LIST_COUNT; i++) {
GPR_ASSERT(t->lists[i].head == NULL);
@@ -162,12 +143,12 @@ static void destruct_transport(grpc_exec_ctx *exec_ctx,
grpc_chttp2_stream_map_destroy(&t->stream_map);
grpc_connectivity_state_destroy(exec_ctx, &t->channel_callback.state_tracker);
- grpc_combiner_destroy(exec_ctx, t->executor.combiner);
+ grpc_combiner_destroy(exec_ctx, t->combiner);
/* callback remaining pings: they're not allowed to call into the transpot,
and maybe they hold resources that need to be freed */
- while (t->global.pings.next != &t->global.pings) {
- grpc_chttp2_outstanding_ping *ping = t->global.pings.next;
+ while (t->pings.next != &t->pings) {
+ grpc_chttp2_outstanding_ping *ping = t->pings.next;
grpc_exec_ctx_sched(exec_ctx, ping->on_recv,
GRPC_ERROR_CREATE("Transport closed"), NULL);
ping->next->prev = ping->prev;
@@ -217,47 +198,41 @@ static void init_transport(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t,
memset(t, 0, sizeof(*t));
t->base.vtable = &vtable;
- t->executor.write_state = GRPC_CHTTP2_WRITES_CORKED;
t->ep = ep;
/* one ref is for destroy */
gpr_ref_init(&t->refs, 1);
/* ref is dropped at transport close() */
gpr_ref_init(&t->shutdown_ep_refs, 1);
- t->executor.combiner = grpc_combiner_create(grpc_endpoint_get_workqueue(ep));
+ t->combiner = grpc_combiner_create(grpc_endpoint_get_workqueue(ep));
t->peer_string = grpc_endpoint_get_peer(ep);
t->endpoint_reading = 1;
- t->global.next_stream_id = is_client ? 1 : 2;
- t->global.is_client = is_client;
- t->writing.outgoing_window = DEFAULT_WINDOW;
- t->global.incoming_window = DEFAULT_WINDOW;
- t->global.stream_lookahead = DEFAULT_WINDOW;
- t->global.connection_window_target = DEFAULT_CONNECTION_WINDOW_TARGET;
- t->global.ping_counter = 1;
- t->global.pings.next = t->global.pings.prev = &t->global.pings;
- t->global.deframe_state =
- is_client ? GRPC_DTS_FH_0 : GRPC_DTS_CLIENT_PREFIX_0;
- t->global.is_first_frame = true;
- t->writing.is_client = is_client;
+ t->next_stream_id = is_client ? 1 : 2;
+ t->is_client = is_client;
+ t->outgoing_window = DEFAULT_WINDOW;
+ t->incoming_window = DEFAULT_WINDOW;
+ t->stream_lookahead = DEFAULT_WINDOW;
+ t->connection_window_target = DEFAULT_CONNECTION_WINDOW_TARGET;
+ t->ping_counter = 1;
+ t->pings.next = t->pings.prev = &t->pings;
+ t->deframe_state = is_client ? GRPC_DTS_FH_0 : GRPC_DTS_CLIENT_PREFIX_0;
+ t->is_first_frame = true;
grpc_connectivity_state_init(
&t->channel_callback.state_tracker, GRPC_CHANNEL_READY,
is_client ? "client_transport" : "server_transport");
- gpr_slice_buffer_init(&t->global.qbuf);
+ gpr_slice_buffer_init(&t->qbuf);
- gpr_slice_buffer_init(&t->writing.outbuf);
- grpc_chttp2_hpack_compressor_init(&t->writing.hpack_compressor);
+ gpr_slice_buffer_init(&t->outbuf);
+ grpc_chttp2_hpack_compressor_init(&t->hpack_compressor);
grpc_closure_init(&t->writing_action, writing_action, t);
grpc_closure_init(&t->reading_action, reading_action, t);
grpc_closure_init(&t->reading_action_locked, reading_action_locked, t);
- grpc_closure_init(&t->initiate_writing, initiate_writing_locked, t);
grpc_closure_init(&t->terminate_writing, terminate_writing_with_lock, t);
grpc_closure_init(&t->initiate_read_flush_locked, initiate_read_flush_locked,
t);
- grpc_closure_init(&t->writing.done_cb, grpc_chttp2_terminate_writing,
- &t->writing);
- grpc_chttp2_goaway_parser_init(&t->global.goaway_parser);
- grpc_chttp2_hpack_parser_init(&t->global.hpack_parser);
+ grpc_chttp2_goaway_parser_init(&t->goaway_parser);
+ grpc_chttp2_hpack_parser_init(&t->hpack_parser);
gpr_slice_buffer_init(&t->read_buffer);
@@ -271,21 +246,19 @@ static void init_transport(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t,
/* copy in initial settings to all setting sets */
for (i = 0; i < GRPC_CHTTP2_NUM_SETTINGS; i++) {
for (j = 0; j < GRPC_NUM_SETTING_SETS; j++) {
- t->global.settings[j][i] =
- grpc_chttp2_settings_parameters[i].default_value;
+ t->settings[j][i] = grpc_chttp2_settings_parameters[i].default_value;
}
}
- t->global.dirtied_local_settings = 1;
+ t->dirtied_local_settings = 1;
/* Hack: it's common for implementations to assume 65536 bytes initial send
window -- this should by rights be 0 */
- t->global.force_send_settings = 1 << GRPC_CHTTP2_SETTINGS_INITIAL_WINDOW_SIZE;
- t->global.sent_local_settings = 0;
+ t->force_send_settings = 1 << GRPC_CHTTP2_SETTINGS_INITIAL_WINDOW_SIZE;
+ t->sent_local_settings = 0;
if (is_client) {
- gpr_slice_buffer_add(
- &t->writing.outbuf,
- gpr_slice_from_copied_string(GRPC_CHTTP2_CLIENT_CONNECT_STRING));
- grpc_chttp2_initiate_write(exec_ctx, &t->global, false, "initial_write");
+ gpr_slice_buffer_add(&t->outbuf, gpr_slice_from_copied_string(
+ GRPC_CHTTP2_CLIENT_CONNECT_STRING));
+ grpc_chttp2_initiate_write(exec_ctx, t, false, "initial_write");
}
/* configure http2 the way we like it */
@@ -317,15 +290,13 @@ static void init_transport(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t,
if (channel_args->args[i].type != GRPC_ARG_INTEGER) {
gpr_log(GPR_ERROR, "%s: must be an integer",
GRPC_ARG_HTTP2_INITIAL_SEQUENCE_NUMBER);
- } else if ((t->global.next_stream_id & 1) !=
+ } else if ((t->next_stream_id & 1) !=
(channel_args->args[i].value.integer & 1)) {
gpr_log(GPR_ERROR, "%s: low bit must be %d on %s",
- GRPC_ARG_HTTP2_INITIAL_SEQUENCE_NUMBER,
- t->global.next_stream_id & 1,
+ GRPC_ARG_HTTP2_INITIAL_SEQUENCE_NUMBER, t->next_stream_id & 1,
is_client ? "client" : "server");
} else {
- t->global.next_stream_id =
- (uint32_t)channel_args->args[i].value.integer;
+ t->next_stream_id = (uint32_t)channel_args->args[i].value.integer;
}
} else if (0 == strcmp(channel_args->args[i].key,
GRPC_ARG_HTTP2_STREAM_LOOKAHEAD_BYTES)) {
@@ -336,8 +307,7 @@ static void init_transport(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t,
gpr_log(GPR_ERROR, "%s: must be at least 5",
GRPC_ARG_HTTP2_STREAM_LOOKAHEAD_BYTES);
} else {
- t->global.stream_lookahead =
- (uint32_t)channel_args->args[i].value.integer;
+ t->stream_lookahead = (uint32_t)channel_args->args[i].value.integer;
}
} else if (0 == strcmp(channel_args->args[i].key,
GRPC_ARG_HTTP2_HPACK_TABLE_SIZE_DECODER)) {
@@ -361,7 +331,7 @@ static void init_transport(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t,
GRPC_ARG_HTTP2_HPACK_TABLE_SIZE_ENCODER);
} else {
grpc_chttp2_hpack_compressor_set_max_usable_size(
- &t->writing.hpack_compressor,
+ &t->hpack_compressor,
(uint32_t)channel_args->args[i].value.integer);
}
} else if (0 == strcmp(channel_args->args[i].key,
@@ -380,8 +350,7 @@ static void init_transport(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t,
}
}
- set_write_state(t, GRPC_CHTTP2_WRITING_INACTIVE, "uncork");
- grpc_chttp2_initiate_write(exec_ctx, &t->global, false, "init");
+ grpc_chttp2_initiate_write(exec_ctx, t, false, "init");
}
static void destroy_transport_locked(grpc_exec_ctx *exec_ctx, void *tp,
@@ -394,7 +363,7 @@ static void destroy_transport_locked(grpc_exec_ctx *exec_ctx, void *tp,
static void destroy_transport(grpc_exec_ctx *exec_ctx, grpc_transport *gt) {
grpc_chttp2_transport *t = (grpc_chttp2_transport *)gt;
- grpc_combiner_execute(exec_ctx, t->executor.combiner,
+ grpc_combiner_execute(exec_ctx, t->combiner,
grpc_closure_create(destroy_transport_locked, t),
GRPC_ERROR_NONE);
}
@@ -416,43 +385,34 @@ static void close_transport_locked(grpc_exec_ctx *exec_ctx,
grpc_chttp2_transport *t,
grpc_error *error) {
if (!t->closed) {
- if (grpc_http_write_state_trace) {
- gpr_log(GPR_DEBUG, "W:%p close transport", t);
- }
t->closed = 1;
- connectivity_state_set(exec_ctx, &t->global, GRPC_CHANNEL_SHUTDOWN,
+ connectivity_state_set(exec_ctx, t, GRPC_CHANNEL_SHUTDOWN,
GRPC_ERROR_REF(error), "close_transport");
allow_endpoint_shutdown_locked(exec_ctx, t);
/* flush writable stream list to avoid dangling references */
- grpc_chttp2_stream_global *stream_global;
- grpc_chttp2_stream_writing *stream_writing;
- while (grpc_chttp2_list_pop_writable_stream(
- &t->global, &t->writing, &stream_global, &stream_writing)) {
- GRPC_CHTTP2_STREAM_UNREF(exec_ctx, stream_global, "chttp2_writing");
+ grpc_chttp2_stream *s;
+ while (grpc_chttp2_list_pop_writable_stream(t, &s)) {
+ GRPC_CHTTP2_STREAM_UNREF(exec_ctx, s, "chttp2_writing");
}
}
GRPC_ERROR_UNREF(error);
}
#ifdef GRPC_STREAM_REFCOUNT_DEBUG
-void grpc_chttp2_stream_ref(grpc_chttp2_stream_global *stream_global,
- const char *reason) {
- grpc_stream_ref(STREAM_FROM_GLOBAL(stream_global)->refcount, reason);
+void grpc_chttp2_stream_ref(grpc_chttp2_stream *s, const char *reason) {
+ grpc_stream_ref(s->refcount, reason);
}
-void grpc_chttp2_stream_unref(grpc_exec_ctx *exec_ctx,
- grpc_chttp2_stream_global *stream_global,
+void grpc_chttp2_stream_unref(grpc_exec_ctx *exec_ctx, grpc_chttp2_stream *s,
const char *reason) {
- grpc_stream_unref(exec_ctx, STREAM_FROM_GLOBAL(stream_global)->refcount,
- reason);
+ grpc_stream_unref(exec_ctx, s->refcount, reason);
}
#else
-void grpc_chttp2_stream_ref(grpc_chttp2_stream_global *stream_global) {
- grpc_stream_ref(STREAM_FROM_GLOBAL(stream_global)->refcount);
+void grpc_chttp2_stream_ref(grpc_chttp2_stream *s) {
+ grpc_stream_ref(s->refcount);
}
-void grpc_chttp2_stream_unref(grpc_exec_ctx *exec_ctx,
- grpc_chttp2_stream_global *stream_global) {
- grpc_stream_unref(exec_ctx, STREAM_FROM_GLOBAL(stream_global)->refcount);
+void grpc_chttp2_stream_unref(grpc_exec_ctx *exec_ctx, grpc_chttp2_stream *s) {
+ grpc_stream_unref(exec_ctx, s->refcount);
}
#endif
@@ -470,28 +430,27 @@ static int init_stream(grpc_exec_ctx *exec_ctx, grpc_transport *gt,
/* We reserve one 'active stream' that's dropped when the stream is
read-closed. The others are for incoming_byte_streams that are actively
reading */
- gpr_ref_init(&s->global.active_streams, 1);
- GRPC_CHTTP2_STREAM_REF(&s->global, "chttp2");
+ gpr_ref_init(&s->active_streams, 1);
+ GRPC_CHTTP2_STREAM_REF(s, "chttp2");
- grpc_chttp2_incoming_metadata_buffer_init(&s->global.metadata_buffer[0]);
- grpc_chttp2_incoming_metadata_buffer_init(&s->global.metadata_buffer[1]);
- grpc_chttp2_data_parser_init(&s->global.data_parser);
- gpr_slice_buffer_init(&s->writing.flow_controlled_buffer);
- s->global.deadline = gpr_inf_future(GPR_CLOCK_MONOTONIC);
+ grpc_chttp2_incoming_metadata_buffer_init(&s->metadata_buffer[0]);
+ grpc_chttp2_incoming_metadata_buffer_init(&s->metadata_buffer[1]);
+ grpc_chttp2_data_parser_init(&s->data_parser);
+ gpr_slice_buffer_init(&s->flow_controlled_buffer);
+ s->deadline = gpr_inf_future(GPR_CLOCK_MONOTONIC);
GRPC_CHTTP2_REF_TRANSPORT(t, "stream");
if (server_data) {
- s->global.id = (uint32_t)(uintptr_t)server_data;
- s->global.outgoing_window =
- t->global.settings[GRPC_PEER_SETTINGS]
- [GRPC_CHTTP2_SETTINGS_INITIAL_WINDOW_SIZE];
- s->global.incoming_window = s->global.max_recv_bytes =
- t->global.settings[GRPC_SENT_SETTINGS]
- [GRPC_CHTTP2_SETTINGS_INITIAL_WINDOW_SIZE];
+ s->id = (uint32_t)(uintptr_t)server_data;
+ s->outgoing_window = t->settings[GRPC_PEER_SETTINGS]
+ [GRPC_CHTTP2_SETTINGS_INITIAL_WINDOW_SIZE];
+ s->incoming_window = s->max_recv_bytes =
+ t->settings[GRPC_SENT_SETTINGS]
+ [GRPC_CHTTP2_SETTINGS_INITIAL_WINDOW_SIZE];
*t->accepting_stream = s;
- grpc_chttp2_stream_map_add(&t->stream_map, s->global.id, s);
- s->global.in_stream_map = true;
+ grpc_chttp2_stream_map_add(&t->stream_map, s->id, s);
+ s->in_stream_map = true;
}
GPR_TIMER_END("init_stream", 0);
@@ -507,42 +466,39 @@ static void destroy_stream_locked(grpc_exec_ctx *exec_ctx, void *sp,
GPR_TIMER_BEGIN("destroy_stream", 0);
- GPR_ASSERT((s->global.write_closed && s->global.read_closed) ||
- s->global.id == 0);
- GPR_ASSERT(!s->global.in_stream_map);
- if (s->global.id != 0) {
- GPR_ASSERT(grpc_chttp2_stream_map_find(&t->stream_map, s->global.id) ==
- NULL);
+ GPR_ASSERT((s->write_closed && s->read_closed) || s->id == 0);
+ GPR_ASSERT(!s->in_stream_map);
+ if (s->id != 0) {
+ GPR_ASSERT(grpc_chttp2_stream_map_find(&t->stream_map, s->id) == NULL);
}
- while (
- (bs = grpc_chttp2_incoming_frame_queue_pop(&s->global.incoming_frames))) {
+ while ((bs = grpc_chttp2_incoming_frame_queue_pop(&s->incoming_frames))) {
incoming_byte_stream_destroy_locked(exec_ctx, bs, GRPC_ERROR_NONE);
}
- grpc_chttp2_list_remove_stalled_by_transport(&t->global, &s->global);
- grpc_chttp2_list_remove_check_read_ops(&t->global, &s->global);
+ grpc_chttp2_list_remove_stalled_by_transport(t, s);
+ grpc_chttp2_list_remove_check_read_ops(t, s);
for (int i = 0; i < STREAM_LIST_COUNT; i++) {
if (s->included[i]) {
gpr_log(GPR_ERROR, "%s stream %d still included in list %d",
- t->global.is_client ? "client" : "server", s->global.id, i);
+ t->is_client ? "client" : "server", s->id, i);
abort();
}
}
- GPR_ASSERT(s->global.send_initial_metadata_finished == NULL);
- GPR_ASSERT(s->global.send_message_finished == NULL);
- GPR_ASSERT(s->global.send_trailing_metadata_finished == NULL);
- GPR_ASSERT(s->global.recv_initial_metadata_ready == NULL);
- GPR_ASSERT(s->global.recv_message_ready == NULL);
- GPR_ASSERT(s->global.recv_trailing_metadata_finished == NULL);
- grpc_chttp2_data_parser_destroy(exec_ctx, &s->global.data_parser);
- grpc_chttp2_incoming_metadata_buffer_destroy(&s->global.metadata_buffer[0]);
- grpc_chttp2_incoming_metadata_buffer_destroy(&s->global.metadata_buffer[1]);
- gpr_slice_buffer_destroy(&s->writing.flow_controlled_buffer);
- GRPC_ERROR_UNREF(s->global.read_closed_error);
- GRPC_ERROR_UNREF(s->global.write_closed_error);
+ GPR_ASSERT(s->send_initial_metadata_finished == NULL);
+ GPR_ASSERT(s->send_message_finished == NULL);
+ GPR_ASSERT(s->send_trailing_metadata_finished == NULL);
+ GPR_ASSERT(s->recv_initial_metadata_ready == NULL);
+ GPR_ASSERT(s->recv_message_ready == NULL);
+ GPR_ASSERT(s->recv_trailing_metadata_finished == NULL);
+ grpc_chttp2_data_parser_destroy(exec_ctx, &s->data_parser);
+ grpc_chttp2_incoming_metadata_buffer_destroy(&s->metadata_buffer[0]);
+ grpc_chttp2_incoming_metadata_buffer_destroy(&s->metadata_buffer[1]);
+ gpr_slice_buffer_destroy(&s->flow_controlled_buffer);
+ GRPC_ERROR_UNREF(s->read_closed_error);
+ GRPC_ERROR_UNREF(s->write_closed_error);
GRPC_CHTTP2_UNREF_TRANSPORT(exec_ctx, t, "stream");
@@ -559,76 +515,47 @@ static void destroy_stream(grpc_exec_ctx *exec_ctx, grpc_transport *gt,
s->destroy_stream_arg = and_free_memory;
grpc_closure_init(&s->destroy_stream, destroy_stream_locked, s);
- grpc_combiner_execute(exec_ctx, t->executor.combiner, &s->destroy_stream,
+ grpc_combiner_execute(exec_ctx, t->combiner, &s->destroy_stream,
GRPC_ERROR_NONE);
GPR_TIMER_END("destroy_stream", 0);
}
-grpc_chttp2_stream_global *grpc_chttp2_parsing_lookup_stream(
- grpc_chttp2_transport_global *transport_global, uint32_t id) {
- grpc_chttp2_transport *t = TRANSPORT_FROM_GLOBAL(transport_global);
- grpc_chttp2_stream *s = grpc_chttp2_stream_map_find(&t->stream_map, id);
- return s ? &s->global : NULL;
+grpc_chttp2_stream *grpc_chttp2_parsing_lookup_stream(grpc_chttp2_transport *t,
+ uint32_t id) {
+ return grpc_chttp2_stream_map_find(&t->stream_map, id);
}
-grpc_chttp2_stream_global *grpc_chttp2_parsing_accept_stream(
- grpc_exec_ctx *exec_ctx, grpc_chttp2_transport_global *transport_global,
- uint32_t id) {
+grpc_chttp2_stream *grpc_chttp2_parsing_accept_stream(grpc_exec_ctx *exec_ctx,
+ grpc_chttp2_transport *t,
+ uint32_t id) {
grpc_chttp2_stream *accepting;
- grpc_chttp2_transport *t = TRANSPORT_FROM_GLOBAL(transport_global);
GPR_ASSERT(t->accepting_stream == NULL);
t->accepting_stream = &accepting;
t->channel_callback.accept_stream(exec_ctx,
t->channel_callback.accept_stream_user_data,
&t->base, (void *)(uintptr_t)id);
t->accepting_stream = NULL;
- return &accepting->global;
+ return accepting;
}
/*******************************************************************************
* LOCK MANAGEMENT
*/
-static const char *write_state_name(grpc_chttp2_write_state state) {
- switch (state) {
- case GRPC_CHTTP2_WRITES_CORKED:
- return "CORKED";
- case GRPC_CHTTP2_WRITING_INACTIVE:
- return "INACTIVE";
- case GRPC_CHTTP2_WRITE_SCHEDULED:
- return "SCHEDULED";
- case GRPC_CHTTP2_WRITING:
- return "WRITING";
- case GRPC_CHTTP2_WRITING_STALE_WITH_POLLER:
- return "WRITING[p=1]";
- case GRPC_CHTTP2_WRITING_STALE_NO_POLLER:
- return "WRITING[p=0]";
- }
- GPR_UNREACHABLE_CODE(return "UNKNOWN");
-}
-
-static void set_write_state(grpc_chttp2_transport *t,
- grpc_chttp2_write_state state, const char *reason) {
- if (grpc_http_write_state_trace) {
- gpr_log(GPR_DEBUG, "W:%p %s -> %s because %s", t,
- write_state_name(t->executor.write_state), write_state_name(state),
- reason);
- }
- t->executor.write_state = state;
-}
-
+#if 0
static void initiate_writing_locked(grpc_exec_ctx *exec_ctx, void *tp,
grpc_error *error) {
grpc_chttp2_transport *t = tp;
GPR_ASSERT(t->executor.write_state == GRPC_CHTTP2_WRITE_SCHEDULED);
start_writing(exec_ctx, t);
}
+#endif
static void initiate_read_flush_locked(grpc_exec_ctx *exec_ctx, void *tp,
grpc_error *error) {
grpc_chttp2_transport *t = tp;
- t->executor.check_read_ops_scheduled = false;
- check_read_ops(exec_ctx, &t->global);
+ t->check_read_ops_scheduled = false;
+ check_read_ops(exec_ctx, t);
GRPC_CHTTP2_UNREF_TRANSPORT(exec_ctx, t, "initiate_read_flush_locked");
}
@@ -637,98 +564,49 @@ static void initiate_read_flush_locked(grpc_exec_ctx *exec_ctx, void *tp,
*/
void grpc_chttp2_initiate_write(grpc_exec_ctx *exec_ctx,
- grpc_chttp2_transport_global *transport_global,
+ grpc_chttp2_transport *t,
bool covered_by_poller, const char *reason) {
GPR_TIMER_BEGIN("grpc_chttp2_initiate_write", 0);
- /* Perform state checks, and transition to a scheduled state if appropriate.
- If we are inactive, schedule a write chain to begin once the transport
- combiner finishes any executions in its current batch (which may be
- scheduled AFTER this code executes). The write chain will:
- - call start_writing, which verifies (under the global lock) that there
- are things that need to be written by calling
- grpc_chttp2_unlocking_check_writes, and if so schedules writing_action
- against the current exec_ctx, to be executed OUTSIDE of the global lock
- - eventually writing_action results in grpc_chttp2_terminate_writing being
- called, which re-takes the global lock, updates state, checks if we need
- to do *another* write immediately, and if so loops back to
- start_writing.
-
- Current problems:
- - too much lock entry/exiting
- - the writing thread can become stuck indefinitely (punt through the
- workqueue periodically to fix) */
-
- grpc_chttp2_transport *t = TRANSPORT_FROM_GLOBAL(transport_global);
- switch (t->executor.write_state) {
- case GRPC_CHTTP2_WRITES_CORKED:
- break;
- case GRPC_CHTTP2_WRITING_INACTIVE:
- set_write_state(t, GRPC_CHTTP2_WRITE_SCHEDULED, reason);
+ switch (t->write_state) {
+ case GRPC_CHTTP2_WRITE_STATE_IDLE:
+ t->write_state = GRPC_CHTTP2_WRITE_STATE_WRITING;
GRPC_CHTTP2_REF_TRANSPORT(t, "writing");
- grpc_combiner_execute_finally(exec_ctx, t->executor.combiner,
- &t->initiate_writing, GRPC_ERROR_NONE,
- covered_by_poller);
- break;
- case GRPC_CHTTP2_WRITE_SCHEDULED:
- if (covered_by_poller) {
- /* upgrade to note poller is available to cover the write */
- grpc_combiner_force_async_finally(t->executor.combiner);
- }
+ grpc_combiner_execute_finally(exec_ctx, t->combiner, &t->writing_action,
+ GRPC_ERROR_NONE, false);
break;
- case GRPC_CHTTP2_WRITING:
- set_write_state(t,
- covered_by_poller ? GRPC_CHTTP2_WRITING_STALE_WITH_POLLER
- : GRPC_CHTTP2_WRITING_STALE_NO_POLLER,
- reason);
+ case GRPC_CHTTP2_WRITE_STATE_WRITING:
+ t->write_state = GRPC_CHTTP2_WRITE_STATE_WRITING_WITH_MORE_TO_COME;
break;
- case GRPC_CHTTP2_WRITING_STALE_WITH_POLLER:
- /* nothing to do: write already requested */
- break;
- case GRPC_CHTTP2_WRITING_STALE_NO_POLLER:
- if (covered_by_poller) {
- /* upgrade to note poller is available to cover the write */
- set_write_state(t, GRPC_CHTTP2_WRITING_STALE_WITH_POLLER, reason);
- }
+ case GRPC_CHTTP2_WRITE_STATE_WRITING_WITH_MORE_TO_COME:
break;
}
GPR_TIMER_END("grpc_chttp2_initiate_write", 0);
}
+void grpc_chttp2_become_writable(grpc_exec_ctx *exec_ctx,
+ grpc_chttp2_transport *t,
+ grpc_chttp2_stream *s, bool covered_by_poller,
+ const char *reason) {
+ if (!t->closed && grpc_chttp2_list_add_writable_stream(t, s)) {
+ GRPC_CHTTP2_STREAM_REF(s, "chttp2_writing");
+ grpc_chttp2_initiate_write(exec_ctx, t, covered_by_poller, reason);
+ }
+}
+
static void start_writing(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t) {
GPR_TIMER_BEGIN("start_writing", 0);
- GPR_ASSERT(t->executor.write_state == GRPC_CHTTP2_WRITE_SCHEDULED);
- if (!t->closed &&
- grpc_chttp2_unlocking_check_writes(exec_ctx, &t->global, &t->writing)) {
- set_write_state(t, GRPC_CHTTP2_WRITING, "start_writing");
+ GPR_ASSERT(t->write_state != GRPC_CHTTP2_WRITE_STATE_IDLE);
+ if (!t->closed && grpc_chttp2_begin_write(exec_ctx, t)) {
prevent_endpoint_shutdown(t);
grpc_exec_ctx_sched(exec_ctx, &t->writing_action, GRPC_ERROR_NONE, NULL);
} else {
- if (t->closed) {
- set_write_state(t, GRPC_CHTTP2_WRITING_INACTIVE,
- "start_writing:transport_closed");
- } else {
- set_write_state(t, GRPC_CHTTP2_WRITING_INACTIVE,
- "start_writing:nothing_to_write");
- }
- end_waiting_for_write(exec_ctx, t, GRPC_ERROR_NONE);
+ t->write_state = GRPC_CHTTP2_WRITE_STATE_IDLE;
GRPC_CHTTP2_UNREF_TRANSPORT(exec_ctx, t, "writing");
}
GPR_TIMER_END("start_writing", 0);
}
-void grpc_chttp2_become_writable(grpc_exec_ctx *exec_ctx,
- grpc_chttp2_transport_global *transport_global,
- grpc_chttp2_stream_global *stream_global,
- bool covered_by_poller, const char *reason) {
- if (!TRANSPORT_FROM_GLOBAL(transport_global)->closed &&
- grpc_chttp2_list_add_writable_stream(transport_global, stream_global)) {
- GRPC_CHTTP2_STREAM_REF(stream_global, "chttp2_writing");
- grpc_chttp2_initiate_write(exec_ctx, transport_global, covered_by_poller,
- reason);
- }
-}
-
static void push_setting(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t,
grpc_chttp2_setting_id id, uint32_t value) {
const grpc_chttp2_setting_parameters *sp =
@@ -738,25 +616,11 @@ static void push_setting(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t,
gpr_log(GPR_INFO, "Requested parameter %s clamped from %d to %d", sp->name,
value, use_value);
}
- if (use_value != t->global.settings[GRPC_LOCAL_SETTINGS][id]) {
- t->global.settings[GRPC_LOCAL_SETTINGS][id] = use_value;
- t->global.dirtied_local_settings = 1;
- grpc_chttp2_initiate_write(exec_ctx, &t->global, false, "push_setting");
- }
-}
-
-/* error may be GRPC_ERROR_NONE if there is no error allocated yet.
- In that case, use "reason" as the text for a new error. */
-static void end_waiting_for_write(grpc_exec_ctx *exec_ctx,
- grpc_chttp2_transport *t, grpc_error *error) {
- grpc_chttp2_stream_global *stream_global;
- while (grpc_chttp2_list_pop_closed_waiting_for_writing(&t->global,
- &stream_global)) {
- fail_pending_writes(exec_ctx, &t->global, stream_global,
- GRPC_ERROR_REF(error));
- GRPC_CHTTP2_STREAM_UNREF(exec_ctx, stream_global, "finish_writes");
+ if (use_value != t->settings[GRPC_LOCAL_SETTINGS][id]) {
+ t->settings[GRPC_LOCAL_SETTINGS][id] = use_value;
+ t->dirtied_local_settings = 1;
+ grpc_chttp2_initiate_write(exec_ctx, t, false, "push_setting");
}
- GRPC_ERROR_UNREF(error);
}
static void terminate_writing_with_lock(grpc_exec_ctx *exec_ctx, void *tp,
@@ -769,34 +633,21 @@ static void terminate_writing_with_lock(grpc_exec_ctx *exec_ctx, void *tp,
drop_connection(exec_ctx, t, GRPC_ERROR_REF(error));
}
- grpc_chttp2_cleanup_writing(exec_ctx, &t->global, &t->writing);
-
- end_waiting_for_write(exec_ctx, t, GRPC_ERROR_REF(error));
+ grpc_chttp2_end_write(exec_ctx, t, GRPC_ERROR_REF(error));
- switch (t->executor.write_state) {
- case GRPC_CHTTP2_WRITES_CORKED:
- case GRPC_CHTTP2_WRITING_INACTIVE:
- case GRPC_CHTTP2_WRITE_SCHEDULED:
+ switch (t->write_state) {
+ case GRPC_CHTTP2_WRITE_STATE_IDLE:
GPR_UNREACHABLE_CODE(break);
- case GRPC_CHTTP2_WRITING:
+ case GRPC_CHTTP2_WRITE_STATE_WRITING:
GPR_TIMER_MARK("state=writing", 0);
- set_write_state(t, GRPC_CHTTP2_WRITING_INACTIVE, "terminate_writing");
+ t->write_state = GRPC_CHTTP2_WRITE_STATE_IDLE;
break;
- case GRPC_CHTTP2_WRITING_STALE_WITH_POLLER:
+ case GRPC_CHTTP2_WRITE_STATE_WRITING_WITH_MORE_TO_COME:
GPR_TIMER_MARK("state=writing_stale_with_poller", 0);
- set_write_state(t, GRPC_CHTTP2_WRITE_SCHEDULED, "terminate_writing");
- GRPC_CHTTP2_REF_TRANSPORT(t, "writing");
- grpc_combiner_execute_finally(exec_ctx, t->executor.combiner,
- &t->initiate_writing, GRPC_ERROR_NONE,
- true);
- break;
- case GRPC_CHTTP2_WRITING_STALE_NO_POLLER:
- GPR_TIMER_MARK("state=writing_stale_no_poller", 0);
- set_write_state(t, GRPC_CHTTP2_WRITE_SCHEDULED, "terminate_writing");
+ t->write_state = GRPC_CHTTP2_WRITE_STATE_WRITING;
GRPC_CHTTP2_REF_TRANSPORT(t, "writing");
- grpc_combiner_execute_finally(exec_ctx, t->executor.combiner,
- &t->initiate_writing, GRPC_ERROR_NONE,
- false);
+ grpc_combiner_execute_finally(exec_ctx, t->combiner, &t->writing_action,
+ GRPC_ERROR_NONE, false);
break;
}
@@ -804,11 +655,11 @@ static void terminate_writing_with_lock(grpc_exec_ctx *exec_ctx, void *tp,
GPR_TIMER_END("terminate_writing_with_lock", 0);
}
-void grpc_chttp2_terminate_writing(grpc_exec_ctx *exec_ctx,
- void *transport_writing, grpc_error *error) {
+static void grpc_chttp2_terminate_writing(grpc_exec_ctx *exec_ctx, void *gt,
+ grpc_error *error) {
GPR_TIMER_BEGIN("grpc_chttp2_terminate_writing", 0);
- grpc_chttp2_transport *t = TRANSPORT_FROM_WRITING(transport_writing);
- grpc_combiner_execute(exec_ctx, t->executor.combiner, &t->terminate_writing,
+ grpc_chttp2_transport *t = gt;
+ grpc_combiner_execute(exec_ctx, t->combiner, &t->terminate_writing,
GRPC_ERROR_REF(error));
GPR_TIMER_END("grpc_chttp2_terminate_writing", 0);
}
@@ -817,22 +668,23 @@ static void writing_action(grpc_exec_ctx *exec_ctx, void *gt,
grpc_error *error) {
grpc_chttp2_transport *t = gt;
GPR_TIMER_BEGIN("writing_action", 0);
- grpc_chttp2_perform_writes(exec_ctx, &t->writing, t->ep);
+ grpc_endpoint_write(exec_ctx, t->ep, &t->outbuf, &t->writing_done_action);
GPR_TIMER_END("writing_action", 0);
}
-void grpc_chttp2_add_incoming_goaway(
- grpc_exec_ctx *exec_ctx, grpc_chttp2_transport_global *transport_global,
- uint32_t goaway_error, gpr_slice goaway_text) {
+void grpc_chttp2_add_incoming_goaway(grpc_exec_ctx *exec_ctx,
+ grpc_chttp2_transport *t,
+ uint32_t goaway_error,
+ gpr_slice goaway_text) {
char *msg = gpr_dump_slice(goaway_text, GPR_DUMP_HEX | GPR_DUMP_ASCII);
GRPC_CHTTP2_IF_TRACING(
gpr_log(GPR_DEBUG, "got goaway [%d]: %s", goaway_error, msg));
gpr_slice_unref(goaway_text);
- transport_global->seen_goaway = 1;
+ t->seen_goaway = 1;
/* lie: use transient failure from the transport to indicate goaway has been
* received */
connectivity_state_set(
- exec_ctx, transport_global, GRPC_CHANNEL_TRANSIENT_FAILURE,
+ exec_ctx, t, GRPC_CHANNEL_TRANSIENT_FAILURE,
grpc_error_set_str(
grpc_error_set_int(GRPC_ERROR_CREATE("GOAWAY received"),
GRPC_ERROR_INT_HTTP2_ERROR,
@@ -842,57 +694,47 @@ void grpc_chttp2_add_incoming_goaway(
gpr_free(msg);
}
-static void maybe_start_some_streams(
- grpc_exec_ctx *exec_ctx, grpc_chttp2_transport_global *transport_global) {
- grpc_chttp2_stream_global *stream_global;
+static void maybe_start_some_streams(grpc_exec_ctx *exec_ctx,
+ grpc_chttp2_transport *t) {
+ grpc_chttp2_stream *s;
uint32_t stream_incoming_window;
/* start streams where we have free grpc_chttp2_stream ids and free
* concurrency */
- while (transport_global->next_stream_id <= MAX_CLIENT_STREAM_ID &&
- grpc_chttp2_stream_map_size(
- &TRANSPORT_FROM_GLOBAL(transport_global)->stream_map) <
- transport_global
- ->settings[GRPC_PEER_SETTINGS]
- [GRPC_CHTTP2_SETTINGS_MAX_CONCURRENT_STREAMS] &&
- grpc_chttp2_list_pop_waiting_for_concurrency(transport_global,
- &stream_global)) {
+ while (t->next_stream_id <= MAX_CLIENT_STREAM_ID &&
+ grpc_chttp2_stream_map_size(&t->stream_map) <
+ t->settings[GRPC_PEER_SETTINGS]
+ [GRPC_CHTTP2_SETTINGS_MAX_CONCURRENT_STREAMS] &&
+ grpc_chttp2_list_pop_waiting_for_concurrency(t, &s)) {
/* safe since we can't (legally) be parsing this stream yet */
GRPC_CHTTP2_IF_TRACING(gpr_log(
GPR_DEBUG, "HTTP:%s: Allocating new grpc_chttp2_stream %p to id %d",
- transport_global->is_client ? "CLI" : "SVR", stream_global,
- transport_global->next_stream_id));
+ t->is_client ? "CLI" : "SVR", s, t->next_stream_id));
- GPR_ASSERT(stream_global->id == 0);
- stream_global->id = transport_global->next_stream_id;
- transport_global->next_stream_id += 2;
+ GPR_ASSERT(s->id == 0);
+ s->id = t->next_stream_id;
+ t->next_stream_id += 2;
- if (transport_global->next_stream_id >= MAX_CLIENT_STREAM_ID) {
- connectivity_state_set(
- exec_ctx, transport_global, GRPC_CHANNEL_TRANSIENT_FAILURE,
- GRPC_ERROR_CREATE("Stream IDs exhausted"), "no_more_stream_ids");
+ if (t->next_stream_id >= MAX_CLIENT_STREAM_ID) {
+ connectivity_state_set(exec_ctx, t, GRPC_CHANNEL_TRANSIENT_FAILURE,
+ GRPC_ERROR_CREATE("Stream IDs exhausted"),
+ "no_more_stream_ids");
}
- stream_global->outgoing_window =
- transport_global->settings[GRPC_PEER_SETTINGS]
- [GRPC_CHTTP2_SETTINGS_INITIAL_WINDOW_SIZE];
- stream_global->incoming_window = stream_incoming_window =
- transport_global->settings[GRPC_SENT_SETTINGS]
- [GRPC_CHTTP2_SETTINGS_INITIAL_WINDOW_SIZE];
- stream_global->max_recv_bytes =
- GPR_MAX(stream_incoming_window, stream_global->max_recv_bytes);
- grpc_chttp2_stream_map_add(
- &TRANSPORT_FROM_GLOBAL(transport_global)->stream_map, stream_global->id,
- STREAM_FROM_GLOBAL(stream_global));
- stream_global->in_stream_map = true;
- grpc_chttp2_become_writable(exec_ctx, transport_global, stream_global, true,
- "new_stream");
+ s->outgoing_window = t->settings[GRPC_PEER_SETTINGS]
+ [GRPC_CHTTP2_SETTINGS_INITIAL_WINDOW_SIZE];
+ s->incoming_window = stream_incoming_window =
+ t->settings[GRPC_SENT_SETTINGS]
+ [GRPC_CHTTP2_SETTINGS_INITIAL_WINDOW_SIZE];
+ s->max_recv_bytes = GPR_MAX(stream_incoming_window, s->max_recv_bytes);
+ grpc_chttp2_stream_map_add(&t->stream_map, s->id, s);
+ s->in_stream_map = true;
+ grpc_chttp2_become_writable(exec_ctx, t, s, true, "new_stream");
}
/* cancel out streams that will never be started */
- while (transport_global->next_stream_id >= MAX_CLIENT_STREAM_ID &&
- grpc_chttp2_list_pop_waiting_for_concurrency(transport_global,
- &stream_global)) {
+ while (t->next_stream_id >= MAX_CLIENT_STREAM_ID &&
+ grpc_chttp2_list_pop_waiting_for_concurrency(t, &s)) {
grpc_chttp2_cancel_stream(
- exec_ctx, transport_global, stream_global,
+ exec_ctx, t, s,
grpc_error_set_int(GRPC_ERROR_CREATE("Stream IDs exhausted"),
GRPC_ERROR_INT_GRPC_STATUS,
GRPC_STATUS_UNAVAILABLE));
@@ -907,10 +749,11 @@ static grpc_closure *add_closure_barrier(grpc_closure *closure) {
return closure;
}
-void grpc_chttp2_complete_closure_step(
- grpc_exec_ctx *exec_ctx, grpc_chttp2_transport_global *transport_global,
- grpc_chttp2_stream_global *stream_global, grpc_closure **pclosure,
- grpc_error *error) {
+void grpc_chttp2_complete_closure_step(grpc_exec_ctx *exec_ctx,
+ grpc_chttp2_transport *t,
+ grpc_chttp2_stream *s,
+ grpc_closure **pclosure,
+ grpc_error *error) {
grpc_closure *closure = *pclosure;
if (closure == NULL) {
GRPC_ERROR_UNREF(error);
@@ -922,33 +765,29 @@ void grpc_chttp2_complete_closure_step(
closure->error =
GRPC_ERROR_CREATE("Error in HTTP transport completing operation");
closure->error = grpc_error_set_str(
- closure->error, GRPC_ERROR_STR_TARGET_ADDRESS,
- TRANSPORT_FROM_GLOBAL(transport_global)->peer_string);
+ closure->error, GRPC_ERROR_STR_TARGET_ADDRESS, t->peer_string);
}
closure->error = grpc_error_add_child(closure->error, error);
}
if (closure->next_data.scratch < CLOSURE_BARRIER_FIRST_REF_BIT) {
if (closure->next_data.scratch & CLOSURE_BARRIER_STATS_BIT) {
- grpc_transport_move_stats(&stream_global->stats,
- stream_global->collecting_stats);
- stream_global->collecting_stats = NULL;
+ grpc_transport_move_stats(&s->stats, s->collecting_stats);
+ s->collecting_stats = NULL;
}
grpc_exec_ctx_sched(exec_ctx, closure, closure->error, NULL);
}
*pclosure = NULL;
}
-static int contains_non_ok_status(
- grpc_chttp2_transport_global *transport_global,
- grpc_metadata_batch *batch) {
+static bool contains_non_ok_status(grpc_metadata_batch *batch) {
grpc_linked_mdelem *l;
for (l = batch->list.head; l; l = l->next) {
if (l->md->key == GRPC_MDSTR_GRPC_STATUS &&
l->md != GRPC_MDELEM_GRPC_STATUS_0) {
- return 1;
+ return true;
}
}
- return 0;
+ return false;
}
static void do_nothing(grpc_exec_ctx *exec_ctx, void *arg, grpc_error *error) {}
@@ -960,8 +799,6 @@ static void perform_stream_op_locked(grpc_exec_ctx *exec_ctx, void *stream_op,
grpc_transport_stream_op *op = stream_op;
grpc_chttp2_transport *t = op->transport_private.args[0];
grpc_chttp2_stream *s = op->transport_private.args[1];
- grpc_chttp2_transport_global *transport_global = &t->global;
- grpc_chttp2_stream_global *stream_global = &s->global;
if (grpc_http_trace) {
char *str = grpc_transport_stream_op_string(op);
@@ -979,39 +816,35 @@ static void perform_stream_op_locked(grpc_exec_ctx *exec_ctx, void *stream_op,
on_complete->error = GRPC_ERROR_NONE;
if (op->collect_stats != NULL) {
- GPR_ASSERT(stream_global->collecting_stats == NULL);
- stream_global->collecting_stats = op->collect_stats;
+ GPR_ASSERT(s->collecting_stats == NULL);
+ s->collecting_stats = op->collect_stats;
on_complete->next_data.scratch |= CLOSURE_BARRIER_STATS_BIT;
}
if (op->cancel_error != GRPC_ERROR_NONE) {
- grpc_chttp2_cancel_stream(exec_ctx, transport_global, stream_global,
- GRPC_ERROR_REF(op->cancel_error));
+ grpc_chttp2_cancel_stream(exec_ctx, t, s, GRPC_ERROR_REF(op->cancel_error));
}
if (op->close_error != GRPC_ERROR_NONE) {
- close_from_api(exec_ctx, transport_global, stream_global,
- GRPC_ERROR_REF(op->close_error));
+ close_from_api(exec_ctx, t, s, GRPC_ERROR_REF(op->close_error));
}
if (op->send_initial_metadata != NULL) {
- GPR_ASSERT(stream_global->send_initial_metadata_finished == NULL);
- stream_global->send_initial_metadata_finished =
- add_closure_barrier(on_complete);
- stream_global->send_initial_metadata = op->send_initial_metadata;
+ GPR_ASSERT(s->send_initial_metadata_finished == NULL);
+ s->send_initial_metadata_finished = add_closure_barrier(on_complete);
+ s->send_initial_metadata = op->send_initial_metadata;
const size_t metadata_size =
grpc_metadata_batch_size(op->send_initial_metadata);
const size_t metadata_peer_limit =
- transport_global->settings[GRPC_PEER_SETTINGS]
- [GRPC_CHTTP2_SETTINGS_MAX_HEADER_LIST_SIZE];
- if (transport_global->is_client) {
- stream_global->deadline =
- gpr_time_min(stream_global->deadline,
- stream_global->send_initial_metadata->deadline);
+ t->settings[GRPC_PEER_SETTINGS]
+ [GRPC_CHTTP2_SETTINGS_MAX_HEADER_LIST_SIZE];
+ if (t->is_client) {
+ s->deadline =
+ gpr_time_min(s->deadline, s->send_initial_metadata->deadline);
}
if (metadata_size > metadata_peer_limit) {
grpc_chttp2_cancel_stream(
- exec_ctx, transport_global, stream_global,
+ exec_ctx, t, s,
grpc_error_set_int(
grpc_error_set_int(
grpc_error_set_int(
@@ -1021,27 +854,24 @@ static void perform_stream_op_locked(grpc_exec_ctx *exec_ctx, void *stream_op,
GRPC_ERROR_INT_LIMIT, (intptr_t)metadata_peer_limit),
GRPC_ERROR_INT_GRPC_STATUS, GRPC_STATUS_RESOURCE_EXHAUSTED));
} else {
- if (contains_non_ok_status(transport_global, op->send_initial_metadata)) {
- stream_global->seen_error = true;
- grpc_chttp2_list_add_check_read_ops(exec_ctx, transport_global,
- stream_global);
+ if (contains_non_ok_status(op->send_initial_metadata)) {
+ s->seen_error = true;
+ grpc_chttp2_list_add_check_read_ops(exec_ctx, t, s);
}
- if (!stream_global->write_closed) {
- if (transport_global->is_client) {
- GPR_ASSERT(stream_global->id == 0);
- grpc_chttp2_list_add_waiting_for_concurrency(transport_global,
- stream_global);
- maybe_start_some_streams(exec_ctx, transport_global);
+ if (!s->write_closed) {
+ if (t->is_client) {
+ GPR_ASSERT(s->id == 0);
+ grpc_chttp2_list_add_waiting_for_concurrency(t, s);
+ maybe_start_some_streams(exec_ctx, t);
} else {
- GPR_ASSERT(stream_global->id != 0);
- grpc_chttp2_become_writable(exec_ctx, transport_global, stream_global,
- true, "op.send_initial_metadata");
+ GPR_ASSERT(s->id != 0);
+ grpc_chttp2_become_writable(exec_ctx, t, s, true,
+ "op.send_initial_metadata");
}
} else {
- stream_global->send_trailing_metadata = NULL;
+ s->send_trailing_metadata = NULL;
grpc_chttp2_complete_closure_step(
- exec_ctx, transport_global, stream_global,
- &stream_global->send_initial_metadata_finished,
+ exec_ctx, t, s, &s->send_initial_metadata_finished,
GRPC_ERROR_CREATE(
"Attempt to send initial metadata after stream was closed"));
}
@@ -1049,36 +879,33 @@ static void perform_stream_op_locked(grpc_exec_ctx *exec_ctx, void *stream_op,
}
if (op->send_message != NULL) {
- GPR_ASSERT(stream_global->send_message_finished == NULL);
- GPR_ASSERT(stream_global->send_message == NULL);
- stream_global->send_message_finished = add_closure_barrier(on_complete);
- if (stream_global->write_closed) {
+ GPR_ASSERT(s->send_message_finished == NULL);
+ GPR_ASSERT(s->send_message == NULL);
+ s->send_message_finished = add_closure_barrier(on_complete);
+ if (s->write_closed) {
grpc_chttp2_complete_closure_step(
- exec_ctx, transport_global, stream_global,
- &stream_global->send_message_finished,
+ exec_ctx, t, s, &s->send_message_finished,
GRPC_ERROR_CREATE("Attempt to send message after stream was closed"));
} else {
- stream_global->send_message = op->send_message;
- if (stream_global->id != 0) {
- grpc_chttp2_become_writable(exec_ctx, transport_global, stream_global,
- true, "op.send_message");
+ s->send_message = op->send_message;
+ if (s->id != 0) {
+ grpc_chttp2_become_writable(exec_ctx, t, s, true, "op.send_message");
}
}
}
if (op->send_trailing_metadata != NULL) {
- GPR_ASSERT(stream_global->send_trailing_metadata_finished == NULL);
- stream_global->send_trailing_metadata_finished =
- add_closure_barrier(on_complete);
- stream_global->send_trailing_metadata = op->send_trailing_metadata;
+ GPR_ASSERT(s->send_trailing_metadata_finished == NULL);
+ s->send_trailing_metadata_finished = add_closure_barrier(on_complete);
+ s->send_trailing_metadata = op->send_trailing_metadata;
const size_t metadata_size =
grpc_metadata_batch_size(op->send_trailing_metadata);
const size_t metadata_peer_limit =
- transport_global->settings[GRPC_PEER_SETTINGS]
- [GRPC_CHTTP2_SETTINGS_MAX_HEADER_LIST_SIZE];
+ t->settings[GRPC_PEER_SETTINGS]
+ [GRPC_CHTTP2_SETTINGS_MAX_HEADER_LIST_SIZE];
if (metadata_size > metadata_peer_limit) {
grpc_chttp2_cancel_stream(
- exec_ctx, transport_global, stream_global,
+ exec_ctx, t, s,
grpc_error_set_int(
grpc_error_set_int(
grpc_error_set_int(
@@ -1088,69 +915,59 @@ static void perform_stream_op_locked(grpc_exec_ctx *exec_ctx, void *stream_op,
GRPC_ERROR_INT_LIMIT, (intptr_t)metadata_peer_limit),
GRPC_ERROR_INT_GRPC_STATUS, GRPC_STATUS_RESOURCE_EXHAUSTED));
} else {
- if (contains_non_ok_status(transport_global,
- op->send_trailing_metadata)) {
- stream_global->seen_error = true;
- grpc_chttp2_list_add_check_read_ops(exec_ctx, transport_global,
- stream_global);
+ if (contains_non_ok_status(op->send_trailing_metadata)) {
+ s->seen_error = true;
+ grpc_chttp2_list_add_check_read_ops(exec_ctx, t, s);
}
- if (stream_global->write_closed) {
- stream_global->send_trailing_metadata = NULL;
+ if (s->write_closed) {
+ s->send_trailing_metadata = NULL;
grpc_chttp2_complete_closure_step(
- exec_ctx, transport_global, stream_global,
- &stream_global->send_trailing_metadata_finished,
+ exec_ctx, t, s, &s->send_trailing_metadata_finished,
grpc_metadata_batch_is_empty(op->send_trailing_metadata)
? GRPC_ERROR_NONE
: GRPC_ERROR_CREATE("Attempt to send trailing metadata after "
"stream was closed"));
- } else if (stream_global->id != 0) {
+ } else if (s->id != 0) {
/* TODO(ctiller): check if there's flow control for any outstanding
bytes before going writable */
- grpc_chttp2_become_writable(exec_ctx, transport_global, stream_global,
- true, "op.send_trailing_metadata");
+ grpc_chttp2_become_writable(exec_ctx, t, s, true,
+ "op.send_trailing_metadata");
}
}
}
if (op->recv_initial_metadata != NULL) {
- GPR_ASSERT(stream_global->recv_initial_metadata_ready == NULL);
- stream_global->recv_initial_metadata_ready =
- op->recv_initial_metadata_ready;
- stream_global->recv_initial_metadata = op->recv_initial_metadata;
- grpc_chttp2_list_add_check_read_ops(exec_ctx, transport_global,
- stream_global);
+ GPR_ASSERT(s->recv_initial_metadata_ready == NULL);
+ s->recv_initial_metadata_ready = op->recv_initial_metadata_ready;
+ s->recv_initial_metadata = op->recv_initial_metadata;
+ grpc_chttp2_list_add_check_read_ops(exec_ctx, t, s);
}
if (op->recv_message != NULL) {
- GPR_ASSERT(stream_global->recv_message_ready == NULL);
- stream_global->recv_message_ready = op->recv_message_ready;
- stream_global->recv_message = op->recv_message;
- if (stream_global->id != 0 &&
- (stream_global->incoming_frames.head == NULL ||
- stream_global->incoming_frames.head->is_tail)) {
- incoming_byte_stream_update_flow_control(
- exec_ctx, transport_global, stream_global,
- transport_global->stream_lookahead, 0);
+ GPR_ASSERT(s->recv_message_ready == NULL);
+ s->recv_message_ready = op->recv_message_ready;
+ s->recv_message = op->recv_message;
+ if (s->id != 0 &&
+ (s->incoming_frames.head == NULL || s->incoming_frames.head->is_tail)) {
+ incoming_byte_stream_update_flow_control(exec_ctx, t, s,
+ t->stream_lookahead, 0);
}
- grpc_chttp2_list_add_check_read_ops(exec_ctx, transport_global,
- stream_global);
+ grpc_chttp2_list_add_check_read_ops(exec_ctx, t, s);
}
if (op->recv_trailing_metadata != NULL) {
- GPR_ASSERT(stream_global->recv_trailing_metadata_finished == NULL);
- stream_global->recv_trailing_metadata_finished =
- add_closure_barrier(on_complete);
- stream_global->recv_trailing_metadata = op->recv_trailing_metadata;
- stream_global->final_metadata_requested = true;
- grpc_chttp2_list_add_check_read_ops(exec_ctx, transport_global,
- stream_global);
+ GPR_ASSERT(s->recv_trailing_metadata_finished == NULL);
+ s->recv_trailing_metadata_finished = add_closure_barrier(on_complete);
+ s->recv_trailing_metadata = op->recv_trailing_metadata;
+ s->final_metadata_requested = true;
+ grpc_chttp2_list_add_check_read_ops(exec_ctx, t, s);
}
- grpc_chttp2_complete_closure_step(exec_ctx, transport_global, stream_global,
- &on_complete, GRPC_ERROR_NONE);
+ grpc_chttp2_complete_closure_step(exec_ctx, t, s, &on_complete,
+ GRPC_ERROR_NONE);
GPR_TIMER_END("perform_stream_op_locked", 0);
- GRPC_CHTTP2_STREAM_UNREF(exec_ctx, &s->global, "perform_stream_op");
+ GRPC_CHTTP2_STREAM_UNREF(exec_ctx, s, "perform_stream_op");
}
static void perform_stream_op(grpc_exec_ctx *exec_ctx, grpc_transport *gt,
@@ -1162,38 +979,36 @@ static void perform_stream_op(grpc_exec_ctx *exec_ctx, grpc_transport *gt,
op);
op->transport_private.args[0] = gt;
op->transport_private.args[1] = gs;
- GRPC_CHTTP2_STREAM_REF(&s->global, "perform_stream_op");
- grpc_combiner_execute(exec_ctx, t->executor.combiner,
- &op->transport_private.closure, GRPC_ERROR_NONE);
+ GRPC_CHTTP2_STREAM_REF(s, "perform_stream_op");
+ grpc_combiner_execute(exec_ctx, t->combiner, &op->transport_private.closure,
+ GRPC_ERROR_NONE);
GPR_TIMER_END("perform_stream_op", 0);
}
static void send_ping_locked(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t,
grpc_closure *on_recv) {
grpc_chttp2_outstanding_ping *p = gpr_malloc(sizeof(*p));
- p->next = &t->global.pings;
+ p->next = &t->pings;
p->prev = p->next->prev;
p->prev->next = p->next->prev = p;
- p->id[0] = (uint8_t)((t->global.ping_counter >> 56) & 0xff);
- p->id[1] = (uint8_t)((t->global.ping_counter >> 48) & 0xff);
- p->id[2] = (uint8_t)((t->global.ping_counter >> 40) & 0xff);
- p->id[3] = (uint8_t)((t->global.ping_counter >> 32) & 0xff);
- p->id[4] = (uint8_t)((t->global.ping_counter >> 24) & 0xff);
- p->id[5] = (uint8_t)((t->global.ping_counter >> 16) & 0xff);
- p->id[6] = (uint8_t)((t->global.ping_counter >> 8) & 0xff);
- p->id[7] = (uint8_t)(t->global.ping_counter & 0xff);
- t->global.ping_counter++;
+ p->id[0] = (uint8_t)((t->ping_counter >> 56) & 0xff);
+ p->id[1] = (uint8_t)((t->ping_counter >> 48) & 0xff);
+ p->id[2] = (uint8_t)((t->ping_counter >> 40) & 0xff);
+ p->id[3] = (uint8_t)((t->ping_counter >> 32) & 0xff);
+ p->id[4] = (uint8_t)((t->ping_counter >> 24) & 0xff);
+ p->id[5] = (uint8_t)((t->ping_counter >> 16) & 0xff);
+ p->id[6] = (uint8_t)((t->ping_counter >> 8) & 0xff);
+ p->id[7] = (uint8_t)(t->ping_counter & 0xff);
+ t->ping_counter++;
p->on_recv = on_recv;
- gpr_slice_buffer_add(&t->global.qbuf, grpc_chttp2_ping_create(0, p->id));
- grpc_chttp2_initiate_write(exec_ctx, &t->global, true, "send_ping");
+ gpr_slice_buffer_add(&t->qbuf, grpc_chttp2_ping_create(0, p->id));
+ grpc_chttp2_initiate_write(exec_ctx, t, true, "send_ping");
}
-void grpc_chttp2_ack_ping(grpc_exec_ctx *exec_ctx,
- grpc_chttp2_transport_global *transport_global,
+void grpc_chttp2_ack_ping(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t,
const uint8_t *opaque_8bytes) {
grpc_chttp2_outstanding_ping *ping;
- for (ping = transport_global->pings.next; ping != &transport_global->pings;
- ping = ping->next) {
+ for (ping = t->pings.next; ping != &t->pings; ping = ping->next) {
if (0 == memcmp(opaque_8bytes, ping->id, 8)) {
grpc_exec_ctx_sched(exec_ctx, ping->on_recv, GRPC_ERROR_NONE, NULL);
ping->next->prev = ping->prev;
@@ -1203,8 +1018,7 @@ void grpc_chttp2_ack_ping(grpc_exec_ctx *exec_ctx,
}
}
char *msg = gpr_dump((const char *)opaque_8bytes, 8, GPR_DUMP_HEX);
- char *from =
- grpc_endpoint_get_peer(TRANSPORT_FROM_GLOBAL(transport_global)->ep);
+ char *from = grpc_endpoint_get_peer(t->ep);
gpr_log(GPR_DEBUG, "Unknown ping response from %s: %s", from, msg);
gpr_free(from);
gpr_free(msg);
@@ -1224,15 +1038,15 @@ static void perform_transport_op_locked(grpc_exec_ctx *exec_ctx,
}
if (op->send_goaway) {
- t->global.sent_goaway = 1;
+ t->sent_goaway = 1;
grpc_chttp2_goaway_append(
- t->global.last_incoming_stream_id,
+ t->last_incoming_stream_id,
(uint32_t)grpc_chttp2_grpc_status_to_http2_error(op->goaway_status),
- gpr_slice_ref(*op->goaway_message), &t->global.qbuf);
+ gpr_slice_ref(*op->goaway_message), &t->qbuf);
close_transport = grpc_chttp2_stream_map_size(&t->stream_map) == 0
? GRPC_ERROR_CREATE("GOAWAY sent")
: GRPC_ERROR_NONE;
- grpc_chttp2_initiate_write(exec_ctx, &t->global, false, "goaway_sent");
+ grpc_chttp2_initiate_write(exec_ctx, t, false, "goaway_sent");
}
if (op->set_accept_stream) {
@@ -1269,85 +1083,77 @@ static void perform_transport_op(grpc_exec_ctx *exec_ctx, grpc_transport *gt,
grpc_closure_init(&op->transport_private.closure, perform_transport_op_locked,
op);
GRPC_CHTTP2_REF_TRANSPORT(t, "transport_op");
- grpc_combiner_execute(exec_ctx, t->executor.combiner,
- &op->transport_private.closure, GRPC_ERROR_NONE);
+ grpc_combiner_execute(exec_ctx, t->combiner, &op->transport_private.closure,
+ GRPC_ERROR_NONE);
}
/*******************************************************************************
* INPUT PROCESSING - GENERAL
*/
-static void check_read_ops(grpc_exec_ctx *exec_ctx,
- grpc_chttp2_transport_global *transport_global) {
+static void check_read_ops(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t) {
GPR_TIMER_BEGIN("check_read_ops", 0);
- grpc_chttp2_stream_global *stream_global;
+ grpc_chttp2_stream *s;
grpc_byte_stream *bs;
- while (
- grpc_chttp2_list_pop_check_read_ops(transport_global, &stream_global)) {
- if (stream_global->recv_initial_metadata_ready != NULL &&
- stream_global->published_metadata[0]) {
- if (stream_global->seen_error) {
+ while (grpc_chttp2_list_pop_check_read_ops(t, &s)) {
+ if (s->recv_initial_metadata_ready != NULL && s->published_metadata[0]) {
+ if (s->seen_error) {
while ((bs = grpc_chttp2_incoming_frame_queue_pop(
- &stream_global->incoming_frames)) != NULL) {
+ &s->incoming_frames)) != NULL) {
incoming_byte_stream_destroy_locked(exec_ctx, bs, GRPC_ERROR_NONE);
}
}
- grpc_chttp2_incoming_metadata_buffer_publish(
- &stream_global->metadata_buffer[0],
- stream_global->recv_initial_metadata);
- grpc_exec_ctx_sched(exec_ctx, stream_global->recv_initial_metadata_ready,
+ grpc_chttp2_incoming_metadata_buffer_publish(&s->metadata_buffer[0],
+ s->recv_initial_metadata);
+ grpc_exec_ctx_sched(exec_ctx, s->recv_initial_metadata_ready,
GRPC_ERROR_NONE, NULL);
- stream_global->recv_initial_metadata_ready = NULL;
+ s->recv_initial_metadata_ready = NULL;
}
- if (stream_global->recv_message_ready != NULL) {
- while (stream_global->final_metadata_requested &&
- stream_global->seen_error &&
- (bs = grpc_chttp2_incoming_frame_queue_pop(
- &stream_global->incoming_frames)) != NULL) {
+ if (s->recv_message_ready != NULL) {
+ while (s->final_metadata_requested && s->seen_error &&
+ (bs = grpc_chttp2_incoming_frame_queue_pop(&s->incoming_frames)) !=
+ NULL) {
incoming_byte_stream_destroy_locked(exec_ctx, bs, GRPC_ERROR_NONE);
}
- if (stream_global->incoming_frames.head != NULL) {
- *stream_global->recv_message = grpc_chttp2_incoming_frame_queue_pop(
- &stream_global->incoming_frames);
- GPR_ASSERT(*stream_global->recv_message != NULL);
- grpc_exec_ctx_sched(exec_ctx, stream_global->recv_message_ready,
- GRPC_ERROR_NONE, NULL);
- stream_global->recv_message_ready = NULL;
- } else if (stream_global->published_metadata[1]) {
- *stream_global->recv_message = NULL;
- grpc_exec_ctx_sched(exec_ctx, stream_global->recv_message_ready,
- GRPC_ERROR_NONE, NULL);
- stream_global->recv_message_ready = NULL;
+ if (s->incoming_frames.head != NULL) {
+ *s->recv_message =
+ grpc_chttp2_incoming_frame_queue_pop(&s->incoming_frames);
+ GPR_ASSERT(*s->recv_message != NULL);
+ grpc_exec_ctx_sched(exec_ctx, s->recv_message_ready, GRPC_ERROR_NONE,
+ NULL);
+ s->recv_message_ready = NULL;
+ } else if (s->published_metadata[1]) {
+ *s->recv_message = NULL;
+ grpc_exec_ctx_sched(exec_ctx, s->recv_message_ready, GRPC_ERROR_NONE,
+ NULL);
+ s->recv_message_ready = NULL;
}
}
- if (stream_global->recv_trailing_metadata_finished != NULL &&
- stream_global->read_closed && stream_global->write_closed) {
- if (stream_global->seen_error) {
+ if (s->recv_trailing_metadata_finished != NULL && s->read_closed &&
+ s->write_closed) {
+ if (s->seen_error) {
while ((bs = grpc_chttp2_incoming_frame_queue_pop(
- &stream_global->incoming_frames)) != NULL) {
+ &s->incoming_frames)) != NULL) {
incoming_byte_stream_destroy_locked(exec_ctx, bs, GRPC_ERROR_NONE);
}
}
- if (stream_global->all_incoming_byte_streams_finished) {
- grpc_chttp2_incoming_metadata_buffer_publish(
- &stream_global->metadata_buffer[1],
- stream_global->recv_trailing_metadata);
- grpc_chttp2_complete_closure_step(
- exec_ctx, transport_global, stream_global,
- &stream_global->recv_trailing_metadata_finished, GRPC_ERROR_NONE);
+ if (s->all_incoming_byte_streams_finished) {
+ grpc_chttp2_incoming_metadata_buffer_publish(&s->metadata_buffer[1],
+ s->recv_trailing_metadata);
+ grpc_chttp2_complete_closure_step(exec_ctx, t, s,
+ &s->recv_trailing_metadata_finished,
+ GRPC_ERROR_NONE);
}
}
}
GPR_TIMER_END("check_read_ops", 0);
}
-static void decrement_active_streams_locked(
- grpc_exec_ctx *exec_ctx, grpc_chttp2_transport_global *transport_global,
- grpc_chttp2_stream_global *stream_global) {
- if ((stream_global->all_incoming_byte_streams_finished =
- gpr_unref(&stream_global->active_streams))) {
- grpc_chttp2_list_add_check_read_ops(exec_ctx, transport_global,
- stream_global);
+static void decrement_active_streams_locked(grpc_exec_ctx *exec_ctx,
+ grpc_chttp2_transport *t,
+ grpc_chttp2_stream *s) {
+ if ((s->all_incoming_byte_streams_finished = gpr_unref(&s->active_streams))) {
+ grpc_chttp2_list_add_check_read_ops(exec_ctx, t, s);
}
}
@@ -1355,31 +1161,29 @@ static void remove_stream(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t,
uint32_t id, grpc_error *error) {
grpc_chttp2_stream *s = grpc_chttp2_stream_map_delete(&t->stream_map, id);
GPR_ASSERT(s);
- s->global.in_stream_map = false;
- if (t->global.incoming_stream == &s->global) {
- t->global.incoming_stream = NULL;
- grpc_chttp2_parsing_become_skip_parser(exec_ctx, &t->global);
+ s->in_stream_map = false;
+ if (t->incoming_stream == s) {
+ t->incoming_stream = NULL;
+ grpc_chttp2_parsing_become_skip_parser(exec_ctx, t);
}
- if (s->global.data_parser.parsing_frame != NULL) {
+ if (s->data_parser.parsing_frame != NULL) {
grpc_chttp2_incoming_byte_stream_finished(
- exec_ctx, s->global.data_parser.parsing_frame, GRPC_ERROR_REF(error),
- 0);
- s->global.data_parser.parsing_frame = NULL;
+ exec_ctx, s->data_parser.parsing_frame, GRPC_ERROR_REF(error), 0);
+ s->data_parser.parsing_frame = NULL;
}
- if (grpc_chttp2_stream_map_size(&t->stream_map) == 0 &&
- t->global.sent_goaway) {
+ if (grpc_chttp2_stream_map_size(&t->stream_map) == 0 && t->sent_goaway) {
close_transport_locked(
exec_ctx, t, GRPC_ERROR_CREATE_REFERENCING(
"Last stream closed after sending GOAWAY", &error, 1));
}
- if (grpc_chttp2_list_remove_writable_stream(&t->global, &s->global)) {
- GRPC_CHTTP2_STREAM_UNREF(exec_ctx, &s->global, "chttp2_writing");
+ if (grpc_chttp2_list_remove_writable_stream(t, s)) {
+ GRPC_CHTTP2_STREAM_UNREF(exec_ctx, s, "chttp2_writing");
}
GRPC_ERROR_UNREF(error);
- maybe_start_some_streams(exec_ctx, &t->global);
+ maybe_start_some_streams(exec_ctx, t);
}
static void status_codes_from_error(grpc_error *error, gpr_timespec deadline,
@@ -1410,22 +1214,19 @@ static void status_codes_from_error(grpc_error *error, gpr_timespec deadline,
}
void grpc_chttp2_cancel_stream(grpc_exec_ctx *exec_ctx,
- grpc_chttp2_transport_global *transport_global,
- grpc_chttp2_stream_global *stream_global,
+ grpc_chttp2_transport *t, grpc_chttp2_stream *s,
grpc_error *due_to_error) {
- if (!stream_global->read_closed || !stream_global->write_closed) {
+ if (!s->read_closed || !s->write_closed) {
grpc_status_code grpc_status;
grpc_chttp2_error_code http_error;
- status_codes_from_error(due_to_error, stream_global->deadline, &http_error,
+ status_codes_from_error(due_to_error, s->deadline, &http_error,
&grpc_status);
- if (stream_global->id != 0) {
+ if (s->id != 0) {
gpr_slice_buffer_add(
- &transport_global->qbuf,
- grpc_chttp2_rst_stream_create(stream_global->id, (uint32_t)http_error,
- &stream_global->stats.outgoing));
- grpc_chttp2_initiate_write(exec_ctx, transport_global, false,
- "rst_stream");
+ &t->qbuf, grpc_chttp2_rst_stream_create(s->id, (uint32_t)http_error,
+ &s->stats.outgoing));
+ grpc_chttp2_initiate_write(exec_ctx, t, false, "rst_stream");
}
const char *msg =
@@ -1436,27 +1237,22 @@ void grpc_chttp2_cancel_stream(grpc_exec_ctx *exec_ctx,
msg = grpc_error_string(due_to_error);
}
gpr_slice msg_slice = gpr_slice_from_copied_string(msg);
- grpc_chttp2_fake_status(exec_ctx, transport_global, stream_global,
- grpc_status, &msg_slice);
+ grpc_chttp2_fake_status(exec_ctx, t, s, grpc_status, &msg_slice);
if (free_msg) grpc_error_free_string(msg);
}
- if (due_to_error != GRPC_ERROR_NONE && !stream_global->seen_error) {
- stream_global->seen_error = true;
- grpc_chttp2_list_add_check_read_ops(exec_ctx, transport_global,
- stream_global);
+ if (due_to_error != GRPC_ERROR_NONE && !s->seen_error) {
+ s->seen_error = true;
+ grpc_chttp2_list_add_check_read_ops(exec_ctx, t, s);
}
- grpc_chttp2_mark_stream_closed(exec_ctx, transport_global, stream_global, 1,
- 1, due_to_error);
+ grpc_chttp2_mark_stream_closed(exec_ctx, t, s, 1, 1, due_to_error);
}
-void grpc_chttp2_fake_status(grpc_exec_ctx *exec_ctx,
- grpc_chttp2_transport_global *transport_global,
- grpc_chttp2_stream_global *stream_global,
- grpc_status_code status, gpr_slice *slice) {
+void grpc_chttp2_fake_status(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t,
+ grpc_chttp2_stream *s, grpc_status_code status,
+ gpr_slice *slice) {
if (status != GRPC_STATUS_OK) {
- stream_global->seen_error = true;
- grpc_chttp2_list_add_check_read_ops(exec_ctx, transport_global,
- stream_global);
+ s->seen_error = true;
+ grpc_chttp2_list_add_check_read_ops(exec_ctx, t, s);
}
/* stream_global->recv_trailing_metadata_finished gives us a
last chance replacement: we've received trailing metadata,
@@ -1464,24 +1260,22 @@ void grpc_chttp2_fake_status(grpc_exec_ctx *exec_ctx,
to the upper layers - drop what we've got, and then publish
what we want - which is safe because we haven't told anyone
about the metadata yet */
- if (!stream_global->published_metadata[1] ||
- stream_global->recv_trailing_metadata_finished != NULL) {
+ if (!s->published_metadata[1] || s->recv_trailing_metadata_finished != NULL) {
char status_string[GPR_LTOA_MIN_BUFSIZE];
gpr_ltoa(status, status_string);
grpc_chttp2_incoming_metadata_buffer_add(
- &stream_global->metadata_buffer[1],
+ &s->metadata_buffer[1],
grpc_mdelem_from_metadata_strings(
GRPC_MDSTR_GRPC_STATUS, grpc_mdstr_from_string(status_string)));
if (slice) {
grpc_chttp2_incoming_metadata_buffer_add(
- &stream_global->metadata_buffer[1],
+ &s->metadata_buffer[1],
grpc_mdelem_from_metadata_strings(
GRPC_MDSTR_GRPC_MESSAGE,
grpc_mdstr_from_slice(gpr_slice_ref(*slice))));
}
- stream_global->published_metadata[1] = true;
- grpc_chttp2_list_add_check_read_ops(exec_ctx, transport_global,
- stream_global);
+ s->published_metadata[1] = true;
+ grpc_chttp2_list_add_check_read_ops(exec_ctx, t, s);
}
if (slice) {
gpr_slice_unref(*slice);
@@ -1500,11 +1294,11 @@ static void add_error(grpc_error *error, grpc_error **refs, size_t *nrefs) {
}
static grpc_error *removal_error(grpc_error *extra_error,
- grpc_chttp2_stream_global *stream_global) {
+ grpc_chttp2_stream *s) {
grpc_error *refs[3];
size_t nrefs = 0;
- add_error(stream_global->read_closed_error, refs, &nrefs);
- add_error(stream_global->write_closed_error, refs, &nrefs);
+ add_error(s->read_closed_error, refs, &nrefs);
+ add_error(s->write_closed_error, refs, &nrefs);
add_error(extra_error, refs, &nrefs);
grpc_error *error = GRPC_ERROR_NONE;
if (nrefs > 0) {
@@ -1516,68 +1310,54 @@ static grpc_error *removal_error(grpc_error *extra_error,
}
static void fail_pending_writes(grpc_exec_ctx *exec_ctx,
- grpc_chttp2_transport_global *transport_global,
- grpc_chttp2_stream_global *stream_global,
+ grpc_chttp2_transport *t, grpc_chttp2_stream *s,
grpc_error *error) {
- error = removal_error(error, stream_global);
- stream_global->send_message = NULL;
- grpc_chttp2_complete_closure_step(
- exec_ctx, transport_global, stream_global,
- &stream_global->send_initial_metadata_finished, GRPC_ERROR_REF(error));
- grpc_chttp2_complete_closure_step(
- exec_ctx, transport_global, stream_global,
- &stream_global->send_trailing_metadata_finished, GRPC_ERROR_REF(error));
- grpc_chttp2_complete_closure_step(exec_ctx, transport_global, stream_global,
- &stream_global->send_message_finished,
+ error = removal_error(error, s);
+ s->send_message = NULL;
+ grpc_chttp2_complete_closure_step(exec_ctx, t, s,
+ &s->send_initial_metadata_finished,
+ GRPC_ERROR_REF(error));
+ grpc_chttp2_complete_closure_step(exec_ctx, t, s,
+ &s->send_trailing_metadata_finished,
+ GRPC_ERROR_REF(error));
+ grpc_chttp2_complete_closure_step(exec_ctx, t, s, &s->send_message_finished,
error);
}
-void grpc_chttp2_mark_stream_closed(
- grpc_exec_ctx *exec_ctx, grpc_chttp2_transport_global *transport_global,
- grpc_chttp2_stream_global *stream_global, int close_reads, int close_writes,
- grpc_error *error) {
- if (stream_global->read_closed && stream_global->write_closed) {
+void grpc_chttp2_mark_stream_closed(grpc_exec_ctx *exec_ctx,
+ grpc_chttp2_transport *t,
+ grpc_chttp2_stream *s, int close_reads,
+ int close_writes, grpc_error *error) {
+ if (s->read_closed && s->write_closed) {
/* already closed */
GRPC_ERROR_UNREF(error);
return;
}
- grpc_chttp2_list_add_check_read_ops(exec_ctx, transport_global,
- stream_global);
- if (close_reads && !stream_global->read_closed) {
- stream_global->read_closed_error = GRPC_ERROR_REF(error);
- stream_global->read_closed = true;
- stream_global->published_metadata[0] = true;
- stream_global->published_metadata[1] = true;
- decrement_active_streams_locked(exec_ctx, transport_global, stream_global);
- }
- if (close_writes && !stream_global->write_closed) {
- stream_global->write_closed_error = GRPC_ERROR_REF(error);
- stream_global->write_closed = true;
- if (TRANSPORT_FROM_GLOBAL(transport_global)->executor.write_state !=
- GRPC_CHTTP2_WRITING_INACTIVE) {
- GRPC_CHTTP2_STREAM_REF(stream_global, "finish_writes");
- grpc_chttp2_list_add_closed_waiting_for_writing(transport_global,
- stream_global);
- } else {
- fail_pending_writes(exec_ctx, transport_global, stream_global,
- GRPC_ERROR_REF(error));
+ grpc_chttp2_list_add_check_read_ops(exec_ctx, t, s);
+ if (close_reads && !s->read_closed) {
+ s->read_closed_error = GRPC_ERROR_REF(error);
+ s->read_closed = true;
+ s->published_metadata[0] = true;
+ s->published_metadata[1] = true;
+ decrement_active_streams_locked(exec_ctx, t, s);
+ }
+ if (close_writes && !s->write_closed) {
+ s->write_closed_error = GRPC_ERROR_REF(error);
+ s->write_closed = true;
+ fail_pending_writes(exec_ctx, t, s, GRPC_ERROR_REF(error));
+ }
+ if (s->read_closed && s->write_closed) {
+ if (s->id != 0) {
+ remove_stream(exec_ctx, t, s->id,
+ removal_error(GRPC_ERROR_REF(error), s));
}
- }
- if (stream_global->read_closed && stream_global->write_closed) {
- if (stream_global->id != 0) {
- remove_stream(exec_ctx, TRANSPORT_FROM_GLOBAL(transport_global),
- stream_global->id,
- removal_error(GRPC_ERROR_REF(error), stream_global));
- }
- GRPC_CHTTP2_STREAM_UNREF(exec_ctx, stream_global, "chttp2");
+ GRPC_CHTTP2_STREAM_UNREF(exec_ctx, s, "chttp2");
}
GRPC_ERROR_UNREF(error);
}
-static void close_from_api(grpc_exec_ctx *exec_ctx,
- grpc_chttp2_transport_global *transport_global,
- grpc_chttp2_stream_global *stream_global,
- grpc_error *error) {
+static void close_from_api(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t,
+ grpc_chttp2_stream *s, grpc_error *error) {
gpr_slice hdr;
gpr_slice status_hdr;
gpr_slice message_pfx;
@@ -1585,12 +1365,11 @@ static void close_from_api(grpc_exec_ctx *exec_ctx,
uint32_t len = 0;
grpc_status_code grpc_status;
grpc_chttp2_error_code http_error;
- status_codes_from_error(error, stream_global->deadline, &http_error,
- &grpc_status);
+ status_codes_from_error(error, s->deadline, &http_error, &grpc_status);
GPR_ASSERT(grpc_status >= 0 && (int)grpc_status < 100);
- if (stream_global->id != 0 && !transport_global->is_client) {
+ if (s->id != 0 && !t->is_client) {
/* Hand roll a header block.
This is unnecessarily ugly - at some point we should find a more elegant
solution.
@@ -1659,23 +1438,22 @@ static void close_from_api(grpc_exec_ctx *exec_ctx,
*p++ = (uint8_t)(len);
*p++ = GRPC_CHTTP2_FRAME_HEADER;
*p++ = GRPC_CHTTP2_DATA_FLAG_END_STREAM | GRPC_CHTTP2_DATA_FLAG_END_HEADERS;
- *p++ = (uint8_t)(stream_global->id >> 24);
- *p++ = (uint8_t)(stream_global->id >> 16);
- *p++ = (uint8_t)(stream_global->id >> 8);
- *p++ = (uint8_t)(stream_global->id);
+ *p++ = (uint8_t)(s->id >> 24);
+ *p++ = (uint8_t)(s->id >> 16);
+ *p++ = (uint8_t)(s->id >> 8);
+ *p++ = (uint8_t)(s->id);
GPR_ASSERT(p == GPR_SLICE_END_PTR(hdr));
- gpr_slice_buffer_add(&transport_global->qbuf, hdr);
- gpr_slice_buffer_add(&transport_global->qbuf, status_hdr);
+ gpr_slice_buffer_add(&t->qbuf, hdr);
+ gpr_slice_buffer_add(&t->qbuf, status_hdr);
if (optional_message) {
- gpr_slice_buffer_add(&transport_global->qbuf, message_pfx);
- gpr_slice_buffer_add(&transport_global->qbuf,
+ gpr_slice_buffer_add(&t->qbuf, message_pfx);
+ gpr_slice_buffer_add(&t->qbuf,
gpr_slice_from_copied_string(optional_message));
}
gpr_slice_buffer_add(
- &transport_global->qbuf,
- grpc_chttp2_rst_stream_create(stream_global->id, GRPC_CHTTP2_NO_ERROR,
- &stream_global->stats.outgoing));
+ &t->qbuf, grpc_chttp2_rst_stream_create(s->id, GRPC_CHTTP2_NO_ERROR,
+ &s->stats.outgoing));
}
const char *msg = grpc_error_get_str(error, GRPC_ERROR_STR_GRPC_MESSAGE);
@@ -1685,14 +1463,11 @@ static void close_from_api(grpc_exec_ctx *exec_ctx,
msg = grpc_error_string(error);
}
gpr_slice msg_slice = gpr_slice_from_copied_string(msg);
- grpc_chttp2_fake_status(exec_ctx, transport_global, stream_global,
- grpc_status, &msg_slice);
+ grpc_chttp2_fake_status(exec_ctx, t, s, grpc_status, &msg_slice);
if (free_msg) grpc_error_free_string(msg);
- grpc_chttp2_mark_stream_closed(exec_ctx, transport_global, stream_global, 1,
- 1, error);
- grpc_chttp2_initiate_write(exec_ctx, transport_global, false,
- "close_from_api");
+ grpc_chttp2_mark_stream_closed(exec_ctx, t, s, 1, 1, error);
+ grpc_chttp2_initiate_write(exec_ctx, t, false, "close_from_api");
}
typedef struct {
@@ -1704,7 +1479,7 @@ typedef struct {
static void cancel_stream_cb(void *user_data, uint32_t key, void *stream) {
cancel_stream_cb_args *args = user_data;
grpc_chttp2_stream *s = stream;
- grpc_chttp2_cancel_stream(args->exec_ctx, &args->t->global, &s->global,
+ grpc_chttp2_cancel_stream(args->exec_ctx, args->t, s,
GRPC_ERROR_REF(args->error));
}
@@ -1735,20 +1510,18 @@ static void update_global_window(void *args, uint32_t id, void *stream) {
update_global_window_args *a = args;
grpc_chttp2_transport *t = a->t;
grpc_chttp2_stream *s = stream;
- grpc_chttp2_transport_global *transport_global = &t->global;
- grpc_chttp2_stream_global *stream_global = &s->global;
int was_zero;
int is_zero;
- int64_t initial_window_update = t->global.initial_window_update;
+ int64_t initial_window_update = t->initial_window_update;
- was_zero = stream_global->outgoing_window <= 0;
- GRPC_CHTTP2_FLOW_CREDIT_STREAM("settings", transport_global, stream_global,
- outgoing_window, initial_window_update);
- is_zero = stream_global->outgoing_window <= 0;
+ was_zero = s->outgoing_window <= 0;
+ GRPC_CHTTP2_FLOW_CREDIT_STREAM("settings", t, s, outgoing_window,
+ initial_window_update);
+ is_zero = s->outgoing_window <= 0;
if (was_zero && !is_zero) {
- grpc_chttp2_become_writable(a->exec_ctx, transport_global, stream_global,
- true, "update_global_window");
+ grpc_chttp2_become_writable(a->exec_ctx, t, s, true,
+ "update_global_window");
}
}
@@ -1764,8 +1537,8 @@ static void reading_action(grpc_exec_ctx *exec_ctx, void *tp,
post_reading_action_locked */
GPR_TIMER_BEGIN("reading_action", 0);
grpc_chttp2_transport *t = tp;
- grpc_combiner_execute(exec_ctx, t->executor.combiner,
- &t->reading_action_locked, GRPC_ERROR_REF(error));
+ grpc_combiner_execute(exec_ctx, t->combiner, &t->reading_action_locked,
+ GRPC_ERROR_REF(error));
GPR_TIMER_END("reading_action", 0);
}
@@ -1801,7 +1574,6 @@ static void reading_action_locked(grpc_exec_ctx *exec_ctx, void *tp,
GPR_TIMER_BEGIN("reading_action_locked", 0);
grpc_chttp2_transport *t = tp;
- grpc_chttp2_transport_global *transport_global = &t->global;
GRPC_ERROR_REF(error);
@@ -1811,8 +1583,8 @@ static void reading_action_locked(grpc_exec_ctx *exec_ctx, void *tp,
grpc_error *errors[3] = {GRPC_ERROR_REF(error), GRPC_ERROR_NONE,
GRPC_ERROR_NONE};
for (; i < t->read_buffer.count && errors[1] == GRPC_ERROR_NONE; i++) {
- errors[1] = grpc_chttp2_perform_read(exec_ctx, &t->global,
- t->read_buffer.slices[i]);
+ errors[1] =
+ grpc_chttp2_perform_read(exec_ctx, t, t->read_buffer.slices[i]);
};
if (errors[1] != GRPC_ERROR_NONE) {
errors[2] = try_http_parsing(exec_ctx, t);
@@ -1826,23 +1598,20 @@ static void reading_action_locked(grpc_exec_ctx *exec_ctx, void *tp,
GPR_TIMER_END("reading_action.parse", 0);
GPR_TIMER_BEGIN("post_parse_locked", 0);
- if (transport_global->initial_window_update != 0) {
+ if (t->initial_window_update != 0) {
update_global_window_args args = {t, exec_ctx};
grpc_chttp2_stream_map_for_each(&t->stream_map, update_global_window,
&args);
- transport_global->initial_window_update = 0;
+ t->initial_window_update = 0;
}
/* handle higher level things */
- if (transport_global->incoming_window <
- transport_global->connection_window_target * 3 / 4) {
- int64_t announce_bytes = transport_global->connection_window_target -
- transport_global->incoming_window;
- GRPC_CHTTP2_FLOW_CREDIT_TRANSPORT(
- "parsed", transport_global, announce_incoming_window, announce_bytes);
- GRPC_CHTTP2_FLOW_CREDIT_TRANSPORT("parsed", transport_global,
- incoming_window, announce_bytes);
- grpc_chttp2_initiate_write(exec_ctx, transport_global, false,
- "global incoming window");
+ if (t->incoming_window < t->connection_window_target * 3 / 4) {
+ int64_t announce_bytes = t->connection_window_target - t->incoming_window;
+ GRPC_CHTTP2_FLOW_CREDIT_TRANSPORT("parsed", t, announce_incoming_window,
+ announce_bytes);
+ GRPC_CHTTP2_FLOW_CREDIT_TRANSPORT("parsed", t, incoming_window,
+ announce_bytes);
+ grpc_chttp2_initiate_write(exec_ctx, t, false, "global incoming window");
}
GPR_TIMER_END("post_parse_locked", 0);
@@ -1856,10 +1625,6 @@ static void reading_action_locked(grpc_exec_ctx *exec_ctx, void *tp,
if (error != GRPC_ERROR_NONE) {
drop_connection(exec_ctx, t, GRPC_ERROR_REF(error));
t->endpoint_reading = 0;
- if (grpc_http_write_state_trace) {
- gpr_log(GPR_DEBUG, "R:%p -> 0 ws=%s", t,
- write_state_name(t->executor.write_state));
- }
} else if (!t->closed) {
keep_reading = true;
GRPC_CHTTP2_REF_TRANSPORT(t, "keep_reading");
@@ -1886,15 +1651,14 @@ static void reading_action_locked(grpc_exec_ctx *exec_ctx, void *tp,
* CALLBACK LOOP
*/
-static void connectivity_state_set(
- grpc_exec_ctx *exec_ctx, grpc_chttp2_transport_global *transport_global,
- grpc_connectivity_state state, grpc_error *error, const char *reason) {
+static void connectivity_state_set(grpc_exec_ctx *exec_ctx,
+ grpc_chttp2_transport *t,
+ grpc_connectivity_state state,
+ grpc_error *error, const char *reason) {
GRPC_CHTTP2_IF_TRACING(
gpr_log(GPR_DEBUG, "set connectivity_state=%d", state));
- grpc_connectivity_state_set(
- exec_ctx,
- &TRANSPORT_FROM_GLOBAL(transport_global)->channel_callback.state_tracker,
- state, error, reason);
+ grpc_connectivity_state_set(exec_ctx, &t->channel_callback.state_tracker,
+ state, error, reason);
}
/*******************************************************************************
@@ -1927,15 +1691,16 @@ 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_global *transport_global,
- grpc_chttp2_stream_global *stream_global, size_t max_size_hint,
- size_t have_already) {
+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;
/* clamp max recv hint to an allowable size */
- if (max_size_hint >= UINT32_MAX - transport_global->stream_lookahead) {
- max_recv_bytes = UINT32_MAX - transport_global->stream_lookahead;
+ if (max_size_hint >= UINT32_MAX - t->stream_lookahead) {
+ max_recv_bytes = UINT32_MAX - t->stream_lookahead;
} else {
max_recv_bytes = (uint32_t)max_size_hint;
}
@@ -1948,20 +1713,18 @@ static void incoming_byte_stream_update_flow_control(
}
/* add some small lookahead to keep pipelines flowing */
- GPR_ASSERT(max_recv_bytes <= UINT32_MAX - transport_global->stream_lookahead);
- max_recv_bytes += transport_global->stream_lookahead;
- if (stream_global->max_recv_bytes < max_recv_bytes) {
- uint32_t add_max_recv_bytes =
- max_recv_bytes - stream_global->max_recv_bytes;
- GRPC_CHTTP2_FLOW_CREDIT_STREAM("op", transport_global, stream_global,
- max_recv_bytes, add_max_recv_bytes);
- GRPC_CHTTP2_FLOW_CREDIT_STREAM("op", transport_global, stream_global,
- incoming_window, add_max_recv_bytes);
- GRPC_CHTTP2_FLOW_CREDIT_STREAM("op", transport_global, stream_global,
+ GPR_ASSERT(max_recv_bytes <= UINT32_MAX - t->stream_lookahead);
+ max_recv_bytes += t->stream_lookahead;
+ if (s->max_recv_bytes < max_recv_bytes) {
+ uint32_t add_max_recv_bytes = max_recv_bytes - s->max_recv_bytes;
+ GRPC_CHTTP2_FLOW_CREDIT_STREAM("op", t, s, max_recv_bytes,
+ add_max_recv_bytes);
+ GRPC_CHTTP2_FLOW_CREDIT_STREAM("op", t, s, incoming_window,
+ add_max_recv_bytes);
+ GRPC_CHTTP2_FLOW_CREDIT_STREAM("op", t, s,
unannounced_incoming_window_for_writing,
add_max_recv_bytes);
- grpc_chttp2_become_writable(exec_ctx, transport_global, stream_global,
- false, "read_incoming_stream");
+ grpc_chttp2_become_writable(exec_ctx, t, s, false, "read_incoming_stream");
}
}
@@ -1969,16 +1732,15 @@ static void incoming_byte_stream_next_locked(grpc_exec_ctx *exec_ctx,
void *argp,
grpc_error *error_ignored) {
grpc_chttp2_incoming_byte_stream *bs = argp;
- grpc_chttp2_transport_global *transport_global = &bs->transport->global;
- grpc_chttp2_stream_global *stream_global = &bs->stream->global;
+ grpc_chttp2_transport *t = bs->transport;
+ grpc_chttp2_stream *s = bs->stream;
if (bs->is_tail) {
gpr_mu_lock(&bs->slice_mu);
size_t cur_length = bs->slices.length;
gpr_mu_unlock(&bs->slice_mu);
incoming_byte_stream_update_flow_control(
- exec_ctx, transport_global, stream_global,
- bs->next_action.max_size_hint, cur_length);
+ exec_ctx, t, s, bs->next_action.max_size_hint, cur_length);
}
gpr_mu_lock(&bs->slice_mu);
if (bs->slices.count > 0) {
@@ -2009,7 +1771,7 @@ static int incoming_byte_stream_next(grpc_exec_ctx *exec_ctx,
bs->next_action.on_complete = on_complete;
grpc_closure_init(&bs->next_action.closure, incoming_byte_stream_next_locked,
bs);
- grpc_combiner_execute(exec_ctx, bs->transport->executor.combiner,
+ grpc_combiner_execute(exec_ctx, bs->transport->combiner,
&bs->next_action.closure, GRPC_ERROR_NONE);
GPR_TIMER_END("incoming_byte_stream_next", 0);
return 0;
@@ -2023,8 +1785,7 @@ static void incoming_byte_stream_destroy_locked(grpc_exec_ctx *exec_ctx,
grpc_error *error_ignored) {
grpc_chttp2_incoming_byte_stream *bs = byte_stream;
GPR_ASSERT(bs->base.destroy == incoming_byte_stream_destroy);
- decrement_active_streams_locked(exec_ctx, &bs->transport->global,
- &bs->stream->global);
+ decrement_active_streams_locked(exec_ctx, bs->transport, bs->stream);
incoming_byte_stream_unref(exec_ctx, bs);
}
@@ -2035,8 +1796,8 @@ static void incoming_byte_stream_destroy(grpc_exec_ctx *exec_ctx,
(grpc_chttp2_incoming_byte_stream *)byte_stream;
grpc_closure_init(&bs->destroy_action, incoming_byte_stream_destroy_locked,
bs);
- grpc_combiner_execute(exec_ctx, bs->transport->executor.combiner,
- &bs->destroy_action, GRPC_ERROR_NONE);
+ grpc_combiner_execute(exec_ctx, bs->transport->combiner, &bs->destroy_action,
+ GRPC_ERROR_NONE);
GPR_TIMER_END("incoming_byte_stream_destroy", 0);
}
@@ -2078,7 +1839,7 @@ void grpc_chttp2_incoming_byte_stream_finished(
if (from_parsing_thread) {
grpc_closure_init(&bs->finished_action,
incoming_byte_stream_finished_locked, bs);
- grpc_combiner_execute(exec_ctx, bs->transport->executor.combiner,
+ grpc_combiner_execute(exec_ctx, bs->transport->combiner,
&bs->finished_action, GRPC_ERROR_REF(error));
} else {
incoming_byte_stream_finished_locked(exec_ctx, bs, error);
@@ -2087,9 +1848,8 @@ void grpc_chttp2_incoming_byte_stream_finished(
}
grpc_chttp2_incoming_byte_stream *grpc_chttp2_incoming_byte_stream_create(
- grpc_exec_ctx *exec_ctx, grpc_chttp2_transport_global *transport_global,
- grpc_chttp2_stream_global *stream_global, uint32_t frame_size,
- uint32_t flags) {
+ grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t, grpc_chttp2_stream *s,
+ uint32_t frame_size, uint32_t flags) {
grpc_chttp2_incoming_byte_stream *incoming_byte_stream =
gpr_malloc(sizeof(*incoming_byte_stream));
incoming_byte_stream->base.length = frame_size;
@@ -2099,14 +1859,14 @@ grpc_chttp2_incoming_byte_stream *grpc_chttp2_incoming_byte_stream_create(
gpr_mu_init(&incoming_byte_stream->slice_mu);
gpr_ref_init(&incoming_byte_stream->refs, 2);
incoming_byte_stream->next_message = NULL;
- incoming_byte_stream->transport = TRANSPORT_FROM_GLOBAL(transport_global);
- incoming_byte_stream->stream = STREAM_FROM_GLOBAL(stream_global);
- gpr_ref(&incoming_byte_stream->stream->global.active_streams);
+ incoming_byte_stream->transport = t;
+ incoming_byte_stream->stream = s;
+ gpr_ref(&incoming_byte_stream->stream->active_streams);
gpr_slice_buffer_init(&incoming_byte_stream->slices);
incoming_byte_stream->on_next = NULL;
incoming_byte_stream->is_tail = 1;
incoming_byte_stream->error = GRPC_ERROR_NONE;
- grpc_chttp2_incoming_frame_queue *q = &stream_global->incoming_frames;
+ grpc_chttp2_incoming_frame_queue *q = &s->incoming_frames;
if (q->head == NULL) {
q->head = incoming_byte_stream;
} else {
diff --git a/src/core/ext/transport/chttp2/transport/frame.h b/src/core/ext/transport/chttp2/transport/frame.h
index 48e3ac6482..0f1e128b3d 100644
--- a/src/core/ext/transport/chttp2/transport/frame.h
+++ b/src/core/ext/transport/chttp2/transport/frame.h
@@ -40,8 +40,8 @@
#include "src/core/lib/iomgr/error.h"
/* defined in internal.h */
-typedef struct grpc_chttp2_stream_global grpc_chttp2_stream_global;
-typedef struct grpc_chttp2_transport_global grpc_chttp2_transport_global;
+typedef struct grpc_chttp2_stream grpc_chttp2_stream;
+typedef struct grpc_chttp2_transport grpc_chttp2_transport;
#define GRPC_CHTTP2_FRAME_DATA 0
#define GRPC_CHTTP2_FRAME_HEADER 1
diff --git a/src/core/ext/transport/chttp2/transport/frame_data.h b/src/core/ext/transport/chttp2/transport/frame_data.h
index c8ff7fe6bc..eb2d97d898 100644
--- a/src/core/ext/transport/chttp2/transport/frame_data.h
+++ b/src/core/ext/transport/chttp2/transport/frame_data.h
@@ -91,10 +91,10 @@ grpc_error *grpc_chttp2_data_parser_begin_frame(grpc_chttp2_data_parser *parser,
/* handle a slice of a data frame - is_last indicates the last slice of a
frame */
-grpc_error *grpc_chttp2_data_parser_parse(
- grpc_exec_ctx *exec_ctx, void *parser,
- grpc_chttp2_transport_global *transport_global,
- grpc_chttp2_stream_global *stream_global, gpr_slice slice, int is_last);
+grpc_error *grpc_chttp2_data_parser_parse(grpc_exec_ctx *exec_ctx, void *parser,
+ grpc_chttp2_transport *t,
+ grpc_chttp2_stream *s,
+ gpr_slice slice, int is_last);
void grpc_chttp2_encode_data(uint32_t id, gpr_slice_buffer *inbuf,
uint32_t write_bytes, int is_eof,
diff --git a/src/core/ext/transport/chttp2/transport/frame_goaway.h b/src/core/ext/transport/chttp2/transport/frame_goaway.h
index f821229931..355104a5a7 100644
--- a/src/core/ext/transport/chttp2/transport/frame_goaway.h
+++ b/src/core/ext/transport/chttp2/transport/frame_goaway.h
@@ -65,10 +65,11 @@ void grpc_chttp2_goaway_parser_init(grpc_chttp2_goaway_parser *p);
void grpc_chttp2_goaway_parser_destroy(grpc_chttp2_goaway_parser *p);
grpc_error *grpc_chttp2_goaway_parser_begin_frame(
grpc_chttp2_goaway_parser *parser, uint32_t length, uint8_t flags);
-grpc_error *grpc_chttp2_goaway_parser_parse(
- grpc_exec_ctx *exec_ctx, void *parser,
- grpc_chttp2_transport_global *transport_global,
- grpc_chttp2_stream_global *stream_global, gpr_slice slice, int is_last);
+grpc_error *grpc_chttp2_goaway_parser_parse(grpc_exec_ctx *exec_ctx,
+ void *parser,
+ grpc_chttp2_transport *t,
+ grpc_chttp2_stream *s,
+ gpr_slice slice, int is_last);
void grpc_chttp2_goaway_append(uint32_t last_stream_id, uint32_t error_code,
gpr_slice debug_data,
diff --git a/src/core/ext/transport/chttp2/transport/frame_ping.h b/src/core/ext/transport/chttp2/transport/frame_ping.h
index c2c4fb2ee5..2071f647fb 100644
--- a/src/core/ext/transport/chttp2/transport/frame_ping.h
+++ b/src/core/ext/transport/chttp2/transport/frame_ping.h
@@ -48,9 +48,9 @@ gpr_slice grpc_chttp2_ping_create(uint8_t ack, uint8_t *opaque_8bytes);
grpc_error *grpc_chttp2_ping_parser_begin_frame(grpc_chttp2_ping_parser *parser,
uint32_t length, uint8_t flags);
-grpc_error *grpc_chttp2_ping_parser_parse(
- grpc_exec_ctx *exec_ctx, void *parser,
- grpc_chttp2_transport_global *transport_global,
- grpc_chttp2_stream_global *stream_global, gpr_slice slice, int is_last);
+grpc_error *grpc_chttp2_ping_parser_parse(grpc_exec_ctx *exec_ctx, void *parser,
+ grpc_chttp2_transport *t,
+ grpc_chttp2_stream *s,
+ gpr_slice slice, int is_last);
#endif /* GRPC_CORE_EXT_TRANSPORT_CHTTP2_TRANSPORT_FRAME_PING_H */
diff --git a/src/core/ext/transport/chttp2/transport/frame_rst_stream.h b/src/core/ext/transport/chttp2/transport/frame_rst_stream.h
index 64a6a27341..5a1f578a29 100644
--- a/src/core/ext/transport/chttp2/transport/frame_rst_stream.h
+++ b/src/core/ext/transport/chttp2/transport/frame_rst_stream.h
@@ -49,9 +49,10 @@ gpr_slice grpc_chttp2_rst_stream_create(uint32_t stream_id, uint32_t code,
grpc_error *grpc_chttp2_rst_stream_parser_begin_frame(
grpc_chttp2_rst_stream_parser *parser, uint32_t length, uint8_t flags);
-grpc_error *grpc_chttp2_rst_stream_parser_parse(
- grpc_exec_ctx *exec_ctx, void *parser,
- grpc_chttp2_transport_global *transport_global,
- grpc_chttp2_stream_global *stream_global, gpr_slice slice, int is_last);
+grpc_error *grpc_chttp2_rst_stream_parser_parse(grpc_exec_ctx *exec_ctx,
+ void *parser,
+ grpc_chttp2_transport *t,
+ grpc_chttp2_stream *s,
+ gpr_slice slice, int is_last);
#endif /* GRPC_CORE_EXT_TRANSPORT_CHTTP2_TRANSPORT_FRAME_RST_STREAM_H */
diff --git a/src/core/ext/transport/chttp2/transport/frame_settings.h b/src/core/ext/transport/chttp2/transport/frame_settings.h
index c19f8067cc..4bfa944cf1 100644
--- a/src/core/ext/transport/chttp2/transport/frame_settings.h
+++ b/src/core/ext/transport/chttp2/transport/frame_settings.h
@@ -95,9 +95,10 @@ gpr_slice grpc_chttp2_settings_ack_create(void);
grpc_error *grpc_chttp2_settings_parser_begin_frame(
grpc_chttp2_settings_parser *parser, uint32_t length, uint8_t flags,
uint32_t *settings);
-grpc_error *grpc_chttp2_settings_parser_parse(
- grpc_exec_ctx *exec_ctx, void *parser,
- grpc_chttp2_transport_global *transport_global,
- grpc_chttp2_stream_global *stream_global, gpr_slice slice, int is_last);
+grpc_error *grpc_chttp2_settings_parser_parse(grpc_exec_ctx *exec_ctx,
+ void *parser,
+ grpc_chttp2_transport *t,
+ grpc_chttp2_stream *s,
+ gpr_slice slice, int is_last);
#endif /* GRPC_CORE_EXT_TRANSPORT_CHTTP2_TRANSPORT_FRAME_SETTINGS_H */
diff --git a/src/core/ext/transport/chttp2/transport/frame_window_update.h b/src/core/ext/transport/chttp2/transport/frame_window_update.h
index 36ae6bd0b1..6e62f31872 100644
--- a/src/core/ext/transport/chttp2/transport/frame_window_update.h
+++ b/src/core/ext/transport/chttp2/transport/frame_window_update.h
@@ -51,8 +51,7 @@ gpr_slice grpc_chttp2_window_update_create(uint32_t id, uint32_t window_delta,
grpc_error *grpc_chttp2_window_update_parser_begin_frame(
grpc_chttp2_window_update_parser *parser, uint32_t length, uint8_t flags);
grpc_error *grpc_chttp2_window_update_parser_parse(
- grpc_exec_ctx *exec_ctx, void *parser,
- grpc_chttp2_transport_global *transport_global,
- grpc_chttp2_stream_global *stream_global, gpr_slice slice, int is_last);
+ grpc_exec_ctx *exec_ctx, void *parser, grpc_chttp2_transport *t,
+ grpc_chttp2_stream *s, gpr_slice slice, int is_last);
#endif /* GRPC_CORE_EXT_TRANSPORT_CHTTP2_TRANSPORT_FRAME_WINDOW_UPDATE_H */
diff --git a/src/core/ext/transport/chttp2/transport/hpack_parser.h b/src/core/ext/transport/chttp2/transport/hpack_parser.h
index 314bd55965..0290c78d5a 100644
--- a/src/core/ext/transport/chttp2/transport/hpack_parser.h
+++ b/src/core/ext/transport/chttp2/transport/hpack_parser.h
@@ -112,9 +112,10 @@ grpc_error *grpc_chttp2_hpack_parser_parse(grpc_exec_ctx *exec_ctx,
/* wraps grpc_chttp2_hpack_parser_parse to provide a frame level parser for
the transport */
-grpc_error *grpc_chttp2_header_parser_parse(
- grpc_exec_ctx *exec_ctx, void *hpack_parser,
- grpc_chttp2_transport_global *transport_global,
- grpc_chttp2_stream_global *stream_global, gpr_slice slice, int is_last);
+grpc_error *grpc_chttp2_header_parser_parse(grpc_exec_ctx *exec_ctx,
+ void *hpack_parser,
+ grpc_chttp2_transport *t,
+ grpc_chttp2_stream *s,
+ gpr_slice slice, int is_last);
#endif /* GRPC_CORE_EXT_TRANSPORT_CHTTP2_TRANSPORT_HPACK_PARSER_H */
diff --git a/src/core/ext/transport/chttp2/transport/internal.h b/src/core/ext/transport/chttp2/transport/internal.h
index 761ed2dad1..1ada0a6a80 100644
--- a/src/core/ext/transport/chttp2/transport/internal.h
+++ b/src/core/ext/transport/chttp2/transport/internal.h
@@ -53,9 +53,6 @@
#include "src/core/lib/transport/connectivity_state.h"
#include "src/core/lib/transport/transport_impl.h"
-typedef struct grpc_chttp2_transport grpc_chttp2_transport;
-typedef struct grpc_chttp2_stream grpc_chttp2_stream;
-
/* streams are kept in various linked lists depending on what things need to
happen to them... this enum labels each list */
typedef enum {
@@ -63,7 +60,6 @@ typedef enum {
GRPC_CHTTP2_LIST_WRITABLE,
GRPC_CHTTP2_LIST_WRITING,
GRPC_CHTTP2_LIST_WRITTEN,
- GRPC_CHTTP2_LIST_CLOSED_WAITING_FOR_WRITING,
GRPC_CHTTP2_LIST_STALLED_BY_TRANSPORT,
/* streams waiting for the outgoing window in the writing path, they will be
* merged to the stalled list or writable list under transport lock. */
@@ -74,6 +70,12 @@ typedef enum {
STREAM_LIST_COUNT /* must be last */
} grpc_chttp2_stream_list_id;
+typedef enum {
+ GRPC_CHTTP2_WRITE_STATE_IDLE,
+ GRPC_CHTTP2_WRITE_STATE_WRITING,
+ GRPC_CHTTP2_WRITE_STATE_WRITING_WITH_MORE_TO_COME,
+} grpc_chttp2_write_state;
+
/* deframer state for the overall http2 stream of bytes */
typedef enum {
/* prefix: one entry per http2 connection prefix byte */
@@ -174,12 +176,76 @@ struct grpc_chttp2_incoming_byte_stream {
grpc_closure finished_action;
};
-struct grpc_chttp2_transport_global {
+struct grpc_chttp2_transport {
+ grpc_transport base; /* must be first */
+ gpr_refcount refs;
+ grpc_endpoint *ep;
+ char *peer_string;
+
+ /** when this drops to zero it's safe to shutdown the endpoint */
+ gpr_refcount shutdown_ep_refs;
+
+ grpc_combiner *combiner;
+
+ /** write execution state of the transport */
+ grpc_chttp2_write_state write_state;
+ /** has a check_read_ops been scheduled */
+ bool check_read_ops_scheduled;
+
+ /** is the transport destroying itself? */
+ uint8_t destroying;
+ /** has the upper layer closed the transport? */
+ uint8_t closed;
+
+ /** is there a read request to the endpoint outstanding? */
+ uint8_t endpoint_reading;
+
+ /** various lists of streams */
+ grpc_chttp2_stream_list lists[STREAM_LIST_COUNT];
+
+ /** maps stream id to grpc_chttp2_stream objects */
+ grpc_chttp2_stream_map stream_map;
+
+ /** closure to execute writing */
+ grpc_closure writing_action;
+ grpc_closure writing_done_action;
+ /** closure to finish writing */
+ grpc_closure terminate_writing;
+ /** closure to start reading from the endpoint */
+ grpc_closure reading_action;
+ grpc_closure reading_action_locked;
+ /** closure to flush read state up the stack */
+ grpc_closure initiate_read_flush_locked;
+
+ /** incoming read bytes */
+ gpr_slice_buffer read_buffer;
+
+ /** address to place a newly accepted stream - set and unset by
+ grpc_chttp2_parsing_accept_stream; used by init_stream to
+ publish the accepted server stream */
+ grpc_chttp2_stream **accepting_stream;
+
+ struct {
+ /* accept stream callback */
+ void (*accept_stream)(grpc_exec_ctx *exec_ctx, void *user_data,
+ grpc_transport *transport, const void *server_data);
+ void *accept_stream_user_data;
+
+ /** connectivity tracking */
+ grpc_connectivity_state_tracker state_tracker;
+ } channel_callback;
+
+ /** data to write now */
+ gpr_slice_buffer outbuf;
+ /** hpack encoding */
+ grpc_chttp2_hpack_compressor hpack_compressor;
+ int64_t outgoing_window;
+ /** is this a client? */
+ uint8_t is_client;
+
/** data to write next write */
gpr_slice_buffer qbuf;
- /** window available for us to send to peer */
- int64_t outgoing_window;
/** window available to announce to peer */
int64_t announce_incoming_window;
/** how much window would we like to have for incoming_window */
@@ -190,8 +256,6 @@ struct grpc_chttp2_transport_global {
/** have we sent a goaway */
uint8_t sent_goaway;
- /** is this transport a client? */
- uint8_t is_client;
/** are the local settings dirty and need to be sent? */
uint8_t dirtied_local_settings;
/** have local settings been sent? */
@@ -246,10 +310,9 @@ struct grpc_chttp2_transport_global {
/* active parser */
void *parser_data;
- grpc_chttp2_stream_global *incoming_stream;
+ grpc_chttp2_stream *incoming_stream;
grpc_error *(*parser)(grpc_exec_ctx *exec_ctx, void *parser_user_data,
- grpc_chttp2_transport_global *transport_global,
- grpc_chttp2_stream_global *stream_global,
+ grpc_chttp2_transport *t, grpc_chttp2_stream *s,
gpr_slice slice, int is_last);
/* goaway data */
@@ -258,109 +321,16 @@ struct grpc_chttp2_transport_global {
gpr_slice goaway_text;
};
-typedef struct {
- /** data to write now */
- gpr_slice_buffer outbuf;
- /** hpack encoding */
- grpc_chttp2_hpack_compressor hpack_compressor;
- int64_t outgoing_window;
- /** is this a client? */
- uint8_t is_client;
- /** callback for when writing is done */
- grpc_closure done_cb;
-} grpc_chttp2_transport_writing;
-
-#if 0
-struct grpc_chttp2_transport_parsing {
-};
-#endif
-
-typedef enum {
- /** no writing activity allowed */
- GRPC_CHTTP2_WRITES_CORKED,
- /** no writing activity */
- GRPC_CHTTP2_WRITING_INACTIVE,
- /** write has been requested and scheduled against the workqueue */
- GRPC_CHTTP2_WRITE_SCHEDULED,
- /** write has been initiated after being reaped from the workqueue */
- GRPC_CHTTP2_WRITING,
- /** write has been initiated, AND another write needs to be started once it's
- done */
- GRPC_CHTTP2_WRITING_STALE_WITH_POLLER,
- GRPC_CHTTP2_WRITING_STALE_NO_POLLER,
-} grpc_chttp2_write_state;
-
-struct grpc_chttp2_transport {
- grpc_transport base; /* must be first */
- gpr_refcount refs;
- grpc_endpoint *ep;
- char *peer_string;
-
- /** when this drops to zero it's safe to shutdown the endpoint */
- gpr_refcount shutdown_ep_refs;
-
- struct {
- grpc_combiner *combiner;
-
- /** write execution state of the transport */
- grpc_chttp2_write_state write_state;
- /** has a check_read_ops been scheduled */
- bool check_read_ops_scheduled;
- } executor;
-
- /** is the transport destroying itself? */
- uint8_t destroying;
- /** has the upper layer closed the transport? */
- uint8_t closed;
-
- /** is there a read request to the endpoint outstanding? */
- uint8_t endpoint_reading;
-
- /** various lists of streams */
- grpc_chttp2_stream_list lists[STREAM_LIST_COUNT];
-
- /** global state for reading/writing */
- grpc_chttp2_transport_global global;
- /** state only accessible by the chain of execution that
- set writing_state >= GRPC_WRITING, and only by the writing closure
- chain. */
- grpc_chttp2_transport_writing writing;
-
- /** maps stream id to grpc_chttp2_stream objects */
- grpc_chttp2_stream_map stream_map;
-
- /** closure to execute writing */
- grpc_closure writing_action;
- /** closure to start reading from the endpoint */
- grpc_closure reading_action;
- grpc_closure reading_action_locked;
- /** closure to initiate writing */
- grpc_closure initiate_writing;
- /** closure to finish writing */
- grpc_closure terminate_writing;
- /** closure to flush read state up the stack */
- grpc_closure initiate_read_flush_locked;
-
- /** incoming read bytes */
- gpr_slice_buffer read_buffer;
-
- /** address to place a newly accepted stream - set and unset by
- grpc_chttp2_parsing_accept_stream; used by init_stream to
- publish the accepted server stream */
- grpc_chttp2_stream **accepting_stream;
+struct grpc_chttp2_stream {
+ grpc_chttp2_transport *t;
+ grpc_stream_refcount *refcount;
- struct {
- /* accept stream callback */
- void (*accept_stream)(grpc_exec_ctx *exec_ctx, void *user_data,
- grpc_transport *transport, const void *server_data);
- void *accept_stream_user_data;
+ grpc_closure destroy_stream;
+ void *destroy_stream_arg;
- /** connectivity tracking */
- grpc_connectivity_state_tracker state_tracker;
- } channel_callback;
-};
+ grpc_chttp2_stream_link links[STREAM_LIST_COUNT];
+ uint8_t included[STREAM_LIST_COUNT];
-struct grpc_chttp2_stream_global {
/** HTTP2 stream id for this stream, or zero if one has not been assigned */
uint32_t id;
@@ -434,50 +404,18 @@ struct grpc_chttp2_stream_global {
grpc_chttp2_data_parser data_parser;
/** number of bytes received - reset at end of parse thread execution */
int64_t received_bytes;
-};
-typedef struct {
/** HTTP2 stream id for this stream, or zero if one has not been assigned */
- uint32_t id;
uint8_t fetching;
bool sent_initial_metadata;
uint8_t sent_message;
uint8_t sent_trailing_metadata;
- uint8_t read_closed;
- /** send this initial metadata */
- grpc_metadata_batch *send_initial_metadata;
- grpc_byte_stream *send_message;
- grpc_metadata_batch *send_trailing_metadata;
- int64_t outgoing_window;
/** how much window should we announce? */
uint32_t announce_window;
gpr_slice_buffer flow_controlled_buffer;
gpr_slice fetching_slice;
size_t stream_fetched;
grpc_closure finished_fetch;
- /** stats gathered during the write */
- grpc_transport_one_way_stats stats;
-} grpc_chttp2_stream_writing;
-
-#if 0
-struct grpc_chttp2_stream_parsing {
-
- /** incoming metadata */
- grpc_chttp2_incoming_metadata_buffer metadata_buffer[2];
-};
-#endif
-
-struct grpc_chttp2_stream {
- grpc_chttp2_transport *t;
- grpc_stream_refcount *refcount;
- grpc_chttp2_stream_global global;
- grpc_chttp2_stream_writing writing;
-
- grpc_closure destroy_stream;
- void *destroy_stream_arg;
-
- grpc_chttp2_stream_link links[STREAM_LIST_COUNT];
- uint8_t included[STREAM_LIST_COUNT];
};
/** Transport writing call flow:
@@ -493,118 +431,84 @@ struct grpc_chttp2_stream {
The actual call chain is documented in the implementation of this function.
*/
void grpc_chttp2_initiate_write(grpc_exec_ctx *exec_ctx,
- grpc_chttp2_transport_global *transport_global,
+ grpc_chttp2_transport *t,
bool covered_by_poller, const char *reason);
/** Someone is unlocking the transport mutex: check to see if writes
- are required, and schedule them if so */
-int grpc_chttp2_unlocking_check_writes(grpc_exec_ctx *exec_ctx,
- grpc_chttp2_transport_global *global,
- grpc_chttp2_transport_writing *writing);
-void grpc_chttp2_perform_writes(
- grpc_exec_ctx *exec_ctx, grpc_chttp2_transport_writing *transport_writing,
- grpc_endpoint *endpoint);
-void grpc_chttp2_terminate_writing(grpc_exec_ctx *exec_ctx,
- void *transport_writing, grpc_error *error);
-void grpc_chttp2_cleanup_writing(grpc_exec_ctx *exec_ctx,
- grpc_chttp2_transport_global *global,
- grpc_chttp2_transport_writing *writing);
+ are required, and frame them if so */
+bool grpc_chttp2_begin_write(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t);
+void grpc_chttp2_end_write(grpc_exec_ctx *exec_ctx, void *transport_writing,
+ grpc_error *error);
/** Process one slice of incoming data; return 1 if the connection is still
viable after reading, or 0 if the connection should be torn down */
-grpc_error *grpc_chttp2_perform_read(
- grpc_exec_ctx *exec_ctx, grpc_chttp2_transport_global *transport_global,
- gpr_slice slice);
+grpc_error *grpc_chttp2_perform_read(grpc_exec_ctx *exec_ctx,
+ grpc_chttp2_transport *t, gpr_slice slice);
-bool grpc_chttp2_list_add_writable_stream(
- grpc_chttp2_transport_global *transport_global,
- grpc_chttp2_stream_global *stream_global);
+bool grpc_chttp2_list_add_writable_stream(grpc_chttp2_transport *t,
+ grpc_chttp2_stream *s);
/** Get a writable stream
returns non-zero if there was a stream available */
-int grpc_chttp2_list_pop_writable_stream(
- grpc_chttp2_transport_global *transport_global,
- grpc_chttp2_transport_writing *transport_writing,
- grpc_chttp2_stream_global **stream_global,
- grpc_chttp2_stream_writing **stream_writing);
+int grpc_chttp2_list_pop_writable_stream(grpc_chttp2_transport *t,
+ grpc_chttp2_stream **s);
bool grpc_chttp2_list_remove_writable_stream(
- grpc_chttp2_transport_global *transport_global,
- grpc_chttp2_stream_global *stream_global) GRPC_MUST_USE_RESULT;
-
-void grpc_chttp2_list_add_writing_stream(
- grpc_chttp2_transport_writing *transport_writing,
- grpc_chttp2_stream_writing *stream_writing);
-int grpc_chttp2_list_have_writing_streams(
- grpc_chttp2_transport_writing *transport_writing);
-int grpc_chttp2_list_pop_writing_stream(
- grpc_chttp2_transport_writing *transport_writing,
- grpc_chttp2_stream_writing **stream_writing);
-
-void grpc_chttp2_list_add_written_stream(
- grpc_chttp2_transport_writing *transport_writing,
- grpc_chttp2_stream_writing *stream_writing);
-int grpc_chttp2_list_pop_written_stream(
- grpc_chttp2_transport_global *transport_global,
- grpc_chttp2_transport_writing *transport_writing,
- grpc_chttp2_stream_global **stream_global,
- grpc_chttp2_stream_writing **stream_writing);
-
-void grpc_chttp2_list_add_waiting_for_concurrency(
- grpc_chttp2_transport_global *transport_global,
- grpc_chttp2_stream_global *stream_global);
-int grpc_chttp2_list_pop_waiting_for_concurrency(
- grpc_chttp2_transport_global *transport_global,
- grpc_chttp2_stream_global **stream_global);
-
-void grpc_chttp2_list_add_check_read_ops(
- grpc_exec_ctx *exec_ctx, grpc_chttp2_transport_global *transport_global,
- grpc_chttp2_stream_global *stream_global);
-bool grpc_chttp2_list_remove_check_read_ops(
- grpc_chttp2_transport_global *transport_global,
- grpc_chttp2_stream_global *stream_global);
-int grpc_chttp2_list_pop_check_read_ops(
- grpc_chttp2_transport_global *transport_global,
- grpc_chttp2_stream_global **stream_global);
-
-void grpc_chttp2_list_add_writing_stalled_by_transport(
- grpc_chttp2_transport_writing *transport_writing,
- grpc_chttp2_stream_writing *stream_writing);
+ grpc_chttp2_transport *t, grpc_chttp2_stream *s) GRPC_MUST_USE_RESULT;
+
+void grpc_chttp2_list_add_writing_stream(grpc_chttp2_transport *t,
+ grpc_chttp2_stream *s);
+int grpc_chttp2_list_have_writing_streams(grpc_chttp2_transport *t);
+int grpc_chttp2_list_pop_writing_stream(grpc_chttp2_transport *t,
+ grpc_chttp2_stream **s);
+
+void grpc_chttp2_list_add_written_stream(grpc_chttp2_transport *t,
+ grpc_chttp2_stream *s);
+int grpc_chttp2_list_pop_written_stream(grpc_chttp2_transport *t,
+ grpc_chttp2_stream **s);
+
+void grpc_chttp2_list_add_waiting_for_concurrency(grpc_chttp2_transport *t,
+ grpc_chttp2_stream *s);
+int grpc_chttp2_list_pop_waiting_for_concurrency(grpc_chttp2_transport *t,
+ grpc_chttp2_stream **s);
+
+void grpc_chttp2_list_add_check_read_ops(grpc_exec_ctx *exec_ctx,
+ grpc_chttp2_transport *t,
+ grpc_chttp2_stream *s);
+bool grpc_chttp2_list_remove_check_read_ops(grpc_chttp2_transport *t,
+ grpc_chttp2_stream *s);
+int grpc_chttp2_list_pop_check_read_ops(grpc_chttp2_transport *t,
+ grpc_chttp2_stream **s);
+
+void grpc_chttp2_list_add_writing_stalled_by_transport(grpc_chttp2_transport *t,
+ grpc_chttp2_stream *s);
bool grpc_chttp2_list_flush_writing_stalled_by_transport(
- grpc_exec_ctx *exec_ctx, grpc_chttp2_transport_writing *transport_writing);
-
-void grpc_chttp2_list_add_stalled_by_transport(
- grpc_chttp2_transport_writing *transport_writing,
- grpc_chttp2_stream_writing *stream_writing);
-int grpc_chttp2_list_pop_stalled_by_transport(
- grpc_chttp2_transport_global *transport_global,
- grpc_chttp2_stream_global **stream_global);
-void grpc_chttp2_list_remove_stalled_by_transport(
- grpc_chttp2_transport_global *transport_global,
- grpc_chttp2_stream_global *stream_global);
-
-void grpc_chttp2_list_add_closed_waiting_for_writing(
- grpc_chttp2_transport_global *transport_global,
- grpc_chttp2_stream_global *stream_global);
-int grpc_chttp2_list_pop_closed_waiting_for_writing(
- grpc_chttp2_transport_global *transport_global,
- grpc_chttp2_stream_global **stream_global);
-
-grpc_chttp2_stream_global *grpc_chttp2_parsing_lookup_stream(
- grpc_chttp2_transport_global *transport_global, uint32_t id);
-grpc_chttp2_stream_global *grpc_chttp2_parsing_accept_stream(
- grpc_exec_ctx *exec_ctx, grpc_chttp2_transport_global *transport_global,
- uint32_t id);
-
-void grpc_chttp2_add_incoming_goaway(
- grpc_exec_ctx *exec_ctx, grpc_chttp2_transport_global *transport_global,
- uint32_t goaway_error, gpr_slice goaway_text);
-
-void grpc_chttp2_parsing_become_skip_parser(
- grpc_exec_ctx *exec_ctx, grpc_chttp2_transport_global *transport_global);
-
-void grpc_chttp2_complete_closure_step(
- grpc_exec_ctx *exec_ctx, grpc_chttp2_transport_global *transport_global,
- grpc_chttp2_stream_global *stream_global, grpc_closure **pclosure,
- grpc_error *error);
+ grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t);
+
+void grpc_chttp2_list_add_stalled_by_transport(grpc_chttp2_transport *t,
+ grpc_chttp2_stream *s);
+int grpc_chttp2_list_pop_stalled_by_transport(grpc_chttp2_transport *t,
+ grpc_chttp2_stream **s);
+void grpc_chttp2_list_remove_stalled_by_transport(grpc_chttp2_transport *t,
+ grpc_chttp2_stream *s);
+
+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,
+ grpc_chttp2_transport *t,
+ uint32_t id);
+
+void grpc_chttp2_add_incoming_goaway(grpc_exec_ctx *exec_ctx,
+ grpc_chttp2_transport *t,
+ uint32_t goaway_error,
+ gpr_slice goaway_text);
+
+void grpc_chttp2_parsing_become_skip_parser(grpc_exec_ctx *exec_ctx,
+ grpc_chttp2_transport *t);
+
+void grpc_chttp2_complete_closure_step(grpc_exec_ctx *exec_ctx,
+ grpc_chttp2_transport *t,
+ grpc_chttp2_stream *s,
+ grpc_closure **pclosure,
+ grpc_error *error);
#define GRPC_CHTTP2_CLIENT_CONNECT_STRING "PRI * HTTP/2.0\r\n\r\nSM\r\n\r\n"
#define GRPC_CHTTP2_CLIENT_CONNECT_STRLEN \
@@ -695,35 +599,30 @@ void grpc_chttp2_flowctl_trace(const char *file, int line, const char *phase,
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_global *transport_global,
- grpc_chttp2_stream_global *stream,
+void grpc_chttp2_fake_status(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t,
+ grpc_chttp2_stream *stream,
grpc_status_code status, gpr_slice *details);
-void grpc_chttp2_mark_stream_closed(
- grpc_exec_ctx *exec_ctx, grpc_chttp2_transport_global *transport_global,
- grpc_chttp2_stream_global *stream_global, int close_reads, int close_writes,
- grpc_error *error);
+void grpc_chttp2_mark_stream_closed(grpc_exec_ctx *exec_ctx,
+ grpc_chttp2_transport *t,
+ grpc_chttp2_stream *s, int close_reads,
+ int close_writes, grpc_error *error);
void grpc_chttp2_start_writing(grpc_exec_ctx *exec_ctx,
- grpc_chttp2_transport_global *transport_global);
+ grpc_chttp2_transport *t);
#ifdef GRPC_STREAM_REFCOUNT_DEBUG
-#define GRPC_CHTTP2_STREAM_REF(stream_global, reason) \
- grpc_chttp2_stream_ref(stream_global, reason)
-#define GRPC_CHTTP2_STREAM_UNREF(exec_ctx, stream_global, reason) \
- grpc_chttp2_stream_unref(exec_ctx, stream_global, reason)
-void grpc_chttp2_stream_ref(grpc_chttp2_stream_global *stream_global,
- const char *reason);
-void grpc_chttp2_stream_unref(grpc_exec_ctx *exec_ctx,
- grpc_chttp2_stream_global *stream_global,
+#define GRPC_CHTTP2_STREAM_REF(stream, reason) \
+ grpc_chttp2_stream_ref(stream, reason)
+#define GRPC_CHTTP2_STREAM_UNREF(exec_ctx, stream, reason) \
+ grpc_chttp2_stream_unref(exec_ctx, stream, reason)
+void grpc_chttp2_stream_ref(grpc_chttp2_stream *s, const char *reason);
+void grpc_chttp2_stream_unref(grpc_exec_ctx *exec_ctx, grpc_chttp2_stream *s,
const char *reason);
#else
-#define GRPC_CHTTP2_STREAM_REF(stream_global, reason) \
- grpc_chttp2_stream_ref(stream_global)
-#define GRPC_CHTTP2_STREAM_UNREF(exec_ctx, stream_global, reason) \
- grpc_chttp2_stream_unref(exec_ctx, stream_global)
-void grpc_chttp2_stream_ref(grpc_chttp2_stream_global *stream_global);
-void grpc_chttp2_stream_unref(grpc_exec_ctx *exec_ctx,
- grpc_chttp2_stream_global *stream_global);
+#define GRPC_CHTTP2_STREAM_REF(stream, reason) grpc_chttp2_stream_ref(stream)
+#define GRPC_CHTTP2_STREAM_UNREF(exec_ctx, stream, reason) \
+ grpc_chttp2_stream_unref(exec_ctx, stream)
+void grpc_chttp2_stream_ref(grpc_chttp2_stream *s);
+void grpc_chttp2_stream_unref(grpc_exec_ctx *exec_ctx, grpc_chttp2_stream *s);
#endif
//#define GRPC_CHTTP2_REFCOUNTING_DEBUG 1
@@ -746,9 +645,8 @@ void grpc_chttp2_ref_transport(grpc_chttp2_transport *t);
#endif
grpc_chttp2_incoming_byte_stream *grpc_chttp2_incoming_byte_stream_create(
- grpc_exec_ctx *exec_ctx, grpc_chttp2_transport_global *transport_global,
- grpc_chttp2_stream_global *stream_global, uint32_t frame_size,
- uint32_t flags);
+ grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t, grpc_chttp2_stream *s,
+ uint32_t frame_size, uint32_t flags);
void grpc_chttp2_incoming_byte_stream_push(grpc_exec_ctx *exec_ctx,
grpc_chttp2_incoming_byte_stream *bs,
gpr_slice slice);
@@ -756,20 +654,18 @@ void grpc_chttp2_incoming_byte_stream_finished(
grpc_exec_ctx *exec_ctx, grpc_chttp2_incoming_byte_stream *bs,
grpc_error *error, int from_parsing_thread);
-void grpc_chttp2_ack_ping(grpc_exec_ctx *exec_ctx,
- grpc_chttp2_transport_global *parsing,
+void grpc_chttp2_ack_ping(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t,
const uint8_t *opaque_8bytes);
/** add a ref to the stream and add it to the writable list;
ref will be dropped in writing.c */
void grpc_chttp2_become_writable(grpc_exec_ctx *exec_ctx,
- grpc_chttp2_transport_global *transport_global,
- grpc_chttp2_stream_global *stream_global,
- bool covered_by_poller, const char *reason);
+ grpc_chttp2_transport *t,
+ grpc_chttp2_stream *s, bool covered_by_poller,
+ const char *reason);
void grpc_chttp2_cancel_stream(grpc_exec_ctx *exec_ctx,
- grpc_chttp2_transport_global *transport_global,
- grpc_chttp2_stream_global *stream_global,
+ grpc_chttp2_transport *t, grpc_chttp2_stream *s,
grpc_error *due_to_error);
#endif /* GRPC_CORE_EXT_TRANSPORT_CHTTP2_TRANSPORT_INTERNAL_H */
--
GitLab