#![cfg_attr(not(test), no_std)]
#![forbid(unsafe_code)]
#![deny(missing_docs)]
#[macro_use]
mod fmt;
use core::cell::Cell;
use core::future::Future;
use core::sync::atomic::Ordering;
use embassy_executor::{SendSpawner, SpawnError, Spawner};
use embassy_futures::select::{Either, select};
use embassy_sync::blocking_mutex::Mutex as BlockingMutex;
use embassy_sync::blocking_mutex::raw::CriticalSectionRawMutex;
#[cfg(feature = "control")]
use embassy_sync::channel::Channel;
use embassy_sync::signal::Signal;
use embassy_time::{Timer, with_timeout};
use portable_atomic::AtomicBool;
#[cfg(any(feature = "trace", feature = "liveness"))]
use portable_atomic::AtomicU32;
#[cfg(feature = "pool")]
static SCALE_REQ: Signal<CriticalSectionRawMutex, ()> = Signal::new();
pub fn request_scale() {
#[cfg(feature = "pool")]
SCALE_REQ.signal(());
}
#[cfg(feature = "pool")]
pub async fn wait_scale() {
SCALE_REQ.wait().await;
}
#[cfg(feature = "control")]
#[derive(Clone, Copy, PartialEq, Eq, Debug)]
pub enum ControlOp {
Activate,
Deactivate,
}
#[cfg(feature = "control")]
#[derive(Clone, Copy, Debug)]
pub struct ControlCommand {
pub node: &'static TaskNode,
pub op: ControlOp,
}
#[cfg(feature = "control")]
static CONTROL_REQ: Channel<CriticalSectionRawMutex, ControlCommand, 4> = Channel::new();
#[cfg(feature = "control")]
#[derive(Clone, Copy, PartialEq, Eq, Debug)]
pub struct ControlQueueFull;
#[cfg(all(feature = "control", feature = "defmt"))]
impl defmt::Format for ControlQueueFull {
fn format(&self, fmt: defmt::Formatter) {
defmt::write!(fmt, "control queue full");
}
}
#[cfg(feature = "control")]
pub async fn request_control(node: &'static TaskNode, op: ControlOp) {
CONTROL_REQ.send(ControlCommand { node, op }).await;
}
#[cfg(feature = "control")]
pub fn try_request_control(node: &'static TaskNode, op: ControlOp) -> Result<(), ControlQueueFull> {
CONTROL_REQ
.try_send(ControlCommand { node, op })
.map_err(|_| ControlQueueFull)
}
#[cfg(feature = "control")]
pub async fn wait_control() -> ControlCommand {
CONTROL_REQ.receive().await
}
const SHUTDOWN_ACK_TIMEOUT_MS: u64 = 2_000;
#[derive(Clone, Copy, Debug)]
pub struct ShutdownTimeout {
pub node: &'static TaskNode,
}
#[cfg(feature = "defmt")]
impl defmt::Format for ShutdownTimeout {
fn format(&self, fmt: defmt::Formatter) {
defmt::write!(fmt, "{} missed shutdown ack", self.node.name);
}
}
#[cfg(any(feature = "pool", feature = "control"))]
#[derive(Clone, Copy, Debug)]
pub enum RunError {
Spawn(SpawnError),
Shutdown(ShutdownTimeout),
}
#[cfg(all(any(feature = "pool", feature = "control"), feature = "defmt"))]
impl defmt::Format for RunError {
fn format(&self, fmt: defmt::Formatter) {
match self {
RunError::Spawn(_) => defmt::write!(fmt, "bring-up spawn failed"),
RunError::Shutdown(e) => defmt::write!(fmt, "{}", e),
}
}
}
#[derive(Clone, Copy, PartialEq, Eq, Debug)]
pub struct Aborted;
#[cfg(feature = "defmt")]
impl defmt::Format for Aborted {
fn format(&self, fmt: defmt::Formatter) {
defmt::write!(fmt, "aborted by shutdown");
}
}
const SLOT_READY_TIMEOUT: embassy_time::Duration = embassy_time::Duration::from_millis(100);
#[derive(Clone, Copy, PartialEq, Eq, Debug)]
pub enum Mode {
Terminate,
Pause,
OnDemand,
}
impl Mode {
pub fn as_str(&self) -> &'static str {
match self {
Mode::Terminate => "terminate",
Mode::Pause => "pause",
Mode::OnDemand => "ondemand",
}
}
}
#[cfg(feature = "defmt")]
impl defmt::Format for Mode {
fn format(&self, f: defmt::Formatter) {
defmt::write!(f, "{}", self.as_str());
}
}
pub struct TaskHandle {
shutdown: AtomicBool,
shutdown_wake: Signal<CriticalSectionRawMutex, ()>,
dropped: AtomicBool,
dropped_wake: Signal<CriticalSectionRawMutex, ()>,
running: AtomicBool,
busy: AtomicBool,
completed: AtomicBool,
resume_wake: Signal<CriticalSectionRawMutex, ()>,
#[cfg(feature = "readiness")]
ready: AtomicBool,
#[cfg(feature = "readiness")]
ready_wake: Signal<CriticalSectionRawMutex, ()>,
#[cfg(feature = "liveness")]
last_beat: AtomicU32,
disabled: AtomicBool,
detached: AtomicBool,
#[cfg(feature = "trace")]
task_id: AtomicU32,
#[cfg(feature = "trace")]
exec_ticks: AtomicU32,
#[cfg(feature = "trace")]
polls: AtomicU32,
#[cfg(feature = "trace")]
max_poll_ticks: AtomicU32,
}
impl TaskHandle {
const fn new(disabled_at_boot: bool) -> Self {
Self {
shutdown: AtomicBool::new(false),
shutdown_wake: Signal::new(),
dropped: AtomicBool::new(false),
dropped_wake: Signal::new(),
running: AtomicBool::new(false),
busy: AtomicBool::new(false),
completed: AtomicBool::new(false),
resume_wake: Signal::new(),
#[cfg(feature = "readiness")]
ready: AtomicBool::new(false),
#[cfg(feature = "readiness")]
ready_wake: Signal::new(),
#[cfg(feature = "liveness")]
last_beat: AtomicU32::new(0),
disabled: AtomicBool::new(disabled_at_boot),
detached: AtomicBool::new(false),
#[cfg(feature = "trace")]
task_id: AtomicU32::new(0),
#[cfg(feature = "trace")]
exec_ticks: AtomicU32::new(0),
#[cfg(feature = "trace")]
polls: AtomicU32::new(0),
#[cfg(feature = "trace")]
max_poll_ticks: AtomicU32::new(0),
}
}
}
pub struct SpawnerSlot {
slot: BlockingMutex<CriticalSectionRawMutex, Cell<Option<SendSpawner>>>,
filled: Signal<CriticalSectionRawMutex, ()>,
}
impl SpawnerSlot {
pub const fn new() -> Self {
Self {
slot: BlockingMutex::new(Cell::new(None)),
filled: Signal::new(),
}
}
pub fn set(&self, spawner: SendSpawner) {
self.slot.lock(|c| c.set(Some(spawner)));
self.filled.signal(());
}
pub fn get(&self) -> Option<SendSpawner> {
self.slot.lock(Cell::get)
}
pub async fn ready(&self) -> SendSpawner {
loop {
if let Some(sp) = self.get() {
return sp;
}
self.filled.wait().await;
}
}
}
impl Default for SpawnerSlot {
fn default() -> Self {
Self::new()
}
}
pub trait ResourceGate: Sync {
fn is_filled(&self) -> bool;
fn filled_signal(&self) -> &Signal<CriticalSectionRawMutex, ()>;
}
pub struct ResourceSlot<T> {
slot: BlockingMutex<CriticalSectionRawMutex, Cell<Option<T>>>,
filled: Signal<CriticalSectionRawMutex, ()>,
}
impl<T> ResourceSlot<T> {
pub const fn new() -> Self {
Self {
slot: BlockingMutex::new(Cell::new(None)),
filled: Signal::new(),
}
}
pub fn provide(&self, value: T) {
self.slot.lock(|c| c.set(Some(value)));
self.filled.signal(());
}
pub fn take(&self) -> Option<T> {
self.slot.lock(Cell::take)
}
pub fn get(&self) -> Option<T>
where
T: Copy,
{
self.slot.lock(|c| {
let v = c.take();
c.set(v);
v
})
}
pub fn restore(&self, value: T) {
self.provide(value);
}
pub async fn wait_take(&self) -> T {
loop {
if let Some(v) = self.take() {
return v;
}
self.filled.wait().await;
}
}
}
impl<T: Send> ResourceGate for ResourceSlot<T> {
fn is_filled(&self) -> bool {
self.slot.lock(|c| {
let v = c.take();
let filled = v.is_some();
c.set(v);
filled
})
}
fn filled_signal(&self) -> &Signal<CriticalSectionRawMutex, ()> {
&self.filled
}
}
impl<T> Default for ResourceSlot<T> {
fn default() -> Self {
Self::new()
}
}
pub struct TaskNode {
pub name: &'static str,
pub mode: Mode,
pub spawn: Option<fn(Spawner) -> Result<(), SpawnError>>,
spawn_slot: Option<&'static SpawnerSlot>,
resource_gates: &'static [&'static dyn ResourceGate],
#[cfg(feature = "readiness")]
ready_deps: &'static [&'static TaskNode],
slot_timeout: embassy_time::Duration,
handle: TaskHandle,
}
impl TaskNode {
pub const fn new(
name: &'static str,
mode: Mode,
spawn: Option<fn(Spawner) -> Result<(), SpawnError>>,
disabled_at_boot: bool,
) -> Self {
Self {
name,
mode,
spawn,
spawn_slot: None,
resource_gates: &[],
#[cfg(feature = "readiness")]
ready_deps: &[],
slot_timeout: SLOT_READY_TIMEOUT,
handle: TaskHandle::new(disabled_at_boot),
}
}
pub const fn with_executor(mut self, slot: &'static SpawnerSlot) -> Self {
self.spawn_slot = Some(slot);
self
}
pub const fn with_resources(mut self, gates: &'static [&'static dyn ResourceGate]) -> Self {
self.resource_gates = gates;
self
}
#[cfg(feature = "readiness")]
pub const fn with_ready_deps(mut self, deps: &'static [&'static TaskNode]) -> Self {
self.ready_deps = deps;
self
}
pub const fn with_slot_timeout(mut self, timeout: embassy_time::Duration) -> Self {
self.slot_timeout = timeout;
self
}
pub fn shutdown_requested(&self) -> bool {
self.handle.shutdown.load(Ordering::Acquire)
}
pub async fn wait_shutdown(&self) {
if self.handle.shutdown.load(Ordering::Acquire) {
return;
}
self.handle.shutdown_wake.wait().await;
}
pub fn ack_dropped(&self) {
self.handle.running.store(false, Ordering::Release);
self.handle.dropped.store(true, Ordering::Release);
self.handle.dropped_wake.signal(());
}
pub fn mark_exited(&self) {
self.handle.completed.store(true, Ordering::Release);
self.ack_dropped();
}
pub fn has_exited(&self) -> bool {
self.handle.completed.load(Ordering::Acquire)
}
#[cfg(feature = "readiness")]
pub fn set_ready(&self) {
self.handle.ready.store(true, Ordering::Release);
self.handle.ready_wake.signal(());
}
#[cfg(feature = "readiness")]
pub fn clear_ready(&self) {
self.handle.ready.store(false, Ordering::Release);
}
#[cfg(feature = "readiness")]
pub fn is_ready(&self) -> bool {
self.handle.ready.load(Ordering::Acquire)
}
#[cfg(feature = "readiness")]
pub async fn wait_ready(&self) {
loop {
if self.is_ready() {
return;
}
self.handle.ready_wake.wait().await;
}
}
#[cfg(all(feature = "pool", feature = "readiness"))]
pub(crate) fn ready_deps_ok(&self) -> bool {
self.ready_deps.iter().all(|d| d.is_ready())
}
#[cfg(all(feature = "pool", not(feature = "readiness")))]
pub(crate) fn ready_deps_ok(&self) -> bool {
true
}
#[cfg(feature = "liveness")]
pub fn beat(&self) {
self.handle.last_beat.store(
embassy_time::Instant::now().as_ticks() as u32,
Ordering::Release,
);
}
#[cfg(feature = "liveness")]
pub fn ticks_since_beat(&self) -> u32 {
(embassy_time::Instant::now().as_ticks() as u32)
.wrapping_sub(self.handle.last_beat.load(Ordering::Acquire))
}
#[cfg(feature = "liveness")]
pub fn is_stale(&self, max_age: embassy_time::Duration) -> bool {
self.is_running() && u64::from(self.ticks_since_beat()) > max_age.as_ticks()
}
pub async fn wait_resume(&self) {
self.handle.resume_wake.wait().await;
}
pub async fn run_cancellable<F: Future>(&self, fut: F) -> Result<F::Output, Aborted> {
match select(fut, self.wait_shutdown()).await {
Either::First(out) => Ok(out),
Either::Second(()) => Err(Aborted),
}
}
pub async fn run_cancellable_acked<F: Future>(&self, fut: F) -> Result<F::Output, Aborted> {
let out = self.run_cancellable(fut).await;
if out.is_err() {
self.ack_dropped();
}
out
}
pub fn mark_busy(&self) {
if !self.handle.busy.swap(true, Ordering::Release) {
request_scale();
}
}
pub fn mark_idle(&self) {
if self.handle.busy.swap(false, Ordering::Release) {
request_scale();
}
}
pub fn is_busy(&self) -> bool {
self.handle.busy.load(Ordering::Acquire)
}
pub fn is_running(&self) -> bool {
self.handle.running.load(Ordering::Acquire)
}
pub fn is_disabled(&self) -> bool {
self.handle.disabled.load(Ordering::Acquire)
}
pub fn set_detached(&self, detached: bool) {
self.handle.detached.store(detached, Ordering::Release);
}
pub fn is_detached(&self) -> bool {
self.handle.detached.load(Ordering::Acquire)
}
#[cfg(feature = "trace")]
pub fn set_task_id(&self, id: u32) {
self.handle.task_id.store(id, Ordering::Release);
}
#[cfg(feature = "trace")]
pub fn adopt<S>(&self, token: &embassy_executor::SpawnToken<S>) {
self.set_task_id(token.id());
#[cfg(feature = "metadata-names")]
self.stamp_name(token);
}
#[cfg(feature = "metadata-names")]
pub fn stamp_name<S>(&self, token: &embassy_executor::SpawnToken<S>) {
token.metadata().set_name(self.name);
}
#[cfg(feature = "trace")]
pub fn task_id(&self) -> u32 {
self.handle.task_id.load(Ordering::Acquire)
}
#[cfg(feature = "trace")]
pub fn exec_ticks(&self) -> u32 {
self.handle.exec_ticks.load(Ordering::Relaxed)
}
#[cfg(feature = "trace")]
pub fn poll_count(&self) -> u32 {
self.handle.polls.load(Ordering::Relaxed)
}
#[cfg(feature = "trace")]
pub fn max_poll_ticks(&self) -> u32 {
self.handle.max_poll_ticks.load(Ordering::Relaxed)
}
pub(crate) fn signal_shutdown(&self) {
self.handle.shutdown.store(true, Ordering::Release);
self.handle.shutdown_wake.signal(());
}
pub(crate) fn signal_resume(&self) {
self.handle.resume_wake.signal(());
}
pub(crate) fn set_running(&self, running: bool) {
self.handle.running.store(running, Ordering::Release);
#[cfg(feature = "liveness")]
if running {
self.handle.last_beat.store(
embassy_time::Instant::now().as_ticks() as u32,
Ordering::Release,
);
}
}
pub fn set_disabled(&self, disabled: bool) {
self.handle.disabled.store(disabled, Ordering::Release);
}
pub(crate) fn has_acked_stop(&self) -> bool {
self.handle.dropped.load(Ordering::Acquire)
&& !self.handle.completed.load(Ordering::Acquire)
}
pub(crate) async fn wait_dropped(&self) {
if self.handle.dropped.load(Ordering::Acquire) {
return;
}
self.handle.dropped_wake.wait().await;
}
pub(crate) fn reset(&self) {
self.handle.shutdown.store(false, Ordering::Release);
self.handle.dropped.store(false, Ordering::Release);
self.handle.busy.store(false, Ordering::Release);
self.handle.completed.store(false, Ordering::Release);
#[cfg(feature = "readiness")]
{
self.handle.ready.store(false, Ordering::Release);
self.handle.ready_wake.reset();
}
self.handle.shutdown_wake.reset();
self.handle.dropped_wake.reset();
}
}
impl core::fmt::Debug for TaskNode {
fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
f.debug_struct("TaskNode")
.field("name", &self.name)
.field("mode", &self.mode)
.field("running", &self.is_running())
.field("busy", &self.is_busy())
.field("disabled", &self.is_disabled())
.field("detached", &self.is_detached())
.finish_non_exhaustive()
}
}
pub struct Graph<const N: usize> {
pub nodes: &'static [Option<&'static TaskNode>; N],
pub deps: &'static [&'static [u8]; N],
pub order: [u8; N],
#[cfg(feature = "pool")]
pub pools: &'static [&'static dyn Pool],
}
pub struct Supervisor<const N: usize> {
nodes: &'static [Option<&'static TaskNode>],
#[cfg(any(feature = "control", feature = "pool"))]
deps: &'static [&'static [u8]],
order: &'static [u8; N],
#[cfg(feature = "pool")]
pools: &'static [&'static dyn Pool],
}
async fn await_spawn_slot(node: &'static TaskNode) -> Result<(), SpawnError> {
if let Some(slot) = node.spawn_slot {
with_timeout(node.slot_timeout, slot.ready())
.await
.map_err(|_| SpawnError::Busy)?;
}
Ok(())
}
async fn await_resources(node: &'static TaskNode) -> Result<(), SpawnError> {
for gate in node.resource_gates {
let wait = async {
loop {
if gate.is_filled() {
break;
}
gate.filled_signal().wait().await;
}
};
with_timeout(node.slot_timeout, wait)
.await
.map_err(|_| SpawnError::Busy)?;
}
Ok(())
}
#[cfg(feature = "readiness")]
async fn await_ready_deps(node: &'static TaskNode) -> Result<(), SpawnError> {
for dep in node.ready_deps {
if with_timeout(node.slot_timeout, dep.wait_ready())
.await
.is_err()
{
warn!(
"supervisor: ready-dep {} not ready within {}ms (spawning {})",
dep.name,
node.slot_timeout.as_millis(),
node.name,
);
return Err(SpawnError::Busy);
}
}
Ok(())
}
#[cfg(not(feature = "readiness"))]
async fn await_ready_deps(_node: &'static TaskNode) -> Result<(), SpawnError> {
Ok(())
}
impl<const N: usize> Supervisor<N> {
pub const fn new(graph: &'static Graph<N>) -> Self {
Self {
nodes: graph.nodes,
#[cfg(any(feature = "control", feature = "pool"))]
deps: graph.deps,
order: &graph.order,
#[cfg(feature = "pool")]
pools: graph.pools,
}
}
pub async fn start(&self, spawner: Spawner) -> Result<(), SpawnError> {
#[cfg(feature = "trace")]
trace::register_graph(self.nodes);
for i in self.order.iter() {
let Some(node) = self.nodes[*i as usize] else {
continue;
};
if matches!(node.mode, Mode::OnDemand) || node.is_disabled() {
continue;
}
if node.is_running() || node.is_detached() {
continue;
}
if matches!(node.mode, Mode::Pause) && node.has_acked_stop() {
node.reset();
info!("supervisor: resuming {} in place", node.name);
node.signal_resume();
node.set_running(true);
continue;
}
node.reset();
info!("supervisor: spawning {} ({})", node.name, node.mode);
if let Some(spawn) = node.spawn {
await_spawn_slot(node).await?;
await_resources(node).await?;
await_ready_deps(node).await?;
spawn(spawner)?;
}
node.set_running(true);
}
Ok(())
}
#[cfg(any(feature = "pool", feature = "control"))]
pub async fn run(&self, spawner: Spawner) -> RunError {
if let Err(e) = self.start(spawner).await {
return RunError::Spawn(e);
}
#[cfg(all(feature = "pool", feature = "control"))]
loop {
match select(self.run_pools(spawner), wait_control()).await {
Either::First(e) => return RunError::Shutdown(e),
Either::Second(cmd) => {
if let Err(e) = self.apply_control(cmd, spawner).await {
return RunError::Shutdown(e);
}
}
}
}
#[cfg(all(feature = "pool", not(feature = "control")))]
return RunError::Shutdown(self.run_pools(spawner).await);
#[cfg(all(feature = "control", not(feature = "pool")))]
loop {
let cmd = wait_control().await;
if let Err(e) = self.apply_control(cmd, spawner).await {
return RunError::Shutdown(e);
}
}
}
pub async fn start_node(
&self,
node: &'static TaskNode,
spawner: Spawner,
) -> Result<(), SpawnError> {
node.reset();
if let Some(spawn) = node.spawn {
await_spawn_slot(node).await?;
await_resources(node).await?;
await_ready_deps(node).await?;
spawn(spawner)?;
}
node.set_running(true);
info!("supervisor: started {}", node.name);
Ok(())
}
async fn shutdown_and_wait(&self, node: &'static TaskNode) -> Result<(), ShutdownTimeout> {
node.signal_shutdown();
if let Either::Second(()) = select(
node.wait_dropped(),
Timer::after_millis(SHUTDOWN_ACK_TIMEOUT_MS),
)
.await
{
warn!(
"supervisor: task {} did not ack shutdown within {}ms",
node.name, SHUTDOWN_ACK_TIMEOUT_MS,
);
return Err(ShutdownTimeout { node });
}
node.set_running(false);
Ok(())
}
pub async fn stop_node(&self, node: &'static TaskNode) -> Result<(), ShutdownTimeout> {
if !node.is_running() || node.is_detached() {
return Ok(());
}
self.shutdown_and_wait(node).await?;
info!("supervisor: stopped {}", node.name);
Ok(())
}
pub async fn teardown(&self) -> Result<(), ShutdownTimeout> {
for i in self.order.iter().rev() {
let Some(node) = self.nodes[*i as usize] else {
continue;
};
if !node.is_running() {
continue;
}
if node.is_detached() {
continue;
}
info!("supervisor: tearing down {}", node.name);
self.shutdown_and_wait(node).await?;
}
Ok(())
}
pub async fn teardown_continue(&self) -> Result<(), ShutdownTimeout> {
let mut first_err = Ok(());
for i in self.order.iter().rev() {
let Some(node) = self.nodes[*i as usize] else {
continue;
};
if !node.is_running() || node.is_detached() {
continue;
}
info!("supervisor: tearing down {}", node.name);
if let Err(e) = self.shutdown_and_wait(node).await {
if first_err.is_ok() {
first_err = Err(e);
}
}
}
first_err
}
pub fn resume_node(&self, node: &'static TaskNode) {
if !matches!(node.mode, Mode::Pause)
|| node.is_disabled()
|| node.is_detached()
|| !node.has_acked_stop()
{
return;
}
node.reset();
info!("supervisor: resuming {}", node.name);
node.signal_resume();
node.set_running(true);
}
pub fn resume_pausable(&self) {
for i in self.order.iter() {
let Some(node) = self.nodes[*i as usize] else {
continue;
};
if matches!(node.mode, Mode::Pause) && !node.is_disabled() && !node.is_detached() {
node.reset();
info!("supervisor: resuming {}", node.name);
node.signal_resume();
node.set_running(true);
}
}
}
pub async fn respawn_terminate(&self, spawner: Spawner) -> Result<(), SpawnError> {
for i in self.order.iter() {
let Some(node) = self.nodes[*i as usize] else {
continue;
};
if matches!(node.mode, Mode::Terminate) && !node.is_disabled() && !node.is_detached() {
node.reset();
info!("supervisor: respawning {}", node.name);
if let Some(spawn) = node.spawn {
await_spawn_slot(node).await?;
await_resources(node).await?;
await_ready_deps(node).await?;
spawn(spawner)?;
}
node.set_running(true);
}
}
Ok(())
}
}
#[cfg(any(feature = "control", feature = "pool"))]
impl<const N: usize> Supervisor<N> {
fn index_of(&self, node: &'static TaskNode) -> Option<usize> {
self.nodes
.iter()
.position(|n| n.is_some_and(|x| core::ptr::eq(x, node)))
}
#[cfg(feature = "pool")]
pub(crate) fn deps_running(&self, node: &'static TaskNode) -> bool {
match self.index_of(node) {
Some(i) => self.deps[i]
.iter()
.all(|&di| self.nodes[di as usize].is_some_and(|n| n.is_running())),
None => false,
}
}
}
#[cfg(feature = "control")]
impl<const N: usize> Supervisor<N> {
fn seed(&self, target: &'static TaskNode, set: &mut [bool; N]) {
if let Some(i) = self.index_of(target) {
set[i] = true;
}
#[cfg(feature = "pool")]
for pool in self.pools {
let members = pool.members();
if members.iter().any(|m| core::ptr::eq(*m, target)) {
for m in members {
if let Some(i) = self.index_of(m) {
set[i] = true;
}
}
}
}
}
pub async fn apply_control(
&self,
cmd: ControlCommand,
spawner: Spawner,
) -> Result<(), ShutdownTimeout> {
match cmd.op {
ControlOp::Deactivate => self.deactivate(cmd.node).await,
ControlOp::Activate => {
self.activate(cmd.node, spawner).await;
Ok(())
}
}
}
pub async fn deactivate(&self, target: &'static TaskNode) -> Result<(), ShutdownTimeout> {
let mut set = [false; N];
self.seed(target, &mut set);
for i in self.order.iter() {
let j = *i as usize;
if set[j] {
continue;
}
let Some(node) = self.nodes[j] else {
continue;
};
if node.is_detached() {
continue;
}
if self.deps[j].iter().any(|&di| set[di as usize]) {
set[j] = true;
}
}
for i in self.order.iter().rev() {
let j = *i as usize;
if !set[j] {
continue;
}
let Some(node) = self.nodes[j] else {
continue;
};
if node.is_detached() {
continue;
}
node.set_disabled(true);
if node.is_running() {
info!("supervisor: control-stop {}", node.name);
self.shutdown_and_wait(node).await?;
}
}
Ok(())
}
pub async fn activate(&self, target: &'static TaskNode, spawner: Spawner) {
let mut set = [false; N];
self.seed(target, &mut set);
for i in self.order.iter().rev() {
let j = *i as usize;
if set[j] && !self.nodes[j].is_some_and(|n| n.is_detached()) {
for &di in self.deps[j] {
set[di as usize] = true;
}
}
}
for i in self.order.iter() {
let j = *i as usize;
if !set[j] {
continue;
}
let Some(node) = self.nodes[j] else {
continue;
};
if node.is_detached() {
continue;
}
node.set_disabled(false);
if node.is_running() {
continue;
}
match node.mode {
Mode::Terminate => {
info!("supervisor: control-start {}", node.name);
let _ = self.start_node(node, spawner).await;
}
Mode::Pause => {
info!("supervisor: control-resume {}", node.name);
node.reset();
node.signal_resume();
node.set_running(true);
}
Mode::OnDemand => {}
}
}
}
}
#[doc(hidden)]
#[must_use]
pub const fn topo_sort_const<const N: usize>(deps: &[&'static [u8]; N]) -> [u8; N] {
assert!(
N <= 256,
"supervisor graph exceeds 256 node slots (indices are u8)"
);
let mut in_degree = [0u8; N];
let mut i = 0;
while i < N {
in_degree[i] = deps[i].len() as u8;
i += 1;
}
let mut queue = [0u8; N];
let mut tail = 0;
i = 0;
while i < N {
if in_degree[i] == 0 {
queue[tail] = i as u8;
tail += 1;
}
i += 1;
}
let mut order = [0u8; N];
let mut produced = 0;
let mut head = 0;
while head < tail {
let node = queue[head] as usize;
head += 1;
order[produced] = node as u8;
produced += 1;
let mut j = 0;
while j < N {
if in_degree[j] != 0 {
let mut depends = false;
let mut k = 0;
while k < deps[j].len() {
if deps[j][k] as usize == node {
depends = true;
}
k += 1;
}
if depends {
in_degree[j] -= 1;
if in_degree[j] == 0 {
queue[tail] = j as u8;
tail += 1;
}
}
}
j += 1;
}
}
if produced != N {
core::panic!("supervisor_graph!: dependency cycle");
}
order
}
#[cfg(feature = "pool")]
mod pool;
#[cfg(feature = "pool")]
pub use pool::*;
#[cfg(feature = "trace")]
pub mod trace;
#[cfg(feature = "macros")]
pub use embassy_supervisor_macros::supervisor_fragment;
#[cfg(feature = "macros")]
pub use embassy_supervisor_macros::supervisor_graph;
#[cfg(feature = "macros")]
#[macro_export]
macro_rules! compose_graph {
(name: $n:ident, fragments: [$f:path $(, $r:path)* $(,)?], graph: {$($g:tt)*}) => {
$f! { @emit $crate::compose_graph, [$($r),*], {name: $n;}, {$($g)*} }
};
(fragments: [$f:path $(, $r:path)* $(,)?], graph: {$($g:tt)*}) => {
$f! { @emit $crate::compose_graph, [$($r),*], {}, {$($g)*} }
};
(@next [], {$($acc:tt)*}, {$($g:tt)*}) => {
$crate::supervisor_graph! { $($acc)* $($g)* }
};
(@next [$f:path $(, $r:path)*], {$($acc:tt)*}, $g:tt) => {
$f! { @emit $crate::compose_graph, [$($r),*], {$($acc)*}, $g }
};
}
#[doc(hidden)]
pub mod _export {
pub use embassy_sync::blocking_mutex::Mutex as BlockingMutex;
pub use embassy_sync::blocking_mutex::raw::CriticalSectionRawMutex;
pub use embassy_sync::signal::Signal;
pub use embassy_time::Duration;
}