[TENT] Add QoS metrics baseline to tebench

docs/source/design/tent/tebench.md

@@ -91,6 +91,56 @@ Each output row corresponds to one benchmark configuration.
 
 A short (~1 second) warmup phase is executed before measurements begin.
 
+### 4.1 QoS Metrics Baseline
+
+Use `--qos_classes` to partition a fixed number of worker threads into QoS
+classes:
+
+```text
+name:threads:slo_us:weight[:isolated_gbps],...
+```
+
+For example, the following closed-loop mixed workload assigns four workers to
+an SLO-constrained foreground class and twelve workers to a best-effort
+checkpoint class:
+
+```bash
+./tebench \
+  --target_seg_name=<SEG> \
+  --backend=tent \
+  --start_num_threads=16 \
+  --max_num_threads=16 \
+  --qos_classes=foreground:4:1000:4:12.5,checkpoint:12:0:1:10.0 \
+  --qos_link_capacity_gbps=25 \
+  --qos_output_jsonl=qos-results.jsonl
+```
+
+The class thread counts must add up to the fixed `start_num_threads` value.
+An `slo_us` of zero marks a best-effort class. For TENT, a non-zero SLO is also
+sent as the request's relative deadline; the classic backend measures the same
+SLO but ignores the scheduling hint.
+
+The human-readable summary and the optional versioned JSONL record report:
+
+| Metric | Definition |
+| ------ | ---------- |
+| `slo_attainment` | Fraction of completed batches whose measured end-to-end transfer time is at most `slo_us` |
+| `p99_us` | P99 end-to-end batch transfer latency for the class |
+| `goodput_gbps` | Class throughput multiplied by SLO attainment; best-effort classes use attainment 1 |
+| `weighted_goodput_gbps` | Sum of `weight × goodput_gbps` |
+| `jain_fairness` | Jain index over per-class `throughput_gbps / weight` |
+| `isolation_leakage` | `max(0, 1 - mixed_throughput / isolated_throughput)` |
+| `total_utilization` | Aggregate measured throughput divided by `qos_link_capacity_gbps` |
+
+Isolation leakage requires a matching class-only baseline, supplied as the
+optional fifth class field. Total utilization requires
+`--qos_link_capacity_gbps`. Missing baselines are emitted as `N/A` in text and
+`null` in JSON rather than being inferred from the mixed run. Run isolated and
+mixed cases with the same block size, batch size, transport, memory type, and
+host pair. JSONL records retain `isolated_throughput_gbps` and
+`link_capacity_gbps` alongside the derived values so every metric can be
+recomputed from one record.
+
 ## 5. Runtime Configuration
 
 This section summarizes the key runtime options that control workload behavior,
@@ -192,3 +242,14 @@ gpu_id + thread_id
 
 * `--metadata_type` : `p2p | etcd | redis | http` (default: `p2p`)
 * `--metadata_url_list` : comma-separated URLs (ignored in `p2p` mode)
+
+### 5.7 QoS Reporting
+
+* `--qos_classes` : class/thread/SLO/weight contract described in Section 4.1
+* `--qos_link_capacity_gbps` : measured usable link capacity in decimal GB/s
+* `--qos_output_jsonl` : append one schema-versioned JSON object per benchmark
+  configuration
+
+QoS mode intentionally requires a fixed thread count. Sweep offered load by
+running explicit cases with different class thread allocations so every output
+record has an unambiguous workload contract.

mooncake-transfer-engine/benchmark/CMakeLists.txt

@@ -63,3 +63,15 @@ endif()
 set_target_properties(
   tebench PROPERTIES BUILD_WITH_INSTALL_RPATH TRUE
                      INSTALL_RPATH "$ORIGIN/../lib:$ORIGIN/../../mooncake-common${TANGRT_RPATH}")
+
+if(BUILD_UNIT_TESTS)
+  add_executable(tebench_qos_metrics_test tests/qos_metrics_test.cpp
+                                          qos_metrics.cpp utils.cpp)
+  target_link_libraries(tebench_qos_metrics_test
+                        PRIVATE transfer_engine gtest gtest_main)
+  target_include_directories(
+    tebench_qos_metrics_test
+    PRIVATE "${CMAKE_CURRENT_SOURCE_DIR}"
+            "${CMAKE_CURRENT_SOURCE_DIR}/../tent/include")
+  add_test(NAME tebench_qos_metrics_test COMMAND tebench_qos_metrics_test)
+endif()

mooncake-transfer-engine/benchmark/bench_runner.h

