Merge branch 'master' of github.com:grpc/grpc into authority_header

13 files changed, 196 insertions, 226 deletions

diff --git a/src/core/ext/transport/chttp2/client/insecure/channel_create_posix.cc b/src/core/ext/transport/chttp2/client/insecure/channel_create_posix.cc
index 479f0da572..b95c9dae53 100644
--- a/src/core/ext/transport/chttp2/client/insecure/channel_create_posix.cc
+++ b/src/core/ext/transport/chttp2/client/insecure/channel_create_posix.cc
@@ -29,7 +29,6 @@
 #include "src/core/ext/transport/chttp2/transport/chttp2_transport.h"
 #include "src/core/lib/channel/channel_args.h"
 #include "src/core/lib/iomgr/endpoint.h"
-#include "src/core/lib/iomgr/exec_ctx.h"
 #include "src/core/lib/iomgr/tcp_client_posix.h"
 #include "src/core/lib/iomgr/tcp_posix.h"
 #include "src/core/lib/surface/api_trace.h"
diff --git a/src/core/ext/transport/chttp2/server/chttp2_server.h b/src/core/ext/transport/chttp2/server/chttp2_server.h
index 7b41972160..6e51001b53 100644
--- a/src/core/ext/transport/chttp2/server/chttp2_server.h
+++ b/src/core/ext/transport/chttp2/server/chttp2_server.h
@@ -23,7 +23,7 @@
 
 #include <grpc/impl/codegen/grpc_types.h>
 
-#include "src/core/lib/iomgr/exec_ctx.h"
+#include "src/core/lib/iomgr/error.h"
 
 /// Adds a port to \a server.  Sets \a port_num to the port number.
 /// Takes ownership of \a args.
diff --git a/src/core/ext/transport/chttp2/server/insecure/server_chttp2.cc b/src/core/ext/transport/chttp2/server/insecure/server_chttp2.cc
index 822236dd2d..99f18cdf39 100644
--- a/src/core/ext/transport/chttp2/server/insecure/server_chttp2.cc
+++ b/src/core/ext/transport/chttp2/server/insecure/server_chttp2.cc
@@ -41,6 +41,5 @@ int grpc_server_add_insecure_http2_port(grpc_server* server, const char* addr) {
 
     GRPC_ERROR_UNREF(err);
   }
