From a0d2c7172cf0bd9bb71037a9cb2a76be978ac315 Mon Sep 17 00:00:00 2001 From: xjdrew Date: Thu, 4 Dec 2014 15:50:19 +0800 Subject: [PATCH 1/3] =?UTF-8?q?=E4=BD=BF=E7=94=A8spinlock=EF=BC=8C?= =?UTF-8?q?=E9=81=BF=E5=85=8Dcas=E4=B8=8D=E8=83=BD=E5=85=85=E5=88=86?= =?UTF-8?q?=E8=B0=83=E5=BA=A6testdeadloop=E8=BF=99=E6=A0=B7=E7=9A=84?= =?UTF-8?q?=E7=94=A8=E4=BE=8B?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- skynet-src/skynet_mq.c | 69 ++++++++++++------------------------------ test/testdeadloop.lua | 10 ++++++ 2 files changed, 30 insertions(+), 49 deletions(-) create mode 100644 test/testdeadloop.lua 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) From 2da2f121c8f8524984792775769f38c36b735aa0 Mon Sep 17 00:00:00 2001 From: xjdrew Date: Thu, 4 Dec 2014 16:16:17 +0800 Subject: [PATCH 2/3] delete unneccessary comment --- skynet-src/skynet_mq.c | 1 - 1 file changed, 1 deletion(-) diff --git a/skynet-src/skynet_mq.c b/skynet-src/skynet_mq.c index 51f47ef1..098cb210 100644 --- a/skynet-src/skynet_mq.c +++ b/skynet-src/skynet_mq.c @@ -48,7 +48,6 @@ void skynet_globalmq_push(struct message_queue * queue) { struct global_queue *q= Q; - // only one thread can set the slot (change q->queue[tail] from NULL to queue) LOCK(q) assert(queue->next == NULL); if(q->tail) { From fe9640e2dd6224b1f7257b08e806a5db3e019b5c Mon Sep 17 00:00:00 2001 From: dpull Date: Thu, 4 Dec 2014 20:10:24 +0800 Subject: [PATCH 3/3] =?UTF-8?q?1=E3=80=81mongo=5Fcollection:createIndex=20?= =?UTF-8?q?=E5=88=9B=E5=BB=BA=E7=B4=A2=E5=BC=95=202=E3=80=81mongo=5Fcollec?= =?UTF-8?q?tion:safe=5Finsert=20=E5=8F=AF=E4=BB=A5=E5=88=A4=E6=96=AD?= =?UTF-8?q?=E8=BF=94=E5=9B=9E=E5=80=BC=E7=9A=84insert=203=E3=80=81mongo=5F?= =?UTF-8?q?collection:findAndModify=20=E6=9F=A5=E8=AF=A2=E5=B9=B6=E4=BF=AE?= =?UTF-8?q?=E6=94=B9?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- lualib/mongo.lua | 48 +++++++++++++++++++++++++++++++++++++++++++++++- 1 file changed, 47 insertions(+), 1 deletion(-) diff --git a/lualib/mongo.lua b/lualib/mongo.lua index 458e1a6c..38ac20dd 100644 --- a/lualib/mongo.lua +++ b/lualib/mongo.lua @@ -227,7 +227,11 @@ function mongo_collection:insert(doc) sock:request(pack) end -function mongo_collection:batch_insert(docs) +function mongo_collection:safe_insert(doc) + return self.database:runCommand("insert", self.name, "documents", {bson_encode(doc)}) +end + +function mongo_collection:batch_insert(docs) for i=1,#docs do if docs[i]._id == nil then docs[i]._id = bson.objectid() @@ -276,6 +280,48 @@ function mongo_collection:find(query, selector) } , cursor_meta) end +-- collection:createIndex({username = 1}, {unique = true}) +function mongo_collection:createIndex(keys, option) + local name + for k, v in pairs(keys) do + assert(v == 1) + name = (name == nil) and k or (name .. "_" .. k) + end + + local doc = {}; + doc.name = name + doc.key = keys + for k, v in pairs(option) do + if v then + doc[k] = true + end + end + return self.database:runCommand("createIndexes", self.name, "indexes", {doc}) +end + +mongo_collection.ensureIndex = mongo_collection.createIndex; + +-- collection:findAndModify({query = {name = "userid"}, update = {["$inc"] = {nextid = 1}}, }) +-- keys, value type +-- query, table +-- sort, table +-- remove, bool +-- update, table +-- new, bool +-- fields, bool +-- upsert, boolean +function mongo_collection:findAndModify(doc) + assert(doc.query) + assert(doc.update or doc.remove) + + local cmd = {"findAndModify", self.name}; + for k, v in pairs(doc) do + table.insert(cmd, k) + table.insert(cmd, v) + end + return self.database:runCommand(unpack(cmd)) +end + function mongo_cursor:hasNext() if self.__ptr == nil then if self.__document == nil then