How to send items through a stream¶

A stream carries a series of items from one process to others, such as the letters a sorting office passes on, and keeps the last few in shared memory so that a reader that falls behind can catch up. Use it when the items are a flow: a box holds one current value that every process reads and writes, while a stream holds items that each reader receives once.

Create the stream¶

The item type can be anything a box field takes, except a reference to a box. This guide sends letters, each a record with an index, a recipient and the scan of its envelope as an array:

import asyncio
import multiprocessing as mp
from dataclasses import dataclass, replace
from typing import Annotated

import numpy as np

from sharedbox import (
    Capacity,
    DType,
    EndOfStream,
    Shape,
    SharedReceiveStream,
    SharedSendStream,
    SharedStream,
)


@dataclass(frozen=True)
class Letter:
    index: int
    recipient: Annotated[str, Capacity(32)]
    scan: Annotated[np.ndarray, Shape(32, 32), DType("uint8")]

The creating process calls create with the item type, a name and a capacity, the number of items the ring holds:

stream = SharedStream.create(Letter, "example-sorting:letters", capacity=8)

A larger capacity lets a reader fall further behind before a lossless reader holds the sender back or a lossy one misses items, and costs memory for that many items. max_readers (8 unless you set it) limits how many readers may be open at once. A stream name follows the same rules as a box name.

Other processes reach the stream in one of two ways. A process you start can take the stream as an argument, and it arrives attached under the same name, as in the example below. Any other process calls attach with the same type and name. When only processes you start use the stream, leave the name out and create gives the stream a new one:

stream = SharedStream.create(Letter, capacity=8)

The sender and the readers cannot be passed to another process: pickling one raises TypeError, and a child made by fork inherits them closed. Pass the stream, and open each end in the process that uses it. Code that is given only an end reaches its stream through stream, to open another reader or read its statistics.

Send items¶

A stream has one sender. You get it from sender() and send with send. Closing the sender ends the stream, and with closes it at the end of the block:

if __name__ == "__main__":
    context = mp.get_context("spawn")
    ready = context.Event()
    with SharedStream.create(Letter, capacity=8) as stream:
        desk = context.Process(target=register, args=(stream, ready))
        desk.start()
        ready.wait(60)
        with stream.sender() as sender:
            for i in range(3):
                sender.send(Letter(i, "Ada", np.full((32, 32), i, np.uint8)))
        desk.join(60)
    SharedStream.unlink(stream.name)

A call to sender() while another live sender exists raises StreamBusyError. Once a sender has closed the stream, a later sender() raises EndOfStream.

Fill an item in place¶

When your code can write an item's arrays itself, such as a device library that fills a buffer it is given, it can write them straight into the stream and leave send no array bytes to copy. Each item lives in a slot of the stream's ring. writing reserves the next slot and yields a SendSlot, whose item is the template item you pass with each of its arrays replaced by an array over the slot. The template's arrays are never read, so an empty array will do. Fill the slot's arrays, then send the item with slot.send, which finds those arrays already in place and does not copy them:

def send_filled(sender: SharedSendStream[Letter], count: int) -> None:
    blank = Letter(0, "Ada", np.empty((0, 0), np.uint8))
    for i in range(count):
        with sender.writing(blank) as slot:
            slot.item.scan.fill(i)
            slot.send(replace(slot.item, index=i))

Readers see nothing of the slot until it is sent. Leaving the block without slot.send, or by an exception, publishes nothing and gives the position to the next send, so the block can also drop an item it finds bad. While the slot is reserved, a send from another thread or task waits for it. In asyncio code, use awriting with async with.

Write the item's arrays only inside the block

The arrays stay usable after the block, but they still view the slot. Writing to them after slot.send changes an item readers may already have received, and nothing tells them. A child process started by fork inherits the arrays as well, and writing to them there changes the parent's slot.

Receive items¶

A reader is opened with reader, and its mode decides what happens when it is slower than the sender:

"lossless" receives every item, and the sender waits for it when it falls capacity items behind. Use it for a consumer that must see everything, such as the registered-mail desk, which logs every letter.
"lossy" skips the items the sender has already overwritten, and missed counts them. Use it for a viewer that can drop items, such as a display board in the hall.
"latest" always receives the newest item. Use it for a display that shows only the current state.

A reader starts at the newest item already sent, or, with start="oldest", at the oldest the stream still holds. Reading with for receives until the stream ends:

for letter in stream.reader(mode="lossy"):
    print(letter.index)

A reader opened late misses earlier items

A reader starts at the newest item already sent, so of the items sent before it opens, the reader receives only the last. Open the reader first and tell the sender it can start, as the ready event does below, or use start="oldest" while the stream still holds what you need.

Read into arrays you already have¶

