magmalake flare_mlake

flare_mlake · API · runtime

flare.runtime.io_uring_driver

io_uring SQ/CQ ring driver.

Sits on top of :mod:flare.runtime.io_uring (syscall FFI + ring setup) and :mod:flare.runtime.io_uring_sqe (SQE/CQE codec). Adds the submission queue + completion queue ring access via mmap(2), the atomic head/tail accessors the kernel-shared SPMC protocol requires, and the high-level submit_sqe / reap_cqe API the upcoming UringReactor calls every poll cycle.

What this module provides

  • mmap + munmap FFI wrappers (libc constants PROT_READ, PROT_WRITE, MAP_SHARED, MAP_FAILED).

  • IORING_OFF_* mmap offsets (IORING_OFF_SQ_RING = 0, IORING_OFF_CQ_RING = 0x8000000, IORING_OFF_SQES = 0x10000000).

  • IoUringDriver(entries) — owning wrapper around IoUringRing that performs the three mmaps the kernel ABI requires:

    1. The SQ ring control region at IORING_OFF_SQ_RING of length sq_off.array + sq_entries * sizeof(__u32).
    2. The CQ ring control region at IORING_OFF_CQ_RING of length cq_off.cqes + cq_entries * sizeof(io_uring_cqe).
    3. The SQE array at IORING_OFF_SQES of length sq_entries * sizeof(io_uring_sqe).

    All three munmap'd on drop. The ring fd itself is owned by the embedded IoUringRing and closed via its destructor.

  • Atomic head/tail accessors — release-store on SQ tail (we're producer), acquire-load on CQ tail (kernel is producer), release-store on CQ head (we're consumer). Used via __atomic_load_n / __atomic_store_n libgcc builtins to get the right barriers per CPU; on x86-64 these compile to plain mov + a compiler fence.

  • High-level submit/reap API:

    • next_sqe() -> Pointer[UInt8, MutUntrackedOrigin] — return a writable pointer to the next free SQE slot (caller fills via prep_* helpers from :mod:flare.runtime.io_uring_sqe).
    • commit_sqe() — advance the SQ tail (after the caller finished writing the SQE).
    • submit_and_wait(min_complete: Int) -> Int — call io_uring_enter with the new SQ tail; returns the number of SQEs the kernel consumed.
    • reap_cqe() -> Optional[IoUringCqe] — return the next pending CQE, or None if the CQ is empty. Advances the CQ head atomically.
    • cqe_count() -> Int — number of pending CQEs (kernel tail - userspace head).
    • submit_nop(user_data) -> Int — convenience: fill an SQE with prep_nop, commit, submit, reap one CQE. Used by test_io_uring_driver for the SQ→CQ round-trip smoke test.
Concurrency contract

The kernel-shared SQ/CQ rings are SPSC (single-producer, single-consumer) per direction:

  • SQ direction: userspace is producer, kernel is consumer. We release-store on tail; kernel acquire-loads on tail.
  • CQ direction: kernel is producer, userspace is consumer. Kernel release-stores on tail; we acquire-load tail and release-store our consumed-up-to head.

flare's per-worker model means each IoUringDriver is owned by exactly one pthread (the worker reactor loop), so there's no inter-worker locking on the rings themselves. The atomic ops are only for the userspace-kernel boundary.

