feat(dfly_bench): allow regulated throughput in 3 modes (#4962)

romange · web-flow · commit 7ffe81296794 · 2025-04-21T09:56:33.000+03:00
* feat(dfly_bench): allow regulated throughput in 3 modes

1. Coordinated omission - with --qps=0, each request is sent and then we wait for the response and so on.
   For pipeline mode, k requests are sent and then we wait for them to return to send another k
2. qps &gt; 0: we schedule sending requests at frequency "qps" per connection but if pending requests count crosses a limit
   we slow down by throttling request sending. This mode enables gentle uncoordinated omission, where the schedule
   converges to the real throughput capacity of the backend (if it's slower than the target throughput).
3. qps &lt; 0, similar as (2) but does not adjust its scheduling and may overload the server
   if target QPS is too high.

Signed-off-by: Roman Gershman &lt;roman@dragonflydb.io&gt;

* chore: change pipelining and coordinated omission logic

Before that the uncoordinated omission only worked without pipelining.
Now, with pipelining mode with send a burst of P requests and then:
a) For coordinated omission - wait for all of them to complete before proceeding
   further
b) For non-coordinated omission - we sleep to pace our single connection throughput as
   defined by the qps setting.

Signed-off-by: Roman Gershman &lt;roman@dragonflydb.io&gt;

---------

Signed-off-by: Roman Gershman &lt;roman@dragonflydb.io&gt;
diff --git a/src/server/dfly_bench.cc b/src/server/dfly_bench.cc
@@ -37,7 +37,12 @@ using std::string;
 
 ABSL_FLAG(uint16_t, p, 6379, "Server port");
 ABSL_FLAG(uint32_t, c, 20, "Number of connections per thread");
-ABSL_FLAG(uint32_t, qps, 20, "QPS schedule at which the generator sends requests to the server");
+ABSL_FLAG(int32_t, qps, 20,
+          "QPS schedule at which the generator sends requests to the server "
+          "per single connection. 0 means - coordinated omission, and positive value will throttle "
+          "the actual qps if server is slower than the target qps. "
+          "negative value means - hard target, without throttling.");
+
 ABSL_FLAG(uint32_t, n, 1000, "Number of requests to send per connection");
 ABSL_FLAG(uint32_t, test_time, 0, "Testing time in seconds");
 ABSL_FLAG(uint32_t, d, 16, "Value size in bytes ");
