tur/reactor

(defopaque EventSourceId :int)

(defopaque Reactor :ptr<void> :linear)

(def READ 1)

event mask bit: fd is readable or an accept is ready.

Since: Phase R2

(def WRITE 2)

event mask bit: fd is writable or a connect has completed.

Since: Phase R2

(def ERROR 4)

event mask bit: error condition on the fd. Always reported; cannot be masked out at registration time.

Since: Phase R2

(def HUP 8)

event mask bit: the peer hung up or EOF was reached.

Since: Phase R2

(reactor-new)

create a new reactor for the current thread.

The underlying I/O backend is chosen at compile time: epoll on Linux, kqueue on macOS/BSD. Returns nil on WASM or if the backend cannot be initialised.

(let [r (reactor-new)]
  (reactor-run r)
  (reactor-free r))

Since: Phase R1

(reactor-free [r :Reactor])

free a reactor and unregister all attached sources.

(reactor-free r)

Since: Phase R1

(reactor-add-fd [^borrow r :Reactor fd :int events :int ^fat cb :(fn [int int ptr<void>] nil) user-data :ptr<void>])

watch a file descriptor for I/O events.

Returns a source id (>= 0) that uniquely identifies this registration. Pass the id to reactor-modify-fd or reactor-remove to change or cancel the watch. Returns -1 on error.

The callback is invoked as (cb source-id events user-data) each time the fd becomes ready. Events is a bitmask of READ, WRITE, ERROR, HUP. Callbacks fire on the thread that calls reactor-poll or reactor-run.

(reactor-add-fd r listen-fd READ
  (fn [id events user]
    (handle-accept listen-fd))
  nil)

Since: Phase R2

(reactor-modify-fd [^borrow r :Reactor id :EventSourceId new-events :int])

change the watched event mask for a source.

(reactor-modify-fd r source-id WRITE)

Since: Phase R2

(reactor-remove [^borrow r :Reactor id :EventSourceId])

unregister a source by id.

After this call the fd is no longer watched and the callback will not fire again. The fd itself is not closed.

(reactor-remove r source-id)

Since: Phase R2

(reactor-add-timer [^borrow r :Reactor delay-ms :int ^fat cb :(fn [int ptr<void>] nil) user-data :ptr<void>])

register a one-shot timer that fires once.

The callback fires after delay-ms milliseconds on the thread that calls reactor-poll or reactor-run. After firing the source is automatically deactivated; it does not need to be removed manually.

The callback is invoked as (cb source-id user-data).

(reactor-add-timer r 500
  (fn [id user] :nil (println "fired"))
  nil)

Since: Phase R4

(reactor-add-interval [^borrow r :Reactor first-ms :int interval-ms :int ^fat cb :(fn [int ptr<void>] nil) user-data :ptr<void>])

register a repeating interval timer.

The callback fires first after first-ms milliseconds, then every interval-ms milliseconds thereafter. The source remains active until reactor-remove is called with its id.

The callback is invoked as (cb source-id user-data) on each tick.

(reactor-add-interval r 0 1000
  (fn [id user] :nil (println "tick"))
  nil)

Since: Phase R4

(reactor-add-signal [^borrow r :Reactor signum :int ^fat cb :(fn [int int ptr<void>] nil) user-data :ptr<void>])

watch an OS signal via the reactor.

Uses signalfd on Linux and a self-pipe + sigaction on macOS/BSD. The signal is masked from normal delivery while the source is active; calling reactor-remove restores the previous signal disposition. Returns -1 on WASM or if registration fails.

The callback is invoked as (cb source-id signum user-data).

(reactor-add-signal r 2
  (fn [id sig user] :nil (println "got SIGINT"))
  nil)

Since: Phase R5

(reactor-add-chan [A])

watch a channel for one incoming value (one-shot).

Fires at most once: when a value is available the reactor dequeues it, the callback receives (source-id, value, user-data), and the source is deactivated. Re-register from the callback for a persistent channel watcher.

The channel is borrowed, but the reactor goes on reading it after this call returns: keep it alive -- do not chan-free it -- until the watch has fired, been removed with reactor-remove, or the reactor has been freed.

Delivery latency: the channel is checked on every reactor-poll / reactor-run iteration, and a channel that already holds a value never lets the poll block, so a value sent before the watch is delivered on the next poll. For prompt wakeup after a chan-send made once the watch is live, call reactor-wake from the sending thread.

The value reaches cb as the channel's carrier word, typed :int. For a (Chan int) that is the int itself. For another element type, leave the slot untyped and convert in the body: (:: v cstr) for a word-sized element, (bits->float v) for a float. Do not annotate the slot with the element type. reactor.c calls cb as (env, int64, int64, int64): a v : float slot would read a register it never loaded, so it is refused at compile time; a v : cstr slot compiles and runs on x86-64 / arm64, but the boxed callback's C type then differs from the one it is called through (a -fsanitize=function trap). int and ptr<void> slots are exact.

