skynet赏析

make linux MALLOC_STATICLIB= SKYNET_DEFINES=-DNOUSE_JEMALLOC

function skynet.dispatch_message(...)

     -- 当前消息处理
    local succ, err = pcall(raw_dispatch_message,...)
    while true do
     -- 顺序执行skynet.fork 创建的协程 
        if fork_queue.h > fork_queue.t then
            -- queue is empty
            fork_queue.h = 1
            fork_queue.t = 0
            break
        end
        -- pop queue
        local h = fork_queue.h
        local co = fork_queue[h]
        fork_queue[h] = nil
        fork_queue.h = h + 1

        local fork_succ, fork_err = pcall(suspend,co,coroutine_resume(co))
        if not fork_succ then
            if succ then
                succ = false
                err = tostring(fork_err)
            else
                err = tostring(err) .. "\n" .. tostring(fork_err)
            end
        end
    end
    assert(succ, tostring(err))
end

local function raw_dispatch_message(prototype, msg, sz, session, source)
    -- skynet.PTYPE_RESPONSE = 1, read skynet.h
    if prototype == 1 then -- 对回应类型的包处理
        local co = session_id_coroutine[session]
        if co == "BREAK" then
            session_id_coroutine[session] = nil
        elseif co == nil then
            unknown_response(session, source, msg, sz)
        else
            local tag = session_coroutine_tracetag[co]
            if tag then c.trace(tag, "resume") end
            session_id_coroutine[session] = nil
            suspend(co, coroutine_resume(co, true, msg, sz, session))
        end
    else
        local p = proto[prototype] -- 找到对应的解析协议
        if p == nil then
            if prototype == skynet.PTYPE_TRACE then
                -- trace next request
                trace_source[source] = c.tostring(msg,sz)
            elseif session ~= 0 then
                c.send(source, skynet.PTYPE_ERROR, session, "")
            else
                unknown_request(session, source, msg, sz, prototype)
            end
            return
        end

        local f = p.dispatch -- 获取处理的函数
        if f then
            local co = co_create(f) -- 获取协程
            session_coroutine_id[co] = session
            session_coroutine_address[co] = source
            local traceflag = p.trace
            if traceflag == false then
                -- force off
                trace_source[source] = nil
                session_coroutine_tracetag[co] = false
            else
                local tag = trace_source[source]
                if tag then
                    trace_source[source] = nil
                    c.trace(tag, "request")
                    session_coroutine_tracetag[co] = tag
                elseif traceflag then
                    -- set running_thread for trace
                    running_thread = co
                    skynet.trace()
                end
            end
            suspend(co, coroutine_resume(co, session,source, p.unpack(msg,sz)))
        else
            trace_source[source] = nil
            if session ~= 0 then
                c.send(source, skynet.PTYPE_ERROR, session, "")
            else
                unknown_request(session, source, msg, sz, proto[prototype].name)
            end
        end
    end
end

