#![cfg_attr(not(test), no_std)]
#![forbid(unsafe_code)]
#![deny(missing_docs)]
#[macro_use]
mod fmt;
use core::cell::Cell;
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(feature = "trace")]
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")]
pub fn request_control(node: &'static TaskNode, op: ControlOp) {
let _ = CONTROL_REQ.try_send(ControlCommand { node, op });
}
#[cfg(feature = "control")]
pub async fn wait_control() -> ControlCommand {
CONTROL_REQ.receive().await
}
const SHUTDOWN_ACK_TIMEOUT_MS: u64 = 2_000;
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,
resume_wake: Signal<CriticalSectionRawMutex, ()>,
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),
resume_wake: Signal::new(),
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 restore(&self, value: T) {
self.provide(value);
}
}
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],
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: &[],
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
}
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 async fn wait_resume(&self) {
self.handle.resume_wake.wait().await;
}
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);
}
pub fn set_disabled(&self, disabled: bool) {
self.handle.disabled.store(disabled, Ordering::Release);
}
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.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>],
deps: &'static [&'static [u8]],
order: [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(SLOT_READY_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(SLOT_READY_TIMEOUT, wait)
.await
.map_err(|_| SpawnError::Busy)?;
}
Ok(())
}
impl<const N: usize> Supervisor<N> {
pub const fn new(graph: &'static Graph<N>) -> Self {
Self {
nodes: graph.nodes,
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;
}
info!("supervisor: spawning {} ({})", node.name, node.mode);
if let Some(spawn) = node.spawn {
await_spawn_slot(node).await?;
await_resources(node).await?;
spawn(spawner)?;
}
node.set_running(true);
}
Ok(())
}
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?;
spawn(spawner)?;
}
node.set_running(true);
info!("supervisor: started {}", node.name);
Ok(())
}
async fn shutdown_and_wait(&self, node: &'static TaskNode) {
node.signal_shutdown();
if let Either::Second(()) = select(
node.wait_dropped(),
Timer::after_millis(SHUTDOWN_ACK_TIMEOUT_MS),
)
.await
{
panic!(
"supervisor: task {} did not ack shutdown within {}ms",
node.name, SHUTDOWN_ACK_TIMEOUT_MS,
);
}
node.set_running(false);
}
pub async fn stop_node(&self, node: &'static TaskNode) {
if !node.is_running() || node.is_detached() {
return;
}
self.shutdown_and_wait(node).await;
info!("supervisor: stopped {}", node.name);
}
pub async fn teardown(&self) {
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;
}
}
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?;
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) {
match cmd.op {
ControlOp::Deactivate => self.deactivate(cmd.node).await,
ControlOp::Activate => self.activate(cmd.node, spawner).await,
}
}
async fn deactivate(&self, target: &'static TaskNode) {
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;
}
}
}
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_graph;
#[cfg(test)]
mod tests {
use super::topo_sort_const;
fn pos<const N: usize>(order: &[u8; N], x: u8) -> usize {
order.iter().position(|&y| y == x).expect("index present")
}
#[test]
fn linear_chain_orders_deps_before_dependents() {
const DEPS: [&[u8]; 3] = [&[], &[0], &[1]];
const ORDER: [u8; 3] = topo_sort_const(&DEPS);
assert_eq!(ORDER, [0, 1, 2]);
}
#[test]
fn diamond_puts_root_first_and_join_last() {
const DEPS: [&[u8]; 4] = [&[], &[0], &[0], &[1, 2]];
const ORDER: [u8; 4] = topo_sort_const(&DEPS);
assert_eq!(ORDER[0], 0, "root first");
assert_eq!(ORDER[3], 3, "join last");
assert!(pos(&ORDER, 1) < pos(&ORDER, 3), "B before D");
assert!(pos(&ORDER, 2) < pos(&ORDER, 3), "C before D");
}
#[test]
fn independent_nodes_all_present() {
const DEPS: [&[u8]; 2] = [&[], &[]];
const ORDER: [u8; 2] = topo_sort_const(&DEPS);
assert!(ORDER.contains(&0) && ORDER.contains(&1));
}
#[test]
fn unsorted_input_is_sorted() {
const DEPS: [&[u8]; 3] = [&[1, 2], &[], &[1]];
const ORDER: [u8; 3] = topo_sort_const(&DEPS);
assert!(pos(&ORDER, 1) < pos(&ORDER, 2), "1 before 2");
assert!(pos(&ORDER, 2) < pos(&ORDER, 0), "2 before 0");
}
#[test]
fn evaluates_at_compile_time() {
const DEPS: [&[u8]; 3] = [&[], &[0], &[1]];
const _: () = {
let order = topo_sort_const(&DEPS);
assert!(order[0] == 0 && order[1] == 1 && order[2] == 2);
};
}
}