From 3d8b32d8b3d7e85d0588064994efa6763d6fec02 Mon Sep 17 00:00:00 2001 From: "Mark D. Roth" Date: Fri, 9 Mar 2018 13:25:40 -0800 Subject: Convert byte_stream API to C++. --- .../transport/chttp2/transport/chttp2_transport.cc | 293 +++++++++------------ .../ext/transport/chttp2/transport/frame_data.cc | 32 +-- .../ext/transport/chttp2/transport/frame_data.h | 9 +- src/core/ext/transport/chttp2/transport/internal.h | 77 ++++-- 4 files changed, 195 insertions(+), 216 deletions(-) (limited to 'src/core/ext/transport/chttp2') diff --git a/src/core/ext/transport/chttp2/transport/chttp2_transport.cc b/src/core/ext/transport/chttp2/transport/chttp2_transport.cc index 56aaada912..a4d616d778 100644 --- a/src/core/ext/transport/chttp2/transport/chttp2_transport.cc +++ b/src/core/ext/transport/chttp2/transport/chttp2_transport.cc @@ -39,6 +39,7 @@ #include "src/core/lib/debug/stats.h" #include "src/core/lib/gpr/env.h" #include "src/core/lib/gpr/string.h" +#include "src/core/lib/gprpp/memory.h" #include "src/core/lib/http/parser.h" #include "src/core/lib/iomgr/executor.h" #include "src/core/lib/iomgr/timer.h" @@ -117,12 +118,6 @@ static void connectivity_state_set(grpc_chttp2_transport* t, grpc_connectivity_state state, grpc_error* error, const char* reason); -static void incoming_byte_stream_destroy_locked(void* byte_stream, - grpc_error* error_ignored); -static void incoming_byte_stream_publish_error( - grpc_chttp2_incoming_byte_stream* bs, grpc_error* error); -static void incoming_byte_stream_unref(grpc_chttp2_incoming_byte_stream* bs); - static void benign_reclaimer_locked(void* t, grpc_error* error); static void destructive_reclaimer_locked(void* t, grpc_error* error); @@ -662,8 +657,8 @@ static int init_stream(grpc_transport* gt, grpc_stream* gs, s->t = t; s->refcount = refcount; /* We reserve one 'active stream' that's dropped when the stream is - read-closed. The others are for incoming_byte_streams that are actively - reading */ + read-closed. The others are for Chttp2IncomingByteStreams that are + actively reading */ GRPC_CHTTP2_STREAM_REF(s, "chttp2"); grpc_chttp2_incoming_metadata_buffer_init(&s->metadata_buffer[0], arena); @@ -1256,8 +1251,7 @@ static void continue_fetching_send_locked(grpc_chttp2_transport* t, abort(); /* TODO(ctiller): what cleanup here? */ return; /* early out */ } - if (s->fetched_send_message_length == s->fetching_send_message->length) { - grpc_byte_stream_destroy(s->fetching_send_message); + if (s->fetched_send_message_length == s->fetching_send_message->length()) { int64_t notify_offset = s->next_message_end_offset; if (notify_offset <= s->flow_controlled_bytes_written) { grpc_chttp2_complete_closure_step( @@ -1274,20 +1268,19 @@ static void continue_fetching_send_locked(grpc_chttp2_transport* t, cb->closure = s->fetching_send_message_finished; s->fetching_send_message_finished = nullptr; grpc_chttp2_write_cb** list = - s->fetching_send_message->flags & GRPC_WRITE_THROUGH + s->fetching_send_message->flags() & GRPC_WRITE_THROUGH ? &s->on_write_finished_cbs : &s->on_flow_controlled_cbs; cb->next = *list; *list = cb; } - s->fetching_send_message = nullptr; + s->fetching_send_message.reset(); return; /* early out */ - } else if (grpc_byte_stream_next(s->fetching_send_message, UINT32_MAX, - &s->complete_fetch_locked)) { - grpc_error* error = - grpc_byte_stream_pull(s->fetching_send_message, &s->fetching_slice); + } else if (s->fetching_send_message->Next(UINT32_MAX, + &s->complete_fetch_locked)) { + grpc_error* error = s->fetching_send_message->Pull(&s->fetching_slice); if (error != GRPC_ERROR_NONE) { - grpc_byte_stream_destroy(s->fetching_send_message); + s->fetching_send_message.reset(); grpc_chttp2_cancel_stream(t, s, error); } else { add_fetched_slice_locked(t, s); @@ -1300,14 +1293,14 @@ static void complete_fetch_locked(void* gs, grpc_error* error) { grpc_chttp2_stream* s = static_cast(gs); grpc_chttp2_transport* t = s->t; if (error == GRPC_ERROR_NONE) { - error = grpc_byte_stream_pull(s->fetching_send_message, &s->fetching_slice); + error = s->fetching_send_message->Pull(&s->fetching_slice); if (error == GRPC_ERROR_NONE) { add_fetched_slice_locked(t, s); continue_fetching_send_locked(t, s); } } if (error != GRPC_ERROR_NONE) { - grpc_byte_stream_destroy(s->fetching_send_message); + s->fetching_send_message.reset(); grpc_chttp2_cancel_stream(t, s, error); } } @@ -1439,7 +1432,7 @@ static void perform_stream_op_locked(void* stream_op, GPR_ASSERT(s->id != 0); grpc_chttp2_mark_stream_writable(t, s); if (!(op->send_message && - (op->payload->send_message.send_message->flags & + (op->payload->send_message.send_message->flags() & GRPC_WRITE_BUFFER_HINT))) { grpc_chttp2_initiate_write( t, GRPC_CHTTP2_INITIATE_WRITE_SEND_INITIAL_METADATA); @@ -1466,7 +1459,7 @@ static void perform_stream_op_locked(void* stream_op, if (op->send_message) { GRPC_STATS_INC_HTTP2_OP_SEND_MESSAGE(); GRPC_STATS_INC_HTTP2_SEND_MESSAGE_SIZE( - op->payload->send_message.send_message->length); + op->payload->send_message.send_message->length()); on_complete->next_data.scratch |= CLOSURE_BARRIER_MAY_COVER_WRITE; s->fetching_send_message_finished = add_closure_barrier(op->on_complete); if (s->write_closed) { @@ -1475,7 +1468,7 @@ static void perform_stream_op_locked(void* stream_op, // streaming call might send another message before getting a // recv_message failure, breaking out of its loop, and then // starting recv_trailing_metadata. - grpc_byte_stream_destroy(op->payload->send_message.send_message); + op->payload->send_message.send_message.reset(); grpc_chttp2_complete_closure_step( t, s, &s->fetching_send_message_finished, t->is_client && s->received_trailing_metadata @@ -1488,14 +1481,15 @@ static void perform_stream_op_locked(void* stream_op, GPR_ASSERT(s->fetching_send_message == nullptr); uint8_t* frame_hdr = grpc_slice_buffer_tiny_add( &s->flow_controlled_buffer, GRPC_HEADER_SIZE_IN_BYTES); - uint32_t flags = op_payload->send_message.send_message->flags; + uint32_t flags = op_payload->send_message.send_message->flags(); frame_hdr[0] = (flags & GRPC_WRITE_INTERNAL_COMPRESS) != 0; - size_t len = op_payload->send_message.send_message->length; + size_t len = op_payload->send_message.send_message->length(); frame_hdr[1] = static_cast(len >> 24); frame_hdr[2] = static_cast(len >> 16); frame_hdr[3] = static_cast(len >> 8); frame_hdr[4] = static_cast(len); - s->fetching_send_message = op_payload->send_message.send_message; + s->fetching_send_message = + std::move(op_payload->send_message.send_message); s->fetched_send_message_length = 0; s->next_message_end_offset = s->flow_controlled_bytes_written + @@ -1947,12 +1941,12 @@ static void remove_stream(grpc_chttp2_transport* t, uint32_t id, } if (s->pending_byte_stream) { if (s->on_next != nullptr) { - grpc_chttp2_incoming_byte_stream* bs = s->data_parser.parsing_frame; + grpc_core::Chttp2IncomingByteStream* bs = s->data_parser.parsing_frame; if (error == GRPC_ERROR_NONE) { error = GRPC_ERROR_CREATE_FROM_STATIC_STRING("Truncated message"); } - incoming_byte_stream_publish_error(bs, error); - incoming_byte_stream_unref(bs); + bs->PublishError(error); + bs->Unref(); s->data_parser.parsing_frame = nullptr; } else { GRPC_ERROR_UNREF(s->byte_stream_error); @@ -2096,10 +2090,7 @@ void grpc_chttp2_fail_pending_writes(grpc_chttp2_transport* t, GRPC_ERROR_REF(error), "send_trailing_metadata_finished"); - if (s->fetching_send_message != nullptr) { - grpc_byte_stream_destroy(s->fetching_send_message); - s->fetching_send_message = nullptr; - } + s->fetching_send_message.reset(); grpc_chttp2_complete_closure_step(t, s, &s->fetching_send_message_finished, GRPC_ERROR_REF(error), "fetching_send_message_finished"); @@ -2715,7 +2706,6 @@ static void set_pollset_set(grpc_transport* gt, grpc_stream* gs, static void reset_byte_stream(void* arg, grpc_error* error) { grpc_chttp2_stream* s = static_cast(arg); - s->pending_byte_stream = false; if (error == GRPC_ERROR_NONE) { grpc_chttp2_maybe_complete_recv_message(s->t, s); @@ -2731,22 +2721,56 @@ static void reset_byte_stream(void* arg, grpc_error* error) { } } -static void incoming_byte_stream_unref(grpc_chttp2_incoming_byte_stream* bs) { - if (gpr_unref(&bs->refs)) { - gpr_free(bs); +namespace grpc_core { + +Chttp2IncomingByteStream::Chttp2IncomingByteStream( + grpc_chttp2_transport* transport, grpc_chttp2_stream* stream, + uint32_t frame_size, uint32_t flags) + : ByteStream(frame_size, flags), + transport_(transport), + stream_(stream), + remaining_bytes_(frame_size) { + gpr_ref_init(&refs_, 2); + GRPC_ERROR_UNREF(stream->byte_stream_error); + stream->byte_stream_error = GRPC_ERROR_NONE; +} + +void Chttp2IncomingByteStream::OrphanLocked(void* arg, + grpc_error* error_ignored) { + Chttp2IncomingByteStream* bs = static_cast(arg); + grpc_chttp2_stream* s = bs->stream_; + grpc_chttp2_transport* t = s->t; + bs->Unref(); + s->pending_byte_stream = false; + grpc_chttp2_maybe_complete_recv_message(t, s); + grpc_chttp2_maybe_complete_recv_trailing_metadata(t, s); +} + +void Chttp2IncomingByteStream::Orphan() { + GPR_TIMER_SCOPE("incoming_byte_stream_destroy", 0); + GRPC_CLOSURE_SCHED( + GRPC_CLOSURE_INIT(&destroy_action_, + &Chttp2IncomingByteStream::OrphanLocked, this, + grpc_combiner_scheduler(transport_->combiner)), + GRPC_ERROR_NONE); +} + +void Chttp2IncomingByteStream::Unref() { + if (gpr_unref(&refs_)) { + Delete(this); } } -static void incoming_byte_stream_next_locked(void* argp, - grpc_error* error_ignored) { - grpc_chttp2_incoming_byte_stream* bs = - static_cast(argp); - grpc_chttp2_transport* t = bs->transport; - grpc_chttp2_stream* s = bs->stream; +void Chttp2IncomingByteStream::Ref() { gpr_ref(&refs_); } +void Chttp2IncomingByteStream::NextLocked(void* arg, + grpc_error* error_ignored) { + Chttp2IncomingByteStream* bs = static_cast(arg); + grpc_chttp2_transport* t = bs->transport_; + grpc_chttp2_stream* s = bs->stream_; size_t cur_length = s->frame_storage.length; if (!s->read_closed) { - s->flow_control->IncomingByteStreamUpdate(bs->next_action.max_size_hint, + s->flow_control->IncomingByteStreamUpdate(bs->next_action_.max_size_hint, cur_length); grpc_chttp2_act_on_flowctl_action(s->flow_control->MakeAction(), t, s); } @@ -2755,22 +2779,22 @@ static void incoming_byte_stream_next_locked(void* argp, grpc_slice_buffer_swap(&s->frame_storage, &s->unprocessed_incoming_frames_buffer); s->unprocessed_incoming_frames_decompressed = false; - GRPC_CLOSURE_SCHED(bs->next_action.on_complete, GRPC_ERROR_NONE); + GRPC_CLOSURE_SCHED(bs->next_action_.on_complete, GRPC_ERROR_NONE); } else if (s->byte_stream_error != GRPC_ERROR_NONE) { - GRPC_CLOSURE_SCHED(bs->next_action.on_complete, + GRPC_CLOSURE_SCHED(bs->next_action_.on_complete, GRPC_ERROR_REF(s->byte_stream_error)); if (s->data_parser.parsing_frame != nullptr) { - incoming_byte_stream_unref(s->data_parser.parsing_frame); + s->data_parser.parsing_frame->Unref(); s->data_parser.parsing_frame = nullptr; } } else if (s->read_closed) { - if (bs->remaining_bytes != 0) { + if (bs->remaining_bytes_ != 0) { s->byte_stream_error = GRPC_ERROR_CREATE_FROM_STATIC_STRING("Truncated message"); - GRPC_CLOSURE_SCHED(bs->next_action.on_complete, + GRPC_CLOSURE_SCHED(bs->next_action_.on_complete, GRPC_ERROR_REF(s->byte_stream_error)); if (s->data_parser.parsing_frame != nullptr) { - incoming_byte_stream_unref(s->data_parser.parsing_frame); + s->data_parser.parsing_frame->Unref(); s->data_parser.parsing_frame = nullptr; } } else { @@ -2778,122 +2802,94 @@ static void incoming_byte_stream_next_locked(void* argp, GPR_ASSERT(false); } } else { - s->on_next = bs->next_action.on_complete; + s->on_next = bs->next_action_.on_complete; } - incoming_byte_stream_unref(bs); + bs->Unref(); } -static bool incoming_byte_stream_next(grpc_byte_stream* byte_stream, - size_t max_size_hint, - grpc_closure* on_complete) { +bool Chttp2IncomingByteStream::Next(size_t max_size_hint, + grpc_closure* on_complete) { GPR_TIMER_SCOPE("incoming_byte_stream_next", 0); - grpc_chttp2_incoming_byte_stream* bs = - reinterpret_cast(byte_stream); - grpc_chttp2_stream* s = bs->stream; - if (s->unprocessed_incoming_frames_buffer.length > 0) { + if (stream_->unprocessed_incoming_frames_buffer.length > 0) { return true; } else { - gpr_ref(&bs->refs); - bs->next_action.max_size_hint = max_size_hint; - bs->next_action.on_complete = on_complete; + Ref(); + next_action_.max_size_hint = max_size_hint; + next_action_.on_complete = on_complete; GRPC_CLOSURE_SCHED( - GRPC_CLOSURE_INIT(&bs->next_action.closure, - incoming_byte_stream_next_locked, bs, - grpc_combiner_scheduler(bs->transport->combiner)), + GRPC_CLOSURE_INIT(&next_action_.closure, + &Chttp2IncomingByteStream::NextLocked, this, + grpc_combiner_scheduler(transport_->combiner)), GRPC_ERROR_NONE); return false; } } -static grpc_error* incoming_byte_stream_pull(grpc_byte_stream* byte_stream, - grpc_slice* slice) { +grpc_error* Chttp2IncomingByteStream::Pull(grpc_slice* slice) { GPR_TIMER_SCOPE("incoming_byte_stream_pull", 0); - grpc_chttp2_incoming_byte_stream* bs = - reinterpret_cast(byte_stream); - grpc_chttp2_stream* s = bs->stream; grpc_error* error; - - if (s->unprocessed_incoming_frames_buffer.length > 0) { - if (!s->unprocessed_incoming_frames_decompressed) { + if (stream_->unprocessed_incoming_frames_buffer.length > 0) { + if (!stream_->unprocessed_incoming_frames_decompressed) { bool end_of_context; - if (!s->stream_decompression_ctx) { - s->stream_decompression_ctx = grpc_stream_compression_context_create( - s->stream_decompression_method); + if (!stream_->stream_decompression_ctx) { + stream_->stream_decompression_ctx = + grpc_stream_compression_context_create( + stream_->stream_decompression_method); } - if (!grpc_stream_decompress(s->stream_decompression_ctx, - &s->unprocessed_incoming_frames_buffer, - &s->decompressed_data_buffer, nullptr, + if (!grpc_stream_decompress(stream_->stream_decompression_ctx, + &stream_->unprocessed_incoming_frames_buffer, + &stream_->decompressed_data_buffer, nullptr, MAX_SIZE_T, &end_of_context)) { error = GRPC_ERROR_CREATE_FROM_STATIC_STRING("Stream decompression error."); return error; } - GPR_ASSERT(s->unprocessed_incoming_frames_buffer.length == 0); - grpc_slice_buffer_swap(&s->unprocessed_incoming_frames_buffer, - &s->decompressed_data_buffer); - s->unprocessed_incoming_frames_decompressed = true; + GPR_ASSERT(stream_->unprocessed_incoming_frames_buffer.length == 0); + grpc_slice_buffer_swap(&stream_->unprocessed_incoming_frames_buffer, + &stream_->decompressed_data_buffer); + stream_->unprocessed_incoming_frames_decompressed = true; if (end_of_context) { - grpc_stream_compression_context_destroy(s->stream_decompression_ctx); - s->stream_decompression_ctx = nullptr; + grpc_stream_compression_context_destroy( + stream_->stream_decompression_ctx); + stream_->stream_decompression_ctx = nullptr; } - if (s->unprocessed_incoming_frames_buffer.length == 0) { + if (stream_->unprocessed_incoming_frames_buffer.length == 0) { *slice = grpc_empty_slice(); } } error = grpc_deframe_unprocessed_incoming_frames( - &s->data_parser, s, &s->unprocessed_incoming_frames_buffer, slice, - nullptr); + &stream_->data_parser, stream_, + &stream_->unprocessed_incoming_frames_buffer, slice, nullptr); if (error != GRPC_ERROR_NONE) { return error; } } else { error = GRPC_ERROR_CREATE_FROM_STATIC_STRING("Truncated message"); - GRPC_CLOSURE_SCHED(&s->reset_byte_stream, GRPC_ERROR_REF(error)); + GRPC_CLOSURE_SCHED(&stream_->reset_byte_stream, GRPC_ERROR_REF(error)); return error; } return GRPC_ERROR_NONE; } -static void incoming_byte_stream_destroy_locked(void* byte_stream, - grpc_error* error_ignored); - -static void incoming_byte_stream_destroy(grpc_byte_stream* byte_stream) { - GPR_TIMER_SCOPE("incoming_byte_stream_destroy", 0); - grpc_chttp2_incoming_byte_stream* bs = - reinterpret_cast(byte_stream); - GRPC_CLOSURE_SCHED( - GRPC_CLOSURE_INIT(&bs->destroy_action, - incoming_byte_stream_destroy_locked, bs, - grpc_combiner_scheduler(bs->transport->combiner)), - GRPC_ERROR_NONE); -} - -static void incoming_byte_stream_publish_error( - grpc_chttp2_incoming_byte_stream* bs, grpc_error* error) { - grpc_chttp2_stream* s = bs->stream; - +void Chttp2IncomingByteStream::PublishError(grpc_error* error) { GPR_ASSERT(error != GRPC_ERROR_NONE); - GRPC_CLOSURE_SCHED(s->on_next, GRPC_ERROR_REF(error)); - s->on_next = nullptr; - GRPC_ERROR_UNREF(s->byte_stream_error); - s->byte_stream_error = GRPC_ERROR_REF(error); - grpc_chttp2_cancel_stream(bs->transport, bs->stream, GRPC_ERROR_REF(error)); + GRPC_CLOSURE_SCHED(stream_->on_next, GRPC_ERROR_REF(error)); + stream_->on_next = nullptr; + GRPC_ERROR_UNREF(stream_->byte_stream_error); + stream_->byte_stream_error = GRPC_ERROR_REF(error); + grpc_chttp2_cancel_stream(transport_, stream_, GRPC_ERROR_REF(error)); } -grpc_error* grpc_chttp2_incoming_byte_stream_push( - grpc_chttp2_incoming_byte_stream* bs, grpc_slice slice, - grpc_slice* slice_out) { - grpc_chttp2_stream* s = bs->stream; - - if (bs->remaining_bytes < GRPC_SLICE_LENGTH(slice)) { +grpc_error* Chttp2IncomingByteStream::Push(grpc_slice slice, + grpc_slice* slice_out) { + if (remaining_bytes_ < GRPC_SLICE_LENGTH(slice)) { grpc_error* error = GRPC_ERROR_CREATE_FROM_STATIC_STRING("Too many bytes in stream"); - - GRPC_CLOSURE_SCHED(&s->reset_byte_stream, GRPC_ERROR_REF(error)); + GRPC_CLOSURE_SCHED(&stream_->reset_byte_stream, GRPC_ERROR_REF(error)); grpc_slice_unref_internal(slice); return error; } else { - bs->remaining_bytes -= static_cast GRPC_SLICE_LENGTH(slice); + remaining_bytes_ -= static_cast GRPC_SLICE_LENGTH(slice); if (slice_out != nullptr) { *slice_out = slice; } @@ -2901,66 +2897,25 @@ grpc_error* grpc_chttp2_incoming_byte_stream_push( } } -grpc_error* grpc_chttp2_incoming_byte_stream_finished( - grpc_chttp2_incoming_byte_stream* bs, grpc_error* error, - bool reset_on_error) { - grpc_chttp2_stream* s = bs->stream; - +grpc_error* Chttp2IncomingByteStream::Finished(grpc_error* error, + bool reset_on_error) { if (error == GRPC_ERROR_NONE) { - if (bs->remaining_bytes != 0) { + if (remaining_bytes_ != 0) { error = GRPC_ERROR_CREATE_FROM_STATIC_STRING("Truncated message"); } } if (error != GRPC_ERROR_NONE && reset_on_error) { - GRPC_CLOSURE_SCHED(&s->reset_byte_stream, GRPC_ERROR_REF(error)); + GRPC_CLOSURE_SCHED(&stream_->reset_byte_stream, GRPC_ERROR_REF(error)); } - incoming_byte_stream_unref(bs); + Unref(); return error; } -static void incoming_byte_stream_shutdown(grpc_byte_stream* byte_stream, - grpc_error* error) { - grpc_chttp2_incoming_byte_stream* bs = - reinterpret_cast(byte_stream); - GRPC_ERROR_UNREF(grpc_chttp2_incoming_byte_stream_finished( - bs, error, true /* reset_on_error */)); +void Chttp2IncomingByteStream::Shutdown(grpc_error* error) { + GRPC_ERROR_UNREF(Finished(error, true /* reset_on_error */)); } -static const grpc_byte_stream_vtable grpc_chttp2_incoming_byte_stream_vtable = { - incoming_byte_stream_next, incoming_byte_stream_pull, - incoming_byte_stream_shutdown, incoming_byte_stream_destroy}; - -static void incoming_byte_stream_destroy_locked(void* byte_stream, - grpc_error* error_ignored) { - grpc_chttp2_incoming_byte_stream* bs = - static_cast(byte_stream); - grpc_chttp2_stream* s = bs->stream; - grpc_chttp2_transport* t = s->t; - - GPR_ASSERT(bs->base.vtable == &grpc_chttp2_incoming_byte_stream_vtable); - incoming_byte_stream_unref(bs); - s->pending_byte_stream = false; - grpc_chttp2_maybe_complete_recv_message(t, s); - grpc_chttp2_maybe_complete_recv_trailing_metadata(t, s); -} - -grpc_chttp2_incoming_byte_stream* grpc_chttp2_incoming_byte_stream_create( - grpc_chttp2_transport* t, grpc_chttp2_stream* s, uint32_t frame_size, - uint32_t flags) { - grpc_chttp2_incoming_byte_stream* incoming_byte_stream = - static_cast( - gpr_malloc(sizeof(*incoming_byte_stream))); - incoming_byte_stream->base.length = frame_size; - incoming_byte_stream->remaining_bytes = frame_size; - incoming_byte_stream->base.flags = flags; - incoming_byte_stream->base.vtable = &grpc_chttp2_incoming_byte_stream_vtable; - gpr_ref_init(&incoming_byte_stream->refs, 2); - incoming_byte_stream->transport = t; - incoming_byte_stream->stream = s; - GRPC_ERROR_UNREF(s->byte_stream_error); - s->byte_stream_error = GRPC_ERROR_NONE; - return incoming_byte_stream; -} +} // namespace grpc_core /******************************************************************************* * RESOURCE QUOTAS diff --git a/src/core/ext/transport/chttp2/transport/frame_data.cc b/src/core/ext/transport/chttp2/transport/frame_data.cc index 0d37a494a2..f8f06f6789 100644 --- a/src/core/ext/transport/chttp2/transport/frame_data.cc +++ b/src/core/ext/transport/chttp2/transport/frame_data.cc @@ -27,6 +27,7 @@ #include #include "src/core/ext/transport/chttp2/transport/internal.h" #include "src/core/lib/gpr/string.h" +#include "src/core/lib/gprpp/memory.h" #include "src/core/lib/slice/slice_internal.h" #include "src/core/lib/slice/slice_string_helpers.h" #include "src/core/lib/transport/transport.h" @@ -39,8 +40,7 @@ grpc_error* grpc_chttp2_data_parser_init(grpc_chttp2_data_parser* parser) { void grpc_chttp2_data_parser_destroy(grpc_chttp2_data_parser* parser) { if (parser->parsing_frame != nullptr) { - GRPC_ERROR_UNREF(grpc_chttp2_incoming_byte_stream_finished( - parser->parsing_frame, + GRPC_ERROR_UNREF(parser->parsing_frame->Finished( GRPC_ERROR_CREATE_FROM_STATIC_STRING("Parser destroyed"), false)); } GRPC_ERROR_UNREF(parser->error); @@ -100,7 +100,7 @@ void grpc_chttp2_encode_data(uint32_t id, grpc_slice_buffer* inbuf, grpc_error* grpc_deframe_unprocessed_incoming_frames( grpc_chttp2_data_parser* p, grpc_chttp2_stream* s, grpc_slice_buffer* slices, grpc_slice* slice_out, - grpc_byte_stream** stream_out) { + grpc_core::OrphanablePtr* stream_out) { grpc_error* error = GRPC_ERROR_NONE; grpc_chttp2_transport* t = s->t; @@ -197,12 +197,11 @@ grpc_error* grpc_deframe_unprocessed_incoming_frames( if (p->is_frame_compressed) { message_flags |= GRPC_WRITE_INTERNAL_COMPRESS; } - p->parsing_frame = grpc_chttp2_incoming_byte_stream_create( + p->parsing_frame = grpc_core::New( t, s, p->frame_size, message_flags); - *stream_out = &p->parsing_frame->base; - if (p->parsing_frame->remaining_bytes == 0) { - GRPC_ERROR_UNREF(grpc_chttp2_incoming_byte_stream_finished( - p->parsing_frame, GRPC_ERROR_NONE, true)); + stream_out->reset(p->parsing_frame); + if (p->parsing_frame->remaining_bytes() == 0) { + GRPC_ERROR_UNREF(p->parsing_frame->Finished(GRPC_ERROR_NONE, true)); p->parsing_frame = nullptr; p->state = GRPC_CHTTP2_DATA_FH_0; } @@ -226,8 +225,7 @@ grpc_error* grpc_deframe_unprocessed_incoming_frames( if (remaining == p->frame_size) { s->stats.incoming.data_bytes += remaining; if (GRPC_ERROR_NONE != - (error = grpc_chttp2_incoming_byte_stream_push( - p->parsing_frame, + (error = p->parsing_frame->Push( grpc_slice_sub(slice, static_cast(cur - beg), static_cast(end - beg)), slice_out))) { @@ -235,8 +233,7 @@ grpc_error* grpc_deframe_unprocessed_incoming_frames( return error; } if (GRPC_ERROR_NONE != - (error = grpc_chttp2_incoming_byte_stream_finished( - p->parsing_frame, GRPC_ERROR_NONE, true))) { + (error = p->parsing_frame->Finished(GRPC_ERROR_NONE, true))) { grpc_slice_unref_internal(slice); return error; } @@ -247,8 +244,7 @@ grpc_error* grpc_deframe_unprocessed_incoming_frames( } else if (remaining < p->frame_size) { s->stats.incoming.data_bytes += remaining; if (GRPC_ERROR_NONE != - (error = grpc_chttp2_incoming_byte_stream_push( - p->parsing_frame, + (error = p->parsing_frame->Push( grpc_slice_sub(slice, static_cast(cur - beg), static_cast(end - beg)), slice_out))) { @@ -261,18 +257,16 @@ grpc_error* grpc_deframe_unprocessed_incoming_frames( GPR_ASSERT(remaining > p->frame_size); s->stats.incoming.data_bytes += p->frame_size; if (GRPC_ERROR_NONE != - (grpc_chttp2_incoming_byte_stream_push( - p->parsing_frame, + p->parsing_frame->Push( grpc_slice_sub( slice, static_cast(cur - beg), static_cast(cur + p->frame_size - beg)), - slice_out))) { + slice_out)) { grpc_slice_unref_internal(slice); return error; } if (GRPC_ERROR_NONE != - (error = grpc_chttp2_incoming_byte_stream_finished( - p->parsing_frame, GRPC_ERROR_NONE, true))) { + (error = p->parsing_frame->Finished(GRPC_ERROR_NONE, true))) { grpc_slice_unref_internal(slice); return error; } diff --git a/src/core/ext/transport/chttp2/transport/frame_data.h b/src/core/ext/transport/chttp2/transport/frame_data.h index 3efbbf9f76..4b0f873644 100644 --- a/src/core/ext/transport/chttp2/transport/frame_data.h +++ b/src/core/ext/transport/chttp2/transport/frame_data.h @@ -40,8 +40,9 @@ typedef enum { GRPC_CHTTP2_DATA_ERROR } grpc_chttp2_stream_state; -typedef struct grpc_chttp2_incoming_byte_stream - grpc_chttp2_incoming_byte_stream; +namespace grpc_core { +class Chttp2IncomingByteStream; +} // namespace grpc_core typedef struct { grpc_chttp2_stream_state state; @@ -50,7 +51,7 @@ typedef struct { grpc_error* error; bool is_frame_compressed; - grpc_chttp2_incoming_byte_stream* parsing_frame; + grpc_core::Chttp2IncomingByteStream* parsing_frame; } grpc_chttp2_data_parser; /* initialize per-stream state for data frame parsing */ @@ -79,6 +80,6 @@ void grpc_chttp2_encode_data(uint32_t id, grpc_slice_buffer* inbuf, grpc_error* grpc_deframe_unprocessed_incoming_frames( grpc_chttp2_data_parser* p, grpc_chttp2_stream* s, grpc_slice_buffer* slices, grpc_slice* slice_out, - grpc_byte_stream** stream_out); + grpc_core::OrphanablePtr* stream_out); #endif /* GRPC_CORE_EXT_TRANSPORT_CHTTP2_TRANSPORT_FRAME_DATA_H */ diff --git a/src/core/ext/transport/chttp2/transport/internal.h b/src/core/ext/transport/chttp2/transport/internal.h index b9431cd311..6d11e5aa31 100644 --- a/src/core/ext/transport/chttp2/transport/internal.h +++ b/src/core/ext/transport/chttp2/transport/internal.h @@ -203,18 +203,58 @@ typedef struct grpc_chttp2_write_cb { struct grpc_chttp2_write_cb* next; } grpc_chttp2_write_cb; -/* forward declared in frame_data.h */ -struct grpc_chttp2_incoming_byte_stream { - grpc_byte_stream base; - gpr_refcount refs; +namespace grpc_core { + +class Chttp2IncomingByteStream : public ByteStream { + public: + Chttp2IncomingByteStream(grpc_chttp2_transport* transport, + grpc_chttp2_stream* stream, uint32_t frame_size, + uint32_t flags); + + void Orphan() override; + + bool Next(size_t max_size_hint, grpc_closure* on_complete) override; + grpc_error* Pull(grpc_slice* slice) override; + void Shutdown(grpc_error* error) override; + + // TODO(roth): When I converted this class to C++, I wanted to make it + // inherit from RefCounted or InternallyRefCounted instead of continuing + // to use its own custom ref-counting code. However, that would require + // using multiple inheritence, which sucks in general. And to make matters + // worse, it causes problems with our New<> and Delete<> wrappers. + // Specifically, unless RefCounted is first in the list of parent classes, + // it will see a different value of the address of the object than the one + // we actually allocated, in which case gpr_free() will be called on a + // different address than the one we got from gpr_malloc(), thus causing a + // crash. Given the fragility of depending on that, as well as a desire to + // avoid multiple inheritence in general, I've decided to leave this + // alone for now. We can revisit this once we're able to link against + // libc++, at which point we can eliminate New<> and Delete<> and + // switch to std::shared_ptr<>. + void Ref(); + void Unref(); + + void PublishError(grpc_error* error); + + grpc_error* Push(grpc_slice slice, grpc_slice* slice_out); - grpc_chttp2_transport* transport; /* immutable */ - grpc_chttp2_stream* stream; /* immutable */ + grpc_error* Finished(grpc_error* error, bool reset_on_error); + + uint32_t remaining_bytes() const { return remaining_bytes_; } + + private: + static void NextLocked(void* arg, grpc_error* error_ignored); + static void OrphanLocked(void* arg, grpc_error* error_ignored); + + grpc_chttp2_transport* transport_; // Immutable. + grpc_chttp2_stream* stream_; // Immutable. + + gpr_refcount refs_; /* Accessed only by transport thread when stream->pending_byte_stream == false * Accessed only by application thread when stream->pending_byte_stream == * true */ - uint32_t remaining_bytes; + uint32_t remaining_bytes_; /* Accessed only by transport thread when stream->pending_byte_stream == false * Accessed only by application thread when stream->pending_byte_stream == @@ -223,11 +263,12 @@ struct grpc_chttp2_incoming_byte_stream { grpc_closure closure; size_t max_size_hint; grpc_closure* on_complete; - } next_action; - grpc_closure destroy_action; - grpc_closure finished_action; + } next_action_; + grpc_closure destroy_action_; }; +} // namespace grpc_core + typedef enum { GRPC_CHTTP2_KEEPALIVE_STATE_WAITING, GRPC_CHTTP2_KEEPALIVE_STATE_PINGING, @@ -456,7 +497,7 @@ struct grpc_chttp2_stream { grpc_metadata_batch* send_trailing_metadata; grpc_closure* send_trailing_metadata_finished; - grpc_byte_stream* fetching_send_message; + grpc_core::OrphanablePtr fetching_send_message; uint32_t fetched_send_message_length; grpc_slice fetching_slice; int64_t next_message_end_offset; @@ -468,7 +509,7 @@ struct grpc_chttp2_stream { grpc_metadata_batch* recv_initial_metadata; grpc_closure* recv_initial_metadata_ready; bool* trailing_metadata_available; - grpc_byte_stream** recv_message; + grpc_core::OrphanablePtr* recv_message; grpc_closure* recv_message_ready; grpc_metadata_batch* recv_trailing_metadata; grpc_closure* recv_trailing_metadata_finished; @@ -719,18 +760,6 @@ void grpc_chttp2_unref_transport(grpc_chttp2_transport* t); void grpc_chttp2_ref_transport(grpc_chttp2_transport* t); #endif -grpc_chttp2_incoming_byte_stream* grpc_chttp2_incoming_byte_stream_create( - grpc_chttp2_transport* t, grpc_chttp2_stream* s, uint32_t frame_size, - uint32_t flags); -grpc_error* grpc_chttp2_incoming_byte_stream_push( - grpc_chttp2_incoming_byte_stream* bs, grpc_slice slice, - grpc_slice* slice_out); -grpc_error* grpc_chttp2_incoming_byte_stream_finished( - grpc_chttp2_incoming_byte_stream* bs, grpc_error* error, - bool reset_on_error); -void grpc_chttp2_incoming_byte_stream_notify( - grpc_chttp2_incoming_byte_stream* bs, grpc_error* error); - void grpc_chttp2_ack_ping(grpc_chttp2_transport* t, uint64_t id); /** Add a new ping strike to ping_recv_state.ping_strikes. If -- cgit v1.2.3