mstar.communication.edge_items

mstar.communication.edge_items#

EDGE_ITEMS: a step’s streamed tensors to one peer worker, as columns.

The graph runtime sends a step’s plain streamed edges this way (frames::EdgeItems in Rust), so a receiver decodes, reads and routes them once per step instead of once per (request, worker) INPUT_SIGNALS. EdgeItemsBuilder is a Python encoder of the same frame, which tests build frames with and check Rust’s against. The reader side turns a frame back into rows, and a row back into what INPUT_SIGNALS carried, for the paths that still want that (a request not known yet, a transport without a batched read).

Functions

frame_dtypes(frame)

The frame's dtypes, by dtype_idx: once per frame, for row_infos.

row_infos(frame, row[, dtypes])

The row's tensor descriptors, as INPUT_SIGNALS carried them (dtypes from frame_dtypes, or computed here).

row_input_signals(frame, row)

The row as the INPUT_SIGNALS that carried it before EDGE_ITEMS.

rows(frame)

Classes

EdgeItemsBuilder(partition_name)

One frame's columns, row by row (frames::EdgeItems in Rust).

StreamRow(index, request_id, signal, ...)

One (request, streamed edge) row of a frame, its index-th: its items are frame.uuids[start:start + count] (and likewise every item column), and their dims and strides frame.dims_flat[dim_start:...].
class mstar.communication.edge_items.EdgeItemsBuilder(partition_name)[source]#

Bases: object

One frame’s columns, row by row (frames::EdgeItems in Rust).

Parameters:

partition_name (str)

build()[source]#
Return type:

EdgeItems

try_add(request_id, edge)[source]#

Add the edge’s row; False, adding nothing, when its descriptors don’t share the frame’s fields (or one walk).

Parameters:
Return type:

bool

class mstar.communication.edge_items.StreamRow(index, request_id, signal, next_node, start, count, dim_start)[source]#

Bases: NamedTuple

One (request, streamed edge) row of a frame, its index-th: its items are frame.uuids[start:start + count] (and likewise every item column), and their dims and strides frame.dims_flat[dim_start:...].

Parameters:
count: int#

Alias for field number 5

dim_start: int#

Alias for field number 6

index: int#

Alias for field number 0

next_node: str#

Alias for field number 3

request_id: str#

Alias for field number 1

signal: str#

Alias for field number 2

start: int#

Alias for field number 4

mstar.communication.edge_items.frame_dtypes(frame)[source]#

The frame’s dtypes, by dtype_idx: once per frame, for row_infos.

Parameters:

frame (EdgeItems)

Return type:

list

mstar.communication.edge_items.row_infos(frame, row, dtypes=None)[source]#

The row’s tensor descriptors, as INPUT_SIGNALS carried them (dtypes from frame_dtypes, or computed here).

Parameters:
Return type:

list[TensorPointerInfo]

mstar.communication.edge_items.row_input_signals(frame, row)[source]#

The row as the INPUT_SIGNALS that carried it before EDGE_ITEMS.

Parameters:
Return type:

InputSignals

mstar.communication.edge_items.rows(frame)[source]#
Parameters:

frame (EdgeItems)

Return type:

Iterator[StreamRow]