//! Benchmark tool for the broker.
//!
//! Measures throughput and latency under load.
//!
//! Supports different routing strategies to test worker selection:
//! - best_price: Route to cheapest worker
//! - best_latency: Route to fastest worker
//! - best_availability: Route to most available worker
//! - round_robin: Distribute evenly across workers
//! - random: Random worker selection
//! - weighted_capacity: Weighted by available resources
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::Arc;
use std::thread;
use std::time::{Duration, Instant};
use crate::async_exec;
use colored::Colorize;
use super::router::RoutingStrategy;
/// Named benchmark workloads
#[derive(Debug, Clone, PartialEq)]
pub enum BenchWorkload {
/// Trivial add(2, 3) — measures pure broker/routing overhead
Add,
/// Recursive fib(28) — CPU-intensive, ~50ms per call
Fib,
}
impl BenchWorkload {
pub fn name(&self) -> &'static str {
match self {
BenchWorkload::Add => "add",
BenchWorkload::Fib => "fib",
}
}
pub fn description(&self) -> &'static str {
match self {
BenchWorkload::Add => "add(2, 3) — routing overhead baseline",
BenchWorkload::Fib => "fib(28) — CPU-intensive recursive Fibonacci",
}
}
}
impl std::str::FromStr for BenchWorkload {
type Err = String;
fn from_str(s: &str) -> Result<Self, Self::Err> {
match s.to_lowercase().as_str() {
"add" => Ok(BenchWorkload::Add),
"fib" | "fibonacci" | "cpu" => Ok(BenchWorkload::Fib),
_ => Err(format!("Unknown benchmark workload: '{}'. Available: add, fib", s)),
}
}
}
/// Benchmark configuration
#[derive(Debug, Clone)]
pub struct BenchConfig {
/// Target broker URL
pub broker_url: String,
/// Number of concurrent workers
pub concurrency: usize,
/// Total number of requests to send
pub requests: u64,
/// Request timeout in seconds
pub timeout_secs: u64,
/// User ID for requests
pub user_id: String,
/// API key for Bearer auth (read from ZAKURO_API_KEY env)
pub api_key: Option<String>,
/// CPU requirement per request
pub cpus: f64,
/// Memory requirement per request (bytes)
pub memory_bytes: u64,
/// Routing strategy to use
pub strategy: RoutingStrategy,
/// Run comparison across all strategies
pub compare_strategies: bool,
/// Named benchmark workload
pub workload: BenchWorkload,
/// Filter by worker type (e.g. "standard", "premium") — routes only to matching workers
pub worker_type: Option<String>,
}
impl Default for BenchConfig {
fn default() -> Self {
Self {
broker_url: "http://127.0.0.1:9000".to_string(),
concurrency: 10,
requests: 1000,
timeout_secs: 30,
user_id: "bench-user".to_string(),
api_key: std::env::var("ZAKURO_API_KEY").ok(),
cpus: 0.1,
memory_bytes: 1024 * 1024 * 100, // 100 MiB
strategy: RoutingStrategy::BestPrice,
compare_strategies: false,
workload: BenchWorkload::Add,
worker_type: None,
}
}
}
/// Benchmark results
#[derive(Debug, Clone)]
pub struct BenchResults {
/// Total requests sent
pub total_requests: u64,
/// Successful requests
pub successful: u64,
/// Failed requests
pub failed: u64,
/// Total duration
pub duration: Duration,
/// Requests per second
pub rps: f64,
/// Latencies in microseconds
pub latencies_us: Vec<u64>,
}
impl BenchResults {
fn new() -> Self {
Self {
total_requests: 0,
successful: 0,
failed: 0,
duration: Duration::ZERO,
rps: 0.0,
latencies_us: Vec::new(),
}
}
fn percentile(&self, p: f64) -> f64 {
if self.latencies_us.is_empty() {
return 0.0;
}
let mut sorted = self.latencies_us.clone();
sorted.sort();
let idx = ((sorted.len() as f64 * p / 100.0) as usize).min(sorted.len() - 1);
sorted[idx] as f64 / 1000.0 // Convert to ms
}
fn avg_latency_ms(&self) -> f64 {
if self.latencies_us.is_empty() {
return 0.0;
}
let sum: u64 = self.latencies_us.iter().sum();
(sum as f64 / self.latencies_us.len() as f64) / 1000.0
}
fn min_latency_ms(&self) -> f64 {
self.latencies_us.iter().min().map(|v| *v as f64 / 1000.0).unwrap_or(0.0)
}
fn max_latency_ms(&self) -> f64 {
self.latencies_us.iter().max().map(|v| *v as f64 / 1000.0).unwrap_or(0.0)
}
pub fn print_report(&self) {
println!();
println!(" {}", "═".repeat(60).cyan());
println!(" {} Benchmark Results", "◆".cyan());
println!(" {}", "═".repeat(60).cyan());
println!();
// Summary
println!(" {}", "Summary".bold());
println!(" {}", "─".repeat(40));
println!(" Total Requests: {}", self.total_requests);
println!(" Successful: {} {}",
self.successful.to_string().green(),
format!("({:.1}%)", self.successful as f64 / self.total_requests as f64 * 100.0).dimmed()
);
println!(" Failed: {} {}",
if self.failed > 0 { self.failed.to_string().red() } else { "0".to_string().green() },
format!("({:.1}%)", self.failed as f64 / self.total_requests as f64 * 100.0).dimmed()
);
println!(" Duration: {:.2}s", self.duration.as_secs_f64());
println!(" Throughput: {} req/s", format!("{:.2}", self.rps).yellow().bold());
println!();
// Latency
println!(" {}", "Latency (ms)".bold());
println!(" {}", "─".repeat(40));
println!(" Min: {:.2}", self.min_latency_ms());
println!(" Avg: {:.2}", self.avg_latency_ms());
println!(" Max: {:.2}", self.max_latency_ms());
println!(" p50: {:.2}", self.percentile(50.0));
println!(" p75: {:.2}", self.percentile(75.0));
println!(" p90: {:.2}", self.percentile(90.0));
println!(" p95: {:.2}", self.percentile(95.0));
println!(" p99: {:.2}", self.percentile(99.0));
println!();
// Histogram
self.print_histogram();
}
fn print_histogram(&self) {
if self.latencies_us.is_empty() {
return;
}
println!(" {}", "Latency Histogram".bold());
println!(" {}", "─".repeat(40));
let mut sorted = self.latencies_us.clone();
sorted.sort();
// Define buckets (in ms)
let buckets = [1.0, 5.0, 10.0, 25.0, 50.0, 100.0, 250.0, 500.0, 1000.0, f64::INFINITY];
let mut counts = vec![0u64; buckets.len()];
for lat_us in &sorted {
let lat_ms = *lat_us as f64 / 1000.0;
for (i, &bucket) in buckets.iter().enumerate() {
if lat_ms <= bucket {
counts[i] += 1;
break;
}
}
}
let max_count = *counts.iter().max().unwrap_or(&1);
let bar_width = 30;
for (i, &bucket) in buckets.iter().enumerate() {
let label = if bucket == f64::INFINITY {
">1000ms".to_string()
} else if i == 0 {
format!("≤{:.0}ms", bucket)
} else {
format!("≤{:.0}ms", bucket)
};
let count = counts[i];
let bar_len = if max_count > 0 {
(count as f64 / max_count as f64 * bar_width as f64) as usize
} else {
0
};
let bar: String = "█".repeat(bar_len);
let pct = count as f64 / self.latencies_us.len() as f64 * 100.0;
println!(" {:>8} [{:<30}] {:>5} ({:>5.1}%)",
label,
bar.cyan(),
count,
pct
);
}
println!();
}
}
/// Query the broker's /health endpoint and return (tailscale_connected, tailscale_ip, node_name).
fn check_tailscale_status(broker_url: &str) -> (bool, Option<String>, Option<String>) {
match ureq::get(&format!("{}/health", broker_url))
.timeout(Duration::from_secs(5))
.call()
{
Ok(resp) => {
if let Ok(body) = resp.into_string() {
if let Ok(v) = serde_json::from_str::<serde_json::Value>(&body) {
let connected = v["tailscale_connected"].as_bool().unwrap_or(false);
let ip = v["tailscale_ip"].as_str().map(|s| s.to_string());
let name = v["node_name"].as_str().map(|s| s.to_string());
return (connected, ip, name);
}
}
(false, None, None)
}
Err(_) => (false, None, None),
}
}
/// Run the benchmark
pub fn run_benchmark(config: BenchConfig) -> BenchResults {
println!();
println!(" {}", "╔═══════════════════════════════════════════╗".cyan());
println!(" {} {} {}", "║".cyan(), "Broker Benchmark".bold().white(), "║".cyan());
println!(" {}", "╚═══════════════════════════════════════════╝".cyan());
println!();
println!(" Target: {}", config.broker_url.cyan());
println!(" Concurrency: {}", config.concurrency);
println!(" Requests: {}", config.requests);
println!(" Workload: {} {}",
config.workload.name().yellow().bold(),
format!("({})", config.workload.description()).dimmed()
);
println!(" Strategy: {} {}",
config.strategy.as_str().yellow().bold(),
format!("({})", config.strategy.description()).dimmed()
);
if let Some(ref wt) = config.worker_type {
println!(" Worker Type: {}", wt.yellow().bold());
}
println!();
// First check if broker is reachable
print!(" Connecting to broker... ");
match ureq::get(&format!("{}/health", config.broker_url))
.timeout(Duration::from_secs(5))
.call()
{
Ok(_) => println!("{}", "OK".green()),
Err(e) => {
println!("{}", "FAILED".red());
println!(" Error: {}", e);
return BenchResults::new();
}
}
// Check Tailscale connectivity — required for cross-node billing
print!(" Checking Tailscale... ");
let (ts_connected, ts_ip, _) = check_tailscale_status(&config.broker_url);
if ts_connected {
println!("{} {}",
"connected".green(),
ts_ip.as_deref().unwrap_or("").dimmed()
);
} else {
println!("{}", "NOT CONNECTED".red());
println!();
println!(" {} Tailscale is not connected on this broker.", "✗".red());
println!(" Remote workers won't be discovered and cross-node billing");
println!(" won't be active. Wait for Tailscale before benchmarking.");
println!();
println!(" Run {} to check mesh status.", "zc info".cyan());
return BenchResults::new();
}
// Check if there are workers
print!(" Checking workers... ");
match ureq::get(&format!("{}/workers", config.broker_url))
.timeout(Duration::from_secs(5))
.call()
{
Ok(resp) => {
if let Ok(body) = resp.into_string() {
if let Ok(json) = serde_json::from_str::<serde_json::Value>(&body) {
let count = json["total"].as_u64().unwrap_or(0);
if count == 0 {
println!("{}", "NO WORKERS".yellow());
println!(" Warning: No workers connected. Benchmark may fail.");
} else {
println!("{} worker(s)", count.to_string().green());
}
}
}
}
Err(_) => println!("{}", "UNKNOWN".yellow()),
}
println!();
println!(" Starting benchmark...");
println!();
// Shared counters
let successful = Arc::new(AtomicU64::new(0));
let failed = Arc::new(AtomicU64::new(0));
let completed = Arc::new(AtomicU64::new(0));
let latencies = Arc::new(std::sync::Mutex::new(Vec::with_capacity(config.requests as usize)));
// Calculate requests per worker
let requests_per_worker = config.requests / config.concurrency as u64;
let extra_requests = config.requests % config.concurrency as u64;
let start = Instant::now();
// Spawn worker threads
let handles: Vec<_> = (0..config.concurrency)
.map(|i| {
let config = config.clone();
let successful = successful.clone();
let failed = failed.clone();
let completed = completed.clone();
let latencies = latencies.clone();
// First few workers get extra requests to handle remainder
let my_requests = if (i as u64) < extra_requests {
requests_per_worker + 1
} else {
requests_per_worker
};
async_exec::spawn_blocking(move || {
run_worker(i, my_requests, &config, &successful, &failed, &completed, &latencies);
})
})
.collect();
// Progress reporter
let total = config.requests;
let completed_for_progress = completed.clone();
let progress_handle = async_exec::spawn_blocking(move || {
let spinner = ['⠋', '⠙', '⠹', '⠸', '⠼', '⠴', '⠦', '⠧', '⠇', '⠏'];
let mut i = 0;
loop {
let done = completed_for_progress.load(Ordering::Relaxed);
let pct = done as f64 / total as f64 * 100.0;
print!("\r {} Progress: {:>6}/{} ({:>5.1}%) ",
spinner[i % spinner.len()].to_string().cyan(),
done, total, pct
);
let _ = std::io::Write::flush(&mut std::io::stdout());
if done >= total {
break;
}
thread::sleep(Duration::from_millis(100));
i += 1;
}
println!();
});
// Wait for all workers
for handle in handles {
let _ = handle.join();
}
let _ = progress_handle.join();
let duration = start.elapsed();
// Collect results
let successful_count = successful.load(Ordering::Relaxed);
let failed_count = failed.load(Ordering::Relaxed);
let latencies_vec = latencies.lock().unwrap().clone();
let rps = config.requests as f64 / duration.as_secs_f64();
BenchResults {
total_requests: config.requests,
successful: successful_count,
failed: failed_count,
duration,
rps,
latencies_us: latencies_vec,
}
}
fn run_worker(
_worker_id: usize,
requests: u64,
config: &BenchConfig,
successful: &AtomicU64,
failed: &AtomicU64,
completed: &AtomicU64,
latencies: &std::sync::Mutex<Vec<u64>>,
) {
let payload = create_test_payload(&config.workload);
let mut requirements = serde_json::json!({
"cpus": config.cpus,
"memory_bytes": config.memory_bytes,
"gpus": 0,
"estimated_duration_secs": 0.1,
"strategy": config.strategy.as_str()
});
if let Some(ref wt) = config.worker_type {
requirements["worker_type"] = serde_json::Value::String(wt.clone());
}
for _ in 0..requests {
let start = Instant::now();
let mut req = ureq::post(&format!("{}/execute", config.broker_url))
.timeout(Duration::from_secs(config.timeout_secs))
.set("Content-Type", "application/octet-stream")
.set("X-Zakuro-User", &config.user_id)
.set("X-Zakuro-Requirements", &requirements.to_string());
if let Some(ref key) = config.api_key {
req = req.set("Authorization", &format!("Bearer {}", key));
}
let result = req.send_bytes(&payload);
let elapsed_us = start.elapsed().as_micros() as u64;
match result {
Ok(_) => {
successful.fetch_add(1, Ordering::Relaxed);
if let Ok(mut lats) = latencies.lock() {
lats.push(elapsed_us);
}
}
Err(_) => {
failed.fetch_add(1, Ordering::Relaxed);
}
}
completed.fetch_add(1, Ordering::Relaxed);
}
}
fn create_test_payload(workload: &BenchWorkload) -> Vec<u8> {
match workload {
BenchWorkload::Add => {
// {"func": add, "args": (2, 3), "kwargs": {}} → returns 5
// def add(a, b): return a + b
// cloudpickle.dumps({"func": add, "args": (2, 3), "kwargs": {}})
hex_decode("800595cd010000000000007d94288c0466756e63948c17636c6f75647069636b6c652e636c6f75647069636b6c65948c0e5f6d616b655f66756e6374696f6e9493942868028c0d5f6275696c74696e5f747970659493948c08436f6465547970659485945294284b024b004b004b024b024b03430c97007c007c017a0000005300944e8594298c0161948c01629486948c083c737472696e673e948c03616464948c03616464944b04430b8000d80b0c88718935804c944300942929749452947d94288c0b5f5f7061636b6167655f5f944e8c085f5f6e616d655f5f948c085f5f6d61696e5f5f94754e4e4e7494529468028c125f66756e6374696f6e5f7365747374617465949394681b7d947d942868188c03616464948c0c5f5f7175616c6e616d655f5f948c03616464948c0f5f5f616e6e6f746174696f6e735f5f947d948c0e5f5f6b7764656661756c74735f5f944e8c0c5f5f64656661756c74735f5f944e8c0a5f5f6d6f64756c655f5f9468198c075f5f646f635f5f944e8c0b5f5f636c6f737572655f5f944e8c175f636c6f75647069636b6c655f7375626d6f64756c6573945d948c0b5f5f676c6f62616c735f5f947d947586948652308c0461726773944b024b0386948c066b7761726773947d94752e")
}
BenchWorkload::Fib => {
// {"func": fib, "args": (28,), "kwargs": {}} → returns 317811
// def fib(n): return n if n <= 1 else fib(n-1) + fib(n-2)
// cloudpickle.dumps({"func": fib, "args": (28,), "kwargs": {}})
hex_decode("8005952f020000000000007d94288c0466756e63948c17636c6f75647069636b6c652e636c6f75647069636b6c65948c0e5f6d616b655f66756e6374696f6e9493942868028c0d5f6275696c74696e5f747970659493948c08436f6465547970659485945294284b014b004b004b014b054b03434a97007c0064016b1a000072027c005300740100000000000000007c0064017a0a0000ab01000000000000740100000000000000007c0064027a0a0000ab010000000000007a0000005300944e4b014b0287948c036669629485948c016e9485948c083c737472696e673e948c0366696294680c4b0543298000d8070888418276d80f108808dc0b0e8871903189758b3a9c039841a00199459b0ad10b22d00422944300942929749452947d94288c0b5f5f7061636b6167655f5f944e8c085f5f6e616d655f5f948c085f5f6d61696e5f5f94754e4e4e7494529468028c125f66756e6374696f6e5f7365747374617465949394681b7d947d942868188c03666962948c0c5f5f7175616c6e616d655f5f948c03666962948c0f5f5f616e6e6f746174696f6e735f5f947d948c0e5f5f6b7764656661756c74735f5f944e8c0c5f5f64656661756c74735f5f944e8c0a5f5f6d6f64756c655f5f9468198c075f5f646f635f5f944e8c0b5f5f636c6f737572655f5f944e8c175f636c6f75647069636b6c655f7375626d6f64756c6573945d948c0b5f5f676c6f62616c735f5f947d94680c681b737586948652308c0461726773944b1c85948c066b7761726773947d94752e")
}
}
}
/// Decode a hex string to bytes at compile time.
fn hex_decode(s: &str) -> Vec<u8> {
(0..s.len())
.step_by(2)
.map(|i| u8::from_str_radix(&s[i..i + 2], 16).unwrap())
.collect()
}
/// Parse benchmark arguments and run
pub fn run_from_args(args: &[String]) {
let mut config = BenchConfig::default();
let mut i = 0;
while i < args.len() {
match args[i].as_str() {
"-c" | "--concurrency" => {
if i + 1 < args.len() {
config.concurrency = args[i + 1].parse().unwrap_or(10);
i += 1;
}
}
"-n" | "--requests" => {
if i + 1 < args.len() {
config.requests = args[i + 1].parse().unwrap_or(1000);
i += 1;
}
}
"-u" | "--url" => {
if i + 1 < args.len() {
config.broker_url = args[i + 1].clone();
i += 1;
}
}
"--user" => {
if i + 1 < args.len() {
config.user_id = args[i + 1].clone();
i += 1;
}
}
"-s" | "--strategy" => {
if i + 1 < args.len() {
match args[i + 1].parse::<RoutingStrategy>() {
Ok(s) => config.strategy = s,
Err(e) => {
eprintln!("Error: {}", e);
eprintln!("Available strategies: best_price, best_latency, best_availability, round_robin, random, weighted_capacity");
return;
}
}
i += 1;
}
}
"-b" | "--benchmark" | "--workload" => {
if i + 1 < args.len() {
match args[i + 1].parse::<BenchWorkload>() {
Ok(w) => config.workload = w,
Err(e) => {
eprintln!("Error: {}", e);
return;
}
}
i += 1;
}
}
"--worker-type" | "--type" => {
if i + 1 < args.len() {
config.worker_type = Some(args[i + 1].clone());
i += 1;
}
}
"--api-key" | "--key" => {
if i + 1 < args.len() {
config.api_key = Some(args[i + 1].clone());
i += 1;
}
}
"--compare" | "--compare-strategies" => {
config.compare_strategies = true;
}
"--list-strategies" => {
print_strategies();
return;
}
"-h" | "--help" => {
print_bench_help();
return;
}
_ => {
// Try to parse as URL if it looks like one
if args[i].starts_with("http") {
config.broker_url = args[i].clone();
}
}
}
i += 1;
}
if config.compare_strategies {
run_strategy_comparison(config);
} else {
let results = run_benchmark(config);
results.print_report();
}
}
/// Print available routing strategies
fn print_strategies() {
println!();
println!("{}", "Available Routing Strategies:".bold());
println!();
let strategies = [
("best_price", "Route to the cheapest worker", "Minimizes cost"),
("best_latency", "Route to the fastest responding worker", "Minimizes response time"),
("best_availability", "Route to the most available worker", "Best for high load"),
("round_robin", "Distribute evenly across all workers", "Even distribution"),
("random", "Random worker selection", "Simple load balancing"),
("weighted_capacity", "Weighted by available resources", "Capacity-aware"),
];
for (name, desc, note) in strategies {
println!(" {:20} - {} {}", name.cyan(), desc, format!("({})", note).dimmed());
}
println!();
}
/// Run benchmark across all strategies for comparison
fn run_strategy_comparison(base_config: BenchConfig) {
println!();
println!(" {}", "╔═══════════════════════════════════════════════════════════════╗".cyan());
println!(" {} {} {}", "║".cyan(), "Strategy Comparison Benchmark".bold().white(), "║".cyan());
println!(" {}", "╚═══════════════════════════════════════════════════════════════╝".cyan());
println!();
// Check Tailscale connectivity before running any strategy
print!(" Checking Tailscale... ");
let (ts_connected, ts_ip, _) = check_tailscale_status(&base_config.broker_url);
if ts_connected {
println!("{} {}",
"connected".green(),
ts_ip.as_deref().unwrap_or("").dimmed()
);
} else {
println!("{}", "NOT CONNECTED".red());
println!();
println!(" {} Tailscale is not connected on this broker.", "✗".red());
println!(" Remote workers won't be discovered and cross-node billing");
println!(" won't be active. Wait for Tailscale before benchmarking.");
println!();
println!(" Run {} to check mesh status.", "zc info".cyan());
return;
}
println!();
println!(" Running {} requests per strategy with {} concurrent workers",
base_config.requests, base_config.concurrency);
println!();
let strategies = [
RoutingStrategy::BestPrice,
RoutingStrategy::BestLatency,
RoutingStrategy::BestAvailability,
RoutingStrategy::RoundRobin,
RoutingStrategy::Random,
RoutingStrategy::WeightedCapacity,
];
let mut results: Vec<(RoutingStrategy, BenchResults)> = Vec::new();
for strategy in strategies {
let mut config = base_config.clone();
config.strategy = strategy;
println!(" {} Testing {}...", "▶".cyan(), strategy.as_str().yellow());
let result = run_benchmark_quiet(config);
results.push((strategy, result));
// Small delay between tests
thread::sleep(Duration::from_millis(500));
}
// Print comparison table
println!();
println!(" {}", "═".repeat(75).cyan());
println!(" {} Comparison Results", "◆".cyan());
println!(" {}", "═".repeat(75).cyan());
println!();
println!(" {:<20} {:>10} {:>10} {:>10} {:>10} {:>10}",
"Strategy".bold(),
"RPS".bold(),
"Avg (ms)".bold(),
"p95 (ms)".bold(),
"p99 (ms)".bold(),
"Success%".bold()
);
println!(" {}", "─".repeat(75));
// Find best values for highlighting
let max_rps = results.iter().map(|(_, r)| r.rps).fold(0.0, f64::max);
let min_latency = results.iter()
.map(|(_, r)| r.avg_latency_ms())
.filter(|l| *l > 0.0)
.fold(f64::INFINITY, f64::min);
for (strategy, result) in &results {
let success_pct = result.successful as f64 / result.total_requests as f64 * 100.0;
let avg_lat = result.avg_latency_ms();
let p95 = result.percentile(95.0);
let p99 = result.percentile(99.0);
// Highlight best values
let rps_str = if (result.rps - max_rps).abs() < 0.01 {
format!("{:.1}", result.rps).green().bold().to_string()
} else {
format!("{:.1}", result.rps)
};
let lat_str = if (avg_lat - min_latency).abs() < 0.01 && avg_lat > 0.0 {
format!("{:.1}", avg_lat).green().bold().to_string()
} else {
format!("{:.1}", avg_lat)
};
println!(" {:<20} {:>10} {:>10} {:>10} {:>10} {:>9.1}%",
strategy.as_str(),
rps_str,
lat_str,
format!("{:.1}", p95),
format!("{:.1}", p99),
success_pct
);
}
println!(" {}", "─".repeat(75));
println!();
// Summary
let best_rps = results.iter().max_by(|a, b| a.1.rps.partial_cmp(&b.1.rps).unwrap()).unwrap();
let best_latency = results.iter()
.filter(|(_, r)| r.avg_latency_ms() > 0.0)
.min_by(|a, b| a.1.avg_latency_ms().partial_cmp(&b.1.avg_latency_ms()).unwrap());
println!(" {}", "Recommendations:".bold());
println!(" Best throughput: {} ({:.1} RPS)",
best_rps.0.as_str().green().bold(),
best_rps.1.rps
);
if let Some((strat, result)) = best_latency {
println!(" Lowest latency: {} ({:.1}ms avg)",
strat.as_str().green().bold(),
result.avg_latency_ms()
);
}
println!();
}
/// Run benchmark without verbose output (for comparison mode)
fn run_benchmark_quiet(config: BenchConfig) -> BenchResults {
// Check broker health first
if ureq::get(&format!("{}/health", config.broker_url))
.timeout(Duration::from_secs(5))
.call()
.is_err()
{
return BenchResults::new();
}
let successful = Arc::new(AtomicU64::new(0));
let failed = Arc::new(AtomicU64::new(0));
let completed = Arc::new(AtomicU64::new(0));
let latencies = Arc::new(std::sync::Mutex::new(Vec::with_capacity(config.requests as usize)));
let requests_per_worker = config.requests / config.concurrency as u64;
let extra_requests = config.requests % config.concurrency as u64;
let start = Instant::now();
let handles: Vec<_> = (0..config.concurrency)
.map(|i| {
let config = config.clone();
let successful = successful.clone();
let failed = failed.clone();
let completed = completed.clone();
let latencies = latencies.clone();
let my_requests = if (i as u64) < extra_requests {
requests_per_worker + 1
} else {
requests_per_worker
};
async_exec::spawn_blocking(move || {
run_worker(i, my_requests, &config, &successful, &failed, &completed, &latencies);
})
})
.collect();
for handle in handles {
let _ = handle.join();
}
let duration = start.elapsed();
let successful_count = successful.load(Ordering::Relaxed);
let failed_count = failed.load(Ordering::Relaxed);
let latencies_vec = latencies.lock().unwrap().clone();
BenchResults {
total_requests: config.requests,
successful: successful_count,
failed: failed_count,
duration,
rps: config.requests as f64 / duration.as_secs_f64(),
latencies_us: latencies_vec,
}
}
fn print_bench_help() {
println!();
println!("{}: zc bench [OPTIONS] [URL]", "Usage".bold());
println!();
println!("Benchmark the broker's request handling performance.");
println!();
println!("{}", "Options:".bold());
println!(" -c, --concurrency <N> Number of concurrent workers (default: 10)");
println!(" -n, --requests <N> Total requests to send (default: 1000)");
println!(" -u, --url <URL> Broker URL (default: http://127.0.0.1:9000)");
println!(" --user <ID> User ID for requests (default: bench-user)");
println!(" --api-key <KEY> API key for Bearer auth (required in P2P mode)");
println!(" -b, --benchmark <NAME> Workload to run (default: add)");
println!(" -s, --strategy <NAME> Routing strategy (default: best_price)");
println!(" --compare Compare all routing strategies");
println!(" --list-strategies List available routing strategies");
println!(" -h, --help Show this help");
println!();
println!("{}", "Workloads:".bold());
println!(" add add(2, 3) — routing overhead baseline (default)");
println!(" fib fib(28) — CPU-intensive recursive Fibonacci");
println!();
println!("{}", "Routing Strategies:".bold());
println!(" best_price Route to cheapest worker");
println!(" best_latency Route to fastest worker");
println!(" best_availability Route to most available worker");
println!(" round_robin Distribute evenly across workers");
println!(" random Random worker selection");
println!(" weighted_capacity Weighted by available resources");
println!();
println!("{}", "Examples:".bold());
println!(" zc bench # Default settings (best_price)");
println!(" zc bench -s best_latency # Use fastest worker strategy");
println!(" zc bench --compare # Compare all strategies");
println!(" zc bench -c 50 -n 10000 # 50 concurrent, 10k requests");
println!(" zc bench http://broker:9000 # Custom broker URL");
println!(" zc bench -c 100 -n 5000 -s round_robin -u http://10.13.13.2:9000");
println!();
}