diff --git a/mooncake-transfer-engine/benchmark/main.cpp b/mooncake-transfer-engine/benchmark/main.cpp index ad756e980b..082a9d6417 100644 --- a/mooncake-transfer-engine/benchmark/main.cpp +++ b/mooncake-transfer-engine/benchmark/main.cpp @@ -22,8 +22,26 @@ #include "tent_backend.h" #endif +#include +#include +#include + using namespace mooncake::tent; +namespace { + +uint64_t steadyClockNs() { + const auto now = std::chrono::steady_clock::now().time_since_epoch(); + return std::chrono::duration_cast(now).count(); +} + +double gbPerSecond(uint64_t bytes, double duration_us) { + if (duration_us <= 0.0) return 0.0; + return static_cast(bytes) / (1000.0 * duration_us); +} + +} // namespace + int processBatchSizes( BenchRunner& runner, size_t block_size, size_t batch_size, int num_threads, const std::vector& qos_classes, @@ -46,6 +64,17 @@ int processBatchSizes( XferBenchStats tight_stats; XferBenchStats loose_stats; std::mutex mutex; + std::atomic measurement_ready{0}; + std::atomic 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 transfer_duration; + std::vector 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,9 +166,14 @@ 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); } } @@ -130,6 +181,7 @@ int processBatchSizes( std::lock_guard 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); diff --git a/mooncake-transfer-engine/benchmark/utils.cpp b/mooncake-transfer-engine/benchmark/utils.cpp index cbc839d3ac..58e82d5b3f 100644 --- a/mooncake-transfer-engine/benchmark/utils.cpp +++ b/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,9 +214,14 @@ 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(block_size * batch_size) / + (avg_instant_gbps * 1000.0); + } // Tabulate print with fixed width for each string // clang-format off @@ -218,6 +229,7 @@ void printStats(size_t block_size, size_t batch_size, XferBenchStats& stats, << 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() diff --git a/mooncake-transfer-engine/benchmark/utils.h b/mooncake-transfer-engine/benchmark/utils.h index 9d6cab1e6c..5dc9833dd6 100644 --- a/mooncake-transfer-engine/benchmark/utils.h +++ b/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 {