embassy_supervisor/data_deps/
backed.rs

1use core::cell::RefCell;
2use core::ops::Deref;
3use core::task::Poll;
4
5use embassy_sync::blocking_mutex::Mutex;
6use embassy_sync::blocking_mutex::raw::CriticalSectionRawMutex;
7use embassy_sync::signal::Signal;
8use embassy_sync::waitqueue::MultiWakerRegistration;
9use embassy_time::Duration;
10use portable_atomic::{AtomicBool, AtomicU32, Ordering};
11
12use super::{Gated, producer_of};
13use crate::{Coupling, Sig, TaskNode};
14
15static SERVING_EVT: Mutex<CriticalSectionRawMutex, RefCell<MultiWakerRegistration<4>>> =
16    Mutex::new(RefCell::new(MultiWakerRegistration::new()));
17
18const GATE_RETRY: embassy_time::Duration = embassy_time::Duration::from_millis(250);
19
20pub(crate) fn notify_serving() {
21    SERVING_EVT.lock(|w| w.borrow_mut().wake());
22}
23
24fn serving(producer: &TaskNode) -> bool {
25    producer.is_running() && producer.is_ready()
26}
27
28/// Wait until the producer is serving, its running state changes, or the
29/// retry interval passes. Returns true if serving. If false, the caller may
30/// request a start so a stopped producer is retried promptly.
31async fn wait_serving(producer: &'static TaskNode) -> bool {
32    use embassy_futures::select::select;
33    let running = producer.is_running();
34    let settled = || serving(producer) || producer.is_running() != running;
35    let woke = core::future::poll_fn(|cx| {
36        if settled() {
37            return Poll::Ready(());
38        }
39        SERVING_EVT.lock(|w| w.borrow_mut().register(cx.waker()));
40        // Registered-then-recheck closes the race against a concurrent
41        // `set_ready` or `ack_dropped`: after this the event cannot fire unseen.
42        if settled() {
43            Poll::Ready(())
44        } else {
45            Poll::Pending
46        }
47    });
48    let _ = select(woke, embassy_time::Timer::after(GATE_RETRY)).await;
49    serving(producer)
50}
51
52/// A signal whose producer is started by the first reader that calls
53/// [`open`](crate::TaskNode::open). The value is not handed out until the
54/// producer is running and ready. The gate counts readers so the producer can
55/// retire once none are left ([`unwatched`](Self::unwatched), [`TaskNode::retire`]).
56#[repr(C)]
57pub struct Backed<T> {
58    inner: T,
59    requested: AtomicBool,
60    openers: AtomicU32,
61    watch: Signal<CriticalSectionRawMutex, ()>,
62}
63
64impl<T> Backed<T> {
65    /// Wrap `inner` as a backed signal.
66    pub const fn new(inner: T) -> Self {
67        Self {
68            inner,
69            requested: AtomicBool::new(false),
70            openers: AtomicU32::new(0),
71            watch: Signal::new(),
72        }
73    }
74
75    /// How many [`Open`] guards are alive right now.
76    pub fn openers(&self) -> u32 {
77        self.openers.load(Ordering::Acquire)
78    }
79
80    /// Resolve once no reader has held the gate for `cooldown`, continuously.
81    pub async fn unwatched(&self, cooldown: Duration) {
82        use embassy_futures::select::{Either, select};
83        loop {
84            while self.openers() > 0 {
85                self.watch.wait().await;
86            }
87            self.watch.reset();
88            if let Either::First(()) =
89                select(embassy_time::Timer::after(cooldown), self.watch.wait()).await
90                && self.openers() == 0
91            {
92                return;
93            }
94        }
95    }
96
97    fn admit_reader(&self) {
98        if self.openers.fetch_add(1, Ordering::AcqRel) == 0 {
99            self.watch.signal(());
100        }
101    }
102
103    fn drop_reader(&self) {
104        if self.openers.fetch_sub(1, Ordering::AcqRel) == 1 {
105            self.watch.signal(());
106        }
107    }
108}
109
110/// A reader's hold on a [`Backed`] signal, from [`open`](crate::TaskNode::open).
111/// `Deref` gives the wrapped signal's API; dropping the guard lets the producer
112/// notice the last reader has left.
113pub struct Open<T: 'static> {
114    target: &'static Backed<T>,
115}
116
117impl<T> Open<T> {
118    /// The wrapped signal with the `'static` lifetime `Deref` cannot lend, for
119    /// APIs taking `&'static self` such as [`Leased::lease`](crate::Leased::lease).
120    /// It outlives the guard, so keeping it past the drop escapes the count.
121    pub fn signal(&self) -> &'static T {
122        &self.target.inner
123    }
124}
125
126impl<T> Deref for Open<T> {
127    type Target = T;
128    fn deref(&self) -> &T {
129        &self.target.inner
130    }
131}
132
133impl<T> Drop for Open<T> {
134    fn drop(&mut self) {
135        self.target.drop_reader();
136    }
137}
138
139impl<T> Deref for Backed<T> {
140    type Target = T;
141    fn deref(&self) -> &T {
142        &self.inner
143    }
144}
145
146impl<T: Sync + 'static> Gated for Backed<T> {
147    type Handle = Open<T>;
148
149    fn admit(&'static self) -> Open<T> {
150        self.admit_reader();
151        Open { target: self }
152    }
153
154    async fn ensure(&'static self, caller: &'static TaskNode, entry: &'static Coupling) {
155        let Some(producer) = producer_of(caller, entry) else {
156            warn!(
157                "supervisor: {} is gated but no graph declares a writer for it",
158                entry.name()
159            );
160            return;
161        };
162        // Name the waits that cannot resolve on their own, once per open.
163        if !producer.is_running() {
164            // `Activate` re-enables an OnDemand node but never spawns it —
165            // that is its pool policy's job.
166            #[cfg(feature = "control")]
167            if matches!(producer.mode(), crate::Mode::OnDemand) {
168                warn!(
169                    "supervisor: {} gates on OnDemand {}, which a control start \
170                     re-enables but does not spawn: this read returns only once \
171                     its pool grows it",
172                    entry.name(),
173                    producer.name()
174                );
175            }
176            #[cfg(not(feature = "control"))]
177            warn!(
178                "supervisor: {} gates on {}, which is not running, and without \
179                 `control` there is nothing that can start it: this read will \
180                 not return",
181                entry.name(),
182                producer.name()
183            );
184        }
185        loop {
186            if serving(producer) {
187                break;
188            }
189            #[cfg(feature = "control")]
190            if !producer.is_running() && !self.requested.swap(true, Ordering::Relaxed) {
191                crate::request_control(producer, crate::ControlOp::Activate).await;
192            }
193            if !wait_serving(producer).await {
194                self.requested.store(false, Ordering::Relaxed);
195            }
196        }
197        self.requested.store(false, Ordering::Relaxed);
198    }
199}
200
201impl TaskNode {
202    /// Wait until no reader has held `s` for `cooldown`, then clear readiness
203    /// and, with `control`, request deactivation. If a reader arrives during
204    /// the wait, restart the cooldown. The readiness handshake keeps the stop
205    /// race-free: readers admitted after `clear_ready` wait for the next
206    /// activation instead of reading a producer that is shutting down.
207    pub async fn retire<T: Sync>(&'static self, s: Sig<Backed<T>>, cooldown: Duration) {
208        loop {
209            s.target.unwatched(cooldown).await;
210            self.clear_ready();
211            if s.target.openers() == 0 {
212                break;
213            }
214            self.set_ready();
215        }
216        #[cfg(feature = "control")]
217        crate::request_control(self, crate::ControlOp::Deactivate).await;
218    }
219}
220
221#[cfg(feature = "coupling-observe")]
222impl<T: crate::Observable> crate::Observable for Backed<T> {
223    fn change_token(&self) -> u32 {
224        self.inner.change_token()
225    }
226}
embassy_supervisor/data_deps/backed.rs

embassy_supervisor/data_deps/
backed.rs
