skynet源碼分析之cluster集群模式

比起slave/harbor集群模式，skynet提供了用的更為廣泛的cluster集群模式，參考官方wiki https://github.com/cloudwu/skynet/wiki/Cluster。cluster模式利用socketchannel庫（http://www.cnblogs.com/RainRill/p/8892648.html） 與其他skynet進程進行交互，每個請求包帶一個唯一的session值，對端回應包附帶session值。cluster集群模式tcp通道是單向的，即skynet進程1（集群中的節點）通過tcp通道向進程2發送請求包，進程2回應包也走這一通道。但是，進程2向進程1發送請求包及進程1的回應包則是另一條tcp通道。

每個集群節點都有一份完整的cluster配置，會啟動一個clusterd的服務，調用loadconfig加載配置。

 第11-19行，加載配置文件（也可以手動傳入配置table tmp）

第20-24行，保存節點名-地址映射關系

-- service/clusterd.lua
 skynet.start(function()
     loadconfig()
     skynet.dispatch("lua", function(session , source, cmd, ...)
         local f = assert(command[cmd])
         f(source, ...)
     end)
 end)
 
 local function loadconfig(tmp)
     if tmp == nil then
         tmp = {}
         if config_name then
             local f = assert(io.open(config_name))
             local source = f:read "*a"
             f:close()
             assert(load(source, "@"..config_name, "t", tmp))()
         end
     end
     for name,address in pairs(tmp) do
         ...
         node_address[name] = address
         ...
     end
 end

以skynet進程1的A服務向skynet進程2的B服務發送請求包及回應為例，說明cluster的工作流程：

 對於進程2，配置了 db = "127.0.0.1:2528"，啟動后調用cluster.open "db"。

第4行，給clusterd服務發送消息。

第12-15行，啟動一個gate服務，然后通知gate服務監聽配置的地址。gate調用socket.listen監聽外部socket連接。

第20行，watchdog就是clusterd服務的地址。

-- lualib/skynet/cluster.lua
function cluster.open(port)
    if type(port) == "string" then
        skynet.call(clusterd, "lua", "listen", port)
    else
        skynet.call(clusterd, "lua", "listen", "0.0.0.0", port)
    end
end

-- service/clusterd.lua
function command.listen(source, addr, port)
    local gate = skynet.newservice("gate")
    ...
    skynet.call(gate, "lua", "open", { address = addr, port = port })
    skynet.ret(skynet.pack(nil))
end

-- servcice/gate.lua
function handler.open(source, conf)
    watchdog = conf.watchdog or source
end

對於進程1，調用cluster.call(db, "A", ...)，給節點名為db（進程2）的A服務發送請求，最終調用到send_request

第9行，請求包帶上唯一的sesssion值

第11行，按cluster定義的模式打包數據

第15行，獲取socketchannel對象，如果第一次請求，會先創建socketchannel對象，並建立tcp連接

第16行，調用socketchannel的request接口發送請求包

-- lualib/skynet/cluster.lua
function cluster.call(node, address, ...)
    -- skynet.pack(...) will free by cluster.core.packrequest
    return skynet.call(clusterd, "lua", "req", node, address, skynet.pack(...))
end

-- service/clusterd.lua
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]

    return c:request(request, session, padding)
end

function command.req(...)
    local ok, msg = pcall(send_request, ...)
    if ok then
        ...
        skynet.ret(msg)
    end
end

創建socket對象時提供了response參數（第6行），所以是采用帶session值的請求-回應模式。

第11行，協程阻塞在socket.read上，此時暫停co，等待回應包

-- service/clusterd
    local host, port = string.match(address, "([^:]+):(.*)$")
    c = sc.channel {
        host = host,
        port = tonumber(port),
        response = read_response,
        nodelay = true,
    }

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

對於進程2，gate服務收到進程1的tcp連接請求后，

第8行，給clusterd服務發送消息

第17-18行，clusterd收到后，新建一個clusteragent服務。注：clusteragent是skynet最近新加的。參考https://blog.codingnow.com/2018/04/skynet_cluster.html#more

第24-28行，clusteragent服務專門處理進程1的cluster模式的請求。每個cluster節點連接都新建一個cluseteragent服務去處理請求包。

-- service/gate.lua
function handler.connect(fd, addr)
    local c = {
        fd = fd,
        ip = addr,
    }
    connection[fd] = c
    skynet.send(watchdog, "lua", "socket", "open", fd, addr)
end

-- service/clusterd.lua
function command.socket(source, subcmd, fd, msg)
    if subcmd == "open" then
        skynet.error(string.format("socket accept from %s", msg))
        -- new cluster agent
        cluster_agent[fd] = false
        local agent = skynet.newservice("clusteragent", skynet.self(), source, fd)
        cluster_agent[fd] = agent
        ...     
end

-- service/clusterdagent.lua
skynet.start(function()
    skynet.register_protocol {
        name = "client",
        id = skynet.PTYPE_CLIENT,
        unpack = cluster.unpackrequest,
        dispatch = dispatch_request,
    }
    ...
end

當gate服務收到請求包后，轉發給對應的clusteragent服務（第7行），

-- service/gate.lua
function handler.message(fd, msg, sz)
    -- recv a package, forward it
    local c = connection[fd]
    local agent = c.agent
    if agent then
        skynet.redirect(agent, c.client, "client", fd, msg, sz)
    else
        skynet.send(watchdog, "lua", "socket", "data", fd, netpack.tostring(msg, sz))
    end
end

clusteragent服務消息分發函數dispatch_request，

第7-9行，如果是push請求，不需要回應，send給目的服務（B服務）后直接返回即可

第11行，如果是call請求，需要回應，給目的服務（B服務）發送消息，然后等待B服務處理完返回。

第14-21行，將消息打包成回應包，通過tcp返回給請求端（skynet進程1）。

進程1收到回應后，重啟協程，返回結果給請求服務（A服務）。這就是cluster模式的調用流程。

-- service/clusteragent.lua
local function dispatch_request(_,_,addr, session, msg, sz, padding, is_push)
    if cluster.isname(addr) then
        addr = register_name[addr]
    end
    if addr then
        if is_push then
            skynet.rawsend(addr, "lua", msg, sz)
            return  -- no response
        else
            ok , msg, sz = pcall(skynet.rawcall, addr, "lua", msg, sz)
        end
    if ok then
        response = cluster.packresponse(session, true, msg, sz)
        if type(response) == "table" then
            for _, v in ipairs(response) do
                socket.lwrite(fd, v)
            end
        else
            socket.write(fd, response)
        end
     ...
end