embassy_supervisor/lib.rs
1// `no_std` for the shipped crate and the embedded build; under `cargo test` the
2// crate is built for the host, where the test harness and the unit tests need `std`.
3#![cfg_attr(not(test), no_std)]
4#![forbid(unsafe_code)]
5#![deny(missing_docs)]
6//! # embassy-supervisor — a task-lifecycle supervisor for [embassy](https://embassy.dev)
7//!
8//! Application- and HAL-agnostic primitives for orchestrating a set of embassy
9//! tasks: bringing them up in dependency order, tearing them down in reverse,
10//! scaling an elastic worker pool with load, placing nodes on interrupt-priority
11//! tiers or a second core, and starting/stopping/pausing/resuming individual
12//! tasks at runtime while keeping the dependency graph consistent. The supervisor
13//! orchestrates task *lifecycle* and leaves the rest — allocation, HAL, power,
14//! what the tasks do — to the application.
15//!
16//! ## The model
17//!
18//! * The graph is declared once with the [`supervisor_graph!`] macro: each
19//! managed task becomes a [`TaskNode`] `static`, and the macro bundles the node
20//! slots, dependency table, and a topological order computed **at compile time**
21//! into a single [`Graph`] (`GRAPH`). The whole graph is validated at compile
22//! time — a dependency cycle, an unknown or duplicate dependency, a duplicate
23//! name, or bad pool bounds are compile errors.
24//! * [`Supervisor::new`] takes `&GRAPH` (no work, no failure) and uses the order
25//! to bring tasks up in dependency order ([`Supervisor::start`]) and tear them
26//! down in reverse ([`Supervisor::teardown`]).
27//! * `executor NAME;` items declare runtime-filled [`SpawnerSlot`]s, and
28//! `executor: NAME` on a node (or a whole pool) routes its spawn through one —
29//! an interrupt-priority tier or the second core. Bring-up *awaits* the slot
30//! (bounded), so an executor that comes up late — or on another core — is a
31//! rendezvous, not a race.
32//! * Two flags span every lifecycle operation: **disabled** (stopped until an
33//! explicit `Activate` — declared `disabled` in the graph or control-stopped;
34//! see [`TaskNode::set_disabled`]) and **detached** (self-managed: after
35//! [`TaskNode::set_detached`] no supervisor operation touches the node).
36//! * Each node carries a `TaskHandle` of per-node atomic flags and
37//! single-consumer `Signal`s. Every node is single-instance — no counts, no
38//! fan-out. See [`TaskHandle`].
39//!
40//! ## Three lifecycles, distinguished by [`Mode`]
41//!
42//! * [`Mode::Terminate`] — the task exits its loop on shutdown and is respawned
43//! on the next bring-up. Stateless services (a network listener, a logger).
44//! * [`Mode::Pause`] — the task acks the shutdown then parks on
45//! `wait_resume()`; it is resumed in place, never respawned. Tasks that
46//! retain a resource across the pause (an open peripheral handle, a socket).
47//! * [`Mode::OnDemand`] — like `Terminate`, but not started at boot and not
48//! auto-respawned; the supervisor brings it up and down at runtime to scale
49//! an elastic worker pool ([`ElasticPool`]) with load.
50//!
51//! ## Writing a supervised task
52//!
53//! A supervised task is a plain `#[embassy_executor::task]` whose first parameter
54//! is its node (the macro's `spawn:` glue passes it; extra arguments come from the
55//! partial-call spawn form). Four rules cover the task side of the protocol:
56//!
57//! 1. select long-lived work against [`TaskNode::wait_shutdown`] — that's how a
58//! stop reaches you;
59//! 2. ack exactly once per stop with [`TaskNode::ack_dropped`]: on exit
60//! (`Terminate`/`OnDemand`), or on each pause (`Pause`) *before* parking on
61//! [`TaskNode::wait_resume`];
62//! 3. an autonomous exit acks too, so the supervisor sees the node as down;
63//! 4. resources follow the mode: a `Terminate` task re-acquires everything on
64//! respawn (drop-on-exit is the cleanup), a `Pause` task keeps what it holds
65//! across the park.
66//!
67//! Pool workers additionally report load with [`TaskNode::mark_busy`] /
68//! [`TaskNode::mark_idle`] (a real transition fires the scale signal itself), and
69//! a self-managed daemon or run-once job opts out of supervision with
70//! [`TaskNode::set_detached`]. The README's *Writing supervised tasks* section has
71//! per-mode skeletons.
72//!
73//! ## What the supervisor does *not* do
74//!
75//! * It does not model any power-state transition (sleep/wake): it reacts to
76//! "teardown" and "bring-up" requests; the application drives them.
77//! * It does not allocate, and does no work at construction: the topological
78//! sort runs at compile time (see the `supervisor_graph!` macro).
79//! * It does not observe task internals. Tasks self-report their drop state via
80//! `ack_dropped()`; a task that fails to ack within a timeout panics the
81//! supervisor with the offending node's name.
82//!
83//! ## Cargo features
84//!
85//! * `control` *(default)* — the runtime control plane: [`ControlOp`],
86//! [`request_control`], [`Supervisor::apply_control`].
87//! * `pool` *(default)* — elastic worker pools: [`ElasticPool`],
88//! [`Supervisor::run_pools`], and the `pools` field of [`Graph`].
89//! * `defmt` — route the supervisor's logs through `defmt`; without it the log
90//! macros are no-ops.
91//! * `trace` family (all opt-in) — `trace`: the [`trace`] recorders consuming
92//! embassy-executor's `_embassy_trace_*` hooks; `trace-hooks`:
93//! `supervisor_graph!` also *defines* the hook symbols; `trace-names`: node
94//! names stamped into task Metadata for external consumers; `trace-nested`:
95//! preemption-exact accounting (a nested higher-tier poll credits its time
96//! back to the window it interrupted).
97//!
98//! Build with `default-features = false` for a minimal core that only does
99//! dependency-ordered bring-up/teardown (drops the control plane and pools,
100//! trimming flash and a couple of statics).
101//!
102//! ## Example
103//!
104//! [`supervisor_graph!`] declares the whole graph once — it generates the node
105//! `static`s and a single [`Graph`] value `GRAPH` bundling the node slots, dep
106//! table, and compile-time topological order (a dependency cycle is a compile
107//! error), which [`Supervisor::new`] consumes.
108//!
109//! ```ignore
110//! use embassy_executor::Spawner;
111//! use embassy_supervisor::{supervisor_graph, Supervisor, wait_control};
112//!
113//! // `app` depends on `net`; each `spawn:` names a task fn spawned with the node.
114//! supervisor_graph! {
115//! node NET = Terminate, deps: [], spawn: net_task;
116//! node APP = Terminate, deps: [NET], spawn: app_task;
117//! }
118//!
119//! #[embassy_executor::task]
120//! async fn supervisor_task(spawner: Spawner) {
121//! let sup = Supervisor::new(&GRAPH);
122//! sup.start(spawner).await.expect("initial spawn"); // brings up `net`, then `app`
123//! loop {
124//! // Apply runtime start/stop/pause/resume requests in dependency order.
125//! let cmd = wait_control().await;
126//! sup.apply_control(cmd, spawner).await;
127//! }
128//! // With the `pool` feature you'd instead drive scaling and control together:
129//! // `select(sup.run_pools(spawner), wait_control())` (see the `firmware` crate).
130//! }
131//! ```
132//!
133//! The `firmware` crate in the [repository](https://github.com/cedrivard/embassy-supervisor)
134//! is a complete working example (USB-net, an HTTP control plane, an elastic pool,
135//! and OTA).
136
137#[macro_use]
138mod fmt;
139
140use core::cell::Cell;
141use core::sync::atomic::Ordering;
142
143use embassy_executor::{SendSpawner, SpawnError, Spawner};
144use embassy_futures::select::{Either, select};
145use embassy_sync::blocking_mutex::Mutex as BlockingMutex;
146use embassy_sync::blocking_mutex::raw::CriticalSectionRawMutex;
147#[cfg(feature = "control")]
148use embassy_sync::channel::Channel;
149use embassy_sync::signal::Signal;
150use embassy_time::{Timer, with_timeout};
151use portable_atomic::AtomicBool;
152#[cfg(feature = "trace")]
153use portable_atomic::AtomicU32;
154
155// ─── Scale-request signal (task → supervisor) ──────────────────────────────
156//
157// Elastic pool workers fire this when their busy/idle status changes; the
158// supervisor's `run_pools` loop awaits it and re-runs the pool policies
159// (`ElasticPool`). Single-consumer `Signal`: many tasks may `signal()`, only the
160// supervisor `wait()`s. This is the *only* path by which task status reaches the
161// supervisor — it never polls.
162#[cfg(feature = "pool")]
163static SCALE_REQ: Signal<CriticalSectionRawMutex, ()> = Signal::new();
164
165/// Fire the scale-request signal. Called by a task on a busy/idle transition.
166/// A no-op when the `pool` feature is disabled (no pools to re-evaluate).
167pub fn request_scale() {
168 #[cfg(feature = "pool")]
169 SCALE_REQ.signal(());
170}
171
172/// Await the next scale request. The supervisor's driver loop selects this
173/// against its other wake sources and runs the scaling policy on each wake.
174#[cfg(feature = "pool")]
175pub async fn wait_scale() {
176 SCALE_REQ.wait().await;
177}
178
179// ─── Runtime control commands (app → supervisor) ───────────────────────────
180//
181// An application's control surface (e.g. a network endpoint) usually can't drive
182// the supervisor directly: the `Supervisor` and the `Spawner` live on the
183// supervisor task's stack, not in a `static`. So control is decoupled via this
184// channel — the caller `request_control()`s a (node, op) pair and returns
185// immediately; the supervisor's driver loop `wait_control()`s it and runs the
186// dependency-honoring `apply_control`. A `Channel` (not a `Signal`) so
187// back-to-back requests aren't coalesced; capacity 4 is ample for hand-driven
188// control and a full channel simply drops the surplus (`try_send`).
189
190/// Which way to drive a node. Higher-level verbs fold onto these two:
191/// `start`/`resume` → `Activate`, `stop`/`pause` → `Deactivate`. The concrete
192/// mechanism (respawn vs resume vs leave-to-pool) is then chosen per node `Mode`
193/// by the supervisor when it applies the command ([`Supervisor::apply_control`]).
194#[cfg(feature = "control")]
195#[derive(Clone, Copy, PartialEq, Eq, Debug)]
196pub enum ControlOp {
197 /// Bring the node up (start a stopped `Terminate` node, resume a `Pause` node).
198 Activate,
199 /// Take the node down (and its dependents, per the graph).
200 Deactivate,
201}
202
203/// A runtime control request: drive `node` (and, per the dependency graph and
204/// pool membership, the nodes it implies) in the `op` direction.
205#[cfg(feature = "control")]
206#[derive(Clone, Copy, Debug)]
207pub struct ControlCommand {
208 /// The node to drive.
209 pub node: &'static TaskNode,
210 /// The direction to drive it.
211 pub op: ControlOp,
212}
213
214/// App → supervisor control mailbox. `&'static TaskNode` is `Copy + Sync`, so
215/// the target rides the channel directly — no name lookup needed supervisor-side.
216#[cfg(feature = "control")]
217static CONTROL_REQ: Channel<CriticalSectionRawMutex, ControlCommand, 4> = Channel::new();
218
219/// Enqueue a control request. Non-blocking; drops if the mailbox is full (4
220/// outstanding), which is harmless for low-frequency manual control. Called by
221/// the application's control surface.
222#[cfg(feature = "control")]
223pub fn request_control(node: &'static TaskNode, op: ControlOp) {
224 let _ = CONTROL_REQ.try_send(ControlCommand { node, op });
225}
226
227/// Await the next control request. Selected by the supervisor's driver loop
228/// against pool scaling and any other application wake sources.
229#[cfg(feature = "control")]
230pub async fn wait_control() -> ControlCommand {
231 CONTROL_REQ.receive().await
232}
233
234/// Per-node timeout for `wait_dropped`. A task that doesn't ack within this
235/// window is a bug (e.g. a missing `ack_dropped()` call) and panics the
236/// supervisor with the offending node's name. 2 s comfortably exceeds a typical
237/// task's poll period and peripheral settle time.
238const SHUTDOWN_ACK_TIMEOUT_MS: u64 = 2_000;
239
240/// How long the supervisor's bring-up waits for a node's `executor:`
241/// [`SpawnerSlot`] to be filled before failing the spawn with
242/// [`SpawnError::Busy`]. A genuine cross-core rendezvous resolves in microseconds;
243/// a slot empty this long is a misconfiguration (the app never registered that
244/// executor's spawner). Bounded, so a misconfigured graph fails loudly instead of
245/// hanging bring-up forever.
246const SLOT_READY_TIMEOUT: embassy_time::Duration = embassy_time::Duration::from_millis(100);
247
248// ─── Mode ────────────────────────────────────────────────────────────────
249
250/// Lifecycle policy for a managed task: what the task does on shutdown and what
251/// the supervisor does to bring it back.
252#[derive(Clone, Copy, PartialEq, Eq, Debug)]
253pub enum Mode {
254 /// Task exits its loop on shutdown. The supervisor respawns it via the
255 /// node's `spawn` fn from `respawn_terminate`.
256 Terminate,
257 /// Task acks shutdown and parks on `wait_resume()`. The supervisor resumes
258 /// it from `resume_pausable`; the task is never respawned, so it keeps any
259 /// resource it holds (a peripheral handle, a socket) across the pause.
260 Pause,
261 /// Like `Terminate` (exits on shutdown), but **not** started at boot and
262 /// **not** auto-respawned. The supervisor brings it up and down at runtime
263 /// via `start_node` / `stop_node` in response to load — see [`ElasticPool`].
264 /// `start()` skips it; `respawn_terminate()` leaves it down (it
265 /// re-grows under demand); `teardown()` only acts on it while it is running.
266 OnDemand,
267}
268
269impl Mode {
270 /// Stable lower-case wire name, used both for serialization (e.g. a JSON
271 /// task-state view) and for `defmt` logging — the single source of these
272 /// strings.
273 pub fn as_str(&self) -> &'static str {
274 match self {
275 Mode::Terminate => "terminate",
276 Mode::Pause => "pause",
277 Mode::OnDemand => "ondemand",
278 }
279 }
280}
281
282#[cfg(feature = "defmt")]
283impl defmt::Format for Mode {
284 fn format(&self, f: defmt::Formatter) {
285 defmt::write!(f, "{}", self.as_str());
286 }
287}
288
289// ─── TaskHandle ──────────────────────────────────────────────────────────
290
291/// Coordination state for one task. Embedded inside [`TaskNode`].
292///
293/// Every node is single-instance, so each field is a per-node atomic flag or a
294/// single-consumer signal — no counts, no fan-out. Written by one side (task or
295/// supervisor) and read by the other:
296/// * `shutdown` / `shutdown_wake` — supervisor requests exit; the task parks
297/// on the signal and reads the flag.
298/// * `dropped` / `dropped_wake` — the task acks its exit; the supervisor
299/// parks on the signal (with a timeout) and reads the flag.
300/// * `resume_wake` — supervisor resumes a parked Pause-mode task.
301/// * `running` — supervisor's record that the node is spawned; `busy` — the
302/// task's active/idle status. Both read by the elastic scaling policy.
303/// * `disabled` — the node has been manually deactivated; see below.
304pub struct TaskHandle {
305 /// Set true by the supervisor when shutdown is requested.
306 /// Cleared by `reset()` before the next spawn.
307 shutdown: AtomicBool,
308 /// Wake source for `wait_shutdown()`. Fired by `signal_shutdown()`.
309 shutdown_wake: Signal<CriticalSectionRawMutex, ()>,
310 /// Set true by the instance when it acks the shutdown (a bool, not a count,
311 /// since every node is single-instance). Cleared by `reset()`.
312 dropped: AtomicBool,
313 /// Wake source for `wait_dropped()`. Fired by `ack_dropped()`.
314 dropped_wake: Signal<CriticalSectionRawMutex, ()>,
315 /// True while the supervisor has the node spawned and it hasn't exited.
316 /// Always-on nodes are set true by `start()`; `OnDemand` nodes are set
317 /// true/false by `start_node()` / `stop_node()`. `teardown()` only acts on
318 /// `running` nodes, so a down `OnDemand` node doesn't stall it.
319 running: AtomicBool,
320 /// True while the task is actively serving (its active/idle status). Set by
321 /// `mark_busy()` / `mark_idle()`; read by the scaling policy.
322 busy: AtomicBool,
323 /// Wake source for `wait_resume()` on Pause-mode tasks. Fired by
324 /// `signal_resume()`.
325 resume_wake: Signal<CriticalSectionRawMutex, ()>,
326 /// True while the node has been manually deactivated (stopped/paused) via the
327 /// runtime control interface (`Supervisor::deactivate`). Unlike the other
328 /// flags this one is **lifecycle-spanning**: it is *not* cleared by
329 /// `reset()`, so a manual stop "sticks" — the automatic bring-up paths
330 /// (`start`, `respawn_terminate`, `resume_pausable`, and the elastic pool's
331 /// grow) skip a node while it is set. Cleared only by `Supervisor::activate`.
332 /// Because it lives in a `static`, it also survives a power-state transition
333 /// that retains RAM (e.g. a warm-resume from deep sleep).
334 disabled: AtomicBool,
335 /// Self-managed: while set, the supervisor never drives this node — teardown,
336 /// deactivate/activate, `stop_node`, respawn, and pause-resume all skip it. Not
337 /// cleared by `reset()`. Full rationale on [`TaskNode::set_detached`].
338 detached: AtomicBool,
339 /// The executor task id currently running this node (`TaskRef::id()`, captured
340 /// from the `SpawnToken` by the macro's spawn glue). `0` = unknown (not yet
341 /// spawned, or a parked/closure-spawned node that never registered). Overwritten
342 /// on every (re)spawn, so — unlike an external tracker — it stays correct across
343 /// respawns without any unlinking.
344 #[cfg(feature = "trace")]
345 task_id: AtomicU32,
346 /// Accumulated executor-poll time for this node, in embassy-time ticks,
347 /// wrapping. Consumers sample twice and `wrapping_sub` to get a rate; the
348 /// crate does no windowing.
349 #[cfg(feature = "trace")]
350 exec_ticks: AtomicU32,
351 /// Number of executor polls of this node, wrapping.
352 #[cfg(feature = "trace")]
353 polls: AtomicU32,
354 /// Longest single poll ever observed, in ticks — the "never yields" watermark.
355 /// A large value names the node that hogged the executor even after the fact,
356 /// which a live check cannot do from the blocked executor itself.
357 #[cfg(feature = "trace")]
358 max_poll_ticks: AtomicU32,
359}
360
361impl TaskHandle {
362 const fn new(disabled_at_boot: bool) -> Self {
363 Self {
364 shutdown: AtomicBool::new(false),
365 shutdown_wake: Signal::new(),
366 dropped: AtomicBool::new(false),
367 dropped_wake: Signal::new(),
368 running: AtomicBool::new(false),
369 busy: AtomicBool::new(false),
370 resume_wake: Signal::new(),
371 disabled: AtomicBool::new(disabled_at_boot),
372 detached: AtomicBool::new(false),
373 #[cfg(feature = "trace")]
374 task_id: AtomicU32::new(0),
375 #[cfg(feature = "trace")]
376 exec_ticks: AtomicU32::new(0),
377 #[cfg(feature = "trace")]
378 polls: AtomicU32::new(0),
379 #[cfg(feature = "trace")]
380 max_poll_ticks: AtomicU32::new(0),
381 }
382 }
383}
384
385// ─── Executor spawner slots ──────────────────────────────────────────────
386
387/// A runtime-filled slot holding the [`SendSpawner`] of an executor other than
388/// the one the supervisor runs on — an `InterruptExecutor` tier, the second
389/// core's executor, any foreign thread executor (via `Spawner::make_send()`).
390///
391/// Declared by the `executor NAME;` item of [`supervisor_graph!`]; nodes carrying
392/// `executor: NAME` are spawned through the slot instead of the supervisor's own
393/// `Spawner`. The application fills it once at startup — before, or concurrently
394/// with, [`Supervisor::start`] (e.g. from the second core's bring-up):
395///
396/// ```ignore
397/// static EXECUTOR_HIGH: InterruptExecutor = InterruptExecutor::new();
398/// HIGH.set(EXECUTOR_HIGH.start(interrupt::SWI_IRQ_0));
399/// sup.start(spawner).await?; // nodes declared `executor: HIGH` spawn on that tier
400/// ```
401///
402/// The supervisor's bring-up (`start` / `start_node` / `respawn_terminate`) awaits
403/// [`ready`](Self::ready) for a node's slot before spawning it, so a tier filled
404/// late — or from another core — is handled without a race; a slot still empty after
405/// the supervisor's bounded wait fails the spawn with [`SpawnError::Busy`] rather
406/// than silently dropping the task. Spawned futures must be `Send` (a non-`Send`
407/// `executor:` task is a compile error at the glue).
408pub struct SpawnerSlot {
409 slot: BlockingMutex<CriticalSectionRawMutex, Cell<Option<SendSpawner>>>,
410 /// Wakes a `ready()` waiter when `set` fills the slot (cross-core safe:
411 /// `Signal` is critical-section based and latches).
412 filled: Signal<CriticalSectionRawMutex, ()>,
413}
414
415impl SpawnerSlot {
416 /// An empty slot (`const` — it lives in a `static` the macro emits).
417 pub const fn new() -> Self {
418 Self {
419 slot: BlockingMutex::new(Cell::new(None)),
420 filled: Signal::new(),
421 }
422 }
423
424 /// Fill the slot (last set wins) and wake a [`ready`](Self::ready) waiter.
425 /// Call before [`Supervisor::start`] — or from the other core's bring-up,
426 /// with the supervisor awaiting `ready()`.
427 pub fn set(&self, spawner: SendSpawner) {
428 self.slot.lock(|c| c.set(Some(spawner)));
429 self.filled.signal(());
430 }
431
432 /// The registered spawner, or `None` while unfilled.
433 pub fn get(&self) -> Option<SendSpawner> {
434 self.slot.lock(Cell::get)
435 }
436
437 /// Await the slot and return the spawner. The rendezvous primitive: the
438 /// supervisor's bring-up awaits this for a node's `executor:` slot before
439 /// spawning it (bounded, see [`Supervisor::start`]), so a tier filled late — or
440 /// from another core — is handled without a race. Returns immediately once the
441 /// slot is filled, so any number of *late* callers are fine (an application can
442 /// gate work on the executor being up). While the slot is still empty, at most
443 /// one task should be parked here: the underlying `Signal` holds a single waker,
444 /// so a second pre-fill waiter would displace the first.
445 pub async fn ready(&self) -> SendSpawner {
446 loop {
447 if let Some(sp) = self.get() {
448 return sp;
449 }
450 // `Signal` latches: a `set()` racing between the check above and
451 // this wait still wakes us.
452 self.filled.wait().await;
453 }
454 }
455}
456
457impl Default for SpawnerSlot {
458 fn default() -> Self {
459 Self::new()
460 }
461}
462
463// ─── TaskNode ────────────────────────────────────────────────────────────
464
465/// A node in the supervisor's task graph.
466///
467/// Designed to live in `static` memory: every field is `Sync`, all constructors
468/// are `const`. Declared by [`supervisor_graph!`], which emits one per managed
469/// task along with the [`Graph`] (`GRAPH`) that [`Supervisor::new`] consumes.
470pub struct TaskNode {
471 /// Human-readable name. Used in defmt logs and panic messages.
472 pub name: &'static str,
473 /// Lifecycle policy. See [`Mode`].
474 pub mode: Mode,
475 /// App-provided spawn function (typically an inline closure at the node's
476 /// declaration). Called once at boot from `Supervisor::start`, again from
477 /// `respawn_terminate` for Terminate nodes, and at runtime from `start_node`
478 /// for `OnDemand` nodes. `None` for a **parked** node the application spawns
479 /// itself (e.g. a `Pause` sensor holding a peripheral handle): the supervisor
480 /// tracks its lifecycle but never spawns it.
481 pub spawn: Option<fn(Spawner) -> Result<(), SpawnError>>,
482 /// The executor [`SpawnerSlot`] this node spawns through (`executor: NAME` in
483 /// the graph), or `None` to spawn on the supervisor's own `Spawner`. When
484 /// `Some`, the supervisor awaits the slot's [`ready`](SpawnerSlot::ready)
485 /// (bounded by [`SLOT_READY_TIMEOUT`]) *before* invoking `spawn`, so the
486 /// generated glue's own non-blocking `SpawnerSlot::get` is already filled. Set
487 /// by the macro via [`with_executor`](Self::with_executor); `const`, zero-cost.
488 spawn_slot: Option<&'static SpawnerSlot>,
489 handle: TaskHandle,
490}
491
492impl TaskNode {
493 /// A single-instance node started at boot (`Terminate`/`Pause`) or on demand
494 /// (`Mode::OnDemand`). Every node is single-instance; an elastic service is
495 /// modelled as several `OnDemand` nodes of the same pooled task fn.
496 ///
497 /// A `TaskNode` carries only its own identity and behaviour; the graph's
498 /// dependency edges live in the compile-time index table that
499 /// [`supervisor_graph!`] emits and [`Supervisor::new`] consumes.
500 /// `disabled_at_boot` seeds the node's disabled flag so a control-started node
501 /// (e.g. an OTA task) can be declared down and started later via a control op.
502 /// `spawn` is `None` for a parked node the application spawns itself.
503 pub const fn new(
504 name: &'static str,
505 mode: Mode,
506 spawn: Option<fn(Spawner) -> Result<(), SpawnError>>,
507 disabled_at_boot: bool,
508 ) -> Self {
509 Self {
510 name,
511 mode,
512 spawn,
513 spawn_slot: None,
514 handle: TaskHandle::new(disabled_at_boot),
515 }
516 }
517
518 /// Route this node's spawn through the given executor [`SpawnerSlot`] (the
519 /// `executor: NAME` graph annotation). The supervisor awaits the slot before
520 /// spawning the node, so a tier filled late — or from another core — is handled
521 /// without a race, and the generated glue's non-blocking `get` is already filled.
522 /// `const` and chainable in a `static` initializer; emitted by [`supervisor_graph!`].
523 pub const fn with_executor(mut self, slot: &'static SpawnerSlot) -> Self {
524 self.spawn_slot = Some(slot);
525 self
526 }
527
528 // ── Task-side API ────────────────────────────────────────────────────
529 //
530 // Called from inside the `#[embassy_executor::task] async fn` body. The
531 // whole task-side protocol is four rules (the README's "Writing supervised
532 // tasks" section has per-mode skeletons):
533 // 1. select long-lived work against `wait_shutdown()`;
534 // 2. `ack_dropped()` exactly once per stop — on exit (Terminate/OnDemand)
535 // or on each pause (Pause), before parking on `wait_resume()`;
536 // 3. an autonomous exit acks too;
537 // 4. resources follow the mode: Terminate re-acquires on respawn, Pause
538 // retains across park.
539
540 /// True iff the supervisor has requested shutdown. Checked at the loop top
541 /// alongside `wait_shutdown()` in a `select`.
542 pub fn shutdown_requested(&self) -> bool {
543 self.handle.shutdown.load(Ordering::Acquire)
544 }
545
546 /// Park until shutdown is requested. Returns immediately if shutdown has
547 /// already been requested. Use this for single-instance tasks in a `select`
548 /// against the task's main work future.
549 pub async fn wait_shutdown(&self) {
550 // Fast path — already requested. (Important because the signal is
551 // edge-triggered: if `signal()` fired before we got here, the bare
552 // `wait()` below would block forever.)
553 if self.handle.shutdown.load(Ordering::Acquire) {
554 return;
555 }
556 self.handle.shutdown_wake.wait().await;
557 }
558
559 /// Mark this instance as having shut down: clears the running flag and acks
560 /// the teardown handshake (so the supervisor's `wait_dropped` completes).
561 /// Every instance must call this exactly once on exit (Terminate/OnDemand
562 /// mode) or on each pause (Pause mode). It also covers an **autonomous** exit
563 /// the supervisor didn't request — e.g. a pool worker backing off — so the
564 /// pool sees the instance as down and can re-grow it under later demand.
565 pub fn ack_dropped(&self) {
566 self.handle.running.store(false, Ordering::Release);
567 self.handle.dropped.store(true, Ordering::Release);
568 self.handle.dropped_wake.signal(());
569 }
570
571 /// Pause-mode only: park until the supervisor signals resume. Call *after*
572 /// [`ack_dropped`](Self::ack_dropped) — ack the pause, then park; held
573 /// resources stay owned across the park.
574 pub async fn wait_resume(&self) {
575 self.handle.resume_wake.wait().await;
576 }
577
578 /// Report that this task started serving a request (active). Fires the
579 /// scale-request signal on a real idle→busy transition so the scaling policy
580 /// can react (e.g. grow the pool); a redundant call doesn't re-signal.
581 pub fn mark_busy(&self) {
582 if !self.handle.busy.swap(true, Ordering::Release) {
583 request_scale();
584 }
585 }
586
587 /// Report that this task finished serving and is idle again. Fires the
588 /// scale-request signal on a real busy→idle transition so the scaling policy
589 /// can react (e.g. shrink the pool); a redundant call doesn't re-signal.
590 pub fn mark_idle(&self) {
591 if self.handle.busy.swap(false, Ordering::Release) {
592 request_scale();
593 }
594 }
595
596 /// True while this task is actively serving. Read by the scaling policy.
597 pub fn is_busy(&self) -> bool {
598 self.handle.busy.load(Ordering::Acquire)
599 }
600
601 /// True while the supervisor has this node spawned (and it hasn't exited).
602 /// Read by the scaling policy to count live instances, and by a task-state
603 /// view.
604 pub fn is_running(&self) -> bool {
605 self.handle.running.load(Ordering::Acquire)
606 }
607
608 /// True while the node is disabled: declared `disabled` in the graph
609 /// (stopped-at-boot, up on an explicit `Activate`), or manually deactivated
610 /// via the control interface and not yet re-activated. Read by a task-state
611 /// view and by the automatic bring-up paths (which skip a disabled node).
612 pub fn is_disabled(&self) -> bool {
613 self.handle.disabled.load(Ordering::Acquire)
614 }
615
616 /// Mark/clear this node as **detached**: a self-managing node the supervisor
617 /// brings up once (via [`start`](Supervisor::start)) and then stops managing
618 /// **entirely**. Every runtime lifecycle operation skips a detached node: full
619 /// [`teardown`](Supervisor::teardown), the control deactivate/activate cascades,
620 /// [`stop_node`](Supervisor::stop_node), [`respawn_terminate`](Supervisor::respawn_terminate),
621 /// and pause-resume. It keeps running (or, for a one-shot, stays exited) across a
622 /// teardown/wake cycle instead of being stopped, re-enabled, or re-spawned. Use it
623 /// for a task that must outlive the teardown it participates in — e.g. a sleep/power
624 /// coordinator that tears the graph down, sleeps, then wakes it — or a self-managed
625 /// one-shot whose `deps:` exist only for start-ordering. The node owns its own
626 /// shutdown; the supervisor will not drive it.
627 pub fn set_detached(&self, detached: bool) {
628 self.handle.detached.store(detached, Ordering::Release);
629 }
630
631 /// True while this node is [detached](Self::set_detached): self-managed, skipped by
632 /// every runtime lifecycle operation (teardown, deactivate/activate, `stop_node`,
633 /// respawn, pause-resume). Only the initial `start` brings it up.
634 pub fn is_detached(&self) -> bool {
635 self.handle.detached.load(Ordering::Acquire)
636 }
637
638 // ── Trace/observability API (features `trace`/`trace-names`) ───────────
639
640 /// Record the executor task id (`SpawnToken::id()` / `TaskRef::id()`) currently
641 /// backing this node, so the [`trace`] recorders can attribute executor polls to
642 /// it. Called automatically by the spawn glue `supervisor_graph!` generates;
643 /// call it manually only for a **parked** node (no `spawn:`) or a verbatim-closure
644 /// `spawn:`, where the macro cannot see the token. Overwrites on every (re)spawn.
645 #[cfg(feature = "trace")]
646 pub fn set_task_id(&self, id: u32) {
647 self.handle.task_id.store(id, Ordering::Release);
648 }
649
650 /// Register an externally-spawned token as this node's live task: records
651 /// the task id for the [`trace`] recorders and (feature `trace-names`)
652 /// stamps the node name into the task Metadata. One call replaces the
653 /// manual [`set_task_id`](Self::set_task_id) dance wherever the macro can't
654 /// see the token — parked nodes and verbatim-closure `spawn:` forms:
655 ///
656 /// ```ignore
657 /// let t = environment_task(i2c_dev)?;
658 /// BME280.adopt(&t);
659 /// high_spawner.spawn(t);
660 /// ```
661 #[cfg(feature = "trace")]
662 pub fn adopt<S>(&self, token: &embassy_executor::SpawnToken<S>) {
663 self.set_task_id(token.id());
664 #[cfg(feature = "trace-names")]
665 token.metadata().set_name(self.name);
666 }
667
668 /// The executor task id last recorded by [`set_task_id`](Self::set_task_id)
669 /// (`0` = never spawned / not registered).
670 #[cfg(feature = "trace")]
671 pub fn task_id(&self) -> u32 {
672 self.handle.task_id.load(Ordering::Acquire)
673 }
674
675 /// Accumulated executor-poll time of this node, in embassy-time ticks. Wrapping:
676 /// sample twice and `wrapping_sub` the readings to get a rate over a window.
677 #[cfg(feature = "trace")]
678 pub fn exec_ticks(&self) -> u32 {
679 self.handle.exec_ticks.load(Ordering::Relaxed)
680 }
681
682 /// Number of executor polls of this node (wrapping counter).
683 #[cfg(feature = "trace")]
684 pub fn poll_count(&self) -> u32 {
685 self.handle.polls.load(Ordering::Relaxed)
686 }
687
688 /// Longest single executor poll of this node ever observed, in ticks — the
689 /// "never yields" watermark. A poll is expected to be microseconds; a large
690 /// value names the node that hogged its executor, even after the fact.
691 #[cfg(feature = "trace")]
692 pub fn max_poll_ticks(&self) -> u32 {
693 self.handle.max_poll_ticks.load(Ordering::Relaxed)
694 }
695
696 // ── Supervisor-side API ──────────────────────────────────────────────
697 //
698 // Driven by the `Supervisor` struct. Kept `pub(crate)` so app code doesn't
699 // accidentally bypass the supervisor's orchestration.
700
701 pub(crate) fn signal_shutdown(&self) {
702 self.handle.shutdown.store(true, Ordering::Release);
703 self.handle.shutdown_wake.signal(());
704 }
705
706 pub(crate) fn signal_resume(&self) {
707 self.handle.resume_wake.signal(());
708 }
709
710 pub(crate) fn set_running(&self, running: bool) {
711 self.handle.running.store(running, Ordering::Release);
712 }
713
714 /// Set/clear the manual-deactivation flag. Set by `Supervisor::deactivate`,
715 /// cleared by `Supervisor::activate`. Deliberately *not* touched by
716 /// `reset()`, so a manual stop survives respawn cycles and RAM-retaining
717 /// power-state transitions.
718 ///
719 /// Public so an application can pre-disable a `Terminate` node *before*
720 /// `Supervisor::start`, making it a stopped-at-boot task that only comes up on
721 /// an explicit `Activate` control (a node started by control rather than at boot).
722 pub fn set_disabled(&self, disabled: bool) {
723 self.handle.disabled.store(disabled, Ordering::Release);
724 }
725
726 /// Wait until the instance has called `ack_dropped()`. Single-instance, so
727 /// one ack ends the wait. The fast-path flag check handles the ack landing
728 /// before this await (the `dropped_wake` signal is edge-triggered).
729 pub(crate) async fn wait_dropped(&self) {
730 if self.handle.dropped.load(Ordering::Acquire) {
731 return;
732 }
733 self.handle.dropped_wake.wait().await;
734 }
735
736 /// Clear the shutdown flag, dropped flag, busy flag, and the shutdown /
737 /// dropped wake-signals so the next cycle starts clean. Doesn't touch
738 /// `running` (managed around spawn/stop), `resume_wake` (`resume_pausable`
739 /// fires that for Pause nodes), or `disabled` (lifecycle-spanning).
740 pub(crate) fn reset(&self) {
741 self.handle.shutdown.store(false, Ordering::Release);
742 self.handle.dropped.store(false, Ordering::Release);
743 self.handle.busy.store(false, Ordering::Release);
744 self.handle.shutdown_wake.reset();
745 self.handle.dropped_wake.reset();
746 }
747}
748
749/// Manual impl: the private `TaskHandle` (Signals + atomics) has no `Debug`, and a
750/// snapshot of the *live* flags is more useful than raw handle internals anyway.
751/// `finish_non_exhaustive` marks the elided fields (`spawn`, the handle).
752impl core::fmt::Debug for TaskNode {
753 fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
754 f.debug_struct("TaskNode")
755 .field("name", &self.name)
756 .field("mode", &self.mode)
757 .field("running", &self.is_running())
758 .field("busy", &self.is_busy())
759 .field("disabled", &self.is_disabled())
760 .field("detached", &self.is_detached())
761 .finish_non_exhaustive()
762 }
763}
764
765// ─── Graph ───────────────────────────────────────────────────────────────
766
767/// The compile-time task graph produced by [`supervisor_graph!`]: the node slots,
768/// the dependency-index table, the topological order, and the elastic pools — the
769/// single value [`Supervisor::new`] consumes. The macro emits one `pub static GRAPH`
770/// of this type. The fields are public so the application can read them directly
771/// (e.g. a status endpoint iterating `GRAPH.nodes` / `GRAPH.deps`).
772///
773/// `N` is capped at 256 (graph indices are `u8`); the macro enforces this at
774/// expansion time.
775pub struct Graph<const N: usize> {
776 /// Node slots, one per declared node. `None` marks a `#[cfg]`-ed-out node.
777 pub nodes: &'static [Option<&'static TaskNode>; N],
778 /// Per-node dependency indices into `nodes` (`deps[i]` lists node `i`'s deps).
779 pub deps: &'static [&'static [u8]; N],
780 /// Topologically sorted indices into `nodes` (dependencies before dependents;
781 /// reverse iteration is the teardown order). A dependency cycle is a compile error.
782 pub order: [u8; N],
783 /// Elastic worker pools to register with the supervisor (empty when unused).
784 #[cfg(feature = "pool")]
785 pub pools: &'static [&'static dyn Pool],
786}
787
788// ─── Supervisor ──────────────────────────────────────────────────────────
789
790/// Orchestrates a set of managed tasks across spawn / teardown / bring-up.
791///
792/// Owned by a single supervisor task. Concurrent access from other tasks goes
793/// through each [`TaskNode`]'s own atomic state, not the `Supervisor` struct.
794pub struct Supervisor<const N: usize> {
795 /// Node slots, one per declared node. `None` marks a slot whose node was
796 /// `#[cfg]`-ed out of the build (feature-gated); every method skips those.
797 nodes: &'static [Option<&'static TaskNode>],
798 /// Per-node dependency indices into `nodes` (`deps[i]` lists the indices of
799 /// the nodes that node `i` depends on). The single runtime source of graph
800 /// topology, generated alongside `order` by the `supervisor_graph!` macro.
801 deps: &'static [&'static [u8]],
802 /// Topologically sorted indices into `nodes`: dependencies before their
803 /// dependents; reverse iteration is the teardown order. Precomputed at
804 /// compile time (a cycle is a compile error), so construction does no work.
805 order: [u8; N],
806 /// Elastic pools, so the control interface can co-control a whole pool from
807 /// any one member (`apply_control` expands the target through
808 /// [`Pool::members`]) — the same registry `run_pools` drives. Taken from
809 /// `GRAPH.pools` at construction (empty when no pool is declared).
810 #[cfg(feature = "pool")]
811 pools: &'static [&'static dyn Pool],
812}
813
814/// Await a node's `executor:` [`SpawnerSlot`] (if it has one), bounded by
815/// [`SLOT_READY_TIMEOUT`]. A slot still empty after the wait yields
816/// [`SpawnError::Busy`] — a loud misconfiguration, not a silent hang. A node with no
817/// slot returns immediately, so a same-executor bring-up never touches the timer.
818async fn await_spawn_slot(node: &'static TaskNode) -> Result<(), SpawnError> {
819 if let Some(slot) = node.spawn_slot {
820 with_timeout(SLOT_READY_TIMEOUT, slot.ready())
821 .await
822 .map_err(|_| SpawnError::Busy)?;
823 }
824 Ok(())
825}
826
827impl<const N: usize> Supervisor<N> {
828 /// Build a supervisor from a precomputed [`Graph`] — the `GRAPH` that
829 /// `supervisor_graph!` emits (node slots, dependency-index table, compile-time
830 /// topological `order`, and the elastic pools). A dependency cycle is a
831 /// *compile* error, so construction is infallible and does no work —
832 /// `start` / `teardown` / `respawn_terminate` just iterate.
833 pub const fn new(graph: &'static Graph<N>) -> Self {
834 Self {
835 nodes: graph.nodes,
836 deps: graph.deps,
837 order: graph.order,
838 #[cfg(feature = "pool")]
839 pools: graph.pools,
840 }
841 }
842
843 /// Spawn every boot node in dependency order. Called once at boot.
844 /// `Mode::OnDemand` nodes are skipped — they're brought up at runtime by
845 /// `start_node`. A **parked** node (no `spawn` fn) is spawned externally by
846 /// `main()` (with hardware handles main owns); it's still marked `running`
847 /// here. Disabled nodes, and `#[cfg]`-ed-out slots, are skipped.
848 ///
849 /// Async because an `executor: NAME` node first awaits its [`SpawnerSlot::ready`]
850 /// (bounded by `SLOT_READY_TIMEOUT` — the rendezvous with a tier or second core
851 /// that comes up asynchronously); a slot still empty at the deadline fails the
852 /// bring-up with [`SpawnError::Busy`]. A node with no `executor:` slot never
853 /// touches the timer.
854 pub async fn start(&self, spawner: Spawner) -> Result<(), SpawnError> {
855 // Register the node slots with the trace recorders.
856 #[cfg(feature = "trace")]
857 trace::register_graph(self.nodes);
858
859 for i in self.order.iter() {
860 let Some(node) = self.nodes[*i as usize] else {
861 continue;
862 };
863 if matches!(node.mode, Mode::OnDemand) || node.is_disabled() {
864 continue;
865 }
866 info!("supervisor: spawning {} ({})", node.name, node.mode);
867 if let Some(spawn) = node.spawn {
868 // For an `executor:` node, wait (bounded) for its slot to be filled
869 // before spawning; a same-executor node has no slot, so this is an
870 // immediate no-op and the bring-up loop stays tight.
871 await_spawn_slot(node).await?;
872 spawn(spawner)?;
873 }
874 node.set_running(true);
875 }
876 Ok(())
877 }
878
879 /// Start a single node at runtime — e.g. growing an elastic pool. Resets the
880 /// handle, spawns one instance via the node's `spawn` fn (which must launch
881 /// exactly one), and marks it `running`. Returns `SpawnError::Busy` if the
882 /// underlying embassy task pool is exhausted (the ceiling), which the caller
883 /// treats as "can't grow".
884 pub async fn start_node(
885 &self,
886 node: &'static TaskNode,
887 spawner: Spawner,
888 ) -> Result<(), SpawnError> {
889 node.reset();
890 if let Some(spawn) = node.spawn {
891 await_spawn_slot(node).await?;
892 spawn(spawner)?;
893 }
894 node.set_running(true);
895 info!("supervisor: started {}", node.name);
896 Ok(())
897 }
898
899 /// Signal `node` to shut down, wait for its ack (panicking on timeout — a
900 /// missing `ack_dropped()` somewhere), then clear `running`. Shared by
901 /// `stop_node` and `teardown`; the caller must have checked `is_running`.
902 async fn shutdown_and_wait(&self, node: &'static TaskNode) {
903 node.signal_shutdown();
904 if let Either::Second(()) = select(
905 node.wait_dropped(),
906 Timer::after_millis(SHUTDOWN_ACK_TIMEOUT_MS),
907 )
908 .await
909 {
910 panic!(
911 "supervisor: task {} did not ack shutdown within {}ms",
912 node.name, SHUTDOWN_ACK_TIMEOUT_MS,
913 );
914 }
915 node.set_running(false);
916 }
917
918 /// Stop a single running node at runtime — e.g. shrinking an elastic pool.
919 /// Signals shutdown, waits for the ack, clears `running`. No-op if the node
920 /// isn't running, or is [detached](TaskNode::set_detached) (self-managed — the
921 /// supervisor never stops it). Panics if it doesn't ack within the timeout.
922 pub async fn stop_node(&self, node: &'static TaskNode) {
923 if !node.is_running() || node.is_detached() {
924 return;
925 }
926 self.shutdown_and_wait(node).await;
927 info!("supervisor: stopped {}", node.name);
928 }
929
930 /// Signal every **running** node to shut down in **reverse** topological
931 /// order, awaiting each node's ack before moving to its dependency. Down
932 /// `OnDemand` nodes are skipped (no instance to ack). Pause-mode nodes ack
933 /// and park on `wait_resume()`; Terminate/OnDemand nodes exit. Panics if a
934 /// running node fails to ack within `SHUTDOWN_ACK_TIMEOUT_MS`.
935 pub async fn teardown(&self) {
936 for i in self.order.iter().rev() {
937 let Some(node) = self.nodes[*i as usize] else {
938 continue;
939 };
940 if !node.is_running() {
941 continue;
942 }
943 // A detached node is self-managed; never tear it down. See
944 // [`TaskNode::set_detached`].
945 if node.is_detached() {
946 continue;
947 }
948 info!("supervisor: tearing down {}", node.name);
949 self.shutdown_and_wait(node).await;
950 }
951 }
952
953 /// Signal every Pause-mode node to resume. Cheap and synchronous — the tasks
954 /// were parked on `wait_resume()` and pick up immediately. Called separately
955 /// from `respawn_terminate` so the application can fire resume independently
956 /// of the respawn step. Disabled (manually-paused) nodes are skipped so a
957 /// manual pause sticks, and detached (self-managed) Pause nodes are left
958 /// parked; there is intentionally no dependency gate here.
959 pub fn resume_pausable(&self) {
960 for i in self.order.iter() {
961 let Some(node) = self.nodes[*i as usize] else {
962 continue;
963 };
964 if matches!(node.mode, Mode::Pause) && !node.is_disabled() && !node.is_detached() {
965 node.reset();
966 info!("supervisor: resuming {}", node.name);
967 node.signal_resume();
968 node.set_running(true);
969 }
970 }
971 }
972
973 /// Reset and re-spawn every Terminate-mode node in dependency order.
974 /// Pause-mode nodes are untouched (use `resume_pausable`); `OnDemand` nodes
975 /// are left down — they re-grow under load via `start_node`. Disabled nodes
976 /// are skipped so a manual stop sticks across the bring-up. Detached nodes are
977 /// skipped too: `teardown` never brought them down, so they are still running
978 /// and re-spawning would double-spawn them (see [`TaskNode::set_detached`]). The
979 /// reset happens before the spawn so newly-running tasks see a clean handle.
980 pub async fn respawn_terminate(&self, spawner: Spawner) -> Result<(), SpawnError> {
981 for i in self.order.iter() {
982 let Some(node) = self.nodes[*i as usize] else {
983 continue;
984 };
985 if matches!(node.mode, Mode::Terminate) && !node.is_disabled() && !node.is_detached() {
986 node.reset();
987 info!("supervisor: respawning {}", node.name);
988 if let Some(spawn) = node.spawn {
989 await_spawn_slot(node).await?;
990 spawn(spawner)?;
991 }
992 node.set_running(true);
993 }
994 }
995 Ok(())
996 }
997}
998
999// ─── Runtime control (dependency- and pool-honoring start/stop) ────────────
1000//
1001// The `apply_control` entry point drives one `ControlCommand` from the
1002// application's control surface. Unlike the pool's bare `start_node`/`stop_node`,
1003// these honor the graph: a stop cascades through dependents (so nothing is left
1004// running without a dependency), a start cascades through deps (so nothing comes
1005// up before what it needs), and either expands across a whole `ElasticPool` so
1006// the pool is controlled as a unit. A manual stop/pause also sets the
1007// lifecycle-spanning `disabled` flag, so it sticks against the elastic policy and
1008// the wake respawn.
1009
1010// Graph-index helpers used by BOTH the control plane and the pool driver, so they
1011// are gated on either feature — `pool` alone (no `control`) must still compile.
1012#[cfg(any(feature = "control", feature = "pool"))]
1013impl<const N: usize> Supervisor<N> {
1014 /// Position of `node` in `self.nodes` (pointer identity — every node is a
1015 /// `&'static`). `None` only if the node isn't in this graph (impossible for
1016 /// targets sourced from `GRAPH.nodes`; treated as a no-op by callers).
1017 fn index_of(&self, node: &'static TaskNode) -> Option<usize> {
1018 self.nodes
1019 .iter()
1020 .position(|n| n.is_some_and(|x| core::ptr::eq(x, node)))
1021 }
1022
1023 /// Whether every dependency of `node` is currently running, resolved through
1024 /// the graph's index table. The pool driver checks this before growing a
1025 /// worker, so a pool member is never spawned while one of its dependencies is
1026 /// down.
1027 #[cfg(feature = "pool")]
1028 pub(crate) fn deps_running(&self, node: &'static TaskNode) -> bool {
1029 match self.index_of(node) {
1030 Some(i) => self.deps[i]
1031 .iter()
1032 .all(|&di| self.nodes[di as usize].is_some_and(|n| n.is_running())),
1033 None => false,
1034 }
1035 }
1036}
1037
1038#[cfg(feature = "control")]
1039impl<const N: usize> Supervisor<N> {
1040 /// Seed a membership set with `target` plus — if `target` belongs to an
1041 /// elastic pool — every member of that pool, so control is applied to the
1042 /// whole pool atomically. Pool membership is read from `GRAPH.pools`; with no
1043 /// pools (the `pool` feature off, or none declared) this is just `{target}`.
1044 fn seed(&self, target: &'static TaskNode, set: &mut [bool; N]) {
1045 if let Some(i) = self.index_of(target) {
1046 set[i] = true;
1047 }
1048 #[cfg(feature = "pool")]
1049 for pool in self.pools {
1050 let members = pool.members();
1051 if members.iter().any(|m| core::ptr::eq(*m, target)) {
1052 for m in members {
1053 if let Some(i) = self.index_of(m) {
1054 set[i] = true;
1055 }
1056 }
1057 }
1058 }
1059 }
1060
1061 /// Apply one control command, honoring pool membership and the dependency
1062 /// graph. Run from the supervisor's driver loop (never concurrently with
1063 /// itself), so the cascade is atomic from the application's perspective.
1064 pub async fn apply_control(&self, cmd: ControlCommand, spawner: Spawner) {
1065 match cmd.op {
1066 ControlOp::Deactivate => self.deactivate(cmd.node).await,
1067 ControlOp::Activate => self.activate(cmd.node, spawner).await,
1068 }
1069 }
1070
1071 /// Bring `target` (and its pool, and every transitive dependent) down, in
1072 /// reverse-topological order so each dependent stops before the dependency it
1073 /// relies on. Marks the whole set `disabled` so the stop sticks against the
1074 /// elastic policy and the wake respawn until a matching `activate`.
1075 async fn deactivate(&self, target: &'static TaskNode) {
1076 let mut set = [false; N];
1077 self.seed(target, &mut set);
1078
1079 // Grow the set to include transitive dependents. `order` is
1080 // dependency-first, so when we reach a node its deps are already decided;
1081 // a node joins if any dep it declares is already in the set.
1082 for i in self.order.iter() {
1083 let j = *i as usize;
1084 if set[j] {
1085 continue;
1086 }
1087 let Some(node) = self.nodes[j] else {
1088 continue;
1089 };
1090 // A detached node declares its dep only for start ordering and intends
1091 // to outlive it, so it's never pulled into the cascade.
1092 if node.is_detached() {
1093 continue;
1094 }
1095 if self.deps[j].iter().any(|&di| set[di as usize]) {
1096 set[j] = true;
1097 }
1098 }
1099
1100 // Tear down in reverse topo order (dependents before their deps).
1101 for i in self.order.iter().rev() {
1102 let j = *i as usize;
1103 if !set[j] {
1104 continue;
1105 }
1106 let Some(node) = self.nodes[j] else {
1107 continue;
1108 };
1109 // A detached node is self-managed — never control-stop it. The growth loop
1110 // keeps detached *dependents* out of the set; this also covers a detached
1111 // node that was seeded directly (or a detached pool member). Without it a
1112 // detached one-shot that already exited (stale `is_running`, no ack path)
1113 // would be signalled a shutdown it can never acknowledge, panicking here.
1114 if node.is_detached() {
1115 continue;
1116 }
1117 node.set_disabled(true);
1118 if node.is_running() {
1119 info!("supervisor: control-stop {}", node.name);
1120 self.shutdown_and_wait(node).await;
1121 }
1122 }
1123 }
1124
1125 /// Bring `target` (and its pool, and every transitive dependency) up, in
1126 /// topological order so each dependency starts before its dependent. Clears
1127 /// `disabled` across the set. `OnDemand` (pool) members are only re-enabled,
1128 /// not force-spawned — the elastic policy re-grows them under load, which is
1129 /// the whole point of the pool.
1130 async fn activate(&self, target: &'static TaskNode, spawner: Spawner) {
1131 let mut set = [false; N];
1132 self.seed(target, &mut set);
1133
1134 // Grow the set to include transitive deps. Walk dependents-first
1135 // (reverse topo); when a set member is seen, pull in its direct deps.
1136 // A detached member's `deps:` are start-ordering only (the node is
1137 // self-managed), so don't expand from it — mirrors deactivate's guard;
1138 // otherwise activating a detached target would un-disable deps that
1139 // were independently disabled.
1140 for i in self.order.iter().rev() {
1141 let j = *i as usize;
1142 if set[j] && !self.nodes[j].is_some_and(|n| n.is_detached()) {
1143 for &di in self.deps[j] {
1144 set[di as usize] = true;
1145 }
1146 }
1147 }
1148
1149 // Bring up in topo order (deps before dependents).
1150 for i in self.order.iter() {
1151 let j = *i as usize;
1152 if !set[j] {
1153 continue;
1154 }
1155 let Some(node) = self.nodes[j] else {
1156 continue;
1157 };
1158 // A detached node is self-managed — the supervisor never re-enables or
1159 // re-starts it, even when it is a dependency of an activated target.
1160 if node.is_detached() {
1161 continue;
1162 }
1163 node.set_disabled(false);
1164 if node.is_running() {
1165 continue;
1166 }
1167 match node.mode {
1168 Mode::Terminate => {
1169 info!("supervisor: control-start {}", node.name);
1170 // SpawnError::Busy (pool exhausted) → can't start, skip.
1171 let _ = self.start_node(node, spawner).await;
1172 }
1173 Mode::Pause => {
1174 info!("supervisor: control-resume {}", node.name);
1175 node.reset();
1176 node.signal_resume();
1177 node.set_running(true);
1178 }
1179 // Pool worker — leave it down; the elastic policy regrows it on
1180 // demand now that `disabled` is cleared.
1181 Mode::OnDemand => {}
1182 }
1183 }
1184 }
1185}
1186
1187// ─── Topological sort (Kahn's algorithm, const) ───────────────────────────
1188//
1189// Computes the topological order at *compile time* over a per-node
1190// dependency-index table; a dependency cycle is a compile error.
1191
1192/// Topologically sort a graph given as a per-node dependency-index table.
1193///
1194/// `deps[i]` lists the indices of the nodes that node `i` depends on; the result
1195/// lists node indices in dependency-first order (a dependency appears before its
1196/// dependents). The supervisor iterates it forward for `start` /
1197/// `respawn_terminate` and in reverse for `teardown`.
1198///
1199/// Evaluated at compile time by the code `supervisor_graph!` generates — a
1200/// dependency **cycle is a compile error** (the `panic!` fires during const
1201/// evaluation). `#[doc(hidden)]`: an engine for the macro, not a user-facing API.
1202///
1203/// Supports at most 256 nodes: indices are `u8`, so a larger `N` would truncate.
1204/// The macro rejects bigger graphs at expansion; the assert below is defense in
1205/// depth for a manual caller (a const-eval panic, i.e. a compile error).
1206#[doc(hidden)]
1207#[must_use]
1208pub const fn topo_sort_const<const N: usize>(deps: &[&'static [u8]; N]) -> [u8; N] {
1209 assert!(
1210 N <= 256,
1211 "supervisor graph exceeds 256 node slots (indices are u8)"
1212 );
1213 // in_degree[i] = number of deps of node i not yet resolved.
1214 let mut in_degree = [0u8; N];
1215 let mut i = 0;
1216 while i < N {
1217 in_degree[i] = deps[i].len() as u8;
1218 i += 1;
1219 }
1220
1221 // Queue (fixed array, head/tail indices) seeded with the dependency-free nodes.
1222 let mut queue = [0u8; N];
1223 let mut tail = 0;
1224 i = 0;
1225 while i < N {
1226 if in_degree[i] == 0 {
1227 queue[tail] = i as u8;
1228 tail += 1;
1229 }
1230 i += 1;
1231 }
1232
1233 let mut order = [0u8; N];
1234 let mut produced = 0;
1235 let mut head = 0;
1236 while head < tail {
1237 let node = queue[head] as usize;
1238 head += 1;
1239 order[produced] = node as u8;
1240 produced += 1;
1241
1242 // Decrement the in-degree of every node that depends on `node`.
1243 let mut j = 0;
1244 while j < N {
1245 if in_degree[j] != 0 {
1246 let mut depends = false;
1247 let mut k = 0;
1248 while k < deps[j].len() {
1249 if deps[j][k] as usize == node {
1250 depends = true;
1251 }
1252 k += 1;
1253 }
1254 if depends {
1255 in_degree[j] -= 1;
1256 if in_degree[j] == 0 {
1257 queue[tail] = j as u8;
1258 tail += 1;
1259 }
1260 }
1261 }
1262 j += 1;
1263 }
1264 }
1265
1266 // A cycle leaves some nodes unproduced. During const eval this panic is a
1267 // compile error, so cyclic graphs are rejected at build time. `core::panic!`
1268 // (not the crate's defmt-shimmed `panic!`) keeps this const-evaluable.
1269 if produced != N {
1270 core::panic!("supervisor_graph!: dependency cycle");
1271 }
1272 order
1273}
1274
1275#[cfg(feature = "pool")]
1276mod pool;
1277#[cfg(feature = "pool")]
1278pub use pool::*;
1279
1280#[cfg(feature = "trace")]
1281pub mod trace;
1282
1283/// Declare a supervised task graph and compute its topological order at compile
1284/// time (single source of nodes, deps, pool, and order). See the
1285/// `embassy-supervisor-macros` crate for the surface syntax.
1286#[cfg(feature = "macros")]
1287pub use embassy_supervisor_macros::supervisor_graph;
1288
1289// ─── Tests (host-only) ─────────────────────────────────────────────────────
1290//
1291// Run on the host: `cargo test -p embassy-supervisor --target x86_64-unknown-linux-gnu`
1292// (the workspace `.cargo/config.toml` pins the embedded target, so `--target` is
1293// required to override it). These exercise the compile-time `topo_sort_const`
1294// over index adjacency tables — exactly what `supervisor_graph!` generates.
1295#[cfg(test)]
1296mod tests {
1297 use super::topo_sort_const;
1298
1299 /// Position of index `x` within `order`.
1300 fn pos<const N: usize>(order: &[u8; N], x: u8) -> usize {
1301 order.iter().position(|&y| y == x).expect("index present")
1302 }
1303
1304 #[test]
1305 fn linear_chain_orders_deps_before_dependents() {
1306 // A=0, B=1 dep A, C=2 dep B.
1307 const DEPS: [&[u8]; 3] = [&[], &[0], &[1]];
1308 const ORDER: [u8; 3] = topo_sort_const(&DEPS);
1309 assert_eq!(ORDER, [0, 1, 2]);
1310 }
1311
1312 #[test]
1313 fn diamond_puts_root_first_and_join_last() {
1314 // A=0; B=1 dep A; C=2 dep A; D=3 dep B,C.
1315 const DEPS: [&[u8]; 4] = [&[], &[0], &[0], &[1, 2]];
1316 const ORDER: [u8; 4] = topo_sort_const(&DEPS);
1317 assert_eq!(ORDER[0], 0, "root first");
1318 assert_eq!(ORDER[3], 3, "join last");
1319 assert!(pos(&ORDER, 1) < pos(&ORDER, 3), "B before D");
1320 assert!(pos(&ORDER, 2) < pos(&ORDER, 3), "C before D");
1321 }
1322
1323 #[test]
1324 fn independent_nodes_all_present() {
1325 const DEPS: [&[u8]; 2] = [&[], &[]];
1326 const ORDER: [u8; 2] = topo_sort_const(&DEPS);
1327 assert!(ORDER.contains(&0) && ORDER.contains(&1));
1328 }
1329
1330 #[test]
1331 fn unsorted_input_is_sorted() {
1332 // Declared out of dependency order: node 0 depends on 1 and 2, node 2 on 1.
1333 const DEPS: [&[u8]; 3] = [&[1, 2], &[], &[1]];
1334 const ORDER: [u8; 3] = topo_sort_const(&DEPS);
1335 assert!(pos(&ORDER, 1) < pos(&ORDER, 2), "1 before 2");
1336 assert!(pos(&ORDER, 2) < pos(&ORDER, 0), "2 before 0");
1337 }
1338
1339 #[test]
1340 fn evaluates_at_compile_time() {
1341 // The sort runs in a `const` context, proving it is const-evaluable.
1342 // (A cyclic table here would be a *compile* error, not a test failure.)
1343 const DEPS: [&[u8]; 3] = [&[], &[0], &[1]];
1344 const _: () = {
1345 let order = topo_sort_const(&DEPS);
1346 assert!(order[0] == 0 && order[1] == 1 && order[2] == 2);
1347 };
1348 }
1349
1350 // Uncommenting this must fail to compile ("dependency cycle"):
1351 // const CYCLE: [&[u8]; 2] = [&[1], &[0]];
1352 // const _BAD: [u8; 2] = topo_sort_const(&CYCLE);
1353}