Pipeline

API reference for the streaming pipeline internals. For the stage builder methods (ctx.source, ctx.pipe, ctx.tap, ctx.pace, ctx.unbatch, ctx.window, ctx.stream in sink mode) see the Context reference. For the concept-level walkthrough, see Streaming Pipelines.

Streaming pipeline

Stages chain with the | operator to form a data-flow graph that runs across GPU ranks. A Stage is a stream transducer with 0..N inputs; a list of streams as the head is fan-in ([a, b] | stage). See the unified Stage DSL.

Reactive control is not on the pipeline — it is on the doc: cheap config is an inline doc.desired.* read, structural change is doc.bind(construct, [deps]) (useMemo). The old pipeline.bind(doc, mapping) field-mapping still works but is discouraged/legacy.

await pipeline.run()

await pipeline.run(*, poll_interval: float = 0.1, **config)

The canonical way to drive a composed pipeline. It starts the pipeline (the stage threads do the work), holds it alive while the session transport is attached, and stops it cleanly on disconnect or when the awaiting task is cancelled. This replaces any hand-rolled while ctx.connected: ... pipeline.update() poll.

run() does not reconcile control itself — reactive control lives on the doc (inline doc.desired.* reads + doc.bind). poll_interval only bounds how quickly a disconnect is observed; it is a sleep granularity, not a reconcile poll. Any **config is forwarded to start as the initial desired-state.

doc = ctx.doc("control")
pipeline = (
    ctx.source(make_frames(models, doc))
    | doc.bind(lambda d: ctx.pipe(make_vae(models, d)), [doc.desired.vae.model])
    | ctx.pace(fps=24) | ctx.stream("video", codec="h264")
)
with ctx.self_test():
    _drive_one_and_assert_shape(pipeline)
await pipeline.run()        # keep-alive until disconnect

pipeline.drive()

pipeline.drive(frames: int = 1) -> list

Drives frames representative items through the composed pipeline synchronously and returns the produced outputs. This is the driver inside ctx.self_test: the self-test block calls pipeline.drive(frames=N) and asserts the result's shape/dtype — the app drives and asserts its own representative pass. The same call runs under the CPU harness (mocked primitives), where reaching the end of the self-test block without raising means the test passed.

with ctx.self_test():
    out = pipeline.drive(frames=1)
    _assert_frames_shaped(out)       # shape/dtype only — no pixel/value read

PipelineItem

from urun.core.streaming.gpu_pipeline import PipelineItem, PipelineMetadata

Wraps a payload with control metadata that survives stage hops and rank boundaries.

@dataclass(frozen=True)
class PipelineItem:
    payload: Any
    metadata: PipelineMetadata

@dataclass(frozen=True)
class PipelineMetadata:
    prompt_generation: int = 0
    flush_generation: int = 0
    cutover: bool = False
    drop_generation: int = 0
    pts: int = 0
    idle: bool = False
    capture_ts: float | None = None
    input_seq: int = 0
    input_at_ms: int = 0
    timeline: tuple[tuple[str, float], ...] = ()

Mouse-to-photon timing

Input-driven pipelines get an always-on, per-hop timing surface with zero app wiring: ctx.input lifts each sample with its correlation fields (input_seq / input_at_ms) and an input.recv hop; ctx.window carries the newest tick's correlation onto each emitted window; and every stage stamps <stage>.in / <stage>.out into timeline. The egress sink folds the timeline into one structured [timing:<stream>] log line per conditioned window (per input_seq, never per frame), exposing per-hop queue-wait vs compute from the user's input to the frame leaving the runtime. Apps can also observe it programmatically via the sink's on_timing hook. Pipelines steered by doc/prompt (no continuous input) still report per cutover, so steer→egress timing comes free there too.

Factory methods:

Most stages handle PipelineItem transparently — they unwrap the payload, process it, and re-wrap the result with the same metadata. You only need PipelineItem directly when implementing custom cutover logic in a source stage.

Rank boundaries

When consecutive stages run on different ranks, urun automatically inserts shared-memory IPC transport:

pipeline = (
    ctx.source(denoise, device="cuda", rank=0)  # rank 0
    | ctx.pipe(vae_decode, device="cuda", rank=1)  # rank 1 — IPC inserted here
    | ctx.pace(fps=24, rank=1)
    | video
)

The IPC transport uses CUDA IPC handles for zero-copy GPU tensor transfer between ranks on the same node. CPU tensors use shared-memory ring buffers.

Pipeline lifecycle

pipeline.update(prompt="new prompt")
pipeline.stop()