diff options
author | 2015-09-15 15:20:26 -0700 | |
---|---|---|
committer | 2015-09-15 15:20:26 -0700 | |
commit | fd2f7331fa97f735879a8759ac9fc8591969011b (patch) | |
tree | da031f982ee5d1d21fd71e6c4bc9d80365c7d6e2 /src/core/transport | |
parent | 1928d496a237c3850365e2557ae41ae73125fc80 (diff) | |
parent | 9ed4be48955e3b65b0540f70060b7191d8c1eef1 (diff) |
Merge branch 'master' of github.com:grpc/grpc into core_creds_plugin
Diffstat (limited to 'src/core/transport')
20 files changed, 422 insertions, 322 deletions
diff --git a/src/core/transport/chttp2/bin_encoder.c b/src/core/transport/chttp2/bin_encoder.c index dee6dbec8b..e21d800083 100644 --- a/src/core/transport/chttp2/bin_encoder.c +++ b/src/core/transport/chttp2/bin_encoder.c @@ -68,7 +68,7 @@ gpr_slice grpc_chttp2_base64_encode(gpr_slice input) { size_t output_length = input_triplets * 4 + tail_xtra[tail_case]; gpr_slice output = gpr_slice_malloc(output_length); gpr_uint8 *in = GPR_SLICE_START_PTR(input); - gpr_uint8 *out = GPR_SLICE_START_PTR(output); + char *out = (char *)GPR_SLICE_START_PTR(output); size_t i; /* encode full triplets */ @@ -100,7 +100,7 @@ gpr_slice grpc_chttp2_base64_encode(gpr_slice input) { break; } - GPR_ASSERT(out == GPR_SLICE_END_PTR(output)); + GPR_ASSERT(out == (char *)GPR_SLICE_END_PTR(output)); GPR_ASSERT(in == GPR_SLICE_END_PTR(input)); return output; } @@ -128,12 +128,13 @@ gpr_slice grpc_chttp2_huffman_compress(gpr_slice input) { while (temp_length > 8) { temp_length -= 8; - *out++ = temp >> temp_length; + *out++ = (gpr_uint8)(temp >> temp_length); } } if (temp_length) { - *out++ = (temp << (8 - temp_length)) | (0xff >> temp_length); + *out++ = (gpr_uint8)(temp << (8u - temp_length)) | + (gpr_uint8)(0xffu >> temp_length); } GPR_ASSERT(out == GPR_SLICE_END_PTR(output)); @@ -150,16 +151,16 @@ typedef struct { static void enc_flush_some(huff_out *out) { while (out->temp_length > 8) { out->temp_length -= 8; - *out->out++ = out->temp >> out->temp_length; + *out->out++ = (gpr_uint8)(out->temp >> out->temp_length); } } static void enc_add2(huff_out *out, gpr_uint8 a, gpr_uint8 b) { b64_huff_sym sa = huff_alphabet[a]; b64_huff_sym sb = huff_alphabet[b]; - out->temp = - (out->temp << (sa.length + sb.length)) | (sa.bits << sb.length) | sb.bits; - out->temp_length += sa.length + sb.length; + out->temp = (out->temp << (sa.length + sb.length)) | + ((gpr_uint32)sa.bits << sb.length) | sb.bits; + out->temp_length += (gpr_uint32)sa.length + (gpr_uint32)sb.length; enc_flush_some(out); } @@ -189,8 +190,9 @@ gpr_slice grpc_chttp2_base64_encode_and_huffman_compress(gpr_slice input) { /* encode full triplets */ for (i = 0; i < input_triplets; i++) { - enc_add2(&out, in[0] >> 2, ((in[0] & 0x3) << 4) | (in[1] >> 4)); - enc_add2(&out, ((in[1] & 0xf) << 2) | (in[2] >> 6), in[2] & 0x3f); + enc_add2(&out, in[0] >> 2, (gpr_uint8)((in[0] & 0x3) << 4) | (in[1] >> 4)); + enc_add2(&out, (gpr_uint8)((in[1] & 0xf) << 2) | (in[2] >> 6), + (gpr_uint8)(in[2] & 0x3f)); in += 3; } @@ -199,19 +201,20 @@ gpr_slice grpc_chttp2_base64_encode_and_huffman_compress(gpr_slice input) { case 0: break; case 1: - enc_add2(&out, in[0] >> 2, (in[0] & 0x3) << 4); + enc_add2(&out, in[0] >> 2, (gpr_uint8)((in[0] & 0x3) << 4)); in += 1; break; case 2: - enc_add2(&out, in[0] >> 2, ((in[0] & 0x3) << 4) | (in[1] >> 4)); - enc_add1(&out, (in[1] & 0xf) << 2); + enc_add2(&out, in[0] >> 2, + (gpr_uint8)((in[0] & 0x3) << 4) | (gpr_uint8)(in[1] >> 4)); + enc_add1(&out, (gpr_uint8)((in[1] & 0xf) << 2)); in += 2; break; } if (out.temp_length) { - *out.out++ = - (out.temp << (8 - out.temp_length)) | (0xff >> out.temp_length); + *out.out++ = (gpr_uint8)(out.temp << (8u - out.temp_length)) | + (gpr_uint8)(0xffu >> out.temp_length); } GPR_ASSERT(out.out <= GPR_SLICE_END_PTR(output)); diff --git a/src/core/transport/chttp2/frame_data.c b/src/core/transport/chttp2/frame_data.c index 474c3d5ee6..403358016d 100644 --- a/src/core/transport/chttp2/frame_data.c +++ b/src/core/transport/chttp2/frame_data.c @@ -146,20 +146,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, - gpr_slice_sub(slice, cur - beg, end - beg)); + grpc_sopb_add_slice( + &p->incoming_sopb, + gpr_slice_sub(slice, (size_t)(cur - beg), (size_t)(end - beg))); 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, cur - beg, cur + p->frame_size - beg)); + grpc_sopb_add_slice(&p->incoming_sopb, + gpr_slice_sub(slice, (size_t)(cur - beg), + (size_t)(cur + p->frame_size - beg))); cur += p->frame_size; goto fh_0; /* loop */ } else { - grpc_sopb_add_slice(&p->incoming_sopb, - gpr_slice_sub(slice, cur - beg, end - beg)); - p->frame_size -= (end - cur); + grpc_sopb_add_slice( + &p->incoming_sopb, + gpr_slice_sub(slice, (size_t)(cur - beg), (size_t)(end - beg))); + GPR_ASSERT(end - cur <= p->frame_size); + p->frame_size -= (gpr_uint32)(end - cur); return GRPC_CHTTP2_PARSE_OK; } } diff --git a/src/core/transport/chttp2/frame_goaway.c b/src/core/transport/chttp2/frame_goaway.c index 1ccbba840c..09d4da234c 100644 --- a/src/core/transport/chttp2/frame_goaway.c +++ b/src/core/transport/chttp2/frame_goaway.c @@ -136,14 +136,15 @@ grpc_chttp2_parse_error grpc_chttp2_goaway_parser_parse( ++cur; /* fallthrough */ case GRPC_CHTTP2_GOAWAY_DEBUG: - memcpy(p->debug_data + p->debug_pos, cur, end - cur); - p->debug_pos += end - cur; + memcpy(p->debug_data + p->debug_pos, cur, (size_t)(end - cur)); + GPR_ASSERT(end - cur < GPR_UINT32_MAX - p->debug_pos); + p->debug_pos += (gpr_uint32)(end - cur); p->state = GRPC_CHTTP2_GOAWAY_DEBUG; if (is_last) { transport_parsing->goaway_received = 1; transport_parsing->goaway_last_stream_index = p->last_stream_id; gpr_slice_unref(transport_parsing->goaway_text); - transport_parsing->goaway_error = p->error_code; + transport_parsing->goaway_error = (grpc_status_code)p->error_code; transport_parsing->goaway_text = gpr_slice_new(p->debug_data, p->debug_length, gpr_free); p->debug_data = NULL; @@ -160,12 +161,14 @@ void grpc_chttp2_goaway_append(gpr_uint32 last_stream_id, gpr_uint32 error_code, gpr_slice_buffer *slice_buffer) { gpr_slice header = gpr_slice_malloc(9 + 4 + 4); gpr_uint8 *p = GPR_SLICE_START_PTR(header); - gpr_uint32 frame_length = 4 + 4 + GPR_SLICE_LENGTH(debug_data); + gpr_uint32 frame_length; + GPR_ASSERT(GPR_SLICE_LENGTH(debug_data) < GPR_UINT32_MAX - 4 - 4); + frame_length = 4 + 4 + (gpr_uint32)GPR_SLICE_LENGTH(debug_data); /* frame header: length */ - *p++ = frame_length >> 16; - *p++ = frame_length >> 8; - *p++ = frame_length; + *p++ = (gpr_uint8)(frame_length >> 16); + *p++ = (gpr_uint8)(frame_length >> 8); + *p++ = (gpr_uint8)(frame_length); /* frame header: type */ *p++ = GRPC_CHTTP2_FRAME_GOAWAY; /* frame header: flags */ @@ -176,15 +179,15 @@ void grpc_chttp2_goaway_append(gpr_uint32 last_stream_id, gpr_uint32 error_code, *p++ = 0; *p++ = 0; /* payload: last stream id */ - *p++ = last_stream_id >> 24; - *p++ = last_stream_id >> 16; - *p++ = last_stream_id >> 8; - *p++ = last_stream_id; + *p++ = (gpr_uint8)(last_stream_id >> 24); + *p++ = (gpr_uint8)(last_stream_id >> 16); + *p++ = (gpr_uint8)(last_stream_id >> 8); + *p++ = (gpr_uint8)(last_stream_id); /* payload: error code */ - *p++ = error_code >> 24; - *p++ = error_code >> 16; - *p++ = error_code >> 8; - *p++ = error_code; + *p++ = (gpr_uint8)(error_code >> 24); + *p++ = (gpr_uint8)(error_code >> 16); + *p++ = (gpr_uint8)(error_code >> 8); + *p++ = (gpr_uint8)(error_code); GPR_ASSERT(p == GPR_SLICE_END_PTR(header)); gpr_slice_buffer_add(slice_buffer, header); gpr_slice_buffer_add(slice_buffer, debug_data); diff --git a/src/core/transport/chttp2/frame_rst_stream.c b/src/core/transport/chttp2/frame_rst_stream.c index a878d936c1..67da245239 100644 --- a/src/core/transport/chttp2/frame_rst_stream.c +++ b/src/core/transport/chttp2/frame_rst_stream.c @@ -47,14 +47,14 @@ gpr_slice grpc_chttp2_rst_stream_create(gpr_uint32 id, gpr_uint32 code) { *p++ = 4; *p++ = GRPC_CHTTP2_FRAME_RST_STREAM; *p++ = 0; - *p++ = id >> 24; - *p++ = id >> 16; - *p++ = id >> 8; - *p++ = id; - *p++ = code >> 24; - *p++ = code >> 16; - *p++ = code >> 8; - *p++ = code; + *p++ = (gpr_uint8)(id >> 24); + *p++ = (gpr_uint8)(id >> 16); + *p++ = (gpr_uint8)(id >> 8); + *p++ = (gpr_uint8)(id); + *p++ = (gpr_uint8)(code >> 24); + *p++ = (gpr_uint8)(code >> 16); + *p++ = (gpr_uint8)(code >> 8); + *p++ = (gpr_uint8)(code); return slice; } diff --git a/src/core/transport/chttp2/frame_settings.c b/src/core/transport/chttp2/frame_settings.c index d42bc000ae..54d3694a5c 100644 --- a/src/core/transport/chttp2/frame_settings.c +++ b/src/core/transport/chttp2/frame_settings.c @@ -61,9 +61,9 @@ const grpc_chttp2_setting_parameters static gpr_uint8 *fill_header(gpr_uint8 *out, gpr_uint32 length, gpr_uint8 flags) { - *out++ = length >> 16; - *out++ = length >> 8; - *out++ = length; + *out++ = (gpr_uint8)(length >> 16); + *out++ = (gpr_uint8)(length >> 8); + *out++ = (gpr_uint8)(length); *out++ = GRPC_CHTTP2_FRAME_SETTINGS; *out++ = flags; *out++ = 0; @@ -76,26 +76,26 @@ static gpr_uint8 *fill_header(gpr_uint8 *out, gpr_uint32 length, gpr_slice grpc_chttp2_settings_create(gpr_uint32 *old, const gpr_uint32 *new, gpr_uint32 force_mask, size_t count) { size_t i; - size_t n = 0; + gpr_uint32 n = 0; gpr_slice output; gpr_uint8 *p; for (i = 0; i < count; i++) { - n += (new[i] != old[i] || (force_mask & (1 << i)) != 0); + n += (new[i] != old[i] || (force_mask & (1u << i)) != 0); } output = gpr_slice_malloc(9 + 6 * n); p = fill_header(GPR_SLICE_START_PTR(output), 6 * n, 0); for (i = 0; i < count; i++) { - if (new[i] != old[i] || (force_mask & (1 << i)) != 0) { + if (new[i] != old[i] || (force_mask & (1u << i)) != 0) { GPR_ASSERT(i); - *p++ = i >> 8; - *p++ = i; - *p++ = new[i] >> 24; - *p++ = new[i] >> 16; - *p++ = new[i] >> 8; - *p++ = new[i]; + *p++ = (gpr_uint8)(i >> 8); + *p++ = (gpr_uint8)(i); + *p++ = (gpr_uint8)(new[i] >> 24); + *p++ = (gpr_uint8)(new[i] >> 16); + *p++ = (gpr_uint8)(new[i] >> 8); + *p++ = (gpr_uint8)(new[i]); old[i] = new[i]; } } @@ -162,7 +162,7 @@ grpc_chttp2_parse_error grpc_chttp2_settings_parser_parse( } return GRPC_CHTTP2_PARSE_OK; } - parser->id = ((gpr_uint16)*cur) << 8; + parser->id = (gpr_uint16)(((gpr_uint16)*cur) << 8); cur++; /* fallthrough */ case GRPC_CHTTP2_SPS_ID1: @@ -170,7 +170,7 @@ grpc_chttp2_parse_error grpc_chttp2_settings_parser_parse( parser->state = GRPC_CHTTP2_SPS_ID1; return GRPC_CHTTP2_PARSE_OK; } - parser->id |= (*cur); + parser->id = (gpr_uint16)(parser->id | (*cur)); cur++; /* fallthrough */ case GRPC_CHTTP2_SPS_VAL0: diff --git a/src/core/transport/chttp2/frame_window_update.c b/src/core/transport/chttp2/frame_window_update.c index d624298ad2..ea13969e8c 100644 --- a/src/core/transport/chttp2/frame_window_update.c +++ b/src/core/transport/chttp2/frame_window_update.c @@ -48,14 +48,14 @@ gpr_slice grpc_chttp2_window_update_create(gpr_uint32 id, *p++ = 4; *p++ = GRPC_CHTTP2_FRAME_WINDOW_UPDATE; *p++ = 0; - *p++ = id >> 24; - *p++ = id >> 16; - *p++ = id >> 8; - *p++ = id; - *p++ = window_update >> 24; - *p++ = window_update >> 16; - *p++ = window_update >> 8; - *p++ = window_update; + *p++ = (gpr_uint8)(id >> 24); + *p++ = (gpr_uint8)(id >> 16); + *p++ = (gpr_uint8)(id >> 8); + *p++ = (gpr_uint8)(id); + *p++ = (gpr_uint8)(window_update >> 24); + *p++ = (gpr_uint8)(window_update >> 16); + *p++ = (gpr_uint8)(window_update >> 8); + *p++ = (gpr_uint8)(window_update); return slice; } diff --git a/src/core/transport/chttp2/hpack_parser.c b/src/core/transport/chttp2/hpack_parser.c index f8bff42ed6..9c40e8a4e6 100644 --- a/src/core/transport/chttp2/hpack_parser.c +++ b/src/core/transport/chttp2/hpack_parser.c @@ -1085,11 +1085,13 @@ static int parse_string_prefix(grpc_chttp2_hpack_parser *p, static void append_bytes(grpc_chttp2_hpack_parser_string *str, const gpr_uint8 *data, size_t length) { if (length + str->length > str->capacity) { - str->capacity = str->length + length; + GPR_ASSERT(str->length + length <= GPR_UINT32_MAX); + str->capacity = (gpr_uint32)(str->length + length); str->str = gpr_realloc(str->str, str->capacity); } memcpy(str->str + str->length, data, length); - str->length += length; + GPR_ASSERT(length <= GPR_UINT32_MAX - str->length); + str->length += (gpr_uint32)length; } static int append_string(grpc_chttp2_hpack_parser *p, const gpr_uint8 *cur, @@ -1099,7 +1101,7 @@ static int append_string(grpc_chttp2_hpack_parser *p, const gpr_uint8 *cur, gpr_uint8 decoded[3]; switch ((binary_state)p->binary) { case NOT_BINARY: - append_bytes(str, cur, end - cur); + append_bytes(str, cur, (size_t)(end - cur)); return 1; b64_byte0: case B64_BYTE0: @@ -1157,9 +1159,9 @@ static int append_string(grpc_chttp2_hpack_parser *p, const gpr_uint8 *cur, goto b64_byte3; p->base64_buffer |= bits; bits = p->base64_buffer; - decoded[0] = bits >> 16; - decoded[1] = bits >> 8; - decoded[2] = bits; + decoded[0] = (gpr_uint8)(bits >> 16); + decoded[1] = (gpr_uint8)(bits >> 8); + decoded[2] = (gpr_uint8)(bits); append_bytes(str, decoded, 3); goto b64_byte0; } @@ -1189,7 +1191,7 @@ static int finish_str(grpc_chttp2_hpack_parser *p) { bits & 0xffff); return 0; } - decoded[0] = bits >> 16; + decoded[0] = (gpr_uint8)(bits >> 16); append_bytes(str, decoded, 1); break; case B64_BYTE3: @@ -1199,8 +1201,8 @@ static int finish_str(grpc_chttp2_hpack_parser *p) { bits & 0xff); return 0; } - decoded[0] = bits >> 16; - decoded[1] = bits >> 8; + decoded[0] = (gpr_uint8)(bits >> 16); + decoded[1] = (gpr_uint8)(bits >> 8); append_bytes(str, decoded, 2); break; } @@ -1249,13 +1251,14 @@ static int add_str_bytes(grpc_chttp2_hpack_parser *p, const gpr_uint8 *cur, static int parse_string(grpc_chttp2_hpack_parser *p, const gpr_uint8 *cur, const gpr_uint8 *end) { size_t remaining = p->strlen - p->strgot; - size_t given = end - cur; + size_t given = (size_t)(end - cur); if (remaining <= given) { return add_str_bytes(p, cur, cur + remaining) && finish_str(p) && parse_next(p, cur + remaining, end); } else { if (!add_str_bytes(p, cur, cur + given)) return 0; - p->strgot += given; + GPR_ASSERT(given <= GPR_UINT32_MAX - p->strgot); + p->strgot += (gpr_uint32)given; p->state = parse_string; return 1; } diff --git a/src/core/transport/chttp2/hpack_parser.h b/src/core/transport/chttp2/hpack_parser.h index c1768d9d5d..4f489d67fb 100644 --- a/src/core/transport/chttp2/hpack_parser.h +++ b/src/core/transport/chttp2/hpack_parser.h @@ -79,7 +79,7 @@ struct grpc_chttp2_hpack_parser { /* number of source bytes read for the currently parsing string */ gpr_uint32 strgot; /* huffman decoding state */ - gpr_uint16 huff_state; + gpr_int16 huff_state; /* is the string being decoded binary? */ gpr_uint8 binary; /* is the current string huffman encoded? */ diff --git a/src/core/transport/chttp2/hpack_table.c b/src/core/transport/chttp2/hpack_table.c index 4fc154380e..e18778ab0b 100644 --- a/src/core/transport/chttp2/hpack_table.c +++ b/src/core/transport/chttp2/hpack_table.c @@ -139,7 +139,7 @@ grpc_mdelem *grpc_chttp2_hptbl_lookup(const grpc_chttp2_hptbl *tbl, /* Otherwise, find the value in the list of valid entries */ index -= (GRPC_CHTTP2_LAST_STATIC_ENTRY + 1); if (index < tbl->num_ents) { - gpr_uint32 offset = (tbl->num_ents - 1 - index + tbl->first_ent) % + gpr_uint32 offset = (tbl->num_ents - 1u - index + tbl->first_ent) % GRPC_CHTTP2_MAX_TABLE_COUNT; return tbl->ents[offset]; } @@ -150,19 +150,22 @@ grpc_mdelem *grpc_chttp2_hptbl_lookup(const grpc_chttp2_hptbl *tbl, /* Evict one element from the table */ static void evict1(grpc_chttp2_hptbl *tbl) { grpc_mdelem *first_ent = tbl->ents[tbl->first_ent]; - tbl->mem_used -= GPR_SLICE_LENGTH(first_ent->key->slice) + - GPR_SLICE_LENGTH(first_ent->value->slice) + - GRPC_CHTTP2_HPACK_ENTRY_OVERHEAD; - tbl->first_ent = (tbl->first_ent + 1) % GRPC_CHTTP2_MAX_TABLE_COUNT; + size_t elem_bytes = GPR_SLICE_LENGTH(first_ent->key->slice) + + GPR_SLICE_LENGTH(first_ent->value->slice) + + GRPC_CHTTP2_HPACK_ENTRY_OVERHEAD; + GPR_ASSERT(elem_bytes <= tbl->mem_used); + tbl->mem_used = (gpr_uint16)(tbl->mem_used - elem_bytes); + tbl->first_ent = + (gpr_uint16)((tbl->first_ent + 1) % GRPC_CHTTP2_MAX_TABLE_COUNT); tbl->num_ents--; GRPC_MDELEM_UNREF(first_ent); } void grpc_chttp2_hptbl_add(grpc_chttp2_hptbl *tbl, grpc_mdelem *md) { /* determine how many bytes of buffer this entry represents */ - gpr_uint16 elem_bytes = GPR_SLICE_LENGTH(md->key->slice) + - GPR_SLICE_LENGTH(md->value->slice) + - GRPC_CHTTP2_HPACK_ENTRY_OVERHEAD; + size_t elem_bytes = GPR_SLICE_LENGTH(md->key->slice) + + GPR_SLICE_LENGTH(md->value->slice) + + GRPC_CHTTP2_HPACK_ENTRY_OVERHEAD; /* we can't add elements bigger than the max table size */ if (elem_bytes > tbl->max_bytes) { @@ -182,7 +185,7 @@ void grpc_chttp2_hptbl_add(grpc_chttp2_hptbl *tbl, grpc_mdelem *md) { } /* evict entries to ensure no overflow */ - while (elem_bytes > tbl->max_bytes - tbl->mem_used) { + while (elem_bytes > (size_t)tbl->max_bytes - tbl->mem_used) { evict1(tbl); } @@ -190,28 +193,30 @@ void grpc_chttp2_hptbl_add(grpc_chttp2_hptbl *tbl, grpc_mdelem *md) { tbl->ents[tbl->last_ent] = md; /* update accounting values */ - tbl->last_ent = (tbl->last_ent + 1) % GRPC_CHTTP2_MAX_TABLE_COUNT; + tbl->last_ent = + (gpr_uint16)((tbl->last_ent + 1) % GRPC_CHTTP2_MAX_TABLE_COUNT); tbl->num_ents++; - tbl->mem_used += elem_bytes; + tbl->mem_used = (gpr_uint16)(tbl->mem_used + elem_bytes); } grpc_chttp2_hptbl_find_result grpc_chttp2_hptbl_find( const grpc_chttp2_hptbl *tbl, grpc_mdelem *md) { grpc_chttp2_hptbl_find_result r = {0, 0}; - int i; + gpr_uint16 i; /* See if the string is in the static table */ for (i = 0; i < GRPC_CHTTP2_LAST_STATIC_ENTRY; i++) { grpc_mdelem *ent = tbl->static_ents[i]; if (md->key != ent->key) continue; - r.index = i + 1; + r.index = (gpr_uint16)(i + 1); r.has_value = md->value == ent->value; if (r.has_value) return r; } /* Scan the dynamic table */ for (i = 0; i < tbl->num_ents; i++) { - int idx = tbl->num_ents - i + GRPC_CHTTP2_LAST_STATIC_ENTRY; + gpr_uint16 idx = + (gpr_uint16)(tbl->num_ents - i + GRPC_CHTTP2_LAST_STATIC_ENTRY); grpc_mdelem *ent = tbl->ents[(tbl->first_ent + i) % GRPC_CHTTP2_MAX_TABLE_COUNT]; if (md->key != ent->key) continue; diff --git a/src/core/transport/chttp2/incoming_metadata.c b/src/core/transport/chttp2/incoming_metadata.c index 974b864ffb..d216c42113 100644 --- a/src/core/transport/chttp2/incoming_metadata.c +++ b/src/core/transport/chttp2/incoming_metadata.c @@ -122,7 +122,7 @@ void grpc_incoming_metadata_buffer_move_to_referencing_sopb( 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_intptr)sopb->ops[i].data.metadata.list.tail); + (void *)(delta + (gpr_uintptr)sopb->ops[i].data.metadata.list.tail); } src->count = 0; } diff --git a/src/core/transport/chttp2/internal.h b/src/core/transport/chttp2/internal.h index 42cf0ecd5b..c8c1abb750 100644 --- a/src/core/transport/chttp2/internal.h +++ b/src/core/transport/chttp2/internal.h @@ -168,7 +168,7 @@ typedef struct { grpc_iomgr_closure *pending_closures_tail; /** window available for us to send to peer */ - gpr_uint32 outgoing_window; + gpr_int64 outgoing_window; /** window available for peer to send to us - updated after parse */ gpr_uint32 incoming_window; /** how much window would we like to have for incoming_window */ @@ -214,6 +214,8 @@ typedef struct { grpc_chttp2_hpack_compressor hpack_compressor; /** is this a client? */ gpr_uint8 is_client; + /** callback for when writing is done */ + grpc_iomgr_closure done_cb; } grpc_chttp2_transport_writing; struct grpc_chttp2_transport_parsing { @@ -278,7 +280,7 @@ struct grpc_chttp2_transport_parsing { gpr_uint32 goaway_last_stream_index; gpr_slice goaway_text; - gpr_uint64 outgoing_window_update; + gpr_int64 outgoing_window_update; /** pings awaiting responses */ grpc_chttp2_outstanding_ping pings; @@ -291,6 +293,9 @@ struct grpc_chttp2_transport { gpr_refcount refs; char *peer_string; + /** when this drops to zero it's safe to shutdown the endpoint */ + gpr_refcount shutdown_ep_refs; + gpr_mu mu; /** is the transport destroying itself? */ @@ -329,8 +334,11 @@ struct grpc_chttp2_transport { /** closure to execute writing */ grpc_iomgr_closure writing_action; - /** closure to start reading from the endpoint */ - grpc_iomgr_closure reading_action; + /** closure to finish reading from the endpoint */ + grpc_iomgr_closure recv_data; + + /** incoming read bytes */ + gpr_slice_buffer read_buffer; /** address to place a newly accepted stream - set and unset by grpc_chttp2_parsing_accept_stream; used by init_stream to @@ -463,8 +471,7 @@ int grpc_chttp2_unlocking_check_writes(grpc_chttp2_transport_global *global, grpc_chttp2_transport_writing *writing); void grpc_chttp2_perform_writes( grpc_chttp2_transport_writing *transport_writing, grpc_endpoint *endpoint); -void grpc_chttp2_terminate_writing( - grpc_chttp2_transport_writing *transport_writing, int success); +void grpc_chttp2_terminate_writing(void *transport_writing, int success); void grpc_chttp2_cleanup_writing(grpc_chttp2_transport_global *global, grpc_chttp2_transport_writing *writing); @@ -602,20 +609,21 @@ 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, context->var, \ - delta); \ +#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, context->var, 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, diff --git a/src/core/transport/chttp2/parsing.c b/src/core/transport/chttp2/parsing.c index dc5eb18e42..f26f446787 100644 --- a/src/core/transport/chttp2/parsing.c +++ b/src/core/transport/chttp2/parsing.c @@ -131,7 +131,7 @@ void grpc_chttp2_publish_reads( published later */ if (transport_parsing->goaway_received) { grpc_chttp2_add_incoming_goaway(transport_global, - transport_parsing->goaway_error, + (gpr_uint32)transport_parsing->goaway_error, transport_parsing->goaway_text); transport_parsing->goaway_text = gpr_empty_slice(); transport_parsing->goaway_received = 0; @@ -194,7 +194,9 @@ void grpc_chttp2_publish_reads( GRPC_CHTTP2_FLOWCTL_TRACE_STREAM( "parsed", transport_parsing, stream_parsing, outgoing_window_update, -(gpr_int64)stream_parsing->outgoing_window_update); - stream_global->outgoing_window += 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) { @@ -211,7 +213,7 @@ void grpc_chttp2_publish_reads( if (stream_parsing->saw_rst_stream) { stream_global->cancelled = 1; stream_global->cancelled_status = grpc_chttp2_http2_error_to_grpc_status( - stream_parsing->rst_stream_reason); + (grpc_chttp2_error_code)stream_parsing->rst_stream_reason); if (stream_parsing->rst_stream_reason == GRPC_CHTTP2_NO_ERROR) { stream_global->published_cancelled = 1; } @@ -379,9 +381,10 @@ int grpc_chttp2_perform_read(grpc_chttp2_transport_parsing *transport_parsing, case GRPC_DTS_FRAME: GPR_ASSERT(cur < end); if ((gpr_uint32)(end - cur) == transport_parsing->incoming_frame_size) { - if (!parse_frame_slice( - transport_parsing, - gpr_slice_sub_no_ref(slice, cur - beg, end - beg), 1)) { + if (!parse_frame_slice(transport_parsing, + gpr_slice_sub_no_ref(slice, (size_t)(cur - beg), + (size_t)(end - beg)), + 1)) { return 0; } transport_parsing->deframe_state = GRPC_DTS_FH_0; @@ -389,11 +392,12 @@ int grpc_chttp2_perform_read(grpc_chttp2_transport_parsing *transport_parsing, return 1; } else if ((gpr_uint32)(end - cur) > transport_parsing->incoming_frame_size) { + size_t cur_offset = (size_t)(cur - beg); if (!parse_frame_slice( transport_parsing, gpr_slice_sub_no_ref( - slice, cur - beg, - cur + transport_parsing->incoming_frame_size - beg), + slice, cur_offset, + cur_offset + transport_parsing->incoming_frame_size), 1)) { return 0; } @@ -401,12 +405,13 @@ int grpc_chttp2_perform_read(grpc_chttp2_transport_parsing *transport_parsing, transport_parsing->incoming_stream = NULL; goto dts_fh_0; /* loop */ } else { - if (!parse_frame_slice( - transport_parsing, - gpr_slice_sub_no_ref(slice, cur - beg, end - beg), 0)) { + if (!parse_frame_slice(transport_parsing, + gpr_slice_sub_no_ref(slice, (size_t)(cur - beg), + (size_t)(end - beg)), + 0)) { return 0; } - transport_parsing->incoming_frame_size -= (end - cur); + transport_parsing->incoming_frame_size -= (gpr_uint32)(end - cur); return 1; } gpr_log(GPR_ERROR, "should never reach here"); @@ -474,14 +479,14 @@ 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) { if (is_header) { - int is_eoh = transport_parsing->expect_continuation_stream_id != 0; + gpr_uint8 is_eoh = transport_parsing->expect_continuation_stream_id != 0; transport_parsing->parser = grpc_chttp2_header_parser_parse; transport_parsing->parser_data = &transport_parsing->hpack_parser; transport_parsing->hpack_parser.on_header = skip_header; transport_parsing->hpack_parser.on_header_user_data = NULL; transport_parsing->hpack_parser.is_boundary = is_eoh; transport_parsing->hpack_parser.is_eof = - is_eoh ? transport_parsing->header_eof : 0; + (gpr_uint8)(is_eoh ? transport_parsing->header_eof : 0); } else { transport_parsing->parser = skip_parser; } @@ -617,8 +622,8 @@ static void on_header(void *tp, grpc_mdelem *md) { static int init_header_frame_parser( grpc_chttp2_transport_parsing *transport_parsing, int is_continuation) { - int is_eoh = (transport_parsing->incoming_frame_flags & - GRPC_CHTTP2_DATA_FLAG_END_HEADERS) != 0; + 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; @@ -691,7 +696,7 @@ static int init_header_frame_parser( transport_parsing->hpack_parser.on_header_user_data = transport_parsing; transport_parsing->hpack_parser.is_boundary = is_eoh; transport_parsing->hpack_parser.is_eof = - is_eoh ? transport_parsing->header_eof : 0; + (gpr_uint8)(is_eoh ? transport_parsing->header_eof : 0); if (!is_continuation && (transport_parsing->incoming_frame_flags & GRPC_CHTTP2_FLAG_HAS_PRIORITY)) { grpc_chttp2_hpack_parser_set_has_priority(&transport_parsing->hpack_parser); diff --git a/src/core/transport/chttp2/stream_encoder.c b/src/core/transport/chttp2/stream_encoder.c index 1ea697f71e..6a22532bc2 100644 --- a/src/core/transport/chttp2/stream_encoder.c +++ b/src/core/transport/chttp2/stream_encoder.c @@ -74,17 +74,18 @@ typedef struct { } framer_state; /* fills p (which is expected to be 9 bytes long) with a data frame header */ -static void fill_header(gpr_uint8 *p, gpr_uint8 type, gpr_uint32 id, - gpr_uint32 len, gpr_uint8 flags) { - *p++ = len >> 16; - *p++ = len >> 8; - *p++ = len; +static void fill_header(gpr_uint8 *p, gpr_uint8 type, gpr_uint32 id, size_t len, + gpr_uint8 flags) { + GPR_ASSERT(len < 16777316); + *p++ = (gpr_uint8)(len >> 16); + *p++ = (gpr_uint8)(len >> 8); + *p++ = (gpr_uint8)(len); *p++ = type; *p++ = flags; - *p++ = id >> 24; - *p++ = id >> 16; - *p++ = id >> 8; - *p++ = id; + *p++ = (gpr_uint8)(id >> 24); + *p++ = (gpr_uint8)(id >> 16); + *p++ = (gpr_uint8)(id >> 8); + *p++ = (gpr_uint8)(id); } /* finish a frame - fill in the previously reserved header */ @@ -105,11 +106,12 @@ static void finish_frame(framer_state *st, int is_header_boundary, case NONE: return; } - 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, - (is_last_in_stream ? GRPC_CHTTP2_DATA_FLAG_END_STREAM : 0) | - (is_header_boundary ? GRPC_CHTTP2_DATA_FLAG_END_HEADERS : 0)); + 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; } @@ -134,7 +136,7 @@ static void begin_new_frame(framer_state *st, frame_type type) { 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, - int need_bytes) { + 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) { @@ -174,7 +176,7 @@ static void add_header_data(framer_state *st, gpr_slice slice) { } } -static gpr_uint8 *add_tiny_header_data(framer_state *st, int len) { +static gpr_uint8 *add_tiny_header_data(framer_state *st, size_t len) { ensure_frame_type(st, HEADER, len); return gpr_slice_buffer_tiny_add(st->output, len); } @@ -185,10 +187,12 @@ static grpc_mdelem *add_elem(grpc_chttp2_hpack_compressor *c, 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; - gpr_uint32 elem_size = 32 + GPR_SLICE_LENGTH(elem->key->slice) + - GPR_SLICE_LENGTH(elem->value->slice); + 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); + /* Reserve space for this element in the remote table: if this overflows the current table, drop elements until it fits, matching the decompressor algorithm */ @@ -200,14 +204,16 @@ static grpc_mdelem *add_elem(grpc_chttp2_hpack_compressor *c, c->table_elem_size[c->tail_remote_index % GRPC_CHTTP2_HPACKC_MAX_TABLE_ELEMS]); GPR_ASSERT(c->table_elems > 0); - c->table_size -= c->table_elem_size[c->tail_remote_index % - GRPC_CHTTP2_HPACKC_MAX_TABLE_ELEMS]; + c->table_size = + (gpr_uint16)(c->table_size - + c->table_elem_size[c->tail_remote_index % + GRPC_CHTTP2_HPACKC_MAX_TABLE_ELEMS]); c->table_elems--; } GPR_ASSERT(c->table_elems < GRPC_CHTTP2_HPACKC_MAX_TABLE_ELEMS); c->table_elem_size[new_index % GRPC_CHTTP2_HPACKC_MAX_TABLE_ELEMS] = - elem_size; - c->table_size += elem_size; + (gpr_uint16)elem_size; + c->table_size = (gpr_uint16)(c->table_size + elem_size); c->table_elems++; /* Store this element into {entries,indices}_elem */ @@ -270,7 +276,7 @@ static grpc_mdelem *add_elem(grpc_chttp2_hpack_compressor *c, static void emit_indexed(grpc_chttp2_hpack_compressor *c, gpr_uint32 index, framer_state *st) { - int len = GRPC_CHTTP2_VARINT_LENGTH(index, 1); + gpr_uint32 len = GRPC_CHTTP2_VARINT_LENGTH(index, 1); GRPC_CHTTP2_WRITE_VARINT(index, 1, 0x80, add_tiny_header_data(st, len), len); } @@ -288,14 +294,16 @@ static gpr_slice get_wire_value(grpc_mdelem *elem, gpr_uint8 *huffman_prefix) { static void emit_lithdr_incidx(grpc_chttp2_hpack_compressor *c, gpr_uint32 key_index, grpc_mdelem *elem, framer_state *st) { - int len_pfx = GRPC_CHTTP2_VARINT_LENGTH(key_index, 2); + gpr_uint32 len_pfx = GRPC_CHTTP2_VARINT_LENGTH(key_index, 2); gpr_uint8 huffman_prefix; gpr_slice value_slice = get_wire_value(elem, &huffman_prefix); - int len_val = GPR_SLICE_LENGTH(value_slice); - int len_val_len = GRPC_CHTTP2_VARINT_LENGTH(len_val, 1); + size_t len_val = GPR_SLICE_LENGTH(value_slice); + gpr_uint32 len_val_len; + GPR_ASSERT(len_val <= GPR_UINT32_MAX); + len_val_len = GRPC_CHTTP2_VARINT_LENGTH((gpr_uint32)len_val, 1); GRPC_CHTTP2_WRITE_VARINT(key_index, 2, 0x40, add_tiny_header_data(st, len_pfx), len_pfx); - GRPC_CHTTP2_WRITE_VARINT(len_val, 1, 0x00, + GRPC_CHTTP2_WRITE_VARINT((gpr_uint32)len_val, 1, 0x00, add_tiny_header_data(st, len_val_len), len_val_len); add_header_data(st, gpr_slice_ref(value_slice)); } @@ -303,26 +311,30 @@ static void emit_lithdr_incidx(grpc_chttp2_hpack_compressor *c, static void emit_lithdr_noidx(grpc_chttp2_hpack_compressor *c, gpr_uint32 key_index, grpc_mdelem *elem, framer_state *st) { - int len_pfx = GRPC_CHTTP2_VARINT_LENGTH(key_index, 4); + gpr_uint32 len_pfx = GRPC_CHTTP2_VARINT_LENGTH(key_index, 4); gpr_uint8 huffman_prefix; gpr_slice value_slice = get_wire_value(elem, &huffman_prefix); - int len_val = GPR_SLICE_LENGTH(value_slice); - int len_val_len = GRPC_CHTTP2_VARINT_LENGTH(len_val, 1); + size_t len_val = GPR_SLICE_LENGTH(value_slice); + gpr_uint32 len_val_len; + GPR_ASSERT(len_val <= GPR_UINT32_MAX); + len_val_len = GRPC_CHTTP2_VARINT_LENGTH((gpr_uint32)len_val, 1); GRPC_CHTTP2_WRITE_VARINT(key_index, 4, 0x00, add_tiny_header_data(st, len_pfx), len_pfx); - GRPC_CHTTP2_WRITE_VARINT(len_val, 1, 0x00, + GRPC_CHTTP2_WRITE_VARINT((gpr_uint32)len_val, 1, 0x00, add_tiny_header_data(st, len_val_len), len_val_len); add_header_data(st, gpr_slice_ref(value_slice)); } static void emit_lithdr_incidx_v(grpc_chttp2_hpack_compressor *c, grpc_mdelem *elem, framer_state *st) { - int len_key = GPR_SLICE_LENGTH(elem->key->slice); + gpr_uint32 len_key = (gpr_uint32)GPR_SLICE_LENGTH(elem->key->slice); gpr_uint8 huffman_prefix; gpr_slice value_slice = get_wire_value(elem, &huffman_prefix); - int len_val = GPR_SLICE_LENGTH(value_slice); - int len_key_len = GRPC_CHTTP2_VARINT_LENGTH(len_key, 1); - int len_val_len = GRPC_CHTTP2_VARINT_LENGTH(len_val, 1); + gpr_uint32 len_val = (gpr_uint32)GPR_SLICE_LENGTH(value_slice); + gpr_uint32 len_key_len = GRPC_CHTTP2_VARINT_LENGTH(len_key, 1); + gpr_uint32 len_val_len = GRPC_CHTTP2_VARINT_LENGTH(len_val, 1); + GPR_ASSERT(len_key <= GPR_UINT32_MAX); + GPR_ASSERT(GPR_SLICE_LENGTH(value_slice) <= GPR_UINT32_MAX); *add_tiny_header_data(st, 1) = 0x40; GRPC_CHTTP2_WRITE_VARINT(len_key, 1, 0x00, add_tiny_header_data(st, len_key_len), len_key_len); @@ -334,12 +346,14 @@ static void emit_lithdr_incidx_v(grpc_chttp2_hpack_compressor *c, static void emit_lithdr_noidx_v(grpc_chttp2_hpack_compressor *c, grpc_mdelem *elem, framer_state *st) { - int len_key = GPR_SLICE_LENGTH(elem->key->slice); + gpr_uint32 len_key = (gpr_uint32)GPR_SLICE_LENGTH(elem->key->slice); gpr_uint8 huffman_prefix; gpr_slice value_slice = get_wire_value(elem, &huffman_prefix); - int len_val = GPR_SLICE_LENGTH(value_slice); - int len_key_len = GRPC_CHTTP2_VARINT_LENGTH(len_key, 1); - int len_val_len = GRPC_CHTTP2_VARINT_LENGTH(len_val, 1); + gpr_uint32 len_val = (gpr_uint32)GPR_SLICE_LENGTH(value_slice); + gpr_uint32 len_key_len = GRPC_CHTTP2_VARINT_LENGTH(len_key, 1); + gpr_uint32 len_val_len = GRPC_CHTTP2_VARINT_LENGTH(len_val, 1); + GPR_ASSERT(len_key <= GPR_UINT32_MAX); + GPR_ASSERT(GPR_SLICE_LENGTH(value_slice) <= GPR_UINT32_MAX); *add_tiny_header_data(st, 1) = 0x00; GRPC_CHTTP2_WRITE_VARINT(len_key, 1, 0x00, add_tiny_header_data(st, len_key_len), len_key_len); @@ -488,7 +502,7 @@ gpr_uint32 grpc_chttp2_preencode(grpc_stream_op *inops, size_t *inops_count, gpr_uint32 flow_controlled_bytes_taken = 0; gpr_uint32 curop = 0; gpr_uint8 *p; - int compressed_flag_set = 0; + gpr_uint8 compressed_flag_set = 0; while (curop < *inops_count) { GPR_ASSERT(flow_controlled_bytes_taken <= max_flow_controlled_bytes); @@ -514,10 +528,10 @@ gpr_uint32 grpc_chttp2_preencode(grpc_stream_op *inops, size_t *inops_count, p = GPR_SLICE_START_PTR(slice); p[0] = compressed_flag_set; - p[1] = op->data.begin_message.length >> 24; - p[2] = op->data.begin_message.length >> 16; - p[3] = op->data.begin_message.length >> 8; - p[4] = op->data.begin_message.length; + 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 */ @@ -541,7 +555,7 @@ gpr_uint32 grpc_chttp2_preencode(grpc_stream_op *inops, size_t *inops_count, grpc_sopb_append(outops, op, 1); curop++; } - flow_controlled_bytes_taken += GPR_SLICE_LENGTH(slice); + flow_controlled_bytes_taken += (gpr_uint32)GPR_SLICE_LENGTH(slice); break; } } @@ -565,7 +579,7 @@ void grpc_chttp2_encode(grpc_stream_op *ops, size_t ops_count, int eof, framer_state st; gpr_slice slice; grpc_stream_op *op; - gpr_uint32 max_take_size; + size_t max_take_size; gpr_uint32 curop = 0; gpr_uint32 unref_op; grpc_mdctx *mdctx = compressor->mdctx; diff --git a/src/core/transport/chttp2/timeout_encoding.c b/src/core/transport/chttp2/timeout_encoding.c index 1dd768ada4..8a9b290ecb 100644 --- a/src/core/transport/chttp2/timeout_encoding.c +++ b/src/core/transport/chttp2/timeout_encoding.c @@ -123,7 +123,7 @@ void grpc_chttp2_encode_timeout(gpr_timespec timeout, char *buffer) { enc_nanos(buffer, timeout.tv_nsec); } else if (timeout.tv_sec < 1000 && timeout.tv_nsec != 0) { enc_micros(buffer, - timeout.tv_sec * 1000000 + + (int)(timeout.tv_sec * 1000000) + (timeout.tv_nsec / 1000 + (timeout.tv_nsec % 1000 != 0))); } else { enc_seconds(buffer, timeout.tv_sec + (timeout.tv_nsec != 0)); @@ -137,14 +137,14 @@ static int is_all_whitespace(const char *p) { int grpc_chttp2_decode_timeout(const char *buffer, gpr_timespec *timeout) { gpr_uint32 x = 0; - const char *p = buffer; + const gpr_uint8 *p = (const gpr_uint8 *)buffer; int have_digit = 0; /* skip whitespace */ for (; *p == ' '; p++) ; /* decode numeric part */ for (; *p >= '0' && *p <= '9'; p++) { - gpr_uint32 xp = x * 10 + *p - '0'; + gpr_uint32 xp = x * 10u + (gpr_uint32)*p - (gpr_uint32)'0'; have_digit = 1; if (xp < x) { *timeout = gpr_inf_future(GPR_CLOCK_REALTIME); @@ -180,5 +180,5 @@ int grpc_chttp2_decode_timeout(const char *buffer, gpr_timespec *timeout) { return 0; } p++; - return is_all_whitespace(p); + return is_all_whitespace((const char *)p); } diff --git a/src/core/transport/chttp2/varint.c b/src/core/transport/chttp2/varint.c index 0722c9ada9..056f68047b 100644 --- a/src/core/transport/chttp2/varint.c +++ b/src/core/transport/chttp2/varint.c @@ -33,7 +33,7 @@ #include "src/core/transport/chttp2/varint.h" -int grpc_chttp2_hpack_varint_length(gpr_uint32 tail_value) { +gpr_uint32 grpc_chttp2_hpack_varint_length(gpr_uint32 tail_value) { if (tail_value < (1 << 7)) { return 2; } else if (tail_value < (1 << 14)) { @@ -48,7 +48,8 @@ int grpc_chttp2_hpack_varint_length(gpr_uint32 tail_value) { } void grpc_chttp2_hpack_write_varint_tail(gpr_uint32 tail_value, - gpr_uint8* target, int tail_length) { + gpr_uint8* target, + gpr_uint32 tail_length) { switch (tail_length) { case 5: target[4] = (gpr_uint8)((tail_value >> 28) | 0x80); diff --git a/src/core/transport/chttp2/varint.h b/src/core/transport/chttp2/varint.h index 0a6fb55248..4dfcc76773 100644 --- a/src/core/transport/chttp2/varint.h +++ b/src/core/transport/chttp2/varint.h @@ -41,10 +41,11 @@ /* length of a value that needs varint tail encoding (it's bigger than can be bitpacked into the opcode byte) - returned value includes the length of the opcode byte */ -int grpc_chttp2_hpack_varint_length(gpr_uint32 tail_value); +gpr_uint32 grpc_chttp2_hpack_varint_length(gpr_uint32 tail_value); void grpc_chttp2_hpack_write_varint_tail(gpr_uint32 tail_value, - gpr_uint8* target, int tail_length); + gpr_uint8* target, + gpr_uint32 tail_length); /* maximum value that can be bitpacked with the opcode if the opcode has a prefix @@ -54,15 +55,15 @@ void grpc_chttp2_hpack_write_varint_tail(gpr_uint32 tail_value, /* length required to bitpack a value */ #define GRPC_CHTTP2_VARINT_LENGTH(n, prefix_bits) \ ((n) < GRPC_CHTTP2_MAX_IN_PREFIX(prefix_bits) \ - ? 1 \ + ? 1u \ : grpc_chttp2_hpack_varint_length( \ (n)-GRPC_CHTTP2_MAX_IN_PREFIX(prefix_bits))) #define GRPC_CHTTP2_WRITE_VARINT(n, prefix_bits, prefix_or, target, length) \ do { \ gpr_uint8* tgt = target; \ - if ((length) == 1) { \ - (tgt)[0] = (prefix_or) | (n); \ + if ((length) == 1u) { \ + (tgt)[0] = (gpr_uint8)((prefix_or) | (n)); \ } else { \ (tgt)[0] = (prefix_or) | GRPC_CHTTP2_MAX_IN_PREFIX(prefix_bits); \ grpc_chttp2_hpack_write_varint_tail( \ diff --git a/src/core/transport/chttp2/writing.c b/src/core/transport/chttp2/writing.c index 123061b3fc..c015e82931 100644 --- a/src/core/transport/chttp2/writing.c +++ b/src/core/transport/chttp2/writing.c @@ -32,12 +32,14 @@ */ #include "src/core/transport/chttp2/internal.h" -#include "src/core/transport/chttp2/http2_errors.h" + +#include <limits.h> #include <grpc/support/log.h> +#include "src/core/transport/chttp2/http2_errors.h" + static void finalize_outbuf(grpc_chttp2_transport_writing *transport_writing); -static void finish_write_cb(void *tw, grpc_endpoint_cb_status write_status); int grpc_chttp2_unlocking_check_writes( grpc_chttp2_transport_global *transport_global, @@ -79,12 +81,13 @@ int grpc_chttp2_unlocking_check_writes( 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_MIN(transport_global->outgoing_window, - stream_global->outgoing_window), - &stream_writing->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, @@ -114,6 +117,10 @@ int grpc_chttp2_unlocking_check_writes( 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( @@ -165,16 +172,15 @@ void grpc_chttp2_perform_writes( GPR_ASSERT(transport_writing->outbuf.count > 0); GPR_ASSERT(endpoint); - switch (grpc_endpoint_write(endpoint, transport_writing->outbuf.slices, - transport_writing->outbuf.count, finish_write_cb, - transport_writing)) { - case GRPC_ENDPOINT_WRITE_DONE: + switch (grpc_endpoint_write(endpoint, &transport_writing->outbuf, + &transport_writing->done_cb)) { + case GRPC_ENDPOINT_DONE: grpc_chttp2_terminate_writing(transport_writing, 1); break; - case GRPC_ENDPOINT_WRITE_ERROR: + case GRPC_ENDPOINT_ERROR: grpc_chttp2_terminate_writing(transport_writing, 0); break; - case GRPC_ENDPOINT_WRITE_PENDING: + case GRPC_ENDPOINT_PENDING: break; } } @@ -198,6 +204,9 @@ static void finalize_outbuf(grpc_chttp2_transport_writing *transport_writing) { &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); stream_writing->announce_window = 0; } if (stream_writing->send_closed == GRPC_SEND_CLOSED_WITH_RST_STREAM) { @@ -209,12 +218,6 @@ static void finalize_outbuf(grpc_chttp2_transport_writing *transport_writing) { } } -static void finish_write_cb(void *tw, grpc_endpoint_cb_status write_status) { - grpc_chttp2_transport_writing *transport_writing = tw; - grpc_chttp2_terminate_writing(transport_writing, - write_status == GRPC_ENDPOINT_CB_OK); -} - void grpc_chttp2_cleanup_writing( grpc_chttp2_transport_global *transport_global, grpc_chttp2_transport_writing *transport_writing) { @@ -243,6 +246,5 @@ void grpc_chttp2_cleanup_writing( grpc_chttp2_list_add_read_write_state_changed(transport_global, stream_global); } - transport_writing->outbuf.count = 0; - transport_writing->outbuf.length = 0; + 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 1bbd210e46..deb2fedf0c 100644 --- a/src/core/transport/chttp2_transport.c +++ b/src/core/transport/chttp2_transport.c @@ -84,15 +84,13 @@ static void unlock_check_read_write_state(grpc_chttp2_transport *t); /* forward declarations of various callbacks that we'll build closures around */ static void writing_action(void *t, int iomgr_success_ignored); -static void reading_action(void *t, int iomgr_success_ignored); /** Set a transport level setting, and push it to our peer */ static void push_setting(grpc_chttp2_transport *t, grpc_chttp2_setting_id id, gpr_uint32 value); /** Endpoint callback to process incoming data */ -static void recv_data(void *tp, gpr_slice *slices, size_t nslices, - grpc_endpoint_cb_status error); +static void recv_data(void *tp, int success); /** Start disconnection chain */ static void drop_connection(grpc_chttp2_transport *t); @@ -143,6 +141,7 @@ static void destruct_transport(grpc_chttp2_transport *t) { grpc_chttp2_hpack_compressor_destroy(&t->writing.hpack_compressor); gpr_slice_buffer_destroy(&t->parsing.qbuf); + gpr_slice_buffer_destroy(&t->read_buffer); grpc_chttp2_hpack_parser_destroy(&t->parsing.hpack_parser); grpc_chttp2_goaway_parser_destroy(&t->parsing.goaway_parser); @@ -210,7 +209,7 @@ static void ref_transport(grpc_chttp2_transport *t) { gpr_ref(&t->refs); } static void init_transport(grpc_chttp2_transport *t, const grpc_channel_args *channel_args, grpc_endpoint *ep, grpc_mdctx *mdctx, - int is_client) { + gpr_uint8 is_client) { size_t i; int j; @@ -223,6 +222,8 @@ static void init_transport(grpc_chttp2_transport *t, t->ep = ep; /* one ref is for destroy, the other for when ep becomes NULL */ gpr_ref_init(&t->refs, 2); + /* ref is dropped at transport close() */ + gpr_ref_init(&t->shutdown_ep_refs, 1); gpr_mu_init(&t->mu); grpc_mdctx_ref(mdctx); t->peer_string = grpc_endpoint_get_peer(ep); @@ -249,12 +250,16 @@ static void init_transport(grpc_chttp2_transport *t, gpr_slice_buffer_init(&t->writing.outbuf); grpc_chttp2_hpack_compressor_init(&t->writing.hpack_compressor, mdctx); grpc_iomgr_closure_init(&t->writing_action, writing_action, t); - grpc_iomgr_closure_init(&t->reading_action, reading_action, t); gpr_slice_buffer_init(&t->parsing.qbuf); grpc_chttp2_goaway_parser_init(&t->parsing.goaway_parser); grpc_chttp2_hpack_parser_init(&t->parsing.hpack_parser, t->metadata_context); + grpc_iomgr_closure_init(&t->writing.done_cb, grpc_chttp2_terminate_writing, + &t->writing); + grpc_iomgr_closure_init(&t->recv_data, recv_data, t); + gpr_slice_buffer_init(&t->read_buffer); + if (is_client) { gpr_slice_buffer_add( &t->global.qbuf, @@ -301,7 +306,7 @@ static void init_transport(grpc_chttp2_transport *t, GRPC_ARG_MAX_CONCURRENT_STREAMS); } else { push_setting(t, GRPC_CHTTP2_SETTINGS_MAX_CONCURRENT_STREAMS, - channel_args->args[i].value.integer); + (gpr_uint32)channel_args->args[i].value.integer); } } else if (0 == strcmp(channel_args->args[i].key, GRPC_ARG_HTTP2_INITIAL_SEQUENCE_NUMBER)) { @@ -315,7 +320,8 @@ static void init_transport(grpc_chttp2_transport *t, t->global.next_stream_id & 1, is_client ? "client" : "server"); } else { - t->global.next_stream_id = channel_args->args[i].value.integer; + t->global.next_stream_id = + (gpr_uint32)channel_args->args[i].value.integer; } } } @@ -333,13 +339,45 @@ static void destroy_transport(grpc_transport *gt) { UNREF_TRANSPORT(t, "destroy"); } +/** block grpc_endpoint_shutdown being called until a paired + allow_endpoint_shutdown is made */ +static void prevent_endpoint_shutdown(grpc_chttp2_transport *t) { + GPR_ASSERT(t->ep); + gpr_ref(&t->shutdown_ep_refs); +} + +static void allow_endpoint_shutdown_locked(grpc_chttp2_transport *t) { + if (gpr_unref(&t->shutdown_ep_refs)) { + if (t->ep) { + grpc_endpoint_shutdown(t->ep); + } + } +} + +static void allow_endpoint_shutdown_unlocked(grpc_chttp2_transport *t) { + if (gpr_unref(&t->shutdown_ep_refs)) { + gpr_mu_lock(&t->mu); + if (t->ep) { + grpc_endpoint_shutdown(t->ep); + } + gpr_mu_unlock(&t->mu); + } +} + +static void destroy_endpoint(grpc_chttp2_transport *t) { + grpc_endpoint_destroy(t->ep); + t->ep = NULL; + UNREF_TRANSPORT( + t, "disconnect"); /* safe because we'll still have the ref for write */ +} + static void close_transport_locked(grpc_chttp2_transport *t) { if (!t->closed) { t->closed = 1; connectivity_state_set(&t->global, GRPC_CHANNEL_FATAL_FAILURE, "close_transport"); if (t->ep) { - grpc_endpoint_shutdown(t->ep); + allow_endpoint_shutdown_locked(t); } } } @@ -468,6 +506,7 @@ static void unlock(grpc_chttp2_transport *t) { t->writing_active = 1; REF_TRANSPORT(t, "writing"); grpc_chttp2_schedule_closure(&t->global, &t->writing_action, 1); + prevent_endpoint_shutdown(t); } run_closures = t->global.pending_closures_head; @@ -502,12 +541,14 @@ static void push_setting(grpc_chttp2_transport *t, grpc_chttp2_setting_id id, } } -void grpc_chttp2_terminate_writing( - grpc_chttp2_transport_writing *transport_writing, int success) { +void grpc_chttp2_terminate_writing(void *transport_writing_ptr, int success) { + grpc_chttp2_transport_writing *transport_writing = transport_writing_ptr; grpc_chttp2_transport *t = TRANSPORT_FROM_WRITING(transport_writing); lock(t); + allow_endpoint_shutdown_locked(t); + if (!success) { drop_connection(t); } @@ -519,10 +560,7 @@ void grpc_chttp2_terminate_writing( from starting */ t->writing_active = 0; if (t->ep && !t->endpoint_reading) { - grpc_endpoint_destroy(t->ep); - t->ep = NULL; - UNREF_TRANSPORT( - t, "disconnect"); /* safe because we'll still have the ref for write */ + destroy_endpoint(t); } unlock(t); @@ -645,6 +683,8 @@ static void perform_stream_op_locked( 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, @@ -653,14 +693,14 @@ static void perform_stream_op_locked( "op", transport_global, stream_global, unannounced_incoming_window, op->max_recv_bytes - stream_global->max_recv_bytes); stream_global->unannounced_incoming_window += - op->max_recv_bytes - stream_global->max_recv_bytes; - stream_global->max_recv_bytes = op->max_recv_bytes; + (gpr_uint32)op->max_recv_bytes - stream_global->max_recv_bytes; + stream_global->max_recv_bytes = (gpr_uint32)op->max_recv_bytes; } 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) { - grpc_chttp2_list_add_read_write_state_changed(transport_global, - stream_global); grpc_chttp2_list_add_writable_stream(transport_global, stream_global); } } @@ -691,14 +731,14 @@ static void send_ping_locked(grpc_chttp2_transport *t, p->next = &t->global.pings; p->prev = p->next->prev; p->prev->next = p->next->prev = p; - p->id[0] = (t->global.ping_counter >> 56) & 0xff; - p->id[1] = (t->global.ping_counter >> 48) & 0xff; - p->id[2] = (t->global.ping_counter >> 40) & 0xff; - p->id[3] = (t->global.ping_counter >> 32) & 0xff; - p->id[4] = (t->global.ping_counter >> 24) & 0xff; - p->id[5] = (t->global.ping_counter >> 16) & 0xff; - p->id[6] = (t->global.ping_counter >> 8) & 0xff; - p->id[7] = t->global.ping_counter & 0xff; + p->id[0] = (gpr_uint8)((t->global.ping_counter >> 56) & 0xff); + p->id[1] = (gpr_uint8)((t->global.ping_counter >> 48) & 0xff); + p->id[2] = (gpr_uint8)((t->global.ping_counter >> 40) & 0xff); + p->id[3] = (gpr_uint8)((t->global.ping_counter >> 32) & 0xff); + p->id[4] = (gpr_uint8)((t->global.ping_counter >> 24) & 0xff); + p->id[5] = (gpr_uint8)((t->global.ping_counter >> 16) & 0xff); + p->id[6] = (gpr_uint8)((t->global.ping_counter >> 8) & 0xff); + p->id[7] = (gpr_uint8)(t->global.ping_counter & 0xff); p->on_recv = on_recv; gpr_slice_buffer_add(&t->global.qbuf, grpc_chttp2_ping_create(0, p->id)); } @@ -723,7 +763,7 @@ static void perform_transport_op(grpc_transport *gt, grpc_transport_op *op) { t->global.sent_goaway = 1; grpc_chttp2_goaway_append( t->global.last_incoming_stream_id, - grpc_chttp2_grpc_status_to_http2_error(op->goaway_status), + (gpr_uint32)grpc_chttp2_grpc_status_to_http2_error(op->goaway_status), gpr_slice_ref(*op->goaway_message), &t->global.qbuf); close_transport = !grpc_chttp2_has_streams(t); } @@ -791,8 +831,9 @@ static void remove_stream(grpc_chttp2_transport *t, gpr_uint32 id) { new_stream_count = grpc_chttp2_stream_map_size(&t->parsing_stream_map) + grpc_chttp2_stream_map_size(&t->new_stream_map); + GPR_ASSERT(new_stream_count <= GPR_UINT32_MAX); if (new_stream_count != t->global.concurrent_stream_count) { - t->global.concurrent_stream_count = new_stream_count; + t->global.concurrent_stream_count = (gpr_uint32)new_stream_count; maybe_start_some_streams(&t->global); } } @@ -905,7 +946,8 @@ static void cancel_from_api(grpc_chttp2_transport_global *transport_global, gpr_slice_buffer_add( &transport_global->qbuf, grpc_chttp2_rst_stream_create( - stream_global->id, grpc_chttp2_grpc_status_to_http2_error(status))); + 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); @@ -951,14 +993,14 @@ static void close_from_api(grpc_chttp2_transport_global *transport_global, *p++ = 's'; if (status < 10) { *p++ = 1; - *p++ = '0' + status; + *p++ = (gpr_uint8)('0' + status); } else { *p++ = 2; - *p++ = '0' + (status / 10); - *p++ = '0' + (status % 10); + *p++ = (gpr_uint8)('0' + (status / 10)); + *p++ = (gpr_uint8)('0' + (status % 10)); } GPR_ASSERT(p == GPR_SLICE_END_PTR(status_hdr)); - len += GPR_SLICE_LENGTH(status_hdr); + len += (gpr_uint32)GPR_SLICE_LENGTH(status_hdr); if (optional_message) { GPR_ASSERT(GPR_SLICE_LENGTH(*optional_message) < 127); @@ -978,23 +1020,23 @@ static void close_from_api(grpc_chttp2_transport_global *transport_global, *p++ = 'a'; *p++ = 'g'; *p++ = 'e'; - *p++ = GPR_SLICE_LENGTH(*optional_message); + *p++ = (gpr_uint8)GPR_SLICE_LENGTH(*optional_message); GPR_ASSERT(p == GPR_SLICE_END_PTR(message_pfx)); - len += GPR_SLICE_LENGTH(message_pfx); - len += GPR_SLICE_LENGTH(*optional_message); + len += (gpr_uint32)GPR_SLICE_LENGTH(message_pfx); + len += (gpr_uint32)GPR_SLICE_LENGTH(*optional_message); } hdr = gpr_slice_malloc(9); p = GPR_SLICE_START_PTR(hdr); - *p++ = len >> 16; - *p++ = len >> 8; - *p++ = len; + *p++ = (gpr_uint8)(len >> 16); + *p++ = (gpr_uint8)(len >> 8); + *p++ = (gpr_uint8)(len); *p++ = GRPC_CHTTP2_FRAME_HEADER; *p++ = GRPC_CHTTP2_DATA_FLAG_END_STREAM | GRPC_CHTTP2_DATA_FLAG_END_HEADERS; - *p++ = stream_global->id >> 24; - *p++ = stream_global->id >> 16; - *p++ = stream_global->id >> 8; - *p++ = stream_global->id; + *p++ = (gpr_uint8)(stream_global->id >> 24); + *p++ = (gpr_uint8)(stream_global->id >> 16); + *p++ = (gpr_uint8)(stream_global->id >> 8); + *p++ = (gpr_uint8)(stream_global->id); GPR_ASSERT(p == GPR_SLICE_END_PTR(hdr)); gpr_slice_buffer_add(&transport_global->qbuf, hdr); @@ -1052,82 +1094,90 @@ static void update_global_window(void *args, gpr_uint32 id, void *stream) { static void read_error_locked(grpc_chttp2_transport *t) { t->endpoint_reading = 0; if (!t->writing_active && t->ep) { - grpc_endpoint_destroy(t->ep); - t->ep = NULL; - /* safe as we still have a ref for read */ - UNREF_TRANSPORT(t, "disconnect"); + destroy_endpoint(t); } } /* tcp read callback */ -static void recv_data(void *tp, gpr_slice *slices, size_t nslices, - grpc_endpoint_cb_status error) { - grpc_chttp2_transport *t = tp; +static int recv_data_loop(grpc_chttp2_transport *t, int *success) { size_t i; - int unref = 0; + int keep_reading = 0; - switch (error) { - case GRPC_ENDPOINT_CB_SHUTDOWN: - case GRPC_ENDPOINT_CB_EOF: - case GRPC_ENDPOINT_CB_ERROR: - lock(t); + lock(t); + i = 0; + GPR_ASSERT(!t->parsing_active); + if (!t->closed) { + t->parsing_active = 1; + /* 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); + gpr_mu_unlock(&t->mu); + for (; i < t->read_buffer.count && + grpc_chttp2_perform_read(&t->parsing, t->read_buffer.slices[i]); + i++) + ; + gpr_mu_lock(&t->mu); + if (i != t->read_buffer.count) { drop_connection(t); - read_error_locked(t); - unlock(t); - unref = 1; - for (i = 0; i < nslices; i++) gpr_slice_unref(slices[i]); - break; - case GRPC_ENDPOINT_CB_OK: - lock(t); - i = 0; - GPR_ASSERT(!t->parsing_active); - if (!t->closed) { - t->parsing_active = 1; - /* 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); - gpr_mu_unlock(&t->mu); - for (; i < nslices && grpc_chttp2_perform_read(&t->parsing, slices[i]); - i++) { - gpr_slice_unref(slices[i]); - } - gpr_mu_lock(&t->mu); - if (i != nslices) { - drop_connection(t); - } - /* merge stream lists */ - grpc_chttp2_stream_map_move_into(&t->new_stream_map, - &t->parsing_stream_map); - t->global.concurrent_stream_count = - grpc_chttp2_stream_map_size(&t->parsing_stream_map); - if (t->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; - } - /* handle higher level things */ - grpc_chttp2_publish_reads(&t->global, &t->parsing); - t->parsing_active = 0; - } - if (i == nslices) { - grpc_chttp2_schedule_closure(&t->global, &t->reading_action, 1); - } else { - read_error_locked(t); - unref = 1; - } - unlock(t); - for (; i < nslices; i++) gpr_slice_unref(slices[i]); - break; + } + /* merge stream lists */ + grpc_chttp2_stream_map_move_into(&t->new_stream_map, + &t->parsing_stream_map); + t->global.concurrent_stream_count = + (gpr_uint32)grpc_chttp2_stream_map_size(&t->parsing_stream_map); + if (t->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; + } + /* handle higher level things */ + grpc_chttp2_publish_reads(&t->global, &t->parsing); + t->parsing_active = 0; + } + if (!*success || i != t->read_buffer.count) { + drop_connection(t); + read_error_locked(t); + } else if (!t->closed) { + keep_reading = 1; + REF_TRANSPORT(t, "keep_reading"); + prevent_endpoint_shutdown(t); } - if (unref) { + gpr_slice_buffer_reset_and_unref(&t->read_buffer); + unlock(t); + + if (keep_reading) { + int ret = -1; + switch (grpc_endpoint_read(t->ep, &t->read_buffer, &t->recv_data)) { + case GRPC_ENDPOINT_DONE: + *success = 1; + ret = 1; + break; + case GRPC_ENDPOINT_ERROR: + *success = 0; + ret = 1; + break; + case GRPC_ENDPOINT_PENDING: + ret = 0; + break; + } + allow_endpoint_shutdown_unlocked(t); + UNREF_TRANSPORT(t, "keep_reading"); + return ret; + } else { UNREF_TRANSPORT(t, "recv_data"); + return 0; } + + gpr_log(GPR_ERROR, "should never reach here"); + abort(); } -static void reading_action(void *pt, int iomgr_success_ignored) { - grpc_chttp2_transport *t = pt; - grpc_endpoint_notify_on_read(t->ep, recv_data, t); +static void recv_data(void *tp, int success) { + grpc_chttp2_transport *t = tp; + + while (recv_data_loop(t, &success)) + ; } /* @@ -1232,7 +1282,7 @@ grpc_transport *grpc_create_chttp2_transport( const grpc_channel_args *channel_args, grpc_endpoint *ep, grpc_mdctx *mdctx, int is_client) { grpc_chttp2_transport *t = gpr_malloc(sizeof(grpc_chttp2_transport)); - init_transport(t, channel_args, ep, mdctx, is_client); + init_transport(t, channel_args, ep, mdctx, is_client != 0); return &t->base; } @@ -1240,5 +1290,6 @@ void grpc_chttp2_transport_start_reading(grpc_transport *transport, gpr_slice *slices, size_t nslices) { grpc_chttp2_transport *t = (grpc_chttp2_transport *)transport; REF_TRANSPORT(t, "recv_data"); /* matches unref inside recv_data */ - recv_data(t, slices, nslices, GRPC_ENDPOINT_CB_OK); + gpr_slice_buffer_addn(&t->read_buffer, slices, nslices); + recv_data(t, 1); } diff --git a/src/core/transport/metadata.c b/src/core/transport/metadata.c index 61638764a6..9d135f4356 100644 --- a/src/core/transport/metadata.c +++ b/src/core/transport/metadata.c @@ -186,7 +186,8 @@ grpc_mdctx *grpc_mdctx_create(void) { /* This seed is used to prevent remote connections from controlling hash table * collisions. It needs to be somewhat unpredictable to a remote connection. */ - return grpc_mdctx_create_with_seed(gpr_now(GPR_CLOCK_REALTIME).tv_nsec); + return grpc_mdctx_create_with_seed( + (gpr_uint32)gpr_now(GPR_CLOCK_REALTIME).tv_nsec); } static void discard_metadata(grpc_mdctx *ctx) { @@ -333,7 +334,7 @@ grpc_mdstr *grpc_mdstr_from_string(grpc_mdctx *ctx, const char *str, grpc_mdstr *ret; for (i = 0; i < len; i++) { if (str[i] >= 'A' && str[i] <= 'Z') { - copy[i] = str[i] - 'A' + 'a'; + copy[i] = (char)(str[i] - 'A' + 'a'); } else { copy[i] = str[i]; } @@ -378,7 +379,7 @@ grpc_mdstr *grpc_mdstr_from_buffer(grpc_mdctx *ctx, const gpr_uint8 *buf, s->slice.refcount = NULL; memcpy(s->slice.data.inlined.bytes, buf, length); s->slice.data.inlined.bytes[length] = 0; - s->slice.data.inlined.length = length; + s->slice.data.inlined.length = (gpr_uint8)length; } else { /* string data goes after the internal_string header, and we +1 for null terminator */ diff --git a/src/core/transport/transport.h b/src/core/transport/transport.h index 92c1f38c5e..6e1ec2f64c 100644 --- a/src/core/transport/transport.h +++ b/src/core/transport/transport.h @@ -75,7 +75,7 @@ typedef struct grpc_transport_stream_op { /** The number of bytes this peer is currently prepared to receive. These bytes will be eventually used to replenish per-stream flow control windows. */ - gpr_uint32 max_recv_bytes; + size_t max_recv_bytes; grpc_iomgr_closure *on_done_recv; grpc_pollset *bind_pollset; |