nemo_rl.data_plane.adapters.noop#

In-memory DataPlaneClient test fixture.

Behaves like a real adapter end-to-end (put → get → clear, consumption counters, field-presence as the stage-done signal) but stores everything in process memory. The ABC contract tests run against this implementation so they don’t require TQ installed.

Production callers must NOT use this — :func:build_data_plane_client intentionally raises when enabled=False rather than returning a NoOp fallback (see factory.py).

Module Contents#

Classes#

_Partition

NoOpDataPlaneClient

Reference in-memory implementation.

Functions#

_reject_non_tensor_leaves

No pickle on the bus. NoOp-only, not a data-plane rule.

API#

nemo_rl.data_plane.adapters.noop._reject_non_tensor_leaves(td: tensordict.TensorDict) → None#

No pickle on the bus. NoOp-only, not a data-plane rule.

The ABC allows NonTensorStack / NonTensorData leaves and both the TQ and local adapters accept them. This adapter keeps rows as plain tensor dicts with no encoder behind it, so it rejects them here instead of storing something it cannot hand back.

Walk the leaves via keys() + indexed lookup rather than items(), because some tensordict versions skip NonTensorData entries from items(leaves_only=True) — they’re “leaves” by structure but not tensor-typed, so they’d silently slip past a naive items() iteration.

class nemo_rl.data_plane.adapters.noop._Partition#
fields: list[str]#

None

num_samples: int#

None

consumer_tasks: list[str]#

None

grpo_group_size: int | None#

None

enums: dict[str, list[str]]#

None

rows: dict[str, dict[str, torch.Tensor]]#

‘field(…)’

tags: dict[str, dict[str, Any]]#

‘field(…)’

consumed: dict[str, set[str]]#

‘field(…)’

class nemo_rl.data_plane.adapters.noop.NoOpDataPlaneClient#

Bases: nemo_rl.data_plane.interfaces.DataPlaneClient

Reference in-memory implementation.

Initialization

register_partition(
partition_id: str,
fields: list[str],
num_samples: int,
consumer_tasks: list[str],
grpo_group_size: int | None = None,
enums: dict[str, list[str]] | None = None,
) → None#
claim_meta(
partition_id: str,
task_name: str,
required_fields: list[str],
batch_size: int,
dp_rank: int | None = None,
blocking: bool = True,
timeout_s: float = 60.0,
) → nemo_rl.data_plane.interfaces.KVBatchMeta#
get_data(
meta: nemo_rl.data_plane.interfaces.KVBatchMeta,
select_fields: list[str] | None = None,
) → tensordict.TensorDict#
check_consumption_status(
partition_id: str,
task_names: list[str],
) → bool#
put_samples(
sample_ids: list[str],
partition_id: str,
fields: tensordict.TensorDict | None = None,
tags: list[dict[str, Any]] | None = None,
) → nemo_rl.data_plane.interfaces.KVBatchMeta#
get_samples(
sample_ids: list[str],
partition_id: str,
select_fields: list[str],
) → tensordict.TensorDict#
list_sample_ids(partition_id: str) → list[str]#

List stored sample IDs without reading their tensor payloads.

clear_samples(sample_ids: list[str] | None, partition_id: str) → None#
save_checkpoint(
checkpoint_dir: str | pathlib.Path,
*,
metadata: dict[str, Any] | None = None,
) → None#

Persist the trusted in-memory fixture for adapter contract tests.

This test-only adapter uses pickle; callers must not load checkpoints from untrusted paths.

load_checkpoint(
checkpoint_dir: str | pathlib.Path,
) → dict[str, Any]#

Restore the in-memory fixture into a clean client.

close() → None#