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);
}
const LOCAL_SLOT_TYPE: &str = "__SvLocalResourceSlot";
#[derive(Clone)]
struct Dep {
cfg: Vec<Attribute>,
ident: 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()?;
deps.push(Dep { cfg, ident });
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>()?; 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>,
}
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: LitInt,
max: LitInt,
}
#[allow(clippy::large_enum_variant)]
enum Item {
Node(NodeItem),
Pool(PoolItem),
Executor(ExecutorItem),
}
struct GraphSpec {
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 mut items = Vec::new();
while !input.is_empty() {
let cfg = input.call(Attribute::parse_outer)?;
if input.peek(kw::node) {
items.push(Item::Node(parse_node(input, cfg)?));
} else if input.peek(kw::pool) {
items.push(Item::Pool(parse_pool(input, cfg)?));
} 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 { 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;
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 {
return Err(input.error(
"expected `spawn:`, `task:`, `pool_size:`, `executor:`, `resources:`, \
`slot_timeout:`, 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((_, 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,
})
}
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.shared.is_none()) {
return Err(syn::Error::new_spanned(
&bad.ident,
"only `shared` resources are supported on `pool` — members \
would contend for a take-kind slot's single instance; declare \
take/consume resources per-node",
));
}
input.parse::<Token![,]>()?;
decls
} else {
Vec::new()
};
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: LitInt = input.parse()?;
input.parse::<Token![,]>()?;
input.parse::<kw::max>()?;
input.parse::<Token![:]>()?;
let max: LitInt = 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,
})
}
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 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>,
}
fn node_spawn(
ident: &Ident,
spawn: &Option<Expr>,
executor: &Option<Ident>,
resources: &[ResourceDecl],
spawn_fn: &TokenStream2,
) -> 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();
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());
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| {
#(#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)?;
#(#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?);)
}
}
fn emit_shell(
owner: &Ident,
cfg: &[Attribute],
worker: &Expr,
pool_size: usize,
resources: &[ResourceDecl],
cr: &TokenStream2,
) -> 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 {
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 res = &r.ident;
let var = format_ident!("__r{}", i);
quote!(#(#cfg)* #res.restore(#var);)
})
.collect();
let mut lead: Vec<TokenStream2> = vec![quote!(__node)];
lead.extend(res_leases);
let call = inject_call_with(worker, &lead)?;
let ps = LitInt::new(&pool_size.to_string(), proc_macro2::Span::call_site());
let allow_unreachable = if restores.is_empty() {
quote!()
} else {
quote!(#[allow(unreachable_code)])
};
let def = quote! {
#(#cfg)*
#[::embassy_executor::task(pool_size = #ps)]
#allow_unreachable
async fn #shell(__node: &'static #cr::TaskNode #(, #res_params)*) {
#call.await;
#(#restores)*
}
};
let path: Expr = syn::parse_quote!(#shell);
Ok((def, path))
}
fn emit_node(
n: &NodeItem,
cr: &TokenStream2,
spawn_fn: &TokenStream2,
) -> 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, cr)?;
(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, spawn_fn)?;
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 = format_ident!("{LOCAL_SLOT_TYPE}");
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 def = quote! {
#res_defs
#shell_def
#(#cfg)*
pub static #ident: #cr::TaskNode =
#cr::TaskNode::new(#name, #cr::Mode::#mode, #spawn, #disabled)
#with_exec #with_res #with_timeout;
};
let slot = Slot {
cfg_pred: cfg_predicate(cfg),
reference: quote!(&#ident),
deps: n.deps.clone(),
};
Ok((def, slot))
}
fn emit_pool(
p: &PoolItem,
cr: &TokenStream2,
spawn_fn: &TokenStream2,
) -> 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 min_v: u8 = p.min.base10_parse()?;
let max_v: u8 = p.max.base10_parse()?;
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 shared values are handed \
to the generated shell as arguments; a `spawn:` task fn manages \
its own arguments",
));
}
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, cr)?,
};
let res_args: Vec<TokenStream2> = p
.resources
.iter()
.enumerate()
.map(|(i, r)| {
let ecfg = &r.cfg;
let var = format_ident!("__r{}", i);
quote!(#(#ecfg)* #var)
})
.collect();
let mut lead: Vec<TokenStream2> = vec![quote!(&#ident[I])];
lead.extend(res_args);
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);
quote! {
#(#ecfg)*
let #var = #res
.get()
.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
#(#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!(),
};
let gates_ident = format_ident!("__SV_GATES_{}", ident);
let member_with_res = if p.resources.is_empty() {
quote!()
} else {
quote!( .with_resources(&#gates_ident) )
};
let gates_def = if p.resources.is_empty() {
quote!()
} else {
let (gates_len, gate_refs) = gate_tokens(&p.resources);
quote! {
#(#cfg)*
static #gates_ident: [&'static dyn #cr::ResourceGate; #gates_len] =
[#(#gate_refs),*];
}
};
defs.push(gates_def);
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}");
quote! {
#cr::TaskNode::new(
#nm, #cr::Mode::#mode,
::core::option::Option::Some((#sp) as #spawn_fn), false,
) #member_with_exec #member_with_res #member_with_timeout
}
});
defs.push(quote! {
#(#cfg)*
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_u, max_u) = (usize::from(min_v), usize::from(max_v));
defs.push(quote! {
#(#cfg)*
#[doc = concat!("Pool `", stringify!(#ident), "`'s `min:` floor (validated at expansion).")]
pub const #min_const: usize = #min_u;
#(#cfg)*
#[doc = concat!("Pool `", stringify!(#ident), "`'s `max:` scaling ceiling — the most members ever running concurrently.")]
pub const #max_const: usize = #max_u;
#(#cfg)*
#[doc = concat!("Pool `", stringify!(#ident), "`'s declared member count (the `[TaskNode; K]` array length).")]
pub const #members_const: usize = #k;
});
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)*
pub static #pool_static: #cr::ElasticPool<#policy_ty> = #cr::ElasticPool {
nodes: &[ #(#member_refs),* ],
min: #min_v,
max: #max_v,
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(),
})
.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
),
));
}
};
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 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_local = graph
.items
.iter()
.any(|item| item_resources(item).iter().any(|r| r.local.is_some()));
if any_local {
let local = format_ident!("{LOCAL_SLOT_TYPE}");
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 executor_names: Vec<String> = graph
.items
.iter()
.filter_map(|i| match i {
Item::Executor(x) => Some(x.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 = format_ident!("{LOCAL_SLOT_TYPE}");
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),
));
}
let (def, slot) = emit_node(n, &cr, &spawn_fn)?;
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)?;
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),
));
}
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") {
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!()
};
Ok(quote! {
#(#defs)*
static NODES: [::core::option::Option<&'static #cr::TaskNode>; #m] = [ #(#all_entries),* ];
const DEPS: [&'static [u8]; #m] = [ #(#deps_entries),* ];
pub static GRAPH: #cr::Graph<#m> = #cr::Graph {
nodes: &NODES,
deps: &DEPS,
order: #cr::topo_sort_const(&DEPS),
#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()
}