skynet1.0阅读笔记2_skynet的消息投递skynet.call

小程序：扫一扫查出行
【扫一扫了解最新限行尾号】
复制小程序

为了了解 skynet.call 的调用过程，需要先看看 skynet的队列是如何把包分到不同工作线程的。看下图

查看 global_queue 的skynet_globalmq_push和skynet_globamq_pop,很容易可以找到两个关键的函数:

skyent_context_push

和

skynet_context_message_dispatch

先来看出口,skynet_context_message_dispatch。在skynet的启动函数中，我们已经知道skynet_start里面的start(config->thread)启动了 worker等线程:

thread_worker(void *p) {
//初始化
...
struct message_queue * q = NULL;
while (!m->quit) {
//循环调用 skynet_context_message_dispatch
q = skynet_context_message_dispatch(sm, q, weight);
if (q == NULL) {
//没包了就挂其线程
...
}
}
return NULL;
}

很清晰的代码，worker线程不断调用 skynet_context_message_dispatch 来读取q里面的skynet_message 队列，并进行分发。我们来看它是怎么分发的。

struct message_queue *
skynet_context_message_dispatch(struct skynet_monitor *sm, struct message_queue *q, int weight) {
//q为空时，重新从global_queue中取下一个 message_queue
if (q == NULL) {
q = skynet_globalmq_pop();
if (q==NULL)
return NULL;
} //当前 message_queue 所属服务的 context 的handle id
uint32_t handle = skynet_mq_handle(q); struct skynet_context * ctx = skynet_handle_grab(handle);
if (ctx == NULL) {
struct drop_t d = { handle };
skynet_mq_release(q, drop_message, &d);
return skynet_globalmq_pop();
} int i,n=;
struct skynet_message msg; for (i=;i<n;i++) {
//从message_queue 中 pop一个msg出来
if (skynet_mq_pop(q,&msg)) {
//若message_queue为空,返回1表示失败,释放ctx的引用次数
skynet_context_release(ctx);
//把返回global_queue里面的下一个message_queue,以供skynet_context_message_dispatch调用
return skynet_globalmq_pop();
} else if (i== && weight >= ) {
n = skynet_mq_length(q);
n >>= weight;
}
int overload = skynet_mq_overload(q);
if (overload) {
skynet_error(ctx, "May overload, message queue length = %d", overload);
} skynet_monitor_trigger(sm, msg.source , handle); //若 ctx->cb不为空，使用dispatch_message调用 ctx->cb
if (ctx->cb == NULL) {
skynet_free(msg.data);
} else {
dispatch_message(ctx, &msg);
} skynet_monitor_trigger(sm, ,);
} assert(q == ctx->queue);
//若global_queue中还有下一个message_queue，返回下一个message_queue供分发，若为空则继续执行当前message_queue的请求
struct message_queue *nq = skynet_globalmq_pop();
if (nq) {
// If global mq is not empty , push q back, and return next queue (nq)
// Else (global mq is empty or block, don't push q back, and return q again (for next dispatch)
skynet_globalmq_push(q);
q = nq;
}
skynet_context_release(ctx); return q;
}

那么，从全局队列最终拿到的 skynet_message包，最后交由了 dispatch_message和ctx-cb来处理了。dispatch_message把msg里面的东西取出来后，调用ctx->cb来进行处理

static void dispatch_message(struct skynet_context *ctx, struct skynet_message *msg) {
...
if (!ctx->cb(ctx, ctx->cb_ud, type, msg->session, msg->source, msg->data, sz)) {
skynet_free(msg->data);
}
CHECKCALLING_END(ctx)
}

在skynet的启动笔记中，已经知道了，首先是：
snlua_init 用 skynet_callback(ctx,l,_launch) 把 ctx->cb注册为 _launch
然后立马投递第一个消息
消息重新进到 dispatch_message,调用 _launch ，把 ctx->cb注册为 skynet.dispatch_message

启动完成后，以后的所有消息其实都进到了 skynet.dispatch_message,然后调用了 raw_dispatch_message

function skynet.dispatch_message(...)
local succ, err = pcall(raw_dispatch_message,...)
...
end

那接着来看 raw_dispatch_message

local function raw_dispatch_message(prototype, msg, sz, session, source, ...)
if prototype == then
...
else
local p = proto[prototype]
...
local f = p.dispatch if f then
...
local co = co_create(f)
...
suspend(co, coroutine.resume(co, session,source, p.unpack(msg,sz, ...)))
end
end
end

接下来要找的是 proto[prototype].disproto所有位置，找出那里定义 dispatch
的。看到这个函数：

function skynet.register_protocol(class)
local name = class.name
local id = class.id
...
proto[name] = class
proto[id] = class
end

以及

