lock free global mq

config

@@ -1,6 +1,5 @@
 root = "./"
 thread = 8
-mqueue = 256
 logger = nil
 harbor = 1
 address = "127.0.0.1:2525"

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;

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);

skynet-src/skynet_mq.c

@@ -7,8 +7,10 @@
 #include <stdlib.h>
 #include <string.h>
 #include <assert.h>
+#include <stdbool.h>
 
 #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;
+	uint32_t head;
-	int head;
+	uint32_t tail;
-	int tail;
-	int lock;
 	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)
+	uint32_t tail = __sync_fetch_and_add(&q->tail,1);
-
+	assert(GP(tail+1) != GP(q->head));
-	q->queue[q->tail] = queue;
+	tail = GP(tail);
-	if (++ q->tail >= q->cap) {
+	q->queue[tail] = queue;
-		q->tail = 0;
+	__sync_synchronize();
-	}
+	q->flag[tail] = true;
-
-	if (q->head == q->tail) {
-		struct message_queue **new_queue = malloc(sizeof(struct message_queue *) * q->cap * 2);
-		int i;
-		for (i=0;i<q->cap;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)
 }
 
 struct message_queue * 
 skynet_globalmq_pop() {
 	struct global_queue *q = Q;
-	struct message_queue * ret = NULL;
+	if (GP(q->head) == GP(q->tail)) {
-	LOCK(q)
+		return NULL;
+	}
+	uint32_t head = __sync_add_and_fetch(&q->head,1);
+	head = GP(head);
 
-	if (q->head != q->tail) {
+	while(!q->flag[head]) {
-		ret = q->queue[q->head];
+		__sync_synchronize();
-		if ( ++ q->head >= q->cap) {
-			q->head = 0;
-		}
 	}
 
-	UNLOCK(q)
+	q->flag[head] = false;
-	
+
-	return ret;
+	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;
+	q->queue = malloc(MAX_GLOBAL_MQ * sizeof(struct message_queue *));
-	while (cap < n) {
+	q->flag = malloc(MAX_GLOBAL_MQ * sizeof(bool));
-		cap *=2;
+	memset(q->flag, 0, sizeof(bool) * MAX_GLOBAL_MQ);
-	}
-	
-	q->cap = cap;
-	q->queue = malloc(cap * sizeof(struct message_queue *));
 	Q=q;
 }
 

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

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();