游戏服务器框架skynet源码分析--协程调度模型

超、凢脫俗 2022-10-09 00:35 152阅读 0赞

linux服务器开发相关视频解析：

[聊聊linux服务器端高级架构—云风的skynet][linux_skynet]

[协程如何使用，与线程有什么区别？][Link 1]

c/c++ linux服务器开发学习地址：[c/c++ linux后台服务器高级架构师][c_c_ linux]

为方便理解，本文贴出的代码部分经过了缩减或展开，与实际skynet代码可能会有所出入。

作为一个skynet actor，在启动脚本被加载的过程中，总是要调用skynet.start和skynet.dispatch的，前者在skynet-os中做一些初始化工作，设置消息的Lua回调，后者则注册针对某协议的解析回调。举个例子：

local skynet = require "skynet"
     
     local function hello()
         skynet.ret(skynet.pack("Hello, World!"))
         print("hello OK")
     end
    
     skynet.start(function()
        skynet.dispatch("lua", function(session, address, cmd, ...)
            assert(cmd == "hello")
            hello()
       end)
    end)

先是调用skynet.start注册初始化回调，在其中调用skynet.dispatch注册针对"lua"协议的解析回调。skynet的基本使用这里我们就不多说了，具体见官方文档。下面我们就从skynet.start（见skynet.lua）开始，逐一分析流程。

function skynet.start(start_func)
         c.callback(skynet.dispatch_message)
         skynet.timeout(0, function()
             skynet.init_service(start_func)
         end)
     end

这里的c来自于require “skynet.core”，它是在lua-skynet.c中注册的，如下：

int luaopen_skynet_core(lua_State *L) { 
         luaL_checkversion(L);
     
         luaL_Reg l[] = { 
             ...
             {  "callback", _callback },
             {  NULL, NULL },
         };
     
         luaL_newlibtable(L, l);
     
         lua_getfield(L, LUA_REGISTRYINDEX, "skynet_context");
         struct skynet_context *ctx = lua_touserdata(L,-1);
         if (ctx == NULL) { 
             return luaL_error(L, "Init skynet context first");
         }
     
         luaL_setfuncs(L,l,1);
    
         return 1;
     }

可以看到，它注册了几个函数，并将skynet\_context实例作为各函数的upvalue，方便调用时获取。skynet.start中调用c.callback，对应的就是lua-skynet.c中的\_callback函数，skynet.dispatch\_message回调就是它的参数：

static int _callback(lua_State *L) { 
         struct skynet_context * context = lua_touserdata(L, lua_upvalueindex(1));
         int forward = lua_toboolean(L, 2);
         luaL_checktype(L,1,LUA_TFUNCTION);
         lua_settop(L,1);
         lua_rawsetp(L, LUA_REGISTRYINDEX, _cb);
     
         lua_rawgeti(L, LUA_REGISTRYINDEX, LUA_RIDX_MAINTHREAD);
         lua_State *gL = lua_tothread(L,-1);
     
         if (forward) { 
             skynet_callback(context, gL, forward_cb);
         } else { 
             skynet_callback(context, gL, _cb);
         }
     
         return 0;
     }

可以看到，其以函数\_cb为key，LUA回调（skynet.dispatch\_message）作为value被注册到全局注册表中。skynet\_callback(在skynet\_server.c中)则设置函数指针\_cb为C层面的消息处理函数：

void skynet_callback(struct skynet_context * context, void *ud, skynet_cb cb) { 
         context->cb = cb;
         context->cb_ud = ud;
     }

【文章福利】  
skynet训练营微课开营啦！！！  
游戏开发大牛带大家2天-4小时搞定skynet框架的核心技术点！

入营地址：[游戏开发核心技术点skynet训练营][skynet]

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwOTg5NzY5_size_16_color_FFFFFF_t_70_pic_center]

先不关注skynet-os内部的线程调度细节，只需要知道，skynet-context接收到消息后会转发给context->cb处理，也就是\_cb函数。在\_cb中，从全局表中取到关联的LUA回调，将type, msg, sz, session, source压栈调用:

