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

Eliminate dual stream maps

Browse Source
pull/7793/head
Craig Tiller 9 years ago
parent 13e4bf8e6a
commit 796f525f7f
2 changed files with 18 additions and 46 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Split View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 50
      src/core/ext/transport/chttp2/transport/chttp2_transport.c
  2. 14
      src/core/ext/transport/chttp2/transport/internal.h

50
src/core/ext/transport/chttp2/transport/chttp2_transport.c
Unescape View File

@ -163,11 +163,9 @@ static void destruct_transport(grpc_exec_ctx *exec_ctx,
GPR_ASSERT(t->lists[i].tail == NULL);
}
GPR_ASSERT(grpc_chttp2_stream_map_size(&t->parsing_stream_map) == 0);
GPR_ASSERT(grpc_chttp2_stream_map_size(&t->new_stream_map) == 0);
GPR_ASSERT(grpc_chttp2_stream_map_size(&t->stream_map) == 0);
grpc_chttp2_stream_map_destroy(&t->parsing_stream_map);
grpc_chttp2_stream_map_destroy(&t->new_stream_map);
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);
@ -277,8 +275,7 @@ static void init_transport(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t,
large enough that the exponential growth should happen nicely when it's
needed.
TODO(ctiller): tune this */
grpc_chttp2_stream_map_init(&t->parsing_stream_map, 8);
grpc_chttp2_stream_map_init(&t->new_stream_map, 8);
grpc_chttp2_stream_map_init(&t->stream_map, 8);
/* copy in initial settings to all setting sets */
for (i = 0; i < GRPC_CHTTP2_NUM_SETTINGS; i++) {
@ -509,7 +506,7 @@ static int init_stream(grpc_exec_ctx *exec_ctx, grpc_transport *gt,
t->global.settings[GRPC_SENT_SETTINGS]
[GRPC_CHTTP2_SETTINGS_INITIAL_WINDOW_SIZE];
*t->accepting_stream = s;
grpc_chttp2_stream_map_add(&t->parsing_stream_map, s->global.id, s);
grpc_chttp2_stream_map_add(&t->stream_map, s->global.id, s);
s->global.in_stream_map = true;
}
@ -540,8 +537,8 @@ static void destroy_stream_locked(grpc_exec_ctx *exec_ctx, void *sp,
GRPC_ERROR_CREATE("Last stream closed after sending goaway"));
}
if (s->global.id != 0) {
GPR_ASSERT(grpc_chttp2_stream_map_find(&t->parsing_stream_map,
s->global.id) == NULL);
GPR_ASSERT(grpc_chttp2_stream_map_find(&t->stream_map, s->global.id) ==
NULL);
}
while (
@ -596,8 +593,7 @@ static void destroy_stream(grpc_exec_ctx *exec_ctx, grpc_transport *gt,
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->parsing_stream_map, id);
grpc_chttp2_stream *s = grpc_chttp2_stream_map_find(&t->stream_map, id);
return s ? &s->global : NULL;
}
@ -878,7 +874,8 @@ static void maybe_start_some_streams(
/* start streams where we have free grpc_chttp2_stream ids and free
* concurrency */
while (transport_global->next_stream_id <= MAX_CLIENT_STREAM_ID &&
transport_global->concurrent_stream_count <
grpc_chttp2_stream_map_size(
&TRANSPORT_FROM_GLOBAL(transport_global)->stream_map) <
transport_global
->settings[GRPC_PEER_SETTINGS]
[GRPC_CHTTP2_SETTINGS_MAX_CONCURRENT_STREAMS] &&
@ -909,10 +906,9 @@ static void maybe_start_some_streams(
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)->new_stream_map,
stream_global->id, STREAM_FROM_GLOBAL(stream_global));
&TRANSPORT_FROM_GLOBAL(transport_global)->stream_map, stream_global->id,
STREAM_FROM_GLOBAL(stream_global));
stream_global->in_stream_map = true;
transport_global->concurrent_stream_count++;
grpc_chttp2_become_writable(exec_ctx, transport_global, stream_global, true,
"new_stream");
}
@ -1397,12 +1393,7 @@ static void decrement_active_streams_locked(
static void remove_stream(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t,
uint32_t id, grpc_error *error) {
size_t new_stream_count;
grpc_chttp2_stream *s =
grpc_chttp2_stream_map_delete(&t->parsing_stream_map, id);
if (!s) {
s = grpc_chttp2_stream_map_delete(&t->new_stream_map, id);
}
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) {
@ -1425,14 +1416,9 @@ static void remove_stream(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t,
GRPC_CHTTP2_STREAM_UNREF(exec_ctx, &s->global, "chttp2_writing");
}
new_stream_count = grpc_chttp2_stream_map_size(&t->parsing_stream_map) +
grpc_chttp2_stream_map_size(&t->new_stream_map);
GPR_ASSERT(new_stream_count <= UINT32_MAX);
if (new_stream_count != t->global.concurrent_stream_count) {
t->global.concurrent_stream_count = (uint32_t)new_stream_count;
maybe_start_some_streams(exec_ctx, &t->global);
}
GRPC_ERROR_UNREF(error);
maybe_start_some_streams(exec_ctx, &t->global);
}
static void status_codes_from_error(grpc_error *error, gpr_timespec deadline,
@ -1859,10 +1845,6 @@ static void reading_action_locked(grpc_exec_ctx *exec_ctx, void *tp,
GRPC_ERROR_REF(error);
if (!t->closed) {
/* merge stream lists */
grpc_chttp2_stream_map_move_into(&t->new_stream_map,
&t->parsing_stream_map);
GPR_TIMER_BEGIN("reading_action.parse", 0);
size_t i = 0;
grpc_error *errors[3] = {GRPC_ERROR_REF(error), GRPC_ERROR_NONE,
@ -1885,8 +1867,8 @@ static void reading_action_locked(grpc_exec_ctx *exec_ctx, void *tp,
GPR_TIMER_BEGIN("post_parse_locked", 0);
if (transport_global->initial_window_update != 0) {
update_global_window_args args = {t, exec_ctx};
grpc_chttp2_stream_map_for_each(&t->parsing_stream_map,
update_global_window, &args);
grpc_chttp2_stream_map_for_each(&t->stream_map, update_global_window,
&args);
transport_global->initial_window_update = 0;
}
/* handle higher level things */

14
src/core/ext/transport/chttp2/transport/internal.h
Unescape View File

@ -217,10 +217,6 @@ struct grpc_chttp2_transport_global {
/** next payload for an outgoing ping */
uint64_t ping_counter;
/** concurrent stream count: updated when not parsing,
so this is a strict over-estimation on the client */
uint32_t concurrent_stream_count;
/** parser for headers */
grpc_chttp2_hpack_parser hpack_parser;
/** simple one shot parsers */
@ -331,14 +327,8 @@ struct grpc_chttp2_transport {
chain. */
grpc_chttp2_transport_writing writing;
/** maps stream id to grpc_chttp2_stream objects;
owned by the parsing thread when parsing */
grpc_chttp2_stream_map parsing_stream_map;
/** streams created by the client (possibly during parsing);
merged with parsing_stream_map during unlock when no
parsing is occurring */
grpc_chttp2_stream_map new_stream_map;
/** maps stream id to grpc_chttp2_stream objects */
grpc_chttp2_stream_map stream_map;
/** closure to execute writing */
grpc_closure writing_action;

Write Preview
Loading…
Cancel
Save