dau_driver.pipeline_execution

Keyed composed-pipeline job execution against the dau_mm_ohlcv_job shell.

The shell runs keyed {value, key} INT32 pair rows through filter -> map -> ordered grouped aggregation and stages 40-byte group records in the output BRAM (see dau_build.shell_tops generate_mm_ohlcv_job_top_sv and the PIPELINE_* register contract).

Attributes

DMA_ALIGN_BYTES

Exceptions

PipelineJobError

A job that timed out or closed out with an error.

Classes

KeyedPipelineConfig

Host-side mirror of the PIPELINE_* configuration registers.

Functions

pack_keyed_rows(rows) → bytes	{value, key} pair rows on the 64-byte DMA grid; the logical length
decode_group_records(data, count) → list[dau_core.operators.GroupAggregate]
configure_pipeline(device, config) → None
_wait_for_job_done(device, *, timeout_seconds, label) → None	Poll JOB_STATUS until DONE, raising on timeout or a hardware error
_read_aligned_records(device, *, address, ...[, ...]) → bytes	Read the 64-byte-aligned record block that RESULT_LENGTH describes.
execute_keyed_pipeline_job(device, rows, *[, ...]) → tuple[list[dau_core.operators.GroupAggregate], int]	Run one batch of (key, value) rows; the pipeline must already be
pack_bar_rows(rows) → bytes	(key, price, shares, frac) rows for the fused bar tile, on the DMA
decode_bar_records(data, count)
execute_bar_job(device, rows, *[, timeout_seconds, ...])	Run one batch of (key, price, shares, frac) rows through the fused
bar_ddr_max_rows_per_job(staging) → int	Rows per job, bounded by the worst case of one record per row so an
stage_bar_ddr_input(device, payload, row_count, *, staging) → None	DMA one pre-packed batch of bar rows into a DDR input region.
start_bar_ddr_job(device, row_count, *, staging[, ...]) → None	Kick a job over rows already staged in DDR.
finish_bar_ddr_job(device, *, staging[, ...]) → tuple[bytes, int]	Wait for the running job and read its records back. Returns the raw
execute_bar_ddr_packed(device, payload, row_count, *, ...) → tuple[bytes, int]	Run one pre-packed batch of bar rows through the DDR-staged fused
bar_noc_lane_output_bytes(staging, lanes) → int	Each lane gets an equal slice of the board's result region.
bar_noc_max_rows_per_job(staging, lanes) → int	Bounded by the worst case of every record landing in ONE lane, so no
device_lane_count(device) → int	The number of NoC lanes the RESIDENT shell advertises.
_advertised(value) → int	All-ones is a word nobody decoded, not a four-billion-row capacity —
device_job_capacity(device) → tuple[int, int]	(input rows, output records) ONE job may stage on the RESIDENT shell.
bar_noc_lane_output_address(staging, lane, lanes) → int
configure_bar_noc_lanes(device, *, staging[, lanes]) → None	Point every lane's writer at its slice of the output region.
execute_bar_noc_packed(device, payload, row_count, *, ...) → list[tuple[bytes, int]]	Run one pre-packed batch through a broadcast-NoC shell: one scan

Module Contents

dau_driver.pipeline_execution.DMA_ALIGN_BYTES = 64

exception dau_driver.pipeline_execution.PipelineJobError(message: str, *, last_error: int | None = None)

Bases: dau_driver.errors.RegisterAccessError

A job that timed out or closed out with an error.

last_error carries the shell’s error code when the job closed out with one, and None when there was none to read (a timeout). Carried on the exception rather than left for the caller to read back, because the register only holds the LAST job’s code: a caller that re-reads it after catching this is racing the next job for the reason this one failed.

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

last_error = None

class dau_driver.pipeline_execution.KeyedPipelineConfig

Host-side mirror of the PIPELINE_* configuration registers.

predicate: dau_core.operators.FilterPredicate

filter_on_key: bool = False

map_config: dau_core.operators.MapConfig | None = None

map_on_key: bool = False

__post_init__() → None

filter_op_word() → int

map_op_word() → int

map_constant_word() → int

dau_driver.pipeline_execution.pack_keyed_rows(rows: list[tuple[int, int]]) → bytes

{value, key} pair rows on the 64-byte DMA grid; the logical length stays len(rows) * 8.

dau_driver.pipeline_execution.decode_group_records(data: bytes, count: int) → list[dau_core.operators.GroupAggregate]

dau_driver.pipeline_execution.configure_pipeline(device, config: KeyedPipelineConfig) → None

dau_driver.pipeline_execution._wait_for_job_done(device, *, timeout_seconds: float, label: str, poll=None) → None

Poll JOB_STATUS until DONE, raising on timeout or a hardware error flag. Shared by every register-polled job runner so the wait/error semantics stay identical across shells. poll, when given, runs each iteration so concurrent work (e.g. overlap staging) can surface its failure instead of hiding behind this wait’s timeout.

dau_driver.pipeline_execution._read_aligned_records(device, *, address: int, result_length: int, record_bytes: int, count: int, count_label: str, prefix: str = '', validate_record_count: bool = True) → bytes

Read the 64-byte-aligned record block that RESULT_LENGTH describes. Returns b"" (and issues no C2H read) when the lane produced no output.

When validate_record_count is set (bar/q1: the count register is an output-record count), the result_length == count * record_bytes invariant is enforced. q6-style aggregate lanes latch an input row count in the count register and emit one fixed-width record, so they pass validate_record_count=False and trust RESULT_LENGTH alone. The empty test keys off RESULT_LENGTH, which is equivalent to count == 0 whenever the invariant holds.

