【专题5】skynet的luaAPI实现专项分析

skynet

自用笔记，仅供参考

参考文档

http://cloudwu.github.io/lua53doc/manual.html#6.4.1

https://github.com/cloudwu/skynet/wiki

【专题4】搞明白skynet的C语言到lua环境建立之一、二、三 篇

背景

skynet一个关键的优势是使用lua语言撰写脚本，而使用脚本语言写逻辑的一个大好处就是可以使用顺序逻辑描述业务。表面的平整之下实际是C语言对lua虚拟机的调度器在起作用。

阻塞API从lua中yield回C代码中，之后有了事件再次resume，看起来实现很简单，但是更加复杂的是错误的处理，API调用不知道会经历多少艰辛，出错、超时如何处理？这个是关键所在。

本篇是 【专题4】搞明白skynet的C语言到lua环境建立之x系列 的延续篇

API研究 sleep()

API sleep()的进入

既然已经弄明白lua运行环境和入口，所以可以直接从skynet的API开始看：

    function skynet.sleep(ti)
        local session = c.intcommand("TIMEOUT",ti)
        local succ, ret = coroutine_yield("SLEEP", session)
        -- 上半部戛然而止

yield 必会回到suspend()中：

    local function raw_dispatch_message(prototype, msg, sz, session, source)
        ...
        suspend(co, coroutine_resume(co, true, msg, sz))
        ...

进而被suspend()下述部分逻辑处理：

    ...
    elseif command == "SLEEP" then
        session_id_coroutine[param] = co
        sleep_session[co] = param
    ...
    dispatch_wakeup()
    dispatch_error_queue()

可见 suspend仅仅将session和co记录在案，注意这个表：sleep_session[]。

然后就如果一切干净，suspend函数便会退出，再回到 raw_dispatch_message() ，再回到 skynet.dispatch_message，然后再回到_cb()，再回归skynet的大循环中去。

API sleep()的退出

退出过程和timeout大致一样：

    dispatch_message() 
    -> pcall(raw_dispatch_message,...) 
    ->...
    -> suspend(co, coroutine_resume(co, true, msg, sz))
    --注意，其中第一个返回值是true

回到 skynet.sleep()后半段，如下：

    function skynet.sleep(ti)
        local session = c.intcommand("TIMEOUT",ti)
        local succ, ret = coroutine_yield("SLEEP", session)
        --上半部截止
        -- coroutine_resume(co, true, msg, sz)  
        -- succ = true
        --从下半段分析  
        sleep_session[coroutine.running()] = nil
        if succ then
            return  --一个正常的sleep完成
        end
        if ret == "BREAK" then
            return "BREAK"
        else
            error(ret)
        end
    end

API研究 skynet.call()

API应用实例：

    skynet.call(gate, "lua", "kick", fd)

命令发出

    function skynet.call(addr, typename, ...)
        local p = proto[typename]
        local session = c.send(addr, p.id , nil , p.pack(...))    --<- 和sleep、timeout之间的区别
        if session == nil then
            error("call to invalid address " .. skynet.address(addr))
        end
        return p.unpack(yield_call(addr, session))
    end

大多数代码很简单，主要就是获取session。
关键是：

    yield_call(addr, session)

函数定义：

    local function yield_call(service, session)
        watching_session[session] = service
        local succ, msg, sz = coroutine_yield("CALL", session)
        watching_session[session] = nil
        if not succ then
            error "call failed"
        end
        return msg,sz
    end

coroutine_yield 返回 true, "CALL", session

回到suspend()中：

        ...
        suspend(co, coroutine_resume(co, true, msg, sz))
        ...

进入其如下逻辑路径：

        if command == "CALL" then
            session_id_coroutine[param] = co

后面如果一切干净，suspend函数便会退出，再回到 raw_dispatch_message() ，再回到 skynet.dispatch_message，然后再回到_cb()，再回归skynet的大循环中去。

命令接收

