Recurrent Nodes & Pipelines

Some computations are loops: a training run repeats an epoch, a simulation repeats a time step, and each iteration reads what the previous one wrote. RecurrentNode expresses that by unrolling the loop into one real node per iteration, each writing its own dataset.

When an iteration is a whole sub-pipeline rather than one node, RecurrentPipeline does the same with one Pipeline per iteration.

pub fn pipeline<'a>(cat: &'a Catalog, params: &'a Params) -> impl Steps<pondrs::error::PondError> + 'a {
    (
        RecurrentNode {
            name: "train",
            state: &cat.checkpoints,
            init: &cat.init_weights,
            input: |prev, _cur| (prev, &params.target, &params.learning_rate),
            output: |cur| (cur,),
            func: step,
        }
        .build(),
        Node {
            name: "report",
            input: (cat.checkpoints.last().expect("at least one iteration"),),
            output: (&cat.report,),
            func: |w: String| (json!({ "final_weights": w.trim().parse::<f64>().unwrap() }),),
        },
    )
}

build() turns the RecurrentNode into an Unrolled group of plain Nodes named 0, 1, … inside a group named train, so their paths are train/0, train/1, … Because each iteration is an ordinary node, everything that works on nodes works on each iteration individually:

• Resume mid-chain: run --from-nodes train/47 runs iterations 47 onward, reading iteration 46’s saved output.
• Checkpoint-aware caching: CacheHook keys each iteration separately, so an unchanged prefix of the chain is skipped.
• Per-iteration progress: node hooks (LoggingHook, VizHook, your own) fire once per iteration.

The state vector is the chain

state holds one element per iteration, so state.len() is the iteration count. Iteration k is handed state[k - 1] as prev and state[k] as cur; iteration 0 gets init as its prev. The input and output projections turn those into the node’s input and output tuples.

An element can be a bare dataset — then the projections are close to the identity, as above — or a sub-catalog carrying per-iteration extras:

#[derive(Serialize, Deserialize)]
struct EpochSlot {
    weights: JsonDataset,
    metrics: JsonDataset,
    lr: Param<f64>,           // a learning-rate schedule, as data
}

RecurrentNode {
    name: "train",
    state: &cat.epochs,       // Vec<EpochSlot>
    init: &cat.init_epoch,    // EpochSlot
    input: |prev, cur| (&prev.weights, &cat.features, &cur.lr),
    output: |cur| (&cur.weights, &cur.metrics),
    func: train_epoch,
}
.build()

With a sub-catalog, init is a whole EpochSlot of which only weights is read, so check reports init_epoch.lr and init_epoch.metrics as UnusedCatalogEntry. That warning is harmless here.

Write the fields in the order shown — state and init before the projections, func last. As with Node, fields are type-checked in written order, and this order keeps a signature mistake reported at the closure. func is cloned once per iteration, so it must be Clone; a plain fn item or a non-capturing closure is.

Where the slots live

Two nodes may not write the same dataset, so every iteration needs its own. A Vec<S> catalog field provides that: each element has its own address, and the catalog indexer names them checkpoints.0, checkpoints.1, … in logs, check and viz.

Listing 100 elements in YAML by hand would be tedious, so pondrs::datasets::expand_indexed accepts a count and a template instead:

#[derive(Serialize, Deserialize)]
pub struct Catalog {
    /// The starting point, read by iteration 0.
    pub init_weights: TextDataset,
    /// One checkpoint per iteration: `checkpoints.0`, `checkpoints.1`, …
    #[serde(deserialize_with = "pondrs::datasets::expand_indexed")]
    pub checkpoints: Vec<TextDataset>,
    pub report: JsonDataset,
}

#[derive(Serialize, Deserialize)]
pub struct Params {
    pub target: Param<f64>,
    pub learning_rate: Param<f64>,
}

checkpoints:
  count: 5
  template:
    path: "ckpt/epoch_{i}.txt"

Element k is the template with {i} replaced by k in every string value; an optional placeholder: key renames {i} when one expanded vector is nested in another’s template. A plain YAML sequence is accepted too.

The catalog is then the single source of truth for the iteration count, and it can be overridden from the command line like any other catalog value:

cargo run --example recurrent_app -- run --catalog checkpoints.count=20

Group inputs and outputs

Unrolled works out its own input and output contract instead of having you declare it. Its inputs are the datasets some iteration reads and no iteration writes (init and any loop-invariant inputs), and its outputs are the datasets some iteration writes and no iteration reads (the final state). So check validates it like any Pipeline, and the parallel runner fires after_pipeline_run once the last iteration is done.

Recurrent pipelines

When one iteration is several nodes rather than one — a forward and a backward pass, a simulation step followed by a measurement — use RecurrentPipeline. It has the same state, init, input and output fields, and a pipe closure in place of func that builds the iteration’s steps from (prev, cur):

#[derive(Serialize, Deserialize)]
struct EpochSlot {
    acts: MemoryDataset<Vec<f64>>,   // intermediate, internal to the iteration
    weights: JsonDataset,
}

RecurrentPipeline {
    name: "train",
    state: &cat.epochs,       // Vec<EpochSlot>
    init: &cat.init_epoch,
    input: |prev, _cur| (&prev.weights, &cat.features),
    output: |cur| (&cur.weights,),
    pipe: |prev, cur| (
        Node { name: "fwd", input: (&prev.weights, &cat.features), output: (&cur.acts,), func: forward },
        Node { name: "bwd", input: (&prev.weights, &cur.acts), output: (&cur.weights,), func: backward },
    ),
}
.build()

build() unrolls it into an Unrolled group of one Pipeline per iteration, named 0, 1, … So the nodes keep their plain names, and their paths tell the iterations apart: train/0/fwd, train/0/bwd, train/1/fwd, … — run --from-nodes train/47/fwd resumes mid-chain, and run --nodes fwd selects every iteration’s forward pass.

Here input and output are each iteration’s declared contract, the same input/output a hand-written Pipeline has, and check holds every iteration to it: a declared output its nodes do not produce, a declared input they do not read, or an external dataset they read without declaring it is an error. pipe builds the steps from (prev, cur) rather than from that contract because of intermediates like cur.acts: each iteration needs its own, so they live in the state element, but they are not part of what the iteration consumes or produces.

The Unrolled group derives its own contract from the iterations’ declared ones, as described above, so acts never shows up as an input or output of the group.