mstar.streaming.stream_buffer

Functions

collate(items)

A chunk's items as its consumer takes them: one item as itself, several stacked.

Classes

StreamBuffer(request_id, edge_name, ...[, ...])	Per-request, per-edge buffer on the CONSUMER worker.
StreamChunk(data, chunk_index[, ...])	A chunk of data popped from a StreamBuffer.
StreamChunkInfo(start_offset, context_items, ...)	A chunk's place in its stream, without its data.
StreamingEdge(name, next_node, uuids, chunk)	A popped chunk as a synthetic input edge into its consumer node: the uuid of each of its items, and its place in the stream.

class mstar.streaming.stream_buffer.StreamBuffer(request_id, edge_name, from_partition, policy, join_dim=None, joiner=None, walk_tagged=False, _waiting_graph_edges=<factory>, ingested_chunk=None, _buffer=<factory>, _walk_runs=<factory>, _tensor_ids_in_order=<factory>, _id_to_tensor=<factory>, _consumed=0, _chunks_popped=0, _delivered_end=0, producer_done=False, _final_chunk_emitted=False, maybe_ready=True)

Bases: object

Per-request, per-edge buffer on the CONSUMER worker.

Items arrive one-by-one via normal routing: tensors, or handles to them (the worker puts each item’s uuid and collates the tensors when the consuming batch is built). The buffer accumulates them and applies a ChunkPolicy to decide when the consuming node has enough data to proceed.

For sliding-window policies the buffer keeps old items so that pop_chunk can return the full window while only advancing by stride.

Parameters:

• request_id (int)

• edge_name (str)

• from_partition (str)

• policy (ChunkPolicy)

• join_dim (int | None)

• joiner (Callable[[list], object] | None)

• walk_tagged (bool)

• _waiting_graph_edges (deque[StreamingEdge])

• ingested_chunk (StreamChunkInfo | None)

• _buffer (list)

• _walk_runs (deque[list])

• _tensor_ids_in_order (deque)

• _id_to_tensor (dict)

• _consumed (int)

• _chunks_popped (int)

• _delivered_end (int)

• producer_done (bool)

• _final_chunk_emitted (bool)

• maybe_ready (bool)

edge_name: str

from_partition: str

has_chunk_ready(consumer_walk=None)

Parameters:

consumer_walk (str | None)

Return type:

bool

ingested_chunk: StreamChunkInfo | None = None

join_dim: int | None = None

joiner: Callable[[list], object] | None = None

maybe_ready: bool = True

peek_walk()

The walk the next chunk runs under, without popping it. None when untagged or nothing is buffered.

Return type:

str | None

policy: ChunkPolicy

pop_chunk()

Pop the next chunk. Only call when has_chunk_ready() is True.

For sliding-window: returns window_size items, advances by stride items, discards items that have fallen out of the window. start_offset is the global position of the first item in the chunk.

Return type:

StreamChunk

pop_waiting_edge()

Return type:

StreamingEdge | None

pre_read_register(tensor_id, walk=None)

Parameters:

• tensor_id (str)

• walk (str | None)

pre_read_register_joined(tensor_ids, walk=None)

Register tensor_ids as the pieces of ONE item.

A chunked producer emits a piece per chunk; without this each would be an item of its own and a policy counting items would hand the consumer only the first. Each piece still counts towards the registered total, which is in the producer’s units (see below).

Parameters:

• tensor_ids (list[str])

• walk (str | None)

Return type:

None

prime_context(context_items)

The first context_items items the producer sends are context only (e.g. a voice clone’s reference frames): the first chunk carries them ahead of its new items, reported as context_items.

Parameters:

context_items (int)

Return type:

None

producer_done: bool = False

put(tensor_id, item)

Called when an item arrives via normal routing.

Parameters:

tensor_id (str)

Return type:

None

request_id: int

signal_done(num_items=0)

Producer signals no more items will arrive after the num_items it sent in total.

Parameters:

num_items (int)

Return type:

None

store_uningested_edge(streaming_edge)

Parameters:

streaming_edge (StreamingEdge)

walk_tagged: bool = False

class mstar.streaming.stream_buffer.StreamChunk(data, chunk_index, start_offset=0, is_final=False, context_items=0, num_items=0, released=<factory>)

Bases: object

A chunk of data popped from a StreamBuffer.

Parameters:

• data (dict[str, Tensor | None])

• chunk_index (int)

• start_offset (int)

• is_final (bool)

• context_items (int)

• num_items (int)

• released (list)

chunk_index: int

context_items: int = 0

data: dict[str, Tensor | None]

property info: StreamChunkInfo

is_final: bool = False

num_items: int = 0

released: list

start_offset: int = 0

class mstar.streaming.stream_buffer.StreamChunkInfo(start_offset, context_items, num_items, is_final)

Bases: NamedTuple

A chunk’s place in its stream, without its data.

Parameters:

• start_offset (int)

• context_items (int)

• num_items (int)

• is_final (bool)

context_items: int

Alias for field number 1

is_final: bool

Alias for field number 3

num_items: int

Alias for field number 2

start_offset: int

Alias for field number 0

class mstar.streaming.stream_buffer.StreamingEdge(name, next_node, uuids, chunk)

Bases: NamedTuple

A popped chunk as a synthetic input edge into its consumer node: the uuid of each of its items, and its place in the stream.

Parameters:

• name (str)

• next_node (str)

• uuids (list[int])

• chunk (StreamChunkInfo)

chunk: StreamChunkInfo

Alias for field number 3

name: str

Alias for field number 0

next_node: str

Alias for field number 1

uuids: list[int]

Alias for field number 2

mstar.streaming.stream_buffer.collate(items)

A chunk’s items as its consumer takes them: one item as itself, several stacked. A consumer must not write into it: a single item is the stored tensor, which other destinations may read too.

Parameters:

items (list[Tensor])

Return type:

Tensor