skynet monitor线程的作用

1. 前言

使用skynet开发时，可能会碰到如下警告：

[:00000000] A message from [ :00000000 ] to [ :0000000a ] maybe in an endless loop (version = 53)

它表示某处代码执行时间过长，可能陷入死循环。

2. 作用

monitor 线程的主要作用是监控系统中的其他线程（如 worker 线程）是否出现卡死或阻塞情况。它通过检测消息队列的状态来确保系统运行的稳定性。如果发现消息队列堵塞，monitor 线程会及时发出警报，以便采取相应的措施来避免系统崩溃或性能下降。这种监控机制对于维持复杂系统的稳定性和可靠性非常重要。

3. 原理

创建worker线程时，从skynet_start.c 中的 start(…) 函数就可以看出， 会为每一个工作线程分配skynet_monitor 的数据结构。

//skynet_monitor.c
struct skynet_monitor {ATOM_INT version;  int check_version; uint32_t source; uint32_t destination; 
};

● version：当前版本
● check_version：检查版本
● source：消息源服务
● destination：消息目标服务

static void
start(int thread) {...struct monitor *m = skynet_malloc(sizeof(*m)); //创建共享monitorm->m = skynet_malloc(thread * sizeof(struct skynet_monitor *)); //为每个工作线程创建一个存储检测信息的结构体int i;for (i=0;i<thread;i++) {m->m[i] = skynet_monitor_new(); //为每个工作线程分配一块检测信息内存}...
}

● 首先，skynet是以消息驱动的，整个skynet的的处理逻辑为多个工作线程不断从消息队列取出消息，然后派发给指定的服务处理，处理完毕后继续处理下一个消息。
● 派发消息的核心函数是skynet_context_message_dispatch，在这个函数中每次真正派发消息之前和之后，都会调用skynet_monitor_trigger函数。
当工作线程处理消息时，会将skynet_monitor 数据结构传入到 skynet_context_message_dispatch(sm, …)。worker线程开始消费次级队列中的消息时，会先通过 skynet_monitor_trigger(…)设置 version（自增id）， 当monitor每5s轮训所有的工作线程，调用 skynet_monitor_check(…)，第一次会将check_version 设置为 version，当下一个5s检测发现 check_version == version 且该消息的发送方为真时（定时器session为0），就会使用 skynet_error(…) 抛出一个错误消息。告知程序可能出现耗时过长的业务或陷入死循环。如果在下一个5s到达前，业务已经处理完，会使用 skynet_monitor_trigger(sm, 0,0) 将版本号重置。

//skynet_start.c
static void *
thread_worker(void *p) {struct worker_parm *wp = p; //当前工作线程参数int id = wp->id; //当前工作线程idint weight = wp->weight;struct monitor *m = wp->m;struct skynet_monitor *sm = m->m[id];skynet_initthread(THREAD_WORKER); //初始化该线程对应的私有数据块struct message_queue * q = NULL;while (!m->quit) {q = skynet_context_message_dispatch(sm, q, weight);  //数据分发...}return NULL;
}

