mirror of
https://github.com/cloudwu/skynet.git
synced 2026-07-22 02:53:09 +00:00
try to fix #988
This commit is contained in:
@@ -3,23 +3,67 @@ local skynet = require "skynet"
|
||||
local clusterd
|
||||
local cluster = {}
|
||||
local sender = {}
|
||||
local task_queue = {}
|
||||
|
||||
local function get_sender(t, node)
|
||||
local c = skynet.call(clusterd, "lua", "sender", node)
|
||||
t[node] = c
|
||||
return c
|
||||
local function request_sender(q, node)
|
||||
local ok, c = pcall(skynet.call, clusterd, "lua", "sender", node)
|
||||
if not ok then
|
||||
skynet.error(c)
|
||||
c = nil
|
||||
end
|
||||
-- run tasks in queue
|
||||
local confirm = coroutine.running()
|
||||
q.confirm = confirm
|
||||
q.sender = c
|
||||
for _, task in ipairs(q) do
|
||||
if type(task) == "table" then
|
||||
if c then
|
||||
skynet.send(c, "lua", "push", table.unpack(task,1,task.n))
|
||||
end
|
||||
else
|
||||
skynet.wakeup(task)
|
||||
skynet.wait(confirm)
|
||||
end
|
||||
end
|
||||
task_queue[node] = nil
|
||||
sender[node] = c
|
||||
end
|
||||
|
||||
setmetatable(sender, { __index = get_sender } )
|
||||
local function get_queue(t, node)
|
||||
local q = {}
|
||||
t[node] = q
|
||||
skynet.fork(request_sender, q, node)
|
||||
return q
|
||||
end
|
||||
|
||||
setmetatable(task_queue, { __index = get_queue } )
|
||||
|
||||
local function get_sender(node)
|
||||
local s = sender[node]
|
||||
if not s then
|
||||
local q = task_queue[node]
|
||||
local task = coroutine.running()
|
||||
table.insert(q, task)
|
||||
skynet.wait(task)
|
||||
skynet.wakeup(q.confirm)
|
||||
return q.sender
|
||||
end
|
||||
return s
|
||||
end
|
||||
|
||||
function cluster.call(node, address, ...)
|
||||
-- skynet.pack(...) will free by cluster.core.packrequest
|
||||
return skynet.call(sender[node], "lua", "req", address, skynet.pack(...))
|
||||
return skynet.call(get_sender(node), "lua", "req", address, skynet.pack(...))
|
||||
end
|
||||
|
||||
function cluster.send(node, address, ...)
|
||||
-- push is the same with req, but no response
|
||||
skynet.send(sender[node], "lua", "push", address, skynet.pack(...))
|
||||
local s = sender[node]
|
||||
if not s then
|
||||
table.insert(task_queue[node], table.pack(address, ...))
|
||||
else
|
||||
skynet.send(sender[node], "lua", "push", address, skynet.pack(...))
|
||||
end
|
||||
end
|
||||
|
||||
function cluster.open(port)
|
||||
@@ -54,7 +98,7 @@ function cluster.register(name, addr)
|
||||
end
|
||||
|
||||
function cluster.query(node, name)
|
||||
return skynet.call(sender[node], "lua", "req", 0, skynet.pack(name))
|
||||
return skynet.call(get_sender(node), "lua", "req", 0, skynet.pack(name))
|
||||
end
|
||||
|
||||
skynet.init(function()
|
||||
|
||||
Reference in New Issue
Block a user