[chttp2] Experiments for rst_stream pushback (#34642)

Experiment 1: On RST_STREAM: reduce MAX_CONCURRENT_STREAMS for one round
trip.
Experiment 2: If a settings frame is outstanding with a lower
MAX_CONCURRENT_STREAMS than is configured, and we receive a new incoming
stream that would exceed the new cap, randomly reject it.

---------

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

BUILD

@@ -4043,11 +4043,13 @@ grpc_cc_library(
         "//src/core:iomgr_fwd",
         "//src/core:iomgr_port",
         "//src/core:match",
+        "//src/core:max_concurrent_streams_policy",
         "//src/core:memory_quota",
         "//src/core:ping_abuse_policy",
         "//src/core:ping_callbacks",
         "//src/core:ping_rate_policy",
         "//src/core:poll",
+        "//src/core:random_early_detection",
         "//src/core:ref_counted",
         "//src/core:resource_quota",
         "//src/core:resource_quota_trace",

src/core/BUILD

@@ -5736,6 +5736,20 @@ grpc_cc_library(
     ],
 )
 
+grpc_cc_library(
+    name = "max_concurrent_streams_policy",
+    srcs = [
+        "ext/transport/chttp2/transport/max_concurrent_streams_policy.cc",
+    ],
+    hdrs = [
+        "ext/transport/chttp2/transport/max_concurrent_streams_policy.h",
+    ],
+    deps = [
+        "//:gpr",
+        "//:gpr_platform",
+    ],
+)
+
 grpc_cc_library(
     name = "huffsyms",
     srcs = [

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

@@ -66,6 +66,7 @@
 #include "src/core/ext/transport/chttp2/transport/http_trace.h"
 #include "src/core/ext/transport/chttp2/transport/internal.h"
 #include "src/core/ext/transport/chttp2/transport/legacy_frame.h"
+#include "src/core/ext/transport/chttp2/transport/max_concurrent_streams_policy.h"
 #include "src/core/ext/transport/chttp2/transport/ping_abuse_policy.h"
 #include "src/core/ext/transport/chttp2/transport/ping_callbacks.h"
 #include "src/core/ext/transport/chttp2/transport/ping_rate_policy.h"
@@ -537,8 +538,12 @@ static void read_channel_args(grpc_chttp2_transport* t,
       const int value = channel_args.GetInt(setting.channel_arg_name)
                             .value_or(setting.default_value);
       if (value >= 0) {
-        queue_setting_update(t, setting.setting_id,
-                             grpc_core::Clamp(value, setting.min, setting.max));
+        const int clamped_value =
+            grpc_core::Clamp(value, setting.min, setting.max);
+        queue_setting_update(t, setting.setting_id, clamped_value);
+        if (setting.setting_id == GRPC_CHTTP2_SETTINGS_MAX_CONCURRENT_STREAMS) {
+          t->max_concurrent_streams_policy.SetTarget(clamped_value);
+        }
       } else if (setting.setting_id ==
                  GRPC_CHTTP2_SETTINGS_MAX_HEADER_LIST_SIZE) {
         // Set value to 1.25 * soft limit if this is larger than

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

@@ -52,6 +52,7 @@
 #include "src/core/ext/transport/chttp2/transport/hpack_parser.h"
 #include "src/core/ext/transport/chttp2/transport/http2_settings.h"
 #include "src/core/ext/transport/chttp2/transport/legacy_frame.h"
+#include "src/core/ext/transport/chttp2/transport/max_concurrent_streams_policy.h"
 #include "src/core/ext/transport/chttp2/transport/ping_abuse_policy.h"
 #include "src/core/ext/transport/chttp2/transport/ping_callbacks.h"
 #include "src/core/ext/transport/chttp2/transport/ping_rate_policy.h"
@@ -372,6 +373,8 @@ struct grpc_chttp2_transport : public grpc_core::KeepsGrpcInitialized {
       delayed_ping_timer_handle;
   grpc_closure retry_initiate_ping_locked;
 
+  grpc_core::Chttp2MaxConcurrentStreamsPolicy max_concurrent_streams_policy;
+
   /// ping acks
   size_t ping_ack_count = 0;
   size_t ping_ack_capacity = 0;

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

@@ -0,0 +1,44 @@
+// Copyright 2023 gRPC authors.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#include <grpc/support/port_platform.h>
+
+#include "src/core/ext/transport/chttp2/transport/max_concurrent_streams_policy.h"
+
+#include <utility>
+
+#include <grpc/support/log.h>
+
+namespace grpc_core {
+
+void Chttp2MaxConcurrentStreamsPolicy::AddDemerit() {
+  ++new_demerits_;
+  ++unacked_demerits_;
+}
+
+void Chttp2MaxConcurrentStreamsPolicy::FlushedSettings() {
+  sent_demerits_ += std::exchange(new_demerits_, 0);
+}
+
+void Chttp2MaxConcurrentStreamsPolicy::AckLastSend() {
+  GPR_ASSERT(unacked_demerits_ >= sent_demerits_);
+  unacked_demerits_ -= std::exchange(sent_demerits_, 0);
+}
+
+uint32_t Chttp2MaxConcurrentStreamsPolicy::AdvertiseValue() const {
+  if (target_ < unacked_demerits_) return 0;
+  return target_ - unacked_demerits_;
+}
+
+}  // namespace grpc_core