mstar.communication.communicator

Module Attributes

EXPECTED_MSTAR_RUST_VERSION	The mstar_rust extension version this tree expects (the vendored rust/ crate's version).
DEPLOYMENT_ANCHOR_ENTITY	the one every deployment has exactly one of.

Functions

make_communicator(*args[, nonblocking_send])	Construct the process's communicator, selecting the transport.
requested_transport()	MSTAR_RUST_ZMQ as set, upper-cased: 0, 1 or AUTO.

Classes

BaseCommunicator()
CommProtocol(*values)
OutboxCommunicator(inner)	A communicator whose sends never block: a send the peer can't take now (at its high-water mark) is queued, in order behind any earlier one to the same peer, and retried by flush.
ZMQCommunicator(my_id, push_ids[, protocol, ...])

class mstar.communication.communicator.BaseCommunicator

Bases: ABC

abstractmethod get_all_new_messages()

Return type:

list

abstractmethod poll_for_messages(timeout_ms=20)

Block until a message is readable, the registered event fires, or timeout_ms elapses, whichever comes first. True when a message is waiting for get_all_new_messages; a fired event is drained.

Parameters:

timeout_ms (int)

Return type:

bool

abstractmethod register_event_for_poll(event)

Have poll_for_messages (and wait_for_work) also end when event is signalled.

Parameters:

event (EventWakeup)

Return type:

None

abstractmethod send(entity_id, msg)

entity_id: worker_xyz, conductor, or api_server

Parameters:

entity_id (str)

try_send_bytes(entity_id, data)

Send a message already encoded with self.codec without blocking: False, with nothing sent, when the peer is at its high-water mark. A loop that must keep reading its own input while a peer is slow sends with this and retries later (see OutboxCommunicator).

Parameters:

• entity_id (str)

• data (bytes)

Return type:

bool

class mstar.communication.communicator.CommProtocol(*values)

Bases: Enum

IPC = 'IPC'

RDMA = 'RDMA'

SHM = 'SHM'

TCP = 'TCP'

mstar.communication.communicator.DEPLOYMENT_ANCHOR_ENTITY = 'conductor'

the one every deployment has exactly one of. Two deployments that agree on this endpoint ARE the same deployment as far as anything keyed off it is concerned – notably the tensor-file namespace in mstar.communication.tensors.

Type:

The entity whose endpoint stands for the whole deployment

mstar.communication.communicator.EXPECTED_MSTAR_RUST_VERSION = '0.1.1'

The mstar_rust extension version this tree expects (the vendored rust/ crate’s version). Under MSTAR_RUST_ZMQ=AUTO a mismatching install - e.g. a stale wheel after an upgrade - takes over the mesh silently, so the factory warns when the imported version differs.

class mstar.communication.communicator.OutboxCommunicator(inner)

Bases: object

A communicator whose sends never block: a send the peer can’t take now (at its high-water mark) is queued, in order behind any earlier one to the same peer, and retried by flush.

For a loop that peers block on while it sends to them (the API server’s preprocess thread, against the conductor and the workers): if it blocked on a full peer it would stop reading the input that peer waits to deliver, and the two would wait on each other for good. Everything else is the wrapped communicator’s.

Parameters:

inner (BaseCommunicator)

flush()

Send what peers can take now; True if anything is still queued.

A queued message whose send fails outright (not just a full peer) is dropped and logged: there is no caller left to raise to, and keeping it would stop every message behind it.

Return type:

bool

send(entity_id, msg)

Parameters:

entity_id (str)

Return type:

None

send_each(entity_ids, msg)

send to each of entity_ids, encoding msg once.

Parameters:

entity_ids (Iterable[str])

Return type:

None

class mstar.communication.communicator.ZMQCommunicator(my_id, push_ids, protocol=CommProtocol.IPC, ipc_socket_path_prefix='/tmp/mstar/', codec=None)

Bases: BaseCommunicator

Parameters:

• my_id (str)

• push_ids (list[str])

• protocol (CommProtocol)

• ipc_socket_path_prefix (str)

• codec (type[Codec] | None)

get_all_new_messages(blocking=False, timeout_s=None)

Return type:

list

poll_for_messages(timeout_ms=20)

Block until a message is readable, a registered wakeup event fires, or timeout_ms elapses — whichever comes first. True when a message is available (left queued for get_all_new_messages); a wakeup ends the poll early with False (the event is drained, exactly as in wait_for_work). Mirrors the Rust communicator’s method so call sites work against either transport.

push_sockets: dict[str, Socket[bytes]]

register_event_for_poll(event)

Have poll_for_messages (and wait_for_work) also end when event is signalled.

Parameters:

event (EventWakeup)

send(entity_id, msg)

entity_id: worker_xyz, conductor, or api_server

Parameters:

entity_id (str)

try_send_bytes(entity_id, data)

Send a message already encoded with self.codec without blocking: False, with nothing sent, when the peer is at its high-water mark. A loop that must keep reading its own input while a peer is slow sends with this and retries later (see OutboxCommunicator).

Parameters:

• entity_id (str)

• data (bytes)

Return type:

bool

wait_for_work(timeout_ms=50)

mstar.communication.communicator.make_communicator(*args, nonblocking_send=False, **kwargs)

Construct the process’s communicator, selecting the transport.

MSTAR_RUST_ZMQ selects it (see docs/environment_variables.rst):

• AUTO (default) — the Rust-backed RustZMQCommunicator (vendored rust/ extension; see communication/rust_communicator.py) when the extension imports successfully, pyzmq otherwise.

• 1 — the Rust communicator; raises if the extension is missing.

• 0 — always the pyzmq ZMQCommunicator.

The two are wire-compatible (same endpoints, same codec), so the flag can be set per-process — one entity at a time — while the rest of the mesh stays on pyzmq.

nonblocking_send is for a communicator OutboxCommunicator wraps, which needs try_send_bytes: an extension built before ZmqCommunicator.try_send falls back to pyzmq under AUTO and raises under 1. Every other caller keeps such a build.

Parameters:

nonblocking_send (bool)

Return type:

BaseCommunicator

mstar.communication.communicator.requested_transport()

MSTAR_RUST_ZMQ as set, upper-cased: 0, 1 or AUTO.

Return type:

str