nordisk/edda

crates/admit/benches/admit.rs

//! **plan + admit + record, 1 → N cores.**
//!
//! One op = what a product does once per transfer: `usage(who)` (the table's
//! 30 loads), `plan` (estimate + the sentence), `admit` (a class slot's
//! `fetch_add`), `record` (one CAS on the day bucket, the striped EWMA, and
//! the slot's `fetch_sub`). Workers are fanned out with the house engine
//! (`gatling_run`, one unit per worker so the dispatcher's counter is touched
//! once per worker, not per op).
//!
//! Three shapes:
//! * **spread** — 4 096 namespaces, each worker walking its own stride of
//!   them: the multi-tenant case;
//! * **hot** — every worker on ONE namespace: the worst case for the one
//!   shared cache line (that namespace's day bucket and the class counter);
//! * **core** — spread, but the plan (and its sentence) built once per worker
//!   and re-used: usage + admit + record only, i.e. the shared data
//!   structures without the per-transfer string formatting.
//!
//! Prints `threads  ns/op (per thread)  Mops/s (total)  scaling`, where
//! scaling is total throughput over the 1-thread run.
//!
//! `ADMIT_BENCH_OPS` overrides the ops per worker (default 200 000).

use std::sync::Arc;
use std::time::Instant;

use edda_admit::{Admit, Consumer, Cost, Settings, Slots, Work, WorkKind};

fn ops_per_worker() -> usize {
    std::env::var("ADMIT_BENCH_OPS").ok().and_then(|v| v.parse().ok()).unwrap_or(200_000)
}

#[derive(Clone, Copy, PartialEq)]
enum Shape {
    Spread,
    Hot,
    Core,
}

fn run(share: &Arc<dyn Admit>, who: &[Consumer], threads: usize, shape: Shape) -> f64 {
    let ops = ops_per_worker();
    let work = Work::new(WorkKind::Clone, Some(1_000_000));
    let cost = Cost { net_out: 1_200_000, cpu_ms: 3, ..Cost::default() };
    let start = Instant::now();
    gatling::gatling_forkjoin::gatling_run(threads, threads, |w| {
        if shape == Shape::Core {
            let plans: Vec<_> = (0..64)
                .map(|i| {
                    let c = &who[(w * 7919 + i) % who.len()];
                    share.plan(&work, c, &share.usage(c))
                })
                .collect();
            for i in 0..ops {
                let plan = &plans[i % plans.len()];
                std::hint::black_box(share.usage(&plan.who));
                if let Ok(g) = share.admit(plan) {
                    share.record(g, &cost);
                }
            }
            return;
        }
        for i in 0..ops {
            let c = if shape == Shape::Hot { &who[0] } else { &who[(w * 7919 + i) % who.len()] };
            let usage = share.usage(c);
            let plan = share.plan(&work, c, &usage);
            if let Ok(g) = share.admit(&plan) {
                share.record(g, &cost);
            }
        }
    });
    start.elapsed().as_nanos() as f64 / ops as f64 // ns per op per thread
}

fn main() {
    let cores = std::thread::available_parallelism().map(|n| n.get()).unwrap_or(1);
    let mut counts = vec![1usize];
    while *counts.last().unwrap() * 2 <= cores {
        counts.push(counts.last().unwrap() * 2);
    }
    if *counts.last().unwrap() != cores {
        counts.push(cores);
    }
    let who: Vec<Consumer> = (0..4096).map(|i| Consumer::Namespace(format!("ns-{i}"))).collect();
    for shape in [Shape::Spread, Shape::Hot, Shape::Core] {
        let share: Arc<dyn Admit> = Settings {
            limit: Some(u64::MAX / 2),
            slots: Some(Slots { full: usize::MAX / 2, anonymous: 16 }),
            ..Settings::default()
        }
        .build();
        // Warm: every consumer seen once, so first-sight inserts are not timed.
        for c in &who {
            let g = share.admit(&share.plan(&Work::new(WorkKind::Clone, None), c, &share.usage(c))).unwrap();
            share.record(g, &Cost::default());
        }
        println!(
            "edda-admit plan+admit+record, {} ({} cores, {} ops/worker)",
            match shape {
                Shape::Spread => "SPREAD: 4096 namespaces",
                Shape::Hot => "HOT: one namespace",
                Shape::Core => "CORE: usage+admit+record, plan re-used",
            },
            cores,
            ops_per_worker()
        );
        println!("{:>8} {:>14} {:>14} {:>9}", "threads", "ns/op/thread", "Mops/s total", "scaling");
        let mut base = 0.0;
        for &n in &counts {
            let ns = run(&share, &who, n, shape);
            let mops = n as f64 * 1e3 / ns;
            if n == 1 {
                base = mops;
            }
            println!("{:>8} {:>14.1} {:>14.2} {:>8.2}x", n, ns, mops, mops / base);
        }
    }
}