smarm/examples/causal_attrib_probe.rs

//! Attribution-efficiency probe (RFC 007 follow-up).
//!
//! Original hypothesis: the ~4pt impact shortfall on the 24-core
//! validation (+29.3/+83.5 vs theoretical +33/+100) is a constant
//! attribution efficiency eff ≈ 0.91 from site-exit tail truncation.
//! The guard-drop flush closed that leak, yet eff held at ~0.93 —
//! RESOLVED (2026-07-13 sweep): the residual is runnable off-CPU time
//! inside the site (~4.9 slice-expiry yields/entry x ~5.6µs runqueue
//! wait), wall time the ground truth below counts but on-CPU
//! attribution correctly skips. The offcpu audit bucket now counts it;
//! `eff+offcpu` printed per window should sit at ~1.00 — the
//! closed-books check.
//!
//! Measurement: same pipeline as `causal_pipeline`, but the `reserve`
//! actor also measures its raw in-site time directly (rdtsc at guard
//! enter/exit) and counts site entries. For each experiment window at
//! pct%:
//!
//!     eff = (Δglobal_delay / (pct/100)) / Δin_site_cycles
//!
//! and the missing time per site entry localizes the leak:
//!
//!     tail_us/entry = (Δin_site − Δglobal_delay/(pct/100)) / Δentries
//!
//! A constant eff across 25/50% with tail/entry in the tens of µs
//! localizes a per-entry mechanism; `eff+offcpu` ≈ 1.00 confirms the
//! runnable-gap account and rules out any remaining silent loss.
//!
//! Run: cargo run --release --example causal_attrib_probe --features smarm-causal

use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::sync::Arc;
use std::time::{Duration, Instant};

fn rdtsc() -> u64 {
    // x86_64 only — same clock the ledger uses.
    unsafe { core::arch::x86_64::_rdtsc() }
}

/// Same fixed-work loop as causal_pipeline (dependent LCG, preemptible).
fn work_iters(iters: u64) {
    let mut acc = 0x2545_f491_4f6c_dd1du64;
    let mut i = 0u64;
    while i < iters {
        let chunk_end = (i + 256).min(iters);
        while i < chunk_end {
            acc = acc.wrapping_mul(6364136223846793005).wrapping_add(i);
            i += 1;
        }
        std::hint::black_box(acc);
        smarm::check!();
    }
}

fn calibrate_iters_per_us() -> u64 {
    let n = 8_000_000u64;
    let t = Instant::now();
    work_iters(n);
    (n / (t.elapsed().as_micros().max(1) as u64)).max(1)
}

static IN_SITE_CYCLES: AtomicU64 = AtomicU64::new(0);
static SITE_ENTRIES: AtomicU64 = AtomicU64::new(0);