References
  • man 2 io_uring_setup — describes the mmap region layout.
  • Jens Axboe, Efficient IO with io_uring (https://kernel.dk/io_uring.pdf) §3 — the SQ/CQ ring protocol.
  • include/uapi/linux/io_uring.h — IORING_OFF_*, io_uring_sqe, io_uring_cqe field layouts.

Structs

IoUringDriver

struct IoUringDriver

High-level driver for one io_uring ring (one fd, one SQ/CQ pair, one SQE array).

Owns one IoUringRing (closes ring fd on drop) and three mmap regions (munmap'd on drop). Exposes the submit/reap API the upcoming UringReactor calls every poll cycle.

Per-worker ownership: each pthread that runs a reactor loop owns exactly one IoUringDriver. The kernel-shared rings are SPSC per direction (kernel + this one userspace thread), so no inter-worker synchronisation is required.

Fields: _ring: Owning IoUringRing (substrate; provides fd + params). _sq_ring_ptr: Mapped SQ ring control region. _sq_ring_len: Length of the SQ ring mapping. _cq_ring_ptr: Mapped CQ ring control region. _cq_ring_len: Length of the CQ ring mapping. _sqes_ptr: Mapped SQE array. _sqes_len: Length of the SQE array mapping. _sq_head_ptr / _sq_tail_ptr / _sq_ring_mask / _sq_array_ptr: Cached pointers + values into the SQ ring control region (no per-call offset arithmetic). _cq_head_ptr / _cq_tail_ptr / _cq_ring_mask / _cq_cqes_ptr: Same for the CQ region. _sq_local_tail: Userspace-side cached SQ tail; bumped by commit_sqe, flushed to the kernel-visible tail by submit_and_wait.

Implements Movable

__init__

def __init__(out self, entries: Int, setup_flags: UInt32 = UInt32(0), sq_thread_cpu: UInt32 = UInt32(0), sq_thread_idle: UInt32 = UInt32(0))

Set up an io_uring with entries SQEs and mmap the three required regions (SQ ring, CQ ring, SQE array).

Args
entries: Number of SQEs (kernel rounds up to a power of two; max 32 768).
setup_flags: IORING_SETUP_* flags forwarded to :class:IoUringRing. Default 0 keeps the historical interrupt-driven, unbatched behaviour. See :data:flare.runtime.io_uring_sqe.IORING_SETUP_COOP_TASKRUN etc. for the bufring path's optimal mix.
sq_thread_cpu: SQPOLL CPU affinity (CPU id the kernel pins the SQPOLL thread to); ignored unless setup_flags includes IORING_SETUP_SQPOLL.
sq_thread_idle: SQPOLL idle timeout in milliseconds before the kernel parks the SQPOLL thread; ignored unless setup_flags includes IORING_SETUP_SQPOLL.

Raises Error: On io_uring_setup failure or any of the three mmap calls failing.

__deinit__

def __deinit__(deinit self)

Unmap the three ring regions; the embedded IoUringRing.__deinit__ then closes the ring fd.

fd

def fd(self) -> Int

Return the underlying ring fd.

sq_entries

def sq_entries(self) -> Int

Return the kernel-allocated SQ size.

cq_entries

def cq_entries(self) -> Int

Return the kernel-allocated CQ size.

sq_ring_mask

def sq_ring_mask(self) -> UInt32

Return the SQ ring index mask (sq_entries - 1).

cq_ring_mask

def cq_ring_mask(self) -> UInt32

Return the CQ ring index mask (cq_entries - 1).

next_sqe

def next_sqe(self) -> Pointer[UInt8, MutUntrackedOrigin]

Return a writable 64-byte pointer to the next free SQE slot.

Caller fills the slot via a prep_* helper from :mod:flare.runtime.io_uring_sqe, then calls :meth:commit_sqe to advance the cached SQ tail. Multiple next_sqe → commit_sqe calls may be interleaved before submit_and_wait flushes them to the kernel.

Returns Pointer() (NULL) if the SQ is full (cached tail - kernel head == sq_entries).

commit_sqe

def commit_sqe(mut self)

Advance the cached SQ tail by one. The kernel-visible tail is not updated until :meth:submit_and_wait is called.

Caller must have written a fully-formed SQE to the slot returned by :meth:next_sqe before calling this.

submit_and_wait

def submit_and_wait(mut self, min_complete: Int) -> Int

Flush the cached SQ tail to the kernel and call io_uring_enter.

Args
min_complete: 0 = non-blocking submit (kernel may process SQEs but the call returns even if no CQEs are ready). Positive = block until at least this many CQEs are ready.

submit_and_wait_timeout

def submit_and_wait_timeout(mut self, min_complete: Int, timeout_ms: Int) -> Int

Like :meth:submit_and_wait, but waits at most timeout_ms.

Needs IORING_ENTER_EXT_ARG (Linux 5.11+), older than every feature the buffer-ring path already requires.

cqe_count

def cqe_count(self) -> Int

Return the number of pending CQEs (kernel-visible tail - userspace head).

reap_cqe

def reap_cqe(mut self) -> Optional[IoUringCqe]

Return the next pending CQE, or None if the CQ is empty.

Advances the CQ head with a release store so the kernel knows the slot is free for the next CQE.

submit_nop

def submit_nop(mut self, user_data: UInt64) -> Int

Submit a single IORING_OP_NOP SQE and return the result-code from io_uring_enter.

Used by the SQ→CQ round-trip smoke test in tests/test_io_uring_driver.mojo and as the simplest sanity check on a fresh driver.

Functions

libc_mmap

def libc_mmap(length: Int, prot: Int, flags: Int, fd: Int, offset: UInt64) -> Pointer[UInt8, MutUntrackedOrigin]

Wrap mmap(2) via libc's mmap symbol.

Args
length: Bytes to map.
prot: PROT_READ | PROT_WRITE for the io_uring use.
flags: MAP_SHARED | MAP_POPULATE for the io_uring use.
fd: IoUringRing.fd().
offset: One of IORING_OFF_SQ_RING / IORING_OFF_CQ_RING / IORING_OFF_SQES.

libc_munmap

def libc_munmap(addr: Pointer[UInt8, MutUntrackedOrigin], length: Int) -> Int

Wrap munmap(2). Returns 0 on success, -errno on failure.

Aliases

comptime PROT_READ = Int(1)

mmap page-protection flag: pages may be read.

comptime PROT_WRITE = Int(2)

mmap page-protection flag: pages may be written.

comptime MAP_SHARED = Int(1)

mmap flag: changes are shared with backing store (required for io_uring's kernel-shared ring regions).

comptime MAP_POPULATE = Int(32768)

mmap flag (Linux-only): pre-fault page tables to avoid the per-page minor-fault cost on first access; the io_uring ring regions are typically a few KB so this is essentially free.

comptime IORING_OFF_SQ_RING = UInt64(0)

mmap offset for the SQ ring control region (head, tail, ring_mask, ring_entries, flags, dropped, array).

comptime IORING_OFF_CQ_RING = UInt64(134217728)

mmap offset for the CQ ring control region (head, tail, ring_mask, ring_entries, overflow, cqes, flags). On kernels with IORING_FEAT_SINGLE_MMAP (5.4+), this region is overlaid on the SQ region and a single mmap(IORING_OFF_SQ_RING, ...) for the larger of the two sizes covers both. flare does the two mmaps separately for kernel-version-portability simplicity; the extra page-table entry is negligible.

comptime IORING_OFF_SQES = UInt64(268435456)

mmap offset for the SQE array (the actual 64-byte SQE slots; the SQ ring's array field is an indirection table of __u32 indices into this array).