parallel2 - a better parallel module

parallel2.lua

--[[

parallel2 - a better parallel module
====================================

@author Lambda Fairy (https://github.com/lfairy)

**parallel2** is a complete rewrite of the ComputerCraft parallel API.
It gives you many features over the original, including:

* Killing threads using `kill()`
* Stepping threads manually using `step()`
* Sending fake events using `feed()`


Installation
------------

This is just another API, so save it in your `api` folder as `parallel2`.


Porting from parallel
---------------------

This:

    parallel.waitForAny(f, g, h)

becomes:

    local ctx = parallel2.group(f, g, h)
    while ctx.living > 0 do
        ctx:step()
    end

This:

    parallel.waitForAll(f, g, h)

becomes:

    parallel2.group(f, g, h):run()


More examples
-------------

You can feed in "fake" events using the `feed()` method:

    local ctx = parallel2.group(...)
    ctx:feed('char', 'h')
    ctx:feed('char', 'i')

Combined with the Shell API, this can be very interesting...

Stopping one thread from another one is easy as well:

    function alice(ctx)
        print('casually killing bob')
        ctx:kill('bob')
    end

    function bob(ctx)
        sleep(0) -- Give Alice a chance to run first
        print('this message will never see the light of day')
    end

    parallel2.group({alice = alice, bob = bob}):run()

Note that to use `kill`, you must associate labels with your functions
using [table notation][].

[table notation]: http://www.lua.org/pil/3.6.html

Nesting contexts (putting a group inside another group) is theoretically
supported, but I haven't tested it. Caveat emptor.

]]

yield = (os and os.pullEvent) or coroutine.yield

function group (...)
    local args = {...}
    if #args == 1 and type(args[1]) == 'table' then
        -- group({alice = ..., bob = ...})
        return Context:new(args[1])
    else
        -- group(alice, bob)
        return Context:new(args)
    end
end

Context = {}

function Context:new (functions)
    local o = {}
    setmetatable(o, self)
    self.__index = self

    o.threads = {}
    o.living = 0
    for k, v in pairs(functions) do
        o.threads[k] = coroutine.create(function() v(o) end)
        o.living = o.living + 1
    end

    o.coroutine = coroutine.create(function () o:run() end)

    return o
end

function Context:run ()
    local ev_data
    local ev_filters = {}

    -- Delete any coroutines that are dead already
    self:_cullDead()

    while self.living > 0 do
        -- Loop through all the coroutines
        for label, co in pairs(self.threads) do
            ev_filters[co] = _stepCoroutine(co, ev_filters[co], ev_data)
        end

        -- Filter out all the dead ones
        self:_cullDead()

        -- Grab the next event
        if self.living > 0 then
            ev_data = pack(yield())
        end
    end
end

Context.__call = Context.run -- Support nested contexts

function Context:_cullDead ()
    self.threads = filter(function (co)
        return coroutine.status(co) ~= 'dead'
    end, self.threads)
    self.living = countkeys(self.threads)
end

function _stepCoroutine (co, ev_filter, ev_data)
    ev_data = ev_data or {}
    local ev_type = ev_data[1]
    -- Only resume a coroutine if it needs the data we're holding
    -- (except for terminate events, which are always sent)
    if ev_filter == nil or ev_filter == ev_type or
            ev_type == 'terminate' then
        local ok, param = coroutine.resume(co, unpack(ev_data))

        if ok then
            return param
        else
            error(param)
        end
    else
        -- If the event doesn't match, don't change anything
        return ev_filter
    end
end

function Context:feedSafe (...)
    return coroutine.resume(self.coroutine, ...)
end

function Context:feed (...)
    local results = pack(self:feedSafe(...))
    if not results[1] then
        error(results[2])
    else
        table.remove(results, 1)
        return unpack(results)
    end
end

function Context:step ()
    return self:feed(yield())
end

function Context:kill (key)
    if self.threads[key] then
        self.threads[key] = nil
        return true
    else
        return false
    end
end

-- Returns its arguments as an array. This is the inverse of unpack().
function pack (...)
    return {...}
end

function countkeys (items)
    local n = 0
    for k, v in pairs(items) do
        n = n + 1
    end
    return n
end

function filter (f, list)
    local result = {}
    for k, v in pairs(list) do
        if f(v) then
            result[k] = v
        end
    end
    return result
end

function testParallel2 ()
    local function testa ()
        local r = coroutine.yield('ham')
        print(r)
    end
    local function testb ()
        for i = 1, 3 do
            local r = coroutine.yield('eggs')
            print('\t' .. r)
        end
    end
    local function testc (ctx)
        local r
        ctx:kill(4) -- testd
        for i = 1, 8 do
            r = coroutine.yield()
            print('\t\t' .. r)
        end
    end
    local function testd ()
        coroutine.yield() -- give other coroutines a chance to run
        error('this should never happen')
    end
    local ctx = group(testa, testb, testc, testd)
    ctx:feed(nil)
    while true do
        for k, v in pairs({ham = true, spam = true, eggs = true, ni = true}) do
            print('== ' .. ctx.living  .. ' running ===========')
            print()
            ctx:feed(k)
            print()
            if ctx.living == 0 then
                return
            end
        end
    end
end

-- Uncomment to see heaps of scrolling text
--testParallel2()