function skynet.dispatch(typename, func)
local p = proto[typename]
if func then
local ret = p.dispatch
p.dispatch = func
return ret
else
return p and p.dispatch
end
end

没错了，就是class.dispatch 。所有的消息，最终进到的就是 class.dispatch。

///////////////////////////////////////////////////////////////////////
skynet.regiser_protocol 和 skynet.dispatch 你会在lua服务中经常看见。以launcher.lua为例子

launcher.lua在启动时，注册一个 name为"text"的table,它的dispatch也定义在下面
所以你应该能看到 skynet.call(".launcher","text",...)这种调用

skynet.register_protocol {
name = "text",
id = skynet.PTYPE_TEXT,
unpack = skynet.tostring,
dispatch = function(session, address , cmd)
if cmd == "" then
command.LAUNCHOK(address)
elseif cmd == "ERROR" then
command.ERROR(address)
else
error ("Invalid text command " .. cmd)
end
end,
} //定义 launcher服务的 proto["lua"] 的dispatch
skynet.dispatch("lua", function(session, address, cmd , ...)
cmd = string.upper(cmd)
local f = command[cmd]
if f then
local ret = f(address, ...)
if ret ~= NORET then
skynet.ret(skynet.pack(ret))
end
else
skynet.ret(skynet.pack {"Unknown command"} )
end
end)

但这里还有一个问题，上面的proto["lua"] 是谁注册的呢？查找skynet.register_protocol,我们能找到这个位置:

--skynet.lua
----- register protocol
do
local REG = skynet.register_protocol REG {
name = "lua",
id = skynet.PTYPE_LUA,
pack = skynet.pack,
unpack = skynet.unpack,
} REG {
name = "response",
id = skynet.PTYPE_RESPONSE,
} REG {
name = "error",
id = skynet.PTYPE_ERROR,
unpack = function(...) return ... end,
dispatch = _error_dispatch,
}
end

在你第一次require "skynet"
的时候，它已经默认帮你注册了"lua","response","error"3种消息，然后你创建新的lua服务时，调用skynet.dispatch 为 proto["lua"] 指定dispatch，之后通过 skynet.call("服务名","lua",...) 调用的消息就能最终投递到你定义的处理函数里面了。

到了这里，从队列取出数据，并分发到指定处理函数dispath的完整流程我们以及看到了。接下来，我们来看消息是如果放入global_queue的。

来看 skynet.call 函数(skynet.send其实也一样的，只是它不管返回)

function skynet.call(addr, typename, ...)
//如proto["lua"] ,消息类型id放入msg中
local p = proto[typename]
local session = c.send(addr, p.id , nil , p.pack(...))
...
//等待返回
return p.unpack(yield_call(addr, session))
end

这里的 c.send 的调用，我们看一下 c 的定义：
local c = require "skynet.core"
这里的 skynet.core ,实际上调用的是 skynet.so ,而从 skynet 的makelog我们可以看到这样一行：

cc -g -O2 -Wall -I3rd/lua -fPIC --shared lualib-src/lua-skynet.c lualib-src/lua-seri.c -o luaclib/skynet.so -Iskynet-src -Iservice-src -Ilualib-src

在 lualib-src/lua-skynet.c 中，我们看到这段代码:

luaL_Reg l[] = {
{ "send" , _send },
{ "genid", _genid },
{ "redirect", _redirect },
{ "command" , _command },
{ "intcommand", _intcommand },
{ "error", _error },
{ "tostring", _tostring },
{ "harbor", _harbor },
{ "pack", _luaseri_pack },
{ "unpack", _luaseri_unpack },
{ "packstring", lpackstring },
{ "trash" , ltrash },
{ "callback", _callback },
{ NULL, NULL },
};

这里的 luaL_Reg 把c函数注册到lua中，从而让lua调用这些函数。
所以 c.send 调用的，就是这里的 _send
_send 调用了 skynet_send ,如果目标在当前进程，将调用 skynet_context_push
然后 skyent_context_push 调用
skynet_mq_push(ctx->queue, message);
把消息放如了全局队列,最后来看看 skynet_mq_push ：

void
skynet_mq_push(struct message_queue *q, struct skynet_message *message) {
assert(message);
SPIN_LOCK(q) //把msg放到队列尾，然后 ++ q->taiskynet_globalmq_pushl
q->queue[q->tail] = *message;
if (++ q->tail >= q->cap) {
q->tail = ;
} if (q->head == q->tail) {
expand_queue(q);
} //若ctx->queue未放入global_queue，放进去
if (q->in_global == ) {
q->in_global = MQ_IN_GLOBAL;
skynet_globalmq_push(q);
} SPIN_UNLOCK(q)
}

值得一提的是，取消息从 ctx->queue 的head开始取，push消息则是从 tail push。
所以先投递的消息会先执行，但由于协程的原因，还是不能保证先投递的消息先执行完。