subetha_cxc/
qos_policy.rs

1//! `QosPolicy`: DDS-inspired Quality-of-Service knobs that sidecar
2//! policies read as input alongside peer counts and workload shape.
3//!
4//! Mirrors the DDS (Data Distribution Service) QoS model that
5//! publishers and subscribers negotiate at connection time: in DDS
6//! the knobs are static; in this substrate they are runtime-mutable
7//! atomics, so a sidecar can re-read them every scan and adapt the
8//! shape / locale / protocol decisions accordingly.
9//!
10//! # Knobs
11//!
12//! - [`Durability`]: Volatile (no persistence), Transient (in-memory
13//!   cross-process), Persistent (disk-backed). Maps to a
14//!   recommended `Locale` for `LocaleAdaptiveRing`.
15//! - [`Reliability`]: BestEffort (drop on backpressure) vs Reliable
16//!   (backpressure-block sender until consumer catches up).
17//! - [`History`]: KeepLastN (bounded buffer of latest N items) vs
18//!   KeepAll (capacity-bounded, no automatic dropping). Drives the
19//!   recommended ring capacity.
20//! - [`max_latency`](QosPolicy::max_latency): caller's wish on
21//!   delivery latency; sidecar policies factor this when choosing
22//!   between batched (lower throughput-cost-per-item) and
23//!   single-item (lower latency) dispatch.
24//!
25//! All four knobs are mutable at runtime via atomic stores. The
26//! sidecar reads them on every scan cycle.
27
28use std::sync::atomic::{AtomicU32, AtomicU64, Ordering as AtomOrd};
29use std::time::Duration;
30
31use crate::locale_adaptive_ring::Locale;
32
33/// Where the substrate stores bytes for the substrate's contract
34/// with the durability knob.
35#[repr(u32)]
36#[derive(Debug, Clone, Copy, PartialEq, Eq)]
37pub enum Durability {
38    /// In-process anonymous memory; data evaporates on process exit.
39    Volatile = 0,
40    /// Cross-process RAM-resident shared memory; survives the
41    /// producer's exit if another holder keeps the named region
42    /// alive, but is not on disk.
43    Transient = 1,
44    /// File-backed mmap; the bytes hit the page cache and can
45    /// persist to disk via flush.
46    Persistent = 2,
47}
48
49impl Durability {
50    /// Map a durability setting to the matching `Locale` member.
51    pub fn recommended_locale(self) -> Locale {
52        match self {
53            Self::Volatile => Locale::Anon,
54            Self::Transient => Locale::ShmFs,
55            Self::Persistent => Locale::File,
56        }
57    }
58
59    fn from_u32(tag: u32) -> Self {
60        match tag {
61            0 => Self::Volatile,
62            1 => Self::Transient,
63            2 => Self::Persistent,
64            _ => panic!("QosPolicy.durability corrupted: {tag}"),
65        }
66    }
67}
68
69/// Whether the substrate drops items under backpressure or blocks
70/// the sender until the consumer catches up.
71#[repr(u32)]
72#[derive(Debug, Clone, Copy, PartialEq, Eq)]
73pub enum Reliability {
74    /// Drop items when the ring is full; sender's `try_send`
75    /// returns `Err(Full)` immediately. Lowest latency on the send
76    /// side; lossy under backpressure.
77    BestEffort = 0,
78    /// Block-spin the sender (or async-yield) until the ring has
79    /// capacity. Lossless; latency rises under backpressure.
80    Reliable = 1,
81}
82
83impl Reliability {
84    fn from_u32(tag: u32) -> Self {
85        match tag {
86            0 => Self::BestEffort,
87            1 => Self::Reliable,
88            _ => panic!("QosPolicy.reliability corrupted: {tag}"),
89        }
90    }
91}
92
93/// History depth policy: how many items the substrate retains for
94/// late-joining subscribers (or for replay).
95#[derive(Debug, Clone, Copy, PartialEq, Eq)]
96pub enum History {
97    /// Keep at most the last N items. Older items get dropped to
98    /// make room when the buffer fills.
99    KeepLastN(u32),
100    /// Keep every item up to ring capacity. When capacity is
101    /// reached, the reliability policy decides what happens.
102    KeepAll,
103}
104
105/// Whether the consumer cares about cross-producer delivery order.
106///
107/// Ordering need is semantic - it lives in the application, not in
108/// the traffic - so the substrate never auto-changes a correctness
109/// property on a heuristic. The caller DECLARES the need here; the
110/// sidecar acts on the declaration: an unstamped
111/// [`AdaptiveRing`](crate::AdaptiveRing) morphs to the Vyukov shape
112/// (the proven global-FIFO structure), a stamped ring flips its
113/// merge flag (the cheap ordered switch). What the substrate
114/// observes on its own is the cross-producer inversion RATE, which
115/// it reports - and acts on only when the caller pre-authorized an
116/// automatic response via an `auto_order` threshold.
117#[repr(u32)]
118#[derive(Debug, Clone, Copy, PartialEq, Eq)]
119pub enum Ordering {
120    /// Items from one producer arrive in that producer's push
121    /// order; no guarantee across producers. The composed shapes'
122    /// native (and cheapest) guarantee.
123    PerProducer = 0,
124    /// Items arrive in global push order across all producers.
125    GlobalFifo = 1,
126}
127
128impl Ordering {
129    fn from_u32(tag: u32) -> Self {
130        match tag {
131            0 => Self::PerProducer,
132            1 => Self::GlobalFifo,
133            _ => panic!("QosPolicy.ordering corrupted: {tag}"),
134        }
135    }
136}
137
138impl History {
139    /// Recommended ring capacity for this history setting. Caller
140    /// can clamp / round up to a power of 2 as needed (the ring
141    /// primitives require pow2 capacity).
142    pub fn recommended_capacity(self) -> usize {
143        match self {
144            Self::KeepLastN(n) => (n as usize).next_power_of_two().max(16),
145            // KeepAll has no inherent ceiling; pick a reasonable
146            // default that the caller can override.
147            Self::KeepAll => 1024,
148        }
149    }
150
151    fn encode(self) -> u64 {
152        match self {
153            Self::KeepLastN(n) => (1u64 << 32) | u64::from(n),
154            Self::KeepAll => 0,
155        }
156    }
157
158    fn decode(raw: u64) -> Self {
159        if raw == 0 {
160            Self::KeepAll
161        } else {
162            let n = (raw & 0xFFFF_FFFF) as u32;
163            Self::KeepLastN(n)
164        }
165    }
166}
167
168/// Runtime-mutable QoS policy. Sidecar policies read these atomics
169/// on every scan; setters publish with `Release`, readers consume
170/// with `Acquire`.
171pub struct QosPolicy {
172    durability_atom: AtomicU32,
173    reliability_atom: AtomicU32,
174    history_atom: AtomicU64,
175    max_latency_nanos: AtomicU64,
176    ordering_atom: AtomicU32,
177}
178
179impl QosPolicy {
180    /// Construct with the four original knobs explicitly. The
181    /// ordering knob starts at [`Ordering::PerProducer`] (the
182    /// composed shapes' native guarantee); declare
183    /// [`Ordering::GlobalFifo`] via
184    /// [`set_ordering`](Self::set_ordering).
185    pub fn new(
186        durability: Durability,
187        reliability: Reliability,
188        history: History,
189        max_latency: Duration,
190    ) -> Self {
191        Self {
192            durability_atom: AtomicU32::new(durability as u32),
193            reliability_atom: AtomicU32::new(reliability as u32),
194            history_atom: AtomicU64::new(history.encode()),
195            max_latency_nanos: AtomicU64::new(
196                max_latency.as_nanos().min(u64::MAX as u128) as u64,
197            ),
198            ordering_atom: AtomicU32::new(Ordering::PerProducer as u32),
199        }
200    }
201
202    /// Default policy: Volatile, BestEffort, KeepLastN(1024),
203    /// max_latency = 100ms. Reasonable for streaming workloads.
204    pub fn streaming_default() -> Self {
205        Self::new(
206            Durability::Volatile,
207            Reliability::BestEffort,
208            History::KeepLastN(1024),
209            Duration::from_millis(100),
210        )
211    }
212
213    /// Reliable-pubsub default: Transient, Reliable, KeepAll,
214    /// max_latency = 1s. Reasonable for cross-process pub/sub where
215    /// every message matters.
216    pub fn reliable_pubsub_default() -> Self {
217        Self::new(
218            Durability::Transient,
219            Reliability::Reliable,
220            History::KeepAll,
221            Duration::from_secs(1),
222        )
223    }
224
225    /// Persistent-log default: Persistent, Reliable, KeepAll,
226    /// max_latency = 5s. Reasonable for durable event logs.
227    pub fn persistent_log_default() -> Self {
228        Self::new(
229            Durability::Persistent,
230            Reliability::Reliable,
231            History::KeepAll,
232            Duration::from_secs(5),
233        )
234    }
235
236    /// Current durability setting.
237    pub fn durability(&self) -> Durability {
238        Durability::from_u32(self.durability_atom.load(AtomOrd::Acquire))
239    }
240
241    /// Current reliability setting.
242    pub fn reliability(&self) -> Reliability {
243        Reliability::from_u32(self.reliability_atom.load(AtomOrd::Acquire))
244    }
245
246    /// Current history setting.
247    pub fn history(&self) -> History {
248        History::decode(self.history_atom.load(AtomOrd::Acquire))
249    }
250
251    /// Current max-latency wish.
252    pub fn max_latency(&self) -> Duration {
253        Duration::from_nanos(self.max_latency_nanos.load(AtomOrd::Acquire))
254    }
255
256    /// Current ordering declaration.
257    pub fn ordering(&self) -> Ordering {
258        Ordering::from_u32(self.ordering_atom.load(AtomOrd::Acquire))
259    }
260
261    /// Replace the durability knob. Sidecars see the change on the
262    /// next scan.
263    pub fn set_durability(&self, durability: Durability) {
264        self.durability_atom.store(durability as u32, AtomOrd::Release);
265    }
266
267    /// Replace the reliability knob.
268    pub fn set_reliability(&self, reliability: Reliability) {
269        self.reliability_atom.store(reliability as u32, AtomOrd::Release);
270    }
271
272    /// Replace the history knob.
273    pub fn set_history(&self, history: History) {
274        self.history_atom.store(history.encode(), AtomOrd::Release);
275    }
276
277    /// Replace the max-latency wish.
278    pub fn set_max_latency(&self, max_latency: Duration) {
279        self.max_latency_nanos.store(
280            max_latency.as_nanos().min(u64::MAX as u128) as u64,
281            AtomOrd::Release,
282        );
283    }
284
285    /// Replace the ordering declaration. Sidecars see the change on
286    /// the next scan and act per the routing in
287    /// [`Ordering`]'s docs (Vyukov morph for unstamped rings, merge
288    /// flag for stamped rings).
289    pub fn set_ordering(&self, ordering: Ordering) {
290        self.ordering_atom.store(ordering as u32, AtomOrd::Release);
291    }
292
293    /// Snapshot: read all five knobs in one method for sidecar use.
294    /// Each load is independently Acquire-ordered; the snapshot is
295    /// NOT a consistent point-in-time view across all five (the
296    /// substrate does not need that property).
297    pub fn snapshot(&self) -> QosSnapshot {
298        QosSnapshot {
299            durability: self.durability(),
300            reliability: self.reliability(),
301            history: self.history(),
302            max_latency: self.max_latency(),
303            ordering: self.ordering(),
304        }
305    }
306}
307
308impl Default for QosPolicy {
309    fn default() -> Self { Self::streaming_default() }
310}
311
312/// Point-in-time snapshot of a `QosPolicy`. Useful for passing to
313/// sidecar policy decisions or for inspecting current settings
314/// without holding a reference to the live atomics.
315#[derive(Debug, Clone, Copy)]
316pub struct QosSnapshot {
317    pub durability: Durability,
318    pub reliability: Reliability,
319    pub history: History,
320    pub max_latency: Duration,
321    pub ordering: Ordering,
322}
323
324impl QosSnapshot {
325    /// Whether this snapshot recommends a locale change relative to
326    /// the currently-active locale.
327    pub fn recommends_locale_change(self, current: Locale) -> Option<Locale> {
328        let recommended = self.durability.recommended_locale();
329        if recommended == current {
330            None
331        } else {
332            Some(recommended)
333        }
334    }
335
336    /// Whether this snapshot's ordering declaration differs from
337    /// the guarantee the caller currently provides. Mirrors
338    /// [`recommends_locale_change`](Self::recommends_locale_change):
339    /// `Some(declared)` means the sidecar should act (Vyukov morph
340    /// or merge-flag flip, per the ring's stampedness).
341    pub fn recommends_ordering_change(self, current: Ordering) -> Option<Ordering> {
342        if self.ordering == current {
343            None
344        } else {
345            Some(self.ordering)
346        }
347    }
348}
349
350#[cfg(test)]
351mod tests {
352    use super::*;
353
354    #[test]
355    fn defaults_load_with_sensible_values() {
356        let qos = QosPolicy::default();
357        assert_eq!(qos.durability(), Durability::Volatile);
358        assert_eq!(qos.reliability(), Reliability::BestEffort);
359        assert!(matches!(qos.history(), History::KeepLastN(_)));
360        assert_eq!(qos.max_latency(), Duration::from_millis(100));
361    }
362
363    #[test]
364    fn setters_update_atomic_fields() {
365        let qos = QosPolicy::default();
366        qos.set_durability(Durability::Persistent);
367        qos.set_reliability(Reliability::Reliable);
368        qos.set_history(History::KeepAll);
369        qos.set_max_latency(Duration::from_secs(10));
370
371        let snap = qos.snapshot();
372        assert_eq!(snap.durability, Durability::Persistent);
373        assert_eq!(snap.reliability, Reliability::Reliable);
374        assert!(matches!(snap.history, History::KeepAll));
375        assert_eq!(snap.max_latency, Duration::from_secs(10));
376    }
377
378    #[test]
379    fn durability_maps_to_locale() {
380        assert_eq!(Durability::Volatile.recommended_locale(), Locale::Anon);
381        assert_eq!(Durability::Transient.recommended_locale(), Locale::ShmFs);
382        assert_eq!(Durability::Persistent.recommended_locale(), Locale::File);
383    }
384
385    #[test]
386    fn history_recommended_capacity_powers_of_two() {
387        assert_eq!(History::KeepLastN(100).recommended_capacity(), 128);
388        assert_eq!(History::KeepLastN(1).recommended_capacity(), 16);
389        assert_eq!(History::KeepLastN(2048).recommended_capacity(), 2048);
390        assert_eq!(History::KeepAll.recommended_capacity(), 1024);
391    }
392
393    #[test]
394    fn snapshot_recommends_locale_change_only_when_different() {
395        let qos = QosPolicy::default();  // Volatile -> Anon
396        let snap = qos.snapshot();
397        assert_eq!(snap.recommends_locale_change(Locale::Anon), None);
398        assert_eq!(snap.recommends_locale_change(Locale::File), Some(Locale::Anon));
399        assert_eq!(snap.recommends_locale_change(Locale::ShmFs), Some(Locale::Anon));
400    }
401
402    #[test]
403    fn ordering_knob_round_trips() {
404        let qos = QosPolicy::default();
405        assert_eq!(qos.ordering(), Ordering::PerProducer,
406                   "ordering must default to the composed shapes' native guarantee");
407        qos.set_ordering(Ordering::GlobalFifo);
408        assert_eq!(qos.ordering(), Ordering::GlobalFifo);
409        assert_eq!(qos.snapshot().ordering, Ordering::GlobalFifo);
410        qos.set_ordering(Ordering::PerProducer);
411        assert_eq!(qos.snapshot().ordering, Ordering::PerProducer);
412    }
413
414    #[test]
415    fn snapshot_recommends_ordering_change_only_when_different() {
416        let qos = QosPolicy::default();
417        let snap = qos.snapshot();
418        assert_eq!(snap.recommends_ordering_change(Ordering::PerProducer), None);
419        assert_eq!(snap.recommends_ordering_change(Ordering::GlobalFifo),
420                   Some(Ordering::PerProducer));
421
422        qos.set_ordering(Ordering::GlobalFifo);
423        let snap = qos.snapshot();
424        assert_eq!(snap.recommends_ordering_change(Ordering::PerProducer),
425                   Some(Ordering::GlobalFifo));
426        assert_eq!(snap.recommends_ordering_change(Ordering::GlobalFifo), None);
427    }
428
429    #[test]
430    fn presets_match_expected_combos() {
431        let streaming = QosPolicy::streaming_default();
432        assert_eq!(streaming.durability(), Durability::Volatile);
433        assert_eq!(streaming.reliability(), Reliability::BestEffort);
434
435        let pubsub = QosPolicy::reliable_pubsub_default();
436        assert_eq!(pubsub.durability(), Durability::Transient);
437        assert_eq!(pubsub.reliability(), Reliability::Reliable);
438        assert!(matches!(pubsub.history(), History::KeepAll));
439
440        let log = QosPolicy::persistent_log_default();
441        assert_eq!(log.durability(), Durability::Persistent);
442        assert_eq!(log.reliability(), Reliability::Reliable);
443    }
444}
subetha_cxc/qos_policy.rs

subetha_cxc/
qos_policy.rs
