diff --git a/skynet-src/skynet_mq.c b/skynet-src/skynet_mq.c index 8cc8f9da..51f47ef1 100644 --- a/skynet-src/skynet_mq.c +++ b/skynet-src/skynet_mq.c @@ -32,12 +32,9 @@ struct message_queue { }; struct global_queue { - uint32_t head; - uint32_t tail; - struct message_queue ** queue; - // We use a separated flag array to ensure the mq is pushed. - // See the comments below. - struct message_queue *list; + struct message_queue *head; + struct message_queue *tail; + int lock; }; static struct global_queue *Q = NULL; @@ -51,57 +48,33 @@ void skynet_globalmq_push(struct message_queue * queue) { struct global_queue *q= Q; - uint32_t tail = GP(__sync_fetch_and_add(&q->tail,1)); - // only one thread can set the slot (change q->queue[tail] from NULL to queue) - if (!__sync_bool_compare_and_swap(&q->queue[tail], NULL, queue)) { - // The queue may full seldom, save queue in list - assert(queue->next == NULL); - struct message_queue * last; - do { - last = q->list; - queue->next = last; - } while(!__sync_bool_compare_and_swap(&q->list, last, queue)); - - return; + LOCK(q) + assert(queue->next == NULL); + if(q->tail) { + q->tail->next = queue; + q->tail = queue; + } else { + q->head = q->tail = queue; } + UNLOCK(q) } struct message_queue * skynet_globalmq_pop() { struct global_queue *q = Q; - uint32_t head = q->head; - if (head == q->tail) { - // The queue is empty. - return NULL; - } - - uint32_t head_ptr = GP(head); - - struct message_queue * list = q->list; - if (list) { - // If q->list is not empty, try to load it back to the queue - struct message_queue *newhead = list->next; - if (__sync_bool_compare_and_swap(&q->list, list, newhead)) { - // try load list only once, if success , push it back to the queue. - list->next = NULL; - skynet_globalmq_push(list); + LOCK(q) + struct message_queue *mq = q->head; + if(mq) { + q->head = mq->next; + if(q->head == NULL) { + assert(mq == q->tail); + q->tail = NULL; } + mq->next = NULL; } - - struct message_queue * mq = q->queue[head_ptr]; - if (mq == NULL) { - // globalmq push not complete - return NULL; - } - if (!__sync_bool_compare_and_swap(&q->head, head, head+1)) { - return NULL; - } - // only one thread can get the slot (change q->queue[head_ptr] to NULL) - if (!__sync_bool_compare_and_swap(&q->queue[head_ptr], mq, NULL)) { - return NULL; - } + UNLOCK(q) return mq; } @@ -243,8 +216,6 @@ void skynet_mq_init() { struct global_queue *q = skynet_malloc(sizeof(*q)); memset(q,0,sizeof(*q)); - q->queue = skynet_malloc(MAX_GLOBAL_MQ * sizeof(struct message_queue *)); - memset(q->queue, 0, sizeof(struct message_queue *) * MAX_GLOBAL_MQ); Q=q; } diff --git a/test/testdeadloop.lua b/test/testdeadloop.lua new file mode 100644 index 00000000..215700e2 --- /dev/null +++ b/test/testdeadloop.lua @@ -0,0 +1,10 @@ +local skynet = require "skynet" +local function dead_loop() + while true do + skynet.sleep(0) + end +end + +skynet.start(function() + skynet.fork(dead_loop) +end)