use proc_macro::TokenStream;
use proc_macro2::TokenStream as TokenStream2;
use quote::{format_ident, quote};
use std::collections::{HashMap, HashSet};
use syn::parse::{Parse, ParseStream};
use syn::punctuated::Punctuated;
use syn::{
Attribute, Expr, Ident, LitInt, Meta, Path, Result as SynResult, Token, Type, bracketed,
};
mod kw {
syn::custom_keyword!(node);
syn::custom_keyword!(pool);
syn::custom_keyword!(deps);
syn::custom_keyword!(spawn);
syn::custom_keyword!(task);
syn::custom_keyword!(pool_size);
syn::custom_keyword!(policy);
syn::custom_keyword!(min);
syn::custom_keyword!(max);
syn::custom_keyword!(disabled);
syn::custom_keyword!(executor);
syn::custom_keyword!(resources);
syn::custom_keyword!(slot_timeout);
syn::custom_keyword!(exit);
syn::custom_keyword!(name);
syn::custom_keyword!(state);
syn::custom_keyword!(fragment);
syn::custom_keyword!(endfragment);
}
const LOCAL_SLOT_TYPE: &str = "__SvLocalResourceSlot";
struct HelperIdents {
local_slot: Ident,
try_box: Ident,
alloc_alias: Ident,
}
impl HelperIdents {
fn new(graph_name: Option<&Ident>) -> Self {
match graph_name {
None => Self {
local_slot: format_ident!("{LOCAL_SLOT_TYPE}"),
try_box: format_ident!("__sv_try_box"),
alloc_alias: format_ident!("__sv_alloc"),
},
Some(n) => {
let lower = n.to_string().to_lowercase();
Self {
local_slot: format_ident!("{LOCAL_SLOT_TYPE}{}", n),
try_box: format_ident!("__sv_try_box_{lower}"),
alloc_alias: format_ident!("__sv_alloc_{lower}"),
}
}
}
}
}
#[derive(Clone)]
struct Dep {
cfg: Vec<Attribute>,
ident: Ident,
ready: Option<Ident>,
}
fn parse_dep_list(input: ParseStream) -> SynResult<Vec<Dep>> {
let content;
bracketed!(content in input);
let mut deps = Vec::new();
while !content.is_empty() {
let cfg = content.call(Attribute::parse_outer)?;
let ident: Ident = content.parse()?;
let ready = if content.peek(Ident) {
let marker: Ident = content.parse()?;
if marker != "ready" {
return Err(syn::Error::new_spanned(
&marker,
format!("expected `,`, `]`, or the `ready` marker, found `{marker}`"),
));
}
if !cfg!(feature = "readiness") {
return Err(syn::Error::new_spanned(
&marker,
"the `ready` dep marker requires the `readiness` feature \
(embassy-supervisor feature `readiness`) — bring-up then \
awaits the dep's set_ready() before spawning this node",
));
}
Some(marker)
} else {
None
};
deps.push(Dep { cfg, ident, ready });
if content.peek(Token![,]) {
content.parse::<Token![,]>()?;
}
}
Ok(deps)
}
fn parse_mode_list(input: ParseStream) -> SynResult<Vec<Ident>> {
let content;
bracketed!(content in input);
let punct = Punctuated::<Ident, Token![,]>::parse_terminated(&content)?;
Ok(punct.into_iter().collect())
}
enum TaskSource {
Spawn(Expr),
Shell(Expr),
}
struct ResourceDecl {
cfg: Vec<Attribute>,
ident: Ident,
ty: Type,
local: Option<Ident>,
consume: Option<Ident>,
shared: Option<Ident>,
}
impl ResourceDecl {
fn shared_signature(&self) -> String {
let ty = &self.ty;
format!(
"{}shared {}",
if self.local.is_some() { "local " } else { "" },
quote!(#ty)
)
}
}
fn peek_kind_marker(content: ParseStream) -> Option<Ident> {
if !content.peek(syn::Ident) {
return None;
}
let fork = content.fork();
let ident: Ident = fork.parse().ok()?;
if ident != "local" && ident != "consume" && ident != "shared" {
return None;
}
if fork.is_empty() || fork.peek(Token![,]) || fork.peek(Token![::]) || fork.peek(Token![<]) {
return None;
}
Some(ident)
}
fn parse_resource_list(input: ParseStream) -> SynResult<Vec<ResourceDecl>> {
let content;
bracketed!(content in input);
let mut resources = Vec::new();
while !content.is_empty() {
let cfg = content.call(Attribute::parse_outer)?;
let ident: Ident = content.parse()?;
content.parse::<Token![:]>()?;
let mut local: Option<Ident> = None;
let mut consume: Option<Ident> = None;
let mut shared: Option<Ident> = None;
while let Some(marker) = peek_kind_marker(&content) {
content.parse::<Ident>()?; if marker == "local" && !cfg!(feature = "local-resources") {
return Err(syn::Error::new_spanned(
&marker,
"`local` resources emit an `unsafe impl Sync` — opt in by \
enabling embassy-supervisor's `local-resources` feature",
));
}
let slot = if marker == "local" {
&mut local
} else if marker == "consume" {
&mut consume
} else {
&mut shared
};
if slot.is_some() {
return Err(syn::Error::new_spanned(
&marker,
format!("duplicate `{marker}` marker"),
));
}
*slot = Some(marker);
}
if let (Some(_), Some(s)) = (&consume, &shared) {
return Err(syn::Error::new_spanned(
s,
"`consume` and `shared` are mutually exclusive — `consume` takes \
the single value out for one owner, `shared` copies it out to \
any number of consumers",
));
}
let ty: Type = content.parse()?;
resources.push(ResourceDecl {
cfg,
ident,
ty,
local,
consume,
shared,
});
if content.peek(Token![,]) {
content.parse::<Token![,]>()?;
}
}
Ok(resources)
}
struct NodeItem {
cfg: Vec<Attribute>,
ident: Ident,
mode: Ident,
deps: Vec<Dep>,
source: Option<TaskSource>,
pool_size: Option<LitInt>,
resources: Vec<ResourceDecl>,
disabled: bool,
executor: Option<Ident>,
slot_timeout: Option<LitInt>,
exit: Option<syn::Type>,
state: Option<(syn::Type, Expr)>,
fragment: Option<String>,
}
struct ExecutorItem {
cfg: Vec<Attribute>,
ident: Ident,
}
struct PoolItem {
cfg: Vec<Attribute>,
ident: Ident,
modes: Vec<Ident>,
deps: Vec<Dep>,
source: TaskSource,
policy: Expr,
policy_ty: Option<Type>,
executor: Option<Ident>,
resources: Vec<ResourceDecl>,
slot_timeout: Option<LitInt>,
min: Expr,
max: Expr,
state: Option<(syn::Type, Expr)>,
fragment: Option<String>,
}
#[allow(clippy::large_enum_variant)]
enum Item {
Node(NodeItem),
Pool(PoolItem),
Executor(ExecutorItem),
}
struct GraphSpec {
name: Option<Ident>,
items: Vec<Item>,
}
fn item_resources(item: &Item) -> &[ResourceDecl] {
match item {
Item::Node(n) => &n.resources,
Item::Pool(p) => &p.resources,
Item::Executor(_) => &[],
}
}
fn item_ident_cfg(item: &Item) -> Option<(&Ident, &[Attribute])> {
match item {
Item::Node(n) => Some((&n.ident, &n.cfg)),
Item::Pool(p) => Some((&p.ident, &p.cfg)),
Item::Executor(_) => None,
}
}
impl Parse for GraphSpec {
fn parse(input: ParseStream) -> SynResult<Self> {
let name = if input.peek(kw::name) && input.peek2(Token![:]) {
input.parse::<kw::name>()?;
input.parse::<Token![:]>()?;
let n: Ident = input.parse()?;
input.parse::<Token![;]>()?;
Some(n)
} else {
None
};
let mut items = Vec::new();
let mut current_fragment: Option<String> = None;
while !input.is_empty() {
if input.peek(Token![@]) {
input.parse::<Token![@]>()?;
if input.peek(kw::fragment) {
input.parse::<kw::fragment>()?;
current_fragment = Some(input.parse::<Ident>()?.to_string());
} else if input.peek(kw::endfragment) {
input.parse::<kw::endfragment>()?;
current_fragment = None;
} else {
return Err(input.error("expected `@fragment NAME;` or `@endfragment;`"));
}
input.parse::<Token![;]>()?;
continue;
}
let cfg = input.call(Attribute::parse_outer)?;
if input.peek(kw::node) {
let mut n = parse_node(input, cfg)?;
n.fragment = current_fragment.clone();
items.push(Item::Node(n));
} else if input.peek(kw::pool) {
let mut p = parse_pool(input, cfg)?;
p.fragment = current_fragment.clone();
items.push(Item::Pool(p));
} else if input.peek(kw::executor) {
input.parse::<kw::executor>()?;
let ident: Ident = input.parse()?;
input.parse::<Token![;]>()?;
items.push(Item::Executor(ExecutorItem { cfg, ident }));
} else {
return Err(input.error(
"expected `node`, `pool`, or `executor` (optionally `#[cfg(...)]`-prefixed)",
));
}
}
Ok(GraphSpec { name, items })
}
}
fn parse_node(input: ParseStream, cfg: Vec<Attribute>) -> SynResult<NodeItem> {
input.parse::<kw::node>()?;
let ident: Ident = input.parse()?;
input.parse::<Token![=]>()?;
let mode: Ident = input.parse()?;
input.parse::<Token![,]>()?;
input.parse::<kw::deps>()?;
input.parse::<Token![:]>()?;
let deps = parse_dep_list(input)?;
let mut spawn = None;
let mut task: Option<(kw::task, Expr)> = None;
let mut pool_size = None;
let mut disabled = false;
let mut executor = None;
let mut resources: Option<(kw::resources, Vec<ResourceDecl>)> = None;
let mut slot_timeout = None;
let mut exit: Option<(kw::exit, syn::Type)> = None;
let mut state: Option<(kw::state, syn::Type, Expr)> = None;
while input.peek(Token![,]) {
input.parse::<Token![,]>()?;
if input.peek(kw::spawn) {
input.parse::<kw::spawn>()?;
input.parse::<Token![:]>()?;
spawn = Some(input.parse::<Expr>()?);
} else if input.peek(kw::task) {
let k = input.parse::<kw::task>()?;
input.parse::<Token![:]>()?;
task = Some((k, input.parse::<Expr>()?));
} else if input.peek(kw::pool_size) {
input.parse::<kw::pool_size>()?;
input.parse::<Token![:]>()?;
pool_size = Some(input.parse::<LitInt>()?);
} else if input.peek(kw::disabled) {
input.parse::<kw::disabled>()?;
disabled = true;
} else if input.peek(kw::executor) {
input.parse::<kw::executor>()?;
input.parse::<Token![:]>()?;
executor = Some(input.parse::<Ident>()?);
} else if input.peek(kw::resources) {
let k = input.parse::<kw::resources>()?;
input.parse::<Token![:]>()?;
resources = Some((k, parse_resource_list(input)?));
} else if input.peek(kw::slot_timeout) {
input.parse::<kw::slot_timeout>()?;
input.parse::<Token![:]>()?;
slot_timeout = Some(input.parse::<LitInt>()?);
} else if input.peek(kw::exit) {
let k = input.parse::<kw::exit>()?;
input.parse::<Token![:]>()?;
exit = Some((k, input.parse::<syn::Type>()?));
} else if input.peek(kw::state) {
let k = input.parse::<kw::state>()?;
input.parse::<Token![:]>()?;
let ty: syn::Type = input.parse()?;
input.parse::<Token![=]>()?;
let init: Expr = input.parse()?;
if !cfg!(feature = "heap-state") {
return Err(syn::Error::new_spanned(
k,
"`state:` requires the `heap-state` feature \
(embassy-supervisor feature `heap-state`) — per-activation \
boxed state, reclaimed on task exit",
));
}
state = Some((k, ty, init));
} else {
return Err(input.error(
"expected `spawn:`, `task:`, `pool_size:`, `executor:`, `resources:`, \
`slot_timeout:`, `exit:`, `state:`, or `disabled`",
));
}
}
input.parse::<Token![;]>()?;
if let Some(st) = &slot_timeout {
if st.base10_parse::<u64>()? == 0 {
return Err(syn::Error::new_spanned(
st,
"`slot_timeout:` must be at least 1 (milliseconds)",
));
}
}
if let (Some(_), Some((k, _))) = (&spawn, &task) {
return Err(syn::Error::new_spanned(
k,
"`task:` and `spawn:` are mutually exclusive — `spawn:` names a \
hand-written `#[embassy_executor::task]` fn, `task:` generates one",
));
}
if let (Some(ps), None) = (&pool_size, &task) {
return Err(syn::Error::new_spanned(
ps,
"`pool_size:` requires `task:` — a `spawn:` task fn sets its own \
`#[embassy_executor::task(pool_size = ...)]`",
));
}
if let Some((k, decls)) = &resources {
if task.is_none() {
return Err(syn::Error::new_spanned(
k,
"`resources:` requires `task:` — resources are handed to the \
generated shell as owned arguments and restored by it; a \
`spawn:` task fn manages its own arguments",
));
}
if decls.is_empty() {
return Err(syn::Error::new_spanned(
k,
"`resources:` must declare at least one `NAME: Type` entry",
));
}
for (i, d) in decls.iter().enumerate() {
if decls[..i].iter().any(|prev| prev.ident == d.ident) {
return Err(syn::Error::new_spanned(
&d.ident,
format!("duplicate resource name `{}`", d.ident),
));
}
}
}
if let Some(ps) = &pool_size {
if ps.base10_parse::<usize>()? == 0 {
return Err(syn::Error::new_spanned(
ps,
"`pool_size:` must be at least 1",
));
}
}
if let Some((k, _, _)) = &state
&& task.is_none()
{
return Err(syn::Error::new_spanned(
k,
"`state:` requires `task:` — the generated shell owns the boxed \
state across the worker call and drops it on exit; a `spawn:` \
task fn can Box its own state",
));
}
if let Some((k, _)) = &exit {
if task.is_none() {
return Err(syn::Error::new_spanned(
k,
"`exit:` requires `task:` — the generated shell is what captures \
the worker's return value; a `spawn:` task fn can provide() into \
a slot itself",
));
}
}
if let (Some((_, decls)), Some(ex)) = (&resources, &executor) {
if let Some(l) = decls.iter().find_map(|d| d.local.as_ref()) {
return Err(syn::Error::new_spanned(
l,
format!(
"`local` resources cannot be combined with `executor: {ex}` — a \
local slot exists to carry `!Send` values, and a node routed \
through a `SpawnerSlot` (`SendSpawner`) must have a `Send` \
future; run the node on the supervisor's own executor"
),
));
}
}
let source = match (spawn, task) {
(Some(e), _) => Some(TaskSource::Spawn(e)),
(None, Some((_, e))) => Some(TaskSource::Shell(e)),
(None, None) => None,
};
Ok(NodeItem {
cfg,
ident,
mode,
deps,
source,
pool_size,
disabled,
executor,
resources: resources.map(|(_, decls)| decls).unwrap_or_default(),
slot_timeout,
exit: exit.map(|(_, ty)| ty),
state: state.map(|(_, ty, init)| (ty, init)),
fragment: None,
})
}
fn parse_pool(input: ParseStream, cfg: Vec<Attribute>) -> SynResult<PoolItem> {
input.parse::<kw::pool>()?;
let ident: Ident = input.parse()?;
input.parse::<Token![=]>()?;
let modes = parse_mode_list(input)?;
input.parse::<Token![,]>()?;
input.parse::<kw::deps>()?;
input.parse::<Token![:]>()?;
let deps = parse_dep_list(input)?;
input.parse::<Token![,]>()?;
let executor = if input.peek(kw::executor) {
input.parse::<kw::executor>()?;
input.parse::<Token![:]>()?;
let ex: Ident = input.parse()?;
input.parse::<Token![,]>()?;
Some(ex)
} else {
None
};
let source = if input.peek(kw::task) {
input.parse::<kw::task>()?;
input.parse::<Token![:]>()?;
TaskSource::Shell(input.parse()?)
} else {
input.parse::<kw::spawn>()?;
input.parse::<Token![:]>()?;
TaskSource::Spawn(input.parse()?)
};
input.parse::<Token![,]>()?;
let resources = if input.peek(kw::resources) {
input.parse::<kw::resources>()?;
input.parse::<Token![:]>()?;
let decls = parse_resource_list(input)?;
if let Some(bad) = decls
.iter()
.find(|d| d.local.is_some() && d.shared.is_none())
{
return Err(syn::Error::new_spanned(
&bad.ident,
"`local` is not supported on take-kind `pool` resources (the single-core \
slot contract + per-member restore is deferred); a `shared local` entry \
works (one pool-wide fan-out slot), or declare the take-kind `local` \
resource on a node",
));
}
input.parse::<Token![,]>()?;
decls
} else {
Vec::new()
};
let state = if input.peek(kw::state) {
let k = input.parse::<kw::state>()?;
input.parse::<Token![:]>()?;
let ty: syn::Type = input.parse()?;
input.parse::<Token![=]>()?;
let init: Expr = input.parse()?;
input.parse::<Token![,]>()?;
if !cfg!(feature = "heap-state") {
return Err(syn::Error::new_spanned(
k,
"`state:` requires the `heap-state` feature \
(embassy-supervisor feature `heap-state`) — per-activation \
boxed state, reclaimed on task exit",
));
}
Some((ty, init))
} else {
None
};
if input.peek(kw::exit) {
let k = input.parse::<kw::exit>()?;
return Err(syn::Error::new_spanned(
k,
"`exit:` is not supported on `pool` — the K members share one shell, \
so per-member exit values need per-member storage; use per-node \
`exit:` declarations, or have the worker provide() into an \
app-declared slot itself",
));
}
input.parse::<kw::policy>()?;
input.parse::<Token![:]>()?;
let policy_ty = {
let fork = input.fork();
if fork.parse::<Type>().is_ok() && fork.peek(Token![=]) {
let ty: Type = input.parse()?;
input.parse::<Token![=]>()?;
Some(ty)
} else {
None
}
};
let policy: Expr = input.parse()?;
input.parse::<Token![,]>()?;
input.parse::<kw::min>()?;
input.parse::<Token![:]>()?;
let min: Expr = input.parse()?;
input.parse::<Token![,]>()?;
input.parse::<kw::max>()?;
input.parse::<Token![:]>()?;
let max: Expr = input.parse()?;
let slot_timeout = if input.peek(Token![,]) && input.peek2(kw::slot_timeout) {
input.parse::<Token![,]>()?;
input.parse::<kw::slot_timeout>()?;
input.parse::<Token![:]>()?;
let st: LitInt = input.parse()?;
if st.base10_parse::<u64>()? == 0 {
return Err(syn::Error::new_spanned(
&st,
"`slot_timeout:` must be at least 1 (milliseconds)",
));
}
Some(st)
} else {
None
};
input.parse::<Token![;]>()?;
if let Some(ex) = &executor {
if let Some(l) = resources.iter().find_map(|d| d.local.as_ref()) {
return Err(syn::Error::new_spanned(
l,
format!(
"`local` resources cannot be combined with `executor: {ex}` — a \
local slot exists to carry `!Send` values, and a pool routed \
through a `SpawnerSlot` (`SendSpawner`) must have `Send` \
futures; run the pool on the supervisor's own executor"
),
));
}
}
Ok(PoolItem {
cfg,
ident,
modes,
deps,
source,
policy,
policy_ty,
executor,
resources,
slot_timeout,
min,
max,
state,
fragment: None,
})
}
fn name_string(ident: &Ident) -> String {
ident.to_string().to_lowercase().replace('_', "-")
}
fn inject_call_with(task: &Expr, lead: &[TokenStream2]) -> SynResult<TokenStream2> {
match task {
Expr::Path(_) => Ok(quote!(#task(#(#lead),*))),
Expr::Call(c) => {
let f = &c.func;
let args = c.args.iter();
Ok(quote!(#f(#(#lead),* #(, #args)*)))
}
other => Err(syn::Error::new_spanned(
other,
"expected a task-fn path or a partial call like `f(extra_args)`",
)),
}
}
fn cfg_predicate(attrs: &[Attribute]) -> Option<TokenStream2> {
let preds: Vec<TokenStream2> = attrs
.iter()
.filter_map(|a| match &a.meta {
Meta::List(ml) if ml.path.is_ident("cfg") => Some(ml.tokens.clone()),
_ => None,
})
.collect();
match preds.len() {
0 => None,
1 => Some(preds[0].clone()),
_ => Some(quote!(all(#(#preds),*))),
}
}
fn gate_tokens(resources: &[ResourceDecl]) -> (TokenStream2, Vec<TokenStream2>) {
let gate_refs: Vec<TokenStream2> = resources
.iter()
.map(|r| {
let cfg = &r.cfg;
let res = &r.ident;
quote!(#(#cfg)* &#res)
})
.collect();
let any_cfg = resources.iter().any(|r| cfg_predicate(&r.cfg).is_some());
let len = if any_cfg {
let terms: Vec<TokenStream2> = resources
.iter()
.map(|r| match cfg_predicate(&r.cfg) {
None => quote!(1usize),
Some(pred) => quote!({
#[cfg(#pred)]
{
1usize
}
#[cfg(not(#pred))]
{
0usize
}
}),
})
.collect();
quote!(0usize #(+ #terms)*)
} else {
let n = resources.len();
quote!(#n)
};
(len, gate_refs)
}
fn fragment_suffix(fragment: &Option<String>) -> String {
match fragment {
Some(f) => format!(" (from fragment `{f}`)"),
None => String::new(),
}
}
fn ready_tokens(
deps: &[Dep],
pool_names: &std::collections::HashSet<String>,
) -> Option<(TokenStream2, Vec<TokenStream2>)> {
let marked: Vec<&Dep> = deps.iter().filter(|d| d.ready.is_some()).collect();
if marked.is_empty() {
return None;
}
let refs: Vec<TokenStream2> = marked
.iter()
.map(|d| {
let cfg = &d.cfg;
let ident = &d.ident;
if pool_names.contains(&ident.to_string()) {
quote!(#(#cfg)* &#ident[0])
} else {
quote!(#(#cfg)* &#ident)
}
})
.collect();
let any_cfg = marked.iter().any(|d| cfg_predicate(&d.cfg).is_some());
let len = if any_cfg {
let terms: Vec<TokenStream2> = marked
.iter()
.map(|d| match cfg_predicate(&d.cfg) {
None => quote!(1usize),
Some(pred) => quote!({
#[cfg(#pred)]
{
1usize
}
#[cfg(not(#pred))]
{
0usize
}
}),
})
.collect();
quote!(0usize #(+ #terms)*)
} else {
let n = marked.len();
quote!(#n)
};
Some((len, refs))
}
fn policy_type(expr: &Expr) -> SynResult<Path> {
if let Expr::Call(call) = expr
&& let Expr::Path(p) = &*call.func
{
let n = p.path.segments.len();
if n >= 2 {
let segs: Punctuated<_, Token![::]> =
p.path.segments.iter().take(n - 1).cloned().collect();
return Ok(Path {
leading_colon: p.path.leading_colon,
segments: segs,
});
}
}
Err(syn::Error::new_spanned(
expr,
"pool `policy:` must be a `Type::new(..)` constructor (e.g. `DeferredShrink::new(..)`), \
or give the type explicitly: `policy: <Type> = <expr>`",
))
}
struct Slot {
cfg_pred: Option<TokenStream2>,
reference: TokenStream2,
deps: Vec<Dep>,
fragment: Option<String>,
}
fn node_spawn(
ident: &Ident,
spawn: &Option<Expr>,
executor: &Option<Ident>,
resources: &[ResourceDecl],
state: Option<&(syn::Type, Expr)>,
spawn_fn: &TokenStream2,
helpers: &HelperIdents,
) -> SynResult<TokenStream2> {
let take_prelude: Vec<TokenStream2> = resources
.iter()
.enumerate()
.map(|(i, r)| {
let cfg = &r.cfg;
let res = &r.ident;
let var = format_ident!("__r{}", i);
let getter = if r.shared.is_some() {
quote!(get)
} else {
quote!(take)
};
quote! {
#(#cfg)*
let #var = #res
.#getter()
.ok_or(::embassy_executor::SpawnError::Busy)?;
}
})
.collect();
let res_args: Vec<TokenStream2> = resources
.iter()
.enumerate()
.map(|(i, r)| {
let cfg = &r.cfg;
let var = format_ident!("__r{}", i);
quote!(#(#cfg)* #var)
})
.collect();
let try_box = &helpers.try_box;
let (state_prelude, state_arg) = match state {
Some((_, init)) => (
quote! {
let __state = #try_box(#init)
.ok_or(::embassy_executor::SpawnError::Busy)?;
},
vec![quote!(__state)],
),
None => (quote!(), vec![]),
};
Ok(match (spawn, executor) {
(None, None) => quote!(::core::option::Option::None),
(None, Some(ex)) => {
return Err(syn::Error::new_spanned(
ex,
"`executor:` requires a `spawn:` (a parked node is spawned by the \
application, which picks its own spawner)",
));
}
(Some(e @ (Expr::Path(_) | Expr::Call(_))), executor) => {
let mut lead: Vec<TokenStream2> = vec![quote!(&#ident)];
lead.extend(res_args.iter().cloned());
lead.extend(state_arg.iter().cloned());
let call = inject_call_with(e, &lead)?;
match executor {
None => {
let stmts = spawn_stmts(&call, "e!(&#ident), "e!(s));
quote!(::core::option::Option::Some(
(|s| {
#state_prelude
#(#take_prelude)*
#stmts
::core::result::Result::Ok(())
}) as #spawn_fn
))
}
Some(ex) => {
let stmts = spawn_stmts(&call, "e!(&#ident), "e!(__sp));
quote!(::core::option::Option::Some(
(|_s| {
let __sp = #ex
.get()
.ok_or(::embassy_executor::SpawnError::Busy)?;
#state_prelude
#(#take_prelude)*
#stmts
::core::result::Result::Ok(())
}) as #spawn_fn
))
}
}
}
(Some(_), Some(ex)) => {
return Err(syn::Error::new_spanned(
ex,
"`executor:` cannot be combined with a verbatim spawn closure (the \
closure owns the spawn; use the named SpawnerSlot inside it instead)",
));
}
(Some(e), None) => quote!(::core::option::Option::Some((#e) as #spawn_fn)),
})
}
fn spawn_stmts(call: &TokenStream2, node_ref: &TokenStream2, sp: &TokenStream2) -> TokenStream2 {
if cfg!(feature = "trace") {
quote! {
let __token = #call?;
(#node_ref).adopt(&__token);
#sp.spawn(__token);
}
} else if cfg!(feature = "metadata-names") {
quote! {
let __token = #call?;
(#node_ref).stamp_name(&__token);
#sp.spawn(__token);
}
} else {
quote!(#sp.spawn(#call?);)
}
}
#[allow(clippy::too_many_arguments)]
fn emit_shell(
owner: &Ident,
cfg: &[Attribute],
worker: &Expr,
pool_size: usize,
resources: &[ResourceDecl],
exit: Option<&syn::Type>,
state: Option<&(syn::Type, Expr)>,
pool_member: bool,
cr: &TokenStream2,
helpers: &HelperIdents,
) -> SynResult<(TokenStream2, Expr)> {
if !matches!(worker, Expr::Path(_) | Expr::Call(_)) {
return Err(syn::Error::new_spanned(
worker,
"`task:` names an async worker fn — a path or a partial call like \
`worker(args)`; for a closure or a ready spawn fn use `spawn:`",
));
}
let shell = format_ident!("__sv_task_{}", owner.to_string().to_lowercase());
let by_value = |r: &ResourceDecl| r.consume.is_some() || r.shared.is_some();
let res_params: Vec<TokenStream2> = resources
.iter()
.enumerate()
.map(|(i, r)| {
let cfg = &r.cfg;
let var = format_ident!("__r{}", i);
let ty = &r.ty;
if by_value(r) {
quote!(#(#cfg)* #var: #ty)
} else if pool_member {
let slot_param = format_ident!("__r{}_slot", i);
quote!(#(#cfg)* mut #var: #ty, #(#cfg)* #slot_param: &'static #cr::ResourceSlot<#ty>)
} else {
quote!(#(#cfg)* mut #var: #ty)
}
})
.collect();
let res_leases: Vec<TokenStream2> = resources
.iter()
.enumerate()
.map(|(i, r)| {
let cfg = &r.cfg;
let var = format_ident!("__r{}", i);
if by_value(r) {
quote!(#(#cfg)* #var)
} else {
quote!(#(#cfg)* &mut #var)
}
})
.collect();
let restores: Vec<TokenStream2> = resources
.iter()
.enumerate()
.filter(|(_, r)| !by_value(r))
.map(|(i, r)| {
let cfg = &r.cfg;
let var = format_ident!("__r{}", i);
if pool_member {
let slot_param = format_ident!("__r{}_slot", i);
quote!(#(#cfg)* #slot_param.restore(#var);)
} else {
let res = &r.ident;
quote!(#(#cfg)* #res.restore(#var);)
}
})
.collect();
let alloc_alias = &helpers.alloc_alias;
let (state_param, state_lease, state_drop) = match state {
Some((ty, _)) => (
quote!(, mut __state: #alloc_alias::boxed::Box<#ty>),
vec![quote!(&mut *__state)],
quote!(::core::mem::drop(__state);),
),
None => (quote!(), vec![], quote!()),
};
let mut lead: Vec<TokenStream2> = vec![quote!(__node)];
lead.extend(res_leases);
lead.extend(state_lease);
let call = inject_call_with(worker, &lead)?;
let ps = LitInt::new(&pool_size.to_string(), proc_macro2::Span::call_site());
let allow_unreachable = quote!(#[allow(unreachable_code)]);
let (bind_out, provide_exit) = match exit {
Some(_) => {
let exit_ident = format_ident!("{}_EXIT", owner);
(quote!(let __out =), quote!(#exit_ident.provide(__out);))
}
None => (quote!(), quote!()),
};
let def = quote! {
#(#cfg)*
#[::embassy_executor::task(pool_size = #ps)]
#allow_unreachable
async fn #shell(__node: &'static #cr::TaskNode #(, #res_params)* #state_param) {
#bind_out #call.await;
#state_drop
#(#restores)*
#provide_exit
__node.mark_exited();
}
};
let path: Expr = syn::parse_quote!(#shell);
Ok((def, path))
}
fn emit_node(
n: &NodeItem,
cr: &TokenStream2,
spawn_fn: &TokenStream2,
pool_names: &std::collections::HashSet<String>,
helpers: &HelperIdents,
) -> SynResult<(TokenStream2, Slot)> {
let ident = &n.ident;
let cfg = &n.cfg;
let mode = &n.mode;
let name = name_string(&n.ident);
let disabled = n.disabled;
let (shell_def, spawn_expr) = match &n.source {
Some(TaskSource::Shell(worker)) => {
let ps = match &n.pool_size {
Some(l) => l.base10_parse::<usize>()?,
None => 1,
};
let (def, path) = emit_shell(
ident,
cfg,
worker,
ps,
&n.resources,
n.exit.as_ref(),
n.state.as_ref(),
false,
cr,
helpers,
)?;
(def, Some(path))
}
Some(TaskSource::Spawn(e)) => (quote!(), Some(e.clone())),
None => (quote!(), None),
};
let spawn = node_spawn(
ident,
&spawn_expr,
&n.executor,
&n.resources,
n.state.as_ref(),
spawn_fn,
helpers,
)?;
let with_exec = match &n.executor {
Some(ex) => quote!( .with_executor(&#ex) ),
None => quote!(),
};
let (res_defs, with_res) = if n.resources.is_empty() {
(quote!(), quote!())
} else {
let gates_ident = format_ident!("__SV_GATES_{}", ident);
let slot_defs = n.resources.iter().filter(|r| r.shared.is_none()).map(|r| {
let ecfg = &r.cfg;
let res = &r.ident;
let ty = &r.ty;
let slot_ty = if r.local.is_some() {
let local = &helpers.local_slot;
quote!(#local<#ty>)
} else {
quote!(#cr::ResourceSlot<#ty>)
};
let doc = if r.consume.is_some() {
format!(
"Resource slot for node `{ident}` (generated by `supervisor_graph!`). \
Move the resource in with `.provide(..)` before `Supervisor::start`. \
`consume`: the worker owns (and may drop) the value, so the slot is \
empty after the task exits — re-`provide()` before any respawn."
)
} else {
format!(
"Resource slot for node `{ident}` (generated by `supervisor_graph!`). \
Move the resource in with `.provide(..)` before `Supervisor::start`."
)
};
quote! {
#(#cfg)*
#(#ecfg)*
#[doc = #doc]
pub static #res: #slot_ty = <#slot_ty>::new();
}
});
let (gates_len, gate_refs) = gate_tokens(&n.resources);
(
quote! {
#(#slot_defs)*
#(#cfg)*
static #gates_ident: [&'static dyn #cr::ResourceGate; #gates_len] =
[#(#gate_refs),*];
},
quote!( .with_resources(&#gates_ident) ),
)
};
let with_timeout = match &n.slot_timeout {
Some(ms) => quote!( .with_slot_timeout(#cr::_export::Duration::from_millis(#ms)) ),
None => quote!(),
};
let (ready_def, with_ready) = match ready_tokens(&n.deps, pool_names) {
Some((len, refs)) => {
let ready_ident = format_ident!("__SV_READY_{}", ident);
(
quote! {
#(#cfg)*
static #ready_ident: [&'static #cr::TaskNode; #len] = [#(#refs),*];
},
quote!( .with_ready_deps(&#ready_ident) ),
)
}
None => (quote!(), quote!()),
};
let exit_def = match &n.exit {
Some(ty) => {
let exit_ident = format_ident!("{}_EXIT", ident);
let doc = format!(
"Exit-value slot for node `{ident}` (generated by `supervisor_graph!`). \
The generated shell `provide()`s the worker's return value here just \
before recording the exit; read it with `.wait_take()` (or `.take()` \
after `has_exited()`). Overwritten by the next completion."
);
quote! {
#(#cfg)*
#[doc = #doc]
pub static #exit_ident: #cr::ResourceSlot<#ty> =
#cr::ResourceSlot::new();
}
}
None => quote!(),
};
let node_doc = format!(
"Supervised node `{ident}` (`{mode}`), generated by `supervisor_graph!`. \
Pass it to the supervisor's per-node verbs (`start_node`, `stop_node`, \
`resume_node`, `activate`/`deactivate`); the worker gets the same \
`&'static TaskNode` for the task-side protocol."
);
let def = quote! {
#res_defs
#exit_def
#ready_def
#shell_def
#(#cfg)*
#[doc = #node_doc]
pub static #ident: #cr::TaskNode =
#cr::TaskNode::new(#name, #cr::Mode::#mode, #spawn, #disabled)
#with_exec #with_res #with_timeout #with_ready;
};
let slot = Slot {
cfg_pred: cfg_predicate(cfg),
reference: quote!(&#ident),
deps: n.deps.clone(),
fragment: n.fragment.clone(),
};
Ok((def, slot))
}
fn emit_pool(
p: &PoolItem,
cr: &TokenStream2,
spawn_fn: &TokenStream2,
pool_names: &std::collections::HashSet<String>,
helpers: &HelperIdents,
) -> SynResult<(Vec<TokenStream2>, TokenStream2, Vec<Slot>)> {
let ident = &p.ident;
let cfg = &p.cfg;
let lname = name_string(&p.ident);
let pool_static = format_ident!("{}_POOL", ident);
let k = p.modes.len();
let lit_bounds = match (&p.min, &p.max) {
(Expr::Lit(lmin), Expr::Lit(lmax)) => match (&lmin.lit, &lmax.lit) {
(syn::Lit::Int(imin), syn::Lit::Int(imax)) => {
Some((imin.base10_parse::<u8>()?, imax.base10_parse::<u8>()?))
}
_ => None,
},
_ => None,
};
if let Some((min_v, max_v)) = lit_bounds {
if min_v > max_v {
return Err(syn::Error::new_spanned(
&p.min,
format!("pool `min:` ({min_v}) must not exceed `max:` ({max_v})"),
));
}
if usize::from(max_v) > k {
return Err(syn::Error::new_spanned(
&p.max,
format!("pool `max:` ({max_v}) exceeds the declared member count ({k})"),
));
}
}
if !p.resources.is_empty() && matches!(p.source, TaskSource::Spawn(_)) {
return Err(syn::Error::new_spanned(
&p.resources[0].ident,
"pool `resources:` requires `task:` — the values are handed to the \
generated shell as arguments (and lend entries restored by it); a \
`spawn:` task fn manages its own arguments",
));
}
if let Some((ty, _)) = &p.state
&& matches!(p.source, TaskSource::Spawn(_))
{
return Err(syn::Error::new_spanned(
ty,
"pool `state:` requires `task:` — the generated shell owns the boxed \
state across the worker call; a `spawn:` task fn can Box its own",
));
}
let (shell_def, member_expr) = match &p.source {
TaskSource::Spawn(e) => (quote!(), e.clone()),
TaskSource::Shell(worker) => emit_shell(
ident,
cfg,
worker,
k,
&p.resources,
None,
p.state.as_ref(),
true,
cr,
helpers,
)?,
};
let res_args: Vec<TokenStream2> = p
.resources
.iter()
.enumerate()
.flat_map(|(i, r)| {
let ecfg = &r.cfg;
let var = format_ident!("__r{}", i);
let res = &r.ident;
if r.shared.is_none() && r.consume.is_none() {
vec![quote!(#(#ecfg)* #var), quote!(#(#ecfg)* &#res[I])]
} else {
vec![quote!(#(#ecfg)* #var)]
}
})
.collect();
let try_box = &helpers.try_box;
let (state_prelude, state_arg) = match &p.state {
Some((_, init)) => (
quote! {
let __state = #try_box(#init)
.ok_or(::embassy_executor::SpawnError::Busy)?;
},
vec![quote!(__state)],
),
None => (quote!(), vec![]),
};
let mut lead: Vec<TokenStream2> = vec![quote!(&#ident[I])];
lead.extend(res_args);
lead.extend(state_arg);
let call = inject_call_with(&member_expr, &lead)?;
let (param, prelude, sp_tokens) = match &p.executor {
None => (quote!(s), quote!(), quote!(s)),
Some(ex) => (
quote!(_s),
quote! {
let __sp = #ex
.get()
.ok_or(::embassy_executor::SpawnError::Busy)?;
},
quote!(__sp),
),
};
let get_prelude: Vec<TokenStream2> = p
.resources
.iter()
.enumerate()
.map(|(i, r)| {
let ecfg = &r.cfg;
let res = &r.ident;
let var = format_ident!("__r{}", i);
if r.shared.is_some() {
quote! {
#(#ecfg)*
let #var = #res
.get()
.ok_or(::embassy_executor::SpawnError::Busy)?;
}
} else {
quote! {
#(#ecfg)*
let #var = #res[I]
.take()
.ok_or(::embassy_executor::SpawnError::Busy)?;
}
}
})
.collect();
let pool_spawn_stmts = spawn_stmts(&call, "e!(&#ident[I]), &sp_tokens);
let wrapper = format_ident!("spawn_{}", lname);
let mut defs: Vec<TokenStream2> = Vec::new();
defs.push(shell_def);
defs.push(quote! {
#(#cfg)*
fn #wrapper<const I: usize>(
#param: ::embassy_executor::Spawner,
) -> ::core::result::Result<(), ::embassy_executor::SpawnError> {
#prelude
#state_prelude
#(#get_prelude)*
#pool_spawn_stmts
::core::result::Result::Ok(())
}
});
let member_spawn: Vec<TokenStream2> = (0..k).map(|j| quote!(#wrapper::<#j>)).collect();
let member_with_exec = match &p.executor {
Some(ex) => quote!( .with_executor(&#ex) ),
None => quote!(),
};
for r in p.resources.iter().filter(|r| r.shared.is_none()) {
let ecfg = &r.cfg;
let res = &r.ident;
let ty = &r.ty;
let doc = format!(
"Per-member resource slots for pool `{ident}` (generated by \
`supervisor_graph!`): member `I` takes/restores element `I`. \
Provide at least the floor members' elements before \
`Supervisor::start`; a member whose element is empty fail-closes \
its (re)spawn with `SpawnError::Busy`."
);
defs.push(quote! {
#(#cfg)*
#(#ecfg)*
#[doc = #doc]
pub static #res: [#cr::ResourceSlot<#ty>; #k] =
[const { #cr::ResourceSlot::new() }; #k];
});
}
let member_with_res: Vec<TokenStream2> = if p.resources.is_empty() {
(0..k).map(|_| quote!()).collect()
} else {
let any_cfg = p.resources.iter().any(|r| cfg_predicate(&r.cfg).is_some());
let gates_len = if any_cfg {
let terms: Vec<TokenStream2> = p
.resources
.iter()
.map(|r| match cfg_predicate(&r.cfg) {
None => quote!(1usize),
Some(pred) => quote!({
#[cfg(#pred)]
{
1usize
}
#[cfg(not(#pred))]
{
0usize
}
}),
})
.collect();
quote!(0usize #(+ #terms)*)
} else {
let n = p.resources.len();
quote!(#n)
};
(0..k)
.map(|j| {
let gates_ident = format_ident!("__SV_GATES_{}_{}", ident, j);
let gate_refs: Vec<TokenStream2> = p
.resources
.iter()
.map(|r| {
let ecfg = &r.cfg;
let res = &r.ident;
if r.shared.is_some() {
quote!(#(#ecfg)* &#res)
} else {
quote!(#(#ecfg)* &#res[#j])
}
})
.collect();
defs.push(quote! {
#(#cfg)*
static #gates_ident: [&'static dyn #cr::ResourceGate; #gates_len] =
[#(#gate_refs),*];
});
quote!( .with_resources(&#gates_ident) )
})
.collect()
};
let member_with_ready = match ready_tokens(&p.deps, pool_names) {
Some((len, refs)) => {
let ready_ident = format_ident!("__SV_READY_{}", ident);
defs.push(quote! {
#(#cfg)*
static #ready_ident: [&'static #cr::TaskNode; #len] = [#(#refs),*];
});
quote!( .with_ready_deps(&#ready_ident) )
}
None => quote!(),
};
let member_with_timeout = match &p.slot_timeout {
Some(ms) => quote!( .with_slot_timeout(#cr::_export::Duration::from_millis(#ms)) ),
None => quote!(),
};
let members = p
.modes
.iter()
.zip(&member_spawn)
.enumerate()
.map(|(j, (mode, sp))| {
let nm = format!("{lname}{j}");
let with_res = &member_with_res[j];
quote! {
#cr::TaskNode::new(
#nm, #cr::Mode::#mode,
::core::option::Option::Some((#sp) as #spawn_fn), false,
) #member_with_exec #with_res #member_with_timeout #member_with_ready
}
});
defs.push(quote! {
#(#cfg)*
#[doc = concat!("Pool `", stringify!(#ident), "`'s members, one `TaskNode` per slot \
(index = member index). Index it for the per-node verbs; the pool itself is \
`", stringify!(#ident), "_POOL`.")]
pub static #ident: [#cr::TaskNode; #k] = [ #(#members),* ];
});
let min_const = format_ident!("{}_MIN", ident);
let max_const = format_ident!("{}_MAX", ident);
let members_const = format_ident!("{}_MEMBERS", ident);
let (min_tokens, max_tokens, bound_asserts) = match lit_bounds {
Some((min_v, max_v)) => {
let (min_u, max_u) = (usize::from(min_v), usize::from(max_v));
(quote!(#min_u), quote!(#max_u), quote!())
}
None => {
let (min_e, max_e) = (&p.min, &p.max);
(
quote!({ #min_e }),
quote!({ #max_e }),
quote! {
#(#cfg)*
const _: () = ::core::assert!(
#min_const <= #max_const,
"pool `min:` must not exceed `max:`",
);
#(#cfg)*
const _: () = ::core::assert!(
#max_const <= #members_const,
"pool `max:` exceeds the declared member count",
);
#(#cfg)*
const _: () = ::core::assert!(
#max_const <= 255,
"pool `max:` exceeds 255 (ElasticPool bounds are u8)",
);
},
)
}
};
defs.push(quote! {
#(#cfg)*
#[doc = concat!("Pool `", stringify!(#ident), "`'s `min:` floor (validated at expansion or by const assert).")]
pub const #min_const: usize = #min_tokens;
#(#cfg)*
#[doc = concat!("Pool `", stringify!(#ident), "`'s `max:` scaling ceiling — the most members ever running concurrently.")]
pub const #max_const: usize = #max_tokens;
#(#cfg)*
#[doc = concat!("Pool `", stringify!(#ident), "`'s declared member count (the `[TaskNode; K]` array length).")]
pub const #members_const: usize = #k;
#bound_asserts
});
let member_refs = (0..k).map(|j| quote!(&#ident[#j]));
let policy = &p.policy;
let policy_ty = match &p.policy_ty {
Some(ty) => quote!(#ty),
None => {
let path = policy_type(policy)?;
quote!(#path)
}
};
defs.push(quote! {
#(#cfg)*
#[doc = concat!("The `ElasticPool` over the `", stringify!(#ident), "` members: \
the `min:`/`max:` bounds and the scaling policy `Supervisor::run_pools` \
drives. Also reachable through `GRAPH.pools`.")]
pub static #pool_static: #cr::ElasticPool<#policy_ty> = #cr::ElasticPool {
nodes: &[ #(#member_refs),* ],
min: #min_const as u8,
max: #max_const as u8,
policy: #policy,
};
});
let pool_entry = quote!( #(#cfg)* &#pool_static );
let pred = cfg_predicate(cfg);
let slots = (0..k)
.map(|j| Slot {
cfg_pred: pred.clone(),
reference: quote!(&#ident[#j]),
deps: p.deps.clone(),
fragment: p.fragment.clone(),
})
.collect();
Ok((defs, pool_entry, slots))
}
fn slot_tables(
slots: &[Slot],
names: &HashMap<String, usize>,
) -> SynResult<(Vec<TokenStream2>, Vec<TokenStream2>)> {
let mut all_entries: Vec<TokenStream2> = Vec::new();
let mut deps_entries: Vec<TokenStream2> = Vec::new();
for slot in slots {
let reference = &slot.reference;
all_entries.push(match &slot.cfg_pred {
None => quote!(::core::option::Option::Some(#reference)),
Some(pred) => quote!({
#[cfg(#pred)]
{ ::core::option::Option::Some(#reference) }
#[cfg(not(#pred))]
{ ::core::option::Option::None }
}),
});
let mut dep_toks: Vec<TokenStream2> = Vec::new();
let mut seen: Vec<(u8, String)> = Vec::new();
for d in &slot.deps {
let idx = match names.get(&d.ident.to_string()) {
Some(&i) => i as u8,
None => {
return Err(syn::Error::new_spanned(
&d.ident,
format!(
"unknown dependency `{}` — not a declared node or pool{}",
d.ident,
fragment_suffix(&slot.fragment),
),
));
}
};
let cfg = &d.cfg;
let cfg_key = quote!( #(#cfg)* ).to_string();
if seen.iter().any(|(i, k)| *i == idx && *k == cfg_key) {
return Err(syn::Error::new_spanned(
&d.ident,
format!("duplicate dependency `{}`", d.ident),
));
}
seen.push((idx, cfg_key));
dep_toks.push(quote!( #(#cfg)* #idx ));
}
deps_entries.push(quote!( &[ #(#dep_toks),* ] ));
}
Ok((all_entries, deps_entries))
}
fn expand(graph: GraphSpec) -> SynResult<TokenStream2> {
let cr = quote!(::embassy_supervisor);
let helpers = HelperIdents::new(graph.name.as_ref());
let spawn_fn = quote!(
fn(
::embassy_executor::Spawner,
) -> ::core::result::Result<(), ::embassy_executor::SpawnError>
);
let mut defs: Vec<TokenStream2> = Vec::new();
let mut pool_entries: Vec<TokenStream2> = Vec::new();
let mut slots: Vec<Slot> = Vec::new();
let mut names: HashMap<String, usize> = HashMap::new();
let any_state = graph.items.iter().any(|item| match item {
Item::Node(n) => n.state.is_some(),
Item::Pool(p) => p.state.is_some(),
Item::Executor(_) => false,
});
if any_state {
let try_box = &helpers.try_box;
let alloc_alias = &helpers.alloc_alias;
defs.push(quote! {
extern crate alloc as #alloc_alias;
#[doc(hidden)]
fn #try_box<T>(init: T) -> ::core::option::Option<#alloc_alias::boxed::Box<T>> {
let layout = ::core::alloc::Layout::new::<T>();
if layout.size() == 0 {
::core::mem::forget(init);
return ::core::option::Option::Some(unsafe {
#alloc_alias::boxed::Box::from_raw(
::core::ptr::NonNull::<T>::dangling().as_ptr(),
)
});
}
unsafe {
let p = #alloc_alias::alloc::alloc(layout) as *mut T;
if p.is_null() {
return ::core::option::Option::None; }
::core::ptr::write(p, init);
::core::option::Option::Some(#alloc_alias::boxed::Box::from_raw(p))
}
}
});
}
let any_local = graph
.items
.iter()
.any(|item| item_resources(item).iter().any(|r| r.local.is_some()));
if any_local {
let local = helpers.local_slot.clone();
let cell = quote!(::core::cell::Cell<::core::option::Option<T>>);
let raw = quote!(#cr::_export::CriticalSectionRawMutex);
let signal = quote!(#cr::_export::Signal<#raw, ()>);
defs.push(quote! {
#[allow(dead_code)]
pub struct #local<T> {
slot: #cr::_export::BlockingMutex<#raw, #cell>,
filled: #signal,
}
unsafe impl<T> ::core::marker::Sync for #local<T> {}
#[allow(dead_code)]
impl<T> #local<T> {
pub const fn new() -> Self {
Self {
slot: #cr::_export::BlockingMutex::new(
::core::cell::Cell::new(::core::option::Option::None),
),
filled: #cr::_export::Signal::new(),
}
}
pub fn provide(&self, value: T) {
self.slot.lock(|c| c.set(::core::option::Option::Some(value)));
self.filled.signal(());
}
pub fn take(&self) -> ::core::option::Option<T> {
self.slot.lock(::core::cell::Cell::take)
}
pub fn restore(&self, value: T) {
self.provide(value);
}
}
#[allow(dead_code)]
impl<T: ::core::marker::Copy> #local<T> {
pub fn get(&self) -> ::core::option::Option<T> {
self.slot.lock(|c| {
let v = c.take();
c.set(v);
v
})
}
}
impl<T> ::core::default::Default for #local<T> {
fn default() -> Self {
Self::new()
}
}
impl<T> #cr::ResourceGate for #local<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 {
&self.filled
}
}
});
}
let helpers = HelperIdents::new(graph.name.as_ref());
let executor_names: Vec<String> = graph
.items
.iter()
.filter_map(|i| match i {
Item::Executor(x) => Some(x.ident.to_string()),
_ => None,
})
.collect();
let pool_names: std::collections::HashSet<String> = graph
.items
.iter()
.filter_map(|i| match i {
Item::Pool(p) => Some(p.ident.to_string()),
_ => None,
})
.collect();
struct SharedPlan<'a> {
decl: &'a ResourceDecl,
sig: String,
preds: Vec<Option<TokenStream2>>,
owners: Vec<String>,
}
let mut shared_plans: Vec<(String, SharedPlan)> = Vec::new();
{
let mut taken: HashSet<String> = HashSet::new();
for item in &graph.items {
let Some((owner, item_cfg)) = item_ident_cfg(item) else {
continue;
};
let item_pred = cfg_predicate(item_cfg);
for r in item_resources(item) {
let key = r.ident.to_string();
if executor_names.contains(&key) {
return Err(syn::Error::new_spanned(
&r.ident,
format!(
"resource name `{}` shadows an `executor {};` slot — \
both are statics at the declaration site",
r.ident, r.ident
),
));
}
let pred = match (item_pred.clone(), cfg_predicate(&r.cfg)) {
(None, None) => None,
(Some(p), None) | (None, Some(p)) => Some(p),
(Some(a), Some(b)) => Some(quote!(all(#a, #b))),
};
if r.shared.is_some() {
if taken.contains(&key) {
return Err(syn::Error::new_spanned(
&r.ident,
format!(
"`{}` is already a take-kind resource elsewhere in \
the graph — a name is either one exclusive slot or \
one `shared` slot, not both",
r.ident
),
));
}
let sig = r.shared_signature();
match shared_plans.iter_mut().find(|(k, _)| *k == key) {
Some((_, plan)) => {
if plan.sig != sig {
return Err(syn::Error::new_spanned(
&r.ident,
format!(
"shared resource `{}` re-declared with a \
different shape: `{}` here vs `{}` on \
`{}` — every declaration of a shared slot \
must repeat the same kind markers and type",
r.ident, sig, plan.sig, plan.owners[0]
),
));
}
plan.preds.push(pred);
plan.owners.push(owner.to_string());
}
None => shared_plans.push((
key,
SharedPlan {
decl: r,
sig,
preds: vec![pred],
owners: vec![owner.to_string()],
},
)),
}
} else {
if !taken.insert(key.clone()) || shared_plans.iter().any(|(k, _)| *k == key) {
return Err(syn::Error::new_spanned(
&r.ident,
format!(
"duplicate resource name `{}` — resource slots are \
statics and must be unique across the graph (only \
`shared` entries may repeat a name)",
r.ident
),
));
}
}
}
}
}
for (_, plan) in &shared_plans {
let res = &plan.decl.ident;
let ty = &plan.decl.ty;
let slot_ty = if plan.decl.local.is_some() {
let local = &helpers.local_slot;
quote!(#local<#ty>)
} else {
quote!(#cr::ResourceSlot<#ty>)
};
let cfg_attr = if plan.preds.iter().any(|p| p.is_none()) {
quote!()
} else {
let preds = plan.preds.iter().flatten();
quote!(#[cfg(any(#(#preds),*))])
};
let doc = format!(
"Shared (fan-out) resource slot declared by `{}` (generated by \
`supervisor_graph!`). `provide()` the `Copy` handle before \
`Supervisor::start`; every consumer's glue copies it out with \
`get()`, so the slot STAYS FILLED — re-`provide()` only to replace \
the handle (e.g. after rebuilding the underlying driver).",
plan.owners.join("`, `"),
);
defs.push(quote! {
#cfg_attr
#[doc = #doc]
pub static #res: #slot_ty = <#slot_ty>::new();
});
}
for item in &graph.items {
match item {
Item::Node(n) => {
if let Some(ex) = &n.executor
&& !executor_names.contains(&ex.to_string())
{
return Err(syn::Error::new_spanned(
ex,
format!(
"unknown executor `{ex}`; declare it in the graph with \
`executor {ex};` (declared: [{}])",
executor_names.join(", ")
),
));
}
if names.insert(n.ident.to_string(), slots.len()).is_some() {
return Err(syn::Error::new_spanned(
&n.ident,
format!(
"duplicate node/pool name `{}`{}",
n.ident,
fragment_suffix(&n.fragment),
),
));
}
let (def, slot) = emit_node(n, &cr, &spawn_fn, &pool_names, &helpers)?;
defs.push(def);
slots.push(slot);
}
Item::Executor(x) => {
let (cfg, ident) = (&x.cfg, &x.ident);
defs.push(quote! {
#(#cfg)*
pub static #ident: #cr::SpawnerSlot = #cr::SpawnerSlot::new();
});
}
Item::Pool(p) => {
if cfg!(feature = "pool") {
if let Some(ex) = &p.executor
&& !executor_names.contains(&ex.to_string())
{
return Err(syn::Error::new_spanned(
ex,
format!(
"unknown executor `{ex}`; declare it in the graph with \
`executor {ex};` (declared: [{}])",
executor_names.join(", ")
),
));
}
let (pool_defs, pool_entry, pool_slots) =
emit_pool(p, &cr, &spawn_fn, &pool_names, &helpers)?;
if names.insert(p.ident.to_string(), slots.len()).is_some() {
return Err(syn::Error::new_spanned(
&p.ident,
format!(
"duplicate node/pool name `{}`{}",
p.ident,
fragment_suffix(&p.fragment),
),
));
}
defs.extend(pool_defs);
pool_entries.push(pool_entry);
slots.extend(pool_slots);
} else {
return Err(syn::Error::new_spanned(
&p.ident,
"a `pool` requires enabling embassy-supervisor's `pool` feature",
));
}
}
}
}
let m = slots.len();
if m > 256 {
return Err(syn::Error::new(
proc_macro2::Span::call_site(),
format!(
"supervisor_graph!: {m} node slots declared, but at most 256 are supported \
(including pool members) — graph indices are `u8`"
),
));
}
let (all_entries, deps_entries) = slot_tables(&slots, &names)?;
let pools_field = if cfg!(feature = "pool") {
quote!( pools: &[ #(#pool_entries),* ], )
} else {
quote!()
};
let trace_hooks = if cfg!(feature = "trace-hooks") && graph.name.is_none() {
quote! {
#[unsafe(no_mangle)]
fn _embassy_trace_poll_start(executor_id: u32) {
#cr::trace::on_poll_start(executor_id);
}
#[unsafe(no_mangle)]
fn _embassy_trace_task_new(_executor_id: u32, _task_id: u32) {}
#[unsafe(no_mangle)]
fn _embassy_trace_task_end(executor_id: u32, task_id: u32) {
#cr::trace::on_task_end(executor_id, task_id);
}
#[unsafe(no_mangle)]
fn _embassy_trace_task_exec_begin(executor_id: u32, task_id: u32) {
#cr::trace::on_task_exec_begin(executor_id, task_id);
}
#[unsafe(no_mangle)]
fn _embassy_trace_task_exec_end(executor_id: u32, task_id: u32) {
#cr::trace::on_task_exec_end(executor_id, task_id);
}
#[unsafe(no_mangle)]
fn _embassy_trace_task_ready_begin(_executor_id: u32, _task_id: u32) {}
#[unsafe(no_mangle)]
fn _embassy_trace_executor_idle(executor_id: u32) {
#cr::trace::on_executor_idle(executor_id);
}
}
} else {
quote!()
};
let graph_ident = graph
.name
.clone()
.unwrap_or_else(|| Ident::new("GRAPH", proc_macro2::Span::call_site()));
let (nodes_ident, deps_ident) = match &graph.name {
Some(n) => (
format_ident!("__SV_NODES_{}", n),
format_ident!("__SV_DEPS_{}", n),
),
None => (
Ident::new("NODES", proc_macro2::Span::call_site()),
Ident::new("DEPS", proc_macro2::Span::call_site()),
),
};
Ok(quote! {
#(#defs)*
static #nodes_ident: [::core::option::Option<&'static #cr::TaskNode>; #m] = [ #(#all_entries),* ];
const #deps_ident: [&'static [u8]; #m] = [ #(#deps_entries),* ];
pub static #graph_ident: #cr::Graph<#m> = #cr::Graph {
nodes: &#nodes_ident,
deps: &#deps_ident,
order: #cr::topo_sort_const(&#deps_ident),
#pools_field
};
#trace_hooks
})
}
#[proc_macro]
pub fn supervisor_graph(input: TokenStream) -> TokenStream {
let graph = syn::parse_macro_input!(input as GraphSpec);
expand(graph)
.unwrap_or_else(syn::Error::into_compile_error)
.into()
}
#[proc_macro]
pub fn supervisor_fragment(input: TokenStream) -> TokenStream {
fragment_expand(input.into())
.unwrap_or_else(syn::Error::into_compile_error)
.into()
}
fn fragment_expand(input: TokenStream2) -> SynResult<TokenStream2> {
struct FragmentSpec {
name: Ident,
items: TokenStream2,
}
impl Parse for FragmentSpec {
fn parse(input: ParseStream) -> SynResult<Self> {
input.parse::<kw::name>()?;
input.parse::<Token![:]>()?;
let name: Ident = input.parse()?;
input.parse::<Token![;]>()?;
let items: TokenStream2 = input.parse()?;
Ok(FragmentSpec { name, items })
}
}
let spec: FragmentSpec = syn::parse2(input)?;
let name = &spec.name;
validate_dollars(spec.items.clone())?;
let substituted = substitute_dollar_crate(spec.items.clone());
syn::parse2::<GraphSpec>(substituted)?;
let items = &spec.items;
let dollar = proc_macro2::Punct::new('$', proc_macro2::Spacing::Alone);
let doc = format!(
"A `supervisor_fragment!` relay (generated). Use from a compose site:\n\
`embassy_supervisor::compose_graph! {{ fragments: [{name}], graph: {{ .. }} }}`\n\
Not for direct invocation."
);
Ok(quote! {
#[doc = #doc]
#[macro_export]
macro_rules! #name {
(@emit #dollar cb:path, [#dollar(#dollar rest:tt)*], {#dollar(#dollar acc:tt)*}, {#dollar(#dollar g:tt)*}) => {
#dollar cb! { @next [#dollar(#dollar rest)*],
{#dollar(#dollar acc)* @fragment #name; #items @endfragment;},
{#dollar(#dollar g)*} }
};
}
})
}
fn validate_dollars(stream: TokenStream2) -> SynResult<()> {
use proc_macro2::TokenTree;
let mut iter = stream.into_iter().peekable();
while let Some(tt) = iter.next() {
match tt {
TokenTree::Group(g) => validate_dollars(g.stream())?,
TokenTree::Punct(p) if p.as_char() == '$' => match iter.peek() {
Some(TokenTree::Ident(i)) if i == "crate" => {}
_ => {
return Err(syn::Error::new(
p.span(),
"only `$crate` is permitted in a fragment — any other `$` \
would be read as a metavariable by the relay macro",
));
}
},
_ => {}
}
}
Ok(())
}
fn substitute_dollar_crate(stream: TokenStream2) -> TokenStream2 {
use proc_macro2::{TokenStream as TS, TokenTree};
let mut out = TS::new();
let mut iter = stream.into_iter().peekable();
while let Some(tt) = iter.next() {
match tt {
TokenTree::Group(g) => {
let inner = substitute_dollar_crate(g.stream());
let mut ng = proc_macro2::Group::new(g.delimiter(), inner);
ng.set_span(g.span());
out.extend([TokenTree::Group(ng)]);
}
TokenTree::Punct(p) if p.as_char() == '$' => {
if let Some(TokenTree::Ident(i)) = iter.peek()
&& i == "crate"
{
let span = iter.next().map(|t| t.span()).unwrap_or_else(|| p.span());
out.extend([TokenTree::Ident(Ident::new("__sv_fragment_crate", span))]);
} else {
out.extend([TokenTree::Punct(p)]);
}
}
other => out.extend([other]),
}
}
out
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn ready_marker_requires_feature() {
let res = syn::parse_str::<GraphSpec>(
"node NET = Terminate, deps: [];\n\
node HTTP = Terminate, deps: [NET ready];",
);
if cfg!(feature = "readiness") {
assert!(res.is_ok(), "marker accepted with the feature");
} else {
match res {
Ok(_) => panic!("marker accepted without the feature"),
Err(err) => assert!(
err.to_string().contains("requires the `readiness` feature"),
"unexpected error: {err}"
),
}
}
}
#[test]
fn unknown_dep_marker_rejected() {
match syn::parse_str::<GraphSpec>("node A = Terminate, deps: [B rdy];") {
Ok(_) => panic!("unknown marker accepted"),
Err(err) => assert!(err.to_string().contains("`ready` marker"), "got: {err}"),
}
}
}