receive builds new arrays for every item. Between processes an item is always copied, so reading into the same arrays each time saves allocating a new one per item. Allocate them once and pass them to receive_into, or to iter_into for a for loop:

def register(stream: SharedStream[Letter], ready: "mp.synchronize.Event") -> None:
    reader = stream.reader(mode="lossless")
    ready.set()
    scan = np.empty((32, 32), np.uint8)
    for letter in reader.iter_into({"scan": scan}):
        print(letter.index, letter.recipient, int(letter.scan[0, 0]))

The reader sets ready once it is open, and the main process waits for it before sending, so no item is sent before the reader exists.

out has the shape of the item. For an item that is an array, out is the array. For a record, it is a mapping from member name to an array, to None for a member decoded as usual, or, for a record or tuple member, to an out of its own. For a tuple, it is a tuple with one entry per member. The returned item holds your arrays. A member that is a list, set or dict takes None, because an array cannot be an element of a collection, and an item that is itself a collection has no arrays to read into.

Run a callback for each item¶

reader.events is a ReceiveEvents group with two psygnal signals: received, emitted with (item, position) for each item, and ended, once when the stream ends:

reader = stream.reader()
reader.events.received.connect(lambda item, position: print(item.index))
reader.events.ended.connect(lambda: print("done"))

The first received callback you connect starts the reader's background thread, and disconnecting the last one stops it. The callbacks run on that thread. To run them on your main thread, connect with thread="main" and call psygnal.emit_queued() in your loop.

A reader that delivers to callbacks cannot also be read from, because one reader cannot hand an item to two consumers. Any receive on it raises RuntimeError until the callbacks disconnect, so open a second reader for the other consumer.

A slow callback loses items or holds the sender

Callbacks run one after another on the reader's thread. While one runs, a lossy or latest reader misses items, and a lossless reader holds the sender back. Do the slow part elsewhere, for example by putting the item on a queue.

Delivery starts on the first connect because sharedbox relies on psygnal calling SignalInstance._append_slot and _remove_slot and on its _lock, three private names that the test suite checks against the installed psygnal.

Wait, poll or give up¶

receive and send take a timeout in seconds and raise TimeoutError when it passes. The _nowait forms, receive_nowait, receive_into_nowait and send_nowait, raise WouldBlock when there is nothing to receive or no room to send. A KeyboardInterrupt that lands while a receive is returning may lose the item it was taking, as queue.Queue.get can.

Use a stream with asyncio¶

A reader works with async for and anext, and a sender has asend. Neither blocks the event loop:

async for letter in reader:
    print(letter.index)

await sender.asend(letter)

To give up after a while, put asyncio.timeout() around the receive:

async with asyncio.timeout(5):
    letter = await anext(reader)

A cancelled receive keeps its item for the next one

When a task is cancelled after the reader had already taken an item, the reader keeps that item and returns it from the next receive, so a lossless reader loses nothing. If two tasks share the reader, that item can come after a later one, and it holds the arrays of the call that took it, not the out of the next. Let one task or thread use a reader at a time.

A cancelled asend may still have sent its item, if room in the stream appeared just as it was cancelled, so sending it again can send it twice.

Both ends can be used with async with, which closes them at the end of the block. In an asyncio.TaskGroup, let the task that sends close the sender, because a reader's async for ends only when the stream does:

async def produce(sender: SharedSendStream[Letter]) -> None:
    async with sender:
        for i in range(3):
            await sender.asend(Letter(i, "Ada", np.full((32, 32), i, np.uint8)))


async def consume(reader: SharedReceiveStream[Letter]) -> None:
    async with reader:
        async for letter in reader:
            print(letter.index)


async def run_tasks(stream: SharedStream[Letter]) -> None:
    async with asyncio.TaskGroup() as group:
        group.create_task(consume(stream.reader()))
        group.create_task(produce(stream.sender()))

The reader is opened before either task starts, so it sees every item.

To wait for whichever of several readers has an item first, receive_future returns a concurrent.futures.Future, which asyncio.wrap_future turns into one the event loop can wait on. A future that has started receiving cannot be cancelled, and an item it takes after you stop waiting on it is lost. Keep one future per reader, and ask a reader for a new one only once its last one is done:

async def print_from_any(readers: list[SharedReceiveStream[Letter]]) -> None:
    pending = {asyncio.wrap_future(r.receive_future()): r for r in readers}
    while pending:
        done, _ = await asyncio.wait(pending, return_when=asyncio.FIRST_COMPLETED)
        for future in done:
            reader = pending.pop(future)
            try:
                letter = future.result()
            except EndOfStream:
                continue
            print(letter.index)
            pending[asyncio.wrap_future(reader.receive_future())] = reader

Close the stream and handle exits¶

