From 3048eed2aaaf3595d7e22ac792a41f60214a8b87 Mon Sep 17 00:00:00 2001
From: Cloud Wu <cloudwu@gmail.com>
Date: Thu, 18 Apr 2019 11:17:07 +0800
Subject: [PATCH] try to fix #988

---
 lualib/skynet/cluster.lua | 60 +++++++++++++++++++++++++++++++++------
 1 file changed, 52 insertions(+), 8 deletions(-)

diff --git a/lualib/skynet/cluster.lua b/lualib/skynet/cluster.lua
index c1633d3e..d8601145 100644
--- a/lualib/skynet/cluster.lua
+++ b/lualib/skynet/cluster.lua
@@ -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()