dau_driver.scan_execution

Streaming scan-job runner: ping-pong staging over two DDR input regions so the next batch’s H2C DMA overlaps the current job’s compute — the OHLCV bar-DDR overlap pattern generalized to the scan-composition register contract (INPUT_ADDRESS/INPUT_LENGTH + JOB_CONTROL/JOB_STATUS).

The caller owns everything shell-specific: lane output addresses and any static configuration are written before the run, and collect reads each job’s results (lane registers + output regions, which are disjoint from input staging, so collection also overlaps the next DMA).

Overlap caveat: H2C writes and the job’s burst reads share one DDR controller on dpv1, so the measured win is bounded by DDR bandwidth headroom — the silicon A/B decides the batch split, not this module.

Classes

ScanBatch	One staged scan batch: the packed row payload and its row count.
_KeyPartitionedScanBatches
_KeyRangeScanBatches	Lazily materialized capacity-bounded batches (see

Functions

key_partitioned_scan_batches(records, *, batch_count, ...) → collections.abc.Iterator[ScanBatch]	Build key-confined batches using longest-processing-time-first
key_range_scan_batches(records, *, key_capacity, key, ...) → collections.abc.Iterable[ScanBatch]	Split records into batches each holding at most key_capacity
run_scan_batches(device, batches, *, per_key_state, ...) → list[Any]	Run scan jobs over batches with ping-pong input staging: batch
run_resident_scan(device, *, address, row_count, ...) → list[Any]	Run the job over rows ALREADY on the device, uploading nothing.

Module Contents

class dau_driver.scan_execution.ScanBatch

One staged scan batch: the packed row payload and its row count.

The payload is row_count * row_bytes bytes; the runner pads the DMA to the 64-byte grid while INPUT_LENGTH stays exact.

payload: bytes

row_count: int

class dau_driver.scan_execution._KeyPartitionedScanBatches[Record](records: collections.abc.Iterable[Record], *, batch_count: int, key: collections.abc.Callable[[Record], collections.abc.Hashable], encode: collections.abc.Callable[[collections.abc.Iterable[Record]], bytes])

Bases: collections.abc.Iterator[ScanBatch]

_records

_batch_count

_key

_encode

_batches

__next__() → ScanBatch

Return the next item from the iterator. When exhausted, raise StopIteration

_build_batches() → collections.abc.Iterator[ScanBatch]

dau_driver.scan_execution.key_partitioned_scan_batches[Record](records: collections.abc.Iterable[Record], *, batch_count: int, key: collections.abc.Callable[[Record], collections.abc.Hashable], encode: collections.abc.Callable[[collections.abc.Iterable[Record]], bytes]) → collections.abc.Iterator[ScanBatch]

Build key-confined batches using longest-processing-time-first greedy assignment: largest key group goes into the currently-smallest batch. Records must be ordered by key, and record order within each key is preserved.

LPT requires every key-group size before assignment, so first iteration consumes one complete grouping pass. Creating this iterable consumes nothing, and each batch’s flattened records and encoded payload are constructed only when that batch is requested. Since key groups are indivisible, the largest group bounds the difference between batch sizes; one hot key can therefore dominate a batch.

class dau_driver.scan_execution._KeyRangeScanBatches[Record](records: collections.abc.Iterable[Record], *, key_capacity: int, key: collections.abc.Callable[[Record], collections.abc.Hashable], encode: collections.abc.Callable[[collections.abc.Iterable[Record]], bytes])

Lazily materialized capacity-bounded batches (see key_range_scan_batches()).

_records

_key_capacity

_key

_encode

__iter__() → collections.abc.Iterator[ScanBatch]

dau_driver.scan_execution.key_range_scan_batches[Record](records: collections.abc.Iterable[Record], *, key_capacity: int, key: collections.abc.Callable[[Record], collections.abc.Hashable], encode: collections.abc.Callable[[collections.abc.Iterable[Record]], bytes]) → collections.abc.Iterable[ScanBatch]

Split records into batches each holding at most key_capacity DISTINCT keys — the host half of the hash-mode grouped tile’s overflow contract.

The tile’s table is a hardware fact: a batch presenting more distinct keys than it holds closes out GROUP_CAPACITY_EXCEEDED and is invalidated. Rather than react to that, size the batches so it cannot happen: partition the key space into ranges of key_capacity keys and emit one batch per range. Each key therefore lands in exactly ONE batch, so no per-key state spans a batch boundary and combining the per-batch group records is a concatenation — a stronger property than the general additive merge (dau_core.operators.merge_hash_grouped), which remains correct and is what a caller should use when batches were produced some other way.

Unlike key_partitioned_scan_batches(), records need NOT arrive in key order and arrival order is preserved within each batch: the hash-mode table needs no key ordering, and imposing one here would give back the staging freedom the tile exists to provide. The trade is a counting pass over every key before the first batch is produced, and batches whose ROW counts may be uneven — key ranges are equal in keys, not in rows. The device’s error close-out remains the safety net for a caller whose key_capacity does not match the built design.

dau_driver.scan_execution.run_scan_batches(device, batches: collections.abc.Iterable[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, poll_yield_seconds: float | None = None) → list[Any]

Run scan jobs over batches with ping-pong input staging: batch i + 1 DMAs into the alternate region (and, for a lazy iterable, is produced) while job i computes. Returns collect(device) results in batch order. batches may be lazy — encoding work in the iterator overlaps compute too. per_key_state is required so a stateful composition cannot silently run position-sliced batches; such jobs accept only the result of key_partitioned_scan_batches().

dau_driver.scan_execution.run_resident_scan(device, *, address: int, row_count: int, row_bytes: int, collect: collections.abc.Callable[[Any], Any], repeats: int = 1, timeout_seconds: float = 60.0) → list[Any]

Run the job over rows ALREADY on the device, uploading nothing.

run_scan_batches() re-stages every batch, which is right when the host owns the data and wrong when the device already holds it. A resident dataset is staged once and queried many times, so the DMA belongs outside the loop – and on the measured platform that DMA is the larger term: the q6 device path is 40.0 ms of which the feed is most of it, against a job of about 10.4 ms.

This is deliberately the smallest thing that proves the idea, and it proves it WITHOUT the shell handle table: the caller uploads once and passes the address it uploaded to. What the handle table adds later is safety (a generation-checked identity instead of a raw address a caller can get wrong) and the ability for the DEVICE to name its own allocations – not speed. Keeping those separable is what lets the amortization be measured before any bitstream changes.

Returns one collect(device) per repeat, so a caller can assert every run is bit-identical rather than trusting that a rerun of the same bytes must be.