lihao/skynet
1
0
Fork 0
mirror of https://github.com/cloudwu/skynet.git synced 2026-07-25 12:43:09 +00:00
Code Issues Packages Projects Releases Wiki Activity

optimize message dispatch

Browse Source
This commit is contained in:
Cloud Wu
2014-05-26 13:40:49 +08:00
parent f77482024e
commit e1a72bd820
3 changed files with 21 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files

26
skynet-src/skynet_server.c
View File

@@ -228,11 +228,13 @@ _dispatch_message(struct skynet_context *ctx, struct skynet_message *msg) {
CHECKCALLING_END(ctx) CHECKCALLING_END(ctx)
} }
int struct message_queue *
skynet_context_message_dispatch(struct skynet_monitor *sm) { skynet_context_message_dispatch(struct skynet_monitor *sm, struct message_queue *q) {
struct message_queue * q = skynet_globalmq_pop(); if (q == NULL) {
if (q==NULL) q = skynet_globalmq_pop();
return 1; if (q==NULL)
return NULL;
}
uint32_t handle = skynet_mq_handle(q); uint32_t handle = skynet_mq_handle(q);
@@ -240,13 +242,13 @@ skynet_context_message_dispatch(struct skynet_monitor *sm) {
if (ctx == NULL) { if (ctx == NULL) {
struct drop_t d = { handle }; struct drop_t d = { handle };
skynet_mq_release(q, drop_message, &d); skynet_mq_release(q, drop_message, &d);
return 0; return skynet_globalmq_pop();
} }
struct skynet_message msg; struct skynet_message msg;
if (skynet_mq_pop(q,&msg)) { if (skynet_mq_pop(q,&msg)) {
skynet_context_release(ctx); skynet_context_release(ctx);
return 0; return skynet_globalmq_pop();
} }
skynet_monitor_trigger(sm, msg.source , handle); skynet_monitor_trigger(sm, msg.source , handle);
@@ -258,12 +260,18 @@ skynet_context_message_dispatch(struct skynet_monitor *sm) {
} }
assert(q == ctx->queue); assert(q == ctx->queue);
skynet_globalmq_push(q); 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); skynet_context_release(ctx);
skynet_monitor_trigger(sm, 0,0); skynet_monitor_trigger(sm, 0,0);
return 0; return q;
} }
static void static void

2
skynet-src/skynet_server.h
View File

@@ -16,7 +16,7 @@ void skynet_context_init(struct skynet_context *, uint32_t handle);
int skynet_context_push(uint32_t handle, struct skynet_message *message); int skynet_context_push(uint32_t handle, struct skynet_message *message);
void skynet_context_send(struct skynet_context * context, void * msg, size_t sz, uint32_t source, int type, int session); void skynet_context_send(struct skynet_context * context, void * msg, size_t sz, uint32_t source, int type, int session);
int skynet_context_newsession(struct skynet_context *); int skynet_context_newsession(struct skynet_context *);
int skynet_context_message_dispatch(struct skynet_monitor *); // return 1 when block struct message_queue * skynet_context_message_dispatch(struct skynet_monitor *, struct message_queue *); // return next queue
int skynet_context_total(); int skynet_context_total();
void skynet_context_endless(uint32_t handle); // for monitor void skynet_context_endless(uint32_t handle); // for monitor

4
skynet-src/skynet_start.c
View File

@@ -121,8 +121,10 @@ _worker(void *p) {
struct monitor *m = wp->m; struct monitor *m = wp->m;
struct skynet_monitor *sm = m->m[id]; struct skynet_monitor *sm = m->m[id];
skynet_initthread(THREAD_WORKER); skynet_initthread(THREAD_WORKER);
struct message_queue * q = NULL;
for (;;) { for (;;) {
if (skynet_context_message_dispatch(sm)) { q = skynet_context_message_dispatch(sm, q);
if (q == NULL) {
CHECK_ABORT CHECK_ABORT
if (pthread_mutex_lock(&m->mutex) == 0) { if (pthread_mutex_lock(&m->mutex) == 0) {
++ m->sleep; ++ m->sleep;