multi cluster sender

lualib/skynet/cluster.lua

@@ -2,15 +2,24 @@ local skynet = require "skynet"
 
 local clusterd
 local cluster = {}
+local sender = {}
+
+local function get_sender(t, node)
+	local c = skynet.call(clusterd, "lua", "sender", node)
+	t[node] = c
+	return c
+end
+
+setmetatable(sender, { __index = get_sender } )
 
 function cluster.call(node, address, ...)
 	-- skynet.pack(...) will free by cluster.core.packrequest
-	return skynet.call(clusterd, "lua", "req", node, address, skynet.pack(...))
+	return skynet.call(sender[node], "lua", "req",  address, skynet.pack(...))
 end
 
 function cluster.send(node, address, ...)
 	-- push is the same with req, but no response
-	skynet.send(clusterd, "lua", "push", node, address, skynet.pack(...))
+	skynet.send(sender[node], "lua", "push", address, skynet.pack(...))
 end
 
 function cluster.open(port)
@@ -45,7 +54,7 @@ function cluster.register(name, addr)
 end
 
 function cluster.query(node, name)
-	return skynet.call(clusterd, "lua", "req", node, 0, skynet.pack(name))
+	return skynet.call(sender[node], "lua", "req", 0, skynet.pack(name))
 end
 
 skynet.init(function()

service/clusterd.lua

@@ -1,4 +1,5 @@
 local skynet = require "skynet"
+require "skynet.manager"
 local sc = require "skynet.socketchannel"
 local socket = require "skynet.socket"
 local cluster = require "skynet.cluster.core"
@@ -6,6 +7,7 @@ local cluster = require "skynet.cluster.core"
 local config_name = skynet.getenv "cluster"
 local node_address = {}
 local node_session = {}
+local node_sender = {}
 local command = {}
 local config = {}
 local nodename = cluster.nodename()
@@ -40,13 +42,20 @@ local function open_channel(t, key)
 	local succ, err, c
 	if address then
 		local host, port = string.match(address, "([^:]+):(.*)$")
-		c = sc.channel {
-			host = host,
-			port = tonumber(port),
-			response = read_response,
-			nodelay = true,
-		}
-		succ, err = pcall(c.connect, c, true)
+		c = node_sender[key]
+		if c == nil then
+			c = skynet.newservice "clustersender"
+			if node_sender[key] then
+				-- double check
+				skynet.kill(c)
+				c = node_sender[key]
+			else
+				node_sender[key] = c
+			end
+		end
+
+		succ = pcall(skynet.call, c, "lua", "changenode", host, port)
+
 		if succ then
 			t[key] = c
 			ct.channel = c
@@ -120,56 +129,8 @@ function command.listen(source, addr, port)
 	skynet.ret(skynet.pack(nil))
 end
 
-local function send_request(source, node, addr, msg, sz)
-	local session = node_session[node] or 1
-	-- msg is a local pointer, cluster.packrequest will free it
-	local request, new_session, padding = cluster.packrequest(addr, session, msg, sz)
-	node_session[node] = new_session
-
-	-- node_channel[node] may yield or throw error
-	local c = node_channel[node]
-
-	local tracetag = skynet.tracetag()
-	if tracetag then
-		if tracetag:sub(1,1) ~= "(" then
-			-- add nodename
-			local newtag = string.format("(%s-%s-%d)%s", nodename, node, session, tracetag)
-			skynet.tracelog(tracetag, string.format("session %s", newtag))
-			tracetag = newtag
-		end
-		skynet.tracelog(tracetag, string.format("cluster %s", node))
-		c:request(cluster.packtrace(tracetag))
-	end
-	return c:request(request, session, padding)
-end
-
-function command.req(...)
-	local ok, msg = pcall(send_request, ...)
-	if ok then
-		if type(msg) == "table" then
-			skynet.ret(cluster.concat(msg))
-		else
-			skynet.ret(msg)
-		end
-	else
-		skynet.error(msg)
-		skynet.response()(false)
-	end
-end
-
-function command.push(source, node, addr, msg, sz)
-	local session = node_session[node] or 1
-	local request, new_session, padding = cluster.packpush(addr, session, msg, sz)
-	if padding then	-- is multi push
-		node_session[node] = new_session
-	end
-
-	-- node_channel[node] may yield or throw error
-	local c = node_channel[node]
-
-	c:request(request, nil, padding)
-
-	-- notice: push may fail where the channel is disconnected or broken.
+function command.sender(source, node)
+	skynet.ret(skynet.pack(node_channel[node]))
 end
 
 local proxy = {}

service/clusterproxy.lua

@@ -22,11 +22,12 @@ skynet.forward_type( forward_map ,function()
 	if n then
 		address = n
 	end
+	local sender = skynet.call(clusterd, "lua", "sender", node)
 	skynet.dispatch("system", function (session, source, msg, sz)
 		if session == 0 then
 			skynet.send(clusterd, "lua", "push", node, address, msg, sz)
 		else
-			skynet.ret(skynet.rawcall(clusterd, "lua", skynet.pack("req", node, address, msg, sz)))
+			skynet.ret(skynet.rawcall(sender, "lua", skynet.pack("req", address, msg, sz)))
 		end
 	end)
 end)

service/clustersender.lua

@@ -0,0 +1,87 @@
+local skynet = require "skynet"
+local sc = require "skynet.socketchannel"
+local socket = require "skynet.socket"
+local cluster = require "skynet.cluster.core"
+local ignoreret = skynet.ignoreret
+
+local channel
+local session = 1
+
+local command = {}
+
+local function send_request(addr, msg, sz)
+	-- msg is a local pointer, cluster.packrequest will free it
+	local current_session = session
+	local request, new_session, padding = cluster.packrequest(addr, session, msg, sz)
+	session = new_session
+
+	-- node_channel[node] may yield or throw error
+	local tracetag = skynet.tracetag()
+	if tracetag then
+		if tracetag:sub(1,1) ~= "(" then
+			-- add nodename
+			local newtag = string.format("(%s-%s-%d)%s", nodename, node, session, tracetag)
+			skynet.tracelog(tracetag, string.format("session %s", newtag))
+			tracetag = newtag
+		end
+		skynet.tracelog(tracetag, string.format("cluster %s", node))
+		channel:request(cluster.packtrace(tracetag))
+	end
+	return channel:request(request, current_session, padding)
+end
+
+function command.req(...)
+	local ok, msg = pcall(send_request, ...)
+	if ok then
+		if type(msg) == "table" then
+			skynet.ret(cluster.concat(msg))
+		else
+			skynet.ret(msg)
+		end
+	else
+		skynet.error(msg)
+		skynet.response()(false)
+	end
+end
+
+function command.push(addr, msg, sz)
+	local request, new_session, padding = cluster.packpush(addr, session, msg, sz)
+	if padding then	-- is multi push
+		session = new_session
+	end
+
+	-- node_channel[node] may yield or throw error
+	channel:request(request, nil, padding)
+end
+
+local function read_response(sock)
+	local sz = socket.header(sock:read(2))
+	local msg = sock:read(sz)
+	return cluster.unpackresponse(msg)	-- session, ok, data, padding
+end
+
+function command.changenode(host, port)
+	local c = sc.channel {
+			host = host,
+			port = tonumber(port),
+			response = read_response,
+			nodelay = true,
+		}
+	succ, err = pcall(c.connect, c, true)
+	if succ then
+		channel = c
+		skynet.ret(skynet.pack(nil))
+	else
+		skynet.response()(false)
+	end
+end
+
+skynet.start(function()
+	skynet.dispatch("lua", function(session , source, cmd, ...)
+		local f = assert(command[cmd])
+		f(...)
+	end)
+end)
+
+
+