zc2 0.0.32

//! Running the adaptive policy alongside the real one, without letting it
//! route anything.
//!
//! A simulation said the policy beats round-robin by a wide margin on a mesh
//! shaped like this one. That is a reason to try it, not a reason to switch:
//! a simulation is only as good as its assumptions, and the mesh is not a
//! simulation. So the policy runs on every real decision, its choice is
//! recorded next to the one actually made, and the difference between them is
//! accumulated. The switch is justified by that ledger or not at all.
//!
//! Cost: `d` constant-time draws per decision. Nothing here touches the
//! network or blocks the request.

use super::policy::{self, Candidate, Weights};
use super::sampler::{AliasTable, Rng};
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::Mutex;

/// What the shadow run has seen so far.
#[derive(Debug, Default, Clone, Copy, PartialEq)]
pub struct Verdict {
    pub decisions: u64,
    /// Times the policy would have picked what was picked anyway.
    pub agreed: u64,
    /// Expected milliseconds the policy's choices would have saved, summed.
    /// Negative means the live routing was doing better.
    pub expected_ms_saved: f64,
}

impl Verdict {
    pub fn agreement(&self) -> f64 {
        match self.decisions {
            0 => 0.0,
            n => self.agreed as f64 / n as f64,
        }
    }

    /// The number that decides the switch: expected milliseconds saved per
    /// task, on this mesh, under live traffic.
    pub fn ms_saved_per_task(&self) -> f64 {
        match self.decisions {
            0 => 0.0,
            n => self.expected_ms_saved / n as f64,
        }
    }
}

/// Accumulates the comparison across decisions.
#[derive(Debug)]
pub struct Shadow {
    decisions: AtomicU64,
    agreed: AtomicU64,
    /// Fixed-point microseconds, so the running total needs no lock.
    saved_us: AtomicU64,
    saved_negative_us: AtomicU64,
    rng: Mutex<Rng>,
    weights: Weights,
}

impl Default for Shadow {
    fn default() -> Self {
        Self::seeded(0x5EED)
    }
}

impl Shadow {
    pub fn seeded(seed: u64) -> Self {
        Shadow {
            decisions: AtomicU64::new(0),
            agreed: AtomicU64::new(0),
            saved_us: AtomicU64::new(0),
            saved_negative_us: AtomicU64::new(0),
            rng: Mutex::new(Rng::seeded(seed)),
            weights: Weights::default(),
        }
    }

    /// Record one real decision and what the policy would have done instead.
    ///
    /// `chosen` indexes `candidates`. Returns what the policy picked, for a
    /// caller that wants to log the pair.
    pub fn observe(&self, candidates: &[Candidate], chosen: usize) -> Option<usize> {
        if candidates.is_empty() || chosen >= candidates.len() {
            return None;
        }
        let weights: Vec<f64> = candidates
            .iter()
            .map(|c| policy::table_weight(c, &self.weights))
            .collect();
        let table = AliasTable::build(&weights)?;
        let mut rng = self.rng.lock().ok()?;
        let would = policy::choose(candidates, &table, &mut rng, 2, &self.weights)?;
        drop(rng);

        self.decisions.fetch_add(1, Ordering::Relaxed);
        if would == chosen {
            self.agreed.fetch_add(1, Ordering::Relaxed);
        }

        // Compare on expected completion, not on a sample: the question is
        // which choice was better on what is known, not which drew luckier.
        let expected = |i: usize| -> f64 {
            let c = &candidates[i];
            let service = c
                .stats
                .expected_service_ms()
                .unwrap_or(self.weights.unmeasured_ms);
            service * (1.0 + c.in_flight as f64)
        };
        let delta = expected(chosen) - expected(would);
        let us = (delta.abs() * 1000.0) as u64;
        match delta >= 0.0 {
            true => self.saved_us.fetch_add(us, Ordering::Relaxed),
            false => self.saved_negative_us.fetch_add(us, Ordering::Relaxed),
        };
        Some(would)
    }

    /// A generator for the routing path to draw from, so switching the policy
    /// on does not need a second RNG threaded through the broker. Seeded from
    /// the shadow's own stream, so a run stays reproducible.
    pub fn rng_for_routing(&self) -> Rng {
        match self.rng.lock() {
            Ok(mut r) => {
                let seed = (r.next_f64() * u64::MAX as f64) as u64;
                Rng::seeded(seed)
            }
            Err(_) => Rng::seeded(1),
        }
    }

