embassy_supervisor_observe/
lib.rs

1#![no_std]
2#![forbid(unsafe_code)]
3#![deny(missing_docs)]
4
5//! Observation facade for [`embassy-supervisor`](https://docs.rs/embassy-supervisor).
6//!
7//! This crate defines the minimal traits a signal library must implement so the
8//! supervisor can verify declared dataflow against live behaviour. The signal
9//! library depends only on this tiny crate, and the supervisor depends only on
10//! this crate for its observation hooks — the same layering used by `log` and
11//! `defmt`.
12
13use portable_atomic::{
14    AtomicBool, AtomicI8, AtomicI16, AtomicI32, AtomicI64, AtomicIsize, AtomicU8, AtomicU16,
15    AtomicU32, AtomicU64, AtomicUsize, Ordering,
16};
17
18/// A type whose writes can be observed without interpreting the value.
19///
20/// `change_token()` returns a `u32` whose only meaning is that two unequal
21/// readings prove the signal was written between them. It is allowed to wrap,
22/// and the supervisor never orders or interprets the numeric value.
23///
24/// For entries that feed a node's heartbeat (`observed beat`), the token must
25/// be counting: each write must advance it, because the heartbeat machinery
26/// folds multiple beat entries into a single wrapping sum. Plain value-as-token
27/// works for ordinary observation but is unsuitable for beats.
28pub trait Observable {
29    /// Return a token that changes when the signal is written.
30    fn change_token(&self) -> u32;
31}
32
33/// A signal that can be written through the supervisor's `put` verb.
34pub trait Sink {
35    /// The value type written into the signal.
36    type Item;
37
38    /// Write `v` into the signal.
39    fn put(&self, v: Self::Item);
40}
41
42/// A signal that can be read through the supervisor's `get` verb.
43pub trait Source {
44    /// The value type read from the signal.
45    type Item;
46
47    /// Return a snapshot of the signal's current value.
48    fn get(&self) -> Self::Item;
49}
50
51macro_rules! value_as_token {
52    ($($(#[$cfg:meta])* $ty:ty),* $(,)?) => {$(
53        $(#[$cfg])*
54        impl Observable for $ty {
55            fn change_token(&self) -> u32 {
56                self.load(Ordering::Relaxed) as u32
57            }
58        }
59    )*};
60}
61value_as_token!(
62    AtomicU8,
63    AtomicU16,
64    AtomicU32,
65    AtomicU64,
66    AtomicUsize,
67    AtomicI8,
68    AtomicI16,
69    AtomicI32,
70    AtomicI64,
71    AtomicIsize,
72);
73value_as_token!(
74    #[cfg(target_has_atomic = "8")]
75    core::sync::atomic::AtomicU8,
76    #[cfg(target_has_atomic = "8")]
77    core::sync::atomic::AtomicI8,
78    #[cfg(target_has_atomic = "16")]
79    core::sync::atomic::AtomicU16,
80    #[cfg(target_has_atomic = "16")]
81    core::sync::atomic::AtomicI16,
82    #[cfg(target_has_atomic = "32")]
83    core::sync::atomic::AtomicU32,
84    #[cfg(target_has_atomic = "32")]
85    core::sync::atomic::AtomicI32,
86    #[cfg(target_has_atomic = "64")]
87    core::sync::atomic::AtomicU64,
88    #[cfg(target_has_atomic = "64")]
89    core::sync::atomic::AtomicI64,
90    #[cfg(target_has_atomic = "ptr")]
91    core::sync::atomic::AtomicUsize,
92    #[cfg(target_has_atomic = "ptr")]
93    core::sync::atomic::AtomicIsize,
94);
95
96impl Observable for AtomicBool {
97    fn change_token(&self) -> u32 {
98        self.load(Ordering::Relaxed) as u32
99    }
100}
101
102#[cfg(target_has_atomic = "8")]
103impl Observable for core::sync::atomic::AtomicBool {
104    fn change_token(&self) -> u32 {
105        self.load(core::sync::atomic::Ordering::Relaxed) as u32
106    }
107}
108
109macro_rules! value_signal {
110    ($($(#[$cfg:meta])* $ty:ty => $item:ty),* $(,)?) => {$(
111        $(#[$cfg])*
112        impl Sink for $ty {
113            type Item = $item;
114            #[inline]
115            fn put(&self, v: $item) {
116                self.store(v, Ordering::Relaxed);
117            }
118        }
119        $(#[$cfg])*
120        impl Source for $ty {
121            type Item = $item;
122            #[inline]
123            fn get(&self) -> $item {
124                self.load(Ordering::Relaxed)
125            }
126        }
127    )*};
128}
129value_signal!(
130    AtomicU8 => u8,
131    AtomicU16 => u16,
132    AtomicU32 => u32,
133    AtomicU64 => u64,
134    AtomicUsize => usize,
135    AtomicI8 => i8,
136    AtomicI16 => i16,
137    AtomicI32 => i32,
138    AtomicI64 => i64,
139    AtomicIsize => isize,
140    AtomicBool => bool,
141);
142value_signal!(
143    #[cfg(target_has_atomic = "8")]
144    core::sync::atomic::AtomicU8 => u8,
145    #[cfg(target_has_atomic = "8")]
146    core::sync::atomic::AtomicI8 => i8,
147    #[cfg(target_has_atomic = "8")]
148    core::sync::atomic::AtomicBool => bool,
149    #[cfg(target_has_atomic = "16")]
150    core::sync::atomic::AtomicU16 => u16,
151    #[cfg(target_has_atomic = "16")]
152    core::sync::atomic::AtomicI16 => i16,
153    #[cfg(target_has_atomic = "32")]
154    core::sync::atomic::AtomicU32 => u32,
155    #[cfg(target_has_atomic = "32")]
156    core::sync::atomic::AtomicI32 => i32,
157    #[cfg(target_has_atomic = "64")]
158    core::sync::atomic::AtomicU64 => u64,
159    #[cfg(target_has_atomic = "64")]
160    core::sync::atomic::AtomicI64 => i64,
161    #[cfg(target_has_atomic = "ptr")]
162    core::sync::atomic::AtomicUsize => usize,
163    #[cfg(target_has_atomic = "ptr")]
164    core::sync::atomic::AtomicIsize => isize,
165);
166
167/// A wrapper that counts accesses to an inner signal.
168///
169/// Use this when the wrapped type does not itself advance a count on every
170/// write. `Counted` increments a counter each time [`w`](Self::w) or
171/// [`r`](Self::r) is called, so even a rewrite carrying the same value
172/// registers as an access. This makes it suitable for heartbeat observation,
173/// where the supervisor must see every write.
174///
175/// Calls through [`inner`](Self::inner) are not counted.
176pub struct Counted<T> {
177    inner: T,
178    writes: AtomicU32,
179    reads: AtomicU32,
180}
181
182impl<T> Counted<T> {
183    /// Wrap `inner` with zero read and write counts.
184    pub const fn new(inner: T) -> Self {
185        Self {
186            inner,
187            writes: AtomicU32::new(0),
188            reads: AtomicU32::new(0),
189        }
190    }
191
192    /// Borrow the inner signal, counting this as one write.
193    pub fn w(&self) -> &T {
194        self.writes.fetch_add(1, Ordering::Relaxed);
195        &self.inner
196    }
197
198    /// Borrow the inner signal, counting this as one read.
199    pub fn r(&self) -> &T {
200        self.reads.fetch_add(1, Ordering::Relaxed);
201        &self.inner
202    }
203
204    /// Borrow the inner signal without changing any count.
205    pub fn inner(&self) -> &T {
206        &self.inner
207    }
208
209    /// Return the number of counted writes.
210    pub fn writes(&self) -> u32 {
211        self.writes.load(Ordering::Relaxed)
212    }
213
214    /// Return the number of counted reads.
215    pub fn reads(&self) -> u32 {
216        self.reads.load(Ordering::Relaxed)
217    }
218}
219
220impl<T> Observable for Counted<T> {
221    fn change_token(&self) -> u32 {
222        self.writes()
223    }
224}
225
226impl<T: Sink> Sink for Counted<T> {
227    type Item = T::Item;
228    fn put(&self, v: T::Item) {
229        self.w().put(v);
230    }
231}
232
233impl<T: Source> Source for Counted<T> {
234    type Item = T::Item;
235    fn get(&self) -> T::Item {
236        self.r().get()
237    }
238}
239
240#[cfg(test)]
241mod tests {
242    use super::*;
243
244    #[test]
245    fn counted_counts_w_and_r_but_not_inner() {
246        let c = Counted::new(5u8);
247        assert_eq!((c.writes(), c.reads()), (0, 0));
248        c.w();
249        c.w();
250        c.r();
251        c.inner();
252        assert_eq!((c.writes(), c.reads()), (2, 1));
253    }
254
255    #[test]
256    fn counted_token_is_the_write_counter() {
257        let c = Counted::new(());
258        assert_eq!(c.change_token(), 0);
259        c.r();
260        assert_eq!(c.change_token(), 0);
261        c.w();
262        assert_eq!(c.change_token(), 1);
263    }
264
265    #[test]
266    fn counted_hands_back_the_inner_signal() {
267        let c = Counted::new(AtomicU32::new(0));
268        c.w().store(7, Ordering::Relaxed);
269        assert_eq!(c.inner().load(Ordering::Relaxed), 7);
270    }
271
272    #[test]
273    fn sink_and_source_move_values_through_atomics() {
274        let a = AtomicU32::new(0);
275        Sink::put(&a, 7);
276        assert_eq!(Source::get(&a), 7);
277        let b = core::sync::atomic::AtomicI32::new(-1);
278        Sink::put(&b, 5);
279        assert_eq!(Source::get(&b), 5);
280    }
281
282    #[test]
283    fn counted_forwards_and_counts_values() {
284        let c = Counted::new(AtomicU32::new(0));
285        c.put(3);
286        assert_eq!(c.get(), 3);
287        assert_eq!((c.writes(), c.reads()), (1, 1));
288        assert_eq!(c.change_token(), 1, "the count is the token, not the value");
289    }
290
291    #[test]
292    fn atomics_use_the_value_as_token() {
293        let a = AtomicU32::new(0);
294        assert_eq!(a.change_token(), 0);
295        a.store(3, Ordering::Relaxed);
296        assert_eq!(a.change_token(), 3);
297        a.store(3, Ordering::Relaxed);
298        assert_eq!(a.change_token(), 3);
299
300        let wide = AtomicU64::new(u64::MAX);
301        assert_eq!(wide.change_token(), u32::MAX);
302
303        let b = AtomicBool::new(false);
304        b.store(true, Ordering::Relaxed);
305        assert_eq!(b.change_token(), 1);
306    }
307}
embassy_supervisor_observe/lib.rs

embassy_supervisor_observe/
lib.rs
