`nemo_rl.data_plane.nixl.blobstore`#

UnitSlabStore: blob bytes on the storage units’ slabs.

The client lays values out as raw bytes in registered local memory and hands the store the segments and the owner unit; on read it hands the store slab ranges and registered destination addresses. Pins, refcounts and the medium are the units’ business: pinned-DRAM slabs on CPU StorageUnit actors, one-sided RDMA over UCX, with an optional SSD tier on each unit.

Module Contents#

Classes#

UnitSlabStore

Pinned-DRAM slabs on NixlStorageUnit actors, one-sided RDMA.

Functions#

`remaining`	`min(cap, deadline - now)`; raise :class:`DataPlaneTimeout` if spent.
`busy_wait`	Sleep one Busy backoff (bounded by `deadline`); return the next one.

Data#

`__all__`
`BUSY_RETRY_S`
`BUSY_RETRY_MAX_S`
`PENDING_POLL_S`

API#

nemo_rl.data_plane.nixl.blobstore.__all__#: [‘LostBlobs’, ‘TransferFailed’, ‘UnitSlabStore’]

nemo_rl.data_plane.nixl.blobstore.BUSY_RETRY_S#: 0.2

nemo_rl.data_plane.nixl.blobstore.BUSY_RETRY_MAX_S#: 2.0

nemo_rl.data_plane.nixl.blobstore.PENDING_POLL_S#: 0.01

exception nemo_rl.data_plane.nixl.blobstore.TransferFailed( blob_ids: list[str], reason: str, *, in_flight: bool = False, )#

Bases: Exception

A transfer or a control call failed, but the unit still has the blobs.

The caller may retry. in_flight: a NIXL transfer may still be running.

Initialization

Initialize self. See help(type(self)) for accurate signature.

nemo_rl.data_plane.nixl.blobstore.remaining(deadline: float | None, cap: float) → float#: min(cap, deadline - now); raise :class:DataPlaneTimeout if spent.

nemo_rl.data_plane.nixl.blobstore.busy_wait(deadline: float | None, backoff: float) → float#: Sleep one Busy backoff (bounded by deadline); return the next one.

class nemo_rl.data_plane.nixl.blobstore.UnitSlabStore( ep: nemo_rl.data_plane.nixl.nixl_io.NixlEndpoint, directory: Any, *, namespace: str, timeout_s: float, ssd: bool = False, lock: Any = None, )#

Pinned-DRAM slabs on NixlStorageUnit actors, one-sided RDMA.

No Ray on the op path: control messages go to the units over ZMQ (nemo_rl/data_plane/nixl/control.py), bytes over NIXL. The caller names the unit of every op (rows live on unit gidx % N).

put: ZMQ alloc + NIXL WRITE + ZMQ commit (publishes the rows).
get: one ZMQ resolve_keys per unit (pins) + one NIXL READ per unit.
clear: one awaited ZMQ release_rows per unit (row-wide).

Ray remains only off the op path: the unit list at start (and when a unit is unknown), and the liveness check after a failure.

Initialization

_refresh_units() → None#

_unit(unit_id: int) → dict[str, Any]#

_addr(unit_id: int) → str#

put_segments( segments: list[tuple[int, int, int]], nbytes: int, n_entries: int, unit_id: int, blob_id: str | None = None, deadline: float | None = None, ) → str#

Write one blob of nbytes to unit unit_id; return its id.

segments are (local_addr, blob_off, n): a value sent zero-copy from the caller’s own (registered) memory, or a run copied into a pool buffer. All go out in one NIXL transfer.

Every wait is bounded by deadline (monotonic). Raises

Class:: UnitFull when the unit refused for lack of space,
Class:: TransferFailed when it failed otherwise, and
Class:: DataPlaneTimeout when the deadline passes (also while the unit keeps answering Busy, design §5.3).

_commit( u: dict[str, Any], blob_id: str, deadline: float | None, errs: list, ) → bool#

Awaited commit: the blob’s rows are published once it returns True.

A blob the unit spilled to disk answers "pending" while its file is written: ask again, inside the put deadline. Commit is idempotent on the unit, so a call whose reply was lost is asked again too. On failure the client aborts, which the unit ignores for a blob it did publish.

put_many( blobs: list[tuple[list[tuple[int, int, int]], int, int, int, str]], deadline: float | None = None, ) → None#

put_segments for several (segments, nbytes, n_entries, unit_id, blob_id).

The first blob of each unit goes out in one round over all units (alloc, one WRITE batch, commit), so a put waits for the slowest unit, not the sum. A blob that does not finish that round (busy, full, spill, no reply, lost) and any further blob of a unit take put_segments, one by one.

_put_round( by_unit: dict[int, tuple], deadline: float | None, ) → set[str]#: One alloc / WRITE / commit round, one blob per unit; the blob ids committed.

_failed(unit_id: int, blobs: list[str], e: Exception) → Exception#: Classify a failed read: lost if the unit is dead, else retryable.

_unit_dead(unit_id: int) → bool#

Ask Ray whether a unit’s actor is gone; mark it DEAD if so.

Only runs after a failure, so the op path stays Ray-free. A unit that is alive but slow is not dead: its blobs are still there.

resolve_keys( by_unit: dict[int, list[str]], deadline: float | None, ) → dict[int, dict[str, Any]]#

One resolve_keys per unit, all units at once; pins what it finds.

Sent to every unit before any reply is read, so a get waits for the slowest unit rather than the sum of them. On a failure, the units that resolved are unpinned before the error is raised (blob_ids of the error are the keys).

read_entries( reads: dict[int, list[tuple[int, int, int]]], keys_of: dict[int, list[str]], deadline: float | None, ) → None#: READ (slab_off, nbytes, local_addr) ranges, all units at once.

unpin_all(resolved: dict[int, dict[str, Any]]) → None#

release_rows( by_unit: dict[int, list[int]], deadline: float | None, ) → None#

Awaited row-wide clear: one release_rows per unit, all units at once.

Rows on a dead unit are already gone, so clearing them succeeds; a live unit that does not answer fails the clear (retryable).

_call_units( op: str, args_by_unit: dict[int, dict[str, Any]], deadline: float | None, ) → tuple[dict[int, Any], list[tuple[int, Exception]]]#

op on every unit at once: results by unit, and (unit, error) failures.

A unit unknown to the directory fails without being asked.

close() → None#

nemo_rl.data_plane.nixl.blobstore#