Implement more missing pieces

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

@ -430,7 +430,8 @@ typedef struct {
/** 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 *incoming_sopb;
grpc_stream_op_buffer *publish_sopb;
grpc_stream_op_buffer incoming_sopb;
/** incoming metadata */
grpc_chttp2_incoming_metadata_buffer incoming_metadata;

@ -183,6 +183,8 @@ void grpc_chttp2_publish_reads(
grpc_chttp2_list_add_read_write_state_changed(transport_global,
stream_global);
}
/* publish incoming stream ops */
}
}

@ -384,7 +384,7 @@ static void destroy_stream(grpc_transport *gt, grpc_stream *gs) {
gpr_mu_unlock(&t->mu);
GPR_ASSERT(s->global.outgoing_sopb == NULL);
GPR_ASSERT(s->global.incoming_sopb == NULL);
GPR_ASSERT(s->global.publish_sopb == NULL);
grpc_sopb_destroy(&s->writing.sopb);
grpc_chttp2_data_parser_destroy(&s->parsing.data_parser);
grpc_chttp2_incoming_metadata_buffer_destroy(&s->parsing.incoming_metadata);
@ -604,11 +604,11 @@ static void perform_op_locked(grpc_chttp2_transport_global *transport_global,
}
if (op->recv_ops) {
GPR_ASSERT(stream_global->incoming_sopb == NULL);
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->incoming_sopb = op->recv_ops;
stream_global->incoming_sopb->nops = 0;
stream_global->publish_sopb = op->recv_ops;
stream_global->publish_sopb->nops = 0;
grpc_chttp2_incoming_metadata_live_op_buffer_end(
&stream_global->outstanding_metadata);
grpc_chttp2_list_add_read_write_state_changed(transport_global,

@ -163,6 +163,19 @@ void grpc_sopb_append(grpc_stream_op_buffer *sopb, grpc_stream_op *ops,
sopb->nops = new_nops;
}
void grpc_sopb_move_to(grpc_stream_op_buffer *src, grpc_stream_op_buffer *dst) {
size_t i;
if (src->nops == 0) {
return;
}
if (dst->nops == 0) {
grpc_sopb_swap(src, dst);
return;
}
grpc_sopb_append(dst, src->ops, src->nops);
src->ops = 0;
}
static void assert_valid_list(grpc_mdelem_list *list) {
#ifndef NDEBUG
grpc_linked_mdelem *l;

@ -159,6 +159,8 @@ void grpc_sopb_add_slice(grpc_stream_op_buffer *sopb, gpr_slice slice);
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 */

Loading…
Cancel
Save