rpc-tp.lua

local levee = require("levee")
local _ = levee._


local RPCThreadPool_mt = {}


function RPCThreadPool_mt:__index(name)
	return function(self, ...)
		local sender, recver = self.h:pipe()
		self.q.sender:send({sender, {name, {...}}})
		return recver
	end
end


local function RPCThreadPool(h, M, n)
	local self = setmetatable({}, RPCThreadPool_mt)

	self.h = h

	self.p = h:pool(
		function()
			local t = h.thread:spawn(
				function(h)
					local err, s = h.parent:recv()
					local M = loadstring(s)()
					while true do
						local err, req = h.parent:recv()
						if err then return err end
						local name, a = unpack(req)
						-- should wrap with a pcall
						h.parent:pass(M[name](unpack(a)))
					end
				end)
			t:send(string.dump(M))
			return t
		end, n)

	self.q = {}
	self.q.sender, self.q.recver = h:queue()

	h:spawn(function()
		for job in self.q.recver do
			local sender, req = unpack(job)
			local err, child = self.p:recv()
			child:send(req)
			h:spawn(function()
				sender:pass(child:recv())
				self.p:send(child)
			end)
		end
	end)

	return self
end


local function main()
	local h = levee.Hub()

	local function M()
		return {
			twice = function(n)
				return nil, n * 2
			end,
			add = function(a, b)
				return nil, a + b
			end,
		}
	end

	local pool = RPCThreadPool(h, M, 3)
	print(pool:twice(4):recv())
	print(pool:add(6, 4):recv())
end


main()