Pipelines#

The DAG carrier. Pipeline is the canonical input to every runner (Scheduler::new, Server::run_blockwise, SerialRunner::run); tasks are pure data with no fields that describe inter-task structure. This is the single biggest API change from daisy 1.x, where each Task carried its own upstream_tasks=[…] field and the DAG was reconstructed by walking those edges.

Why move the DAG off the Task#

In daisy 1.x:

a = daisy.Task(task_id="extract", ...)
b = daisy.Task(task_id="predict", upstream_tasks=[a], ...)
c = daisy.Task(task_id="relabel", upstream_tasks=[b], ...)
daisy.run_blockwise([a, b, c])

This works for simple chains but has structural problems:

  • A task’s identity depends on its position in a DAG. You can’t have a predict task that runs in two unrelated pipelines without re-instantiating it with different upstream_tasks=.

  • Reusing a sub-pipeline is awkward. If extract → predict → relabel is one logical pipeline, you can’t refer to “that pipeline” — only to the leaf task and its transitive parents.

  • Pure data is hard to reason about. A Task that holds references to other Tasks isn’t a value in the algebraic-data sense; it’s a graph node with stateful relationships.

In daisy v2, Task is pure data — no upstream references, no DAG knowledge — and a separate Pipeline value type owns the edges:

extract = daisy.Task(task_id="extract", ...)
predict = daisy.Task(task_id="predict", ...)
relabel = daisy.Task(task_id="relabel", ...)

pipeline = extract + predict + relabel
daisy.run_blockwise(pipeline)

extract, predict, and relabel are reusable across pipelines — you can build extract + predict in one place and extract + something_else in another, and extract itself is unchanged.

The data structure#

In Rust (daisy-core/src/pipeline.rs):

pub struct Pipeline {
    pub tasks: Vec<Arc<Task>>,
    graph: petgraph::DiGraph<(), ()>,   // NodeIndex(i) ↔ tasks[i]
}

The graph is a petgraph::DiGraph whose NodeIndex(i) matches the position of tasks[i]. Edges carry no weight; the structure is purely topological.

Three invariants enforced by Pipeline::new:

  1. Task ids are unique within the pipeline.

  2. Every edge endpoint is one of the listed tasks.

  3. The resulting graph is acyclic (petgraph::algo::is_cyclic_directed).

Sources (no incoming edges) and outputs (no outgoing edges) are derived on demand via graph.externals(Direction::Incoming/Outgoing) — no separate fields. The edges() method returns an iterator of (upstream_task_id, downstream_task_id) pairs derived from graph.edge_indices().

In Python (daisy-py/src/py_pipeline.rs), Pipeline is a #[pyclass] that holds Vec<Py<PyAny>> (the original Python Task references) plus an edge list of index pairs:

pub struct PyPipeline {
    pub tasks: Vec<Py<PyAny>>,
    pub edges: Vec<(usize, usize)>,
    pub sources: Vec<usize>,
    pub outputs: Vec<usize>,
}

tasks holds the Python objects directly (not the unwrapped Rust Tasks) so that subclasses of _rs.Task — most importantly the v1.x compat subclass daisy.v1_compat.Task — survive composition. The conversion to daisy_core::Pipeline happens once at run time via to_core(py), which calls PyTask::convert_task on each member.

Composition operators#

+ (sequential) and | (parallel) compose pipelines. Both are implemented in py_pipeline.rs; the operators on Task delegate to PyPipeline::from_task then call __add__ / __or__.

a + b — sequential#

Every output of a becomes a block-level upstream of every source of b. Tasks are deduplicated by Python identity:

(extract + predict) + relabel
= Pipeline(
    tasks=[extract, predict, relabel],
    edges=[(extract, predict), (predict, relabel)],
    sources=[extract],
    outputs=[relabel],
  )

Fan-out and fan-in fall out naturally:

a + (b | c) + d
# edges = [(a,b), (a,c), (b,d), (c,d)]
# Diamond: a fans out to b and c, both fan in to d.

a | b — parallel#

Union of two DAGs with no new edges. Sources and outputs are unioned:

(extract_left + predict) | (extract_right + predict)
# Two parallel chains. `predict` appears once if it's the same Python object,
# twice if they're separately constructed (Python identity).

Identity-based deduplication#

merge_tasks in py_pipeline.rs deduplicates by Py<PyAny>::as_ptr() — the raw Python object pointer. This means:

  • The same Python Task instance composed multiple times appears once in the merged pipeline.

  • Two Tasks with the same task_id but different Python identities are two distinct nodes and will collide at to_core() time (the underlying Rust Pipeline::new rejects duplicate task ids with "duplicate task_id in pipeline").

The second point is intentional: mistakenly building the same task twice is almost always a bug, and we’d rather hard-error at run time than silently merge.

