aio — primitive asyncio-like coroutines scheduler

aio.lua

local M = {}

M.tasks = {}


function M.yield(...)

    return coroutine.yield({args=table.pack(...)})

end


function M.sleep(x)

    return coroutine.yield({sleep=M.time() + x})

end


function M.resume(task, ...)

    local ok, state = coroutine.resume(task.coro, ...)
    if not ok or not state then

        return ok, state

    end

    while state.sleep do

        ok, state = coroutine.resume(task.coro, coroutine.yield(state))
        if not ok or not state then

            return ok, state

        end

    end

    return ok, table.unpack(state.args)

end


function M.wait_for(condition, period)

    while not condition() do

        M.sleep(period or 0)

    end

end


function M.create_task(coro, ...)

    if type(coro) ~= "thread" then

        coro = coroutine.create(coro)

    end

    return {
        coro=coro,
        args={...},
        time=0
    }

end


function M.add_task(task, ...)

    if type(task) ~= "table" then

        task = M.create_task(task, ...)

    end
    table.insert(M.tasks, task)
    return task

end


function M.remove_task(task)

    for i, curr in ipairs(M.tasks) do

        if curr == task then

            return table.remove(M.tasks, i)

        end

    end

end


function M.loop(time, error_callback)

    M.time = time
    empty = {}
    while #M.tasks ~= 0 do

        table.sort(M.tasks, function (a, b) return a.time > b.time end)
        local task = table.remove(M.tasks)
        while task.time > M.time() do end

        local ok, state = coroutine.resume(task.coro, table.unpack(task.args))
        task.args = empty

        if not ok then

            if error_callback then

                error_callback(state)

            else

                print("loop got error:", state)

            end

        elseif state and state.sleep then

            task.time = state.sleep
            M.add_task(task)

        end

    end

end

return M

result.txt

Lua 5.3
foo: init       'some init'
bar: init       resuming        bar
foo: received from bar  true    123     from    bar
foo: sleep started      1447671968
alive init      1
alive 1 -> 1447671968
alive init      3
alive 3 -> 1447671968
alive init      2
alive 2 -> 1447671968
alive 1 -> 1447671969
alive 1 -> 1447671970
BOOM!
loop got error:    test.lua:24: attempt to index a nil value (global 'x')
alive 1 -> 1447671971
alive 3 -> 1447671971
alive 1 -> 1447671972
removing alive 3
alive   1       done
foo: sleep finished     1447671978
bar: received from foo  last resume
bar: sleep started      1447671978
bar: sleep finished     1447671988
bar: done
foo: done

test.lua

aio = require("aio")

print(_VERSION)


local function alive(period)

    print("alive init", period)
    for i = 1, 5 do

        local t = os.time()
        print(string.format("alive %d -> %d", period, t))
        aio.sleep(period)
        if period == 1 and i == 5 then

            print("removing alive 3")
            aio.remove_task(three)

        end

        if period == 2 then

            print("BOOM!")
            local y = x.z

        end

    end
    print("alive", period, "done")

end


local function foo(...)

    print("foo: init", ...)
    print("foo: received from bar", aio.resume(bar, "resuming", "bar"))
    print("foo: sleep started", os.time())
    aio.sleep(10)
    print("foo: sleep finished", os.time())
    aio.resume(bar, "last resume")
    print("foo: done")

end

bar = aio.create_task(function (...)

    print("bar: init", ...)
    print("bar: received from foo", aio.yield(123, "from", "bar"))
    print("bar: sleep started", os.time())
    aio.sleep(10)
    print("bar: sleep finished", os.time())
    print("bar: done")

end)


aio.add_task(alive, 1)
three = aio.add_task(alive, 3)
aio.add_task(alive, 2)
aio.add_task(foo, "'some init'")
aio.loop(os.time)