diff --git a/ffi.nim b/ffi.nim index a95898f..c268b8f 100644 --- a/ffi.nim +++ b/ffi.nim @@ -2,10 +2,13 @@ import std/[atomics, sysatomics, tables] import chronos, chronicles import ffi/internal/[ffi_library, ffi_macro], - ffi/[alloc, ffi_types, ffi_context, ffi_context_pool, ffi_thread_request, cbor_serial] + ffi/[ + alloc, ffi_types, ffi_events, ffi_context, ffi_context_pool, ffi_thread_request, + cbor_serial, + ] export atomics, sysatomics, tables export chronos, chronicles export - atomics, alloc, ffi_library, ffi_macro, ffi_types, ffi_context, ffi_context_pool, - ffi_thread_request, cbor_serial + atomics, alloc, ffi_library, ffi_macro, ffi_types, ffi_events, ffi_context, + ffi_context_pool, ffi_thread_request, cbor_serial diff --git a/ffi/ffi_events.nim b/ffi/ffi_events.nim new file mode 100644 index 0000000..cfa1b02 --- /dev/null +++ b/ffi/ffi_events.nim @@ -0,0 +1,151 @@ +## Multi-listener registry primitive for FFI library-initiated events. +## +## Each event name maps to a `seq` of listeners; the empty event name `""` +## is the wildcard channel and receives every dispatched event in addition +## to its own per-name subscribers. +## +## This module ships the data structure only. Dispatch templates that +## consume the registry, plus the FFIContext wiring that exposes the +## registry to {.ffi.}-generated code, land in follow-up PRs. The +## registry is thread-safe via an embedded `Lock`; the future dispatch +## path acquires the lock only long enough to copy out the listener +## slice for the event being dispatched, so re-entrant add/remove from +## within a handler is deadlock-free by construction. + +import std/[locks, tables] +import ./ffi_types + +# --------------------------------------------------------------------------- +# Registry types +# --------------------------------------------------------------------------- + +type + FFIEventListener* = object + id*: uint64 + callback*: FFICallBack + userData*: pointer + + FFIEventRegistry* = object + ## Per-context multi-listener registry. `lock` guards every mutation; + ## readers (dispatch path) acquire it only long enough to copy out the + ## listener slice for the event being dispatched. + lock*: Lock + nextId*: uint64 + ## Monotonic id source. 0 is reserved as "invalid"; ids start at 1. + byEvent*: Table[string, seq[FFIEventListener]] + wildcard*: seq[FFIEventListener] + +const WildcardEventName* = "" + ## Empty string registers a wildcard listener that receives every event. + +# --------------------------------------------------------------------------- +# Registry lifecycle and mutation +# --------------------------------------------------------------------------- + +proc initEventRegistry*(reg: var FFIEventRegistry) = + ## Must be called exactly once on the owning thread before the registry + ## is shared. The embedded `Lock` wraps a platform primitive that cannot + ## be safely double-initialised, so concurrent callers would hit UB at + ## the OS layer — the lock itself can't defend against its own init. + reg.lock.initLock() + reg.nextId = 0'u64 + reg.byEvent = initTable[string, seq[FFIEventListener]]() + reg.wildcard.setLen(0) + +proc deinitEventRegistry*(reg: var FFIEventRegistry) = + ## Mirror of `initEventRegistry`: must be called exactly once, by the + ## same thread that owns the registry, after all other threads have + ## stopped using it. `deinitLock` on a platform primitive that any + ## thread might still be holding or about to acquire is UB at the OS + ## layer. + reg.lock.deinitLock() + reg.byEvent.clear() + reg.wildcard.setLen(0) + +proc addEventListener*( + reg: var FFIEventRegistry, + eventName: string, + callback: FFICallBack, + userData: pointer, +): uint64 {.raises: [].} = + ## Registers `callback` for `eventName` and returns the listener's stable + ## id (always non-zero on success). `eventName == ""` registers a wildcard + ## listener that receives every dispatched event. Returns 0 if `callback` + ## is nil — the only documented failure mode. + if callback.isNil(): + return 0 + + var assigned: uint64 = 0 + + withLock reg.lock: + reg.nextId.inc() + assigned = reg.nextId + let listener = + FFIEventListener(id: assigned, callback: callback, userData: userData) + if eventName.len == 0: + reg.wildcard.add(listener) + else: + reg.byEvent.mgetOrPut(eventName, @[]).add(listener) + return assigned + +proc removeEventListener*(reg: var FFIEventRegistry, id: uint64): bool {.raises: [].} = + ## Removes the listener with `id`. Returns true on success, false if no + ## listener with that id exists. Safe to call from inside a dispatch: + ## the in-flight snapshot still delivers exactly once to the listener + ## being removed. + if id == 0'u64: + return false + + var removed = false + + withLock reg.lock: + for i in 0 ..< reg.wildcard.len: + if reg.wildcard[i].id == id: + reg.wildcard.delete(i) + removed = true + break + if not removed: + var emptyKey = "" + var prune = false + for key, listeners in reg.byEvent.mpairs: + var idx = -1 + for i in 0 ..< listeners.len: + if listeners[i].id == id: + idx = i + break + if idx >= 0: + listeners.delete(idx) + removed = true + if listeners.len == 0: + emptyKey = key + prune = true + break + if prune: + reg.byEvent.del(emptyKey) + return removed + +proc removeAllEventListeners*(reg: var FFIEventRegistry) {.raises: [].} = + ## Drops every registered listener (per-event and wildcard). Does not + ## reset the listener-id counter — subsequent `addEventListener` calls + ## still return strictly increasing ids. + withLock reg.lock: + reg.wildcard.setLen(0) + reg.byEvent.clear() + +proc snapshotListeners*( + reg: var FFIEventRegistry, eventName: string +): seq[FFIEventListener] {.raises: [].} = + ## Returns a copy of the listener slice for `eventName`, plus every + ## wildcard listener. The copy is what makes re-entrant add/remove from + ## inside a handler deadlock-free: dispatch holds the lock only for the + ## duration of the copy, then iterates the copy outside the lock. + var snap: seq[FFIEventListener] = @[] + withLock reg.lock: + if eventName.len > 0: + # `getOrDefault` returns an empty seq when the key is absent — + # avoids the raising `[]` operator path. + for l in reg.byEvent.getOrDefault(eventName): + snap.add(l) + for l in reg.wildcard: + snap.add(l) + return snap diff --git a/tests/unit/test_event_listener.nim b/tests/unit/test_event_listener.nim new file mode 100644 index 0000000..432b6ba --- /dev/null +++ b/tests/unit/test_event_listener.nim @@ -0,0 +1,181 @@ +## Unit tests for the `FFIEventRegistry` primitive — the multi-listener +## data structure that will back `_add_event_listener` / +## `_remove_event_listener` once the dispatch wiring lands. +## +## These tests exercise the registry directly (no FFI thread, no dispatch +## templates) so they stay fast and pin down the registry's mutation and +## snapshot semantics in isolation. + +import std/locks +import unittest2 +import ffi + +# --------------------------------------------------------------------------- +# Tiny helpers — a thread-safe sink each listener writes into so we can +# assert which callbacks would fire and in what order once dispatch lands. +# Today only `tagCb`'s presence is exercised; the recorder is also used to +# make sure listener bookkeeping doesn't accidentally invoke callbacks. +# --------------------------------------------------------------------------- + +type Recorder = object + lock: Lock + hits: seq[string] # tag captured from `userData` per invocation + retCodes: seq[cint] + payloads: seq[string] + +proc initRecorder(r: var Recorder) = + r.lock.initLock() + +proc deinitRecorder(r: var Recorder) = + r.lock.deinitLock() + +proc record(r: var Recorder, tag: string, retCode: cint, payload: string) = + acquire(r.lock) + r.hits.add(tag) + r.retCodes.add(retCode) + r.payloads.add(payload) + release(r.lock) + +# Each listener is identified by a `Tag` passed through `userData`. +type Tag = object + name: string + rec: ptr Recorder + +proc tagCb( + retCode: cint, msg: ptr cchar, len: csize_t, userData: pointer +) {.cdecl, gcsafe, raises: [].} = + let t = cast[ptr Tag](userData) + var payload = newString(int(len)) + if len > 0 and not msg.isNil: + copyMem(addr payload[0], msg, int(len)) + record(t[].rec[], t[].name, retCode, payload) + +# --------------------------------------------------------------------------- +# Tests +# --------------------------------------------------------------------------- + +suite "FFIEventRegistry mutation": + test "addEventListener assigns monotonically increasing non-zero ids": + var reg: FFIEventRegistry + initEventRegistry(reg) + defer: + deinitEventRegistry(reg) + var rec: Recorder + initRecorder(rec) + defer: + deinitRecorder(rec) + var t = Tag(name: "a", rec: addr rec) + + let id1 = addEventListener(reg, "evt", tagCb, addr t) + let id2 = addEventListener(reg, "evt", tagCb, addr t) + let id3 = addEventListener(reg, "", tagCb, addr t) + check id1 == 1'u64 + check id2 == 2'u64 + check id3 == 3'u64 + + test "addEventListener returns 0 when callback is nil": + var reg: FFIEventRegistry + initEventRegistry(reg) + defer: + deinitEventRegistry(reg) + let id = addEventListener(reg, "evt", nil, nil) + check id == 0'u64 + + test "removeEventListener returns false for unknown ids": + var reg: FFIEventRegistry + initEventRegistry(reg) + defer: + deinitEventRegistry(reg) + check not removeEventListener(reg, 0'u64) + check not removeEventListener(reg, 99'u64) + + test "removeEventListener removes from per-event seq and wildcard": + var reg: FFIEventRegistry + initEventRegistry(reg) + defer: + deinitEventRegistry(reg) + var rec: Recorder + initRecorder(rec) + defer: + deinitRecorder(rec) + var t = Tag(name: "a", rec: addr rec) + + let id1 = addEventListener(reg, "evt", tagCb, addr t) + let id2 = addEventListener(reg, "", tagCb, addr t) + + check removeEventListener(reg, id1) + check removeEventListener(reg, id2) + # Second remove of the same id is a no-op. + check not removeEventListener(reg, id1) + + let snap = snapshotListeners(reg, "evt") + check snap.len == 0 + +suite "FFIEventRegistry snapshot semantics": + test "snapshot includes both per-event listeners and wildcards": + var reg: FFIEventRegistry + initEventRegistry(reg) + defer: + deinitEventRegistry(reg) + var rec: Recorder + initRecorder(rec) + defer: + deinitRecorder(rec) + var a = Tag(name: "a", rec: addr rec) + var b = Tag(name: "b", rec: addr rec) + var c = Tag(name: "c", rec: addr rec) + + discard addEventListener(reg, "evt", tagCb, addr a) + discard addEventListener(reg, "other", tagCb, addr b) + discard addEventListener(reg, "", tagCb, addr c) + + let snapEvt = snapshotListeners(reg, "evt") + check snapEvt.len == 2 # listener for "evt" + wildcard + + let snapOther = snapshotListeners(reg, "other") + check snapOther.len == 2 # listener for "other" + wildcard + + let snapUnknown = snapshotListeners(reg, "no-subscriber") + check snapUnknown.len == 1 # only the wildcard + + test "snapshot is a copy: post-snapshot mutation does not affect it": + var reg: FFIEventRegistry + initEventRegistry(reg) + defer: + deinitEventRegistry(reg) + var rec: Recorder + initRecorder(rec) + defer: + deinitRecorder(rec) + var t = Tag(name: "a", rec: addr rec) + + let id1 = addEventListener(reg, "evt", tagCb, addr t) + let snap = snapshotListeners(reg, "evt") + check snap.len == 1 + + # Mutating the registry after the snapshot must not retroactively + # shrink or grow the snapshot we already captured. + check removeEventListener(reg, id1) + discard addEventListener(reg, "evt", tagCb, addr t) + check snap.len == 1 + check snap[0].id == id1 + +suite "removeAllEventListeners": + test "drops every listener (per-event and wildcard)": + var reg: FFIEventRegistry + initEventRegistry(reg) + defer: + deinitEventRegistry(reg) + var rec: Recorder + initRecorder(rec) + defer: + deinitRecorder(rec) + var a = Tag(name: "a", rec: addr rec) + var b = Tag(name: "b", rec: addr rec) + + discard addEventListener(reg, "evt", tagCb, addr a) + discard addEventListener(reg, WildcardEventName, tagCb, addr b) + removeAllEventListeners(reg) + + check snapshotListeners(reg, "evt").len == 0 + check snapshotListeners(reg, WildcardEventName).len == 0