mstar.communication.communicator

Module Attributes

EXPECTED_MSTAR_RUST_VERSION

The mstar_rust extension version this tree expects (the vendored rust/ crate's version).

Functions

make_communicator(*args, **kwargs)

Construct the process's communicator, selecting the transport.

Classes

BaseCommunicator()
CommProtocol(*values)
ZMQCommunicator(my_id, push_ids[, protocol, ...])

class mstar.communication.communicator.BaseCommunicator

Bases: ABC

abstractmethod get_all_new_messages()

Return type:

list

abstractmethod send(entity_id, msg)

entity_id: worker_xyz, conductor, or api_server

Parameters:

entity_id (str)

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

Bases: Enum

IPC = 'IPC'

RDMA = 'RDMA'

SHM = 'SHM'

TCP = 'TCP'

mstar.communication.communicator.EXPECTED_MSTAR_RUST_VERSION = '0.1.0'

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.ZMQCommunicator(my_id, push_ids, protocol=CommProtocol.IPC, ipc_socket_path_prefix='/tmp/mstar/')

Bases: BaseCommunicator

Parameters:

• my_id (str)

• push_ids (list[str])

• protocol (CommProtocol)

• ipc_socket_path_prefix (str)

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)

Parameters:

event (EventWakeup)

send(entity_id, msg)

entity_id: worker_xyz, conductor, or api_server

Parameters:

entity_id (str)

wait_for_work(timeout_ms=50)

mstar.communication.communicator.make_communicator(*args, **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 pickle frames), so the flag can be set per-process — one entity at a time — while the rest of the mesh stays on pyzmq.

Return type:

BaseCommunicator