embassy_supervisor_macros/lib.rs
1//! Proc-macro for `embassy-supervisor`.
2//!
3//! `supervisor_graph!` is the **single source** of a task graph: it declares the
4//! nodes (and an optional elastic pool), generates their `static`s, and computes
5//! the topological order at **compile time**. A dependency cycle is a compile
6//! error; an unknown dependency name is a compile error.
7//!
8//! Surface (each item may be `#[cfg(...)]`-prefixed):
9//! ```text
10//! node NAME = Mode, deps: [A, B], spawn: <spawn>[, executor: EXEC][, disabled];
11//! node NAME = Mode, deps: [A, B], task: <worker>[, pool_size: N][, executor: EXEC]
12//! [, resources: [[#[cfg(..)]] RES: [local] [shared|consume] Type, ..]]
13//! [, slot_timeout: MS][, cancel][, disabled];
14//! node NAME = Mode, deps: [A]; // neither => a parked node the app spawns
15//! executor EXEC; // runtime-filled SendSpawner slot
16//! pool NAME = [Mode, ..], deps: [A][, executor: EXEC], spawn: <fn> | task: <worker>,
17//! [resources: [RES: [local] shared Type, ..],]
18//! policy: [<Ty> =] <expr>, min: N, max: M[, slot_timeout: MS][, cancel];
19//! ```
20//! `deps:` entries name a `node` or a `pool`; a `pool` dep resolves to that pool's floor
21//! member (member 0, the `min`-kept one), i.e. "start after the pool is up". A repeated
22//! dep or a redeclared node/pool name is a compile error.
23//!
24//! An `executor NAME;` slot may carry `#[cfg(...)]`, but validation does not model cfg
25//! predicates: a node referencing a slot that is cfg'd *out* while the node is cfg'd
26//! *in* surfaces as rustc's `cannot find value NAME`, not a macro error — don't gate an
27//! executor slot more restrictively than the nodes that reference it.
28//! `executor EXEC;` emits a `pub static EXEC: SpawnerSlot`; the app fills it with a
29//! `SendSpawner` (`InterruptExecutor::start`, `Spawner::make_send`) before
30//! `Supervisor::start`, and nodes carrying `executor: EXEC` spawn through it instead
31//! of the supervisor's own executor (their futures must be `Send`; an unfilled slot
32//! fails the spawn with `SpawnError::Busy`).
33//! A pool is emitted as `ElasticPool<P>`, so the macro needs the policy type `P`. By
34//! default it derives `P` from a `Ty::new(..)`-shaped `policy:` value (e.g.
35//! `DeferredShrink::new(..)` => `P = DeferredShrink`). Give `policy: <Ty> = <expr>` to
36//! state `P` explicitly when the value isn't that shape — a const, a free fn, a builder
37//! chain (`X::new(..).with(..)`), or a qualified path.
38//! `spawn:` takes a path or a partial call to a task fn taking the node **first**
39//! (`spawn: f` => `s.spawn(f(&NAME)?)`; `spawn: f(a)` => `s.spawn(f(&NAME, a)?)`), or,
40//! for a node, a closure / ready spawn fn emitted verbatim (for anything that doesn't
41//! fit that shape). A pool's `spawn:` is the same path/partial-call form with `&POOL[j]`
42//! injected first, via a generated `spawn_<pool>::<j>` glue fn; a pool has no closure
43//! form (members are instantiated per index).
44//!
45//! `task:` takes the same path/partial-call forms but names a **plain async worker
46//! fn** — possibly generic (turbofish or inferred) — instead of a hand-written
47//! `#[embassy_executor::task]`. The macro stamps a concrete shell task per
48//! declaration (embassy forbids generic tasks: one static `TaskPool` per concrete
49//! future type), sized by `pool_size:` on a node (default 1) or by the member count
50//! on a pool. Worker args are evaluated **inside the shell** — at the task's first
51//! poll, on the node's own executor — so cross-node data should go through awaited
52//! accessors, and a cross-core node builds its resources on its own core. `task:`
53//! and `spawn:` are mutually exclusive; `pool_size:` requires `task:`.
54//!
55//! `cancel` (a bare flag on `task:` items) makes the shell own the shutdown
56//! race: the worker is driven under `TaskNode::run_cancellable` and does NOT
57//! receive the node (resources become the first arguments), so a plain
58//! supervisor-unaware `async fn` — even a diverging one — binds directly. On
59//! stop/teardown its future is dropped in place and the shell still runs its
60//! full tail (state drop, resource restores, exit record). With `exit:` the
61//! value is provided only on a real completion — an aborted worker leaves the
62//! exit slot empty. Rejected on `spawn:` (that fn owns its body) and on
63//! `Mode::Pause` (a Pause worker must survive the stop and park on
64//! `wait_resume()`; `cancel` would record an exit nothing resumes). On a `pool`
65//! it is the trailing flag (after `max:`/`slot_timeout:`) and applies to the one
66//! shared shell, i.e. to every member: a shrink drops that member's future in
67//! place and its per-member resources are restored to its own slot index.
68//!
69//! **Prefer `task:`** — no attribute boilerplate, generic workers, auto-sized pool
70//! shells, and it is the only form supporting `resources:`; the shell inlines into
71//! the same poll and its `TaskPool` replaces the one the attribute would emit.
72//! `spawn:` remains for: a fn that already carries `#[embassy_executor::task]` and
73//! can't be de-attributed (another crate); a task also spawned outside the graph
74//! (sharing its one `TaskPool` instead of duplicating it as a shell); the verbatim
75//! closure form (custom spawn-time logic); and args that must be evaluated at
76//! spawn time on the supervisor's executor rather than at the shell's first poll.
77//! Worked examples: README "`spawn:` vs `task:` — which to use".
78//!
79//! Two `task:` footguns, spelled out because nothing warns about either:
80//!
81//! * a partial-call **extra that can be missing at first poll is a task-side
82//! panic**, not a failed spawn — extras are for infallible accessors. A value
83//! that might not exist yet belongs in `resources:` (a `shared` entry for a
84//! fan-out handle), where the pre-spawn gate turns "missing" into a clean
85//! `SpawnError::Busy`;
86//! * a **verbatim-closure `spawn:` node is invisible to the trace/name glue** —
87//! the closure owns the `SpawnToken`, so `adopt`/`stamp_name` is YOUR job
88//! inside it, and a stable proc-macro cannot emit a warning when you forget.
89//!
90//! `resources: [RES: [local] [shared|consume] Type, ..]` (requires `task:`)
91//! threads **owned resources from `main`** into the worker instead of re-acquiring
92//! them inside the task (`Peripherals::steal()`). Each entry emits a
93//! `pub static RES` slot at the declaration site; `main` moves the
94//! resource in with `RES.provide(..)` (consuming the `Peripherals` field — the
95//! compile-time exclusive-ownership guarantee), the generated glue `take()`s it just
96//! before the spawn (an unprovided slot fails `Supervisor::start` with
97//! `SpawnError::Busy` after a bounded wait — fail-closed, not a task-side panic),
98//! and the shell passes the worker `&mut Type` (after the node arg, in declared
99//! order, before any partial-call extras) and `restore()`s the value after the
100//! worker returns, so a Terminate respawn re-takes the *same instance*. Take-kind
101//! slot names are statics: unique across the graph. Entries may carry per-entry
102//! `#[cfg(...)]` (the slot, gate, glue, shell param, and worker-call argument all
103//! follow it — gate the worker fn's matching parameter with the same `#[cfg]`).
104//!
105//! Per-entry kind markers refine that default (order-free; `local` composes with
106//! either of the mutually-exclusive `consume`/`shared`):
107//!
108//! * `consume` — the worker receives the value **by value** and no restore is
109//! emitted: the slot stays empty after the task exits, so the worker may *drop*
110//! the resource at teardown (a driver whose `Drop` releases pins/DMA) and a
111//! respawn fail-closes (`SpawnError::Busy`) until the application `provide()`s
112//! a fresh value — the pattern for resources rebuilt each run (e.g. radio
113//! driver objects that go stale across a power cycle).
114//! * `shared` — a fan-out slot for a `Copy` handle (an `embassy_net::Stack`, a
115//! `&'static` shared-bus ref): the glue copies the value out non-destructively
116//! (`get()` — `T: Copy` enforced by its bound), the worker receives it by
117//! value, no restore, and the slot STAYS FILLED — so any number of nodes
118//! (and whole `task:` pools, where it stays ONE pool-wide slot — take kinds
119//! become per-member arrays there instead) may declare the SAME slot name.
120//! The static is emitted once, with the union of the declaring sites' cfg
121//! predicates; every re-declaration must repeat the kinds + type verbatim.
122//! * `local` — **requires the non-default `local-resources` feature** (of the
123//! supervisor crate, forwarded here): the slot is the graph-site
124//! `__SvLocalResourceSlot` type instead of `ResourceSlot` — same protocol, no
125//! `T: Send` bound, for `!Send` driver handles
126//! (`RefCell`-/`NoopRawMutex`-based). Feature-gated because it is the one
127//! graph form that emits `unsafe` code into the CONSUMER'S crate: an
128//! `unsafe impl Sync` whose soundness is the **single-core contract** (all
129//! `provide`/`take`/`restore` of the slot on one core). It cannot combine
130//! with `executor:` (a `SendSpawner`-routed node needs a `Send` future —
131//! macro error), and a consumer crate forbidding `unsafe_code` cannot use
132//! `local` (the assertion lands in *its* code, like the `trace-hooks`
133//! symbols).
134//!
135//! `slot_timeout: MS` (node and pool; milliseconds ≥ 1) overrides the node's
136//! pre-spawn wait bound for its `executor:` slot and `resources:` gates (default
137//! 100 ms — sized for "provided before `start()`"). Raise it for consumers of a
138//! **provider node** — a first-in-topo node whose worker *builds* the resources
139//! at runtime and `provide()`s them (the graph-native `hw_init`): size the
140//! timeout to the provider's async build time and the gate wait becomes a
141//! rendezvous instead of a `Busy`. See the README's provider-node recipe.
142//!
143//! A graph holds at most **256 node slots** (including pool members): all graph
144//! indices are `u8`, and the macro rejects a larger declaration at expansion.
145//!
146//! Nodes, pools, and individual deps may carry `#[cfg(...)]` attributes. A
147//! proc-macro can't evaluate `cfg`, so the node array is a fixed-length
148//! `[Option<&TaskNode>; M]` over all declared slots (each entry `Some`/`None` via a
149//! cfg-expression), the dep table is cfg-aware per-dep, and the order runs through
150//! `topo_sort_const` at const-eval (after cfg). Absent nodes are skipped at runtime.
151//!
152//! Generated items (at the call site): one `pub static` per `node`, a `[TaskNode; K]`
153//! array + `spawn_<pool>` glue fn + `<POOL>_POOL` `ElasticPool` + the structural
154//! `pub const`s `<POOL>_MIN` / `<POOL>_MAX` / `<POOL>_MEMBERS` (usize; for
155//! const-context sizing downstream — a `const` cannot read them off the member
156//! `static`) per `pool`, one slot `pub static` per `resources:` entry (shared
157//! entries: one per unique name; plus, iff any entry is `local`, the
158//! `__SvLocalResourceSlot` type), plus a
159//! single `pub static GRAPH: Graph<M>` bundling the node slots, the dependency table,
160//! the topological order, and (with the `pool` feature) the pools — pass `&GRAPH` to
161//! `Supervisor::new`. The backing tables are private; read them through `GRAPH.nodes`
162//! / `GRAPH.deps` / `GRAPH.order` / `GRAPH.pools` (node count is `GRAPH.nodes.len()`).
163//!
164//! With the supervisor's `trace` feature (forwarded here) the generated spawn glue
165//! also captures each `SpawnToken`'s task id into its node (`set_task_id`); with
166//! `metadata-names` it stamps the node name into the task Metadata. These are
167//! independent: `metadata-names` without `trace` emits a name-only spawn path
168//! (`stamp_name`, no id capture, no `_embassy_trace_*` dependency), so node names
169//! reach external tooling (rtos-trace/SystemView) without the trace recorders.
170//! With `trace-hooks` the macro additionally defines the seven `_embassy_trace_*`
171//! hook symbols at the declaration site (the supervisor crate is
172//! `forbid(unsafe_code)` and cannot), forwarding to the supervisor's `trace`
173//! recorders — requires an edition-2024 consumer, and exactly one graph declaration
174//! (or hook set) per binary.
175//!
176//! Types are referenced absolutely (`::embassy_supervisor::…`), so the consuming
177//! crate must depend on `embassy-supervisor` under its real name (not aliased).
178
179use proc_macro::TokenStream;
180use proc_macro2::TokenStream as TokenStream2;
181use quote::{format_ident, quote};
182use std::collections::{HashMap, HashSet};
183use syn::parse::{Parse, ParseStream};
184use syn::punctuated::Punctuated;
185use syn::spanned::Spanned;
186use syn::{
187 Attribute, Expr, Ident, LitInt, Meta, Path, Result as SynResult, Token, Type, bracketed,
188};
189
190mod kw {
191 syn::custom_keyword!(node);
192 syn::custom_keyword!(pool);
193 syn::custom_keyword!(deps);
194 syn::custom_keyword!(spawn);
195 syn::custom_keyword!(task);
196 syn::custom_keyword!(pool_size);
197 syn::custom_keyword!(policy);
198 syn::custom_keyword!(min);
199 syn::custom_keyword!(max);
200 syn::custom_keyword!(disabled);
201 syn::custom_keyword!(executor);
202 syn::custom_keyword!(resources);
203 syn::custom_keyword!(slot_timeout);
204 syn::custom_keyword!(exit);
205 syn::custom_keyword!(name);
206 syn::custom_keyword!(state);
207 syn::custom_keyword!(cancel);
208 syn::custom_keyword!(fragment);
209 syn::custom_keyword!(endfragment);
210}
211
212/// The graph-site slot type emitted (once per graph, iff any `resources:` entry
213/// is `local`-marked) for `!Send` resources. A single shared name: `emit_node`
214/// types the slot statics with it and `expand` emits its definition. Like the
215/// fixed `GRAPH` static, at most one `supervisor_graph!` per module.
216const LOCAL_SLOT_TYPE: &str = "__SvLocalResourceSlot";
217
218/// Per-graph idents for the generated helper items. UNNAMED graphs keep the
219/// historical fixed names; a `name: X;` graph suffixes them so several graphs
220/// coexist, even in one module.
221struct HelperIdents {
222 /// The graph-site `!Send` slot type (`local` kind).
223 local_slot: Ident,
224 /// The fallible-boxing fn (`state:` clauses).
225 try_box: Ident,
226 /// The `extern crate alloc as …` alias the try-box helper and shells use.
227 alloc_alias: Ident,
228}
229
230impl HelperIdents {
231 fn new(graph_name: Option<&Ident>) -> Self {
232 match graph_name {
233 None => Self {
234 local_slot: format_ident!("{LOCAL_SLOT_TYPE}"),
235 try_box: format_ident!("__sv_try_box"),
236 alloc_alias: format_ident!("__sv_alloc"),
237 },
238 Some(n) => {
239 let lower = n.to_string().to_lowercase();
240 Self {
241 local_slot: format_ident!("{LOCAL_SLOT_TYPE}{}", n),
242 try_box: format_ident!("__sv_try_box_{lower}"),
243 alloc_alias: format_ident!("__sv_alloc_{lower}"),
244 }
245 }
246 }
247 }
248}
249
250/// A dependency reference: a node ident, optionally `#[cfg(...)]`-gated.
251#[derive(Clone)]
252struct Dep {
253 cfg: Vec<Attribute>,
254 ident: Ident,
255 /// `deps: [NET ready]` — bring-up additionally awaits the dep's
256 /// task-asserted readiness (`set_ready`), bounded by the dependent's
257 /// `slot_timeout`. The `Ident` is kept for its span (feature errors).
258 ready: Option<Ident>,
259}
260
261/// `deps: [a, #[cfg(feature = "x")] b, c ready, …]`
262fn parse_dep_list(input: ParseStream) -> SynResult<Vec<Dep>> {
263 let content;
264 bracketed!(content in input);
265 let mut deps = Vec::new();
266 while !content.is_empty() {
267 let cfg = content.call(Attribute::parse_outer)?;
268 let ident: Ident = content.parse()?;
269 // Optional contextual `ready` marker: a dep entry is otherwise a lone
270 // ident, so a following ident can only be the marker.
271 let ready = if content.peek(Ident) {
272 let marker: Ident = content.parse()?;
273 if marker != "ready" {
274 return Err(syn::Error::new_spanned(
275 &marker,
276 format!("expected `,`, `]`, or the `ready` marker, found `{marker}`"),
277 ));
278 }
279 if !cfg!(feature = "readiness") {
280 return Err(syn::Error::new_spanned(
281 &marker,
282 "the `ready` dep marker requires the `readiness` feature \
283 (embassy-supervisor feature `readiness`) — bring-up then \
284 awaits the dep's set_ready() before spawning this node",
285 ));
286 }
287 Some(marker)
288 } else {
289 None
290 };
291 deps.push(Dep { cfg, ident, ready });
292 if content.peek(Token![,]) {
293 content.parse::<Token![,]>()?;
294 }
295 }
296 Ok(deps)
297}
298
299/// `[Terminate, OnDemand, …]` — the bracketed mode list.
300fn parse_mode_list(input: ParseStream) -> SynResult<Vec<Ident>> {
301 let content;
302 bracketed!(content in input);
303 let punct = Punctuated::<Ident, Token![,]>::parse_terminated(&content)?;
304 Ok(punct.into_iter().collect())
305}
306
307/// How a node/pool member gets its task: `spawn:` names a hand-written
308/// `#[embassy_executor::task]` fn (path / partial call / verbatim closure), while
309/// `task:` names a **plain async fn** — possibly generic — for which the macro
310/// emits a concrete `#[embassy_executor::task]` shell (embassy forbids generic
311/// tasks: one static `TaskPool` per concrete future type, so per-type shells are
312/// the only way — the macro stamps them so the user doesn't).
313enum TaskSource {
314 /// `spawn: <expr>` — the expr *is* (or produces) the task fn.
315 Spawn(Expr),
316 /// `task: <path | partial call>` — wrap in a generated shell; args are
317 /// evaluated inside the shell (at the task's first poll, on its own executor).
318 Shell(Expr),
319}
320
321/// One `[#[cfg(..)]] NAME: [local] [shared|consume] Type` entry of a
322/// `resources:` clause. The macro emits a `pub static NAME` slot at the
323/// declaration site (`ResourceSlot<Type>`, or the graph-site local slot type
324/// for `local` entries); `main` moves the resource in with `NAME.provide(..)`
325/// (consuming the `Peripherals` field — the compile-time ownership guarantee),
326/// the generated spawn glue `take()`s (or, for `shared`, copies via `get()`) it
327/// before the spawn, and the generated shell `restore()`s it after the worker
328/// returns so a respawn re-takes the same instance (unless `consume`/`shared`).
329struct ResourceDecl {
330 /// Per-entry `#[cfg(...)]` attributes: the slot static, gate entry, glue
331 /// take/get, shell param, worker-call argument, and restore all carry them,
332 /// so a feature-varying resource set works within one node (the worker fn
333 /// must gate its matching parameter with the same `#[cfg]`).
334 cfg: Vec<Attribute>,
335 ident: Ident,
336 ty: Type,
337 /// `local` marker: the slot holds a `!Send`-capable value (`Rc`-, `RefCell`-,
338 /// `NoopRawMutex`-based driver handles). Kept as the marker `Ident` for
339 /// span-attached errors (`local` composes with neither `executor:` nor a
340 /// multi-core provider — see `parse_node`).
341 local: Option<Ident>,
342 /// `consume` marker: the worker receives the value **by value** and the shell
343 /// emits no restore — the slot is left empty when the worker exits, so a
344 /// respawn gates on an explicit re-`provide()`. For resources that must be
345 /// *dropped* at teardown (a driver whose `Drop` releases pins/DMA) or that go
346 /// stale across a power cycle and must be rebuilt each run.
347 consume: Option<Ident>,
348 /// `shared` marker: a fan-out slot for a `Copy` handle. The glue copies the
349 /// value out non-destructively (`get()`), the worker receives it **by
350 /// value**, no restore — so any number of nodes (and whole pools) may
351 /// declare the SAME slot name (the static is emitted once; re-declarations
352 /// must repeat kinds + type exactly). Mutually exclusive with `consume`.
353 shared: Option<Ident>,
354}
355
356impl ResourceDecl {
357 /// The kinds+type signature every re-declaration of a `shared` slot must
358 /// repeat verbatim (compared as token strings — same-name shared slots are
359 /// ONE static, so their declared shapes must agree).
360 fn shared_signature(&self) -> String {
361 let ty = &self.ty;
362 format!(
363 "{}shared {}",
364 if self.local.is_some() { "local " } else { "" },
365 quote!(#ty)
366 )
367 }
368}
369
370/// Peek whether the next token of a `resources:` entry is a kind *marker*
371/// (`local` / `consume` / `shared`) rather than the start of the resource
372/// `Type` itself. Contextual-keyword rule (no reserved words): the ident is a
373/// marker only when something else of the entry still follows it — i.e. it is
374/// NOT a marker when followed by `::` or `<` (it starts a path/generic type
375/// like `local::Foo` or `local<T>`) or by `,` / end-of-list (it IS the whole
376/// type, a type literally named `local`). Same fork-and-peek disambiguation as
377/// the pool `policy:` type annotation.
378fn peek_kind_marker(content: ParseStream) -> Option<Ident> {
379 if !content.peek(syn::Ident) {
380 return None;
381 }
382 let fork = content.fork();
383 let ident: Ident = fork.parse().ok()?;
384 if ident != "local" && ident != "consume" && ident != "shared" {
385 return None;
386 }
387 if fork.is_empty() || fork.peek(Token![,]) || fork.peek(Token![::]) || fork.peek(Token![<]) {
388 return None;
389 }
390 Some(ident)
391}
392
393/// `resources: [LED: Output<'static>, RUNNER: local consume Runner, …]`
394fn parse_resource_list(input: ParseStream) -> SynResult<Vec<ResourceDecl>> {
395 let content;
396 bracketed!(content in input);
397 let mut resources = Vec::new();
398 while !content.is_empty() {
399 let cfg = content.call(Attribute::parse_outer)?;
400 let ident: Ident = content.parse()?;
401 content.parse::<Token![:]>()?;
402 // Kind markers between the colon and the type, order-free: `local`
403 // plus at most one of `consume` / `shared`. A repeated marker is a
404 // declaration bug; `consume` (exclusive take, slot empty after exit)
405 // and `shared` (non-destructive fan-out copy) contradict each other.
406 let mut local: Option<Ident> = None;
407 let mut consume: Option<Ident> = None;
408 let mut shared: Option<Ident> = None;
409 while let Some(marker) = peek_kind_marker(&content) {
410 content.parse::<Ident>()?; // commit the peeked marker
411 // `local` is the one kind whose slot type carries an `unsafe impl
412 // Sync` — injecting unsafe code is an explicit opt-in, so the
413 // marker is rejected unless the (non-default) `local-resources`
414 // feature forwarded by the supervisor crate is enabled.
415 if marker == "local" && !cfg!(feature = "local-resources") {
416 return Err(syn::Error::new_spanned(
417 &marker,
418 "`local` resources emit an `unsafe impl Sync` — opt in by \
419 enabling embassy-supervisor's `local-resources` feature",
420 ));
421 }
422 let slot = if marker == "local" {
423 &mut local
424 } else if marker == "consume" {
425 &mut consume
426 } else {
427 &mut shared
428 };
429 if slot.is_some() {
430 return Err(syn::Error::new_spanned(
431 &marker,
432 format!("duplicate `{marker}` marker"),
433 ));
434 }
435 *slot = Some(marker);
436 }
437 if let (Some(_), Some(s)) = (&consume, &shared) {
438 return Err(syn::Error::new_spanned(
439 s,
440 "`consume` and `shared` are mutually exclusive — `consume` takes \
441 the single value out for one owner, `shared` copies it out to \
442 any number of consumers",
443 ));
444 }
445 let ty: Type = content.parse()?;
446 resources.push(ResourceDecl {
447 cfg,
448 ident,
449 ty,
450 local,
451 consume,
452 shared,
453 });
454 if content.peek(Token![,]) {
455 content.parse::<Token![,]>()?;
456 }
457 }
458 Ok(resources)
459}
460
461struct NodeItem {
462 cfg: Vec<Attribute>,
463 ident: Ident,
464 mode: Ident,
465 deps: Vec<Dep>,
466 /// `None` = a parked node the app spawns itself (neither `spawn:` nor `task:`).
467 source: Option<TaskSource>,
468 /// `pool_size: N` on a `task:` node — sizes the generated shell's `TaskPool`
469 /// (headroom for a respawn while the previous instance is still draining).
470 pool_size: Option<LitInt>,
471 /// `resources: [NAME: Type, ..]` on a `task:` node — owned values threaded
472 /// from `main` through macro-emitted `ResourceSlot` statics into the
473 /// generated shell (which hands the worker `&mut Type` and restores the
474 /// value on exit). Empty for `spawn:`/parked nodes (enforced at parse).
475 resources: Vec<ResourceDecl>,
476 disabled: bool,
477 /// `executor: NAME` — spawn through the named [`SpawnerSlot`] (a
478 /// `SendSpawner` the app registers at runtime) instead of the supervisor's
479 /// own `Spawner`. `None` = the default executor.
480 executor: Option<Ident>,
481 /// `slot_timeout: N` (milliseconds) — overrides the node's pre-spawn
482 /// slot/gate wait bound (default 100 ms). Needed when the node's resources
483 /// are filled by a **provider node** at runtime: size it to the provider's
484 /// async build time.
485 slot_timeout: Option<LitInt>,
486 /// `exit: Type` on a `task:` node — the worker's return value is
487 /// `provide()`d into a generated `pub static <NODE>_EXIT: ResourceSlot<Type>`
488 /// just before the shell records the exit, so `<NODE>_EXIT.wait_take()`
489 /// observes the completion value (idiomatically `Result<R, Aborted>` out of
490 /// `run_cancellable` for completed-vs-cancelled). A worker that can never
491 /// return rejects the clause: the provide is dead code, and the shell denies
492 /// `unreachable_code` on it (spanned here) rather than emit a slot nothing
493 /// could ever fill.
494 exit: Option<syn::Type>,
495 /// `state: Type = init_expr` (feature `heap-state`) — per-activation heap
496 /// state: the glue fallibly boxes `init_expr` (alloc failure =
497 /// `SpawnError::Busy`, retryable), the shell lends the worker `&mut Type`,
498 /// and the Box drops on task exit — allocated fresh each activation,
499 /// reclaimed on every exit.
500 state: Option<(syn::Type, Expr)>,
501 /// `cancel` on a `task:` node — the shell drives the worker under
502 /// [`TaskNode::run_cancellable`] instead of awaiting it directly, and does
503 /// NOT pass the node to it. For the common shape the supervisor otherwise
504 /// can't take: a plain `async fn` that loops forever and knows nothing about
505 /// any supervisor. On shutdown its future is dropped in place and the shell
506 /// runs its usual restore/exit-provide/`mark_exited` tail.
507 cancel: bool,
508 /// The `supervisor_fragment!` this item was forwarded from (via the
509 /// `@fragment NAME;` marker), for error attribution across the relay.
510 fragment: Option<String>,
511}
512
513/// `executor NAME;` — declares a `pub static NAME: SpawnerSlot` the application
514/// fills with a `SendSpawner` before (or concurrently with) `Supervisor::start` (an
515/// InterruptExecutor tier, core1, ...). Nodes reference it with `executor: NAME`; the
516/// supervisor awaits the slot before spawning such a node (bounded by
517/// `SLOT_READY_TIMEOUT`, then `SpawnError::Busy`).
518struct ExecutorItem {
519 cfg: Vec<Attribute>,
520 ident: Ident,
521}
522
523struct PoolItem {
524 cfg: Vec<Attribute>,
525 ident: Ident,
526 modes: Vec<Ident>,
527 deps: Vec<Dep>,
528 /// The member task. Either a bare path (`http_task`) or a partial call carrying
529 /// extra args (`mcp_server_task(stack())`); the macro spawns member `j` as
530 /// `s.spawn(<fn>(&POOL[j] [, extra args])?)` — the node is always the first arg.
531 /// No closure form (members are instantiated per index), unlike a node's `spawn:`.
532 /// The `Shell` variant (`task:`) wraps a plain — possibly generic — async fn in
533 /// ONE generated `#[embassy_executor::task(pool_size = K)]` shell shared by all
534 /// members (they share one concrete future type, like a `spawn:` pool).
535 source: TaskSource,
536 /// The scaling policy value, emitted as the `policy:` field of the `ElasticPool`
537 /// static. The static is typed `ElasticPool<P>`, so the macro needs the policy
538 /// *type* `P`: when `policy_ty` is `None` it derives `P` from this expr via
539 /// `policy_type` (requires a `Type::new(..)` shape); when `policy_ty` is `Some`
540 /// the caller stated `P` explicitly and this expr can be any value of that type.
541 policy: Expr,
542 /// Optional explicit policy type from the `policy: <Type> = <expr>` form. `Some`
543 /// bypasses `policy_type` derivation, allowing a value the deriver can't handle
544 /// (a free fn, a const, a builder chain, a qualified path).
545 policy_ty: Option<Type>,
546 /// `executor: NAME` — spawn every member through the named [`SpawnerSlot`]
547 /// (e.g. a worker pool on the second core, scaled by this core's supervisor).
548 executor: Option<Ident>,
549 /// Pool `resources:` — **`shared` entries only** (enforced at parse): each
550 /// member's glue copies the same `Copy` handle out non-destructively, so
551 /// members don't contend (the reason non-shared kinds stay rejected).
552 resources: Vec<ResourceDecl>,
553 /// `slot_timeout: N` (milliseconds) — applied to every member (see the
554 /// node field of the same name).
555 slot_timeout: Option<LitInt>,
556 /// `min:`/`max:` — the scaling floor/ceiling. Any const-evaluable `usize`
557 /// expression: integer literals validate at parse time (best spans); other
558 /// exprs become the emitted `<POOL>_MIN`/`<POOL>_MAX` consts guarded by
559 /// const asserts (min <= max <= member count <= 255). The member count
560 /// itself (the mode list) stays structural — a proc macro cannot
561 /// const-evaluate, and the count drives how many nodes/shells/names are
562 /// emitted, so it cannot come from a const.
563 min: Expr,
564 max: Expr,
565 /// `state: Type = init_expr` (feature `heap-state`) — per-activation heap
566 /// state, one fresh Box per member per activation (see the node field).
567 state: Option<(syn::Type, Expr)>,
568 /// `cancel` on a `task:` pool — the one shared shell drives each member's
569 /// worker under [`TaskNode::run_cancellable`] and does not lead its arguments
570 /// with `&POOL[I]`, so a plain worker is shrunk (and torn down) by having its
571 /// future dropped in place. See the node field of the same name.
572 cancel: bool,
573 /// The `supervisor_fragment!` this item was forwarded from (via the
574 /// `@fragment NAME;` marker), for error attribution across the relay.
575 fragment: Option<String>,
576}
577
578// Both variants embed a large `syn::Expr` (and `PoolItem` a bit more), so their sizes
579// are close but unequal — enough for `large_enum_variant` to flag the gap. This AST is
580// parsed once and lives only briefly in a `Vec` during expansion, so boxing a variant
581// to shave a few bytes per element buys nothing real; suppress the lint instead of
582// paying a heap allocation.
583#[allow(clippy::large_enum_variant)]
584enum Item {
585 Node(NodeItem),
586 Pool(PoolItem),
587 Executor(ExecutorItem),
588}
589
590/// The parsed macro input: the list of `node`/`pool` declarations, in source order.
591/// Named `GraphSpec` (not `Graph`) to stay distinct from the *emitted* public type
592/// [`embassy_supervisor::Graph`] that `expand` produces as the `GRAPH` static.
593struct GraphSpec {
594 /// `name: IDENT;` as the FIRST item — the emitted graph static's ident
595 /// (default `GRAPH`). Named graphs suffix every generated helper ident so
596 /// several graphs coexist (even in one module); only the UNNAMED graph may
597 /// emit the once-per-binary `_embassy_trace_*` hook symbols.
598 name: Option<Ident>,
599 items: Vec<Item>,
600}
601
602/// An item's `resources:` entries (nodes and pools both carry them; an
603/// `executor` slot has none) — for the graph-wide pre-passes in `expand`.
604fn item_resources(item: &Item) -> &[ResourceDecl] {
605 match item {
606 Item::Node(n) => &n.resources,
607 Item::Pool(p) => &p.resources,
608 Item::Executor(_) => &[],
609 }
610}
611
612/// An item's own name + cfg attributes (for shared-slot bookkeeping/docs).
613fn item_ident_cfg(item: &Item) -> Option<(&Ident, &[Attribute])> {
614 match item {
615 Item::Node(n) => Some((&n.ident, &n.cfg)),
616 Item::Pool(p) => Some((&p.ident, &p.cfg)),
617 Item::Executor(_) => None,
618 }
619}
620
621impl Parse for GraphSpec {
622 fn parse(input: ParseStream) -> SynResult<Self> {
623 // Optional `name: IDENT;` first (same shape as a fragment's header).
624 let name = if input.peek(kw::name) && input.peek2(Token![:]) {
625 input.parse::<kw::name>()?;
626 input.parse::<Token![:]>()?;
627 let n: Ident = input.parse()?;
628 input.parse::<Token![;]>()?;
629 Some(n)
630 } else {
631 None
632 };
633 let mut items = Vec::new();
634 // Set while parsing items forwarded through a `supervisor_fragment!`
635 // relay: `@fragment NAME;` opens the span, `@endfragment;` closes it.
636 // Purely for error attribution — the items themselves are ordinary.
637 let mut current_fragment: Option<String> = None;
638 while !input.is_empty() {
639 if input.peek(Token![@]) {
640 input.parse::<Token![@]>()?;
641 if input.peek(kw::fragment) {
642 input.parse::<kw::fragment>()?;
643 current_fragment = Some(input.parse::<Ident>()?.to_string());
644 } else if input.peek(kw::endfragment) {
645 input.parse::<kw::endfragment>()?;
646 current_fragment = None;
647 } else {
648 return Err(input.error("expected `@fragment NAME;` or `@endfragment;`"));
649 }
650 input.parse::<Token![;]>()?;
651 continue;
652 }
653 let cfg = input.call(Attribute::parse_outer)?;
654 if input.peek(kw::node) {
655 let mut n = parse_node(input, cfg)?;
656 n.fragment = current_fragment.clone();
657 items.push(Item::Node(n));
658 } else if input.peek(kw::pool) {
659 let mut p = parse_pool(input, cfg)?;
660 p.fragment = current_fragment.clone();
661 items.push(Item::Pool(p));
662 } else if input.peek(kw::executor) {
663 // `executor NAME;` — a runtime-filled SendSpawner slot; nodes
664 // carrying `executor: NAME` spawn through it (the supervisor awaits
665 // the slot before spawning them).
666 input.parse::<kw::executor>()?;
667 let ident: Ident = input.parse()?;
668 input.parse::<Token![;]>()?;
669 items.push(Item::Executor(ExecutorItem { cfg, ident }));
670 } else {
671 return Err(input.error(
672 "expected `node`, `pool`, or `executor` (optionally `#[cfg(...)]`-prefixed)",
673 ));
674 }
675 }
676 Ok(GraphSpec { name, items })
677 }
678}
679
680// node IDENT = MODE, deps: [..] [, spawn: <expr>] [, disabled];
681fn parse_node(input: ParseStream, cfg: Vec<Attribute>) -> SynResult<NodeItem> {
682 input.parse::<kw::node>()?;
683 let ident: Ident = input.parse()?;
684 input.parse::<Token![=]>()?;
685 let mode: Ident = input.parse()?;
686 input.parse::<Token![,]>()?;
687 input.parse::<kw::deps>()?;
688 input.parse::<Token![:]>()?;
689 let deps = parse_dep_list(input)?;
690
691 let mut spawn = None;
692 let mut task: Option<(kw::task, Expr)> = None;
693 let mut pool_size = None;
694 let mut disabled = false;
695 let mut executor = None;
696 let mut resources: Option<(kw::resources, Vec<ResourceDecl>)> = None;
697 let mut slot_timeout = None;
698 let mut exit: Option<(kw::exit, syn::Type)> = None;
699 let mut state: Option<(kw::state, syn::Type, Expr)> = None;
700 let mut cancel: Option<kw::cancel> = None;
701 while input.peek(Token![,]) {
702 input.parse::<Token![,]>()?;
703 if input.peek(kw::spawn) {
704 input.parse::<kw::spawn>()?;
705 input.parse::<Token![:]>()?;
706 spawn = Some(input.parse::<Expr>()?);
707 } else if input.peek(kw::task) {
708 let k = input.parse::<kw::task>()?;
709 input.parse::<Token![:]>()?;
710 task = Some((k, input.parse::<Expr>()?));
711 } else if input.peek(kw::pool_size) {
712 input.parse::<kw::pool_size>()?;
713 input.parse::<Token![:]>()?;
714 pool_size = Some(input.parse::<LitInt>()?);
715 } else if input.peek(kw::disabled) {
716 input.parse::<kw::disabled>()?;
717 disabled = true;
718 } else if input.peek(kw::cancel) {
719 cancel = Some(input.parse::<kw::cancel>()?);
720 } else if input.peek(kw::executor) {
721 input.parse::<kw::executor>()?;
722 input.parse::<Token![:]>()?;
723 executor = Some(input.parse::<Ident>()?);
724 } else if input.peek(kw::resources) {
725 let k = input.parse::<kw::resources>()?;
726 input.parse::<Token![:]>()?;
727 resources = Some((k, parse_resource_list(input)?));
728 } else if input.peek(kw::slot_timeout) {
729 input.parse::<kw::slot_timeout>()?;
730 input.parse::<Token![:]>()?;
731 slot_timeout = Some(input.parse::<LitInt>()?);
732 } else if input.peek(kw::exit) {
733 let k = input.parse::<kw::exit>()?;
734 input.parse::<Token![:]>()?;
735 exit = Some((k, input.parse::<syn::Type>()?));
736 } else if input.peek(kw::state) {
737 let k = input.parse::<kw::state>()?;
738 input.parse::<Token![:]>()?;
739 let ty: syn::Type = input.parse()?;
740 input.parse::<Token![=]>()?;
741 let init: Expr = input.parse()?;
742 if !cfg!(feature = "heap-state") {
743 return Err(syn::Error::new_spanned(
744 k,
745 "`state:` requires the `heap-state` feature \
746 (embassy-supervisor feature `heap-state`) — per-activation \
747 boxed state, reclaimed on task exit",
748 ));
749 }
750 state = Some((k, ty, init));
751 } else {
752 return Err(input.error(
753 "expected `spawn:`, `task:`, `pool_size:`, `executor:`, `resources:`, \
754 `slot_timeout:`, `exit:`, `state:`, `cancel`, or `disabled`",
755 ));
756 }
757 }
758 input.parse::<Token![;]>()?;
759
760 // `slot_timeout: 0` would make every gated spawn fail instantly — reject it
761 // as the declaration bug it is (`base10_parse::<u64>` also rejects suffixed
762 // or oversized literals with a span-attached error).
763 if let Some(st) = &slot_timeout
764 && st.base10_parse::<u64>()? == 0
765 {
766 return Err(syn::Error::new_spanned(
767 st,
768 "`slot_timeout:` must be at least 1 (milliseconds)",
769 ));
770 }
771
772 // Exactly one of `spawn:` / `task:` may pick the node's task.
773 if let (Some(_), Some((k, _))) = (&spawn, &task) {
774 return Err(syn::Error::new_spanned(
775 k,
776 "`task:` and `spawn:` are mutually exclusive — `spawn:` names a \
777 hand-written `#[embassy_executor::task]` fn, `task:` generates one",
778 ));
779 }
780 // `pool_size:` sizes the generated shell's TaskPool; without `task:` there is
781 // no generated shell to size (a `spawn:` task fn declares its own).
782 if let (Some(ps), None) = (&pool_size, &task) {
783 return Err(syn::Error::new_spanned(
784 ps,
785 "`pool_size:` requires `task:` — a `spawn:` task fn sets its own \
786 `#[embassy_executor::task(pool_size = ...)]`",
787 ));
788 }
789 // `resources:` only makes sense with `task:`: the generated shell is what
790 // takes the values out of their slots at spawn and restores them after the
791 // worker returns. A hand-written `spawn:` fn (or a parked node) manages its
792 // own arguments.
793 if let Some((k, decls)) = &resources {
794 if task.is_none() {
795 return Err(syn::Error::new_spanned(
796 k,
797 "`resources:` requires `task:` — resources are handed to the \
798 generated shell as owned arguments and restored by it; a \
799 `spawn:` task fn manages its own arguments",
800 ));
801 }
802 if decls.is_empty() {
803 return Err(syn::Error::new_spanned(
804 k,
805 "`resources:` must declare at least one `NAME: Type` entry",
806 ));
807 }
808 // Duplicate names within one node would emit two statics with the same
809 // ident; catch it here with a clearer message than rustc's E0428.
810 for (i, d) in decls.iter().enumerate() {
811 if decls[..i].iter().any(|prev| prev.ident == d.ident) {
812 return Err(syn::Error::new_spanned(
813 &d.ident,
814 format!("duplicate resource name `{}`", d.ident),
815 ));
816 }
817 }
818 }
819 if let Some(ps) = &pool_size
820 && ps.base10_parse::<usize>()? == 0
821 {
822 return Err(syn::Error::new_spanned(
823 ps,
824 "`pool_size:` must be at least 1",
825 ));
826 }
827 // `state:` lives in the generated shell (it owns the Box across the worker
828 // call and drops it on exit); a `spawn:` fn can box its own state.
829 if let Some((k, _, _)) = &state
830 && task.is_none()
831 {
832 return Err(syn::Error::new_spanned(
833 k,
834 "`state:` requires `task:` — the generated shell owns the boxed \
835 state across the worker call and drops it on exit; a `spawn:` \
836 task fn can Box its own state",
837 ));
838 }
839 // `exit:` captures the worker's return value in the generated shell — only
840 // `task:` has one. A `spawn:` fn (or a parked node) owns its body and can
841 // `provide()` into any slot itself.
842 if let Some((k, _)) = &exit
843 && task.is_none()
844 {
845 return Err(syn::Error::new_spanned(
846 k,
847 "`exit:` requires `task:` — the generated shell is what captures \
848 the worker's return value; a `spawn:` task fn can provide() into \
849 a slot itself",
850 ));
851 }
852 // `cancel` rewrites how the generated shell drives the worker; a `spawn:` fn
853 // (or a parked node) owns its own body and can call `run_cancellable` itself.
854 if let Some(k) = &cancel {
855 if task.is_none() {
856 return Err(syn::Error::new_spanned(
857 k,
858 "`cancel` requires `task:` — it wraps the generated shell's call \
859 to the worker; a `spawn:` task fn can call \
860 `node.run_cancellable(..)` itself",
861 ));
862 }
863 // A Pause worker is supposed to SURVIVE a stop (ack, park on
864 // `wait_resume()`, keep its resources); `cancel` drops its future and
865 // lets the shell record an exit, after which nothing resumes it. That is
866 // the opposite of the mode, so reject the pair rather than silently
867 // turning a Pause node into a one-shot.
868 if mode == "Pause" {
869 return Err(syn::Error::new_spanned(
870 k,
871 "`cancel` cannot be combined with `Mode::Pause` — a Pause worker \
872 must survive the stop and park on `wait_resume()`, but `cancel` \
873 drops its future and records an exit; use `Mode::Terminate` (or \
874 `OnDemand`), or drive the pause by hand in the worker",
875 ));
876 }
877 }
878 // A `local` resource makes the shell future hold a `!Send`-capable value, and
879 // an `executor:`-routed node spawns through a `SendSpawner`, whose `spawn`
880 // requires a `Send` future. Reject here with the reason instead of letting
881 // rustc surface it as an opaque `F: Send` bound failure deep in the glue.
882 if let (Some((_, decls)), Some(ex)) = (&resources, &executor)
883 && let Some(l) = decls.iter().find_map(|d| d.local.as_ref())
884 {
885 return Err(syn::Error::new_spanned(
886 l,
887 format!(
888 "`local` resources cannot be combined with `executor: {ex}` — a \
889 local slot exists to carry `!Send` values, and a node routed \
890 through a `SpawnerSlot` (`SendSpawner`) must have a `Send` \
891 future; run the node on the supervisor's own executor"
892 ),
893 ));
894 }
895 let source = match (spawn, task) {
896 (Some(e), _) => Some(TaskSource::Spawn(e)),
897 (None, Some((_, e))) => Some(TaskSource::Shell(e)),
898 (None, None) => None,
899 };
900
901 Ok(NodeItem {
902 cfg,
903 ident,
904 mode,
905 deps,
906 source,
907 pool_size,
908 disabled,
909 executor,
910 resources: resources.map(|(_, decls)| decls).unwrap_or_default(),
911 slot_timeout,
912 exit: exit.map(|(_, ty)| ty),
913 state: state.map(|(_, ty, init)| (ty, init)),
914 cancel: cancel.is_some(),
915 fragment: None,
916 })
917}
918
919// pool IDENT = [MODES], deps: [..][, executor: EXEC], spawn: <fn>, policy: EXPR, min: N, max: M;
920fn parse_pool(input: ParseStream, cfg: Vec<Attribute>) -> SynResult<PoolItem> {
921 input.parse::<kw::pool>()?;
922 let ident: Ident = input.parse()?;
923 input.parse::<Token![=]>()?;
924 let modes = parse_mode_list(input)?;
925 input.parse::<Token![,]>()?;
926 input.parse::<kw::deps>()?;
927 input.parse::<Token![:]>()?;
928 let deps = parse_dep_list(input)?;
929 input.parse::<Token![,]>()?;
930 // Optional `executor: NAME,` — run the whole pool on the named SpawnerSlot's
931 // executor (e.g. a worker pool on the second core, scaled from this one).
932 let executor = if input.peek(kw::executor) {
933 input.parse::<kw::executor>()?;
934 input.parse::<Token![:]>()?;
935 let ex: Ident = input.parse()?;
936 input.parse::<Token![,]>()?;
937 Some(ex)
938 } else {
939 None
940 };
941 // The member task: a path, or a partial call supplying extra args (the macro
942 // injects `&POOL[j]` as the first argument in either case). `spawn:` names a
943 // hand-written `#[embassy_executor::task(pool_size = K)]` fn; `task:` names a
944 // plain (possibly generic) async fn the macro wraps in ONE generated shell
945 // task sized `pool_size = K`.
946 let source = if input.peek(kw::task) {
947 input.parse::<kw::task>()?;
948 input.parse::<Token![:]>()?;
949 TaskSource::Shell(input.parse()?)
950 } else {
951 input.parse::<kw::spawn>()?;
952 input.parse::<Token![:]>()?;
953 TaskSource::Spawn(input.parse()?)
954 };
955 input.parse::<Token![,]>()?;
956 // Pool `resources:` — `shared` entries only. A take-kind slot holds ONE
957 // value and pool members all run the same worker: they would contend for
958 // that single instance and every member past the first would fail its
959 // spawn. A `shared` entry is a non-destructive fan-out copy, so members
960 // don't contend — each glue `get()`s the same `Copy` handle.
961 let resources = if input.peek(kw::resources) {
962 input.parse::<kw::resources>()?;
963 input.parse::<Token![:]>()?;
964 let decls = parse_resource_list(input)?;
965 // Take-kind entries (lend/consume) become per-member SLOT ARRAYS
966 // (`[ResourceSlot<T>; K]`, member `I` takes/restores index `I`), so
967 // members no longer contend. Only TAKE-KIND `local` stays rejected:
968 // its single-core provide/take/restore contract interacts with
969 // per-member restore in ways deferred for now. A `shared local`
970 // entry is fine — it rides the pool-wide shared-slot path (one
971 // graph-site slot, non-destructive `get()`, no restore), exactly as
972 // before per-member resources existed.
973 if let Some(bad) = decls
974 .iter()
975 .find(|d| d.local.is_some() && d.shared.is_none())
976 {
977 return Err(syn::Error::new_spanned(
978 &bad.ident,
979 "`local` is not supported on take-kind `pool` resources (the single-core \
980 slot contract + per-member restore is deferred); a `shared local` entry \
981 works (one pool-wide fan-out slot), or declare the take-kind `local` \
982 resource on a node",
983 ));
984 }
985 input.parse::<Token![,]>()?;
986 decls
987 } else {
988 Vec::new()
989 };
990 // Optional `state: Type = expr,` — per-member per-activation boxed state.
991 let state = if input.peek(kw::state) {
992 let k = input.parse::<kw::state>()?;
993 input.parse::<Token![:]>()?;
994 let ty: syn::Type = input.parse()?;
995 input.parse::<Token![=]>()?;
996 let init: Expr = input.parse()?;
997 input.parse::<Token![,]>()?;
998 if !cfg!(feature = "heap-state") {
999 return Err(syn::Error::new_spanned(
1000 k,
1001 "`state:` requires the `heap-state` feature \
1002 (embassy-supervisor feature `heap-state`) — per-activation \
1003 boxed state, reclaimed on task exit",
1004 ));
1005 }
1006 Some((ty, init))
1007 } else {
1008 None
1009 };
1010 // `exit:` would land here positionally; reject it with the reason instead of
1011 // the generic "expected `policy`" the positional grammar produces.
1012 if input.peek(kw::exit) {
1013 let k = input.parse::<kw::exit>()?;
1014 return Err(syn::Error::new_spanned(
1015 k,
1016 "`exit:` is not supported on `pool` — the K members share one shell, \
1017 so per-member exit values need per-member storage; use per-node \
1018 `exit:` declarations, or have the worker provide() into an \
1019 app-declared slot itself",
1020 ));
1021 }
1022 input.parse::<kw::policy>()?;
1023 input.parse::<Token![:]>()?;
1024 // Optional explicit policy type: `policy: <Ty> = <expr>`. Fork to see if a `Type`
1025 // is followed by `=`; if so it's an annotation (commit on the real stream + eat the
1026 // `=`), otherwise rewind and treat the whole thing as the value expr (type derived
1027 // from it in `emit_pool`). For the common `Ty::new(..)` value the fork parses only a
1028 // partial type and then sees `(`, not `=`, so it correctly falls back to the derive
1029 // path — this keeps the bare form working unchanged.
1030 let policy_ty = {
1031 let fork = input.fork();
1032 if fork.parse::<Type>().is_ok() && fork.peek(Token![=]) {
1033 let ty: Type = input.parse()?;
1034 input.parse::<Token![=]>()?;
1035 Some(ty)
1036 } else {
1037 None
1038 }
1039 };
1040 let policy: Expr = input.parse()?;
1041 input.parse::<Token![,]>()?;
1042 input.parse::<kw::min>()?;
1043 input.parse::<Token![:]>()?;
1044 let min: Expr = input.parse()?;
1045 input.parse::<Token![,]>()?;
1046 input.parse::<kw::max>()?;
1047 input.parse::<Token![:]>()?;
1048 let max: Expr = input.parse()?;
1049 // Optional trailing `, slot_timeout: N` (milliseconds, ≥ 1) — every member's
1050 // pre-spawn slot/gate wait bound (see the node clause of the same name).
1051 let slot_timeout = if input.peek(Token![,]) && input.peek2(kw::slot_timeout) {
1052 input.parse::<Token![,]>()?;
1053 input.parse::<kw::slot_timeout>()?;
1054 input.parse::<Token![:]>()?;
1055 let st: LitInt = input.parse()?;
1056 if st.base10_parse::<u64>()? == 0 {
1057 return Err(syn::Error::new_spanned(
1058 &st,
1059 "`slot_timeout:` must be at least 1 (milliseconds)",
1060 ));
1061 }
1062 Some(st)
1063 } else {
1064 None
1065 };
1066 // Optional trailing `, cancel` — the pool's one shared shell owns the
1067 // shutdown race for every member (see the node clause of the same name).
1068 // Shrink and teardown already signal each member; `cancel` is what makes a
1069 // worker that never returns answer them.
1070 let cancel = if input.peek(Token![,]) && input.peek2(kw::cancel) {
1071 input.parse::<Token![,]>()?;
1072 Some(input.parse::<kw::cancel>()?)
1073 } else {
1074 None
1075 };
1076 input.parse::<Token![;]>()?;
1077 if let Some(k) = &cancel {
1078 // Same reason as the node: `cancel` rewrites how the GENERATED shell
1079 // drives the worker, so there must be one.
1080 if matches!(source, TaskSource::Spawn(_)) {
1081 return Err(syn::Error::new_spanned(
1082 k,
1083 "`cancel` requires `task:` — it wraps the generated shell's call to \
1084 the member worker; a `spawn:` member fn can call \
1085 `node.run_cancellable(..)` itself",
1086 ));
1087 }
1088 // A Pause member must survive its stop and park on `wait_resume()`;
1089 // `cancel` drops its future and records an exit instead. Reject the pair
1090 // per member rather than silently turning parked members into one-shots.
1091 if let Some(m) = modes.iter().find(|m| *m == "Pause") {
1092 return Err(syn::Error::new_spanned(
1093 m,
1094 "`cancel` cannot be combined with a `Pause` member — a Pause worker \
1095 must survive the stop and park on `wait_resume()`, but `cancel` \
1096 drops its future and records an exit; use `Terminate` (or \
1097 `OnDemand`) members, or drive the pause by hand in the worker",
1098 ));
1099 }
1100 }
1101 // Same `Send` reasoning as the node-side check: a `local` (i.e. `!Send`able)
1102 // resource cannot ride members routed through a `SendSpawner`.
1103 if let Some(ex) = &executor
1104 && let Some(l) = resources.iter().find_map(|d| d.local.as_ref())
1105 {
1106 return Err(syn::Error::new_spanned(
1107 l,
1108 format!(
1109 "`local` resources cannot be combined with `executor: {ex}` — a \
1110 local slot exists to carry `!Send` values, and a pool routed \
1111 through a `SpawnerSlot` (`SendSpawner`) must have `Send` \
1112 futures; run the pool on the supervisor's own executor"
1113 ),
1114 ));
1115 }
1116 Ok(PoolItem {
1117 cfg,
1118 ident,
1119 modes,
1120 deps,
1121 source,
1122 policy,
1123 policy_ty,
1124 executor,
1125 resources,
1126 slot_timeout,
1127 min,
1128 max,
1129 state,
1130 cancel: cancel.is_some(),
1131 fragment: None,
1132 })
1133}
1134
1135/// The node/pool name string: ident lowercased with `_`→`-` (`WIFI_CTRL` → "wifi-ctrl").
1136fn name_string(ident: &Ident) -> String {
1137 ident.to_string().to_lowercase().replace('_', "-")
1138}
1139
1140/// Build a task-call expression with leading arguments injected ahead of the
1141/// user-supplied extras — the node ref (`&NODE` / `&POOL[i]`) first, then the
1142/// item's threaded `resources:` values: a bare path `f` => `f(lead..)`; a
1143/// partial call `f(a, b)` => `f(lead.., a, b)`.
1144///
1145/// The lead may be EMPTY: a `cancel` shell suppresses the node ref, so a node
1146/// with no `resources:`/`state:` leads with nothing at all. Both groups are
1147/// therefore joined as ONE list — a separator hard-coded between them would
1148/// emit `f(, a, b)`.
1149fn inject_call_with(task: &Expr, lead: &[TokenStream2]) -> SynResult<TokenStream2> {
1150 match task {
1151 Expr::Path(_) => Ok(quote!(#task(#(#lead),*))),
1152 Expr::Call(c) => {
1153 let f = &c.func;
1154 let mut args: Vec<TokenStream2> = lead.to_vec();
1155 args.extend(c.args.iter().map(|a| quote!(#a)));
1156 Ok(quote!(#f(#(#args),*)))
1157 }
1158 other => Err(syn::Error::new_spanned(
1159 other,
1160 "expected a task-fn path or a partial call like `f(extra_args)`",
1161 )),
1162 }
1163}
1164
1165/// Combine an item's `#[cfg(...)]` attributes into one predicate (`all(..)` if
1166/// several), used to gate its `GRAPH.nodes` slot to `Some`/`None`. `None` = always present.
1167fn cfg_predicate(attrs: &[Attribute]) -> Option<TokenStream2> {
1168 let preds: Vec<TokenStream2> = attrs
1169 .iter()
1170 .filter_map(|a| match &a.meta {
1171 Meta::List(ml) if ml.path.is_ident("cfg") => Some(ml.tokens.clone()),
1172 _ => None,
1173 })
1174 .collect();
1175 match preds.len() {
1176 0 => None,
1177 1 => Some(preds[0].clone()),
1178 _ => Some(quote!(all(#(#preds),*))),
1179 }
1180}
1181
1182/// Gate-array tokens for a `resources:` list: the element list (each entry
1183/// `#[cfg]`-gated — cfg on array elements is stable, same as the deps table)
1184/// and a matching length expression. A cfg'd-out element must also subtract
1185/// from the fixed array length, so with any per-entry cfg the length becomes a
1186/// sum of cfg-block 1/0 terms (the `GRAPH.nodes` Some/None trick, in const
1187/// position); without, it stays the plain count.
1188fn gate_tokens(resources: &[ResourceDecl]) -> (TokenStream2, Vec<TokenStream2>) {
1189 let gate_refs: Vec<TokenStream2> = resources
1190 .iter()
1191 .map(|r| {
1192 let cfg = &r.cfg;
1193 let res = &r.ident;
1194 quote!(#(#cfg)* &#res)
1195 })
1196 .collect();
1197 let any_cfg = resources.iter().any(|r| cfg_predicate(&r.cfg).is_some());
1198 let len = if any_cfg {
1199 let terms: Vec<TokenStream2> = resources
1200 .iter()
1201 .map(|r| match cfg_predicate(&r.cfg) {
1202 None => quote!(1usize),
1203 Some(pred) => quote!({
1204 #[cfg(#pred)]
1205 {
1206 1usize
1207 }
1208 #[cfg(not(#pred))]
1209 {
1210 0usize
1211 }
1212 }),
1213 })
1214 .collect();
1215 quote!(0usize #(+ #terms)*)
1216 } else {
1217 let n = resources.len();
1218 quote!(#n)
1219 };
1220 (len, gate_refs)
1221}
1222
1223/// `" (from fragment \`X\`)"` when the item was forwarded through a
1224/// `supervisor_fragment!` relay, else empty — error-message attribution.
1225fn fragment_suffix(fragment: &Option<String>) -> String {
1226 match fragment {
1227 Some(f) => format!(" (from fragment `{f}`)"),
1228 None => String::new(),
1229 }
1230}
1231
1232/// Build the `[&'static TaskNode; n]` element and length tokens for a node's or
1233/// pool's `ready`-marked deps, cfg-aware like `gate_tokens`. A dep naming a pool
1234/// resolves to the pool's floor member (`&POOL[0]`), matching how `deps: [POOL]`
1235/// resolves for spawn ordering.
1236fn ready_tokens(
1237 deps: &[Dep],
1238 pool_names: &std::collections::HashSet<String>,
1239) -> Option<(TokenStream2, Vec<TokenStream2>)> {
1240 let marked: Vec<&Dep> = deps.iter().filter(|d| d.ready.is_some()).collect();
1241 if marked.is_empty() {
1242 return None;
1243 }
1244 let refs: Vec<TokenStream2> = marked
1245 .iter()
1246 .map(|d| {
1247 let cfg = &d.cfg;
1248 let ident = &d.ident;
1249 if pool_names.contains(&ident.to_string()) {
1250 quote!(#(#cfg)* &#ident[0])
1251 } else {
1252 quote!(#(#cfg)* &#ident)
1253 }
1254 })
1255 .collect();
1256 let any_cfg = marked.iter().any(|d| cfg_predicate(&d.cfg).is_some());
1257 let len = if any_cfg {
1258 let terms: Vec<TokenStream2> = marked
1259 .iter()
1260 .map(|d| match cfg_predicate(&d.cfg) {
1261 None => quote!(1usize),
1262 Some(pred) => quote!({
1263 #[cfg(#pred)]
1264 {
1265 1usize
1266 }
1267 #[cfg(not(#pred))]
1268 {
1269 0usize
1270 }
1271 }),
1272 })
1273 .collect();
1274 quote!(0usize #(+ #terms)*)
1275 } else {
1276 let n = marked.len();
1277 quote!(#n)
1278 };
1279 Some((len, refs))
1280}
1281
1282/// Extract the policy *type* from a `Type::new(..)` constructor expression. Only used
1283/// on the derive path (no explicit `policy: <Ty> = ..` annotation); the type is the
1284/// call's path minus its last segment (`DeferredShrink::new` -> `DeferredShrink`).
1285fn policy_type(expr: &Expr) -> SynResult<Path> {
1286 if let Expr::Call(call) = expr
1287 && let Expr::Path(p) = &*call.func
1288 {
1289 let n = p.path.segments.len();
1290 if n >= 2 {
1291 let segs: Punctuated<_, Token![::]> =
1292 p.path.segments.iter().take(n - 1).cloned().collect();
1293 return Ok(Path {
1294 leading_colon: p.path.leading_colon,
1295 segments: segs,
1296 });
1297 }
1298 }
1299 Err(syn::Error::new_spanned(
1300 expr,
1301 "pool `policy:` must be a `Type::new(..)` constructor (e.g. `DeferredShrink::new(..)`), \
1302 or give the type explicitly: `policy: <Type> = <expr>`",
1303 ))
1304}
1305
1306/// One emitted node slot, in final index order.
1307struct Slot {
1308 /// Presence predicate (`None` = unconditional), gates the node slot (`GRAPH.nodes`) entry.
1309 cfg_pred: Option<TokenStream2>,
1310 /// `&NODE` or `&POOL[j]`.
1311 reference: TokenStream2,
1312 /// Raw deps, resolved to indices in the second pass.
1313 deps: Vec<Dep>,
1314 /// The `supervisor_fragment!` the owning item came from, for error
1315 /// attribution when a dep fails to resolve across the relay.
1316 fragment: Option<String>,
1317}
1318
1319/// The `Option<fn(..)>` spawn expression for a node. `None` (no `spawn:`) is a
1320/// parked node the app spawns itself. A path or partial call is a task fn taking
1321/// `&NODE` first (plus any given args); the macro wraps it as
1322/// `|s| { s.spawn(<task>(&NODE, ..)?); Ok(()) }`. Anything else (a closure, or a
1323/// ready spawn fn) is emitted verbatim. Every form is cast to `spawn_fn` so it
1324/// coerces cleanly inside `Option::Some(..)`.
1325fn node_spawn(
1326 ident: &Ident,
1327 spawn: &Option<Expr>,
1328 executor: &Option<Ident>,
1329 resources: &[ResourceDecl],
1330 // `state:`: fallibly box the init value in the glue, BEFORE the resource
1331 // takes (a failed alloc strands nothing) — `SpawnError::Busy`, retryable.
1332 state: Option<&(syn::Type, Expr)>,
1333 spawn_fn: &TokenStream2,
1334 helpers: &HelperIdents,
1335) -> SynResult<TokenStream2> {
1336 // `resources:` take-prelude + the taken values as extra shell arguments.
1337 // Taking here — in the glue, BEFORE the spawn — is the point: an unprovided
1338 // slot fails `Supervisor::start` with `SpawnError::Busy` (the supervisor
1339 // logs the node name), instead of panicking inside an already-spawned task.
1340 // The values ride into the task as ordinary `#[embassy_executor::task]`
1341 // arguments (embassy stores them in the shell's TaskPool slot). A `shared`
1342 // entry copies the value out non-destructively (`get()` — the slot stays
1343 // filled for the other consumers) instead of `take()`ing it; `get`'s
1344 // `T: Copy` bound is what enforces "shared handles must be Copy".
1345 let take_prelude: Vec<TokenStream2> = resources
1346 .iter()
1347 .enumerate()
1348 .map(|(i, r)| {
1349 let cfg = &r.cfg;
1350 let res = &r.ident;
1351 let var = format_ident!("__r{}", i);
1352 let getter = if r.shared.is_some() {
1353 quote!(get)
1354 } else {
1355 quote!(take)
1356 };
1357 quote! {
1358 #(#cfg)*
1359 let #var = #res
1360 .#getter()
1361 .ok_or(::embassy_executor::SpawnError::Busy)?;
1362 }
1363 })
1364 .collect();
1365 // Per-entry `#[cfg]` rides on the call ARGUMENT too (stable in call
1366 // position, like the cfg'd array elements in the deps table), so a
1367 // cfg'd-out entry vanishes from the glue, the shell signature, and the
1368 // worker call consistently.
1369 let res_args: Vec<TokenStream2> = resources
1370 .iter()
1371 .enumerate()
1372 .map(|(i, r)| {
1373 let cfg = &r.cfg;
1374 let var = format_ident!("__r{}", i);
1375 quote!(#(#cfg)* #var)
1376 })
1377 .collect();
1378 let try_box = &helpers.try_box;
1379 let (state_prelude, state_arg) = match state {
1380 Some((_, init)) => (
1381 quote! {
1382 let __state = #try_box(#init)
1383 .ok_or(::embassy_executor::SpawnError::Busy)?;
1384 },
1385 vec![quote!(__state)],
1386 ),
1387 None => (quote!(), vec![]),
1388 };
1389 Ok(match (spawn, executor) {
1390 (None, None) => quote!(::core::option::Option::None),
1391 // `executor:` needs the macro to perform the spawn, so it composes only
1392 // with the path / partial-call `spawn:` forms below.
1393 (None, Some(ex)) => {
1394 return Err(syn::Error::new_spanned(
1395 ex,
1396 "`executor:` requires a `spawn:` (a parked node is spawned by the \
1397 application, which picks its own spawner)",
1398 ));
1399 }
1400 // A path or a partial call: a task fn taking `&NODE` first (plus any
1401 // given args); generate `|s| { s.spawn(<task>(&NODE, ..)?); Ok(()) }`.
1402 // With `executor: NAME` the glue ignores the supervisor's `Spawner` and
1403 // spawns through the named `SpawnerSlot` (a `SendSpawner` the app
1404 // registers at runtime): an unfilled slot fails the spawn with
1405 // `SpawnError::Busy` — loud misconfiguration, not a missing task. The
1406 // task future must then be `Send` (enforced by `SendSpawner::spawn`).
1407 (Some(e @ (Expr::Path(_) | Expr::Call(_))), executor) => {
1408 let mut lead: Vec<TokenStream2> = vec![quote!(&#ident)];
1409 lead.extend(res_args.iter().cloned());
1410 lead.extend(state_arg.iter().cloned());
1411 let call = inject_call_with(e, &lead)?;
1412 match executor {
1413 None => {
1414 let stmts = spawn_stmts(&call, "e!(&#ident), "e!(s));
1415 quote!(::core::option::Option::Some(
1416 (|s| {
1417 #state_prelude
1418 #(#take_prelude)*
1419 #stmts
1420 ::core::result::Result::Ok(())
1421 }) as #spawn_fn
1422 ))
1423 }
1424 Some(ex) => {
1425 let stmts = spawn_stmts(&call, "e!(&#ident), "e!(__sp));
1426 quote!(::core::option::Option::Some(
1427 (|_s| {
1428 // The supervisor awaits this slot's `ready()` before
1429 // invoking the glue (the node carries `.with_executor(&EX)`
1430 // and the bring-up bounds the wait), so `get()` is already
1431 // filled; `ok_or` is the belt-and-braces unfilled guard.
1432 // Resources are taken AFTER the spawner guard, so an
1433 // unfilled executor never consumes (and strands) them.
1434 let __sp = #ex
1435 .get()
1436 .ok_or(::embassy_executor::SpawnError::Busy)?;
1437 #state_prelude
1438 #(#take_prelude)*
1439 #stmts
1440 ::core::result::Result::Ok(())
1441 }) as #spawn_fn
1442 ))
1443 }
1444 }
1445 }
1446 (Some(_), Some(ex)) => {
1447 return Err(syn::Error::new_spanned(
1448 ex,
1449 "`executor:` cannot be combined with a verbatim spawn closure (the \
1450 closure owns the spawn; use the named SpawnerSlot inside it instead)",
1451 ));
1452 }
1453 // Anything else (a closure, or a ready spawn fn) is emitted verbatim.
1454 // NOTE: with the `trace` feature such a node is not auto-mapped — the
1455 // closure owns the SpawnToken; call `adopt`/`set_task_id` in it yourself.
1456 (Some(e), None) => quote!(::core::option::Option::Some((#e) as #spawn_fn)),
1457 })
1458}
1459
1460/// The spawn statement(s) for the generated glue. Plain `s.spawn(<call>?)`
1461/// normally; with the `trace` feature the `SpawnToken` is bound first so its task
1462/// id can be captured into the node (`set_task_id`) — the id→node mapping the
1463/// supervisor's `trace` recorders resolve against (in embassy-executor 0.10 the
1464/// task-fn call returns `Result<SpawnToken, SpawnError>` and `Spawner::spawn`
1465/// itself is infallible, so the token is available between the two).
1466///
1467/// Three shapes, resolved at expansion by the macro crate's own features:
1468/// * `trace` on → bind the token and `adopt` it (`set_task_id` + name stamp under
1469/// `metadata-names`).
1470/// * `trace` off but `metadata-names` on → bind the token and `stamp_name` only:
1471/// the node name reaches the task Metadata (for rtos-trace/SystemView) with no id
1472/// capture and no dependency on the `_embassy_trace_*` hooks.
1473/// * neither → plain infallible spawn.
1474fn spawn_stmts(call: &TokenStream2, node_ref: &TokenStream2, sp: &TokenStream2) -> TokenStream2 {
1475 if cfg!(feature = "trace") {
1476 // `adopt` = set_task_id + (under metadata-names) Metadata name stamp.
1477 quote! {
1478 let __token = #call?;
1479 (#node_ref).adopt(&__token);
1480 #sp.spawn(__token);
1481 }
1482 } else if cfg!(feature = "metadata-names") {
1483 // Name-only path: stamp the node name into the task Metadata, nothing else.
1484 quote! {
1485 let __token = #call?;
1486 (#node_ref).stamp_name(&__token);
1487 #sp.spawn(__token);
1488 }
1489 } else {
1490 quote!(#sp.spawn(#call?);)
1491 }
1492}
1493
1494/// Emit the `#[embassy_executor::task]` shell for a `task:` clause: a concrete,
1495/// non-generic task fn that takes only the node and awaits the user's worker with
1496/// the node injected first. This is how a **generic** worker becomes spawnable —
1497/// embassy forbids generic tasks (one static `TaskPool` per concrete future type),
1498/// so a monomorphized shell is stamped per declaration. Worker args are evaluated
1499/// inside the shell — at the task's first poll, on the node's own executor — so
1500/// the DSL never needs the arg types and a cross-core node builds its resources on
1501/// the core that runs them.
1502///
1503/// Returns the shell item and a path `Expr` naming it, which feeds the ordinary
1504/// `spawn:` path-form glue (executor routing and trace `adopt` compose unchanged).
1505// One argument per independent codegen input; a bundling struct would only
1506// rename the coupling.
1507#[allow(clippy::too_many_arguments)]
1508fn emit_shell(
1509 owner: &Ident,
1510 cfg: &[Attribute],
1511 worker: &Expr,
1512 pool_size: usize,
1513 resources: &[ResourceDecl],
1514 exit: Option<&syn::Type>,
1515 // `state: Type = ..`: the shell owns the glue-boxed state across the worker
1516 // call (worker sees `&mut Type`) and DROPS it first thing after the worker
1517 // returns — reclaimed before restores/exit-provide/mark_exited.
1518 state: Option<&(syn::Type, Expr)>,
1519 // Pool shells restore lend entries to a slot REFERENCE parameter (the
1520 // member's own array element, passed by the wrapper) instead of a slot
1521 // named statically — restore-to-same-index by construction.
1522 pool_member: bool,
1523 // `cancel`: drive the worker under `run_cancellable` and DON'T lead its
1524 // arguments with the node — the worker is a plain future that never returns
1525 // on its own, so the shell owns the shutdown race on its behalf.
1526 cancel: bool,
1527 cr: &TokenStream2,
1528 helpers: &HelperIdents,
1529) -> SynResult<(TokenStream2, Expr)> {
1530 if !matches!(worker, Expr::Path(_) | Expr::Call(_)) {
1531 return Err(syn::Error::new_spanned(
1532 worker,
1533 "`task:` names an async worker fn — a path or a partial call like \
1534 `worker(args)`; for a closure or a ready spawn fn use `spawn:`",
1535 ));
1536 }
1537 let shell = format_ident!("__sv_task_{}", owner.to_string().to_lowercase());
1538 // `resources:` values arrive as owned task arguments (the spawn glue took
1539 // them out of their slots); the shell keeps ownership, lends the worker
1540 // `&mut`, and restores each value to its slot after the worker returns —
1541 // i.e. after the worker's clean shutdown ack — so a Terminate respawn
1542 // re-takes the SAME instance instead of re-acquiring hardware. A `Pause`
1543 // worker parks instead of returning, so it simply retains its resources
1544 // (the restore lines below are unreachable for it — correct, same as a
1545 // hand-written parked task holding its arguments).
1546 //
1547 // A `consume` entry is forwarded to the worker BY VALUE instead — the worker
1548 // owns it (it can drop it at teardown, e.g. a driver whose `Drop` releases
1549 // pins/DMA) and no restore is emitted: the slot stays empty until the app
1550 // re-`provide()`s, which the supervisor's pre-respawn gate wait turns into
1551 // fail-closed `SpawnError::Busy` rather than a stale-value reuse.
1552 //
1553 // A `shared` entry is also by value with no restore — but because the glue
1554 // COPIED it out (`get()`), the slot stays filled; the worker's value is its
1555 // own copy of the fan-out handle.
1556 //
1557 // Per-entry `#[cfg]` rides on params, worker-call arguments, and restore
1558 // statements alike, so a cfg'd-out entry disappears from the whole chain
1559 // (the worker fn must gate its matching parameter with the same `#[cfg]`).
1560 let by_value = |r: &ResourceDecl| r.consume.is_some() || r.shared.is_some();
1561 let res_params: Vec<TokenStream2> = resources
1562 .iter()
1563 .enumerate()
1564 .map(|(i, r)| {
1565 let cfg = &r.cfg;
1566 let var = format_ident!("__r{}", i);
1567 let ty = &r.ty;
1568 if by_value(r) {
1569 quote!(#(#cfg)* #var: #ty)
1570 } else if pool_member {
1571 // Lend entry of a pool: value + the member's own slot element.
1572 let slot_param = format_ident!("__r{}_slot", i);
1573 quote!(#(#cfg)* mut #var: #ty, #(#cfg)* #slot_param: &'static #cr::ResourceSlot<#ty>)
1574 } else {
1575 quote!(#(#cfg)* mut #var: #ty)
1576 }
1577 })
1578 .collect();
1579 let res_leases: Vec<TokenStream2> = resources
1580 .iter()
1581 .enumerate()
1582 .map(|(i, r)| {
1583 let cfg = &r.cfg;
1584 let var = format_ident!("__r{}", i);
1585 if by_value(r) {
1586 quote!(#(#cfg)* #var)
1587 } else {
1588 quote!(#(#cfg)* &mut #var)
1589 }
1590 })
1591 .collect();
1592 let restores: Vec<TokenStream2> = resources
1593 .iter()
1594 .enumerate()
1595 .filter(|(_, r)| !by_value(r))
1596 .map(|(i, r)| {
1597 let cfg = &r.cfg;
1598 let var = format_ident!("__r{}", i);
1599 if pool_member {
1600 let slot_param = format_ident!("__r{}_slot", i);
1601 quote!(#(#cfg)* #slot_param.restore(#var);)
1602 } else {
1603 let res = &r.ident;
1604 quote!(#(#cfg)* #res.restore(#var);)
1605 }
1606 })
1607 .collect();
1608 let alloc_alias = &helpers.alloc_alias;
1609 let (state_param, state_lease, state_drop) = match state {
1610 Some((ty, _)) => (
1611 quote!(, mut __state: #alloc_alias::boxed::Box<#ty>),
1612 vec![quote!(&mut *__state)],
1613 // Reclaim the bulk FIRST: before restores, exit-provide, and the
1614 // completion record, so has_exited() implies the heap is back.
1615 quote!(::core::mem::drop(__state);),
1616 ),
1617 None => (quote!(), vec![], quote!()),
1618 };
1619 // `cancel` workers take no node: the shell holds it and races the worker's
1620 // future against the shutdown signal itself, which is the whole point of the
1621 // flag — the worker stays a plain async fn with no supervisor in its
1622 // signature.
1623 let mut lead: Vec<TokenStream2> = if cancel {
1624 Vec::new()
1625 } else {
1626 vec![quote!(__node)]
1627 };
1628 lead.extend(res_leases);
1629 lead.extend(state_lease);
1630 let call = inject_call_with(worker, &lead)?;
1631 // Unsuffixed literal: `#[task]`'s own parser wants a plain integer.
1632 let ps = LitInt::new(&pool_size.to_string(), proc_macro2::Span::call_site());
1633 // A diverging (`-> !`) worker makes the trailing statements unreachable —
1634 // legitimate (a detached/`Pause` worker retains its resources forever), so
1635 // silence rustc's `unreachable_code` lint on the generated body. Always
1636 // emitted: the completion record below is an unconditional trailing
1637 // statement.
1638 let allow_unreachable = quote!(#[allow(unreachable_code)]);
1639 // `exit: Type`: bind the worker's return value and provide() it into the
1640 // node's exit slot BEFORE mark_exited, so has_exited() implies the value is
1641 // present. A worker whose return type mismatches the declared `exit:` fails
1642 // at this provide with a plain rustc type error on the shell.
1643 // Under `cancel` the worker may not have returned at all — the shell holds a
1644 // `Result<Output, Aborted>` — so the exit value is provided only on a real
1645 // completion. An aborted worker leaves `<NODE>_EXIT` empty (and
1646 // `shutdown_requested()` set), which is how a waiter tells "it finished" from
1647 // "it was stopped".
1648 //
1649 // A DIVERGING worker (`-> !`) makes that provide dead code: its future has
1650 // no output, so the slot could never be filled and every `wait_take()` on
1651 // it would hang forever. The blanket allow above would hide that, so the
1652 // provide re-DENIES `unreachable_code` on itself — the one statement in the
1653 // shell where unreachability is a declaration error rather than a
1654 // legitimate parked/detached worker. Spanned on the declared `exit:` type,
1655 // so rustc points at the clause the user has to remove (a bare diverging
1656 // worker stays legal: that is what `cancel` is for).
1657 let exit_ident = format_ident!("{}_EXIT", owner);
1658 let provide = |exit: &syn::Type| {
1659 // Every token of the statement carries the `exit:` type's span, so the
1660 // lint's own label lands on that clause instead of the whole item.
1661 let slot = Ident::new(&exit_ident.to_string(), exit.span());
1662 quote::quote_spanned!(exit.span()=>
1663 #[deny(unreachable_code)]
1664 #slot.provide(__out);
1665 )
1666 };
1667 // The `cancel` arms pin the worker into the shell's own frame and hand
1668 // `run_cancellable` a `Pin<&mut _>`: the shell then stores the worker's state
1669 // machine ONCE, whatever rustc decides to do with the callee's arguments
1670 // (rust-lang/rust#62958 doubles a future passed by value into an `async fn`,
1671 // and this is static task storage, so the doubling was per node, per binary).
1672 // The `pin!` lives in its own block so the worker is dropped at the same point
1673 // it always was — before the restores below, which move the lent resources
1674 // back out.
1675 let (drive, provide_exit) = match (cancel, exit) {
1676 (false, Some(ty)) => {
1677 let provide = provide(ty);
1678 (quote!(let __out = #call.await;), provide)
1679 }
1680 (false, None) => (quote!(#call.await;), quote!()),
1681 (true, Some(ty)) => {
1682 let provide = provide(ty);
1683 (
1684 quote!(let __res = { let __fut = ::core::pin::pin!(#call); __node.run_cancellable(__fut).await };),
1685 quote!(if let ::core::result::Result::Ok(__out) = __res {
1686 #provide
1687 }),
1688 )
1689 }
1690 (true, None) => (
1691 quote!({
1692 let __fut = ::core::pin::pin!(#call);
1693 let _ = __node.run_cancellable(__fut).await;
1694 }),
1695 quote!(),
1696 ),
1697 };
1698 let def = quote! {
1699 #(#cfg)*
1700 #[::embassy_executor::task(pool_size = #ps)]
1701 #allow_unreachable
1702 async fn #shell(__node: &'static #cr::TaskNode #(, #res_params)* #state_param) {
1703 #drive
1704 #state_drop
1705 #(#restores)*
1706 #provide_exit
1707 // Record the completion (and ack any pending shutdown handshake):
1708 // a worker that returns on its own reads as down, not running
1709 // forever, and a control Activate can respawn it.
1710 __node.mark_exited();
1711 }
1712 };
1713 let path: Expr = syn::parse_quote!(#shell);
1714 Ok((def, path))
1715}
1716
1717/// Emit a `node`: its `pub static #ident: TaskNode` definition and its `Slot`. The
1718/// caller assigns the slot index and records the name, so this touches neither.
1719/// A `task:` node additionally emits its generated shell ahead of the static.
1720fn emit_node(
1721 n: &NodeItem,
1722 cr: &TokenStream2,
1723 spawn_fn: &TokenStream2,
1724 // Threaded to ready_tokens: a ready dep naming a pool refs its floor member.
1725 pool_names: &std::collections::HashSet<String>,
1726 helpers: &HelperIdents,
1727) -> SynResult<(TokenStream2, Slot)> {
1728 let ident = &n.ident;
1729 let cfg = &n.cfg;
1730 let mode = &n.mode;
1731 let name = name_string(&n.ident);
1732 let disabled = n.disabled;
1733 let (shell_def, spawn_expr) = match &n.source {
1734 Some(TaskSource::Shell(worker)) => {
1735 let ps = match &n.pool_size {
1736 Some(l) => l.base10_parse::<usize>()?,
1737 None => 1,
1738 };
1739 let (def, path) = emit_shell(
1740 ident,
1741 cfg,
1742 worker,
1743 ps,
1744 &n.resources,
1745 n.exit.as_ref(),
1746 n.state.as_ref(),
1747 false,
1748 n.cancel,
1749 cr,
1750 helpers,
1751 )?;
1752 (def, Some(path))
1753 }
1754 Some(TaskSource::Spawn(e)) => (quote!(), Some(e.clone())),
1755 None => (quote!(), None),
1756 };
1757 let spawn = node_spawn(
1758 ident,
1759 &spawn_expr,
1760 &n.executor,
1761 &n.resources,
1762 n.state.as_ref(),
1763 spawn_fn,
1764 helpers,
1765 )?;
1766 // `executor: NAME` routes the node through that SpawnerSlot; the supervisor
1767 // awaits the slot before spawning (see `TaskNode::with_executor`).
1768 let with_exec = match &n.executor {
1769 Some(ex) => quote!( .with_executor(&#ex) ),
1770 None => quote!(),
1771 };
1772 // `resources: [NAME: Type, ..]` — one `pub static NAME: ResourceSlot<Type>`
1773 // per entry (main moves the resource in with `NAME.provide(..)`), plus a
1774 // type-erased gate array wired into the node so the supervisor can await
1775 // provisioning/restore before each (re)spawn (see `TaskNode::with_resources`).
1776 // The unsized coercion `&NAME` -> `&dyn ResourceGate` happens in the static
1777 // initializer, where it is allowed.
1778 let (res_defs, with_res) = if n.resources.is_empty() {
1779 (quote!(), quote!())
1780 } else {
1781 let gates_ident = format_ident!("__SV_GATES_{}", ident);
1782 // `shared` slots are emitted once per graph in `expand` (several items
1783 // may declare the same one); only this node's exclusive (take-kind)
1784 // slots are emitted here.
1785 let slot_defs = n.resources.iter().filter(|r| r.shared.is_none()).map(|r| {
1786 let ecfg = &r.cfg;
1787 let res = &r.ident;
1788 let ty = &r.ty;
1789 // `local` entries use the graph-site slot type (emitted once per
1790 // graph in `expand`): same provide/take protocol as `ResourceSlot`
1791 // but without its `T: Send` bound, for `!Send` driver handles on a
1792 // single-core system. `consume` changes only shell codegen (by-value
1793 // arg, no restore) — the slot type is the same either way.
1794 let slot_ty = if r.local.is_some() {
1795 let local = &helpers.local_slot;
1796 quote!(#local<#ty>)
1797 } else {
1798 quote!(#cr::ResourceSlot<#ty>)
1799 };
1800 let doc = if r.consume.is_some() {
1801 format!(
1802 "Resource slot for node `{ident}` (generated by `supervisor_graph!`). \
1803 Move the resource in with `.provide(..)` before `Supervisor::start`. \
1804 `consume`: the worker owns (and may drop) the value, so the slot is \
1805 empty after the task exits — re-`provide()` before any respawn."
1806 )
1807 } else {
1808 format!(
1809 "Resource slot for node `{ident}` (generated by `supervisor_graph!`). \
1810 Move the resource in with `.provide(..)` before `Supervisor::start`."
1811 )
1812 };
1813 quote! {
1814 #(#cfg)*
1815 #(#ecfg)*
1816 #[doc = #doc]
1817 pub static #res: #slot_ty = <#slot_ty>::new();
1818 }
1819 });
1820 let (gates_len, gate_refs) = gate_tokens(&n.resources);
1821 (
1822 quote! {
1823 #(#slot_defs)*
1824 #(#cfg)*
1825 static #gates_ident: [&'static dyn #cr::ResourceGate; #gates_len] =
1826 [#(#gate_refs),*];
1827 },
1828 quote!( .with_resources(&#gates_ident) ),
1829 )
1830 };
1831 // `slot_timeout: N` — override the node's pre-spawn slot/gate wait bound
1832 // (see `TaskNode::with_slot_timeout`; sized to a provider node's build time).
1833 let with_timeout = match &n.slot_timeout {
1834 Some(ms) => quote!( .with_slot_timeout(#cr::_export::Duration::from_millis(#ms)) ),
1835 None => quote!(),
1836 };
1837 // `deps: [X ready, ..]` — the ready-marked subset becomes a per-node
1838 // `[&'static TaskNode; n]` array wired via `.with_ready_deps`: bring-up
1839 // awaits each one's set_ready() (bounded by slot_timeout) after the
1840 // resource gates. Spawn-order deps are unaffected (same DEPS table).
1841 let (ready_def, with_ready) = match ready_tokens(&n.deps, pool_names) {
1842 Some((len, refs)) => {
1843 let ready_ident = format_ident!("__SV_READY_{}", ident);
1844 (
1845 quote! {
1846 #(#cfg)*
1847 static #ready_ident: [&'static #cr::TaskNode; #len] = [#(#refs),*];
1848 },
1849 quote!( .with_ready_deps(&#ready_ident) ),
1850 )
1851 }
1852 None => (quote!(), quote!()),
1853 };
1854 // `exit: Type` — one `pub static <NODE>_EXIT: ResourceSlot<Type>` the shell
1855 // provide()s the worker's return value into just before mark_exited. Plain
1856 // `ResourceSlot` on purpose: it is an outbound mailbox, not a gated input,
1857 // so it joins no gate array (an empty exit slot must not block a spawn).
1858 let exit_def = match &n.exit {
1859 Some(ty) => {
1860 let exit_ident = format_ident!("{}_EXIT", ident);
1861 let doc = format!(
1862 "Exit-value slot for node `{ident}` (generated by `supervisor_graph!`). \
1863 The generated shell `provide()`s the worker's return value here just \
1864 before recording the exit; read it with `.wait_take()` (or `.take()` \
1865 after `has_exited()`). Overwritten by the next completion."
1866 );
1867 quote! {
1868 #(#cfg)*
1869 #[doc = #doc]
1870 pub static #exit_ident: #cr::ResourceSlot<#ty> =
1871 #cr::ResourceSlot::new();
1872 }
1873 }
1874 None => quote!(),
1875 };
1876 // Every emitted `pub` item carries a doc string: a consumer crate may be
1877 // `#![deny(missing_docs)]`, and the lint fires on macro-generated items.
1878 let node_doc = format!(
1879 "Supervised node `{ident}` (`{mode}`), generated by `supervisor_graph!`. \
1880 Pass it to the supervisor's per-node verbs (`start_node`, `stop_node`, \
1881 `resume_node`, `activate`/`deactivate`); the worker gets the same \
1882 `&'static TaskNode` for the task-side protocol."
1883 );
1884 let def = quote! {
1885 #res_defs
1886 #exit_def
1887 #ready_def
1888 #shell_def
1889 #(#cfg)*
1890 #[doc = #node_doc]
1891 pub static #ident: #cr::TaskNode =
1892 #cr::TaskNode::new(#name, #cr::Mode::#mode, #spawn, #disabled)
1893 #with_exec #with_res #with_timeout #with_ready;
1894 };
1895 let slot = Slot {
1896 cfg_pred: cfg_predicate(cfg),
1897 reference: quote!(&#ident),
1898 deps: n.deps.clone(),
1899 fragment: n.fragment.clone(),
1900 };
1901 Ok((def, slot))
1902}
1903
1904/// Emit a `pool`: the member `[TaskNode; K]` array, the `spawn_<pool>` glue fn, and
1905/// the `ElasticPool` static (returned as `defs`, in that emission order), plus the
1906/// pool-registry entry (for `GRAPH.pools`) and one `Slot` per member (members occupy
1907/// slots but aren't name-addressable, so no name is recorded).
1908fn emit_pool(
1909 p: &PoolItem,
1910 cr: &TokenStream2,
1911 spawn_fn: &TokenStream2,
1912 // Threaded to ready_tokens: a ready dep naming a pool refs its floor member.
1913 pool_names: &std::collections::HashSet<String>,
1914 helpers: &HelperIdents,
1915) -> SynResult<(Vec<TokenStream2>, TokenStream2, Vec<Slot>)> {
1916 let ident = &p.ident;
1917 let cfg = &p.cfg;
1918 let lname = name_string(&p.ident);
1919 let pool_static = format_ident!("{}_POOL", ident);
1920 let k = p.modes.len();
1921
1922 // Validate the scaling bounds. Two paths:
1923 // - both int literals (the common case): validated HERE, at expansion time,
1924 // with the best possible spans. `base10_parse::<u8>` also rejects values
1925 // > 255 (the `ElasticPool` fields are `u8`).
1926 // - otherwise (paths, const exprs — e.g. `min: HTTP_FLOOR`): the emitted
1927 // `<POOL>_MIN`/`<POOL>_MAX` consts become the source of truth and
1928 // `const _: () = assert!(..)` guards enforce min <= max <= members <= 255
1929 // at const-eval time (rendered like the cycle error, with rust-src spans).
1930 // `min > max` makes the policy contradict itself; `max > k` is a ceiling the
1931 // pool can never reach (only `k` member slots exist) — declaration bugs
1932 // either way. `max < k` is allowed (spare declared members below the
1933 // ceiling), as is `min: 0` (scale to zero when idle). The member count `k`
1934 // itself stays a structural literal: it drives how many nodes, shells, name
1935 // strings and graph slots are EMITTED, which a proc macro cannot derive
1936 // from a const it can't evaluate.
1937 let lit_bounds = match (&p.min, &p.max) {
1938 (Expr::Lit(lmin), Expr::Lit(lmax)) => match (&lmin.lit, &lmax.lit) {
1939 (syn::Lit::Int(imin), syn::Lit::Int(imax)) => {
1940 Some((imin.base10_parse::<u8>()?, imax.base10_parse::<u8>()?))
1941 }
1942 _ => None,
1943 },
1944 _ => None,
1945 };
1946 if let Some((min_v, max_v)) = lit_bounds {
1947 if min_v > max_v {
1948 return Err(syn::Error::new_spanned(
1949 &p.min,
1950 format!("pool `min:` ({min_v}) must not exceed `max:` ({max_v})"),
1951 ));
1952 }
1953 if usize::from(max_v) > k {
1954 return Err(syn::Error::new_spanned(
1955 &p.max,
1956 format!("pool `max:` ({max_v}) exceeds the declared member count ({k})"),
1957 ));
1958 }
1959 }
1960
1961 // Pool `resources:` (all `shared`, enforced at parse) additionally require
1962 // `task:` — same rule as nodes: the generated shell is what receives the
1963 // values as arguments (a hand-written `spawn:` task fn manages its own).
1964 if !p.resources.is_empty() && matches!(p.source, TaskSource::Spawn(_)) {
1965 return Err(syn::Error::new_spanned(
1966 &p.resources[0].ident,
1967 "pool `resources:` requires `task:` — the values are handed to the \
1968 generated shell as arguments (and lend entries restored by it); a \
1969 `spawn:` task fn manages its own arguments",
1970 ));
1971 }
1972 if let Some((ty, _)) = &p.state
1973 && matches!(p.source, TaskSource::Spawn(_))
1974 {
1975 return Err(syn::Error::new_spanned(
1976 ty,
1977 "pool `state:` requires `task:` — the generated shell owns the boxed \
1978 state across the worker call; a `spawn:` task fn can Box its own",
1979 ));
1980 }
1981
1982 // Resolve the member task: `spawn:` uses the given expr directly; `task:`
1983 // first stamps ONE generated shell sized `pool_size = K` (all members share a
1984 // single concrete future type) and targets that. Shared resources become
1985 // by-value shell parameters, exactly like a node's.
1986 let (shell_def, member_expr) = match &p.source {
1987 TaskSource::Spawn(e) => (quote!(), e.clone()),
1988 TaskSource::Shell(worker) => emit_shell(
1989 ident,
1990 cfg,
1991 worker,
1992 k,
1993 &p.resources,
1994 None,
1995 p.state.as_ref(),
1996 true,
1997 p.cancel,
1998 cr,
1999 helpers,
2000 )?,
2001 };
2002 // Build member `I`'s spawn call from the member task, injecting `&POOL[I]`
2003 // as the first argument, then the shared resource copies (see
2004 // `inject_call_with`).
2005 let res_args: Vec<TokenStream2> = p
2006 .resources
2007 .iter()
2008 .enumerate()
2009 .flat_map(|(i, r)| {
2010 let ecfg = &r.cfg;
2011 let var = format_ident!("__r{}", i);
2012 let res = &r.ident;
2013 if r.shared.is_none() && r.consume.is_none() {
2014 // Lend: value + the member's own slot element, so the shell
2015 // restores to the same index it was taken from.
2016 vec![quote!(#(#ecfg)* #var), quote!(#(#ecfg)* &#res[I])]
2017 } else {
2018 vec![quote!(#(#ecfg)* #var)]
2019 }
2020 })
2021 .collect();
2022 let try_box = &helpers.try_box;
2023 let (state_prelude, state_arg) = match &p.state {
2024 Some((_, init)) => (
2025 quote! {
2026 let __state = #try_box(#init)
2027 .ok_or(::embassy_executor::SpawnError::Busy)?;
2028 },
2029 vec![quote!(__state)],
2030 ),
2031 None => (quote!(), vec![]),
2032 };
2033 let mut lead: Vec<TokenStream2> = vec![quote!(&#ident[I])];
2034 lead.extend(res_args);
2035 lead.extend(state_arg);
2036 let call = inject_call_with(&member_expr, &lead)?;
2037 // Per-member spawn fn: a generated `spawn_<pool>::<I>` wrapper. Same optional
2038 // trace capture as a node's closure, against member `I`'s slot. With
2039 // `executor: NAME` the wrapper ignores the supervisor's `Spawner` and spawns
2040 // through the named SpawnerSlot; each member node carries `.with_executor(&EX)`,
2041 // so the supervisor awaits the slot (bounded) before invoking the wrapper and
2042 // the wrapper's `get()` is already filled (`SpawnError::Busy` guards a never-
2043 // filled slot; member futures must be `Send`). A whole worker pool can thus live
2044 // on another executor — e.g. the second core — while this core scales it.
2045 let (param, prelude, sp_tokens) = match &p.executor {
2046 None => (quote!(s), quote!(), quote!(s)),
2047 Some(ex) => (
2048 quote!(_s),
2049 quote! {
2050 let __sp = #ex
2051 .get()
2052 .ok_or(::embassy_executor::SpawnError::Busy)?;
2053 },
2054 quote!(__sp),
2055 ),
2056 };
2057 // Resource prelude, kind-aware. `shared`: copy the fan-out handle out
2058 // non-destructively (slot stays filled for the next member/consumer).
2059 // Take kinds (lend/consume): take from THIS member's array element —
2060 // `RES[I]`, per-member exclusive by construction. Either way an unprovided
2061 // slot fail-closes the member's spawn with `SpawnError::Busy`. After the
2062 // executor-slot guard, same ordering rationale as a node's glue.
2063 let get_prelude: Vec<TokenStream2> = p
2064 .resources
2065 .iter()
2066 .enumerate()
2067 .map(|(i, r)| {
2068 let ecfg = &r.cfg;
2069 let res = &r.ident;
2070 let var = format_ident!("__r{}", i);
2071 if r.shared.is_some() {
2072 quote! {
2073 #(#ecfg)*
2074 let #var = #res
2075 .get()
2076 .ok_or(::embassy_executor::SpawnError::Busy)?;
2077 }
2078 } else {
2079 quote! {
2080 #(#ecfg)*
2081 let #var = #res[I]
2082 .take()
2083 .ok_or(::embassy_executor::SpawnError::Busy)?;
2084 }
2085 }
2086 })
2087 .collect();
2088 let pool_spawn_stmts = spawn_stmts(&call, "e!(&#ident[I]), &sp_tokens);
2089 let wrapper = format_ident!("spawn_{}", lname);
2090 let mut defs: Vec<TokenStream2> = Vec::new();
2091 defs.push(shell_def);
2092 defs.push(quote! {
2093 #(#cfg)*
2094 fn #wrapper<const I: usize>(
2095 #param: ::embassy_executor::Spawner,
2096 ) -> ::core::result::Result<(), ::embassy_executor::SpawnError> {
2097 #prelude
2098 #state_prelude
2099 #(#get_prelude)*
2100 #pool_spawn_stmts
2101 ::core::result::Result::Ok(())
2102 }
2103 });
2104 let member_spawn: Vec<TokenStream2> = (0..k).map(|j| quote!(#wrapper::<#j>)).collect();
2105
2106 // `executor: NAME` on the pool routes every member through that SpawnerSlot; the
2107 // supervisor awaits it before spawning each member (see `TaskNode::with_executor`).
2108 let member_with_exec = match &p.executor {
2109 Some(ex) => quote!( .with_executor(&#ex) ),
2110 None => quote!(),
2111 };
2112 // Take-kind entries (lend/consume) get per-member SLOT ARRAYS: member `I`
2113 // takes/restores index `I` exclusively, so members don't contend and the
2114 // elastic floor can come up with only floor-many elements provided. The
2115 // shared slot statics are emitted once per graph in `expand`, as for nodes.
2116 for r in p.resources.iter().filter(|r| r.shared.is_none()) {
2117 let ecfg = &r.cfg;
2118 let res = &r.ident;
2119 let ty = &r.ty;
2120 let doc = format!(
2121 "Per-member resource slots for pool `{ident}` (generated by \
2122 `supervisor_graph!`): member `I` takes/restores element `I`. \
2123 Provide at least the floor members' elements before \
2124 `Supervisor::start`; a member whose element is empty fail-closes \
2125 its (re)spawn with `SpawnError::Busy`."
2126 );
2127 defs.push(quote! {
2128 #(#cfg)*
2129 #(#ecfg)*
2130 #[doc = #doc]
2131 pub static #res: [#cr::ResourceSlot<#ty>; #k] =
2132 [const { #cr::ResourceSlot::new() }; #k];
2133 });
2134 }
2135 // Per-member gate arrays: member `j` gates on ITS OWN take-kind elements
2136 // plus the pool-wide shared slots. Same cfg-aware length for every member.
2137 let member_with_res: Vec<TokenStream2> = if p.resources.is_empty() {
2138 (0..k).map(|_| quote!()).collect()
2139 } else {
2140 let any_cfg = p.resources.iter().any(|r| cfg_predicate(&r.cfg).is_some());
2141 let gates_len = if any_cfg {
2142 let terms: Vec<TokenStream2> = p
2143 .resources
2144 .iter()
2145 .map(|r| match cfg_predicate(&r.cfg) {
2146 None => quote!(1usize),
2147 Some(pred) => quote!({
2148 #[cfg(#pred)]
2149 {
2150 1usize
2151 }
2152 #[cfg(not(#pred))]
2153 {
2154 0usize
2155 }
2156 }),
2157 })
2158 .collect();
2159 quote!(0usize #(+ #terms)*)
2160 } else {
2161 let n = p.resources.len();
2162 quote!(#n)
2163 };
2164 (0..k)
2165 .map(|j| {
2166 let gates_ident = format_ident!("__SV_GATES_{}_{}", ident, j);
2167 let gate_refs: Vec<TokenStream2> = p
2168 .resources
2169 .iter()
2170 .map(|r| {
2171 let ecfg = &r.cfg;
2172 let res = &r.ident;
2173 if r.shared.is_some() {
2174 quote!(#(#ecfg)* &#res)
2175 } else {
2176 quote!(#(#ecfg)* &#res[#j])
2177 }
2178 })
2179 .collect();
2180 defs.push(quote! {
2181 #(#cfg)*
2182 static #gates_ident: [&'static dyn #cr::ResourceGate; #gates_len] =
2183 [#(#gate_refs),*];
2184 });
2185 quote!( .with_resources(&#gates_ident) )
2186 })
2187 .collect()
2188 };
2189 // `deps: [X ready, ..]` — ONE shared ready-dep array for the whole pool
2190 // (markers apply to every member; growth also checks it synchronously).
2191 let member_with_ready = match ready_tokens(&p.deps, pool_names) {
2192 Some((len, refs)) => {
2193 let ready_ident = format_ident!("__SV_READY_{}", ident);
2194 defs.push(quote! {
2195 #(#cfg)*
2196 static #ready_ident: [&'static #cr::TaskNode; #len] = [#(#refs),*];
2197 });
2198 quote!( .with_ready_deps(&#ready_ident) )
2199 }
2200 None => quote!(),
2201 };
2202 // `slot_timeout: N` — every member's pre-spawn slot/gate wait bound.
2203 let member_with_timeout = match &p.slot_timeout {
2204 Some(ms) => quote!( .with_slot_timeout(#cr::_export::Duration::from_millis(#ms)) ),
2205 None => quote!(),
2206 };
2207 let members = p
2208 .modes
2209 .iter()
2210 .zip(&member_spawn)
2211 .enumerate()
2212 .map(|(j, (mode, sp))| {
2213 let nm = format!("{lname}{j}");
2214 let with_res = &member_with_res[j];
2215 quote! {
2216 #cr::TaskNode::new(
2217 #nm, #cr::Mode::#mode,
2218 ::core::option::Option::Some((#sp) as #spawn_fn), false,
2219 ) #member_with_exec #with_res #member_with_timeout #member_with_ready
2220 }
2221 });
2222 defs.push(quote! {
2223 #(#cfg)*
2224 #[doc = concat!("Pool `", stringify!(#ident), "`'s members, one `TaskNode` per slot \
2225 (index = member index). Index it for the per-node verbs; the pool itself is \
2226 `", stringify!(#ident), "_POOL`.")]
2227 pub static #ident: [#cr::TaskNode; #k] = [ #(#members),* ];
2228 });
2229
2230 // Structural constants, for downstream compile-time sizing (e.g. a socket
2231 // budget: `const BUDGET: usize = HTTP_MAX + 1`). Emitted because user code
2232 // can't derive them from the member array — a `const` cannot refer to a
2233 // `static` (E0013), so `HTTP.len()` is unusable in const context and the
2234 // count would otherwise have to be duplicated by hand next to the DSL.
2235 let min_const = format_ident!("{}_MIN", ident);
2236 let max_const = format_ident!("{}_MAX", ident);
2237 let members_const = format_ident!("{}_MEMBERS", ident);
2238 // Literal path: emit the *validated* u8 values. Expr path: the consts ARE
2239 // the source of truth (any const-evaluable usize expr) and const asserts
2240 // enforce what the literal path checked at expansion.
2241 let (min_tokens, max_tokens, bound_asserts) = match lit_bounds {
2242 Some((min_v, max_v)) => {
2243 let (min_u, max_u) = (usize::from(min_v), usize::from(max_v));
2244 (quote!(#min_u), quote!(#max_u), quote!())
2245 }
2246 None => {
2247 let (min_e, max_e) = (&p.min, &p.max);
2248 (
2249 quote!({ #min_e }),
2250 quote!({ #max_e }),
2251 quote! {
2252 #(#cfg)*
2253 const _: () = ::core::assert!(
2254 #min_const <= #max_const,
2255 "pool `min:` must not exceed `max:`",
2256 );
2257 #(#cfg)*
2258 const _: () = ::core::assert!(
2259 #max_const <= #members_const,
2260 "pool `max:` exceeds the declared member count",
2261 );
2262 #(#cfg)*
2263 const _: () = ::core::assert!(
2264 #max_const <= 255,
2265 "pool `max:` exceeds 255 (ElasticPool bounds are u8)",
2266 );
2267 },
2268 )
2269 }
2270 };
2271 defs.push(quote! {
2272 #(#cfg)*
2273 #[doc = concat!("Pool `", stringify!(#ident), "`'s `min:` floor (validated at expansion or by const assert).")]
2274 pub const #min_const: usize = #min_tokens;
2275 #(#cfg)*
2276 #[doc = concat!("Pool `", stringify!(#ident), "`'s `max:` scaling ceiling — the most members ever running concurrently.")]
2277 pub const #max_const: usize = #max_tokens;
2278 #(#cfg)*
2279 #[doc = concat!("Pool `", stringify!(#ident), "`'s declared member count (the `[TaskNode; K]` array length).")]
2280 pub const #members_const: usize = #k;
2281 #bound_asserts
2282 });
2283
2284 let member_refs = (0..k).map(|j| quote!(&#ident[#j]));
2285 let policy = &p.policy;
2286 // The `ElasticPool<P>` type argument: honor an explicit `policy: <Ty> = ..`
2287 // annotation, else derive `P` from the constructor expr (`Ty::new(..)` shape).
2288 let policy_ty = match &p.policy_ty {
2289 Some(ty) => quote!(#ty),
2290 None => {
2291 let path = policy_type(policy)?;
2292 quote!(#path)
2293 }
2294 };
2295 // The u8 fields come from the emitted consts, which both paths validate
2296 // (parse-time for literals, const asserts otherwise) — so the `as u8` casts
2297 // cannot truncate. Going through the consts also keeps a suffixed literal
2298 // like `min: 3usize` working (the const is usize either way).
2299 defs.push(quote! {
2300 #(#cfg)*
2301 #[doc = concat!("The `ElasticPool` over the `", stringify!(#ident), "` members: \
2302 the `min:`/`max:` bounds and the scaling policy `Supervisor::run_pools` \
2303 drives. Also reachable through `GRAPH.pools`.")]
2304 pub static #pool_static: #cr::ElasticPool<#policy_ty> = #cr::ElasticPool {
2305 nodes: &[ #(#member_refs),* ],
2306 min: #min_const as u8,
2307 max: #max_const as u8,
2308 policy: #policy,
2309 };
2310 });
2311
2312 let pool_entry = quote!( #(#cfg)* &#pool_static );
2313
2314 let pred = cfg_predicate(cfg);
2315 let slots = (0..k)
2316 .map(|j| Slot {
2317 cfg_pred: pred.clone(),
2318 reference: quote!(&#ident[#j]),
2319 deps: p.deps.clone(),
2320 fragment: p.fragment.clone(),
2321 })
2322 .collect();
2323
2324 Ok((defs, pool_entry, slots))
2325}
2326
2327/// Second pass: build the node-slot entries for `GRAPH.nodes` (`Option`, cfg-gated) and
2328/// the cfg-aware dep-index entries for `GRAPH.deps`. Runs after every slot + name is
2329/// known, since a dep may forward-reference a node declared later. An unknown dep name
2330/// is a compile error.
2331fn slot_tables(
2332 slots: &[Slot],
2333 names: &HashMap<String, usize>,
2334) -> SynResult<(Vec<TokenStream2>, Vec<TokenStream2>)> {
2335 let mut all_entries: Vec<TokenStream2> = Vec::new();
2336 let mut deps_entries: Vec<TokenStream2> = Vec::new();
2337 for slot in slots {
2338 let reference = &slot.reference;
2339 all_entries.push(match &slot.cfg_pred {
2340 None => quote!(::core::option::Option::Some(#reference)),
2341 Some(pred) => quote!({
2342 #[cfg(#pred)]
2343 { ::core::option::Option::Some(#reference) }
2344 #[cfg(not(#pred))]
2345 { ::core::option::Option::None }
2346 }),
2347 });
2348
2349 let mut dep_toks: Vec<TokenStream2> = Vec::new();
2350 // Duplicate deps are a compile error: `deps: [A, A]` would emit a doubled
2351 // index, which `topo_sort_const` counts twice in the in-degree but decrements
2352 // once — misreported as a dependency cycle. Compared by *resolved* slot index
2353 // (so a repeated pool name trips it too); two cfg-gated variants of the same
2354 // dep are allowed only when their cfg predicates differ.
2355 let mut seen: Vec<(u8, String)> = Vec::new();
2356 for d in &slot.deps {
2357 let idx = match names.get(&d.ident.to_string()) {
2358 Some(&i) => i as u8,
2359 None => {
2360 return Err(syn::Error::new_spanned(
2361 &d.ident,
2362 format!(
2363 "unknown dependency `{}` — not a declared node or pool{}",
2364 d.ident,
2365 fragment_suffix(&slot.fragment),
2366 ),
2367 ));
2368 }
2369 };
2370 let cfg = &d.cfg;
2371 let cfg_key = quote!( #(#cfg)* ).to_string();
2372 if seen.iter().any(|(i, k)| *i == idx && *k == cfg_key) {
2373 return Err(syn::Error::new_spanned(
2374 &d.ident,
2375 format!("duplicate dependency `{}`", d.ident),
2376 ));
2377 }
2378 seen.push((idx, cfg_key));
2379 dep_toks.push(quote!( #(#cfg)* #idx ));
2380 }
2381 deps_entries.push(quote!( &[ #(#dep_toks),* ] ));
2382 }
2383 Ok((all_entries, deps_entries))
2384}
2385
2386fn expand(graph: GraphSpec) -> SynResult<TokenStream2> {
2387 let cr = quote!(::embassy_supervisor);
2388 let helpers = HelperIdents::new(graph.name.as_ref());
2389 // The node spawn fn-pointer type. Spawn exprs (closures / const-generic fns) are
2390 // cast to this so they coerce cleanly inside `Option::Some(..)`.
2391 let spawn_fn = quote!(
2392 fn(
2393 ::embassy_executor::Spawner,
2394 ) -> ::core::result::Result<(), ::embassy_executor::SpawnError>
2395 );
2396
2397 // First pass: emit the statics/glue in declaration order, assign stable slot
2398 // indices, and record each slot + its raw deps. `names` maps a dep-addressable ident
2399 // to its slot index for dep resolution — keyed on the *raw* ident (not the runtime
2400 // `name_string`). A `node` maps to its own slot; a `pool` maps to its floor member's
2401 // slot (so `deps: [POOL]` = "after the pool is up"). Individual pool members are not
2402 // separately name-addressable.
2403 let mut defs: Vec<TokenStream2> = Vec::new();
2404 let mut pool_entries: Vec<TokenStream2> = Vec::new();
2405 let mut slots: Vec<Slot> = Vec::new();
2406 let mut names: HashMap<String, usize> = HashMap::new();
2407
2408 // Iff any `resources:` entry is `local`-marked, emit the local slot TYPE once
2409 // per graph (the per-entry statics in `emit_node` reference it by name). It
2410 // mirrors `embassy_supervisor::ResourceSlot` — same provide/take/restore
2411 // protocol, same critical-section interior, same `ResourceGate` view — but
2412 // WITHOUT the `T: Send` bound, so it can carry the `!Send` driver handles
2413 // (`RefCell`-/`NoopRawMutex`-based: `embassy_net::Stack` runners,
2414 // `cyw43::Control`, …) that a single-core system hands between its own tasks.
2415 // That requires asserting `Sync` for a `!Send` payload, so like the
2416 // `trace-hooks` symbols it is emitted here, at the graph declaration site,
2417 // where the application owns the soundness contract (see the SAFETY note).
2418 // `state:` anywhere in the graph: emit the fallible-boxing helper ONCE, at
2419 // the graph site (like the local slot type). This is the `heap-state`
2420 // feature's ENTIRE unsafe surface, and it lives in the CONSUMER crate (the
2421 // `local-resources` precedent): raw alloc + null check + ptr::write +
2422 // Box::from_raw — after which it is a NORMAL Box, freed by ordinary drop
2423 // when the shell drops it on task exit. Alloc failure returns None (the
2424 // glue maps it to SpawnError::Busy; the init value is dropped normally).
2425 let any_state = graph.items.iter().any(|item| match item {
2426 Item::Node(n) => n.state.is_some(),
2427 Item::Pool(p) => p.state.is_some(),
2428 Item::Executor(_) => false,
2429 });
2430 if any_state {
2431 let try_box = &helpers.try_box;
2432 let alloc_alias = &helpers.alloc_alias;
2433 defs.push(quote! {
2434 extern crate alloc as #alloc_alias;
2435 /// Fallible boxing for `state:` clauses (generated by
2436 /// `supervisor_graph!`). Returns `None` when the global allocator
2437 /// is out of memory — surfaced by the spawn glue as
2438 /// `SpawnError::Busy`, retryable once heap frees up. The value is
2439 /// written to the heap allocation directly; note the INIT argument
2440 /// itself is materialized in this call's frame first (rustc may or
2441 /// may not elide the copy) — keep `state:` types reasonably sized
2442 /// or box internal bulk.
2443 #[doc(hidden)]
2444 fn #try_box<T>(init: T) -> ::core::option::Option<#alloc_alias::boxed::Box<T>> {
2445 let layout = ::core::alloc::Layout::new::<T>();
2446 if layout.size() == 0 {
2447 // ZST: no allocation. Box<ZST> from a dangling well-aligned
2448 // pointer is the documented representation; `init` is
2449 // forgotten so T's drop (if any) runs exactly once, via the
2450 // Box.
2451 ::core::mem::forget(init);
2452 // SAFETY: dangling NonNull is valid for a ZST Box.
2453 return ::core::option::Option::Some(unsafe {
2454 #alloc_alias::boxed::Box::from_raw(
2455 ::core::ptr::NonNull::<T>::dangling().as_ptr(),
2456 )
2457 });
2458 }
2459 // SAFETY: `layout` has non-zero size. On success the pointer is
2460 // valid for writes of `T` and exclusively ours; `write`
2461 // initializes it; `from_raw` then owns an allocation made with
2462 // the global allocator and `T`'s layout — a normal `Box`.
2463 unsafe {
2464 let p = #alloc_alias::alloc::alloc(layout) as *mut T;
2465 if p.is_null() {
2466 return ::core::option::Option::None; // `init` drops here
2467 }
2468 ::core::ptr::write(p, init);
2469 ::core::option::Option::Some(#alloc_alias::boxed::Box::from_raw(p))
2470 }
2471 }
2472 });
2473 }
2474
2475 let any_local = graph
2476 .items
2477 .iter()
2478 .any(|item| item_resources(item).iter().any(|r| r.local.is_some()));
2479 if any_local {
2480 let local = helpers.local_slot.clone();
2481 // `Cell<Option<T>>` spelled through absolute paths (macro output must not
2482 // rely on the caller's prelude/imports); the mutex/signal types come from
2483 // the supervisor's `_export` shim so the consumer needs no direct
2484 // `embassy-sync` dependency.
2485 let cell = quote!(::core::cell::Cell<::core::option::Option<T>>);
2486 let raw = quote!(#cr::_export::CriticalSectionRawMutex);
2487 let signal = quote!(#cr::_export::Signal<#raw, ()>);
2488 defs.push(quote! {
2489 /// One-value handoff cell for a `local`-marked `resources:` entry
2490 /// (generated by `supervisor_graph!`). Protocol and fail-closed
2491 /// semantics of `embassy_supervisor::ResourceSlot`, minus its
2492 /// `T: Send` bound — for `!Send` driver handles on a single core.
2493 ///
2494 /// Contract (see the `unsafe impl Sync` below): every `provide` /
2495 /// `take` / `restore` of a given slot must happen on the SAME core.
2496 // `dead_code`/`missing_docs` in the consumer: the type is emitted
2497 // whenever a `local` entry is *declared*, even if every declaring
2498 // node is `#[cfg]`-compiled out of this build.
2499 #[allow(dead_code)]
2500 pub struct #local<T> {
2501 slot: #cr::_export::BlockingMutex<#raw, #cell>,
2502 filled: #signal,
2503 }
2504 // SAFETY: the payload is intentionally NOT `Send` — this assertion is
2505 // exactly the single-core contract: the value only ever moves between
2506 // executors/tasks of one core (interrupt-safe via the critical-section
2507 // mutex around every access), never across cores. The macro rejects
2508 // `local` + `executor:` so a slot cannot feed a `SendSpawner`-routed
2509 // node, and a multi-core application must not `provide`/`take` a given
2510 // slot from different cores.
2511 unsafe impl<T> ::core::marker::Sync for #local<T> {}
2512 #[allow(dead_code)]
2513 impl<T> #local<T> {
2514 /// An empty slot (`const` — it lives in the generated `static`s).
2515 pub const fn new() -> Self {
2516 Self {
2517 slot: #cr::_export::BlockingMutex::new(
2518 ::core::cell::Cell::new(::core::option::Option::None),
2519 ),
2520 filled: #cr::_export::Signal::new(),
2521 }
2522 }
2523 /// Move the resource in and wake the supervisor's pre-spawn wait.
2524 pub fn provide(&self, value: T) {
2525 self.slot.lock(|c| c.set(::core::option::Option::Some(value)));
2526 self.filled.signal(());
2527 }
2528 /// Take the resource out, leaving the slot empty (spawn glue).
2529 pub fn take(&self) -> ::core::option::Option<T> {
2530 self.slot.lock(::core::cell::Cell::take)
2531 }
2532 /// Put the resource back for the next spawn (generated shell;
2533 /// not emitted for `consume` entries).
2534 pub fn restore(&self, value: T) {
2535 self.provide(value);
2536 }
2537 }
2538 #[allow(dead_code)]
2539 impl<T: ::core::marker::Copy> #local<T> {
2540 /// Copy the value out WITHOUT emptying the slot — the `shared`
2541 /// kind's fan-out read (any number of consumers, slot stays
2542 /// filled). `T: Copy` only.
2543 pub fn get(&self) -> ::core::option::Option<T> {
2544 self.slot.lock(|c| {
2545 let v = c.take();
2546 c.set(v);
2547 v
2548 })
2549 }
2550 }
2551 impl<T> ::core::default::Default for #local<T> {
2552 fn default() -> Self {
2553 Self::new()
2554 }
2555 }
2556 impl<T> #cr::ResourceGate for #local<T> {
2557 fn is_filled(&self) -> bool {
2558 // Peek without consuming: `Cell` has no `&T` access, so
2559 // take-and-put-back under the same critical section.
2560 self.slot.lock(|c| {
2561 let v = c.take();
2562 let filled = v.is_some();
2563 c.set(v);
2564 filled
2565 })
2566 }
2567 fn filled_signal(&self) -> &#signal {
2568 &self.filled
2569 }
2570 }
2571 });
2572 }
2573
2574 // Pre-pass: collect the declared `executor NAME;` slots so a node's
2575 // `executor:` reference can be validated regardless of declaration order.
2576 let helpers = HelperIdents::new(graph.name.as_ref());
2577 let executor_names: Vec<String> = graph
2578 .items
2579 .iter()
2580 .filter_map(|i| match i {
2581 Item::Executor(x) => Some(x.ident.to_string()),
2582 _ => None,
2583 })
2584 .collect();
2585 // Pool idents, known up front: a `ready`-marked dep naming a pool resolves
2586 // to the pool's floor member (`&POOL[0]`), and forward references are legal.
2587 let pool_names: std::collections::HashSet<String> = graph
2588 .items
2589 .iter()
2590 .filter_map(|i| match i {
2591 Item::Pool(p) => Some(p.ident.to_string()),
2592 _ => None,
2593 })
2594 .collect();
2595
2596 // Pre-pass: `resources:` slot names become `pub static`s at the declaration
2597 // site, so take-kind names must be unique across the whole graph — and no
2598 // resource may shadow an `executor NAME;` static. `shared` entries are the
2599 // deliberate exception: the SAME name on several items is one fan-out slot,
2600 // emitted once (below, with the union of the declaring sites' cfg
2601 // predicates so it exists whenever any consumer does) — provided every
2602 // re-declaration repeats the kinds + type verbatim. Caught here with
2603 // targeted messages instead of rustc's downstream duplicate-static E0428.
2604 struct SharedPlan<'a> {
2605 /// First declaration — supplies the emitted static's ident (span), type,
2606 /// and `local` flag.
2607 decl: &'a ResourceDecl,
2608 /// Kinds+type token string every re-declaration must match.
2609 sig: String,
2610 /// One entry per declaring site: `None` = unconditional (the slot is
2611 /// then unconditional too), `Some(pred)` = that site's combined
2612 /// item-level + entry-level cfg predicate.
2613 preds: Vec<Option<TokenStream2>>,
2614 /// Declaring node/pool names, for the generated doc comment.
2615 owners: Vec<String>,
2616 }
2617 let mut shared_plans: Vec<(String, SharedPlan)> = Vec::new();
2618 {
2619 let mut taken: HashSet<String> = HashSet::new();
2620 for item in &graph.items {
2621 let Some((owner, item_cfg)) = item_ident_cfg(item) else {
2622 continue;
2623 };
2624 let item_pred = cfg_predicate(item_cfg);
2625 for r in item_resources(item) {
2626 let key = r.ident.to_string();
2627 if executor_names.contains(&key) {
2628 return Err(syn::Error::new_spanned(
2629 &r.ident,
2630 format!(
2631 "resource name `{}` shadows an `executor {};` slot — \
2632 both are statics at the declaration site",
2633 r.ident, r.ident
2634 ),
2635 ));
2636 }
2637 // A site's presence predicate: the item's cfg AND the entry's.
2638 let pred = match (item_pred.clone(), cfg_predicate(&r.cfg)) {
2639 (None, None) => None,
2640 (Some(p), None) | (None, Some(p)) => Some(p),
2641 (Some(a), Some(b)) => Some(quote!(all(#a, #b))),
2642 };
2643 if r.shared.is_some() {
2644 if taken.contains(&key) {
2645 return Err(syn::Error::new_spanned(
2646 &r.ident,
2647 format!(
2648 "`{}` is already a take-kind resource elsewhere in \
2649 the graph — a name is either one exclusive slot or \
2650 one `shared` slot, not both",
2651 r.ident
2652 ),
2653 ));
2654 }
2655 let sig = r.shared_signature();
2656 match shared_plans.iter_mut().find(|(k, _)| *k == key) {
2657 Some((_, plan)) => {
2658 if plan.sig != sig {
2659 return Err(syn::Error::new_spanned(
2660 &r.ident,
2661 format!(
2662 "shared resource `{}` re-declared with a \
2663 different shape: `{}` here vs `{}` on \
2664 `{}` — every declaration of a shared slot \
2665 must repeat the same kind markers and type",
2666 r.ident, sig, plan.sig, plan.owners[0]
2667 ),
2668 ));
2669 }
2670 plan.preds.push(pred);
2671 plan.owners.push(owner.to_string());
2672 }
2673 None => shared_plans.push((
2674 key,
2675 SharedPlan {
2676 decl: r,
2677 sig,
2678 preds: vec![pred],
2679 owners: vec![owner.to_string()],
2680 },
2681 )),
2682 }
2683 } else {
2684 if !taken.insert(key.clone()) || shared_plans.iter().any(|(k, _)| *k == key) {
2685 return Err(syn::Error::new_spanned(
2686 &r.ident,
2687 format!(
2688 "duplicate resource name `{}` — resource slots are \
2689 statics and must be unique across the graph (only \
2690 `shared` entries may repeat a name)",
2691 r.ident
2692 ),
2693 ));
2694 }
2695 }
2696 }
2697 }
2698 }
2699 // Emit each shared slot once. Presence: unconditional if ANY declaring site
2700 // is, else `#[cfg(any(<site preds>))]` — the slot exists whenever at least
2701 // one consumer does.
2702 for (_, plan) in &shared_plans {
2703 let res = &plan.decl.ident;
2704 let ty = &plan.decl.ty;
2705 let slot_ty = if plan.decl.local.is_some() {
2706 let local = &helpers.local_slot;
2707 quote!(#local<#ty>)
2708 } else {
2709 quote!(#cr::ResourceSlot<#ty>)
2710 };
2711 let cfg_attr = if plan.preds.iter().any(|p| p.is_none()) {
2712 quote!()
2713 } else {
2714 let preds = plan.preds.iter().flatten();
2715 quote!(#[cfg(any(#(#preds),*))])
2716 };
2717 let doc = format!(
2718 "Shared (fan-out) resource slot declared by `{}` (generated by \
2719 `supervisor_graph!`). `provide()` the `Copy` handle before \
2720 `Supervisor::start`; every consumer's glue copies it out with \
2721 `get()`, so the slot STAYS FILLED — re-`provide()` only to replace \
2722 the handle (e.g. after rebuilding the underlying driver).",
2723 plan.owners.join("`, `"),
2724 );
2725 defs.push(quote! {
2726 #cfg_attr
2727 #[doc = #doc]
2728 pub static #res: #slot_ty = <#slot_ty>::new();
2729 });
2730 }
2731
2732 for item in &graph.items {
2733 match item {
2734 Item::Node(n) => {
2735 if let Some(ex) = &n.executor
2736 && !executor_names.contains(&ex.to_string())
2737 {
2738 return Err(syn::Error::new_spanned(
2739 ex,
2740 format!(
2741 "unknown executor `{ex}`; declare it in the graph with \
2742 `executor {ex};` (declared: [{}])",
2743 executor_names.join(", ")
2744 ),
2745 ));
2746 }
2747 // The index is the slot's position, taken *before* the push.
2748 // A redeclared name is a hard error here (not just the downstream
2749 // `duplicate definition of static`): deps resolve through this map,
2750 // so a silent overwrite would silently rewire earlier `deps:` edges.
2751 if names.insert(n.ident.to_string(), slots.len()).is_some() {
2752 return Err(syn::Error::new_spanned(
2753 &n.ident,
2754 format!(
2755 "duplicate node/pool name `{}`{}",
2756 n.ident,
2757 fragment_suffix(&n.fragment),
2758 ),
2759 ));
2760 }
2761 let (def, slot) = emit_node(n, &cr, &spawn_fn, &pool_names, &helpers)?;
2762 defs.push(def);
2763 slots.push(slot);
2764 }
2765 Item::Executor(x) => {
2766 let (cfg, ident) = (&x.cfg, &x.ident);
2767 // A runtime-filled SendSpawner slot: the app registers the
2768 // executor's spawner before `Supervisor::start`; nodes declared
2769 // `executor: NAME` spawn through it. Occupies no graph slot.
2770 defs.push(quote! {
2771 #(#cfg)*
2772 /// Spawner slot for the graph's `executor:`-annotated nodes
2773 /// (generated by `supervisor_graph!`). Fill with
2774 /// `SpawnerSlot::set` before `Supervisor::start`.
2775 pub static #ident: #cr::SpawnerSlot = #cr::SpawnerSlot::new();
2776 });
2777 }
2778 Item::Pool(p) => {
2779 // Pools are only meaningful with the supervisor's `pool` feature (which
2780 // forwards to this crate). Without it, `Graph` has no `pools` field and
2781 // `ElasticPool` doesn't exist — so refuse a `pool` with a clear message
2782 // rather than emitting dangling references.
2783 if cfg!(feature = "pool") {
2784 if let Some(ex) = &p.executor
2785 && !executor_names.contains(&ex.to_string())
2786 {
2787 return Err(syn::Error::new_spanned(
2788 ex,
2789 format!(
2790 "unknown executor `{ex}`; declare it in the graph with \
2791 `executor {ex};` (declared: [{}])",
2792 executor_names.join(", ")
2793 ),
2794 ));
2795 }
2796 let (pool_defs, pool_entry, pool_slots) =
2797 emit_pool(p, &cr, &spawn_fn, &pool_names, &helpers)?;
2798 // A dep on the pool NAME resolves to the pool's floor member (member 0
2799 // — the `min`-kept, always-started member): `deps: [POOL]` means "after
2800 // the pool is up". `slots.len()` here is that member's slot index, taken
2801 // *before* the extend below (pool_slots[0] lands at exactly this index).
2802 // A redeclared name errors, same as the node arm.
2803 if names.insert(p.ident.to_string(), slots.len()).is_some() {
2804 return Err(syn::Error::new_spanned(
2805 &p.ident,
2806 format!(
2807 "duplicate node/pool name `{}`{}",
2808 p.ident,
2809 fragment_suffix(&p.fragment),
2810 ),
2811 ));
2812 }
2813 defs.extend(pool_defs);
2814 pool_entries.push(pool_entry);
2815 slots.extend(pool_slots);
2816 } else {
2817 return Err(syn::Error::new_spanned(
2818 &p.ident,
2819 "a `pool` requires enabling embassy-supervisor's `pool` feature",
2820 ));
2821 }
2822 }
2823 }
2824 }
2825
2826 let m = slots.len();
2827 // Every graph index (dep entries, `topo_sort_const`'s queue/order) is a `u8`, so
2828 // more than 256 slots would silently truncate (`i as u8`) and corrupt the order.
2829 // 256 slots means max index 255 and max per-node dep count 255 — both fit exactly.
2830 if m > 256 {
2831 return Err(syn::Error::new(
2832 proc_macro2::Span::call_site(),
2833 format!(
2834 "supervisor_graph!: {m} node slots declared, but at most 256 are supported \
2835 (including pool members) — graph indices are `u8`"
2836 ),
2837 ));
2838 }
2839 let (all_entries, deps_entries) = slot_tables(&slots, &names)?;
2840
2841 // `Graph.pools` is `#[cfg(feature = "pool")]`; emit that field iff this macro was
2842 // built with pool support (forwarded from the supervisor's `pool` feature).
2843 let pools_field = if cfg!(feature = "pool") {
2844 quote!( pools: &[ #(#pool_entries),* ], )
2845 } else {
2846 quote!()
2847 };
2848
2849 // embassy-executor's trace hooks (declared `unsafe extern "Rust"` in the
2850 // executor), defined once here at the graph declaration site — the supervisor
2851 // crate is `forbid(unsafe_code)` and cannot carry `#[unsafe(no_mangle)]` items.
2852 // They forward to the supervisor's `trace` recorders. `task_new` and
2853 // `task_ready_begin` carry nothing the recorders need (the id→node mapping
2854 // comes from the spawn glue above), so they are no-ops. Exactly one definition
2855 // of each may exist per binary: enable `trace-hooks` OR write your own set.
2856 // Requires an edition-2024 consumer (`#[unsafe(no_mangle)]` syntax).
2857 // Named graphs never emit the hook symbols: `no_mangle` items exist once
2858 // per binary, and a multi-graph binary's PRIMARY (unnamed) graph carries
2859 // them; the recorders resolve every registered graph's nodes regardless.
2860 let trace_hooks = if cfg!(feature = "trace-hooks") && graph.name.is_none() {
2861 quote! {
2862 #[unsafe(no_mangle)]
2863 fn _embassy_trace_poll_start(executor_id: u32) {
2864 #cr::trace::on_poll_start(executor_id);
2865 }
2866 #[unsafe(no_mangle)]
2867 fn _embassy_trace_task_new(_executor_id: u32, _task_id: u32) {}
2868 #[unsafe(no_mangle)]
2869 fn _embassy_trace_task_end(executor_id: u32, task_id: u32) {
2870 #cr::trace::on_task_end(executor_id, task_id);
2871 }
2872 #[unsafe(no_mangle)]
2873 fn _embassy_trace_task_exec_begin(executor_id: u32, task_id: u32) {
2874 #cr::trace::on_task_exec_begin(executor_id, task_id);
2875 }
2876 #[unsafe(no_mangle)]
2877 fn _embassy_trace_task_exec_end(executor_id: u32, task_id: u32) {
2878 #cr::trace::on_task_exec_end(executor_id, task_id);
2879 }
2880 #[unsafe(no_mangle)]
2881 fn _embassy_trace_task_ready_begin(_executor_id: u32, _task_id: u32) {}
2882 #[unsafe(no_mangle)]
2883 fn _embassy_trace_executor_idle(executor_id: u32) {
2884 #cr::trace::on_executor_idle(executor_id);
2885 }
2886 }
2887 } else {
2888 quote!()
2889 };
2890
2891 // `name: X;` renames the emitted graph static and suffixes the private
2892 // backing tables, so several graphs coexist — even in one module. Unnamed
2893 // keeps the historical `GRAPH`/`NODES`/`DEPS` idents.
2894 let graph_ident = graph
2895 .name
2896 .clone()
2897 .unwrap_or_else(|| Ident::new("GRAPH", proc_macro2::Span::call_site()));
2898 let (nodes_ident, deps_ident) = match &graph.name {
2899 Some(n) => (
2900 format_ident!("__SV_NODES_{}", n),
2901 format_ident!("__SV_DEPS_{}", n),
2902 ),
2903 None => (
2904 Ident::new("NODES", proc_macro2::Span::call_site()),
2905 Ident::new("DEPS", proc_macro2::Span::call_site()),
2906 ),
2907 };
2908 Ok(quote! {
2909 #(#defs)*
2910
2911 // Private backing tables — the application uses the graph static. The
2912 // topological order and pools are inlined into its literal below; the
2913 // node count is `.nodes.len()`.
2914 static #nodes_ident: [::core::option::Option<&'static #cr::TaskNode>; #m] = [ #(#all_entries),* ];
2915 const #deps_ident: [&'static [u8]; #m] = [ #(#deps_entries),* ];
2916
2917 /// The compile-time task graph — node slots, dependency table, topological order,
2918 /// and (with the `pool` feature) the elastic pools. Pass to `Supervisor::new`.
2919 pub static #graph_ident: #cr::Graph<#m> = #cr::Graph {
2920 nodes: &#nodes_ident,
2921 deps: &#deps_ident,
2922 order: #cr::topo_sort_const(&#deps_ident),
2923 #pools_field
2924 };
2925
2926 #trace_hooks
2927 })
2928}
2929
2930/// Declare a supervised task graph; see the crate docs for the surface syntax.
2931#[proc_macro]
2932pub fn supervisor_graph(input: TokenStream) -> TokenStream {
2933 let graph = syn::parse_macro_input!(input as GraphSpec);
2934 expand(graph)
2935 .unwrap_or_else(syn::Error::into_compile_error)
2936 .into()
2937}
2938
2939/// Declare a **graph fragment**: `supervisor_fragment! { name: NET_FRAG; <items> }`
2940/// emits a `#[macro_export] macro_rules! NET_FRAG` relay that forwards the items
2941/// (verbatim, wrapped in `@fragment`/`@endfragment` attribution markers) into the
2942/// single `supervisor_graph!` expansion a `compose_graph!` call site assembles —
2943/// so every whole-graph compile-time pass (name map, u8 slot indices, topo order,
2944/// shared-slot dedup, the 256 cap) still sees ALL items, across crates.
2945///
2946/// Item syntax is validated here, with fragment-site spans; dep/executor NAMES
2947/// resolve at the compose site (cross-fragment references are the point).
2948/// Fragment authors reference their own workers/types via `$crate::…` (which
2949/// hygienically resolves to the fragment's crate at every compose site) or a
2950/// fully-qualified `::crate_name::…` path; a bare `crate::…` would resolve at
2951/// the COMPOSE crate and is a bug. No `$` other than `$crate` is permitted.
2952/// `#[cfg(...)]` inside a fragment is evaluated against the COMPOSE crate's
2953/// features (the tokens expand there) — export differently-named fragment
2954/// variants instead of feature-gating items.
2955#[proc_macro]
2956pub fn supervisor_fragment(input: TokenStream) -> TokenStream {
2957 fragment_expand(input.into())
2958 .unwrap_or_else(syn::Error::into_compile_error)
2959 .into()
2960}
2961
2962fn fragment_expand(input: TokenStream2) -> SynResult<TokenStream2> {
2963 struct FragmentSpec {
2964 name: Ident,
2965 items: TokenStream2,
2966 }
2967 impl Parse for FragmentSpec {
2968 fn parse(input: ParseStream) -> SynResult<Self> {
2969 input.parse::<kw::name>()?;
2970 input.parse::<Token![:]>()?;
2971 let name: Ident = input.parse()?;
2972 input.parse::<Token![;]>()?;
2973 let items: TokenStream2 = input.parse()?;
2974 Ok(FragmentSpec { name, items })
2975 }
2976 }
2977 let spec: FragmentSpec = syn::parse2(input)?;
2978 let name = &spec.name;
2979
2980 // Only `$crate` may appear (it resolves to the fragment's own crate in the
2981 // emitted macro_rules RHS); any other `$` would be interpreted as a
2982 // metavariable by the relay and mangle the forwarded tokens.
2983 validate_dollars(spec.items.clone())?;
2984
2985 // Syntax validation with fragment-site spans: parse the items as a graph,
2986 // with `$crate` substituted by a placeholder ident so paths parse. Name
2987 // RESOLUTION (deps, executors) is deliberately skipped — targets may live
2988 // in other fragments and resolve at the compose site.
2989 let substituted = substitute_dollar_crate(spec.items.clone());
2990 syn::parse2::<GraphSpec>(substituted)?;
2991
2992 let items = &spec.items;
2993 let dollar = proc_macro2::Punct::new('$', proc_macro2::Spacing::Alone);
2994 let doc = format!(
2995 "A `supervisor_fragment!` relay (generated). Use from a compose site:\n\
2996 `embassy_supervisor::compose_graph! {{ fragments: [{name}], graph: {{ .. }} }}`\n\
2997 Not for direct invocation."
2998 );
2999 Ok(quote! {
3000 #[doc = #doc]
3001 #[macro_export]
3002 macro_rules! #name {
3003 (@emit #dollar cb:path, [#dollar(#dollar rest:tt)*], {#dollar(#dollar acc:tt)*}, {#dollar(#dollar g:tt)*}) => {
3004 #dollar cb! { @next [#dollar(#dollar rest)*],
3005 {#dollar(#dollar acc)* @fragment #name; #items @endfragment;},
3006 {#dollar(#dollar g)*} }
3007 };
3008 }
3009 })
3010}
3011
3012/// Reject any `$` not immediately followed by `crate`, recursively through
3013/// groups. `$crate` is the one dollar token with meaning in the emitted
3014/// macro_rules RHS (fragment-crate paths); anything else would be read as a
3015/// metavariable.
3016fn validate_dollars(stream: TokenStream2) -> SynResult<()> {
3017 use proc_macro2::TokenTree;
3018 let mut iter = stream.into_iter().peekable();
3019 while let Some(tt) = iter.next() {
3020 match tt {
3021 TokenTree::Group(g) => validate_dollars(g.stream())?,
3022 TokenTree::Punct(p) if p.as_char() == '$' => match iter.peek() {
3023 Some(TokenTree::Ident(i)) if i == "crate" => {}
3024 _ => {
3025 return Err(syn::Error::new(
3026 p.span(),
3027 "only `$crate` is permitted in a fragment — any other `$` \
3028 would be read as a metavariable by the relay macro",
3029 ));
3030 }
3031 },
3032 _ => {}
3033 }
3034 }
3035 Ok(())
3036}
3037
3038/// Replace every `$crate` pair with a placeholder ident so the items parse as a
3039/// `GraphSpec` for validation. The ORIGINAL tokens (with `$crate` intact) are
3040/// what get forwarded.
3041fn substitute_dollar_crate(stream: TokenStream2) -> TokenStream2 {
3042 use proc_macro2::{TokenStream as TS, TokenTree};
3043 let mut out = TS::new();
3044 let mut iter = stream.into_iter().peekable();
3045 while let Some(tt) = iter.next() {
3046 match tt {
3047 TokenTree::Group(g) => {
3048 let inner = substitute_dollar_crate(g.stream());
3049 let mut ng = proc_macro2::Group::new(g.delimiter(), inner);
3050 ng.set_span(g.span());
3051 out.extend([TokenTree::Group(ng)]);
3052 }
3053 TokenTree::Punct(p) if p.as_char() == '$' => {
3054 if let Some(TokenTree::Ident(i)) = iter.peek()
3055 && i == "crate"
3056 {
3057 let span = iter.next().map(|t| t.span()).unwrap_or_else(|| p.span());
3058 out.extend([TokenTree::Ident(Ident::new("__sv_fragment_crate", span))]);
3059 } else {
3060 out.extend([TokenTree::Punct(p)]);
3061 }
3062 }
3063 other => out.extend([other]),
3064 }
3065 }
3066 out
3067}
3068
3069#[cfg(test)]
3070mod tests {
3071 use super::*;
3072
3073 /// The `ready` dep marker's feature rejection. Not UI-testable: the
3074 /// dev-dependency supervisor carries `readiness` for the trybuild pass
3075 /// cases, and the scratch project's feature unification re-enables this
3076 /// crate's feature through the supervisor's weak forward — so the
3077 /// no-feature path can only be exercised here, on the parser directly.
3078 #[test]
3079 fn ready_marker_requires_feature() {
3080 let res = syn::parse_str::<GraphSpec>(
3081 "node NET = Terminate, deps: [];\n\
3082 node HTTP = Terminate, deps: [NET ready];",
3083 );
3084 if cfg!(feature = "readiness") {
3085 assert!(res.is_ok(), "marker accepted with the feature");
3086 } else {
3087 match res {
3088 Ok(_) => panic!("marker accepted without the feature"),
3089 Err(err) => assert!(
3090 err.to_string().contains("requires the `readiness` feature"),
3091 "unexpected error: {err}"
3092 ),
3093 }
3094 }
3095 }
3096
3097 /// A stray ident after a dep name is rejected as an unknown marker in both
3098 /// feature states.
3099 #[test]
3100 fn unknown_dep_marker_rejected() {
3101 match syn::parse_str::<GraphSpec>("node A = Terminate, deps: [B rdy];") {
3102 Ok(_) => panic!("unknown marker accepted"),
3103 Err(err) => assert!(err.to_string().contains("`ready` marker"), "got: {err}"),
3104 }
3105 }
3106}