benchmark: align --request_interval_us help text with actual pacing semantics

mooncake-transfer-engine/benchmark/main.cpp

@@ -22,8 +22,26 @@
 #include "tent_backend.h"
 #endif
 
+#include <atomic>
+#include <mutex>
+#include <thread>
+
 using namespace mooncake::tent;
 
+namespace {
+
+uint64_t steadyClockNs() {
+    const auto now = std::chrono::steady_clock::now().time_since_epoch();
+    return std::chrono::duration_cast<std::chrono::nanoseconds>(now).count();
+}
+
+double gbPerSecond(uint64_t bytes, double duration_us) {
+    if (duration_us <= 0.0) return 0.0;
+    return static_cast<double>(bytes) / (1000.0 * duration_us);
+}
+
+}  // namespace
+
 int processBatchSizes(
     BenchRunner& runner, size_t block_size, size_t batch_size, int num_threads,
     const std::vector<QosClassConfig>& qos_classes,
@@ -46,6 +64,17 @@ int processBatchSizes(
     XferBenchStats tight_stats;
     XferBenchStats loose_stats;
     std::mutex mutex;
+    std::atomic<int> measurement_ready{0};
+    std::atomic<bool> measurement_started{false};
+    auto paceRequest = [&]() {
+        if (XferBenchConfig::request_interval_us == 0) return;
+        const uint64_t interval_ns =
+            XferBenchConfig::request_interval_us * 1000ull;
+        const uint64_t target_ns = steadyClockNs() + interval_ns;
+        while (steadyClockNs() < target_ns) {
+            std::this_thread::yield();
+        }
+    };
     size_t address_stride_bytes =
         XferBenchConfig::max_block_size * XferBenchConfig::max_batch_size;
     if (!workload_classes.empty()) {
@@ -94,24 +123,41 @@ int processBatchSizes(
                                      thread_batch_size, opcode, deadlineNs(),
                                      intent_type);
         }
+        if (measurement_ready.fetch_add(1, std::memory_order_acq_rel) + 1 ==
+            num_threads) {
+            measurement_started.store(true, std::memory_order_release);
+        } else {
+            while (!measurement_started.load(std::memory_order_acquire)) {
+                std::this_thread::yield();
+            }
+        }
         timer.reset();
         std::vector<double> transfer_duration;
+        std::vector<double> thread_instant_bandwidth;
         if (mixed_opcode) {
             while (timer.lap_us(false) <
                    XferBenchConfig::duration * 1000000ull) {
+                const uint64_t batch_bytes =
+                    thread_block_size * thread_batch_size;
                 uint8_t pattern = 0;
                 if (XferBenchConfig::check_consistency)
                     pattern = fillData((void*)local_addr,
                                        thread_block_size * thread_batch_size);
+                paceRequest();
                 auto val = runner.runSingleTransfer(
                     local_addr, target_addr, thread_block_size,
                     thread_batch_size, WRITE, deadlineNs(), intent_type);
+                thread_instant_bandwidth.push_back(gbPerSecond(batch_bytes,
+                                                               val));
                 transfer_duration.push_back(val);
                 fillData((void*)local_addr,
                          thread_block_size * thread_batch_size);
+                paceRequest();
                 val = runner.runSingleTransfer(
                     local_addr, target_addr, thread_block_size,
                     thread_batch_size, READ, deadlineNs(), intent_type);
+                thread_instant_bandwidth.push_back(gbPerSecond(batch_bytes,
+                                                               val));
                 if (XferBenchConfig::check_consistency)
                     verifyData((void*)local_addr,
                                thread_block_size * thread_batch_size, pattern);
@@ -120,16 +166,22 @@ int processBatchSizes(
         } else {
             while (timer.lap_us(false) <
                    XferBenchConfig::duration * 1000000ull) {
+                const uint64_t batch_bytes =
+                    thread_block_size * thread_batch_size;
+                paceRequest();
                 auto val = runner.runSingleTransfer(
                     local_addr, target_addr, thread_block_size,
                     thread_batch_size, opcode, deadlineNs(), intent_type);
+                thread_instant_bandwidth.push_back(gbPerSecond(batch_bytes,
+                                                               val));
                 transfer_duration.push_back(val);
             }
         }
         auto total_duration = timer.lap_us();
         std::lock_guard<std::mutex> lock(mutex);
         stats.total_duration.add(total_duration);
         stats.transfer_duration.add(transfer_duration);
+        stats.instant_bandwidth.add(thread_instant_bandwidth);
         if (qos_enabled) {
             qos_stats[qos_class].total_duration.add(total_duration);
             qos_stats[qos_class].transfer_duration.add(transfer_duration);

mooncake-transfer-engine/benchmark/utils.cpp

@@ -60,6 +60,9 @@ DEFINE_double(qos_link_capacity_gbps, 0.0,
               "Link capacity in GB/s for total utilization (0 reports N/A).");
 DEFINE_string(qos_output_jsonl, "",
               "Append versioned QoS metric records to this JSONL file.");
+DEFINE_uint64(request_interval_us, 0,
+              "Per-thread delay before each issued transfer batch, in "
+              "microseconds. 0 disables pacing.");
 DEFINE_uint64(deadline_us, 0,
               "tent only: relative per-transfer deadline in microseconds for "
               "tight worker threads (0 disables deadline tagging); cannot be "
@@ -115,6 +118,7 @@ std::string XferBenchConfig::qos_classes_json;
 std::string XferBenchConfig::workload_classes_json;
 double XferBenchConfig::qos_link_capacity_gbps = 0.0;
 std::string XferBenchConfig::qos_output_jsonl;
+uint64_t XferBenchConfig::request_interval_us = 0;
 uint64_t XferBenchConfig::deadline_us = 0;
 int XferBenchConfig::deadline_tight_threads = 0;
 bool XferBenchConfig::deadline_bw_arbitration = false;
@@ -151,6 +155,7 @@ void XferBenchConfig::loadFromFlags() {
     workload_classes_json = FLAGS_workload_classes_json;
     qos_link_capacity_gbps = FLAGS_qos_link_capacity_gbps;
     qos_output_jsonl = FLAGS_qos_output_jsonl;
+    request_interval_us = FLAGS_request_interval_us;
     deadline_us = FLAGS_deadline_us;
     deadline_tight_threads = FLAGS_deadline_tight_threads;
     deadline_bw_arbitration = FLAGS_deadline_bw_arbitration;
@@ -191,7 +196,8 @@ void printStatsHeader() {
     std::cout << std::left
               << std::setw(14) << "BlkSize (B)"
               << std::setw(8) << "Batch"
-              << std::setw(14) << "BW (GB/S)"
+              << std::setw(14) << "BW (GB/s)"
+              << std::setw(18) << "Avg Inst GB/s"
               << std::setw(14) << "Avg Lat (us)"
               << std::setw(14) << "Avg Tx (us)"
               << std::setw(14) << "P99 Tx (us)"
@@ -208,16 +214,22 @@ void printStats(size_t block_size, size_t batch_size, XferBenchStats& stats,
     auto num_ops = stats.transfer_duration.count();
     double total_duration = stats.total_duration.avg();
     total_data_transferred = ((block_size * batch_size) * num_ops);
-    avg_latency = (total_duration * num_threads / num_ops);
     throughput_gb = (((double)total_data_transferred / (1000 * 1000 * 1000)) /
                      (total_duration / 1e6));  // In GB/Sec
+    const double avg_instant_gbps = stats.instant_bandwidth.avg();
+    if (avg_instant_gbps > 0.0) {
+        avg_latency =
+            static_cast<double>(block_size * batch_size) /
+            (avg_instant_gbps * 1000.0);
+    }
 
     // Tabulate print with fixed width for each string
     // clang-format off
     std::cout << std::left << std::fixed << std::setprecision(6)
               << std::setw(14) << block_size
               << std::setw(8)  << batch_size
               << std::setw(14) << throughput_gb
+              << std::setw(18) << avg_instant_gbps
               << std::setprecision(1)
               << std::setw(14) << avg_latency
               << std::setw(14) << stats.transfer_duration.avg()

mooncake-transfer-engine/benchmark/utils.h

@@ -76,6 +76,7 @@ struct XferBenchConfig {
     static std::string workload_classes_json;
     static double qos_link_capacity_gbps;
     static std::string qos_output_jsonl;
+    static uint64_t request_interval_us;
     static uint64_t deadline_us;
     static int deadline_tight_threads;
     static bool deadline_bw_arbitration;
@@ -147,6 +148,7 @@ struct XferMetricStats {
 struct XferBenchStats {
     XferMetricStats total_duration;
     XferMetricStats transfer_duration;
+    XferMetricStats instant_bandwidth;
 };
 
 class XferBenchTimer {