lihao/skynet
1
0
Fork 0
mirror of https://github.com/cloudwu/skynet.git synced 2026-07-24 03:53:09 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'cluster'

Browse Source
This commit is contained in:
Cloud Wu
2019-03-26 09:46:57 +08:00
parent 2c247b511e 1c3b563ef8
commit 7cb4bbb960
4 changed files with 140 additions and 61 deletions
Download Patch File Download Diff File Expand all files Collapse all files

75
service/clusterd.lua
View File

@@ -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 = {}

3
service/clusterproxy.lua
View File

@@ -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)

108
service/clustersender.lua Normal file
View File

@@ -0,0 +1,108 @@
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 waiting = {}
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
local function wait()
local co = coroutine.running()
table.insert(waiting, co)
skynet.wait(co)
end
function command.req(...)
if channel == nil then
wait()
end
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)
if channel == nil then
wait()
end
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 channel then
channel:close()
end
if succ then
channel = c
for k, co in ipairs(waiting) do
waiting[k] = nil
skynet.wakeup(co)
end
skynet.ret(skynet.pack(nil))
else
channel = nil -- reset channel
skynet.response()(false)
end
end
skynet.start(function()
skynet.dispatch("lua", function(session , source, cmd, ...)
local f = assert(command[cmd])
f(...)
end)
end)