lihao/skynet
1
0
Fork 0
mirror of https://github.com/cloudwu/skynet.git synced 2026-07-24 20:23:06 +00:00
Code Issues Packages Projects Releases Wiki Activity

concurrence bugfix

Browse Source
This commit is contained in:
Cloud Wu
2014-05-22 11:24:36 +08:00
parent 1fbee1ac0b
commit 90cd967333
1 changed files with 9 additions and 20 deletions
Download Patch File Download Diff File Expand all files Collapse all files

29
skynet-src/skynet_mq.c
View File

@@ -34,7 +34,6 @@ struct global_queue {
struct message_queue ** queue; struct message_queue ** queue;
// We use a separated flag array to ensure the mq is pushed. // We use a separated flag array to ensure the mq is pushed.
// See the comments below. // See the comments below.
bool * flag;
struct message_queue *list; struct message_queue *list;
}; };
@@ -49,7 +48,10 @@ static void
skynet_globalmq_push(struct message_queue * queue) { skynet_globalmq_push(struct message_queue * queue) {
struct global_queue *q= Q; struct global_queue *q= Q;
if (q->flag[GP(q->tail)]) { 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 // The queue may full seldom, save queue in list
assert(queue->next == NULL); assert(queue->next == NULL);
struct message_queue * last; struct message_queue * last;
@@ -60,14 +62,6 @@ skynet_globalmq_push(struct message_queue * queue) {
return; return;
} }
uint32_t tail = GP(__sync_fetch_and_add(&q->tail,1));
// The thread would suspend here, and the q->queue[tail] is last version ,
// but the queue tail is increased.
// So we set q->flag[tail] after changing q->queue[tail].
q->queue[tail] = queue;
__sync_synchronize();
q->flag[tail] = true;
} }
struct message_queue * struct message_queue *
@@ -93,18 +87,14 @@ skynet_globalmq_pop() {
} }
} }
// Check the flag first, if the flag is false, the pushing may not complete.
if(!q->flag[head_ptr]) {
return NULL;
}
__sync_synchronize();
struct message_queue * mq = q->queue[head_ptr]; struct message_queue * mq = q->queue[head_ptr];
if (!__sync_bool_compare_and_swap(&q->head, head, head+1)) { if (!__sync_bool_compare_and_swap(&q->head, head, head+1)) {
return NULL; return NULL;
} }
q->flag[head_ptr] = false; // 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;
}
return mq; return mq;
} }
@@ -220,8 +210,7 @@ skynet_mq_init() {
struct global_queue *q = skynet_malloc(sizeof(*q)); struct global_queue *q = skynet_malloc(sizeof(*q));
memset(q,0,sizeof(*q)); memset(q,0,sizeof(*q));
q->queue = skynet_malloc(MAX_GLOBAL_MQ * sizeof(struct message_queue *)); q->queue = skynet_malloc(MAX_GLOBAL_MQ * sizeof(struct message_queue *));
q->flag = skynet_malloc(MAX_GLOBAL_MQ * sizeof(bool)); memset(q->queue, 0, sizeof(struct message_queue *) * MAX_GLOBAL_MQ);
memset(q->flag, 0, sizeof(bool) * MAX_GLOBAL_MQ);
Q=q; Q=q;
} }