decdn_node/load_shed/
resource_pressure.rs

1//! The `ResourcePressure` load-shed policy: hysteresis over node-wide
2//! concurrency, a hard egress ceiling, hit-before-miss priority, and a
3//! per-client fair-share cap that engages only under pressure.
4
5use std::sync::atomic::{AtomicBool, Ordering};
6
7use super::{LoadShedPolicy, PressureSnapshot, RequestClass, ShedDecision, ShedReason};
8
9/// Tunable thresholds, resolved from node config (see `ResolvedLoadShed`).
10#[derive(Debug, Clone, Copy)]
11pub struct Params {
12    /// Measured egress ceiling in bytes/sec. `0` disables the egress gate.
13    pub egress_budget_bps: u64,
14    /// Concurrency high-water mark: at or above it, the node becomes pressured.
15    pub max_serves_high: u32,
16    /// Concurrency low-water mark: at or below it, pressure clears (hysteresis).
17    pub max_serves_low: u32,
18    /// Per-client concurrent-serve ceiling, enforced ONLY while pressured.
19    /// `0` disables per-client fairness.
20    pub per_client_cap: u32,
21}
22
23/// Sheds the miss tier under concurrency pressure and any class under egress
24/// saturation, while capping a single client's share only during real
25/// contention. Holds one bit of hysteresis state.
26#[derive(Debug)]
27pub struct ResourcePressure {
28    params: Params,
29    /// Latched pressure state: set at the high-water mark, cleared at the
30    /// low-water mark, so the node does not flap at a single boundary.
31    pressured: AtomicBool,
32}
33
34impl ResourcePressure {
35    /// A policy with these thresholds, starting unpressured.
36    #[must_use]
37    pub const fn new(params: Params) -> Self {
38        Self {
39            params,
40            pressured: AtomicBool::new(false),
41        }
42    }
43
44    /// Update and read the hysteresis latch from the live concurrency count.
45    fn update_pressure(&self, streams_in_flight: u32) -> bool {
46        if streams_in_flight >= self.params.max_serves_high {
47            self.pressured.store(true, Ordering::Relaxed);
48        } else if streams_in_flight <= self.params.max_serves_low {
49            self.pressured.store(false, Ordering::Relaxed);
50        }
51        self.pressured.load(Ordering::Relaxed)
52    }
53}
54
55impl LoadShedPolicy for ResourcePressure {
56    fn decide(&self, class: RequestClass, snap: &PressureSnapshot) -> ShedDecision {
57        let pressured = self.update_pressure(snap.streams_in_flight);
58
59        // Per-client fairness engages ONLY under pressure (fairness invariant):
60        // a client is capped for holding more than its share during real
61        // contention, never to reserve a slot for a client that might arrive.
62        if pressured
63            && self.params.per_client_cap > 0
64            && snap.client_streams_in_flight >= self.params.per_client_cap
65        {
66            return ShedDecision::Shed(ShedReason::ClientAtCapacity);
67        }
68
69        // Hard egress ceiling: when the pipe is full, admitting another stream
70        // — hit or miss — creates no bandwidth, so shed it. `0` disables.
71        if self.params.egress_budget_bps > 0 && snap.egress_bps >= self.params.egress_budget_bps {
72            return ShedDecision::Shed(ShedReason::EgressSaturated);
73        }
74
75        // Concurrency pressure sheds the miss tier first (it fronts upstream
76        // cost and stresses disk/CPU); hits ride until egress is the constraint.
77        if pressured && class == RequestClass::CacheMiss {
78            return ShedDecision::Shed(ShedReason::NodeAtCapacity);
79        }
80
81        ShedDecision::Admit
82    }
83
84    fn pressure_active(&self) -> bool {
85        self.pressured.load(Ordering::Relaxed)
86    }
87}
88
89#[cfg(test)]
90mod tests;
decdn_node/load_shed/resource_pressure.rs

decdn_node/load_shed/
resource_pressure.rs
