aboutsummaryrefslogtreecommitdiffhomepage
path: root/src/core
diff options
context:
space:
mode:
authorGravatar Craig Tiller <ctiller@google.com>2015-05-20 10:33:37 -0700
committerGravatar Craig Tiller <ctiller@google.com>2015-05-20 10:33:37 -0700
commitbf6bd2303c2693373c4df161e11b6ce93f5d6ae3 (patch)
tree3196fcfb73f5df28815df87ba90d398809a46e3b /src/core
parent1e0452b63b335c5f00d35d3774a4b4a60a21b80d (diff)
parenta42c1fe8aeddff6e1fe13b7ee2d990999f2867c6 (diff)
Merge github.com:grpc/grpc into mmm-mmm-mmm-mmm
Diffstat (limited to 'src/core')
-rw-r--r--src/core/iomgr/tcp_posix.c4
-rw-r--r--src/core/transport/chttp2/frame.h2
-rw-r--r--src/core/transport/chttp2/frame_rst_stream.c40
-rw-r--r--src/core/transport/chttp2/frame_rst_stream.h11
-rw-r--r--src/core/transport/chttp2_transport.c116
5 files changed, 131 insertions, 42 deletions
diff --git a/src/core/iomgr/tcp_posix.c b/src/core/iomgr/tcp_posix.c
index f7dae5f86c..cd6b2ecae6 100644
--- a/src/core/iomgr/tcp_posix.c
+++ b/src/core/iomgr/tcp_posix.c
@@ -138,8 +138,10 @@ static void slice_state_remove_prefix(grpc_tcp_slice_state *state,
native "trim the first N bytes" operation to splice */
/* TODO(klempner): This really shouldn't be modifying the current slice
unless we own the slices array. */
- *current_slice = gpr_slice_split_tail(current_slice, prefix_bytes);
+ gpr_slice tail;
+ tail = gpr_slice_split_tail(current_slice, prefix_bytes);
gpr_slice_unref(*current_slice);
+ *current_slice = tail;
return;
} else {
gpr_slice_unref(*current_slice);
diff --git a/src/core/transport/chttp2/frame.h b/src/core/transport/chttp2/frame.h
index ac76c4cc9c..c9e3e13042 100644
--- a/src/core/transport/chttp2/frame.h
+++ b/src/core/transport/chttp2/frame.h
@@ -53,12 +53,14 @@ typedef struct {
gpr_uint8 send_ping_ack;
gpr_uint8 process_ping_reply;
gpr_uint8 goaway;
+ gpr_uint8 rst_stream;
gpr_int64 initial_window_update;
gpr_uint32 window_update;
gpr_uint32 goaway_last_stream_index;
gpr_uint32 goaway_error;
gpr_slice goaway_text;
+ gpr_uint32 rst_stream_reason;
} grpc_chttp2_parse_state;
#define GRPC_CHTTP2_FRAME_DATA 0
diff --git a/src/core/transport/chttp2/frame_rst_stream.c b/src/core/transport/chttp2/frame_rst_stream.c
index 368ca86481..3016aac7a2 100644
--- a/src/core/transport/chttp2/frame_rst_stream.c
+++ b/src/core/transport/chttp2/frame_rst_stream.c
@@ -32,6 +32,9 @@
*/
#include "src/core/transport/chttp2/frame_rst_stream.h"
+
+#include <grpc/support/log.h>
+
#include "src/core/transport/chttp2/frame.h"
gpr_slice grpc_chttp2_rst_stream_create(gpr_uint32 id, gpr_uint32 code) {
@@ -54,3 +57,40 @@ gpr_slice grpc_chttp2_rst_stream_create(gpr_uint32 id, gpr_uint32 code) {
return slice;
}
+
+grpc_chttp2_parse_error grpc_chttp2_rst_stream_parser_begin_frame(
+ grpc_chttp2_rst_stream_parser *parser, gpr_uint32 length, gpr_uint8 flags) {
+ if (length != 4) {
+ gpr_log(GPR_ERROR, "invalid rst_stream: length=%d, flags=%02x", length, flags);
+ return GRPC_CHTTP2_CONNECTION_ERROR;
+ }
+ parser->byte = 0;
+ return GRPC_CHTTP2_PARSE_OK;
+}
+
+grpc_chttp2_parse_error grpc_chttp2_rst_stream_parser_parse(
+ void *parser, grpc_chttp2_parse_state *state, gpr_slice slice,
+ int is_last) {
+ gpr_uint8 *const beg = GPR_SLICE_START_PTR(slice);
+ gpr_uint8 *const end = GPR_SLICE_END_PTR(slice);
+ gpr_uint8 *cur = beg;
+ grpc_chttp2_rst_stream_parser *p = parser;
+
+ while (p->byte != 4 && cur != end) {
+ p->reason_bytes[p->byte] = *cur;
+ cur++;
+ p->byte++;
+ }
+
+ if (p->byte == 4) {
+ GPR_ASSERT(is_last);
+ state->rst_stream = 1;
+ state->rst_stream_reason =
+ (((gpr_uint32)p->reason_bytes[0]) << 24) |
+ (((gpr_uint32)p->reason_bytes[1]) << 16) |
+ (((gpr_uint32)p->reason_bytes[2]) << 8) |
+ (((gpr_uint32)p->reason_bytes[3]));
+ }
+
+ return GRPC_CHTTP2_PARSE_OK;
+}
diff --git a/src/core/transport/chttp2/frame_rst_stream.h b/src/core/transport/chttp2/frame_rst_stream.h
index 2d3ee18637..07a3c98d03 100644
--- a/src/core/transport/chttp2/frame_rst_stream.h
+++ b/src/core/transport/chttp2/frame_rst_stream.h
@@ -35,7 +35,18 @@
#define GRPC_INTERNAL_CORE_TRANSPORT_CHTTP2_FRAME_RST_STREAM_H
#include <grpc/support/slice.h>
+#include "src/core/transport/chttp2/frame.h"
+
+typedef struct {
+ gpr_uint8 byte;
+ gpr_uint8 reason_bytes[4];
+} grpc_chttp2_rst_stream_parser;
gpr_slice grpc_chttp2_rst_stream_create(gpr_uint32 stream_id, gpr_uint32 code);
+grpc_chttp2_parse_error grpc_chttp2_rst_stream_parser_begin_frame(
+ grpc_chttp2_rst_stream_parser *parser, gpr_uint32 length, gpr_uint8 flags);
+grpc_chttp2_parse_error grpc_chttp2_rst_stream_parser_parse(
+ void *parser, grpc_chttp2_parse_state *state, gpr_slice slice, int is_last);
+
#endif /* GRPC_INTERNAL_CORE_TRANSPORT_CHTTP2_FRAME_RST_STREAM_H */
diff --git a/src/core/transport/chttp2_transport.c b/src/core/transport/chttp2_transport.c
index a6f9f782a1..678b361036 100644
--- a/src/core/transport/chttp2_transport.c
+++ b/src/core/transport/chttp2_transport.c
@@ -154,7 +154,13 @@ typedef enum {
WRITE_STATE_OPEN,
WRITE_STATE_QUEUED_CLOSE,
WRITE_STATE_SENT_CLOSE
-} WRITE_STATE;
+} write_state;
+
+typedef enum {
+ DONT_SEND_CLOSED = 0,
+ SEND_CLOSED,
+ SEND_CLOSED_WITH_RST_STREAM
+} send_closed;
typedef struct {
stream *head;
@@ -267,6 +273,7 @@ struct transport {
grpc_chttp2_window_update_parser window_update;
grpc_chttp2_settings_parser settings;
grpc_chttp2_ping_parser ping;
+ grpc_chttp2_rst_stream_parser rst_stream;
} simple_parsers;
/* goaway */
@@ -312,8 +319,8 @@ struct stream {
/* 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' */
- WRITE_STATE write_state;
- gpr_uint8 send_closed;
+ write_state write_state;
+ send_closed send_closed;
gpr_uint8 read_closed;
gpr_uint8 cancelled;
@@ -937,7 +944,11 @@ static int prepare_write(transport *t) {
if (s->write_state == WRITE_STATE_QUEUED_CLOSE &&
s->outgoing_sopb->nops == 0) {
- s->send_closed = 1;
+ if (!t->is_client && !s->read_closed) {
+ s->send_closed = SEND_CLOSED_WITH_RST_STREAM;
+ } else {
+ s->send_closed = SEND_CLOSED;
+ }
}
if (s->writing_sopb.nops > 0 || s->send_closed) {
stream_list_join(t, s, WRITING);
@@ -982,9 +993,12 @@ static void finalize_outbuf(transport *t) {
while ((s = stream_list_remove_head(t, WRITING))) {
grpc_chttp2_encode(s->writing_sopb.ops, s->writing_sopb.nops,
- s->send_closed, s->id, &t->hpack_compressor, &t->outbuf);
+ s->send_closed != DONT_SEND_CLOSED, s->id, &t->hpack_compressor, &t->outbuf);
s->writing_sopb.nops = 0;
- if (s->send_closed) {
+ if (s->send_closed == SEND_CLOSED_WITH_RST_STREAM) {
+ gpr_slice_buffer_add(&t->outbuf, grpc_chttp2_rst_stream_create(s->id, GRPC_CHTTP2_NO_ERROR));
+ }
+ if (s->send_closed != DONT_SEND_CLOSED) {
stream_list_join(t, s, WRITTEN_CLOSED);
}
}
@@ -999,9 +1013,10 @@ static void finish_write_common(transport *t, int success) {
}
while ((s = stream_list_remove_head(t, WRITTEN_CLOSED))) {
s->write_state = WRITE_STATE_SENT_CLOSE;
- if (1||!s->cancelled) {
- maybe_finish_read(t, s);
+ if (!t->is_client) {
+ s->read_closed = 1;
}
+ maybe_finish_read(t, s);
}
t->outbuf.count = 0;
t->outbuf.length = 0;
@@ -1214,12 +1229,14 @@ static void cancel_stream_inner(transport *t, stream *s, gpr_uint32 id,
if (s) {
/* clear out any unreported input & output: nobody cares anymore */
had_outgoing = s->outgoing_sopb && s->outgoing_sopb->nops != 0;
- schedule_nuke_sopb(t, &s->parser.incoming_sopb);
- if (s->outgoing_sopb) {
- schedule_nuke_sopb(t, s->outgoing_sopb);
- s->outgoing_sopb = NULL;
- stream_list_remove(t, s, WRITABLE);
- schedule_cb(t, s->send_done_closure, 0);
+ if (error_code != GRPC_CHTTP2_NO_ERROR) {
+ schedule_nuke_sopb(t, &s->parser.incoming_sopb);
+ if (s->outgoing_sopb) {
+ schedule_nuke_sopb(t, s->outgoing_sopb);
+ s->outgoing_sopb = NULL;
+ stream_list_remove(t, s, WRITABLE);
+ schedule_cb(t, s->send_done_closure, 0);
+ }
}
if (s->cancelled) {
send_rst = 0;
@@ -1228,31 +1245,34 @@ static void cancel_stream_inner(transport *t, stream *s, gpr_uint32 id,
s->cancelled = 1;
stream_list_join(t, s, CANCELLED);
- gpr_ltoa(local_status, buffer);
- add_incoming_metadata(
- t, s,
- grpc_mdelem_from_strings(t->metadata_context, "grpc-status", buffer));
- if (!optional_message) {
- switch (local_status) {
- case GRPC_STATUS_CANCELLED:
- add_incoming_metadata(
- t, s, grpc_mdelem_from_strings(t->metadata_context,
- "grpc-message", "Cancelled"));
- break;
- default:
- break;
- }
- } else {
+ if (error_code != GRPC_CHTTP2_NO_ERROR) {
+ /* synthesize a status if we don't believe we'll get one */
+ gpr_ltoa(local_status, buffer);
add_incoming_metadata(
t, s,
- grpc_mdelem_from_metadata_strings(
- t->metadata_context,
- grpc_mdstr_from_string(t->metadata_context, "grpc-message"),
- grpc_mdstr_ref(optional_message)));
+ grpc_mdelem_from_strings(t->metadata_context, "grpc-status", buffer));
+ if (!optional_message) {
+ switch (local_status) {
+ case GRPC_STATUS_CANCELLED:
+ add_incoming_metadata(
+ t, s, grpc_mdelem_from_strings(t->metadata_context,
+ "grpc-message", "Cancelled"));
+ break;
+ default:
+ break;
+ }
+ } else {
+ add_incoming_metadata(
+ t, s,
+ grpc_mdelem_from_metadata_strings(
+ t->metadata_context,
+ grpc_mdstr_from_string(t->metadata_context, "grpc-message"),
+ grpc_mdstr_ref(optional_message)));
+ }
+ add_metadata_batch(t, s);
}
- add_metadata_batch(t, s);
- maybe_finish_read(t, s);
}
+ maybe_finish_read(t, s);
}
if (!id) send_rst = 0;
if (send_rst) {
@@ -1527,6 +1547,19 @@ static int init_ping_parser(transport *t) {
return ok;
}
+static int init_rst_stream_parser(transport *t) {
+ int ok = GRPC_CHTTP2_PARSE_OK ==
+ grpc_chttp2_rst_stream_parser_begin_frame(&t->simple_parsers.rst_stream,
+ t->incoming_frame_size,
+ t->incoming_frame_flags);
+ if (!ok) {
+ drop_connection(t);
+ }
+ t->parser = grpc_chttp2_rst_stream_parser_parse;
+ t->parser_data = &t->simple_parsers.rst_stream;
+ return ok;
+}
+
static int init_goaway_parser(transport *t) {
int ok =
GRPC_CHTTP2_PARSE_OK ==
@@ -1581,12 +1614,7 @@ static int init_frame_parser(transport *t) {
gpr_log(GPR_ERROR, "Unexpected CONTINUATION frame");
return 0;
case GRPC_CHTTP2_FRAME_RST_STREAM:
- /* TODO(ctiller): actually parse the reason */
- cancel_stream_id(
- t, t->incoming_stream_id,
- grpc_chttp2_http2_error_to_grpc_status(GRPC_CHTTP2_CANCEL),
- GRPC_CHTTP2_CANCEL, 0);
- return init_skip_frame(t, 0);
+ return init_rst_stream_parser(t);
case GRPC_CHTTP2_FRAME_SETTINGS:
return init_settings_frame_parser(t);
case GRPC_CHTTP2_FRAME_WINDOW_UPDATE:
@@ -1650,6 +1678,12 @@ static int parse_frame_slice(transport *t, gpr_slice slice, int is_last) {
if (st.goaway) {
add_goaway(t, st.goaway_error, st.goaway_text);
}
+ if (st.rst_stream) {
+ cancel_stream_id(
+ t, t->incoming_stream_id,
+ grpc_chttp2_http2_error_to_grpc_status(st.rst_stream_reason),
+ st.rst_stream_reason, 0);
+ }
if (st.process_ping_reply) {
for (i = 0; i < t->ping_count; i++) {
if (0 ==