static int _cb(struct skynet_context * context, void * ud, int type, int session, uint32_t source, const void * msg, size_t sz) { 
          lua_State *L = ud;
         int trace = 1;
         int r;
         int top = lua_gettop(L);
         if (top == 0) { 
            lua_pushcfunction(L, traceback);
             lua_rawgetp(L, LUA_REGISTRYINDEX, _cb);
         } else { 
             assert(top == 2);
         }
        lua_pushvalue(L,2);
    
         lua_pushinteger(L, type);
        lua_pushlightuserdata(L, (void *)msg);
         lua_pushinteger(L,sz);
         lua_pushinteger(L, session);
         lua_pushinteger(L, source);
     
         r = lua_pcall(L, 5, 0 , trace);
     
        if (r == LUA_OK) { 
            return 0;
        }
     }

此时调用流程正式转到skynet.lua中的skynet.dispatch\_message：

function skynet.dispatch_message(...)
         local succ, err = pcall(raw_dispatch_message,...)
         while true do
             local key,co = next(fork_queue)
             fork_queue[key] = nil
             pcall(suspend,co,coroutine_resume(co))
         end
     end

首先是将msg交由raw\_dispatch\_message作分发，然后开始处理fork\_queue中缓存的fork协程：

pcall(suspend, co, coroutine_resume(co))

这行代码是我们今天关注的重点。在继续之前，我假设你对lua的协程有一定的了解，了解coroutine.resume，coroutine.yield的基本用法。coroutine就是lua里的线程，它拥有自己的函数栈，但与我们平常接触的大多数操作系统里的线程不同，是非抢占式的。skynet对lua的coroutine作了封装（详见lua-profile.c），主要是增加了starttime和totaltime的监测，最终还是交由lua的coroutine库来处理的。既然这里分析到了fork\_queue，那我们就先以skynet.fork为例，看看它作了什么：

function skynet.fork(func,...)
         local args = table.pack(...)
         local co = co_create(function()
             func(table.unpack(args,1,args.n))
         end)
         table.insert(fork_queue, co)
         return co
     end

skynet.fork做的事情很简单，通过co\_create创建一个coroutine并将其入队fork\_queue。看看co\_create是如何创建协程的：

local function co_create(f)
          local co = table.remove(coroutine_pool)
         if co == nil then
              co = coroutine.create(function(...)
                  f(...)
                 while true do
                      f = nil
                      coroutine_pool[#coroutine_pool+1] = co
                      f = coroutine_yield "EXIT"
                     f(coroutine_yield())
                 end
             end)
         else
             coroutine_resume(co, f)
         end
         return co
     end

调用co\_create时，如果coroutine\_pool为空，它会创建一个新的co。co在第一次被resume时，会执行f，接着便进入一个使用和回收的无限循环。在这个循环中，先是收回co到coroutine\_pool中，接着便yield "EXIT"到上一次的resume点A。当下一次被resume在点B唤醒时，会先将函数f传递过来，接着再次yield到点B，等待下一次在点D被resume唤醒时，传递需要的参数过来加以执行，完毕后回收，如此反复。这样看来，co的执行似乎相当简单。但是实际上要复杂一些，因为在执行f的过程中，可以再反复地yield和resume。下面我们举个简单的例子：

skynet.fork(function()
             print ("skynet fork: <1>")
             skynet.sleep(10)
             print ("skynet fork: <2>")
         end)

我们把skynet.sleep展开：

skynet.fork(function()
             print ("skynet fork: <1>")
             coroutine_yield("SLEEP", COMMAND_TIMEOUT(10))
             print ("skynet fork: <2>")
         end)

下面开始分析调用流程。fork-co入队，主co在skynet.dispatch\_message中分发消息后取出fork-co，调用resume开始进入fork-co的函数f执行，如下图所示，如果fork-co是第一次执行，是走圈1，如果是复用，则走圈1\*（如果是复用的话，调用co\_create时，会先coroutine\_resume(co,f)一次进入fork-co，将用户函数传递给while循环中的coroutine\_yield "EXIT"点之后，接着fork-co再次yield让出，等待实际传参的调用）。接着进入用户函数，COMMAND\_TIMEOUT会先向skynet-kernal发送TIMEOUT命令，如圈2所示。然后yield "SLEEP"到主co的resume点1之后继续执行，如圈3所示，按圈4的指向，调用suspend进入"SLEEP"分支，记录下TIMEOUT-session与fork-co的映射关系。此时主co回到skynet.dispatch\_message中继续下一个fork-co的处理。当TIMEOUT消息回来时，会由主co再次进入skynet.dispatch\_message并调用raw\_dispatch\_message分发，这时通过session拿到之前映射的fork-co，再次resume，按照圈5的指向，会跳转到fork-co的yield "SLEEP"点之后继续向下处理。用户函数处理完毕后，回到上层调用，即圈6所指，回收fork-co，接着yield "EXIT"到主co所在raw\_dispatch\_message中的resume点之后，如圈7所示。进入suspend后，无额外命令，raw\_dispatch\_message处理结束，继续主co的消息处理流程。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwOTg5NzY5_size_16_color_FFFFFF_t_70]  
由以上分析可以看到，实际的协程跳转过程是比较复杂的，也更显得小小的LUA在skynet中的精巧运用。为方便理解，顺便贴出suspend的代码（只列出了我们关注的几个命令，并做了删减）：

