pipeline-dsl

The static front-end of the pipeline family: write ordinary functions, annotate them, and let the #[pipeline] macro derive a dependency DAG and a deterministically ordered compute().

Functions + names ⇒ DAG ⇒ deterministic compute. Edges are inferred from name-based parameter binding and mutability (&T reads, &mut T writes); the macro enforces single-writer, detects missing producers, topologically sorts, and emits crisp compile-time errors.

Setup

A single dependency — pipeline-dsl re-exports both the macros and the value layer:

[dependencies]
pipeline-dsl = "0.8"

(Optionally also depend on pipeline-core to refer to the value types under the shared pipeline:: name; they're the same re-exported types.)

Quick start

use pipeline_dsl::{pipeline, stage};

#[derive(Default)]
struct Db { count: i32 }
#[derive(Default)]
struct Cache { total: i32 }

#[pipeline(name = "App", context = "db, cache")]
mod app {
    use super::*;

    #[stage]
    pub fn tick(db: &mut Db) { db.count += 1; }

    #[stage]
    pub fn sum(cache: &mut Cache, db: &Db) {
        cache.total += db.count; // reads db (after tick), writes cache
    }
}

fn main() -> Result<(), pipeline::Error> {
    let mut app = App::new();
    let (mut db, mut cache) = (Db::default(), Cache::default());
    app.compute(&mut db, &mut cache)?; // stages run in dependency order
    Ok(())
}

Runtime value types (Vector, Value, Buckets, Reset) are re-exported by this crate, so you can name them as pipeline_dsl::Vector etc., and the code generated by #[pipeline] resolves its runtime support through pipeline-dsl — no second dependency required.

Attribute summary

• #[pipeline(name="…", args="…", context="…", external="…", error="…", controlflow_break="…", constructor="…", stats)] — declares the pipeline container; args/context/external describe inputs; the bare stats flag opts into per-stage stats (below).
• #[stage] — marks a function as a stage; #[stage(skip_when_clean)] opts it into demand-driven scheduling (below). Parameter attributes: #[rename = "field"], #[skip_reset], #[unused], #[state].

Parameter kinds

A stage parameter is one base kind (where its storage lives and who fills it):

…optionally tweaked by a modifier: #[rename] (bind to a different field name), #[skip_reset] (a slot that persists instead of resetting), #[unused] (keep a parameter that's deliberately disconnected from the graph).

Field visibility. Stage outputs and external inputs are pub (callers read results and feed inputs). args and #[state] are pipeline-internal, so they're private fields — reachable from the module that defines the pipeline (e.g. to seed state post-new()), but not by downstream crates.

#[state] is the only off-graph kind: per-stage-private, persistent, plain T the pipeline owns and exactly one stage mutates — for an accumulator, cache, or scratch that must survive across cycles. It is never reset, never read by another stage, and does not appear in dot() / html_diagram(). It requires &mut T; sharing one state across two stages is an error (use a Value<T> slot instead).

In the dynamic pipeline-graph front-end there is no #[state]: stages are closures, so a stage simply captures its own persistent mutable state (with a Cell/RefCell for interior mutability). #[state] is the static front-end's equivalent, since its stages are free fns that can't capture.

Constructors

When every field is Default (the usual case), the macro generates Pipeline::new(args…); if there are no args it also derives Default. Each Default-initialized field is bounded by Default, so a field whose type isn't Default produces a clear T: Default is not satisfied error rather than one buried in generated code. In that case, pass constructor = "manual" and write your own constructor:

#[pipeline(name = "P", constructor = "manual")]
mod p { /* … */ }

impl P {
    pub fn with_caps(n: usize) -> Self { /* build all fields yourself */ }
}

Demand-driven scheduling (opt-in)

By default every stage runs each compute(). Mark one #[stage(skip_when_clean)] and the generated compute() skips it in any cycle where none of its input slots changed (is_updated()) — a skipped stage doesn't run and doesn't write, so "unchanged" propagates to its readers (and a value→invalid transition, being dirty, wakes them). args and context don't count as dirty triggers; a stage whose only inputs are those is a compile error (it would never run). Same contract as the dynamic front-end: the body must be a pure function of its declared inputs.

#[stage(skip_when_clean)]
pub fn refine(cfg: &Config, src: &Vector<u32>, dst: &mut Vector<u32>) { /* recompute dst from src */ }

Add the stats flag to measure it — #[pipeline(name = "P", stats)] generates collect_stats(bool) (off by default), stats() -> &[StageStats] ({ name, ran, skipped, time }), reset_stats(), and stats_age(), so you can see how often each stage actually does work. Without stats, no stats fields or methods are generated (zero cost); skipping still works. See the runnable demand_driven example.

See the repository for the full guide (binding rules, multiple contexts, generics, diagrams, diagnostics).

Safety

This front-end is fully checked at compile time and uses no unsafe — mis-wiring (a missing producer, two writers, a cycle) is a compile error. The dynamic pipeline-graph front-end trades those compile-time guarantees for runtime flexibility, validating wiring at build() and using a small encapsulated unsafe core. If a fixed graph fits your problem, this crate gives you the stronger guarantees.

Related crates

Part of the pipeline family — a shared value layer with two front-ends:

Need the graph wired at runtime (stages/implementations chosen dynamically)? See pipeline-graph.

License

MIT OR Apache-2.0.