diff options
Diffstat (limited to 'src/core')
22 files changed, 1841 insertions, 1421 deletions
diff --git a/src/core/transport/byte_stream.c b/src/core/transport/byte_stream.c new file mode 100644 index 0000000000..81e8e77ccb --- /dev/null +++ b/src/core/transport/byte_stream.c @@ -0,0 +1,76 @@ +/* + * + * Copyright 2015, Google Inc. + * All rights reserved. + * + * Redistribution and use in source and binary forms, with or without + * modification, are permitted provided that the following conditions are + * met: + * + * * Redistributions of source code must retain the above copyright + * notice, this list of conditions and the following disclaimer. + * * Redistributions in binary form must reproduce the above + * copyright notice, this list of conditions and the following disclaimer + * in the documentation and/or other materials provided with the + * distribution. + * * Neither the name of Google Inc. nor the names of its + * contributors may be used to endorse or promote products derived from + * this software without specific prior written permission. + * + * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS + * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT + * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR + * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT + * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, + * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT + * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, + * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY + * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT + * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE + * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. + * + */ + +#include "src/core/transport/byte_stream.h" + +#include <stdlib.h> + +#include <grpc/support/log.h> + +int grpc_byte_stream_next(grpc_exec_ctx *exec_ctx, + grpc_byte_stream *byte_stream, gpr_slice *slice, + size_t max_size_hint, grpc_closure *on_complete) { + return byte_stream->next(exec_ctx, byte_stream, slice, max_size_hint, + on_complete); +} + +void grpc_byte_stream_destroy(grpc_byte_stream *byte_stream) { + byte_stream->destroy(byte_stream); +} + +/* slice_buffer_stream */ + +static int slice_buffer_stream_next(grpc_exec_ctx *exec_ctx, + grpc_byte_stream *byte_stream, + gpr_slice *slice, size_t max_size_hint, + grpc_closure *on_complete) { + grpc_slice_buffer_stream *stream = (grpc_slice_buffer_stream *)byte_stream; + GPR_ASSERT(stream->cursor < stream->backing_buffer->count); + *slice = gpr_slice_ref(stream->backing_buffer->slices[stream->cursor]); + stream->cursor++; + return 1; +} + +static void slice_buffer_stream_destroy(grpc_byte_stream *byte_stream) {} + +void grpc_slice_buffer_stream_init(grpc_slice_buffer_stream *stream, + gpr_slice_buffer *slice_buffer, + gpr_uint32 flags) { + GPR_ASSERT(slice_buffer->length <= GPR_UINT32_MAX); + stream->base.length = (gpr_uint32)slice_buffer->length; + stream->base.flags = flags; + stream->base.next = slice_buffer_stream_next; + stream->base.destroy = slice_buffer_stream_destroy; + stream->backing_buffer = slice_buffer; + stream->cursor = 0; +} diff --git a/src/core/transport/byte_stream.h b/src/core/transport/byte_stream.h new file mode 100644 index 0000000000..c94d8ff275 --- /dev/null +++ b/src/core/transport/byte_stream.h @@ -0,0 +1,88 @@ +/* + * + * Copyright 2015, Google Inc. + * All rights reserved. + * + * Redistribution and use in source and binary forms, with or without + * modification, are permitted provided that the following conditions are + * met: + * + * * Redistributions of source code must retain the above copyright + * notice, this list of conditions and the following disclaimer. + * * Redistributions in binary form must reproduce the above + * copyright notice, this list of conditions and the following disclaimer + * in the documentation and/or other materials provided with the + * distribution. + * * Neither the name of Google Inc. nor the names of its + * contributors may be used to endorse or promote products derived from + * this software without specific prior written permission. + * + * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS + * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT + * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR + * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT + * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, + * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT + * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, + * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY + * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT + * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE + * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. + * + */ + +#ifndef GRPC_INTERNAL_CORE_TRANSPORT_BYTE_STREAM_H +#define GRPC_INTERNAL_CORE_TRANSPORT_BYTE_STREAM_H + +#include "src/core/iomgr/exec_ctx.h" +#include <grpc/support/slice_buffer.h> + +/** Internal bit flag for grpc_begin_message's \a flags signaling the use of + * compression for the message */ +#define GRPC_WRITE_INTERNAL_COMPRESS (0x80000000u) +/** Mask of all valid internal flags. */ +#define GRPC_WRITE_INTERNAL_USED_MASK (GRPC_WRITE_INTERNAL_COMPRESS) + +struct grpc_byte_stream; +typedef struct grpc_byte_stream grpc_byte_stream; + +struct grpc_byte_stream { + gpr_uint32 length; + gpr_uint32 flags; + int (*next)(grpc_exec_ctx *exec_ctx, grpc_byte_stream *byte_stream, + gpr_slice *slice, size_t max_size_hint, + grpc_closure *on_complete); + void (*destroy)(grpc_byte_stream *byte_stream); +}; + +/* returns 1 if the bytes are available immediately (in which case + * on_complete will not be called), 0 if the bytes will be available + * asynchronously. + * + * on entry, *remaining can be set as a hint as to the maximum number + * of bytes that would be acceptable to read. + * + * fills *buffer, *length, *remaining with the bytes, length of bytes + * and length of data remaining to be read before either returning 1 + * or calling on_complete. + * + * once a slice is returned into *slice, it is owned by the caller. + */ +int grpc_byte_stream_next(grpc_exec_ctx *exec_ctx, + grpc_byte_stream *byte_stream, gpr_slice *slice, + size_t max_size_hint, grpc_closure *on_complete); + +void grpc_byte_stream_destroy(grpc_byte_stream *byte_stream); + +/* grpc_byte_stream that wraps a slice buffer */ +typedef struct grpc_slice_buffer_stream { + grpc_byte_stream base; + gpr_slice_buffer *backing_buffer; + size_t cursor; +} grpc_slice_buffer_stream; + +void grpc_slice_buffer_stream_init(grpc_slice_buffer_stream *stream, + gpr_slice_buffer *slice_buffer, + gpr_uint32 flags); + +#endif /* GRPC_INTERNAL_CORE_TRANSPORT_BYTE_STREAM_H */ diff --git a/src/core/transport/chttp2/frame_data.c b/src/core/transport/chttp2/frame_data.c index 07179a4571..7287f97aaa 100644 --- a/src/core/transport/chttp2/frame_data.c +++ b/src/core/transport/chttp2/frame_data.c @@ -45,12 +45,15 @@ grpc_chttp2_parse_error grpc_chttp2_data_parser_init( grpc_chttp2_data_parser *parser) { parser->state = GRPC_CHTTP2_DATA_FH_0; - grpc_sopb_init(&parser->incoming_sopb); return GRPC_CHTTP2_PARSE_OK; } void grpc_chttp2_data_parser_destroy(grpc_chttp2_data_parser *parser) { - grpc_sopb_destroy(&parser->incoming_sopb); + grpc_byte_stream *bs; + while ( + (bs = grpc_chttp2_incoming_frame_queue_pop(&parser->incoming_frames))) { + grpc_byte_stream_destroy(bs); + } } grpc_chttp2_parse_error grpc_chttp2_data_parser_begin_frame( @@ -69,6 +72,62 @@ grpc_chttp2_parse_error grpc_chttp2_data_parser_begin_frame( return GRPC_CHTTP2_PARSE_OK; } +void grpc_chttp2_incoming_frame_queue_merge( + grpc_chttp2_incoming_frame_queue *head_dst, + grpc_chttp2_incoming_frame_queue *tail_src) { + if (tail_src->head == NULL) { + return; + } + + if (head_dst->head == NULL) { + *head_dst = *tail_src; + memset(tail_src, 0, sizeof(*tail_src)); + return; + } + + head_dst->tail->next_message = tail_src->head; + head_dst->tail = tail_src->tail; + memset(tail_src, 0, sizeof(*tail_src)); +} + +grpc_byte_stream *grpc_chttp2_incoming_frame_queue_pop( + grpc_chttp2_incoming_frame_queue *q) { + grpc_byte_stream *out; + if (q->head == NULL) { + return NULL; + } + out = &q->head->base; + if (q->head == q->tail) { + memset(q, 0, sizeof(*q)); + } else { + q->head = q->head->next_message; + } + return out; +} + +void grpc_chttp2_encode_data(gpr_uint32 id, gpr_slice_buffer *inbuf, + gpr_uint32 write_bytes, int is_eof, + gpr_slice_buffer *outbuf) { + gpr_slice hdr; + gpr_uint8 *p; + + hdr = gpr_slice_malloc(9); + p = GPR_SLICE_START_PTR(hdr); + GPR_ASSERT(write_bytes < 16777316); + *p++ = (gpr_uint8)(write_bytes >> 16); + *p++ = (gpr_uint8)(write_bytes >> 8); + *p++ = (gpr_uint8)(write_bytes); + *p++ = GRPC_CHTTP2_FRAME_DATA; + *p++ = is_eof ? GRPC_CHTTP2_DATA_FLAG_END_STREAM : 0; + *p++ = (gpr_uint8)(id >> 24); + *p++ = (gpr_uint8)(id >> 16); + *p++ = (gpr_uint8)(id >> 8); + *p++ = (gpr_uint8)(id); + gpr_slice_buffer_add(outbuf, hdr); + + gpr_slice_buffer_move_first(inbuf, write_bytes, outbuf); +} + grpc_chttp2_parse_error grpc_chttp2_data_parser_parse( grpc_exec_ctx *exec_ctx, void *parser, grpc_chttp2_transport_parsing *transport_parsing, @@ -77,7 +136,8 @@ grpc_chttp2_parse_error grpc_chttp2_data_parser_parse( gpr_uint8 *const end = GPR_SLICE_END_PTR(slice); gpr_uint8 *cur = beg; grpc_chttp2_data_parser *p = parser; - gpr_uint32 message_flags = 0; + gpr_uint32 message_flags; + grpc_chttp2_incoming_byte_stream *incoming_byte_stream; if (is_last && p->is_last_frame) { stream_parsing->received_close = 1; @@ -132,11 +192,14 @@ grpc_chttp2_parse_error grpc_chttp2_data_parser_parse( p->frame_size |= ((gpr_uint32)*cur); p->state = GRPC_CHTTP2_DATA_FRAME; ++cur; + message_flags = 0; if (p->is_frame_compressed) { message_flags |= GRPC_WRITE_INTERNAL_COMPRESS; } - grpc_sopb_add_begin_message(&p->incoming_sopb, p->frame_size, - message_flags); + p->parsing_frame = incoming_byte_stream = + grpc_chttp2_incoming_byte_stream_create( + transport_parsing, stream_parsing, p->frame_size, message_flags, + &p->incoming_frames); /* fallthrough */ case GRPC_CHTTP2_DATA_FRAME: if (cur == end) { @@ -147,20 +210,23 @@ grpc_chttp2_parse_error grpc_chttp2_data_parser_parse( grpc_chttp2_list_add_parsing_seen_stream(transport_parsing, stream_parsing); if ((gpr_uint32)(end - cur) == p->frame_size) { - grpc_sopb_add_slice( - &p->incoming_sopb, + grpc_chttp2_incoming_byte_stream_push( + exec_ctx, p->parsing_frame, gpr_slice_sub(slice, (size_t)(cur - beg), (size_t)(end - beg))); + grpc_chttp2_incoming_byte_stream_finished(exec_ctx, p->parsing_frame); p->state = GRPC_CHTTP2_DATA_FH_0; return GRPC_CHTTP2_PARSE_OK; } else if ((gpr_uint32)(end - cur) > p->frame_size) { - grpc_sopb_add_slice(&p->incoming_sopb, - gpr_slice_sub(slice, (size_t)(cur - beg), - (size_t)(cur + p->frame_size - beg))); + grpc_chttp2_incoming_byte_stream_push( + exec_ctx, p->parsing_frame, + gpr_slice_sub(slice, (size_t)(cur - beg), + (size_t)(cur + p->frame_size - beg))); + grpc_chttp2_incoming_byte_stream_finished(exec_ctx, p->parsing_frame); cur += p->frame_size; goto fh_0; /* loop */ } else { - grpc_sopb_add_slice( - &p->incoming_sopb, + grpc_chttp2_incoming_byte_stream_push( + exec_ctx, p->parsing_frame, gpr_slice_sub(slice, (size_t)(cur - beg), (size_t)(end - beg))); GPR_ASSERT((size_t)(end - cur) <= p->frame_size); p->frame_size -= (gpr_uint32)(end - cur); diff --git a/src/core/transport/chttp2/frame_data.h b/src/core/transport/chttp2/frame_data.h index 6762484e5b..bc32f29d97 100644 --- a/src/core/transport/chttp2/frame_data.h +++ b/src/core/transport/chttp2/frame_data.h @@ -39,7 +39,7 @@ #include "src/core/iomgr/exec_ctx.h" #include <grpc/support/slice.h> #include <grpc/support/slice_buffer.h> -#include "src/core/transport/stream_op.h" +#include "src/core/transport/byte_stream.h" #include "src/core/transport/chttp2/frame.h" typedef enum { @@ -51,6 +51,14 @@ typedef enum { GRPC_CHTTP2_DATA_FRAME } grpc_chttp2_stream_state; +typedef struct grpc_chttp2_incoming_byte_stream + grpc_chttp2_incoming_byte_stream; + +typedef struct grpc_chttp2_incoming_frame_queue { + grpc_chttp2_incoming_byte_stream *head; + grpc_chttp2_incoming_byte_stream *tail; +} grpc_chttp2_incoming_frame_queue; + typedef struct { grpc_chttp2_stream_state state; gpr_uint8 is_last_frame; @@ -58,9 +66,16 @@ typedef struct { gpr_uint32 frame_size; int is_frame_compressed; - grpc_stream_op_buffer incoming_sopb; + grpc_chttp2_incoming_frame_queue incoming_frames; + grpc_chttp2_incoming_byte_stream *parsing_frame; } grpc_chttp2_data_parser; +void grpc_chttp2_incoming_frame_queue_merge( + grpc_chttp2_incoming_frame_queue *head_dst, + grpc_chttp2_incoming_frame_queue *tail_src); +grpc_byte_stream *grpc_chttp2_incoming_frame_queue_pop( + grpc_chttp2_incoming_frame_queue *q); + /* initialize per-stream state for data frame parsing */ grpc_chttp2_parse_error grpc_chttp2_data_parser_init( grpc_chttp2_data_parser *parser); @@ -81,4 +96,8 @@ grpc_chttp2_parse_error grpc_chttp2_data_parser_parse( /* create a slice with an empty data frame and is_last set */ gpr_slice grpc_chttp2_data_frame_create_empty_close(gpr_uint32 id); +void grpc_chttp2_encode_data(gpr_uint32 id, gpr_slice_buffer *inbuf, + gpr_uint32 write_bytes, int is_eof, + gpr_slice_buffer *outbuf); + #endif /* GRPC_INTERNAL_CORE_TRANSPORT_CHTTP2_FRAME_DATA_H */ diff --git a/src/core/transport/chttp2/frame_window_update.c b/src/core/transport/chttp2/frame_window_update.c index 91bbcfe2c1..74ca29baf9 100644 --- a/src/core/transport/chttp2/frame_window_update.c +++ b/src/core/transport/chttp2/frame_window_update.c @@ -89,7 +89,8 @@ grpc_chttp2_parse_error grpc_chttp2_window_update_parser_parse( } if (p->byte == 4) { - if (p->amount == 0 || (p->amount & 0x80000000u)) { + gpr_uint32 received_update = p->amount; + if (received_update == 0 || (received_update & 0x80000000u)) { gpr_log(GPR_ERROR, "invalid window update bytes: %d", p->amount); return GRPC_CHTTP2_CONNECTION_ERROR; } @@ -97,17 +98,15 @@ grpc_chttp2_parse_error grpc_chttp2_window_update_parser_parse( if (transport_parsing->incoming_stream_id != 0) { if (stream_parsing != NULL) { - GRPC_CHTTP2_FLOWCTL_TRACE_STREAM("update", transport_parsing, - stream_parsing, outgoing_window_update, - p->amount); - stream_parsing->outgoing_window_update += p->amount; + GRPC_CHTTP2_FLOW_CREDIT_STREAM("parse", transport_parsing, + stream_parsing, outgoing_window, + received_update); grpc_chttp2_list_add_parsing_seen_stream(transport_parsing, stream_parsing); } } else { - GRPC_CHTTP2_FLOWCTL_TRACE_TRANSPORT("update", transport_parsing, - outgoing_window_update, p->amount); - transport_parsing->outgoing_window_update += p->amount; + GRPC_CHTTP2_FLOW_CREDIT_TRANSPORT("parse", transport_parsing, + outgoing_window, received_update); } } diff --git a/src/core/transport/chttp2/stream_encoder.c b/src/core/transport/chttp2/hpack_encoder.c index 24a5d958b8..6c7c00b7c3 100644 --- a/src/core/transport/chttp2/stream_encoder.c +++ b/src/core/transport/chttp2/hpack_encoder.c @@ -31,7 +31,7 @@ * */ -#include "src/core/transport/chttp2/stream_encoder.h" +#include "src/core/transport/chttp2/hpack_encoder.h" #include <assert.h> #include <string.h> @@ -54,18 +54,13 @@ /* don't consider adding anything bigger than this to the hpack table */ #define MAX_DECODER_SPACE_USAGE 512 -/* what kind of frame our we encoding? */ -typedef enum { HEADER, DATA, NONE } frame_type; - typedef struct { - frame_type cur_frame_type; + int is_first_frame; /* number of bytes in 'output' when we started the frame - used to calculate frame length */ size_t output_length_at_start_of_frame; /* index (in output) of the header for the current frame */ size_t header_idx; - /* was the last frame emitted a header? (if yes, we'll need a CONTINUATION */ - gpr_uint8 last_was_header; /* have we seen a regular (non-colon-prefixed) header yet? */ gpr_uint8 seen_regular_header; /* output stream id */ @@ -92,58 +87,35 @@ static void fill_header(gpr_uint8 *p, gpr_uint8 type, gpr_uint32 id, size_t len, static void finish_frame(framer_state *st, int is_header_boundary, int is_last_in_stream) { gpr_uint8 type = 0xff; - switch (st->cur_frame_type) { - case HEADER: - type = st->last_was_header ? GRPC_CHTTP2_FRAME_CONTINUATION - : GRPC_CHTTP2_FRAME_HEADER; - st->last_was_header = 1; - break; - case DATA: - type = GRPC_CHTTP2_FRAME_DATA; - st->last_was_header = 0; - is_header_boundary = 0; - break; - case NONE: - return; - } + type = st->is_first_frame ? GRPC_CHTTP2_FRAME_HEADER + : GRPC_CHTTP2_FRAME_CONTINUATION; fill_header( GPR_SLICE_START_PTR(st->output->slices[st->header_idx]), type, st->stream_id, st->output->length - st->output_length_at_start_of_frame, (gpr_uint8)( (is_last_in_stream ? GRPC_CHTTP2_DATA_FLAG_END_STREAM : 0) | (is_header_boundary ? GRPC_CHTTP2_DATA_FLAG_END_HEADERS : 0))); - st->cur_frame_type = NONE; + st->is_first_frame = 0; } /* begin a new frame: reserve off header space, remember how many bytes we'd output before beginning */ -static void begin_frame(framer_state *st, frame_type type) { - GPR_ASSERT(type != NONE); - GPR_ASSERT(st->cur_frame_type == NONE); - st->cur_frame_type = type; +static void begin_frame(framer_state *st) { st->header_idx = gpr_slice_buffer_add_indexed(st->output, gpr_slice_malloc(9)); st->output_length_at_start_of_frame = st->output->length; } -static void begin_new_frame(framer_state *st, frame_type type) { - finish_frame(st, 1, 0); - st->last_was_header = 0; - begin_frame(st, type); -} - /* make sure that the current frame is of the type desired, and has sufficient space to add at least about_to_add bytes -- finishes the current frame if needed */ -static void ensure_frame_type(framer_state *st, frame_type type, - size_t need_bytes) { - if (st->cur_frame_type == type && - st->output->length - st->output_length_at_start_of_frame + need_bytes <= - GRPC_CHTTP2_MAX_PAYLOAD_LENGTH) { +static void ensure_space(framer_state *st, size_t need_bytes) { + if (st->output->length - st->output_length_at_start_of_frame + need_bytes <= + GRPC_CHTTP2_MAX_PAYLOAD_LENGTH) { return; } - finish_frame(st, type != HEADER, 0); - begin_frame(st, type); + finish_frame(st, 0, 0); + begin_frame(st); } /* increment a filter count, halve all counts if one element reaches max */ @@ -165,31 +137,30 @@ static void add_header_data(framer_state *st, gpr_slice slice) { size_t len = GPR_SLICE_LENGTH(slice); size_t remaining; if (len == 0) return; - ensure_frame_type(st, HEADER, 1); remaining = GRPC_CHTTP2_MAX_PAYLOAD_LENGTH + st->output_length_at_start_of_frame - st->output->length; if (len <= remaining) { gpr_slice_buffer_add(st->output, slice); } else { gpr_slice_buffer_add(st->output, gpr_slice_split_head(&slice, remaining)); + finish_frame(st, 0, 0); + begin_frame(st); add_header_data(st, slice); } } static gpr_uint8 *add_tiny_header_data(framer_state *st, size_t len) { - ensure_frame_type(st, HEADER, len); + ensure_space(st, len); return gpr_slice_buffer_tiny_add(st->output, len); } -/* add an element to the decoder table: returns metadata element to unref */ -static grpc_mdelem *add_elem(grpc_chttp2_hpack_compressor *c, - grpc_mdelem *elem) { +/* add an element to the decoder table */ +static void add_elem(grpc_chttp2_hpack_compressor *c, grpc_mdelem *elem) { gpr_uint32 key_hash = elem->key->hash; gpr_uint32 elem_hash = GRPC_MDSTR_KV_HASH(key_hash, elem->value->hash); gpr_uint32 new_index = c->tail_remote_index + c->table_elems + 1; size_t elem_size = 32 + GPR_SLICE_LENGTH(elem->key->slice) + GPR_SLICE_LENGTH(elem->value->slice); - grpc_mdelem *elem_to_unref; GPR_ASSERT(elem_size < 65536); @@ -220,31 +191,27 @@ static grpc_mdelem *add_elem(grpc_chttp2_hpack_compressor *c, if (c->entries_elems[HASH_FRAGMENT_2(elem_hash)] == elem) { /* already there: update with new index */ c->indices_elems[HASH_FRAGMENT_2(elem_hash)] = new_index; - elem_to_unref = elem; } else if (c->entries_elems[HASH_FRAGMENT_3(elem_hash)] == elem) { /* already there (cuckoo): update with new index */ c->indices_elems[HASH_FRAGMENT_3(elem_hash)] = new_index; - elem_to_unref = elem; } else if (c->entries_elems[HASH_FRAGMENT_2(elem_hash)] == NULL) { /* not there, but a free element: add */ - c->entries_elems[HASH_FRAGMENT_2(elem_hash)] = elem; + c->entries_elems[HASH_FRAGMENT_2(elem_hash)] = GRPC_MDELEM_REF(elem); c->indices_elems[HASH_FRAGMENT_2(elem_hash)] = new_index; - elem_to_unref = NULL; } else if (c->entries_elems[HASH_FRAGMENT_3(elem_hash)] == NULL) { /* not there (cuckoo), but a free element: add */ - c->entries_elems[HASH_FRAGMENT_3(elem_hash)] = elem; + c->entries_elems[HASH_FRAGMENT_3(elem_hash)] = GRPC_MDELEM_REF(elem); c->indices_elems[HASH_FRAGMENT_3(elem_hash)] = new_index; - elem_to_unref = NULL; } else if (c->indices_elems[HASH_FRAGMENT_2(elem_hash)] < c->indices_elems[HASH_FRAGMENT_3(elem_hash)]) { /* not there: replace oldest */ - elem_to_unref = c->entries_elems[HASH_FRAGMENT_2(elem_hash)]; - c->entries_elems[HASH_FRAGMENT_2(elem_hash)] = elem; + GRPC_MDELEM_UNREF(c->entries_elems[HASH_FRAGMENT_2(elem_hash)]); + c->entries_elems[HASH_FRAGMENT_2(elem_hash)] = GRPC_MDELEM_REF(elem); c->indices_elems[HASH_FRAGMENT_2(elem_hash)] = new_index; } else { /* not there: replace oldest */ - elem_to_unref = c->entries_elems[HASH_FRAGMENT_3(elem_hash)]; - c->entries_elems[HASH_FRAGMENT_3(elem_hash)] = elem; + GRPC_MDELEM_UNREF(c->entries_elems[HASH_FRAGMENT_3(elem_hash)]); + c->entries_elems[HASH_FRAGMENT_3(elem_hash)] = GRPC_MDELEM_REF(elem); c->indices_elems[HASH_FRAGMENT_3(elem_hash)] = new_index; } @@ -270,8 +237,6 @@ static grpc_mdelem *add_elem(grpc_chttp2_hpack_compressor *c, c->entries_keys[HASH_FRAGMENT_3(key_hash)] = GRPC_MDSTR_REF(elem->key); c->indices_keys[HASH_FRAGMENT_3(key_hash)] = new_index; } - - return elem_to_unref; } static void emit_indexed(grpc_chttp2_hpack_compressor *c, gpr_uint32 elem_index, @@ -370,9 +335,9 @@ static gpr_uint32 dynidx(grpc_chttp2_hpack_compressor *c, c->table_elems - elem_index; } -/* encode an mdelem; returns metadata element to unref */ -static grpc_mdelem *hpack_enc(grpc_chttp2_hpack_compressor *c, - grpc_mdelem *elem, framer_state *st) { +/* encode an mdelem */ +static void hpack_enc(grpc_chttp2_hpack_compressor *c, grpc_mdelem *elem, + framer_state *st) { gpr_uint32 key_hash = elem->key->hash; gpr_uint32 elem_hash = GRPC_MDSTR_KV_HASH(key_hash, elem->value->hash); size_t decoder_space_usage; @@ -397,7 +362,7 @@ static grpc_mdelem *hpack_enc(grpc_chttp2_hpack_compressor *c, /* HIT: complete element (first cuckoo hash) */ emit_indexed(c, dynidx(c, c->indices_elems[HASH_FRAGMENT_2(elem_hash)]), st); - return elem; + return; } if (c->entries_elems[HASH_FRAGMENT_3(elem_hash)] == elem && @@ -405,7 +370,7 @@ static grpc_mdelem *hpack_enc(grpc_chttp2_hpack_compressor *c, /* HIT: complete element (second cuckoo hash) */ emit_indexed(c, dynidx(c, c->indices_elems[HASH_FRAGMENT_3(elem_hash)]), st); - return elem; + return; } /* should this elem be in the table? */ @@ -423,12 +388,13 @@ static grpc_mdelem *hpack_enc(grpc_chttp2_hpack_compressor *c, /* HIT: key (first cuckoo hash) */ if (should_add_elem) { emit_lithdr_incidx(c, dynidx(c, indices_key), elem, st); - return add_elem(c, elem); + add_elem(c, elem); + return; } else { emit_lithdr_noidx(c, dynidx(c, indices_key), elem, st); - return elem; + return; } - GPR_UNREACHABLE_CODE(return NULL); + GPR_UNREACHABLE_CODE(return ); } indices_key = c->indices_keys[HASH_FRAGMENT_3(key_hash)]; @@ -437,24 +403,26 @@ static grpc_mdelem *hpack_enc(grpc_chttp2_hpack_compressor *c, /* HIT: key (first cuckoo hash) */ if (should_add_elem) { emit_lithdr_incidx(c, dynidx(c, indices_key), elem, st); - return add_elem(c, elem); + add_elem(c, elem); + return; } else { emit_lithdr_noidx(c, dynidx(c, indices_key), elem, st); - return elem; + return; } - GPR_UNREACHABLE_CODE(return NULL); + GPR_UNREACHABLE_CODE(return ); } /* no elem, key in the table... fall back to literal emission */ if (should_add_elem) { emit_lithdr_incidx_v(c, elem, st); - return add_elem(c, elem); + add_elem(c, elem); + return; } else { emit_lithdr_noidx_v(c, elem, st); - return elem; + return; } - GPR_UNREACHABLE_CODE(return NULL); + GPR_UNREACHABLE_CODE(return ); } #define STRLEN_LIT(x) (sizeof(x) - 1) @@ -469,8 +437,8 @@ static void deadline_enc(grpc_chttp2_hpack_compressor *c, gpr_timespec deadline, mdelem = grpc_mdelem_from_metadata_strings( c->mdctx, GRPC_MDSTR_REF(c->timeout_key_str), grpc_mdstr_from_string(c->mdctx, timeout_str)); - mdelem = hpack_enc(c, mdelem, st); - if (mdelem) GRPC_MDELEM_UNREF(mdelem); + hpack_enc(c, mdelem, st); + GRPC_MDELEM_UNREF(mdelem); } gpr_slice grpc_chttp2_data_frame_create_empty_close(gpr_uint32 id) { @@ -495,169 +463,34 @@ void grpc_chttp2_hpack_compressor_destroy(grpc_chttp2_hpack_compressor *c) { GRPC_MDSTR_UNREF(c->timeout_key_str); } -gpr_uint32 grpc_chttp2_preencode(grpc_stream_op *inops, size_t *inops_count, - gpr_uint32 max_flow_controlled_bytes, - grpc_stream_op_buffer *outops) { - gpr_slice slice; - grpc_stream_op *op; - gpr_uint32 max_take_size; - gpr_uint32 flow_controlled_bytes_taken = 0; - gpr_uint32 curop = 0; - gpr_uint8 *p; - gpr_uint8 compressed_flag_set = 0; - - while (curop < *inops_count) { - GPR_ASSERT(flow_controlled_bytes_taken <= max_flow_controlled_bytes); - op = &inops[curop]; - switch (op->type) { - case GRPC_NO_OP: - /* skip */ - curop++; - break; - case GRPC_OP_METADATA: - grpc_metadata_batch_assert_ok(&op->data.metadata); - /* these just get copied as they don't impact the number of flow - controlled bytes */ - grpc_sopb_append(outops, op, 1); - curop++; - break; - case GRPC_OP_BEGIN_MESSAGE: - /* begin op: for now we just convert the op to a slice and fall - through - this lets us reuse the slice framing code below */ - compressed_flag_set = - (op->data.begin_message.flags & GRPC_WRITE_INTERNAL_COMPRESS) != 0; - slice = gpr_slice_malloc(5); - - p = GPR_SLICE_START_PTR(slice); - p[0] = compressed_flag_set; - p[1] = (gpr_uint8)(op->data.begin_message.length >> 24); - p[2] = (gpr_uint8)(op->data.begin_message.length >> 16); - p[3] = (gpr_uint8)(op->data.begin_message.length >> 8); - p[4] = (gpr_uint8)(op->data.begin_message.length); - op->type = GRPC_OP_SLICE; - op->data.slice = slice; - /* fallthrough */ - case GRPC_OP_SLICE: - slice = op->data.slice; - if (!GPR_SLICE_LENGTH(slice)) { - /* skip zero length slices */ - gpr_slice_unref(slice); - curop++; - break; - } - max_take_size = max_flow_controlled_bytes - flow_controlled_bytes_taken; - if (max_take_size == 0) { - goto exit_loop; - } - if (GPR_SLICE_LENGTH(slice) > max_take_size) { - slice = gpr_slice_split_head(&op->data.slice, max_take_size); - grpc_sopb_add_slice(outops, slice); - } else { - /* consume this op immediately */ - grpc_sopb_append(outops, op, 1); - curop++; - } - flow_controlled_bytes_taken += (gpr_uint32)GPR_SLICE_LENGTH(slice); - break; - } - } -exit_loop: - *inops_count -= curop; - memmove(inops, inops + curop, *inops_count * sizeof(grpc_stream_op)); - - for (curop = 0; curop < *inops_count; curop++) { - if (inops[curop].type == GRPC_OP_METADATA) { - grpc_metadata_batch_assert_ok(&inops[curop].data.metadata); - } - } - - return flow_controlled_bytes_taken; -} - -void grpc_chttp2_encode(grpc_stream_op *ops, size_t ops_count, int eof, - gpr_uint32 stream_id, - grpc_chttp2_hpack_compressor *compressor, - gpr_slice_buffer *output) { +void grpc_chttp2_encode_header(grpc_chttp2_hpack_compressor *c, + gpr_uint32 stream_id, + grpc_metadata_batch *metadata, int is_eof, + gpr_slice_buffer *outbuf) { framer_state st; - gpr_slice slice; - grpc_stream_op *op; - size_t max_take_size; - gpr_uint32 curop = 0; - gpr_uint32 unref_op; grpc_linked_mdelem *l; - int need_unref = 0; gpr_timespec deadline; GPR_ASSERT(stream_id != 0); - st.cur_frame_type = NONE; - st.last_was_header = 0; st.seen_regular_header = 0; st.stream_id = stream_id; - st.output = output; - - while (curop < ops_count) { - op = &ops[curop]; - switch (op->type) { - case GRPC_NO_OP: - case GRPC_OP_BEGIN_MESSAGE: - gpr_log( - GPR_ERROR, - "These stream ops should be filtered out by grpc_chttp2_preencode"); - abort(); - case GRPC_OP_METADATA: - /* Encode a metadata batch; store the returned values, representing - a metadata element that needs to be unreffed back into the metadata - slot. THIS MAY NOT BE THE SAME ELEMENT (if a decoder table slot got - updated). After this loop, we'll do a batch unref of elements. */ - begin_new_frame(&st, HEADER); - need_unref |= op->data.metadata.garbage.head != NULL; - grpc_metadata_batch_assert_ok(&op->data.metadata); - for (l = op->data.metadata.list.head; l; l = l->next) { - l->md = hpack_enc(compressor, l->md, &st); - need_unref |= l->md != NULL; - } - deadline = op->data.metadata.deadline; - if (gpr_time_cmp(deadline, gpr_inf_future(deadline.clock_type)) != 0) { - deadline_enc(compressor, deadline, &st); - } - curop++; - break; - case GRPC_OP_SLICE: - slice = op->data.slice; - if (st.cur_frame_type == DATA && - st.output->length - st.output_length_at_start_of_frame == - GRPC_CHTTP2_MAX_PAYLOAD_LENGTH) { - finish_frame(&st, 0, 0); - } - ensure_frame_type(&st, DATA, 1); - max_take_size = GRPC_CHTTP2_MAX_PAYLOAD_LENGTH + - st.output_length_at_start_of_frame - st.output->length; - if (GPR_SLICE_LENGTH(slice) > max_take_size) { - slice = gpr_slice_split_head(&op->data.slice, max_take_size); - } else { - /* consume this op immediately */ - curop++; - } - gpr_slice_buffer_add(output, slice); - break; - } + st.output = outbuf; + st.is_first_frame = 1; + + /* Encode a metadata batch; store the returned values, representing + a metadata element that needs to be unreffed back into the metadata + slot. THIS MAY NOT BE THE SAME ELEMENT (if a decoder table slot got + updated). After this loop, we'll do a batch unref of elements. */ + begin_frame(&st); + grpc_metadata_batch_assert_ok(metadata); + for (l = metadata->list.head; l; l = l->next) { + hpack_enc(c, l->md, &st); } - if (eof && st.cur_frame_type == NONE) { - begin_frame(&st, DATA); - } - finish_frame(&st, 1, eof); - - if (need_unref) { - for (unref_op = 0; unref_op < curop; unref_op++) { - op = &ops[unref_op]; - if (op->type != GRPC_OP_METADATA) continue; - for (l = op->data.metadata.list.head; l; l = l->next) { - if (l->md) GRPC_MDELEM_UNREF(l->md); - } - for (l = op->data.metadata.garbage.head; l; l = l->next) { - GRPC_MDELEM_UNREF(l->md); - } - } + deadline = metadata->deadline; + if (gpr_time_cmp(deadline, gpr_inf_future(deadline.clock_type)) != 0) { + deadline_enc(c, deadline, &st); } + + finish_frame(&st, 1, is_eof); } diff --git a/src/core/transport/chttp2/stream_encoder.h b/src/core/transport/chttp2/hpack_encoder.h index db52f2a0f6..59b122dfda 100644 --- a/src/core/transport/chttp2/stream_encoder.h +++ b/src/core/transport/chttp2/hpack_encoder.h @@ -31,12 +31,12 @@ * */ -#ifndef GRPC_INTERNAL_CORE_TRANSPORT_CHTTP2_STREAM_ENCODER_H -#define GRPC_INTERNAL_CORE_TRANSPORT_CHTTP2_STREAM_ENCODER_H +#ifndef GRPC_INTERNAL_CORE_TRANSPORT_CHTTP2_HPACK_ENCODER_H +#define GRPC_INTERNAL_CORE_TRANSPORT_CHTTP2_HPACK_ENCODER_H #include "src/core/transport/chttp2/frame.h" #include "src/core/transport/metadata.h" -#include "src/core/transport/stream_op.h" +#include "src/core/transport/metadata_batch.h" #include <grpc/support/port_platform.h> #include <grpc/support/slice.h> #include <grpc/support/slice_buffer.h> @@ -78,16 +78,8 @@ void grpc_chttp2_hpack_compressor_init(grpc_chttp2_hpack_compressor *c, grpc_mdctx *mdctx); void grpc_chttp2_hpack_compressor_destroy(grpc_chttp2_hpack_compressor *c); -/* select stream ops to be encoded, moving them from inops to outops, and - moving subsequent ops in inops forward in the queue */ -gpr_uint32 grpc_chttp2_preencode(grpc_stream_op *inops, size_t *inops_count, - gpr_uint32 max_flow_controlled_bytes, - grpc_stream_op_buffer *outops); +void grpc_chttp2_encode_header(grpc_chttp2_hpack_compressor *c, gpr_uint32 id, + grpc_metadata_batch *metadata, int is_eof, + gpr_slice_buffer *outbuf); -/* encode stream ops to output */ -void grpc_chttp2_encode(grpc_stream_op *ops, size_t ops_count, int eof, - gpr_uint32 stream_id, - grpc_chttp2_hpack_compressor *compressor, - gpr_slice_buffer *output); - -#endif /* GRPC_INTERNAL_CORE_TRANSPORT_CHTTP2_STREAM_ENCODER_H */ +#endif /* GRPC_INTERNAL_CORE_TRANSPORT_CHTTP2_HPACK_ENCODER_H */ diff --git a/src/core/transport/chttp2/hpack_parser.c b/src/core/transport/chttp2/hpack_parser.c index 20d8312d54..6eebfc3ce4 100644 --- a/src/core/transport/chttp2/hpack_parser.c +++ b/src/core/transport/chttp2/hpack_parser.c @@ -38,13 +38,15 @@ #include <string.h> #include <assert.h> -#include "src/core/transport/chttp2/bin_encoder.h" -#include "src/core/support/string.h" #include <grpc/support/alloc.h> #include <grpc/support/log.h> #include <grpc/support/port_platform.h> #include <grpc/support/useful.h> +#include "src/core/profiling/timers.h" +#include "src/core/support/string.h" +#include "src/core/transport/chttp2/bin_encoder.h" + typedef enum { NOT_BINARY, B64_BYTE0, @@ -1379,20 +1381,23 @@ grpc_chttp2_parse_error grpc_chttp2_header_parser_parse( grpc_chttp2_transport_parsing *transport_parsing, grpc_chttp2_stream_parsing *stream_parsing, gpr_slice slice, int is_last) { grpc_chttp2_hpack_parser *parser = hpack_parser; + GPR_TIMER_BEGIN("grpc_chttp2_hpack_parser_parse", 0); if (!grpc_chttp2_hpack_parser_parse(parser, GPR_SLICE_START_PTR(slice), GPR_SLICE_END_PTR(slice))) { + GPR_TIMER_END("grpc_chttp2_hpack_parser_parse", 0); return GRPC_CHTTP2_CONNECTION_ERROR; } if (is_last) { if (parser->is_boundary && parser->state != parse_begin) { gpr_log(GPR_ERROR, "end of header frame not aligned with a hpack record boundary"); + GPR_TIMER_END("grpc_chttp2_hpack_parser_parse", 0); return GRPC_CHTTP2_CONNECTION_ERROR; } if (parser->is_boundary) { - grpc_chttp2_incoming_metadata_buffer_place_metadata_batch_into( - &stream_parsing->incoming_metadata, - &stream_parsing->data_parser.incoming_sopb); + stream_parsing + ->got_metadata_on_parse[stream_parsing->header_frames_received] = 1; + stream_parsing->header_frames_received++; grpc_chttp2_list_add_parsing_seen_stream(transport_parsing, stream_parsing); } @@ -1404,5 +1409,6 @@ grpc_chttp2_parse_error grpc_chttp2_header_parser_parse( parser->is_boundary = 0xde; parser->is_eof = 0xde; } + GPR_TIMER_END("grpc_chttp2_hpack_parser_parse", 0); return GRPC_CHTTP2_PARSE_OK; } diff --git a/src/core/transport/chttp2/incoming_metadata.c b/src/core/transport/chttp2/incoming_metadata.c index 10c64f3356..956afc8e9d 100644 --- a/src/core/transport/chttp2/incoming_metadata.c +++ b/src/core/transport/chttp2/incoming_metadata.c @@ -48,14 +48,27 @@ void grpc_chttp2_incoming_metadata_buffer_init( void grpc_chttp2_incoming_metadata_buffer_destroy( grpc_chttp2_incoming_metadata_buffer *buffer) { size_t i; + if (!buffer->published) { + for (i = 0; i < buffer->count; i++) { + GRPC_MDELEM_UNREF(buffer->elems[i].md); + } + } + gpr_free(buffer->elems); +} + +void grpc_chttp2_incoming_metadata_buffer_reset( + grpc_chttp2_incoming_metadata_buffer *buffer) { + size_t i; + GPR_ASSERT(!buffer->published); for (i = 0; i < buffer->count; i++) { GRPC_MDELEM_UNREF(buffer->elems[i].md); } - gpr_free(buffer->elems); + buffer->count = 0; } void grpc_chttp2_incoming_metadata_buffer_add( grpc_chttp2_incoming_metadata_buffer *buffer, grpc_mdelem *elem) { + GPR_ASSERT(!buffer->published); if (buffer->capacity == buffer->count) { buffer->capacity = GPR_MAX(8, 2 * buffer->capacity); buffer->elems = @@ -66,117 +79,36 @@ void grpc_chttp2_incoming_metadata_buffer_add( void grpc_chttp2_incoming_metadata_buffer_set_deadline( grpc_chttp2_incoming_metadata_buffer *buffer, gpr_timespec deadline) { + GPR_ASSERT(!buffer->published); buffer->deadline = deadline; } -void grpc_chttp2_incoming_metadata_live_op_buffer_end( - grpc_chttp2_incoming_metadata_live_op_buffer *buffer) { - gpr_free(buffer->elems); - buffer->elems = NULL; -} - -void grpc_chttp2_incoming_metadata_buffer_place_metadata_batch_into( - grpc_chttp2_incoming_metadata_buffer *buffer, grpc_stream_op_buffer *sopb) { - grpc_metadata_batch b; - - b.list.head = NULL; - /* Store away the last element of the list, so that in patch_metadata_ops - we can reconstitute the list. - We can't do list building here as later incoming metadata may reallocate - the underlying array. */ - b.list.tail = (void *)(gpr_intptr)buffer->count; - b.garbage.head = b.garbage.tail = NULL; - b.deadline = buffer->deadline; - buffer->deadline = gpr_inf_future(GPR_CLOCK_REALTIME); - - grpc_sopb_add_metadata(sopb, b); -} - void grpc_chttp2_incoming_metadata_buffer_swap( grpc_chttp2_incoming_metadata_buffer *a, grpc_chttp2_incoming_metadata_buffer *b) { + GPR_ASSERT(!a->published); + GPR_ASSERT(!b->published); GPR_SWAP(grpc_chttp2_incoming_metadata_buffer, *a, *b); } -void grpc_incoming_metadata_buffer_move_to_referencing_sopb( - grpc_chttp2_incoming_metadata_buffer *src, - grpc_chttp2_incoming_metadata_buffer *dst, grpc_stream_op_buffer *sopb) { - size_t delta; - size_t i; - dst->deadline = gpr_time_min(src->deadline, dst->deadline); - - if (src->count == 0) { - return; - } - if (dst->count == 0) { - grpc_chttp2_incoming_metadata_buffer_swap(src, dst); - return; - } - delta = dst->count; - if (dst->capacity < src->count + dst->count) { - dst->capacity = GPR_MAX(dst->capacity * 2, src->count + dst->count); - dst->elems = gpr_realloc(dst->elems, dst->capacity * sizeof(*dst->elems)); - } - memcpy(dst->elems + dst->count, src->elems, src->count * sizeof(*src->elems)); - dst->count += src->count; - for (i = 0; i < sopb->nops; i++) { - if (sopb->ops[i].type != GRPC_OP_METADATA) continue; - sopb->ops[i].data.metadata.list.tail = - (void *)(delta + (gpr_uintptr)sopb->ops[i].data.metadata.list.tail); - } - src->count = 0; -} - -void grpc_chttp2_incoming_metadata_buffer_postprocess_sopb_and_begin_live_op( - grpc_chttp2_incoming_metadata_buffer *buffer, grpc_stream_op_buffer *sopb, - grpc_chttp2_incoming_metadata_live_op_buffer *live_op_buffer) { - grpc_stream_op *ops = sopb->ops; - size_t nops = sopb->nops; - size_t i; - size_t j; - size_t mdidx = 0; - size_t last_mdidx; - int found_metadata = 0; - - /* rework the array of metadata into a linked list, making use - of the breadcrumbs we left in metadata batches during - add_metadata_batch */ - for (i = 0; i < nops; i++) { - grpc_stream_op *op = &ops[i]; - if (op->type != GRPC_OP_METADATA) continue; - found_metadata = 1; - /* we left a breadcrumb indicating where the end of this list is, - and since we add sequentially, we know from the end of the last - segment where this segment begins */ - last_mdidx = (size_t)(gpr_intptr)(op->data.metadata.list.tail); - GPR_ASSERT(last_mdidx > mdidx); - GPR_ASSERT(last_mdidx <= buffer->count); - /* turn the array into a doubly linked list */ - op->data.metadata.list.head = &buffer->elems[mdidx]; - op->data.metadata.list.tail = &buffer->elems[last_mdidx - 1]; - for (j = mdidx + 1; j < last_mdidx; j++) { - buffer->elems[j].prev = &buffer->elems[j - 1]; - buffer->elems[j - 1].next = &buffer->elems[j]; +void grpc_chttp2_incoming_metadata_buffer_publish( + grpc_chttp2_incoming_metadata_buffer *buffer, grpc_metadata_batch *batch) { + GPR_ASSERT(!buffer->published); + buffer->published = 1; + if (buffer->count > 0) { + size_t i; + for (i = 1; i < buffer->count; i++) { + buffer->elems[i].prev = &buffer->elems[i - 1]; } - buffer->elems[mdidx].prev = NULL; - buffer->elems[last_mdidx - 1].next = NULL; - /* track where we're up to */ - mdidx = last_mdidx; - } - if (found_metadata) { - live_op_buffer->elems = buffer->elems; - if (mdidx != buffer->count) { - /* we have a partially read metadata batch still in incoming_metadata */ - size_t new_count = buffer->count - mdidx; - size_t copy_bytes = sizeof(*buffer->elems) * new_count; - GPR_ASSERT(mdidx < buffer->count); - buffer->elems = gpr_malloc(copy_bytes); - memcpy(buffer->elems, live_op_buffer->elems + mdidx, copy_bytes); - buffer->count = buffer->capacity = new_count; - } else { - buffer->elems = NULL; - buffer->count = 0; - buffer->capacity = 0; + for (i = 0; i < buffer->count - 1; i++) { + buffer->elems[i].next = &buffer->elems[i + 1]; } + buffer->elems[0].prev = NULL; + buffer->elems[buffer->count - 1].next = NULL; + batch->list.head = &buffer->elems[0]; + batch->list.tail = &buffer->elems[buffer->count - 1]; + } else { + batch->list.head = batch->list.tail = NULL; } + batch->deadline = buffer->deadline; } diff --git a/src/core/transport/chttp2/incoming_metadata.h b/src/core/transport/chttp2/incoming_metadata.h index 2f1de411ba..0e1dabe825 100644 --- a/src/core/transport/chttp2/incoming_metadata.h +++ b/src/core/transport/chttp2/incoming_metadata.h @@ -41,12 +41,9 @@ typedef struct { size_t count; size_t capacity; gpr_timespec deadline; + int published; } grpc_chttp2_incoming_metadata_buffer; -typedef struct { - grpc_linked_mdelem *elems; -} grpc_chttp2_incoming_metadata_live_op_buffer; - /** assumes everything initially zeroed */ void grpc_chttp2_incoming_metadata_buffer_init( grpc_chttp2_incoming_metadata_buffer *buffer); @@ -54,27 +51,12 @@ void grpc_chttp2_incoming_metadata_buffer_destroy( grpc_chttp2_incoming_metadata_buffer *buffer); void grpc_chttp2_incoming_metadata_buffer_reset( grpc_chttp2_incoming_metadata_buffer *buffer); +void grpc_chttp2_incoming_metadata_buffer_publish( + grpc_chttp2_incoming_metadata_buffer *buffer, grpc_metadata_batch *batch); void grpc_chttp2_incoming_metadata_buffer_add( grpc_chttp2_incoming_metadata_buffer *buffer, grpc_mdelem *elem); void grpc_chttp2_incoming_metadata_buffer_set_deadline( grpc_chttp2_incoming_metadata_buffer *buffer, gpr_timespec deadline); -/** extend sopb with a metadata batch; this must be post-processed by - grpc_chttp2_incoming_metadata_buffer_postprocess_sopb before being handed - out of the transport */ -void grpc_chttp2_incoming_metadata_buffer_place_metadata_batch_into( - grpc_chttp2_incoming_metadata_buffer *buffer, grpc_stream_op_buffer *sopb); - -void grpc_incoming_metadata_buffer_move_to_referencing_sopb( - grpc_chttp2_incoming_metadata_buffer *src, - grpc_chttp2_incoming_metadata_buffer *dst, grpc_stream_op_buffer *sopb); - -void grpc_chttp2_incoming_metadata_buffer_postprocess_sopb_and_begin_live_op( - grpc_chttp2_incoming_metadata_buffer *buffer, grpc_stream_op_buffer *sopb, - grpc_chttp2_incoming_metadata_live_op_buffer *live_op_buffer); - -void grpc_chttp2_incoming_metadata_live_op_buffer_end( - grpc_chttp2_incoming_metadata_live_op_buffer *live_op_buffer); - #endif /* GRPC_INTERNAL_CORE_CHTTP2_INCOMING_METADATA_H */ diff --git a/src/core/transport/chttp2/internal.h b/src/core/transport/chttp2/internal.h index b35f8b5d88..b53c9dee0b 100644 --- a/src/core/transport/chttp2/internal.h +++ b/src/core/transport/chttp2/internal.h @@ -34,6 +34,8 @@ #ifndef GRPC_INTERNAL_CORE_CHTTP2_INTERNAL_H #define GRPC_INTERNAL_CORE_CHTTP2_INTERNAL_H +#include <assert.h> + #include "src/core/iomgr/endpoint.h" #include "src/core/transport/chttp2/frame.h" #include "src/core/transport/chttp2/frame_data.h" @@ -42,9 +44,9 @@ #include "src/core/transport/chttp2/frame_rst_stream.h" #include "src/core/transport/chttp2/frame_settings.h" #include "src/core/transport/chttp2/frame_window_update.h" +#include "src/core/transport/chttp2/hpack_encoder.h" #include "src/core/transport/chttp2/hpack_parser.h" #include "src/core/transport/chttp2/incoming_metadata.h" -#include "src/core/transport/chttp2/stream_encoder.h" #include "src/core/transport/chttp2/stream_map.h" #include "src/core/transport/connectivity_state.h" #include "src/core/transport/transport_impl.h" @@ -56,14 +58,14 @@ typedef struct grpc_chttp2_stream grpc_chttp2_stream; happen to them... this enum labels each list */ typedef enum { GRPC_CHTTP2_LIST_ALL_STREAMS, - GRPC_CHTTP2_LIST_READ_WRITE_STATE_CHANGED, + GRPC_CHTTP2_LIST_CHECK_READ_OPS, + GRPC_CHTTP2_LIST_UNANNOUNCED_INCOMING_WINDOW_AVAILABLE, GRPC_CHTTP2_LIST_WRITABLE, GRPC_CHTTP2_LIST_WRITING, GRPC_CHTTP2_LIST_WRITTEN, GRPC_CHTTP2_LIST_PARSING_SEEN, GRPC_CHTTP2_LIST_CLOSED_WAITING_FOR_PARSING, - GRPC_CHTTP2_LIST_CANCELLED_WAITING_FOR_WRITING, - GRPC_CHTTP2_LIST_INCOMING_WINDOW_UPDATED, + GRPC_CHTTP2_LIST_STALLED_BY_TRANSPORT, /** streams that are waiting to start because there are too many concurrent streams on the connection */ GRPC_CHTTP2_LIST_WAITING_FOR_CONCURRENCY, @@ -113,22 +115,6 @@ typedef enum { GRPC_DTS_FRAME } grpc_chttp2_deframe_transport_state; -typedef enum { - GRPC_WRITE_STATE_OPEN, - GRPC_WRITE_STATE_QUEUED_CLOSE, - GRPC_WRITE_STATE_SENT_CLOSE -} grpc_chttp2_write_state; - -/* flags that can be or'd into stream_global::writing_now */ -#define GRPC_CHTTP2_WRITING_DATA 1 -#define GRPC_CHTTP2_WRITING_WINDOW 2 - -typedef enum { - GRPC_DONT_SEND_CLOSED = 0, - GRPC_SEND_CLOSED, - GRPC_SEND_CLOSED_WITH_RST_STREAM -} grpc_chttp2_send_closed; - typedef struct { grpc_chttp2_stream *head; grpc_chttp2_stream *tail; @@ -160,14 +146,28 @@ typedef struct grpc_chttp2_outstanding_ping { struct grpc_chttp2_outstanding_ping *prev; } grpc_chttp2_outstanding_ping; +/* forward declared in frame_data.h */ +struct grpc_chttp2_incoming_byte_stream { + grpc_byte_stream base; + gpr_refcount refs; + struct grpc_chttp2_incoming_byte_stream *next_message; + + grpc_chttp2_transport *transport; + grpc_chttp2_stream *stream; + int is_tail; + gpr_slice_buffer slices; + grpc_closure *on_next; + gpr_slice *next; +}; + typedef struct { /** data to write next write */ gpr_slice_buffer qbuf; /** window available for us to send to peer */ gpr_int64 outgoing_window; - /** window available for peer to send to us - updated after parse */ - gpr_uint32 incoming_window; + /** window available to announce to peer */ + gpr_int64 announce_incoming_window; /** how much window would we like to have for incoming_window */ gpr_uint32 connection_window_target; @@ -209,6 +209,7 @@ typedef struct { gpr_slice_buffer outbuf; /** hpack encoding */ grpc_chttp2_hpack_compressor hpack_compressor; + gpr_int64 outgoing_window; /** is this a client? */ gpr_uint8 is_client; /** callback for when writing is done */ @@ -233,6 +234,7 @@ struct grpc_chttp2_transport_parsing { gpr_slice_buffer qbuf; /* metadata object cache */ grpc_mdstr *str_grpc_timeout; + grpc_mdelem *elem_grpc_status_ok; /** parser for headers */ grpc_chttp2_hpack_parser hpack_parser; /** simple one shot parsers */ @@ -246,8 +248,7 @@ struct grpc_chttp2_transport_parsing { grpc_chttp2_goaway_parser goaway_parser; /** window available for peer to send to us */ - gpr_uint32 incoming_window; - gpr_uint32 incoming_window_delta; + gpr_int64 incoming_window; /** next stream id available at the time of beginning parsing */ gpr_uint32 next_stream_id; @@ -278,7 +279,7 @@ struct grpc_chttp2_transport_parsing { gpr_uint32 goaway_last_stream_index; gpr_slice goaway_text; - gpr_int64 outgoing_window_update; + gpr_int64 outgoing_window; /** pings awaiting responses */ grpc_chttp2_outstanding_ping pings; @@ -345,8 +346,8 @@ struct grpc_chttp2_transport { struct { /* accept stream callback */ - void (*accept_stream)(void *user_data, grpc_transport *transport, - const void *server_data); + 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 */ @@ -358,9 +359,6 @@ typedef struct { /** HTTP2 stream id for this stream, or zero if one has not been assigned */ gpr_uint32 id; - grpc_closure *send_done_closure; - grpc_closure *recv_done_closure; - /** window available for us to send to peer */ gpr_int64 outgoing_window; /** The number of bytes the upper layers have offered to receive. @@ -371,54 +369,66 @@ typedef struct { not yet announced to HTTP2 flow control. As the upper layers offer to read more bytes, this value increases. As we advertise incoming flow control window, this value decreases. */ - gpr_uint32 unannounced_incoming_window; - /** The number of bytes of HTTP2 flow control we have advertised. - As we advertise incoming flow control window, this value increases. - As bytes are read, this value decreases. - Updated after parse. */ - gpr_uint32 incoming_window; - /** stream ops the transport user would like to send */ - grpc_stream_op_buffer *outgoing_sopb; + gpr_uint32 unannounced_incoming_window_for_parse; + gpr_uint32 unannounced_incoming_window_for_writing; + /** things the upper layers would like to send */ + grpc_metadata_batch *send_initial_metadata; + grpc_closure *send_initial_metadata_finished; + grpc_byte_stream *send_message; + grpc_closure *send_message_finished; + grpc_metadata_batch *send_trailing_metadata; + grpc_closure *send_trailing_metadata_finished; + + grpc_metadata_batch *recv_initial_metadata; + grpc_closure *recv_initial_metadata_finished; + grpc_byte_stream **recv_message; + grpc_closure *recv_message_ready; + grpc_metadata_batch *recv_trailing_metadata; + grpc_closure *recv_trailing_metadata_finished; + /** when the application requests writes be closed, the write_closed is 'queued'; when the close is flow controlled into the send path, we are 'sending' it; when the write has been performed it is 'sent' */ - grpc_chttp2_write_state write_state; - /** is this stream closed (boolean) */ + gpr_uint8 write_closed; + /** is this stream reading half-closed (boolean) */ gpr_uint8 read_closed; - /** has this stream been cancelled? (boolean) */ - gpr_uint8 cancelled; - grpc_status_code cancelled_status; - /** have we told the upper layer that this stream is cancelled? */ - gpr_uint8 published_cancelled; + /** is this stream finished closing (and reportably closed) */ + gpr_uint8 finished_close; /** is this stream in the stream map? (boolean) */ gpr_uint8 in_stream_map; - /** bitmask of GRPC_CHTTP2_WRITING_xxx above */ - gpr_uint8 writing_now; - /** has anything been written to this stream? */ - gpr_uint8 written_anything; - - /** stream state already published to the upper layer */ - grpc_stream_state published_state; - /** address to publish next stream state to */ - grpc_stream_state *publish_state; - /** pointer to sop buffer to fill in with new stream ops */ - grpc_stream_op_buffer *publish_sopb; - grpc_stream_op_buffer incoming_sopb; + /** has this stream seen an error? if 1, then pending incoming frames + can be thrown away */ + gpr_uint8 seen_error; - /** incoming metadata */ - grpc_chttp2_incoming_metadata_buffer incoming_metadata; - grpc_chttp2_incoming_metadata_live_op_buffer outstanding_metadata; + gpr_uint8 published_initial_metadata; + gpr_uint8 published_trailing_metadata; + gpr_uint8 faked_trailing_metadata; + + grpc_chttp2_incoming_metadata_buffer received_initial_metadata; + grpc_chttp2_incoming_metadata_buffer received_trailing_metadata; + + grpc_chttp2_incoming_frame_queue incoming_frames; } grpc_chttp2_stream_global; typedef struct { /** HTTP2 stream id for this stream, or zero if one has not been assigned */ gpr_uint32 id; - /** sops that have passed flow control to be written */ - grpc_stream_op_buffer sopb; - /** how strongly should we indicate closure with the next write */ - grpc_chttp2_send_closed send_closed; + gpr_uint8 fetching; + gpr_uint8 sent_initial_metadata; + gpr_uint8 sent_message; + gpr_uint8 sent_trailing_metadata; + gpr_uint8 read_closed; + /** send this initial metadata */ + grpc_metadata_batch *send_initial_metadata; + grpc_byte_stream *send_message; + grpc_metadata_batch *send_trailing_metadata; + gpr_int64 outgoing_window; /** how much window should we announce? */ gpr_uint32 announce_window; + gpr_slice_buffer flow_controlled_buffer; + gpr_slice fetching_slice; + size_t stream_fetched; + grpc_closure finished_fetch; } grpc_chttp2_stream_writing; struct grpc_chttp2_stream_parsing { @@ -428,22 +438,29 @@ struct grpc_chttp2_stream_parsing { gpr_uint8 received_close; /** saw a rst_stream */ gpr_uint8 saw_rst_stream; - /** incoming_window has been reduced by this much during parsing */ - gpr_uint32 incoming_window_delta; + /** how many header frames have we received? */ + gpr_uint8 header_frames_received; + /** which metadata did we get (on this parse) */ + gpr_uint8 got_metadata_on_parse[2]; + /** should we raise the seen_error flag in transport_global */ + gpr_uint8 seen_error; /** window available for peer to send to us */ - gpr_uint32 incoming_window; + gpr_int64 incoming_window; /** parsing state for data frames */ grpc_chttp2_data_parser data_parser; /** reason give to rst_stream */ gpr_uint32 rst_stream_reason; - /* amount of window given */ - gpr_uint64 outgoing_window_update; + /** amount of window given */ + gpr_int64 outgoing_window; + /** number of bytes received - reset at end of parse thread execution */ + gpr_int64 received_bytes; /** incoming metadata */ - grpc_chttp2_incoming_metadata_buffer incoming_metadata; + grpc_chttp2_incoming_metadata_buffer metadata_buffer[2]; }; struct grpc_chttp2_stream { + grpc_stream_refcount *refcount; grpc_chttp2_stream_global global; grpc_chttp2_stream_writing writing; grpc_chttp2_stream_parsing parsing; @@ -504,21 +521,10 @@ void grpc_chttp2_list_remove_writable_stream( grpc_chttp2_transport_global *transport_global, grpc_chttp2_stream_global *stream_global); -void grpc_chttp2_list_add_incoming_window_updated( - grpc_chttp2_transport_global *transport_global, - grpc_chttp2_stream_global *stream_global); -int grpc_chttp2_list_pop_incoming_window_updated( - grpc_chttp2_transport_global *transport_global, - grpc_chttp2_transport_parsing *transport_parsing, - grpc_chttp2_stream_global **stream_global, - grpc_chttp2_stream_parsing **stream_parsing); -void grpc_chttp2_list_remove_incoming_window_updated( - grpc_chttp2_transport_global *transport_global, - grpc_chttp2_stream_global *stream_global); - -void grpc_chttp2_list_add_writing_stream( +/* returns 1 if stream added, 0 if it was already present */ +int grpc_chttp2_list_add_writing_stream( grpc_chttp2_transport_writing *transport_writing, - grpc_chttp2_stream_writing *stream_writing); + grpc_chttp2_stream_writing *stream_writing) GRPC_MUST_USE_RESULT; int grpc_chttp2_list_have_writing_streams( grpc_chttp2_transport_writing *transport_writing); int grpc_chttp2_list_pop_writing_stream( @@ -550,31 +556,44 @@ 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_closed_waiting_for_parsing( +void grpc_chttp2_list_add_check_read_ops( grpc_chttp2_transport_global *transport_global, grpc_chttp2_stream_global *stream_global); -int grpc_chttp2_list_pop_closed_waiting_for_parsing( +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_cancelled_waiting_for_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_add_unannounced_incoming_window_available( grpc_chttp2_transport_global *transport_global, grpc_chttp2_stream_global *stream_global); -int grpc_chttp2_list_pop_cancelled_waiting_for_writing( +void grpc_chttp2_list_remove_unannounced_incoming_window_available( grpc_chttp2_transport_global *transport_global, - grpc_chttp2_stream_global **stream_global); + grpc_chttp2_stream_global *stream_global); +int grpc_chttp2_list_pop_unannounced_incoming_window_available( + grpc_chttp2_transport_global *transport_global, + grpc_chttp2_transport_parsing *transport_parsing, + grpc_chttp2_stream_global **stream_global, + grpc_chttp2_stream_parsing **stream_parsing); -void grpc_chttp2_list_add_read_write_state_changed( +void grpc_chttp2_list_add_closed_waiting_for_parsing( grpc_chttp2_transport_global *transport_global, grpc_chttp2_stream_global *stream_global); -int grpc_chttp2_list_pop_read_write_state_changed( +int grpc_chttp2_list_pop_closed_waiting_for_parsing( grpc_chttp2_transport_global *transport_global, grpc_chttp2_stream_global **stream_global); grpc_chttp2_stream_parsing *grpc_chttp2_parsing_lookup_stream( grpc_chttp2_transport_parsing *transport_parsing, gpr_uint32 id); grpc_chttp2_stream_parsing *grpc_chttp2_parsing_accept_stream( - grpc_chttp2_transport_parsing *transport_parsing, gpr_uint32 id); + grpc_exec_ctx *exec_ctx, grpc_chttp2_transport_parsing *transport_parsing, + gpr_uint32 id); void grpc_chttp2_add_incoming_goaway( grpc_exec_ctx *exec_ctx, grpc_chttp2_transport_global *transport_global, @@ -592,7 +611,10 @@ void grpc_chttp2_for_all_streams( grpc_chttp2_stream_global *stream_global)); void grpc_chttp2_parsing_become_skip_parser( - grpc_chttp2_transport_parsing *transport_parsing); + grpc_exec_ctx *exec_ctx, grpc_chttp2_transport_parsing *transport_parsing); + +void grpc_chttp2_complete_closure_step(grpc_exec_ctx *exec_ctx, + grpc_closure **pclosure, int success); #define GRPC_CHTTP2_CLIENT_CONNECT_STRING "PRI * HTTP/2.0\r\n\r\nSM\r\n\r\n" #define GRPC_CHTTP2_CLIENT_CONNECT_STRLEN \ @@ -607,26 +629,122 @@ extern int grpc_flowctl_trace; else \ stmt -#define GRPC_CHTTP2_FLOWCTL_TRACE_STREAM(reason, transport, context, var, \ - delta) \ - if (!(grpc_flowctl_trace)) { \ - } else { \ - grpc_chttp2_flowctl_trace(__FILE__, __LINE__, reason, #context, #var, \ - transport->is_client, context->id, \ - (gpr_int64)(context->var), (gpr_int64)(delta)); \ - } - -#define GRPC_CHTTP2_FLOWCTL_TRACE_TRANSPORT(reason, context, var, delta) \ - if (!(grpc_flowctl_trace)) { \ - } else { \ - grpc_chttp2_flowctl_trace(__FILE__, __LINE__, reason, #context, #var, \ - context->is_client, 0, \ - (gpr_int64)(context->var), (gpr_int64)(delta)); \ - } - -void grpc_chttp2_flowctl_trace(const char *file, int line, const char *reason, - const char *context, const char *var, - int is_client, gpr_uint32 stream_id, - gpr_int64 current_value, gpr_int64 delta); +typedef enum { + GRPC_CHTTP2_FLOWCTL_MOVE, + GRPC_CHTTP2_FLOWCTL_CREDIT, + GRPC_CHTTP2_FLOWCTL_DEBIT +} grpc_chttp2_flowctl_op; + +#define GRPC_CHTTP2_FLOW_MOVE_COMMON(phase, transport, id1, id2, dst_context, \ + dst_var, src_context, src_var) \ + do { \ + assert(id1 == id2); \ + if (grpc_flowctl_trace) { \ + grpc_chttp2_flowctl_trace( \ + __FILE__, __LINE__, phase, GRPC_CHTTP2_FLOWCTL_MOVE, #dst_context, \ + #dst_var, #src_context, #src_var, transport->is_client, id1, \ + dst_context->dst_var, src_context->src_var); \ + } \ + dst_context->dst_var += src_context->src_var; \ + src_context->src_var = 0; \ + } while (0) + +#define GRPC_CHTTP2_FLOW_MOVE_STREAM(phase, transport, dst_context, dst_var, \ + src_context, src_var) \ + GRPC_CHTTP2_FLOW_MOVE_COMMON(phase, transport, dst_context->id, \ + src_context->id, dst_context, dst_var, \ + src_context, src_var) +#define GRPC_CHTTP2_FLOW_MOVE_TRANSPORT(phase, dst_context, dst_var, \ + src_context, src_var) \ + GRPC_CHTTP2_FLOW_MOVE_COMMON(phase, dst_context, 0, 0, dst_context, dst_var, \ + src_context, src_var) + +#define GRPC_CHTTP2_FLOW_CREDIT_COMMON(phase, transport, id, dst_context, \ + dst_var, amount) \ + do { \ + if (grpc_flowctl_trace) { \ + grpc_chttp2_flowctl_trace(__FILE__, __LINE__, phase, \ + GRPC_CHTTP2_FLOWCTL_CREDIT, #dst_context, \ + #dst_var, NULL, #amount, transport->is_client, \ + id, dst_context->dst_var, amount); \ + } \ + dst_context->dst_var += amount; \ + } while (0) + +#define GRPC_CHTTP2_FLOW_CREDIT_STREAM(phase, transport, dst_context, dst_var, \ + amount) \ + GRPC_CHTTP2_FLOW_CREDIT_COMMON(phase, transport, dst_context->id, \ + dst_context, dst_var, amount) +#define GRPC_CHTTP2_FLOW_CREDIT_TRANSPORT(phase, dst_context, dst_var, amount) \ + GRPC_CHTTP2_FLOW_CREDIT_COMMON(phase, dst_context, 0, dst_context, dst_var, \ + amount) + +#define GRPC_CHTTP2_FLOW_DEBIT_COMMON(phase, transport, id, dst_context, \ + dst_var, amount) \ + do { \ + if (grpc_flowctl_trace) { \ + grpc_chttp2_flowctl_trace(__FILE__, __LINE__, phase, \ + GRPC_CHTTP2_FLOWCTL_DEBIT, #dst_context, \ + #dst_var, NULL, #amount, transport->is_client, \ + id, dst_context->dst_var, amount); \ + } \ + dst_context->dst_var -= amount; \ + } while (0) + +#define GRPC_CHTTP2_FLOW_DEBIT_STREAM(phase, transport, dst_context, dst_var, \ + amount) \ + GRPC_CHTTP2_FLOW_DEBIT_COMMON(phase, transport, dst_context->id, \ + dst_context, dst_var, amount) +#define GRPC_CHTTP2_FLOW_DEBIT_TRANSPORT(phase, dst_context, dst_var, amount) \ + GRPC_CHTTP2_FLOW_DEBIT_COMMON(phase, dst_context, 0, dst_context, dst_var, \ + amount) + +void grpc_chttp2_flowctl_trace(const char *file, int line, const char *phase, + grpc_chttp2_flowctl_op op, const char *context1, + const char *var1, const char *context2, + const char *var2, int is_client, + gpr_uint32 stream_id, gpr_int64 val1, + gpr_int64 val2); + +void grpc_chttp2_fake_status(grpc_exec_ctx *exec_ctx, + grpc_chttp2_transport_global *transport_global, + grpc_chttp2_stream_global *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); +void grpc_chttp2_start_writing(grpc_exec_ctx *exec_ctx, + grpc_chttp2_transport_global *transport_global); + +#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, + 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); +#endif + +grpc_chttp2_incoming_byte_stream *grpc_chttp2_incoming_byte_stream_create( + 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, + grpc_chttp2_incoming_byte_stream *bs, + gpr_slice slice); +void grpc_chttp2_incoming_byte_stream_finished( + grpc_exec_ctx *exec_ctx, grpc_chttp2_incoming_byte_stream *bs); #endif diff --git a/src/core/transport/chttp2/parsing.c b/src/core/transport/chttp2/parsing.c index 5d4d8e70c4..8cef8fbb77 100644 --- a/src/core/transport/chttp2/parsing.c +++ b/src/core/transport/chttp2/parsing.c @@ -42,22 +42,28 @@ #include <grpc/support/alloc.h> #include <grpc/support/log.h> +#include <grpc/support/string_util.h> -static int init_frame_parser(grpc_chttp2_transport_parsing *transport_parsing); +static int init_frame_parser(grpc_exec_ctx *exec_ctx, + grpc_chttp2_transport_parsing *transport_parsing); static int init_header_frame_parser( - grpc_chttp2_transport_parsing *transport_parsing, int is_continuation); + grpc_exec_ctx *exec_ctx, grpc_chttp2_transport_parsing *transport_parsing, + int is_continuation); static int init_data_frame_parser( - grpc_chttp2_transport_parsing *transport_parsing); + grpc_exec_ctx *exec_ctx, grpc_chttp2_transport_parsing *transport_parsing); static int init_rst_stream_parser( - grpc_chttp2_transport_parsing *transport_parsing); + grpc_exec_ctx *exec_ctx, grpc_chttp2_transport_parsing *transport_parsing); static int init_settings_frame_parser( - grpc_chttp2_transport_parsing *transport_parsing); + grpc_exec_ctx *exec_ctx, grpc_chttp2_transport_parsing *transport_parsing); static int init_window_update_frame_parser( - grpc_chttp2_transport_parsing *transport_parsing); -static int init_ping_parser(grpc_chttp2_transport_parsing *transport_parsing); -static int init_goaway_parser(grpc_chttp2_transport_parsing *transport_parsing); + grpc_exec_ctx *exec_ctx, grpc_chttp2_transport_parsing *transport_parsing); +static int init_ping_parser(grpc_exec_ctx *exec_ctx, + grpc_chttp2_transport_parsing *transport_parsing); +static int init_goaway_parser(grpc_exec_ctx *exec_ctx, + grpc_chttp2_transport_parsing *transport_parsing); static int init_skip_frame_parser( - grpc_chttp2_transport_parsing *transport_parsing, int is_header); + grpc_exec_ctx *exec_ctx, grpc_chttp2_transport_parsing *transport_parsing, + int is_header); static int parse_frame_slice(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport_parsing *transport_parsing, @@ -74,23 +80,11 @@ void grpc_chttp2_prepare_to_read( transport_parsing->next_stream_id = transport_global->next_stream_id; /* update the parsing view of incoming window */ - if (transport_parsing->incoming_window != transport_global->incoming_window) { - GRPC_CHTTP2_FLOWCTL_TRACE_TRANSPORT( - "parse", transport_parsing, incoming_window, - (gpr_int64)transport_global->incoming_window - - (gpr_int64)transport_parsing->incoming_window); - transport_parsing->incoming_window = transport_global->incoming_window; - } - while (grpc_chttp2_list_pop_incoming_window_updated( + while (grpc_chttp2_list_pop_unannounced_incoming_window_available( transport_global, transport_parsing, &stream_global, &stream_parsing)) { - stream_parsing->id = stream_global->id; - if (stream_parsing->incoming_window != stream_global->incoming_window) { - GRPC_CHTTP2_FLOWCTL_TRACE_STREAM( - "parse", transport_parsing, stream_parsing, incoming_window, - (gpr_int64)stream_global->incoming_window - - (gpr_int64)stream_parsing->incoming_window); - stream_parsing->incoming_window = stream_global->incoming_window; - } + GRPC_CHTTP2_FLOW_MOVE_STREAM("parse", transport_parsing, stream_parsing, + incoming_window, stream_global, + unannounced_incoming_window_for_parse); } GPR_TIMER_END("grpc_chttp2_prepare_to_read", 0); @@ -101,6 +95,8 @@ void grpc_chttp2_publish_reads( grpc_chttp2_transport_parsing *transport_parsing) { grpc_chttp2_stream_global *stream_global; grpc_chttp2_stream_parsing *stream_parsing; + int was_zero; + int is_zero; /* transport_parsing->last_incoming_stream_id is used as last-grpc_chttp2_stream-id when @@ -144,98 +140,102 @@ void grpc_chttp2_publish_reads( } /* propagate flow control tokens to global state */ - if (transport_parsing->outgoing_window_update) { - GRPC_CHTTP2_FLOWCTL_TRACE_TRANSPORT( - "parsed", transport_global, outgoing_window, - transport_parsing->outgoing_window_update); - GRPC_CHTTP2_FLOWCTL_TRACE_TRANSPORT( - "parsed", transport_parsing, outgoing_window_update, - -(gpr_int64)transport_parsing->outgoing_window_update); - transport_global->outgoing_window += - transport_parsing->outgoing_window_update; - transport_parsing->outgoing_window_update = 0; - } - - if (transport_parsing->incoming_window_delta) { - GRPC_CHTTP2_FLOWCTL_TRACE_TRANSPORT( - "parsed", transport_global, incoming_window, - -(gpr_int64)transport_parsing->incoming_window_delta); - GRPC_CHTTP2_FLOWCTL_TRACE_TRANSPORT( - "parsed", transport_parsing, incoming_window_delta, - -(gpr_int64)transport_parsing->incoming_window_delta); - transport_global->incoming_window -= - transport_parsing->incoming_window_delta; - transport_parsing->incoming_window_delta = 0; + was_zero = transport_global->outgoing_window <= 0; + GRPC_CHTTP2_FLOW_MOVE_TRANSPORT("parsed", transport_global, outgoing_window, + transport_parsing, outgoing_window); + is_zero = transport_global->outgoing_window <= 0; + if (was_zero && !is_zero) { + while (grpc_chttp2_list_pop_stalled_by_transport(transport_global, + &stream_global)) { + grpc_chttp2_list_add_writable_stream(transport_global, stream_global); + } + } + + if (transport_parsing->incoming_window < + transport_global->connection_window_target * 3 / 4) { + gpr_int64 announce_bytes = transport_global->connection_window_target - + transport_parsing->incoming_window; + GRPC_CHTTP2_FLOW_CREDIT_TRANSPORT("parsed", transport_global, + announce_incoming_window, announce_bytes); + GRPC_CHTTP2_FLOW_CREDIT_TRANSPORT("parsed", transport_parsing, + incoming_window, announce_bytes); } /* for each stream that saw an update, fixup global state */ while (grpc_chttp2_list_pop_parsing_seen_stream( transport_global, transport_parsing, &stream_global, &stream_parsing)) { - /* update incoming flow control window */ - if (stream_parsing->incoming_window_delta) { - GRPC_CHTTP2_FLOWCTL_TRACE_STREAM( - "parsed", transport_parsing, stream_global, incoming_window, - -(gpr_int64)stream_parsing->incoming_window_delta); - GRPC_CHTTP2_FLOWCTL_TRACE_STREAM( - "parsed", transport_parsing, stream_parsing, incoming_window_delta, - -(gpr_int64)stream_parsing->incoming_window_delta); - GRPC_CHTTP2_FLOWCTL_TRACE_STREAM( - "parsed", transport_parsing, stream_global, max_recv_bytes, - -(gpr_int64)stream_parsing->incoming_window_delta); - stream_global->incoming_window -= stream_parsing->incoming_window_delta; - GPR_ASSERT(stream_global->max_recv_bytes >= - stream_parsing->incoming_window_delta); - stream_global->max_recv_bytes -= stream_parsing->incoming_window_delta; - stream_parsing->incoming_window_delta = 0; - grpc_chttp2_list_add_writable_stream(transport_global, stream_global); + if (stream_parsing->seen_error) { + stream_global->seen_error = 1; + grpc_chttp2_list_add_check_read_ops(transport_global, stream_global); } /* update outgoing flow control window */ - if (stream_parsing->outgoing_window_update) { - int was_zero = stream_global->outgoing_window <= 0; - int is_zero; - GRPC_CHTTP2_FLOWCTL_TRACE_STREAM("parsed", transport_parsing, - stream_global, outgoing_window, - stream_parsing->outgoing_window_update); - GRPC_CHTTP2_FLOWCTL_TRACE_STREAM( - "parsed", transport_parsing, stream_parsing, outgoing_window_update, - -(gpr_int64)stream_parsing->outgoing_window_update); - GPR_ASSERT(stream_parsing->outgoing_window_update <= GPR_UINT32_MAX); - stream_global->outgoing_window += - (gpr_uint32)stream_parsing->outgoing_window_update; - stream_parsing->outgoing_window_update = 0; - is_zero = stream_global->outgoing_window <= 0; - if (was_zero && !is_zero) { - grpc_chttp2_list_add_writable_stream(transport_global, stream_global); - } + was_zero = stream_global->outgoing_window <= 0; + GRPC_CHTTP2_FLOW_MOVE_STREAM("parsed", transport_global, stream_global, + outgoing_window, stream_parsing, + outgoing_window); + is_zero = stream_global->outgoing_window <= 0; + if (was_zero && !is_zero) { + grpc_chttp2_list_add_writable_stream(transport_global, stream_global); } - /* updating closed status */ - if (stream_parsing->received_close) { - stream_global->read_closed = 1; - grpc_chttp2_list_add_read_write_state_changed(transport_global, - stream_global); + stream_global->max_recv_bytes -= (gpr_uint32)GPR_MIN( + stream_global->max_recv_bytes, stream_parsing->received_bytes); + stream_parsing->received_bytes = 0; + + /* publish incoming stream ops */ + if (stream_global->incoming_frames.tail != NULL) { + stream_global->incoming_frames.tail->is_tail = 0; + } + if (stream_parsing->data_parser.incoming_frames.head != NULL) { + grpc_chttp2_list_add_check_read_ops(transport_global, stream_global); + } + grpc_chttp2_incoming_frame_queue_merge( + &stream_global->incoming_frames, + &stream_parsing->data_parser.incoming_frames); + if (stream_global->incoming_frames.tail != NULL) { + stream_global->incoming_frames.tail->is_tail = 1; } + + if (!stream_global->published_initial_metadata && + stream_parsing->got_metadata_on_parse[0]) { + stream_parsing->got_metadata_on_parse[0] = 0; + stream_global->published_initial_metadata = 1; + GPR_SWAP(grpc_chttp2_incoming_metadata_buffer, + stream_parsing->metadata_buffer[0], + stream_global->received_initial_metadata); + grpc_chttp2_list_add_check_read_ops(transport_global, stream_global); + } + if (!stream_global->published_trailing_metadata && + stream_parsing->got_metadata_on_parse[1]) { + stream_parsing->got_metadata_on_parse[1] = 0; + stream_global->published_trailing_metadata = 1; + GPR_SWAP(grpc_chttp2_incoming_metadata_buffer, + stream_parsing->metadata_buffer[1], + stream_global->received_trailing_metadata); + grpc_chttp2_list_add_check_read_ops(transport_global, stream_global); + } + if (stream_parsing->saw_rst_stream) { - stream_global->cancelled = 1; - stream_global->cancelled_status = grpc_chttp2_http2_error_to_grpc_status( - (grpc_chttp2_error_code)stream_parsing->rst_stream_reason); - if (stream_parsing->rst_stream_reason == GRPC_CHTTP2_NO_ERROR) { - stream_global->published_cancelled = 1; + if (stream_parsing->rst_stream_reason != GRPC_CHTTP2_NO_ERROR) { + grpc_status_code status_code = grpc_chttp2_http2_error_to_grpc_status( + (grpc_chttp2_error_code)stream_parsing->rst_stream_reason); + char *status_details; + gpr_slice slice_details; + gpr_asprintf(&status_details, "Received RST_STREAM err=%d", + stream_parsing->rst_stream_reason); + slice_details = gpr_slice_from_copied_string(status_details); + gpr_free(status_details); + grpc_chttp2_fake_status(exec_ctx, transport_global, stream_global, + status_code, &slice_details); } - grpc_chttp2_list_add_read_write_state_changed(transport_global, - stream_global); + grpc_chttp2_mark_stream_closed(exec_ctx, transport_global, stream_global, + 1, 1); } - /* publish incoming stream ops */ - if (stream_parsing->data_parser.incoming_sopb.nops > 0) { - grpc_incoming_metadata_buffer_move_to_referencing_sopb( - &stream_parsing->incoming_metadata, &stream_global->incoming_metadata, - &stream_parsing->data_parser.incoming_sopb); - grpc_sopb_move_to(&stream_parsing->data_parser.incoming_sopb, - &stream_global->incoming_sopb); - grpc_chttp2_list_add_read_write_state_changed(transport_global, - stream_global); + if (stream_parsing->received_close) { + grpc_chttp2_mark_stream_closed(exec_ctx, transport_global, stream_global, + 1, 0); } } } @@ -363,7 +363,7 @@ int grpc_chttp2_perform_read(grpc_exec_ctx *exec_ctx, GPR_ASSERT(cur < end); transport_parsing->incoming_stream_id |= ((gpr_uint32)*cur); transport_parsing->deframe_state = GRPC_DTS_FRAME; - if (!init_frame_parser(transport_parsing)) { + if (!init_frame_parser(exec_ctx, transport_parsing)) { return 0; } if (transport_parsing->incoming_stream_id) { @@ -428,7 +428,8 @@ int grpc_chttp2_perform_read(grpc_exec_ctx *exec_ctx, GPR_UNREACHABLE_CODE(return 0); } -static int init_frame_parser(grpc_chttp2_transport_parsing *transport_parsing) { +static int init_frame_parser(grpc_exec_ctx *exec_ctx, + grpc_chttp2_transport_parsing *transport_parsing) { if (transport_parsing->expect_continuation_stream_id != 0) { if (transport_parsing->incoming_frame_type != GRPC_CHTTP2_FRAME_CONTINUATION) { @@ -445,30 +446,30 @@ static int init_frame_parser(grpc_chttp2_transport_parsing *transport_parsing) { transport_parsing->incoming_stream_id); return 0; } - return init_header_frame_parser(transport_parsing, 1); + return init_header_frame_parser(exec_ctx, transport_parsing, 1); } switch (transport_parsing->incoming_frame_type) { case GRPC_CHTTP2_FRAME_DATA: - return init_data_frame_parser(transport_parsing); + return init_data_frame_parser(exec_ctx, transport_parsing); case GRPC_CHTTP2_FRAME_HEADER: - return init_header_frame_parser(transport_parsing, 0); + return init_header_frame_parser(exec_ctx, transport_parsing, 0); case GRPC_CHTTP2_FRAME_CONTINUATION: gpr_log(GPR_ERROR, "Unexpected CONTINUATION frame"); return 0; case GRPC_CHTTP2_FRAME_RST_STREAM: - return init_rst_stream_parser(transport_parsing); + return init_rst_stream_parser(exec_ctx, transport_parsing); case GRPC_CHTTP2_FRAME_SETTINGS: - return init_settings_frame_parser(transport_parsing); + return init_settings_frame_parser(exec_ctx, transport_parsing); case GRPC_CHTTP2_FRAME_WINDOW_UPDATE: - return init_window_update_frame_parser(transport_parsing); + return init_window_update_frame_parser(exec_ctx, transport_parsing); case GRPC_CHTTP2_FRAME_PING: - return init_ping_parser(transport_parsing); + return init_ping_parser(exec_ctx, transport_parsing); case GRPC_CHTTP2_FRAME_GOAWAY: - return init_goaway_parser(transport_parsing); + return init_goaway_parser(exec_ctx, transport_parsing); default: gpr_log(GPR_ERROR, "Unknown frame type %02x", transport_parsing->incoming_frame_type); - return init_skip_frame_parser(transport_parsing, 0); + return init_skip_frame_parser(exec_ctx, transport_parsing, 0); } } @@ -482,7 +483,8 @@ static grpc_chttp2_parse_error skip_parser( static void skip_header(void *tp, grpc_mdelem *md) { GRPC_MDELEM_UNREF(md); } static int init_skip_frame_parser( - grpc_chttp2_transport_parsing *transport_parsing, int is_header) { + grpc_exec_ctx *exec_ctx, grpc_chttp2_transport_parsing *transport_parsing, + int is_header) { if (is_header) { gpr_uint8 is_eoh = transport_parsing->expect_continuation_stream_id != 0; transport_parsing->parser = grpc_chttp2_header_parser_parse; @@ -499,65 +501,51 @@ static int init_skip_frame_parser( } void grpc_chttp2_parsing_become_skip_parser( - grpc_chttp2_transport_parsing *transport_parsing) { + grpc_exec_ctx *exec_ctx, grpc_chttp2_transport_parsing *transport_parsing) { init_skip_frame_parser( - transport_parsing, + exec_ctx, transport_parsing, transport_parsing->parser == grpc_chttp2_header_parser_parse); } static grpc_chttp2_parse_error update_incoming_window( - grpc_chttp2_transport_parsing *transport_parsing, + grpc_exec_ctx *exec_ctx, grpc_chttp2_transport_parsing *transport_parsing, grpc_chttp2_stream_parsing *stream_parsing) { - if (transport_parsing->incoming_frame_size > - transport_parsing->incoming_window) { + gpr_uint32 incoming_frame_size = transport_parsing->incoming_frame_size; + if (incoming_frame_size > transport_parsing->incoming_window) { gpr_log(GPR_ERROR, "frame of size %d overflows incoming window of %d", transport_parsing->incoming_frame_size, transport_parsing->incoming_window); return GRPC_CHTTP2_CONNECTION_ERROR; } - if (transport_parsing->incoming_frame_size > - stream_parsing->incoming_window) { + if (incoming_frame_size > stream_parsing->incoming_window) { gpr_log(GPR_ERROR, "frame of size %d overflows incoming window of %d", transport_parsing->incoming_frame_size, stream_parsing->incoming_window); return GRPC_CHTTP2_CONNECTION_ERROR; } - GRPC_CHTTP2_FLOWCTL_TRACE_TRANSPORT( - "data", transport_parsing, incoming_window, - -(gpr_int64)transport_parsing->incoming_frame_size); - GRPC_CHTTP2_FLOWCTL_TRACE_TRANSPORT("data", transport_parsing, - incoming_window_delta, - transport_parsing->incoming_frame_size); - GRPC_CHTTP2_FLOWCTL_TRACE_STREAM( - "data", transport_parsing, stream_parsing, incoming_window, - -(gpr_int64)transport_parsing->incoming_frame_size); - GRPC_CHTTP2_FLOWCTL_TRACE_STREAM("data", transport_parsing, stream_parsing, - incoming_window_delta, - transport_parsing->incoming_frame_size); - - transport_parsing->incoming_window -= transport_parsing->incoming_frame_size; - transport_parsing->incoming_window_delta += - transport_parsing->incoming_frame_size; - stream_parsing->incoming_window -= transport_parsing->incoming_frame_size; - stream_parsing->incoming_window_delta += - transport_parsing->incoming_frame_size; + GRPC_CHTTP2_FLOW_DEBIT_TRANSPORT("parse", transport_parsing, incoming_window, + incoming_frame_size); + GRPC_CHTTP2_FLOW_DEBIT_STREAM("parse", transport_parsing, stream_parsing, + incoming_window, incoming_frame_size); + stream_parsing->received_bytes += incoming_frame_size; + grpc_chttp2_list_add_parsing_seen_stream(transport_parsing, stream_parsing); return GRPC_CHTTP2_PARSE_OK; } static int init_data_frame_parser( - grpc_chttp2_transport_parsing *transport_parsing) { + grpc_exec_ctx *exec_ctx, grpc_chttp2_transport_parsing *transport_parsing) { grpc_chttp2_stream_parsing *stream_parsing = grpc_chttp2_parsing_lookup_stream(transport_parsing, transport_parsing->incoming_stream_id); grpc_chttp2_parse_error err = GRPC_CHTTP2_PARSE_OK; if (!stream_parsing || stream_parsing->received_close) - return init_skip_frame_parser(transport_parsing, 0); + return init_skip_frame_parser(exec_ctx, transport_parsing, 0); if (err == GRPC_CHTTP2_PARSE_OK) { - err = update_incoming_window(transport_parsing, stream_parsing); + err = update_incoming_window(exec_ctx, transport_parsing, stream_parsing); } if (err == GRPC_CHTTP2_PARSE_OK) { err = grpc_chttp2_data_parser_begin_frame( @@ -577,7 +565,7 @@ static int init_data_frame_parser( &transport_parsing->qbuf, grpc_chttp2_rst_stream_create(transport_parsing->incoming_stream_id, GRPC_CHTTP2_PROTOCOL_ERROR)); - return init_skip_frame_parser(transport_parsing, 0); + return init_skip_frame_parser(exec_ctx, transport_parsing, 0); case GRPC_CHTTP2_CONNECTION_ERROR: return 0; } @@ -586,11 +574,13 @@ static int init_data_frame_parser( static void free_timeout(void *p) { gpr_free(p); } -static void on_header(void *tp, grpc_mdelem *md) { +static void on_initial_header(void *tp, grpc_mdelem *md) { grpc_chttp2_transport_parsing *transport_parsing = tp; grpc_chttp2_stream_parsing *stream_parsing = transport_parsing->incoming_stream; + GPR_TIMER_BEGIN("on_initial_header", 0); + GPR_ASSERT(stream_parsing); GRPC_CHTTP2_IF_TRACING(gpr_log( @@ -598,6 +588,12 @@ static void on_header(void *tp, grpc_mdelem *md) { transport_parsing->is_client ? "CLI" : "SVR", grpc_mdstr_as_c_string(md->key), grpc_mdstr_as_c_string(md->value))); + if (md->key == transport_parsing->elem_grpc_status_ok->key && + md != transport_parsing->elem_grpc_status_ok) { + /* TODO(ctiller): check for a status like " 0" */ + stream_parsing->seen_error = 1; + } + if (md->key == transport_parsing->str_grpc_timeout) { gpr_timespec *cached_timeout = grpc_mdelem_get_user_data(md, free_timeout); if (!cached_timeout) { @@ -612,24 +608,57 @@ static void on_header(void *tp, grpc_mdelem *md) { grpc_mdelem_set_user_data(md, free_timeout, cached_timeout); } grpc_chttp2_incoming_metadata_buffer_set_deadline( - &stream_parsing->incoming_metadata, + &stream_parsing->metadata_buffer[0], gpr_time_add(gpr_now(GPR_CLOCK_MONOTONIC), *cached_timeout)); GRPC_MDELEM_UNREF(md); } else { - grpc_chttp2_incoming_metadata_buffer_add(&stream_parsing->incoming_metadata, - md); + grpc_chttp2_incoming_metadata_buffer_add( + &stream_parsing->metadata_buffer[0], md); } grpc_chttp2_list_add_parsing_seen_stream(transport_parsing, stream_parsing); + + GPR_TIMER_END("on_initial_header", 0); +} + +static void on_trailing_header(void *tp, grpc_mdelem *md) { + grpc_chttp2_transport_parsing *transport_parsing = tp; + grpc_chttp2_stream_parsing *stream_parsing = + transport_parsing->incoming_stream; + + GPR_TIMER_BEGIN("on_trailing_header", 0); + + GPR_ASSERT(stream_parsing); + + GRPC_CHTTP2_IF_TRACING(gpr_log( + GPR_INFO, "HTTP:%d:TRL:%s: %s: %s", stream_parsing->id, + transport_parsing->is_client ? "CLI" : "SVR", + grpc_mdstr_as_c_string(md->key), grpc_mdstr_as_c_string(md->value))); + + if (md->key == transport_parsing->elem_grpc_status_ok->key && + md != transport_parsing->elem_grpc_status_ok) { + /* TODO(ctiller): check for a status like " 0" */ + stream_parsing->seen_error = 1; + } + + grpc_chttp2_incoming_metadata_buffer_add(&stream_parsing->metadata_buffer[1], + md); + + grpc_chttp2_list_add_parsing_seen_stream(transport_parsing, stream_parsing); + + GPR_TIMER_END("on_trailing_header", 0); } static int init_header_frame_parser( - grpc_chttp2_transport_parsing *transport_parsing, int is_continuation) { + grpc_exec_ctx *exec_ctx, grpc_chttp2_transport_parsing *transport_parsing, + int is_continuation) { gpr_uint8 is_eoh = (transport_parsing->incoming_frame_flags & GRPC_CHTTP2_DATA_FLAG_END_HEADERS) != 0; int via_accept = 0; grpc_chttp2_stream_parsing *stream_parsing; + /* TODO(ctiller): when to increment header_frames_received? */ + if (is_eoh) { transport_parsing->expect_continuation_stream_id = 0; } else { @@ -649,7 +678,7 @@ static int init_header_frame_parser( if (is_continuation) { gpr_log(GPR_ERROR, "grpc_chttp2_stream disbanded before CONTINUATION received"); - return init_skip_frame_parser(transport_parsing, 1); + return init_skip_frame_parser(exec_ctx, transport_parsing, 1); } if (transport_parsing->is_client) { if ((transport_parsing->incoming_stream_id & 1) && @@ -660,7 +689,7 @@ static int init_header_frame_parser( gpr_log(GPR_ERROR, "ignoring new grpc_chttp2_stream creation on client"); } - return init_skip_frame_parser(transport_parsing, 1); + return init_skip_frame_parser(exec_ctx, transport_parsing, 1); } else if (transport_parsing->last_incoming_stream_id > transport_parsing->incoming_stream_id) { gpr_log(GPR_ERROR, @@ -669,19 +698,19 @@ static int init_header_frame_parser( "id=%d, new grpc_chttp2_stream id=%d", transport_parsing->last_incoming_stream_id, transport_parsing->incoming_stream_id); - return init_skip_frame_parser(transport_parsing, 1); + return init_skip_frame_parser(exec_ctx, transport_parsing, 1); } else if ((transport_parsing->incoming_stream_id & 1) == 0) { gpr_log(GPR_ERROR, "ignoring grpc_chttp2_stream with non-client generated index %d", transport_parsing->incoming_stream_id); - return init_skip_frame_parser(transport_parsing, 1); + return init_skip_frame_parser(exec_ctx, transport_parsing, 1); } stream_parsing = transport_parsing->incoming_stream = grpc_chttp2_parsing_accept_stream( - transport_parsing, transport_parsing->incoming_stream_id); + exec_ctx, transport_parsing, transport_parsing->incoming_stream_id); if (stream_parsing == NULL) { gpr_log(GPR_ERROR, "grpc_chttp2_stream not accepted"); - return init_skip_frame_parser(transport_parsing, 1); + return init_skip_frame_parser(exec_ctx, transport_parsing, 1); } via_accept = 1; } else { @@ -691,11 +720,21 @@ static int init_header_frame_parser( if (stream_parsing->received_close) { gpr_log(GPR_ERROR, "skipping already closed grpc_chttp2_stream header"); transport_parsing->incoming_stream = NULL; - return init_skip_frame_parser(transport_parsing, 1); + return init_skip_frame_parser(exec_ctx, transport_parsing, 1); } transport_parsing->parser = grpc_chttp2_header_parser_parse; transport_parsing->parser_data = &transport_parsing->hpack_parser; - transport_parsing->hpack_parser.on_header = on_header; + switch (stream_parsing->header_frames_received) { + case 0: + transport_parsing->hpack_parser.on_header = on_initial_header; + break; + case 1: + transport_parsing->hpack_parser.on_header = on_trailing_header; + break; + case 2: + gpr_log(GPR_ERROR, "too many header frames received"); + return init_skip_frame_parser(exec_ctx, transport_parsing, 1); + } transport_parsing->hpack_parser.on_header_user_data = transport_parsing; transport_parsing->hpack_parser.is_boundary = is_eoh; transport_parsing->hpack_parser.is_eof = @@ -708,7 +747,7 @@ static int init_header_frame_parser( } static int init_window_update_frame_parser( - grpc_chttp2_transport_parsing *transport_parsing) { + grpc_exec_ctx *exec_ctx, grpc_chttp2_transport_parsing *transport_parsing) { int ok = GRPC_CHTTP2_PARSE_OK == grpc_chttp2_window_update_parser_begin_frame( &transport_parsing->simple.window_update, transport_parsing->incoming_frame_size, @@ -722,7 +761,8 @@ static int init_window_update_frame_parser( return ok; } -static int init_ping_parser(grpc_chttp2_transport_parsing *transport_parsing) { +static int init_ping_parser(grpc_exec_ctx *exec_ctx, + grpc_chttp2_transport_parsing *transport_parsing) { int ok = GRPC_CHTTP2_PARSE_OK == grpc_chttp2_ping_parser_begin_frame( &transport_parsing->simple.ping, transport_parsing->incoming_frame_size, @@ -733,7 +773,7 @@ static int init_ping_parser(grpc_chttp2_transport_parsing *transport_parsing) { } static int init_rst_stream_parser( - grpc_chttp2_transport_parsing *transport_parsing) { + grpc_exec_ctx *exec_ctx, grpc_chttp2_transport_parsing *transport_parsing) { int ok = GRPC_CHTTP2_PARSE_OK == grpc_chttp2_rst_stream_parser_begin_frame( &transport_parsing->simple.rst_stream, transport_parsing->incoming_frame_size, @@ -741,7 +781,7 @@ static int init_rst_stream_parser( transport_parsing->incoming_stream = grpc_chttp2_parsing_lookup_stream( transport_parsing, transport_parsing->incoming_stream_id); if (!transport_parsing->incoming_stream) { - return init_skip_frame_parser(transport_parsing, 0); + return init_skip_frame_parser(exec_ctx, transport_parsing, 0); } transport_parsing->parser = grpc_chttp2_rst_stream_parser_parse; transport_parsing->parser_data = &transport_parsing->simple.rst_stream; @@ -749,7 +789,7 @@ static int init_rst_stream_parser( } static int init_goaway_parser( - grpc_chttp2_transport_parsing *transport_parsing) { + grpc_exec_ctx *exec_ctx, grpc_chttp2_transport_parsing *transport_parsing) { int ok = GRPC_CHTTP2_PARSE_OK == grpc_chttp2_goaway_parser_begin_frame( &transport_parsing->goaway_parser, transport_parsing->incoming_frame_size, @@ -760,7 +800,7 @@ static int init_goaway_parser( } static int init_settings_frame_parser( - grpc_chttp2_transport_parsing *transport_parsing) { + grpc_exec_ctx *exec_ctx, grpc_chttp2_transport_parsing *transport_parsing) { int ok; if (transport_parsing->incoming_stream_id != 0) { @@ -806,7 +846,7 @@ static int parse_frame_slice(grpc_exec_ctx *exec_ctx, } return 1; case GRPC_CHTTP2_STREAM_ERROR: - grpc_chttp2_parsing_become_skip_parser(transport_parsing); + grpc_chttp2_parsing_become_skip_parser(exec_ctx, transport_parsing); if (stream_parsing) { stream_parsing->saw_rst_stream = 1; stream_parsing->rst_stream_reason = GRPC_CHTTP2_PROTOCOL_ERROR; diff --git a/src/core/transport/chttp2/stream_lists.c b/src/core/transport/chttp2/stream_lists.c index 781db7b0d6..a81ffcce79 100644 --- a/src/core/transport/chttp2/stream_lists.c +++ b/src/core/transport/chttp2/stream_lists.c @@ -142,12 +142,13 @@ static void stream_list_add_tail(grpc_chttp2_transport *t, s->included[id] = 1; } -static void stream_list_add(grpc_chttp2_transport *t, grpc_chttp2_stream *s, - grpc_chttp2_stream_list_id id) { +static int stream_list_add(grpc_chttp2_transport *t, grpc_chttp2_stream *s, + grpc_chttp2_stream_list_id id) { if (s->included[id]) { - return; + return 0; } stream_list_add_tail(t, s, id); + return 1; } /* wrappers for specializations */ @@ -192,12 +193,12 @@ void grpc_chttp2_list_remove_writable_stream( GRPC_CHTTP2_LIST_WRITABLE); } -void grpc_chttp2_list_add_writing_stream( +int grpc_chttp2_list_add_writing_stream( grpc_chttp2_transport_writing *transport_writing, grpc_chttp2_stream_writing *stream_writing) { - stream_list_add(TRANSPORT_FROM_WRITING(transport_writing), - STREAM_FROM_WRITING(stream_writing), - GRPC_CHTTP2_LIST_WRITING); + return stream_list_add(TRANSPORT_FROM_WRITING(transport_writing), + STREAM_FROM_WRITING(stream_writing), + GRPC_CHTTP2_LIST_WRITING); } int grpc_chttp2_list_have_writing_streams( @@ -241,6 +242,40 @@ int grpc_chttp2_list_pop_written_stream( return r; } +void grpc_chttp2_list_add_unannounced_incoming_window_available( + grpc_chttp2_transport_global *transport_global, + grpc_chttp2_stream_global *stream_global) { + GPR_ASSERT(stream_global->id != 0); + stream_list_add(TRANSPORT_FROM_GLOBAL(transport_global), + STREAM_FROM_GLOBAL(stream_global), + GRPC_CHTTP2_LIST_UNANNOUNCED_INCOMING_WINDOW_AVAILABLE); +} + +void grpc_chttp2_list_remove_unannounced_incoming_window_available( + grpc_chttp2_transport_global *transport_global, + grpc_chttp2_stream_global *stream_global) { + stream_list_maybe_remove( + TRANSPORT_FROM_GLOBAL(transport_global), + STREAM_FROM_GLOBAL(stream_global), + GRPC_CHTTP2_LIST_UNANNOUNCED_INCOMING_WINDOW_AVAILABLE); +} + +int grpc_chttp2_list_pop_unannounced_incoming_window_available( + grpc_chttp2_transport_global *transport_global, + grpc_chttp2_transport_parsing *transport_parsing, + grpc_chttp2_stream_global **stream_global, + grpc_chttp2_stream_parsing **stream_parsing) { + grpc_chttp2_stream *stream; + int r = + stream_list_pop(TRANSPORT_FROM_GLOBAL(transport_global), &stream, + GRPC_CHTTP2_LIST_UNANNOUNCED_INCOMING_WINDOW_AVAILABLE); + if (r != 0) { + *stream_global = &stream->global; + *stream_parsing = &stream->parsing; + } + return r; +} + void grpc_chttp2_list_add_parsing_seen_stream( grpc_chttp2_transport_parsing *transport_parsing, grpc_chttp2_stream_parsing *stream_parsing) { @@ -284,91 +319,60 @@ int grpc_chttp2_list_pop_waiting_for_concurrency( return r; } -void grpc_chttp2_list_add_closed_waiting_for_parsing( +void grpc_chttp2_list_add_check_read_ops( grpc_chttp2_transport_global *transport_global, grpc_chttp2_stream_global *stream_global) { stream_list_add(TRANSPORT_FROM_GLOBAL(transport_global), STREAM_FROM_GLOBAL(stream_global), - GRPC_CHTTP2_LIST_CLOSED_WAITING_FOR_PARSING); + GRPC_CHTTP2_LIST_CHECK_READ_OPS); } -int grpc_chttp2_list_pop_closed_waiting_for_parsing( +int grpc_chttp2_list_pop_check_read_ops( grpc_chttp2_transport_global *transport_global, grpc_chttp2_stream_global **stream_global) { grpc_chttp2_stream *stream; int r = stream_list_pop(TRANSPORT_FROM_GLOBAL(transport_global), &stream, - GRPC_CHTTP2_LIST_CLOSED_WAITING_FOR_PARSING); + GRPC_CHTTP2_LIST_CHECK_READ_OPS); if (r != 0) { *stream_global = &stream->global; } return r; } -void grpc_chttp2_list_add_cancelled_waiting_for_writing( - grpc_chttp2_transport_global *transport_global, - grpc_chttp2_stream_global *stream_global) { - stream_list_add(TRANSPORT_FROM_GLOBAL(transport_global), - STREAM_FROM_GLOBAL(stream_global), - GRPC_CHTTP2_LIST_CANCELLED_WAITING_FOR_WRITING); +void grpc_chttp2_list_add_stalled_by_transport( + grpc_chttp2_transport_writing *transport_writing, + grpc_chttp2_stream_writing *stream_writing) { + stream_list_add(TRANSPORT_FROM_WRITING(transport_writing), + STREAM_FROM_WRITING(stream_writing), + GRPC_CHTTP2_LIST_STALLED_BY_TRANSPORT); } -int grpc_chttp2_list_pop_cancelled_waiting_for_writing( +int grpc_chttp2_list_pop_stalled_by_transport( grpc_chttp2_transport_global *transport_global, grpc_chttp2_stream_global **stream_global) { grpc_chttp2_stream *stream; int r = stream_list_pop(TRANSPORT_FROM_GLOBAL(transport_global), &stream, - GRPC_CHTTP2_LIST_CANCELLED_WAITING_FOR_WRITING); + GRPC_CHTTP2_LIST_STALLED_BY_TRANSPORT); if (r != 0) { *stream_global = &stream->global; } return r; } -void grpc_chttp2_list_add_incoming_window_updated( - grpc_chttp2_transport_global *transport_global, - grpc_chttp2_stream_global *stream_global) { - stream_list_add(TRANSPORT_FROM_GLOBAL(transport_global), - STREAM_FROM_GLOBAL(stream_global), - GRPC_CHTTP2_LIST_INCOMING_WINDOW_UPDATED); -} - -int grpc_chttp2_list_pop_incoming_window_updated( - grpc_chttp2_transport_global *transport_global, - grpc_chttp2_transport_parsing *transport_parsing, - grpc_chttp2_stream_global **stream_global, - grpc_chttp2_stream_parsing **stream_parsing) { - grpc_chttp2_stream *stream; - int r = stream_list_pop(TRANSPORT_FROM_GLOBAL(transport_global), &stream, - GRPC_CHTTP2_LIST_INCOMING_WINDOW_UPDATED); - if (r != 0) { - *stream_global = &stream->global; - *stream_parsing = &stream->parsing; - } - return r; -} - -void grpc_chttp2_list_remove_incoming_window_updated( - grpc_chttp2_transport_global *transport_global, - grpc_chttp2_stream_global *stream_global) { - stream_list_maybe_remove(TRANSPORT_FROM_GLOBAL(transport_global), - STREAM_FROM_GLOBAL(stream_global), - GRPC_CHTTP2_LIST_INCOMING_WINDOW_UPDATED); -} - -void grpc_chttp2_list_add_read_write_state_changed( +void grpc_chttp2_list_add_closed_waiting_for_parsing( grpc_chttp2_transport_global *transport_global, grpc_chttp2_stream_global *stream_global) { stream_list_add(TRANSPORT_FROM_GLOBAL(transport_global), STREAM_FROM_GLOBAL(stream_global), - GRPC_CHTTP2_LIST_READ_WRITE_STATE_CHANGED); + GRPC_CHTTP2_LIST_CLOSED_WAITING_FOR_PARSING); } -int grpc_chttp2_list_pop_read_write_state_changed( +int grpc_chttp2_list_pop_closed_waiting_for_parsing( grpc_chttp2_transport_global *transport_global, grpc_chttp2_stream_global **stream_global) { grpc_chttp2_stream *stream; int r = stream_list_pop(TRANSPORT_FROM_GLOBAL(transport_global), &stream, - GRPC_CHTTP2_LIST_READ_WRITE_STATE_CHANGED); + GRPC_CHTTP2_LIST_CLOSED_WAITING_FOR_PARSING); if (r != 0) { *stream_global = &stream->global; } diff --git a/src/core/transport/chttp2/writing.c b/src/core/transport/chttp2/writing.c index 69ad8854ba..353e476e40 100644 --- a/src/core/transport/chttp2/writing.c +++ b/src/core/transport/chttp2/writing.c @@ -40,15 +40,16 @@ #include "src/core/profiling/timers.h" #include "src/core/transport/chttp2/http2_errors.h" -static void finalize_outbuf(grpc_chttp2_transport_writing *transport_writing); +static void finalize_outbuf(grpc_exec_ctx *exec_ctx, + grpc_chttp2_transport_writing *transport_writing); int grpc_chttp2_unlocking_check_writes( grpc_chttp2_transport_global *transport_global, grpc_chttp2_transport_writing *transport_writing) { grpc_chttp2_stream_global *stream_global; grpc_chttp2_stream_writing *stream_writing; - grpc_chttp2_stream_global *first_reinserted_stream = NULL; - gpr_uint32 window_delta; + + GPR_TIMER_BEGIN("grpc_chttp2_unlocking_check_writes", 0); /* simple writes are queued to qbuf, and flushed here */ gpr_slice_buffer_swap(&transport_global->qbuf, &transport_writing->outbuf); @@ -67,98 +68,103 @@ int grpc_chttp2_unlocking_check_writes( transport_global->sent_local_settings = 1; } + GRPC_CHTTP2_FLOW_MOVE_TRANSPORT("write", transport_writing, outgoing_window, + transport_global, outgoing_window); + /* for each grpc_chttp2_stream that's become writable, frame it's data (according to available window sizes) and add to the output buffer */ while (grpc_chttp2_list_pop_writable_stream( transport_global, transport_writing, &stream_global, &stream_writing)) { - if (stream_global == first_reinserted_stream) { - /* prevent infinite loop */ - grpc_chttp2_list_add_first_writable_stream(transport_global, - stream_global); - break; - } + gpr_uint8 sent_initial_metadata; stream_writing->id = stream_global->id; - stream_writing->send_closed = GRPC_DONT_SEND_CLOSED; - - if (stream_global->outgoing_sopb) { - window_delta = grpc_chttp2_preencode( - stream_global->outgoing_sopb->ops, - &stream_global->outgoing_sopb->nops, - (gpr_uint32)GPR_MIN(GPR_MIN(transport_global->outgoing_window, - stream_global->outgoing_window), - GPR_UINT32_MAX), - &stream_writing->sopb); - GRPC_CHTTP2_FLOWCTL_TRACE_TRANSPORT( - "write", transport_global, outgoing_window, -(gpr_int64)window_delta); - GRPC_CHTTP2_FLOWCTL_TRACE_STREAM("write", transport_global, stream_global, - outgoing_window, - -(gpr_int64)window_delta); - transport_global->outgoing_window -= window_delta; - stream_global->outgoing_window -= window_delta; - - if (stream_global->write_state == GRPC_WRITE_STATE_QUEUED_CLOSE && - stream_global->outgoing_sopb->nops == 0) { - if (!transport_global->is_client && !stream_global->read_closed) { - stream_writing->send_closed = GRPC_SEND_CLOSED_WITH_RST_STREAM; + stream_writing->read_closed = stream_global->read_closed; + + GRPC_CHTTP2_FLOW_MOVE_STREAM("write", transport_writing, stream_writing, + outgoing_window, stream_global, + outgoing_window); + + sent_initial_metadata = stream_writing->sent_initial_metadata; + if (!sent_initial_metadata && stream_global->send_initial_metadata) { + stream_writing->send_initial_metadata = + stream_global->send_initial_metadata; + stream_global->send_initial_metadata = NULL; + if (grpc_chttp2_list_add_writing_stream(transport_writing, + stream_writing)) { + GRPC_CHTTP2_STREAM_REF(stream_global, "chttp2_writing"); + } + sent_initial_metadata = 1; + } + if (sent_initial_metadata) { + if (stream_global->send_message != NULL) { + gpr_slice hdr = gpr_slice_malloc(5); + gpr_uint8 *p = GPR_SLICE_START_PTR(hdr); + gpr_uint32 len = stream_global->send_message->length; + GPR_ASSERT(stream_writing->send_message == NULL); + p[0] = (stream_global->send_message->flags & + GRPC_WRITE_INTERNAL_COMPRESS) != 0; + p[1] = (gpr_uint8)(len >> 24); + p[2] = (gpr_uint8)(len >> 16); + p[3] = (gpr_uint8)(len >> 8); + p[4] = (gpr_uint8)(len); + gpr_slice_buffer_add(&stream_writing->flow_controlled_buffer, hdr); + if (stream_global->send_message->length > 0) { + stream_writing->send_message = stream_global->send_message; } else { - stream_writing->send_closed = GRPC_SEND_CLOSED; + stream_writing->send_message = NULL; } + stream_writing->stream_fetched = 0; + stream_global->send_message = NULL; } - - if (stream_global->outgoing_window > 0 && - stream_global->outgoing_sopb->nops != 0) { - grpc_chttp2_list_add_writable_stream(transport_global, stream_global); - if (first_reinserted_stream == NULL && - transport_global->outgoing_window == 0) { - first_reinserted_stream = stream_global; + if ((stream_writing->send_message != NULL || + stream_writing->flow_controlled_buffer.length > 0) && + stream_writing->outgoing_window > 0) { + if (transport_writing->outgoing_window > 0) { + if (grpc_chttp2_list_add_writing_stream(transport_writing, + stream_writing)) { + GRPC_CHTTP2_STREAM_REF(stream_global, "chttp2_writing"); + } + } else { + grpc_chttp2_list_add_stalled_by_transport(transport_writing, + stream_writing); + } + } + if (stream_global->send_trailing_metadata) { + stream_writing->send_trailing_metadata = + stream_global->send_trailing_metadata; + stream_global->send_trailing_metadata = NULL; + if (grpc_chttp2_list_add_writing_stream(transport_writing, + stream_writing)) { + GRPC_CHTTP2_STREAM_REF(stream_global, "chttp2_writing"); } } } if (!stream_global->read_closed && - stream_global->unannounced_incoming_window > 0) { - GPR_ASSERT(stream_writing->announce_window == 0); - GRPC_CHTTP2_FLOWCTL_TRACE_STREAM( - "write", transport_writing, stream_writing, announce_window, - stream_global->unannounced_incoming_window); - stream_writing->announce_window = - stream_global->unannounced_incoming_window; - GRPC_CHTTP2_FLOWCTL_TRACE_STREAM( - "write", transport_global, stream_global, incoming_window, - stream_global->unannounced_incoming_window); - GRPC_CHTTP2_FLOWCTL_TRACE_STREAM( - "write", transport_global, stream_global, unannounced_incoming_window, - -(gpr_int64)stream_global->unannounced_incoming_window); - stream_global->incoming_window += - stream_global->unannounced_incoming_window; - stream_global->unannounced_incoming_window = 0; - grpc_chttp2_list_add_incoming_window_updated(transport_global, - stream_global); - stream_global->writing_now |= GRPC_CHTTP2_WRITING_WINDOW; - } - if (stream_writing->sopb.nops > 0 || - stream_writing->send_closed != GRPC_DONT_SEND_CLOSED) { - stream_global->writing_now |= GRPC_CHTTP2_WRITING_DATA; - } - if (stream_global->writing_now != 0) { - grpc_chttp2_list_add_writing_stream(transport_writing, stream_writing); + stream_global->unannounced_incoming_window_for_writing > 1024) { + GRPC_CHTTP2_FLOW_MOVE_STREAM("write", transport_global, stream_writing, + announce_window, stream_global, + unannounced_incoming_window_for_writing); + if (grpc_chttp2_list_add_writing_stream(transport_writing, + stream_writing)) { + GRPC_CHTTP2_STREAM_REF(stream_global, "chttp2_writing"); + } } } /* if the grpc_chttp2_transport is ready to send a window update, do so here also; 3/4 is a magic number that will likely get tuned soon */ - if (transport_global->incoming_window < - transport_global->connection_window_target * 3 / 4) { - window_delta = transport_global->connection_window_target - - transport_global->incoming_window; + if (transport_global->announce_incoming_window > 0) { + gpr_uint32 announced = (gpr_uint32)GPR_MIN( + transport_global->announce_incoming_window, GPR_UINT32_MAX); + GRPC_CHTTP2_FLOW_DEBIT_TRANSPORT("write", transport_global, + announce_incoming_window, announced); gpr_slice_buffer_add(&transport_writing->outbuf, - grpc_chttp2_window_update_create(0, window_delta)); - GRPC_CHTTP2_FLOWCTL_TRACE_TRANSPORT("write", transport_global, - incoming_window, window_delta); - transport_global->incoming_window += window_delta; + grpc_chttp2_window_update_create(0, announced)); } + GPR_TIMER_END("grpc_chttp2_unlocking_check_writes", 0); + return transport_writing->outbuf.count > 0 || grpc_chttp2_list_have_writing_streams(transport_writing); } @@ -169,50 +175,146 @@ void grpc_chttp2_perform_writes( GPR_ASSERT(transport_writing->outbuf.count > 0 || grpc_chttp2_list_have_writing_streams(transport_writing)); - finalize_outbuf(transport_writing); + finalize_outbuf(exec_ctx, transport_writing); - GPR_ASSERT(transport_writing->outbuf.count > 0); GPR_ASSERT(endpoint); - grpc_endpoint_write(exec_ctx, endpoint, &transport_writing->outbuf, - &transport_writing->done_cb); + if (transport_writing->outbuf.count > 0) { + grpc_endpoint_write(exec_ctx, endpoint, &transport_writing->outbuf, + &transport_writing->done_cb); + } else { + grpc_exec_ctx_enqueue(exec_ctx, &transport_writing->done_cb, 1); + } } -static void finalize_outbuf(grpc_chttp2_transport_writing *transport_writing) { +static void finalize_outbuf(grpc_exec_ctx *exec_ctx, + grpc_chttp2_transport_writing *transport_writing) { grpc_chttp2_stream_writing *stream_writing; GPR_TIMER_BEGIN("finalize_outbuf", 0); while ( grpc_chttp2_list_pop_writing_stream(transport_writing, &stream_writing)) { - if (stream_writing->sopb.nops > 0 || - stream_writing->send_closed != GRPC_DONT_SEND_CLOSED) { - grpc_chttp2_encode(stream_writing->sopb.ops, stream_writing->sopb.nops, - stream_writing->send_closed != GRPC_DONT_SEND_CLOSED, - stream_writing->id, - &transport_writing->hpack_compressor, - &transport_writing->outbuf); - stream_writing->sopb.nops = 0; + gpr_uint32 max_outgoing = + (gpr_uint32)GPR_MIN(GRPC_CHTTP2_MAX_PAYLOAD_LENGTH, + GPR_MIN(stream_writing->outgoing_window, + transport_writing->outgoing_window)); + /* send initial metadata if it's available */ + if (stream_writing->send_initial_metadata != NULL) { + grpc_chttp2_encode_header( + &transport_writing->hpack_compressor, stream_writing->id, + stream_writing->send_initial_metadata, 0, &transport_writing->outbuf); + stream_writing->send_initial_metadata = NULL; + stream_writing->sent_initial_metadata = 1; } - if (stream_writing->announce_window > 0) { + /* send any window updates */ + if (stream_writing->announce_window > 0 && + stream_writing->send_initial_metadata == NULL) { + gpr_uint32 announce = stream_writing->announce_window; gpr_slice_buffer_add( &transport_writing->outbuf, grpc_chttp2_window_update_create(stream_writing->id, stream_writing->announce_window)); - GRPC_CHTTP2_FLOWCTL_TRACE_STREAM( - "write", transport_writing, stream_writing, announce_window, - -(gpr_int64)stream_writing->announce_window); + GRPC_CHTTP2_FLOW_DEBIT_STREAM("write", transport_writing, stream_writing, + announce_window, announce); stream_writing->announce_window = 0; } - if (stream_writing->send_closed == GRPC_SEND_CLOSED_WITH_RST_STREAM) { - gpr_slice_buffer_add(&transport_writing->outbuf, - grpc_chttp2_rst_stream_create(stream_writing->id, - GRPC_CHTTP2_NO_ERROR)); + /* fetch any body bytes */ + while (!stream_writing->fetching && stream_writing->send_message && + stream_writing->flow_controlled_buffer.length < max_outgoing && + stream_writing->stream_fetched < + stream_writing->send_message->length) { + if (grpc_byte_stream_next(exec_ctx, stream_writing->send_message, + &stream_writing->fetching_slice, max_outgoing, + &stream_writing->finished_fetch)) { + stream_writing->stream_fetched += + GPR_SLICE_LENGTH(stream_writing->fetching_slice); + if (stream_writing->stream_fetched == + stream_writing->send_message->length) { + stream_writing->send_message = NULL; + } + gpr_slice_buffer_add(&stream_writing->flow_controlled_buffer, + stream_writing->fetching_slice); + } else { + stream_writing->fetching = 1; + } + } + /* send any body bytes */ + if (stream_writing->flow_controlled_buffer.length > 0) { + if (max_outgoing > 0) { + gpr_uint32 send_bytes = (gpr_uint32)GPR_MIN( + max_outgoing, stream_writing->flow_controlled_buffer.length); + int is_last_data_frame = + stream_writing->send_message == NULL && + send_bytes == stream_writing->flow_controlled_buffer.length; + int is_last_frame = is_last_data_frame && + stream_writing->send_trailing_metadata != NULL && + grpc_metadata_batch_is_empty( + stream_writing->send_trailing_metadata); + grpc_chttp2_encode_data( + stream_writing->id, &stream_writing->flow_controlled_buffer, + send_bytes, is_last_frame, &transport_writing->outbuf); + GRPC_CHTTP2_FLOW_DEBIT_STREAM("write", transport_writing, + stream_writing, outgoing_window, + send_bytes); + GRPC_CHTTP2_FLOW_DEBIT_TRANSPORT("write", transport_writing, + outgoing_window, send_bytes); + if (is_last_frame) { + stream_writing->send_trailing_metadata = NULL; + stream_writing->sent_trailing_metadata = 1; + } + if (is_last_data_frame) { + GPR_ASSERT(stream_writing->send_message == NULL); + stream_writing->sent_message = 1; + } + } else if (transport_writing->outgoing_window == 0) { + grpc_chttp2_list_add_stalled_by_transport(transport_writing, + stream_writing); + grpc_chttp2_list_add_written_stream(transport_writing, stream_writing); + } + } + /* send trailing metadata if it's available and we're ready for it */ + if (stream_writing->send_message == NULL && + stream_writing->flow_controlled_buffer.length == 0 && + stream_writing->send_trailing_metadata != NULL) { + if (grpc_metadata_batch_is_empty( + stream_writing->send_trailing_metadata)) { + grpc_chttp2_encode_data(stream_writing->id, + &stream_writing->flow_controlled_buffer, 0, 1, + &transport_writing->outbuf); + } else { + grpc_chttp2_encode_header(&transport_writing->hpack_compressor, + stream_writing->id, + stream_writing->send_trailing_metadata, 1, + &transport_writing->outbuf); + } + if (!transport_writing->is_client && !stream_writing->read_closed) { + gpr_slice_buffer_add(&transport_writing->outbuf, + grpc_chttp2_rst_stream_create( + stream_writing->id, GRPC_CHTTP2_NO_ERROR)); + } + stream_writing->send_trailing_metadata = NULL; + stream_writing->sent_trailing_metadata = 1; + } + /* if there's more to write, then loop, otherwise prepare to finish the + * write */ + if ((stream_writing->flow_controlled_buffer.length > 0 || + (stream_writing->send_message && !stream_writing->fetching)) && + stream_writing->outgoing_window > 0) { + if (transport_writing->outgoing_window > 0) { + if (grpc_chttp2_list_add_writing_stream(transport_writing, + stream_writing)) { + /* do nothing - already reffed */ + } + } else { + grpc_chttp2_list_add_stalled_by_transport(transport_writing, + stream_writing); + grpc_chttp2_list_add_written_stream(transport_writing, stream_writing); + } + } else { + grpc_chttp2_list_add_written_stream(transport_writing, stream_writing); } - grpc_chttp2_list_add_written_stream(transport_writing, stream_writing); } - - GPR_TIMER_END("finalize_outbuf", 0); } void grpc_chttp2_cleanup_writing( @@ -223,24 +325,26 @@ void grpc_chttp2_cleanup_writing( while (grpc_chttp2_list_pop_written_stream( transport_global, transport_writing, &stream_global, &stream_writing)) { - GPR_ASSERT(stream_global->writing_now != 0); - if (stream_writing->send_closed != GRPC_DONT_SEND_CLOSED) { - stream_global->write_state = GRPC_WRITE_STATE_SENT_CLOSE; - if (!transport_global->is_client) { - stream_global->read_closed = 1; - } + if (stream_writing->sent_trailing_metadata) { + grpc_chttp2_mark_stream_closed(exec_ctx, transport_global, stream_global, + !transport_global->is_client, 1); } - if (stream_global->writing_now & GRPC_CHTTP2_WRITING_DATA) { - if (stream_global->outgoing_sopb != NULL && - stream_global->outgoing_sopb->nops == 0) { - GPR_ASSERT(stream_global->write_state != GRPC_WRITE_STATE_QUEUED_CLOSE); - stream_global->outgoing_sopb = NULL; - grpc_exec_ctx_enqueue(exec_ctx, stream_global->send_done_closure, 1); - } + if (stream_writing->sent_initial_metadata) { + grpc_chttp2_complete_closure_step( + exec_ctx, &stream_global->send_initial_metadata_finished, 1); + } + if (stream_writing->sent_message) { + GPR_ASSERT(stream_writing->send_message == NULL); + GPR_ASSERT(stream_global->send_message_finished); + grpc_chttp2_complete_closure_step( + exec_ctx, &stream_global->send_message_finished, 1); + stream_writing->sent_message = 0; + } + if (stream_writing->sent_trailing_metadata) { + grpc_chttp2_complete_closure_step( + exec_ctx, &stream_global->send_trailing_metadata_finished, 1); } - stream_global->writing_now = 0; - grpc_chttp2_list_add_read_write_state_changed(transport_global, - stream_global); + GRPC_CHTTP2_STREAM_UNREF(exec_ctx, stream_global, "chttp2_writing"); } gpr_slice_buffer_reset_and_unref(&transport_writing->outbuf); } diff --git a/src/core/transport/chttp2_transport.c b/src/core/transport/chttp2_transport.c index effc3c4b3b..10f52f4923 100644 --- a/src/core/transport/chttp2_transport.c +++ b/src/core/transport/chttp2_transport.c @@ -52,6 +52,7 @@ #include "src/core/transport/transport_impl.h" #define DEFAULT_WINDOW 65535 +#define GRPC_CHTTP2_STREAM_LOOKAHEAD DEFAULT_WINDOW #define DEFAULT_CONNECTION_WINDOW_TARGET (1024 * 1024) #define MAX_WINDOW 0x7fffffffu @@ -75,14 +76,14 @@ int grpc_flowctl_trace = 0; #define STREAM_FROM_GLOBAL(sg) \ ((grpc_chttp2_stream *)((char *)(sg)-offsetof(grpc_chttp2_stream, global))) +#define STREAM_FROM_PARSING(sg) \ + ((grpc_chttp2_stream *)((char *)(sg)-offsetof(grpc_chttp2_stream, parsing))) + static const grpc_transport_vtable vtable; static void lock(grpc_chttp2_transport *t); static void unlock(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t); -static void unlock_check_read_write_state(grpc_exec_ctx *exec_ctx, - grpc_chttp2_transport *t); - /* forward declarations of various callbacks that we'll build closures around */ static void writing_action(grpc_exec_ctx *exec_ctx, void *t, int iomgr_success_ignored); @@ -103,11 +104,13 @@ static void perform_stream_op_locked( grpc_chttp2_stream_global *stream_global, grpc_transport_stream_op *op); /** Cancel a stream: coming from the transport API */ -static void cancel_from_api(grpc_chttp2_transport_global *transport_global, +static void cancel_from_api(grpc_exec_ctx *exec_ctx, + grpc_chttp2_transport_global *transport_global, grpc_chttp2_stream_global *stream_global, grpc_status_code status); -static void close_from_api(grpc_chttp2_transport_global *transport_global, +static void close_from_api(grpc_exec_ctx *exec_ctx, + grpc_chttp2_transport_global *transport_global, grpc_chttp2_stream_global *stream_global, grpc_status_code status, gpr_slice *optional_message); @@ -128,6 +131,9 @@ static void connectivity_state_set( grpc_exec_ctx *exec_ctx, grpc_chttp2_transport_global *transport_global, grpc_connectivity_state state, const char *reason); +static void check_read_ops(grpc_exec_ctx *exec_ctx, + grpc_chttp2_transport_global *transport_global); + /* * CONSTRUCTION/DESTRUCTION/REFCOUNTING */ @@ -151,6 +157,7 @@ static void destruct_transport(grpc_exec_ctx *exec_ctx, grpc_chttp2_goaway_parser_destroy(&t->parsing.goaway_parser); GRPC_MDSTR_UNREF(t->parsing.str_grpc_timeout); + GRPC_MDELEM_UNREF(t->parsing.elem_grpc_status_ok); for (i = 0; i < STREAM_LIST_COUNT; i++) { GPR_ASSERT(t->lists[i].head == NULL); @@ -236,14 +243,16 @@ static void init_transport(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t, t->endpoint_reading = 1; t->global.next_stream_id = is_client ? 1 : 2; t->global.is_client = is_client; - t->global.outgoing_window = DEFAULT_WINDOW; - t->global.incoming_window = DEFAULT_WINDOW; + t->writing.outgoing_window = DEFAULT_WINDOW; + t->parsing.incoming_window = 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->parsing.is_client = is_client; t->parsing.str_grpc_timeout = grpc_mdstr_from_string(t->metadata_context, "grpc-timeout"); + t->parsing.elem_grpc_status_ok = + grpc_mdelem_from_strings(t->metadata_context, "grpc-status", "0"); t->parsing.deframe_state = is_client ? GRPC_DTS_FH_0 : GRPC_DTS_CLIENT_PREFIX_0; t->writing.is_client = is_client; @@ -392,19 +401,46 @@ static void close_transport_locked(grpc_exec_ctx *exec_ctx, } } +#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_unref(grpc_exec_ctx *exec_ctx, + grpc_chttp2_stream_global *stream_global, + const char *reason) { + grpc_stream_unref(exec_ctx, STREAM_FROM_GLOBAL(stream_global)->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_unref(grpc_exec_ctx *exec_ctx, + grpc_chttp2_stream_global *stream_global) { + grpc_stream_unref(exec_ctx, STREAM_FROM_GLOBAL(stream_global)->refcount); +} +#endif + static int init_stream(grpc_exec_ctx *exec_ctx, grpc_transport *gt, - grpc_stream *gs, const void *server_data, - grpc_transport_stream_op *initial_op) { + grpc_stream *gs, grpc_stream_refcount *refcount, + const void *server_data) { grpc_chttp2_transport *t = (grpc_chttp2_transport *)gt; grpc_chttp2_stream *s = (grpc_chttp2_stream *)gs; memset(s, 0, sizeof(*s)); - grpc_chttp2_incoming_metadata_buffer_init(&s->parsing.incoming_metadata); - grpc_chttp2_incoming_metadata_buffer_init(&s->global.incoming_metadata); - grpc_sopb_init(&s->writing.sopb); - grpc_sopb_init(&s->global.incoming_sopb); + s->refcount = refcount; + GRPC_CHTTP2_STREAM_REF(&s->global, "chttp2"); + + grpc_chttp2_incoming_metadata_buffer_init(&s->parsing.metadata_buffer[0]); + grpc_chttp2_incoming_metadata_buffer_init(&s->parsing.metadata_buffer[1]); + grpc_chttp2_incoming_metadata_buffer_init( + &s->global.received_initial_metadata); + grpc_chttp2_incoming_metadata_buffer_init( + &s->global.received_trailing_metadata); grpc_chttp2_data_parser_init(&s->parsing.data_parser); + gpr_slice_buffer_init(&s->writing.flow_controlled_buffer); REF_TRANSPORT(t, "stream"); @@ -413,20 +449,17 @@ static int init_stream(grpc_exec_ctx *exec_ctx, grpc_transport *gt, if (server_data) { GPR_ASSERT(t->parsing_active); s->global.id = (gpr_uint32)(gpr_uintptr)server_data; + s->parsing.id = s->global.id; s->global.outgoing_window = t->global.settings[GRPC_PEER_SETTINGS] [GRPC_CHTTP2_SETTINGS_INITIAL_WINDOW_SIZE]; - s->global.max_recv_bytes = s->parsing.incoming_window = - s->global.incoming_window = - t->global.settings[GRPC_SENT_SETTINGS] - [GRPC_CHTTP2_SETTINGS_INITIAL_WINDOW_SIZE]; + s->parsing.incoming_window = s->global.max_recv_bytes = + 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); s->global.in_stream_map = 1; } - - if (initial_op) - perform_stream_op_locked(exec_ctx, &t->global, &s->global, initial_op); unlock(exec_ctx, t); return 0; @@ -437,10 +470,13 @@ static void destroy_stream(grpc_exec_ctx *exec_ctx, grpc_transport *gt, grpc_chttp2_transport *t = (grpc_chttp2_transport *)gt; grpc_chttp2_stream *s = (grpc_chttp2_stream *)gs; int i; + grpc_byte_stream *bs; + + GPR_TIMER_BEGIN("destroy_stream", 0); gpr_mu_lock(&t->mu); - GPR_ASSERT(s->global.published_state == GRPC_STREAM_CLOSED || + GPR_ASSERT((s->global.write_closed && s->global.read_closed) || s->global.id == 0); GPR_ASSERT(!s->global.in_stream_map); if (grpc_chttp2_unregister_stream(t, s) && t->global.sent_goaway) { @@ -451,8 +487,9 @@ static void destroy_stream(grpc_exec_ctx *exec_ctx, grpc_transport *gt, s->global.id) == NULL); } - grpc_chttp2_list_remove_incoming_window_updated(&t->global, &s->global); grpc_chttp2_list_remove_writable_stream(&t->global, &s->global); + grpc_chttp2_list_remove_unannounced_incoming_window_available(&t->global, + &s->global); gpr_mu_unlock(&t->mu); @@ -464,17 +501,29 @@ static void destroy_stream(grpc_exec_ctx *exec_ctx, grpc_transport *gt, } } - GPR_ASSERT(s->global.outgoing_sopb == NULL); - GPR_ASSERT(s->global.publish_sopb == NULL); - grpc_sopb_destroy(&s->writing.sopb); - grpc_sopb_destroy(&s->global.incoming_sopb); + while ( + (bs = grpc_chttp2_incoming_frame_queue_pop(&s->global.incoming_frames))) { + grpc_byte_stream_destroy(bs); + } + + 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_finished == NULL); + GPR_ASSERT(s->global.recv_message_ready == NULL); + GPR_ASSERT(s->global.recv_trailing_metadata_finished == NULL); grpc_chttp2_data_parser_destroy(&s->parsing.data_parser); - grpc_chttp2_incoming_metadata_buffer_destroy(&s->parsing.incoming_metadata); - grpc_chttp2_incoming_metadata_buffer_destroy(&s->global.incoming_metadata); - grpc_chttp2_incoming_metadata_live_op_buffer_end( - &s->global.outstanding_metadata); + grpc_chttp2_incoming_metadata_buffer_destroy(&s->parsing.metadata_buffer[0]); + grpc_chttp2_incoming_metadata_buffer_destroy(&s->parsing.metadata_buffer[1]); + grpc_chttp2_incoming_metadata_buffer_destroy( + &s->global.received_initial_metadata); + grpc_chttp2_incoming_metadata_buffer_destroy( + &s->global.received_trailing_metadata); + gpr_slice_buffer_destroy(&s->writing.flow_controlled_buffer); UNREF_TRANSPORT(exec_ctx, t, "stream"); + + GPR_TIMER_END("destroy_stream", 0); } grpc_chttp2_stream_parsing *grpc_chttp2_parsing_lookup_stream( @@ -486,12 +535,14 @@ grpc_chttp2_stream_parsing *grpc_chttp2_parsing_lookup_stream( } grpc_chttp2_stream_parsing *grpc_chttp2_parsing_accept_stream( - grpc_chttp2_transport_parsing *transport_parsing, gpr_uint32 id) { + grpc_exec_ctx *exec_ctx, grpc_chttp2_transport_parsing *transport_parsing, + gpr_uint32 id) { grpc_chttp2_stream *accepting; grpc_chttp2_transport *t = TRANSPORT_FROM_PARSING(transport_parsing); GPR_ASSERT(t->accepting_stream == NULL); t->accepting_stream = &accepting; - t->channel_callback.accept_stream(t->channel_callback.accept_stream_user_data, + t->channel_callback.accept_stream(exec_ctx, + t->channel_callback.accept_stream_user_data, &t->base, (void *)(gpr_uintptr)id); t->accepting_stream = NULL; return &accepting->parsing; @@ -511,7 +562,6 @@ static void lock(grpc_chttp2_transport *t) { gpr_mu_lock(&t->mu); } static void unlock(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t) { GPR_TIMER_BEGIN("unlock", 0); - unlock_check_read_write_state(exec_ctx, t); if (!t->writing_active && !t->closed && grpc_chttp2_unlocking_check_writes(&t->global, &t->writing)) { t->writing_active = 1; @@ -519,6 +569,7 @@ static void unlock(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t) { grpc_exec_ctx_enqueue(exec_ctx, &t->writing_action, 1); prevent_endpoint_shutdown(t); } + check_read_ops(exec_ctx, &t->global); gpr_mu_unlock(&t->mu); GPR_TIMER_END("unlock", 0); @@ -558,7 +609,6 @@ void grpc_chttp2_terminate_writing(grpc_exec_ctx *exec_ctx, drop_connection(exec_ctx, t); } - /* cleanup writing related jazz */ grpc_chttp2_cleanup_writing(exec_ctx, &t->global, &t->writing); /* leave the writing flag up on shutdown to prevent further writes in unlock() @@ -598,6 +648,7 @@ void grpc_chttp2_add_incoming_goaway( static void maybe_start_some_streams( grpc_exec_ctx *exec_ctx, grpc_chttp2_transport_global *transport_global) { grpc_chttp2_stream_global *stream_global; + gpr_uint32 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 && @@ -607,13 +658,16 @@ static void maybe_start_some_streams( [GRPC_CHTTP2_SETTINGS_MAX_CONCURRENT_STREAMS] && grpc_chttp2_list_pop_waiting_for_concurrency(transport_global, &stream_global)) { + /* safe since we can't (legally) be parsing this stream yet */ + grpc_chttp2_stream_parsing *stream_parsing = + &STREAM_FROM_GLOBAL(stream_global)->parsing; 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)); GPR_ASSERT(stream_global->id == 0); - stream_global->id = transport_global->next_stream_id; + stream_global->id = stream_parsing->id = transport_global->next_stream_id; transport_global->next_stream_id += 2; if (transport_global->next_stream_id >= MAX_CLIENT_STREAM_ID) { @@ -625,103 +679,166 @@ static void maybe_start_some_streams( stream_global->outgoing_window = transport_global->settings[GRPC_PEER_SETTINGS] [GRPC_CHTTP2_SETTINGS_INITIAL_WINDOW_SIZE]; - stream_global->incoming_window = + stream_parsing->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_global->incoming_window, 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)); stream_global->in_stream_map = 1; transport_global->concurrent_stream_count++; - grpc_chttp2_list_add_incoming_window_updated(transport_global, - stream_global); grpc_chttp2_list_add_writable_stream(transport_global, stream_global); } /* 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)) { - cancel_from_api(transport_global, stream_global, GRPC_STATUS_UNAVAILABLE); + cancel_from_api(exec_ctx, transport_global, stream_global, + GRPC_STATUS_UNAVAILABLE); + } +} + +static grpc_closure *add_closure_barrier(grpc_closure *closure) { + closure->final_data += 2; + return closure; +} + +void grpc_chttp2_complete_closure_step(grpc_exec_ctx *exec_ctx, + grpc_closure **pclosure, int success) { + grpc_closure *closure = *pclosure; + if (closure == NULL) { + return; + } + closure->final_data -= 2; + if (!success) { + closure->final_data |= 1; + } + if (closure->final_data < 2) { + grpc_exec_ctx_enqueue(exec_ctx, closure, closure->final_data == 0); } + *pclosure = NULL; +} + +static int contains_non_ok_status( + grpc_chttp2_transport_global *transport_global, + grpc_metadata_batch *batch) { + grpc_mdelem *ok_elem = + TRANSPORT_FROM_GLOBAL(transport_global)->parsing.elem_grpc_status_ok; + grpc_linked_mdelem *l; + for (l = batch->list.head; l; l = l->next) { + if (l->md->key == ok_elem->key && l->md != ok_elem) { + return 1; + } + } + return 0; } static void perform_stream_op_locked( grpc_exec_ctx *exec_ctx, grpc_chttp2_transport_global *transport_global, grpc_chttp2_stream_global *stream_global, grpc_transport_stream_op *op) { + grpc_closure *on_complete; + GPR_TIMER_BEGIN("perform_stream_op_locked", 0); + + on_complete = op->on_complete; + /* use final_data as a barrier until enqueue time; the inital counter is + dropped at the end of this function */ + on_complete->final_data = 2; + if (op->cancel_with_status != GRPC_STATUS_OK) { - cancel_from_api(transport_global, stream_global, op->cancel_with_status); + cancel_from_api(exec_ctx, transport_global, stream_global, + op->cancel_with_status); } if (op->close_with_status != GRPC_STATUS_OK) { - close_from_api(transport_global, stream_global, op->close_with_status, - op->optional_close_message); - } - - if (op->send_ops) { - GPR_ASSERT(stream_global->outgoing_sopb == NULL); - stream_global->send_done_closure = op->on_done_send; - if (!stream_global->cancelled) { - stream_global->written_anything = 1; - stream_global->outgoing_sopb = op->send_ops; - if (op->is_last_send && - stream_global->write_state == GRPC_WRITE_STATE_OPEN) { - stream_global->write_state = GRPC_WRITE_STATE_QUEUED_CLOSE; - } - if (stream_global->id == 0) { - GRPC_CHTTP2_IF_TRACING(gpr_log( - GPR_DEBUG, - "HTTP:%s: New grpc_chttp2_stream %p waiting for concurrency", - transport_global->is_client ? "CLI" : "SVR", stream_global)); + close_from_api(exec_ctx, transport_global, stream_global, + op->close_with_status, op->optional_close_message); + } + + 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; + if (contains_non_ok_status(transport_global, op->send_initial_metadata)) { + stream_global->seen_error = 1; + grpc_chttp2_list_add_check_read_ops(transport_global, stream_global); + } + 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); - } else if (stream_global->outgoing_window > 0) { + } else { + GPR_ASSERT(stream_global->id != 0); grpc_chttp2_list_add_writable_stream(transport_global, stream_global); } } else { - grpc_sopb_reset(op->send_ops); - grpc_exec_ctx_enqueue(exec_ctx, stream_global->send_done_closure, 0); + grpc_chttp2_complete_closure_step( + exec_ctx, &stream_global->send_initial_metadata_finished, 0); } } - if (op->recv_ops) { - GPR_ASSERT(stream_global->publish_sopb == NULL); - GPR_ASSERT(stream_global->published_state != GRPC_STREAM_CLOSED); - stream_global->recv_done_closure = op->on_done_recv; - stream_global->publish_sopb = op->recv_ops; - stream_global->publish_sopb->nops = 0; - stream_global->publish_state = op->recv_state; - /* clamp max recv bytes */ - op->max_recv_bytes = GPR_MIN(op->max_recv_bytes, GPR_UINT32_MAX); - if (stream_global->max_recv_bytes < op->max_recv_bytes) { - GRPC_CHTTP2_FLOWCTL_TRACE_STREAM( - "op", transport_global, stream_global, max_recv_bytes, - op->max_recv_bytes - stream_global->max_recv_bytes); - GRPC_CHTTP2_FLOWCTL_TRACE_STREAM( - "op", transport_global, stream_global, unannounced_incoming_window, - op->max_recv_bytes - stream_global->max_recv_bytes); - stream_global->unannounced_incoming_window += - (gpr_uint32)op->max_recv_bytes - stream_global->max_recv_bytes; - stream_global->max_recv_bytes = (gpr_uint32)op->max_recv_bytes; + 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) { + grpc_chttp2_complete_closure_step( + exec_ctx, &stream_global->send_message_finished, 0); + } else if (stream_global->id != 0) { + stream_global->send_message = op->send_message; + grpc_chttp2_list_add_writable_stream(transport_global, stream_global); } - grpc_chttp2_incoming_metadata_live_op_buffer_end( - &stream_global->outstanding_metadata); - grpc_chttp2_list_add_read_write_state_changed(transport_global, - stream_global); - if (stream_global->id != 0) { + } + + 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; + if (contains_non_ok_status(transport_global, op->send_trailing_metadata)) { + stream_global->seen_error = 1; + grpc_chttp2_list_add_check_read_ops(transport_global, stream_global); + } + if (stream_global->write_closed) { + grpc_chttp2_complete_closure_step( + exec_ctx, &stream_global->send_trailing_metadata_finished, 0); + } else if (stream_global->id != 0) { + /* TODO(ctiller): check if there's flow control for any outstanding + bytes before going writable */ grpc_chttp2_list_add_writable_stream(transport_global, stream_global); } } - if (op->bind_pollset) { - add_to_pollset_locked(exec_ctx, TRANSPORT_FROM_GLOBAL(transport_global), - op->bind_pollset); + if (op->recv_initial_metadata != NULL) { + GPR_ASSERT(stream_global->recv_initial_metadata_finished == NULL); + stream_global->recv_initial_metadata_finished = + add_closure_barrier(on_complete); + stream_global->recv_initial_metadata = op->recv_initial_metadata; + grpc_chttp2_list_add_check_read_ops(transport_global, stream_global); } - grpc_exec_ctx_enqueue(exec_ctx, op->on_consumed, 1); + 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; + grpc_chttp2_list_add_check_read_ops(transport_global, stream_global); + } + + 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; + grpc_chttp2_list_add_check_read_ops(transport_global, stream_global); + } + + grpc_chttp2_complete_closure_step(exec_ctx, &on_complete, 1); + GPR_TIMER_END("perform_stream_op_locked", 0); } @@ -811,12 +928,49 @@ static void perform_transport_op(grpc_exec_ctx *exec_ctx, grpc_transport *gt, * INPUT PROCESSING */ -static grpc_stream_state compute_state(gpr_uint8 write_closed, - gpr_uint8 read_closed) { - if (write_closed && read_closed) return GRPC_STREAM_CLOSED; - if (write_closed) return GRPC_STREAM_SEND_CLOSED; - if (read_closed) return GRPC_STREAM_RECV_CLOSED; - return GRPC_STREAM_OPEN; +static void check_read_ops(grpc_exec_ctx *exec_ctx, + grpc_chttp2_transport_global *transport_global) { + grpc_chttp2_stream_global *stream_global; + grpc_byte_stream *bs; + while ( + grpc_chttp2_list_pop_check_read_ops(transport_global, &stream_global)) { + if (stream_global->recv_initial_metadata_finished != NULL && + stream_global->published_initial_metadata) { + grpc_chttp2_incoming_metadata_buffer_publish( + &stream_global->received_initial_metadata, + stream_global->recv_initial_metadata); + grpc_chttp2_complete_closure_step( + exec_ctx, &stream_global->recv_initial_metadata_finished, 1); + } + if (stream_global->recv_message_ready != NULL) { + 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_enqueue(exec_ctx, stream_global->recv_message_ready, 1); + stream_global->recv_message_ready = NULL; + } else if (stream_global->published_trailing_metadata) { + *stream_global->recv_message = NULL; + grpc_exec_ctx_enqueue(exec_ctx, stream_global->recv_message_ready, 1); + stream_global->recv_message_ready = NULL; + } + } + if (stream_global->recv_trailing_metadata_finished != NULL && + stream_global->read_closed && stream_global->write_closed) { + while (stream_global->seen_error && + (bs = grpc_chttp2_incoming_frame_queue_pop( + &stream_global->incoming_frames)) != NULL) { + grpc_byte_stream_destroy(bs); + } + if (stream_global->incoming_frames.head == NULL) { + grpc_chttp2_incoming_metadata_buffer_publish( + &stream_global->received_trailing_metadata, + stream_global->recv_trailing_metadata); + grpc_chttp2_complete_closure_step( + exec_ctx, &stream_global->recv_trailing_metadata_finished, 1); + } + } + } } static void remove_stream(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t, @@ -832,7 +986,7 @@ static void remove_stream(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t, s->global.in_stream_map = 0; if (t->parsing.incoming_stream == &s->parsing) { t->parsing.incoming_stream = NULL; - grpc_chttp2_parsing_become_skip_parser(&t->parsing); + grpc_chttp2_parsing_become_skip_parser(exec_ctx, &t->parsing); } if (grpc_chttp2_unregister_stream(t, s) && t->global.sent_goaway) { close_transport_locked(exec_ctx, t); @@ -847,109 +1001,10 @@ static void remove_stream(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t, } } -static void unlock_check_read_write_state(grpc_exec_ctx *exec_ctx, - grpc_chttp2_transport *t) { - grpc_chttp2_transport_global *transport_global = &t->global; - grpc_chttp2_stream_global *stream_global; - grpc_stream_state state; - - if (!t->parsing_active) { - /* if a stream is in the stream map, and gets cancelled, we need to ensure - we are not parsing before continuing the cancellation to keep things in - a sane state */ - while (grpc_chttp2_list_pop_closed_waiting_for_parsing(transport_global, - &stream_global)) { - GPR_ASSERT(stream_global->in_stream_map); - GPR_ASSERT(stream_global->write_state != GRPC_WRITE_STATE_OPEN); - GPR_ASSERT(stream_global->read_closed); - remove_stream(exec_ctx, t, stream_global->id); - grpc_chttp2_list_add_read_write_state_changed(transport_global, - stream_global); - } - } - - if (!t->writing_active) { - while (grpc_chttp2_list_pop_cancelled_waiting_for_writing(transport_global, - &stream_global)) { - grpc_chttp2_list_add_read_write_state_changed(transport_global, - stream_global); - } - } - - while (grpc_chttp2_list_pop_read_write_state_changed(transport_global, - &stream_global)) { - if (stream_global->cancelled) { - if (t->writing_active && - stream_global->write_state != GRPC_WRITE_STATE_SENT_CLOSE) { - grpc_chttp2_list_add_cancelled_waiting_for_writing(transport_global, - stream_global); - } else { - stream_global->write_state = GRPC_WRITE_STATE_SENT_CLOSE; - if (stream_global->outgoing_sopb != NULL) { - grpc_sopb_reset(stream_global->outgoing_sopb); - stream_global->outgoing_sopb = NULL; - grpc_exec_ctx_enqueue(exec_ctx, stream_global->send_done_closure, 1); - } - stream_global->read_closed = 1; - if (!stream_global->published_cancelled) { - char buffer[GPR_LTOA_MIN_BUFSIZE]; - gpr_ltoa(stream_global->cancelled_status, buffer); - grpc_chttp2_incoming_metadata_buffer_add( - &stream_global->incoming_metadata, - grpc_mdelem_from_strings(t->metadata_context, "grpc-status", - buffer)); - grpc_chttp2_incoming_metadata_buffer_place_metadata_batch_into( - &stream_global->incoming_metadata, &stream_global->incoming_sopb); - stream_global->published_cancelled = 1; - } - } - } - if (stream_global->write_state == GRPC_WRITE_STATE_SENT_CLOSE && - stream_global->read_closed && stream_global->in_stream_map) { - if (t->parsing_active) { - grpc_chttp2_list_add_closed_waiting_for_parsing(transport_global, - stream_global); - } else { - remove_stream(exec_ctx, t, stream_global->id); - } - } - if (!stream_global->publish_sopb) { - continue; - } - if (stream_global->writing_now != 0) { - continue; - } - /* FIXME(ctiller): we include in_stream_map in our computation of - whether the stream is write-closed. This is completely bogus, - but has the effect of delaying stream-closed until the stream - is indeed evicted from the stream map, making it safe to delete. - To fix this will require having an edge after stream-closed - indicating that the stream is closed AND safe to delete. */ - state = compute_state( - stream_global->write_state == GRPC_WRITE_STATE_SENT_CLOSE && - !stream_global->in_stream_map, - stream_global->read_closed); - if (stream_global->incoming_sopb.nops == 0 && - state == stream_global->published_state) { - continue; - } - grpc_chttp2_incoming_metadata_buffer_postprocess_sopb_and_begin_live_op( - &stream_global->incoming_metadata, &stream_global->incoming_sopb, - &stream_global->outstanding_metadata); - grpc_sopb_swap(stream_global->publish_sopb, &stream_global->incoming_sopb); - stream_global->published_state = *stream_global->publish_state = state; - grpc_exec_ctx_enqueue(exec_ctx, stream_global->recv_done_closure, 1); - stream_global->recv_done_closure = NULL; - stream_global->publish_sopb = NULL; - stream_global->publish_state = NULL; - } -} - -static void cancel_from_api(grpc_chttp2_transport_global *transport_global, +static void cancel_from_api(grpc_exec_ctx *exec_ctx, + grpc_chttp2_transport_global *transport_global, grpc_chttp2_stream_global *stream_global, grpc_status_code status) { - stream_global->cancelled = 1; - stream_global->cancelled_status = status; if (stream_global->id != 0) { gpr_slice_buffer_add( &transport_global->qbuf, @@ -957,11 +1012,89 @@ static void cancel_from_api(grpc_chttp2_transport_global *transport_global, stream_global->id, (gpr_uint32)grpc_chttp2_grpc_status_to_http2_error(status))); } - grpc_chttp2_list_add_read_write_state_changed(transport_global, - stream_global); + grpc_chttp2_fake_status(exec_ctx, transport_global, stream_global, status, + NULL); + grpc_chttp2_mark_stream_closed(exec_ctx, transport_global, stream_global, 1, + 1); +} + +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) { + if (status != GRPC_STATUS_OK) { + stream_global->seen_error = 1; + grpc_chttp2_list_add_check_read_ops(transport_global, stream_global); + } + if (stream_global->published_trailing_metadata && + stream_global->recv_trailing_metadata_finished != NULL) { + /* last chance replacement: we've received trailing metadata, + but something more important has become available to signal + 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 */ + grpc_chttp2_incoming_metadata_buffer_reset( + &stream_global->received_trailing_metadata); + stream_global->published_trailing_metadata = 0; + } + if (!stream_global->published_trailing_metadata) { + grpc_mdctx *mdctx = + TRANSPORT_FROM_GLOBAL(transport_global)->metadata_context; + char status_string[GPR_LTOA_MIN_BUFSIZE]; + gpr_ltoa(status, status_string); + grpc_chttp2_incoming_metadata_buffer_add( + &stream_global->received_trailing_metadata, + grpc_mdelem_from_strings(mdctx, "grpc-status", status_string)); + if (slice) { + grpc_chttp2_incoming_metadata_buffer_add( + &stream_global->received_trailing_metadata, + grpc_mdelem_from_metadata_strings( + mdctx, grpc_mdstr_from_string(mdctx, "grpc-message"), + grpc_mdstr_from_slice(mdctx, gpr_slice_ref(*slice)))); + } + stream_global->published_trailing_metadata = 1; + grpc_chttp2_list_add_check_read_ops(transport_global, stream_global); + } + if (slice) { + gpr_slice_unref(*slice); + } +} + +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) { + if (stream_global->read_closed && stream_global->write_closed) { + /* already closed */ + return; + } + grpc_chttp2_list_add_check_read_ops(transport_global, stream_global); + if (close_reads && !stream_global->read_closed) { + stream_global->read_closed = 1; + stream_global->published_initial_metadata = 1; + stream_global->published_trailing_metadata = 1; + } + if (close_writes && !stream_global->write_closed) { + stream_global->write_closed = 1; + } + if (stream_global->read_closed && stream_global->write_closed) { + if (stream_global->id != 0 && + TRANSPORT_FROM_GLOBAL(transport_global)->parsing_active) { + grpc_chttp2_list_add_closed_waiting_for_parsing(transport_global, + stream_global); + } else { + if (stream_global->id != 0) { + remove_stream(exec_ctx, TRANSPORT_FROM_GLOBAL(transport_global), + stream_global->id); + } + stream_global->finished_close = 1; + GRPC_CHTTP2_STREAM_UNREF(exec_ctx, stream_global, "chttp2"); + } + } } -static void close_from_api(grpc_chttp2_transport_global *transport_global, +static void close_from_api(grpc_exec_ctx *exec_ctx, + grpc_chttp2_transport_global *transport_global, grpc_chttp2_stream_global *stream_global, grpc_status_code status, gpr_slice *optional_message) { @@ -973,10 +1106,7 @@ static void close_from_api(grpc_chttp2_transport_global *transport_global, GPR_ASSERT(status >= 0 && (int)status < 100); - stream_global->cancelled = 1; - stream_global->cancelled_status = status; GPR_ASSERT(stream_global->id != 0); - GPR_ASSERT(!stream_global->written_anything); /* Hand roll a header block. This is unnecessarily ugly - at some point we should find a more elegant @@ -1059,23 +1189,30 @@ static void close_from_api(grpc_chttp2_transport_global *transport_global, &transport_global->qbuf, grpc_chttp2_rst_stream_create(stream_global->id, GRPC_CHTTP2_NO_ERROR)); - grpc_chttp2_list_add_read_write_state_changed(transport_global, - stream_global); + if (optional_message) { + gpr_slice_ref(*optional_message); + } + grpc_chttp2_fake_status(exec_ctx, transport_global, stream_global, status, + optional_message); + grpc_chttp2_mark_stream_closed(exec_ctx, transport_global, stream_global, 1, + 1); } static void cancel_stream_cb(grpc_chttp2_transport_global *transport_global, void *user_data, grpc_chttp2_stream_global *stream_global) { - cancel_from_api(transport_global, stream_global, GRPC_STATUS_UNAVAILABLE); + cancel_from_api(user_data, transport_global, stream_global, + GRPC_STATUS_UNAVAILABLE); } -static void end_all_the_calls(grpc_chttp2_transport *t) { - grpc_chttp2_for_all_streams(&t->global, NULL, cancel_stream_cb); +static void end_all_the_calls(grpc_exec_ctx *exec_ctx, + grpc_chttp2_transport *t) { + grpc_chttp2_for_all_streams(&t->global, exec_ctx, cancel_stream_cb); } static void drop_connection(grpc_exec_ctx *exec_ctx, grpc_chttp2_transport *t) { close_transport_locked(exec_ctx, t); - end_all_the_calls(t); + end_all_the_calls(exec_ctx, t); } /** update window from a settings change */ @@ -1086,12 +1223,11 @@ static void update_global_window(void *args, gpr_uint32 id, void *stream) { grpc_chttp2_stream_global *stream_global = &s->global; int was_zero; int is_zero; + gpr_int64 initial_window_update = t->parsing.initial_window_update; - GRPC_CHTTP2_FLOWCTL_TRACE_STREAM("settings", transport_global, stream_global, - outgoing_window, - t->parsing.initial_window_update); was_zero = stream_global->outgoing_window <= 0; - stream_global->outgoing_window += t->parsing.initial_window_update; + GRPC_CHTTP2_FLOW_CREDIT_STREAM("settings", transport_global, stream_global, + outgoing_window, initial_window_update); is_zero = stream_global->outgoing_window <= 0; if (was_zero && !is_zero) { @@ -1112,6 +1248,9 @@ static void recv_data(grpc_exec_ctx *exec_ctx, void *tp, int success) { size_t i; int keep_reading = 0; grpc_chttp2_transport *t = tp; + grpc_chttp2_transport_global *transport_global = &t->global; + grpc_chttp2_transport_parsing *transport_parsing = &t->parsing; + grpc_chttp2_stream_global *stream_global; GPR_TIMER_BEGIN("recv_data", 0); @@ -1123,11 +1262,11 @@ static void recv_data(grpc_exec_ctx *exec_ctx, void *tp, int success) { /* merge stream lists */ grpc_chttp2_stream_map_move_into(&t->new_stream_map, &t->parsing_stream_map); - grpc_chttp2_prepare_to_read(&t->global, &t->parsing); + grpc_chttp2_prepare_to_read(transport_global, transport_parsing); gpr_mu_unlock(&t->mu); GPR_TIMER_BEGIN("recv_data.parse", 0); for (; i < t->read_buffer.count && - grpc_chttp2_perform_read(exec_ctx, &t->parsing, + grpc_chttp2_perform_read(exec_ctx, transport_parsing, t->read_buffer.slices[i]); i++) ; @@ -1139,16 +1278,28 @@ static void recv_data(grpc_exec_ctx *exec_ctx, void *tp, int success) { /* merge stream lists */ grpc_chttp2_stream_map_move_into(&t->new_stream_map, &t->parsing_stream_map); - t->global.concurrent_stream_count = + transport_global->concurrent_stream_count = (gpr_uint32)grpc_chttp2_stream_map_size(&t->parsing_stream_map); - if (t->parsing.initial_window_update != 0) { + if (transport_parsing->initial_window_update != 0) { grpc_chttp2_stream_map_for_each(&t->parsing_stream_map, update_global_window, t); - t->parsing.initial_window_update = 0; + transport_parsing->initial_window_update = 0; } /* handle higher level things */ - grpc_chttp2_publish_reads(exec_ctx, &t->global, &t->parsing); + grpc_chttp2_publish_reads(exec_ctx, transport_global, transport_parsing); t->parsing_active = 0; + /* if a stream is in the stream map, and gets cancelled, we need to ensure + * we are not parsing before continuing the cancellation to keep things in + * a sane state */ + while (grpc_chttp2_list_pop_closed_waiting_for_parsing(transport_global, + &stream_global)) { + GPR_ASSERT(stream_global->in_stream_map); + GPR_ASSERT(stream_global->write_closed); + GPR_ASSERT(stream_global->read_closed); + remove_stream(exec_ctx, t, stream_global->id); + stream_global->finished_close = 1; + GRPC_CHTTP2_STREAM_UNREF(exec_ctx, stream_global, "chttp2"); + } } if (!success || i != t->read_buffer.count || t->closed) { drop_connection(exec_ctx, t); @@ -1206,35 +1357,216 @@ static void add_to_pollset_set_locked(grpc_exec_ctx *exec_ctx, } } +static void set_pollset(grpc_exec_ctx *exec_ctx, grpc_transport *gt, + grpc_stream *gs, grpc_pollset *pollset) { + grpc_chttp2_transport *t = (grpc_chttp2_transport *)gt; + lock(t); + add_to_pollset_locked(exec_ctx, t, pollset); + unlock(exec_ctx, t); +} + /* - * TRACING + * BYTE STREAM */ -void grpc_chttp2_flowctl_trace(const char *file, int line, const char *reason, - const char *context, const char *var, - int is_client, gpr_uint32 stream_id, - gpr_int64 current_value, gpr_int64 delta) { - char *identifier; - char *context_scope; - char *context_thread; - char *underscore_pos = strchr(context, '_'); - GPR_ASSERT(underscore_pos); - context_thread = gpr_strdup(underscore_pos + 1); - context_scope = gpr_strdup(context); - context_scope[underscore_pos - context] = 0; - if (stream_id) { - gpr_asprintf(&identifier, "%s[%d]", context_scope, stream_id); +static int incoming_byte_stream_next(grpc_exec_ctx *exec_ctx, + grpc_byte_stream *byte_stream, + gpr_slice *slice, size_t max_size_hint, + grpc_closure *on_complete) { + grpc_chttp2_incoming_byte_stream *bs = + (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); + } + } + if (bs->slices.count > 0) { + *slice = gpr_slice_buffer_take_first(&bs->slices); + unlock(exec_ctx, bs->transport); + return 1; + } else { + bs->on_next = on_complete; + bs->next = slice; + unlock(exec_ctx, bs->transport); + return 0; + } +} + +static void incoming_byte_stream_unref(grpc_chttp2_incoming_byte_stream *bs) { + if (gpr_unref(&bs->refs)) { + gpr_slice_buffer_destroy(&bs->slices); + gpr_free(bs); + } +} + +static void incoming_byte_stream_destroy(grpc_byte_stream *byte_stream) { + incoming_byte_stream_unref((grpc_chttp2_incoming_byte_stream *)byte_stream); +} + +void grpc_chttp2_incoming_byte_stream_push(grpc_exec_ctx *exec_ctx, + grpc_chttp2_incoming_byte_stream *bs, + gpr_slice slice) { + gpr_mu_lock(&bs->transport->mu); + if (bs->on_next != NULL) { + *bs->next = slice; + grpc_exec_ctx_enqueue(exec_ctx, bs->on_next, 1); + bs->on_next = NULL; + } else { + gpr_slice_buffer_add(&bs->slices, slice); + } + gpr_mu_unlock(&bs->transport->mu); +} + +void grpc_chttp2_incoming_byte_stream_finished( + grpc_exec_ctx *exec_ctx, grpc_chttp2_incoming_byte_stream *bs) { + incoming_byte_stream_unref(bs); +} + +grpc_chttp2_incoming_byte_stream *grpc_chttp2_incoming_byte_stream_create( + 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 = + gpr_malloc(sizeof(*incoming_byte_stream)); + incoming_byte_stream->base.length = frame_size; + incoming_byte_stream->base.flags = flags; + incoming_byte_stream->base.next = incoming_byte_stream_next; + incoming_byte_stream->base.destroy = incoming_byte_stream_destroy; + gpr_ref_init(&incoming_byte_stream->refs, 2); + incoming_byte_stream->next_message = NULL; + incoming_byte_stream->transport = TRANSPORT_FROM_PARSING(transport_parsing); + incoming_byte_stream->stream = STREAM_FROM_PARSING(stream_parsing); + gpr_slice_buffer_init(&incoming_byte_stream->slices); + incoming_byte_stream->on_next = NULL; + incoming_byte_stream->is_tail = 1; + if (add_to_queue->head == NULL) { + add_to_queue->head = incoming_byte_stream; } else { - identifier = gpr_strdup(context_scope); - } - gpr_log(GPR_INFO, - "FLOWCTL: %s %-10s %8s %-27s %8lld %c %8lld = %8lld %-10s [%s:%d]", - is_client ? "client" : "server", identifier, context_thread, var, - current_value, delta < 0 ? '-' : '+', delta < 0 ? -delta : delta, - current_value + delta, reason, file, line); - gpr_free(identifier); - gpr_free(context_thread); - gpr_free(context_scope); + add_to_queue->tail->is_tail = 0; + add_to_queue->tail->next_message = incoming_byte_stream; + } + add_to_queue->tail = incoming_byte_stream; + return incoming_byte_stream; +} + +/* + * TRACING + */ + +static char *format_flowctl_context_var(const char *context, const char *var, + gpr_int64 val, gpr_uint32 id, + char **scope) { + char *underscore_pos; + char *result; + if (context == NULL) { + *scope = NULL; + gpr_asprintf(&result, "%s(%lld)", var, val); + return result; + } + underscore_pos = strchr(context, '_'); + *scope = gpr_strdup(context); + (*scope)[underscore_pos - context] = 0; + if (id != 0) { + char *tmp = *scope; + gpr_asprintf(scope, "%s[%d]", tmp, id); + gpr_free(tmp); + } + gpr_asprintf(&result, "%s.%s(%lld)", underscore_pos + 1, var, val); + return result; +} + +static int samestr(char *a, char *b) { + if (a == NULL) { + return b == NULL; + } + if (b == NULL) { + return 0; + } + return 0 == strcmp(a, b); +} + +void grpc_chttp2_flowctl_trace(const char *file, int line, const char *phase, + grpc_chttp2_flowctl_op op, const char *context1, + const char *var1, const char *context2, + const char *var2, int is_client, + gpr_uint32 stream_id, gpr_int64 val1, + gpr_int64 val2) { + char *scope1; + char *scope2; + char *label1 = + format_flowctl_context_var(context1, var1, val1, stream_id, &scope1); + char *label2 = + format_flowctl_context_var(context2, var2, val2, stream_id, &scope2); + char *clisvr = is_client ? "client" : "server"; + char *prefix; + + gpr_asprintf(&prefix, "FLOW % 8s: %s % 11s ", phase, clisvr, scope1); + + switch (op) { + case GRPC_CHTTP2_FLOWCTL_MOVE: + GPR_ASSERT(samestr(scope1, scope2)); + if (val2 != 0) { + gpr_log(file, line, GPR_LOG_SEVERITY_DEBUG, + "%sMOVE % 40s <- % 40s giving %d", prefix, label1, label2, + val1 + val2); + } + break; + case GRPC_CHTTP2_FLOWCTL_CREDIT: + GPR_ASSERT(val2 >= 0); + if (val2 != 0) { + gpr_log(file, line, GPR_LOG_SEVERITY_DEBUG, + "%sCREDIT % 40s by % 40s giving %d", prefix, label1, label2, + val1 + val2); + } + break; + case GRPC_CHTTP2_FLOWCTL_DEBIT: + GPR_ASSERT(val2 >= 0); + if (val2 != 0) { + gpr_log(file, line, GPR_LOG_SEVERITY_DEBUG, + "%sDEBIT % 40s by % 40s giving %d", prefix, label1, label2, + val1 - val2); + } + break; + } + + gpr_free(scope1); + gpr_free(scope2); + gpr_free(label1); + gpr_free(label2); + gpr_free(prefix); } /* @@ -1246,7 +1578,7 @@ static char *chttp2_get_peer(grpc_exec_ctx *exec_ctx, grpc_transport *t) { } static const grpc_transport_vtable vtable = { - sizeof(grpc_chttp2_stream), init_stream, perform_stream_op, + sizeof(grpc_chttp2_stream), init_stream, set_pollset, perform_stream_op, perform_transport_op, destroy_stream, destroy_transport, chttp2_get_peer}; grpc_transport *grpc_create_chttp2_transport( diff --git a/src/core/transport/metadata.c b/src/core/transport/metadata.c index 68f23177eb..a72dee4399 100644 --- a/src/core/transport/metadata.c +++ b/src/core/transport/metadata.c @@ -41,9 +41,10 @@ #include <grpc/support/alloc.h> #include <grpc/support/atm.h> #include <grpc/support/log.h> +#include <grpc/support/time.h> +#include "src/core/profiling/timers.h" #include "src/core/support/murmur_hash.h" #include "src/core/transport/chttp2/bin_encoder.h" -#include <grpc/support/time.h> #define INITIAL_STRTAB_CAPACITY 4 #define INITIAL_MDTAB_CAPACITY 4 @@ -232,24 +233,32 @@ static void metadata_context_destroy_locked(grpc_mdctx *ctx) { } void grpc_mdctx_ref(grpc_mdctx *ctx) { + GPR_TIMER_BEGIN("grpc_mdctx_ref", 0); lock(ctx); GPR_ASSERT(ctx->refs > 0); ctx->refs++; unlock(ctx); + GPR_TIMER_END("grpc_mdctx_ref", 0); } void grpc_mdctx_unref(grpc_mdctx *ctx) { + GPR_TIMER_BEGIN("grpc_mdctx_unref", 0); lock(ctx); GPR_ASSERT(ctx->refs > 0); ctx->refs--; unlock(ctx); + GPR_TIMER_END("grpc_mdctx_unref", 0); } static void grow_strtab(grpc_mdctx *ctx) { size_t capacity = ctx->strtab_capacity * 2; size_t i; - internal_string **strtab = gpr_malloc(sizeof(internal_string *) * capacity); + internal_string **strtab; internal_string *s, *next; + + GPR_TIMER_BEGIN("grow_strtab", 0); + + strtab = gpr_malloc(sizeof(internal_string *) * capacity); memset(strtab, 0, sizeof(internal_string *) * capacity); for (i = 0; i < ctx->strtab_capacity; i++) { @@ -263,12 +272,15 @@ static void grow_strtab(grpc_mdctx *ctx) { gpr_free(ctx->strtab); ctx->strtab = strtab; ctx->strtab_capacity = capacity; + + GPR_TIMER_END("grow_strtab", 0); } static void internal_destroy_string(internal_string *is) { internal_string **prev_next; internal_string *cur; grpc_mdctx *ctx = is->context; + GPR_TIMER_BEGIN("internal_destroy_string", 0); if (is->has_base64_and_huffman_encoded) { gpr_slice_unref(is->base64_and_huffman); } @@ -279,6 +291,7 @@ static void internal_destroy_string(internal_string *is) { *prev_next = cur->bucket_next; ctx->strtab_count--; gpr_free(is); + GPR_TIMER_END("internal_destroy_string", 0); } static void internal_string_ref(internal_string *s DEBUG_ARGS) { @@ -304,18 +317,22 @@ static void slice_ref(void *p) { internal_string *is = (internal_string *)((char *)p - offsetof(internal_string, refcount)); grpc_mdctx *ctx = is->context; + GPR_TIMER_BEGIN("slice_ref", 0); lock(ctx); INTERNAL_STRING_REF(is); unlock(ctx); + GPR_TIMER_END("slice_ref", 0); } static void slice_unref(void *p) { internal_string *is = (internal_string *)((char *)p - offsetof(internal_string, refcount)); grpc_mdctx *ctx = is->context; + GPR_TIMER_BEGIN("slice_unref", 0); lock(ctx); INTERNAL_STRING_UNREF(is); unlock(ctx); + GPR_TIMER_END("slice_unref", 0); } grpc_mdstr *grpc_mdstr_from_string(grpc_mdctx *ctx, const char *str) { @@ -334,6 +351,7 @@ grpc_mdstr *grpc_mdstr_from_buffer(grpc_mdctx *ctx, const gpr_uint8 *buf, gpr_uint32 hash = gpr_murmur_hash3(buf, length, ctx->hash_seed); internal_string *s; + GPR_TIMER_BEGIN("grpc_mdstr_from_buffer", 0); lock(ctx); /* search for an existing string */ @@ -342,6 +360,7 @@ grpc_mdstr *grpc_mdstr_from_buffer(grpc_mdctx *ctx, const gpr_uint8 *buf, 0 == memcmp(buf, GPR_SLICE_START_PTR(s->slice), length)) { INTERNAL_STRING_REF(s); unlock(ctx); + GPR_TIMER_END("grpc_mdstr_from_buffer", 0); return (grpc_mdstr *)s; } } @@ -382,6 +401,7 @@ grpc_mdstr *grpc_mdstr_from_buffer(grpc_mdctx *ctx, const gpr_uint8 *buf, } unlock(ctx); + GPR_TIMER_END("grpc_mdstr_from_buffer", 0); return (grpc_mdstr *)s; } @@ -391,6 +411,7 @@ static void gc_mdtab(grpc_mdctx *ctx) { internal_metadata **prev_next; internal_metadata *md, *next; + GPR_TIMER_BEGIN("gc_mdtab", 0); for (i = 0; i < ctx->mdtab_capacity; i++) { prev_next = &ctx->mdtab[i]; for (md = ctx->mdtab[i]; md; md = next) { @@ -412,17 +433,19 @@ static void gc_mdtab(grpc_mdctx *ctx) { } } } - - GPR_ASSERT(ctx->mdtab_free == 0); + GPR_TIMER_END("gc_mdtab", 0); } static void grow_mdtab(grpc_mdctx *ctx) { size_t capacity = ctx->mdtab_capacity * 2; size_t i; - internal_metadata **mdtab = - gpr_malloc(sizeof(internal_metadata *) * capacity); + internal_metadata **mdtab; internal_metadata *md, *next; gpr_uint32 hash; + + GPR_TIMER_BEGIN("grow_mdtab", 0); + + mdtab = gpr_malloc(sizeof(internal_metadata *) * capacity); memset(mdtab, 0, sizeof(internal_metadata *) * capacity); for (i = 0; i < ctx->mdtab_capacity; i++) { @@ -437,6 +460,8 @@ static void grow_mdtab(grpc_mdctx *ctx) { gpr_free(ctx->mdtab); ctx->mdtab = mdtab; ctx->mdtab_capacity = capacity; + + GPR_TIMER_END("grow_mdtab", 0); } static void rehash_mdtab(grpc_mdctx *ctx) { @@ -458,6 +483,8 @@ grpc_mdelem *grpc_mdelem_from_metadata_strings(grpc_mdctx *ctx, GPR_ASSERT(key->context == ctx); GPR_ASSERT(value->context == ctx); + GPR_TIMER_BEGIN("grpc_mdelem_from_metadata_strings", 0); + lock(ctx); /* search for an existing pair */ @@ -467,6 +494,7 @@ grpc_mdelem *grpc_mdelem_from_metadata_strings(grpc_mdctx *ctx, INTERNAL_STRING_UNREF(key); INTERNAL_STRING_UNREF(value); unlock(ctx); + GPR_TIMER_END("grpc_mdelem_from_metadata_strings", 0); return (grpc_mdelem *)md; } } @@ -496,6 +524,8 @@ grpc_mdelem *grpc_mdelem_from_metadata_strings(grpc_mdctx *ctx, unlock(ctx); + GPR_TIMER_END("grpc_mdelem_from_metadata_strings", 0); + return (grpc_mdelem *)md; } @@ -542,6 +572,7 @@ grpc_mdelem *grpc_mdelem_ref(grpc_mdelem *gmd DEBUG_ARGS) { void grpc_mdelem_unref(grpc_mdelem *gmd DEBUG_ARGS) { internal_metadata *md = (internal_metadata *)gmd; + if (!md) return; #ifdef GRPC_METADATA_REFCOUNT_DEBUG gpr_log(file, line, GPR_LOG_SEVERITY_DEBUG, "ELM UNREF:%p:%d->%d: '%s' = '%s'", md, @@ -552,12 +583,14 @@ void grpc_mdelem_unref(grpc_mdelem *gmd DEBUG_ARGS) { #endif if (2 == gpr_atm_full_fetch_add(&md->refcnt, -1)) { grpc_mdctx *ctx = md->context; + GPR_TIMER_BEGIN("grpc_mdelem_unref.to_zero", 0); lock(ctx); if (1 == gpr_atm_no_barrier_load(&md->refcnt)) { ctx->mdtab_free++; gpr_atm_no_barrier_store(&md->refcnt, 0); } unlock(ctx); + GPR_TIMER_END("grpc_mdelem_unref.to_zero", 0); } } diff --git a/src/core/transport/stream_op.c b/src/core/transport/metadata_batch.c index 6493e77bc5..c5d39e0c9f 100644 --- a/src/core/transport/stream_op.c +++ b/src/core/transport/metadata_batch.c @@ -31,7 +31,7 @@ * */ -#include "src/core/transport/stream_op.h" +#include "src/core/transport/metadata_batch.h" #include <string.h> @@ -40,143 +40,6 @@ #include "src/core/profiling/timers.h" -/* Exponential growth function: Given x, return a larger x. - Currently we grow by 1.5 times upon reallocation. */ -#define GROW(x) (3 * (x) / 2) - -void grpc_sopb_init(grpc_stream_op_buffer *sopb) { - sopb->ops = sopb->inlined_ops; - sopb->nops = 0; - sopb->capacity = GRPC_SOPB_INLINE_ELEMENTS; -} - -void grpc_sopb_destroy(grpc_stream_op_buffer *sopb) { - grpc_stream_ops_unref_owned_objects(sopb->ops, sopb->nops); - if (sopb->ops != sopb->inlined_ops) gpr_free(sopb->ops); -} - -void grpc_sopb_reset(grpc_stream_op_buffer *sopb) { - grpc_stream_ops_unref_owned_objects(sopb->ops, sopb->nops); - sopb->nops = 0; -} - -void grpc_sopb_swap(grpc_stream_op_buffer *a, grpc_stream_op_buffer *b) { - GPR_SWAP(size_t, a->nops, b->nops); - GPR_SWAP(size_t, a->capacity, b->capacity); - - if (a->ops == a->inlined_ops) { - if (b->ops == b->inlined_ops) { - /* swap contents of inlined buffer */ - grpc_stream_op temp[GRPC_SOPB_INLINE_ELEMENTS]; - memcpy(temp, a->ops, b->nops * sizeof(grpc_stream_op)); - memcpy(a->ops, b->ops, a->nops * sizeof(grpc_stream_op)); - memcpy(b->ops, temp, b->nops * sizeof(grpc_stream_op)); - } else { - /* a is inlined, b is not - copy a inlined into b, fix pointers */ - a->ops = b->ops; - b->ops = b->inlined_ops; - memcpy(b->ops, a->inlined_ops, b->nops * sizeof(grpc_stream_op)); - } - } else if (b->ops == b->inlined_ops) { - /* b is inlined, a is not - copy b inlined int a, fix pointers */ - b->ops = a->ops; - a->ops = a->inlined_ops; - memcpy(a->ops, b->inlined_ops, a->nops * sizeof(grpc_stream_op)); - } else { - /* no inlining: easy swap */ - GPR_SWAP(grpc_stream_op *, a->ops, b->ops); - } -} - -void grpc_stream_ops_unref_owned_objects(grpc_stream_op *ops, size_t nops) { - size_t i; - for (i = 0; i < nops; i++) { - switch (ops[i].type) { - case GRPC_OP_SLICE: - gpr_slice_unref(ops[i].data.slice); - break; - case GRPC_OP_METADATA: - grpc_metadata_batch_destroy(&ops[i].data.metadata); - break; - case GRPC_NO_OP: - case GRPC_OP_BEGIN_MESSAGE: - break; - } - } -} - -static void expandto(grpc_stream_op_buffer *sopb, size_t new_capacity) { - sopb->capacity = new_capacity; - if (sopb->ops == sopb->inlined_ops) { - sopb->ops = gpr_malloc(sizeof(grpc_stream_op) * new_capacity); - memcpy(sopb->ops, sopb->inlined_ops, sopb->nops * sizeof(grpc_stream_op)); - } else { - sopb->ops = gpr_realloc(sopb->ops, sizeof(grpc_stream_op) * new_capacity); - } -} - -static grpc_stream_op *add(grpc_stream_op_buffer *sopb) { - grpc_stream_op *out; - - GPR_ASSERT(sopb->nops <= sopb->capacity); - if (sopb->nops == sopb->capacity) { - expandto(sopb, GROW(sopb->capacity)); - } - out = sopb->ops + sopb->nops; - sopb->nops++; - return out; -} - -void grpc_sopb_add_no_op(grpc_stream_op_buffer *sopb) { - add(sopb)->type = GRPC_NO_OP; -} - -void grpc_sopb_add_begin_message(grpc_stream_op_buffer *sopb, gpr_uint32 length, - gpr_uint32 flags) { - grpc_stream_op *op = add(sopb); - op->type = GRPC_OP_BEGIN_MESSAGE; - op->data.begin_message.length = length; - op->data.begin_message.flags = flags; -} - -void grpc_sopb_add_metadata(grpc_stream_op_buffer *sopb, - grpc_metadata_batch b) { - grpc_stream_op *op = add(sopb); - op->type = GRPC_OP_METADATA; - op->data.metadata = b; -} - -void grpc_sopb_add_slice(grpc_stream_op_buffer *sopb, gpr_slice slice) { - grpc_stream_op *op = add(sopb); - op->type = GRPC_OP_SLICE; - op->data.slice = slice; -} - -void grpc_sopb_append(grpc_stream_op_buffer *sopb, grpc_stream_op *ops, - size_t nops) { - size_t orig_nops = sopb->nops; - size_t new_nops = orig_nops + nops; - - if (new_nops > sopb->capacity) { - expandto(sopb, GPR_MAX(GROW(sopb->capacity), new_nops)); - } - - memcpy(sopb->ops + orig_nops, ops, sizeof(grpc_stream_op) * nops); - sopb->nops = new_nops; -} - -void grpc_sopb_move_to(grpc_stream_op_buffer *src, grpc_stream_op_buffer *dst) { - if (src->nops == 0) { - return; - } - if (dst->nops == 0) { - grpc_sopb_swap(src, dst); - return; - } - grpc_sopb_append(dst, src->ops, src->nops); - src->nops = 0; -} - static void assert_valid_list(grpc_mdelem_list *list) { #ifndef NDEBUG grpc_linked_mdelem *l; @@ -200,13 +63,11 @@ static void assert_valid_list(grpc_mdelem_list *list) { #ifndef NDEBUG void grpc_metadata_batch_assert_ok(grpc_metadata_batch *batch) { assert_valid_list(&batch->list); - assert_valid_list(&batch->garbage); } #endif /* NDEBUG */ void grpc_metadata_batch_init(grpc_metadata_batch *batch) { - batch->list.head = batch->list.tail = batch->garbage.head = - batch->garbage.tail = NULL; + batch->list.head = batch->list.tail = NULL; batch->deadline = gpr_inf_future(GPR_CLOCK_REALTIME); } @@ -215,9 +76,6 @@ void grpc_metadata_batch_destroy(grpc_metadata_batch *batch) { for (l = batch->list.head; l; l = l->next) { GRPC_MDELEM_UNREF(l->md); } - for (l = batch->garbage.head; l; l = l->next) { - GRPC_MDELEM_UNREF(l->md); - } } void grpc_metadata_batch_add_head(grpc_metadata_batch *batch, @@ -283,10 +141,6 @@ void grpc_metadata_batch_merge(grpc_metadata_batch *target, next = l->next; link_tail(&target->list, l); } - for (l = to_add->garbage.head; l; l = next) { - next = l->next; - link_tail(&target->garbage, l); - } } void grpc_metadata_batch_move(grpc_metadata_batch *dst, @@ -305,7 +159,6 @@ void grpc_metadata_batch_filter(grpc_metadata_batch *batch, GPR_TIMER_BEGIN("grpc_metadata_batch_filter", 0); assert_valid_list(&batch->list); - assert_valid_list(&batch->garbage); for (l = batch->list.head; l; l = next) { grpc_mdelem *orig = l->md; grpc_mdelem *filt = filter(user_data, orig); @@ -324,14 +177,28 @@ void grpc_metadata_batch_filter(grpc_metadata_batch *batch, batch->list.tail = l->prev; } assert_valid_list(&batch->list); - link_head(&batch->garbage, l); + GRPC_MDELEM_UNREF(l->md); } else if (filt != orig) { GRPC_MDELEM_UNREF(orig); l->md = filt; } } assert_valid_list(&batch->list); - assert_valid_list(&batch->garbage); GPR_TIMER_END("grpc_metadata_batch_filter", 0); } + +static grpc_mdelem *no_metadata_for_you(void *user_data, grpc_mdelem *elem) { + return NULL; +} + +void grpc_metadata_batch_clear(grpc_metadata_batch *batch) { + batch->deadline = gpr_inf_future(GPR_CLOCK_REALTIME); + grpc_metadata_batch_filter(batch, no_metadata_for_you, NULL); +} + +int grpc_metadata_batch_is_empty(grpc_metadata_batch *batch) { + return batch->list.head == NULL && + gpr_time_cmp(gpr_inf_future(batch->deadline.clock_type), + batch->deadline) == 0; +} diff --git a/src/core/transport/stream_op.h b/src/core/transport/metadata_batch.h index 37f18b02d9..fc3a46004f 100644 --- a/src/core/transport/stream_op.h +++ b/src/core/transport/metadata_batch.h @@ -40,39 +40,6 @@ #include <grpc/support/time.h> #include "src/core/transport/metadata.h" -/* this many stream ops are inlined into a sopb before allocating */ -#define GRPC_SOPB_INLINE_ELEMENTS 4 - -/* Operations that can be performed on a stream. - Used by grpc_stream_op. */ -typedef enum grpc_stream_op_code { - /* Do nothing code. Useful if rewriting a batch to exclude some operations. - Must be ignored by receivers */ - GRPC_NO_OP, - GRPC_OP_METADATA, - /* Begin a message/metadata element/status - as defined by - grpc_message_type. */ - GRPC_OP_BEGIN_MESSAGE, - /* Add a slice of data to the current message/metadata element/status. - Must not overflow the forward declared length. */ - GRPC_OP_SLICE -} grpc_stream_op_code; - -/** Internal bit flag for grpc_begin_message's \a flags signaling the use of - * compression for the message */ -#define GRPC_WRITE_INTERNAL_COMPRESS (0x80000000u) -/** Mask of all valid internal flags. */ -#define GRPC_WRITE_INTERNAL_USED_MASK (GRPC_WRITE_INTERNAL_COMPRESS) - -/* Arguments for GRPC_OP_BEGIN_MESSAGE */ -typedef struct grpc_begin_message { - /* How many bytes of data will this message contain */ - gpr_uint32 length; - /* Write flags for the message: see grpc.h GRPC_WRITE_* for the public bits, - * GRPC_WRITE_INTERNAL_* for the internal ones. */ - gpr_uint32 flags; -} grpc_begin_message; - typedef struct grpc_linked_mdelem { grpc_mdelem *md; struct grpc_linked_mdelem *next; @@ -88,10 +55,6 @@ typedef struct grpc_mdelem_list { typedef struct grpc_metadata_batch { /** Metadata elements in this batch */ grpc_mdelem_list list; - /** Elements that have been removed from the batch, but have - not yet been unreffed - used to allow collecting garbage - under a single metadata context lock */ - grpc_mdelem_list garbage; /** Used to calculate grpc-timeout at the point of sending, or gpr_inf_future if this batch does not need to send a grpc-timeout */ @@ -102,6 +65,8 @@ void grpc_metadata_batch_init(grpc_metadata_batch *batch); void grpc_metadata_batch_destroy(grpc_metadata_batch *batch); void grpc_metadata_batch_merge(grpc_metadata_batch *target, grpc_metadata_batch *add); +void grpc_metadata_batch_clear(grpc_metadata_batch *batch); +int grpc_metadata_batch_is_empty(grpc_metadata_batch *batch); /** Moves the metadata information from \a src to \a dst. Upon return, \a src is * zeroed. */ @@ -159,54 +124,4 @@ void grpc_metadata_batch_assert_ok(grpc_metadata_batch *comd); } while (0) #endif -/* Represents a single operation performed on a stream/transport */ -typedef struct grpc_stream_op { - /* the operation to be applied */ - enum grpc_stream_op_code type; - /* the arguments to this operation. union fields are named according to the - associated op-code */ - union { - grpc_begin_message begin_message; - grpc_metadata_batch metadata; - gpr_slice slice; - } data; -} grpc_stream_op; - -/** A stream op buffer is a wrapper around stream operations that is - * dynamically extendable. */ -typedef struct grpc_stream_op_buffer { - grpc_stream_op *ops; - size_t nops; - size_t capacity; - grpc_stream_op inlined_ops[GRPC_SOPB_INLINE_ELEMENTS]; -} grpc_stream_op_buffer; - -/* Initialize a stream op buffer */ -void grpc_sopb_init(grpc_stream_op_buffer *sopb); -/* Destroy a stream op buffer */ -void grpc_sopb_destroy(grpc_stream_op_buffer *sopb); -/* Reset a sopb to no elements */ -void grpc_sopb_reset(grpc_stream_op_buffer *sopb); -/* Swap two sopbs */ -void grpc_sopb_swap(grpc_stream_op_buffer *a, grpc_stream_op_buffer *b); - -void grpc_stream_ops_unref_owned_objects(grpc_stream_op *ops, size_t nops); - -/* Append a GRPC_NO_OP to a buffer */ -void grpc_sopb_add_no_op(grpc_stream_op_buffer *sopb); -/* Append a GRPC_OP_BEGIN to a buffer */ -void grpc_sopb_add_begin_message(grpc_stream_op_buffer *sopb, gpr_uint32 length, - gpr_uint32 flags); -void grpc_sopb_add_metadata(grpc_stream_op_buffer *sopb, - grpc_metadata_batch metadata); -/* Append a GRPC_SLICE to a buffer - does not ref/unref the slice */ -void grpc_sopb_add_slice(grpc_stream_op_buffer *sopb, gpr_slice slice); -/* Append a buffer to a buffer - does not ref/unref any internal objects */ -void grpc_sopb_append(grpc_stream_op_buffer *sopb, grpc_stream_op *ops, - size_t nops); - -void grpc_sopb_move_to(grpc_stream_op_buffer *src, grpc_stream_op_buffer *dst); - -char *grpc_sopb_string(grpc_stream_op_buffer *sopb); - #endif /* GRPC_INTERNAL_CORE_TRANSPORT_STREAM_OP_H */ diff --git a/src/core/transport/transport.c b/src/core/transport/transport.c index 828d212cfe..f2bebc62f3 100644 --- a/src/core/transport/transport.c +++ b/src/core/transport/transport.c @@ -33,9 +33,36 @@ #include "src/core/transport/transport.h" #include <grpc/support/alloc.h> +#include <grpc/support/atm.h> #include <grpc/support/log.h> #include "src/core/transport/transport_impl.h" +#ifdef GRPC_STREAM_REFCOUNT_DEBUG +void grpc_stream_ref(grpc_stream_refcount *refcount, const char *reason) { + gpr_atm val = gpr_atm_no_barrier_load(&refcount->refs.count); + gpr_log(GPR_DEBUG, "STREAM %p:%p REF %d->%d %s", refcount, + refcount->destroy.cb_arg, val, val + 1, reason); +#else +void grpc_stream_ref(grpc_stream_refcount *refcount) { +#endif + gpr_ref(&refcount->refs); +} + +#ifdef GRPC_STREAM_REFCOUNT_DEBUG +void grpc_stream_unref(grpc_exec_ctx *exec_ctx, grpc_stream_refcount *refcount, + const char *reason) { + gpr_atm val = gpr_atm_no_barrier_load(&refcount->refs.count); + gpr_log(GPR_DEBUG, "STREAM %p:%p UNREF %d->%d %s", refcount, + refcount->destroy.cb_arg, val, val - 1, reason); +#else +void grpc_stream_unref(grpc_exec_ctx *exec_ctx, + grpc_stream_refcount *refcount) { +#endif + if (gpr_unref(&refcount->refs)) { + grpc_exec_ctx_enqueue(exec_ctx, &refcount->destroy, 1); + } +} + size_t grpc_transport_stream_size(grpc_transport *transport) { return transport->vtable->sizeof_stream; } @@ -47,10 +74,10 @@ void grpc_transport_destroy(grpc_exec_ctx *exec_ctx, int grpc_transport_init_stream(grpc_exec_ctx *exec_ctx, grpc_transport *transport, grpc_stream *stream, - const void *server_data, - grpc_transport_stream_op *initial_op) { - return transport->vtable->init_stream(exec_ctx, transport, stream, - server_data, initial_op); + grpc_stream_refcount *refcount, + const void *server_data) { + return transport->vtable->init_stream(exec_ctx, transport, stream, refcount, + server_data); } void grpc_transport_perform_stream_op(grpc_exec_ctx *exec_ctx, @@ -66,6 +93,12 @@ void grpc_transport_perform_op(grpc_exec_ctx *exec_ctx, transport->vtable->perform_op(exec_ctx, transport, op); } +void grpc_transport_set_pollset(grpc_exec_ctx *exec_ctx, + grpc_transport *transport, grpc_stream *stream, + grpc_pollset *pollset) { + transport->vtable->set_pollset(exec_ctx, transport, stream, pollset); +} + void grpc_transport_destroy_stream(grpc_exec_ctx *exec_ctx, grpc_transport *transport, grpc_stream *stream) { @@ -79,9 +112,8 @@ char *grpc_transport_get_peer(grpc_exec_ctx *exec_ctx, void grpc_transport_stream_op_finish_with_failure( grpc_exec_ctx *exec_ctx, grpc_transport_stream_op *op) { - grpc_exec_ctx_enqueue(exec_ctx, op->on_done_recv, 0); - grpc_exec_ctx_enqueue(exec_ctx, op->on_done_send, 0); - grpc_exec_ctx_enqueue(exec_ctx, op->on_consumed, 0); + grpc_exec_ctx_enqueue(exec_ctx, op->recv_message_ready, 0); + grpc_exec_ctx_enqueue(exec_ctx, op->on_complete, 0); } void grpc_transport_stream_op_add_cancellation(grpc_transport_stream_op *op, @@ -129,9 +161,9 @@ void grpc_transport_stream_op_add_close(grpc_transport_stream_op *op, if (optional_message) { cmd = gpr_malloc(sizeof(*cmd)); cmd->message = *optional_message; - cmd->then_call = op->on_consumed; + cmd->then_call = op->on_complete; grpc_closure_init(&cmd->closure, free_message, cmd); - op->on_consumed = &cmd->closure; + op->on_complete = &cmd->closure; op->optional_close_message = &cmd->message; } op->close_with_status = status; diff --git a/src/core/transport/transport.h b/src/core/transport/transport.h index d4cee03862..f296ce8251 100644 --- a/src/core/transport/transport.h +++ b/src/core/transport/transport.h @@ -38,7 +38,8 @@ #include "src/core/iomgr/pollset.h" #include "src/core/iomgr/pollset_set.h" -#include "src/core/transport/stream_op.h" +#include "src/core/transport/metadata_batch.h" +#include "src/core/transport/byte_stream.h" #include "src/core/channel/context.h" /* forward declarations */ @@ -49,36 +50,35 @@ typedef struct grpc_transport grpc_transport; for a stream. */ typedef struct grpc_stream grpc_stream; -/* Represents the send/recv closed state of a stream. */ -typedef enum grpc_stream_state { - /* the stream is open for sends and receives */ - GRPC_STREAM_OPEN, - /* the stream is closed for sends, but may still receive data */ - GRPC_STREAM_SEND_CLOSED, - /* the stream is closed for receives, but may still send data */ - GRPC_STREAM_RECV_CLOSED, - /* the stream is closed for both sends and receives */ - GRPC_STREAM_CLOSED -} grpc_stream_state; +typedef struct grpc_stream_refcount { + gpr_refcount refs; + grpc_closure destroy; +} grpc_stream_refcount; + +/*#define GRPC_STREAM_REFCOUNT_DEBUG*/ +#ifdef GRPC_STREAM_REFCOUNT_DEBUG +void grpc_stream_ref(grpc_stream_refcount *refcount, const char *reason); +void grpc_stream_unref(grpc_exec_ctx *exec_ctx, grpc_stream_refcount *refcount, + const char *reason); +#else +void grpc_stream_ref(grpc_stream_refcount *refcount); +void grpc_stream_unref(grpc_exec_ctx *exec_ctx, grpc_stream_refcount *refcount); +#endif /* Transport stream op: a set of operations to perform on a transport against a single stream */ typedef struct grpc_transport_stream_op { - grpc_closure *on_consumed; + grpc_metadata_batch *send_initial_metadata; + grpc_metadata_batch *send_trailing_metadata; - grpc_stream_op_buffer *send_ops; - int is_last_send; - grpc_closure *on_done_send; + grpc_byte_stream *send_message; - grpc_stream_op_buffer *recv_ops; - grpc_stream_state *recv_state; - /** The number of bytes this peer is currently prepared to receive. - These bytes will be eventually used to replenish per-stream flow control - windows. */ - size_t max_recv_bytes; - grpc_closure *on_done_recv; + grpc_metadata_batch *recv_initial_metadata; + grpc_byte_stream **recv_message; + grpc_closure *recv_message_ready; + grpc_metadata_batch *recv_trailing_metadata; - grpc_pollset *bind_pollset; + grpc_closure *on_complete; /** If != GRPC_STATUS_OK, cancel this stream */ grpc_status_code cancel_with_status; @@ -110,8 +110,8 @@ typedef struct grpc_transport_op { gpr_slice *goaway_message; /** set the callback for accepting new streams; this is a permanent callback, unlike the other one-shot closures */ - void (*set_accept_stream)(void *user_data, grpc_transport *transport, - const void *server_data); + void (*set_accept_stream)(grpc_exec_ctx *exec_ctx, void *user_data, + grpc_transport *transport, const void *server_data); void *set_accept_stream_user_data; /** add this transport to a pollset */ grpc_pollset *bind_pollset; @@ -136,8 +136,12 @@ size_t grpc_transport_stream_size(grpc_transport *transport); supplied from the accept_stream callback function */ int grpc_transport_init_stream(grpc_exec_ctx *exec_ctx, grpc_transport *transport, grpc_stream *stream, - const void *server_data, - grpc_transport_stream_op *initial_op); + grpc_stream_refcount *refcount, + const void *server_data); + +void grpc_transport_set_pollset(grpc_exec_ctx *exec_ctx, + grpc_transport *transport, grpc_stream *stream, + grpc_pollset *pollset); /* Destroy transport data for a stream. diff --git a/src/core/transport/transport_impl.h b/src/core/transport/transport_impl.h index 900c6340ff..40bfb4b13a 100644 --- a/src/core/transport/transport_impl.h +++ b/src/core/transport/transport_impl.h @@ -43,8 +43,12 @@ typedef struct grpc_transport_vtable { /* implementation of grpc_transport_init_stream */ int (*init_stream)(grpc_exec_ctx *exec_ctx, grpc_transport *self, - grpc_stream *stream, const void *server_data, - grpc_transport_stream_op *initial_op); + grpc_stream *stream, grpc_stream_refcount *refcount, + const void *server_data); + + /* implementation of grpc_transport_set_pollset */ + void (*set_pollset)(grpc_exec_ctx *exec_ctx, grpc_transport *self, + grpc_stream *stream, grpc_pollset *pollset); /* implementation of grpc_transport_perform_stream_op */ void (*perform_stream_op)(grpc_exec_ctx *exec_ctx, grpc_transport *self, diff --git a/src/core/transport/transport_op_string.c b/src/core/transport/transport_op_string.c index f62c340e97..f3b6db29d6 100644 --- a/src/core/transport/transport_op_string.c +++ b/src/core/transport/transport_op_string.c @@ -69,42 +69,6 @@ static void put_metadata_list(gpr_strvec *b, grpc_metadata_batch md) { } } -char *grpc_sopb_string(grpc_stream_op_buffer *sopb) { - char *out; - char *tmp; - size_t i; - gpr_strvec b; - gpr_strvec_init(&b); - - for (i = 0; i < sopb->nops; i++) { - grpc_stream_op *op = &sopb->ops[i]; - if (i > 0) gpr_strvec_add(&b, gpr_strdup(", ")); - switch (op->type) { - case GRPC_NO_OP: - gpr_strvec_add(&b, gpr_strdup("NO_OP")); - break; - case GRPC_OP_BEGIN_MESSAGE: - gpr_asprintf(&tmp, "BEGIN_MESSAGE:%d", op->data.begin_message.length); - gpr_strvec_add(&b, tmp); - break; - case GRPC_OP_SLICE: - gpr_asprintf(&tmp, "SLICE:%d", GPR_SLICE_LENGTH(op->data.slice)); - gpr_strvec_add(&b, tmp); - break; - case GRPC_OP_METADATA: - gpr_strvec_add(&b, gpr_strdup("METADATA{")); - put_metadata_list(&b, op->data.metadata); - gpr_strvec_add(&b, gpr_strdup("}")); - break; - } - } - - out = gpr_strvec_flatten(&b, NULL); - gpr_strvec_destroy(&b); - - return out; -} - char *grpc_transport_stream_op_string(grpc_transport_stream_op *op) { char *tmp; char *out; @@ -113,42 +77,52 @@ char *grpc_transport_stream_op_string(grpc_transport_stream_op *op) { gpr_strvec b; gpr_strvec_init(&b); - if (op->send_ops) { + if (op->send_initial_metadata != NULL) { if (!first) gpr_strvec_add(&b, gpr_strdup(" ")); first = 0; - gpr_asprintf(&tmp, "SEND%s:%p", op->is_last_send ? "_LAST" : "", - op->on_done_send); - gpr_strvec_add(&b, tmp); - gpr_strvec_add(&b, gpr_strdup("[")); - gpr_strvec_add(&b, grpc_sopb_string(op->send_ops)); - gpr_strvec_add(&b, gpr_strdup("]")); + gpr_strvec_add(&b, gpr_strdup("SEND_INITIAL_METADATA{")); + put_metadata_list(&b, *op->send_initial_metadata); + gpr_strvec_add(&b, gpr_strdup("}")); } - if (op->recv_ops) { + if (op->send_message != NULL) { if (!first) gpr_strvec_add(&b, gpr_strdup(" ")); first = 0; - gpr_asprintf(&tmp, "RECV:%p:max_recv_bytes=%d", op->on_done_recv, - op->max_recv_bytes); + gpr_asprintf(&tmp, "SEND_MESSAGE:flags=0x%08x:len=%d", + op->send_message->flags, op->send_message->length); gpr_strvec_add(&b, tmp); } - if (op->bind_pollset) { + if (op->send_trailing_metadata != NULL) { if (!first) gpr_strvec_add(&b, gpr_strdup(" ")); first = 0; - gpr_strvec_add(&b, gpr_strdup("BIND")); + gpr_strvec_add(&b, gpr_strdup("SEND_TRAILING_METADATA{")); + put_metadata_list(&b, *op->send_trailing_metadata); + gpr_strvec_add(&b, gpr_strdup("}")); } - if (op->cancel_with_status != GRPC_STATUS_OK) { + if (op->recv_initial_metadata != NULL) { if (!first) gpr_strvec_add(&b, gpr_strdup(" ")); first = 0; - gpr_asprintf(&tmp, "CANCEL:%d", op->cancel_with_status); - gpr_strvec_add(&b, tmp); + gpr_strvec_add(&b, gpr_strdup("RECV_INITIAL_METADATA")); + } + + if (op->recv_message != NULL) { + if (!first) gpr_strvec_add(&b, gpr_strdup(" ")); + first = 0; + gpr_strvec_add(&b, gpr_strdup("RECV_MESSAGE")); } - if (op->on_consumed != NULL) { + if (op->recv_trailing_metadata != NULL) { if (!first) gpr_strvec_add(&b, gpr_strdup(" ")); first = 0; - gpr_asprintf(&tmp, "ON_CONSUMED:%p", op->on_consumed); + gpr_strvec_add(&b, gpr_strdup("RECV_TRAILING_METADATA")); + } + + if (op->cancel_with_status != GRPC_STATUS_OK) { + if (!first) gpr_strvec_add(&b, gpr_strdup(" ")); + first = 0; + gpr_asprintf(&tmp, "CANCEL:%d", op->cancel_with_status); gpr_strvec_add(&b, tmp); } |