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 909cf17..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 = op.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 64fa01f..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 = op.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 2d5f50c..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 = op.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 b04bc1a..e849e63 100644 --- a/examples/1-basic-usage/context-examples.lua +++ b/examples/1-basic-usage/context-examples.lua @@ -1,11 +1,10 @@ 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 = op.perform +local perform = require 'fibers.performer'.perform -- Simulated work function local function do_work(task_name, duration) @@ -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 43a7012..36f3182 100644 --- a/examples/2-lua-http/cq-http-fiber-test.lua +++ b/examples/2-lua-http/cq-http-fiber-test.lua @@ -4,16 +4,16 @@ 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" 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() @@ -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/fibers/fiber.lua b/fibers/fiber.lua deleted file mode 100644 index a3c7d94..0000000 --- a/fibers/fiber.lua +++ /dev/null @@ -1,157 +0,0 @@ --- (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. --- @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 - -return { - current_scheduler = current_scheduler, - spawn = spawn, - now = now, - suspend = suspend, - yield = yield, - stop = stop, - main = main -} diff --git a/src/fibers.lua b/src/fibers.lua new file mode 100644 index 0000000..8a61e1b --- /dev/null +++ b/src/fibers.lua @@ -0,0 +1,78 @@ +-- fibers.lua +--- +-- Top-level facade for the fibers library. +-- Provides a small, convenient surface over the lower-level modules: +-- - 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 fiber. +-- +-- @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 pack = rawget(table, "pack") or function(...) + return { n = select("#", ...), ... } +end + +local fibers = {} + +-- Perform an event under the current scope. +fibers.perform = performer.perform + +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 = { ... } + + 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) + + runtime.main() +end + +--- Spawn a fiber under the current scope. +function fibers.spawn(fn, ...) + local s = scope_mod.current() + return s:spawn(fn, ...) +end + +fibers.sleep = sleep_mod.sleep +fibers.channel = channel.new + +fibers.scope = scope_mod +fibers.op = op + +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/fibers/alarm.lua b/src/fibers/alarm.lua similarity index 94% rename from fibers/alarm.lua rename to src/fibers/alarm.lua index 70590a3..2281720 100644 --- a/fibers/alarm.lua +++ b/src/fibers/alarm.lua @@ -3,11 +3,11 @@ -- 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' -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 @@ -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/fibers/channel.lua b/src/fibers/channel.lua similarity index 97% rename from fibers/channel.lua rename to src/fibers/channel.lua index 4da9516..f665da3 100644 --- a/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. @@ -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/fibers/cond.lua b/src/fibers/cond.lua similarity index 95% rename from fibers/cond.lua rename to src/fibers/cond.lua index 2c7583c..41255c8 100644 --- a/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/fibers/context.lua b/src/fibers/context.lua similarity index 94% rename from fibers/context.lua rename to src/fibers/context.lua index 21dd922..b8fb040 100644 --- a/fibers/context.lua +++ b/src/fibers/context.lua @@ -13,10 +13,12 @@ --- cancellation. -- @module context -local fiber = require "fibers.fiber" +local runtime = require "fibers.runtime" 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. @@ -43,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 @@ -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) @@ -159,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 @@ -168,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/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 99% rename from fibers/exec.lua rename to src/fibers/exec.lua index d327080..30f0247 100644 --- a/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/fibers/op.lua b/src/fibers/op.lua similarity index 83% rename from fibers/op.lua rename to src/fibers/op.lua index fa20ded..c51599b 100644 --- a/fibers/op.lua +++ b/src/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(...) @@ -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 - local suspended = pack(fiber.suspend(block_choice_op, leaves)) + -- 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 @@ -567,7 +577,8 @@ end return { perform = perform, - new_base_op = new_base_op, -- primitive event constructor + perform_raw = perform, + new_primitive = new_primitive, -- primitive event constructor choice = choice, guard = guard, with_nack = with_nack, @@ -575,7 +586,6 @@ return { bracket = bracket, always = always, never = never, - -- wrap_handler = wrap_handler, - -- finally = finally, + Event = Event, -- 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..269ec62 --- /dev/null +++ b/src/fibers/performer.lua @@ -0,0 +1,32 @@ +-- fibers/performer.lua +--- +-- Scope-aware performer. +-- 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 op = require 'fibers.op' + +local scope_mod + +local M = {} + +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) + 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/fibers/pollio.lua b/src/fibers/pollio.lua similarity index 93% rename from fibers/pollio.lua rename to src/fibers/pollio.lua index e8c2ddd..37cd2cc 100644 --- a/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/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/src/fibers/runtime.lua b/src/fibers/runtime.lua new file mode 100644 index 0000000..055351a --- /dev/null +++ b/src/fibers/runtime.lua @@ -0,0 +1,122 @@ +-- 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_fiber() + return _current_fiber +end + +local function now() + return current_scheduler:now() +end + +local function suspend(block_fn, ...) + 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(function(scheduler, fiber) + scheduler:schedule(fiber) + end) +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_fiber, + + -- time and suspension + now = now, + suspend = suspend, + yield = yield, + + -- fiber management + spawn = spawn, + stop = stop, + main = main, +} 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/src/fibers/scope.lua b/src/fibers/scope.lua new file mode 100644 index 0000000..9827ace --- /dev/null +++ b/src/fibers/scope.lua @@ -0,0 +1,446 @@ +-- fibers/scope.lua +--- +-- Scope module (structured concurrency). +-- +-- Provides a tree of Scope objects and a per-fiber “current scope”. +-- +-- Semantics: +-- - A single root scope exists for the process. +-- - Each fiber has an associated current scope (defaulting to root). +-- - scope.current() returns the current scope for the fiber, +-- or the process-wide current scope when not in a fiber. +-- - scope.run(fn, ...) runs fn in a fresh child scope in the +-- current context (fiber or non-fiber), waits for its children, +-- runs defers, and then returns or raises based on scope status. +-- - s:spawn(fn, ...) spawns a new fiber whose current scope is s +-- 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 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 op = require 'fibers.op' +local waitgroup = require 'fibers.waitgroup' + +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 fiber. +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 = {}, + + -- 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 + 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 + +--- 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. +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. +-- 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_obj + + local res = pack(pcall(fn, scope_obj, ...)) + fiber_scopes[fib] = prev + return res + else + local prev = global_scope or root() + global_scope = scope_obj + + local res = pack(pcall(fn, scope_obj, ...)) + global_scope = prev + 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) + + -- 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 (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 +-- +-- 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 + 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 + -- 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 + +--- 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 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, + with_ev = with_ev, + Scope = Scope, +} diff --git a/fibers/sleep.lua b/src/fibers/sleep.lua similarity index 88% rename from fibers/sleep.lua rename to src/fibers/sleep.lua index 788843e..af57529 100644 --- a/fibers/sleep.lua +++ b/src/fibers/sleep.lua @@ -5,9 +5,9 @@ -- @module fibers.sleep local op = require 'fibers.op' -local fiber = require 'fibers.fiber' +local runtime = require 'fibers.runtime' -local perform = op.perform +local perform = require 'fibers.performer'.perform --- Timeout class. -- Represents a timeout for a fiber. @@ -20,12 +20,12 @@ local perform = op.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/fibers/stream.lua b/src/fibers/stream.lua similarity index 99% rename from fibers/stream.lua rename to src/fibers/stream.lua index c8935a5..3daa769 100644 --- a/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 @@ -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/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 99% rename from fibers/stream/file.lua rename to src/fibers/stream/file.lua index 40b3677..5c83e52 100644 --- a/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/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 92% rename from fibers/waitgroup.lua rename to src/fibers/waitgroup.lua index f0b1aa0..a0dbaa2 100644 --- a/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 @@ -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 abc8374..a7f87ae 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 = '-' @@ -12,7 +12,7 @@ local modules = { { 'stream', 'socket' }, { 'timer' }, { 'sched' }, - { 'fiber' }, + { 'runtime' }, { 'channel' }, { 'queue' }, { 'cond' }, @@ -22,7 +22,8 @@ local modules = { { '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..1a930ef 100644 --- a/tests/test_alarm.lua +++ b/tests/test_alarm.lua @@ -2,9 +2,9 @@ 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 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 0c1bbc9..bb255a5 100644 --- a/tests/test_channel.lua +++ b/tests/test_channel.lua @@ -1,21 +1,21 @@ print('testing: fibers.channel') -package.path = "../?.lua;" .. package.path +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 af27293..8ffd57e 100644 --- a/tests/test_cond.lua +++ b/tests/test_cond.lua @@ -2,10 +2,10 @@ 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' +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 ebcb61a..1ca847a 100644 --- a/tests/test_context.lua +++ b/tests/test_context.lua @@ -2,14 +2,13 @@ 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' -local op = require 'fibers.op' +local fibers = require 'fibers' local sleep = require 'fibers.sleep' -local perform = op.perform +local perform = require 'fibers.performer'.perform -- Test Background Context local function test_background() @@ -25,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") @@ -56,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) @@ -75,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") @@ -89,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") @@ -99,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() @@ -108,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_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..6cbc10f 100644 --- a/tests/test_exec.lua +++ b/tests/test_exec.lua @@ -1,7 +1,9 @@ -- test_fibers_exec.lua -package.path = "../../?.lua;../?.lua;" .. package.path -local fiber = require 'fibers.fiber' +-- look one level up +package.path = "../src/?.lua;" .. package.path + +local fibers = require 'fibers' local sleep = require 'fibers.sleep' local channel = require "fibers.channel" local exec = require 'fibers.exec' @@ -68,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() @@ -116,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() @@ -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() - - fiber.spawn(function () - local cmd = exec.command('sleep', '1') - cmd:setprdeathsig(sc.SIGKILL) - local err = cmd:start() - if err then - error(err) + package.path = "../src/?.lua;" .. package.path + 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() + + local function main() + 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 - io.stdout:write(tostring(cmd.process.pid) .. "\n") - io.stdout:flush() - sleep.sleep(0.01) - print(obj.obj) - end) - fiber.main() + fibers.run(main) ]] -- Write the script to a temporary file @@ -217,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() @@ -239,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) @@ -250,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 eb54407..3715fea 100644 --- a/tests/test_op.lua +++ b/tests/test_op.lua @@ -2,12 +2,12 @@ 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' +local op = require 'fibers.op' +local runtime = require 'fibers.runtime' -local perform, choice = op.perform, op.choice +local perform, choice = require 'fibers.performer'.perform, op.choice local always = op.always local never = op.never @@ -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 4e1ac66..5ff04b1 100644 --- a/tests/test_pollio.lua +++ b/tests/test_pollio.lua @@ -2,11 +2,11 @@ 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' -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 b0822ac..6d432d0 100644 --- a/tests/test_queue.lua +++ b/tests/test_queue.lua @@ -2,17 +2,17 @@ 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' +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 67% rename from tests/test_fiber.lua rename to tests/test_runtime.lua index 39513b9..dae321a 100644 --- a/tests/test_fiber.lua +++ b/tests/test_runtime.lua @@ -2,27 +2,27 @@ 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 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_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 new file mode 100644 index 0000000..5cfa538 --- /dev/null +++ b/tests/test_scope.lua @@ -0,0 +1,407 @@ +--- Tests the Scope implementation. +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 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 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 + + 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() + 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) + 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() + 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 + 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 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() + local root = scope.root() + + local child_in_fiber + local grandchild_in_fiber + + -- Spawn a fiber anchored to the root scope. + root:spawn(function(s) + -- 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 fiber, current() should be root initially") + + -- Create a child scope inside the fiber + scope.run(function(child) + child_in_fiber = child + assert(scope.current() == child, "inside scope.run in fiber, current() should be child") + assert(child:parent() == root, "child-in-fiber parent must be root") + + -- Create a grandchild scope + scope.run(function(grandchild) + grandchild_in_fiber = 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 fiber, current() should be child again") + end) + + -- After inner run, current() should be back to root for this fiber + assert(scope.current() == root, "after scope.run in fiber, current() should be root again") + + -- Stop the scheduler once all fiber-local tests have run + runtime.stop() + end) + + -- Drive the scheduler so the spawned fiber runs + runtime.main() + + -- After main() returns we are back outside fibers; + -- current() should again be the process-wide current scope (root). + assert(scope.current() == root, "after runtime.main, current() outside fibers should be root") + + -- 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 the child created in this fiber. + local rc = root:children() + 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 + +------------------------------------------------------------------------------- +-- 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 + +main() diff --git a/tests/test_sleep.lua b/tests/test_sleep.lua index 210e396..bcdf4c6 100644 --- a/tests/test_sleep.lua +++ b/tests/test_sleep.lua @@ -2,35 +2,28 @@ 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' +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-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..0680e04 100644 --- a/tests/test_stream-file.lua +++ b/tests/test_stream-file.lua @@ -2,11 +2,10 @@ 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 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 = op.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-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..11dee3a 100644 --- a/tests/test_stream-socket.lua +++ b/tests/test_stream-socket.lua @@ -2,9 +2,9 @@ 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 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 b7236e6..70083e7 100644 --- a/tests/test_stream.lua +++ b/tests/test_stream.lua @@ -2,9 +2,9 @@ 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 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_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..9bea596 100644 --- a/tests/test_waitgroup.lua +++ b/tests/test_waitgroup.lua @@ -2,16 +2,15 @@ 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' +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 = op.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)