Streams¶

Name	What it does
`SharedStream`	a named ring of typed items in shared memory, sent by one process and received by up to `max_readers` readers
`SharedSendStream`	the one sender of a stream, with blocking, non-blocking and asyncio sends
`SendSlot`	a slot reserved by `writing`: its item, whose arrays are filled in place, and `send`
`SharedReceiveStream`	a reader of a stream, in mode lossless, lossy or latest
`ReceiveEvents`	the psygnal signal group of a reader: `received` for each item and `ended` once
`SharedStreamStatistics`	a stream's state at one moment: what was sent, what the ring holds and who reads it
`ReceiveStreamStatistics`	one open reader of a stream: its mode, position, lag and process

SharedStream ¶

Bases: Generic[T]

A ring of items in shared memory, sent by one process and received by up to max_readers readers.

Create it with create in one process, and open it with attach in others or pass it to them, as a process argument for instance: a pickled stream attaches again, and refuses a stream created since under the same name with SchemaMismatchError. Every item has the type given there: any type a SharedBox field takes, except a reference to a box.

create `classmethod` ¶

create(item_type: type[T], name: str | None = None, *, capacity: int, max_readers: int = 8) -> SharedStream[T]

create(item_type: object, name: str | None = None, *, capacity: int, max_readers: int = 8) -> SharedStream[Any]

create(item_type: object, name: str | None = None, *, capacity: int, max_readers: int = 8) -> SharedStream[Any]

Create the stream name holding capacity items of item_type.

Parameters:

Name	Type	Description	Default
`name`	`str \| None`	Without one, the stream gets a new name of 16 hex digits. A process the stream is passed to opens it without being told the name, so only code that calls `attach` needs to know it.	`None`
`capacity`	`int`	Items the ring holds, at least 2. A lossless reader may fall this many items behind before `send` waits for it.	required
`max_readers`	`int`	Readers that may be open at once, 1 to 4095.	`8`

Raises:

Type	Description
`SegmentExistsError`	If the name is taken.
`TypeError`	If a box field could not hold `item_type`.
`ValueError`	If the name breaks the naming rules, or `capacity` or `max_readers` is out of range.

attach `classmethod` ¶

attach(item_type: type[T], name: str) -> SharedStream[T]

attach(item_type: object, name: str) -> SharedStream[Any]

attach(item_type: object, name: str) -> SharedStream[Any]

Open the stream name, whose items must have the type item_type.

Raises:

Type	Description
`SegmentNotFoundError`	If no stream has that name, or it does not become one within 1 s.
`KindMismatchError`	If the name holds a segment of another kind, such as a box.
`SchemaMismatchError`	If the stream holds items of another type.

unlink `staticmethod` ¶

unlink(name: str) -> None

Remove the name name on Linux, as SharedBox.unlink does; a no-op on Windows.

sender ¶

sender() -> SharedSendStream[T]

Return the stream's sender for this process.

Raises:

Type	Description
`StreamBusyError`	If another live end holds the sender.
`EndOfStream`	If the stream has ended: its sender closed, or a reader found the sender's process dead. There is no second sender after that.

Notes

A sender whose process died is replaced, unless a reader already found it dead, which ended the stream.

The stream keeps its sender until it is closed, so dropping the returned object does not end the stream, and another sender() call in this process raises StreamBusyError until then.

reader ¶

reader(mode: Mode = 'lossless', start: Start = 'newest') -> SharedReceiveStream[T]

Open a reader.

Parameters:

Name	Type	Description	Default
`mode`	`Mode`	`"lossless"` receives every item and holds the sender back when a full ring behind; `"lossy"` skips the items the sender overwrote; `"latest"` always receives the newest item.	`'lossless'`
`start`	`Start`	`"newest"` begins at the newest item sent, `"oldest"` at the oldest the ring still holds.	`'newest'`

Raises:

Type	Description
`ValueError`	If `mode` or `start` is none of the values above.
`WaiterSlotsFullError`	If `max_readers` readers are open.

statistics ¶

statistics() -> SharedStreamStatistics

Return what was sent, what the ring holds, and each open reader's state.

close ¶

close() -> None

Close every sender and reader made from this object, then let go of the stream.

Closing the sender ends the stream for its readers.

Other processes keep the stream. Using a closed stream raises StreamClosedError.

SharedSendStream ¶

SharedSendStream(stream: SharedStream[Any], native: Sender)

Bases: End, Generic[T]