call底层也是msg的交互，在另外的coroutine会接收到msg：

skynet.dispatch_message 
-> raw_dispatch_message

进入raw_dispatch_message函数的下述逻辑路径：

        local p = proto[prototype]
        ...
        local f = p.dispatch    --调用的func
        if f then
            local ref = watching_service[source]
            if ref then
                watching_service[source] = ref + 1
            else
                watching_service[source] = 1
            end
            --新建一个coroutine
            local co = co_create(f)
            --标注
            session_coroutine_id[co] = session
            session_coroutine_address[co] = source --这个重要，将来源地址挂在本coroutine上
            suspend(co, coroutine_resume(co, session,source, p.unpack(msg,sz)))
        ...

注意，上面resume之前，还将来源加入关注表，新建立的coroutine 标注了来源地址：

    watching_service[source] = ref + 1
    ...
    session_coroutine_address[co] = source

最后一个： coroutine_resume(co, session,source, p.unpack(msg,sz))

在新coroutine中调用了本服务使用skynet.dispatch()安装的回调函数。示例代码类如：

    skynet.dispatch("lua", function(_,_, command, ...)
        local f = CMD[command]
        skynet.ret(skynet.pack(f(...)))
    end)

命令返回

命令返回调用 skynet.ret 来完成。

    function skynet.ret(msg, sz)
        msg = msg or ""
        return coroutine_yield("RETURN", msg, sz)
    end

异常简单，yield "RETURN"

看 suspend() 处理后续逻辑的代码上，：

    elseif command == "RETURN" then
        local co_session = session_coroutine_id[co]
        --本coroutine上挂载的是来源地址，见上面，是在dispatch函数被resume之前安装的
        local co_address = session_coroutine_address[co]
        if param == nil or session_response[co] then
            error(debug.traceback(co))
        end
        session_response[co] = true
        local ret
        if not dead_service[co_address] then
            --上面的逻辑都是在整理参数，检查什么的，下面才是重点：
            ret = c.send(co_address, skynet.PTYPE_RESPONSE, co_session, param, size) ~= nil
            if not ret then
                -- If the package is too large, returns nil. so we should report error back
                c.send(co_address, skynet.PTYPE_ERROR, co_session, "")
            end
        elseif size ~= nil then
            c.trash(param, size)
            ret = false
        end
        return suspend(co, coroutine_resume(co, ret))

注意最后有一个resume。wiki中LuaAPI章节中 skynet.ret被声明为非阻塞，果然是如此。

也就是说，skynet.ret调用之后，还会继续运行下去。

命令返回闭包 skynet.response

代码简单：

    function skynet.response(pack)
        pack = pack or skynet.pack
        return coroutine_yield("RESPONSE", pack)
    end

suspend() 处理后续逻辑的代码上：

    elseif command == "RESPONSE" then
        --获得session和来源的addr
        local co_session = session_coroutine_id[co]
        local co_address = session_coroutine_address[co]
        ...
        --下面拿到 yield的pack函数
        local f = param
        --这里定义了一个function，太长省略
        local function response(ok, ...)
            ...
        end
        -- 下面的部分和skynet.ret基本类似了
        watching_service[co_address] = watching_service[co_address] + 1
        session_response[co] = true
        unresponse[response] = true
        --关键在resume的结果就是上面定义的函数
        return suspend(co, coroutine_resume(co, response))

后面，我们先复习一下wiki中对该返回函数的说明：

skynet.response 返回的闭包可用于延迟回应。调用它时，第一个参数通常是 true表示是一个正常的回应，之后的参数是需要回应的数据。如果是 false，则给请求者抛出一个异常。它的返回值表示回应的地址是否还有效。如果你仅仅想知道回应地址的有效性，那么可以在第一个参数传入 "TEST" 用于检测。