-
   return port_num;
 }
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<grpc_chttp2_stream*>(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<uint8_t>(len >> 24);
       frame_hdr[2] = static_cast<uint8_t>(len >> 16);
       frame_hdr[3] = static_cast<uint8_t>(len >> 8);
       frame_hdr[4] = static_cast<uint8_t>(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<grpc_chttp2_stream*>(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<Chttp2IncomingByteStream*>(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<grpc_chttp2_incoming_byte_stream*>(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<Chttp2IncomingByteStream*>(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<grpc_chttp2_incoming_byte_stream*>(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<grpc_chttp2_incoming_byte_stream*>(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<grpc_chttp2_incoming_byte_stream*>(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<uint32_t> GRPC_SLICE_LENGTH(slice);
+    remaining_bytes_ -= static_cast<uint32_t> 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<grpc_chttp2_incoming_byte_stream*>(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<grpc_chttp2_incoming_byte_stream*>(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<grpc_chttp2_incoming_byte_stream*>(
-          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 <grpc/support/string_util.h>
 #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<grpc_core::ByteStream>* 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<grpc_core::Chttp2IncomingByteStream>(
             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<size_t>(cur - beg),
                                   static_cast<size_t>(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<size_t>(cur - beg),
                                   static_cast<size_t>(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<size_t>(cur - beg),
                       static_cast<size_t>(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..e5d01f764e 100644
--- a/src/core/ext/transport/chttp2/transport/frame_data.h
+++ b/src/core/ext/transport/chttp2/transport/frame_data.h
@@ -26,7 +26,6 @@
 #include <grpc/slice.h>
 #include <grpc/slice_buffer.h>
 #include "src/core/ext/transport/chttp2/transport/frame.h"
-#include "src/core/lib/iomgr/exec_ctx.h"
 #include "src/core/lib/transport/byte_stream.h"
 #include "src/core/lib/transport/transport.h"
 
@@ -40,8 +39,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 +50,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 +79,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<grpc_core::ByteStream>* stream_out);
 
 #endif /* GRPC_CORE_EXT_TRANSPORT_CHTTP2_TRANSPORT_FRAME_DATA_H */
diff --git a/src/core/ext/transport/chttp2/transport/frame_goaway.h b/src/core/ext/transport/chttp2/transport/frame_goaway.h
index e17ed8d563..66c7a68bef 100644
--- a/src/core/ext/transport/chttp2/transport/frame_goaway.h
+++ b/src/core/ext/transport/chttp2/transport/frame_goaway.h
@@ -24,7 +24,6 @@
 #include <grpc/slice.h>
 #include <grpc/slice_buffer.h>
 #include "src/core/ext/transport/chttp2/transport/frame.h"
-#include "src/core/lib/iomgr/exec_ctx.h"
 
 typedef enum {
   GRPC_CHTTP2_GOAWAY_LSI0,
diff --git a/src/core/ext/transport/chttp2/transport/frame_ping.h b/src/core/ext/transport/chttp2/transport/frame_ping.h
index 8718d6a097..55a4499ad5 100644
--- a/src/core/ext/transport/chttp2/transport/frame_ping.h
+++ b/src/core/ext/transport/chttp2/transport/frame_ping.h
@@ -23,7 +23,6 @@
 
 #include <grpc/slice.h>
 #include "src/core/ext/transport/chttp2/transport/frame.h"
-#include "src/core/lib/iomgr/exec_ctx.h"
 
 typedef struct {
   uint8_t byte;
diff --git a/src/core/ext/transport/chttp2/transport/frame_rst_stream.h b/src/core/ext/transport/chttp2/transport/frame_rst_stream.h
index bb2d34f918..6bcf9c4479 100644
--- a/src/core/ext/transport/chttp2/transport/frame_rst_stream.h
+++ b/src/core/ext/transport/chttp2/transport/frame_rst_stream.h
@@ -23,7 +23,6 @@
 
 #include <grpc/slice.h>
 #include "src/core/ext/transport/chttp2/transport/frame.h"
-#include "src/core/lib/iomgr/exec_ctx.h"
 #include "src/core/lib/transport/transport.h"
 
 typedef struct {
diff --git a/src/core/ext/transport/chttp2/transport/frame_settings.h b/src/core/ext/transport/chttp2/transport/frame_settings.h
index df19627194..8d8d9b1a91 100644
--- a/src/core/ext/transport/chttp2/transport/frame_settings.h
+++ b/src/core/ext/transport/chttp2/transport/frame_settings.h
@@ -24,7 +24,6 @@
 #include <grpc/slice.h>
 #include "src/core/ext/transport/chttp2/transport/frame.h"
 #include "src/core/ext/transport/chttp2/transport/http2_settings.h"
-#include "src/core/lib/iomgr/exec_ctx.h"
 
 typedef enum {
   GRPC_CHTTP2_SPS_ID0,
diff --git a/src/core/ext/transport/chttp2/transport/frame_window_update.h b/src/core/ext/transport/chttp2/transport/frame_window_update.h
index 30667c77e1..3d2391f637 100644
--- a/src/core/ext/transport/chttp2/transport/frame_window_update.h
+++ b/src/core/ext/transport/chttp2/transport/frame_window_update.h
@@ -23,7 +23,6 @@
 
 #include <grpc/slice.h>
 #include "src/core/ext/transport/chttp2/transport/frame.h"
-#include "src/core/lib/iomgr/exec_ctx.h"
 #include "src/core/lib/transport/transport.h"
 
 typedef struct {
diff --git a/src/core/ext/transport/chttp2/transport/hpack_parser.h b/src/core/ext/transport/chttp2/transport/hpack_parser.h
index b3b8018b98..3e05de4b92 100644
--- a/src/core/ext/transport/chttp2/transport/hpack_parser.h
+++ b/src/core/ext/transport/chttp2/transport/hpack_parser.h
@@ -25,7 +25,6 @@
 
 #include "src/core/ext/transport/chttp2/transport/frame.h"
 #include "src/core/ext/transport/chttp2/transport/hpack_table.h"
-#include "src/core/lib/iomgr/exec_ctx.h"
 #include "src/core/lib/transport/metadata.h"
 
 typedef struct grpc_chttp2_hpack_parser grpc_chttp2_hpack_parser;
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<grpc_core::ByteStream> 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<grpc_core::ByteStream>* 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