The sender of a stream: one per stream, in the process that holds it.

SharedStream.sender returns it. The stream keeps it open until close(), or until the stream is closed or the process dies; dropping the object does not close it. Closing it ends the stream: readers receive what is buffered, then EndOfStream.

stream `property` ¶

stream: SharedStream[T]

The stream this sender sends to.

send ¶

send(item: T, timeout: float | None = None) -> None

Send item, waiting while a lossless reader is a full ring behind.

The item is encoded straight into its slot once the slot is free, so an item that cannot be encoded raises only after that wait.

Raises:

Type	Description
`TimeoutError`	If `timeout` seconds pass first; the item is not sent.
`StreamClosedError`	If this sender is closed, before or during the wait.
`RuntimeError`	If called inside a block of this sender's `writing`.

Notes

An interrupt such as Ctrl-C during the call may leave the item sent although the call raised.

asend `async` ¶

asend(item: T) -> None

Send item, as send does, without blocking the event loop.

A cancelled asend may still have sent its item, if a reader made room just as it was cancelled.

Raises:

Type	Description
`StreamClosedError`	If this sender is closed, before or during the wait.
`RuntimeError`	If called inside a block of this sender's `writing`.

send_nowait ¶

send_nowait(item: T) -> None

Send item, or raise WouldBlock if that needs a wait.

It would wait while a lossless reader is a full ring behind, or while another thread or task holds the slot from writing or awriting.

Raises:

Type	Description
`RuntimeError`	If called inside a block of this sender's `writing`.

writing ¶

writing(template: T | None = None, timeout: float | None = None) -> Generator[SendSlot[T], None, None]

Reserve the next slot and yield it, to fill the item's arrays in place and send it.

slot.item is an item built from template, with each array it reaches through records and tuples replaced by a writable array over the slot. The template's own arrays are never read, so they may be empty. The block fills the slot's arrays and sends the item with slot.send, which does not copy them:

blank = Letter(0, "", np.empty((0, 0), np.uint8))
with sender.writing(blank) as slot:
    scanner.read_into(slot.item.scan)
    slot.send(replace(slot.item, index=n))

An item that is an array needs no template. Leaving the block without send, or by an exception, gives the slot up: nothing is published, and the next send takes its position. Until the slot is sent or the block ends, readers see nothing new, and a send or writing on this sender from another thread or task waits. That includes a task started inside the block by asyncio.wait_for or create_task.

Raises:

Type	Description
`TypeError`	If `template` is `None` and the item is not an array, or `template` is not of the item type.
`ValueError`	If a value of `template` outside its arrays does not fit.
`TimeoutError`	If `timeout` seconds pass while a lossless reader is a full ring behind, or while another thread or task holds the slot from `writing` or `awriting`.
`StreamClosedError`	If this sender is closed, before or during the wait.
`RuntimeError`	If called inside a block of this sender's `writing`.

Notes

An interrupt such as Ctrl-C just before the block starts may leave the slot reserved with no block to give it up, so later sends on this sender wait until they time out. Closing the sender ends them.

awriting `async` ¶

awriting(template: T | None = None) -> AsyncGenerator[SendSlot[T], None]

Reserve the next slot, as writing does, without blocking the event loop.

A cancelled awriting reserves nothing.

Raises:

Type	Description
`TypeError`	If `template` is `None` and the item is not an array, or `template` is not of the item type.
`StreamClosedError`	If this sender is closed, before or during the wait.
`RuntimeError`	If called inside a block of this sender's `writing`.

close ¶

close() -> None

Close the sender and end the stream. Calling it again does nothing.

SendSlot ¶

SendSlot(sender: Sender, template: _T | None)

Bases: Slot[T]

A slot of a stream that SharedSendStream.writing reserved for one item.

writing and awriting make it. Its item holds arrays over the slot, which the block fills, and send publishes an item.

item `property` ¶

item: _T

The item to fill: the template's members, with arrays over the slot.

Each array the item reaches through records and tuples is a writable array of its member's declared type, dtype and shape, which views the slot in shared memory and starts with the bytes of an older item. The arrays keep the stream's mapping alive, so using them after the block or after the stream is closed cannot crash, but writing to them after send changes an item readers may have received and tells no one.

send ¶

send(item: _T) -> None

Encode item into the slot and publish it.

An array of item that already is its place in the slot, such as one of item's arrays, is not copied; any other array is copied into the slot.

Raises:

