`flare.runtime.scheduler`

Multicore scheduler: 1 shared listener, N reactors, N workers.

Runs N single-threaded reactor loops in parallel, one per worker thread, all sharing a single listener fd. Each worker registers the listener fd in its own reactor with EPOLLEXCLUSIVE (on Linux

= 4.5) so the kernel wakes exactly one worker per accept event. On macOS / kqueue the flag is unavailable and registration falls back to plain register — the practical behaviour is similar because non-blocking accept returns EAGAIN on the losers.

This is the redesign of the multicore path. the prior implementation used N independent listeners bound with SO_REUSEPORT; the kernel then distributed accepted connections across workers by hashing each connection's 4-tuple to one of the N listeners. That scheme has a known fairness gap under bursty arrival: a 256-connection storm from a wrk2-style load generator can land 80+ conns on one worker and 30 on another, producing hundreds-of-millisecond head-of-line tail latency on the overloaded worker(s). benchmark/configs/throughput_mc shows the failure mode at p99 ≈ 1.7 s; the EPOLLEXCLUSIVE-based shared-listener design collapses the same workload's p99 to the millisecond range.

Shutdown path: shutdown() flips a heap-allocated Bool that every worker polls on each reactor iteration. A stable heap address is used (instead of a field on the Scheduler struct) so the flag survives the NRVO-or-not move from Scheduler.start back to the caller. The main thread joins all workers, and only then closes the shared listener fd (once), before shutdown() returns.

The listener is never closed to wake the workers. On Linux an fd that is also registered in a worker's epoll instance is kept alive by the kernel, so close() from another thread triggers no EPOLLHUP and the workers stay blocked in epoll_wait until the 100 ms poll timeout fires; and closing it early let a worker that polled a batch before it saw the flag accept on a reused number (CONC-05). The heap flag is what breaks the loop.

This module is :trait:Frontend-generic: every worker calls :meth:Frontend.run_worker once with its assigned listener fd and the shared stopping flag. The frontend value is moved (per-worker copies are made via F.copy(); if that's expensive users should wrap their handler's expensive state behind an Pointer or a similar shared-reference holder).

The scheduler used to import directly from :mod:flare.http._server_reactor_impl, :mod:flare.http._unified_reactor_impl, :mod:flare.http.handler, :mod:flare.http.server, :mod:flare.http.static_response, and :mod:flare.http2.server. That layering violation (runtime depending on protocol modules) is gone now; the scheduler depends only on the :trait:flare.runtime.Frontend trait and concrete impls live where their protocol does (:class:flare.http.HttpFrontend / :class:flare.http.StaticHttpFrontend).

Known limitations:

The stopping flag is a heap-allocated byte written from the main thread and read from each worker through Atomic[UInt8] release-store / acquire-load (store_stop_flag / load_stop_flag). This gives the workers a proper happens-before edge on shutdown and lowers to a plain mov on x86-64 (TSO) and to stlr / ldar on ARM64. The per-iteration re-materialisation of the pointer inside the frontend's serving loop still defeats the optimiser's LICM so every iteration re-reads the flag.
Worker panics that escape :meth:Frontend.run_worker are caught and discarded in _worker_entry because pthread has no exception channel. is_running() still reports True until the ThreadHandle is joined, which is a mildly wrong signal for a worker that crashed rather than shut down cleanly. Plumbing a per-worker error cell back to the Scheduler is also scheduled for a future release. .

Structs

`ShutdownReport`

struct ShutdownReport

Per-worker drain summary returned by :meth:Scheduler.drain.

Originally defined under flare.http.server and imported back by the scheduler; promoted into :mod:flare.runtime to break the runtime → http import cycle. flare.http.server re-exports this type so the public flare.http.ShutdownReport surface is unchanged.

The per-worker registry that would let us count individual in-flight connections lives on each worker's stack today; until that registry is published back through a shared atomic, in_flight_at_deadline is the coarse 0/1 signal driven by whether the worker's join completed inside the budget.

Fields: drained: Connections that completed their in-flight work inside the drain timeout. Best-effort. timed_out: Connections that were force-closed because the drain timeout elapsed before they finished. Best-effort. in_flight_at_deadline: Connections still alive at the instant the timeout fired (== timed_out after the force-close completes). crashed: 1 if this worker's last run exited crashed (poll failure) rather than cleanly, else 0. Mirrors the per-worker slot behind crashed_worker_count() so a caller can attribute a crash to a specific worker.

Implements Copyable, ImplicitlyCopyable, Movable

Fields: drained: Int; timed_out: Int; in_flight_at_deadline: Int; crashed: Int

`Scheduler`

struct Scheduler[F: Frontend]

Owns num_workers workers, each running a frontend's serving loop sharing a single listener fd (or its own SO_REUSEPORT listener when the strategy demands).

Usage: mojo from flare.http import HttpFrontend var frontend = HttpFrontend(handler^, config^) var s = Scheduler[HttpFrontend[MyHandler]].start( addr, frontend^, num_workers=4 ) # ... server running ... s.shutdown()

