dau-driver v0.1.0

How to move data to and from a device

This guide shows you how to get bytes across the host link: raw DMA in both directions, the shape rules the device enforces, and the stream codec that turns typed columns into the batch format the shells read.

Two address spaces are in play and confusing them is the classic mistake. DMA device addresses are device memory addresses. Register offsets are AXI-lite offsets in the register window. read_dma(device_address=0x1000) and read_register(0x1000) have nothing to do with each other.

Write and read raw bytes

transfer = device.write_dma(device_address=0x0000_0000, data=payload)
readback = device.read_dma(device_address=0x0000_0000, byte_count=len(payload))
assert readback == payload

write_dma returns the DmaTransfer it validated and performed; read_dma returns the bytes. Both block until the transfer completes: the Linux channel devices have no timeout, and the package does not promise one it cannot keep.

Get the shape right the first time

Every transfer is validated against the device’s own advertised DMA constraints before a channel is opened, so a bad shape costs you nothing:

from dau_driver import DmaDirection

transfer = device.prepare_transfer(DmaDirection.H2C, device_address=0x40, byte_count=128)

TransferValidationError is raised when the byte count is not positive, the device address is negative, the byte count exceeds max_transfer_bytes, the address or byte count is not a multiple of alignment_bytes, or the direction’s channel is not available.

Read the constraints from the device rather than hard-coding them:

dma = device.capabilities.dma
dma.alignment_bytes
dma.max_transfer_bytes

For a maximum-sized loopback, use dma.max_aligned_transfer_bytes rather than computing the descriptor limit yourself.

Guard against addressing past the end of memory

An address beyond the device’s memory does not fail. It wraps to the bottom, and you read or overwrite the wrong region without any error. Tell the handle how big its memory is, then program addresses through the method form:

import dataclasses

from dau_core.registers import RegisterOffset

device = dataclasses.replace(device, memory_bytes=platform.memory.size_bytes)
device.write_address_register(RegisterOffset.INPUT_ADDRESS_LOW, RegisterOffset.INPUT_ADDRESS_HIGH, address)

RegisterAccessError is then raised for an address at or past the limit. Left as None, the caller is trusted: the guard is only as good as the fact, and an invented default would either reject valid addresses or give false confidence.

The module-level write_address function reads memory_bytes off the device by default. Pass an integer to override the limit, or None to disable the check entirely, but only on purpose: the default is what makes the guard apply to callers who never thought about it.

Program a 64-bit address into a register pair

Device addresses are wider than a register, so they are written as a LOW/HIGH pair:

from dau_core.registers import RegisterOffset
from dau_driver.device import write_address

write_address(device, RegisterOffset.INPUT_ADDRESS_LOW, RegisterOffset.INPUT_ADDRESS_HIGH, address)

Always write both halves, which is what write_address does even when the high half is zero. A design whose master is wider than 32 bits keeps the high half across a LOW write, so writing only LOW reuses whatever the previous job left there and addresses the wrong region.

Pass the offsets explicitly rather than deriving low + 4. The job pair is adjacent, but a NoC lane’s is not: its high half is at 0x14, because 0x04 is the result length.

write_address is a function, not a method, because devices here are duck-typed: test fakes and alternative transports implement write_register without subclassing DauDevice.

Encode typed columns as a stream batch

The shells read a DAU stream batch: a header, one descriptor per column, optional operation descriptors, and 64-byte-aligned payloads. Build one from fixed-width, null-free columns:

import struct

from dau_core.stream import LogicalType, OperationCode, OperationDescriptor, output_type_for
from dau_driver.stream_codec import RecordBatchColumn, StreamEncoder

symbols = [1, 1, 2]
prices = [1050, 1075, 990]

columns = (
    RecordBatchColumn(field_id=0, logical_type=LogicalType.INT32, payload=struct.pack(f"<{len(symbols)}i", *symbols)),
    RecordBatchColumn(field_id=1, logical_type=LogicalType.INT32, payload=struct.pack(f"<{len(prices)}i", *prices)),
)

operation = OperationDescriptor(
    operation_id=0,
    opcode=OperationCode.SUM,
    input_column_id=1,
    output_field_id=0,
    output_type=output_type_for(OperationCode.SUM),
    accumulator_width_bits=64,
)

encoded = StreamEncoder.encode_record_batch(columns, row_count=len(prices), operations=(operation,))

For a single column, StreamEncoder.encode_column(logical_type, payload, row_count=..., operations=...) is the one-column form.

Nulls are rejected: the format has no validity bitmap, so null_count must be zero. Drop or fill upstream.

Encode straight from PyArrow

import pyarrow as pa
from dau_driver.stream_codec import StreamDecoder, StreamEncoder

batch = pa.RecordBatch.from_arrays(
    [pa.array([1, 1, 2], type=pa.int32()), pa.array([1050, 1075, 990], type=pa.int32())],
    names=["symbol", "price"],
)

encoded = StreamEncoder.encode_arrow_batch(batch)
restored = StreamDecoder.decode_record_batch_to_arrow(encoded, names=["symbol", "price"])
assert restored.equals(batch)

Column data is taken as a zero-copy, offset-aware view of the primitive buffer, so slicing an array before encoding is cheap. A column with nulls raises InvalidStreamError naming the column.

Decode what comes back

decoded = StreamDecoder.decode_record_batch(encoded)
decoded.header.row_count
decoded.column_payload(1)          # by field_id, not by position

For a single-column batch use decode_column; for the one-row result of a scalar aggregation use decode_scalar_result, which returns the typed value directly.

Decoding validates as it goes (version, header length, alignment, payload windows, per-column widths and row counts, and the operation descriptors) and raises InvalidStreamError on any mismatch. A truncated or misaligned read fails there rather than coming back as a plausible wrong number.

Run one aggregation as a single call

from dau_driver import execute_scalar_operation

result = execute_scalar_operation(
    device,
    LogicalType.INT32,
    struct.pack(f"<{len(prices)}i", *prices),
    row_count=len(prices),
    operation=operation,
    input_device_address=0x0000_0000,
    output_device_address=0x0010_0000,
    output_byte_count=192,
    trigger=my_trigger,
)
print(result.value)

The operation is checked against the device’s advertised capabilities before any DMA: an unsupported opcode or a mismatched output type raises StreamExecutionError with nothing staged. trigger is the hook that starts the job between the H2C write and the C2H read; supply the register sequence your shell expects, or use dau_driver.testing.golden_aggregation_trigger against a fake.

Next

© the dau-driver authorsCreated using Sphinx 8.2.3.