function suspend(co, result, command, param, size)
          if command == "CALL" then
              session_id_coroutine[param] = co
          elseif command == "SLEEP" then
              session_id_coroutine[param] = co
              sleep_session[co] = param
          elseif command == "RETURN" then
              local co_session = session_coroutine_id[co]
              local co_address = session_coroutine_address[co]
             session_response[co] = true
             c.send(co_address, skynet.PTYPE_RESPONSE, co_session, param, size)
             return suspend(co, coroutine_resume(co, ret))
         elseif command == "EXIT" then
             -- coroutine exit
             local address = session_coroutine_address[co]
             session_coroutine_id[co] = nil
             session_coroutine_address[co] = nil
             session_response[co] = nil
         elseif command == nil then
             -- debug trace
             return
         end
     end

看完skynet.fork，我们再回过头来，看一看skynet.dispatch\_message中消息分发raw\_dispatch\_message的具体细节：

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]
             session_id_coroutine[session] = nil
              suspend(co, coroutine_resume(co, true, msg, sz))
          else
              local p = proto[prototype]
              local f = p.dispatch
             local co = co_create(f)
             session_coroutine_id[co] = session
             session_coroutine_address[co] = source
             suspend(co, coroutine_resume(co, session,source, p.unpack(msg,sz)))
         end
     end

RESPONSE的处理先就不说了，在叙述skynet.sleep时已经有所讨论。消息会根据它的prototype查找proto，接着调用co\_create取得协程user-co，将message解包后resume到user-co进入用户函数f。而后续的流程则与上文我们讨论的fork是一样的了。比如，在f中调用skynet.call时，会向目标发送消息，接着会yield到主co，回到这里的resume点，接着进入suspend的"CALL"分支，记录session与user-co的映射关系。下一次response消息回来时，会查找到user-co并resume唤醒，在skynet.call后继续执行，用户函数f结束后进入上层调用，回收user-co，等待新的调用。

[linux_skynet]: https://www.bilibili.com/video/BV1Nv411B7Q7/
[Link 1]: https://www.bilibili.com/video/BV1CV411d7uk/
[c_c_ linux]: https://ke.qq.com/course/417774?flowToken=1013300
[skynet]: https://ke.qq.com/course/2806743?flowToken=1025549
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwOTg5NzY5_size_16_color_FFFFFF_t_70_pic_center]: /images/20221005/f27f387b69dc42e8b1291ac92ba9131d.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQwOTg5NzY5_size_16_color_FFFFFF_t_70]: /images/20221005/0437f1ccc8d8417bb97c0d58466749df.png