Fixing stuff

pull/2149/head
Craig Tiller 10 years ago
parent 6905b7ce0c
commit f73fcd1cb9
  1. 3
      src/core/transport/chttp2/internal.h
  2. 5
      src/core/transport/chttp2/parsing.c
  3. 34
      src/core/transport/chttp2_transport.c
  4. 3
      src/core/transport/stream_op.c

@ -595,6 +595,9 @@ int grpc_chttp2_list_pop_cancelled_waiting_for_parsing(
void grpc_chttp2_list_add_read_write_state_changed(
grpc_chttp2_transport_global *transport_global,
grpc_chttp2_stream_global *stream_global);
int grpc_chttp2_list_pop_read_write_state_changed(
grpc_chttp2_transport_global *transport_global,
grpc_chttp2_stream_global **stream_global);
void grpc_chttp2_list_add_incoming_window_state_changed(
grpc_chttp2_transport_global *transport_global,

@ -185,6 +185,11 @@ void grpc_chttp2_publish_reads(
}
/* publish incoming stream ops */
if (stream_parsing->data_parser.incoming_sopb.nops > 0) {
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);
}
}
}

@ -80,6 +80,7 @@ static void unlock(grpc_chttp2_transport *t);
static void unlock_check_cancellations(grpc_chttp2_transport *t);
static void unlock_check_channel_callbacks(grpc_chttp2_transport *t);
static void unlock_check_reads(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);
@ -345,6 +346,7 @@ static int init_stream(grpc_transport *gt, grpc_stream *gs,
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);
grpc_chttp2_data_parser_init(&s->parsing.data_parser);
ref_transport(t);
@ -446,6 +448,7 @@ static void unlock(grpc_chttp2_transport *t) {
grpc_chttp2_schedule_closure(&t->global, &t->writing_action, 1);
}
unlock_check_cancellations(t);
unlock_check_reads(t);
/* unlock_check_parser(t); */
unlock_check_channel_callbacks(t);
@ -695,6 +698,27 @@ static void unlock_check_cancellations(grpc_chttp2_transport *t) {
#endif
}
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 unlock_check_reads(grpc_chttp2_transport *t) {
grpc_chttp2_stream_global *stream_global;
while (grpc_chttp2_pop_read_write_state_changed(&t->global, &stream_global)) {
if (!stream_global->publish_sopb) {
continue;
}
grpc_sopb_swap(stream_global->publish_sopb, &stream_global->incoming_sopb);
/* TODO(ctiller): we need to not publish closed until !writing, or define a new STREAM_DELETABLE state */
stream_global->published_state = *stream_global->publish_state = compute_state(stream_global->write_closed, stream_global->read_closed && !stream_global->in_stream_map);
}
}
static void cancel_from_api(grpc_chttp2_transport_global *transport_global,
grpc_chttp2_stream_global *stream_global,
grpc_status_code status) {
@ -928,16 +952,6 @@ static void recv_data(void *tp, gpr_slice *slices, size_t nslices,
* CALLBACK LOOP
*/
#if 0
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;
}
#endif
typedef struct {
grpc_chttp2_transport *t;
gpr_uint32 error;

@ -164,7 +164,6 @@ void grpc_sopb_append(grpc_stream_op_buffer *sopb, grpc_stream_op *ops,
}
void grpc_sopb_move_to(grpc_stream_op_buffer *src, grpc_stream_op_buffer *dst) {
size_t i;
if (src->nops == 0) {
return;
}
@ -173,7 +172,7 @@ void grpc_sopb_move_to(grpc_stream_op_buffer *src, grpc_stream_op_buffer *dst) {
return;
}
grpc_sopb_append(dst, src->ops, src->nops);
src->ops = 0;
src->nops = 0;
}
static void assert_valid_list(grpc_mdelem_list *list) {

Loading…
Cancel
Save