`nemo_rl.data_plane.nixl.control`#

ZMQ control plane between clients and storage units (no Ray on the op path).

Same shape as TransferQueue’s SimpleStorage: each storage unit binds a ZMQ ROUTER on node_ip:<free port> and serves small control messages from one thread; clients talk to it over DEALER sockets lent by TQ’s socket pool. Ray is used only to start the units and to publish their addresses once (bootstrap).

Messages are msgpack [op, request_id, args]; replies [request_id, ok, result_or_error]. notify sends ops that need no reply (unpin, abort); the unit sends nothing back for those.

An error reply is {"kind", "msg", "reason", "blobs"} so the client can rebuild the exception the unit raised instead of reporting every failure the same way:

======== =================================== ========================== kind client exception meaning ======== =================================== ========================== UnitFull UnitFull put fails (SSD on: spills here) Busy DataPlaneBusy(reason) retry this unit later Missing LostBlobs(blobs) the unit no longer has them Protocol TransferError any other server exception ======== =================================== ==========================

Module Contents#

Classes#

`UnitServer`	ROUTER server thread for one storage unit.
`UnitClient`	Requests to storage units over DEALER sockets lent by TransferQueue’s pool.

Functions#

`error_reply`	The wire form of a server-side exception (see the module docstring).
`reply_error`	Rebuild the exception behind an error reply.
`node_ip`	This node’s routable IP (what Ray advertises), falling back to a UDP probe.
`_peer`	A unit’s `tcp://ip:port` as the `ZMQServerInfo` TQ’s pool dials.

Data#

`log`
`_enc`
`_dec`
`NO_REPLY`

API#

nemo_rl.data_plane.nixl.control.log#: ‘getLogger(…)’

nemo_rl.data_plane.nixl.control._enc#: ‘Encoder(…)’

nemo_rl.data_plane.nixl.control._dec#: ‘Decoder(…)’

nemo_rl.data_plane.nixl.control.NO_REPLY#: None

nemo_rl.data_plane.nixl.control.error_reply(e: BaseException) → dict[str, Any]#: The wire form of a server-side exception (see the module docstring).

nemo_rl.data_plane.nixl.control.reply_error(address: str, op: str, err: Any) → Exception#: Rebuild the exception behind an error reply.

nemo_rl.data_plane.nixl.control.node_ip() → str#: This node’s routable IP (what Ray advertises), falling back to a UDP probe.

class nemo_rl.data_plane.nixl.control.UnitServer( handlers: dict[str, Callable[..., Any]], lock: threading.Lock, )#

ROUTER server thread for one storage unit.

handlers maps op name -> callable(**args). They run on this thread under lock, which the unit also takes for its (rare) Ray-called methods.

Initialization

_serve() → None#

close() → None#

class nemo_rl.data_plane.nixl.control.UnitClient(timeout_s: float = 30.0)#

Requests to storage units over DEALER sockets lent by TransferQueue’s pool.

A socket goes back to the pool only after a complete reply with its request’s id; after anything else a late reply may still arrive, so it is closed.

Initialization

call(

address: str,

op: str,

*,

timeout_s: float | None = None,

**args: Any,

) → Any#

Request/reply.

Raises what the unit raised (see the module docstring), and TransferError when no reply comes within timeout_s (default: the client’s timeout; callers pass their remaining operation budget).

call_many( op: str, args_by_address: dict[str, dict[str, Any]], *, timeout_s: float | None = None, ) → dict[str, Any]#

call to every address at once: all requests go out before any reply is read.

Returns each address’s result, or the exception its call raised, so the caller decides what one failure means for the others. timeout_s bounds the whole batch, not each call.

_timeout_ms(timeout_s: float | None) → int#

static _send(s: zmq.Socket, op: str, args: dict[str, Any]) → str#

static _recv( s: zmq.Socket, address: str, op: str, rid: str, end: float, ) → Any#

notify(address: str, op: str, **args: Any) → None#: Fire-and-forget (no reply is sent for NO_REPLY ops).

close() → None#

nemo_rl.data_plane.nixl.control._peer(address: str) → transfer_queue.utils.zmq_utils.ZMQServerInfo#: A unit’s tcp://ip:port as the ZMQServerInfo TQ’s pool dials.

nemo_rl.data_plane.nixl.control#