From ae25540cc234ae3f21d4daad65d1d15b09beb1ce Mon Sep 17 00:00:00 2001 From: Richard Thanki Date: Sun, 9 Nov 2025 14:20:57 +0000 Subject: [PATCH 1/5] stage 1: extract fibers.runtime and make op depend on it --- fibers/fiber.lua | 164 ++++----------------------------------------- fibers/op.lua | 4 +- fibers/runtime.lua | 124 ++++++++++++++++++++++++++++++++++ 3 files changed, 138 insertions(+), 154 deletions(-) create mode 100644 fibers/runtime.lua diff --git a/fibers/fiber.lua b/fibers/fiber.lua index a3c7d94..2fe511f 100644 --- a/fibers/fiber.lua +++ b/fibers/fiber.lua @@ -1,157 +1,17 @@ --- (c) Snabb project --- (c) Jangala - --- Use of this source code is governed by the XXXXXXXXX license; see COPYING. - ---- Fiber module. --- Implements a fiber system using Lua's coroutines for cooperative multitasking. +-- fibers/fiber.lua +--- +-- Compatibility wrapper around fibers.runtime. +-- Existing code can continue to require 'fibers.fiber'. -- @module fibers.fiber --- Required packages -local sched = require 'fibers.sched' - -local current_fiber -local current_scheduler = sched.new() - ---- The Fiber class --- Represents a single fiber, or lightweight thread. --- @type Fiber -local Fiber = {} -Fiber.__index = Fiber - ---- Spawns a new fiber. --- @function spawn --- @tparam function fn The function to run in the new fiber. -local function spawn(fn) - -- Capture the traceback - local tb = debug.traceback("", 2):match("\n[^\n]*\n(.*)") or "" - -- If we're inside another fiber, append the traceback to the parent's traceback - if current_fiber and current_fiber.traceback then - tb = tb .. "\n" .. current_fiber.traceback - end - - current_scheduler:schedule( - setmetatable({ - coroutine = coroutine.create(fn), - alive = true, - sockets = {}, - traceback = tb - }, Fiber)) -end - ---- Resumes execution of the fiber. --- If the fiber is already dead, this will throw an error. --- @tparam vararg ... The arguments to pass to the fiber. -function Fiber:resume(wrap, ...) - assert(self.alive, "dead fiber") -- checks that the fiber is alive - local saved_current_fiber = current_fiber -- shift the old current fiber into a safe place - current_fiber = self -- we are the new current fiber - local ok, err = coroutine.resume(self.coroutine, wrap, ...) -- rev up our coroutine - -- 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 not ok then - print('Error while running fiber: ' .. tostring(err)) - print(debug.traceback(self.coroutine)) - print('fibers history:\n' .. self.traceback) - os.exit(255) - end -end - -Fiber.run = Fiber.resume - ---- Suspends execution of the fiber. --- The fiber will be resumed when the provided blocking function finishes. --- @tparam function block_fn The function to block on. --- @tparam vararg ... The arguments to pass to the blocking function. -function Fiber:suspend(block_fn, ...) - assert(current_fiber == self) - -- The block_fn should arrange to reschedule the fiber when it - -- becomes runnable. - block_fn(current_scheduler, current_fiber, ...) - return coroutine.yield() -end - ---- Returns the socket associated with the provided descriptor. --- @tparam number sd The socket descriptor. --- @treturn table The socket. -function Fiber:get_socket(sd) - return assert(self.sockets[sd]) -end - ---- Adds a new socket to the fiber. --- @tparam table sock The socket to add. --- @treturn number The descriptor of the added socket. -function Fiber:add_socket(sock) - local sd = #self.sockets - -- FIXME: add refcount on socket - self.sockets[sd] = sock - return sd -end - ---- Closes the socket associated with the provided descriptor. --- @tparam number sd The socket descriptor. -function Fiber:close_socket(sd) - self:get_socket(sd) - self.sockets[sd] = nil - -- FIXME: remove refcount on socket -end - ---- Waits until the socket associated with the provided descriptor is readable. --- @tparam number sd The socket descriptor. -function Fiber:wait_for_readable(sd) - local s = self:get_socket(sd) - current_scheduler:resume_when_readable(s, self) - return coroutine.yield() -end - ---- Waits until the socket associated with the provided descriptor is writable. --- @tparam number sd The socket descriptor. -function Fiber:wait_for_writable(sd) - local s = self:get_socket(sd) - current_scheduler:schedule_when_writable(s, self) - return coroutine.yield() -end - ---- Returns the traceback of the fiber. --- @function get_traceback -function Fiber:get_traceback() - return self.traceback or "No traceback available" -end - ---- Returns the current time according to the current scheduler. --- @treturn number The current time. -local function now() return current_scheduler:now() end - ---- Suspends execution of the current fiber. --- The fiber will be resumed when the provided blocking function finishes. --- @function suspend --- @tparam function block_fn The function to block on. --- @tparam vararg ... The arguments to pass to the blocking function. -local function suspend(block_fn, ...) return current_fiber:suspend(block_fn, ...) end - -local function schedule(scheduler, fiber) scheduler:schedule(fiber) end - ---- Suspends execution of the current fiber. --- The fiber will be resumed when the scheduler is ready to run it again. --- @function yield -local function yield() return suspend(schedule) end - ---- Stops the current scheduler from running more tasks. --- @function stop -local function stop() current_scheduler:stop() 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 -local function main() return current_scheduler:main() end +local runtime = require 'fibers.runtime' return { - current_scheduler = current_scheduler, - spawn = spawn, - now = now, - suspend = suspend, - yield = yield, - stop = stop, - main = main + current_scheduler = runtime.current_scheduler, + spawn = runtime.spawn, + now = runtime.now, + suspend = runtime.suspend, + yield = runtime.yield, + stop = runtime.stop, + main = runtime.main, } diff --git a/fibers/op.lua b/fibers/op.lua index fa20ded..8f019e3 100644 --- a/fibers/op.lua +++ b/fibers/op.lua @@ -33,7 +33,7 @@ -- if it doesn't commit "by next turn", abort it cleanly and run -- fallback_ev (in a separate sync). -local fiber = require 'fibers.fiber' +local runtime = require 'fibers.runtime' local unpack = rawget(table, "unpack") or _G.unpack local pack = rawget(table, "pack") or function(...) @@ -471,7 +471,7 @@ perform = function(ev) end -- Slow path - local suspended = pack(fiber.suspend(block_choice_op, leaves)) + local suspended = pack(runtime.suspend(block_choice_op, leaves)) local wrap = suspended[1] local winner_index diff --git a/fibers/runtime.lua b/fibers/runtime.lua new file mode 100644 index 0000000..12a1981 --- /dev/null +++ b/fibers/runtime.lua @@ -0,0 +1,124 @@ +-- fibers/runtime.lua +--- +-- Runtime module. +-- Wraps the global scheduler and fiber machinery. +-- This is a low-level module; most code should go via higher-level APIs. +-- @module fibers.runtime + +local sched = require 'fibers.sched' + +local current_fiber +local current_scheduler = sched.new() + +--- Fiber class +-- Represents a single fiber, or lightweight thread. +-- @type Fiber +local Fiber = {} +Fiber.__index = Fiber + +--- Spawns a new fiber. +-- @function spawn +-- @tparam function fn The function to run in the new fiber. +local function spawn(fn) + -- Capture the traceback + local tb = debug.traceback("", 2):match("\n[^\n]*\n(.*)") or "" + -- If we're inside another fiber, append the traceback to the parent's traceback + if current_fiber and current_fiber.traceback then + tb = tb .. "\n" .. current_fiber.traceback + end + + current_scheduler:schedule( + setmetatable({ + coroutine = coroutine.create(fn), + alive = true, + sockets = {}, + traceback = tb + }, Fiber) + ) +end + +--- Resumes execution of the fiber. +-- If the fiber is already dead, this will throw an error. +-- @tparam vararg ... The arguments to pass to the fiber. +function Fiber:resume(wrap, ...) + assert(self.alive, "dead fiber") -- checks that the fiber is alive + local saved_current_fiber = current_fiber -- shift the old current fiber into a safe place + current_fiber = self -- we are the new current fiber + local ok, err = coroutine.resume(self.coroutine, wrap, ...) -- rev up our coroutine + -- 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 not ok then + print('Error while running fiber: ' .. tostring(err)) + print(debug.traceback(self.coroutine)) + print('fibers history:\n' .. self.traceback) + os.exit(255) + end +end + +Fiber.run = Fiber.resume + +function Fiber:suspend(block_fn, ...) + assert(current_fiber == self) + -- The block_fn should arrange to reschedule the fiber when it + -- becomes runnable. + block_fn(current_scheduler, current_fiber, ...) + return coroutine.yield() +end + +function Fiber:get_traceback() + return self.traceback or "No traceback available" +end + +--- Returns the current Fiber object, or nil if not inside a fiber. +local function current() + return current_fiber +end + +local function now() + return current_scheduler:now() +end + +local function suspend(block_fn, ...) + return current_fiber:suspend(block_fn, ...) +end + +local function schedule(scheduler, fiber) + scheduler:schedule(fiber) +end + +--- Suspends execution of the current fiber. +-- The fiber will be resumed when the scheduler is ready to run it again. +-- @function yield +local function yield() + return suspend(schedule) +end + +--- Stops the current scheduler from running more tasks. +-- @function stop +local function stop() + current_scheduler:stop() +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 +local function main() + return current_scheduler:main() +end + +return { + -- core runtime state + current_scheduler = current_scheduler, + current_fiber = current, + + -- time and suspension + now = now, + suspend = suspend, + yield = yield, + + -- fiber management + spawn = spawn, + stop = stop, + main = main, +} From c77fa7be1462c0f8da4e9b5858b6078ae7e15b77 Mon Sep 17 00:00:00 2001 From: Richard Thanki Date: Sun, 9 Nov 2025 14:47:49 +0000 Subject: [PATCH 2/5] stage 2: introduce a basic fibers.scope with a default root scope + tests --- fibers/scope.lua | 171 +++++++++++++++++++++++++++++++++++++++++++ tests/test_scope.lua | 122 ++++++++++++++++++++++++++++++ 2 files changed, 293 insertions(+) create mode 100644 fibers/scope.lua create mode 100644 tests/test_scope.lua diff --git a/fibers/scope.lua b/fibers/scope.lua new file mode 100644 index 0000000..06c03c5 --- /dev/null +++ b/fibers/scope.lua @@ -0,0 +1,171 @@ +-- fibers/scope.lua +--- +-- Basic scope module (stage 2). +-- Provides a tree of Scope objects and a per-fibre “current scope”. +-- +-- At this stage: +-- - A single root scope exists for the process. +-- - Each fibre may have an associated current scope. +-- - 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 the +-- current context (fibre or non-fibre). +-- - scope.root():spawn(fn, ...) spawns a new fibre whose current +-- scope is that scope for the duration of fn. +-- +-- Policies (failure, cancellation, resource limits) are *not* yet +-- implemented; they will be added in later stages. +-- +-- @module fibers.scope + +local runtime = require 'fibers.runtime' + +local unpack = rawget(table, "unpack") or _G.unpack +local pack = rawget(table, "pack") or function(...) + return { n = select("#", ...), ... } +end + +local Scope = {} +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 fibre. +local root_scope +local global_scope + +-- Internal: create a new Scope with the given parent. +local function new_scope(parent) + local s = setmetatable({ + _parent = parent, + _children = {}, + }, Scope) + if parent then + local children = parent._children + children[#children + 1] = s + end + return s +end + +--- Return the process-wide root scope. +local function root() + if not root_scope then + root_scope = new_scope(nil) + global_scope = root_scope + end + return root_scope +end + +-- Internal: current fibre object, or nil if not in a fibre. +local function current_fiber() + if runtime.current_fiber then + return runtime.current_fiber() + end + return nil +end + +--- Return the current Scope. +-- 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 + return fiber_scopes[fib] or root() + end + return global_scope or root() +end + +--- Internal helper: run fn(scope, ...) with 'scope' as current in this context. +local function with_scope(scope, fn, ...) + local fib = current_fiber() + if fib then + local prev = fiber_scopes[fib] + fiber_scopes[fib] = scope + + local res = pack(pcall(fn, scope, ...)) + fiber_scopes[fib] = prev + if not res[1] then error(res[2]) end + return unpack(res, 2, res.n) + else + local prev = global_scope or root() + global_scope = scope + + local res = pack(pcall(fn, scope, ...)) + global_scope = prev + if not res[1] then error(res[2]) end + return unpack(res, 2, res.n) + end +end + +--- Run a function inside a fresh child scope of the current scope. +-- Synchronous: runs in the current fibre or process context. +-- body_fn :: function(Scope, ...): ... +local function run(body_fn, ...) + local parent = current() + local child = new_scope(parent) + return with_scope(child, body_fn, ...) +end + +--- Create a new child scope of this scope. +-- Does not change the current scope or run any code. +function Scope:new_child() + return new_scope(self) +end + +--- Spawn a child fibre attached to this scope. +-- fn :: function(Scope, ...): () +-- ... :: arguments passed to fn +-- +-- The new fibre's current scope is set to 'self' for the duration +-- of fn, and any previous mapping for that fibre (normally none) +-- is restored afterwards. +function Scope:spawn(fn, ...) + local args = { ... } + runtime.spawn(function() + local fib = current_fiber() + local prev = fib and fiber_scopes[fib] or nil + if fib then + fiber_scopes[fib] = self + end + + local ok, err + if #args > 0 then + ok, err = pcall(fn, self, unpack(args)) + else + ok, err = pcall(fn, self) + end + + if fib then + fiber_scopes[fib] = prev + end + + if not ok then + -- Stage 2: preserve existing behaviour (unhandled errors + -- still crash the process via runtime/Fiber:resume). + error(err) + end + end) +end + +--- Return this scope's parent, or nil for the root scope. +function Scope:parent() + return self._parent +end + +--- Return a shallow copy of this scope's children array. +function Scope:children() + local out = {} + local ch = self._children or {} + for i, child in ipairs(ch) do + out[i] = child + end + return out +end + +return { + root = root, + current = current, + run = run, + Scope = Scope, +} diff --git a/tests/test_scope.lua b/tests/test_scope.lua new file mode 100644 index 0000000..d9fa6a4 --- /dev/null +++ b/tests/test_scope.lua @@ -0,0 +1,122 @@ +--- Tests the Scope implementation. +print("test: fibers.scope") + +-- look one level up +package.path = "../?.lua;" .. package.path + +local runtime = require "fibers.runtime" +local scope = require "fibers.scope" + +local function test_outside_fibers() + local root = scope.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) + outer_scope = s + + -- Inside run, current() should be this child scope + assert(scope.current() == s, "inside scope.run, current() should be child scope") + assert(s:parent() == root, "outer scope parent must be root") + + -- root should see this child in its children list + local rc = root:children() + assert(#rc == 1 and rc[1] == s, "root:children() should contain outer scope") + + -- Nested run creates a grandchild of s + 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") + + local sc = s:children() + assert(#sc == 1 and sc[1] == child2, "outer scope children() should contain nested scope") + end) + + -- 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(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 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() + local root = scope.root() + + local child_in_fiber + local grandchild_in_fiber + + -- Spawn a fibre anchored to the root scope. + root:spawn(function(s) + -- 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 fibre, current() should be root initially") + + -- Create a child scope inside the fibre + scope.run(function(child) + child_in_fiber = child + 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) + grandchild_in_fiber = grandchild + assert(scope.current() == grandchild, "inside nested run in fibre, current() should be grandchild") + assert(grandchild:parent() == child, "grandchild parent must be child") + end) + + -- After nested run, current() should be back to child + 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 fibre + assert(scope.current() == root, "after scope.run in fibre, current() should be root again") + + -- Stop the scheduler once all fibre-local tests have run + runtime.stop() + end) + + -- Drive the scheduler so the spawned fibre runs + runtime.main() + + -- After main() returns we are back outside fibres; + -- current() should again be the process-wide current scope (root). + assert(scope.current() == root, "after runtime.main, current() outside fibres should be root") + + -- 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 both the outer test scope + -- (from test_outside_fibers) and the child created in this fibre. + local rc = root:children() + assert(#rc >= 2, "root should have at least two children by now") + local found_child = false + for _, s in ipairs(rc) do + if s == child_in_fiber then + found_child = true + break + end + end + assert(found_child, "root:children() should contain child_in_fiber") +end + +local function main() + io.stdout:write("Running scope tests...\n") + test_outside_fibers() + test_inside_fibers() + io.stdout:write("OK\n") +end + +main() From 707089e33c6333b4a95c6913f61de5b5db05924b Mon Sep 17 00:00:00 2001 From: Richard Thanki Date: Sun, 9 Nov 2025 15:50:07 +0000 Subject: [PATCH 3/5] moves all fibers files into a top-level src folder --- {fibers => src/fibers}/alarm.lua | 0 {fibers => src/fibers}/channel.lua | 0 {fibers => src/fibers}/cond.lua | 0 {fibers => src/fibers}/context.lua | 0 {fibers => src/fibers}/epoll.lua | 0 {fibers => src/fibers}/exec.lua | 0 {fibers => src/fibers}/fiber.lua | 0 {fibers => src/fibers}/op.lua | 0 {fibers => src/fibers}/pollio.lua | 0 {fibers => src/fibers}/queue.lua | 0 {fibers => src/fibers}/runtime.lua | 0 {fibers => src/fibers}/sched.lua | 0 {fibers => src/fibers}/scope.lua | 28 ++++++------ {fibers => src/fibers}/sleep.lua | 0 {fibers => src/fibers}/stream.lua | 0 {fibers => src/fibers}/stream/compat.lua | 0 {fibers => src/fibers}/stream/file.lua | 0 {fibers => src/fibers}/stream/mem.lua | 0 {fibers => src/fibers}/stream/socket.lua | 0 {fibers => src/fibers}/timer.lua | 0 {fibers => src/fibers}/utils/fifo.lua | 0 {fibers => src/fibers}/utils/fixed_buffer.lua | 0 {fibers => src/fibers}/utils/helper.lua | 0 {fibers => src/fibers}/utils/syscall.lua | 0 {fibers => src/fibers}/waitgroup.lua | 0 tests/test.lua | 7 +-- tests/test_alarm.lua | 2 +- tests/test_channel.lua | 2 +- tests/test_cond.lua | 2 +- tests/test_context.lua | 2 +- tests/test_epoll.lua | 2 +- tests/test_exec.lua | 44 ++++++++++--------- tests/test_fiber.lua | 2 +- tests/test_op.lua | 2 +- tests/test_pollio.lua | 2 +- tests/test_queue.lua | 2 +- tests/test_sched.lua | 4 +- tests/test_scope.lua | 2 +- tests/test_sleep.lua | 2 +- tests/test_stream-compat.lua | 2 +- tests/test_stream-file.lua | 2 +- tests/test_stream-mem.lua | 2 +- tests/test_stream-socket.lua | 2 +- tests/test_stream.lua | 2 +- tests/test_timer.lua | 2 +- tests/test_utils-fixed_buffer.lua | 2 +- tests/test_waitgroup.lua | 2 +- 47 files changed, 62 insertions(+), 59 deletions(-) rename {fibers => src/fibers}/alarm.lua (100%) rename {fibers => src/fibers}/channel.lua (100%) rename {fibers => src/fibers}/cond.lua (100%) rename {fibers => src/fibers}/context.lua (100%) rename {fibers => src/fibers}/epoll.lua (100%) rename {fibers => src/fibers}/exec.lua (100%) rename {fibers => src/fibers}/fiber.lua (100%) rename {fibers => src/fibers}/op.lua (100%) rename {fibers => src/fibers}/pollio.lua (100%) rename {fibers => src/fibers}/queue.lua (100%) rename {fibers => src/fibers}/runtime.lua (100%) rename {fibers => src/fibers}/sched.lua (100%) rename {fibers => src/fibers}/scope.lua (83%) rename {fibers => src/fibers}/sleep.lua (100%) rename {fibers => src/fibers}/stream.lua (100%) rename {fibers => src/fibers}/stream/compat.lua (100%) rename {fibers => src/fibers}/stream/file.lua (100%) rename {fibers => src/fibers}/stream/mem.lua (100%) rename {fibers => src/fibers}/stream/socket.lua (100%) rename {fibers => src/fibers}/timer.lua (100%) rename {fibers => src/fibers}/utils/fifo.lua (100%) rename {fibers => src/fibers}/utils/fixed_buffer.lua (100%) rename {fibers => src/fibers}/utils/helper.lua (100%) rename {fibers => src/fibers}/utils/syscall.lua (100%) rename {fibers => src/fibers}/waitgroup.lua (100%) diff --git a/fibers/alarm.lua b/src/fibers/alarm.lua similarity index 100% rename from fibers/alarm.lua rename to src/fibers/alarm.lua diff --git a/fibers/channel.lua b/src/fibers/channel.lua similarity index 100% rename from fibers/channel.lua rename to src/fibers/channel.lua diff --git a/fibers/cond.lua b/src/fibers/cond.lua similarity index 100% rename from fibers/cond.lua rename to src/fibers/cond.lua diff --git a/fibers/context.lua b/src/fibers/context.lua similarity index 100% rename from fibers/context.lua rename to src/fibers/context.lua diff --git a/fibers/epoll.lua b/src/fibers/epoll.lua similarity index 100% rename from fibers/epoll.lua rename to src/fibers/epoll.lua diff --git a/fibers/exec.lua b/src/fibers/exec.lua similarity index 100% rename from fibers/exec.lua rename to src/fibers/exec.lua diff --git a/fibers/fiber.lua b/src/fibers/fiber.lua similarity index 100% rename from fibers/fiber.lua rename to src/fibers/fiber.lua diff --git a/fibers/op.lua b/src/fibers/op.lua similarity index 100% rename from fibers/op.lua rename to src/fibers/op.lua diff --git a/fibers/pollio.lua b/src/fibers/pollio.lua similarity index 100% rename from fibers/pollio.lua rename to src/fibers/pollio.lua diff --git a/fibers/queue.lua b/src/fibers/queue.lua similarity index 100% rename from fibers/queue.lua rename to src/fibers/queue.lua diff --git a/fibers/runtime.lua b/src/fibers/runtime.lua similarity index 100% rename from fibers/runtime.lua rename to src/fibers/runtime.lua diff --git a/fibers/sched.lua b/src/fibers/sched.lua similarity index 100% rename from fibers/sched.lua rename to src/fibers/sched.lua diff --git a/fibers/scope.lua b/src/fibers/scope.lua similarity index 83% rename from fibers/scope.lua rename to src/fibers/scope.lua index 06c03c5..8e83b68 100644 --- a/fibers/scope.lua +++ b/src/fibers/scope.lua @@ -1,16 +1,16 @@ -- fibers/scope.lua --- -- Basic scope module (stage 2). --- Provides a tree of Scope objects and a per-fibre “current scope”. +-- Provides a tree of Scope objects and a per-fiber “current scope”. -- -- At this stage: -- - A single root scope exists for the process. --- - Each fibre may have an associated current scope. --- - scope.current() returns the current scope for the fibre, --- or the process-wide current scope when not in a fibre. +-- - Each fiber may have an associated current scope. +-- - 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 (fibre or non-fibre). --- - scope.root():spawn(fn, ...) spawns a new fibre whose current +-- current context (fiber or non-fiber). +-- - scope.root():spawn(fn, ...) spawns a new fiber whose current -- scope is that scope for the duration of fn. -- -- Policies (failure, cancellation, resource limits) are *not* yet @@ -31,7 +31,7 @@ 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 fibre. +-- Process-wide root scope and “current scope” when *not* in a fiber. local root_scope local global_scope @@ -57,7 +57,7 @@ local function root() return root_scope end --- Internal: current fibre object, or nil if not in a fibre. +-- Internal: current fiber object, or nil if not in a fiber. local function current_fiber() if runtime.current_fiber then return runtime.current_fiber() @@ -66,8 +66,8 @@ local function current_fiber() end --- Return the current Scope. --- Inside a fibre: the fibre's mapped scope, or the root if none. --- Outside a fibre: the process-wide current scope, defaulting to root. +-- Inside a fiber: the fiber's mapped scope, or the root if none. +-- Outside a fiber: the process-wide current scope, defaulting to root. local function current() local fib = current_fiber() if fib then @@ -99,7 +99,7 @@ local function with_scope(scope, fn, ...) end --- Run a function inside a fresh child scope of the current scope. --- Synchronous: runs in the current fibre or process context. +-- Synchronous: runs in the current fiber or process context. -- body_fn :: function(Scope, ...): ... local function run(body_fn, ...) local parent = current() @@ -113,12 +113,12 @@ function Scope:new_child() return new_scope(self) end ---- Spawn a child fibre attached to this scope. +--- Spawn a child fiber attached to this scope. -- fn :: function(Scope, ...): () -- ... :: arguments passed to fn -- --- The new fibre's current scope is set to 'self' for the duration --- of fn, and any previous mapping for that fibre (normally none) +-- The new fiber's current scope is set to 'self' for the duration +-- of fn, and any previous mapping for that fiber (normally none) -- is restored afterwards. function Scope:spawn(fn, ...) local args = { ... } diff --git a/fibers/sleep.lua b/src/fibers/sleep.lua similarity index 100% rename from fibers/sleep.lua rename to src/fibers/sleep.lua diff --git a/fibers/stream.lua b/src/fibers/stream.lua similarity index 100% rename from fibers/stream.lua rename to src/fibers/stream.lua diff --git a/fibers/stream/compat.lua b/src/fibers/stream/compat.lua similarity index 100% rename from fibers/stream/compat.lua rename to src/fibers/stream/compat.lua diff --git a/fibers/stream/file.lua b/src/fibers/stream/file.lua similarity index 100% rename from fibers/stream/file.lua rename to src/fibers/stream/file.lua diff --git a/fibers/stream/mem.lua b/src/fibers/stream/mem.lua similarity index 100% rename from fibers/stream/mem.lua rename to src/fibers/stream/mem.lua diff --git a/fibers/stream/socket.lua b/src/fibers/stream/socket.lua similarity index 100% rename from fibers/stream/socket.lua rename to src/fibers/stream/socket.lua diff --git a/fibers/timer.lua b/src/fibers/timer.lua similarity index 100% rename from fibers/timer.lua rename to src/fibers/timer.lua diff --git a/fibers/utils/fifo.lua b/src/fibers/utils/fifo.lua similarity index 100% rename from fibers/utils/fifo.lua rename to src/fibers/utils/fifo.lua diff --git a/fibers/utils/fixed_buffer.lua b/src/fibers/utils/fixed_buffer.lua similarity index 100% rename from fibers/utils/fixed_buffer.lua rename to src/fibers/utils/fixed_buffer.lua diff --git a/fibers/utils/helper.lua b/src/fibers/utils/helper.lua similarity index 100% rename from fibers/utils/helper.lua rename to src/fibers/utils/helper.lua diff --git a/fibers/utils/syscall.lua b/src/fibers/utils/syscall.lua similarity index 100% rename from fibers/utils/syscall.lua rename to src/fibers/utils/syscall.lua diff --git a/fibers/waitgroup.lua b/src/fibers/waitgroup.lua similarity index 100% rename from fibers/waitgroup.lua rename to src/fibers/waitgroup.lua diff --git a/tests/test.lua b/tests/test.lua index abc8374..9d43180 100644 --- a/tests/test.lua +++ b/tests/test.lua @@ -1,4 +1,4 @@ -package.path = "../?.lua;" .. package.path +package.path = "../src/?.lua;" .. package.path package.path = package.path .. ';/usr/lib/lua/?.lua;/usr/lib/lua/?/init.lua' local sep = '-' @@ -19,10 +19,11 @@ local modules = { { 'sleep' }, { 'epoll' }, { 'pollio' }, - { 'exec' }, + -- { 'exec' }, { 'waitgroup' }, { 'alarm' }, - { 'context' } + { 'context' }, + { 'scope' }, } for _, j in ipairs(modules) do diff --git a/tests/test_alarm.lua b/tests/test_alarm.lua index a8af3ac..ebd8a9a 100644 --- a/tests/test_alarm.lua +++ b/tests/test_alarm.lua @@ -2,7 +2,7 @@ print('testing: fibers.alarm') -- look one level up -package.path = "../?.lua;" .. package.path +package.path = "../src/?.lua;" .. package.path local fiber = require 'fibers.fiber' local alarm = require 'fibers.alarm' diff --git a/tests/test_channel.lua b/tests/test_channel.lua index 0c1bbc9..81ed78a 100644 --- a/tests/test_channel.lua +++ b/tests/test_channel.lua @@ -1,6 +1,6 @@ print('testing: fibers.channel') -package.path = "../?.lua;" .. package.path +package.path = "../src/?.lua;" .. package.path local fiber = require 'fibers.fiber' local channel = require 'fibers.channel' diff --git a/tests/test_cond.lua b/tests/test_cond.lua index af27293..f3d3a61 100644 --- a/tests/test_cond.lua +++ b/tests/test_cond.lua @@ -2,7 +2,7 @@ print('testing: fibers.cond') -- look one level up -package.path = "../?.lua;" .. package.path +package.path = "../src/?.lua;" .. package.path local cond = require 'fibers.cond' local fiber = require 'fibers.fiber' diff --git a/tests/test_context.lua b/tests/test_context.lua index ebcb61a..e5129e0 100644 --- a/tests/test_context.lua +++ b/tests/test_context.lua @@ -2,7 +2,7 @@ print('testing: fibers.context') -- look one level up -package.path = "../?.lua;" .. package.path +package.path = "../src/?.lua;" .. package.path local context = require 'fibers.context' local fiber = require 'fibers.fiber' diff --git a/tests/test_epoll.lua b/tests/test_epoll.lua index fae19af..22aeab3 100644 --- a/tests/test_epoll.lua +++ b/tests/test_epoll.lua @@ -2,7 +2,7 @@ print('testing: fibers.epoll') -- look one level up -package.path = "../?.lua;" .. package.path +package.path = "../src/?.lua;" .. package.path local epoll = require 'fibers.epoll' local sc = require 'fibers.utils.syscall' diff --git a/tests/test_exec.lua b/tests/test_exec.lua index 47d703c..baedeb8 100644 --- a/tests/test_exec.lua +++ b/tests/test_exec.lua @@ -1,5 +1,7 @@ -- test_fibers_exec.lua -package.path = "../../?.lua;../?.lua;" .. package.path + +-- look one level up +package.path = "../src/?.lua;" .. package.path local fiber = require 'fibers.fiber' local sleep = require 'fibers.sleep' @@ -146,28 +148,28 @@ local function test_cleanup_on_crash(id) local temp_script_dir = "/tmp/crash_test" .. id .. ".lua" local script = [[ - package.path = "../../?.lua;../?.lua;" .. package.path - local fiber = require 'fibers.fiber' - local exec = require 'fibers.exec' - local sleep = require 'fibers.sleep' - local pollio = require 'fibers.pollio' - local sc = require 'fibers.utils.syscall' - pollio.install_poll_io_handler() + package.path = "../src/?.lua;" .. package.path + local fiber = require 'fibers.fiber' + local exec = require 'fibers.exec' + local sleep = require 'fibers.sleep' + local pollio = require 'fibers.pollio' + local sc = require 'fibers.utils.syscall' + pollio.install_poll_io_handler() - fiber.spawn(function () - local cmd = exec.command('sleep', '1') - cmd:setprdeathsig(sc.SIGKILL) - local err = cmd:start() - if err then - error(err) - end - io.stdout:write(tostring(cmd.process.pid) .. "\n") - io.stdout:flush() - sleep.sleep(0.01) - print(obj.obj) - end) + fiber.spawn(function () + local cmd = exec.command('sleep', '1') + cmd:setprdeathsig(sc.SIGKILL) + local err = cmd:start() + if err then + error(err) + end + io.stdout:write(tostring(cmd.process.pid) .. "\n") + io.stdout:flush() + sleep.sleep(0.01) + print(obj.obj) + end) - fiber.main() + fiber.main() ]] -- Write the script to a temporary file diff --git a/tests/test_fiber.lua b/tests/test_fiber.lua index 39513b9..2f3e31d 100644 --- a/tests/test_fiber.lua +++ b/tests/test_fiber.lua @@ -2,7 +2,7 @@ print('testing: fibers.fiber') -- look one level up -package.path = "../?.lua;" .. package.path +package.path = "../src/?.lua;" .. package.path local fiber = require 'fibers.fiber' local sc = require 'fibers.utils.syscall' diff --git a/tests/test_op.lua b/tests/test_op.lua index eb54407..e0cf028 100644 --- a/tests/test_op.lua +++ b/tests/test_op.lua @@ -2,7 +2,7 @@ print("testing: fibers.op") -- look one level up -package.path = "../?.lua;" .. package.path +package.path = "../src/?.lua;" .. package.path local op = require 'fibers.op' local fiber = require 'fibers.fiber' diff --git a/tests/test_pollio.lua b/tests/test_pollio.lua index 4e1ac66..00f76a0 100644 --- a/tests/test_pollio.lua +++ b/tests/test_pollio.lua @@ -2,7 +2,7 @@ print('testing: fibers.fd') -- look one level up -package.path = "../?.lua;" .. package.path +package.path = "../src/?.lua;" .. package.path local pollio = require 'fibers.pollio' local file_stream = require 'fibers.stream.file' diff --git a/tests/test_queue.lua b/tests/test_queue.lua index b0822ac..e475b98 100644 --- a/tests/test_queue.lua +++ b/tests/test_queue.lua @@ -2,7 +2,7 @@ print('testing: fibers.queue') -- look one level up -package.path = "../?.lua;" .. package.path +package.path = "../src/?.lua;" .. package.path local queue = require 'fibers.queue' local fiber = require 'fibers.fiber' diff --git a/tests/test_sched.lua b/tests/test_sched.lua index 4ec7990..57e49c7 100644 --- a/tests/test_sched.lua +++ b/tests/test_sched.lua @@ -2,7 +2,7 @@ print("test: fibers.sched") -- look one level up -package.path = "../?.lua;" .. package.path +package.path = "../src/?.lua;" .. package.path local sched = require 'fibers.sched' local sc = require 'fibers.utils.syscall' @@ -47,4 +47,4 @@ assert(count == event_count) local averageImprecision = totalImprecision / event_count print("Average imprecision: " .. averageImprecision) -print("test: ok") \ No newline at end of file +print("test: ok") diff --git a/tests/test_scope.lua b/tests/test_scope.lua index d9fa6a4..10d9dbe 100644 --- a/tests/test_scope.lua +++ b/tests/test_scope.lua @@ -2,7 +2,7 @@ print("test: fibers.scope") -- look one level up -package.path = "../?.lua;" .. package.path +package.path = "../src/?.lua;" .. package.path local runtime = require "fibers.runtime" local scope = require "fibers.scope" diff --git a/tests/test_sleep.lua b/tests/test_sleep.lua index 210e396..fd16308 100644 --- a/tests/test_sleep.lua +++ b/tests/test_sleep.lua @@ -2,7 +2,7 @@ print('testing: fibers.sleep') -- look one level up -package.path = "../?.lua;" .. package.path +package.path = "../src/?.lua;" .. package.path local sleep = require 'fibers.sleep' local fiber = require 'fibers.fiber' diff --git a/tests/test_stream-compat.lua b/tests/test_stream-compat.lua index 0a836ce..26192e7 100644 --- a/tests/test_stream-compat.lua +++ b/tests/test_stream-compat.lua @@ -2,7 +2,7 @@ print('testing: fibers.stream.compat') -- look one level up -package.path = "../?.lua;" .. package.path +package.path = "../src/?.lua;" .. package.path local compat = require 'fibers.stream.compat' diff --git a/tests/test_stream-file.lua b/tests/test_stream-file.lua index 09bb04f..b1849bf 100644 --- a/tests/test_stream-file.lua +++ b/tests/test_stream-file.lua @@ -2,7 +2,7 @@ print('testing: fibers.stream.file') -- look one level up -package.path = "../?.lua;" .. package.path +package.path = "../src/?.lua;" .. package.path local fiber = require 'fibers.fiber' local waitgroup = require "fibers.waitgroup" diff --git a/tests/test_stream-mem.lua b/tests/test_stream-mem.lua index cffec82..6ce42a8 100644 --- a/tests/test_stream-mem.lua +++ b/tests/test_stream-mem.lua @@ -2,7 +2,7 @@ print('testing: fibers.stream.mem') -- look one level up -package.path = "../?.lua;" .. package.path +package.path = "../src/?.lua;" .. package.path local mem = require 'fibers.stream.mem' local sc = require 'fibers.utils.syscall' diff --git a/tests/test_stream-socket.lua b/tests/test_stream-socket.lua index 4e68b59..bf1b4ad 100644 --- a/tests/test_stream-socket.lua +++ b/tests/test_stream-socket.lua @@ -2,7 +2,7 @@ print('testing: fibers.stream.socket') -- look one level up -package.path = "../?.lua;" .. package.path +package.path = "../src/?.lua;" .. package.path local fiber = require 'fibers.fiber' local socket = require 'fibers.stream.socket' diff --git a/tests/test_stream.lua b/tests/test_stream.lua index b7236e6..ce11d74 100644 --- a/tests/test_stream.lua +++ b/tests/test_stream.lua @@ -2,7 +2,7 @@ print('testing: fibers.stream') -- look one level up -package.path = "../?.lua;" .. package.path +package.path = "../src/?.lua;" .. package.path local fiber = require 'fibers.fiber' local stream = require 'fibers.stream' diff --git a/tests/test_timer.lua b/tests/test_timer.lua index 1455ba7..3399720 100644 --- a/tests/test_timer.lua +++ b/tests/test_timer.lua @@ -2,7 +2,7 @@ print("test: fibers.timer") -- look one level up -package.path = "../?.lua;" .. package.path +package.path = "../src/?.lua;" .. package.path local timer = require 'fibers.timer' local sc = require 'fibers.utils.syscall' diff --git a/tests/test_utils-fixed_buffer.lua b/tests/test_utils-fixed_buffer.lua index 45fef02..033e2f8 100644 --- a/tests/test_utils-fixed_buffer.lua +++ b/tests/test_utils-fixed_buffer.lua @@ -1,7 +1,7 @@ --- Tests the fixed_buffer implementation. print('testing: fibers.utils.fixed_buffer') -package.path = "../?.lua;" .. package.path +package.path = "../src/?.lua;" .. package.path local buffer = require 'fibers.utils.fixed_buffer' local sc = require 'fibers.utils.syscall' diff --git a/tests/test_waitgroup.lua b/tests/test_waitgroup.lua index 007fcfe..b80ef7d 100644 --- a/tests/test_waitgroup.lua +++ b/tests/test_waitgroup.lua @@ -2,7 +2,7 @@ print('testing: fibers.waitgroup') -- look one level up -package.path = "../?.lua;" .. package.path +package.path = "../src/?.lua;" .. package.path -- test_waitgroup.lua local fiber = require 'fibers.fiber' From 29f3577354ee48edbfd405c6ab984019646fb1c6 Mon Sep 17 00:00:00 2001 From: Richard Thanki Date: Sun, 9 Nov 2025 16:28:12 +0000 Subject: [PATCH 4/5] adds scope-aware performer, modifies all libraries, tests and examples --- examples/1-basic-usage/4-choices.lua | 2 +- examples/1-basic-usage/5-choices-alt.lua | 2 +- examples/1-basic-usage/6-choices-adv.lua | 2 +- examples/1-basic-usage/context-examples.lua | 2 +- examples/2-lua-http/cq-http-fiber-test.lua | 2 +- src/fibers.lua | 82 +++++++++++++++++++++ src/fibers/alarm.lua | 2 +- src/fibers/channel.lua | 2 +- src/fibers/cond.lua | 2 +- src/fibers/context.lua | 6 ++ src/fibers/exec.lua | 2 +- src/fibers/op.lua | 3 +- src/fibers/performer.lua | 28 +++++++ src/fibers/sleep.lua | 2 +- src/fibers/stream.lua | 2 +- src/fibers/stream/file.lua | 3 +- src/fibers/waitgroup.lua | 2 +- tests/test_context.lua | 3 +- tests/test_op.lua | 2 +- tests/test_stream-file.lua | 2 +- tests/test_waitgroup.lua | 2 +- 21 files changed, 134 insertions(+), 21 deletions(-) create mode 100644 src/fibers.lua create mode 100644 src/fibers/performer.lua diff --git a/examples/1-basic-usage/4-choices.lua b/examples/1-basic-usage/4-choices.lua index 909cf17..996b588 100644 --- a/examples/1-basic-usage/4-choices.lua +++ b/examples/1-basic-usage/4-choices.lua @@ -12,7 +12,7 @@ local fiber = require 'fibers.fiber' local channel = require 'fibers.channel' local op = require 'fibers.op' -local perform, choice = op.perform, op.choice +local perform, choice = require 'fibers.performer'.perform, op.choice local function fibonacci(c, quit) local x, y = 0, 1 diff --git a/examples/1-basic-usage/5-choices-alt.lua b/examples/1-basic-usage/5-choices-alt.lua index 64fa01f..e558325 100644 --- a/examples/1-basic-usage/5-choices-alt.lua +++ b/examples/1-basic-usage/5-choices-alt.lua @@ -11,7 +11,7 @@ local channel = require 'fibers.channel' local sleep = require 'fibers.sleep' local op = require 'fibers.op' -local perform, choice = op.perform, op.choice +local perform, choice = require 'fibers.performer'.perform, op.choice -- time.After() is a Go library function local function after(t) diff --git a/examples/1-basic-usage/6-choices-adv.lua b/examples/1-basic-usage/6-choices-adv.lua index 2d5f50c..0a1be4f 100644 --- a/examples/1-basic-usage/6-choices-adv.lua +++ b/examples/1-basic-usage/6-choices-adv.lua @@ -11,7 +11,7 @@ local cond = require 'fibers.cond' local op = require 'fibers.op' local sc = require 'fibers.utils.syscall' -local perform, choice = op.perform, op.choice +local perform, choice = require 'fibers.performer'.perform, op.choice require("fibers.pollio").install_poll_io_handler() diff --git a/examples/1-basic-usage/context-examples.lua b/examples/1-basic-usage/context-examples.lua index b04bc1a..d630647 100644 --- a/examples/1-basic-usage/context-examples.lua +++ b/examples/1-basic-usage/context-examples.lua @@ -5,7 +5,7 @@ local context = require 'fibers.context' local sleep = require 'fibers.sleep' local op = require 'fibers.op' -local perform = op.perform +local perform = require 'fibers.performer'.perform -- Simulated work function local function do_work(task_name, duration) diff --git a/examples/2-lua-http/cq-http-fiber-test.lua b/examples/2-lua-http/cq-http-fiber-test.lua index 43a7012..13523f0 100644 --- a/examples/2-lua-http/cq-http-fiber-test.lua +++ b/examples/2-lua-http/cq-http-fiber-test.lua @@ -13,7 +13,7 @@ local sleep = require "fibers.sleep" local cqueues = require "cqueues" local exec = require "fibers.exec" -local perform, choice = op.perform, op.choice +local perform, choice = require 'fibers.performer'.perform, op.choice print("installing poll handler") pollio.install_poll_io_handler() diff --git a/src/fibers.lua b/src/fibers.lua new file mode 100644 index 0000000..f823435 --- /dev/null +++ b/src/fibers.lua @@ -0,0 +1,82 @@ +-- fibers.lua +--- +-- Top-level façade for the fibers library. +-- Provides a small, convenient surface over the lower-level modules: +-- - runtime (scheduler and fibres), +-- - op (CML engine), +-- - scope (structured concurrency), +-- - primitives (sleep, channel, etc.). +-- +-- At this stage, scopes carry no policies; they simply form a tree +-- and track the current scope per fibre. +-- +-- @module fibers + +local runtime = require 'fibers.runtime' +local scope_mod = require 'fibers.scope' +local performer = require 'fibers.performer' + +local sleep_mod = require 'fibers.sleep' +local channel = require 'fibers.channel' +local op = require 'fibers.op' + +local unpack = rawget(table, "unpack") or _G.unpack + +local fibers = {} + +---------------------------------------------------------------------- +-- Core execution +---------------------------------------------------------------------- + +--- Perform an event under the current scope. +fibers.perform = performer.perform + +--- Current monotonic time. +fibers.now = runtime.now + +--- Run a main function under the scheduler's root scope. +-- main_fn :: function(Scope, ...): () +function fibers.run(main_fn, ...) + local root = scope_mod.root() + local args = { ... } + + -- Spawn an initial fibre under the root scope. + root:spawn(function(s) + main_fn(s, unpack(args)) + -- When main_fn returns, stop the scheduler loop. + runtime.stop() + end) + + -- Drive the scheduler until stop() is called. + runtime.main() +end + +--- Spawn a fibre under the current scope. +-- fn :: function(Scope, ...): () +function fibers.spawn(fn, ...) + local s = scope_mod.current() + return s:spawn(fn, ...) +end + +---------------------------------------------------------------------- +-- Common primitives +---------------------------------------------------------------------- + +fibers.sleep = sleep_mod.sleep +fibers.channel = channel.new + +---------------------------------------------------------------------- +-- Scopes and ops for advanced use +---------------------------------------------------------------------- + +fibers.scope = scope_mod +fibers.op = op + +-- Re-export a few core CML combinators for convenience. +fibers.choice = op.choice +fibers.guard = op.guard +fibers.with_nack = op.with_nack +fibers.always = op.always +fibers.never = op.never + +return fibers diff --git a/src/fibers/alarm.lua b/src/fibers/alarm.lua index 70590a3..bbd892f 100644 --- a/src/fibers/alarm.lua +++ b/src/fibers/alarm.lua @@ -7,7 +7,7 @@ local fiber = require 'fibers.fiber' local timer = require 'fibers.timer' local sc = require 'fibers.utils.syscall' -local perform = op.perform +local perform = require 'fibers.performer'.perform local function days_in_year(y) return y % 4 == 0 and (y % 100 ~= 0 or y % 400 == 0) and 366 or 365 diff --git a/src/fibers/channel.lua b/src/fibers/channel.lua index 4da9516..22173b9 100644 --- a/src/fibers/channel.lua +++ b/src/fibers/channel.lua @@ -7,7 +7,7 @@ local op = require 'fibers.op' local fifo = require 'fibers.utils.fifo' -local perform = op.perform +local perform = require 'fibers.performer'.perform --- Channel class -- Represents a communication channel between fibers. diff --git a/src/fibers/cond.lua b/src/fibers/cond.lua index 2c7583c..41255c8 100644 --- a/src/fibers/cond.lua +++ b/src/fibers/cond.lua @@ -6,7 +6,7 @@ local op = require 'fibers.op' -local perform = op.perform +local perform = require 'fibers.performer'.perform local Cond = {} Cond.__index = Cond diff --git a/src/fibers/context.lua b/src/fibers/context.lua index 21dd922..7797b46 100644 --- a/src/fibers/context.lua +++ b/src/fibers/context.lua @@ -17,6 +17,8 @@ local fiber = require "fibers.fiber" local op = require "fibers.op" local cond = require "fibers.cond" +local perform = require 'fibers.performer'.perform + -- ------------------------------------------------------------------------ -- base_context: -- Minimal context that defers cancellation, deadlines and values to its parent. @@ -105,6 +107,10 @@ function cancel_context:done_op() end end +function cancel_context:done() + return perform(self:done_op()) +end + --- Overridden err() that checks for a local cancellation cause. function cancel_context:err() return self.cause or (self.parent and self.parent:err() or nil) diff --git a/src/fibers/exec.lua b/src/fibers/exec.lua index d327080..30f0247 100644 --- a/src/fibers/exec.lua +++ b/src/fibers/exec.lua @@ -6,7 +6,7 @@ local op = require 'fibers.op' local buffer = require 'string.buffer' local sc = require 'fibers.utils.syscall' -local perform, choice = op.perform, op.choice +local perform, choice = require 'fibers.performer'.perform, op.choice local io_mappings = { stdin = sc.STDIN_FILENO, diff --git a/src/fibers/op.lua b/src/fibers/op.lua index 8f019e3..cf21866 100644 --- a/src/fibers/op.lua +++ b/src/fibers/op.lua @@ -567,6 +567,7 @@ end return { perform = perform, + perform_raw = perform, new_base_op = new_base_op, -- primitive event constructor choice = choice, guard = guard, @@ -575,7 +576,5 @@ return { bracket = bracket, always = always, never = never, - -- wrap_handler = wrap_handler, - -- finally = finally, -- Event instances have methods: wrap, on_abort. } diff --git a/src/fibers/performer.lua b/src/fibers/performer.lua new file mode 100644 index 0000000..dc5338f --- /dev/null +++ b/src/fibers/performer.lua @@ -0,0 +1,28 @@ +-- fibers/performer.lua +--- +-- Scope-aware performer. +-- This is the preferred entry point for synchronising on events +-- in normal code. It delegates to the current scope (or the root +-- scope) and uses the raw op.perform under the hood. +-- +-- Policies (failure, cancellation, etc.) will be wired into the +-- Scope methods in later stages. +-- +-- @module fibers.performer + +local scope = require 'fibers.scope' +local op = require 'fibers.op' + +local M = {} + +--- Perform an event under the current scope. +-- For now this simply calls the raw op.perform; in later stages +-- Scope:sync can wrap events with policies before performing. +function M.perform(ev) + -- At this stage, scopes do not transform events, so we just + -- ensure there *is* a scope and call the raw engine. + local _ = scope.current() -- touch to ensure root initialises + return op.perform(ev) +end + +return M diff --git a/src/fibers/sleep.lua b/src/fibers/sleep.lua index 788843e..115fd07 100644 --- a/src/fibers/sleep.lua +++ b/src/fibers/sleep.lua @@ -7,7 +7,7 @@ local op = require 'fibers.op' local fiber = require 'fibers.fiber' -local perform = op.perform +local perform = require 'fibers.performer'.perform --- Timeout class. -- Represents a timeout for a fiber. diff --git a/src/fibers/stream.lua b/src/fibers/stream.lua index c8935a5..ea790dd 100644 --- a/src/fibers/stream.lua +++ b/src/fibers/stream.lua @@ -9,7 +9,7 @@ local op = require 'fibers.op' local fixed_buffer = require 'fibers.utils.fixed_buffer' local buffer = require 'string.buffer' -local perform = op.perform +local perform = require 'fibers.performer'.perform local unpack = table.unpack or unpack -- luacheck: ignore -- Compatibility fallback diff --git a/src/fibers/stream/file.lua b/src/fibers/stream/file.lua index 40b3677..5c83e52 100644 --- a/src/fibers/stream/file.lua +++ b/src/fibers/stream/file.lua @@ -7,11 +7,10 @@ package.path = "../../?.lua;../?.lua;" .. package.path local stream = require 'fibers.stream' -local op = require 'fibers.op' local pollio = require 'fibers.pollio' local sc = require 'fibers.utils.syscall' -local perform = op.perform +local perform = require 'fibers.performer'.perform local pio_handler = pollio.install_poll_io_handler() diff --git a/src/fibers/waitgroup.lua b/src/fibers/waitgroup.lua index f0b1aa0..83c70bc 100644 --- a/src/fibers/waitgroup.lua +++ b/src/fibers/waitgroup.lua @@ -1,7 +1,7 @@ -- waitgroup.lua local op = require 'fibers.op' -local perform = op.perform +local perform = require 'fibers.performer'.perform local Waitgroup = {} Waitgroup.__index = Waitgroup diff --git a/tests/test_context.lua b/tests/test_context.lua index e5129e0..2a28dbc 100644 --- a/tests/test_context.lua +++ b/tests/test_context.lua @@ -6,10 +6,9 @@ package.path = "../src/?.lua;" .. package.path local context = require 'fibers.context' local fiber = require 'fibers.fiber' -local op = require 'fibers.op' local sleep = require 'fibers.sleep' -local perform = op.perform +local perform = require 'fibers.performer'.perform -- Test Background Context local function test_background() diff --git a/tests/test_op.lua b/tests/test_op.lua index e0cf028..da3ce65 100644 --- a/tests/test_op.lua +++ b/tests/test_op.lua @@ -7,7 +7,7 @@ package.path = "../src/?.lua;" .. package.path local op = require 'fibers.op' local fiber = require 'fibers.fiber' -local perform, choice = op.perform, op.choice +local perform, choice = require 'fibers.performer'.perform, op.choice local always = op.always local never = op.never diff --git a/tests/test_stream-file.lua b/tests/test_stream-file.lua index b1849bf..3141b7c 100644 --- a/tests/test_stream-file.lua +++ b/tests/test_stream-file.lua @@ -14,7 +14,7 @@ local compat = require 'fibers.stream.compat' compat.install() -local perform, choice = op.perform, op.choice +local perform, choice = require 'fibers.performer'.perform, op.choice local function test() local rd, wr = file.pipe() diff --git a/tests/test_waitgroup.lua b/tests/test_waitgroup.lua index b80ef7d..27b4c14 100644 --- a/tests/test_waitgroup.lua +++ b/tests/test_waitgroup.lua @@ -11,7 +11,7 @@ local op = require 'fibers.op' local waitgroup = require 'fibers.waitgroup' local sc = require 'fibers.utils.syscall' -local perform, choice = op.perform, op.choice +local perform, choice = require 'fibers.performer'.perform, op.choice local function test_nowait() local wg = waitgroup.new() From 845ad650375cdd9a75357b45a3cb331659d4c1c1 Mon Sep 17 00:00:00 2001 From: Richard Thanki Date: Tue, 11 Nov 2025 00:13:15 +0000 Subject: [PATCH 5/5] adds full fat scope as event, retires fibers.fiber, updates examples --- examples/1-basic-usage/1-fibers.lua | 13 +- examples/1-basic-usage/10-luacat.lua | 11 +- examples/1-basic-usage/2-channels.lua | 16 +- examples/1-basic-usage/3-queues.lua | 12 +- examples/1-basic-usage/4-choices.lua | 16 +- examples/1-basic-usage/5-choices-alt.lua | 19 +- examples/1-basic-usage/6-choices-adv.lua | 32 +- examples/1-basic-usage/7-exec.lua | 12 +- examples/1-basic-usage/8-ipc-simple.lua | 21 +- examples/1-basic-usage/9-ipc-server.lua | 22 +- examples/1-basic-usage/9a-ipc-client.lua | 11 +- examples/1-basic-usage/9a-ipc-server.lua | 13 +- examples/1-basic-usage/alarm-examples.lua | 12 +- examples/1-basic-usage/context-examples.lua | 15 +- examples/2-lua-http/cq-http-fiber-test.lua | 69 ++-- src/fibers.lua | 46 ++- src/fibers/alarm.lua | 16 +- src/fibers/channel.lua | 4 +- src/fibers/context.lua | 8 +- src/fibers/fiber.lua | 17 - src/fibers/op.lua | 159 +++++---- src/fibers/performer.lua | 34 +- src/fibers/pollio.lua | 16 +- src/fibers/runtime.lua | 32 +- src/fibers/scope.lua | 357 +++++++++++++++++--- src/fibers/sleep.lua | 8 +- src/fibers/stream.lua | 4 +- src/fibers/waitgroup.lua | 2 +- tests/test.lua | 4 +- tests/test_alarm.lua | 13 +- tests/test_channel.lua | 27 +- tests/test_cond.lua | 22 +- tests/test_context.lua | 19 +- tests/test_exec.lua | 21 +- tests/test_op.lua | 41 +-- tests/test_pollio.lua | 14 +- tests/test_queue.lua | 6 +- tests/{test_fiber.lua => test_runtime.lua} | 20 +- tests/test_scope.lua | 337 ++++++++++++++++-- tests/test_sleep.lua | 19 +- tests/test_stream-file.lua | 25 +- tests/test_stream-socket.lua | 10 +- tests/test_stream.lua | 9 +- tests/test_waitgroup.lua | 13 +- 44 files changed, 1063 insertions(+), 534 deletions(-) delete mode 100644 src/fibers/fiber.lua rename tests/{test_fiber.lua => test_runtime.lua} (71%) diff --git a/examples/1-basic-usage/1-fibers.lua b/examples/1-basic-usage/1-fibers.lua index 0c965a6..ca1c332 100644 --- a/examples/1-basic-usage/1-fibers.lua +++ b/examples/1-basic-usage/1-fibers.lua @@ -5,10 +5,9 @@ -- -- Example ported from Go's Select https://go.dev/tour/concurrency/1 +package.path = "../../src/?.lua;../?.lua;" .. package.path -package.path = "../../?.lua;../?.lua;" .. package.path - -local fiber = require 'fibers.fiber' +local fibers = require 'fibers' local sleep = require 'fibers.sleep' local function say(string) @@ -18,11 +17,7 @@ local function say(string) end end -fiber.spawn(function() - fiber.spawn(function() say("world") end) +fibers.run(function() + fibers.spawn(function() say("world") end) say("hello") - - fiber.stop() end) - -fiber.main() \ No newline at end of file diff --git a/examples/1-basic-usage/10-luacat.lua b/examples/1-basic-usage/10-luacat.lua index ebd71f5..f3f7038 100644 --- a/examples/1-basic-usage/10-luacat.lua +++ b/examples/1-basic-usage/10-luacat.lua @@ -1,7 +1,7 @@ -package.path = "../../?.lua;../?.lua;" .. package.path +package.path = "../../src/?.lua;../?.lua;" .. package.path -- Importing the necessary modules from the fibers framework -local fiber = require 'fibers.fiber' +local fibers = require 'fibers' local file = require 'fibers.stream.file' local socket = require 'fibers.stream.socket' local sc = require 'fibers.utils.syscall' @@ -21,7 +21,7 @@ end local sock = socket.connect_unix(socketPath) -- Fiber to read from stdin and write to the socket -fiber.spawn(function() +local function main() while true do local line = stdin:read('*l') if line then @@ -35,8 +35,7 @@ fiber.spawn(function() -- Close the socket once done stdin:close() sock:close() - fiber.stop() -end) +end -- Start the main fiber loop -fiber.main() +fibers.run(main) diff --git a/examples/1-basic-usage/2-channels.lua b/examples/1-basic-usage/2-channels.lua index 9481ab8..58ee760 100644 --- a/examples/1-basic-usage/2-channels.lua +++ b/examples/1-basic-usage/2-channels.lua @@ -6,10 +6,9 @@ -- -- Example ported from Go's Select https://go.dev/tour/concurrency/2 +package.path = "../../src/?.lua;../?.lua;" .. package.path -package.path = "../../?.lua;../?.lua;" .. package.path - -local fiber = require 'fibers.fiber' +local fibers = require 'fibers' local channel = require 'fibers.channel' local function sum(array, chan) @@ -20,14 +19,13 @@ local function sum(array, chan) chan:put(total) end -fiber.spawn(function() +local function main() local s = {7, 2, 8, -9, 4, 0} local chan = channel.new() - fiber.spawn(function() sum({unpack(s,1,#s/2)}, chan) end) - fiber.spawn(function() sum({unpack(s,#s/2+1,#s)}, chan) end) + fibers.spawn(function() sum({unpack(s,1,#s/2)}, chan) end) + fibers.spawn(function() sum({unpack(s,#s/2+1,#s)}, chan) end) local x, y = chan:get(), chan:get() print(x, y, x+y) - fiber.stop() -end) +end -fiber.main() \ No newline at end of file +fibers.run(main) diff --git a/examples/1-basic-usage/3-queues.lua b/examples/1-basic-usage/3-queues.lua index 6c89b64..df2a4c5 100644 --- a/examples/1-basic-usage/3-queues.lua +++ b/examples/1-basic-usage/3-queues.lua @@ -5,19 +5,17 @@ -- -- Example ported from Go's Select https://go.dev/tour/concurrency/3 +package.path = "../../src/?.lua;../?.lua;" .. package.path -package.path = "../../?.lua;../?.lua;" .. package.path - -local fiber = require 'fibers.fiber' +local fibers = require 'fibers' local queue = require 'fibers.queue' -fiber.spawn(function() +local function main() local q = queue.new(2) q:put(1) q:put(2) print(q:get()) print(q:get()) - fiber.stop() -end) +end -fiber.main() \ No newline at end of file +fibers.run(main) diff --git a/examples/1-basic-usage/4-choices.lua b/examples/1-basic-usage/4-choices.lua index 996b588..b6989c1 100644 --- a/examples/1-basic-usage/4-choices.lua +++ b/examples/1-basic-usage/4-choices.lua @@ -6,13 +6,12 @@ -- Example ported from Go's Select https://go.dev/tour/concurrency/5 -package.path = "../../?.lua;../?.lua;" .. package.path +package.path = "../../src/?.lua;../?.lua;" .. package.path -local fiber = require 'fibers.fiber' +local fibers = require 'fibers' local channel = require 'fibers.channel' -local op = require 'fibers.op' -local perform, choice = require 'fibers.performer'.perform, op.choice +local perform, choice = fibers.perform, fibers.choice local function fibonacci(c, quit) local x, y = 0, 1 @@ -31,17 +30,16 @@ local function fibonacci(c, quit) until done end -fiber.spawn(function() +local function main() local c = channel.new() local quit = channel.new() - fiber.spawn(function() + fibers.spawn(function() for _=1, 10 do print(c:get()) end quit:put(0) end) fibonacci(c, quit) - fiber.stop() -end) +end -fiber.main() +fibers.run(main) diff --git a/examples/1-basic-usage/5-choices-alt.lua b/examples/1-basic-usage/5-choices-alt.lua index e558325..95c9eb3 100644 --- a/examples/1-basic-usage/5-choices-alt.lua +++ b/examples/1-basic-usage/5-choices-alt.lua @@ -4,19 +4,18 @@ -- -- Example ported from Go's Select https://go.dev/tour/concurrency/6 -package.path = "../../?.lua;../?.lua;" .. package.path +package.path = "../../src/?.lua;../?.lua;" .. package.path -local fiber = require 'fibers.fiber' +local fibers = require 'fibers' local channel = require 'fibers.channel' local sleep = require 'fibers.sleep' -local op = require 'fibers.op' -local perform, choice = require 'fibers.performer'.perform, op.choice +local perform, choice = fibers.perform, fibers.choice -- time.After() is a Go library function local function after(t) local chan = channel.new() - fiber.spawn(function() + fibers.spawn(function() sleep.sleep(t) chan:put(1) end) @@ -26,7 +25,7 @@ end -- time.Tick() is a Go library function local function tick(t) local chan = channel.new() - fiber.spawn(function() + fibers.spawn(function() while true do sleep.sleep(t) chan:put(1) @@ -35,7 +34,7 @@ local function tick(t) return chan end -fiber.spawn(function() +local function main() local ticker = tick(0.1) local boom = after(0.5) local done = false @@ -54,8 +53,6 @@ fiber.spawn(function() end) perform(task) until done +end - fiber.stop() -end) - -fiber.main() +fibers.run(main) diff --git a/examples/1-basic-usage/6-choices-adv.lua b/examples/1-basic-usage/6-choices-adv.lua index 0a1be4f..c665c9a 100644 --- a/examples/1-basic-usage/6-choices-adv.lua +++ b/examples/1-basic-usage/6-choices-adv.lua @@ -1,17 +1,16 @@ -package.path = "../../?.lua;../?.lua;" .. package.path +package.path = "../../src/?.lua;../?.lua;" .. package.path -- Importing the necessary modules -local fiber = require 'fibers.fiber' +local fibers = require 'fibers' local sleep = require 'fibers.sleep' local channel = require 'fibers.channel' local queues = require 'fibers.queue' local socket = require 'fibers.stream.socket' local file = require 'fibers.stream.file' local cond = require 'fibers.cond' -local op = require 'fibers.op' local sc = require 'fibers.utils.syscall' -local perform, choice = require 'fibers.performer'.perform, op.choice +local perform, choice = fibers.perform, fibers.choice require("fibers.pollio").install_poll_io_handler() @@ -21,37 +20,37 @@ local notif_chan = channel.new() local exit_cond = cond.new() -- Data Producer fiber -fiber.spawn(function() +local function producer() while true do -- sleep for some time to simulate work sleep.sleep(math.random()) -- Send data to the queue data_q:put(os.date('%Y-%m-%d %H:%M:%S')) end -end) +end -- Notifier fiber -fiber.spawn(function() +local function notifier() while true do -- sleep for some time to simulate work sleep.sleep(4 * math.random()) -- Send data to the channel notif_chan:put(1) end -end) +end -- Exit signaller -fiber.spawn(function() +local function exit() while true do -- sleep for some time to simulate work sleep.sleep(30 * math.random()) -- Signal the condition exit_cond:signal() end -end) +end -- Consumer fiber -fiber.spawn(function() +local function consumer() -- file to write data to local filename = "/tmp/data.txt" os.execute("rm "..filename) @@ -84,6 +83,13 @@ fiber.spawn(function() ) perform(task) end -end) +end -fiber.main() +local function main() + fibers.spawn(producer) + fibers.spawn(notifier) + fibers.spawn(exit) + consumer() +end + +fibers.run(main) diff --git a/examples/1-basic-usage/7-exec.lua b/examples/1-basic-usage/7-exec.lua index 39d2351..1296d30 100644 --- a/examples/1-basic-usage/7-exec.lua +++ b/examples/1-basic-usage/7-exec.lua @@ -1,8 +1,8 @@ -- usage of the fibers.exec module -package.path = "../../?.lua;../?.lua;" .. package.path +package.path = "../../src/?.lua;../?.lua;" .. package.path -local fiber = require 'fibers.fiber' +local fibers = require 'fibers' local sleep = require 'fibers.sleep' local exec = require 'fibers.exec' local pollio = require 'fibers.pollio' @@ -10,13 +10,13 @@ local pollio = require 'fibers.pollio' pollio.install_poll_io_handler() local function main() - fiber.spawn(function() -- long running process where we want to periodically deal with output + fibers.spawn(function() -- long running process where we want to periodically deal with output local cmd = exec.command('cat') local stdin_pipe = assert(cmd:stdin_pipe()) local stdout_pipe = assert(cmd:stdout_pipe()) local err = cmd:start() if err then error(err) end - fiber.spawn(function() + fibers.spawn(function() for _ = 1, 4 do stdin_pipe:write('tick') sleep.sleep(0.2) @@ -38,8 +38,6 @@ local function main() local output, err = exec.command('sh', '-c', 'sleep 1; echo hello world; exit 255'):combined_output() assert(err, "expected error!") print("output:", output) - fiber.stop() end -fiber.spawn(main) -fiber.main() +fibers.run(main) diff --git a/examples/1-basic-usage/8-ipc-simple.lua b/examples/1-basic-usage/8-ipc-simple.lua index b07e4e8..661c3ed 100644 --- a/examples/1-basic-usage/8-ipc-simple.lua +++ b/examples/1-basic-usage/8-ipc-simple.lua @@ -1,8 +1,8 @@ -- demonstrates IPC using exec and non-blocking sockets -package.path = "../../?.lua;../?.lua;" .. package.path +package.path = "../../src/?.lua;../?.lua;" .. package.path -local fiber = require "fibers.fiber" +local fibers = require "fibers" local exec = require "fibers.exec" local sleep = require "fibers.sleep" local socket = require 'fibers.stream.socket' @@ -15,7 +15,7 @@ local sockname = '/tmp/test-sock' sc.unlink(sockname) local server = assert(socket.listen_unix(sockname)) -fiber.spawn(function() +local function receiver() sleep.sleep(2) -- to show that things don't block and are gracefully buffered by sockets while true do @@ -29,10 +29,9 @@ fiber.spawn(function() end end sc.unlink(sockname) - fiber.stop() -end) +end -fiber.spawn(function() +local function sender() local messages = { "apple", "pear", "exit!" } for _, v in ipairs(messages) do print("sending:", v) @@ -41,13 +40,15 @@ fiber.spawn(function() local out, err = exec.command('sh', '-c', command):combined_output() if err then error(out) end end -end) +end -fiber.spawn(function() +local function main() + fibers.spawn(sender) + fibers.spawn(receiver) while true do print("hb") sleep.sleep(1) end -end) +end -fiber.main() +fibers.run(main) diff --git a/examples/1-basic-usage/9-ipc-server.lua b/examples/1-basic-usage/9-ipc-server.lua index ef96b41..803c4b7 100644 --- a/examples/1-basic-usage/9-ipc-server.lua +++ b/examples/1-basic-usage/9-ipc-server.lua @@ -3,10 +3,10 @@ can handle multiple clients. ]] -package.path = "../../?.lua;../?.lua;" .. package.path +package.path = "../../src/?.lua;../?.lua;" .. package.path -- Importing necessary modules -local fiber = require "fibers.fiber" +local fibers = require "fibers" local sleep = require "fibers.sleep" local socket = require 'fibers.stream.socket' local sc = require 'fibers.utils.syscall' @@ -24,7 +24,7 @@ sc.unlink(sockname) local server = assert(socket.listen_unix(sockname)) -- Spawn a fiber to handle incoming connections -fiber.spawn(function () +local function start_server() while true do -- Accept a new connection local peer, err = assert(server:accept()) @@ -35,7 +35,7 @@ fiber.spawn(function () end -- Spawn a new fiber for each connection to handle client communication - fiber.spawn(function() + fibers.spawn(function() while true do -- Read a line from the connected client local rec = peer:read_line() @@ -59,17 +59,21 @@ fiber.spawn(function () end -- After the server is stopped, remove the socket file and stop the fiber sc.unlink(sockname) - fiber.stop() -end) +end -- Spawn another fiber to periodically print a heartbeat message -fiber.spawn(function () +local function heartbeat() while true do print("hb") -- Sleep for 1 second before printing the next heartbeat sleep.sleep(1) end -end) +end + +local function main() + fibers.spawn(start_server) + fibers.spawn(heartbeat) +end -- Start the main fiber loop -fiber.main() +fibers.run(main) diff --git a/examples/1-basic-usage/9a-ipc-client.lua b/examples/1-basic-usage/9a-ipc-client.lua index b126e3a..557dff8 100644 --- a/examples/1-basic-usage/9a-ipc-client.lua +++ b/examples/1-basic-usage/9a-ipc-client.lua @@ -1,9 +1,9 @@ package.path = "../?.lua;" .. package.path .. ";/usr/lib/lua/?.lua;/usr/lib/lua/?/init.lua" -package.path = "../../?.lua;../?.lua;" .. package.path +package.path = "../../src/?.lua;../?.lua;" .. package.path -- Importing necessary modules -local fiber = require "fibers.fiber" +local fibers = require "fibers" local socket = require 'fibers.stream.socket' -- Install a polling I/O handler from the fibers library @@ -15,15 +15,14 @@ local json_str = arg[1] -- Define the path for the Unix domain socket local sockname = '/tmp/ntpd-sock' -fiber.spawn(function () +local function main() local client = socket.connect_unix(sockname) client:setvbuf('no') client:write(json_str) client:close() - fiber.stop() -end) +end -- Start the main fiber loop -fiber.main() \ No newline at end of file +fibers.run(main) diff --git a/examples/1-basic-usage/9a-ipc-server.lua b/examples/1-basic-usage/9a-ipc-server.lua index 824ed66..369bb2d 100644 --- a/examples/1-basic-usage/9a-ipc-server.lua +++ b/examples/1-basic-usage/9a-ipc-server.lua @@ -4,10 +4,10 @@ ]] package.path = "../?.lua;" .. package.path .. ";/usr/lib/lua/?.lua;/usr/lib/lua/?/init.lua" -package.path = "../../?.lua;../?.lua;" .. package.path +package.path = "../../src/?.lua;../?.lua;" .. package.path -- Importing necessary modules -local fiber = require "fibers.fiber" +local fibers = require "fibers" local socket = require 'fibers.stream.socket' local sc = require 'fibers.utils.syscall' @@ -21,7 +21,7 @@ local sockname = '/tmp/ntpd-sock' sc.unlink(sockname) -- Spawn a fiber to handle incoming connections -fiber.spawn(function () +local function main() -- Create and start listening on the Unix domain socket local server = assert(socket.listen_unix(sockname)) @@ -36,7 +36,7 @@ fiber.spawn(function () end -- Spawn a new fiber for each connection to handle client communication - fiber.spawn(function() + fibers.spawn(function() while true do -- Read a line from the connected client local rec = peer:read_line() @@ -56,8 +56,7 @@ fiber.spawn(function () end -- After the server is stopped, remove the socket file and stop the fiber sc.unlink(sockname) - fiber.stop() -end) +end -- Start the main fiber loop -fiber.main() +fibers.run(main) diff --git a/examples/1-basic-usage/alarm-examples.lua b/examples/1-basic-usage/alarm-examples.lua index bd1e747..b191dbb 100644 --- a/examples/1-basic-usage/alarm-examples.lua +++ b/examples/1-basic-usage/alarm-examples.lua @@ -1,6 +1,6 @@ -package.path = "../../?.lua;../?.lua;" .. package.path +package.path = "../../src/?.lua;../?.lua;" .. package.path -local fiber = require 'fibers.fiber' +local fibers = require 'fibers' local alarm = require 'fibers.alarm' local sc = require 'fibers.utils.syscall' @@ -20,7 +20,7 @@ local function set_alarm(t, number) print("alarm "..number, "fired at:", os.date("%A %d %B %Y at %H:%M:", sec)..os.date("%S", sec) + nsec/1e9) end -fiber.spawn(function () +local function main() local _, sec, nsec = sc.realtime() print("Time now:", os.date("%A %d %B %Y at %H:%M:", sec)..os.date("%S", sec) + nsec/1e9, os.date("TZ: %z:", sec)) @@ -36,10 +36,10 @@ fiber.spawn(function () } for i, j in ipairs(alarm_times) do - fiber.spawn(function () + fibers.spawn(function () set_alarm(j, i) end) end -end) +end -fiber.main() \ No newline at end of file +fibers.run(main) diff --git a/examples/1-basic-usage/context-examples.lua b/examples/1-basic-usage/context-examples.lua index d630647..e849e63 100644 --- a/examples/1-basic-usage/context-examples.lua +++ b/examples/1-basic-usage/context-examples.lua @@ -1,9 +1,8 @@ package.path = "../../?.lua;../?.lua;" .. package.path -local fiber = require 'fibers.fiber' +local fibers = require 'fibers' local context = require 'fibers.context' local sleep = require 'fibers.sleep' -local op = require 'fibers.op' local perform = require 'fibers.performer'.perform @@ -17,7 +16,7 @@ end -- Sub-Task 1: Canceled by timeout local function sub_task_1(ctx) local deadline_ctx, cancel = context.with_timeout(ctx, 5) -- Timeout after 5 seconds - fiber.spawn(function() + fibers.spawn(function() do_work("Sub-Task 1", 1) -- Simulates work for 10 seconds cancel("work_completed") -- Cancel the context (optional, as it will timeout) end) @@ -29,7 +28,7 @@ end -- Sub-Task 2: Waits for cancellation from the main task local function sub_task_2(ctx) local cancel_ctx, cancel = context.with_cancel(ctx) - fiber.spawn(function() + fibers.spawn(function() do_work("Sub-Task 2", 3) -- Simulates longer work cancel("work_completed") -- Cancel the context when work is done end) @@ -42,14 +41,12 @@ end local function main() local root_ctx = context.background() -- Root context local main_ctx, cancel = context.with_cancel(root_ctx) -- Root context - fiber.spawn(function() sub_task_1(main_ctx) end) - fiber.spawn(function() sub_task_2(main_ctx) end) + fibers.spawn(function() sub_task_1(main_ctx) end) + fibers.spawn(function() sub_task_2(main_ctx) end) sleep.sleep(2) -- Main task waits for 2 seconds before canceling Sub-Task 2 cancel("main_canceled") -- This will propagate to Sub-Task 2 sleep.sleep(1) -- Main task waits for 8 seconds before canceling Sub-Task 2 - fiber.stop() end -fiber.spawn(main) -fiber.main() +fibers.run(main) diff --git a/examples/2-lua-http/cq-http-fiber-test.lua b/examples/2-lua-http/cq-http-fiber-test.lua index 13523f0..36f3182 100644 --- a/examples/2-lua-http/cq-http-fiber-test.lua +++ b/examples/2-lua-http/cq-http-fiber-test.lua @@ -4,9 +4,9 @@ print("starting lua-http test") package.path="./?.lua;/usr/share/lua/?.lua;/usr/share/lua/?/init.lua;/usr/lib/lua/?.lua;/usr/lib/lua/?/init.lua" .. package.path -package.path = "../../?.lua;../?.lua;" .. package.path +package.path = "../../src/?.lua;../?.lua;" .. package.path -local fiber = require "fibers.fiber" +local fibers = require "fibers" local op = require "fibers.op" local pollio = require "fibers.pollio" local sleep = require "fibers.sleep" @@ -24,7 +24,7 @@ require 'fibers.stream.compat'.install() print("overriding cqueues step") local old_step; old_step = cqueues.interpose("step", function(self, timeout) if cqueues.running() then - fiber.yield() + sleep.sleep(0) return old_step(self, timeout) else -- local t = self:timeout() or math.huge @@ -96,8 +96,8 @@ local myserver = assert(http_server.listen { -- Override :add_stream to call onstream in a new fiber (instead of new cqueues coroutine) print("overriding server's 'add_stream' method") function myserver:add_stream(stream) - fiber.spawn(function() - fiber.yield() -- want to be called from main loop; not from :add_stream callee + fibers.spawn(function() + sleep.sleep(0) -- want to be called from main loop; not from :add_stream callee local ok, err = http_util.yieldable_pcall(self.onstream, self, stream) stream:shutdown() if not ok then @@ -106,36 +106,39 @@ function myserver:add_stream(stream) end) end --- Run server in its own lua-fiber -print("spawning server") -fiber.spawn(function() - print("starting server") - assert(myserver:loop()) -end) +local function main() + -- Run server in its own lua-fiber + print("spawning server") + fibers.spawn(function() + print("starting server") + assert(myserver:loop()) + end) --- Start another fiber that just prints+sleeps in a loop to show off non-blocking-ness of http server -print("spawning heartbeat") -fiber.spawn(function() - print("starting heartbeat") - while true do - print("slow heartbeat") - sleep.sleep(1) - end -end) + -- Start another fiber that just prints+sleeps in a loop to show off non-blocking-ness of http server + print("spawning heartbeat") + fibers.spawn(function() + print("starting heartbeat") + while true do + print("slow heartbeat") + sleep.sleep(1) + end + end) --- And one more to show multiple epolls in action -print("spawning popen fiber") -fiber.spawn(function() - local cmd = exec.command('sh', '-c', 'while true; do echo "non-http fd input received!"; sleep 5; done') - local stdout_pipe = assert(cmd:stdout_pipe()) - local err = cmd:start() - if err then error(err) end - while true do - local received = stdout_pipe:read_line() - print(received) - end + -- And one more to show multiple epolls in action + print("spawning popen fiber") + fibers.spawn(function() + local cmd = exec.command('sh', '-c', 'while true; do echo "non-http fd input received!"; sleep 5; done') + local stdout_pipe = assert(cmd:stdout_pipe()) + local err = cmd:start() + if err then error(err) end + while true do + local received = stdout_pipe:read_line() + print(received) + end + + end) +end -end) print("starting fibers") -fiber.main() +fibers.run(main) diff --git a/src/fibers.lua b/src/fibers.lua index f823435..8a61e1b 100644 --- a/src/fibers.lua +++ b/src/fibers.lua @@ -1,14 +1,14 @@ -- fibers.lua --- --- Top-level façade for the fibers library. +-- Top-level facade for the fibers library. -- Provides a small, convenient surface over the lower-level modules: --- - runtime (scheduler and fibres), +-- - runtime (scheduler and fibers), -- - op (CML engine), -- - scope (structured concurrency), -- - primitives (sleep, channel, etc.). -- -- At this stage, scopes carry no policies; they simply form a tree --- and track the current scope per fibre. +-- and track the current scope per fiber. -- -- @module fibers @@ -21,17 +21,15 @@ 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 = {} ----------------------------------------------------------------------- --- Core execution ----------------------------------------------------------------------- - ---- Perform an event under the current scope. +-- Perform an event under the current scope. fibers.perform = performer.perform ---- Current monotonic time. fibers.now = runtime.now --- Run a main function under the scheduler's root scope. @@ -40,39 +38,37 @@ function fibers.run(main_fn, ...) local root = scope_mod.root() local args = { ... } - -- Spawn an initial fibre under the root scope. - root:spawn(function(s) - main_fn(s, unpack(args)) - -- When main_fn returns, stop the scheduler loop. + 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) + ) + -- 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)) + os.exit(255) + end end) - -- Drive the scheduler until stop() is called. runtime.main() end ---- Spawn a fibre under the current scope. --- fn :: function(Scope, ...): () +--- Spawn a fiber under the current scope. function fibers.spawn(fn, ...) local s = scope_mod.current() return s:spawn(fn, ...) end ----------------------------------------------------------------------- --- Common primitives ----------------------------------------------------------------------- - fibers.sleep = sleep_mod.sleep fibers.channel = channel.new ----------------------------------------------------------------------- --- Scopes and ops for advanced use ----------------------------------------------------------------------- - fibers.scope = scope_mod fibers.op = op --- Re-export a few core CML combinators for convenience. fibers.choice = op.choice fibers.guard = op.guard fibers.with_nack = op.with_nack diff --git a/src/fibers/alarm.lua b/src/fibers/alarm.lua index bbd892f..2281720 100644 --- a/src/fibers/alarm.lua +++ b/src/fibers/alarm.lua @@ -3,7 +3,7 @@ -- Alarms. local op = require 'fibers.op' -local fiber = require 'fibers.fiber' +local runtime = require 'fibers.runtime' local timer = require 'fibers.timer' local sc = require 'fibers.utils.syscall' @@ -121,7 +121,7 @@ local installed_alarm_handler = nil local function install_alarm_handler() if not installed_alarm_handler then installed_alarm_handler = new_alarm_handler() - fiber.current_scheduler:add_task_source(installed_alarm_handler) + runtime.current_scheduler:add_task_source(installed_alarm_handler) end return installed_alarm_handler end @@ -130,9 +130,9 @@ end -- This should be called to clean up when the Alarm Handler is no longer needed. local function uninstall_alarm_handler() if installed_alarm_handler then - for i, source in ipairs(fiber.current_scheduler.sources) do + for i, source in ipairs(runtime.current_scheduler.sources) do if source == installed_alarm_handler then - table.remove(fiber.current_scheduler.sources, i) + table.remove(runtime.current_scheduler.sources, i) break end end @@ -154,8 +154,8 @@ function AlarmHandler:schedule_tasks(sched) end function AlarmHandler:block(time_to_start, t, task) - if time_to_start < fiber.current_scheduler.maxsleep then - fiber.current_scheduler:schedule_after_sleep(time_to_start, task) + if time_to_start < runtime.current_scheduler.maxsleep then + runtime.current_scheduler:schedule_after_sleep(time_to_start, task) else self.abs_timer:add_absolute(t, task) end @@ -196,7 +196,7 @@ function AlarmHandler:wait_absolute_op(t) end self:block(time_to_start, t, task) end - return op.new_base_op(nil, try, block) + return op.new_primitive(nil, try, block) end function AlarmHandler:wait_next_op(t) @@ -212,7 +212,7 @@ function AlarmHandler:wait_next_op(t) local target, _ = calculate_next(t, now) self:block(target-now, target, task) end - return op.new_base_op(nil, try, block) + return op.new_primitive(nil, try, block) end --- Indicates to the Alarm Handler that time synchronisation has been achieved (through NTP or other methods). diff --git a/src/fibers/channel.lua b/src/fibers/channel.lua index 22173b9..f665da3 100644 --- a/src/fibers/channel.lua +++ b/src/fibers/channel.lua @@ -66,7 +66,7 @@ function Channel:put_op(val) -- No space in buffer and no receivers, so block putq:push({ suspension = suspension, wrap = wrap_fn, val = val }) end - return op.new_base_op(nil, try, block) + return op.new_primitive(nil, try, block) end --- Create a get operation for the Channel. @@ -108,7 +108,7 @@ function Channel:get_op() -- Block this receiver getq:push({ suspension = suspension, wrap = wrap_fn }) end - return op.new_base_op(nil, try, block) + return op.new_primitive(nil, try, block) end --- Put a message into the Channel. diff --git a/src/fibers/context.lua b/src/fibers/context.lua index 7797b46..b8fb040 100644 --- a/src/fibers/context.lua +++ b/src/fibers/context.lua @@ -13,7 +13,7 @@ --- cancellation. -- @module context -local fiber = require "fibers.fiber" +local runtime = require "fibers.runtime" local op = require "fibers.op" local cond = require "fibers.cond" @@ -45,7 +45,7 @@ function base_context:done_op() return self.parent:done_op() else -- Background context: never cancelled, so done_op never completes. - return op.new_base_op(nil, function() return false end, function() end) + return op.new_primitive(nil, function() return false end, function() end) end end @@ -165,7 +165,7 @@ end -- @return The new with_deadline_context and a function to cancel it. local function with_deadline(parent, deadline) local ctx, cancel_fn = with_cancel(parent) - fiber.current_scheduler:schedule_at_time(deadline, { run = function() ctx:cancel("deadline_exceeded") end }) + runtime.current_scheduler:schedule_at_time(deadline, { run = function() ctx:cancel("deadline_exceeded") end }) return ctx, cancel_fn end @@ -174,7 +174,7 @@ end -- @param The timeout at which the context will be cancelled. -- @return The new with_timeout_context and a function to cancel it. local function with_timeout(parent, timeout) - return with_deadline(parent, fiber.now() + timeout) + return with_deadline(parent, runtime.now() + timeout) end --- Returns a value_context that stores a key/value pair. diff --git a/src/fibers/fiber.lua b/src/fibers/fiber.lua deleted file mode 100644 index 2fe511f..0000000 --- a/src/fibers/fiber.lua +++ /dev/null @@ -1,17 +0,0 @@ --- fibers/fiber.lua ---- --- Compatibility wrapper around fibers.runtime. --- Existing code can continue to require 'fibers.fiber'. --- @module fibers.fiber - -local runtime = require 'fibers.runtime' - -return { - current_scheduler = runtime.current_scheduler, - spawn = runtime.spawn, - now = runtime.now, - suspend = runtime.suspend, - yield = runtime.yield, - stop = runtime.stop, - main = runtime.main, -} diff --git a/src/fibers/op.lua b/src/fibers/op.lua index cf21866..c51599b 100644 --- a/src/fibers/op.lua +++ b/src/fibers/op.lua @@ -118,7 +118,7 @@ local perform -- Primitive event (leaf). -- try_fn() -> success:boolean, ... -- block_fn(suspension, wrap_fn) sets up async completion. -local function new_base_op(wrap_fn, try_fn, block_fn) +local function new_primitive(wrap_fn, try_fn, block_fn) return setmetatable( { kind = 'prim', @@ -157,42 +157,31 @@ local function with_nack(g) return setmetatable({ kind = 'with_nack', builder = g }, Event) end --- next_turn_op: primitive event that is never ready in the fast path; --- if forced to block, it completes itself on the *next scheduler turn*. -local function next_turn_op() +local function always(...) + local results = { ... } local function try() - return false + return true, unpack(results) end - - local function block(suspension, wrap_fn) - local task = suspension:complete_task(wrap_fn) - suspension.sched:schedule(task) - end - - return new_base_op(nil, try, block) + local function block() error("always: block_fn should never run") end -- never reached + return new_primitive(nil, try, block) end -local function always(value) - return new_base_op(nil, - function() return true, value end, - function() error("always: block_fn should never run") end) -end local function never() -- An event that never becomes ready - return new_base_op(nil, + return new_primitive(nil, function() return false end, function() end) end -function Event:or_else(fallback_thunk) - return choice( - self, - next_turn_op():wrap(function() - return fallback_thunk() - 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. @@ -247,24 +236,23 @@ local function new_cond(opts) suspension:complete_task(wrap_fn) end end - return new_base_op(nil, try, block) + return new_primitive(nil, try, block) end local function signal() if state.triggered then return end state.triggered = true - -- wake waiters, if any for i = 1, #state.waiters do local task = state.waiters[i] state.waiters[i] = nil if task - and task.suspension - and task.suspension:waiting() + and task.suspension + and task.suspension:waiting() then - task.suspension.sched:schedule(task) + -- Run the completion *now*, in this turn. + task:run() end end - -- fire abort handler, if any if state.abort_fn then pcall(state.abort_fn) end @@ -341,40 +329,34 @@ local function compile_event(ev, outer_wrap, out, nacks, handlers) compile_event(ev.inner, outer_wrap, out, nacks, child_handlers) else -- 'prim' - -- Core wrap chain for this leaf (no exception handling yet). - local function core(...) - return outer_wrap(ev.wrap_fn(...)) - end + local function wrapped(...) + local function core(...) + return outer_wrap(ev.wrap_fn(...)) + end - -- If there are handlers, wrap the core in them, innermost first. - local wrapped = core - if #handlers > 0 then - for i = #handlers, 1, -1 do - local h = handlers[i] - local prev = wrapped - wrapped = function(...) - local res = pack(pcall(prev, ...)) - local ok = res[1] - if ok then - return unpack(res, 2, res.n) + 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 - -- res[2] is the exception (CML: pass exn to handler) - local ex = res[2] - local hev = h(ex) - -- Handler returns an event; synchronise on it. - return perform(hev) end end - end - - local final_wrap = function(...) - return wrapped(...) + return f(...) end out[#out + 1] = { try_fn = ev.try_fn, block_fn = ev.block_fn, - wrap = final_wrap, + wrap = wrapped, nacks = nacks, } end @@ -392,28 +374,30 @@ end -- - For each loser leaf, signal any nacks not in the winner set. -- - Each cond object is responsible for idempotence. local function trigger_nacks(ops, winner_index) - assert(winner_index, "trigger_nacks: no winner_index (internal error)") - - local winner_set = {} - local wnacks = ops[winner_index].nacks - if wnacks then - for i = 1, #wnacks do - winner_set[wnacks[i]] = true + local winner_set + if winner_index then + winner_set = {} + local wnacks = ops[winner_index].nacks + if wnacks then + for i = 1, #wnacks do + winner_set[wnacks[i]] = true + end end end - local signaled = {} + local function is_winner_cond(cond) + return winner_set and winner_set[cond] or false + end + + local signalled = {} for i = 1, #ops do - if i ~= winner_index then + if not winner_index or i ~= winner_index then local nacks = ops[i].nacks if nacks then for j = #nacks, 1, -1 do local cond = nacks[j] - if cond - and not winner_set[cond] - and not signaled[cond] - then - signaled[cond] = true + if cond and not is_winner_cond(cond) and not signalled[cond] then + signalled[cond] = true cond.signal() end end @@ -444,6 +428,31 @@ local function apply_wrap(wrap, retval) return wrap(unpack(retval, 2, retval.n)) end +function Event:or_else(fallback_thunk) + assert(type(fallback_thunk) == "function", "or_else expects a function") + + return guard(function() + -- Compile `self` once for this synchronisation. + local leaves = compile_event(self) + + -- Non-blocking attempt to commit to `self`. + local idx, retval = try_ready(leaves) + if idx then + -- Normal CML semantics: `self` wins, fire nacks for losers. + trigger_nacks(leaves, idx) + local results = { apply_wrap(leaves[idx].wrap, retval) } + return always(unpack(results)) + end + + -- No leaf of `self` is ready now → `self` loses as a whole. + -- Fire all nacks/abort handlers hanging off `self`. + trigger_nacks(leaves, nil) + + local results = { fallback_thunk() } + return always(unpack(results)) + end) +end + ---------------------------------------------------------------------- -- Blocking choice path ---------------------------------------------------------------------- @@ -463,17 +472,18 @@ end perform = function(ev) local leaves = compile_event(ev) - -- Fast path + -- Fast path: try once using the same semantics as fast_commit(). local idx, retval = try_ready(leaves) if idx then trigger_nacks(leaves, idx) return apply_wrap(leaves[idx].wrap, retval) end - -- Slow path + -- Slow path: we now block all leaves using block_choice_op. local suspended = pack(runtime.suspend(block_choice_op, leaves)) local wrap = suspended[1] + -- Identify winning leaf by its wrap function. local winner_index for i, leaf in ipairs(leaves) do if leaf.wrap == wrap then @@ -568,7 +578,7 @@ end return { perform = perform, perform_raw = perform, - new_base_op = new_base_op, -- primitive event constructor + new_primitive = new_primitive, -- primitive event constructor choice = choice, guard = guard, with_nack = with_nack, @@ -576,5 +586,6 @@ return { bracket = bracket, always = always, never = never, + Event = Event, -- Event instances have methods: wrap, on_abort. } diff --git a/src/fibers/performer.lua b/src/fibers/performer.lua index dc5338f..269ec62 100644 --- a/src/fibers/performer.lua +++ b/src/fibers/performer.lua @@ -1,28 +1,32 @@ -- fibers/performer.lua --- -- Scope-aware performer. --- This is the preferred entry point for synchronising on events --- in normal code. It delegates to the current scope (or the root --- scope) and uses the raw op.perform under the hood. --- --- Policies (failure, cancellation, etc.) will be wired into the --- Scope methods in later stages. +-- Preferred entry point for synchronising on events in normal code. +-- Delegates to the current scope if available, otherwise falls back +-- to the raw op.perform. -- -- @module fibers.performer -local scope = require 'fibers.scope' -local op = require 'fibers.op' +local op = require 'fibers.op' + +local scope_mod local M = {} ---- Perform an event under the current scope. --- For now this simply calls the raw op.perform; in later stages --- Scope:sync can wrap events with policies before performing. +local function current_scope() + if not scope_mod then + scope_mod = require 'fibers.scope' + end + return scope_mod.current and scope_mod.current() or nil +end + function M.perform(ev) - -- At this stage, scopes do not transform events, so we just - -- ensure there *is* a scope and call the raw engine. - local _ = scope.current() -- touch to ensure root initialises - return op.perform(ev) + local s = current_scope() + if s and s.sync then + return s:sync(ev) + else + return op.perform_raw(ev) + end end return M diff --git a/src/fibers/pollio.lua b/src/fibers/pollio.lua index e8c2ddd..37cd2cc 100644 --- a/src/fibers/pollio.lua +++ b/src/fibers/pollio.lua @@ -3,12 +3,12 @@ -- File events. local op = require 'fibers.op' -local fiber = require 'fibers.fiber' +local runtime = require 'fibers.runtime' local epoll = require 'fibers.epoll' local sc = require 'fibers.utils.syscall' local bit = rawget(_G, "bit") or require 'bit32' -local perform = op.perform +local perform = require 'fibers.performer'.perform local PollIOHandler = {} PollIOHandler.__index = PollIOHandler @@ -58,14 +58,14 @@ function PollIOHandler:fd_readable_op(fd) local function try() return false end local block = make_block_fn( fd, self.waiting_for_readable, self.epoll, epoll.RD) - return op.new_base_op(nil, try, block) + return op.new_primitive(nil, try, block) end function PollIOHandler:fd_writable_op(fd) local function try() return false end local block = make_block_fn( fd, self.waiting_for_writable, self.epoll, epoll.WR) - return op.new_base_op(nil, try, block) + return op.new_primitive(nil, try, block) end function PollIOHandler:stream_readable_op(stream) @@ -73,7 +73,7 @@ function PollIOHandler:stream_readable_op(stream) local function try() return not stream.rx:is_empty() end local block = make_block_fn( fd, self.waiting_for_readable, self.epoll, epoll.RD) - return op.new_base_op(nil, try, block) + return op.new_primitive(nil, try, block) end -- A stream_writable_op is the same as fd_writable_op, as a stream's @@ -146,7 +146,7 @@ local function install_poll_io_handler() if installed == 1 then installed_poll_handler = new_poll_io_handler() -- file.set_blocking_handler(installed_poll_handler) - fiber.current_scheduler:add_task_source(installed_poll_handler) + runtime.current_scheduler:add_task_source(installed_poll_handler) end return installed_poll_handler end @@ -156,9 +156,9 @@ local function uninstall_poll_io_handler() if installed == 0 then -- file.set_blocking_handler(nil) -- FIXME: Remove task source. - for i, source in ipairs(fiber.current_scheduler.sources) do + for i, source in ipairs(runtime.current_scheduler.sources) do if source == installed_poll_handler then - table.remove(fiber.current_scheduler.sources, i) + table.remove(runtime.current_scheduler.sources, i) break end end diff --git a/src/fibers/runtime.lua b/src/fibers/runtime.lua index 12a1981..055351a 100644 --- a/src/fibers/runtime.lua +++ b/src/fibers/runtime.lua @@ -7,7 +7,7 @@ local sched = require 'fibers.sched' -local current_fiber +local _current_fiber local current_scheduler = sched.new() --- Fiber class @@ -23,8 +23,8 @@ local function spawn(fn) -- Capture the traceback local tb = debug.traceback("", 2):match("\n[^\n]*\n(.*)") or "" -- If we're inside another fiber, append the traceback to the parent's traceback - if current_fiber and current_fiber.traceback then - tb = tb .. "\n" .. current_fiber.traceback + if _current_fiber and _current_fiber.traceback then + tb = tb .. "\n" .. _current_fiber.traceback end current_scheduler:schedule( @@ -42,12 +42,12 @@ end -- @tparam vararg ... The arguments to pass to the fiber. function Fiber:resume(wrap, ...) assert(self.alive, "dead fiber") -- checks that the fiber is alive - local saved_current_fiber = current_fiber -- shift the old current fiber into a safe place - current_fiber = self -- we are the new current fiber + local saved_current_fiber = _current_fiber -- shift the old current fiber into a safe place + _current_fiber = self -- we are the new current fiber local ok, err = coroutine.resume(self.coroutine, wrap, ...) -- rev up our coroutine -- 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 + _current_fiber = saved_current_fiber if not ok then print('Error while running fiber: ' .. tostring(err)) print(debug.traceback(self.coroutine)) @@ -59,10 +59,10 @@ end Fiber.run = Fiber.resume function Fiber:suspend(block_fn, ...) - assert(current_fiber == self) + assert(_current_fiber == self) -- The block_fn should arrange to reschedule the fiber when it -- becomes runnable. - block_fn(current_scheduler, current_fiber, ...) + block_fn(current_scheduler, _current_fiber, ...) return coroutine.yield() end @@ -71,8 +71,8 @@ function Fiber:get_traceback() end --- Returns the current Fiber object, or nil if not inside a fiber. -local function current() - return current_fiber +local function current_fiber() + return _current_fiber end local function now() @@ -80,18 +80,16 @@ local function now() end local function suspend(block_fn, ...) - return current_fiber:suspend(block_fn, ...) -end - -local function schedule(scheduler, fiber) - scheduler:schedule(fiber) + return _current_fiber:suspend(block_fn, ...) end --- Suspends execution of the current fiber. -- The fiber will be resumed when the scheduler is ready to run it again. -- @function yield local function yield() - return suspend(schedule) + return suspend(function(scheduler, fiber) + scheduler:schedule(fiber) + end) end --- Stops the current scheduler from running more tasks. @@ -110,7 +108,7 @@ end return { -- core runtime state current_scheduler = current_scheduler, - current_fiber = current, + current_fiber = current_fiber, -- time and suspension now = now, diff --git a/src/fibers/scope.lua b/src/fibers/scope.lua index 8e83b68..9827ace 100644 --- a/src/fibers/scope.lua +++ b/src/fibers/scope.lua @@ -1,24 +1,35 @@ -- fibers/scope.lua --- --- Basic scope module (stage 2). +-- Scope module (structured concurrency). +-- -- Provides a tree of Scope objects and a per-fiber “current scope”. -- --- At this stage: +-- Semantics: -- - A single root scope exists for the process. --- - Each fiber may have an associated current scope. +-- - 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). --- - scope.root():spawn(fn, ...) spawns a new fiber whose current --- scope is that scope for the duration of fn. +-- 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 +-- for the duration of fn. +-- - scope.with_ev(build_ev) creates a *first-class Event* that +-- represents running a child scope whose body is build_ev(child). -- --- Policies (failure, cancellation, resource limits) are *not* yet --- implemented; they will be added in later stages. +-- 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. -- -- @module fibers.scope -local runtime = require 'fibers.runtime' +local runtime = require 'fibers.runtime' +local op = require 'fibers.op' +local waitgroup = require 'fibers.waitgroup' local unpack = rawget(table, "unpack") or _G.unpack local pack = rawget(table, "pack") or function(...) @@ -35,12 +46,32 @@ local fiber_scopes = setmetatable({}, { __mode = "k" }) local root_scope local global_scope +-- Internal: current fiber object, or nil if not in a fiber. +local function current_fiber() + return runtime.current_fiber() +end + -- Internal: create a new Scope with the given parent. local function new_scope(parent) local s = setmetatable({ - _parent = parent, - _children = {}, + _parent = parent, + _children = {}, + + -- Status and failure tracking + _status = "running", -- "running" | "ok" | "failed" | "cancelled" + _error = nil, -- primary error / cancellation cause + _failures = {}, -- additional failures + failure_mode = "fail_fast", -- only fail_fast for now + + -- Concurrency tracking + _wg = waitgroup.new(), -- waitgroup for child fibers + _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 }, Scope) + if parent then local children = parent._children children[#children + 1] = s @@ -57,14 +88,6 @@ local function root() return root_scope end --- Internal: current fiber object, or nil if not in a fiber. -local function current_fiber() - if runtime.current_fiber then - return runtime.current_fiber() - end - return nil -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. @@ -77,51 +100,131 @@ local function current() end --- Internal helper: run fn(scope, ...) with 'scope' as current in this context. -local function with_scope(scope, fn, ...) +-- Returns a packed result table: { n = ..., [1] = ok, [2..n] = values }. +local function with_scope(scope_obj, fn, ...) local fib = current_fiber() if fib then local prev = fiber_scopes[fib] - fiber_scopes[fib] = scope + fiber_scopes[fib] = scope_obj - local res = pack(pcall(fn, scope, ...)) + local res = pack(pcall(fn, scope_obj, ...)) fiber_scopes[fib] = prev - if not res[1] then error(res[2]) end - return unpack(res, 2, res.n) + return res else local prev = global_scope or root() - global_scope = scope + global_scope = scope_obj - local res = pack(pcall(fn, scope, ...)) + local res = pack(pcall(fn, scope_obj, ...)) global_scope = prev - if not res[1] then error(res[2]) end - return unpack(res, 2, res.n) + return res end end +---------------------------------------------------------------------- +-- Core scope API +---------------------------------------------------------------------- + --- 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) - return with_scope(child, body_fn, ...) + + -- 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) + else + error(child._error) + end end ---- Create a new child scope of this scope. --- Does not change the current scope or run any code. +--- Create a new child scope of this scope (no body, no current() change). function Scope:new_child() return new_scope(self) end +--- Register a deferred handler to run at scope exit (LIFO). +-- handler :: function(Scope) +function Scope:defer(handler) + local defers = self._defers + defers[#defers + 1] = handler +end + +--- Cancel this scope and its children with an optional reason. +-- Idempotent: multiple calls are safe. +function Scope:cancel(reason) + local r = reason or self._error or "scope cancelled" + + if self._status == "running" 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. + self._cancel_cond.signal() + + -- Propagate to children. + local children = self._children + for i = 1, #children do + local c = children[i] + if c then + c:cancel(r) + end + end +end + --- Spawn a child fiber attached to this scope. -- fn :: function(Scope, ...): () -- ... :: arguments passed to fn -- --- The new fiber's current scope is set to 'self' for the duration --- of fn, and any previous mapping for that fiber (normally none) --- is restored afterwards. +-- Fail-fast semantics: +-- - If fn raises, this scope's status becomes "failed", its primary +-- error is set (if not already), and cancel() is invoked. function Scope:spawn(fn, ...) local args = { ... } + self._wg:add(1) + runtime.spawn(function() local fib = current_fiber() local prev = fib and fiber_scopes[fib] or nil @@ -141,10 +244,23 @@ function Scope:spawn(fn, ...) end if not ok then - -- Stage 2: preserve existing behaviour (unhandled errors - -- still crash the process via runtime/Fiber:resume). - error(err) + -- 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 @@ -163,9 +279,168 @@ function Scope:children() return out end +--- Return this scope's status and primary error. +-- status :: "running" | "ok" | "failed" | "cancelled" +-- err :: any | nil +function Scope:status() + return self._status, self._error +end + +--- Return a shallow copy of additional failures. +function Scope:failures() + local out = {} + local f = self._failures or {} + for i, v in ipairs(f) do + out[i] = v + end + return out +end + +--- Event that fires once the scope has reached a terminal status. +-- Returns (status, error) when synchronised. +function Scope:join_ev() + local ev = self._join_cond.wait_op() + return ev:wrap(function() + return self._status, self._error + end) +end + +--- Event that fires when the scope is cancelled or fails. +-- Returns the cancellation/failure reason when synchronised. +function Scope:done_ev() + local ev = self._cancel_cond.wait_op() + return ev:wrap(function() + return self._error or "scope cancelled" + end) +end + +---------------------------------------------------------------------- +-- Failure + cancellation wrapping for Events +---------------------------------------------------------------------- + +-- Internal: cancellation event used when running events under this scope. +local function cancel_event(self) + local ev = self._cancel_cond.wait_op() + return ev:wrap(function() + error(self._error or "scope cancelled") + end) +end + +--- Transform an event to obey this scope's failure and cancellation policy. +-- Returns a new Event; does not perform it. +function Scope:run_ev(ev) + local cancel_ev = cancel_event(self) + return op.choice(ev, cancel_ev) +end + +--- Synchronise on an event under this scope. +-- Equivalent to op.perform(self:run_ev(ev)). +function Scope:sync(ev) + return op.perform(self:run_ev(ev)) +end + +---------------------------------------------------------------------- +-- Scope as an *event*: scope.with_ev +---------------------------------------------------------------------- + +--- Event-level API: create a child scope whose body is an Event. +-- +-- build_ev :: function(child_scope :: Scope) -> Event +-- +-- The returned Event, when performed, will: +-- * 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, +-- and signal join_ev(); +-- * propagate the inner Event's result or error. +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] + fiber_scopes[fib] = child + return { kind = "fiber", fib = fib, prev = prev } + else + local prev = global_scope or root() + global_scope = child + return { kind = "global", prev = prev } + end + end + + local function release(token, aborted) + -- Restore the previous current scope. + if token.kind == "fiber" then + fiber_scopes[token.fib] = token.prev + else + global_scope = token.prev + end + + -- If the event was aborted and the scope is still running, + -- treat that as cancellation of the scope. + if aborted and child._status == "running" then + child._status = "cancelled" + child._error = child._error or "scope aborted" + 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() + 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 + end + + return op.bracket(acquire, release, use) + end) +end + return { - root = root, - current = current, - run = run, - Scope = Scope, + root = root, + current = current, + run = run, + with_ev = with_ev, + Scope = Scope, } diff --git a/src/fibers/sleep.lua b/src/fibers/sleep.lua index 115fd07..af57529 100644 --- a/src/fibers/sleep.lua +++ b/src/fibers/sleep.lua @@ -5,7 +5,7 @@ -- @module fibers.sleep local op = require 'fibers.op' -local fiber = require 'fibers.fiber' +local runtime = require 'fibers.runtime' local perform = require 'fibers.performer'.perform @@ -20,12 +20,12 @@ local perform = require 'fibers.performer'.perform -- @treturn operation The created operation. local function sleep_until_op(t) local function try() - return t <= fiber.now() + return t <= runtime.now() end local function block(suspension, wrap_fn) suspension.sched:schedule_at_time(t, suspension:complete_task(wrap_fn)) end - return op.new_base_op(nil, try, block) + return op.new_primitive(nil, try, block) end --- Put the current fiber to sleep until time t. @@ -42,7 +42,7 @@ local function sleep_op(dt) local function block(suspension, wrap_fn) suspension.sched:schedule_after_sleep(dt, suspension:complete_task(wrap_fn)) end - return op.new_base_op(nil, try, block) + return op.new_primitive(nil, try, block) end --- Put the current fiber to sleep for a duration dt. diff --git a/src/fibers/stream.lua b/src/fibers/stream.lua index ea790dd..3daa769 100644 --- a/src/fibers/stream.lua +++ b/src/fibers/stream.lua @@ -126,7 +126,7 @@ local function core_write_op(stream, buf, count, flush_needed) return ... end - return op.new_base_op(wrap, try, block) + return op.new_primitive(wrap, try, block) end @@ -217,7 +217,7 @@ local function core_read_op(stream, buf, min, max, terminator) return ... end - return op.new_base_op(wrap, try, block) + return op.new_primitive(wrap, try, block) end function Stream:partial_read() diff --git a/src/fibers/waitgroup.lua b/src/fibers/waitgroup.lua index 83c70bc..a0dbaa2 100644 --- a/src/fibers/waitgroup.lua +++ b/src/fibers/waitgroup.lua @@ -45,7 +45,7 @@ function Waitgroup:wait_op() end end - return op.new_base_op(nil, try, block) + return op.new_primitive(nil, try, block) end function Waitgroup:wait() diff --git a/tests/test.lua b/tests/test.lua index 9d43180..a7f87ae 100644 --- a/tests/test.lua +++ b/tests/test.lua @@ -12,14 +12,14 @@ local modules = { { 'stream', 'socket' }, { 'timer' }, { 'sched' }, - { 'fiber' }, + { 'runtime' }, { 'channel' }, { 'queue' }, { 'cond' }, { 'sleep' }, { 'epoll' }, { 'pollio' }, - -- { 'exec' }, + { 'exec' }, { 'waitgroup' }, { 'alarm' }, { 'context' }, diff --git a/tests/test_alarm.lua b/tests/test_alarm.lua index ebd8a9a..1a930ef 100644 --- a/tests/test_alarm.lua +++ b/tests/test_alarm.lua @@ -4,7 +4,7 @@ print('testing: fibers.alarm') -- look one level up package.path = "../src/?.lua;" .. package.path -local fiber = require 'fibers.fiber' +local fibers = require 'fibers' local alarm = require 'fibers.alarm' local sleep = require 'fibers.sleep' local sc = require 'fibers.utils.syscall' @@ -124,7 +124,7 @@ local function buffer_test() -- first absolutes - easy to test local t_sleep, t_sleep_before_realtime = 1, 2 alarm.clock_desynced() - fiber.spawn(function () + fibers.spawn(function () sleep.sleep(t_sleep_before_realtime) alarm.clock_synced() end) @@ -135,7 +135,7 @@ local function buffer_test() -- next let's do nexts local msec_target = 333 alarm.clock_desynced() - fiber.spawn(function () + fibers.spawn(function () sleep.sleep(t_sleep_before_realtime) alarm.clock_synced() end) @@ -228,7 +228,7 @@ local function test_next_calc() print("complete!") end -fiber.spawn(function () +local function main() abs_test(2) next_error() next_msec_test(666) @@ -242,7 +242,6 @@ fiber.spawn(function () buffer_test() validate_next_table_test() test_next_calc() - fiber.stop() -end) +end -fiber.main() +fibers.run(main) diff --git a/tests/test_channel.lua b/tests/test_channel.lua index 81ed78a..bb255a5 100644 --- a/tests/test_channel.lua +++ b/tests/test_channel.lua @@ -2,20 +2,20 @@ print('testing: fibers.channel') package.path = "../src/?.lua;" .. package.path -local fiber = require 'fibers.fiber' +local fibers = require 'fibers' local channel = require 'fibers.channel' local function test_unbuffered() local chan = channel.new() - fiber.spawn(function() chan:put(42) end) + fibers.spawn(function() chan:put(42) end) assert(chan:get() == 42, "basic transfer") local received, signal = true, channel.new() - fiber.spawn(function() + fibers.spawn(function() received = chan:get() signal:put(true) end) - fiber.spawn(function() + fibers.spawn(function() chan:put(nil) signal:put(true) end) @@ -28,20 +28,20 @@ end local function test_buffered() local chan, signal = channel.new(2), channel.new() - fiber.spawn(function() + fibers.spawn(function() for i = 1, 4 do chan:put(i) end signal:put(true) end) - fiber.spawn(function() + fibers.spawn(function() for i = 1, 2 do assert(chan:get() == i) end end) signal:get() - fiber.spawn(function() + fibers.spawn(function() chan:put(5) signal:put(true) end) @@ -58,7 +58,7 @@ local function test_unbounded() for i = 1, 1000 do assert(chan:get() == i) end local blocked = true - fiber.spawn(function() + fibers.spawn(function() chan:get() blocked = false end) @@ -68,11 +68,11 @@ end local function test_concurrent() local chan, signal, results = channel.new(1), channel.new(), {} - fiber.spawn(function() + fibers.spawn(function() for i = 1, 11 do chan:put(i) end signal:put() end) - fiber.spawn(function() + fibers.spawn(function() for _ = 1, 10 do table.insert(results, chan:get()) end signal:put() end) @@ -82,13 +82,12 @@ local function test_concurrent() print("Concurrent passed") end -fiber.spawn(function() +local function main() test_unbuffered() test_buffered() test_unbounded() test_concurrent() print("All channel tests passed!") - fiber.stop() -end) +end -fiber.main() +fibers.run(main) diff --git a/tests/test_cond.lua b/tests/test_cond.lua index f3d3a61..8ffd57e 100644 --- a/tests/test_cond.lua +++ b/tests/test_cond.lua @@ -5,7 +5,7 @@ print('testing: fibers.cond') package.path = "../src/?.lua;" .. package.path local cond = require 'fibers.cond' -local fiber = require 'fibers.fiber' +local runtime = require 'fibers.runtime' local sleep = require 'fibers.sleep' local sc = require 'fibers.utils.syscall' @@ -14,22 +14,22 @@ local equal = require 'fibers.utils.helper'.equal local c, log = cond.new(), {} local function record(x) table.insert(log, x) end -fiber.spawn(function() +runtime.spawn(function() record('a'); c:wait(); record('b') end) -fiber.spawn(function() +runtime.spawn(function() record('c'); c:signal(); record('d') end) assert(equal(log, {})) -fiber.current_scheduler:run() -assert(equal(log, { 'a', 'c', 'd' })) -fiber.current_scheduler:run() -assert(equal(log, { 'a', 'c', 'd', 'b' })) +runtime.current_scheduler:run() +-- assert(equal(log, { 'a', 'c', 'd' })) +runtime.current_scheduler:run() +assert(equal(log, { 'a', 'c', 'b', 'd' })) -fiber.spawn(function() +runtime.spawn(function() local fiber_count = 1e3 for _ = 1, fiber_count do - fiber.spawn(function() c:wait(); end) + runtime.spawn(function() c:wait(); end) end sleep.sleep(1) @@ -39,8 +39,8 @@ fiber.spawn(function() local end_time = sc.monotime() print("Time taken to signal fiber: ", (end_time - start_time) / fiber_count) - fiber.stop() + runtime.stop() end) -fiber.main() +runtime.main() print('test: ok') diff --git a/tests/test_context.lua b/tests/test_context.lua index 2a28dbc..1ca847a 100644 --- a/tests/test_context.lua +++ b/tests/test_context.lua @@ -5,7 +5,7 @@ print('testing: fibers.context') package.path = "../src/?.lua;" .. package.path local context = require 'fibers.context' -local fiber = require 'fibers.fiber' +local fibers = require 'fibers' local sleep = require 'fibers.sleep' local perform = require 'fibers.performer'.perform @@ -24,13 +24,13 @@ local function test_with_cancel() local child_ctx, _ = context.with_cancel(ctx) local is_cancelled = false - fiber.spawn(function() + fibers.spawn(function() perform(child_ctx:done_op()) is_cancelled = true end) cancel() - fiber.yield() -- Give time for cancellation to propagate + sleep.sleep(0.1) -- Give time for cancellation to propagate assert(is_cancelled, "Child context should be cancelled when parent is cancelled") print("test_with_cancel passed") @@ -55,7 +55,7 @@ local function test_with_timeout() local ctx, _ = context.with_timeout(parent, 0.01) local is_cancelled = false - fiber.spawn(function() + fibers.spawn(function() perform(ctx:done_op()) is_cancelled = true end) @@ -74,7 +74,7 @@ local function test_custom_cause() local custom_cause = "Custom Cancel Reason" cancel(custom_cause) - fiber.yield() -- Give time for cancellation to propagate + sleep.sleep(0.01) -- Give time for cancellation to propagate assert(ctx:err() == custom_cause, "Context should have the custom cancel cause") assert(ctx_2:err() == custom_cause, "Child context should have the custom cancel cause") @@ -88,7 +88,7 @@ local function test_cancel_on_with_value() local ctx_3, _ = context.with_value(ctx_2, "key2", "value2") cancel('cancelled') - fiber.yield() -- Give time for cancellation to propagate + sleep.sleep(0.01) -- Give time for cancellation to propagate assert(ctx:err() == 'cancelled', "Context should have cancel cause") assert(ctx_2:err() == 'cancelled', "Child context should have cancel cause") @@ -98,7 +98,7 @@ local function test_cancel_on_with_value() assert(ctx_3:value('key2') == 'value2', "Child context should have the value") end -- Run all tests -fiber.spawn(function() +local function main() test_background() test_with_cancel() test_with_value() @@ -107,7 +107,6 @@ fiber.spawn(function() test_cancel_on_with_value() print("All tests passed") - fiber.stop() -end) +end -fiber.main() +fibers.run(main) diff --git a/tests/test_exec.lua b/tests/test_exec.lua index baedeb8..6cbc10f 100644 --- a/tests/test_exec.lua +++ b/tests/test_exec.lua @@ -3,7 +3,7 @@ -- look one level up package.path = "../src/?.lua;" .. package.path -local fiber = require 'fibers.fiber' +local fibers = require 'fibers' local sleep = require 'fibers.sleep' local channel = require "fibers.channel" local exec = require 'fibers.exec' @@ -70,7 +70,7 @@ local function test_io_redirection() local err = cmd:start() assert(cmd:start(), "Expected error on starting command twice") assert(err == nil, "Expected no error but got:", err) - fiber.spawn(function () + fibers.spawn(function () for _, v in ipairs(msgs) do stdin_pipe:write(v) signal_chan:get() @@ -118,7 +118,7 @@ local function test_cancel_during_output() local cmd = exec.command_context(ctx, '/bin/sh', '-c', 'for i in $(seq 1 10000); do echo y; sleep 0.001; done') :setpgid(true) - fiber.spawn(function() + fibers.spawn(function() -- Let it run for a short moment, then cancel sleep.sleep(0.00001) cancel() @@ -149,14 +149,14 @@ local function test_cleanup_on_crash(id) local script = [[ package.path = "../src/?.lua;" .. package.path - local fiber = require 'fibers.fiber' + local fibers = require 'fibers' local exec = require 'fibers.exec' local sleep = require 'fibers.sleep' local pollio = require 'fibers.pollio' local sc = require 'fibers.utils.syscall' pollio.install_poll_io_handler() - fiber.spawn(function () + local function main() local cmd = exec.command('sleep', '1') cmd:setprdeathsig(sc.SIGKILL) local err = cmd:start() @@ -167,9 +167,9 @@ local function test_cleanup_on_crash(id) io.stdout:flush() sleep.sleep(0.01) print(obj.obj) - end) + end - fiber.main() + fibers.run(main) ]] -- Write the script to a temporary file @@ -219,6 +219,7 @@ local function test_cleanup_on_crash(id) assert(not is_running, "Child sleep command is still running") end + -- Main test function local function main() local pid = sc.getpid() @@ -241,7 +242,7 @@ local function main() local wg = waitgroup.new() for i = 1, reps do wg:add(1) - fiber.spawn(function () + fibers.spawn(function () v(i) wg:done() end) @@ -252,8 +253,6 @@ local function main() print(k..": passed!") end print("test: ok") - fiber.stop() end -fiber.spawn(main) -fiber.main() +fibers.run(main) diff --git a/tests/test_op.lua b/tests/test_op.lua index da3ce65..3715fea 100644 --- a/tests/test_op.lua +++ b/tests/test_op.lua @@ -4,8 +4,8 @@ print("testing: fibers.op") -- look one level up package.path = "../src/?.lua;" .. package.path -local op = require 'fibers.op' -local fiber = require 'fibers.fiber' +local op = require 'fibers.op' +local runtime = require 'fibers.runtime' local perform, choice = require 'fibers.performer'.perform, op.choice local always = op.always @@ -26,7 +26,7 @@ local function async_task(val) local t = suspension:complete_task(wrap_fn, val) suspension.sched:schedule(t) end - local ev = op.new_base_op(nil, try_fn, block_fn) + local ev = op.new_primitive(nil, try_fn, block_fn) return ev, function() return tries end end @@ -34,7 +34,7 @@ end -- Run all tests inside a single top-level fiber ------------------------------------------------------------ -fiber.spawn(function() +runtime.spawn(function() -------------------------------------------------------- -- 1) Base event: perform, or_else, wrap @@ -54,7 +54,7 @@ fiber.spawn(function() local palt2 = perform(ev2) assert(palt2 == 99, "base: or_else should use fallback when event can't commit") - -- or_else: async event should still win (gets a chance next turn) + -- or_else: async event is not ready now, so fallback wins do local fallback_called = false local ev_async, tries = async_task(123) @@ -63,10 +63,11 @@ fiber.spawn(function() return -1 end) local r = perform(ev) - assert(r == 123, "or_else(async): expected main event to win") - assert(tries() == 1, "async_task: try_fn not called exactly once") - assert(fallback_called == false, - "or_else(async): fallback should not run when event commits") + + assert(r == -1, "or_else(async): expected fallback to win") + assert(tries() == 1, "async_task: try_fn should be called exactly once") + assert(fallback_called == true, + "or_else(async): fallback should run when event is not ready") end -- nested wrap: ((x + 1) * 2) @@ -142,7 +143,7 @@ fiber.spawn(function() local guarded_nack = op.guard(function() guard_calls = guard_calls + 1 return op.with_nack(function(nack_ev) - fiber.spawn(function() + runtime.spawn(function() perform(nack_ev) cancelled = true end) @@ -154,7 +155,7 @@ fiber.spawn(function() local v2 = perform(ev2) assert(v2 == "OK", "guard+with_nack: wrong winner") - fiber.yield() + runtime.yield() assert(cancelled == true, "guard+with_nack: nack not fired") assert(guard_calls == 1, "guard+with_nack: builder not once") end @@ -185,7 +186,7 @@ fiber.spawn(function() do local cancelled = false local with_nack_ev = op.with_nack(function(nack_ev) - fiber.spawn(function() + runtime.spawn(function() perform(nack_ev) cancelled = true end) @@ -196,7 +197,7 @@ fiber.spawn(function() local v = perform(ev) assert(v == "WIN", "with_nack win: wrong winner") - fiber.yield() + runtime.yield() assert(cancelled == false, "with_nack win: nack fired unexpectedly") end @@ -204,7 +205,7 @@ fiber.spawn(function() do local cancelled = false local with_nack_ev = op.with_nack(function(nack_ev) - fiber.spawn(function() + runtime.spawn(function() perform(nack_ev) cancelled = true end) @@ -215,7 +216,7 @@ fiber.spawn(function() local v = perform(ev) assert(v == "OTHER", "with_nack loss: wrong winner") - fiber.yield() + runtime.yield() assert(cancelled == true, "with_nack loss: nack did not fire") end @@ -225,13 +226,13 @@ fiber.spawn(function() local outer_cancelled, inner_cancelled = false, false local outer = op.with_nack(function(outer_nack_ev) - fiber.spawn(function() + runtime.spawn(function() perform(outer_nack_ev) outer_cancelled = true end) return op.with_nack(function(inner_nack_ev) - fiber.spawn(function() + runtime.spawn(function() perform(inner_nack_ev) inner_cancelled = true end) @@ -244,7 +245,7 @@ fiber.spawn(function() assert(v == "INNER_WIN", "nested with_nack: wrong result") - fiber.yield() + runtime.yield() assert(outer_cancelled == false, "nested with_nack: outer nack fired unexpectedly") assert(inner_cancelled == false, @@ -511,7 +512,7 @@ fiber.spawn(function() end end print("fibers.op tests: ok") - fiber.stop() + runtime.stop() end) -fiber.main() +runtime.main() diff --git a/tests/test_pollio.lua b/tests/test_pollio.lua index 00f76a0..5ff04b1 100644 --- a/tests/test_pollio.lua +++ b/tests/test_pollio.lua @@ -6,7 +6,7 @@ package.path = "../src/?.lua;" .. package.path local pollio = require 'fibers.pollio' local file_stream = require 'fibers.stream.file' -local fiber = require 'fibers.fiber' +local runtime = require 'fibers.runtime' local equal = require 'fibers.utils.helper'.equal local log = {} @@ -14,21 +14,21 @@ local function record(x) table.insert(log, x) end -- local handler = new_poll_io_handler() -- file.set_blocking_handler(handler) --- fiber.current_scheduler:add_task_source(handler) +-- runtime.current_scheduler:add_task_source(handler) pollio.install_poll_io_handler() -fiber.current_scheduler:run() +runtime.current_scheduler:run() assert(equal(log, {})) local rd, wr = file_stream.pipe() local message = "hello, world\n" -fiber.spawn(function() +runtime.spawn(function() record('rd-a') local str = rd:read_some_chars() record('rd-b') record(str) end) -fiber.spawn(function() +runtime.spawn(function() record('wr-a') wr:write(message) record('wr-b') @@ -36,9 +36,9 @@ fiber.spawn(function() record('wr-c') end) -fiber.current_scheduler:run() +runtime.current_scheduler:run() assert(equal(log, { 'rd-a', 'wr-a', 'wr-b', 'wr-c' })) -fiber.current_scheduler:run() +runtime.current_scheduler:run() assert(equal(log, { 'rd-a', 'wr-a', 'wr-b', 'wr-c', 'rd-b', message })) print('test: ok') diff --git a/tests/test_queue.lua b/tests/test_queue.lua index e475b98..6d432d0 100644 --- a/tests/test_queue.lua +++ b/tests/test_queue.lua @@ -5,14 +5,14 @@ print('testing: fibers.queue') package.path = "../src/?.lua;" .. package.path local queue = require 'fibers.queue' -local fiber = require 'fibers.fiber' +local runtime = require 'fibers.runtime' local helper = require 'fibers.utils.helper' local equal = helper.equal local log = {} local function record(x) table.insert(log, x) end -fiber.spawn(function() +runtime.spawn(function() local q = queue.new() record('a') q:put('b') @@ -29,7 +29,7 @@ end) local function run(...) log = {} - fiber.current_scheduler:run() + runtime.current_scheduler:run() assert(equal(log, { ... })) end diff --git a/tests/test_fiber.lua b/tests/test_runtime.lua similarity index 71% rename from tests/test_fiber.lua rename to tests/test_runtime.lua index 2f3e31d..dae321a 100644 --- a/tests/test_fiber.lua +++ b/tests/test_runtime.lua @@ -4,25 +4,25 @@ print('testing: fibers.fiber') -- look one level up package.path = "../src/?.lua;" .. package.path -local fiber = require 'fibers.fiber' +local runtime = require 'fibers.runtime' local sc = require 'fibers.utils.syscall' local equal = require 'fibers.utils.helper'.equal local log = {} local function record(x) table.insert(log, x) end -fiber.spawn(function() - record('a'); fiber.yield(); record('b'); fiber.yield(); record('c') +runtime.spawn(function() + record('a'); runtime.yield(); record('b'); runtime.yield(); record('c') end) assert(equal(log, {})) -fiber.current_scheduler:run() +runtime.current_scheduler:run() assert(equal(log, {'a'})) -fiber.current_scheduler:run() +runtime.current_scheduler:run() assert(equal(log, {'a', 'b'})) -fiber.current_scheduler:run() +runtime.current_scheduler:run() assert(equal(log, {'a', 'b', 'c'})) -fiber.current_scheduler:run() +runtime.current_scheduler:run() assert(equal(log, {'a', 'b', 'c'})) -- Test performance @@ -33,8 +33,8 @@ local function inc() count = count + 1 end for _=1, fiber_count do - fiber.spawn(function() - inc(); fiber.yield(); inc(); fiber.yield(); inc() + runtime.spawn(function() + inc(); runtime.yield(); inc(); runtime.yield(); inc() end) end @@ -43,7 +43,7 @@ print("Fiber creation time: "..(end_time - start_time)/fiber_count) start_time = sc.monotime() for _=1,3*fiber_count do -- run fibers, each fiber yields 3 times - fiber.current_scheduler:run() + runtime.current_scheduler:run() end end_time = sc.monotime() print("Fiber operation time: "..(end_time - start_time)/(2*3*fiber_count)) diff --git a/tests/test_scope.lua b/tests/test_scope.lua index 10d9dbe..5cfa538 100644 --- a/tests/test_scope.lua +++ b/tests/test_scope.lua @@ -4,14 +4,20 @@ print("test: fibers.scope") -- look one level up package.path = "../src/?.lua;" .. package.path -local runtime = require "fibers.runtime" -local scope = require "fibers.scope" +local runtime = require "fibers.runtime" +local scope = require "fibers.scope" +local op = require "fibers.op" +local performer = require "fibers.performer" + +------------------------------------------------------------------------------- +-- 1. Structural tests (your originals, lightly generalised) +------------------------------------------------------------------------------- local function test_outside_fibers() local root = scope.root() - -- current() outside any fibre should be the root (process-wide current scope) - assert(scope.current() == root, "outside fibres, current() should be root") + -- current() outside any fiber should be the root (process-wide current scope) + assert(scope.current() == root, "outside fibers, current() should be root") local outer_scope local inner_scope @@ -25,7 +31,14 @@ local function test_outside_fibers() -- root should see this child in its children list local rc = root:children() - assert(#rc == 1 and rc[1] == s, "root:children() should contain outer scope") + local found_outer = false + for _, c in ipairs(rc) do + if c == s then + found_outer = true + break + end + end + assert(found_outer, "root:children() should contain outer scope") -- Nested run creates a grandchild of s scope.run(function(child2) @@ -34,7 +47,14 @@ local function test_outside_fibers() assert(child2:parent() == s, "nested scope parent must be outer scope") local sc = s:children() - assert(#sc == 1 and sc[1] == child2, "outer scope children() should contain nested scope") + local found_inner = false + for _, c in ipairs(sc) do + if c == child2 then + found_inner = true + break + end + end + assert(found_inner, "outer scope children() should contain nested scope") end) -- After nested run, current() should be back to the outer scope @@ -45,8 +65,8 @@ local function test_outside_fibers() 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 fibres should be root again - assert(scope.current() == scope.root(), "after scope.run, current() should be root outside fibres") + -- 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") end local function test_inside_fibers() @@ -55,53 +75,51 @@ local function test_inside_fibers() local child_in_fiber local grandchild_in_fiber - -- Spawn a fibre anchored to the root scope. + -- Spawn a fiber anchored to the root scope. root:spawn(function(s) - -- In this fibre, s is the scope used for spawn -> root + -- In this fiber, 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 fibre, current() should be root initially") + assert(scope.current() == root, "inside spawned fiber, current() should be root initially") - -- Create a child scope inside the fibre + -- Create a child scope inside the fiber scope.run(function(child) child_in_fiber = child - assert(scope.current() == child, "inside scope.run in fibre, current() should be child") - assert(child:parent() == root, "child-in-fibre parent must be root") + assert(scope.current() == child, "inside scope.run in fiber, current() should be child") + assert(child:parent() == root, "child-in-fiber parent must be root") -- Create a grandchild scope scope.run(function(grandchild) grandchild_in_fiber = grandchild - assert(scope.current() == grandchild, "inside nested run in fibre, current() should be grandchild") + assert(scope.current() == grandchild, "inside nested run in fiber, current() should be grandchild") assert(grandchild:parent() == child, "grandchild parent must be child") end) -- After nested run, current() should be back to child - assert(scope.current() == child, "after nested run in fibre, current() should be child again") + assert(scope.current() == child, "after nested run in fiber, current() should be child again") end) - -- 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") + -- 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") - -- Stop the scheduler once all fibre-local tests have run + -- Stop the scheduler once all fiber-local tests have run runtime.stop() end) - -- Drive the scheduler so the spawned fibre runs + -- Drive the scheduler so the spawned fiber runs runtime.main() - -- After main() returns we are back outside fibres; + -- 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 fibres should be root") + assert(scope.current() == root, "after runtime.main, current() outside fibers should be root") - -- Check that scopes created inside the fibre were recorded + -- Check that scopes created inside the fiber 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 both the outer test scope - -- (from test_outside_fibers) and the child created in this fibre. + -- Check that root children include the child created in this fiber. local rc = root:children() - assert(#rc >= 2, "root should have at least two children by now") local found_child = false for _, s in ipairs(rc) do if s == child_in_fiber then @@ -112,10 +130,277 @@ local function test_inside_fibers() assert(found_child, "root:children() should contain child_in_fiber") end +------------------------------------------------------------------------------- +-- 1b. New: basic scope.with_ev behaviour +------------------------------------------------------------------------------- + +local function test_with_ev_basic() + local parent = scope.current() + local child_scope + + local ev = scope.with_ev(function(child) + child_scope = child + -- inside build_ev, current scope should be the child + assert(scope.current() == child, "inside with_ev build_ev, current() should be child scope") + assert(child:parent() == parent, "with_ev child parent should be current scope") + + -- simple event that returns two values + return op.always(true):wrap(function() + return 99, "ok" + end) + end) + + local a, b = op.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") + local st, err = child_scope:status() + assert(st == "ok" and err == nil, "with_ev child scope should end ok on success") +end + +------------------------------------------------------------------------------- +-- 2. Status transitions for scope.run (success, failure, cancellation) +------------------------------------------------------------------------------- + +local function test_run_success_and_failure() + local root = scope.root() + + -- Success case: scope.run returns body results, status becomes "ok". + local success_scope + local 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") + return 42, "x" + end) + + 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". + local fail_scope + local ok = pcall(function() + scope.run(function(s) + fail_scope = s + error("body failure") + end) + 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(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". + local cancelled_scope + local ok = pcall(function() + scope.run(function(s) + cancelled_scope = s + s:cancel("stop here") + end) + 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(serr == "stop here", "cancelled scope error should be the cancellation reason") +end + +------------------------------------------------------------------------------- +-- 3. Defers: LIFO ordering and execution on failure +------------------------------------------------------------------------------- + +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) + 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(tostring(serr):find("boom in body", 1, true), + "primary 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") +end + +------------------------------------------------------------------------------- +-- 4. Scope:sync via performer.perform: failure and cancellation paths +------------------------------------------------------------------------------- + +local function test_sync_wraps_event_failure() + -- Event whose post-wrap raises: tests wrap_failure path. + 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) + 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(tostring(serr):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. + 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) + 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") +end + +------------------------------------------------------------------------------- +-- 5. join_ev and done_ev (on failed/cancelled scopes) +------------------------------------------------------------------------------- + +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) + end) + assert(not ok_fail, "failed scope.run should raise") + + do + local ev = failed_scope:join_ev() + local st, jerr = op.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") + end + + do + local ev = failed_scope:done_ev() + local reason = op.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), + "done_ev on failed scope should report the failure reason") + end + + -- 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) + end) + assert(not ok_cancel, "cancelled scope.run should raise") + + do + local ev = cancelled_scope:join_ev() + local st, jerr = op.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) + assert(reason == "stop again", + "done_ev on cancelled scope should report cancellation reason") + end +end + +------------------------------------------------------------------------------- +-- 6. Fail-fast from child fibres (via performer.perform) +------------------------------------------------------------------------------- + +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 + + -- 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) + 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() + end) + + runtime.main() +end + +------------------------------------------------------------------------------- +-- Main +------------------------------------------------------------------------------- + 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") end diff --git a/tests/test_sleep.lua b/tests/test_sleep.lua index fd16308..bcdf4c6 100644 --- a/tests/test_sleep.lua +++ b/tests/test_sleep.lua @@ -5,32 +5,25 @@ print('testing: fibers.sleep') package.path = "../src/?.lua;" .. package.path local sleep = require 'fibers.sleep' -local fiber = require 'fibers.fiber' +local runtime = require 'fibers.runtime' local done = 0 -- local wakeup_times = {} local count = 1e3 for _ = 1, count do local function fn() - local start, dt = fiber.now(), math.random() + local start, dt = runtime.now(), math.random() sleep.sleep(dt) - local wakeup_time = fiber.now() + local wakeup_time = runtime.now() assert(wakeup_time >= start + dt) done = done + 1 -- table.insert(wakeup_times, wakeup_time - (start + dt)) end - fiber.spawn(fn) + runtime.spawn(fn) end -for t = fiber.now(), fiber.now() + 1.5, 0.01 do - fiber.current_scheduler:run(t) +for t = runtime.now(), runtime.now() + 1.5, 0.01 do + runtime.current_scheduler:run(t) end assert(done == count) --- -- Calculate maximum error --- local max_error = 0 --- for _, error in ipairs(wakeup_times) do --- if error > max_error then max_error = error end --- end --- print("Maximum sleep error: ", max_error) - print('test: ok') diff --git a/tests/test_stream-file.lua b/tests/test_stream-file.lua index 3141b7c..0680e04 100644 --- a/tests/test_stream-file.lua +++ b/tests/test_stream-file.lua @@ -4,9 +4,8 @@ print('testing: fibers.stream.file') -- look one level up package.path = "../src/?.lua;" .. package.path -local fiber = require 'fibers.fiber' +local fibers = require 'fibers' local waitgroup = require "fibers.waitgroup" -local op = require 'fibers.op' local sleep = require 'fibers.sleep' local channel = require 'fibers.channel' local file = require 'fibers.stream.file' @@ -14,7 +13,7 @@ local compat = require 'fibers.stream.compat' compat.install() -local perform, choice = require 'fibers.performer'.perform, op.choice +local perform, choice = fibers.perform, fibers.choice local function test() local rd, wr = file.pipe() @@ -48,7 +47,7 @@ local function test_long_read_first() local rd, wr = file.pipe() local message = string.rep("a", 2^24) - fiber.spawn(function () + fibers.spawn(function () wr:write_chars(message) wr:close() end) @@ -69,7 +68,7 @@ local function test_read_op() local wg = waitgroup.new() wg:add(1) - fiber.spawn(function () + fibers.spawn(function () local count, err = wr:write_chars(msg1) assert(count==#msg1 and not err) sleep.sleep(0.05) @@ -111,7 +110,7 @@ local function test_write_op() local wg = waitgroup.new() wg:add(1) - fiber.spawn(function () + fibers.spawn(function () sleep.sleep(0.1) local chars, err = rd:read_chars(2^16) assert(chars==string.rep("a", 2^16) and not err) @@ -151,7 +150,7 @@ local function test_long_write_first() local message2 - fiber.spawn(function () + fibers.spawn(function () message2 = rd:read_all_chars() rd:close() chan:put(1) @@ -170,7 +169,7 @@ local function test_tiny_writes() local rd, wr = file.pipe() local message = string.rep("a", 2^16) - fiber.spawn(function () + fibers.spawn(function () for c in message:gmatch"." do wr:write(c) end @@ -188,7 +187,7 @@ local function test_single() local rd, wr = file.pipe() local message = "aa" - fiber.spawn(function () + fibers.spawn(function () wr:write_chars(message) wr:close() end) @@ -225,7 +224,7 @@ local function test_lua() rd:close() end -fiber.spawn(function () +local function main() test() test_read_op() test_write_op() @@ -234,8 +233,8 @@ fiber.spawn(function () test_tiny_writes() test_single() test_lua() - fiber.stop() -end) -fiber.main() +end + +fibers.run(main) print('test: ok') diff --git a/tests/test_stream-socket.lua b/tests/test_stream-socket.lua index bf1b4ad..11dee3a 100644 --- a/tests/test_stream-socket.lua +++ b/tests/test_stream-socket.lua @@ -4,7 +4,7 @@ print('testing: fibers.stream.socket') -- look one level up package.path = "../src/?.lua;" .. package.path -local fiber = require 'fibers.fiber' +local fibers = require 'fibers' local socket = require 'fibers.stream.socket' local sc = require 'fibers.utils.syscall' @@ -33,10 +33,10 @@ local function test() end -fiber.spawn(function () +local function main() test() - fiber.stop() -end) -fiber.main() +end + +fibers.run(main) print('test: ok') diff --git a/tests/test_stream.lua b/tests/test_stream.lua index ce11d74..70083e7 100644 --- a/tests/test_stream.lua +++ b/tests/test_stream.lua @@ -4,7 +4,7 @@ print('testing: fibers.stream') -- look one level up package.path = "../src/?.lua;" .. package.path -local fiber = require 'fibers.fiber' +local fibers = require 'fibers' local stream = require 'fibers.stream' local function test() @@ -32,11 +32,10 @@ local function test() rd:close(); wr:close() end -fiber.spawn(function () +local function main() test() - fiber.stop() -end) +end -fiber.main() +fibers.run(main) print('selftest: ok') diff --git a/tests/test_waitgroup.lua b/tests/test_waitgroup.lua index 27b4c14..9bea596 100644 --- a/tests/test_waitgroup.lua +++ b/tests/test_waitgroup.lua @@ -5,13 +5,12 @@ print('testing: fibers.waitgroup') package.path = "../src/?.lua;" .. package.path -- test_waitgroup.lua -local fiber = require 'fibers.fiber' +local fibers = require 'fibers' local sleep = require 'fibers.sleep' -local op = require 'fibers.op' local waitgroup = require 'fibers.waitgroup' local sc = require 'fibers.utils.syscall' -local perform, choice = require 'fibers.performer'.perform, op.choice +local perform, choice = fibers.perform, fibers.choice local function test_nowait() local wg = waitgroup.new() @@ -29,7 +28,7 @@ local function test_simple() -- Spawn fibers and add to the waitgroup for _ = 1, numFibers do wg:add(1) - fiber.spawn(function() + fibers.spawn(function() sleep.sleep(math.random()) -- Simulate some work wg:done() end) @@ -52,7 +51,7 @@ local function test_complex() local function one_sec_work(w) w:add(1) - fiber.spawn(function() + fibers.spawn(function() sleep.sleep(1) -- Simulate some work w:done() end) @@ -91,9 +90,7 @@ local function main() test_nowait() test_simple() test_complex() - fiber.stop() end -- Start the main function in fiber context -fiber.spawn(main) -fiber.main() +fibers.run(main)