再详细看看这个response()的实现：

        local function response(ok, ...)
            --TEST 命令是用来测试目标是否还在的
            if ok == "TEST" then
                if dead_service[co_address] then
                    release_watching(co_address)
                    unresponse[response] = nil
                    f = false
                    return false
                else
                    return true
                end
            end
            ...
            local ret
            if not dead_service[co_address] then
                if ok then
                    ret = c.send(co_address, skynet.PTYPE_RESPONSE, co_session, f(...)) ~= nil
                    --记得，f()是pack函数
                    ...
                else
                    ret = c.send(co_address, skynet.PTYPE_ERROR, co_session, "") ~= nil
                end
            else
                ret = false
            end
            release_watching(co_address)    --减少ref值 为0则清空
            unresponse[response] = nil
            f = nil
            return ret
        end

可见，skynet.response()的实现充分利用了lua函数闭包的特性，所有相关数据随身携带，自然随心所欲啦。

API研究 skynet.newservice()

之前分析了 snlua 加载了lua代码之后，如何构造运行环境，并且运营coroutine的过程。后面，看一下lua环境下如何引导其他的lua程序。

    function skynet.newservice(name, ...)
        return skynet.call(".launcher", "lua" , "LAUNCH", "snlua", name, ...)
    end

可见这里依赖一个launcher服务，这个服务在哪里呢？ 这肯定是一个早期加载的服务，可以看看 bootstrap.lua

    skynet.start(function()
        ...
        local launcher = assert(skynet.launch("snlua","launcher"))
        skynet.name(".launcher", launcher)
        ...
    )

skynet.launch()的实现在 lualib/skynet/manager.lua中：

    function skynet.launch(...)
        local addr = c.command("LAUNCH", table.concat({...}," "))
        --这里返回十六进制的handle，就是服务 地址了
        if addr then
            return tonumber("0x" .. string.sub(addr , 2))
        end
    end

可见又调用了 lua_skynet.c中的内容，关键函数如下：

    static const char *cmd_launch(struct skynet_context * context, const char * param) {
        size_t sz = strlen(param);
        char tmp[sz+1];
        strcpy(tmp,param);
        char * args = tmp;
        char * mod = strsep(&args, " \t\r\n");
        args = strsep(&args, "\r\n");
        struct skynet_context * inst = skynet_context_new(mod,args);
        if (inst == NULL) {
            return NULL;
        } else {
            id_to_hex(context->result, inst->handle);
            return context->result; //这个重要，返回了服务地址
        }
    }

这里调用了 skynet_context_new(mod,args) 可见这里加载了一个C模块。注意之前命令是 snlua xxxxx，所以和第一篇一样，这里也是先加载 snlua.so之后，通过loader.lua加载目标 launcher.lua文件。

在继续分析下去之前，先总结一下，skynet.launch()可以直接通过snlua加载lua文件——这是独立skynet_context的服务。

回到原先话题，launcher.lua 是在bootstrap.lua中被使用skynet.launch() 加载的。

之后便可以提供服务，skynet.newservice()函数，就通过 向其 发布“LAUNCH”命令来实施。

看看 ".launcher"服务的实现，该launcher.lua文件中，负责这个LAUNCH指令的代码：

    function command.LAUNCH(_, service, ...)
        launch_service(service, ...)
        return NORET
    end
    local function launch_service(service, ...)
        local param = table.concat({...}, " ")
        --可见，又一次调用了skynet.launch()方法，和bootstrap.lua中引导“.launcher”一样。
        local inst = skynet.launch(service, param)
        -- inst返回值是 服务地址数值
        local response = skynet.response()  --记住这个是闭包
        if inst then
            services[inst] = service .. " " .. param
            instance[inst] = response
        else
            response(false)
            return
        end
        return inst
    end

