ctiller · Oct 10, 2023 · Sep 15, 2023 · Sep 15, 2023 · Sep 15, 2023 · Sep 15, 2023
diff --git a/bazel/experiments.bzl b/bazel/experiments.bzl
diff --git a/src/core/ext/transport/chttp2/transport/chttp2_transport.cc b/src/core/ext/transport/chttp2/transport/chttp2_transport.cc
@@ -412,6 +412,17 @@ static void read_channel_args(grpc_chttp2_transport* t,
           .GetBool(GRPC_ARG_EXPERIMENTAL_HTTP2_PREFERRED_CRYPTO_FRAME_SIZE)
           .value_or(false);
 
+  const auto max_requests_per_read =
+      channel_args.GetInt("grpc.http2.max_requests_per_read");
+  if (max_requests_per_read.has_value()) {
+    t->max_requests_per_read =
+        grpc_core::Clamp(*max_requests_per_read, 1, 10000);
+  } else if (grpc_core::IsChttp2BatchRequestsEnabled()) {
+    t->max_requests_per_read = 32;
+  } else {
+    t->max_requests_per_read = std::numeric_limits<size_t>::max();
+  }
+
   if (channel_args.GetBool(GRPC_ARG_ENABLE_CHANNELZ)
           .value_or(GRPC_ENABLE_CHANNELZ_DEFAULT)) {
     t->channelz_socket =
@@ -2500,21 +2511,34 @@ static void read_action(grpc_core::RefCountedPtr<grpc_chttp2_transport> t,
                     error);
 }
 
-static void read_action_locked(
+static void read_action_parse_loop_locked(
     grpc_core::RefCountedPtr<grpc_chttp2_transport> t,
     grpc_error_handle error) {
-  grpc_error_handle err = error;
-  if (!err.ok()) {
-    err = grpc_error_set_int(
-        GRPC_ERROR_CREATE_REFERENCING("Endpoint read failed", &err, 1),
-        grpc_core::StatusIntProperty::kOccurredDuringWrite, t->write_state);
-  }
-  std::swap(err, error);
   if (t->closed_with_error.ok()) {
-    size_t i = 0;
     grpc_error_handle errors[3] = {error, absl::OkStatus(), absl::OkStatus()};
-    for (; i < t->read_buffer.count && errors[1] == absl::OkStatus(); i++) {
-      errors[1] = grpc_chttp2_perform_read(t.get(), t->read_buffer.slices[i]);
+    size_t requests_started = 0;
+    for (size_t i = 0;
+         i < t->read_buffer.count && errors[1] == absl::OkStatus(); i++) {
+      auto r = grpc_chttp2_perform_read(t.get(), t->read_buffer.slices[i],
+                                        requests_started);
+      if (auto* partial_read_size = absl::get_if<size_t>(&r)) {
+        for (size_t j = 0; j < i; j++) {
+          grpc_core::CSliceUnref(grpc_slice_buffer_take_first(&t->read_buffer));
+        }
+        grpc_slice_buffer_sub_first(
+            &t->read_buffer, *partial_read_size,
+            GRPC_SLICE_LENGTH(t->read_buffer.slices[0]));
+        t->combiner->ForceOffload();
+        auto* tp = t.get();
+        tp->combiner->Run(
+            grpc_core::InitTransportClosure<read_action_parse_loop_locked>(
+                std::move(t), &tp->read_action_locked),
+            std::move(errors[0]));
+        // Early return: we queued to retry later.
+        return;
+      } else {
+        errors[1] = std::move(absl::get<absl::Status>(r));
+      }
     }
     if (errors[1] != absl::OkStatus()) {
       errors[2] = try_http_parsing(t.get());
@@ -2573,6 +2597,19 @@ static void read_action_locked(
   }
 }
 
+static void read_action_locked(
+    grpc_core::RefCountedPtr<grpc_chttp2_transport> t,
+    grpc_error_handle error) {
+  grpc_error_handle err = error;
+  if (!err.ok()) {
+    err = grpc_error_set_int(
+        GRPC_ERROR_CREATE_REFERENCING("Endpoint read failed", &err, 1),
+        grpc_core::StatusIntProperty::kOccurredDuringWrite, t->write_state);
+  }
+  std::swap(err, error);
+  read_action_parse_loop_locked(std::move(t), std::move(err));
+}
+
 static void continue_read_action_locked(
     grpc_core::RefCountedPtr<grpc_chttp2_transport> t) {
   const bool urgent = !t->goaway_error.ok();

diff --git a/src/core/ext/transport/chttp2/transport/internal.h b/src/core/ext/transport/chttp2/transport/internal.h
@@ -307,6 +307,8 @@ struct grpc_chttp2_transport : public grpc_core::KeepsGrpcInitialized {
   /// data to write next write
   grpc_slice_buffer qbuf;
 
+  size_t max_requests_per_read;
+
   /// Set to a grpc_error object if a goaway frame is received. By default, set
   /// to absl::OkStatus()
   grpc_error_handle goaway_error;
@@ -630,10 +632,14 @@ grpc_chttp2_begin_write_result grpc_chttp2_begin_write(
     grpc_chttp2_transport* t);
 void grpc_chttp2_end_write(grpc_chttp2_transport* t, grpc_error_handle error);
 
-/// Process one slice of incoming data; return 1 if the connection is still
-/// viable after reading, or 0 if the connection should be torn down
-grpc_error_handle grpc_chttp2_perform_read(grpc_chttp2_transport* t,
-                                           const grpc_slice& slice);
+/// Process one slice of incoming data
+/// Returns:
+///  - a count of parsed bytes in the event of a partial read: the caller should
+///    offload responsibilities to another thread to continue parsing.
+///  - or a status in the case of a completed read
+absl::variant<size_t, absl::Status> grpc_chttp2_perform_read(
+    grpc_chttp2_transport* t, const grpc_slice& slice,
+    size_t& requests_started);
 
 bool grpc_chttp2_list_add_writable_stream(grpc_chttp2_transport* t,
                                           grpc_chttp2_stream* s);

diff --git a/src/core/ext/transport/chttp2/transport/parsing.cc b/src/core/ext/transport/chttp2/transport/parsing.cc
@@ -70,9 +70,11 @@ using grpc_core::HPackParser;
 
 grpc_core::TraceFlag grpc_trace_chttp2_new_stream(false, "chttp2_new_stream");
 
-static grpc_error_handle init_frame_parser(grpc_chttp2_transport* t);
+static grpc_error_handle init_frame_parser(grpc_chttp2_transport* t,
+                                           size_t& requests_started);
 static grpc_error_handle init_header_frame_parser(grpc_chttp2_transport* t,
-                                                  int is_continuation);
+                                                  int is_continuation,
+                                                  size_t& requests_started);
 static grpc_error_handle init_data_frame_parser(grpc_chttp2_transport* t);
 static grpc_error_handle init_rst_stream_parser(grpc_chttp2_transport* t);
 static grpc_error_handle init_settings_frame_parser(grpc_chttp2_transport* t);
@@ -194,8 +196,9 @@ std::string FrameTypeString(uint8_t frame_type, uint8_t flags) {
 }
 }  // namespace
 
-grpc_error_handle grpc_chttp2_perform_read(grpc_chttp2_transport* t,
-                                           const grpc_slice& slice) {
+absl::variant<size_t, absl::Status> grpc_chttp2_perform_read(
+    grpc_chttp2_transport* t, const grpc_slice& slice,
+    size_t& requests_started) {
   const uint8_t* beg = GRPC_SLICE_START_PTR(slice);
   const uint8_t* end = GRPC_SLICE_END_PTR(slice);
   const uint8_t* cur = beg;
@@ -249,6 +252,10 @@ grpc_error_handle grpc_chttp2_perform_read(grpc_chttp2_transport* t,
         return absl::OkStatus();
       }
     dts_fh_0:
+      if (requests_started >= t->max_requests_per_read) {
+        t->deframe_state = GRPC_DTS_FH_0;
+        return static_cast<size_t>(cur - beg);
+      }
       ABSL_FALLTHROUGH_INTENDED;
     case GRPC_DTS_FH_0:
       GPR_DEBUG_ASSERT(cur < end);
@@ -324,7 +331,7 @@ grpc_error_handle grpc_chttp2_perform_read(grpc_chttp2_transport* t,
                 t->incoming_frame_size, t->incoming_stream_id);
       }
       t->deframe_state = GRPC_DTS_FRAME;
-      err = init_frame_parser(t);
+      err = init_frame_parser(t, requests_started);
       if (!err.ok()) {
         return err;
       }
@@ -378,6 +385,10 @@ grpc_error_handle grpc_chttp2_perform_read(grpc_chttp2_transport* t,
         }
         cur += t->incoming_frame_size;
         t->incoming_stream = nullptr;
+        if (t->incoming_frame_type == GRPC_CHTTP2_FRAME_RST_STREAM &&
+            grpc_core::IsChttp2OffloadOnRstStreamEnabled()) {
+          requests_started = std::numeric_limits<size_t>::max();
+        }
         goto dts_fh_0;  // loop
       } else {
         err = parse_frame_slice(
@@ -397,7 +408,8 @@ grpc_error_handle grpc_chttp2_perform_read(grpc_chttp2_transport* t,
   GPR_UNREACHABLE_CODE(return absl::OkStatus());
 }
 
-static grpc_error_handle init_frame_parser(grpc_chttp2_transport* t) {
+static grpc_error_handle init_frame_parser(grpc_chttp2_transport* t,
+                                           size_t& requests_started) {
   if (t->is_first_frame &&
       t->incoming_frame_type != GRPC_CHTTP2_FRAME_SETTINGS) {
     return GRPC_ERROR_CREATE(absl::StrCat(
@@ -417,13 +429,13 @@ static grpc_error_handle init_frame_parser(grpc_chttp2_transport* t) {
           "grpc_chttp2_stream %08x",
           t->expect_continuation_stream_id, t->incoming_stream_id));
     }
-    return init_header_frame_parser(t, 1);
+    return init_header_frame_parser(t, 1, requests_started);
   }
   switch (t->incoming_frame_type) {
     case GRPC_CHTTP2_FRAME_DATA:
       return init_data_frame_parser(t);
     case GRPC_CHTTP2_FRAME_HEADER:
-      return init_header_frame_parser(t, 0);
+      return init_header_frame_parser(t, 0, requests_started);
     case GRPC_CHTTP2_FRAME_CONTINUATION:
       return GRPC_ERROR_CREATE("Unexpected CONTINUATION frame");
     case GRPC_CHTTP2_FRAME_RST_STREAM:
@@ -567,7 +579,8 @@ static grpc_error_handle init_data_frame_parser(grpc_chttp2_transport* t) {
 }
 
 static grpc_error_handle init_header_frame_parser(grpc_chttp2_transport* t,
-                                                  int is_continuation) {
+                                                  int is_continuation,
+                                                  size_t& requests_started) {
   const bool is_eoh =
       (t->incoming_frame_flags & GRPC_CHTTP2_DATA_FLAG_END_HEADERS) != 0;
   grpc_chttp2_stream* s;
@@ -643,6 +656,7 @@ static grpc_error_handle init_header_frame_parser(grpc_chttp2_transport* t,
     t->last_new_stream_id = t->incoming_stream_id;
     s = t->incoming_stream =
         grpc_chttp2_parsing_accept_stream(t, t->incoming_stream_id);
+    ++requests_started;
     if (GPR_UNLIKELY(s == nullptr)) {
       GRPC_CHTTP2_IF_TRACING(
           gpr_log(GPR_ERROR, "grpc_chttp2_stream not accepted"));

diff --git a/src/core/lib/experiments/experiments.cc b/src/core/lib/experiments/experiments.cc