Type	Description
`RuntimeError`	If this slot was sent or given up already.
`StreamClosedError`	If the sender was closed after the slot was reserved; nothing is published.
`TypeError`	If `item` does not have the stream's item type; the slot is given up.
`ValueError`	If a value in `item` does not fit; the slot is given up.

abandon ¶

abandon() -> None

Give the slot up unless it was sent: nothing is published, and the next send takes its position.

Leaving the block of writing does this.

SharedReceiveStream ¶

SharedReceiveStream(stream: SharedStream[Any], native: Reader)

Bases: End, Iterator[T], AsyncIterator[T], Generic[T]

A reader of a stream, in the process that opened it.

SharedStream.reader returns it. for item in reader and async for item in reader receive until the stream ends.

Notes

Receives made with receive and the like must not overlap receives made with receive_future or async for on the same reader: an item that an awaiting task gave up is kept and returned by the next receive, which may then come after a later item.

stream `property` ¶

stream: SharedStream[T]

The stream this reader receives from.

position `property` ¶

position: int | None

The position of the last item received, None before the first.

ReceiveStreamStatistics.position is the position of the item received next, so right after a receive it is one more than this.

missed `property` ¶

missed: int

Items a lossy or latest reader skipped; always 0 for a lossless reader.

events `property` ¶

events: ReceiveEvents

The reader's psygnal signals: received as (item, position) for each item, and ended once per delivery.

Connecting the first received callback starts delivery on the reader's background thread, and disconnecting the last one stops it. While delivery runs, every other way of receiving from this reader raises RuntimeError, also after the stream has ended, until the callbacks disconnect: one reader cannot hand an item to two consumers, so open a second reader for that. Callbacks run on the background thread; connect with thread="main" and call psygnal.emit_queued() to run them on the main thread. A callback that raises is logged to the sharedbox logger and delivery goes on, unless the receive itself fails: that is logged too, and callbacks then receive nothing more until a received callback connects or disconnects. Delivery runs only while received has a callback, and ended fires only then. A slow callback makes a lossy or latest reader miss items and holds the sender back for a lossless one. Right after delivery stops, the _nowait forms may raise WouldBlock until the delivering call has returned.

Notes

Starting and stopping delivery relies on psygnal calling the private methods SignalInstance._append_slot and _remove_slot.

receive ¶

receive(timeout: float | None = None) -> T

Return the next item, with new arrays.

Raises:

Type	Description
`EndOfStream`	If the sender closed or died and every item it sent was received.
`TimeoutError`	If `timeout` seconds pass first.
`StreamClosedError`	If this reader is closed, before or during the wait.

Notes

A blocking receive first waits for calls already on the reader's background thread, such as an outstanding receive_future, and returns the item after theirs; that wait counts against timeout. An interrupt such as Ctrl-C during the call may lose the item being received.

receive_into ¶

receive_into(out: Any, /, *, timeout: float | None = None) -> T

Return the next item, with its arrays read into the arrays in out.

out has the item's shape: an array for an item that is an array, a tuple with an entry per member for a tuple item, and a mapping from member name to entry for a record item. An entry is an array, None for a member decoded as usual, or, for a member that is a record or a tuple, an out of its own shape. A name left out of a mapping is decoded as usual. The item returned holds your arrays, filled in place. Allocate the arrays once and pass them to every call:

scan = np.empty((32, 32), np.uint8)
letter = reader.receive_into({"scan": scan})

An item that a cancelled receive_future or anext had already received is returned by the next receive of any kind, and holds the arrays of the call that received it, not out.

A member that is a list, set, frozenset, dict or tuple[T, ...] takes None, since an array cannot be a collection element, and an item that is a collection has no array to read into. Every check runs before anything is received, so a refused call takes no item.

Raises:

Type	Description
`TypeError`	If `out` does not have the item's shape, names a member the item does not have, gives an array for a member that is not an array, or holds an array that cannot be written; or if the item holds no array at all.
`ValueError`	If an array has another dtype or shape than its member.

Notes

A blocking receive first waits for calls already on the reader's background thread, such as an outstanding receive_future, and returns the item after theirs; that wait counts against timeout. After an exception the contents of out are unspecified. out must not be written to or resized from another thread during the call. An interrupt such as Ctrl-C during the call may lose the item being received.

receive_nowait ¶

receive_nowait() -> T

Return the next item, or raise WouldBlock if none is waiting.

It also raises WouldBlock while a call on the reader's background thread is still running, such as a cancelled anext or a delivery stopping after the last events.received callback disconnected, so an item that call hands back is received first.

