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
:doc:`/guide/context` for the typed ``pld_spec`` contract bits.

Two ways to stream
------------------

.. margin:: It's a ``trio.abc.Channel``

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

- **Bidirectional, context-based**: open a
  :class:`tractor.Context` to a peer task then enter
  ``ctx.open_stream()`` for a full-duplex
  :class:`tractor.MsgStream`. This is the modern core API, taught
  end-to-end in :doc:`/guide/context`; we won't re-teach it here.

- **One-way, portal-based**: point
  :meth:`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:

.. literalinclude:: ../../examples/asynchronous_generators.py
   :caption: examples/asynchronous_generators.py
   :language: python

Each ``yield`` crosses the process boundary as one msg and feeds
the parent's ``async for``. When the consumer ``break``\ s 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 :func:`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.

.. d2:: diagrams/streaming_pipeline.d2
   :caption: Four actors, three streams, one deduped feed.
   :alt: two streamer actors fan in to an aggregator then root

.. literalinclude:: ../../examples/full_fledged_streaming_service.py
   :caption: examples/full_fledged_streaming_service.py
   :language: python

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 :func:`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:

.. literalinclude:: ../../examples/multiple_streams_one_portal.py
   :caption: examples/multiple_streams_one_portal.py
   :language: python

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
:class:`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
:meth:`tractor.MsgStream.subscribe` to get a
``BroadcastReceiver`` — a tokio-style broadcast channel from
``tractor.trionics`` — which copies every received value to each
subscribed task:

.. literalinclude:: ../../examples/streaming_broadcast_fanout.py
   :caption: examples/streaming_broadcast_fanout.py
   :language: python

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 :exc:`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 :doc:`/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 :exc:`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
  :exc:`tractor.ContextCancelled`, a boxed
  :exc:`tractor.RemoteActorError`, etc. See the cancellation
  section of :doc:`/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 :class:`tractor.Context`
(they're cheap).

.. seealso::

   - :doc:`/guide/context` — the full ``Context`` lifecycle: the
     handshake, results, cancellation semantics and the
     overrun/backpressure knobs.

   - :class:`tractor.MsgStream` and
     :meth:`tractor.Portal.open_stream_from` API docs.

   - The zguide chapters our wire philosophy is named after:
     "`cheap or nasty`_" and `(un)protocol`_\ s.

.. _structured concurrency: https://en.wikipedia.org/wiki/Structured_concurrency
.. _cheap or nasty: https://zguide.zeromq.org/docs/chapter7/#The-Cheap-or-Nasty-Pattern
.. _(un)protocol: https://zguide.zeromq.org/docs/chapter7/#Unprotocols
.. _msgspec: https://jcristharif.com/msgspec/