diff --git a/lualib/skynet/cluster.lua b/lualib/skynet/cluster.lua index dd3b9dde..c1633d3e 100644 --- a/lualib/skynet/cluster.lua +++ b/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() diff --git a/service/clusterd.lua b/service/clusterd.lua index 9a062796..b73b0743 100644 --- a/service/clusterd.lua +++ b/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 = {} diff --git a/service/clusterproxy.lua b/service/clusterproxy.lua index e0cc5daa..d82ca44c 100644 --- a/service/clusterproxy.lua +++ b/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) diff --git a/service/clustersender.lua b/service/clustersender.lua new file mode 100644 index 00000000..d9fa29ab --- /dev/null +++ b/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) + + +