dau_driver.pipeline_execution.execute_keyed_pipeline_job(device, rows: list[tuple[int, int]], *, timeout_seconds: float = 5.0, contract: dau_core.registers.StreamJobRegisterContract | None = None, max_rows: int | None = None) → tuple[list[dau_core.operators.GroupAggregate], int]

Run one batch of (key, value) rows; the pipeline must already be configured (configure_pipeline). Returns (group records, selected row count).

max_rows defaults to what the DEVICE advertises, not to a constant: the staging budget is the resident shell’s BRAM depth, and only the shell knows it.

dau_driver.pipeline_execution.pack_bar_rows(rows: list[tuple[int, int, int, int]]) → bytes

(key, price, shares, frac) rows for the fused bar tile, on the DMA grid; the logical length stays len(rows) * 16.

dau_driver.pipeline_execution.decode_bar_records(data: bytes, count: int)

dau_driver.pipeline_execution.execute_bar_job(device, rows: list[tuple[int, int, int, int]], *, timeout_seconds: float = 5.0, contract: dau_core.registers.StreamJobRegisterContract | None = None, max_rows: int | None = None)

Run one batch of (key, price, shares, frac) rows through the fused bar shell. Returns the decoded bar records.

max_rows defaults to what the DEVICE advertises; the bar shell’s staging depth is its own, not the grouped shell’s.

dau_driver.pipeline_execution.bar_ddr_max_rows_per_job(staging) → int

Rows per job, bounded by the worst case of one record per row so an overrun is impossible by construction. An actual overflow would fault the job rather than corrupt input, but refusing early says which budget was exceeded.

dau_driver.pipeline_execution.stage_bar_ddr_input(device, payload: bytes, row_count: int, *, staging, input_address: int | None = None) → None

DMA one pre-packed batch of bar rows into a DDR input region.

dau_driver.pipeline_execution.start_bar_ddr_job(device, row_count: int, *, staging, input_address: int | None = None, output_address: int | None = None) → None

Kick a job over rows already staged in DDR.

dau_driver.pipeline_execution.finish_bar_ddr_job(device, *, staging, output_address: int | None = None, timeout_seconds: float = 60.0) → tuple[bytes, int]

Wait for the running job and read its records back. Returns the raw record bytes and the bar count — decoding stays with the caller so hosts can use vectorized decoders.

dau_driver.pipeline_execution.execute_bar_ddr_packed(device, payload: bytes, row_count: int, *, staging, timeout_seconds: float = 60.0) → tuple[bytes, int]

Run one pre-packed batch of bar rows through the DDR-staged fused bar shell (dau_mm_bar_ddr_job): stage, start, finish.

dau_driver.pipeline_execution.bar_noc_lane_output_bytes(staging, lanes: int) → int

Each lane gets an equal slice of the board’s result region.

dau_driver.pipeline_execution.bar_noc_max_rows_per_job(staging, lanes: int) → int

Bounded by the worst case of every record landing in ONE lane, so no partition can overrun its slice however skewed the keys are.

dau_driver.pipeline_execution.device_lane_count(device) → int

The number of NoC lanes the RESIDENT shell advertises.

The shell publishes its lane count in the identity registers, so the host has no business assuming one. A composition built with eight lanes driven by a host that assumes four writes four lane windows and silently drops the rest of the result – the kind of wrong answer that looks like a correct short read.

Refuses when the shell advertises nothing (0, which fixed binding tops and any pre-0.2 shell report) rather than falling back to a default: a caller running such a shell knows its geometry and must say so.

dau_driver.pipeline_execution._advertised(value: int) → int

All-ones is a word nobody decoded, not a four-billion-row capacity — the same normalization every capability reader applies.

dau_driver.pipeline_execution.device_job_capacity(device) → tuple[int, int]

(input rows, output records) ONE job may stage on the RESIDENT shell.

These are the shell’s BRAM depths, and the host used to carry them as module constants – one pair for the grouped shell and one for the bar shell, both compiled in. Every shell was therefore batched against the depths of whichever shell the constants were written for, and a shell built with a different depth would have been sliced to the wrong size with nothing to say so. The shell advertises them from register map 0.3.

Refuses when the shell advertises nothing, exactly like the lane count: a 0.2 or earlier shell does not decode these offsets, and a host that guessed here would be reintroducing the constants this replaced.

Reads the two capacity words directly rather than taking a whole identity snapshot: this sits on the per-job path, and a snapshot is ~25 MMIO reads where three will do. Not cached, because a device can be reflashed inside one session – the flywheel does exactly that – and a cached capacity would then describe the previous shell.

dau_driver.pipeline_execution.bar_noc_lane_output_address(staging, lane: int, lanes: int) → int

dau_driver.pipeline_execution.configure_bar_noc_lanes(device, *, staging, lanes: int | None = None) → None

Point every lane’s writer at its slice of the output region.

lanes defaults to what the DEVICE advertises, not to a constant: the shell knows its own geometry and the host should ask.

dau_driver.pipeline_execution.execute_bar_noc_packed(device, payload: bytes, row_count: int, *, staging, lanes: int | None = None, record_bytes: int = BAR_RECORD_BYTES, validate_record_count: bool = True, timeout_seconds: float = 60.0) → list[tuple[bytes, int]]

Run one pre-packed batch through a broadcast-NoC shell: one scan computes every lane’s partition concurrently. The lanes must already be configured (configure_bar_noc_lanes). Returns raw record bytes and each lane’s RECORD_COUNT_LOW value.

lanes defaults to the count the DEVICE advertises; the key-partitioned scan shells reuse this runner by passing their own lane count and per-record width when they run a shell that advertises none.

q1’s grouped shell keeps the default validate_record_count (its count register is an output group-record count). q6’s aggregate shell latches an input row count and emits one fixed-width [sum, count] record per lane, so it passes validate_record_count=False and relies on RESULT_LENGTH.