Skynet_处理队列消息

Skynet_处理队列消息

工作线程 thread_worker 在设置完参数后，服务队列的消息在 while 循环中不断被处理和分发：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
static void *
thread_worker(void *p) {
	struct worker_parm *wp = p;
	int id = wp->id;
	int 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);//next
		if (q == NULL) {//没有消息处理 就休息一下 等待唤醒
			if (pthread_mutex_lock(&m->mutex) == 0) {
				++ m->sleep;
				// “虚假唤醒”是无害的，
				// 因为skynet_context_message_dispatch（）可以随时调用。
				if (!m->quit)
					pthread_cond_wait(&m->cond, &m->mutex);//这个函数调用时释放锁,返回时获取锁 
				-- m->sleep;
				if (pthread_mutex_unlock(&m->mutex)) {
					fprintf(stderr, "unlock mutex error");
					exit(1);
				}
			}
		}
	}
	return NULL;
}

服务队列的消息被消息分发函数 skynet_context_message_dispatch 进行处理：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
struct message_queue * //next
skynet_context_message_dispatch(struct skynet_monitor *sm, struct message_queue *q, int weight) {
	if (q == NULL) {
		q = skynet_globalmq_pop();//从全局队列中取出一个服务队列
		if (q==NULL)
			return NULL;
	}

	uint32_t handle = skynet_mq_handle(q);//通过队列获取他对应的handle

	struct skynet_context * ctx = skynet_handle_grab(handle);//从cxt仓库中获取ctx

	int i,n=1;
	struct skynet_message msg;

	for (i=0;i<n;i++) {
		if (skynet_mq_pop(q,&msg)) {//skynet_mq_pop返回值为0 表示成功 否则表示队列是空的
			skynet_context_release(ctx);//表示服务可能标记退出,如果不是,注意 此时队列没有重新push到全局队列
			return skynet_globalmq_pop();
		} else if (i==0 && weight >= 0) {
			n = skynet_mq_length(q);//当前队列等待处理的消息个数
			n >> = weight;//假设此时weight是2 队列长度是16 那么此时n==4 即这次要处理4个消息
		}
		int overload = skynet_mq_overload(q);//overload包含两个意思 1.是否过载 2.当前过载数,也就是当前队列等待处理的消息个数
		if (overload) {//过载
			skynet_error(ctx, "May overload, message queue length = %d", overload);
		}

		skynet_monitor_trigger(sm, msg.source , handle);//如果339行一直得不到执行 那么 监听管理线程 就会报告消息处理进入死循环

		if (ctx->cb == NULL) {
			skynet_free(msg.data);
		} else {
			dispatch_message(ctx, &msg);//把消息交给服务内部的回调函数处理 next
		}

		skynet_monitor_trigger(sm, 0,0);
	}

	assert(q == ctx->queue);
	struct message_queue *nq = skynet_globalmq_pop();//再次弹出一个队列
	if (nq) {//说明全局队列不为空 那么把刚刚处理过的队列push进全局队列 并返回新拿到的队列为下一轮消息处理做准备
		// 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;
}

工作线程的每一轮处理过程是：

  1. 从全局队列中取出一个队列，然后取出队列里面的消息进行处理。
  2. 队列处理完成后,把队列重新加入到全局队列的尾部。接着处理下一轮的消息。
  3. 每次拿到一个队列时,可能会处理其中的一个消息,也可能处理其中的几个消息。（这个与当前线程分配的权重有关。这个权重在工作线程启动时就设置好了）。如下所示
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
static int weight[] = { 
		-1, -1, -1, -1, 0, 0, 0, 0,
		1, 1, 1, 1, 1, 1, 1, 1, 
		2, 2, 2, 2, 2, 2, 2, 2, 
		3, 3, 3, 3, 3, 3, 3, 3, };
	struct worker_parm wp[thread];
	for (i=0;i<thread;i++) {
		wp[i].m = m;
		wp[i].id = i;
		if (i < sizeof(weight)/sizeof(weight[0])) {//设置前面32个工作线程的 weight
			wp[i].weight= weight[i];
		} else {//第32个工作线程之后的 weight
			wp[i].weight = 0;
		}
		create_thread(&pid[i+3], thread_worker, &wp[i]);//给每个工作线程传递不同参数
	}

工作线程中，队列消息的权重：

Skynet_消息权重

当工作线程确定处理某条消息的时候调用，通过 dispatch_message 函数把消息交给服务内部的回调函数处理：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
static void
dispatch_message(struct skynet_context *ctx, struct skynet_message *msg) {
	assert(ctx->init);
	CHECKCALLING_BEGIN(ctx)
	pthread_setspecific(G_NODE.handle_key, (void *)(uintptr_t)(ctx->handle));//把handle保存到线程局部存储 是为获取当前服务内存情况服务的
	int type = msg->sz >> MESSAGE_TYPE_SHIFT;//获取消息的类型 比如 PTYPE_TEXT PTYPE_SOCKET 等等
	size_t sz = msg->sz & MESSAGE_TYPE_MASK; //获取消息的真实sz 实质上去掉了sz头部的高八位代表的消息类型
	FILE *f = (FILE *)ATOM_LOAD(&ctx->logfile);//每个服务可以有自己专属的日志文件 这个有开关设置 注意这不同于skynet.error这种通用日志
	if (f) {
		skynet_log_output(f, msg->source, type, msg->session, msg->data, sz);
	}
	++ctx->message_count;//递增当前服务已经处理的消息个数
	int reserve_msg;
	if (ctx->profile) {
		ctx->cpu_start = skynet_thread_time();
		reserve_msg = ctx->cb(ctx, ctx->cb_ud, type, msg->session, msg->source, msg->data, sz);	//调用ctx的cb处理
		uint64_t cost_time = skynet_thread_time() - ctx->cpu_start;	//计算花费的时间
		ctx->cpu_cost += cost_time;	//记录CPU花费的时间
	} else {
		reserve_msg = ctx->cb(ctx, ctx->cb_ud, type, msg->session, msg->source, msg->data, sz);	//调用ctx的cb处理
	}
	if (!reserve_msg) {//reserve_msg是 0 则释放内存
		skynet_free(msg->data);
	}
	CHECKCALLING_END(ctx)
}

snlua服务的回调函数

对于snula服务来说 ，最终的回调函数是在lua层调用 skynet.start 时注册的

1
2
3
4
5
6
7
function skynet.start(start_func)
	c.callback(skynet.dispatch_message) -- 这里注册的回调函数
	init_thread = skynet.timeout(0, function()
		skynet.init_service(start_func)
		init_thread = nil
	end)
end

查看 c.callback(skynet.dispatch_message) 回调函数的注册过程 lcallback 

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
static int
lcallback(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);//相当于 注册表[_cb] = fun

	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);//lua服务注册回调函数 在回调函数里面会调用lua应用层的消息分发函数skynet.dispatch_message
	}

	return 0;
}

最终的底层回调函数是 _cb 

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
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);//调用lua层的skynet.dispatch_message 其参数是prototype, msg, sz, session, source

	if (r == LUA_OK) {
		return 0;
	}
	const char * self = skynet_command(context, "REG", NULL);
	switch (r) {
	case LUA_ERRRUN:
		skynet_error(context, "lua call [%x to %s : %d msgsz = %d] error : " KRED "%s" KNRM, source , self, session, sz, lua_tostring(L,-1));
		break;
	case LUA_ERRMEM:
		skynet_error(context, "lua memory error : [%x to %s : %d]", source , self, session);
		break;
	case LUA_ERRERR:
		skynet_error(context, "lua error in error : [%x to %s : %d]", source , self, session);
		break;
	};

	lua_pop(L,1);

	return 0;
}

最终的返回值是 0，表示lua层处理完后，底层会释放 msg指向的内存