Closing the sender ends the stream: readers receive what is buffered, then get EndOfStream. A for loop treats that as its end, and a direct receive raises it. Closing a reader, or the stream with close, raises StreamClosedError in a call still waiting on it. On Linux, SharedStream.unlink(stream.name) removes the name.

When a process dies, the others notice it during a wait. Readers end after a dead sender, and the sender stops waiting for a dead lossless reader. A reader that only polls with receive_nowait never waits, so it keeps raising WouldBlock and never sees the stream end; give it a timeout instead.

sharedbox registers an atexit hook in each process that stops its box watcher threads and its stream worker threads. atexit runs handlers last-registered first, so a handler you register before importing sharedbox runs after sharedbox's and finds the ends closed. Spawned child processes exit through sys.exit and run the hook. Children made by fork or forkserver exit with os._exit and do not, so their daemon threads end with the process, and the other processes free a reader ended that way after a short delay.

A child made by fork inherits the parent's ends, but they are closed in the child: using one raises StreamClosedError. Open new ends in the child from the inherited stream instead. Do not connect callbacks to an inherited reader's events either: if the parent was delivering to callbacks when it forked, the child can block on the lock the parent's delivery thread held.

The whole script

"""The script of the guide "How to send items through a stream"."""

import asyncio
import multiprocessing as mp
from dataclasses import dataclass, replace
from typing import Annotated

import numpy as np

from sharedbox import (
    Capacity,
    DType,
    EndOfStream,
    Shape,
    SharedReceiveStream,
    SharedSendStream,
    SharedStream,
)


@dataclass(frozen=True)
class Letter:
    index: int
    recipient: Annotated[str, Capacity(32)]
    scan: Annotated[np.ndarray, Shape(32, 32), DType("uint8")]




def register(stream: SharedStream[Letter], ready: "mp.synchronize.Event") -> None:
    reader = stream.reader(mode="lossless")
    ready.set()
    scan = np.empty((32, 32), np.uint8)
    for letter in reader.iter_into({"scan": scan}):
        print(letter.index, letter.recipient, int(letter.scan[0, 0]))




async def produce(sender: SharedSendStream[Letter]) -> None:
    async with sender:
        for i in range(3):
            await sender.asend(Letter(i, "Ada", np.full((32, 32), i, np.uint8)))


async def consume(reader: SharedReceiveStream[Letter]) -> None:
    async with reader:
        async for letter in reader:
            print(letter.index)


async def run_tasks(stream: SharedStream[Letter]) -> None:
    async with asyncio.TaskGroup() as group:
        group.create_task(consume(stream.reader()))
        group.create_task(produce(stream.sender()))




async def print_from_any(readers: list[SharedReceiveStream[Letter]]) -> None:
    pending = {asyncio.wrap_future(r.receive_future()): r for r in readers}
    while pending:
        done, _ = await asyncio.wait(pending, return_when=asyncio.FIRST_COMPLETED)
        for future in done:
            reader = pending.pop(future)
            try:
                letter = future.result()
            except EndOfStream:
                continue
            print(letter.index)
            pending[asyncio.wrap_future(reader.receive_future())] = reader




def send_filled(sender: SharedSendStream[Letter], count: int) -> None:
    blank = Letter(0, "Ada", np.empty((0, 0), np.uint8))
    for i in range(count):
        with sender.writing(blank) as slot:
            slot.item.scan.fill(i)
            slot.send(replace(slot.item, index=i))




if __name__ == "__main__":
    context = mp.get_context("spawn")
    ready = context.Event()
    with SharedStream.create(Letter, capacity=8) as stream:
        desk = context.Process(target=register, args=(stream, ready))
        desk.start()
        ready.wait(60)
        with stream.sender() as sender:
            for i in range(3):
                sender.send(Letter(i, "Ada", np.full((32, 32), i, np.uint8)))
        desk.join(60)
    SharedStream.unlink(stream.name)


if __name__ == "__main__":
    with SharedStream.create(Letter, "example-sorting:tasks", capacity=8) as stream:
        asyncio.run(run_tasks(stream))
    SharedStream.unlink("example-sorting:tasks")
    with SharedStream.create(Letter, "example-sorting:merged", capacity=8) as stream:
        readers = [stream.reader(), stream.reader()]
        with stream.sender() as sender:
            for i in range(2):
                sender.send(Letter(i, "Ada", np.full((32, 32), i, np.uint8)))
        asyncio.run(print_from_any(readers))
    SharedStream.unlink("example-sorting:merged")
    with SharedStream.create(Letter, "example-sorting:filled", capacity=8) as stream:
        reader = stream.reader()
        with stream.sender() as sender:
            send_filled(sender, 2)
        for letter in reader:
            print(letter.index, int(letter.scan[0, 0]))
    SharedStream.unlink("example-sorting:filled")