diff --git a/control_plane/src/emit/monitor.rs b/control_plane/src/emit/monitor.rs index a8ea1133..ceb6e73f 100644 --- a/control_plane/src/emit/monitor.rs +++ b/control_plane/src/emit/monitor.rs @@ -25,6 +25,11 @@ pub enum Functional { Sum, CmsPoint, LinearBuckets, + /// Whole-sketch second frequency moment F2 = ‖f‖₂². Monitors the entire + /// sketch's L2 mass (no per-point key) so any future point query stays within + /// ε — see `data_plane::monitor` module docs for the whole-sketch-vs-point + /// decision rule. The edge reports its local F2 = Σ_x f_i(x)² as the value. + F2, } impl Functional { @@ -33,6 +38,7 @@ impl Functional { Functional::Sum => "sum", Functional::CmsPoint => "cms_point", Functional::LinearBuckets => "linear_buckets", + Functional::F2 => "f2", } } @@ -41,6 +47,7 @@ impl Functional { match s { "cms_point" => Functional::CmsPoint, "linear_buckets" => Functional::LinearBuckets, + "f2" | "l2" => Functional::F2, _ => Functional::Sum, } } @@ -118,6 +125,7 @@ pub fn edge_threshold_block(intent: &MonitorIntent) -> Value { pub fn streaming_config_monitor_entry(intent: &MonitorIntent) -> serde_json::Value { serde_json::json!({ "agg_id": agg_id_for_metric(&intent.metric), + "functional": intent.functional.as_str(), "key": intent.key, "tau": intent.tau, "epsilon": intent.epsilon, diff --git a/crates/asap_types/src/streaming_config.rs b/crates/asap_types/src/streaming_config.rs index 255ff7eb..ca39b495 100644 --- a/crates/asap_types/src/streaming_config.rs +++ b/crates/asap_types/src/streaming_config.rs @@ -22,7 +22,13 @@ use crate::query_requirements::QueryRequirements; #[derive(Debug, Clone, Serialize, Deserialize)] pub struct MonitorSpec { pub agg_id: u64, - /// CMS point-frequency key x; empty (default) for Sum / whole-stream. + /// Additive readout the edge reports: "sum" (default), "cms_point", "f2". + /// Pass-through metadata so the edge can auto-learn its reporting mode from + /// the pushed config; the coordinator allocation is value-driven and does not + /// branch on it (p_i ∝ √(value/rate) is the F2 allocation when value=‖f‖²). + #[serde(default)] + pub functional: String, + /// CMS point-frequency key x; empty (default) for Sum / whole-stream / F2. #[serde(default)] pub key: String, /// Threshold τ (authoritative here, not at the edge). diff --git a/data_plane/src/monitor/coordinator.rs b/data_plane/src/monitor/coordinator.rs index b09a523e..546b445b 100644 --- a/data_plane/src/monitor/coordinator.rs +++ b/data_plane/src/monitor/coordinator.rs @@ -30,7 +30,7 @@ use std::collections::HashMap; -use super::sampling_alloc::{allocate_sample_rates, epsilon_sample_floor}; +use super::sampling_alloc::epsilon_sample_floor; /// Static configuration for one monitor, sourced from the streaming-config /// `monitors:` section (τ authoritative here, not at the edge). @@ -235,42 +235,31 @@ impl Monitor { /// 1/√rate_i`). With <2 edges or all rates unknown the allocation degenerates /// to `p=1` everywhere (no sampling). fn allocate_p(&self) -> HashMap { - // Stable edge order so the rate / freq vectors align with the p vector. - let ids: Vec<&String> = self.edges.keys().collect(); - let rates: Vec = ids.iter().map(|id| self.edges[*id].rate).collect(); - // Per-key frequency `f_i` = the monitored functional's value this edge - // knows this epoch (`known_value`: the cms_point key frequency, or the - // edge's contribution to the monitored sum), which is DISTINCT from the - // edge's total update `rate`. Feeding `f_i` (not `rates`) into the KKT - // allocation is what makes `p_i ∝ √(f_i/rate_i)` differentiate by the - // monitored key's *share* of each edge's stream. Previously this passed - // `rates` as both vectors → `√(rate_i/rate_i)=1` for every edge → a - // uniform allocation, with ALL differentiation left to the ε-floor; - // now the allocation itself is freq-driven (an edge that carries more of - // the monitored key at equal total rate keeps a higher p). An edge with - // no monitored-key mass yet (`known_value=0`) falls back to `p_i=1` - // (allocate_sample_rates: `freqs_i ≤ 0 ⇒ p_i=1`), i.e. nothing to sample. - let freqs: Vec = ids.iter().map(|id| self.edges[*id].known_value).collect(); - let var_budget = { - let band = self.cfg.epsilon * self.cfg.tau; - band * band - }; - let mut p_vec = allocate_sample_rates(&rates, &freqs, var_budget); - // Enforce the CDM-threshold coupling floor per edge: never sample so hard - // that ε_s = √((1−p)/(p·rate)) exceeds the agg's ε. - for (i, &rate) in rates.iter().enumerate() { - let floor = epsilon_sample_floor(self.cfg.epsilon, rate); - if p_vec[i] < floor { - p_vec[i] = floor; - } - } - ids.into_iter() - .cloned() - .zip(p_vec) - .map(|(id, p)| { - // Single edge or unknown rate ⇒ no sampling (p=1). - let p = if p > 0.0 && p <= 1.0 { p } else { 1.0 }; - (id, p) + // Coordinated update-sampling for a SKETCH is governed by the whole-sketch + // ε-floor — a single law, NOT a per-key `√(f_i/rate_i)` KKT allocation. + // + // Why: the sampling protects the warm SKETCH's accuracy. A sketch point/L2 + // estimate's error is bounded by the sketch NORM (‖f‖, rate-spread over + // buckets — NitroSketch), never by a single key's `f(x)`. So the only + // accuracy a per-edge `p_i` can buy is "keep this edge's L2 contribution + // within ε", i.e. ε_s = √((1−p)/(p·rate)) ≤ ε, which gives + // p_i = 1/(1 + ε²·rate_i) (the CDM-threshold coupling floor). + // The `√(freq/rate)` allocation only holds when a key is EXACT-counted + // OUTSIDE the sketch (then sampling that one counter is pointless anyway), + // so it is not a valid sketch-sampling regime — it has been retired. The + // monitored functional (cms_point / f2 / sum) still drives the THRESHOLD + // (`known_value` → `global_estimate`/alert); it no longer drives sampling. + // See `monitor` module docs for the derivation. + self.edges + .iter() + .map(|(id, e)| { + // Unknown rate ⇒ no sampling (p=1). + let p = if e.rate > 0.0 { + epsilon_sample_floor(self.cfg.epsilon, e.rate) + } else { + 1.0 + }; + (id.clone(), p) }) .collect() } @@ -484,61 +473,42 @@ mod tests { } #[test] - fn skewed_rates_yield_differentiated_sample_p_above_floor() { - // Two edges, skewed reported rates: the hot edge should be sampled - // harder (smaller p) than the quiet edge, and BOTH must sit at/above the - // ε-derived coupling floor. τ is large so the monitor stays in the grant - // (not alert) regime while we exercise the allocation. + fn skewed_rates_yield_sample_p_at_the_epsilon_floor() { + // The whole-sketch sampling law: p_i = ε-floor(rate_i). The hot (high-rate) + // edge is sampled harder (smaller p) than the quiet edge, and each p sits + // EXACTLY at its rate's ε-floor. τ is large so no alert fires. let mut m = Monitor::new(cfg(1_000_000.0)); // epsilon 0.05 m.on_register("e1", 60_000, 0); m.on_register("e2", 60_000, 0); - // e1 hot (100k items/win), e2 quiet (1k/win). Small local values keep us - // far from τ so no alert fires. - m.on_report("e1", 0, 1.0, 1, 100_000.0); - let a = m.on_report("e2", 0, 1.0, 1, 1_000.0); + m.on_report("e1", 0, 1.0, 1, 100_000.0); // hot: 100k items/win + let a = m.on_report("e2", 0, 1.0, 1, 1_000.0); // quiet: 1k/win let p_hot = sample_p_in(&a, "e1"); let p_quiet = sample_p_in(&a, "e2"); - assert!( - p_hot < p_quiet, - "hot edge p {p_hot} should be < quiet edge p {p_quiet}" - ); - // Both respect the ε-derived floor for their rate. let eps = m.cfg.epsilon; - assert!(p_hot >= epsilon_sample_floor(eps, 100_000.0) - 1e-9); - assert!(p_quiet >= epsilon_sample_floor(eps, 1_000.0) - 1e-9); - assert!(p_hot > 0.0 && p_hot <= 1.0); - assert!(p_quiet > 0.0 && p_quiet <= 1.0); + assert!(p_hot < p_quiet, "hot p {p_hot} should be < quiet p {p_quiet}"); + assert!((p_hot - epsilon_sample_floor(eps, 100_000.0)).abs() < 1e-12); + assert!((p_quiet - epsilon_sample_floor(eps, 1_000.0)).abs() < 1e-12); // Slack countdown unaffected: both grants carry the same (positive) slack. assert!(slack_in(&a, "e1") > 0.0); assert_eq!(slack_in(&a, "e1"), slack_in(&a, "e2")); } #[test] - fn per_key_freq_differentiates_at_equal_rate() { - // EQUAL total rate, but different monitored-key frequency (`known_value`): - // the edge carrying more of the monitored key keeps a HIGHER p (sampled - // less), since p_i ∝ √(f_i/rate_i) and rate cancels. Under the old - // freqs==rates this returned a UNIFORM p (same rate ⇒ same floor ⇒ tie), - // so this test fails on the pre-fix code — it pins the freq-driven branch. - // High rate ⇒ low ε-floor, modest τ ⇒ tight var_budget, so the allocation - // (not the floor) is the operative differentiator. - let mut m = Monitor::new(cfg(2_000.0)); // epsilon 0.05 ⇒ var_budget=(100)^2 + fn sampling_is_rate_floor_independent_of_monitored_freq() { + // Unified law: p_i = ε-floor(rate_i), the whole-sketch sampling law — it + // depends ONLY on rate, never on the monitored-key frequency + // (`known_value`). EQUAL rate but very different known_value ⇒ EQUAL p. + // (The retired `√(f_i/rate_i)` allocation would have differentiated here; + // it does not, because sketch accuracy is rate/L2-bounded, not per-key.) + let mut m = Monitor::new(cfg(2_000.0)); m.on_register("e1", 60_000, 0); m.on_register("e2", 60_000, 0); - // same rate (100k/win); e1 sees the monitored key 10× more than e2. m.on_report("e1", 0, 900.0, 1, 100_000.0); // known_value=900 - let a = m.on_report("e2", 0, 90.0, 1, 100_000.0); // known_value=90 + let a = m.on_report("e2", 0, 90.0, 1, 100_000.0); // known_value=90, same rate let p_e1 = sample_p_in(&a, "e1"); let p_e2 = sample_p_in(&a, "e2"); - let floor = epsilon_sample_floor(m.cfg.epsilon, 100_000.0); - assert!( - p_e1 > p_e2, - "high-freq edge p {p_e1} should exceed low-freq p {p_e2} (freq-driven, not floor)" - ); - // Both sit ABOVE the (equal) rate-floor, proving the allocation — not the - // floor — produced the spread. - assert!(p_e1 > floor + 1e-6 && p_e2 > floor + 1e-6, - "both p ({p_e1}, {p_e2}) should exceed the equal rate-floor {floor}"); + assert_eq!(p_e1, p_e2, "equal rate ⇒ equal p regardless of monitored freq"); + assert!((p_e1 - epsilon_sample_floor(m.cfg.epsilon, 100_000.0)).abs() < 1e-12); } #[test] diff --git a/data_plane/src/monitor/f2.rs b/data_plane/src/monitor/f2.rs index 0527d5a3..e1d5b37e 100644 --- a/data_plane/src/monitor/f2.rs +++ b/data_plane/src/monitor/f2.rs @@ -31,7 +31,6 @@ use std::collections::HashMap; -use super::sampling_alloc::{allocate_sample_rates, epsilon_sample_floor}; /// murmur3 64-bit finalizer (`fmix64`). #[inline] @@ -183,12 +182,12 @@ pub enum F2Action { #[derive(Clone, Debug)] struct EdgeF2 { sketch: CountSketchF2, - rate: f64, } -/// Distributed F2 threshold monitor + coordinated-sampling allocator. Holds each -/// edge's latest Count-Sketch + rate, merges on report, estimates the global -/// `F2̂`, fires at `F2̂ ≥ (1−ε)τ`, and allocates `p_i ∝ √(F2̂_i / rate_i)`. +/// Distributed F2 **threshold** monitor. Holds each edge's latest Count-Sketch, +/// linearly merges on report, estimates the global `F2̂`, and fires at +/// `F2̂ ≥ (1−ε)τ`. F2 is a MONITORED quantity only — per-edge update-sampling is +/// the single whole-sketch ε-floor law (`sampling_alloc`), not driven by F2. pub struct DistributedF2Monitor { tau: f64, epsilon: f64, @@ -237,15 +236,16 @@ impl DistributedF2Monitor { false } - /// Ingest one edge's `(Count-Sketch, rate)` for `window_start_ms`. Replaces - /// that edge's prior report, re-estimates the merged global F2, and returns - /// an Alert iff it crossed `(1−ε)τ`. + /// Ingest one edge's Count-Sketch for `window_start_ms`. Replaces that edge's + /// prior report, re-estimates the merged global F2, and returns an Alert iff + /// it crossed `(1−ε)τ`. (`rate` is part of the edge wire report but the F2 + /// threshold needs only the sketch; sampling consumes rate elsewhere.) pub fn on_report( &mut self, edge_id: &str, window_start_ms: u64, sketch: CountSketchF2, - rate: f64, + _rate: f64, ) -> Option { if sketch.dims() != (self.d, self.w, self.seed) { return None; // mis-dimensioned — drop @@ -253,7 +253,7 @@ impl DistributedF2Monitor { if !self.ensure_epoch(window_start_ms) { return None; // stale epoch } - self.edges.insert(edge_id.to_string(), EdgeF2 { sketch, rate }); + self.edges.insert(edge_id.to_string(), EdgeF2 { sketch }); let f2 = self.global_f2(); if !self.alerted && f2 >= (1.0 - self.epsilon) * self.tau { self.alerted = true; @@ -275,37 +275,10 @@ impl DistributedF2Monitor { merged.estimate_f2() } - /// L2 sampling variance budget `V = (ε·‖f‖₂)² = ε²·F2̂` (vs L1's `(ε·τ)²`). - pub fn sampling_var_budget(&self) -> f64 { - self.epsilon * self.epsilon * self.global_f2() - } - - /// Coordinated per-edge sampling probability `p_i ∝ √(F2̂_i / rate_i)`: the - /// freq weight is each edge's LOCAL F2 (from its own sketch), the rate is its - /// reported update rate, the budget is `ε²·F2̂_global`. An edge contributing - /// more L2 mass is sampled less (higher p) to preserve the F2 estimate. - /// Mirrors the L1 `coordinator::allocate_p` but with L2 freq/budget. - pub fn allocate_p(&self) -> HashMap { - let ids: Vec<&String> = self.edges.keys().collect(); - let rates: Vec = ids.iter().map(|id| self.edges[*id].rate).collect(); - let freqs: Vec = ids - .iter() - .map(|id| self.edges[*id].sketch.estimate_f2()) // local F2̂_i - .collect(); - let var_budget = self.sampling_var_budget(); - let mut p = allocate_sample_rates(&rates, &freqs, var_budget); - for (i, &rate) in rates.iter().enumerate() { - let floor = epsilon_sample_floor(self.epsilon, rate); - if p[i] < floor { - p[i] = floor; - } - } - ids.into_iter() - .cloned() - .zip(p) - .map(|(id, pi)| (id, if pi > 0.0 && pi <= 1.0 { pi } else { 1.0 })) - .collect() - } + // NOTE: F2 here is a MONITORED quantity (threshold/alert), NOT a sampling + // driver. Per-edge update-sampling is the single whole-sketch ε-floor law + // (`sampling_alloc::epsilon_sample_floor`); the old `√(F2/rate)` allocation + // was retired (it is not a valid sketch-sampling regime — see `monitor` docs). } /// Geometric (Sharfman–Schuster–Keren) safe-zone layer for **communication-efficient** @@ -492,42 +465,6 @@ mod tests { assert!(matches!(r_lo, F2Action::Alert { .. }), "above tau ⇒ Alert, got {r_lo:?}"); } - #[test] - fn allocate_p_higher_local_f2_keeps_higher_p_at_equal_rate() { - // Equal rate; edge A carries far more L2 mass than edge B (disjoint keys). - // p_i ∝ √(F2_i/rate) ⇒ A (high F2) keeps a HIGHER p (sampled less) to - // preserve the F2 estimate. High rate ⇒ low floor so the allocation, not - // the floor, drives the spread. - let seed = 0x1357_2468; - let mut mon = DistributedF2Monitor::new(1.0e12, 0.1, 9, 4096, seed); - let mut a = mon.new_edge_sketch(); - let mut b = mon.new_edge_sketch(); - fill(&mut a, &[(1u64, 100i64)]); // local F2 ≈ 10000 - fill(&mut b, &[(2u64, 1i64)]); // local F2 ≈ 1 - mon.on_report("a", 0, a, 1_000_000.0); - mon.on_report("b", 0, b, 1_000_000.0); - let p = mon.allocate_p(); - let pa = p["a"]; - let pb = p["b"]; - let floor = epsilon_sample_floor(0.1, 1_000_000.0); - assert!(pa > pb, "high-F2 edge p {pa} should exceed low-F2 p {pb}"); - assert!(pa > floor + 1e-9 && pb > floor + 1e-9, "p ({pa},{pb}) above floor {floor}"); - } - - #[test] - fn sampling_var_budget_is_eps2_times_f2() { - let seed = 0x9999_1111; - let eps = 0.2; - let mut mon = DistributedF2Monitor::new(1.0e12, eps, 9, 4096, seed); - let mut e = mon.new_edge_sketch(); - fill(&mut e, &(1..=100u64).map(|k| (k, k as i64)).collect::>()); - mon.on_report("e", 0, e, 1000.0); - let f2 = mon.global_f2(); - let v = mon.sampling_var_budget(); - assert!((v - eps * eps * f2).abs() < 1e-6); - assert!(v > 0.0); - } - #[test] fn epoch_advance_clears_edges() { let mut mon = DistributedF2Monitor::new(1_000.0, 0.1, 5, 1024, 7); diff --git a/data_plane/src/monitor/mod.rs b/data_plane/src/monitor/mod.rs index 4196b44e..20ea896b 100644 --- a/data_plane/src/monitor/mod.rs +++ b/data_plane/src/monitor/mod.rs @@ -13,6 +13,26 @@ //! - [`epoch`] — tumbling-epoch alignment (matches the edge formula). //! - [`alert`] — alert egress via the control-plane `Violation` sink. //! - [`server`] — the tonic bidi-streaming `MonitorService` server. +//! - [`sampling_alloc`] — the coordinated update-sampling law (ε-floor). +//! - [`f2`] — whole-sketch L2/F2 *threshold* monitor (Count-Sketch + +//! geometric safe-zone). NOTE: F2 here is a MONITORED quantity (alert), not a +//! sampling driver — see the sampling law below. +//! +//! # Two orthogonal axes — DON'T conflate them +//! +//! 1. **What to monitor / alert on** (the `functional`): `sum`, `cms_point` +//! (a declared point `f(x)`), or `f2` (whole-sketch L2). This sets the +//! THRESHOLD `g vs τ` and what `known_value` means. Identity is `(agg_id,key)` +//! — `cms_point` carries a key; `sum`/`f2` are whole-stream (`key=""`). +//! +//! 2. **How hard to sample** (the per-edge `p_i`): a SINGLE law, the whole-sketch +//! **ε-floor** `p_i = 1/(1+ε²·rate_i)` (see [`sampling_alloc`]). It depends +//! ONLY on each edge's rate, NEVER on the monitored functional — because the +//! sampling protects a SKETCH, whose accuracy is bounded by the stream norm +//! (rate/L2), not by any single key. A per-key `√(f/rate)` allocation is NOT a +//! valid sketch-sampling regime (it needs exact-counting outside the sketch), +//! so it has been retired. Applies to additive sketches — Count-Min, Count- +//! Sketch, DDSketch, KLL, Sum — but NOT HLL (sampling biases cardinality). pub mod alert; pub mod coordinator; @@ -23,5 +43,5 @@ pub mod server; pub use alert::{global_threshold_violation, AlertSink}; pub use coordinator::{Action, Monitor, MonitorConfig}; -pub use sampling_alloc::{allocate_sample_rates, epsilon_sample_floor, uniform_sample_rate}; +pub use sampling_alloc::epsilon_sample_floor; pub use server::{MonitorCoordinator, MonitorServiceImpl}; diff --git a/data_plane/src/monitor/sampling_alloc.rs b/data_plane/src/monitor/sampling_alloc.rs index 3e8a8793..e356e101 100644 --- a/data_plane/src/monitor/sampling_alloc.rs +++ b/data_plane/src/monitor/sampling_alloc.rs @@ -1,133 +1,46 @@ -//! Distributed-NitroSketch coordinated update-sampling allocation — the -//! coordinator's per-edge sampling-probability `p_i` computation. Faithful Rust -//! port of the Go reference `asap-precompute-go/monitor/sampling_alloc.go` -//! (`AllocateSampleRates`), plus the CDM-threshold coupling floor from -//! `docs/distributed-nitrosketch-coordinated-sampling.md` -//! ("Combining with the CDM threshold"). +//! Coordinated update-sampling: the per-edge sampling-probability `p_i` law. //! -//! Each edge reports its observed per-window `rate` (items/window) over the -//! reverse `MonitorService` channel (`MonitorReport.rate`); the coordinator -//! holds the full per-edge rate vector for a monitor and allocates a `p_i` that -//! MINIMIZES total edge update work `Σ rate_i·p_i` subject to a merged -//! sampling-variance budget `V ≥ Σ f_i·(1−p_i)/p_i`. The KKT solution is +//! For a SKETCH-based warm tier the law is a single, closed form — the +//! **whole-sketch ε-floor** //! //! ```text -//! p_i = clamp( √λ · √(f_i / rate_i), 0, 1 ) +//! p_i = 1 / (1 + ε²·rate_i) //! ``` //! -//! i.e. sample harder (smaller `p`) on high-rate edges where local accuracy is -//! cheap, keep `p≈1` on low-rate edges. `√λ` is the single scale that makes the -//! variance constraint bind; solved by bisection (variance is monotone -//! decreasing in the scale). The result rides each edge's `SlackGrant.sample_p`. - -/// Compute per-edge update-sampling probabilities `p_i`. +//! and that is all this module provides ([`epsilon_sample_floor`]). +//! +//! Why one law (and not a per-key `√(f_i/rate_i)` allocation): the sampling +//! protects the warm sketch's accuracy, and a sketch's point/L2 estimate error +//! is bounded by the sketch NORM (‖f‖, the rate spread over buckets — +//! NitroSketch), never by a single key's `f(x)`. So the only accuracy a per-edge +//! `p_i` buys is "keep this edge's L2/mass contribution within ε": +//! `ε_s = √((1−p)/(p·rate)) ≤ ε`, which solves to the floor above. The +//! `√(f_i/rate_i)` KKT allocation (minimise `Σ rate_i·p_i` s.t. +//! `Σ f_i·(1−p_i)/p_i ≤ V`) only holds when a key is EXACT-counted OUTSIDE the +//! sketch — and sampling a single exact counter is pointless — so it is not a +//! valid sketch-sampling regime and has been retired. +//! +//! **Which sketches:** the law applies to any additive/linear sketch whose +//! estimate is unbiased (after a 1/p rescale) under update-sampling, and whose +//! error is norm- or rank-bounded: **Count-Min (L1), Count-Sketch (L2), +//! DDSketch & KLL (quantile rank; rank-preserving, no 1/p rescale needed), Sum**. +//! It does NOT apply to **HLL / cardinality**: update-sampling systematically +//! under-counts distinct items (you measure the sample's cardinality, not the +//! full set) and is not 1/p-correctable, so HLL needs a separate treatment. + +/// The whole-sketch coordinated-sampling probability for an edge with the given +/// per-window `rate` (items/window): `p = 1/(1 + ε²·rate)`. /// -/// `rates[i]` = edge i's items/window; `freqs[i]` = edge i's mass of the queried -/// quantity (pass `rates` as a proxy when the per-key split is unknown — then -/// `p_i ∝ 1/√rate_i`). `var_budget` = `V` in the bound above. A `rate_i ≤ 0` or -/// `freqs_i ≤ 0` yields `p_i = 1` (nothing to sample). -pub fn allocate_sample_rates(rates: &[f64], freqs: &[f64], var_budget: f64) -> Vec { - let n = rates.len(); - let mut p = vec![0.0_f64; n]; - if n == 0 { - return p; - } - // base[i] = √(f_i/rate_i); 0 forces p_i = 1 (no rate ⇒ no sampling benefit). - let mut base = vec![0.0_f64; n]; - let mut max_base = 0.0_f64; - for i in 0..n { - if rates[i] <= 0.0 || freqs[i] <= 0.0 { - base[i] = 0.0; - continue; - } - base[i] = (freqs[i] / rates[i]).sqrt(); - if base[i] > max_base { - max_base = base[i]; - } - } - if max_base == 0.0 || var_budget <= 0.0 { - return vec![1.0; n]; - } - - // variance(scale) = Σ f_i·(1−p_i)/p_i with p_i = min(1, scale·base_i). - // Monotone decreasing in scale: scale→0 ⇒ p→0 ⇒ variance→∞; at - // scale = 1/min(positive base_i) all p_i hit 1 ⇒ variance = 0. - let variance = |scale: f64| -> f64 { - let mut v = 0.0; - for i in 0..n { - if base[i] == 0.0 { - continue; // p_i = 1, contributes 0 - } - let pi = scale * base[i]; - if pi >= 1.0 { - continue; - } - v += freqs[i] * (1.0 - pi) / pi; - } - v - }; - - // scale_hi: every positive-base edge clamped to p_i = 1 (variance 0 ≤ V). - let mut min_base = f64::INFINITY; - for &b in &base { - if b > 0.0 && b < min_base { - min_base = b; - } - } - let scale_hi = 1.0 / min_base; - if variance(scale_hi) >= var_budget { - // Even no sampling (all p=1) can't meet V → don't sample at all. - return vec![1.0; n]; - } - // Bisect for the smallest scale (= most sampling, least CPU) with - // variance(scale) ≤ var_budget. - let mut lo = 0.0_f64; - let mut hi = scale_hi; - for _ in 0..100 { - let mid = 0.5 * (lo + hi); - if mid <= 0.0 { - lo = mid; - continue; - } - if variance(mid) > var_budget { - lo = mid; // too much sampling (variance too high) → raise scale - } else { - hi = mid; - } - } - let scale = hi; - for i in 0..n { - if base[i] == 0.0 { - p[i] = 1.0; - } else { - p[i] = (scale * base[i]).min(1.0); - } - } - p -} - -/// The single uniform sampling probability `p` meeting the same variance budget -/// `V` with one rate everywhere (the per-edge-independent NitroSketch baseline): -/// `(1−p)/p·Σf = V ⇒ p = Σf/(Σf+V)`. Used to quantify the coordination win. -pub fn uniform_sample_rate(freqs: &[f64], var_budget: f64) -> f64 { - let sum_f: f64 = freqs.iter().filter(|&&f| f > 0.0).sum(); - if sum_f <= 0.0 || var_budget <= 0.0 { - return 1.0; - } - (sum_f / (sum_f + var_budget)).min(1.0) -} - -/// The CDM-threshold coupling floor on `p_i`: never sample so hard that the -/// sampling noise `ε_s = √((1−p)/(p·rate))` exceeds the CDM threshold tolerance -/// `ε` (the agg's epsilon). Solving `ε ≥ √((1−p)/(p·rate))` for `p`: +/// Derivation — never sample so hard that the per-edge sampling noise +/// `ε_s = √((1−p)/(p·rate))` exceeds the CDM threshold tolerance `ε`: /// /// ```text /// ε²·p·rate ≥ 1 − p ⇒ p·(ε²·rate + 1) ≥ 1 ⇒ p ≥ 1/(1 + ε²·rate) /// ``` /// -/// With `rate ≤ 0` or `epsilon ≤ 0` the floor is 1.0 (no sampling permitted). The -/// floor is always in `(0,1]` and grows toward 1 as the edge's rate falls (a -/// low-`N` edge can't absorb sampling noise within the band). +/// With `rate ≤ 0` or `epsilon ≤ 0` the result is 1.0 (no sampling). The value is +/// always in `(0,1]` and grows toward 1 as the edge's rate falls (a low-`N` edge +/// can't absorb sampling noise within the band). pub fn epsilon_sample_floor(epsilon: f64, rate: f64) -> f64 { if rate <= 0.0 || epsilon <= 0.0 { return 1.0; @@ -139,91 +52,6 @@ pub fn epsilon_sample_floor(epsilon: f64, rate: f64) -> f64 { mod tests { use super::*; - fn merged_variance(freqs: &[f64], p: &[f64]) -> f64 { - let mut v = 0.0; - for i in 0..freqs.len() { - if p[i] > 0.0 && p[i] < 1.0 { - v += freqs[i] * (1.0 - p[i]) / p[i]; - } - } - v - } - fn total_cpu(rates: &[f64], p: &[f64]) -> f64 { - rates.iter().zip(p).map(|(r, p)| r * p).sum() - } - - /// Mirrors Go `TestAllocateSampleRates_SkewedFleetBeatsUniform`: on a skewed - /// fleet (one hot edge, a quiet tail) the coordinated p_i ∝ √(f_i/rate_i) - /// allocation hits the same merged-variance budget at strictly LESS total - /// edge update work than per-edge-uniform NitroSketch. - #[test] - fn skewed_fleet_beats_uniform() { - let rates = [100000.0, 1000.0, 1000.0, 1000.0, 1000.0]; // 1 hot + 4 quiet - let freqs = [100.0, 100.0, 100.0, 100.0, 100.0]; // key ~uniform across edges - let v = 2000.0; - - let p_coord = allocate_sample_rates(&rates, &freqs, v); - let p_uni = uniform_sample_rate(&freqs, v); - let uni = vec![p_uni; rates.len()]; - - let vc = merged_variance(&freqs, &p_coord); - let vu = merged_variance(&freqs, &uni); - assert!(vc <= v * 1.02, "coordinated variance {vc} exceeds budget {v}"); - assert!( - (vu - v).abs() <= v * 0.02, - "uniform variance {vu} should ≈ budget {v}" - ); - let cc = total_cpu(&rates, &p_coord); - let cu = total_cpu(&rates, &uni); - assert!( - cc < cu, - "coordinated CPU {cc} should be < uniform CPU {cu} at equal variance" - ); - // Hot edge sampled hard; quiet edges near p=1. - assert!( - p_coord[0] < p_coord[1], - "hot edge should get smaller p than quiet edges: {p_coord:?}" - ); - } - - /// Mirrors Go `TestAllocateSampleRates_FlatFleetNoWin`: on a uniform fleet - /// the coordinated allocation collapses to ≈ the uniform rate (no win). - #[test] - fn flat_fleet_no_win() { - let rates = [2000.0, 2000.0, 2000.0, 2000.0]; - let v = 3000.0; - let p_coord = allocate_sample_rates(&rates, &rates, v); - let p_uni = uniform_sample_rate(&rates, v); - let cc = total_cpu(&rates, &p_coord); - let cu = p_uni * (2000.0 * 4.0); - assert!( - (cc - cu).abs() <= 0.05 * cu, - "flat fleet: coordinated CPU {cc} should ≈ uniform {cu} (no win)" - ); - } - - #[test] - fn zero_or_negative_rate_gets_no_sampling() { - let rates = [0.0, -5.0, 1000.0]; - let freqs = [100.0, 100.0, 100.0]; - let p = allocate_sample_rates(&rates, &freqs, 50.0); - assert_eq!(p[0], 1.0); - assert_eq!(p[1], 1.0); - assert!(p[2] <= 1.0 && p[2] > 0.0); - } - - #[test] - fn empty_input() { - assert!(allocate_sample_rates(&[], &[], 100.0).is_empty()); - } - - #[test] - fn zero_budget_means_no_sampling() { - let rates = [1000.0, 2000.0]; - let p = allocate_sample_rates(&rates, &rates, 0.0); - assert_eq!(p, vec![1.0, 1.0]); - } - #[test] fn epsilon_floor_monotone_and_bounded() { // Floor in (0,1], rises toward 1 as rate falls. @@ -241,12 +69,12 @@ mod tests { } #[test] - fn floor_enforced_keeps_sampling_noise_within_epsilon() { - // A floored p_i keeps ε_s = √((1−p)/(p·rate)) ≤ ε. + fn floor_keeps_sampling_noise_within_epsilon() { + // The floored p keeps ε_s = √((1−p)/(p·rate)) ≤ ε. let eps = 0.05; let rate = 100000.0; - let floor = epsilon_sample_floor(eps, rate); - let eps_s = ((1.0 - floor) / (floor * rate)).sqrt(); + let p = epsilon_sample_floor(eps, rate); + let eps_s = ((1.0 - p) / (p * rate)).sqrt(); assert!(eps_s <= eps + 1e-9, "ε_s {eps_s} should be ≤ ε {eps}"); } }