receive_into_nowait ¶

receive_into_nowait(out: Any) -> T

As receive_into, raising WouldBlock if no item is waiting.

iter_into ¶

iter_into(out: Any) -> IterInto[T]

Return an iterator of items read into the arrays in out, as receive_into takes it, for for.

close ¶

close() -> None

Close this reader and drop items kept from cancelled receives.

receive_future ¶

receive_future(out: Any = None) -> Future[T]

Return a future of the next item, received by this reader's background thread.

The future is a concurrent.futures.Future, so concurrent.futures.wait and asyncio.wrap_future take it. Its exception is EndOfStream once the stream has ended, and StreamClosedError if the reader closes first. out, when given, is what receive_into takes.

Cancel the future before it starts to drop it. A running call cannot be cancelled: cancel() returns False, as for any Future, and the call goes on. An item it takes after its future was abandoned, or after an asyncio.wrap_future around it was cancelled, is lost. To wait on several readers, keep one outstanding future per reader and submit a new one only for the readers whose future completed.

ReceiveEvents ¶

ReceiveEvents(reader: SharedReceiveStream[Any] | None = None)

Bases: SignalGroup

The psygnal signal group of a stream reader.

It is the type of SharedReceiveStream.events.

Create the signal group of a stream reader.

Parameters:

Name	Type	Description	Default
`reader`	`SharedReceiveStream[Any] \| None`	The reader whose delivery starts and stops with the `received` callbacks; None for a group tied to no reader.	`None`

received `class-attribute` `instance-attribute` ¶

received = Signal(object, int, signal_instance_class=CountedSignal)

Emitted as (item, position) for each item the reader receives.

ended `class-attribute` `instance-attribute` ¶

ended = Signal()

Emitted once per delivery, when it reaches the end of the stream.

SharedStreamStatistics `dataclass` ¶

SharedStreamStatistics(capacity: int, sent: int, buffered: int, ended: bool, sender_pid: int, readers: tuple[ReceiveStreamStatistics, ...])

A stream's state at one moment: what was sent, what the ring holds, and who reads it.

buffered `instance-attribute` ¶

buffered: int

Items the ring holds: the last capacity sent, or fewer.

sender_pid `instance-attribute` ¶

sender_pid: int

The process holding the sender, 0 when none does.

ReceiveStreamStatistics `dataclass` ¶

ReceiveStreamStatistics(mode: Mode, position: int, lag: int, pid: int)

One open reader of a stream, as SharedStream.statistics found it.

position `instance-attribute` ¶

position: int

The position of the item the reader receives next, at most sent.

Right after a receive it is one more than SharedReceiveStream.position, the position of the item received, except for the moment before the sender counts that item in sent.

lag `instance-attribute` ¶

lag: int

Items sent that the reader has not received or skipped yet.

Streams¶

SharedStream ¶

create classmethod ¶

attach classmethod ¶

unlink staticmethod ¶

sender ¶

reader ¶

statistics ¶

close ¶

SharedSendStream ¶

stream property ¶

send ¶

asend async ¶

send_nowait ¶

writing ¶

awriting async ¶

close ¶

SendSlot ¶

item property ¶

send ¶

abandon ¶

SharedReceiveStream ¶

stream property ¶

position property ¶

missed property ¶

events property ¶

receive ¶

receive_into ¶

receive_nowait ¶

receive_into_nowait ¶

iter_into ¶

close ¶

receive_future ¶

ReceiveEvents ¶

received class-attribute instance-attribute ¶

ended class-attribute instance-attribute ¶

SharedStreamStatistics dataclass ¶

buffered instance-attribute ¶

sender_pid instance-attribute ¶

ReceiveStreamStatistics dataclass ¶

position instance-attribute ¶

lag instance-attribute ¶

create `classmethod` ¶

attach `classmethod` ¶

unlink `staticmethod` ¶

stream `property` ¶

asend `async` ¶

awriting `async` ¶

item `property` ¶

stream `property` ¶

position `property` ¶

missed `property` ¶

events `property` ¶

received `class-attribute` `instance-attribute` ¶

ended `class-attribute` `instance-attribute` ¶

SharedStreamStatistics `dataclass` ¶

buffered `instance-attribute` ¶

sender_pid `instance-attribute` ¶

ReceiveStreamStatistics `dataclass` ¶

position `instance-attribute` ¶

lag `instance-attribute` ¶