From aed83ea5e87e47cc3f0086e773d1dc745bac8029 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E4=BA=91=E9=A3=8E?= Date: Thu, 11 Oct 2012 19:25:13 +0800 Subject: [PATCH] lock free global mq --- config | 1 - skynet-src/skynet_imp.h | 1 - skynet-src/skynet_main.c | 1 - skynet-src/skynet_mq.c | 73 ++++++++++++++++----------------------- skynet-src/skynet_mq.h | 2 +- skynet-src/skynet_start.c | 3 +- 6 files changed, 32 insertions(+), 49 deletions(-) diff --git a/config b/config index f5f5c448..e3fc0db8 100644 --- a/config +++ b/config @@ -1,6 +1,5 @@ root = "./" thread = 8 -mqueue = 256 logger = nil harbor = 1 address = "127.0.0.1:2525" diff --git a/skynet-src/skynet_imp.h b/skynet-src/skynet_imp.h index bd76b1bd..a5c7e078 100644 --- a/skynet-src/skynet_imp.h +++ b/skynet-src/skynet_imp.h @@ -3,7 +3,6 @@ struct skynet_config { int thread; - int mqueue_size; int harbor; const char * logger; const char * module_path; diff --git a/skynet-src/skynet_main.c b/skynet-src/skynet_main.c index 1cd53a67..49db5147 100644 --- a/skynet-src/skynet_main.c +++ b/skynet-src/skynet_main.c @@ -102,7 +102,6 @@ main(int argc, char *argv[]) { optstring("luaservice","./service/?.lua"); config.thread = optint("thread",8); - config.mqueue_size = optint("mqueue",256); config.module_path = optstring("cpath","./service/?.so"); config.logger = optstring("logger",NULL); config.harbor = optint("harbor", 1); diff --git a/skynet-src/skynet_mq.c b/skynet-src/skynet_mq.c index a86fcbd6..ffa3a11c 100644 --- a/skynet-src/skynet_mq.c +++ b/skynet-src/skynet_mq.c @@ -7,8 +7,10 @@ #include #include #include +#include #define DEFAULT_QUEUE_SIZE 64; +#define MAX_GLOBAL_MQ 0x10000 struct message_queue { uint32_t handle; @@ -23,11 +25,11 @@ struct message_queue { }; struct global_queue { - int cap; - int head; - int tail; - int lock; + uint32_t head; + uint32_t tail; struct message_queue ** queue; + bool * flag; + }; static struct global_queue *Q = NULL; @@ -35,49 +37,36 @@ static struct global_queue *Q = NULL; #define LOCK(q) while (__sync_lock_test_and_set(&(q)->lock,1)) {} #define UNLOCK(q) __sync_lock_release(&(q)->lock); +#define GP(p) ((p) % (MAX_GLOBAL_MQ-1)) + static void skynet_globalmq_push(struct message_queue * queue) { + struct global_queue *q= Q; - LOCK(q) - - q->queue[q->tail] = queue; - if (++ q->tail >= q->cap) { - q->tail = 0; - } - - if (q->head == q->tail) { - struct message_queue **new_queue = malloc(sizeof(struct message_queue *) * q->cap * 2); - int i; - for (i=0;icap;i++) { - new_queue[i] = q->queue[(q->head + i) % q->cap]; - } - q->head = 0; - q->tail = q->cap; - q->cap *= 2; - - free(q->queue); - q->queue = new_queue; - } - - UNLOCK(q) + uint32_t tail = __sync_fetch_and_add(&q->tail,1); + assert(GP(tail+1) != GP(q->head)); + tail = GP(tail); + q->queue[tail] = queue; + __sync_synchronize(); + q->flag[tail] = true; } struct message_queue * skynet_globalmq_pop() { struct global_queue *q = Q; - struct message_queue * ret = NULL; - LOCK(q) + if (GP(q->head) == GP(q->tail)) { + return NULL; + } + uint32_t head = __sync_add_and_fetch(&q->head,1); + head = GP(head); - if (q->head != q->tail) { - ret = q->queue[q->head]; - if ( ++ q->head >= q->cap) { - q->head = 0; - } + while(!q->flag[head]) { + __sync_synchronize(); } - UNLOCK(q) - - return ret; + q->flag[head] = false; + + return q->queue[head]; } struct message_queue * @@ -203,16 +192,12 @@ skynet_mq_lock(struct message_queue *q, int session) { } void -skynet_mq_init(int n) { +skynet_mq_init() { struct global_queue *q = malloc(sizeof(*q)); memset(q,0,sizeof(*q)); - int cap = 2; - while (cap < n) { - cap *=2; - } - - q->cap = cap; - q->queue = malloc(cap * sizeof(struct message_queue *)); + q->queue = malloc(MAX_GLOBAL_MQ * sizeof(struct message_queue *)); + q->flag = malloc(MAX_GLOBAL_MQ * sizeof(bool)); + memset(q->flag, 0, sizeof(bool) * MAX_GLOBAL_MQ); Q=q; } diff --git a/skynet-src/skynet_mq.h b/skynet-src/skynet_mq.h index deadd1e7..bd71e91f 100644 --- a/skynet-src/skynet_mq.h +++ b/skynet-src/skynet_mq.h @@ -28,6 +28,6 @@ void skynet_mq_lock(struct message_queue *q, int session); void skynet_mq_force_push(struct message_queue *q); void skynet_mq_pushglobal(struct message_queue *q); -void skynet_mq_init(int cap); +void skynet_mq_init(); #endif diff --git a/skynet-src/skynet_start.c b/skynet-src/skynet_start.c index 20a03caa..28e65781 100644 --- a/skynet-src/skynet_start.c +++ b/skynet-src/skynet_start.c @@ -1,3 +1,4 @@ +#include "skynet.h" #include "skynet_server.h" #include "skynet_imp.h" #include "skynet_mq.h" @@ -96,7 +97,7 @@ skynet_start(struct skynet_config * config) { skynet_group_init(); skynet_harbor_init(config->harbor); skynet_handle_init(config->harbor); - skynet_mq_init(config->mqueue_size); + skynet_mq_init(); skynet_module_init(config->module_path); skynet_timer_init();