//skynet_server.c
struct message_queue * 
skynet_context_message_dispatch(struct skynet_monitor *sm, struct message_queue *q, int weight) {...skynet_monitor_trigger(sm, msg.source , handle);if (ctx->cb == NULL) {skynet_free(msg.data);} else {dispatch_message(ctx, &msg); //执行业务}skynet_monitor_trigger(sm, 0,0);...
}

检测worker线程代码如下：

//skynet_start.c
static void *
thread_monitor(void *p) {struct monitor * m = p;int i;int n = m->count; //工作线程的数量skynet_initthread(THREAD_MONITOR);for (;;) {CHECK_ABORTfor (i=0;i<n;i++) {skynet_monitor_check(m->m[i]); //m->m[i]是为每个工作线程分配的监测信息}for (i=0;i<5;i++) {CHECK_ABORTsleep(1);}}return NULL;
}

每5s对所有工作线程进行一次检测，调用skynet_monitor_check函数检测线程是否有卡住在某条消息处理。

//skynet_monitor.c
//监测某个工作线程是否卡主在某一条消息（死循环或者执行时间过长）
void 
skynet_monitor_check(struct skynet_monitor *sm) {if (sm->version == sm->check_version) { //当前处理的消息和已经监测到的消息是否相等if (sm->destination) { //判断消息是否已经被处理skynet_context_endless(sm->destination); //标记相应的服务陷入死循环skynet_error(NULL, "error: A message from [ :%08x ] to [ :%08x ] maybe in an endless loop (version = %d)", sm->source , sm->destination, sm->version);}} else {sm->check_version = sm->version;}
}

4. 如何打断业务

4.1. 在lua.h中定义了lua_checksig宏

/* Add by skynet */LUA_API lua_State * skynet_sig_L;
LUA_API void (lua_checksig_)(lua_State *L);
#define lua_checksig(L) if (skynet_sig_L) { lua_checksig_(L); }

可以看到，当触发信号时，会将skynet_sig_L 置为 l->L; 那么skynet_sig_L是干嘛用的呢？在lua的 lvm.c中有如下定义：

/* Add by skynet */
lua_State * skynet_sig_L = NULL;LUA_API void
lua_checksig_(lua_State *L) {if (skynet_sig_L == G(L)->mainthread) {skynet_sig_L = NULL;lua_pushnil(L);lua_error(L);}
}

查看此宏的引用点，这里会修改Lua源码

vmcase(OP_JMP) {lua_checksig(L);dojump(ci, i, 0);vmbreak;}vmcase(OP_CALL) {int b = GETARG_B(i);int nresults = GETARG_C(i) - 1;lua_checksig(L);...
}vmcase(OP_TAILCALL) {int b = GETARG_B(i);lua_checksig(L);...}vmcase(OP_TAILCALL) {int b = GETARG_B(i);lua_checksig(L);...}vmcase(OP_FORLOOP) {lua_checksig(L);...}vmcase(OP_TFORLOOP) {l_tforloop:lua_checksig(L);...}

在虚拟机指令：跳转OP_JMP、尾调用OP_TAILCALL、for循环OP_FORLOOP、OP_TFORLOOP处都会做checksig检查。除了for循环之外的其它循环，比如while或repeat，在lua中都是通过条件判断结合JMP跳转来实现的，因此也是可以被检查报错的。对于无限递归的情况，如果递归函数可以被优化成尾调用的话，那么会在TAILCALL中被检查并报错。
游戏处理得更加粗暴，在skynet_monitor_check函数里直接调用skynet_command强制退出死循环，然后再通过运维工具监控错误消息，发送警报通知。

4.2. 利用 lua_sethook

使用 lua_sethook 去实现跳出死循环。就不需要修改lua源码以及增加 skynet_sig_L 变量。代码如下，可以看出，当snlua收到信号时，会向lua虚拟机，会使用**luaL_error(…)**抛出一个错误，打断业务的执行。

void
snlua_signal(struct snlua *l, int signal) {skynet_error(l->ctx, "recv a signal %d", signal);if (signal == 0) {if (ATOM_LOAD(&l->trap) == 0) {// only one thread can set trap ( l->trap 0->1 )if (!ATOM_CAS(&l->trap, 0, 1))return;lua_sethook (l->activeL, signal_hook, LUA_MASKCOUNT, 1);  //当执行完一条指令后，就调用signal_hook// finish set ( l->trap 1 -> -1 )ATOM_CAS(&l->trap, 1, -1);}} else if (signal == 1) {skynet_error(l->ctx, "Current Memory %.3fK", (float)l->mem / 1024);}
}static void
signal_hook(lua_State *L, lua_Debug *ar) {void *ud = NULL;lua_getallocf(L, &ud);struct snlua *l = (struct snlua *)ud;lua_sethook (L, NULL, 0, 0);if (ATOM_LOAD(&l->trap)) {ATOM_STORE(&l->trap , 0);luaL_error(L, "signal 0"); //抛出异常，结束循环}
}

如想了解更多skynet知识，可关注微信公众号 GameGuide