# flare.runtime.uring_reactor > A module of flare_mlake's API. ``UringReactor``: io_uring-native event-loop wrapper. Sits on top of :mod:`flare.runtime.io_uring_driver` (which owns the ``io_uring_setup`` + the three SQ/CQ/SQE-array mmaps, the atomic head/tail accessors, and the raw submit/reap surface) and adds the **operation-tagged** API that the upcoming uring-backed server reactor calls: * ``arm_listener_multishot(fd, conn_id)`` — submit one ``IORING_OP_ACCEPT`` SQE with ``IORING_ACCEPT_MULTISHOT`` so the kernel keeps re-arming the accept after every completion. * ``arm_recv_multishot(fd, buf_ptr, buf_len, conn_id)`` — submit one ``IORING_OP_RECV`` SQE with ``IORING_RECV_MULTISHOT`` so the kernel posts a CQE every time data lands without re-arming. (Caller is responsible for buffer ownership; flare uses :mod:`flare.runtime.buffer_pool`.) * ``submit_send(fd, buf_ptr, buf_len, conn_id)`` — fire-and- forget ``IORING_OP_SEND``. The CQE confirms the kernel enqueued the bytes; no per-byte loop. * ``submit_close(fd, conn_id)`` — graceful close via ``IORING_OP_CLOSE``. Posted with ``IOSQE_CQE_SKIP_SUCCESS`` so the success case doesn't even round-trip a CQE — close-on- success is async fire-and-forget. * ``cancel_conn(conn_id)`` — submit an ``IORING_OP_ASYNC_CANCEL`` targeting any in-flight SQE bearing ``conn_id``'s tag. Used by deadline / shutdown paths. * ``poll(min_complete, out)`` — call ``io_uring_enter(min_complete=min_complete)`` then drain the CQ into ``out`` as a list of ``UringCompletion`` records. * ``wakeup()`` — write 1 to a per-reactor eventfd registered as a poll target on the ring (so ``poll`` returns from ``io_uring_enter`` as soon as any thread calls ``wakeup``). Operation tagging ----------------- Each SQE is tagged with a 64-bit ``user_data`` that the kernel returns verbatim on the matching CQE. We pack two fields into that 64-bit slot: | bits 63..56 | bits 55..0 | | op_kind | conn_id | * ``op_kind`` (8 bits) — one of :data:`URING_OP_ACCEPT`, :data:`URING_OP_RECV`, :data:`URING_OP_SEND`, :data:`URING_OP_CLOSE`, :data:`URING_OP_CANCEL`, :data:`URING_OP_WAKEUP`. * ``conn_id`` (56 bits) — caller-defined connection identifier. flare passes a connection-pool slot index. Two helpers — :func:`pack_user_data` and :func:`unpack_user_data` — round-trip the encoding and are unit-tested in the dedicated test module. Why an operation-tagged surface (not register/poll like epoll) -------------------------------------------------------------- io_uring's value proposition is the **opposite** of epoll: instead of "tell me when fd is ready" → "do op manually", it's "do op for me; tell me when it completed". Flattening that into an epoll-style ``register(fd, READ); on poll, recv()`` surface re-introduces the syscall-per-event cost epoll has and ``io_uring`` was built to avoid. So ``UringReactor`` exposes the io_uring-native submit/reap API directly. The two backends (``epoll/kqueue`` ``Reactor`` vs ``UringReactor``) are selected at the **server** layer by a comptime branch — the ``_server_reactor_impl`` state machine talks to whichever backend is selected via a thin trait surface. Concurrency ----------- One ``UringReactor`` per worker pthread (matching the existing ``Reactor`` ownership model). All SQ/CQ ring atomics happen on the single owning thread; the only cross-thread hook is ``wakeup`` which writes 1 byte into the per-reactor eventfd. ## Structs ### `UringReactor` ```mojo struct UringReactor ``` One io_uring ring + a wakeup eventfd, exposed as the submit/reap surface flare's per-worker server reactor uses on the io_uring backend. Per-worker ownership model: each pthread that runs a server reactor loop owns exactly one ``UringReactor``; the kernel- shared rings are SPSC per direction so no inter-worker synchronisation is needed beyond the ``wakeup`` eventfd write. Fields: _driver: The owning ``IoUringDriver`` (closes ring fd + unmaps SQ/CQ/SQE on drop). _wake_fd: Per-reactor eventfd. ``poll`` arms a ``IORING_OP_RECV`` against it (the eventfd's read side is byte-stream-shaped) so the next ``wakeup`` from any thread surfaces a ``URING_OP_WAKEUP`` CQE. _wake_buf: Pinned 8-byte buffer the wakeup recv writes into. _io: Cached ``libflare_tls`` raw-IO handles for the ``write`` syscall used by ``wakeup``. _wake_armed: True after the first ``poll`` arms the wakeup recv; we re-arm lazily after each wakeup CQE. Implements: `Movable` #### `__init__` ```mojo def __init__(out self, entries: Int = Int(256), setup_flags: UInt32 = UInt32(0), sq_thread_cpu: UInt32 = UInt32(0), sq_thread_idle: UInt32 = UInt32(0), enable_wakeup: Bool = True) ``` Set up the ring + (optionally) the wakeup eventfd. Args: - `entries`: SQE count (kernel rounds up to a power of two; default 256 matches the epoll ``max_events`` budget so the per-worker memory footprint is comparable). - `setup_flags`: Bitwise-OR of ``IORING_SETUP_*`` flags forwarded to :class:`IoUringDriver`. Default 0 preserves the original interrupt-driven, no- batching behaviour. The bufring dispatch path opts into ``COOP_TASKRUN | TASKRUN_FLAG | SUBMIT_ALL`` and (on kernel >= 6.1) ``SINGLE_ISSUER | DEFER_TASKRUN`` so the kernel batches task work to enter boundaries instead of running it IPI-style mid-syscall. - `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``. - `enable_wakeup`: Default True (the historical behaviour). Set False to skip the wakeup eventfd creation + the lazy-arm of the wakeup ``IORING_OP_READ`` SQE in :func:`poll`. This eliminates one always-pending SQE on the hot path AND removes the eventfd file descriptor. Safe to set False when the reactor's owner thread will never call :func:`wakeup` from another thread (which is the case for the bufring dispatch -- each worker owns its own ring + drives it from one pthread, and shutdown is signalled via a heap-allocated flag the worker checks each iteration). Raises: Error: On ``io_uring_setup`` failure (see :class:`IoUringDriver`) or ``eventfd`` failure. #### `__deinit__` ```mojo def __deinit__(deinit self) ``` Free the wakeup buffer + close the wakeup fd; the ``IoUringDriver`` destructor handles ring teardown. #### `fd` ```mojo def fd(self) -> Int ``` Return the underlying io_uring ring fd. #### `sq_entries` ```mojo def sq_entries(self) -> Int ``` Kernel-allocated SQ size. #### `cq_entries` ```mojo def cq_entries(self) -> Int ``` Kernel-allocated CQ size. #### `arm_listener_multishot` ```mojo def arm_listener_multishot(mut self, listener_fd: Int, conn_id: UInt64 = UInt64(0)) ``` Submit one multishot accept SQE so the kernel posts a CQE for every accepted connection without re-arming. The CQE handler should look at ``cqe.has_more`` — when unset, the multishot has terminated (e.g. listener closed) and the caller should re-arm. Args: - `listener_fd`: Listener socket fd (must be non-blocking or kernel ≥ 6.0 for blocking-listener support). - `conn_id`: Tag returned in every accept CQE; defaults to 0 when there's only one listener. Raises: Error: If the SQ is full (hot-path error; caller should ``poll`` to drain CQEs first). #### `arm_recv_multishot` ```mojo def arm_recv_multishot(mut self, fd: Int, buf: Pointer[UInt8, MutUntrackedOrigin], buf_len: Int, conn_id: UInt64) ``` Submit one recv SQE that re-fires once when data arrives. **NOTE on naming**: the original intent was true multishot (``IORING_RECV_MULTISHOT``) but the Linux kernel rejects multishot recv with ``-EINVAL`` unless ``IOSQE_BUFFER_SELECT`` is set on the SQE (kernel needs a buffer pool to pick from on each completion -- it can't pre-allocate space for an unbounded stream into a single fixed buffer). This helper targets a single user-owned buffer, so it's effectively one-shot at the kernel level. For TRUE multishot recv, use :func:`arm_recv_buffer_select` with a registered ``IORING_REGISTER_PBUF_RING`` buffer ring. Args: - `fd`: Connected socket fd. - `buf`: Receive buffer (caller-owned; flare uses a BufferHandle from the per-worker pool). - `buf_len`: Buffer capacity in bytes. - `conn_id`: Tag returned in the recv CQE. #### `arm_provide_buffers` ```mojo def arm_provide_buffers(mut self, addr: UInt64, nbytes_per_buf: Int, nbufs: Int, bgid: UInt16, bid: UInt16 = UInt16(0)) ``` Submit one ``IORING_OP_PROVIDE_BUFFERS`` SQE that hands the kernel a contiguous run of ``nbufs`` buffers of ``nbytes_per_buf`` bytes each, starting at ``addr`` with ids ``[bid, bid + nbufs)`` in buffer-group ``bgid``. After this SQE completes (one CQE tagged ``URING_OP_PROVIDE_BUFFERS`` with ``conn_id = bgid`` and ``res = number of buffers actually accepted``), subsequent ``arm_recv_buffer_select`` calls with the same ``bgid`` will have the kernel auto-pick a free buffer for each recv. flare's recv-buffer-ring dispatch loop typically calls this once at startup with N×8 KiB buffers per worker, then re-arms the same buffer id after each recv CQE is processed (one PROVIDE_BUFFERS SQE with ``nbufs=1`` per recv CQE -- a cheap re-fill that amortises into the next ``io_uring_enter``). Args: - `addr`: Pointer to the first buffer in the run (typically the worker's owned heap allocation, or the slot inside it being re-fed). - `nbytes_per_buf`: Per-buffer size in bytes. - `nbufs`: Number of contiguous buffers (must be > 0). - `bgid`: Buffer-group id; used by the matching recv SQE's ``buf_index`` and reported in the CQE's ``conn_id``. - `bid`: Starting buffer id (default 0). #### `register_pbuf_ring` ```mojo def register_pbuf_ring(mut self, bgid: UInt16, ring_entries: Int) -> Int ``` Register a kernel-mapped provided-buffer ring of ``ring_entries`` slots in buffer-group ``bgid`` via ``IORING_REGISTER_PBUF_RING`` (kernel 5.19+). The 2.7x faster successor to :func:`arm_provide_buffers` per Linux kernel benchmarks. Once registered, the ring is populated by writing ``struct io_uring_buf`` entries and bumping the tail pointer in shared memory -- NO SQE per refill, NO io_uring_enter per refill. The kernel reads the tail with acquire semantics on every recv that selects from this group, so applications can replenish single buffers cheaply. Returns the ring memory address. Caller is responsible for :func:`pbuf_ring_add_buffer` + :func:`pbuf_ring_advance` to seed the ring; flare's bufring dispatch typically does this once with all N buffers at startup, then re-adds one buffer per recv CQE before the next io_uring_enter. Args: - `bgid`: Buffer-group id matching subsequent ``arm_recv_buffer_select(... bgid ...)`` calls. - `ring_entries`: Number of slots in the ring. Must be a power of two; max 2^15 (32768). Raises: Error: If ring_entries is not a power of two, exceeds 2^15, or the kernel rejects the registration. #### `unregister_pbuf_ring` ```mojo def unregister_pbuf_ring(mut self, bgid: UInt16, ring_addr: Int, ring_entries: Int) ``` Drop a ring previously registered via :func:`register_pbuf_ring`. Kernel side cleans up its reference; userspace side ``munmap``s the ring memory. #### `arm_recv_buffer_select` ```mojo def arm_recv_buffer_select(mut self, fd: Int, bgid: UInt16, conn_id: UInt64, multishot: Bool = True) ``` Submit one ``IORING_OP_RECV`` SQE with ``IOSQE_BUFFER_SELECT`` set; the kernel picks a buffer from the ``bgid`` pool at recv time. This is the production HTTP server recv shape every Rust io_uring HTTP server uses. Combined with ``multishot=True`` (default), one SQE per accepted connection drives an unbounded stream of recv CQEs; each CQE points at a fresh kernel-picked buffer (id in the high 16 bits of ``UringCompletion.flags``, decoded via :func:`IoUringCqe.buffer_id`). No per-conn buffer ownership, no per-CQE re-arm, no recv syscall round-trip. Args: - `fd`: Connected socket fd. - `bgid`: Buffer-group id matching a prior ``arm_provide_buffers`` call. - `conn_id`: Tag returned in every recv CQE; identifies the connection this recv belongs to. - `multishot`: When True (default), set ``IORING_RECV_MULTISHOT`` so the kernel keeps the recv armed across CQEs. Only valid in combination with ``IOSQE_BUFFER_SELECT`` (which this helper always sets), since multishot recv requires a buffer-group source. #### `submit_send` ```mojo def submit_send(mut self, fd: Int, buf: Pointer[UInt8], buf_len: Int, conn_id: UInt64) ``` Submit one ``IORING_OP_SEND`` SQE. Args: - `fd`: Connected socket fd. - `buf`: Send buffer; caller must keep it alive until the matching CQE is reaped. - `buf_len`: Bytes to send. - `conn_id`: Tag returned in the send CQE. #### `submit_close` ```mojo def submit_close(mut self, fd: Int, conn_id: UInt64) ``` Submit one ``IORING_OP_CLOSE`` SQE. Uses ``IOSQE_CQE_SKIP_SUCCESS`` so the kernel only posts a CQE on failure — the typical close path is async fire-and-forget. #### `arm_poll_readable_multishot` ```mojo def arm_poll_readable_multishot(mut self, fd: Int, conn_id: UInt64, poll_mask: UInt32 = UInt32(8193)) ``` Submit a multishot ``IORING_OP_POLL_ADD`` against ``fd``. Kernel posts a CQE every time the fd's readiness matches any bit in ``poll_mask``, without re-arming. This is the io_uring analog of ``epoll_ctl(EPOLL_CTL_ADD, fd, EPOLLIN | EPOLLET)`` and is the substrate the upcoming server-loop dispatch swap (B0 wire-in) uses to replace ``epoll_wait`` on the io_uring backend. The CQE is tagged ``URING_OP_POLL`` so the dispatch loop can route it to the same code path that handles ``EVENT_READABLE`` on the epoll backend; the existing ``ConnHandle.on_readable`` then runs its own ``recv`` syscall. The buffer-ring path (``IORING_OP_RECV`` + ``IORING_RECV_MULTISHOT`` against a registered ``IORING_REGISTER_PBUF_RING``) is a separate dispatch loop opt-in via ``FLARE_BUFRING_HANDLER=1``. Args: - `fd`: Connected socket fd. ``debug_assert`` checks ``fd >= 0``. - `conn_id`: Caller-defined connection slot id (e.g. the fd itself when 1:1 with a ConnHandle). - `poll_mask`: Defaults to ``POLLIN | POLLRDHUP`` so peer-closed connections surface alongside data- available ones; pass ``POLLOUT`` for write-side readiness or any combination of the ``POLL*`` constants. Raises: Error: If the SQ is full (hot-path; caller should drain CQEs first). #### `cancel_poll` ```mojo def cancel_poll(mut self, conn_id: UInt64) ``` Submit ``IORING_OP_POLL_REMOVE`` cancelling the multishot poll registered for ``conn_id``. After the kernel processes the remove SQE, two CQEs arrive: one tagged ``URING_OP_POLL_REMOVE`` (this SQE's own completion, ``res = 0`` on success or ``-ENOENT`` if nothing matched), and one final tagged ``URING_OP_POLL`` without ``IORING_CQE_F_MORE`` for the cancelled poll itself. #### `cancel_conn` ```mojo def cancel_conn(mut self, conn_id: UInt64) ``` Submit an ``IORING_OP_ASYNC_CANCEL`` for any in-flight op tagged with ``conn_id`` (the kernel cancels the first match). #### `poll` ```mojo def poll(mut self, min_complete: Int, mut out: List[UringCompletion], max_completions: Int = Int(64), timeout_ms: Int = Int(-1)) -> Int ``` Submit any pending SQEs, drain ready CQEs, then optionally block for more. Peek-then-block dispatch (matches the ``tokio-uring`` / ``monoio`` pattern). Three phases per call: 1. **Submit**: flush any pending SQEs to the kernel via ``io_uring_enter(submit, 0, GETEVENTS)`` -- this does NOT block, but does collect any CQEs the kernel has already posted asynchronously (e.g. multishot recv completions that fired between the prior poll and this one). On kernels with DEFER_TASKRUN (negotiated by the bufring setup-flag probe), this call also runs deferred task work which can produce the CQEs we then drain in step 2. 2. **Drain**: pull every ready CQE out of the CQ into ``out``, up to ``max_completions``. No syscall in this phase -- pure userspace ring head/tail manipulation. This is where multishot recv CQEs land in batches under sustained load; an always-block design would only pull one at a time because ``submit_and_wait(1)`` returns as soon as one CQE is ready. 3. **Block (only if needed)**: if we drained fewer than ``min_complete`` CQEs, call ``submit_and_wait(...)`` with the remaining count to actually block. The common case under sustained load is that step 2 returned plenty of CQEs and step 3 is skipped entirely. Returns the number of completions appended to ``out``. Args: - `min_complete`: 0 = non-blocking poll; positive = block until at least this many CQEs are ready across the drain + optional block phases. - `out`: Output list (cleared on entry). - `max_completions`: Per-poll budget so one slow op doesn't starve the reactor (matches epoll's ``max_events``). - `timeout_ms`: Longest the blocking phase may wait. Negative (the default) waits for ``min_complete`` completions however long they take; non-negative bounds the wait and returns whatever arrived, so a loop can re-read its stop flag (a ring with no wakeup channel needs this). #### `wakeup` ```mojo def wakeup(self) ``` Cross-thread wakeup: write 1 to the eventfd. Safe from any thread. The next ``poll`` will return because the multishot recv on the eventfd posts a CQE. No-op when the reactor was constructed with ``enable_wakeup=False`` -- single-issuer bufring rings opt out because they don't need cross-thread signalling. ## Functions ### `use_uring_backend` ```mojo def use_uring_backend() -> Bool ``` Comptime+runtime predicate: True iff the host kernel exposes io_uring **and** the build target is Linux. The flare server's reactor branch consults this once at startup to decide whether to construct a ``UringReactor`` (io_uring path) or fall back to the existing ``Reactor`` (epoll path on Linux without io_uring, kqueue on macOS). The decision is per-process: a long-running server picks one backend at boot and never switches. The choice can be forced off via the ``FLARE_DISABLE_IO_URING=1`` environment variable for A/B benchmarking; flare's default is "use io_uring when available".