skynet源碼閱讀<3>--網關分析

    繼上一篇介紹了skynet的網絡部分之后，這一篇以網關gate.lua為例，簡單分析下其串接和處理流程。

    在官方給出的范例中，是以examples/main.lua作為啟動腳本的，在此過程中會創建watchdog服務：

    local watchdog = skynet.newservice("watchdog")
    skynet.call(watchdog, "lua", "start", {
        port = 8888,
        maxclient = max_client,
        nodelay = true,
    })

    首先加載watchdog.lua腳本。而在watchdog.lua的加載過程中，創建了gate服務。加載gate.lua過程中，調用gateserver.start(gate)，gateserver會向skynet注冊socket協議的處理：

    skynet.register_protocol {
        name = "socket",
        id = skynet.PTYPE_SOCKET,    -- PTYPE_SOCKET = 6
        unpack = function ( msg, sz )
            return netpack.filter( queue, msg, sz)
        end,
        dispatch = function (_, _, q, type, ...)
            queue = q
            if type then
                MSG[type](...)
            end
        end
    }

    另外gateserver也會注冊lua協議的處理，這里就不展開了。gateserver中會攔截skynet_socket_message；也會攔截部分lua消息（一般是由watchdog轉發而來），並調用gate注冊進來的回調。注意gateserver才是skynet消息的入口，gate只不過是個回調而已。至此，gate服務加載完畢。

    watchdog服務加載完畢后，main.lua中接着調用watchdog的start方法，其參數分別指定了偵聽的端口、最大客戶端連接數、是否延遲等。看下watchdog的start方法：

function CMD.start(conf)
    skynet.call(gate, "lua", "open" , conf)
end

    其緊接着調用gate的open方法，而這個方法在gateserver中被攔截了：

    function CMD.open( source, conf )
        assert(not socket)
        local address = conf.address or "0.0.0.0"
        local port = assert(conf.port)
        maxclient = conf.maxclient or 1024
        nodelay = conf.nodelay
        skynet.error(string.format("Listen on %s:%d", address, port))
        socket = socketdriver.listen(address, port)
        socketdriver.start(socket)
        if handler.open then
            return handler.open(source, conf)
        end
    end

    可以看到，在open方法中創建了socket並開始了偵聽過程。回憶上篇，socket操作的LuaAPI作為socketdriver被實現在lua-socket.c文件中，看一眼這里的listen是如何交互的：

static int
llisten(lua_State *L) {
    const char * host = luaL_checkstring(L,1);
    int port = luaL_checkinteger(L,2);
    int backlog = luaL_optinteger(L,3,BACKLOG);
    struct skynet_context * ctx = lua_touserdata(L, lua_upvalueindex(1));
    int id = skynet_socket_listen(ctx, host,port,backlog);
    if (id < 0) {
        return luaL_error(L, "Listen error");
    }

    lua_pushinteger(L,id);
    return 1;
}

    析取參數，獲取關聯的skynet-context之后，調用skynet_socket.c的skynet_socket_listen：

int 
skynet_socket_listen(struct skynet_context *ctx, const char *host, int port, int backlog) {
    uint32_t source = skynet_context_handle(ctx);
    return socket_server_listen(SOCKET_SERVER, source, host, port, backlog);
}

    拿到context-handle，這個handle在后續創建socket時會被關聯起來。handle作為參數opaque傳遞入socket_server.c中的socket_server_listen方法中：

int 
socket_server_listen(struct socket_server *ss, uintptr_t opaque, const char * addr, int port, int backlog) {
    int fd = do_listen(addr, port, backlog);
    if (fd < 0) {
        return -1;
    }
    struct request_package request;
    int id = reserve_id(ss);
    if (id < 0) {
        close(fd);
        return id;
    }
    request.u.listen.opaque = opaque;
    request.u.listen.id = id;
    request.u.listen.fd = fd;
    send_request(ss, &request, 'L', sizeof(request.u.listen));
    return id;
}

    在作了bind和listen之后，將此socket描述符打包為request寫入socket_server的讀管道，由socket_server_poll輪循處理。至此，gate服務中listen的流程已經非常清晰了。

    再回到gateserver.CMD.open方法中，在socketdriver.listen之后，緊接着調用socketdriver.start開始網絡事件的處理（具體細節請按上述流程參照源碼），最后調用gate.lua的回調handler.open。在這個過程中，gate服務的skynet-context-handle已經與對應的socket綁定了，后續關於此socket的SOCKET消息都會轉發到gate服務中來，具體到代碼則是由gateserver接收過濾后，再做進一步的分發處理。

    那么一個新的客戶端連接是如何被接收創建的呢？

    在socket_server_poll過程中，會檢查是否有網絡事件產生（以下是簡化的代碼）：

