dau_driver.fleet

Concurrent scan scheduling across independent DAU devices.

Exceptions

FleetScanError	Failure from one device pipeline, annotated with fleet provenance.
_BatchIterationError	Common base class for all non-exit exceptions.
_WorkerError	Common base class for all non-exit exceptions.

Classes

FleetScanResult	One collected batch result with its global index and exact device.
_DeviceTracker
_TrackedDevice
_SharedBatches
_DeviceBatches
_Progress

Functions

run_scan_batches_fleet(devices, batches, *, ...[, ...]) → list[FleetScanResult]	Run one independent scan pipeline per device over a shared lazy input.
_validate_configuration(input_regions, row_bytes) → None
_validate_weights(weights, device_count) → tuple[float, ...] | None

Module Contents

class dau_driver.fleet.FleetScanResult

One collected batch result with its global index and exact device.

Keeping the device object, rather than only its display name, makes provenance unambiguous even when callers construct devices with duplicate names.

batch_index: int

device: Any

value: Any

exception dau_driver.fleet.FleetScanError(batch_index: int | None, device: Any, cause: Exception)

Bases: RuntimeError

Failure from one device pipeline, annotated with fleet provenance.

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

batch_index

device

class dau_driver.fleet._DeviceTracker

indices: list[int] = []

staged_count: int = 0

completed_count: int = 0

explicit_error_index: int | None = None

lock: threading.Lock

pulled(batch_index: int) → None

staged() → None

completed() → None

mark_error(batch_index: int) → None

result_indices() → list[int]

failure_index() → int | None

class dau_driver.fleet._TrackedDevice(device: Any, tracker: _DeviceTracker)

_device

_tracker

__getattr__(name: str) → Any

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

exception dau_driver.fleet._BatchIterationError(batch_index: int, cause: Exception)

Bases: Exception

Common base class for all non-exit exceptions.

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

batch_index

cause

class dau_driver.fleet._SharedBatches(batches: collections.abc.Iterable[dau_driver.scan_execution.ScanBatch], device_count: int, weights: collections.abc.Sequence[float] | None)

_iterator

_condition

_next_index = 0

_cancelled = False

_exhausted = False

_weights

_weighted_current

pull(device_index: int, tracker: _DeviceTracker) → dau_driver.scan_execution.ScanBatch

_preferred_device() → int | None

_advance_weighted_schedule(selected: int) → None

cancel() → None

class dau_driver.fleet._DeviceBatches(source: _SharedBatches, device_index: int, tracker: _DeviceTracker)

_source

_device_index

_tracker

__iter__() → _DeviceBatches

__next__() → dau_driver.scan_execution.ScanBatch

class dau_driver.fleet._Progress(callback: collections.abc.Callable[[int], Any] | None)

_callback

_completed = 0

_lock

report() → None

exception dau_driver.fleet._WorkerError(batch_index: int | None, device: Any, cause: Exception)

Bases: Exception

Common base class for all non-exit exceptions.

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

batch_index

device

cause

dau_driver.fleet.run_scan_batches_fleet(devices, batches: collections.abc.Iterable[dau_driver.scan_execution.ScanBatch], *, per_key_state: bool, input_regions: tuple[int, int], row_bytes: int, collect: collections.abc.Callable[[Any], Any], timeout_seconds: float = 60.0, progress=None, weights: collections.abc.Sequence[float] | None = None, poll_yield_seconds: float | None = None) → list[FleetScanResult]

Run one independent scan pipeline per device over a shared lazy input.

Each batch is pulled exactly once by whichever device pipeline demands it. With weights=None this is pure demand-driven scheduling and naturally self-balances by device throughput. Weights use weighted round-robin only to pre-bias each shared pull toward the requested relative device shares; batches remain lazy and are never partitioned or queued ahead.

Results are returned in global batch order as frozen FleetScanResult records. Each record retains the exact device object so the caller can recover which card produced its collect value. input_regions is reused as device-local addresses, never shared memory.

per_key_state is required and must be False. A stateful composition cannot run on a fleet at all: batches are pulled from ONE shared source by whichever device is free, so a given key’s rows land on whichever cards happen to be idle and per-key state is split across devices that never see each other’s rows. Key-partitioning the input does not help — the affinity would have to be device-wide, and demand-driven scheduling is what makes the fleet self-balancing in the first place. Refusing is the same guard run_scan_batches applies to position-sliced batches, one level up.

poll_yield_seconds is passed through to each pipeline’s JOB_STATUS poll and defaults to None (spin). The real transports release the GIL in their register read (os.pread on Linux, ctypes on macOS), so concurrent pipelines interleave without it and a sleep would only add latency per poll. It exists for a pure-Python transport — a fake in a concurrent test — where the spin WOULD hold the GIL and flatten self-balancing toward an even split regardless of device speed (on fakes with a large speed gap the spinning split was close to even and took twice as long).

Known limitation: cancelling on a device failure wakes pipelines blocked pulling a batch, but one already inside its JOB_STATUS wait runs to its own completion or timeout first, so an early failure can surface only after its peers finish.

An empty batch iterable returns []. An empty device sequence raises ValueError because no pipeline can consume even a potentially lazy iterable. A single device invokes the same run_scan_batches pipeline as the multi-device workers and adds only ordered provenance records.

dau_driver.fleet._validate_configuration(input_regions: tuple[int, int], row_bytes: int) → None

dau_driver.fleet._validate_weights(weights: collections.abc.Sequence[float] | None, device_count: int) → tuple[float, ...] | None