fn main() {
    let per_us = calibrate_iters_per_us();
    println!("calibration: {per_us} work iters/µs");
    let work_us = move |us: u64| work_iters(us * per_us);

    let cores = std::thread::available_parallelism().map(|n| n.get()).unwrap_or(1);
    println!("cores: {cores}");
    if cores < 4 {
        println!("probe: SKIPPED (needs the stages in parallel)");
        return;
    }

    smarm::init(smarm::Config::default()).run(move || {
        let stop = Arc::new(AtomicBool::new(false));
        let (tx_ab, rx_ab) = smarm::channel::<u64>();
        let (tx_bc, rx_bc) = smarm::channel::<u64>();

        let stop_p = stop.clone();
        let producer = smarm::spawn(move || {
            let mut i = 0u64;
            while !stop_p.load(Ordering::Relaxed) {
                {
                    let _g = smarm::causal_site!("serialize");
                    work_us(200);
                }
                if tx_ab.send(i).is_err() {
                    break;
                }
                i += 1;
            }
        });

        // Reserve: the target — instrumented with ground-truth in-site time.
        let reserve = smarm::spawn(move || {
            while let Ok(item) = rx_ab.recv() {
                {
                    let t0 = rdtsc();
                    let _g = smarm::causal_site!("reserve");
                    work_us(400);
                    // Measured before guard drop: exactly the span the
                    // ledger should be attributing.
                    IN_SITE_CYCLES.fetch_add(rdtsc().saturating_sub(t0), Ordering::Relaxed);
                    SITE_ENTRIES.fetch_add(1, Ordering::Relaxed);
                }
                if tx_bc.send(item).is_err() {
                    break;
                }
            }
        });

        let notify = smarm::spawn(move || {
            while rx_bc.recv().is_ok() {
                {
                    let _g = smarm::causal_site!("notify");
                    work_us(50);
                }
                smarm::progress!("orders-processed");
            }
        });

        let stop_bg = stop.clone();
        let background = smarm::spawn(move || {
            while !stop_bg.load(Ordering::Relaxed) {
                let _g = smarm::causal_site!("background-compaction");
                work_us(500);
            }
        });

        smarm::sleep(Duration::from_millis(300));

        let hz = smarm::causal::tsc_hz();
        println!("tsc_hz: {:.3} GHz", hz / 1e9);

        // Manual windows so ledger/ground-truth snapshots align exactly.
        for &pct in &[25u32, 50, 50, 25] {
            let g0 = smarm::causal::global_delay_cycles();
            let s0 = IN_SITE_CYCLES.load(Ordering::Relaxed);
            let e0 = SITE_ENTRIES.load(Ordering::Relaxed);
            let a0 = smarm::causal::ledger_counters();
            smarm::causal::begin_experiment_for_test("reserve", pct);
            smarm::sleep(Duration::from_millis(1000));
            smarm::causal::end_experiment_for_test();
            let audit = smarm::causal::ledger_counters().delta_since(&a0);
            let injected = smarm::causal::global_delay_cycles() - g0;
            let in_site = IN_SITE_CYCLES.load(Ordering::Relaxed) - s0;
            let entries = SITE_ENTRIES.load(Ordering::Relaxed) - e0;

            let attributed = injected as f64 / (pct as f64 / 100.0);
            let eff = attributed / in_site as f64;
            // Books-closure check: add back the runnable off-CPU gaps the
            // audit counted (delta terms -> raw via /pct) — should be ~1.00.
            let eff_closed = (injected as f64 + audit.offcpu_in_site_cycles as f64)
                / (pct as f64 / 100.0)
                / in_site as f64;
            let missing = in_site as f64 - attributed;
            let tail_us = if entries > 0 {
                missing / entries as f64 / hz * 1e6
            } else {
                f64::NAN
            };
            println!(
                "pct {pct:>2}%  in_site {:>8.1}ms  attributed {:>8.1}ms  eff {eff:.3}  eff+offcpu {eff_closed:.3}  entries {entries}  missing/entry {tail_us:.1}µs",
                in_site as f64 / hz * 1e3,
                attributed / hz * 1e3,
            );
            // RFC 007 deficit hunt: name the losses. Drop/discard columns are
            // in would-be delta terms — divide by pct/100 to compare with the
            // missing attribution above.
            let ms = |c: u64| c as f64 / hz * 1e3;
            println!(
                "        audit: absorbed {:>7.1}ms  forgiven {:>6.1}ms  drop park {:>5.2}ms/{:<5} yield {:>5.2}ms/{:<5} offcpu {:>6.2}ms/{:<5} discard >max {:>5.2}ms/{:<3} unarmed {}",
                ms(audit.spin_absorbed_cycles),
                ms(audit.park_forgiven_cycles),
                ms(audit.drop_park_cycles),
                audit.drop_park_n,
                ms(audit.drop_yield_cycles),
                audit.drop_yield_n,
                ms(audit.offcpu_in_site_cycles),
                audit.offcpu_in_site_n,
                ms(audit.discard_overmax_cycles),
                audit.discard_overmax_n,
                audit.discard_unarmed_n
            );
            smarm::sleep(Duration::from_millis(150));
        }

        stop.store(true, Ordering::Relaxed);
        producer.join().unwrap();
        reserve.join().unwrap();
        notify.join().unwrap();
        background.join().unwrap();
        println!("probe: DONE");
    });
}