metrale_telemetry/
seqcell.rs

1// SPDX-License-Identifier: MIT OR Apache-2.0
2
3//! 2026-09-26: A sequence-locked cell of `W` words: readers take no lock and
4//! never see a mix of two stores.
5//!
6//! A device sample is [`crate::device::SAMPLE_WORDS`] words describing one
7//! instant; one atomic per word would let a reader pair one sample's power
8//! with another sample's energy counter. A writer makes the sequence word
9//! odd, stores the words and makes it even; a reader retries until it saw the
10//! same even value before and after reading. Every access is an atomic.
11//!
12//! Owner: telemetry.
13//! Invariants: `load` returns only a value that a single `store` wrote in full.
14
15use std::sync::atomic::{AtomicU64, Ordering, fence};
16
17#[derive(Debug)]
18pub struct SeqCell<const W: usize> {
19    seq: AtomicU64,
20    words: [AtomicU64; W],
21}
22
23impl<const W: usize> Default for SeqCell<W> {
24    fn default() -> Self {
25        Self::new()
26    }
27}
28
29impl<const W: usize> SeqCell<W> {
30    pub const fn new() -> Self {
31        Self {
32            seq: AtomicU64::new(0),
33            words: [const { AtomicU64::new(0) }; W],
34        }
35    }
36
37    /// 2026-09-26: Replace the value. Concurrent writers serialise on the
38    /// sequence word (a CAS from even to odd), so two writers cannot
39    /// interleave words.
40    pub fn store(&self, value: [u64; W]) {
41        let mut s = self.seq.load(Ordering::Relaxed);
42        loop {
43            if s & 1 == 1 {
44                std::hint::spin_loop();
45                s = self.seq.load(Ordering::Relaxed);
46                continue;
47            }
48            match self
49                .seq
50                .compare_exchange_weak(s, s + 1, Ordering::Acquire, Ordering::Relaxed)
51            {
52                Ok(_) => break,
53                Err(now) => s = now,
54            }
55        }
56        fence(Ordering::Release);
57        for (w, v) in self.words.iter().zip(value) {
58            w.store(v, Ordering::Relaxed);
59        }
60        self.seq.store(s + 2, Ordering::Release);
61    }
62
63    /// 2026-09-26: A consistent copy, and the number of stores it reflects
64    /// (0 = never written).
65    pub fn load(&self) -> (u64, [u64; W]) {
66        loop {
67            let s1 = self.seq.load(Ordering::Acquire);
68            if s1 & 1 == 1 {
69                std::hint::spin_loop();
70                continue;
71            }
72            let mut out = [0u64; W];
73            for (o, w) in out.iter_mut().zip(&self.words) {
74                *o = w.load(Ordering::Relaxed);
75            }
76            fence(Ordering::Acquire);
77            if self.seq.load(Ordering::Relaxed) == s1 {
78                return (s1 / 2, out);
79            }
80        }
81    }
82}
83
84#[cfg(test)]
85mod tests {
86    use super::*;
87    use std::sync::Arc;
88    use std::sync::atomic::AtomicBool;
89
90    #[test]
91    fn a_fresh_cell_reports_no_stores() {
92        let c = SeqCell::<3>::new();
93        assert_eq!(c.load(), (0, [0; 3]));
94        c.store([1, 2, 3]);
95        assert_eq!(c.load(), (1, [1, 2, 3]));
96    }
97
98    /// 2026-09-26: Every value written has all words equal, so any mixture of
99    /// two values shows as unequal words. Two writers and four readers race
100    /// over 200k stores; a single torn read fails the test.
101    #[test]
102    fn concurrent_readers_never_see_a_torn_value() {
103        const W: usize = 15;
104        let cell = Arc::new(SeqCell::<W>::new());
105        let stop = Arc::new(AtomicBool::new(false));
106        let writers: Vec<_> = (0..2u64)
107            .map(|id| {
108                let cell = cell.clone();
109                std::thread::spawn(move || {
110                    for k in 0..100_000u64 {
111                        let v = k * 2 + id + 1;
112                        cell.store([v; W]);
113                    }
114                })
115            })
116            .collect();
117        let readers: Vec<_> = (0..4)
118            .map(|_| {
119                let (cell, stop) = (cell.clone(), stop.clone());
120                std::thread::spawn(move || {
121                    let mut reads = 0u64;
122                    while !stop.load(Ordering::Relaxed) {
123                        let (_, w) = cell.load();
124                        assert!(w.iter().all(|x| *x == w[0]), "torn read: {w:?}");
125                        reads += 1;
126                    }
127                    reads
128                })
129            })
130            .collect();
131        for w in writers {
132            w.join().unwrap();
133        }
134        stop.store(true, Ordering::Relaxed);
135        let reads: u64 = readers.into_iter().map(|r| r.join().unwrap()).sum();
136        assert!(reads > 0, "the readers ran");
137        assert_eq!(cell.load().0, 200_000, "every store is counted once");
138    }
139}
metrale_telemetry/seqcell.rs

metrale_telemetry/
seqcell.rs