@@ -57,10 +57,10 @@ class BenchRunner {
 
     virtual double runSingleTransfer(uint64_t local_addr, uint64_t target_addr,
                                      uint64_t block_size, uint64_t batch_size,
-                                     OpCode opcode) = 0;
+                                     OpCode opcode, uint64_t deadline_ns) = 0;
 };
 
 }  // namespace tent
 }  // namespace mooncake
 
-#endif  // BENCH_RUNNER_H
+#endif  // BENCH_RUNNER_H

mooncake-transfer-engine/benchmark/main.cpp

@@ -15,6 +15,7 @@
 #include "utils.h"
 
 #include "bench_runner.h"
+#include "qos_metrics.h"
 #include "te_backend.h"
 #ifdef USE_TENT
 #include "tent_backend.h"
@@ -23,7 +24,8 @@
 using namespace mooncake::tent;
 
 int processBatchSizes(BenchRunner& runner, size_t block_size, size_t batch_size,
-                      int num_threads) {
+                      int num_threads,
+                      const std::vector<QosClassConfig>& qos_classes) {
     bool mixed_opcode = false;
     OpCode opcode = READ;
     if (XferBenchConfig::check_consistency || XferBenchConfig::op_type == "mix")
@@ -38,6 +40,9 @@ int processBatchSizes(BenchRunner& runner, size_t block_size, size_t batch_size,
     }
 
     XferBenchStats stats;
+    XferBenchStats tight_stats;
+    XferBenchStats loose_stats;
+    std::vector<XferBenchStats> qos_stats(qos_classes.size());
     std::mutex mutex;
     int rc = runner.runInitiatorTasks([&](int thread_id) -> int {
         runner.pinThread(thread_id);
@@ -49,11 +54,27 @@ int processBatchSizes(BenchRunner& runner, size_t block_size, size_t batch_size,
             local_gpu_offset + thread_id, max_block_size, max_batch_size);
         uint64_t target_addr = runner.getTargetBufferBase(
             target_gpu_offset + thread_id, max_block_size, max_batch_size);
+        const bool qos_enabled = !qos_classes.empty();
+        const size_t qos_class =
+            qos_enabled ? qosClassForThread(qos_classes, thread_id) : 0;
+        const bool tight = !qos_enabled && XferBenchConfig::deadline_us > 0 &&
+                           thread_id < XferBenchConfig::deadline_tight_threads;
+        const uint64_t relative_deadline_us =
+            qos_enabled ? qos_classes[qos_class].slo_us
+                        : (tight ? XferBenchConfig::deadline_us : 0);
+        auto deadlineNs = [&]() -> uint64_t {
+            if (relative_deadline_us == 0) return 0;
+            const auto now =
+                std::chrono::steady_clock::now().time_since_epoch();
+            return std::chrono::duration_cast<std::chrono::nanoseconds>(now)
+                       .count() +
+                   relative_deadline_us * 1000ull;
+        };
 
         XferBenchTimer timer;
         while (timer.lap_us(false) < 1000000ull) {
             runner.runSingleTransfer(local_addr, target_addr, block_size,
-                                     batch_size, opcode);
+                                     batch_size, opcode, deadlineNs());
         }
         timer.reset();
         std::vector<double> transfer_duration;
@@ -64,12 +85,14 @@ int processBatchSizes(BenchRunner& runner, size_t block_size, size_t batch_size,
                 if (XferBenchConfig::check_consistency)
                     pattern =
                         fillData((void*)local_addr, block_size * batch_size);
-                auto val = runner.runSingleTransfer(
-                    local_addr, target_addr, block_size, batch_size, WRITE);
+                auto val = runner.runSingleTransfer(local_addr, target_addr,
+                                                    block_size, batch_size,
+                                                    WRITE, deadlineNs());
                 transfer_duration.push_back(val);
                 fillData((void*)local_addr, block_size * batch_size);
                 val = runner.runSingleTransfer(local_addr, target_addr,
-                                               block_size, batch_size, READ);
+                                               block_size, batch_size, READ,
+                                               deadlineNs());
                 if (XferBenchConfig::check_consistency)
                     verifyData((void*)local_addr, block_size * batch_size,
                                pattern);
@@ -78,21 +101,51 @@ int processBatchSizes(BenchRunner& runner, size_t block_size, size_t batch_size,
         } else {
             while (timer.lap_us(false) <
                    XferBenchConfig::duration * 1000000ull) {
-                auto val = runner.runSingleTransfer(
-                    local_addr, target_addr, block_size, batch_size, opcode);
+                auto val = runner.runSingleTransfer(local_addr, target_addr,
+                                                    block_size, batch_size,
+                                                    opcode, deadlineNs());
                 transfer_duration.push_back(val);
             }
         }
         auto total_duration = timer.lap_us();
-        mutex.lock();
+        std::lock_guard<std::mutex> lock(mutex);
         stats.total_duration.add(total_duration);
         for (auto val : transfer_duration) stats.transfer_duration.add(val);
-        mutex.unlock();
+        auto& group_stats = tight ? tight_stats : loose_stats;
+        group_stats.total_duration.add(total_duration);
+        for (auto val : transfer_duration)
+            group_stats.transfer_duration.add(val);
+        if (qos_enabled) {
+            qos_stats[qos_class].total_duration.add(total_duration);
+            for (auto val : transfer_duration)
+                qos_stats[qos_class].transfer_duration.add(val);
+        }
         return 0;
     });
 
     if (rc != 0) return -1;
     printStats(block_size, batch_size, stats, num_threads);
