style(bench): rustfmt the blob/pool/runtime bench examples
These example targets landed unformatted on main and fail the Rustfmt CI check (`cargo fmt --all --check`); reformat them so this PR's checks pass. No logic changes.
This commit is contained in:
@@ -47,13 +47,20 @@ use oxicloud::application::ports::blob_storage_ports::BlobStorageBackend;
|
|||||||
use oxicloud::infrastructure::services::local_blob_backend::LocalBlobBackend;
|
use oxicloud::infrastructure::services::local_blob_backend::LocalBlobBackend;
|
||||||
|
|
||||||
fn env_or<T: std::str::FromStr>(key: &str, default: T) -> T {
|
fn env_or<T: std::str::FromStr>(key: &str, default: T) -> T {
|
||||||
env::var(key).ok().and_then(|v| v.parse().ok()).unwrap_or(default)
|
env::var(key)
|
||||||
|
.ok()
|
||||||
|
.and_then(|v| v.parse().ok())
|
||||||
|
.unwrap_or(default)
|
||||||
}
|
}
|
||||||
|
|
||||||
fn env_list_usize(key: &str, default: &[usize]) -> Vec<usize> {
|
fn env_list_usize(key: &str, default: &[usize]) -> Vec<usize> {
|
||||||
env::var(key)
|
env::var(key)
|
||||||
.ok()
|
.ok()
|
||||||
.map(|s| s.split(',').filter_map(|x| x.trim().parse().ok()).collect::<Vec<_>>())
|
.map(|s| {
|
||||||
|
s.split(',')
|
||||||
|
.filter_map(|x| x.trim().parse().ok())
|
||||||
|
.collect::<Vec<_>>()
|
||||||
|
})
|
||||||
.filter(|v: &Vec<usize>| !v.is_empty())
|
.filter(|v: &Vec<usize>| !v.is_empty())
|
||||||
.unwrap_or_else(|| default.to_vec())
|
.unwrap_or_else(|| default.to_vec())
|
||||||
}
|
}
|
||||||
@@ -130,7 +137,11 @@ async fn run_once(
|
|||||||
// Coarse token-bucket: only sleep once the accumulated owed time clears a
|
// Coarse token-bucket: only sleep once the accumulated owed time clears a
|
||||||
// 2 ms floor, so the throttle models a rate-limited socket without drowning
|
// 2 ms floor, so the throttle models a rate-limited socket without drowning
|
||||||
// the measurement in sub-ms timer noise.
|
// the measurement in sub-ms timer noise.
|
||||||
let per_byte_secs = if throttle_bps > 0.0 { 1.0 / throttle_bps } else { 0.0 };
|
let per_byte_secs = if throttle_bps > 0.0 {
|
||||||
|
1.0 / throttle_bps
|
||||||
|
} else {
|
||||||
|
0.0
|
||||||
|
};
|
||||||
let mut owed = Duration::ZERO;
|
let mut owed = Duration::ZERO;
|
||||||
|
|
||||||
while let Some(item) = byte_stream.next().await {
|
while let Some(item) = byte_stream.next().await {
|
||||||
@@ -196,15 +207,18 @@ async fn main() {
|
|||||||
"# production LocalBlobBackend.read_prefetch() = {}",
|
"# production LocalBlobBackend.read_prefetch() = {}",
|
||||||
backend.read_prefetch()
|
backend.read_prefetch()
|
||||||
);
|
);
|
||||||
println!("# reps/cell: {reps} (median MB/s reported) cold-cache: {}", {
|
println!(
|
||||||
if !COLD_SUPPORTED {
|
"# reps/cell: {reps} (median MB/s reported) cold-cache: {}",
|
||||||
"unsupported (non-Linux) → warm only"
|
{
|
||||||
} else if want_cold {
|
if !COLD_SUPPORTED {
|
||||||
"yes (posix_fadvise DONTNEED, best-effort)"
|
"unsupported (non-Linux) → warm only"
|
||||||
} else {
|
} else if want_cold {
|
||||||
"disabled (BENCH_COLD=0)"
|
"yes (posix_fadvise DONTNEED, best-effort)"
|
||||||
|
} else {
|
||||||
|
"disabled (BENCH_COLD=0)"
|
||||||
|
}
|
||||||
}
|
}
|
||||||
});
|
);
|
||||||
println!("# N=1 is current production ('antes'); higher N is the candidate ('después')");
|
println!("# N=1 is current production ('antes'); higher N is the candidate ('después')");
|
||||||
println!("############################################################\n");
|
println!("############################################################\n");
|
||||||
println!(
|
println!(
|
||||||
|
|||||||
@@ -29,7 +29,10 @@ use oxicloud::infrastructure::services::thumbnail_service::ThumbnailService;
|
|||||||
use tokio::sync::Semaphore;
|
use tokio::sync::Semaphore;
|
||||||
|
|
||||||
fn env_or<T: std::str::FromStr>(key: &str, default: T) -> T {
|
fn env_or<T: std::str::FromStr>(key: &str, default: T) -> T {
|
||||||
std::env::var(key).ok().and_then(|v| v.parse().ok()).unwrap_or(default)
|
std::env::var(key)
|
||||||
|
.ok()
|
||||||
|
.and_then(|v| v.parse().ok())
|
||||||
|
.unwrap_or(default)
|
||||||
}
|
}
|
||||||
|
|
||||||
#[cfg(target_os = "linux")]
|
#[cfg(target_os = "linux")]
|
||||||
@@ -181,7 +184,9 @@ fn main() {
|
|||||||
);
|
);
|
||||||
println!(
|
println!(
|
||||||
"# available_parallelism = {} effective_parallelism = {} (= the 'after' permit count)",
|
"# available_parallelism = {} effective_parallelism = {} (= the 'after' permit count)",
|
||||||
std::thread::available_parallelism().map(|n| n.get()).unwrap_or(0),
|
std::thread::available_parallelism()
|
||||||
|
.map(|n| n.get())
|
||||||
|
.unwrap_or(0),
|
||||||
eff
|
eff
|
||||||
);
|
);
|
||||||
println!("# run under `taskset -c 0,1` to model a 2-core quota");
|
println!("# run under `taskset -c 0,1` to model a 2-core quota");
|
||||||
@@ -190,7 +195,10 @@ fn main() {
|
|||||||
"| {:>8} | {:>9} | {:>10} | {:>9} | {:>9} |",
|
"| {:>8} | {:>9} | {:>10} | {:>9} | {:>9} |",
|
||||||
"permits", "renders", "renders/s", "p50 ms", "p99 ms"
|
"permits", "renders", "renders/s", "p50 ms", "p99 ms"
|
||||||
);
|
);
|
||||||
println!("|{:-<10}|{:-<11}|{:-<12}|{:-<11}|{:-<11}|", "", "", "", "", "");
|
println!(
|
||||||
|
"|{:-<10}|{:-<11}|{:-<12}|{:-<11}|{:-<11}|",
|
||||||
|
"", "", "", "", ""
|
||||||
|
);
|
||||||
|
|
||||||
// Warm up (also triggers corpus generation / codec init).
|
// Warm up (also triggers corpus generation / codec init).
|
||||||
let _ = bench_k(&rt, img.clone(), 2, producers, 1);
|
let _ = bench_k(&rt, img.clone(), 2, producers, 1);
|
||||||
@@ -206,7 +214,10 @@ fn main() {
|
|||||||
|
|
||||||
// ── Part B: peak RSS for K concurrent decodes (the real over-permit cost) ──
|
// ── Part B: peak RSS for K concurrent decodes (the real over-permit cost) ──
|
||||||
println!("\n[B] Peak RSS with K concurrent decodes (one wave)\n");
|
println!("\n[B] Peak RSS with K concurrent decodes (one wave)\n");
|
||||||
println!("| {:>8} | {:>14} | {:>12} |", "permits", "peak RSS MiB", "vs effective");
|
println!(
|
||||||
|
"| {:>8} | {:>14} | {:>12} |",
|
||||||
|
"permits", "peak RSS MiB", "vs effective"
|
||||||
|
);
|
||||||
println!("|{:-<10}|{:-<16}|{:-<14}|", "", "", "");
|
println!("|{:-<10}|{:-<16}|{:-<14}|", "", "", "");
|
||||||
let mut eff_rss: Option<u64> = None;
|
let mut eff_rss: Option<u64> = None;
|
||||||
for &k in &k_list {
|
for &k in &k_list {
|
||||||
|
|||||||
@@ -35,7 +35,10 @@ use std::time::{Duration, Instant};
|
|||||||
use oxicloud::common::runtime::{cgroup_cpu_quota, effective_parallelism, runtime_pool_sizes};
|
use oxicloud::common::runtime::{cgroup_cpu_quota, effective_parallelism, runtime_pool_sizes};
|
||||||
|
|
||||||
fn env_or<T: std::str::FromStr>(key: &str, default: T) -> T {
|
fn env_or<T: std::str::FromStr>(key: &str, default: T) -> T {
|
||||||
env::var(key).ok().and_then(|v| v.parse().ok()).unwrap_or(default)
|
env::var(key)
|
||||||
|
.ok()
|
||||||
|
.and_then(|v| v.parse().ok())
|
||||||
|
.unwrap_or(default)
|
||||||
}
|
}
|
||||||
|
|
||||||
#[cfg(target_os = "linux")]
|
#[cfg(target_os = "linux")]
|
||||||
@@ -204,9 +207,7 @@ fn main() {
|
|||||||
|
|
||||||
// ── Part A ──────────────────────────────────────────────────────────────
|
// ── Part A ──────────────────────────────────────────────────────────────
|
||||||
println!("\n[A] Worker over-subscription under CPU contention");
|
println!("\n[A] Worker over-subscription under CPU contention");
|
||||||
println!(
|
println!(" workload: {concurrency} concurrent requests, {burn_kb} KiB BLAKE3 each, {secs}s");
|
||||||
" workload: {concurrency} concurrent requests, {burn_kb} KiB BLAKE3 each, {secs}s"
|
|
||||||
);
|
|
||||||
println!(" (run under `taskset -c 0,1` to model a 2-core quota)\n");
|
println!(" (run under `taskset -c 0,1` to model a 2-core quota)\n");
|
||||||
println!(
|
println!(
|
||||||
"| {:<26} | {:>8} | {:>10} | {:>8} | {:>8} |",
|
"| {:<26} | {:>8} | {:>10} | {:>8} | {:>8} |",
|
||||||
@@ -227,7 +228,13 @@ fn main() {
|
|||||||
before.2,
|
before.2,
|
||||||
before.3
|
before.3
|
||||||
);
|
);
|
||||||
let after = bench_workers(workers_after, def_max_blocking, concurrency, secs, burn_bytes);
|
let after = bench_workers(
|
||||||
|
workers_after,
|
||||||
|
def_max_blocking,
|
||||||
|
concurrency,
|
||||||
|
secs,
|
||||||
|
burn_bytes,
|
||||||
|
);
|
||||||
println!(
|
println!(
|
||||||
"| {:<26} | {:>8.0} | {:>10} | {:>8} | {:>8} |",
|
"| {:<26} | {:>8.0} | {:>10} | {:>8} | {:>8} |",
|
||||||
format!("after: {workers_after} workers"),
|
format!("after: {workers_after} workers"),
|
||||||
@@ -259,7 +266,8 @@ fn main() {
|
|||||||
"| {:<26} | {:>12} | {:>12} |",
|
"| {:<26} | {:>12} | {:>12} |",
|
||||||
"before: 512 (tokio default)", peak_def, "—"
|
"before: 512 (tokio default)", peak_def, "—"
|
||||||
);
|
);
|
||||||
let (peak_cap, _base_cap) = bench_blocking_rss(max_blocking_after, blocking_tasks, alloc_mb, hold_ms);
|
let (peak_cap, _base_cap) =
|
||||||
|
bench_blocking_rss(max_blocking_after, blocking_tasks, alloc_mb, hold_ms);
|
||||||
let saved = peak_def as i64 - peak_cap as i64;
|
let saved = peak_def as i64 - peak_cap as i64;
|
||||||
println!(
|
println!(
|
||||||
"| {:<26} | {:>12} | {:>12} |",
|
"| {:<26} | {:>12} | {:>12} |",
|
||||||
|
|||||||
Reference in New Issue
Block a user