@@ -604,24 +609,56 @@ void Driver::Connect(unsigned index, const tcp::endpoint& ep) {
 
 void Driver::Run(uint64_t* cycle_ns, CommandGenerator* cmd_gen) {
   start_ns_ = absl::GetCurrentTimeNanos();
-  unsigned pipeline = GetFlag(FLAGS_pipeline);
+  uint32_t pipeline = std::max<uint32_t>(GetFlag(FLAGS_pipeline), 1u);
+  bool should_throttle = GetFlag(FLAGS_qps) > 0;
 
   stats_.num_clients++;
   int64_t time_limit_ns =
       time_limit_ > 0 ? int64_t(time_limit_) * 1'000'000'000 + start_ns_ : INT64_MAX;
-
+  int64_t now = start_ns_;
   SlotRange slot_range{0, kNumSlots - 1};
+  CHECK_GT(num_reqs_, 0u);
 
-  for (unsigned i = 0; i < num_reqs_; ++i) {
-    int64_t now = absl::GetCurrentTimeNanos();
+  uint32_t num_batches = ((num_reqs_ - 1) / pipeline) + 1;
 
-    if (now > time_limit_ns) {
-      break;
+  for (unsigned i = 0; i < num_batches && now < time_limit_ns; ++i) {
+    if (i == num_batches - 1) {  // last batch
+      pipeline = num_reqs_ - i * pipeline;
+    }
+
+    for (unsigned j = 0; j < pipeline; ++j) {
+      // TODO: this skews the distribution if slot ranges are uneven.
+      // Ideally we would like to pick randomly a single slot from all the ranges we have
+      // and pass it to cmd_gen->Next below.
+      if (!shard_slots_.Empty()) {
+        slot_range = shard_slots_.NextSlotRange(ep_, i);
+      }
+
+      string cmd = cmd_gen->Next(slot_range);
+
+      Req req;
+      req.start = absl::GetCurrentTimeNanos();
+      req.might_hit = cmd_gen->might_hit();
+
+      reqs_.push(req);
+
+      error_code ec = socket_->Write(io::Buffer(cmd));
+      if (ec && FiberSocketBase::IsConnClosed(ec)) {
+        // TODO: report failure
+        VLOG(1) << "Connection closed";
+        break;
+      }
+      CHECK(!ec) << ec.message();
+      if (cmd_gen->noreply()) {
+        PopRequest();
+      }
     }
+
+    now = absl::GetCurrentTimeNanos();
     if (cycle_ns) {
       int64_t target_ts = start_ns_ + i * (*cycle_ns);
       int64_t sleep_ns = target_ts - now;
-      if (reqs_.size() > 10 && sleep_ns <= 0) {
+      if (reqs_.size() > pipeline * 2 && should_throttle && sleep_ns <= 0) {
         sleep_ns = 10'000;
       }
 
@@ -630,7 +667,7 @@ void Driver::Run(uint64_t* cycle_ns, CommandGenerator* cmd_gen) {
         // There is no point in sending more requests if they are piled up in the server.
         do {
           ThisFiber::SleepFor(chrono::nanoseconds(sleep_ns));
-        } while (reqs_.size() > 10);
+        } while (should_throttle && reqs_.size() > pipeline * 2);
       } else if (i % 256 == 255) {
         ThisFiber::Yield();
         VLOG(5) << "Behind QPS schedule";
@@ -639,33 +676,7 @@ void Driver::Run(uint64_t* cycle_ns, CommandGenerator* cmd_gen) {
       // Coordinated omission.
 
       fb2::NoOpLock lk;
-      cnd_.wait(lk, [this, pipeline] { return reqs_.size() < pipeline; });
-    }
-
-    // TODO: this skews the distribution if slot ranges are uneven.
-    // Ideally we would like to pick randomly a single slot from all the ranges we have
-    // and pass it to cmd_gen->Next below.
-    if (!shard_slots_.Empty()) {
-      slot_range = shard_slots_.NextSlotRange(ep_, i);
-    }
-
-    string cmd = cmd_gen->Next(slot_range);
-
-    Req req;
-    req.start = absl::GetCurrentTimeNanos();
-    req.might_hit = cmd_gen->might_hit();
-
-    reqs_.push(req);
-
-    error_code ec = socket_->Write(io::Buffer(cmd));
-    if (ec && FiberSocketBase::IsConnClosed(ec)) {
-      // TODO: report failure
-      VLOG(1) << "Connection closed";
-      break;
-    }
-    CHECK(!ec) << ec.message();
-    if (cmd_gen->noreply()) {
-      PopRequest();
+      cnd_.wait(lk, [this] { return reqs_.empty(); });
     }
   }
 
@@ -908,12 +919,15 @@ void WatchFiber(size_t num_shards, atomic_bool* finish_signal, ProactorPool* pp)
   num_shards = max<size_t>(num_shards, 1u);
   uint64_t resp_goal = GetFlag(FLAGS_c) * pp->size() * GetFlag(FLAGS_n) * num_shards;
   uint32_t time_limit = GetFlag(FLAGS_test_time);
+  bool should_throttle = GetFlag(FLAGS_qps) > 0;
 
   while (*finish_signal == false) {
     // we sleep with resolution of 1s but print with lower frequency to be more responsive
     // when benchmark finishes.
     ThisFiber::SleepFor(1s);
-    pp->AwaitBrief([](auto, auto*) { client->AdjustCycle(); });
+    if (should_throttle) {
+      pp->AwaitBrief([](auto, auto*) { client->AdjustCycle(); });
+    }
 
     int64_t now = absl::GetCurrentTimeNanos();
     if (now - last_print < 5000'000'000LL)  // 5s
@@ -1084,9 +1098,9 @@ int main(int argc, char* argv[]) {
   if (protocol == RESP) {
     shards = proactor->Await([&] { return FetchClusterInfo(ep, proactor); });
   }
-  LOG(INFO) << "Connecting threads to "
-            << (shards.empty() ? string("single node ")
-                               : absl::StrCat(shards.size(), " shard cluster"));
+  CONSOLE_INFO << "Connecting to "
+               << (shards.empty() ? string("single node ")
+                                  : absl::StrCat(shards.size(), " shard cluster"));
 
   if (!shards.empty() && !GetFlag(FLAGS_command).empty() && GetFlag(FLAGS_cluster_skip_tags)) {
     // For custom commands we may need to use the same hashtag for multiple keys.
@@ -1112,9 +1126,11 @@ int main(int argc, char* argv[]) {
   CHECK_LE(key_minimum, key_maximum);
 
   uint32_t thread_key_step = 0;
-  const uint32_t qps = GetFlag(FLAGS_qps);
+  uint32_t qps = abs(GetFlag(FLAGS_qps));
+  bool throttle = GetFlag(FLAGS_qps) > 0;
   const int64_t interval = qps ? 1'000'000'000LL / qps : 0;
   uint64_t num_reqs = GetFlag(FLAGS_n);
+
   uint64_t total_conn_num = GetFlag(FLAGS_c) * pp->size();
   uint64_t total_requests = num_reqs * total_conn_num;
   uint32_t time_limit = GetFlag(FLAGS_test_time);
@@ -1130,11 +1146,12 @@ int main(int argc, char* argv[]) {
 
   if (!time_limit) {
     CONSOLE_INFO << "Running " << pp->size() << " threads, sending " << num_reqs
-                 << " requests per each connection, or " << total_requests << " requests overall";
+                 << " requests per each connection, or " << total_requests << " requests overall "
+                 << (throttle ? "with" : "without") << " throttling";
   }
   if (interval) {
-    CONSOLE_INFO << "At a rate of " << GetFlag(FLAGS_qps)
-                 << " rps per connection, i.e. request every " << interval / 1000 << "us";
+    CONSOLE_INFO << "At a rate of " << qps << " rps per connection, i.e. request every "
+                 << interval / 1000 << "us";
     CONSOLE_INFO << "Overall scheduled RPS: " << qps * total_conn_num;
   } else {
     CONSOLE_INFO << "Coordinated omission mode - the rate is determined by the server";
diff --git a/tools/local/monitoring/grafana/provisioning/dashboards/dragonfly.json b/tools/local/monitoring/grafana/provisioning/dashboards/dragonfly.json
@@ -1143,6 +1143,109 @@
       "title": "Network I/O",
       "type": "timeseries"
     },
+    {
+      "datasource": {
+        "type": "prometheus",
+        "uid": "${DS_PROMETHEUS}"
+      },
+      "fieldConfig": {
+        "defaults": {
+          "color": {
+            "mode": "palette-classic"
+          },
+          "custom": {
+            "axisCenteredZero": false,
+            "axisColorMode": "text",
+            "axisLabel": "",
+            "axisPlacement": "auto",
+            "barAlignment": 0,
+            "drawStyle": "line",
+            "fillOpacity": 0,
+            "gradientMode": "none",
+            "hideFrom": {
+              "legend": false,
+              "tooltip": false,
+              "viz": false
+            },
+            "insertNulls": false,
+            "lineInterpolation": "linear",
+            "lineWidth": 1,
+            "pointSize": 5,
+            "scaleDistribution": {
+              "type": "linear"
+            },
+            "showPoints": "auto",
+            "spanNulls": false,
+            "stacking": {
+              "group": "A",
+              "mode": "none"
+            },
+            "thresholdsStyle": {
+              "mode": "off"
+            }
+          },
+          "links": [],
+          "mappings": [],
+          "thresholds": {
+            "mode": "absolute",
+            "steps": [
+              {
+                "color": "green",
+                "value": null
+              },
+              {
+                "color": "red",
+                "value": 80
+              }
+            ]
+          },
+          "unit": "s"
+        },
+        "overrides": []
+      },
+      "gridPos": {
+        "h": 7,
+        "w": 12,
+        "x": 0,
+        "y": 29
+      },
+      "id": 26,
+      "options": {
+        "alertThreshold": true,
+        "legend": {
+          "calcs": [],
+          "displayMode": "list",
+          "placement": "bottom",
+          "showLegend": true
+        },
+        "tooltip": {
+          "mode": "single",
+          "sort": "none"
+        }
+      },
+      "pluginVersion": "10.1.10",
+      "targets": [
+        {
+          "datasource": {
+            "type": "prometheus",
+            "uid": "${DS_PROMETHEUS}"
+          },
+          "editorMode": "code",
+          "exemplar": true,
+          "expr":
+              "irate(dragonfly_reply_duration_seconds{namespace=\"$namespace\",pod=~\"$pod_name\"}[$__interval]) / irate(dragonfly_reply_total{namespace=\"$namespace\",pod=~\"$pod_name\"}[$__interval])",
+          "format": "time_series",
+          "interval": "",
+          "intervalFactor": 2,
+          "legendFormat": "{{ pod }} input",
+          "range": true,
+          "refId": "A",
+          "step": 240
+        }
+      ],
+      "title": "Reply Latency",
+      "type": "timeseries"
+    },
     {
       "datasource": {
         "type": "prometheus",
@@ -1422,7 +1525,8 @@
             "mode": "absolute",
             "steps": [
               {
-                "color": "green"
+                "color": "green",
+                "value": null
               },
               {
                 "color": "red",
@@ -1536,7 +1640,8 @@
             "mode": "absolute",
             "steps": [
               {
-                "color": "green"
+                "color": "green",
+                "value": null
               },
               {
                 "color": "red",
@@ -1574,8 +1679,7 @@
             "uid": "${DS_PROMETHEUS}"
           },
           "editorMode": "code",
-          "expr":
-              "dragonfly_pipeline_queue_length{namespace=\"$namespace\",pod=~\"$pod_name\"}/dragonfly_connected_clients{namespace=\"$namespace\",pod=~\"$pod_name\"}",
+          "expr": "dragonfly_pipeline_queue_length{namespace=\"$namespace\",pod=~\"$pod_name\"}",
           "instant": false,
           "legendFormat": "avr_pipeline_depth",
           "range": true,
@@ -1631,7 +1735,8 @@
             "mode": "absolute",
             "steps": [
               {
-                "color": "green"
+                "color": "green",
+                "value": null
               },
               {
                 "color": "red",
@@ -1740,7 +1845,8 @@
             "mode": "absolute",
             "steps": [
               {
-                "color": "green"
+                "color": "green",
+                "value": null
               },
               {
                 "color": "red",
