Pending requests: measure time spend waiting on a full queue

integration/integration_test_fixtures.py

@@ -80,6 +80,12 @@ def getNighthawkCounterMapFromJson(self, parsed_json):
         for counter in parsed_json["results"][0]["counters"]
     }
 
+  def getNighthawkGlobalHistogramsbyIdFromJson(self, parsed_json):
+    """
+    Utility method to get the global histograms from the json indexed by id.
+    """
+    return {statistic["id"]: statistic for statistic in parsed_json["results"][0]["statistics"]}
+
   def getTestServerRootUri(self, https=False):
     """
     Utility for getting the http://host:port/ that can be used to query the server we started in setUp()

integration/test_integration_basics.py

@@ -29,6 +29,31 @@ def test_h1(self):
     self.assertEqual(counters["upstream_rq_total"], 25)
     self.assertEqual(len(counters), 9)
 
+  def mini_stress_test_h1(self, args):
+    # run a test with more rps then we can handle, and a very small client-side queue.
+    # we should observe both lots of successfull requests as well as time spend in blocking mode.,
+    parsed_json = self.runNighthawkClient(args)
+    counters = self.getNighthawkCounterMapFromJson(parsed_json)
+    self.assertGreater(counters["benchmark.http_2xx"], 1000)
+    self.assertEqual(counters["upstream_cx_http1_total"], 1)
+    global_histograms = self.getNighthawkGlobalHistogramsbyIdFromJson(parsed_json)
+    self.assertGreater(int(global_histograms["sequencer.blocking"]["count"]), 1000)
+    self.assertGreater(
+        int(global_histograms["benchmark_http_client.request_to_response"]["count"]), 1000)
+    return counters
+
+  def test_h1_mini_stress_test_with_client_side_queueing(self):
+    counters = self.mini_stress_test_h1([
+        self.getTestServerRootUri(), "--rps", "999999", "--max-pending-requests", "10",
+        "--duration 2"
+    ])
+    self.assertGreater(counters["upstream_rq_pending_total"], 100)
+
+  def test_h1_mini_stress_test_without_client_side_queueing(self):
+    counters = self.mini_stress_test_h1(
+        [self.getTestServerRootUri(), "--rps", "999999", "--duration 2"])
+    self.assertEqual(counters["upstream_rq_pending_total"], 1)
+
   def test_h2(self):
     parsed_json = self.runNighthawkClient(["--h2", self.getTestServerRootUri()])
     counters = self.getNighthawkCounterMapFromJson(parsed_json)

source/client/benchmark_client_impl.cc

@@ -189,16 +189,19 @@ void BenchmarkClientHttpImpl::setRequestHeader(absl::string_view key, absl::stri
 }
 
 bool BenchmarkClientHttpImpl::tryStartOne(std::function<void()> caller_completion_callback) {
-  // When no client side queueing is specified (via max_pending_requests_), we are in closed loop
-  // mode. In closed loop mode we want to be able to control the pacing as exactly as possible. In
-  // open-loop mode we probably want to skip this. NOTE(oschaaf): We can't consistently rely on
-  // resourceManager()::requests() because that isn't used for h/1 (it is used in tcp and h2
-  // though).
-  if (max_pending_requests_ == 1 &&
-      (!cluster_->resourceManager(Envoy::Upstream::ResourcePriority::Default)
-            .pendingRequests()
-            .canCreate() ||
-       ((requests_initiated_ - requests_completed_) >= connection_limit_))) {
+  // When we allow client-side queuing, we want to have a sense of time spend waiting on that queue.
+  // So we return false here to indicate we couldn't initiate a new request.
+  if (!cluster_->resourceManager(Envoy::Upstream::ResourcePriority::Default)
+           .pendingRequests()
+           .canCreate()) {
+    return false;
+  }
+  // When no client side queueing is disabled (max_pending equals 1) we control the pacing as
+  // exactly as possible here.
+  // NOTE: We can't consistently rely on resourceManager()::requests()
+  // because that isn't used for h/1 (it is used in tcp and h2 though).
+  if ((max_pending_requests_ == 1 &&
+       (requests_initiated_ - requests_completed_) >= connection_limit_)) {
     return false;
   }