Lua Thread Pool

thread.lua

local thread = {}

local threadpoolt = {
  __index = {}
}

function threadpoolt.__index:fork(forked)
  local co = coroutine.create(forked)
  self[co] = {}
  return co
end

function threadpoolt.__index:on(co, callback)
  table.insert(self[co], callback)
  return co
end

function threadpoolt.__index:success(co, callback)
  return self:on(co, function(status, ...)
      if status then callback(...) end
    end)
end

function threadpoolt.__index:failure(co, callback)
  return self:on(co, function(status, ...)
      if not status then callback(...) end
    end)
end

local function call(callbacks, ...)
  local indices = {}
  for index, callback in ipairs(callbacks) do
    if not pcall(callback, ...) then
      table.insert(indices, 1, index)
    end
  end
  for _, index in ipairs(indices) do
    table.remove(callbacks, index)
  end
  return ...
end

function threadpoolt.__index:call(co, ...)
  return call(self[co], coroutine.resume(co, ...))
end

function threadpoolt.__index:continue()
  for co, callbacks in pairs(self) do
    call(callbacks, coroutine.resume(co))
  end
end

function threadpoolt.__index:reap()
  for co, _ in pairs(self) do
    if coroutine.status(co) == "dead" then
      self[co] = nil
    end
  end
  return next(self) == nil
end

function thread.pool()
  return setmetatable({}, threadpoolt)
end

return thread