local function co_create(f)
    local co = tremove(coroutine_pool) -- 从协程池中获取协程
    if co == nil then --如果没有了
        co = coroutine_create(function(...) -- 创建新的
            f(...) --执行回调函数，不会立马执行只会调用coroutine.resume时候才会执行
            while true do -- 为了能够复用刚创建的协成，下面需要对协程进行初始化和回收
                local session = session_coroutine_id[co]
                if session and session ~= 0 then
                    local source = debug.getinfo(f,"S")
                    skynet.error(string.format("Maybe forgot response session %s from %s : %s:%d",
                        session,
                        skynet.address(session_coroutine_address[co]),
                        source.source, source.linedefined))
                end
                -- coroutine exit
                local tag = session_coroutine_tracetag[co]
                if tag ~= nil then
                    if tag then c.trace(tag, "end") end
                    session_coroutine_tracetag[co] = nil
                end
                local address = session_coroutine_address[co]
                if address then
                    session_coroutine_id[co] = nil
                    session_coroutine_address[co] = nil
                end

                -- recycle co into pool
                f = nil
                coroutine_pool[#coroutine_pool+1] = co
                -- recv new main function f
                f = coroutine_yield "SUSPEND"
                f(coroutine_yield())
            end
        end)
    else
        -- pass the main function f to coroutine, and restore running thread
        local running = running_thread
        coroutine_resume(co, f)
        running_thread = running
    end
    return co
end

-- suspend is local function
function suspend(co, result, command)
    if not result then -- 执行co失败以后的处理
        local session = session_coroutine_id[co]
        if session then -- coroutine may fork by others (session is nil)
            local addr = session_coroutine_address[co]
            if session ~= 0 then
                -- only call response error
                local tag = session_coroutine_tracetag[co]
                if tag then c.trace(tag, "error") end
                c.send(addr, skynet.PTYPE_ERROR, session, "")
            end
            session_coroutine_id[co] = nil
        end
        session_coroutine_address[co] = nil
        session_coroutine_tracetag[co] = nil
        skynet.fork(function() end) -- trigger command "SUSPEND"
        local tb = traceback(co,tostring(command))
        coroutine.close(co)
        error(tb)
    end
    if command == "SUSPEND" then -- 挂起操作
        return dispatch_wakeup() -- 如果有能够被唤醒的协程，就wakeup
    elseif command == "QUIT" then
        coroutine.close(co)
        -- service exit
        return
    elseif command == "USER" then
        -- See skynet.coutine for detail
        error("Call skynet.coroutine.yield out of skynet.coroutine.resume\n" .. traceback(co))
    elseif command == nil then
        -- debug trace
        return
    else
        error("Unknown command : " .. command .. "\n" .. traceback(co))
    end
end				

cluster.call(node, address, ...) --远程调用node中的addr
cluster.send(node, address, ...) --send调用远程node的addr
cluster.open(port) --本地打开(监听)一个cluster结点，使其能在cluster中的其他结点发现
cluster.reload(config) --重载远程结点配置表，表中的cluster结点都open过，则可以通讯
cluster.proxy(node, name) --设置远程结点的代理，使得可以像调用本地RPC一样调用远程结点
cluster.snax(node, name, address) --生成一个远程的snax服务对象
cluster.register(name, addr) --注册一个cluster结点
cluster.query(node, name) --查找远程结点中注册过的结点是否存在

harbor.link(id) --用来监控一个 slave 是否断开。如果 harbor id 对应的 slave 正常，这个 api 将阻塞。当 slave 断开时，会立刻返回。
harbor.linkmaster() --用来在 slave 上监控和 master 的连接是否正常。这个 api 多用于异常时的安全退出（因为当 slave 和 master 断开后，没有手段可以恢复）。
harbor.connect(id) --和 harbor.link 相反。如果 harbor id 对应的 slave 没有连接，这个 api 将阻塞，一直到它连上来才返回。
harbor.queryname(name) --可以用来查询全局名字或本地名字对应的服务地址。它是一个阻塞调用。
harbor.globalname(name, handle) --注册一个全局名字。如果 handle 为空，则注册自己。skynet.name 和 skynet.register 是用其实现的。

skynet.getenv(varName) --conf配置信息已经写入到注册表中，通过该函数获取注册表的变量值
skynet.setenv(varName, varValue) --设置注册表信息，varValue一般是number或string，但是不能设置已经存在的varname
skynet.error(...) --打印函数
skynet.start(func) --用 func 函数初始化服务，并将消息处理函数注册到 C 层，让该服务可以工作。
skynet.init(func) --若服务尚未初始化完成，则注册一个函数等服务初始化阶段再执行；若服务已经初始化完成，则立刻运行该函数。
skynet.exit() --结束当前服务
skynet.self() --获取当前服务的句柄handler
skynet.address(handler) --将handle转换成字符串
skynet.abort() --退出skynet进程
skynet.kill(address) ----强制杀死其他服务。可以用来强制关闭别的服务。但强烈不推荐这样做。因为对象会在任意一条消息处理完毕后，毫无征兆的退出。所以推荐的做法是，发送一条消息，让对方自己善后以及调用 skynet.exit 。注：skynet.kill(skynet.self()) 不完全等价于 skynet.exit() ，后者更安全。			

skynet.newservice(luaServerName, ...)		

skynet.uniqueservice(servicename, ...) --当前的skynet节点全局唯一
skynet.uniqueservice(true, servicename, ...) --所有的节点全局唯一

skynet.queryservice(servicename, ...) --当前的skynet节点中查找
skynet.queryservice(true, servicename, ...) --所有的节点中查找

skynet.register(aliasname) --给当前服务定一个别名，可以是全局别名，也可以是本地别名
skynet.name(aliasname, servicehandler) --给指定servicehandler的服务定一个别名，可以是全局别名，也可以是本地别名

--[[
查询别名为aliasname的服务,可以是全局别名也可以是本地别名，
1、当查询本地别名时，返回servicehandler，不存在就返回nil
2、当查询全局别名时，返回servicehandler，不存在就阻塞等待到该服务初始化完成
]]--
skynet.harbor.queryname(aliasname)
skynet.localname(aliasname) --查询本地别名为aliasname的服务，返回servicehandler，不存在就返回nil
skynet.kill(handle) --杀死带别名服务

skynet.sleep(time) --让当前的任务等待 time * 0.01s 。
skynet.fork(func, ...) --启动一个新的任务去执行函数 func , 其实就是开了一个协程，函数调用完成将返回线程句柄 虽然你也可以使用原生的coroutine.create来创建协程，但是会打乱skynet的工作流程
skynet.yield() --让出当前的任务执行流程，使本服务内其它任务有机会执行，随后会继续运行。
skynet.wait() --让出当前的任务执行流程，直到用 wakeup 唤醒它。
skynet.wakeup(co) --唤醒用 wait 或 sleep 处于等待状态的任务。    
skynet.timeout(time, func) --设定一个定时触发函数 func ，在 time * 0.01s 后触发。
skynet.starttime() --返回当前进程的启动 UTC 时间（秒）。
skynet.now() --返回当前进程启动后经过的时间 (0.01 秒) 。
skynet.time() --通过 starttime 和 now 计算出当前 UTC 时间（秒）。		

#define PTYPE_TEXT 0        --文本
#define PTYPE_RESPONSE 1    --表示一个回应包
#define PTYPE_MULTICAST 2   --广播消息
#define PTYPE_CLIENT 3      --用来处理网络客户端的请求消息
#define PTYPE_SYSTEM 4      --系统消息
#define PTYPE_HARBOR 5      --集群内其他的 skynet 节点发来的消息
#define PTYPE_SOCKET 6    --套接字消息
#define PTYPE_ERROR 7     --错误消息，一般服务退出的时候会发送error消息给关联的服务
#define PTYPE_QUEUE 8     --队列方式
#define PTYPE_DEBUG 9     --调试
#define PTYPE_LUA 10   --lua类型的消息，最常用
#define PTYPE_SNAX 11  --snax服务消息

#define PTYPE_TAG_DONTCOPY 0x10000 --禁止拷贝
#define PTYPE_TAG_ALLOCSESSION 0x20000 --分配新的 session

skynet.pack(...) --打包
skynet.unpack(msg, sz) --解包		

 -- 发送无需响应的消息
 skynet.send(addr, type, ...)  --用 type 类型向 addr 发送未打包的消息。该函数会自动把...参数列表进行打包，默认情况下lua消息使用skynet.pack打包。addr可以是服务句柄也可以是别名。自动打包与解包。）
 skynet.rawsend(addr, type, msg, sz)  --用 type 类型向 addr 发送一个打包好的消息。addr可以是服务句柄也可以是别名。（需要自己打包与解包）


 -- 发送必须响应的消息
skynet.call(addr, type, ...)  --用默认函数打包消息，向addr发送type类型的消息并等待返回响应，并对回应信息进行解包。（自动打包与解包。）
skynet.rawcall(addr, type, msg, sz) --直接向addr发送type类型的msg,sz并等待返回响应，不对回应信息解包。（需要自己打包与解包）

-- 同一个协成处理
skynet.ret() --目标服务消息处理后需要通过该函数将结果返回
skynet.retpack(...) --将消息用skynet.pack 打包，并调用 ret 回应。

--不在一个协成处理
local response = skynet.response(pack)--参数pack指定应答打包函数，不填默认使用skynet.pack, 必须根据接收到消息的打包函数一致 返回值是一个闭包函数
response(ok, ...) --参数ok的值可以是 "test"、true、false，为"test"时表示检查接收响应的服务是否存在，为true时表示发送应答PTYPE_RESPONSE，为false时表示发送PTYPE_ERROR错误消息。

skynet.queue() --帮助你回避这些服务重入或者伪并发引起的复杂性,但是明显降低了服务的并发处理能力，所以使用执行队列的时候尽量缩小临界区的颗粒度大小

skynet.forward_type() --需要提供一张消息转换映射表forward_map, 其他的方法与skynet.start一样

skynet.redirect(dest,source,typename, session, msg, sz) --使用source服务地址，发送typename类型的消息给dest服务，不需要接收响应，（source，dest只能是服务ID）msg sz一般使用skynet.pack打包生成

skynet.multicast -- 当组播的数据量较大时候可以节省内部的带宽

--建立一个 TCP 连接。返回一个数字 id 。
socket.open(address, port)      
--关闭一个连接，这个 API 有可能阻塞住执行流。因为如果有其它 coroutine
--正在阻塞读这个 id 对应的连接，会先驱使读操作结束，close 操作才返回。
socket.close(id)
--在极其罕见的情况下，需要粗暴的直接关闭某个连接，而避免 socket.close 的阻塞等待流程，可以使用它。
socket.close_fd(id)
--强行关闭一个连接。和 close 不同的是，它不会等待可能存在的其它 coroutine 的读操作。
--一般不建议使用这个 API ，但如果你需要在 __gc 元方法中关闭连接的话，
--shutdown 是一个比 close 更好的选择（因为在 gc 过程中无法切换 coroutine）。与close_fd类似
socket.shutdown(id)
--[[
    从一个 socket 上读 sz 指定的字节数。
    如果读到了指定长度的字符串，它把这个字符串返回。
    如果连接断开导致字节数不够，将返回一个 false 加上读到的字符串。
    如果 sz 为 nil ，则返回尽可能多的字节数，但至少读一个字节（若无新数据，会阻塞）。
--]]
socket.read(id, sz)
--从一个 socket 上读所有的数据，直到 socket 主动断开，或在其它 coroutine 用 socket.close 关闭它。
socket.readall(id)
--从一个 socket 上读一行数据。sep 指行分割符。默认的 sep 为 "\n"。读到的字符串是不包含这个分割符的。
--如果另外一端就关闭了，那么这个时候会返回一个nil，如果buffer中有未读数据则作为第二个返回值返回。
socket.readline(id, sep)
--等待一个 socket 可读。
socket.block(id)
--把一个字符串置入正常的写队列，skynet 框架会在 socket 可写时发送它。
socket.write(id, str)
--把字符串写入低优先级队列。如果正常的写队列还有写操作未完成时，低优先级队列上的数据永远不会被发出。
--只有在正常写队列为空时，才会处理低优先级队列。但是，每次写的字符串都可以看成原子操作。
--不会只发送一半，然后转去发送正常写队列的数据。
socket.lwrite(id, str)
--监听一个端口，返回一个 id ，供 start 使用。
socket.listen(address, port)
--[[
    accept 是一个函数。每当一个监听的 id 对应的 socket 上有连接接入的时候，都会调用 accept 函数。
这个函数会得到接入连接的 id 以及 ip 地址。你可以做后续操作。
    每当 accept 函数获得一个新的 socket id 后，并不会立即收到这个 socket 上的数据。
这是因为，我们有时会希望把这个 socket 的操作权转让给别的服务去处理。accept(id, addr)
]]--
socket.start(id , accept)
--[[
    任何一个服务只有在调用 socket.start(id) 之后，才可以读到这个 socket 上的数据。
向一个 socket id 写数据也需要先调用 start 。
    socket 的 id 对于整个 skynet 节点都是公开的。也就是说，你可以把 id 这个数字
通过消息发送给其它服务，其他服务也可以去操作它。skynet 框架是根据调用 start 这个
api 的位置来决定把对应 socket 上的数据转发到哪里去的。
--]]
socket.start(id)
--清除 socket id 在本服务内的数据结构，但并不关闭这个 socket 。
--这可以用于你把 id 发送给其它服务，以转交 socket 的控制权。
socket.abandon(id)
--[[
    当 id 对应的 socket 上待发的数据超过 1M 字节后，系统将回调 callback 以示警告。
function callback(id, size) 回调函数接收两个参数 id 和 size ，size 的单位是 K 。
    如果你不设回调，那么将每增加 64K 利用 skynet.error 写一行错误信息。
--]]
socket.warning(id, callback)

用来支持双向传输，异步非堵塞处理数据

skynet.dns --调用了系统 api getaddrinfo ，有可能阻塞住整个 socket 线程 所以skynet封装了这个接口来解决dns查询时候造成的线程堵塞问题

https://blog.csdn.net/gneveek/article/details/78940693