diff --git a/docs/source/design/tent/tebench.md b/docs/source/design/tent/tebench.md index c25745b5bd..4e97ad4369 100644 --- a/docs/source/design/tent/tebench.md +++ b/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= \ + --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. diff --git a/mooncake-transfer-engine/benchmark/CMakeLists.txt b/mooncake-transfer-engine/benchmark/CMakeLists.txt index b17a8a803d..090051d9db 100644 --- a/mooncake-transfer-engine/benchmark/CMakeLists.txt +++ b/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() diff --git a/mooncake-transfer-engine/benchmark/bench_runner.h b/mooncake-transfer-engine/benchmark/bench_runner.h index f0bd69e4e2..cd0dc9958f 100644 --- a/mooncake-transfer-engine/benchmark/bench_runner.h +++ b/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 \ No newline at end of file +#endif // BENCH_RUNNER_H diff --git a/mooncake-transfer-engine/benchmark/main.cpp b/mooncake-transfer-engine/benchmark/main.cpp index 6e9d76cb0e..368df08c83 100644 --- a/mooncake-transfer-engine/benchmark/main.cpp +++ b/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& 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 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(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 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 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 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 runner; if (XferBenchConfig::backend == "classic") { runner = std::make_unique(); @@ -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; } } diff --git a/mooncake-transfer-engine/benchmark/qos_metrics.cpp b/mooncake-transfer-engine/benchmark/qos_metrics.cpp new file mode 100644 index 0000000000..a1daf9b576 --- /dev/null +++ b/mooncake-transfer-engine/benchmark/qos_metrics.cpp @@ -0,0 +1,297 @@ +// Copyright 2026 KVCache.AI +// +// 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 "qos_metrics.h" + +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include "tent/thirdparty/nlohmann/json.h" + +namespace mooncake { +namespace tent { +namespace { + +std::vector split(const std::string& value, char delimiter) { + std::vector parts; + std::stringstream stream(value); + std::string part; + while (std::getline(stream, part, delimiter)) parts.push_back(part); + return parts; +} + +template +bool parseNumber(const std::string& value, T* result) { + if (value.empty()) return false; + std::istringstream stream(value); + stream >> std::noskipws >> *result; + return stream.eof() && !stream.fail(); +} + +double jainIndex(const std::vector& values) { + if (values.empty()) return 0.0; + double sum = 0.0; + double squared_sum = 0.0; + for (double value : values) { + sum += value; + squared_sum += value * value; + } + if (squared_sum == 0.0) return 0.0; + return sum * sum / (values.size() * squared_sum); +} + +nlohmann::json optionalJson(const std::optional& value) { + return value ? nlohmann::json(*value) : nlohmann::json(nullptr); +} + +} // namespace + +bool parseQosClasses(const std::string& spec, + std::vector* classes, std::string* error) { + classes->clear(); + if (spec.empty()) { + *error = "qos_classes must not be empty"; + return false; + } + + std::set names; + for (const auto& entry : split(spec, ',')) { + const auto fields = split(entry, ':'); + if (fields.size() != 4 && fields.size() != 5) { + *error = + "each qos class must be " + "name:threads:slo_us:weight[:isolated_gbps]"; + return false; + } + + QosClassConfig config; + config.name = fields[0]; + if (config.name.empty() || !names.insert(config.name).second) { + *error = "qos class names must be non-empty and unique"; + return false; + } + if (!parseNumber(fields[1], &config.threads) || config.threads <= 0) { + *error = "qos class threads must be a positive integer"; + return false; + } + if (fields[2].empty() || fields[2].front() == '-' || + !parseNumber(fields[2], &config.slo_us)) { + *error = "qos class slo_us must be a non-negative integer"; + return false; + } + if (!parseNumber(fields[3], &config.weight) || config.weight <= 0.0 || + !std::isfinite(config.weight)) { + *error = "qos class weight must be finite and positive"; + return false; + } + if (fields.size() == 5) { + double isolated_gbps = 0.0; + if (!parseNumber(fields[4], &isolated_gbps) || + isolated_gbps <= 0.0 || !std::isfinite(isolated_gbps)) { + *error = "isolated_gbps must be finite and positive"; + return false; + } + config.isolated_throughput_gbps = isolated_gbps; + } + classes->push_back(std::move(config)); + } + return true; +} + +bool validateQosClasses(const std::vector& classes, + int num_threads, std::string* error) { + int configured_threads = 0; + for (const auto& config : classes) configured_threads += config.threads; + if (configured_threads != num_threads) { + std::ostringstream stream; + stream << "qos_classes configures " << configured_threads + << " threads, but tebench runs " << num_threads; + *error = stream.str(); + return false; + } + return true; +} + +size_t qosClassForThread(const std::vector& classes, + int thread_id) { + int boundary = 0; + for (size_t i = 0; i < classes.size(); ++i) { + boundary += classes[i].threads; + if (thread_id < boundary) return i; + } + return classes.size(); +} + +QosMetricsReport calculateQosMetrics(size_t block_size, size_t batch_size, + int num_threads, + const std::vector& classes, + std::vector* stats, + double link_capacity_gbps) { + QosMetricsReport report; + report.block_size = block_size; + report.batch_size = batch_size; + report.num_threads = num_threads; + + std::vector normalized_throughput; + std::optional max_leakage; + for (size_t i = 0; i < classes.size(); ++i) { + const auto& config = classes[i]; + auto& class_stats = (*stats)[i]; + QosClassMetrics metrics; + metrics.name = config.name; + metrics.threads = config.threads; + metrics.slo_us = config.slo_us; + metrics.weight = config.weight; + metrics.isolated_throughput_gbps = config.isolated_throughput_gbps; + metrics.operations = class_stats.transfer_duration.count(); + metrics.p99_us = class_stats.transfer_duration.p99(); + + const double duration_s = class_stats.total_duration.avg() / 1e6; + const double bytes = + static_cast(block_size) * batch_size * metrics.operations; + if (duration_s > 0.0) + metrics.throughput_gbps = bytes / 1e9 / duration_s; + + double attainment = 1.0; + if (config.slo_us != 0) { + attainment = class_stats.transfer_duration.fractionAtOrBelow( + static_cast(config.slo_us)); + metrics.slo_attainment = attainment; + } + metrics.goodput_gbps = metrics.throughput_gbps * attainment; + metrics.weighted_goodput_gbps = metrics.goodput_gbps * config.weight; + normalized_throughput.push_back(metrics.throughput_gbps / + config.weight); + + if (config.isolated_throughput_gbps) { + metrics.isolation_leakage = + std::max(0.0, 1.0 - metrics.throughput_gbps / + *config.isolated_throughput_gbps); + max_leakage = + max_leakage ? std::max(*max_leakage, *metrics.isolation_leakage) + : metrics.isolation_leakage; + } + + report.aggregate_throughput_gbps += metrics.throughput_gbps; + report.weighted_goodput_gbps += metrics.weighted_goodput_gbps; + report.classes.push_back(std::move(metrics)); + } + + report.jain_fairness = jainIndex(normalized_throughput); + report.max_isolation_leakage = max_leakage; + if (link_capacity_gbps > 0.0) { + report.link_capacity_gbps = link_capacity_gbps; + report.total_utilization = + report.aggregate_throughput_gbps / link_capacity_gbps; + } + return report; +} + +void printQosMetrics(const QosMetricsReport& report) { + std::cout << " [qos-summary] throughput=" << std::fixed + << std::setprecision(6) << report.aggregate_throughput_gbps + << " GB/s weighted_goodput=" << report.weighted_goodput_gbps + << " GB/s jain_fairness=" << report.jain_fairness + << " max_isolation_leakage="; + if (report.max_isolation_leakage) { + std::cout << *report.max_isolation_leakage; + } else { + std::cout << "N/A"; + } + std::cout << " total_utilization="; + if (report.total_utilization) { + std::cout << *report.total_utilization; + } else { + std::cout << "N/A"; + } + std::cout << std::endl; + + for (const auto& metrics : report.classes) { + std::cout << " [qos-class] name=" << metrics.name + << " threads=" << metrics.threads + << " operations=" << metrics.operations + << " throughput=" << metrics.throughput_gbps + << " GB/s p99_us=" << std::setprecision(1) << metrics.p99_us + << " slo_attainment="; + std::cout << std::setprecision(6); + if (metrics.slo_attainment) { + std::cout << std::setprecision(6) << *metrics.slo_attainment; + } else { + std::cout << "N/A"; + } + std::cout << " isolation_leakage="; + if (metrics.isolation_leakage) { + std::cout << *metrics.isolation_leakage; + } else { + std::cout << "N/A"; + } + std::cout << std::endl; + } +} + +bool appendQosMetricsJsonl(const std::string& path, + const QosMetricsReport& report, std::string* error) { + nlohmann::json root = { + {"schema_version", 1}, + {"block_size", report.block_size}, + {"batch_size", report.batch_size}, + {"num_threads", report.num_threads}, + {"aggregate_throughput_gbps", report.aggregate_throughput_gbps}, + {"weighted_goodput_gbps", report.weighted_goodput_gbps}, + {"jain_fairness", report.jain_fairness}, + {"max_isolation_leakage", optionalJson(report.max_isolation_leakage)}, + {"link_capacity_gbps", optionalJson(report.link_capacity_gbps)}, + {"total_utilization", optionalJson(report.total_utilization)}, + {"classes", nlohmann::json::array()}, + }; + for (const auto& metrics : report.classes) { + root["classes"].push_back({ + {"name", metrics.name}, + {"threads", metrics.threads}, + {"slo_us", metrics.slo_us}, + {"weight", metrics.weight}, + {"operations", metrics.operations}, + {"throughput_gbps", metrics.throughput_gbps}, + {"p99_us", metrics.p99_us}, + {"slo_attainment", optionalJson(metrics.slo_attainment)}, + {"goodput_gbps", metrics.goodput_gbps}, + {"weighted_goodput_gbps", metrics.weighted_goodput_gbps}, + {"isolated_throughput_gbps", + optionalJson(metrics.isolated_throughput_gbps)}, + {"isolation_leakage", optionalJson(metrics.isolation_leakage)}, + }); + } + + std::ofstream output(path, std::ios::app); + if (!output) { + *error = "failed to open QoS JSONL output: " + path; + return false; + } + output << root.dump() << '\n'; + if (!output) { + *error = "failed to write QoS JSONL output: " + path; + return false; + } + return true; +} + +} // namespace tent +} // namespace mooncake diff --git a/mooncake-transfer-engine/benchmark/qos_metrics.h b/mooncake-transfer-engine/benchmark/qos_metrics.h new file mode 100644 index 0000000000..f052e15e49 --- /dev/null +++ b/mooncake-transfer-engine/benchmark/qos_metrics.h @@ -0,0 +1,92 @@ +// Copyright 2026 KVCache.AI +// +// 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. + +#ifndef TEBENCH_QOS_METRICS_H +#define TEBENCH_QOS_METRICS_H + +#include +#include +#include +#include +#include + +#include "utils.h" + +namespace mooncake { +namespace tent { + +// One entry in --qos_classes: +// name:threads:slo_us:weight[:isolated_gbps] +// isolated_gbps is an optional result from a matching isolated run. It is +// deliberately an input: isolation loss cannot be inferred from a mixed run. +struct QosClassConfig { + std::string name; + int threads = 0; + uint64_t slo_us = 0; + double weight = 1.0; + std::optional isolated_throughput_gbps; +}; + +struct QosClassMetrics { + std::string name; + int threads = 0; + uint64_t slo_us = 0; + double weight = 1.0; + size_t operations = 0; + double throughput_gbps = 0.0; + double p99_us = 0.0; + std::optional slo_attainment; + double goodput_gbps = 0.0; + double weighted_goodput_gbps = 0.0; + std::optional isolated_throughput_gbps; + std::optional isolation_leakage; +}; + +struct QosMetricsReport { + size_t block_size = 0; + size_t batch_size = 0; + int num_threads = 0; + double aggregate_throughput_gbps = 0.0; + double weighted_goodput_gbps = 0.0; + double jain_fairness = 0.0; + std::optional max_isolation_leakage; + std::optional link_capacity_gbps; + std::optional total_utilization; + std::vector classes; +}; + +bool parseQosClasses(const std::string& spec, + std::vector* classes, std::string* error); + +bool validateQosClasses(const std::vector& classes, + int num_threads, std::string* error); + +size_t qosClassForThread(const std::vector& classes, + int thread_id); + +QosMetricsReport calculateQosMetrics(size_t block_size, size_t batch_size, + int num_threads, + const std::vector& classes, + std::vector* stats, + double link_capacity_gbps); + +void printQosMetrics(const QosMetricsReport& report); + +bool appendQosMetricsJsonl(const std::string& path, + const QosMetricsReport& report, std::string* error); + +} // namespace tent +} // namespace mooncake + +#endif // TEBENCH_QOS_METRICS_H diff --git a/mooncake-transfer-engine/benchmark/te_backend.cpp b/mooncake-transfer-engine/benchmark/te_backend.cpp index 7b742e94ac..11165efeff 100644 --- a/mooncake-transfer-engine/benchmark/te_backend.cpp +++ b/mooncake-transfer-engine/benchmark/te_backend.cpp @@ -349,7 +349,9 @@ int TEBenchRunner::runInitiatorTasks( double TEBenchRunner::runSingleTransfer(uint64_t local_addr, uint64_t target_addr, uint64_t block_size, - uint64_t batch_size, OpCode opcode) { + uint64_t batch_size, OpCode opcode, + uint64_t deadline_ns) { + (void)deadline_ns; auto batch_id = engine_->allocateBatchID(batch_size); std::vector requests; for (uint64_t i = 0; i < batch_size; ++i) { diff --git a/mooncake-transfer-engine/benchmark/te_backend.h b/mooncake-transfer-engine/benchmark/te_backend.h index 7d48a1d96d..ea2a5d3eab 100644 --- a/mooncake-transfer-engine/benchmark/te_backend.h +++ b/mooncake-transfer-engine/benchmark/te_backend.h @@ -69,7 +69,7 @@ class TEBenchRunner : public BenchRunner { double runSingleTransfer(uint64_t local_addr, uint64_t target_addr, uint64_t block_size, uint64_t batch_size, - OpCode opcode); + OpCode opcode, uint64_t deadline_ns); private: int allocateBuffers(); diff --git a/mooncake-transfer-engine/benchmark/tent_backend.cpp b/mooncake-transfer-engine/benchmark/tent_backend.cpp index 3aa24db0d4..e91a3b13ce 100644 --- a/mooncake-transfer-engine/benchmark/tent_backend.cpp +++ b/mooncake-transfer-engine/benchmark/tent_backend.cpp @@ -51,6 +51,8 @@ std::shared_ptr loadConfig() { config->set("metadata_type", XferBenchConfig::metadata_type); config->set("metadata_servers", XferBenchConfig::metadata_url_list); config->set("rpc_server_port", XferBenchConfig::rpc_server_port); + config->set("transports/rdma/deadline_bw_arbitration", + XferBenchConfig::deadline_bw_arbitration); // Configure transport types based on xport_type parameter if (!XferBenchConfig::xport_type.empty()) { @@ -337,7 +339,8 @@ int TENTBenchRunner::runInitiatorTasks( double TENTBenchRunner::runSingleTransfer(uint64_t local_addr, uint64_t target_addr, uint64_t block_size, - uint64_t batch_size, OpCode opcode) { + uint64_t batch_size, OpCode opcode, + uint64_t deadline_ns) { auto batch_id = engine_->allocateBatch(batch_size); std::vector requests; for (uint64_t i = 0; i < batch_size; ++i) { @@ -348,6 +351,7 @@ double TENTBenchRunner::runSingleTransfer(uint64_t local_addr, entry.target_id = handle_; entry.target_offset = target_addr + block_size * i; entry.transport_hint = transport_hint_; + entry.deadline_ns = deadline_ns; requests.emplace_back(entry); } XferBenchTimer timer; diff --git a/mooncake-transfer-engine/benchmark/tent_backend.h b/mooncake-transfer-engine/benchmark/tent_backend.h index 1648ad9269..e82d576c9d 100644 --- a/mooncake-transfer-engine/benchmark/tent_backend.h +++ b/mooncake-transfer-engine/benchmark/tent_backend.h @@ -72,7 +72,7 @@ class TENTBenchRunner : public BenchRunner { double runSingleTransfer(uint64_t local_addr, uint64_t target_addr, uint64_t block_size, uint64_t batch_size, - OpCode opcode); + OpCode opcode, uint64_t deadline_ns); private: int allocateBuffers(); @@ -99,4 +99,4 @@ class TENTBenchRunner : public BenchRunner { } // namespace tent } // namespace mooncake -#endif // TEV1_BACKEND_H \ No newline at end of file +#endif // TEV1_BACKEND_H diff --git a/mooncake-transfer-engine/benchmark/tests/qos_metrics_test.cpp b/mooncake-transfer-engine/benchmark/tests/qos_metrics_test.cpp new file mode 100644 index 0000000000..9f57f20ff7 --- /dev/null +++ b/mooncake-transfer-engine/benchmark/tests/qos_metrics_test.cpp @@ -0,0 +1,130 @@ +// Copyright 2026 KVCache.AI +// +// 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 "qos_metrics.h" + +#include +#include +#include +#include + +#include + +#include "tent/thirdparty/nlohmann/json.h" + +namespace mooncake { +namespace tent { +namespace { + +TEST(QosMetricsTest, ParsesClassContract) { + std::vector classes; + std::string error; + ASSERT_TRUE(parseQosClasses("fg:4:1000:2:12.5,bg:12:0:1", &classes, &error)) + << error; + ASSERT_EQ(classes.size(), 2); + EXPECT_EQ(classes[0].name, "fg"); + EXPECT_EQ(classes[0].threads, 4); + EXPECT_EQ(classes[0].slo_us, 1000); + EXPECT_DOUBLE_EQ(classes[0].weight, 2.0); + ASSERT_TRUE(classes[0].isolated_throughput_gbps); + EXPECT_DOUBLE_EQ(*classes[0].isolated_throughput_gbps, 12.5); + EXPECT_FALSE(classes[1].isolated_throughput_gbps); + EXPECT_TRUE(validateQosClasses(classes, 16, &error)); + EXPECT_EQ(qosClassForThread(classes, 0), 0); + EXPECT_EQ(qosClassForThread(classes, 3), 0); + EXPECT_EQ(qosClassForThread(classes, 4), 1); + EXPECT_EQ(qosClassForThread(classes, 15), 1); +} + +TEST(QosMetricsTest, RejectsAmbiguousOrInvalidContracts) { + std::vector classes; + std::string error; + EXPECT_FALSE(parseQosClasses("fg:1:100:1,fg:1:0:1", &classes, &error)); + EXPECT_FALSE(parseQosClasses("fg:0:100:1", &classes, &error)); + EXPECT_FALSE(parseQosClasses("fg:1::1", &classes, &error)); + EXPECT_FALSE(parseQosClasses("fg:1:-1:1", &classes, &error)); + EXPECT_FALSE(parseQosClasses("fg:1:100:0", &classes, &error)); + EXPECT_FALSE(parseQosClasses("fg:1:100:1:0", &classes, &error)); + ASSERT_TRUE(parseQosClasses("fg:1:100:1", &classes, &error)); + EXPECT_FALSE(validateQosClasses(classes, 2, &error)); +} + +TEST(QosMetricsTest, CalculatesSloFairnessIsolationAndUtilization) { + std::vector classes = { + {"foreground", 1, 100, 2.0, 0.006}, + {"background", 1, 0, 1.0, 0.004}, + }; + std::vector stats(2); + stats[0].total_duration.add(1000.0); + stats[0].transfer_duration.add(50.0); + stats[0].transfer_duration.add(100.0); + stats[0].transfer_duration.add(150.0); + stats[1].total_duration.add(1000.0); + stats[1].transfer_duration.add(100.0); + stats[1].transfer_duration.add(100.0); + + const auto report = calculateQosMetrics(1000, 1, 2, classes, &stats, 0.01); + ASSERT_EQ(report.classes.size(), 2); + EXPECT_NEAR(report.aggregate_throughput_gbps, 0.005, 1e-12); + EXPECT_NEAR(report.weighted_goodput_gbps, 0.006, 1e-12); + EXPECT_NEAR(report.jain_fairness, 0.98, 1e-12); + ASSERT_TRUE(report.max_isolation_leakage); + EXPECT_NEAR(*report.max_isolation_leakage, 0.5, 1e-12); + ASSERT_TRUE(report.total_utilization); + EXPECT_NEAR(*report.total_utilization, 0.5, 1e-12); + ASSERT_TRUE(report.link_capacity_gbps); + EXPECT_NEAR(*report.link_capacity_gbps, 0.01, 1e-12); + + const auto& foreground = report.classes[0]; + EXPECT_EQ(foreground.operations, 3); + EXPECT_NEAR(foreground.p99_us, 149.0, 1e-12); + ASSERT_TRUE(foreground.slo_attainment); + EXPECT_NEAR(*foreground.slo_attainment, 2.0 / 3.0, 1e-12); + EXPECT_NEAR(foreground.goodput_gbps, 0.002, 1e-12); + EXPECT_NEAR(foreground.weighted_goodput_gbps, 0.004, 1e-12); + ASSERT_TRUE(foreground.isolated_throughput_gbps); + EXPECT_NEAR(*foreground.isolated_throughput_gbps, 0.006, 1e-12); + + EXPECT_FALSE(report.classes[1].slo_attainment); +} + +TEST(QosMetricsTest, UsesNullForUnavailableJsonMetrics) { + std::vector classes = { + {"best_effort", 1, 0, 1.0, std::nullopt}}; + std::vector stats(1); + stats[0].total_duration.add(1000.0); + stats[0].transfer_duration.add(100.0); + const auto report = calculateQosMetrics(1000, 1, 1, classes, &stats, 0.0); + + const std::string path = "tebench_qos_metrics_test.jsonl"; + std::remove(path.c_str()); + std::string error; + ASSERT_TRUE(appendQosMetricsJsonl(path, report, &error)) << error; + + std::ifstream input(path); + nlohmann::json record; + ASSERT_NO_THROW(input >> record); + EXPECT_EQ(record["schema_version"], 1); + EXPECT_TRUE(record["total_utilization"].is_null()); + EXPECT_TRUE(record["link_capacity_gbps"].is_null()); + EXPECT_TRUE(record["max_isolation_leakage"].is_null()); + EXPECT_TRUE(record["classes"][0]["slo_attainment"].is_null()); + EXPECT_TRUE(record["classes"][0]["isolation_leakage"].is_null()); + EXPECT_TRUE(record["classes"][0]["isolated_throughput_gbps"].is_null()); + std::remove(path.c_str()); +} + +} // namespace +} // namespace tent +} // namespace mooncake diff --git a/mooncake-transfer-engine/benchmark/utils.cpp b/mooncake-transfer-engine/benchmark/utils.cpp index d907a2bd68..89ddc46638 100644 --- a/mooncake-transfer-engine/benchmark/utils.cpp +++ b/mooncake-transfer-engine/benchmark/utils.cpp @@ -35,6 +35,22 @@ DEFINE_int32(start_num_threads, 1, "Start number of concurrent worker threads."); DEFINE_int32(max_num_threads, 1, "Maximum number of concurrent worker threads."); +DEFINE_uint64(deadline_us, 0, + "tent only: relative per-transfer deadline in microseconds for " + "tight worker threads (0 disables deadline tagging)."); +DEFINE_int32(deadline_tight_threads, 0, + "tent only: workers [0, N) that carry --deadline_us; remaining " + "workers have no deadline."); +DEFINE_bool(deadline_bw_arbitration, false, + "tent only: enable deadline-aware RDMA bandwidth arbitration."); +DEFINE_string( + qos_classes, "", + "QoS classes as name:threads:slo_us:weight[:isolated_gbps],...; " + "enables per-class QoS metrics and requires a fixed thread count."); +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_int32(local_gpu_id, 0, "Local GPU ID to be used, -1 for all GPUs"); DEFINE_int32(target_gpu_id, 0, "Target GPU ID to be used, -1 for all GPUs"); DEFINE_string(metadata_type, "p2p", @@ -70,6 +86,12 @@ size_t XferBenchConfig::max_batch_size = 0; int XferBenchConfig::duration = 0; int XferBenchConfig::max_num_threads = 0; int XferBenchConfig::start_num_threads = 0; +uint64_t XferBenchConfig::deadline_us = 0; +int XferBenchConfig::deadline_tight_threads = 0; +bool XferBenchConfig::deadline_bw_arbitration = false; +std::string XferBenchConfig::qos_classes; +double XferBenchConfig::qos_link_capacity_gbps = 0.0; +std::string XferBenchConfig::qos_output_jsonl; std::string XferBenchConfig::metadata_type; std::string XferBenchConfig::metadata_url_list; @@ -96,6 +118,12 @@ void XferBenchConfig::loadFromFlags() { max_batch_size = FLAGS_max_batch_size; start_num_threads = FLAGS_start_num_threads; max_num_threads = FLAGS_max_num_threads; + deadline_us = FLAGS_deadline_us; + deadline_tight_threads = FLAGS_deadline_tight_threads; + deadline_bw_arbitration = FLAGS_deadline_bw_arbitration; + qos_classes = FLAGS_qos_classes; + qos_link_capacity_gbps = FLAGS_qos_link_capacity_gbps; + qos_output_jsonl = FLAGS_qos_output_jsonl; duration = FLAGS_duration; metadata_type = FLAGS_metadata_type; @@ -168,5 +196,20 @@ void printStats(size_t block_size, size_t batch_size, XferBenchStats& stats, // clang-format on } +void printDeadlineGroupStats(const char* group, size_t block_size, + size_t batch_size, XferBenchStats& stats, + int num_threads, uint64_t deadline_us) { + if (num_threads <= 0 || stats.transfer_duration.count() == 0) return; + const double duration_s = stats.total_duration.avg() / 1e6; + const double bytes = static_cast(block_size) * batch_size * + stats.transfer_duration.count(); + const double throughput_gbs = bytes / 1e9 / duration_s; + std::cout << " [deadline-" << group << "] threads=" << num_threads; + if (deadline_us != 0) std::cout << " deadline_us=" << deadline_us; + std::cout << " operations=" << stats.transfer_duration.count() + << " throughput=" << std::fixed << std::setprecision(6) + << throughput_gbs << " GB/s" << std::endl; +} + } // namespace tent } // namespace mooncake diff --git a/mooncake-transfer-engine/benchmark/utils.h b/mooncake-transfer-engine/benchmark/utils.h index 6c862cd6c5..960b1dcf17 100644 --- a/mooncake-transfer-engine/benchmark/utils.h +++ b/mooncake-transfer-engine/benchmark/utils.h @@ -68,6 +68,12 @@ struct XferBenchConfig { static int duration; static int max_num_threads; static int start_num_threads; + static uint64_t deadline_us; + static int deadline_tight_threads; + static bool deadline_bw_arbitration; + static std::string qos_classes; + static double qos_link_capacity_gbps; + static std::string qos_output_jsonl; static std::string metadata_type; static std::string metadata_url_list; @@ -107,6 +113,14 @@ struct XferMetricStats { double p999() { return percentile(99.9); } + double fractionAtOrBelow(double threshold) const { + if (samples.empty()) return 0.0; + const auto count = std::count_if( + samples.begin(), samples.end(), + [threshold](double value) { return value <= threshold; }); + return static_cast(count) / samples.size(); + } + void add(double value) { samples.push_back(value); } void clear() { samples.clear(); } @@ -153,6 +167,10 @@ void printStatsHeader(); void printStats(size_t block_size, size_t batch_size, XferBenchStats& stats, int num_threads); +void printDeadlineGroupStats(const char* group, size_t block_size, + size_t batch_size, XferBenchStats& stats, + int num_threads, uint64_t deadline_us); + #if defined(USE_CUDA) || defined(USE_SUNRISE) static inline bool isCudaMemory(void* ptr) { cudaPointerAttributes attr; diff --git a/mooncake-transfer-engine/tent/include/tent/transport/rdma/bw_arbitration.h b/mooncake-transfer-engine/tent/include/tent/transport/rdma/bw_arbitration.h new file mode 100644 index 0000000000..046ff6a341 --- /dev/null +++ b/mooncake-transfer-engine/tent/include/tent/transport/rdma/bw_arbitration.h @@ -0,0 +1,83 @@ +// Copyright 2026 KVCache.AI +// +// 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. +// +// Deadline-aware NIC bandwidth arbitration WITHIN a priority tier (RFC #2792). +// +// TENT's QoS is otherwise vertical: SL/TC and priority tiers separate business +// classes. But when several flows in the SAME tier contend for one NIC, the +// NIC's bandwidth is split blindly/equally (measured: a ~388 Gb/s NIC gives +// ~97 Gb/s to each of 4 contending flows). There is no way to let a flow that +// is about to miss its deadline claim a larger share. +// +// This header isolates the pure ordering decision so it can be unit-tested +// without the RDMA stack: given the contending slices' (deadline_ns, length) +// and a bandwidth estimate, order them most-urgent-first by predicted MLU +// (predicted transfer time / remaining deadline window) — the same MLU used by +// the admission layer (#2618/#2764). Opt-in: when disabled, the original order +// is preserved exactly (byte-identical to today's equal split). + +#pragma once + +#include +#include +#include +#include +#include +#include + +namespace mooncake { +namespace tent { + +// Minimal view of a contending slice for arbitration. Kept free of RdmaSlice +// so the policy is unit-testable in isolation. +struct ArbFlow { + uint64_t deadline_ns; // 0 == no deadline + size_t length; // bytes to transfer +}; + +// Predicted Missed-Latency-per-Unit: predicted transfer time / remaining +// deadline window. Higher == more urgent (closer to / past its deadline). +// A flow with no deadline (deadline_ns == 0) is least urgent (MLU 0). A flow +// already past its deadline, or that cannot fit its window at the given +// bandwidth, gets a very high MLU so it sorts first. bw_bps <= 0 disables +// prediction (returns 0 for everyone == no reordering). +inline double PredictedMlu(const ArbFlow& f, uint64_t now_ns, double bw_bps) { + if (f.deadline_ns == 0 || bw_bps <= 0.0) return 0.0; + if (f.deadline_ns <= now_ns) return std::numeric_limits::max(); + const double window_s = (f.deadline_ns - now_ns) / 1e9; + const double predicted_time_s = static_cast(f.length) / bw_bps; + return predicted_time_s / window_s; +} + +// Return the indices of `flows` ordered most-urgent-first (highest predicted +// MLU first). Ties and the no-deadline case keep original (FIFO) order — a +// stable sort — so a symmetric set degrades to today's behavior. This does not +// drop or admit anything; it only reorders selection among already-eligible, +// same-tier flows. +inline std::vector OrderByUrgency(const std::vector& flows, + uint64_t now_ns, double bw_bps) { + std::vector idx(flows.size()); + std::iota(idx.begin(), idx.end(), size_t{0}); + std::vector mlu(flows.size()); + for (size_t i = 0; i < flows.size(); ++i) { + mlu[i] = PredictedMlu(flows[i], now_ns, bw_bps); + } + std::stable_sort(idx.begin(), idx.end(), [&](size_t a, size_t b) { + return mlu[a] > mlu[b]; // higher MLU first; stable keeps FIFO on ties + }); + return idx; +} + +} // namespace tent +} // namespace mooncake diff --git a/mooncake-transfer-engine/tent/include/tent/transport/rdma/workers.h b/mooncake-transfer-engine/tent/include/tent/transport/rdma/workers.h index c18d5de130..0fc772f49b 100644 --- a/mooncake-transfer-engine/tent/include/tent/transport/rdma/workers.h +++ b/mooncake-transfer-engine/tent/include/tent/transport/rdma/workers.h @@ -226,6 +226,9 @@ class Workers { // per-worker/per-peer RailMonitor instances. std::string rail_topo_json_; bool always_tier1_ = false; + // Opt-in deadline-aware bandwidth arbitration within a priority tier + // (RFC #2792). Default false = original FIFO order (equal bandwidth split). + bool deadline_bw_arbitration_ = false; }; } // namespace tent } // namespace mooncake diff --git a/mooncake-transfer-engine/tent/src/transport/rdma/workers.cpp b/mooncake-transfer-engine/tent/src/transport/rdma/workers.cpp index 22e95b66af..d5bf3f05c5 100644 --- a/mooncake-transfer-engine/tent/src/transport/rdma/workers.cpp +++ b/mooncake-transfer-engine/tent/src/transport/rdma/workers.cpp @@ -16,10 +16,12 @@ #include +#include #include #include #include +#include "tent/transport/rdma/bw_arbitration.h" #include "tent/transport/rdma/endpoint_store.h" #include "tent/transport/rdma/promotion_policy.h" #include "tent/transport/rdma/shared_quota.h" @@ -33,6 +35,12 @@ namespace tent { thread_local int tl_wid = -1; namespace { +struct ArbitrationEntry { + RdmaSlice* slice; + double mlu; + size_t order; +}; + // Look up (or create) the RailMonitor for `machine_id` on this worker's // map. Returning a stable reference is safe because the map stores values // via unique_ptr -- rehashes move the pointer slot, not the RailMonitor. @@ -143,6 +151,11 @@ Workers::Workers(RdmaTransport* transport) conf->get("transports/rdma/priority_promotion_timeout_us", 10000) * 1000ull; + // Opt-in deadline-aware bandwidth arbitration within a priority tier + // (RFC #2792). Default false = original FIFO order. + deadline_bw_arbitration_ = + conf->get("transports/rdma/deadline_bw_arbitration", false); + // Opt-in per-entry promotion (issue #2528). Default false = historical // head-only "flush the tier" behavior. priority_promotion_per_entry_ = @@ -392,6 +405,48 @@ void Workers::asyncPostSend() { continue; } + // RFC #2792 (opt-in): these slices all contend for one NIC path + // (local NIC -> remote NIC). submitSlices posts as many as the QP + // budget allows and returns num_submitted; the rest wait for the next + // round. Ordering by deadline urgency here means a flow about to miss + // its deadline claims the shared NIC's QP slots ahead of looser flows. + // Default (deadline_bw_arbitration_ == false) leaves order untouched, + // so behavior is byte-identical to today's FIFO / equal split. + if (deadline_bw_arbitration_ && slices.size() > 1) { + const uint64_t now_ns = getCurrentTimeInNano(); + const double bw_bps = device_selector_ + ? device_selector_->getSchedulingParams() + .default_bandwidth_gbps * + 1e9 / 8.0 + : 0.0; + if (bw_bps > 0.0) { + thread_local std::vector scratch; + scratch.clear(); + scratch.reserve(slices.size()); + + for (size_t i = 0; i < slices.size(); ++i) { + const RdmaSlice* s = slices[i]; + ArbFlow flow{0, 0}; + if (s && s->task) { + flow = ArbFlow{s->task->request.deadline_ns, s->length}; + } + scratch.push_back(ArbitrationEntry{ + slices[i], PredictedMlu(flow, now_ns, bw_bps), i}); + } + + std::sort( + scratch.begin(), scratch.end(), + [](const ArbitrationEntry& a, const ArbitrationEntry& b) { + if (a.mlu > b.mlu) return true; + if (a.mlu < b.mlu) return false; + return a.order < b.order; + }); + for (size_t i = 0; i < scratch.size(); ++i) { + slices[i] = scratch[i].slice; + } + } + } + int num_submitted = endpoint->submitSlices(slices, tl_wid); for (int id = 0; id < num_submitted; ++id) { auto slice = slices[id]; diff --git a/mooncake-transfer-engine/tent/tests/CMakeLists.txt b/mooncake-transfer-engine/tent/tests/CMakeLists.txt index d04805f1ae..818212019f 100644 --- a/mooncake-transfer-engine/tent/tests/CMakeLists.txt +++ b/mooncake-transfer-engine/tent/tests/CMakeLists.txt @@ -27,6 +27,12 @@ target_include_directories(admission_queue_test PRIVATE ${CMAKE_CURRENT_SOURCE_DIR}/../include) add_test(NAME admission_queue_test COMMAND admission_queue_test) +add_executable(bw_arbitration_test bw_arbitration_test.cpp) +target_link_libraries(bw_arbitration_test PRIVATE tent_common gtest gtest_main) +target_include_directories(bw_arbitration_test + PRIVATE ${CMAKE_CURRENT_SOURCE_DIR}/../include) +add_test(NAME bw_arbitration_test COMMAND bw_arbitration_test) + # Reproducible hot-path microbenchmark; intentionally not registered with # ctest. Run manually when changing deadline promotion partitioning. add_executable(deadline_promotion_bench deadline_promotion_bench.cpp diff --git a/mooncake-transfer-engine/tent/tests/bw_arbitration_test.cpp b/mooncake-transfer-engine/tent/tests/bw_arbitration_test.cpp new file mode 100644 index 0000000000..3664362ba0 --- /dev/null +++ b/mooncake-transfer-engine/tent/tests/bw_arbitration_test.cpp @@ -0,0 +1,111 @@ +// Copyright 2026 KVCache.AI +// +// 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. +// +// Unit tests for the deadline-aware bandwidth arbitration ordering (#2792). + +#include "tent/transport/rdma/bw_arbitration.h" + +#include + +#include +#include + +namespace mooncake { +namespace tent { +namespace { + +constexpr uint64_t kNow = 1'000'000'000; // 1s in ns +constexpr double kBw = 1e9; // 1 GB/s -> 16 B takes 16 ns + +// A flow whose window is `window_ns` from now, transferring `len` bytes. +ArbFlow flow(uint64_t window_ns, size_t len) { + return ArbFlow{kNow + window_ns, len}; +} + +TEST(BwArbitrationTest, TighterDeadlineSortsFirst) { + std::vector flows = { + flow(1'000'000, 4096), // idx0: loose (1ms window) + flow(10'000, 4096), // idx1: tight (10us window) -> most urgent + flow(100'000, 4096), // idx2: medium + }; + auto order = OrderByUrgency(flows, kNow, kBw); + ASSERT_EQ(order.size(), 3u); + EXPECT_EQ(order[0], 1u); // tightest first + EXPECT_EQ(order[1], 2u); + EXPECT_EQ(order[2], 0u); // loosest last +} + +TEST(BwArbitrationTest, NoDeadlineSortsLast) { + std::vector flows = { + ArbFlow{0, 4096}, // idx0: no deadline -> least urgent + flow(50'000, 4096), // idx1: has deadline -> first + }; + auto order = OrderByUrgency(flows, kNow, kBw); + EXPECT_EQ(order[0], 1u); + EXPECT_EQ(order[1], 0u); +} + +TEST(BwArbitrationTest, PastDeadlineSortsFirst) { + std::vector flows = { + flow(50'000, 4096), // idx0: still feasible + ArbFlow{kNow - 1, 4096}, // idx1: already past -> most urgent + }; + auto order = OrderByUrgency(flows, kNow, kBw); + EXPECT_EQ(order[0], 1u); + EXPECT_EQ(order[1], 0u); +} + +TEST(BwArbitrationTest, TiesKeepFifoOrder) { + // Identical deadlines/lengths -> stable sort preserves original order. + std::vector flows = { + flow(50'000, 4096), // idx0 + flow(50'000, 4096), // idx1 + flow(50'000, 4096), // idx2 + }; + auto order = OrderByUrgency(flows, kNow, kBw); + EXPECT_EQ(order, (std::vector{0, 1, 2})); +} + +TEST(BwArbitrationTest, AllNoDeadlineKeepsFifoOrder) { + // No flow has a deadline -> byte-identical to today's order (no reorder). + std::vector flows = {ArbFlow{0, 1}, ArbFlow{0, 2}, ArbFlow{0, 3}}; + auto order = OrderByUrgency(flows, kNow, kBw); + EXPECT_EQ(order, (std::vector{0, 1, 2})); +} + +TEST(BwArbitrationTest, ZeroBandwidthDisablesReorder) { + // bw<=0 -> prediction disabled -> original order preserved. + std::vector flows = { + flow(1'000'000, 4096), + flow(10'000, 4096), + }; + auto order = OrderByUrgency(flows, kNow, /*bw_bps=*/0.0); + EXPECT_EQ(order, (std::vector{0, 1})); +} + +TEST(BwArbitrationTest, LongerTransferIsMoreUrgentAtSameDeadline) { + // Same window, but a bigger transfer has higher predicted MLU (needs more + // of the shared bandwidth to finish in time). + std::vector flows = { + flow(50'000, 4096), // idx0: small + flow(50'000, 65536), // idx1: large -> more urgent + }; + auto order = OrderByUrgency(flows, kNow, kBw); + EXPECT_EQ(order[0], 1u); + EXPECT_EQ(order[1], 0u); +} + +} // namespace +} // namespace tent +} // namespace mooncake