From 6437db038943789733550a8627719c01e2c504dc Mon Sep 17 00:00:00 2001 From: Richard Thanki Date: Fri, 7 Nov 2025 21:34:13 +0000 Subject: [PATCH] add bracket, reworked op and top-level perform --- README.md | 36 +- examples/1-basic-usage/4-choices.lua | 9 +- examples/1-basic-usage/5-choices-alt.lua | 9 +- examples/1-basic-usage/6-choices-adv.lua | 11 +- examples/1-basic-usage/context-examples.lua | 8 +- examples/2-lua-http/cq-http-fiber-test.lua | 17 +- fibers/alarm.lua | 8 +- fibers/channel.lua | 6 +- fibers/cond.lua | 4 +- fibers/exec.lua | 8 +- fibers/op.lua | 254 +++++++--- fibers/pollio.lua | 6 +- fibers/sleep.lua | 6 +- fibers/stream.lua | 20 +- fibers/stream/file.lua | 7 +- fibers/waitgroup.lua | 4 +- tests/test_context.lua | 7 +- tests/test_op.lua | 495 +++++++++----------- tests/test_stream-file.lua | 12 +- tests/test_waitgroup.lua | 24 +- 20 files changed, 534 insertions(+), 417 deletions(-) diff --git a/README.md b/README.md index bc1d231..99a6f5a 100644 --- a/README.md +++ b/README.md @@ -7,15 +7,17 @@ local function fibonacci(c, quit) local x, y = 0, 1 local done = false repeat - op.choice( - c:put_op(x):wrap(function(value) - x, y = y, x+y - end), - quit:get_op():wrap(function(value) - print("quit") - done = true - end) - ):perform() + perform( + choice( + c:put_op(x):wrap(function(value) + x, y = y, x+y + end), + quit:get_op():wrap(function(value) + print("quit") + done = true + end) + ) + ) until done end @@ -35,7 +37,7 @@ end) fiber.main() ``` -Ported from the Snabb Project's [`fibers`](https://github.com/snabbco/snabb/tree/master/src/lib/fibers) and [`streams`](https://github.com/snabbco/snabb/tree/master/src/lib/stream) libraries, written by +Ported from the Snabb Project's [`fibers`](https://github.com/snabbco/snabb/tree/master/src/lib/fibers) and [`streams`](https://github.com/snabbco/snabb/tree/master/src/lib/stream) libraries, written by Andy Wingo as an implementation of Reppy et al's Concurrent ML(CML), and a fibers-based reimplementation of Lua's streams enabling smooth non-blocking access to files and sockets. Inspired by Andy's [blog post](https://wingolog.org/archives/2018/05/16/lightweight-concurrency-in-lua) introducing fibers on Lua @@ -55,7 +57,7 @@ with the following points to bear in mind: ## Installation This is a pure Lua (with FFI) module with the following dependencies: - - lua-posix (for micro/nano timing and sleeping options, for forking and + - lua-posix (for micro/nano timing and sleeping options, for forking and other syscall operations) - libffi and lua-cffi (if not using LuaJIT) - lua-bit32 (if not using LuaJIT) @@ -76,7 +78,7 @@ These dependencies will be installed in a VScode devcontainer automatically. To ### Installation with LuaJIT on OpenWRT -This is the simplest set up for running on OpenWRT. +This is the simplest set up for running on OpenWRT. `opkg update; opkg install luajit; opkg install luaposix` @@ -111,15 +113,15 @@ This library implements a simple version of Concurrent ML(CML), and provides pri ## Progress -All of the Snabb 'fibers' modules the following have so far been ported and -tested. All of Snabb's 'stream' module has also been ported, which can do -non-blocking reads and writes from file descriptors using familiar `line` -and `all` approaches. +All of the Snabb 'fibers' modules the following have so far been ported and +tested. All of Snabb's 'stream' module has also been ported, which can do +non-blocking reads and writes from file descriptors using familiar `line` +and `all` approaches. We use the `cffi` module to port Wingo's `luajit` C ffi based buffers implementation in a way that will work across multiple architectures, as Lua versions of these buffers would be inefficient and lead to allocation and -garbage collection without a substantial investment of time. +garbage collection without a substantial investment of time. ## Structured concurrency diff --git a/examples/1-basic-usage/4-choices.lua b/examples/1-basic-usage/4-choices.lua index cdad53e..909cf17 100644 --- a/examples/1-basic-usage/4-choices.lua +++ b/examples/1-basic-usage/4-choices.lua @@ -12,11 +12,13 @@ local fiber = require 'fibers.fiber' local channel = require 'fibers.channel' local op = require 'fibers.op' +local perform, choice = op.perform, op.choice + local function fibonacci(c, quit) local x, y = 0, 1 local done = false repeat - op.choice( + local task = choice( c:put_op(x):wrap(function() x, y = y, x+y end), @@ -24,7 +26,8 @@ local function fibonacci(c, quit) print("quit") done = true end) - ):perform() + ) + perform(task) until done end @@ -41,4 +44,4 @@ fiber.spawn(function() fiber.stop() end) -fiber.main() \ No newline at end of file +fiber.main() diff --git a/examples/1-basic-usage/5-choices-alt.lua b/examples/1-basic-usage/5-choices-alt.lua index d1e3b5d..64fa01f 100644 --- a/examples/1-basic-usage/5-choices-alt.lua +++ b/examples/1-basic-usage/5-choices-alt.lua @@ -11,6 +11,8 @@ local channel = require 'fibers.channel' local sleep = require 'fibers.sleep' local op = require 'fibers.op' +local perform, choice = op.perform, op.choice + -- time.After() is a Go library function local function after(t) local chan = channel.new() @@ -38,7 +40,7 @@ fiber.spawn(function() local boom = after(0.5) local done = false repeat - op.choice( + local task = choice( ticker:get_op():wrap(function() print("tick.") end), @@ -46,13 +48,14 @@ fiber.spawn(function() print("BOOM!") done = true end) - ):perform_alt(function() + ):or_else(function() print(" .") sleep.sleep(0.05) end) + perform(task) until done fiber.stop() end) -fiber.main() \ No newline at end of file +fiber.main() diff --git a/examples/1-basic-usage/6-choices-adv.lua b/examples/1-basic-usage/6-choices-adv.lua index 3aaffbd..2d5f50c 100644 --- a/examples/1-basic-usage/6-choices-adv.lua +++ b/examples/1-basic-usage/6-choices-adv.lua @@ -11,6 +11,8 @@ local cond = require 'fibers.cond' local op = require 'fibers.op' local sc = require 'fibers.utils.syscall' +local perform, choice = op.perform, op.choice + require("fibers.pollio").install_poll_io_handler() -- Set up the queues, channel and condition variable @@ -60,8 +62,8 @@ fiber.spawn(function() socket.socket(sc.AF_UNIX, sc.SOCK_STREAM, 0):listen_unix(sockname) local sock = socket.connect_unix(sockname) while true do - -- Use op.choice to handle multiple potentially blocking actions - op.choice( + -- Use choice to handle multiple potentially blocking actions + local task = choice( data_q:get_op():wrap(function(value) print("data received - writing to socket") sock:write(value .. "\n") @@ -79,8 +81,9 @@ fiber.spawn(function() sleep.sleep_op(0.5):wrap(function() print("yawn - nothing happening") end) - ):perform() + ) + perform(task) end end) -fiber.main() \ No newline at end of file +fiber.main() diff --git a/examples/1-basic-usage/context-examples.lua b/examples/1-basic-usage/context-examples.lua index a028eb4..b04bc1a 100644 --- a/examples/1-basic-usage/context-examples.lua +++ b/examples/1-basic-usage/context-examples.lua @@ -5,6 +5,8 @@ local context = require 'fibers.context' local sleep = require 'fibers.sleep' local op = require 'fibers.op' +local perform = op.perform + -- Simulated work function local function do_work(task_name, duration) print(task_name .. " started") @@ -20,8 +22,7 @@ local function sub_task_1(ctx) cancel("work_completed") -- Cancel the context (optional, as it will timeout) end) - local done_op = deadline_ctx:done_op() - op.choice(done_op):perform() -- Wait for the context to be done + perform(deadline_ctx:done_op()) -- Wait for the context to be done print("Sub-Task 1 status: " .. (deadline_ctx:err() or "completed")) end @@ -33,8 +34,7 @@ local function sub_task_2(ctx) cancel("work_completed") -- Cancel the context when work is done end) - local done_op = cancel_ctx:done_op() - op.choice(done_op):perform() -- Wait for the context to be done + perform(cancel_ctx:done_op()) -- Wait for the context to be done print("Sub-Task 2 status: " .. (cancel_ctx:err() or "completed")) end diff --git a/examples/2-lua-http/cq-http-fiber-test.lua b/examples/2-lua-http/cq-http-fiber-test.lua index 9bb843a..43a7012 100644 --- a/examples/2-lua-http/cq-http-fiber-test.lua +++ b/examples/2-lua-http/cq-http-fiber-test.lua @@ -13,6 +13,8 @@ local sleep = require "fibers.sleep" local cqueues = require "cqueues" local exec = require "fibers.exec" +local perform, choice = op.perform, op.choice + print("installing poll handler") pollio.install_poll_io_handler() @@ -32,14 +34,15 @@ local old_step; old_step = cqueues.interpose("step", function(self, timeout) local events = self:events() -- messy if events == 'r' then - pollio.fd_readable_op(self:pollfd()):perform() + perform(pollio.fd_readable_op(self:pollfd())) elseif events == 'w' then - pollio.fd_writable_op(self:pollfd()):perform() + perform(pollio.fd_writable_op(self:pollfd())) elseif events == 'rw' then - op.choice( - pollio.fd_readable_op(self:pollfd()), - pollio.fd_writable_op(self:pollfd()) - ):perform() + perform( + choice( + pollio.fd_readable_op(self:pollfd()), + pollio.fd_writable_op(self:pollfd()) + )) end return old_step(self, 0.0) end @@ -135,4 +138,4 @@ fiber.spawn(function() end) print("starting fibers") -fiber.main() \ No newline at end of file +fiber.main() diff --git a/fibers/alarm.lua b/fibers/alarm.lua index 7445c36..70590a3 100644 --- a/fibers/alarm.lua +++ b/fibers/alarm.lua @@ -7,6 +7,8 @@ local fiber = require 'fibers.fiber' local timer = require 'fibers.timer' local sc = require 'fibers.utils.syscall' +local perform = op.perform + local function days_in_year(y) return y % 4 == 0 and (y % 100 ~= 0 or y % 400 == 0) and 366 or 365 end @@ -240,7 +242,7 @@ end -- Wrapper for `absolute_op` that immediately performs the operation. -- @param t The absolute time (epoch) for the alarm. local function wait_absolute(t) - return wait_absolute_op(t):perform() + return perform(wait_absolute_op(t)) end --- Creates an operation for a next (relative) alarm. @@ -260,7 +262,7 @@ end -- @return An error if the time table is invalid. local function wait_next(t) local _, err = validate_next_table(t) - return err or assert(installed_alarm_handler):wait_next_op(t):perform() + return err or perform(assert(installed_alarm_handler):wait_next_op(t)) end -- Public API @@ -275,4 +277,4 @@ return { wait_next = wait_next, validate_next_table = validate_next_table, calculate_next = calculate_next -} \ No newline at end of file +} diff --git a/fibers/channel.lua b/fibers/channel.lua index 09cc181..4da9516 100644 --- a/fibers/channel.lua +++ b/fibers/channel.lua @@ -7,6 +7,8 @@ local op = require 'fibers.op' local fifo = require 'fibers.utils.fifo' +local perform = op.perform + --- Channel class -- Represents a communication channel between fibers. -- @type Channel @@ -115,7 +117,7 @@ end -- continue. Otherwise, block until a receiver becomes available. -- @tparam any message The message to put into the Channel. function Channel:put(message) - self:put_op(message):perform() + return perform(self:put_op(message)) end --- Get a message from the Channel. @@ -125,7 +127,7 @@ end -- available. -- @treturn any The message retrieved from the Channel. function Channel:get() - return self:get_op():perform() + return perform(self:get_op()) end --- @export diff --git a/fibers/cond.lua b/fibers/cond.lua index 03fc4cc..2c7583c 100644 --- a/fibers/cond.lua +++ b/fibers/cond.lua @@ -6,6 +6,8 @@ local op = require 'fibers.op' +local perform = op.perform + local Cond = {} Cond.__index = Cond @@ -24,7 +26,7 @@ end --- Put the fiber into a wait state on the condition variable. function Cond:wait() - return self:wait_op():perform() + return perform(self:wait_op()) end --- Wake up all fibers that are waiting on this condition variable. diff --git a/fibers/exec.lua b/fibers/exec.lua index e61a594..d327080 100644 --- a/fibers/exec.lua +++ b/fibers/exec.lua @@ -6,6 +6,8 @@ local op = require 'fibers.op' local buffer = require 'string.buffer' local sc = require 'fibers.utils.syscall' +local perform, choice = op.perform, op.choice + local io_mappings = { stdin = sc.STDIN_FILENO, stdout = sc.STDOUT_FILENO, @@ -106,7 +108,7 @@ function Cmd:_output_collector(pipes) ops[#ops + 1] = self.ctx:done_op():wrap(close_pipes) end - op.choice(unpack(ops)):perform() + perform(choice(unpack(ops))) end return buf:tostring(), self:wait() @@ -228,11 +230,11 @@ function Cmd:wait() if self.ctx then ops[#ops + 1] = self.ctx:done_op():wrap(function() self:kill() - pollio.fd_readable_op(self.process.pidfd):perform() + perform(pollio.fd_readable_op(self.process.pidfd)) end) end - op.choice(unpack(ops)):perform() + perform(choice(unpack(ops))) local _, _, status = sc.waitpid(self.process.pid) self.process.state = status sc.close(self.process.pidfd) diff --git a/fibers/op.lua b/fibers/op.lua index 67e924d..acd264d 100644 --- a/fibers/op.lua +++ b/fibers/op.lua @@ -1,7 +1,34 @@ --- fibers.op module -- Provides Concurrent ML style operations for managing concurrency. -- Events are CML-style: primitive leaves, choices, guards, with_nack, --- and wraps. Synchronization compiles an event tree into primitive leaves. +-- wraps, and an extra abort combinator (on_abort). +-- +-- Core event AST kinds: +-- prim : primitive leaf { try_fn, block_fn, wrap_fn } +-- choice : non-empty list of events +-- guard : delayed event builder (run once per sync) +-- with_nack : CML-style nack combinator +-- wrap : post-commit mapper (composed at compile time) +-- abort : attach abort handler to an event (run if this arm loses) +-- +-- Semantics sketch +-- ---------------- +-- We keep CML-style semantics for with_nack: +-- - with_nack g gets a nack event that becomes enabled iff the +-- *entire* resulting event loses in an enclosing choice. +-- - nested with_nack behaves correctly: outer nacks only fire when +-- the outer event loses, not when internal subchoices resolve. +-- +-- `on_abort(ev, f)` is implemented in terms of the same "nack" machinery: +-- - each abort scope behaves like a nack-cond whose signal() runs f(). +-- - after a choice commits, we figure out which conds are associated +-- exclusively with losing arms and signal those once. +-- +-- We also provide: +-- - bracket(acquire, release, use): RAII-style resource protocol. +-- - else_next_turn(ev, fallback_ev): biased choice; prefer ev, but +-- if it doesn't commit "by next turn", abort it cleanly and run +-- fallback_ev (in a separate sync). local fiber = require 'fibers.fiber' @@ -75,6 +102,7 @@ end -- kind = 'guard' : { builder = function() -> Event } -- kind = 'with_nack' : { builder = function(nack_ev) -> Event } -- kind = 'wrap' : { inner = Event, wrap_fn = f } +-- kind = 'abort' : { inner = Event, abort_fn = f } ---------------------------------------------------------------------- local Event = {} @@ -116,13 +144,50 @@ local function guard(g) return setmetatable({ kind = 'guard', builder = g }, Event) end --- with_nack g: delayed event; g(nack_ev) evaluated once per synchronization. --- nack_ev is an Event that becomes ready iff this with_nack is *not* chosen. +-- CML-style with_nack: builder gets a nack event that becomes ready +-- iff this event participates in a choice and *loses*. local function with_nack(g) return setmetatable({ kind = 'with_nack', builder = g }, Event) end --- Wrap event with a post-processing function f. +-- 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 try() + return false + 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) +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, + 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 + +-- Wrap event with a post-processing function f (commit phase). -- This is another node in the tree; composed at compile time. function Event:wrap(f) return setmetatable( @@ -131,17 +196,29 @@ function Event:wrap(f) ) end +-- Attach an abort handler to this event. +-- f() is run iff this event participates in a choice and *does not win*. +function Event:on_abort(f) + assert(type(f) == 'function', "on_abort expects a function") + return setmetatable( + { kind = 'abort', inner = self, abort_fn = f }, + Event + ) +end + ---------------------------------------------------------------------- -- Simple one-shot condition primitive (used for with_nack; also exported) ---------------------------------------------------------------------- -local function new_cond() +local function new_cond(opts) local state = { triggered = false, - waiters = {}, -- array of CompleteTask + waiters = {}, -- optional + abort_fn = opts and opts.abort_fn or nil, } local function wait_op() + assert(not state.abort_fn, "abort-only cond has no wait_op") local function try() return state.triggered end @@ -149,7 +226,8 @@ local function new_cond() if state.triggered then suspension:complete(wrap_fn) else - state.waiters[#state.waiters + 1] = suspension:complete_task(wrap_fn) + state.waiters[#state.waiters + 1] = + suspension:complete_task(wrap_fn) end end return new_base_op(nil, try, block) @@ -158,17 +236,25 @@ local function new_cond() 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() then + if task + and task.suspension + and task.suspension:waiting() + then task.suspension.sched:schedule(task) end end + -- fire abort handler, if any + if state.abort_fn then + pcall(state.abort_fn) + end end return { - wait_op = wait_op, + wait_op = state.abort_fn and nil or wait_op, signal = signal, } end @@ -181,8 +267,13 @@ end -- try_fn, -- block_fn, -- wrap, -- final wrap function for this leaf --- nacks = {...}, -- list of all active with_nack conds on this path +-- nacks = {...}, -- list of all active nack/abort conds on this path -- } +-- +-- Semantics: +-- - Each with_nack or abort node adds a cond to the nacks list. +-- - After a winner leaf is chosen, we find which conds appear only +-- on losing paths and signal those once (CML-style nack). ---------------------------------------------------------------------- local function compile_event(ev, outer_wrap, out, nacks) @@ -203,12 +294,11 @@ local function compile_event(ev, outer_wrap, out, nacks) elseif kind == 'with_nack' then local cond = new_cond() - local nack_ev = cond.wait_op() -- Event + local nack_ev = cond.wait_op() local inner = ev.builder(nack_ev) - -- Extend the current nack list for this subtree - local child_nacks = { unpack(nacks) } - child_nacks[#child_nacks + 1] = cond + local child_nacks = { unpack(nacks) } + child_nacks[#child_nacks + 1] = cond compile_event(inner, outer_wrap, out, child_nacks) elseif kind == 'wrap' then @@ -218,7 +308,14 @@ local function compile_event(ev, outer_wrap, out, nacks) end compile_event(ev.inner, new_outer, out, nacks) + elseif kind == 'abort' then + local cond = new_cond{ abort_fn = ev.abort_fn } + local child_nacks = { unpack(nacks) } + child_nacks[#child_nacks + 1] = cond + compile_event(ev.inner, outer_wrap, out, child_nacks) + else -- 'prim' + -- Each leaf gets a unique final_wrap closure so identity comparison in local final_wrap = function(...) return outer_wrap(ev.wrap_fn(...)) end @@ -237,29 +334,36 @@ end -- Nack triggering and non-blocking attempt ---------------------------------------------------------------------- --- Signal all with_nack conds that belong exclusively to losing arms. +-- Signal all conds that belong exclusively to losing arms. +-- This is the original CML-style logic: +-- - Build set of nacks on the winner path. +-- - For each loser leaf, signal any nacks not in the winner set. +-- - Each cond object is responsible for idempotence. local function trigger_nacks(ops, winner_index) - -- Build a set of conds to *skip* (all on the winner path). - local winner = {} - if winner_index then - local wnacks = ops[winner_index].nacks - if wnacks then - for i = 1, #wnacks do - winner[wnacks[i]] = true - end + 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 end end - -- Signal each losing cond once. local signaled = {} for i = 1, #ops do - local nacks = ops[i].nacks - if nacks then - for j = 1, #nacks do - local cond = nacks[j] - if cond and not winner[cond] and not signaled[cond] then - signaled[cond] = true - cond.signal() + if 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 + cond.signal() + end end end end @@ -300,54 +404,71 @@ local function block_choice_op(sched, fib, ops) end ---------------------------------------------------------------------- --- Event methods: perform, poll, perform_alt +-- Event methods: perform, perform_alt ---------------------------------------------------------------------- -- Perform this event (primitive or composite), possibly blocking. -function Event:perform() - local ops = compile_event(self) +local function perform(ev) + local leaves = compile_event(ev) - -- Fast path: non-blocking attempt. - local idx, retval = try_ready(ops) + -- Fast path + local idx, retval = try_ready(leaves) if idx then - trigger_nacks(ops, idx) - return apply_wrap(ops[idx].wrap, retval) + trigger_nacks(leaves, idx) + return apply_wrap(leaves[idx].wrap, retval) end - -- Slow path: block on all compiled leaves. - local suspended = pack(fiber.suspend(block_choice_op, ops)) + -- Slow path + local suspended = pack(fiber.suspend(block_choice_op, leaves)) local wrap = suspended[1] - -- Find the winning leaf by matching its wrap function. local winner_index - for i, op in ipairs(ops) do - if op.wrap == wrap then + for i, leaf in ipairs(leaves) do + if leaf.wrap == wrap then winner_index = i break end end - trigger_nacks(ops, winner_index) + trigger_nacks(leaves, winner_index) return wrap(unpack(suspended, 2, suspended.n)) end --- poll: non-blocking synchronization attempt. --- Returns (true, ...results) if some arm commits, or (false) otherwise. -function Event:poll() - local ops = compile_event(self) - local idx, retval = try_ready(ops) - if not idx then return false end - trigger_nacks(ops, idx) - return true, apply_wrap(ops[idx].wrap, retval) -end +---------------------------------------------------------------------- +-- bracket : (acquire, release, use) -> 'a event +-- +-- acquire() : -> resource +-- release(resource, aborted:boolean) +-- use(resource) : -> Event (the main action) +-- +-- Semantics: +-- * acquire is run once, at sync time (inside a guard). +-- * if the resulting event `ev` WINS: +-- - its result is returned +-- - release(res, false) is called (best-effort, pcall) +-- * if the resulting event PARTICIPATES in a choice but LOSES: +-- - release(res, true) is called (via on_abort / nack machinery) +-- +-- This is *purely an Event combinator*; no new fiber is spawned. +---------------------------------------------------------------------- --- perform_alt: non-blocking; if no arm ready, call f(). -function Event:perform_alt(f) - local res = pack(self:poll()) - if res[1] then - return unpack(res, 2, res.n) +local function bracket(acquire, release, use) + return guard(function() + local res = acquire() + local ok, ev = pcall(use, res) + if not ok then + pcall(release, res, true) -- ensure cleanup on builder failure + error(ev) end - return f() + return ev + :wrap(function(...) + pcall(release, res, false) + return ... + end) + :on_abort(function() + pcall(release, res, true) + end) + end) end ---------------------------------------------------------------------- @@ -355,9 +476,14 @@ end ---------------------------------------------------------------------- return { - new_base_op = new_base_op, -- primitive event constructor - choice = choice, - guard = guard, - with_nack = with_nack, - new_cond = new_cond, + perform = perform, + new_base_op = new_base_op, -- primitive event constructor + choice = choice, + guard = guard, + with_nack = with_nack, + new_cond = new_cond, + bracket = bracket, + always = always, + never = never, + -- Event instances have methods: wrap, on_abort. } diff --git a/fibers/pollio.lua b/fibers/pollio.lua index d569af0..e8c2ddd 100644 --- a/fibers/pollio.lua +++ b/fibers/pollio.lua @@ -8,6 +8,8 @@ local epoll = require 'fibers.epoll' local sc = require 'fibers.utils.syscall' local bit = rawget(_G, "bit") or require 'bit32' +local perform = op.perform + local PollIOHandler = {} PollIOHandler.__index = PollIOHandler @@ -172,13 +174,13 @@ local function fd_readable_op(fd) return assert(installed_poll_handler):fd_readable_op(fd) end local function fd_readable(fd) - return fd_readable_op(fd):perform() + return perform(fd_readable_op(fd)) end local function fd_writable_op(fd) return assert(installed_poll_handler):fd_writable_op(fd) end local function fd_writable(fd) - return fd_writable_op(fd):perform() + return perform(fd_writable_op(fd)) end local function stream_readable_op(stream) return assert(installed_poll_handler):stream_readable_op(stream) diff --git a/fibers/sleep.lua b/fibers/sleep.lua index 808d9cb..788843e 100644 --- a/fibers/sleep.lua +++ b/fibers/sleep.lua @@ -7,6 +7,8 @@ local op = require 'fibers.op' local fiber = require 'fibers.fiber' +local perform = op.perform + --- Timeout class. -- Represents a timeout for a fiber. -- @type Timeout @@ -29,7 +31,7 @@ end --- Put the current fiber to sleep until time t. -- @tparam number t The time to sleep until. local function sleep_until(t) - return sleep_until_op(t):perform() + return perform(sleep_until_op(t)) end --- Create a new operation that puts the current fiber to sleep for a duration dt. @@ -46,7 +48,7 @@ end --- Put the current fiber to sleep for a duration dt. -- @tparam number dt The duration to sleep. local function sleep(dt) - return sleep_op(dt):perform() + return perform(sleep_op(dt)) end return { diff --git a/fibers/stream.lua b/fibers/stream.lua index e8639da..c8935a5 100644 --- a/fibers/stream.lua +++ b/fibers/stream.lua @@ -9,6 +9,8 @@ local op = require 'fibers.op' local fixed_buffer = require 'fibers.utils.fixed_buffer' local buffer = require 'string.buffer' +local perform = op.perform + local unpack = table.unpack or unpack -- luacheck: ignore -- Compatibility fallback local Stream = {} @@ -148,7 +150,7 @@ end -- @return number of bytes written -- @return error encountered during the write function Stream:write_chars(str) - return self:write_chars_op(str):perform() + return perform(self:write_chars_op(str)) end local function core_read_op(stream, buf, min, max, terminator) @@ -244,7 +246,7 @@ end -- @return string containing the characters read -- @return error during read, if any function Stream:read_chars(count) - return self:read_chars_op(count):perform() + return perform(self:read_chars_op(count)) end --- Operation to read up to a specified number of characters from the stream. @@ -263,7 +265,7 @@ end -- @return string containing the characters read -- @return error during read, if any function Stream:read_some_chars(count) - return self:read_some_chars_op(count):perform() + return perform(self:read_some_chars_op(count)) end --- Operation to read all characters from the stream. @@ -279,7 +281,7 @@ end -- @return string containing all characters read -- @return error during read, if any function Stream:read_all_chars() - return self:read_all_chars_op():perform() + return perform(self:read_all_chars_op()) end --- Operation to read a single character from the stream. @@ -295,7 +297,7 @@ end -- @return the character read, or nil if at end of file -- @return error during read, if any function Stream:read_char() - return self:read_char_op():perform() + return perform(self:read_char_op()) end --- Operation to read a line from the stream. @@ -316,7 +318,7 @@ end -- @return the line read, or nil if at end of file -- @return error during read, if any function Stream:read_line(style) - return self:read_line_op(style):perform() + return perform(self:read_line_op(style)) end function Stream:flush_input() @@ -335,7 +337,7 @@ end --- Flush the output buffer, writing all buffered data to the underlying IO. function Stream:flush_output() - return self:flush_output_op():perform() + return perform(self:flush_output_op()) end Stream.flush = Stream.flush_output @@ -397,7 +399,7 @@ end -- @return the data read from the stream according to the specified format, or nil on end of file -- @return error during read, if any function Stream:read(...) - return self:read_op(...):perform() + return perform(self:read_op(...)) end --- Get or set the file position. @@ -492,7 +494,7 @@ end -- @param ... data to write (strings or numbers) -- @return true on success, or nil plus an error message on failure function Stream:write(...) - return self:write_op(...):perform() + return perform(self:write_op(...)) end -- The result may be nil. diff --git a/fibers/stream/file.lua b/fibers/stream/file.lua index e82e467..40b3677 100644 --- a/fibers/stream/file.lua +++ b/fibers/stream/file.lua @@ -7,9 +7,12 @@ 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 pio_handler = pollio.install_poll_io_handler() local bit = rawget(_G, "bit") or require 'bit32' @@ -88,11 +91,11 @@ end function File:wait_for_readable_op() return pollio.fd_readable_op(self.fd) end -function File:wait_for_readable() self:wait_for_readable_op():perform() end +function File:wait_for_readable() perform(self:wait_for_readable_op()) end function File:wait_for_writable_op() return pollio.fd_writable_op(self.fd) end -function File:wait_for_writable() self:wait_for_writable_op():perform() end +function File:wait_for_writable() perform(self:wait_for_writable_op()) end function File:task_on_readable(t) if self.fd then pio_handler:task_on_readable(self.fd, t) end end diff --git a/fibers/waitgroup.lua b/fibers/waitgroup.lua index c01ffb5..f0b1aa0 100644 --- a/fibers/waitgroup.lua +++ b/fibers/waitgroup.lua @@ -1,6 +1,8 @@ -- waitgroup.lua local op = require 'fibers.op' +local perform = op.perform + local Waitgroup = {} Waitgroup.__index = Waitgroup @@ -47,7 +49,7 @@ function Waitgroup:wait_op() end function Waitgroup:wait() - self:wait_op():perform() + return perform(self:wait_op()) end return { diff --git a/tests/test_context.lua b/tests/test_context.lua index 11ac773..ebcb61a 100644 --- a/tests/test_context.lua +++ b/tests/test_context.lua @@ -6,8 +6,11 @@ package.path = "../?.lua;" .. package.path local context = require 'fibers.context' local fiber = require 'fibers.fiber' +local op = require 'fibers.op' local sleep = require 'fibers.sleep' +local perform = op.perform + -- Test Background Context local function test_background() local ctx = context.background() @@ -23,7 +26,7 @@ local function test_with_cancel() local is_cancelled = false fiber.spawn(function() - child_ctx:done_op():perform() + perform(child_ctx:done_op()) is_cancelled = true end) @@ -54,7 +57,7 @@ local function test_with_timeout() local is_cancelled = false fiber.spawn(function() - ctx:done_op():perform() + perform(ctx:done_op()) is_cancelled = true end) diff --git a/tests/test_op.lua b/tests/test_op.lua index 3d5a03e..6d30eb6 100644 --- a/tests/test_op.lua +++ b/tests/test_op.lua @@ -1,32 +1,20 @@ --- fibers/op comprehensive test -print("testing: fibers.op (CML events)") +-- fibers/op compact but comprehensive test (no poll) +print("testing: fibers.op") -- look one level up package.path = "../?.lua;" .. package.path -local op = require 'fibers.op' +local op = require 'fibers.op' local fiber = require 'fibers.fiber' +local perform, choice = op.perform, op.choice +local always = op.always +local never = op.never + ------------------------------------------------------------ -- Helpers ------------------------------------------------------------ --- Ready primitive event that returns val. -local function task(val) - local wrap_fn = function(x) return x end - local try_fn = function() return true, val end - local block_fn = function() end - return op.new_base_op(wrap_fn, try_fn, block_fn) -end - --- Primitive that never becomes ready (try=false, no block). -local function never_op() - local wrap_fn = function(x) return x end - local try_fn = function() return false end - local block_fn = function() end - return op.new_base_op(wrap_fn, try_fn, block_fn) -end - -- Primitive that *forces* the blocking path then completes once. local function async_task(val) local tries = 0 @@ -35,7 +23,6 @@ local function async_task(val) return false end local function block_fn(suspension, wrap_fn) - -- Complete on next scheduler turn. local t = suspension:complete_task(wrap_fn, val) suspension.sched:schedule(t) end @@ -50,19 +37,44 @@ end fiber.spawn(function() -------------------------------------------------------- - -- 1) Base op: perform, perform_alt + -- 1) Base event: perform, or_else, wrap -------------------------------------------------------- do - local base = task(1) - assert(base:perform() == 1, "base: perform failed") + local base = always(1) + assert(perform(base) == 1, "base: perform failed") + + -- or_else: event wins, fallback ignored + local base2 = always(2) + local ev1 = base2:or_else(function() return 9 end) + local palt1 = perform(ev1) + assert(palt1 == 2, "base: or_else should use event result") + + -- or_else: never-ready event → fallback wins + local ev2 = never():or_else(function() return 99 end) + local palt2 = perform(ev2) + assert(palt2 == 99, "base: or_else should use fallback when event can't commit") - local base2 = task(2) - assert(base2:perform_alt(function() return 9 end) == 2, - "base: perform_alt should use event result") + -- or_else: async event should still win (gets a chance next turn) + do + local fallback_called = false + local ev_async, tries = async_task(123) + local ev = ev_async:or_else(function() + fallback_called = true + 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") + end - local never = never_op() - assert(never:perform_alt(function() return 99 end) == 99, - "base: perform_alt should use fallback when not ready") + -- nested wrap: ((x + 1) * 2) + local ev3 = always(5) + :wrap(function(x) return x + 1 end) + :wrap(function(y) return y * 2 end) + local v3 = perform(ev3) + assert(v3 == 12, "nested wrap: wrong result") end -------------------------------------------------------- @@ -70,361 +82,292 @@ fiber.spawn(function() -------------------------------------------------------- do local ev, tries = async_task(42) - local v = ev:perform() + local v = perform(ev) assert(v == 42, "async_task: wrong result") assert(tries() == 1, "async_task: try_fn not called exactly once") end -------------------------------------------------------- - -- 3) Choice over multiple ready events - -- (we don't assume fairness, only that results are valid) + -- 3) Choice + wrap -------------------------------------------------------- do - local choice_ev = op.choice(task(1), task(2), task(3)) + -- choice over multiple ready events + local choice_ev = choice(always(1), always(2), always(3)) for _ = 1, 5 do - local v = choice_ev:perform() + local v = perform(choice_ev) assert(v == 1 or v == 2 or v == 3, "choice(ready): result not in {1,2,3}") end - end - -------------------------------------------------------- - -- 4) Wrap (including nested and on choice) - -------------------------------------------------------- - do -- wrap on choice - local ev = op.choice(task(1), task(2)):wrap(function(x) + local ev = choice(always(1), always(2)):wrap(function(x) return x * 10 end) - local v = ev:perform() + local v = perform(ev) assert(v == 10 or v == 20, "wrap(choice): wrong result") - - -- nested wrap: ((x + 1) * 2) - local ev2 = task(5) - :wrap(function(x) return x + 1 end) - :wrap(function(y) return y * 2 end) - local v2 = ev2:perform() - assert(v2 == 12, "nested wrap: wrong result") end -------------------------------------------------------- - -- 5) Guard: basic, in choice, nested, perform_alt + -- 4) Guard: basic + in choice + with with_nack -------------------------------------------------------- do - -- basic + -- basic guard local calls = 0 local g = function() calls = calls + 1 - return task(42) + return always(42) end local ev = op.guard(g) - local v = ev:perform() + local v = perform(ev) assert(v == 42, "guard basic: wrong result") assert(calls == 1, "guard basic: builder not called once") - -- guard inside choice + -- guard in choice (ensures builder runs each sync) local calls2 = 0 - local g2 = function() + local guarded = op.guard(function() calls2 = calls2 + 1 - return task(10) - end - local guarded = op.guard(g2) - local choice_ev = op.choice(guarded, task(20)) + return always(10) + end) + local choice_ev = choice(guarded, always(20)) local runs = 5 for _ = 1, runs do - local r = choice_ev:perform() + local r = perform(choice_ev) assert(r == 10 or r == 20, "guard in choice: result not in {10,20}") end assert(calls2 == runs, "guard in choice: builder call mismatch") - -- guard + wrap - local calls3 = 0 - local g3 = function() - calls3 = calls3 + 1 - return task(5) - end - local ev3 = op.guard(g3):wrap(function(x) return x * 2 end) - local v3 = ev3:perform() - assert(v3 == 10, "guard wrap: wrong result") - assert(calls3 == 1, "guard wrap: builder not called once") - - -- guard + perform_alt (inner not ready) - local calls4 = 0 - local g4 = function() - calls4 = calls4 + 1 - return never_op() - end - local ev4 = op.guard(g4) - local v4 = ev4:perform_alt(function() return 99 end) - assert(v4 == 99, "guard perform_alt: wrong fallback") - assert(calls4 == 1, "guard perform_alt: builder not called once") - - -- nested guards - local outer_calls, inner_calls = 0, 0 - local inner_g = function() - inner_calls = inner_calls + 1 - return task(7) - end - local outer_g = function() - outer_calls = outer_calls + 1 - return op.guard(inner_g) - end - local ev5 = op.guard(outer_g):wrap(function(x) return x + 1 end) - local v5 = ev5:perform() - assert(v5 == 8, "nested guard: wrong result") - assert(outer_calls == 1, "nested guard: outer builder count") - assert(inner_calls == 1, "nested guard: inner builder count") + -- guard + with_nack (builder returns a with_nack event) + local guard_calls, cancelled = 0, false + local guarded_nack = op.guard(function() + guard_calls = guard_calls + 1 + return op.with_nack(function(nack_ev) + fiber.spawn(function() + perform(nack_ev) + cancelled = true + end) + return never() + end) + end) + + local ev2 = choice(guarded_nack, always("OK")) + local v2 = perform(ev2) + assert(v2 == "OK", "guard+with_nack: wrong winner") + + fiber.yield() + assert(cancelled == true, "guard+with_nack: nack not fired") + assert(guard_calls == 1, "guard+with_nack: builder not once") end -------------------------------------------------------- - -- 6) poll() and perform_alt() on composite events + -- 5) or_else on composite events -------------------------------------------------------- do - -- poll on ready event - local ok, v = task(123):poll() - assert(ok and v == 123, "poll(ready): wrong result") - - -- poll on never-ready - local ok2, v2 = never_op():poll() - assert(ok2 == false and v2 == nil, "poll(never): expected (false)") - - -- poll on composite none-ready - local comp = op.choice(never_op(), never_op()) - local ok3, v3 = comp:poll() - assert(ok3 == false and v3 == nil, "poll(composite none): expected (false)") - - -- perform_alt on composite (ready) - local comp_ready = op.choice(task(1), task(2)) - local r1 = comp_ready:perform_alt(function() return 99 end) + -- composite ready: choice(always, always) + local comp_ready = choice(always(1), always(2)) + local ev1 = comp_ready:or_else(function() return 99 end) + local r1 = perform(ev1) assert(r1 == 1 or r1 == 2, - "perform_alt(composite ready): wrong result") + "or_else(composite ready): wrong result") - -- perform_alt on composite (none ready) - local comp_never = op.choice(never_op(), never_op()) - local r2 = comp_never:perform_alt(function() return 42 end) - assert(r2 == 42, "perform_alt(composite none): fallback not used") + -- composite never-ready: choice(never, never) → fallback + local comp_never = choice(never(), never()) + local ev2 = comp_never:or_else(function() return 42 end) + local r2 = perform(ev2) + assert(r2 == 42, "or_else(composite none): fallback not used") end -------------------------------------------------------- - -- 7) with_nack: winner vs loser vs poll, plus guard+with_nack + -- 6) with_nack: winner vs loser, basic nesting -------------------------------------------------------- do - -- 7.1 with_nack branch wins: nack must NOT fire + -- 6.1 with_nack branch wins: nack must NOT fire do local cancelled = false local with_nack_ev = op.with_nack(function(nack_ev) fiber.spawn(function() - nack_ev:perform() + perform(nack_ev) cancelled = true end) - return task("WIN") + return always("WIN") end) - -- opposing arm never ready - local ev = op.choice(with_nack_ev, never_op()) - local v = ev:perform() + local ev = choice(with_nack_ev, never()) + local v = perform(ev) assert(v == "WIN", "with_nack win: wrong winner") fiber.yield() assert(cancelled == false, "with_nack win: nack fired unexpectedly") end - -- 7.2 with_nack branch loses: nack MUST fire + -- 6.2 with_nack branch loses: nack MUST fire do local cancelled = false local with_nack_ev = op.with_nack(function(nack_ev) fiber.spawn(function() - nack_ev:perform() + perform(nack_ev) cancelled = true end) - return never_op() + return never() end) - local ev = op.choice(with_nack_ev, task("OTHER")) - local v = ev:perform() + local ev = choice(with_nack_ev, always("OTHER")) + local v = perform(ev) assert(v == "OTHER", "with_nack loss: wrong winner") fiber.yield() assert(cancelled == true, "with_nack loss: nack did not fire") end - -- 7.3 with_nack + poll(): some arm ready → nack fires for loser - do - local cancelled = false - local with_nack_ev = op.with_nack(function(nack_ev) - fiber.spawn(function() - nack_ev:perform() - cancelled = true - end) - return never_op() - end) - - local ev = op.choice(with_nack_ev, task("WINNER")) - local ok, v = ev:poll() - assert(ok and v == "WINNER", "with_nack+poll: wrong result") - - fiber.yield() - assert(cancelled == true, "with_nack+poll: nack not fired") - end - - -- 7.4 with_nack + poll(): no arm ready → NO commit, NO nack - do - local cancelled = false - local with_nack_ev = op.with_nack(function(nack_ev) - fiber.spawn(function() - nack_ev:perform() - cancelled = true - end) - return never_op() - end) - - local ev = op.choice(with_nack_ev, never_op()) - local ok, v = ev:poll() - assert(ok == false and v == nil, - "with_nack+poll(no ready): expected no commit") - - fiber.yield() - assert(cancelled == false, - "with_nack+poll(no ready): nack fired unexpectedly") - end - - -- 7.5 guard + with_nack interaction - do - local guard_calls = 0 - local cancelled = false - - local guarded = op.guard(function() - guard_calls = guard_calls + 1 - return op.with_nack(function(nack_ev) - fiber.spawn(function() - nack_ev:perform() - cancelled = true - end) - return never_op() - end) - end) - - local ev = op.choice(guarded, task("OK")) - local v = ev:perform() - assert(v == "OK", "guard+with_nack: wrong winner") - - fiber.yield() - assert(cancelled == true, "guard+with_nack: nack not fired") - assert(guard_calls == 1, "guard+with_nack: builder not once") - end - end - - -------------------------------------------------------- - -- 8) Nested with_nack trees - -------------------------------------------------------- - do - -- 8.1 winner has BOTH outer and inner with_nack: - -- neither outer nor inner nack should fire. + -- 6.3 basic nested with_nack: + -- outer subtree wins via inner leaf → neither nack fires. do local outer_cancelled, inner_cancelled = false, false local outer = op.with_nack(function(outer_nack_ev) - -- watcher for outer nack fiber.spawn(function() - outer_nack_ev:perform() + perform(outer_nack_ev) outer_cancelled = true end) - -- inner with_nack subtree return op.with_nack(function(inner_nack_ev) - -- watcher for inner nack fiber.spawn(function() - inner_nack_ev:perform() + perform(inner_nack_ev) inner_cancelled = true end) - -- this leaf *wins*, so both nacks are on the winner path - return task("INNER_WIN") + return always("INNER_WIN") end) end) - -- Only competitor is never_ready, so outer subtree wins. - local ev = op.choice(outer, never_op()) - local v = ev:perform() + local ev = choice(outer, never()) + local v = perform(ev) assert(v == "INNER_WIN", - "nested with_nack (inner winner): wrong result") + "nested with_nack: wrong result") fiber.yield() assert(outer_cancelled == false, - "nested with_nack (inner winner): outer nack fired unexpectedly") + "nested with_nack: outer nack fired unexpectedly") assert(inner_cancelled == false, - "nested with_nack (inner winner): inner nack fired unexpectedly") + "nested with_nack: inner nack fired unexpectedly") end + end - -- 8.2 winner is in OUTER subtree but NOT in inner subtree: - -- outer nack must NOT fire, inner nack MUST fire. + -------------------------------------------------------- + -- 7) bracket: RAII-style resource management over events + -------------------------------------------------------- + do + ---------------------------------------------------- + -- 7.1 basic success: inner event wins → aborted=false + ---------------------------------------------------- do - local outer_cancelled, inner_cancelled = false, false - - local outer = op.with_nack(function(outer_nack_ev) - fiber.spawn(function() - outer_nack_ev:perform() - outer_cancelled = true - end) - - return op.choice( - -- This leaf wins: path has ONLY outer's nack. - task("OUTER_ONLY"), - -- This leaf loses: path has OUTER and INNER nacks. - op.with_nack(function(inner_nack_ev) - fiber.spawn(function() - inner_nack_ev:perform() - inner_cancelled = true - end) - return never_op() - end) - ) - end) - - local ev = op.choice(outer, never_op()) - local v = ev:perform() - assert(v == "OUTER_ONLY", - "nested with_nack (outer-only winner): wrong result") - - fiber.yield() - assert(outer_cancelled == false, - "nested with_nack (outer-only winner): outer nack fired unexpectedly") - assert(inner_cancelled == true, - "nested with_nack (outer-only winner): inner nack did not fire") + local acq_count = 0 + local rel_count = 0 + local use_count = 0 + local last_res, last_aborted + + local ev = op.bracket( + function() + acq_count = acq_count + 1 + return "RESOURCE" + end, + function(res, aborted) + rel_count = rel_count + 1 + last_res = res + last_aborted = aborted + end, + function(res) + use_count = use_count + 1 + assert(res == "RESOURCE", "bracket basic: wrong resource") + return always(99) + end + ) + + local v = perform(ev) + assert(v == 99, "bracket basic: wrong result") + + assert(acq_count == 1, "bracket basic: acquire not once") + assert(use_count == 1, "bracket basic: use not once") + assert(rel_count == 1, "bracket basic: release not once") + assert(last_res == "RESOURCE", "bracket basic: wrong res in release") + assert(last_aborted == false, + "bracket basic: aborted flag should be false on success") end - -- 8.3 winner is OUTSIDE the outer with_nack subtree: - -- both outer and inner nacks MUST fire. + ---------------------------------------------------- + -- 7.2 losing branch in choice → aborted=true + ---------------------------------------------------- do - local outer_cancelled, inner_cancelled = false, false - - local outer = op.with_nack(function(outer_nack_ev) - fiber.spawn(function() - outer_nack_ev:perform() - outer_cancelled = true - end) + local acq_count, use_count, rel_count = 0, 0, 0 + local last_aborted + + local bracket_ev = op.bracket( + function() + acq_count = acq_count + 1 + return "R" + end, + function(_, aborted) + rel_count = rel_count + 1 + last_aborted = aborted + end, + function(r) + use_count = use_count + 1 + assert(r == "R") + return never() + end + ) + + local ev = choice(bracket_ev, always("WIN")) + local v = perform(ev) + assert(v == "WIN", "bracket choice: wrong winner") + + assert(acq_count == 1, "bracket choice: acquire not once") + assert(use_count == 1, "bracket choice: use not once") + assert(rel_count == 1, "bracket choice: release not once") + assert(last_aborted == true, + "bracket choice: aborted flag should be true when losing") + end - -- Entire outer subtree never becomes ready. - return op.with_nack(function(inner_nack_ev) - fiber.spawn(function() - inner_nack_ev:perform() - inner_cancelled = true - end) - return never_op() - end) + ---------------------------------------------------- + -- 7.3 bracket + or_else: + -- bracket arm never commits → aborted=true, fallback used + ---------------------------------------------------- + do + local acq_count, use_count, rel_count = 0, 0, 0 + local last_aborted, fallback_called + + local bracket_ev = op.bracket( + function() + acq_count = acq_count + 1 + return "R" + end, + function(_, aborted) + rel_count = rel_count + 1 + last_aborted = aborted + end, + function(r) + use_count = use_count + 1 + assert(r == "R") + return never() + end + ) + + local ev = bracket_ev:or_else(function() + fallback_called = true + return "FALLBACK" end) - -- Top-level winner is outside the outer subtree. - local ev = op.choice(outer, task("TOP_WIN")) - local v = ev:perform() - assert(v == "TOP_WIN", - "nested with_nack (outer loses): wrong winner") + local res = perform(ev) - fiber.yield() - assert(outer_cancelled == true, - "nested with_nack (outer loses): outer nack did not fire") - assert(inner_cancelled == true, - "nested with_nack (outer loses): inner nack did not fire") + assert(res == "FALLBACK", + "bracket+or_else: expected fallback result") + assert(fallback_called == true, + "bracket+or_else: fallback thunk not called") + + assert(acq_count == 1, "bracket+or_else: acquire once") + assert(use_count == 1, "bracket+or_else: use once") + assert(rel_count == 1, "bracket+or_else: release once") + assert(last_aborted == true, + "bracket+or_else: aborted flag should be true when losing") end end diff --git a/tests/test_stream-file.lua b/tests/test_stream-file.lua index bdd9104..09bb04f 100644 --- a/tests/test_stream-file.lua +++ b/tests/test_stream-file.lua @@ -14,6 +14,8 @@ local compat = require 'fibers.stream.compat' compat.install() +local perform, choice = op.perform, op.choice + local function test() local rd, wr = file.pipe() local message = "hello, world\n" @@ -77,10 +79,11 @@ local function test_read_op() wg:done() end) - local chars, err = op.choice( + local task = choice( rd:read_all_chars_op(), sleep.sleep_op(0.01):wrap(function () return nil, 'timeout' end) - ):perform() + ) + local chars, err = perform(task) assert(not chars and err == 'timeout') @@ -118,10 +121,11 @@ local function test_write_op() wg:done() end) - local written, err = op.choice( + local task = choice( wr:write_chars_op(msg..msg2), sleep.sleep_op(0.01):wrap(function () return nil, 'timeout' end) - ):perform() + ) + local written, err = perform(task) assert(not written and err=='timeout') diff --git a/tests/test_waitgroup.lua b/tests/test_waitgroup.lua index 1b87a10..007fcfe 100644 --- a/tests/test_waitgroup.lua +++ b/tests/test_waitgroup.lua @@ -11,11 +11,14 @@ local op = require 'fibers.op' local waitgroup = require 'fibers.waitgroup' local sc = require 'fibers.utils.syscall' +local perform, choice = op.perform, op.choice + local function test_nowait() local wg = waitgroup.new() - wg:wait_op():perform_alt(function() + local task = wg:wait_op():or_else(function() error("blocked on empty waitgroup") end) + perform(task) print("No wait test: ok") end @@ -32,9 +35,12 @@ local function test_simple() end) end - wg:wait_op():wrap(function() - error("waitgroup didn't block when it should have") - end):perform_alt(function() end) + perform( + wg:wait_op():wrap(function() + error("waitgroup didn't block when it should have") + end) + :or_else(function() end) + ) wg:wait() print("Simple test: ok") @@ -68,10 +74,12 @@ local function test_complex() end while not done do - op.choice( - wg:wait_op():wrap(function() done = true end), - sleep.sleep_op(0.9):wrap(extra_work) - ):perform() + perform( + choice( + wg:wait_op():wrap(function() done = true end), + sleep.sleep_op(0.9):wrap(extra_work) + ) + ) end assert(sc.monotime() - start > 1.5)