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);
}
#[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 {
ident: Ident,
ty: Type,
}
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 ident: Ident = content.parse()?;
content.parse::<Token![:]>()?;
let ty: Type = content.parse()?;
resources.push(ResourceDecl { ident, ty });
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>,
}
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>,
min: LitInt,
max: LitInt,
}
#[allow(clippy::large_enum_variant)]
enum Item {
Node(NodeItem),
Pool(PoolItem),
Executor(ExecutorItem),
}
struct GraphSpec {
items: Vec<Item>,
}
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;
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 {
return Err(input.error(
"expected `spawn:`, `task:`, `pool_size:`, `executor:`, `resources:`, or `disabled`",
));
}
}
input.parse::<Token![;]>()?;
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",
));
}
}
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(),
})
}
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![,]>()?;
if input.peek(kw::resources) {
let k = input.parse::<kw::resources>()?;
return Err(syn::Error::new_spanned(
k,
"`resources:` is not supported on `pool` — members would contend \
for a single instance; declare per-node",
));
}
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()?;
input.parse::<Token![;]>()?;
Ok(PoolItem {
cfg,
ident,
modes,
deps,
source,
policy,
policy_ty,
executor,
min,
max,
})
}
fn name_string(ident: &Ident) -> String {
ident.to_string().to_lowercase().replace('_', "-")
}
fn inject_node_call(task: &Expr, node_ref: &TokenStream2) -> SynResult<TokenStream2> {
inject_call_with(task, core::slice::from_ref(node_ref))
}
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 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 res = &r.ident;
let var = format_ident!("__r{}", i);
quote! {
let #var = #res
.take()
.ok_or(::embassy_executor::SpawnError::Busy)?;
}
})
.collect();
let res_args: Vec<TokenStream2> = (0..resources.len())
.map(|i| {
let var = format_ident!("__r{}", i);
quote!(#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 res_params: Vec<TokenStream2> = resources
.iter()
.enumerate()
.map(|(i, r)| {
let var = format_ident!("__r{}", i);
let ty = &r.ty;
quote!(mut #var: #ty)
})
.collect();
let res_leases: Vec<TokenStream2> = (0..resources.len())
.map(|i| {
let var = format_ident!("__r{}", i);
quote!(&mut #var)
})
.collect();
let restores: Vec<TokenStream2> = resources
.iter()
.enumerate()
.map(|(i, r)| {
let res = &r.ident;
let var = format_ident!("__r{}", i);
quote!(#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 def = quote! {
#(#cfg)*
#[::embassy_executor::task(pool_size = #ps)]
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().map(|r| {
let res = &r.ident;
let ty = &r.ty;
let doc = format!(
"Resource slot for node `{ident}` (generated by `supervisor_graph!`). \
Move the resource in with `.provide(..)` before `Supervisor::start`."
);
quote! {
#(#cfg)*
#[doc = #doc]
pub static #res: #cr::ResourceSlot<#ty> = #cr::ResourceSlot::new();
}
});
let gate_refs = n.resources.iter().map(|r| {
let res = &r.ident;
quote!(&#res)
});
let gate_count = n.resources.len();
(
quote! {
#(#slot_defs)*
#(#cfg)*
static #gates_ident: [&'static dyn #cr::ResourceGate; #gate_count] =
[#(#gate_refs),*];
},
quote!( .with_resources(&#gates_ident) ),
)
};
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;
};
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})"),
));
}
let (shell_def, member_expr) = match &p.source {
TaskSource::Spawn(e) => (quote!(), e.clone()),
TaskSource::Shell(worker) => emit_shell(ident, cfg, worker, k, &[], cr)?,
};
let call = inject_node_call(&member_expr, "e!(&#ident[I]))?;
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 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
#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 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
}
});
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 executor_names: Vec<String> = graph
.items
.iter()
.filter_map(|i| match i {
Item::Executor(x) => Some(x.ident.to_string()),
_ => None,
})
.collect();
{
let mut seen: HashSet<String> = HashSet::new();
for item in &graph.items {
if let Item::Node(n) = item {
for r in &n.resources {
let key = r.ident.to_string();
if !seen.insert(key.clone()) || executor_names.contains(&key) {
return Err(syn::Error::new_spanned(
&r.ident,
format!(
"duplicate resource name `{}` — resource slots are \
statics and must be unique across the graph",
r.ident
),
));
}
}
}
}
}
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()
}