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

Handle cancelling writes whilst writing

Browse Source
pull/4451/head
Craig Tiller 9 years ago
parent 7954e36059
commit abb2e3dc00
3 changed files with 57 additions and 10 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Split View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 13
      src/core/transport/chttp2/internal.h
  2. 20
      src/core/transport/chttp2/stream_lists.c
  3. 34
      src/core/transport/chttp2_transport.c

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

@ -65,6 +65,7 @@ typedef enum {
GRPC_CHTTP2_LIST_WRITTEN,
GRPC_CHTTP2_LIST_PARSING_SEEN,
GRPC_CHTTP2_LIST_CLOSED_WAITING_FOR_PARSING,
GRPC_CHTTP2_LIST_CLOSED_WAITING_FOR_WRITING,
GRPC_CHTTP2_LIST_STALLED_BY_TRANSPORT,
/** streams that are waiting to start because there are too many concurrent
streams on the connection */
@ -283,6 +284,9 @@ struct grpc_chttp2_transport_parsing {
gpr_slice goaway_text;
gpr_int64 outgoing_window;
/** pings awaiting responses */
grpc_chttp2_outstanding_ping pings;
};
struct grpc_chttp2_transport {
@ -391,8 +395,6 @@ typedef struct {
gpr_uint8 write_closed;
/** is this stream reading half-closed (boolean) */
gpr_uint8 read_closed;
/** 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;
/** has this stream seen an error? if 1, then pending incoming frames
@ -586,6 +588,13 @@ int grpc_chttp2_list_pop_closed_waiting_for_parsing(
grpc_chttp2_transport_global *transport_global,
grpc_chttp2_stream_global **stream_global);
void grpc_chttp2_list_add_closed_waiting_for_writing(
grpc_chttp2_transport_global *transport_global,
grpc_chttp2_stream_global *stream_global);
int grpc_chttp2_list_pop_closed_waiting_for_writing(
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(

20
src/core/transport/chttp2/stream_lists.c
Unescape View File

@ -353,6 +353,26 @@ int grpc_chttp2_list_pop_closed_waiting_for_parsing(
return r;
}
void grpc_chttp2_list_add_closed_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_CLOSED_WAITING_FOR_WRITING);
}
int grpc_chttp2_list_pop_closed_waiting_for_writing(
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_WRITING);
if (r != 0) {
*stream_global = &stream->global;
}
return r;
}
void grpc_chttp2_register_stream(grpc_chttp2_transport *t,
grpc_chttp2_stream *s) {
stream_list_add_tail(t, s, GRPC_CHTTP2_LIST_ALL_STREAMS);

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

@ -134,6 +134,9 @@ static void connectivity_state_set(
static void check_read_ops(grpc_exec_ctx *exec_ctx,
grpc_chttp2_transport_global *transport_global);
static void fail_pending_writes(grpc_exec_ctx *exec_ctx,
grpc_chttp2_stream_global *stream_global);
/*
* CONSTRUCTION/DESTRUCTION/REFCOUNTING
*/
@ -625,6 +628,7 @@ void grpc_chttp2_terminate_writing(grpc_exec_ctx *exec_ctx,
void *transport_writing_ptr, int success) {
grpc_chttp2_transport_writing *transport_writing = transport_writing_ptr;
grpc_chttp2_transport *t = TRANSPORT_FROM_WRITING(transport_writing);
grpc_chttp2_stream_global *stream_global;
GPR_TIMER_BEGIN("grpc_chttp2_terminate_writing", 0);
@ -638,6 +642,11 @@ void grpc_chttp2_terminate_writing(grpc_exec_ctx *exec_ctx,
grpc_chttp2_cleanup_writing(exec_ctx, &t->global, &t->writing);
while (grpc_chttp2_list_pop_closed_waiting_for_writing(&t->global, &stream_global)) {
fail_pending_writes(exec_ctx, stream_global);
GRPC_CHTTP2_STREAM_UNREF(exec_ctx, stream_global, "finish_writes");
}
/* leave the writing flag up on shutdown to prevent further writes in unlock()
from starting */
t->writing_active = 0;
@ -1107,6 +1116,16 @@ void grpc_chttp2_fake_status(grpc_exec_ctx *exec_ctx,
}
}
static void fail_pending_writes(grpc_exec_ctx *exec_ctx,
grpc_chttp2_stream_global *stream_global) {
grpc_chttp2_complete_closure_step(
exec_ctx, &stream_global->send_initial_metadata_finished, 0);
grpc_chttp2_complete_closure_step(
exec_ctx, &stream_global->send_trailing_metadata_finished, 0);
grpc_chttp2_complete_closure_step(exec_ctx,
&stream_global->send_message_finished, 0);
}
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,
@ -1123,12 +1142,13 @@ void grpc_chttp2_mark_stream_closed(
}
if (close_writes && !stream_global->write_closed) {
stream_global->write_closed = 1;
grpc_chttp2_complete_closure_step(
exec_ctx, &stream_global->send_initial_metadata_finished, 0);
grpc_chttp2_complete_closure_step(
exec_ctx, &stream_global->send_trailing_metadata_finished, 0);
grpc_chttp2_complete_closure_step(exec_ctx,
&stream_global->send_message_finished, 0);
if (TRANSPORT_FROM_GLOBAL(transport_global)->writing_active) {
GRPC_CHTTP2_STREAM_REF(stream_global, "finish_writes");
grpc_chttp2_list_add_closed_waiting_for_writing(transport_global,
stream_global);
} else {
fail_pending_writes(exec_ctx, stream_global);
}
}
if (stream_global->read_closed && stream_global->write_closed) {
if (stream_global->id != 0 &&
@ -1140,7 +1160,6 @@ void grpc_chttp2_mark_stream_closed(
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");
}
}
@ -1354,7 +1373,6 @@ static void recv_data(grpc_exec_ctx *exec_ctx, void *tp, int success) {
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");
}
}

Write Preview
Loading…
Cancel
Save