From ad35dfd782ea08bace394e18f285fc216cf7f880 Mon Sep 17 00:00:00 2001 From: Richard Thanki Date: Sun, 16 Nov 2025 03:04:51 +0000 Subject: [PATCH 1/8] make op.cond less eager --- src/fibers/op.lua | 27 +++++++++++++++++---------- tests/test_cond.lua | 4 ++-- 2 files changed, 19 insertions(+), 12 deletions(-) diff --git a/src/fibers/op.lua b/src/fibers/op.lua index c51599b1..3f614856 100644 --- a/src/fibers/op.lua +++ b/src/fibers/op.lua @@ -219,40 +219,47 @@ end local function new_cond(opts) local state = { triggered = false, - waiters = {}, -- optional + waiters = {}, -- list of { suspension = ..., wrap = ... } abort_fn = opts and opts.abort_fn or nil, } local function wait_op() assert(not state.abort_fn, "abort-only cond has no wait_op") + local function try() return state.triggered end + local function block(suspension, wrap_fn) if state.triggered then + -- Already triggered: complete immediately via the scheduler. suspension:complete(wrap_fn) else - state.waiters[#state.waiters + 1] = - suspension:complete_task(wrap_fn) + -- Record this suspension + wrap for later signalling. + state.waiters[#state.waiters + 1] = { + suspension = suspension, + wrap = wrap_fn, + } end end + return new_primitive(nil, try, block) end local function signal() if state.triggered then return end state.triggered = true + + -- Complete all recorded waiters via the scheduler. for i = 1, #state.waiters do - local task = state.waiters[i] + local remote = state.waiters[i] state.waiters[i] = nil - if task - and task.suspension - and task.suspension:waiting() - then - -- Run the completion *now*, in this turn. - task:run() + + if remote and remote.suspension and remote.suspension:waiting() then + remote.suspension:complete(remote.wrap) end end + if state.abort_fn then pcall(state.abort_fn) end diff --git a/tests/test_cond.lua b/tests/test_cond.lua index 8ffd57e4..888a6290 100644 --- a/tests/test_cond.lua +++ b/tests/test_cond.lua @@ -22,9 +22,9 @@ runtime.spawn(function() end) assert(equal(log, {})) runtime.current_scheduler:run() --- assert(equal(log, { 'a', 'c', 'd' })) +assert(equal(log, { 'a', 'c', 'd' })) runtime.current_scheduler:run() -assert(equal(log, { 'a', 'c', 'b', 'd' })) +assert(equal(log, { 'a', 'c', 'd', 'b' })) runtime.spawn(function() local fiber_count = 1e3 From aef12282a44cd07619208f2064ea1cfb26f56d0f Mon Sep 17 00:00:00 2001 From: Richard Thanki Date: Sun, 16 Nov 2025 04:36:10 +0000 Subject: [PATCH 2/8] adds constants to syscall --- src/fibers/utils/syscall.lua | 2 ++ 1 file changed, 2 insertions(+) diff --git a/src/fibers/utils/syscall.lua b/src/fibers/utils/syscall.lua index 8d027272..d684f81b 100644 --- a/src/fibers/utils/syscall.lua +++ b/src/fibers/utils/syscall.lua @@ -71,6 +71,8 @@ M.ECONNRESET = p_errno.ECONNRESET M.ECONNREFUSED = p_errno.ECONNREFUSED M.ENETUNREACH = p_errno.ENETUNREACH M.EHOSTUNREACH = p_errno.EHOSTUNREACH +M.EBADF = p_errno.EBADF +M.ENOENT = p_errno.ENOENT M.SIGKILL = p_signal.SIGKILL M.SIGTERM = p_signal.SIGTERM From 9e4e434ac59ac352a4a20f75af0271e92093236c Mon Sep 17 00:00:00 2001 From: Richard Thanki Date: Sun, 16 Nov 2025 05:30:04 +0000 Subject: [PATCH 3/8] pcall free op --- src/fibers/op.lua | 264 +++++++++++++++++++++++----------------------- tests/test_op.lua | 142 +++++-------------------- 2 files changed, 158 insertions(+), 248 deletions(-) diff --git a/src/fibers/op.lua b/src/fibers/op.lua index 3f614856..b2f50e9a 100644 --- a/src/fibers/op.lua +++ b/src/fibers/op.lua @@ -1,37 +1,22 @@ --- fibers.op module --- Provides Concurrent ML style operations for managing concurrency. +-- Concurrent ML style operations for managing concurrency. +-- -- Events are CML-style: primitive leaves, choices, guards, with_nack, --- wraps, and an extra abort combinator (on_abort). +-- wraps, and an abort combinator (on_abort). -- -- Core event AST kinds: --- prim : primitive leaf { try_fn, block_fn, wrap_fn } --- choice : non-empty list of events --- guard : delayed event builder (run once per sync) --- with_nack : CML-style nack combinator --- wrap : post-commit mapper (composed at compile time) --- abort : attach abort handler to an event (run if this arm loses) --- wrap_handler : CML-style exception handler (see wrap_handler below) --- --- Semantics sketch --- ---------------- --- We keep CML-style semantics for with_nack: --- - with_nack g gets a nack event that becomes enabled iff the --- *entire* resulting event loses in an enclosing choice. --- - nested with_nack behaves correctly: outer nacks only fire when --- the outer event loses, not when internal subchoices resolve. --- --- `on_abort(ev, f)` is implemented in terms of the same "nack" machinery: --- - each abort scope behaves like a nack-cond whose signal() runs f(). --- - after a choice commits, we figure out which conds are associated --- exclusively with losing arms and signal those once. +-- prim : primitive leaf { try_fn, block_fn, wrap_fn } +-- choice : non-empty list of events +-- guard : delayed event builder (run once per sync) +-- with_nack : CML-style nack combinator +-- wrap : post-commit mapper (composed at compile time) +-- abort : attach abort handler to an event (run if this arm loses) -- --- We also provide: --- - bracket(acquire, release, use): RAII-style resource protocol. --- - wrap_handler(ev, h): CML-style wrapHandler (exn -> event). --- - finally(ev, cleanup): derived "always run cleanup" combinator. --- - else_next_turn(ev, fallback_ev): biased choice; prefer ev, but --- if it doesn't commit "by next turn", abort it cleanly and run --- fallback_ev (in a separate sync). +-- Important design note: +-- This module is *exception-neutral*. It does not interpret Lua +-- errors as part of event semantics. Any uncaught error in a wrap +-- or primitive is treated as a bug and will be surfaced by the +-- surrounding scope / fibre machinery. local runtime = require 'fibers.runtime' @@ -90,29 +75,36 @@ function CompleteTask:run() end end --- A CompleteTask can be cancelled, completing with an error. +-- A CompleteTask can be cancelled. In the non-exceptional model, this +-- completes the suspension with a special "cancel" wrap that returns +-- a tagged result (false, reason) rather than raising. function CompleteTask:cancel(reason) if self.suspension:waiting() then - self.suspension:complete(error, reason or 'cancelled') + local msg = reason or 'cancelled' + local function cancelled_wrap() + -- Convention: (ok:boolean, value_or_reason:any) + return false, msg + end + self.suspension:complete(cancelled_wrap) end end ---------------------------------------------------------------------- -- Event type (unifies primitive and composite events) -- --- kind = 'prim' : { try_fn, block_fn, wrap_fn } --- kind = 'choice' : { events = { Event, ... } } --- kind = 'guard' : { builder = function() -> Event } --- kind = 'with_nack' : { builder = function(nack_ev) -> Event } --- kind = 'wrap' : { inner = Event, wrap_fn = f } --- kind = 'abort' : { inner = Event, abort_fn = f } --- kind = 'wrap_handler' : { inner = Event, handler = function(ex) -> Event } +-- kind = 'prim' : { try_fn, block_fn, wrap_fn } +-- kind = 'choice' : { events = { Event, ... } } +-- kind = 'guard' : { builder = function() -> Event } +-- kind = 'with_nack' : { builder = function(nack_ev) -> Event } +-- kind = 'wrap' : { inner = Event, wrap_fn = f } +-- kind = 'abort' : { inner = Event, abort_fn = f } ---------------------------------------------------------------------- local Event = {} Event.__index = Event --- forward declaration so compile_event can call perform +-- forward declaration so compile_event can call perform if needed in +-- future extension; currently perform does not use exceptions. local perform -- Primitive event (leaf). @@ -166,7 +158,6 @@ local function always(...) return new_primitive(nil, try, block) end - local function never() -- An event that never becomes ready return new_primitive(nil, @@ -174,15 +165,6 @@ local function never() function() end) end --- function Event:or_else(fallback_thunk) --- return choice( --- self, --- next_turn_op():wrap(function() --- return fallback_thunk() --- end) --- ) --- end - -- Wrap event with a post-processing function f (commit phase). -- This is another node in the tree; composed at compile time. function Event:wrap(f) @@ -202,16 +184,6 @@ function Event:on_abort(f) ) end --- Attach an exception handler for post-synchronisation --- actions. h(ex) must return a replacement event to synchronise on. -function Event:wrap_handler(handler) - assert(type(handler) == 'function', "wrap_handler expects a function") - return setmetatable( - { kind = 'wrap_handler', inner = self, handler = handler }, - Event - ) -end - ---------------------------------------------------------------------- -- Simple one-shot condition primitive (used for with_nack; also exported) ---------------------------------------------------------------------- @@ -255,7 +227,10 @@ local function new_cond(opts) local remote = state.waiters[i] state.waiters[i] = nil - if remote and remote.suspension and remote.suspension:waiting() then + if remote + and remote.suspension + and remote.suspension:waiting() + then remote.suspension:complete(remote.wrap) end end @@ -284,28 +259,24 @@ end -- -- Semantics: -- - Each with_nack or abort node adds a cond to the nacks list. --- - Each wrap_handler node adds a handler to the handlers list. --- - At the leaf, we build a final wrap function that: --- * runs the normal wrap chain --- * then applies the wrap_handler chain (innermost first) using pcall. +-- - wrap nodes compose their functions into the final wrap. ---------------------------------------------------------------------- -local function compile_event(ev, outer_wrap, out, nacks, handlers) +local function compile_event(ev, outer_wrap, out, nacks) out = out or {} outer_wrap = outer_wrap or id_wrap nacks = nacks or {} - handlers = handlers or {} local kind = ev.kind if kind == 'choice' then for _, sub in ipairs(ev.events) do - compile_event(sub, outer_wrap, out, nacks, handlers) + compile_event(sub, outer_wrap, out, nacks) end elseif kind == 'guard' then local inner = ev.builder() - compile_event(inner, outer_wrap, out, nacks, handlers) + compile_event(inner, outer_wrap, out, nacks) elseif kind == 'with_nack' then local cond = new_cond() @@ -314,50 +285,26 @@ local function compile_event(ev, outer_wrap, out, nacks, handlers) local child_nacks = { unpack(nacks) } child_nacks[#child_nacks + 1] = cond - compile_event(inner, outer_wrap, out, child_nacks, handlers) + compile_event(inner, outer_wrap, out, child_nacks) elseif kind == 'wrap' then local f = ev.wrap_fn local new_outer = function(...) return outer_wrap(f(...)) end - compile_event(ev.inner, new_outer, out, nacks, handlers) + compile_event(ev.inner, new_outer, out, nacks) elseif kind == 'abort' then local cond = new_cond{ abort_fn = ev.abort_fn } local child_nacks = { unpack(nacks) } child_nacks[#child_nacks + 1] = cond - compile_event(ev.inner, outer_wrap, out, child_nacks, handlers) - - elseif kind == 'wrap_handler' then - -- Accumulate handlers; innermost handler should see the exception first. - local child_handlers = { unpack(handlers) } - child_handlers[#child_handlers + 1] = ev.handler - compile_event(ev.inner, outer_wrap, out, nacks, child_handlers) + compile_event(ev.inner, outer_wrap, out, child_nacks) else -- 'prim' local function wrapped(...) - local function core(...) - return outer_wrap(ev.wrap_fn(...)) - end - - local f = core - if #handlers > 0 then - for i = #handlers, 1, -1 do - local h = handlers[i] - local prev = f - f = function(...) - local res = pack(pcall(prev, ...)) - if res[1] then - return unpack(res, 2, res.n) - end - local ex = res[2] - local hev = h(ex) - return perform(hev) - end - end - end - return f(...) + -- No exception machinery here; any Lua error is treated + -- as a bug and handled by the surrounding scope/fibre. + return outer_wrap(ev.wrap_fn(...)) end out[#out + 1] = { @@ -376,7 +323,6 @@ end ---------------------------------------------------------------------- -- Signal all conds that belong exclusively to losing arms. --- This is the original CML-style logic: -- - Build set of nacks on the winner path. -- - For each loser leaf, signal any nacks not in the winner set. -- - Each cond object is responsible for idempotence. @@ -435,6 +381,10 @@ local function apply_wrap(wrap, retval) return wrap(unpack(retval, 2, retval.n)) end +---------------------------------------------------------------------- +-- or_else: biased, non-blocking choice +---------------------------------------------------------------------- + function Event:or_else(fallback_thunk) assert(type(fallback_thunk) == "function", "or_else expects a function") @@ -476,10 +426,12 @@ end ---------------------------------------------------------------------- -- Perform this event (primitive or composite), possibly blocking. +-- Any Lua error raised during wraps or primitives is not caught here; +-- it will abort the current fibre and be handled by the scope layer. perform = function(ev) local leaves = compile_event(ev) - -- Fast path: try once using the same semantics as fast_commit(). + -- Fast path: non-blocking attempt. local idx, retval = try_ready(leaves) if idx then trigger_nacks(leaves, idx) @@ -490,7 +442,7 @@ perform = function(ev) local suspended = pack(runtime.suspend(block_choice_op, leaves)) local wrap = suspended[1] - -- Identify winning leaf by its wrap function. + -- Identify winning leaf by its wrap function, if any. local winner_index for i, leaf in ipairs(leaves) do if leaf.wrap == wrap then @@ -506,37 +458,34 @@ end ---------------------------------------------------------------------- -- finally : (ev, cleanup) -> ev' -- --- cleanup(aborted:boolean, exn:any|nil) +-- cleanup(aborted:boolean) -- -- Semantics: -- * on normal post-sync completion: --- cleanup(false, nil) is called (best-effort, protected). --- * if a post-sync action raises: --- cleanup(true, exn) is called (best-effort) and the same --- exception is re-raised. --- * Exceptions from guard/with_nack builders are not intercepted. +-- cleanup(false) is called (best-effort, protected). +-- * if the event *loses* in a choice (via on_abort): +-- cleanup(true) is called (best-effort, protected). +-- +-- Exceptions from ev's wrap or primitives are not intercepted here; +-- they are handled by the surrounding scope as fibre failures. ---------------------------------------------------------------------- function Event:finally(cleanup) - -- Success path: only runs if the entire event (including any inner - -- wrap_handler handlers) completes without raising. + assert(type(cleanup) == "function", "finally expects a function") + + -- Success path: only runs if this event wins and completes its wraps + -- without raising. local function success_wrap(...) - pcall(cleanup, false, nil) + pcall(cleanup, false) return ... end - local function handler(ex) - -- Error path: run cleanup(true, ex) and rethrow. - return always(true):wrap(function() - pcall(cleanup, true, ex) - error(ex) - end) + -- Abort path: runs if this event participates in a choice and loses. + local function abort_action() + pcall(cleanup, true) end - -- Important: wrap_handler on ev, then a wrap on top. - -- This ensures success_wrap runs only on the success path, - -- and handler is responsible for the error path. - return self:wrap_handler(handler):wrap(success_wrap) + return self:wrap(success_wrap):on_abort(abort_action) end ---------------------------------------------------------------------- @@ -554,36 +503,83 @@ end -- * if the resulting event PARTICIPATES in a choice but LOSES: -- - release(res, true) is called (via on_abort / nack machinery) -- --- This is *purely an Event combinator*; no new fiber is spawned. +-- This combinator does not interpret Lua errors from acquire/use as +-- normal control flow. Any uncaught error there fails the running +-- fibre and is recorded at the scope level. ---------------------------------------------------------------------- local function bracket(acquire, release, use) + assert(type(acquire) == "function", "bracket: acquire must be a function") + assert(type(release) == "function", "bracket: release must be a function") + assert(type(use) == "function", "bracket: use must be a function") + return guard(function() local res = acquire() - local ok, ev = pcall(use, res) - if not ok then - local ex = ev - pcall(release, res, true) - error(ex) - end + -- If use(res) throws, that is a bug; scope machinery will handle it. + local ev = use(res) - local wrapped = ev:finally(function(aborted, _) - pcall(release, res, aborted) + -- Success path: event wins and completes → release(res, false) once. + local wrapped = ev:wrap(function(...) + pcall(release, res, false) + return ... end) + -- Losing path: event participates in a choice but loses → release(res, true) once. return wrapped:on_abort(function() pcall(release, res, true) end) end) end +---------------------------------------------------------------------- +-- Higher-level choice helpers (built entirely from choice + wrap) +---------------------------------------------------------------------- + +local function race(events, on_win) + assert(type(on_win) == "function", "race expects on_win callback") + local wrapped = {} + for i, ev in ipairs(events) do + wrapped[i] = ev:wrap(function(...) + return on_win(i, ...) + end) + end + return choice(unpack(wrapped)) +end + +local function first_ready(events) + return race(events, function(i, ...) + return i, ... + end) +end + +local function named_choice(arms) + -- arms is a map { name = Event, ... } + local events, names = {}, {} + for name, ev in pairs(arms) do + names[#names + 1] = name + events[#events + 1] = ev + end + return race(events, function(i, ...) + return names[i], ... + end) +end + +local function boolean_choice(ev_true, ev_false) + return race({ ev_true, ev_false }, function(i, ...) + if i == 1 then + return true, ... + else + return false, ... + end + end) +end + ---------------------------------------------------------------------- -- Public API ---------------------------------------------------------------------- return { - perform = perform, perform_raw = perform, new_primitive = new_primitive, -- primitive event constructor choice = choice, @@ -594,5 +590,11 @@ return { always = always, never = never, Event = Event, - -- Event instances have methods: wrap, on_abort. + -- Event instances have methods: wrap, on_abort, finally, or_else. + + -- higher-level helpers + race = race, + first_ready = first_ready, + named_choice = named_choice, + boolean_choice = boolean_choice, } diff --git a/tests/test_op.lua b/tests/test_op.lua index 3715feab..befd4ba2 100644 --- a/tests/test_op.lua +++ b/tests/test_op.lua @@ -4,7 +4,7 @@ print("testing: fibers.op") -- look one level up package.path = "../src/?.lua;" .. package.path -local op = require 'fibers.op' +local op = require 'fibers.op' local runtime = require 'fibers.runtime' local perform, choice = require 'fibers.performer'.perform, op.choice @@ -373,144 +373,52 @@ runtime.spawn(function() end -------------------------------------------------------- - -- 8) wrap_handler: exception in post-sync, plus pre-sync error + -- 8) finally: cleanup on success and on abort + -- + -- In the new model: + -- finally(cleanup) is about lifetime: + -- * cleanup(false) on normal success + -- * cleanup(true) if the event participates in a choice and loses + -- It does not intercept Lua errors; those are handled by scopes. -------------------------------------------------------- do - -- 8.1 error in a post-synchronisation wrap is caught and - -- mapped to a recovery event. - do - local handler_called = false - - local base = always(10) - - local ev = base - :wrap_handler(function(ex) - handler_called = true - assert(tostring(ex):match("boom"), - "wrap_handler: unexpected exception value") - return always("recovered") - end) - :wrap(function() - -- post-sync action that fails - error("boom") - end) - - local r = perform(ev) - assert(r == "recovered", - "wrap_handler: expected recovery result") - assert(handler_called, - "wrap_handler: handler was not invoked") - end - - -- 8.2 guard builder error is *not* caught by wrap_handler - do - local g_ev = op.guard(function() - error("builder-fail") - end) - - local handled = g_ev:wrap_handler(function(_) - return always("ignored") - end) - - local ok, err = pcall(function() - perform(handled) - end) - assert(not ok, "wrap_handler: should not catch guard builder errors") - assert(tostring(err):match("builder%-fail"), - "wrap_handler: wrong error propagated for guard builder") - end - - -- 8.3 wrap_handler: nesting order (innermost first) - do - local log = {} - - -- Base event whose post-sync action fails. - local base = always(1):wrap(function() - error("boom-inner") - end) - - -- Inner handler: sees the original exception and rethrows via a new event. - local ev_inner = base:wrap_handler(function(ex) - table.insert(log, "inner:" .. tostring(ex)) - -- Rethrow as a different error so the outer handler can distinguish it. - return always(true):wrap(function() - error("inner-rethrow") - end) - end) - - -- Outer handler: should see the *rethrown* exception, not the original. - local ev = ev_inner:wrap_handler(function(ex) - table.insert(log, "outer:" .. tostring(ex)) - return always("ok") - end) - - local res = perform(ev) - - assert(res == "ok", - "wrap_handler nesting: final result mismatch") - assert(#log == 2, - "wrap_handler nesting: expected two handlers invoked") - - -- Innermost handler must see the original error first. - assert(log[1]:match("^inner:.*boom%-inner"), - "wrap_handler nesting: inner handler did not see original exception first") - - -- Outermost handler must see the rethrown error. - assert(log[2]:match("^outer:.*inner%-rethrow"), - "wrap_handler nesting: outer handler did not see rethrown exception") - end - end - - - -------------------------------------------------------- - -- 9) finally: cleanup on success and on failure - -------------------------------------------------------- - do - -- 9.1 success path: cleanup(false, nil) once, result propagated + -- 8.1 success path: cleanup(false) once, result propagated do local calls = {} local base = always(7) - local ev = base:finally(function(aborted, exn) - calls[#calls + 1] = { aborted = aborted, exn = exn } + local ev = base:finally(function(aborted) + calls[#calls + 1] = aborted end) local r = perform(ev) assert(r == 7, "finally(success): wrong result") assert(#calls == 1, "finally(success): cleanup not called once") - assert(calls[1].aborted == false, + assert(calls[1] == false, "finally(success): aborted should be false") - assert(calls[1].exn == nil, - "finally(success): exn should be nil") end - -- 9.2 failure in post-sync action: cleanup(true, exn), then rethrow + -- 8.2 abort path: event loses in a choice → cleanup(true) do local calls = {} - local base = always(1):wrap(function(_) - -- simulate user post-sync failure - error("post-sync-fail") + -- This event never commits, so in choice it always loses. + local base = never():finally(function(aborted) + calls[#calls + 1] = aborted end) - local ev = base:finally(function(aborted, exn) - calls[#calls + 1] = { aborted = aborted, exn = exn } - end) + local ev = choice(base, always("WIN")) + local r = perform(ev) - local ok, err = pcall(function() - perform(ev) - end) - assert(not ok, "finally(error): expected re-raise of exception") - assert(tostring(err):match("post%-sync%-fail"), - "finally(error): wrong exception propagated") - - assert(#calls == 1, "finally(error): cleanup not called once") - assert(calls[1].aborted == true, - "finally(error): aborted should be true") - assert(calls[1].exn ~= nil, - "finally(error): exn should be non-nil") + assert(r == "WIN", + "finally(abort): wrong winner") + assert(#calls == 1, + "finally(abort): cleanup not called once") + assert(calls[1] == true, + "finally(abort): aborted should be true") end end + print("fibers.op tests: ok") runtime.stop() end) From af8d2ae192433a8d5c657b284b4f5e646aac37c7 Mon Sep 17 00:00:00 2001 From: Richard Thanki Date: Sun, 16 Nov 2025 10:18:43 +0000 Subject: [PATCH 4/8] remove the use of pcall/exception to trigger scope failure --- src/fibers.lua | 18 +-- src/fibers/op.lua | 8 +- src/fibers/runtime.lua | 68 +++++++- src/fibers/scope.lua | 342 +++++++++++++++++++++++------------------ tests/test_scope.lua | 305 +++++++++++++++++++----------------- 5 files changed, 432 insertions(+), 309 deletions(-) diff --git a/src/fibers.lua b/src/fibers.lua index 8a61e1ba..2dbb381b 100644 --- a/src/fibers.lua +++ b/src/fibers.lua @@ -33,23 +33,19 @@ fibers.perform = performer.perform fibers.now = runtime.now --- Run a main function under the scheduler's root scope. --- main_fn :: function(Scope, ...): () +-- main_fn :: function(Scope, ...): ... function fibers.run(main_fn, ...) local root = scope_mod.root() local args = { ... } - root:spawn(function() - -- Run main_fn inside a child scope of the current scope (root). - local res = pack( - pcall(function() - return scope_mod.run(main_fn, unpack(args)) - end) - ) + -- Run main_fn inside a child scope of the root, in its own fibre. + root:spawn(function(s) + local status, err = scope_mod.run(main_fn, unpack(args)) -- In all cases, stop the scheduler so runtime.main() returns. runtime.stop() - -- If the main scope failed, treat as fatal for the process. - if not res[1] then - print(unpack(res, 2, res.n)) + -- Treat non-ok main scope as fatal for the process. + if status ~= "ok" then + print(err) os.exit(255) end end) diff --git a/src/fibers/op.lua b/src/fibers/op.lua index b2f50e9a..8696595c 100644 --- a/src/fibers/op.lua +++ b/src/fibers/op.lua @@ -476,13 +476,13 @@ function Event:finally(cleanup) -- Success path: only runs if this event wins and completes its wraps -- without raising. local function success_wrap(...) - pcall(cleanup, false) + cleanup(false) return ... end -- Abort path: runs if this event participates in a choice and loses. local function abort_action() - pcall(cleanup, true) + cleanup(true) end return self:wrap(success_wrap):on_abort(abort_action) @@ -521,13 +521,13 @@ local function bracket(acquire, release, use) -- Success path: event wins and completes → release(res, false) once. local wrapped = ev:wrap(function(...) - pcall(release, res, false) + release(res, false) return ... end) -- Losing path: event participates in a choice but loses → release(res, true) once. return wrapped:on_abort(function() - pcall(release, res, true) + release(res, true) end) end) end diff --git a/src/fibers/runtime.lua b/src/fibers/runtime.lua index 055351ad..fb1da7cf 100644 --- a/src/fibers/runtime.lua +++ b/src/fibers/runtime.lua @@ -7,6 +7,23 @@ local sched = require 'fibers.sched' +local function id(...) + return ... +end + +-- Queue of uncaught fibre errors to be consumed by supervisors. +local error_queue = {} +local error_waiters = {} + +-- Task object used to wake a fibre waiting for an error. +local WaiterTask = {} +WaiterTask.__index = WaiterTask + +function WaiterTask:run() + -- Resume the waiting fibre with (wrap, fiber, err). + self.waiter:resume(id, self.err_fiber, self.err) +end + local _current_fiber local current_scheduler = sched.new() @@ -48,11 +65,24 @@ function Fiber:resume(wrap, ...) -- current_fiber = saved_current_fiber the KEY bit, we only get here when the coroutine above has yielded, -- but we then pop back in the fiber we previously displaced _current_fiber = saved_current_fiber + if coroutine.status(self.coroutine) == "dead" then + self.alive = false + end if not ok then - print('Error while running fiber: ' .. tostring(err)) - print(debug.traceback(self.coroutine)) - print('fibers history:\n' .. self.traceback) - os.exit(255) + -- Report uncaught error to error consumers. + if #error_waiters > 0 then + local waiter = table.remove(error_waiters, 1) + current_scheduler:schedule(setmetatable({ + waiter = waiter.fiber, + err_fiber = self, + err = err, + }, WaiterTask)) + else + error_queue[#error_queue + 1] = { + fiber = self, + err = err, + } + end end end @@ -98,6 +128,29 @@ local function stop() current_scheduler:stop() end +local function wait_fiber_error() + -- Fast path: if an error is already queued, return it immediately. + if #error_queue > 0 then + local rec = table.remove(error_queue, 1) + return rec.fiber, rec.err + end + + -- Otherwise, we must be in a fibre and suspend until an error arrives. + assert(_current_fiber, "wait_fiber_error must be called from within a fiber") + + local function block_fn(sched, fib) + -- Record this fibre as waiting for an error. When an error + -- arrives, the failing fibre will arrange to schedule a task + -- that resumes this fibre with (fiber, err). + error_waiters[#error_waiters + 1] = { fiber = fib } + end + + local wrap, err_fiber, err = _current_fiber:suspend(block_fn) + -- wrap should be the identity function; ignore it. + return err_fiber, err +end + + --- Runs the main event loop of the current scheduler. -- The scheduler will continue to run tasks and wait for events until stopped. -- @function main @@ -111,9 +164,10 @@ return { current_fiber = current_fiber, -- time and suspension - now = now, - suspend = suspend, - yield = yield, + now = now, + suspend = suspend, + yield = yield, + wait_fiber_error = wait_fiber_error, -- fiber management spawn = spawn, diff --git a/src/fibers/scope.lua b/src/fibers/scope.lua index 9827ace1..bd714edf 100644 --- a/src/fibers/scope.lua +++ b/src/fibers/scope.lua @@ -1,29 +1,39 @@ --- fibers/scope.lua --- -- Scope module (structured concurrency). -- --- Provides a tree of Scope objects and a per-fiber “current scope”. +-- Scopes form a tree of supervision domains. -- -- Semantics: -- - A single root scope exists for the process. --- - Each fiber has an associated current scope (defaulting to root). --- - scope.current() returns the current scope for the fiber, --- or the process-wide current scope when not in a fiber. --- - scope.run(fn, ...) runs fn in a fresh child scope in the --- current context (fiber or non-fiber), waits for its children, --- runs defers, and then returns or raises based on scope status. --- - s:spawn(fn, ...) spawns a new fiber whose current scope is s +-- - Each fibre has an associated current scope (defaulting to root). +-- - scope.current() returns the current scope for the fibre, +-- or the process-wide current scope when not in a fibre. +-- - scope.run(fn, ...) runs fn in a fresh child scope in its *own* +-- fibre and then returns (status, error, ...body_results). +-- - s:spawn(fn, ...) spawns a new fibre whose current scope is s -- for the duration of fn. --- - scope.with_ev(build_ev) creates a *first-class Event* that +-- - scope.with_ev(build_ev) creates a first-class Event that -- represents running a child scope whose body is build_ev(child). -- --- Policies implemented: --- - Status: "running" | "ok" | "failed" | "cancelled". --- - Fail-fast failure propagation by default. --- - Cancellation via Scope:cancel(reason) and a cancellation event. --- - Child fibers tracked via a waitgroup; scope exit waits for them. --- - Scope-level defers (LIFO) run at scope exit. --- - Scope:run_ev(ev) wraps an Event with failure + cancellation. +-- Status model: +-- - "running" : scope is active; children may be running. +-- - "ok" : all fibres finished without uncaught errors and +-- no explicit cancellation. +-- - "failed" : at least one fibre ended with an uncaught error. +-- - "cancelled" : explicit cancellation, or abort via with_ev. +-- +-- Failure handling: +-- - Uncaught Lua errors from any fibre are reported by runtime. +-- - The scope owning that fibre records a failure and cancels +-- its children (fail-fast). +-- - Callers observe this via status(), join_ev(), run(), etc. +-- +-- Cancellation in events: +-- - Scope:run_ev(ev) races ev against a cancellation event. +-- - If cancellation wins, the event result is: +-- ok = false, +-- value1 = reason, +-- value2 = nil. -- -- @module fibers.scope @@ -42,11 +52,15 @@ Scope.__index = Scope -- Weak-keyed table mapping Fiber objects to their current Scope. local fiber_scopes = setmetatable({}, { __mode = "k" }) --- Process-wide root scope and “current scope” when *not* in a fiber. +-- Process-wide root scope and “current scope” when *not* in a fibre. local root_scope local global_scope --- Internal: current fiber object, or nil if not in a fiber. +---------------------------------------------------------------------- +-- Internal helpers +---------------------------------------------------------------------- + +-- Internal: current fibre object, or nil if not in a fibre. local function current_fiber() return runtime.current_fiber() end @@ -61,21 +75,25 @@ local function new_scope(parent) _status = "running", -- "running" | "ok" | "failed" | "cancelled" _error = nil, -- primary error / cancellation cause _failures = {}, -- additional failures - failure_mode = "fail_fast", -- only fail_fast for now + failure_mode = "fail_fast", -- placeholder for future policies -- Concurrency tracking - _wg = waitgroup.new(), -- waitgroup for child fibers + _wg = waitgroup.new(), -- waitgroup for child fibres _defers = {}, -- LIFO list of deferred handlers - -- Cancellation and join - _cancel_cond = op.new_cond(), -- one-shot cond; signalled on cancel/failure - _join_cond = op.new_cond(), -- one-shot cond; signalled on scope exit + -- Cancellation and join conditions + _cancel_cond = op.new_cond(), -- signalled on cancel/failure + _join_cond = op.new_cond(), -- signalled when scope is closed + + -- Join worker flag + _join_worker_started = false, }, Scope) if parent then local children = parent._children children[#children + 1] = s end + return s end @@ -84,13 +102,36 @@ local function root() if not root_scope then root_scope = new_scope(nil) global_scope = root_scope + + -- Supervisor fibre: translate uncaught fibre errors into + -- scope failures and cancellations. + runtime.spawn(function() + while true do + local fib, err = runtime.wait_fiber_error() + if not fib then + -- If runtime chooses to return nil, terminate. + break + end + local s = fiber_scopes[fib] + if s then + -- mark failure + cancel children + s:_record_failure(err) + -- ensure the scope's waitgroup is decremented for this fibre + s._wg:done() + else + -- Unscoped fibre failure: treat as fatal for now. + print("Unscoped fibre error: " .. tostring(err)) + os.exit(255) + end + end + end) end return root_scope end --- Return the current Scope. --- Inside a fiber: the fiber's mapped scope, or the root if none. --- Outside a fiber: the process-wide current scope, defaulting to root. +-- Inside a fibre: the fibre's mapped scope, or the root if none. +-- Outside a fibre: the process-wide current scope, defaulting to root. local function current() local fib = current_fiber() if fib then @@ -100,76 +141,42 @@ local function current() end --- Internal helper: run fn(scope, ...) with 'scope' as current in this context. --- Returns a packed result table: { n = ..., [1] = ok, [2..n] = values }. +-- Returns the raw multiple values from fn. local function with_scope(scope_obj, fn, ...) local fib = current_fiber() if fib then local prev = fiber_scopes[fib] fiber_scopes[fib] = scope_obj - local res = pack(pcall(fn, scope_obj, ...)) + local res = { fn(scope_obj, ...) } fiber_scopes[fib] = prev - return res + return unpack(res) else local prev = global_scope or root() global_scope = scope_obj - local res = pack(pcall(fn, scope_obj, ...)) + local res = { fn(scope_obj, ...) } global_scope = prev - return res + return unpack(res) end end ---------------------------------------------------------------------- --- Core scope API +-- Scope methods: lifecycle and failure ---------------------------------------------------------------------- ---- Run a function inside a fresh child scope of the current scope. --- Synchronous: runs in the current fiber or process context. --- Returns the body_fn results on success. --- On failure or cancellation, raises the scope's primary error. --- body_fn :: function(Scope, ...): ... -local function run(body_fn, ...) - local parent = current() - local child = new_scope(parent) - - -- Run the body with 'child' as current scope. - local res = with_scope(child, body_fn, ...) - local ok = res[1] - - -- If the body itself raised (outside Event machinery), - -- mark failure and cancel the scope (to trigger done_ev, etc.). - if child._status == "running" and not ok then - child._status = "failed" - child._error = res[2] - child:cancel(child._error) -- signals _cancel_cond, cancels children - end - - -- Wait for child fibers to complete (even after fail_fast). - op.perform(child._wg:wait_op()) - - -- If still running and not cancelled/failed, mark as ok. - if child._status == "running" then - child._status = "ok" - child._error = nil - end - - -- Run defers in LIFO order. - local defers = child._defers - for i = #defers, 1, -1 do - local f = defers[i] - defers[i] = nil - pcall(f, child) - end - - -- Signal join completion. - child._join_cond.signal() - - -- Propagate outcome to caller. - if child._status == "ok" then - return unpack(res, 2, res.n) +--- Internal: record a failure in this scope and cancel children. +function Scope:_record_failure(err) + if self._status == "running" or self._status == "cancelled" then + self._status = "failed" + if self._error == nil then + self._error = err + end + -- Fail-fast: cancel this scope and descendants. + self:cancel(self._error) else - error(child._error) + local failures = self._failures + failures[#failures + 1] = err end end @@ -178,7 +185,7 @@ function Scope:new_child() return new_scope(self) end ---- Register a deferred handler to run at scope exit (LIFO). +--- Register a deferred handler to run at scope close (LIFO). -- handler :: function(Scope) function Scope:defer(handler) local defers = self._defers @@ -190,15 +197,11 @@ end function Scope:cancel(reason) local r = reason or self._error or "scope cancelled" - if self._status == "running" then + if self._status == "running" or self._status == "ok" then self._status = "cancelled" if self._error == nil then self._error = r end - elseif self._status == "ok" then - -- Explicit cancellation after success: treat as cancelled. - self._status = "cancelled" - self._error = r end -- Signal cancellation to any waiters. @@ -214,13 +217,14 @@ function Scope:cancel(reason) end end ---- Spawn a child fiber attached to this scope. +--- Spawn a child fibre attached to this scope. -- fn :: function(Scope, ...): () -- ... :: arguments passed to fn -- -- Fail-fast semantics: --- - If fn raises, this scope's status becomes "failed", its primary --- error is set (if not already), and cancel() is invoked. +-- - If fn raises, the fibre dies, runtime reports the failure, +-- and this scope's status becomes "failed" with cancellation +-- propagated to children. function Scope:spawn(fn, ...) local args = { ... } self._wg:add(1) @@ -232,34 +236,16 @@ function Scope:spawn(fn, ...) fiber_scopes[fib] = self end - local ok, err if #args > 0 then - ok, err = pcall(fn, self, unpack(args)) + fn(self, unpack(args)) else - ok, err = pcall(fn, self) + fn(self) end if fib then fiber_scopes[fib] = prev end - if not ok then - -- Fail-fast policy: record failure and cancel the scope. - if self._status == "running" then - self._status = "failed" - if self._error == nil then - self._error = err - end - self:cancel(self._error) - else - -- Additional failures are recorded but do not change - -- the primary status at this stage. - local failures = self._failures - failures[#failures + 1] = err - end - -- Do not rethrow; errors are handled via scope status. - end - self._wg:done() end) end @@ -296,9 +282,49 @@ function Scope:failures() return out end +---------------------------------------------------------------------- +-- Join and done events +---------------------------------------------------------------------- + +-- Internal: start a join worker fibre that: +-- - waits for this scope's waitgroup to reach zero; +-- - sets final status to "ok" if still "running"; +-- - runs defers; and +-- - signals _join_cond. +function Scope:_start_join_worker() + if self._join_worker_started then return end + self._join_worker_started = true + + -- System fibre: not attached to this scope, so its operation is + -- not affected by this scope's cancellation. It runs under the + -- root scope's cancellation policy. + runtime.spawn(function() + -- Wait for child fibres of this scope to complete. + op.perform_raw(self._wg:wait_op()) + + -- If still running and not cancelled/failed, mark as ok. + if self._status == "running" then + self._status = "ok" + self._error = nil + end + + -- Run defers in LIFO order. + local defers = self._defers + for i = #defers, 1, -1 do + local f = defers[i] + defers[i] = nil + f(self) + end + + -- Signal join completion. + self._join_cond.signal() + end) +end + --- Event that fires once the scope has reached a terminal status. -- Returns (status, error) when synchronised. function Scope:join_ev() + self:_start_join_worker() local ev = self._join_cond.wait_op() return ev:wrap(function() return self._status, self._error @@ -319,10 +345,12 @@ end ---------------------------------------------------------------------- -- Internal: cancellation event used when running events under this scope. +-- Convention for cancellable events: +-- ok:boolean, value1_or_reason, value2 local function cancel_event(self) local ev = self._cancel_cond.wait_op() return ev:wrap(function() - error(self._error or "scope cancelled") + return false, self._error or "scope cancelled", nil end) end @@ -334,9 +362,9 @@ function Scope:run_ev(ev) end --- Synchronise on an event under this scope. --- Equivalent to op.perform(self:run_ev(ev)). +-- Equivalent to op.perform_raw(self:run_ev(ev)). function Scope:sync(ev) - return op.perform(self:run_ev(ev)) + return op.perform_raw(self:run_ev(ev)) end ---------------------------------------------------------------------- @@ -351,20 +379,16 @@ end -- * create a child scope of scope.current(); -- * install it as the current scope while build_ev runs; -- * run build_ev(child) as an Event under normal CML semantics --- (i.e. whoever performs this Event controls cancellation etc.); --- * on conclusion or abort, wait for child fibers, run defers, +-- (whoever performs this Event controls cancellation etc.); +-- * on conclusion or abort, wait for child fibres, run defers, -- and signal join_ev(); --- * propagate the inner Event's result or error. +-- * propagate the inner Event's result or error as usual. local function with_ev(build_ev) return op.guard(function() local parent = current() local child = new_scope(parent) - -- bracket acquires "current scope = child", and guarantees - -- we restore the previous current scope and run scope cleanup - -- exactly once, whether the event wins, errors, or is aborted. local function acquire() - -- Install child as current, remember what to restore. local fib = current_fiber() if fib then local prev = fiber_scopes[fib] @@ -393,50 +417,70 @@ local function with_ev(build_ev) child:cancel(child._error) end - -- Wait for child fibers to complete. - op.perform(child._wg:wait_op()) - - -- If still running and not cancelled/failed, mark as ok. - if child._status == "running" then - child._status = "ok" - child._error = nil - end - - -- Run defers in LIFO order. - local defers = child._defers - for i = #defers, 1, -1 do - local f = defers[i] - defers[i] = nil - pcall(f, child) - end - - -- Signal join completion. - child._join_cond.signal() + -- Ensure the child scope is closed and defers run. + op.perform_raw(child:join_ev()) end local function use() -- Here the child is already installed as current(). -- build_ev must return an Event, and must not perform it. - local ok, ev = pcall(build_ev, child) - if not ok then - local ex = ev - -- mark failure & cancel the scope - if child._status == "running" then - child._status = "failed" - child._error = ex - child:cancel(ex) - end - error(ex) - end - -- The inner event itself may fail; that is handled by whoever - -- is performing this with_ev event (typically via Scope:run_ev). - return ev + return build_ev(child) end return op.bracket(acquire, release, use) end) end +---------------------------------------------------------------------- +-- scope.run: run a child scope in its own fibre +---------------------------------------------------------------------- + +--- Run a function inside a fresh child scope of the current scope. +-- +-- body_fn :: function(Scope, ...): ...results... +-- +-- Behaviour: +-- * A child scope of scope.current() is created. +-- * body_fn is run in a *separate fibre* with that child as current. +-- * All fibres spawned under that child are tracked. +-- * When the child scope has closed (ok/failed/cancelled, defers run), +-- this function returns: +-- status, error, ...results_from_body_fn... +-- +-- status :: "ok" | "failed" | "cancelled" +-- error :: primary error / cancellation reason, or nil on "ok". +-- +-- On failure or cancellation, no Lua error is thrown here; callers +-- should branch on status. +local function run(body_fn, ...) + local parent = current() + local child = new_scope(parent) + local args = { ... } + + -- store body results on the child scope + child._result = nil + + -- body fibre under the child scope + child:spawn(function(s) + local res = { body_fn(s, unpack(args)) } + s._result = res + end) + + -- wait for the child scope to reach a terminal state + local status, err = op.perform_raw(child:join_ev()) + + local res = child._result + if res then + return status, err, unpack(res) + else + return status, err + end +end + +---------------------------------------------------------------------- +-- Public API +---------------------------------------------------------------------- + return { root = root, current = current, diff --git a/tests/test_scope.lua b/tests/test_scope.lua index 5cfa538f..bdfc77ae 100644 --- a/tests/test_scope.lua +++ b/tests/test_scope.lua @@ -10,19 +10,19 @@ local op = require "fibers.op" local performer = require "fibers.performer" ------------------------------------------------------------------------------- --- 1. Structural tests (your originals, lightly generalised) +-- 1. Structural tests ------------------------------------------------------------------------------- local function test_outside_fibers() local root = scope.root() - -- current() outside any fiber should be the root (process-wide current scope) - assert(scope.current() == root, "outside fibers, current() should be root") + -- current() outside any fibre should be the root (process-wide current scope) + assert(scope.current() == root, "outside fibres, current() should be root") local outer_scope local inner_scope - scope.run(function(s) + local st, err = scope.run(function(s) outer_scope = s -- Inside run, current() should be this child scope @@ -41,7 +41,7 @@ local function test_outside_fibers() assert(found_outer, "root:children() should contain outer scope") -- Nested run creates a grandchild of s - scope.run(function(child2) + local st2, err2 = scope.run(function(child2) inner_scope = child2 assert(scope.current() == child2, "inside nested run, current() should be nested child") assert(child2:parent() == s, "nested scope parent must be outer scope") @@ -57,16 +57,21 @@ local function test_outside_fibers() assert(found_inner, "outer scope children() should contain nested scope") end) + assert(st2 == "ok" and err2 == nil, + "nested scope.run should complete with status ok") + -- After nested run, current() should be back to the outer scope assert(scope.current() == s, "after nested run, current() should be outer scope again") end) + assert(st == "ok" and err == nil, "outer scope.run should complete with status ok") + assert(outer_scope ~= nil, "outer_scope should have been set") assert(inner_scope ~= nil, "inner_scope should have been set") assert(outer_scope ~= inner_scope, "outer and inner scopes must differ") - -- After scope.run returns, current() outside fibers should be root again - assert(scope.current() == scope.root(), "after scope.run, current() should be root outside fibers") + -- After scope.run returns, current() outside fibres should be root again + assert(scope.current() == scope.root(), "after scope.run, current() should be root outside fibres") end local function test_inside_fibers() @@ -75,50 +80,57 @@ local function test_inside_fibers() local child_in_fiber local grandchild_in_fiber - -- Spawn a fiber anchored to the root scope. + -- Use a cond to wait for the spawned fibre to finish. + local done = op.new_cond() + + -- Spawn a fibre anchored to the root scope. root:spawn(function(s) - -- In this fiber, s is the scope used for spawn -> root + -- In this fibre, s is the scope used for spawn -> root assert(s == root, "spawn(fn) on root should pass root as scope") - assert(scope.current() == root, "inside spawned fiber, current() should be root initially") + assert(scope.current() == root, "inside spawned fibre, current() should be root initially") - -- Create a child scope inside the fiber - scope.run(function(child) + -- Create a child scope inside the fibre + local st, err = scope.run(function(child) child_in_fiber = child - assert(scope.current() == child, "inside scope.run in fiber, current() should be child") - assert(child:parent() == root, "child-in-fiber parent must be root") + assert(scope.current() == child, "inside scope.run in fibre, current() should be child") + assert(child:parent() == root, "child-in-fibre parent must be root") -- Create a grandchild scope - scope.run(function(grandchild) + local st2, err2 = scope.run(function(grandchild) grandchild_in_fiber = grandchild - assert(scope.current() == grandchild, "inside nested run in fiber, current() should be grandchild") + assert(scope.current() == grandchild, "inside nested run in fibre, current() should be grandchild") assert(grandchild:parent() == child, "grandchild parent must be child") end) + assert(st2 == "ok" and err2 == nil, + "nested scope.run in fibre should complete with status ok") + -- After nested run, current() should be back to child - assert(scope.current() == child, "after nested run in fiber, current() should be child again") + assert(scope.current() == child, "after nested run in fibre, current() should be child again") end) - -- After inner run, current() should be back to root for this fiber - assert(scope.current() == root, "after scope.run in fiber, current() should be root again") + assert(st == "ok" and err == nil, + "scope.run in fibre should complete with status ok") - -- Stop the scheduler once all fiber-local tests have run - runtime.stop() + -- After inner run, current() should be back to root for this fibre + assert(scope.current() == root, "after scope.run in fibre, current() should be root again") + + done.signal() end) - -- Drive the scheduler so the spawned fiber runs - runtime.main() + -- Drive until the child fibre finishes. + performer.perform(done.wait_op()) - -- After main() returns we are back outside fibers; - -- current() should again be the process-wide current scope (root). - assert(scope.current() == root, "after runtime.main, current() outside fibers should be root") + -- After that, we are still inside the test fibre; current() should be root. + assert(scope.current() == root, "after inner fibre completes, current() should be root in test fibre") - -- Check that scopes created inside the fiber were recorded + -- Check that scopes created inside the fibre were recorded assert(child_in_fiber ~= nil, "child_in_fiber should have been set") assert(grandchild_in_fiber ~= nil, "grandchild_in_fiber should have been set") assert(child_in_fiber:parent() == root, "child_in_fiber parent must be root") assert(grandchild_in_fiber:parent() == child_in_fiber, "grandchild_in_fiber parent must be child_in_fiber") - -- Check that root children include the child created in this fiber. + -- Check that root children include the child created in this fibre. local rc = root:children() local found_child = false for _, s in ipairs(rc) do @@ -131,7 +143,7 @@ local function test_inside_fibers() end ------------------------------------------------------------------------------- --- 1b. New: basic scope.with_ev behaviour +-- 1b. basic scope.with_ev behaviour ------------------------------------------------------------------------------- local function test_with_ev_basic() @@ -150,7 +162,7 @@ local function test_with_ev_basic() end) end) - local a, b = op.perform(ev) + local a, b = performer.perform(ev) assert(a == 99 and b == "ok", "with_ev should propagate child event results") assert(child_scope ~= nil, "with_ev should have created a child scope") @@ -165,51 +177,51 @@ end local function test_run_success_and_failure() local root = scope.root() - -- Success case: scope.run returns body results, status becomes "ok". + -- Success case: scope.run returns status ok and body results. local success_scope - local a, b = scope.run(function(s) + local st, err, a, b = scope.run(function(s) success_scope = s - local st, err = s:status() - assert(st == "running" and err == nil, "inside body, status should be running") + local st0, err0 = s:status() + assert(st0 == "running" and err0 == nil, "inside body, status should be running") return 42, "x" end) + assert(st == "ok" and err == nil, + "scope.run should report status ok on success") assert(a == 42 and b == "x", "scope.run should return body results on success") + local st_ok, err_ok = success_scope:status() assert(st_ok == "ok" and err_ok == nil, "successful scope should end with status ok and no error") assert(success_scope:parent() == root, "success scope parent should be root") - -- Failure case: body error propagates, status becomes "failed". + -- Failure case: body error becomes scope failure; scope.run does not throw. local fail_scope - local ok = pcall(function() - scope.run(function(s) - fail_scope = s - error("body failure") - end) + local st_fail, err_fail = scope.run(function(s) + fail_scope = s + error("body failure") end) - assert(not ok, "scope.run should rethrow body error on failure") - local st_fail, err_fail = fail_scope:status() - assert(st_fail == "failed", "failed scope should have status 'failed'") - assert(type(err_fail) == "string" or err_fail ~= nil, - "failed scope should have a primary error recorded") + + assert(st_fail == "failed", "scope.run should report status failed on body error") + assert(err_fail ~= nil, "failed scope should have a primary error recorded") assert(tostring(err_fail):find("body failure", 1, true), "failed scope primary error should mention the body failure") end local function test_run_explicit_cancel() - -- If the body explicitly cancels the scope, scope.run should raise - -- the cancellation reason and status should be "cancelled". + -- If the body explicitly cancels the scope, scope.run should + -- report status 'cancelled' and the cancellation reason. local cancelled_scope - local ok = pcall(function() - scope.run(function(s) - cancelled_scope = s - s:cancel("stop here") - end) + local st, serr = scope.run(function(s) + cancelled_scope = s + s:cancel("stop here") end) - assert(not ok, "scope.run should raise when scope is cancelled inside body") - local st, serr = cancelled_scope:status() - assert(st == "cancelled", "cancelled scope should have status 'cancelled'") + + assert(st == "cancelled", "scope.run should report cancelled when scope is cancelled inside body") assert(serr == "stop here", "cancelled scope error should be the cancellation reason") + + local st2, serr2 = cancelled_scope:status() + assert(st2 == "cancelled", "cancelled scope should have status 'cancelled'") + assert(serr2 == "stop here", "cancelled scope error should be the cancellation reason") end ------------------------------------------------------------------------------- @@ -220,20 +232,22 @@ local function test_defers_lifo_and_failure() local order = {} local scope_ref - local ok = pcall(function() - scope.run(function(s) - scope_ref = s - s:defer(function() table.insert(order, "first") end) - s:defer(function() table.insert(order, "second") end) - error("boom in body") - end) + local st, serr = scope.run(function(s) + scope_ref = s + s:defer(function() table.insert(order, "first") end) + s:defer(function() table.insert(order, "second") end) + error("boom in body") end) - assert(not ok, "scope.run should propagate body failure") - local st, serr = scope_ref:status() - assert(st == "failed", "scope should be failed after body error") + assert(st == "failed", "scope.run should report failure when body errors") assert(tostring(serr):find("boom in body", 1, true), "primary error should mention the body error") + + local st2, serr2 = scope_ref:status() + assert(st2 == "failed", "scope should be failed after body error") + assert(tostring(serr2):find("boom in body", 1, true), + "scope error should mention the body error") + assert(#order == 2, "two defers should have run") assert(order[1] == "second" and order[2] == "first", "defers should run in LIFO order even on failure") @@ -244,47 +258,58 @@ end ------------------------------------------------------------------------------- local function test_sync_wraps_event_failure() - -- Event whose post-wrap raises: tests wrap_failure path. + -- Event whose post-wrap raises: tests that scope sees failure. local ev = op.always(123):wrap(function(v) assert(v == 123, "inner always should pass its value") error("event post-wrap failure") end) local failed_scope - local ok = pcall(function() - scope.run(function(s) - failed_scope = s - -- This synchronisation should trigger fail-fast handling via performer. - performer.perform(ev) - end) + local st, serr = scope.run(function(s) + failed_scope = s + -- This synchronisation will cause this fibre to fail; + -- scope should record status 'failed'. + performer.perform(ev) end) - assert(not ok, "performer.perform on failing event should raise") - local st, serr = failed_scope:status() - assert(st == "failed", "scope should be failed after event failure") + assert(st == "failed", "scope.run should report failure when event post-wrap fails") assert(tostring(serr):find("event post-wrap failure", 1, true), "scope error should mention the event failure") + + local st2, serr2 = failed_scope:status() + assert(st2 == "failed", "scope should be failed after event failure") + assert(tostring(serr2):find("event post-wrap failure", 1, true), + "scope error should mention the event failure") end local function test_sync_respects_cancellation() - -- Race a never-ready event against cancellation. + -- Race a never-ready event against cancellation; cancellation should win + -- and be reflected as (ok=false, reason, nil) at the event level, and + -- as status 'cancelled' at the scope level. local ev = op.never() local cancelled_scope - local ok = pcall(function() - scope.run(function(s) - cancelled_scope = s - s:cancel("cancel before sync") - -- This should immediately raise via the cancellation event, - -- rather than blocking on never(). - performer.perform(ev) - end) + local st, serr, ok_ev, reason_ev = scope.run(function(s) + cancelled_scope = s + s:cancel("cancel before sync") + + local ok2, reason2 = performer.perform(ev) + assert(ok2 == false, "performer.perform should return ok=false after cancellation") + assert(reason2 == "cancel before sync", + "performer.perform should return cancellation reason") + return ok2, reason2 end) - assert(not ok, "performer.perform on never() after cancel should raise") - local st, serr = cancelled_scope:status() assert(st == "cancelled", "scope should be cancelled") assert(serr == "cancel before sync", "cancellation reason should be preserved") + + assert(ok_ev == false, "scope.run should return the event ok flag from body") + assert(reason_ev == "cancel before sync", + "scope.run should return the cancellation reason from body") + + local st2, serr2 = cancelled_scope:status() + assert(st2 == "cancelled", "cancelled_scope should be cancelled") + assert(serr2 == "cancel before sync", "cancelled_scope error should be the cancellation reason") end ------------------------------------------------------------------------------- @@ -294,17 +319,18 @@ end local function test_join_and_done_events() -- Failed scope: body error. local failed_scope - local ok_fail = pcall(function() - scope.run(function(s) - failed_scope = s - error("join test failure") - end) + local st_fail, err_fail = scope.run(function(s) + failed_scope = s + error("join test failure") end) - assert(not ok_fail, "failed scope.run should raise") + + assert(st_fail == "failed", "failed scope.run should report failed") + assert(tostring(err_fail):find("join test failure", 1, true), + "failed scope error should mention the body failure") do local ev = failed_scope:join_ev() - local st, jerr = op.perform(ev) + local st, jerr = performer.perform(ev) assert(st == "failed", "join_ev on failed scope should report 'failed'") assert(tostring(jerr):find("join test failure", 1, true), "join_ev error should mention the body failure") @@ -312,7 +338,7 @@ local function test_join_and_done_events() do local ev = failed_scope:done_ev() - local reason = op.perform(ev) + local reason = performer.perform(ev) -- For a failed scope we also call cancel(error), so done_ev -- should be triggered and report the same error. assert(tostring(reason):find("join test failure", 1, true), @@ -321,24 +347,25 @@ local function test_join_and_done_events() -- Cancelled scope (explicit cancel, not body error). local cancelled_scope - local ok_cancel = pcall(function() - scope.run(function(s) - cancelled_scope = s - s:cancel("stop again") - end) + local st_cancel, err_cancel = scope.run(function(s) + cancelled_scope = s + s:cancel("stop again") end) - assert(not ok_cancel, "cancelled scope.run should raise") + + assert(st_cancel == "cancelled", "cancelled scope.run should report cancelled") + assert(err_cancel == "stop again", + "cancelled scope.run should report the cancellation reason") do local ev = cancelled_scope:join_ev() - local st, jerr = op.perform(ev) + local st, jerr = performer.perform(ev) assert(st == "cancelled" and jerr == "stop again", "join_ev on cancelled scope should report 'cancelled' and reason") end do local ev = cancelled_scope:done_ev() - local reason = op.perform(ev) + local reason = performer.perform(ev) assert(reason == "stop again", "done_ev on cancelled scope should report cancellation reason") end @@ -352,37 +379,31 @@ local function test_fail_fast_from_child_fibre() local root = scope.root() local test_scope - root:spawn(function() - -- Create a child scope under root in this fibre. - local ok = pcall(function() - scope.run(function(s) - test_scope = s + local st, serr = scope.run(function(s) + test_scope = s - -- Use a condition to ensure the child fibre runs before we exit the body. - local cond = op.new_cond() + -- Use a condition to ensure the child fibre runs before we exit the body. + local cond = op.new_cond() - -- Spawn a child fibre that signals, then fails. - s:spawn(function(_) - cond.signal() - error("child fibre failure") - end) - - -- Wait for the cond via performer, so we do not exit - -- the body until after the child has signalled. - performer.perform(cond.wait_op()) - end) + -- Spawn a child fibre that signals, then fails. + s:spawn(function(_) + cond.signal() + error("child fibre failure") end) - assert(not ok, "scope.run should raise when a child fibre fails") - local st, serr = test_scope:status() - assert(st == "failed", "scope status should be failed after child fibre failure") - assert(tostring(serr):find("child fibre failure", 1, true), - "primary error should mention child fibre failure") - - runtime.stop() + -- Wait for the cond via performer, so we do not exit + -- the body until after the child has signalled. + performer.perform(cond.wait_op()) end) - runtime.main() + assert(st == "failed", "scope.run should report failed when a child fibre fails") + assert(tostring(serr):find("child fibre failure", 1, true), + "primary error should mention child fibre failure") + + local st2, serr2 = test_scope:status() + assert(st2 == "failed", "scope status should be failed after child fibre failure") + assert(tostring(serr2):find("child fibre failure", 1, true), + "scope primary error should mention child fibre failure") end ------------------------------------------------------------------------------- @@ -391,17 +412,25 @@ end local function main() io.stdout:write("Running scope tests...\n") - test_outside_fibers() - test_inside_fibers() - test_with_ev_basic() - test_run_success_and_failure() - test_run_explicit_cancel() - test_defers_lifo_and_failure() - test_sync_wraps_event_failure() - test_sync_respects_cancellation() - test_join_and_done_events() - test_fail_fast_from_child_fibre() - io.stdout:write("OK\n") + + -- Run all tests inside a single top-level fibre so that scope.run + -- and performer.perform are always called from within the scheduler. + runtime.spawn(function() + test_outside_fibers() + test_inside_fibers() + test_with_ev_basic() + test_run_success_and_failure() + test_run_explicit_cancel() + test_defers_lifo_and_failure() + test_sync_wraps_event_failure() + test_sync_respects_cancellation() + test_join_and_done_events() + test_fail_fast_from_child_fibre() + io.stdout:write("OK\n") + runtime.stop() + end) + + runtime.main() end main() From 7c01d995492909d9755fd7d98af9f46d7b1bb289 Mon Sep 17 00:00:00 2001 From: Richard Thanki Date: Mon, 17 Nov 2025 02:34:45 +0000 Subject: [PATCH 5/8] adds coxpcall, protects core lib calls only outside scopes, prevents dead scopes from spawning --- src/coxpcall.lua | 106 +++++++++++++++++++++++++++++++++++++++++++ src/fibers/op.lua | 3 +- src/fibers/scope.lua | 57 ++++++++++++----------- 3 files changed, 136 insertions(+), 30 deletions(-) create mode 100644 src/coxpcall.lua diff --git a/src/coxpcall.lua b/src/coxpcall.lua new file mode 100644 index 00000000..e0d9ac53 --- /dev/null +++ b/src/coxpcall.lua @@ -0,0 +1,106 @@ +-- coxpcall.lua + +local M = {} + +------------------------------------------------------------------------------- +-- Checks if (x)pcall function is coroutine safe +------------------------------------------------------------------------------- +local function isCoroutineSafe(func) + local co = coroutine.create(function() + return func(coroutine.yield, function() end) + end) + + coroutine.resume(co) + return coroutine.resume(co) +end + +-- Fast path: environment already has coroutine-safe pcall/xpcall +if isCoroutineSafe(pcall) and isCoroutineSafe(xpcall) then + -- No globals; just return plain ones + M.pcall = pcall + M.xpcall = xpcall + M.running = coroutine.running + return M +end + +------------------------------------------------------------------------------- +-- Implements xpcall with coroutines +------------------------------------------------------------------------------- + +local performResume, handleReturnValue +local oldpcall, oldxpcall = pcall, xpcall +local pack = table.pack or function(...) return { n = select("#", ...), ... } end +local unpack = table.unpack or unpack +local running = coroutine.running +local coromap = setmetatable({}, { __mode = "k" }) + +function handleReturnValue(err, co, status, ...) + if not status then + return false, err(debug.traceback(co, (...)), ...) + end + if coroutine.status(co) == 'suspended' then + return performResume(err, co, coroutine.yield(...)) + else + return true, ... + end +end + +function performResume(err, co, ...) + return handleReturnValue(err, co, coroutine.resume(co, ...)) +end + +local function id(trace, ...) + return trace +end + +local function coxpcall(f, err, ...) + local current = running() + if not current then + -- Not in a coroutine: fall back to normal pcall/xpcall + if err == id then + return oldpcall(f, ...) + else + if select("#", ...) > 0 then + local oldf, params = f, pack(...) + f = function() return oldf(unpack(params, 1, params.n)) end + end + return oldxpcall(f, err) + end + else + local res, co = oldpcall(coroutine.create, f) + if not res then + local newf = function(...) return f(...) end + co = coroutine.create(newf) + end + coromap[co] = current + return performResume(err, co, ...) + end +end + +local function corunning(coro) + if coro ~= nil then + assert(type(coro) == "thread", + "Bad argument; expected thread, got: " .. type(coro)) + else + coro = running() + end + while coromap[coro] do + coro = coromap[coro] + end + if coro == "mainthread" then return nil end + return coro +end + +------------------------------------------------------------------------------- +-- Implements pcall with coroutines +------------------------------------------------------------------------------- + +local function copcall(f, ...) + return coxpcall(f, id, ...) +end + +M.pcall = copcall +M.xpcall = coxpcall +M.running = corunning + +return M diff --git a/src/fibers/op.lua b/src/fibers/op.lua index 8696595c..e8144e5d 100644 --- a/src/fibers/op.lua +++ b/src/fibers/op.lua @@ -19,6 +19,7 @@ -- surrounding scope / fibre machinery. local runtime = require 'fibers.runtime' +local safe = require 'coxpcall' local unpack = rawget(table, "unpack") or _G.unpack local pack = rawget(table, "pack") or function(...) @@ -236,7 +237,7 @@ local function new_cond(opts) end if state.abort_fn then - pcall(state.abort_fn) + safe.pcall(state.abort_fn) end end diff --git a/src/fibers/scope.lua b/src/fibers/scope.lua index bd714edf..2422edf4 100644 --- a/src/fibers/scope.lua +++ b/src/fibers/scope.lua @@ -41,10 +41,9 @@ local runtime = require 'fibers.runtime' local op = require 'fibers.op' local waitgroup = require 'fibers.waitgroup' +local safe = require 'coxpcall' + local unpack = rawget(table, "unpack") or _G.unpack -local pack = rawget(table, "pack") or function(...) - return { n = select("#", ...), ... } -end local Scope = {} Scope.__index = Scope @@ -140,27 +139,6 @@ local function current() return global_scope or root() end ---- Internal helper: run fn(scope, ...) with 'scope' as current in this context. --- Returns the raw multiple values from fn. -local function with_scope(scope_obj, fn, ...) - local fib = current_fiber() - if fib then - local prev = fiber_scopes[fib] - fiber_scopes[fib] = scope_obj - - local res = { fn(scope_obj, ...) } - fiber_scopes[fib] = prev - return unpack(res) - else - local prev = global_scope or root() - global_scope = scope_obj - - local res = { fn(scope_obj, ...) } - global_scope = prev - return unpack(res) - end -end - ---------------------------------------------------------------------- -- Scope methods: lifecycle and failure ---------------------------------------------------------------------- @@ -226,6 +204,7 @@ end -- and this scope's status becomes "failed" with cancellation -- propagated to children. function Scope:spawn(fn, ...) + assert(self._status == "running", "cannot spawn on a non-running scope") local args = { ... } self._wg:add(1) @@ -295,10 +274,14 @@ function Scope:_start_join_worker() if self._join_worker_started then return end self._join_worker_started = true - -- System fibre: not attached to this scope, so its operation is - -- not affected by this scope's cancellation. It runs under the - -- root scope's cancellation policy. runtime.spawn(function() + -- Attach this worker to the scope + local fib = current_fiber() + local prev = fib and fiber_scopes[fib] + if fib then + fiber_scopes[fib] = self + end + -- Wait for child fibres of this scope to complete. op.perform_raw(self._wg:wait_op()) @@ -308,12 +291,28 @@ function Scope:_start_join_worker() self._error = nil end - -- Run defers in LIFO order. + -- Run defers in LIFO order, protected. local defers = self._defers for i = #defers, 1, -1 do local f = defers[i] defers[i] = nil - f(self) + + local ok, err = safe.pcall(f, self) + if not ok then + -- Treat defer failures as scope failures, but do not crash process. + if self._status == "ok" then + self._status = "failed" + self._error = self._error or err + else + local failures = self._failures + failures[#failures + 1] = err + end + end + end + + -- Restore previous scope mapping for this fibre + if fib then + fiber_scopes[fib] = prev end -- Signal join completion. From 85a79389e23dc60f77462f58f0b3304763385d18 Mon Sep 17 00:00:00 2001 From: Richard Thanki Date: Mon, 17 Nov 2025 02:36:29 +0000 Subject: [PATCH 6/8] small tidyups --- src/fibers.lua | 5 +---- 1 file changed, 1 insertion(+), 4 deletions(-) diff --git a/src/fibers.lua b/src/fibers.lua index 2dbb381b..1ae6b090 100644 --- a/src/fibers.lua +++ b/src/fibers.lua @@ -21,9 +21,6 @@ local channel = require 'fibers.channel' local op = require 'fibers.op' local unpack = rawget(table, "unpack") or _G.unpack -local pack = rawget(table, "pack") or function(...) - return { n = select("#", ...), ... } -end local fibers = {} @@ -39,7 +36,7 @@ function fibers.run(main_fn, ...) local args = { ... } -- Run main_fn inside a child scope of the root, in its own fibre. - root:spawn(function(s) + root:spawn(function() local status, err = scope_mod.run(main_fn, unpack(args)) -- In all cases, stop the scheduler so runtime.main() returns. runtime.stop() From ee10e6660282930d53774eb743b64c0810ea6e0e Mon Sep 17 00:00:00 2001 From: Richard Thanki Date: Mon, 17 Nov 2025 02:37:56 +0000 Subject: [PATCH 7/8] errors on performing non-events --- src/fibers/performer.lua | 8 ++++++++ 1 file changed, 8 insertions(+) diff --git a/src/fibers/performer.lua b/src/fibers/performer.lua index 269ec62c..5bcf48b2 100644 --- a/src/fibers/performer.lua +++ b/src/fibers/performer.lua @@ -20,7 +20,15 @@ local function current_scope() return scope_mod.current and scope_mod.current() or nil end +local function assert_event(ev) + if type(ev) ~= "table" or getmetatable(ev) ~= op.Event then + error(("perform: expected Event, got %s (%s)"):format(type(ev), tostring(ev)), 3) + end +end + function M.perform(ev) + assert_event(ev) + local s = current_scope() if s and s.sync then return s:sync(ev) From 1924ab66b5ec16cc853f1da18eed7b9a71332156 Mon Sep 17 00:00:00 2001 From: Richard Thanki Date: Mon, 17 Nov 2025 03:10:56 +0000 Subject: [PATCH 8/8] yes Mr Linter --- src/coxpcall.lua | 8 +++++--- src/fibers/runtime.lua | 4 ++-- tests/test_scope.lua | 8 ++++---- 3 files changed, 11 insertions(+), 9 deletions(-) diff --git a/src/coxpcall.lua b/src/coxpcall.lua index e0d9ac53..466d8179 100644 --- a/src/coxpcall.lua +++ b/src/coxpcall.lua @@ -29,8 +29,10 @@ end local performResume, handleReturnValue local oldpcall, oldxpcall = pcall, xpcall -local pack = table.pack or function(...) return { n = select("#", ...), ... } end -local unpack = table.unpack or unpack +local unpack = rawget(table, "unpack") or _G.unpack +local pack = rawget(table, "pack") or function(...) + return { n = select("#", ...), ... } +end local running = coroutine.running local coromap = setmetatable({}, { __mode = "k" }) @@ -49,7 +51,7 @@ function performResume(err, co, ...) return handleReturnValue(err, co, coroutine.resume(co, ...)) end -local function id(trace, ...) +local function id(trace) return trace end diff --git a/src/fibers/runtime.lua b/src/fibers/runtime.lua index fb1da7cf..6c5774ff 100644 --- a/src/fibers/runtime.lua +++ b/src/fibers/runtime.lua @@ -138,14 +138,14 @@ local function wait_fiber_error() -- Otherwise, we must be in a fibre and suspend until an error arrives. assert(_current_fiber, "wait_fiber_error must be called from within a fiber") - local function block_fn(sched, fib) + local function block_fn(_, fib) -- Record this fibre as waiting for an error. When an error -- arrives, the failing fibre will arrange to schedule a task -- that resumes this fibre with (fiber, err). error_waiters[#error_waiters + 1] = { fiber = fib } end - local wrap, err_fiber, err = _current_fiber:suspend(block_fn) + local _, err_fiber, err = _current_fiber:suspend(block_fn) -- wrap should be the identity function; ignore it. return err_fiber, err end diff --git a/tests/test_scope.lua b/tests/test_scope.lua index bdfc77ae..c88369c6 100644 --- a/tests/test_scope.lua +++ b/tests/test_scope.lua @@ -195,9 +195,9 @@ local function test_run_success_and_failure() assert(success_scope:parent() == root, "success scope parent should be root") -- Failure case: body error becomes scope failure; scope.run does not throw. - local fail_scope - local st_fail, err_fail = scope.run(function(s) - fail_scope = s + -- local fail_scope + local st_fail, err_fail = scope.run(function() + -- fail_scope = s error("body failure") end) @@ -376,7 +376,7 @@ end ------------------------------------------------------------------------------- local function test_fail_fast_from_child_fibre() - local root = scope.root() + -- local root = scope.root() local test_scope local st, serr = scope.run(function(s)