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

Posted linux大本营

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了游戏服务器框架skynet源码分析--协程调度模型相关的知识,希望对你有一定的参考价值。

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

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

协程如何使用,与线程有什么区别?

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-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的消息处理流程。
在这里插入图片描述
由以上分析可以看到,实际的协程跳转过程是比较复杂的,也更显得小小的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,等待新的调用。

以上是关于游戏服务器框架skynet源码分析--协程调度模型的主要内容,如果未能解决你的问题,请参考以下文章

游戏开发实战教你Unity通过sproto协议与Skynet框架的服务端通信,附工程源码(Unity | Sproto | 协议 | Skynet)

三个问题带你看懂多核并发框架skynet源码

使用skynet 的一个简单范例,附分析

不同服务器之间的数据同步(skynet 和 c++)

游戏开发实战手把手教你从零跑一个Skynet,详细教程,含案例讲解(服务端 | Skynet | Ubuntu)

游戏开发实战手把手教你从零跑一个Skynet,详细教程,含案例讲解(服务端 | Skynet | Ubuntu)