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: RustZMQCommunicator, on the vendored rust/ extension (see communication/rust_communicator.py).

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.

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). Every process sends on the extension’s transport, so a mismatching install - e.g. a stale wheel after an upgrade - is refused at startup rather than left to carry the mesh.

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

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

Construct the process’s communicator: RustZMQCommunicator, on the vendored rust/ extension (see communication/rust_communicator.py).

Raises when the extension is missing or is not the version this tree expects. nonblocking_send is for a communicator OutboxCommunicator wraps, which needs try_send_bytes: an extension built before ZmqCommunicator.try_send is refused for it.

Parameters:

nonblocking_send (bool)

Return type:

BaseCommunicator