int socket_server_poll(struct socket_server *ss, struct socket_message * result, int * more) {
    for (;;) {
        // 管道select
        ......

        // socket event check
        ......

        // s: socket
        switch (s->type) {
        case SOCKET_TYPE_CONNECTING:
            return report_connect(ss, s, result);
        case SOCKET_TYPE_LISTEN: {
            int ok = report_accept(ss, s, result);
            if (ok > 0) { return SOCKET_ACCEPT; }
17             if (ok < 0) { return SOCKET_ERROR; }
            // when ok == 0, retry
            break;
        }
23         // other
24         ......25     }
26 }

    對於正在listen的socket（SOCKET_TYPE_LISTEN），當發生事件（即偵測到有新的連接）時，會在report_accept中accept得連接描述符fd並創建socket結構，然后返回SOCKET_ACCEPT交由上層的skynet_socket.c:skynet_socket_poll處理，后者會封裝類型為SKYNET_SOCKET_TYPE_ACCEPT的skynet-message並推入到gate服務的隊列中去，最終轉發到gateserver中所注冊的SOCKET協議入口。

    回到gateserver中來（見上述gateserver摘錄的代碼），在接收到網絡消息時，先是在unpack中通過netpack（見lua-netpack.c）合並過濾消息，比如TCP消息粘包等（注意skynet_socket_message如果其padding為true，則表示非數據的命令，比如SKYNET_SOCKET_TYPE_ACCEPT）。對於SKYNET_SOCKET_TYPE_ACCEPT命令，netpack會解析並轉換為open命令，最后gateserver會調用到MSG.open：

    function MSG.open(fd, msg)
        if client_number >= maxclient then
            socketdriver.close(fd)
            return
        end
        if nodelay then
            socketdriver.nodelay(fd)
        end
        connection[fd] = true
        client_number = client_number + 1
        handler.connect(fd, msg)
    end

    回調到gate.lua中的connect：

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

    watchdog.lua中的SOCKET.open：

function SOCKET.open(fd, addr)
    skynet.error("New client from : " .. addr)
    agent[fd] = skynet.newservice("agent")
    skynet.call(agent[fd], "lua", "start", { gate = gate, client = fd, watchdog = skynet.self() })
end

    此時，會創建玩家agent服務，並調用start方法：

function CMD.start(conf)
    local fd = conf.client
    local gate = conf.gate
    WATCHDOG = conf.watchdog
    -- slot 1,2 set at main.lua
    host = sprotoloader.load(1):host "package"
    send_request = host:attach(sprotoloader.load(2))
    skynet.fork(function()
        while true do
            send_package(send_request "heartbeat")
            skynet.sleep(500)
        end
    end)

    client_fd = fd
    skynet.call(gate, "lua", "forward", fd)
end

    agent拿到gate服務的標識后，調用forward將自己的標識注冊到gate服務中來：

function CMD.forward(source, fd, client, address)
    local c = assert(connection[fd])
    unforward(c)
    c.client = client or 0
    c.agent = address or source
    forwarding[c.agent] = c
    gateserver.openclient(fd)
end

    gateserver.openclient開始偵聽此socket的事件：

function gateserver.openclient(fd)
    if connection[fd] then
        socketdriver.start(fd)
    end
end

    至此，一個新連接的建立流程就結束了。那么連接建立后網絡數據又是如何轉發進來的呢？流程依然是一致的，socket_server_poll偵測到READ后讀取數據並轉發到gateserver中來，后者調用netpack對數據粘包，此時會調用MSG.more或MSG.data將數據轉交給gate.message：

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", 0, msg, sz)
    else
        skynet.send(watchdog, "lua", "socket", "data", fd, netpack.tostring(msg, sz))
    end
end

    直接將數據轉發給之前通過forward注冊進來的agent，采用的協議是"client"。agent中需注冊此協議的處理，拿到字節流並根據上層的業務協議對數據轉碼，做進一步的處理。這樣，數據接收和轉發的流程就結束了。其它方面，比如數據發送，關閉socket等等，流程上都是一致的，具體細節不再詳述。