(reactor-add-chan r stop-chan
  (fn [id v user] :nil (reactor-stop r))
  nil)

Since: Phase R6

(reactor-add-async-chan [A])

reactor-add-chan for an AsyncChan.

The same one-shot watch over the same ChanBlock layout; only the handle type differs. The keep-alive rule and the carrier-word value are the same as reactor-add-chan's: do not async-chan-free the channel while the watch is live.

(reactor-add-async-chan r jobs
  (fn [id job user] :nil (run-job job))
  nil)

Since: Phase R6

(reactor-poll [^borrow r :Reactor timeout-ms :int])

poll for events once, dispatching ready callbacks.

Blocks up to timeout-ms milliseconds waiting for events. Pass -1 to block indefinitely; pass 0 for a non-blocking check. Returns after all callbacks for this batch have been dispatched.

(reactor-poll r 100)  ; => 0..n

Since: Phase R1

(reactor-run [^borrow r :Reactor])

run the event loop until stopped or all sources removed.

Calls reactor-poll in a loop until reactor-stop is called from a callback or another thread, or until no registered sources remain.

(reactor-run r)

Since: Phase R1

(reactor-stop [^borrow r :Reactor])

ask reactor-run to return after the current dispatch round.

Thread-safe: safe to call from a callback inside reactor-run or from another OS thread. reactor-run will return once the current batch of callbacks is complete.

(reactor-stop r)

Since: Phase R1

(reactor-wake [^borrow r :Reactor])

interrupt a reactor blocked in reactor-poll or reactor-run.

Thread-safe: causes the blocking poll to return immediately so the calling thread can check external state (e.g. a new fd to register) before re-entering the loop.

(reactor-wake r)

Since: Phase R1

(defopaque LocalFiberGroup :ptr<void> :linear)

a linear handle to a fiber group bound to one reactor.

(local-fiber-group-new [^borrow r :Reactor])

create a fiber group bound to a reactor.

The group borrows the reactor; do not free the reactor while the group is alive. One group is bound to one reactor for its lifetime and is not thread-safe (same rule as the reactor itself).

(let [r (reactor-new)
      g (local-fiber-group-new r)]
  (reactor-run-fibers g)
  (local-fiber-group-free g)
  (reactor-free r))

Since: Phase F2

(local-fiber-group-free [g :LocalFiberGroup])

free a fiber group and cancel any parked fibers.

Any fiber still parked is cancelled: its park source is removed from the bound reactor and its stack is freed without unwinding (no cleanup hooks). The bound reactor is not freed.

(local-fiber-group-free g)

Since: Phase F2

(local-spawn [^borrow g :LocalFiberGroup ^fat body :(fn [ptr<void>] nil) user-data :ptr<void>])

spawn a fiber into a group.

The fiber first runs on the next reactor-run-fibers tick. The body is a closure taking the user-data pointer; its return value is ignored.

(local-spawn g
  (fn [user] :nil (println "hello from a fiber"))
  nil)

Since: Phase F3

(reactor-run-fibers [^borrow g :LocalFiberGroup])

drive the reactor and pump the group's fibers.

Runs until either the group is empty and the bound reactor has no remaining sources, or reactor-stop is called on the bound reactor. Resumed fibers run until they park or return; parked fibers wake when their reactor source fires.

Calling this re-entrantly from inside a fiber of the same group is a hard error and returns -1 immediately (it would deadlock the pump).

(local-spawn g (fn [u] :nil (println "hi")) nil)
(reactor-run-fibers g)  ; => 1

Since: Phase F3

(local-park-fd [^borrow g :LocalFiberGroup fd :int events :int timeout-ms :int])

park the running fiber until an fd is ready.

Registers fd on the bound reactor for events (a READ/WRITE bitmask) and suspends the current fiber. The fiber resumes when the reactor fires the source; the source is removed before it runs again (one-shot). If timeout-ms is >= 0 and elapses first, the fiber resumes with -2.

Must be called from inside a fiber spawned into g. Outside of one it returns -1 and leaves the caller running.

(local-park-fd g read-fd READ -1)  ; => 1 when readable

Since: Phase F4

(local-park-chan [A])

park the running fiber until a channel has a value.

Registers ch on the bound reactor (one-shot) and suspends the current fiber. The fiber resumes with the received value when one is available. As with reactor-add-chan, prompt same-thread delivery requires the sender to call reactor-wake after chan-send (the driver otherwise blocks in poll).

The channel is borrowed, so the caller keeps the handle; keep it alive until the fiber has resumed. Must be called from inside a fiber spawned into g. Outside of one it returns (none) and leaves the caller running.

(match (local-park-chan g ch)
  (Some v) (handle v)
  (None)   (not-in-a-fiber))

Since: Phase F5

(local-park-async-chan [A])

local-park-chan for an AsyncChan.

The same park over the same ChanBlock layout; only the handle type differs.

(local-park-async-chan g jobs)  ; => (some job)

Since: Phase F5