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

Fix flow control for 0-byte messages

Browse Source
pull/4136/head
Craig Tiller 9 years ago
parent 27226f5999
commit 20df14ef6d
4 changed files with 54 additions and 37 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. 2
      src/core/surface/call.c
  2. 4
      src/core/transport/chttp2/frame_data.c
  3. 2
      src/core/transport/chttp2/internal.h
  4. 83
      src/core/transport/chttp2_transport.c

2
src/core/surface/call.c
Unescape View File

@ -944,12 +944,12 @@ static void post_batch_completion(grpc_exec_ctx *exec_ctx,
batch_control *bctl) {
grpc_call *call = bctl->call;
if (bctl->is_notify_tag_closure) {
grpc_exec_ctx_enqueue(exec_ctx, bctl->notify_tag, bctl->success);
gpr_mu_lock(&call->mu);
bctl->call->used_batches =
(gpr_uint8)(bctl->call->used_batches &
~(gpr_uint8)(1 << (bctl - bctl->call->active_batches)));
gpr_mu_unlock(&call->mu);
grpc_exec_ctx_enqueue(exec_ctx, bctl->notify_tag, bctl->success);
GRPC_CALL_INTERNAL_UNREF(exec_ctx, call, "completion");
} else {
grpc_cq_end_op(exec_ctx, bctl->call->cq, bctl->notify_tag, bctl->success,

4
src/core/transport/chttp2/frame_data.c
Unescape View File

@ -198,8 +198,8 @@ grpc_chttp2_parse_error grpc_chttp2_data_parser_parse(
}
p->parsing_frame = incoming_byte_stream =
grpc_chttp2_incoming_byte_stream_create(
transport_parsing, stream_parsing, p->frame_size, message_flags,
&p->incoming_frames);
exec_ctx, transport_parsing, stream_parsing, p->frame_size,
message_flags, &p->incoming_frames);
/* fallthrough */
case GRPC_CHTTP2_DATA_FRAME:
if (cur == end) {

2
src/core/transport/chttp2/internal.h
Unescape View File

@ -738,7 +738,7 @@ void grpc_chttp2_stream_unref(grpc_exec_ctx *exec_ctx,
#endif
grpc_chttp2_incoming_byte_stream *grpc_chttp2_incoming_byte_stream_create(
grpc_chttp2_transport_parsing *transport_parsing,
grpc_exec_ctx *exec_ctx, grpc_chttp2_transport_parsing *transport_parsing,
grpc_chttp2_stream_parsing *stream_parsing, gpr_uint32 frame_size,
gpr_uint32 flags, grpc_chttp2_incoming_frame_queue *add_to_queue);
void grpc_chttp2_incoming_byte_stream_push(grpc_exec_ctx *exec_ctx,

83
src/core/transport/chttp2_transport.c
Unescape View File

@ -1364,6 +1364,46 @@ static void set_pollset(grpc_exec_ctx *exec_ctx, grpc_transport *gt,
* BYTE STREAM
*/
static void incoming_byte_stream_update_flow_control(
grpc_chttp2_transport_global *transport_global,
grpc_chttp2_stream_global *stream_global, size_t max_size_hint,
size_t have_already) {
gpr_uint32 max_recv_bytes;
/* clamp max recv hint to an allowable size */
if (max_size_hint >= GPR_UINT32_MAX - GRPC_CHTTP2_STREAM_LOOKAHEAD) {
max_recv_bytes = GPR_UINT32_MAX - GRPC_CHTTP2_STREAM_LOOKAHEAD;
} else {
max_recv_bytes = (gpr_uint32)max_size_hint;
}
/* account for bytes already received but unknown to higher layers */
if (max_recv_bytes >= have_already) {
max_recv_bytes -= (gpr_uint32)have_already;
} else {
max_recv_bytes = 0;
}
/* add some small lookahead to keep pipelines flowing */
GPR_ASSERT(max_recv_bytes <= GPR_UINT32_MAX - GRPC_CHTTP2_STREAM_LOOKAHEAD);
max_recv_bytes += GRPC_CHTTP2_STREAM_LOOKAHEAD;
if (stream_global->max_recv_bytes < max_recv_bytes) {
gpr_uint32 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,
unannounced_incoming_window_for_parse,
add_max_recv_bytes);
GRPC_CHTTP2_FLOW_CREDIT_STREAM("op", transport_global, stream_global,
unannounced_incoming_window_for_writing,
add_max_recv_bytes);
grpc_chttp2_list_add_unannounced_incoming_window_available(transport_global,
stream_global);
grpc_chttp2_list_add_writable_stream(transport_global, stream_global);
}
}
static int incoming_byte_stream_next(grpc_exec_ctx *exec_ctx,
grpc_byte_stream *byte_stream,
gpr_slice *slice, size_t max_size_hint,
@ -1372,41 +1412,11 @@ static int incoming_byte_stream_next(grpc_exec_ctx *exec_ctx,
(grpc_chttp2_incoming_byte_stream *)byte_stream;
grpc_chttp2_transport_global *transport_global = &bs->transport->global;
grpc_chttp2_stream_global *stream_global = &bs->stream->global;
gpr_uint32 max_recv_bytes;
lock(bs->transport);
if (bs->is_tail) {
/* clamp max recv hint to an allowable size */
if (max_size_hint >= GPR_UINT32_MAX - GRPC_CHTTP2_STREAM_LOOKAHEAD) {
max_recv_bytes = GPR_UINT32_MAX - GRPC_CHTTP2_STREAM_LOOKAHEAD;
} else {
max_recv_bytes = (gpr_uint32)max_size_hint;
}
/* account for bytes already received but unknown to higher layers */
if (max_recv_bytes >= bs->slices.length) {
max_recv_bytes -= (gpr_uint32)bs->slices.length;
} else {
max_recv_bytes = 0;
}
/* add some small lookahead to keep pipelines flowing */
GPR_ASSERT(max_recv_bytes <= GPR_UINT32_MAX - GRPC_CHTTP2_STREAM_LOOKAHEAD);
max_recv_bytes += GRPC_CHTTP2_STREAM_LOOKAHEAD;
if (stream_global->max_recv_bytes < max_recv_bytes) {
gpr_uint32 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,
unannounced_incoming_window_for_parse,
add_max_recv_bytes);
GRPC_CHTTP2_FLOW_CREDIT_STREAM("op", transport_global, stream_global,
unannounced_incoming_window_for_writing,
add_max_recv_bytes);
grpc_chttp2_list_add_unannounced_incoming_window_available(
transport_global, stream_global);
grpc_chttp2_list_add_writable_stream(transport_global, stream_global);
}
incoming_byte_stream_update_flow_control(transport_global, stream_global,
max_size_hint, bs->slices.length);
}
if (bs->slices.count > 0) {
*slice = gpr_slice_buffer_take_first(&bs->slices);
@ -1451,7 +1461,7 @@ void grpc_chttp2_incoming_byte_stream_finished(
}
grpc_chttp2_incoming_byte_stream *grpc_chttp2_incoming_byte_stream_create(
grpc_chttp2_transport_parsing *transport_parsing,
grpc_exec_ctx *exec_ctx, grpc_chttp2_transport_parsing *transport_parsing,
grpc_chttp2_stream_parsing *stream_parsing, gpr_uint32 frame_size,
gpr_uint32 flags, grpc_chttp2_incoming_frame_queue *add_to_queue) {
grpc_chttp2_incoming_byte_stream *incoming_byte_stream =
@ -1474,6 +1484,13 @@ grpc_chttp2_incoming_byte_stream *grpc_chttp2_incoming_byte_stream_create(
add_to_queue->tail->next_message = incoming_byte_stream;
}
add_to_queue->tail = incoming_byte_stream;
if (frame_size == 0) {
lock(TRANSPORT_FROM_PARSING(transport_parsing));
incoming_byte_stream_update_flow_control(
&TRANSPORT_FROM_PARSING(transport_parsing)->global,
&STREAM_FROM_PARSING(stream_parsing)->global, 0, 0);
unlock(exec_ctx, TRANSPORT_FROM_PARSING(transport_parsing));
}
return incoming_byte_stream;
}

Write Preview
Loading…
Cancel
Save