+    if (!qos_classes.empty()) {
+        auto report = calculateQosMetrics(
+            block_size, batch_size, num_threads, qos_classes, &qos_stats,
+            XferBenchConfig::qos_link_capacity_gbps);
+        printQosMetrics(report);
+        if (!XferBenchConfig::qos_output_jsonl.empty()) {
+            std::string error;
+            if (!appendQosMetricsJsonl(XferBenchConfig::qos_output_jsonl,
+                                       report, &error)) {
+                LOG(ERROR) << error;
+                return -1;
+            }
+        }
+    } else if (XferBenchConfig::deadline_us > 0) {
+        const int tight_threads =
+            std::min(num_threads, XferBenchConfig::deadline_tight_threads);
+        printDeadlineGroupStats("tight", block_size, batch_size, tight_stats,
+                                tight_threads, XferBenchConfig::deadline_us);
+        printDeadlineGroupStats("loose", block_size, batch_size, loose_stats,
+                                num_threads - tight_threads, 0);
+    }
     return 0;
 }
 
@@ -102,6 +155,48 @@ int main(int argc, char* argv[]) {
         "Usage: ./tebench [options]");
     gflags::ParseCommandLineFlags(&argc, &argv, true);
     XferBenchConfig::loadFromFlags();
+    std::vector<QosClassConfig> qos_classes;
+    if (!XferBenchConfig::qos_classes.empty()) {
+        std::string error;
+        if (!parseQosClasses(XferBenchConfig::qos_classes, &qos_classes,
+                             &error)) {
+            LOG(ERROR) << "Invalid --qos_classes: " << error;
+            return EXIT_FAILURE;
+        }
+        if (XferBenchConfig::start_num_threads !=
+            XferBenchConfig::max_num_threads) {
+            LOG(ERROR) << "--qos_classes requires start_num_threads == "
+                          "max_num_threads";
+            return EXIT_FAILURE;
+        }
+        if (!validateQosClasses(qos_classes, XferBenchConfig::start_num_threads,
+                                &error)) {
+            LOG(ERROR) << "Invalid --qos_classes: " << error;
+            return EXIT_FAILURE;
+        }
+        if (XferBenchConfig::deadline_us != 0 ||
+            XferBenchConfig::deadline_tight_threads != 0) {
+            LOG(ERROR) << "--qos_classes cannot be combined with "
+                          "--deadline_us or --deadline_tight_threads";
+            return EXIT_FAILURE;
+        }
+    }
+    if (XferBenchConfig::qos_link_capacity_gbps < 0.0 ||
+        !std::isfinite(XferBenchConfig::qos_link_capacity_gbps)) {
+        LOG(ERROR) << "qos_link_capacity_gbps must be finite and non-negative";
+        return EXIT_FAILURE;
+    }
+    if (XferBenchConfig::deadline_tight_threads < 0 ||
+        XferBenchConfig::deadline_tight_threads >
+            XferBenchConfig::max_num_threads) {
+        LOG(ERROR) << "deadline_tight_threads must be in [0, max_num_threads]";
+        return EXIT_FAILURE;
+    }
+    if (XferBenchConfig::deadline_us > 0 &&
+        XferBenchConfig::backend != "tent") {
+        LOG(ERROR) << "deadline tagging is supported only by the tent backend";
+        return EXIT_FAILURE;
+    }
     std::unique_ptr<BenchRunner> runner;
     if (XferBenchConfig::backend == "classic") {
         runner = std::make_unique<TEBenchRunner>();
@@ -145,7 +240,7 @@ int main(int argc, char* argv[]) {
                               << " batch_size " << batch_size;
                 } else {
                     if (processBatchSizes(*runner, block_size, batch_size,
-                                          num_threads) != 0)
+                                          num_threads, qos_classes) != 0)
                         interrupted = true;
                 }
             }