Oxicloud/examples/bench_pool_concurrency.rs

//! CPU pool concurrency benchmark — thumbnail decode under a CPU quota.
//!
//! Measures what `effective_parallelism()` changes for the image pools
//! (`ThumbnailService::max_concurrent_decodes`, `image_transcode_service`,
//! `di.rs` video): the number of concurrent CPU-heavy renders permitted. Those
//! pools used to size from `available_parallelism()`, which ignores the CFS
//! quota (`--cpus` / cgroup `cpu.max`), so under a container quota they permit
//! one render per *host* core onto cores the scheduler can't actually give.
//!
//! It drives the **real service path** — a `Semaphore(K)` gating
//! `spawn_blocking(ThumbnailService::bench_render_all)` — with a gallery of
//! concurrent requests, and sweeps the permit count K. Run pinned to the quota's
//! cores to reproduce the pathology:
//!   taskset -c 0,1 cargo run --release --features bench --example bench_pool_concurrency
//! Under `taskset -c 0,1`, `effective_parallelism()` = 2 (the "after"); the
//! higher K rows are what bare `available_parallelism()` would permit on a
//! many-core host under a 2-core quota (the "before").
//!
//! No Postgres needed. Tunables (env):
//!   BENCH_K_LIST (1,2,4,8,16)  BENCH_GALLERY (48)  BENCH_SECONDS (4)

use std::sync::Arc;
use std::time::{Duration, Instant};

use oxicloud::bench_support;
use oxicloud::common::runtime::effective_parallelism;
use oxicloud::infrastructure::services::thumbnail_service::ThumbnailService;
use tokio::sync::Semaphore;

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)
}

fn percentile(sorted: &[u64], p: f64) -> u64 {
    if sorted.is_empty() {
        return 0;
    }
    let idx = ((p / 100.0) * (sorted.len() as f64 - 1.0)).round() as usize;
    sorted[idx.min(sorted.len() - 1)]
}

/// One sweep cell: gallery of `producers` callers, `k` permits, `secs` window.
/// Returns (renders, renders/s, p50_ms, p99_ms).
fn bench_k(
    rt: &tokio::runtime::Runtime,
    img: Arc<Vec<u8>>,
    k: usize,
    producers: usize,
    secs: u64,
) -> (u64, f64, f64, f64) {
    rt.block_on(async move {
        let sem = Arc::new(Semaphore::new(k));
        let deadline = Instant::now() + Duration::from_secs(secs);
        let mut handles = Vec::with_capacity(producers);
        for _ in 0..producers {
            let sem = sem.clone();
            let img = img.clone();
            handles.push(tokio::spawn(async move {
                let mut count = 0u64;
                let mut lats: Vec<u64> = Vec::with_capacity(256);
                while Instant::now() < deadline {
                    // Real path: acquire a decode permit, render off-reactor.
                    let t = Instant::now();
                    let permit = sem.clone().acquire_owned().await.unwrap();
                    let img2 = img.clone();
                    let _ = tokio::task::spawn_blocking(move || {
                        ThumbnailService::bench_render_all(&img2).expect("render_all")
                    })
                    .await;
                    drop(permit);
                    lats.push(t.elapsed().as_micros() as u64);
                    count += 1;
                }
                (count, lats)
            }));
        }
        let mut total = 0u64;
        let mut all: Vec<u64> = Vec::new();
        for h in handles {
            let (c, l) = h.await.expect("join");
            total += c;
            all.extend_from_slice(&l);
        }
        all.sort_unstable();
        let rps = total as f64 / secs as f64;
        (
            total,
            rps,
            percentile(&all, 50.0) as f64 / 1000.0,
            percentile(&all, 99.0) as f64 / 1000.0,
        )
    })
}

fn main() {
    let k_list: Vec<usize> = std::env::var("BENCH_K_LIST")
        .ok()
        .map(|s| s.split(',').filter_map(|x| x.trim().parse().ok()).collect())
        .filter(|v: &Vec<usize>| !v.is_empty())
        .unwrap_or_else(|| vec![1, 2, 4, 8, 16]);
    let producers: usize = env_or("BENCH_GALLERY", 48);
    let secs: u64 = env_or("BENCH_SECONDS", 4);

    // Pick the heaviest corpus image — the decode cost the pool gates.
    let corpus = bench_support::load_or_generate();
    let case = corpus
        .iter()
        .max_by(|a, b| a.megapixels().partial_cmp(&b.megapixels()).unwrap())
        .expect("corpus non-empty");
    let img = Arc::new(case.bytes.clone());

    // The renders run on spawn_blocking; give the blocking pool plenty of room
    // so the Semaphore(K) — not the runtime — is the binding constraint.
    let rt = tokio::runtime::Builder::new_multi_thread()
        .worker_threads(2)
        .max_blocking_threads(64)
        .enable_all()
        .build()
        .expect("runtime");

    let eff = effective_parallelism();
    println!("\n############################################################");
    println!("# CPU pool concurrency — thumbnail decode under a CPU quota");
    println!(
        "# image: {} ({:.1} MP, {} KiB)   gallery: {} concurrent callers   window: {}s",
        case.name,
        case.megapixels(),
        case.bytes.len() / 1024,
        producers,
        secs
    );
    println!(
        "# available_parallelism = {}   effective_parallelism = {}  (= the 'after' permit count)",
        std::thread::available_parallelism().map(|n| n.get()).unwrap_or(0),
        eff
    );
    println!("# run under `taskset -c 0,1` to model a 2-core quota");
    println!("############################################################\n");
    println!(
        "| {:>8} | {:>9} | {:>10} | {:>9} | {:>9} |",
        "permits", "renders", "renders/s", "p50 ms", "p99 ms"
    );
    println!("|{:-<10}|{:-<11}|{:-<12}|{:-<11}|{:-<11}|", "", "", "", "", "");

    // Warm up (also triggers corpus generation / codec init).
    let _ = bench_k(&rt, img.clone(), 2, producers, 1);

    let mut base_rps: Option<f64> = None;
    for &k in &k_list {
        let (renders, rps, p50, p99) = bench_k(&rt, img.clone(), k, producers, secs);
        let tag = if k == eff { "  ← effective" } else { "" };
        let _ = base_rps.get_or_insert(rps);
        println!(
            "| {:>8} | {:>9} | {:>10.1} | {:>9.1} | {:>9.1} |{}",
            k, renders, rps, p50, p99, tag
        );
    }
    println!(
        "\nThroughput is CPU-bound (≈ flat past the core count); the signal is p99:\n\
         over-subscribing the decode permits past the *effective* cores inflates\n\
         per-request tail latency (gallery responsiveness) with no throughput gain.\n"
    );
}