http2: migrates DATA frame payloads to the visitor API

changelogs/current.yaml

@@ -16,6 +16,10 @@ behavior_changes:
   change: |
     Changes the default value of ``envoy.reloadable_features.http2_use_oghttp2`` to true. This changes the codec used for HTTP/2
     requests and responses. This behavior can be reverted by setting the feature to false.
+- area: http2
+  change: |
+    Passes HTTP/2 DATA frames through a different codec API. This behavior can be temporarily disabled by setting the runtime
+    feature ``envoy.reloadable_features.http2_use_visitor_for_data`` to false.
 - area: proxy_protocol
   change: |
     Populate typed metadata by default in proxy protocol listener. Typed metadata can be consumed as

source/common/http/http2/codec_impl.cc

@@ -663,8 +663,12 @@ bool ConnectionImpl::StreamDataFrameSource::Send(absl::string_view frame_header,
 
 void ConnectionImpl::ClientStreamImpl::submitHeaders(const HeaderMap& headers, bool end_stream) {
   ASSERT(stream_id_ == -1);
+  const bool skip_frame_source =
+      end_stream ||
+      Runtime::runtimeFeatureEnabled("envoy.reloadable_features.http2_use_visitor_for_data");
   stream_id_ = parent_.adapter_->SubmitRequest(
-      buildHeaders(headers), end_stream ? nullptr : std::make_unique<StreamDataFrameSource>(*this),
+      buildHeaders(headers),
+      skip_frame_source ? nullptr : std::make_unique<StreamDataFrameSource>(*this), end_stream,
       base());
   ASSERT(stream_id_ > 0);
 }
@@ -685,9 +689,12 @@ void ConnectionImpl::ClientStreamImpl::advanceHeadersState() {
 
 void ConnectionImpl::ServerStreamImpl::submitHeaders(const HeaderMap& headers, bool end_stream) {
   ASSERT(stream_id_ != -1);
-  parent_.adapter_->SubmitResponse(stream_id_, buildHeaders(headers),
-                                   end_stream ? nullptr
-                                              : std::make_unique<StreamDataFrameSource>(*this));
+  const bool skip_frame_source =
+      end_stream ||
+      Runtime::runtimeFeatureEnabled("envoy.reloadable_features.http2_use_visitor_for_data");
+  parent_.adapter_->SubmitResponse(
+      stream_id_, buildHeaders(headers),
+      skip_frame_source ? nullptr : std::make_unique<StreamDataFrameSource>(*this), end_stream);
 }
 
 Status ConnectionImpl::ServerStreamImpl::onBeginHeaders() {
@@ -1692,6 +1699,58 @@ int64_t ConnectionImpl::Http2Visitor::OnReadyToSend(absl::string_view serialized
                              serialized.size());
 }
 
+ConnectionImpl::Http2Visitor::DataFrameHeaderInfo
+ConnectionImpl::Http2Visitor::OnReadyToSendDataForStream(Http2StreamId stream_id,
+                                                         size_t max_length) {
+  StreamImpl* stream = connection_->getStream(stream_id);
+  if (stream == nullptr) {
+    return {/*payload_length=*/-1, /*end_data=*/false, /*end_stream=*/false};
+  }
+  if (stream->pending_send_data_->length() == 0 && !stream->local_end_stream_) {
+    stream->data_deferred_ = true;
+    return {/*payload_length=*/0, /*end_data=*/false, /*end_stream=*/false};
+  }
+  const size_t length = std::min<size_t>(max_length, stream->pending_send_data_->length());
+  bool end_data = false;
+  bool end_stream = false;
+  if (stream->local_end_stream_ && length == stream->pending_send_data_->length()) {
+    end_data = true;
+    if (stream->pending_trailers_to_encode_) {
+      stream->submitTrailers(*stream->pending_trailers_to_encode_);
+      stream->pending_trailers_to_encode_.reset();
+    } else {
+      end_stream = true;
+    }
+  }
+  return {static_cast<int64_t>(length), end_data, end_stream};
+}
+
+bool ConnectionImpl::Http2Visitor::SendDataFrame(Http2StreamId stream_id,
+                                                 absl::string_view frame_header,
+                                                 size_t payload_length) {
+  connection_->protocol_constraints_.incrementOutboundDataFrameCount();
+
+  StreamImpl* stream = connection_->getStream(stream_id);
+  if (stream == nullptr) {
+    ENVOY_CONN_LOG(error, "error sending data frame: stream {} not found", connection_->connection_,
+                   stream_id);
+    return false;
+  }
+  Buffer::OwnedImpl output;
+  connection_->addOutboundFrameFragment(
+      output, reinterpret_cast<const uint8_t*>(frame_header.data()), frame_header.size());
+  if (!connection_->protocol_constraints_.checkOutboundFrameLimits().ok()) {
+    ENVOY_CONN_LOG(debug, "error sending data frame: Too many frames in the outbound queue",
+                   connection_->connection_);
+    stream->setDetails(Http2ResponseCodeDetails::get().outbound_frame_flood);
+  }
+
+  connection_->stats_.pending_send_bytes_.sub(payload_length);
+  output.move(*stream->pending_send_data_, payload_length);
+  connection_->connection_.write(output, false);
+  return true;
+}
+
 bool ConnectionImpl::Http2Visitor::OnFrameHeader(Http2StreamId stream_id, size_t length,
                                                  uint8_t type, uint8_t flags) {
   ENVOY_CONN_LOG(debug, "Http2Visitor::OnFrameHeader({}, {}, {}, {})", connection_->connection_,

source/common/http/http2/codec_impl.h

@@ -176,6 +176,10 @@ class ConnectionImpl : public virtual Connection,
       stream_close_listener_ = std::move(f);
     }
     int64_t OnReadyToSend(absl::string_view serialized) override;
+    DataFrameHeaderInfo OnReadyToSendDataForStream(Http2StreamId stream_id,
+                                                   size_t max_length) override;
+    bool SendDataFrame(Http2StreamId stream_id, absl::string_view frame_header,
+                       size_t payload_bytes) override;
     void OnConnectionError(ConnectionError /*error*/) override {}
     bool OnFrameHeader(Http2StreamId stream_id, size_t length, uint8_t type,
                        uint8_t flags) override;
@@ -466,6 +470,8 @@ class ConnectionImpl : public virtual Connection,
   };
 
   // Encapsulates the logic for sending DATA frames on a given stream.
+  // Deprecated. Remove when removing
+  // `envoy_reloadable_features_http2_use_visitor_for_data`.
   class StreamDataFrameSource : public http2::adapter::DataFrameSource {
   public:
     explicit StreamDataFrameSource(StreamImpl& stream) : stream_(stream) {}

source/common/runtime/runtime_features.cc

@@ -59,6 +59,7 @@ RUNTIME_GUARD(envoy_reloadable_features_http2_decode_metadata_with_quiche);
 RUNTIME_GUARD(envoy_reloadable_features_http2_discard_host_header);
 // Ignore the automated "remove this flag" issue: we should keep this for 1 year.
 RUNTIME_GUARD(envoy_reloadable_features_http2_use_oghttp2);
+RUNTIME_GUARD(envoy_reloadable_features_http2_use_visitor_for_data);
 RUNTIME_GUARD(envoy_reloadable_features_http2_validate_authority_with_quiche);
 RUNTIME_GUARD(envoy_reloadable_features_http_allow_partial_urls_in_referer);
 RUNTIME_GUARD(envoy_reloadable_features_http_filter_avoid_reentrant_local_reply);