Notes:

The scheduler stores the workers' ThreadHandle values and a heap-allocated stopping flag. shutdown() writes through that heap address and joins all workers.
The stopping flag lives on the heap (not on this struct) so its address survives the move from Scheduler.start back to the caller; every worker captures that address once at spawn time.
Listener strategy. Default: each worker pre-binds its own SO_REUSEPORT listener (the kernel hashes new 4-tuples to one of N pre-bound listeners; matches actix_web's listener strategy and gives the highest steady-state throughput on dev-box workloads). Opt out by exporting FLARE_REUSEPORT_WORKERS=0 before start to switch to a single shared listener bound via bind_shared and registered with Reactor.register_exclusive (EPOLLEXCLUSIVE on Linux >= 4.5) — the kernel wakes one worker per accept event, idle workers absorb spikes, p99.99 σ is uniformly tighter under sustained load for 7-22 % less req/s depending on path (see docs/benchmark.md). Frontends that require per-worker listeners (the io_uring buffer-ring frontend) override this via :meth:Frontend.requires_per_worker_listener.
The frontend value is cloned into each worker via F.copy().

Implements Movable

`init`

def __init__(out self)

Build an empty scheduler; use Scheduler.start instead.

`start`

def start(addr: SocketAddr, var frontend: F, num_workers: Int, pin_cores: Bool = True, extra_addrs: List[SocketAddr] = List()) -> Self

Spawn num_workers threads sharing one listener.

The scheduler binds its listener(s) according to :meth:Frontend.requires_per_worker_listener plus the FLARE_REUSEPORT_WORKERS env knob, then hands one fd per worker into :meth:Frontend.run_worker. The frontend encapsulates everything protocol-specific (handler, request config, HTTP/2 settings, auto-protocol toggle, backend selection); the runtime layer no longer knows any of those flags exist.

Args: addr: Address the listener(s) bind.; frontend: Per-worker run target. Cloned per worker via :meth:Frontend.copy.; num_workers: Number of worker threads. Must be in 1..=256; values outside that range raise. The upper bound is a defensive guard against runaway pthread_create + heap allocation.; pin_cores: If True (default), pin worker N to core N % num_cpus. No-op on macOS.; extra_addrs: Additional addresses to serve. Each worker gets its own SO_REUSEPORT listener on every one, so N addresses x M workers binds N*M listeners, all serially on this thread. Empty (default) preserves the single-address behaviour exactly. Every extra address must be bindable with SO_REUSEPORT, so any earlier listener on it must already be closed.

Raises Error: If num_workers is outside 1..=256, if the listener fails to bind, or if spawning a worker fails; partially-started workers are best-effort joined before re-raising.

`shutdown`

def shutdown(mut self)

Signal every worker to stop and wait for them to join.

Flips the heap-allocated stopping flag (what breaks the workers' loops), joins all worker threads, records the crash count, and frees every worker context, listener (the shared one is closed here, after the join), stats cell, and the stopping-flag heap cell. Idempotent — a second call finds the state empty and is a no-op.

`is_running`

def is_running(self) -> Bool

Return True if any worker has not yet joined.

Note: this only tracks join state. To tell a crash from a clean shutdown after the workers have joined, read crashed_worker_count() (populated by shutdown / drain from the per-worker exit-status cells).

`crashed_worker_count`

def crashed_worker_count(self) -> Int

Number of workers whose last run exited crashed (poll failure), captured during the most recent shutdown / drain. 0 after a clean shutdown.

`drain`

def drain(mut self, timeout_ms: Int) -> List[ShutdownReport]

Graceful multi-worker shutdown.

Broadcasts the stopping flag to every worker, closes every worker's listener socket, waits up to timeout_ms for workers to drain in-flight work, then joins. Returns one ShutdownReport per worker — best-effort counts based on whether the worker joined inside the timeout.

Each worker publishes its live-connection count to a shared per-worker stats cell on every reactor iteration (release store), so once the worker joins the Scheduler reads a real in_flight_at_deadline (acquire load) instead of the old fabricated 0 / 1. Those remaining connections are force-closed during the worker's graceful-shutdown pass, so timed_out mirrors that count. drained stays a coarse budget signal (1 when a positive drain budget was given, 0 on a hard cut): the reactor does not count naturally-completed connections, so an exact "finished vs cut" split is not available in this force-close model.

timeout_ms bounds the wait. A worker whose thread has not returned by then (a handler that will not finish) is detached and reported with drained == 0; its context, stats cell and listeners (its own, not the joined workers'), the shared listener (opt-out shape), and the shared stop flag, are left allocated, since the thread may still be using them. timeout_ms <= 0 is a hard stop: no drain window, and every worker is joined, as shutdown() does.

Args: timeout_ms: Max ms to wait for the workers to drain.

Functions

`default_worker_count`

def default_worker_count() -> Int

Sensible default worker count: num_cpus().

For IO-bound HTTP plaintext the best throughput is usually num_cpus workers; CPU-heavy handlers may prefer num_cpus // 2 to leave headroom for the kernel network stack.