mstar.streaming.stream_buffer

mstar.streaming.stream_buffer#

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`(edge, chunk)	A synthetic streaming-input edge and the chunk it carries.

class mstar.streaming.stream_buffer.StreamBuffer(request_id, edge_name, from_partition, policy, 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)[source]#

Bases: object

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

Tensors arrive one-by-one via normal RDMA routing. 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)
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)

edge_name: str#

from_partition: str#

has_chunk_ready(consumer_walk=None)[source]#

Parameters:: consumer_walk (str | None)
Return type:: bool

ingested_chunk: StreamChunkInfo | None = None#

peek_walk()[source]#

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()[source]#

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()[source]#

Return type:: StreamingEdge | None

pre_read_register(tensor_id, walk=None)[source]#

Parameters:

tensor_id (str)
walk (str | None)

prime_context(context_items)[source]#

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)[source]#

Called when a tensor arrives via normal RDMA routing.

Parameters:

tensor_id (str)
item (Tensor)

Return type:

None

request_id: int#

signal_done(num_items=0)[source]#

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)[source]#

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)[source]#