UringReactor
struct UringReactorOne 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__
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 epollmax_eventsbudget so the per-worker memory footprint is comparable).setup_flags: Bitwise-OR ofIORING_SETUP_*flags forwarded to :class:IoUringDriver. Default 0 preserves the original interrupt-driven, no- batching behaviour. The bufring dispatch path opts intoCOOP_TASKRUN | TASKRUN_FLAG | SUBMIT_ALLand (on kernel >= 6.1)SINGLE_ISSUER | DEFER_TASKRUNso 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 unlesssetup_flagsincludesIORING_SETUP_SQPOLL.sq_thread_idle: SQPOLL idle timeout in milliseconds before the kernel parks the SQPOLL thread; ignored unlesssetup_flagsincludesIORING_SETUP_SQPOLL.enable_wakeup: Default True (the historical behaviour). Set False to skip the wakeup eventfd creation + the lazy-arm of the wakeupIORING_OP_READSQE 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:wakeupfrom 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__
def __deinit__(deinit self) Free the wakeup buffer + close the wakeup fd; the IoUringDriver destructor handles ring teardown.
fd
def fd(self) -> Int Return the underlying io_uring ring fd.
sq_entries
def sq_entries(self) -> Int Kernel-allocated SQ size.
cq_entries
def cq_entries(self) -> Int Kernel-allocated CQ size.
arm_listener_multishot
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
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
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'sbuf_indexand reported in the CQE'sconn_id.bid: Starting buffer id (default 0).
register_pbuf_ring
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 subsequentarm_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
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 munmaps the ring memory.
arm_recv_buffer_select
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 priorarm_provide_bufferscall.conn_id: Tag returned in every recv CQE; identifies the connection this recv belongs to.multishot: When True (default), setIORING_RECV_MULTISHOTso the kernel keeps the recv armed across CQEs. Only valid in combination withIOSQE_BUFFER_SELECT(which this helper always sets), since multishot recv requires a buffer-group source.
submit_send
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
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
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_assertchecksfd >= 0.conn_id: Caller-defined connection slot id (e.g. the fd itself when 1:1 with a ConnHandle).poll_mask: Defaults toPOLLIN | POLLRDHUPso peer-closed connections surface alongside data- available ones; passPOLLOUTfor write-side readiness or any combination of thePOLL*constants.
Raises Error: If the SQ is full (hot-path; caller should drain CQEs first).
cancel_poll
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
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
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:
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.Drain: pull every ready CQE out of the CQ into
out, up tomax_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 becausesubmit_and_wait(1)returns as soon as one CQE is ready.Block (only if needed): if we drained fewer than
min_completeCQEs, callsubmit_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'smax_events).timeout_ms: Longest the blocking phase may wait. Negative (the default) waits formin_completecompletions 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
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.