tractor/docs/guide/streaming.rst

Cross-process streaming

Spawning processes is the boring half of tractor: the real cool stuff is the native support for cross-process streaming. Yes, you saw it here first — 2-way msg streams with reliable, transitive setup/teardown semantics, wired straight into the runtime's structured concurrency (SC) supervision machinery so that how a stream ends is part of the protocol.

No broker, no topic exchange, no IDL compiler. The IPC layer is a deliberately "cheap or nasty" (un)protocol: a tiny set of msgspec-typed msgs over a transport (TCP or UDS today) with payload typing opt-in per dialog — handshake msgs get the nasty treatment (strict validation) while high-rate stream payloads stay cheap (receiver-side checks only). See /guide/context for the typed pld_spec contract bits.

Two ways to stream

It's a trio.abc.Channel

tractor.MsgStream implements trio.abc.Channel — send(), receive(), async-iteration, aclose() — so trio-generic channel code drives an IPC stream unchanged.

• Bidirectional, context-based: open a tractor.Context to a peer task then enter ctx.open_stream() for a full-duplex tractor.MsgStream. This is the modern core API, taught end-to-end in /guide/context; we won't re-teach it here.
• One-way, portal-based: point tractor.Portal.open_stream_from at a plain async generator fn in the peer actor. Legacy, but perfectly fine for simple produce/consume pipelines — and it powers the classic examples below.

Rule of thumb: if the consumer ever needs to talk back — acks, control msgs, a final result — use a context. If it's a pure pipeline stage, either works and the one-way form is less typing.

One-way streaming from an async generator

The OG api. Write an async generator in the target actor's module; iterate its yields from the spawning side:

../../examples/asynchronous_generators.py

Each yield crosses the process boundary as one msg and feeds the parent's async for. When the consumer breaks out and exits the open_stream_from() block the far-end generator task is cancelled for you: the producer's lifetime is coupled to the consumer's scope so a one-way stream can never leak a remote task.

Any extra kwargs (stream_data, seed=100 style) are forwarded to the remote generator's call, and a non-async-gen target is rejected up front with a TypeError.

Note

No decorator required — any plain async-gen fn works. You may still meet @tractor.stream in the wild; it's the legacy marker for one-way endpoints and sticks around only for compat (heads up: the param name ctx is reserved for @context endpoints nowadays, so legacy fns should call theirs stream). New code wanting anything fancier than a one-way pipe should use tractor.context + ctx.open_stream().

Warning

One-way means one way: there's no sending to the generator side and no graceful consumer-to-producer stop msg — the teardown above is cancel-based. Needing upstream control flow is the sign you've outgrown this API.

A full-fledged streaming service

Now let's get fancy: compose one-way streams through a nested actor tree and you've got yourself a fan-in pipeline.

../../examples/full_fledged_streaming_service.py

What's going on?

• the root actor spawns 'aggregator' which opens its own actor nursery and spawns 'streamer_1' + 'streamer_2': 4 processes total, supervision nested two levels deep with zero special casing.
• aggregate() opens a one-way stream from each streamer and fans both into a single trio.open_memory_channel via one local trio task per portal — in-actor fan-in riding trio's built-in backpressure end-to-end.
• duplicates get dropped via a set and the deduped sequence is re-yielded upward: aggregate() is itself an async gen being consumed over IPC by the root. Streams compose.
• when the seed runs out the streamer gens finish, the memory channel drains closed, the aggregator's gen returns and the root's async for ends; await an.cancel() then reaps the subtree. Every exit is awaited — if you can produce a zombie process from this, it is a bug.

Watch the tree breathe while it runs, using the README's signature process-monitor incantation:

$TERM -e watch -n 0.1  "pstree -a $$" \
    & python examples/full_fledged_streaming_service.py \
    && kill $!

No extra threads, no fancy semaphores, no futures; all we need is tractor's IPC.

Two streams, one portal

Every open_stream_from() call starts its own remote task — even through the same portal — so two local consumer tasks can independently stream the same generator fn concurrently, both dialogs multiplexed over the single underlying IPC channel:

../../examples/multiple_streams_one_portal.py

The add-else-remove trick on the shared consumed list is the proof: each value arrives in both streams, getting appended by whichever task sees it first and removed by the other, so the list always ends up empty. Two streams, same data, zero interference.

This works because every dialog is keyed by its own context id (Context.cid) — any number of concurrent streams, contexts and one-shot RPCs share a single underlying tractor.Channel per peer pair.

Fan-out inside an actor: MsgStream.subscribe()

The inverse pattern: one IPC stream feeding many local tasks. Instead of paying for N redundant cross-process streams, call tractor.MsgStream.subscribe to get a BroadcastReceiver — a tokio-style broadcast channel from tractor.trionics — which copies every received value to each subscribed task:

../../examples/streaming_broadcast_fanout.py

Each task entering stream.subscribe() receives its own copy of everything sent from that point on. The underlying stream keeps pace with the fastest subscriber; a task falling more than the buffered window behind has its next receive raise tractor.trionics.Lagged to say it lost data.

The broadcast handle stays duplex btw: it proxies send() through to the underlying stream, so each subscriber task can keep talking upstream while consuming its fan-out copy.

Warning

.subscribe() is idempotent and non-reversible: the first call permanently swaps the stream's receive machinery over to the internally allocated broadcaster. There's no un-subscribing back to the raw stream, so make sure you're ok with the (theoretical) overhead before opting in.

Consuming: async for and friends

async for msg in stream: is just sugar over repeated await stream.receive(). The receive-side surface:

• receive() — next msg, or raises trio.EndOfChannel on a graceful far-end close (async for translates that into a clean loop exit for you).
• receive_nowait() — opportunistic, non-blocking drain.
• closed — property flagging an already-ended stream.

Send-side it's just await stream.send(data) — one Yield msg per call carrying any msgspec-encodable payload (or whatever your pld_spec permits, see /guide/context).

End-of-stream: close vs. cancel

How a stream ends is part of the protocol; the runtime keeps the polite case and the violent case distinct:

• graceful close: the far side exits its stream block, its async gen returns, or it calls await stream.aclose(). A Stop msg is sent so your async for simply ends (StopAsyncIteration, via trio.EndOfChannel under the hood). A normal, non-error ending — the dialog's result phase proceeds as usual.
• cancel or error: no Stop is sent. Instead the cancel/error itself is relayed so the far end knows the dialog did not end on purpose and raises accordingly — a tractor.ContextCancelled, a boxed tractor.RemoteActorError, etc. See the cancellation section of /guide/context for exactly who raises what.

Tying it together: every MsgStream is one-shot use. Both endings are final — once closed a stream can't be re-opened and the supported "retry" is opening a fresh tractor.Context (they're cheap).

• /guide/context — the full Context lifecycle: the handshake, results, cancellation semantics and the overrun/backpressure knobs.
• tractor.MsgStream and tractor.Portal.open_stream_from API docs.
• The zguide chapters our wire philosophy is named after: "cheap or nasty" and (un)protocols.