    pub fn verdict(&self) -> Verdict {
        let saved = self.saved_us.load(Ordering::Relaxed) as f64;
        let lost = self.saved_negative_us.load(Ordering::Relaxed) as f64;
        Verdict {
            decisions: self.decisions.load(Ordering::Relaxed),
            agreed: self.agreed.load(Ordering::Relaxed),
            expected_ms_saved: (saved - lost) / 1000.0,
        }
    }

    /// One line for the broker log, or `None` when there is nothing yet to say.
    pub fn summary(&self) -> Option<String> {
        let v = self.verdict();
        (v.decisions > 0).then(|| {
            format!(
                "[SHADOW] {} decisions · adaptive agreed {:.0}% · would save {:.0} ms/task",
                v.decisions,
                v.agreement() * 100.0,
                v.ms_saved_per_task()
            )
        })
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::broker::estimator::Stats;

    fn worker(median_ms: f64, n: usize, in_flight: u32) -> Candidate {
        let mut stats = Stats::new();
        for _ in 0..n {
            stats.observe(median_ms, true);
        }
        Candidate {
            stats,
            in_flight,
            price_per_hour: 3.6,
        }
    }

    /// The ledger has to be able to say "switching would have helped" in a
    /// number, or the switch is a matter of taste.
    #[test]
    fn it_counts_what_the_live_choice_cost_against_what_the_policy_would_have_picked() {
        let shadow = Shadow::seeded(1);
        // Four fast workers and nine slow ones: the mesh this was measured on.
        let mut fleet: Vec<Candidate> = (0..4).map(|_| worker(254.0, 40, 0)).collect();
        fleet.extend((0..9).map(|_| worker(1000.0, 40, 0)));

        // Live routing is round-robin, so it lands on the slow ones most of
        // the time.
        for t in 0..260 {
            shadow.observe(&fleet, t % fleet.len());
        }
        let v = shadow.verdict();
        assert_eq!(v.decisions, 260);
        assert!(
            v.ms_saved_per_task() > 200.0,
            "the policy should look clearly better here: {:.0} ms/task",
            v.ms_saved_per_task()
        );
        assert!(
            v.agreement() < 0.5,
            "it should disagree with round-robin often: {:.0}%",
            v.agreement() * 100.0
        );
    }

    /// And it has to be able to say "no", or it is not evidence. When the live
    /// routing is already picking the best worker, the ledger must not claim a
    /// saving.
    #[test]
    fn it_reports_no_saving_when_the_live_choice_is_already_the_best() {
        let shadow = Shadow::seeded(2);
        let mut fleet = vec![worker(50.0, 40, 0)];
        fleet.extend((0..8).map(|_| worker(900.0, 40, 0)));

        for _ in 0..200 {
            shadow.observe(&fleet, 0); // always the fast one
        }
        let v = shadow.verdict();
        assert!(
            v.ms_saved_per_task() <= 0.0,
            "nothing to gain, got {:.1} ms/task",
            v.ms_saved_per_task()
        );
        assert!(
            v.agreement() > 0.8,
            "it should mostly agree: {:.0}%",
            v.agreement() * 100.0
        );
    }

    #[test]
    fn an_empty_or_out_of_range_decision_is_ignored_rather_than_panicking() {
        let shadow = Shadow::seeded(3);
        assert_eq!(shadow.observe(&[], 0), None);
        assert_eq!(shadow.observe(&[worker(10.0, 5, 0)], 7), None);
        assert_eq!(shadow.verdict(), Verdict::default());
        assert!(shadow.summary().is_none(), "nothing to report yet");
    }

    #[test]
    fn the_summary_says_what_was_learned() {
        let shadow = Shadow::seeded(4);
        let fleet = vec![worker(50.0, 40, 0), worker(900.0, 40, 0)];
        for _ in 0..50 {
            shadow.observe(&fleet, 1); // always the slow one
        }
        let line = shadow.summary().expect("something to say");
        assert!(line.contains("[SHADOW]"), "{line}");
        assert!(line.contains("50 decisions"), "{line}");
        assert!(line.contains("ms/task"), "{line}");
    }
}