Rework fibres runtime for coroutine-safe errors, scopes, and channel/waitgroup semantics

src/coxpcall.lua

@@ -0,0 +1,108 @@
+-- coxpcall.lua
+
+local M = {}
+
+-------------------------------------------------------------------------------
+-- Checks if (x)pcall function is coroutine safe
+-------------------------------------------------------------------------------
+local function isCoroutineSafe(func)
+    local co = coroutine.create(function()
+        return func(coroutine.yield, function() end)
+    end)
+
+    coroutine.resume(co)
+    return coroutine.resume(co)
+end
+
+-- Fast path: environment already has coroutine-safe pcall/xpcall
+if isCoroutineSafe(pcall) and isCoroutineSafe(xpcall) then
+    -- No globals; just return plain ones
+    M.pcall   = pcall
+    M.xpcall  = xpcall
+    M.running = coroutine.running
+    return M
+end
+
+-------------------------------------------------------------------------------
+-- Implements xpcall with coroutines
+-------------------------------------------------------------------------------
+
+local performResume, handleReturnValue
+local oldpcall, oldxpcall = pcall, xpcall
+local unpack = rawget(table, "unpack") or _G.unpack
+local pack   = rawget(table, "pack") or function(...)
+    return { n = select("#", ...), ... }
+end
+local running = coroutine.running
+local coromap = setmetatable({}, { __mode = "k" })
+
+function handleReturnValue(err, co, status, ...)
+    if not status then
+        return false, err(debug.traceback(co, (...)), ...)
+    end
+    if coroutine.status(co) == 'suspended' then
+        return performResume(err, co, coroutine.yield(...))
+    else
+        return true, ...
+    end
+end
+
+function performResume(err, co, ...)
+    return handleReturnValue(err, co, coroutine.resume(co, ...))
+end
+
+local function id(trace)
+    return trace
+end
+
+local function coxpcall(f, err, ...)
+    local current = running()
+    if not current then
+        -- Not in a coroutine: fall back to normal pcall/xpcall
+        if err == id then
+            return oldpcall(f, ...)
+        else
+            if select("#", ...) > 0 then
+                local oldf, params = f, pack(...)
+                f = function() return oldf(unpack(params, 1, params.n)) end
+            end
+            return oldxpcall(f, err)
+        end
+    else
+        local res, co = oldpcall(coroutine.create, f)
+        if not res then
+            local newf = function(...) return f(...) end
+            co = coroutine.create(newf)
+        end
+        coromap[co] = current
+        return performResume(err, co, ...)
+    end
+end
+
+local function corunning(coro)
+    if coro ~= nil then
+        assert(type(coro) == "thread",
+               "Bad argument; expected thread, got: " .. type(coro))
+    else
+        coro = running()
+    end
+    while coromap[coro] do
+        coro = coromap[coro]
+    end
+    if coro == "mainthread" then return nil end
+    return coro
+end
+
+-------------------------------------------------------------------------------
+-- Implements pcall with coroutines
+-------------------------------------------------------------------------------
+
+local function copcall(f, ...)
+    return coxpcall(f, id, ...)
+end
+
+M.pcall   = copcall
+M.xpcall  = coxpcall
+M.running = corunning
+
+return M

src/fibers.lua

@@ -21,9 +21,6 @@ local channel   = require 'fibers.channel'
 local op        = require 'fibers.op'
 
 local unpack = rawget(table, "unpack") or _G.unpack
-local pack   = rawget(table, "pack") or function(...)
-    return { n = select("#", ...), ... }
-end
 
 local fibers = {}
 
@@ -33,23 +30,19 @@ fibers.perform = performer.perform
 fibers.now = runtime.now
 
 --- Run a main function under the scheduler's root scope.
---   main_fn :: function(Scope, ...): ()
+--   main_fn :: function(Scope, ...): ...
 function fibers.run(main_fn, ...)
     local root = scope_mod.root()
     local args = { ... }
 
+    -- Run main_fn inside a child scope of the root, in its own fibre.
     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)
-        )
+        local status, err = scope_mod.run(main_fn, unpack(args))
         -- In all cases, stop the scheduler so runtime.main() returns.
         runtime.stop()
-        -- If the main scope failed, treat as fatal for the process.
-        if not res[1] then
-            print(unpack(res, 2, res.n))
+        -- Treat non-ok main scope as fatal for the process.
+        if status ~= "ok" then
+            print(err)
             os.exit(255)
         end
     end)