[chttp2] Alternative protection for too many streams in the system (#34697)

Cancel streams if we have too much concurrency on a single channel to
allow that server to recover.

There seems to be a convergence in the HTTP2 community about this being
a reasonable thing to do, so I'd like to try it in some real scenarios.

If this pans out well then I'll likely drop the
`red_max_concurrent_streams` and the `rstpit` experiments in preference
to this.

I'm also considering tying in resource quota so that under high memory
pressure we just default to this path.

---------

Co-authored-by: ctiller <ctiller@users.noreply.github.com>

bazel/experiments.bzl

@@ -86,6 +86,7 @@ EXPERIMENTS = {
                 "chttp2_batch_requests",
                 "chttp2_offload_on_rst_stream",
                 "lazier_stream_updates",
+                "overload_protection",
                 "write_size_cap",
                 "write_size_policy",
             ],
@@ -173,6 +174,7 @@ EXPERIMENTS = {
                 "chttp2_batch_requests",
                 "chttp2_offload_on_rst_stream",
                 "lazier_stream_updates",
+                "overload_protection",
                 "write_size_cap",
                 "write_size_policy",
             ],
@@ -270,6 +272,7 @@ EXPERIMENTS = {
                 "chttp2_batch_requests",
                 "chttp2_offload_on_rst_stream",
                 "lazier_stream_updates",
+                "overload_protection",
                 "write_size_cap",
                 "write_size_policy",
             ],

src/core/ext/transport/chttp2/transport/chttp2_transport.cc

@@ -24,6 +24,7 @@
 #include <string.h>
 
 #include <algorithm>
+#include <atomic>
 #include <limits>
 #include <memory>
 #include <new>
@@ -840,6 +841,7 @@ grpc_chttp2_stream::grpc_chttp2_stream(grpc_chttp2_transport* t,
       initial_metadata_buffer(arena),
       trailing_metadata_buffer(arena),
       flow_control(&t->flow_control) {
+  t->streams_allocated.fetch_add(1, std::memory_order_relaxed);
   if (server_data) {
     id = static_cast<uint32_t>(reinterpret_cast<uintptr_t>(server_data));
     if (grpc_http_trace.enabled()) {
@@ -856,6 +858,7 @@ grpc_chttp2_stream::grpc_chttp2_stream(grpc_chttp2_transport* t,
 }
 
 grpc_chttp2_stream::~grpc_chttp2_stream() {
+  t->streams_allocated.fetch_sub(1, std::memory_order_relaxed);
   grpc_chttp2_list_remove_stalled_by_stream(t.get(), this);
   grpc_chttp2_list_remove_stalled_by_transport(t.get(), this);
 

src/core/ext/transport/chttp2/transport/internal.h

@@ -24,6 +24,7 @@
 #include <stddef.h>
 #include <stdint.h>
 
+#include <atomic>
 #include <memory>
 #include <utility>
 
@@ -475,6 +476,8 @@ struct grpc_chttp2_transport final
   /// grace period to wait for data after sending a ping before keepalives
   /// timeout
   grpc_core::Duration keepalive_timeout;
+  /// number of stream objects currently allocated by this transport
+  std::atomic<size_t> streams_allocated{0};
   /// keep-alive state machine state
   grpc_chttp2_keepalive_state keepalive_state;
   // Soft limit on max header size.

src/core/ext/transport/chttp2/transport/parsing.cc

@@ -21,6 +21,7 @@
 #include <inttypes.h>
 #include <string.h>
 
+#include <atomic>
 #include <initializer_list>
 #include <limits>
 #include <memory>
@@ -661,6 +662,18 @@ static grpc_error_handle init_header_frame_parser(grpc_chttp2_transport* t,
       } else {
         return GRPC_ERROR_CREATE("Max stream count exceeded");
       }
+    } else if (GPR_UNLIKELY(
+                   grpc_core::IsOverloadProtectionEnabled() &&
+                   t->streams_allocated.load(std::memory_order_relaxed) >
+                       t->max_concurrent_streams_policy.AdvertiseValue())) {
+      // We have more streams allocated than we'd like, so apply some pushback
+      // by refusing this stream.
+      ++t->num_pending_induced_frames;
+      grpc_slice_buffer_add(&t->qbuf, grpc_chttp2_rst_stream_create(
+                                          t->incoming_stream_id,
+                                          GRPC_HTTP2_REFUSED_STREAM, nullptr));
+      grpc_chttp2_initiate_write(t, GRPC_CHTTP2_INITIATE_WRITE_RST_STREAM);
+      return init_header_skip_frame_parser(t, priority_type, is_eoh);
     } else if (GPR_UNLIKELY(
                    grpc_core::IsRedMaxConcurrentStreamsEnabled() &&
                    t->stream_map.size() >=

src/core/lib/experiments/experiments.cc

@@ -83,6 +83,10 @@ const char* const additional_constraints_monitoring_experiment = "{}";
 const char* const description_multiping =
     "Allow more than one ping to be in flight at a time by default.";
 const char* const additional_constraints_multiping = "{}";
+const char* const description_overload_protection =
+    "If chttp2 has more streams than it can handle open, send RST_STREAM "
+    "immediately on new streams appearing.";
+const char* const additional_constraints_overload_protection = "{}";
 const char* const description_peer_state_based_framing =
     "If set, the max sizes of frames sent to lower layers is controlled based "
     "on the peer's memory pressure which is reflected in its max http2 frame "
@@ -235,6 +239,8 @@ const ExperimentMetadata g_experiment_metadata[] = {
      additional_constraints_monitoring_experiment, true, true},
     {"multiping", description_multiping, additional_constraints_multiping,
      false, true},
+    {"overload_protection", description_overload_protection,
+     additional_constraints_overload_protection, true, true},
     {"peer_state_based_framing", description_peer_state_based_framing,
      additional_constraints_peer_state_based_framing, false, true},
     {"pick_first_happy_eyeballs", description_pick_first_happy_eyeballs,
@@ -358,6 +364,10 @@ const char* const additional_constraints_monitoring_experiment = "{}";
 const char* const description_multiping =
     "Allow more than one ping to be in flight at a time by default.";
 const char* const additional_constraints_multiping = "{}";
+const char* const description_overload_protection =
+    "If chttp2 has more streams than it can handle open, send RST_STREAM "
+    "immediately on new streams appearing.";
+const char* const additional_constraints_overload_protection = "{}";
 const char* const description_peer_state_based_framing =
     "If set, the max sizes of frames sent to lower layers is controlled based "
     "on the peer's memory pressure which is reflected in its max http2 frame "
@@ -510,6 +520,8 @@ const ExperimentMetadata g_experiment_metadata[] = {
      additional_constraints_monitoring_experiment, true, true},
     {"multiping", description_multiping, additional_constraints_multiping,
      false, true},
+    {"overload_protection", description_overload_protection,
+     additional_constraints_overload_protection, true, true},
     {"peer_state_based_framing", description_peer_state_based_framing,
      additional_constraints_peer_state_based_framing, false, true},
     {"pick_first_happy_eyeballs", description_pick_first_happy_eyeballs,
@@ -633,6 +645,10 @@ const char* const additional_constraints_monitoring_experiment = "{}";
 const char* const description_multiping =
     "Allow more than one ping to be in flight at a time by default.";
 const char* const additional_constraints_multiping = "{}";
+const char* const description_overload_protection =
+    "If chttp2 has more streams than it can handle open, send RST_STREAM "
+    "immediately on new streams appearing.";
+const char* const additional_constraints_overload_protection = "{}";
 const char* const description_peer_state_based_framing =
     "If set, the max sizes of frames sent to lower layers is controlled based "
     "on the peer's memory pressure which is reflected in its max http2 frame "
@@ -785,6 +801,8 @@ const ExperimentMetadata g_experiment_metadata[] = {
      additional_constraints_monitoring_experiment, true, true},
     {"multiping", description_multiping, additional_constraints_multiping,
      false, true},
+    {"overload_protection", description_overload_protection,
+     additional_constraints_overload_protection, true, true},
     {"peer_state_based_framing", description_peer_state_based_framing,
      additional_constraints_peer_state_based_framing, false, true},
     {"pick_first_happy_eyeballs", description_pick_first_happy_eyeballs,

src/core/lib/experiments/experiments.h

@@ -91,6 +91,8 @@ inline bool IsMemoryPressureControllerEnabled() { return false; }
 #define GRPC_EXPERIMENT_IS_INCLUDED_MONITORING_EXPERIMENT
 inline bool IsMonitoringExperimentEnabled() { return true; }
 inline bool IsMultipingEnabled() { return false; }
+#define GRPC_EXPERIMENT_IS_INCLUDED_OVERLOAD_PROTECTION
+inline bool IsOverloadProtectionEnabled() { return true; }
 inline bool IsPeerStateBasedFramingEnabled() { return false; }
 #define GRPC_EXPERIMENT_IS_INCLUDED_PICK_FIRST_HAPPY_EYEBALLS
 inline bool IsPickFirstHappyEyeballsEnabled() { return true; }
@@ -164,6 +166,8 @@ inline bool IsMemoryPressureControllerEnabled() { return false; }
 #define GRPC_EXPERIMENT_IS_INCLUDED_MONITORING_EXPERIMENT
 inline bool IsMonitoringExperimentEnabled() { return true; }
 inline bool IsMultipingEnabled() { return false; }
+#define GRPC_EXPERIMENT_IS_INCLUDED_OVERLOAD_PROTECTION
+inline bool IsOverloadProtectionEnabled() { return true; }
 inline bool IsPeerStateBasedFramingEnabled() { return false; }
 #define GRPC_EXPERIMENT_IS_INCLUDED_PICK_FIRST_HAPPY_EYEBALLS
 inline bool IsPickFirstHappyEyeballsEnabled() { return true; }
@@ -237,6 +241,8 @@ inline bool IsMemoryPressureControllerEnabled() { return false; }
 #define GRPC_EXPERIMENT_IS_INCLUDED_MONITORING_EXPERIMENT
 inline bool IsMonitoringExperimentEnabled() { return true; }
 inline bool IsMultipingEnabled() { return false; }
+#define GRPC_EXPERIMENT_IS_INCLUDED_OVERLOAD_PROTECTION
+inline bool IsOverloadProtectionEnabled() { return true; }
 inline bool IsPeerStateBasedFramingEnabled() { return false; }
 #define GRPC_EXPERIMENT_IS_INCLUDED_PICK_FIRST_HAPPY_EYEBALLS
 inline bool IsPickFirstHappyEyeballsEnabled() { return true; }
@@ -296,6 +302,7 @@ enum ExperimentIds {
   kExperimentIdMemoryPressureController,
   kExperimentIdMonitoringExperiment,
   kExperimentIdMultiping,
+  kExperimentIdOverloadProtection,
   kExperimentIdPeerStateBasedFraming,
   kExperimentIdPickFirstHappyEyeballs,
   kExperimentIdPingOnRstStream,
@@ -397,6 +404,10 @@ inline bool IsMonitoringExperimentEnabled() {
 inline bool IsMultipingEnabled() {
   return IsExperimentEnabled(kExperimentIdMultiping);
 }
+#define GRPC_EXPERIMENT_IS_INCLUDED_OVERLOAD_PROTECTION
+inline bool IsOverloadProtectionEnabled() {
+  return IsExperimentEnabled(kExperimentIdOverloadProtection);
+}
 #define GRPC_EXPERIMENT_IS_INCLUDED_PEER_STATE_BASED_FRAMING
 inline bool IsPeerStateBasedFramingEnabled() {
   return IsExperimentEnabled(kExperimentIdPeerStateBasedFraming);

src/core/lib/experiments/experiments.yaml

@@ -144,6 +144,13 @@
   expiry: 2024/01/15
   owner: ctiller@google.com
   test_tags: [flow_control_test]
+- name: overload_protection
+  description:
+    If chttp2 has more streams than it can handle open, send RST_STREAM immediately
+    on new streams appearing.
+  expiry: 2024/03/03
+  owner: ctiller@google.com
+  test_tags: [flow_control_test]
 - name: peer_state_based_framing
   description:
     If set, the max sizes of frames sent to lower layers is controlled based

src/core/lib/experiments/rollouts.yaml

@@ -84,6 +84,8 @@
   default: false
 - name: monitoring_experiment
   default: true
+- name: overload_protection
+  default: true
 - name: peer_state_based_framing
   default: false
 - name: pick_first_happy_eyeballs