test_fleet

Attributes

REGIONS
ROW_BYTES

Classes

TimedTransport	In-memory DeviceTransport: records DMA traffic and register writes.
BlockingTransport	In-memory DeviceTransport: records DMA traffic and register writes.

Functions

_device(tmp_path, name, transport)
_batches(count)
_collect(device)
test_fleet_preserves_order_and_fast_device_takes_more_batches(...)
test_single_device_matches_scan_runner(tmp_path)
test_fleet_consumes_batches_only_as_device_pipelines_demand_them(...)
test_fleet_failure_includes_global_batch_and_device(...)
test_weights_pre_bias_batch_assignment(tmp_path)
test_fleet_defines_empty_inputs(tmp_path)
test_poll_yield_reaches_every_pipeline_status_wait(...) → None	poll_yield_seconds exists for pure-Python transports, whose
test_a_stateful_composition_cannot_run_on_a_fleet(tmp_path) → None	Batches are pulled from ONE shared source by whichever device is free,

Module Contents

test_fleet.REGIONS = (0, 134217728)

test_fleet.ROW_BYTES = 64

class test_fleet.TimedTransport(delay: float = 0.0, *, fail_write: bool = False)

Bases: dau_driver.testing.FakeTransport

In-memory DeviceTransport: records DMA traffic and register writes.

Advertises a current (register map 0.3) shell by default, including the job capacity a host batches against — an all-zero register file models a shell nobody builds any more, and every job test would be exercising the not-advertised refusal instead of the path it means to test. Tests that want an older shell clear or lower these.

delay = 0.0

fail_write = False

write_dma(device_address: int, data: bytes | bytearray | memoryview) → int

write_u32(offset: int, value: int) → None

read_u32(offset: int) → int

class test_fleet.BlockingTransport(started: threading.Event, release: threading.Event)

Bases: TimedTransport

In-memory DeviceTransport: records DMA traffic and register writes.

Advertises a current (register map 0.3) shell by default, including the job capacity a host batches against — an all-zero register file models a shell nobody builds any more, and every job test would be exercising the not-advertised refusal instead of the path it means to test. Tests that want an older shell clear or lower these.

started

release

write_u32(offset: int, value: int) → None

read_u32(offset: int) → int

test_fleet._device(tmp_path, name: str, transport: dau_driver.testing.FakeTransport)

test_fleet._batches(count: int)

test_fleet._collect(device)

test_fleet.test_fleet_preserves_order_and_fast_device_takes_more_batches(tmp_path)

test_fleet.test_single_device_matches_scan_runner(tmp_path)

test_fleet.test_fleet_consumes_batches_only_as_device_pipelines_demand_them(tmp_path)

test_fleet.test_fleet_failure_includes_global_batch_and_device(tmp_path)

test_fleet.test_weights_pre_bias_batch_assignment(tmp_path)

test_fleet.test_fleet_defines_empty_inputs(tmp_path)

test_fleet.test_poll_yield_reaches_every_pipeline_status_wait(tmp_path) → None

poll_yield_seconds exists for pure-Python transports, whose register read would otherwise hold the GIL and serialize co-running pipelines. The real transports release it (os.pread / ctypes), so the default is to spin — assert the plumbing deterministically rather than a wall-clock split, which is not a reliable signal.

test_fleet.test_a_stateful_composition_cannot_run_on_a_fleet(tmp_path) → None

Batches are pulled from ONE shared source by whichever device is free, so a key’s rows land on whichever cards happen to be idle and per-key state splits across devices that never see each other’s rows. Key-partitioning the input does not save it — the affinity would have to be device-wide, and demand-driven scheduling is what makes the fleet self-balancing. This is the same guard run_scan_batches applies to position-sliced batches, one level up.