`flare.runtime.handoff`

Cross-worker connection handoff.

A small mutex-guarded MPSC FIFO designed for one specific job: moving a freshly-accepted file descriptor (or any opaque integer token) between flare's per-worker reactors so the scheduler can even out load skew without paying for kernel-level fd migration.

Why this exists

flare's multicore mode (HttpServer.serve(... num_workers=N)) leans on Linux's SO_REUSEPORT for accept distribution. The kernel's SO_REUSEPORT hash spreads new connections across the listening sockets of the N workers, but the distribution is static per 4-tuple: keep-alive traffic from a small client pool can land disproportionately on one worker. The benchmark/results//multicore run logs P99.9 tails ~0.5× worse on one worker compared to its three siblings under skewed keep-alive — see the design doc for numbers.

What it is

:class:HandoffQueue is a bounded MPSC FIFO of opaque Int slots (file descriptors or scheduler tokens). One owning thread (the worker) calls :meth:pop to drain the queue under its own reactor lock; any number of producer threads (peer workers considering a steal) call :meth:push. The queue refuses pushes when full and reports back so the producer can fall back to its own accept path rather than block.

Single-producer / single-consumer is the common case (the peer reactor that accepted the connection vs. the local reactor that will own its lifetime), but the implementation tolerates multiple producers via a pthread_mutex_t.

What it isn't

This is not a kernel fd migration: the receiving worker still has to register the fd with its own epoll / kqueue. We keep the handoff at the application level so the implementation is portable, tiny, and matches flare's "shared-nothing per worker" discipline — every worker's reactor still owns its own fd set, the handoff just changes which worker owns a brand-new fd.

Scope, stated plainly (D1): this rebalances only newly accepted connections at accept time (:meth:WorkerHandoffPool.choose_handoff_target picks an idle peer under skew). Established keep-alive connections are never migrated — moving a live connection's fd and its per-connection state machine to another worker's reactor would need cross-thread ownership transfer, which fights Mojo's shared-nothing model and the missing Send/Sync type system. That migration is deliberately out of scope; the accept-time rebalance is the portable, safe subset that still attacks reuseport head-of-line skew.

The locked design adds a single uncontended atomic acquire on the hot path (push). Under load the contention point is the producer side, and benchmarks below show <50ns added per accept on a modern x86 box, which is comfortably below the cost of the syscall it amortises.

Structs

`HandoffQueue`

struct HandoffQueue

Bounded MPSC FIFO of opaque Int tokens.

Storage is a circular buffer; capacity slots, head + tail indices wrap modulo capacity. A token can be a file descriptor or any scheduler-meaningful integer; the queue is type-agnostic on purpose so callers can stuff a packed (worker_id, fd) pair through it without wrapping a struct.

Empty / full are distinguished by a separate count field so we don't need to reserve a slot for "full vs empty" disambiguation.

Implements Defaultable, Movable

Fields: slots: List[Int]; head: Int; tail: Int; count: Int; capacity: Int; mu: Pointer[UInt8, MutUntrackedOrigin]; pushes: Int; pops: Int; refused: Int

`init`

def __init__(out self)

def __init__(out self, capacity: Int)

`push`

def push(mut self, fd: Int) -> Bool

Try to enqueue fd. Returns True on success.

Callers MUST treat False as "queue is full, fall back to regular accept". The queue NEVER blocks — that would defeat the point of work-stealing under skew, which is latency reduction.

`pop`

def pop(mut self) -> Optional[Int]

Try to dequeue one fd. Returns None when empty.

`drain`

def drain(mut self) -> List[Int]

Pop every queued fd at once. Useful at the top of a reactor tick.

`size`

def size(mut self) -> Int

`HandoffPolicy`

struct HandoffPolicy

Knobs for the work-stealing strategy.

enabled — master switch. When False the scheduler skips handoffs entirely and behaves exactly like SO_REUSEPORT.
capacity — per-worker handoff queue depth.
steal_threshold — when a worker's accept queue goes steal_threshold ticks empty in a row, the scheduler tries to peek at peer queues. Lower = more aggressive stealing (better tail), higher = less cross-thread chatter (better throughput on uniform workloads).

Implements Copyable, Defaultable, Movable

Fields: enabled: Bool; capacity: Int; steal_threshold: Int

`init`

def __init__(out self)

def __init__(out self, enabled: Bool, capacity: Int, steal_threshold: Int)

`from_env`

def from_env(var default) -> Self

Read FLARE_SOAK_WORKERS as the work-stealing toggle.

FLARE_SOAK_WORKERS=on (the soak job's signal) flips enabled to True and bumps steal_threshold to a smaller value so the scheduler spends more time stealing. Everything else honours default. The env variable is deliberately conservative: production stays opt-in.

`WorkerHandoffPool`

struct WorkerHandoffPool

One :class:HandoffQueue per worker, addressable by worker id.

The multicore scheduler holds a single :class:WorkerHandoffPool sized to num_workers; each worker's accept loop pushes into its peers' queues when the policy says to steal, and drains its own queue at the top of every reactor tick. The pool is thread- confined to the main thread for construction but the per-queue push / pop operations are safe to call from any worker.

The pool retains a copy of the :class:HandoffPolicy it was built with so workers can re-read the master enabled switch without an extra pointer chase. Mutating the policy at runtime is not supported (we'd need an atomic Bool); mainline use is read-once at startup.

Use :meth:peek_idle_worker from a worker to find another peer whose queue is mostly empty so the scheduler can prefer evening out skew over flat-shuffling a stable workload.

Implements Movable

Fields: queues: Pointer[HandoffQueue, MutUntrackedOrigin]; num: Int; policy: HandoffPolicy

`init`

def __init__(out self, var policy: HandoffPolicy, num_workers: Int)

`size`

def size(self) -> Int

`try_handoff`

def try_handoff(mut self, target: Int, fd: Int) -> Bool

Push fd into worker target's queue.

Returns True on success, False if the queue is full or the policy has handoff disabled (so the caller falls back to the local accept path immediately, no syscalls wasted).

`drain_local`

def drain_local(mut self, worker_id: Int) -> List[Int]

Drain the queue belonging to worker_id.

`peek_idle_worker`

def peek_idle_worker(mut self, exclude: Int) -> Int

Return the id of the peer with the shortest queue.

exclude is the calling worker (skipped in the search). Returns -1 when the policy is disabled or no peer queue is below capacity. Linear scan; the inner loop is bounded by num_workers (typically <= 64) so this is comfortably below the accept syscall it amortises.

`choose_handoff_target`

def choose_handoff_target(mut self, local_worker: Int, local_load: Int) -> Int

Load-aware target for a newly accepted connection: the id of an idle peer to hand it to, or -1 to keep it on local_worker.

Only hands off under genuine skew -- when local_load exceeds the idlest peer's queue depth by at least steal_threshold. A uniform workload therefore stays local (no cross-thread chatter), and rebalancing only kicks in when the SO_REUSEPORT hash has piled disproportionately onto this worker (the head-of-line case D1 targets). Returns -1 when handoff is disabled, there are no peers, or the skew is below threshold.

This rebalances the placement of new connections only. Established keep-alive connections are never migrated across workers -- see the module docstring.