Runtime entry points#

Every runner accepts Pipeline | Task (a singleton task is auto-promoted to a one-task pipeline):

daisy.run_blockwise(pipeline_or_task, ...)
daisy.run_blockwise(pipeline_or_task, multiprocessing=False)  # serial
daisy.Server().run_blockwise(pipeline_or_task)
daisy.Scheduler(pipeline_or_task)  # debug only

The coercion is _to_pipeline(x) in daisy/_task.py:

  • Pipeline → as-is.

  • Task → Pipeline.from_task(task) (singleton).

  • list | tuple of tasks → _build_pipeline_from_tasks(tasks). This is the v1.x bridge — see below.

At the FFI boundary, coerce_pipeline_or_task in daisy-py/src/py_server.rs does the same coercion on the Rust side, then PyPipeline::to_core produces the daisy_core::Pipeline that Scheduler::new consumes.

v1.x bridge: list-of-tasks calling convention#

Existing daisy 1.x code uses Task(upstream_tasks=[…]) and run_blockwise([a, b, c]). We bridge this without breaking compatibility:

  1. daisy.v1_compat.Task is a Python subclass of _rs.Task. Its __new__ accepts upstream_tasks=[…] and stores the list as a Python attribute (_v1_upstream_tasks) on the subclass instance. PyO3’s #[pyclass(subclass)] ensures Python subclass instances get their own __dict__, so the attribute lives on the subclass — the underlying Rust Task is untouched.

  2. run_blockwise([a, b, c]) routes through _to_pipeline → _build_pipeline_from_tasks. That function walks each task’s _v1_upstream_tasks attribute, transitively collects every reachable task, and builds a Pipeline by parallel-unioning each task and then sequential-edging each (upstream, downstream) pair:

seen = transitive_closure(tasks, via=_get_task_upstream)
pipe = Pipeline.from_task(seen[0])
for t in seen[1:]:
    pipe = pipe | Pipeline.from_task(t)
for up, down in edges_from_side_table:
    pipe = pipe | (Pipeline.from_task(up) + Pipeline.from_task(down))

The | (rather than +) for the pair-pipeline is critical — it avoids creating spurious edges from pipe’s current outputs to up. Each edge contributes exactly the (up, down) edge to the merged DAG without re-introducing any others.

  1. Each Task(upstream_tasks=…) call emits a DeprecationWarning pointing at v2 Pipeline composition. The compat path remains supported indefinitely.

The result is that daisy.v1_compat.run_blockwise([a, b, c]) produces exactly the same Pipeline you’d get from a + b + c in the v2-native API, but without forcing existing 1.x users to rewrite their scripts.

Reset and re-run#

Pipeline.reset() calls reset() on every member task. There’s no cascade or topological order — each task’s reset() clears that task’s done-marker independently. This matches user expectations: “reset the pipeline” means “clear everything”; users who want partial resets call task.reset() per task.

Why composition is in Rust#

Earlier drafts had the + / | operators in pure Python — Python is the natural place for “DSL ergonomics,” and the Rust side just consumed a flat task list with explicit edges. We moved composition into Rust because:

  • The edge-merging logic for (a + b) | (c + d) and friends gets non-trivial. Doing it in Rust keeps every composition O(tasks + edges) with predictable allocation behavior and lets the type system enforce the invariants (PyPipeline::new validates the edge list before construction).

  • The Python wrapper around a Rust Task previously had to mirror state (its own tasks list, edge list, source list, output list), and keeping those in sync across + / | chains was a class of bugs we saw during the v2 refactor.

  • Pipeline construction is on the run-blockwise hot path for very-large-DAG users; doing it in Rust shaves overhead.

The Python operators on Task (__add__, __or__) are PyO3 methods that promote to PyPipeline::from_task and then call into Rust. The full composition logic is merge_tasks + edge remapping in py_pipeline.rs.

Tests#

  • tests/test_pipeline.py — composition operator semantics: sequential, parallel, fan-in, fan-out, diamond, non-mutation of sub-pipelines, cycle rejection.

  • tests/test_topo_order.py — topological order over composed pipelines (alphabetical tiebreaker on the ready set).

  • tests/daisy_compat/test_scheduler.py and friends — the v1.x list-of-tasks path with Task(upstream_tasks=[…]).

Where to look#

  • daisy-core/src/pipeline.rs — the Rust value type, validation, from_task, sources, outputs.

  • daisy-py/src/py_pipeline.rs — PyPipeline, the + / | operators, identity-based dedup, to_core.

  • daisy-py/python/daisy/_task.py — _to_pipeline, _build_pipeline_from_tasks, the v1.x side-table walk.

  • daisy-py/python/daisy/v1_compat.py — the Task(upstream_tasks=…) subclass and the deprecation warnings.