可见，LAUNCH这个命令，并非直接回应，而是制作了闭包挂在instance表中，为啥尼？
原因是，它想等被引导的lua程序确认自己运行正常，再回复。如何实现的？看一下skynet.lua中的两个函数：

    function skynet.start(start_func)
        c.callback(skynet.dispatch_message)
        skynet.timeout(0, function()
            skynet.init_service(start_func)
        end)
    end
    --上面的函数不陌生吧，继续
    function skynet.init_service(start)
        local ok, err = skynet.pcall(start) --谨慎运行用户 start代码
        if not ok then
            skynet.error("init service failed: " .. tostring(err))
            skynet.send(".launcher","lua", "ERROR")
            skynet.exit()
        else
            --看这里，若是成功，则调用.launcher的“LAUNCHOK”命令
            skynet.send(".launcher","lua", "LAUNCHOK")
        end
    end

而在launcher.lua中

    function command.LAUNCHOK(address)
        -- init notice
        local response = instance[address]
        if response then
            response(true, address) --这里就回应很早以前引导我的恩人了
            instance[address] = nil
        end
        return NORET
    end

launch过程分析完毕，总结

初期条件简陋，引导lua代码（例如bootstrap.lua）， 直接采用c语言，用“skynet_context_new()”函数加载snlua，再引导loader.lua，再引导bootstrap.lua。

bootstrap.lua中要去引导launcher.lua，会采用skynet.launch()函数，其调用c语言扩展，也通过“skynet_context_new()”函数实施引导。

而一旦 .launcher 服务运转起来，我们就可以使用 skynet.newservice()来引导应用了。如下：

    function skynet.newservice(name, ...)
        return skynet.call(".launcher", "lua" , "LAUNCH", "snlua", name, ...)
    end

用 .launcher 加载代码可以判断加载运行是否顺利，引导者可以通知launcher引导目标，目标运行起来正常之后，会发消息给.launcher报个平安，.launcher收到后便回报引导者调用成功。

哦，另外，点开头的服务名称，在云风wiki上有所说明：

. 开头的名字是在同一 skynet 节点下有效的，跨节点的 skynet 服务对别的节点下的 . 开头的名字不可见。不同的 skynet 节点可以定义相同的 . 开头的名字。

API研究 skynet.error

错误处理实际是一个大话题，我们看看这个函数的作用和如何被实现。

在分析 simpleweb的时候我看到不少 skynet.error()的使用，追溯一下：

in skynet.lua

    local c = require "skynet.core"
    ...
    skynet.error = c.error

lua-skynet.c中定义了 _error()，

    static int
    _error(lua_State *L) {
        struct skynet_context * context = lua_touserdata(L, lua_upvalueindex(1));
        skynet_error(context, "%s", luaL_checkstring(L,1));
        return 0;
    }

然后在 skynet_error()中做了实现。

    void 
    skynet_error(struct skynet_context * context, const char *msg, ...) {
        ...
            logger = skynet_handle_findname("logger");
        ...
        char tmp[LOG_MESSAGE_SIZE];
        char *data = NULL;
        va_list ap;
        va_start(ap,msg);
        int len = vsnprintf(tmp, LOG_MESSAGE_SIZE, msg, ap);
        va_end(ap);
        ...
            data = skynet_strdup(tmp);
        ...
        struct skynet_message smsg;
        ...
            smsg.source = skynet_context_handle(context);
        ...
        smsg.session = 0;
        smsg.data = data;
        smsg.sz = len | ((size_t)PTYPE_TEXT << MESSAGE_TYPE_SHIFT);
        skynet_context_push(logger, &smsg);
    }

可见，该函数从参数中找出string作为参数，输出错误信息，发送给log模块。

API研究 skynet.fork ，skynet.wait，skynet.wakeup

TBC

logger

TBC

coroutine = require "skynet.coroutine"

TBC

研究 wiki “CriticalSection” 实现

TBC

研究 wiki “DataCenter” 实现

TBC

测试并研究 wiki “DebugConsole” 实现

TBC

研究 wiki “http” 实现

TBC

研究 wiki “multicast” 实现

TBC

研究 wiki “ShareData” 实现

TBC