diff --git a/lualib/skynet.lua b/lualib/skynet.lua index 589a9f97..b7f30b0a 100644 --- a/lualib/skynet.lua +++ b/lualib/skynet.lua @@ -174,6 +174,13 @@ function skynet.call(addr, typename, ...) return p.unpack(coroutine.yield("CALL", session)) end +function skynet.blockcall(addr, typename , ...) + local p = proto[typename] + local session = c.send(addr, p.id , nil , p.pack(...)) + c.command("LOCK",tostring(session)) + return p.unpack(coroutine.yield("CALL", session)) +end + function skynet.rawcall(addr, typename, msg, sz) local p = proto[typename] local session = c.send(addr, p.id , nil , msg, sz) diff --git a/service/pingserver.lua b/service/pingserver.lua new file mode 100644 index 00000000..e58226f6 --- /dev/null +++ b/service/pingserver.lua @@ -0,0 +1,18 @@ +local skynet = require "skynet" + +local command = {} + +function command.PING(hello) + skynet.ret(skynet.pack(hello)) +end + +function command.HELLO() + skynet.sleep(100) + skynet.ret(skynet.pack("hello")) +end + +skynet.start(function() + skynet.dispatch("lua", function(session,addr, cmd, ...) + command[cmd](...) + end) +end) diff --git a/service/testblockcall.lua b/service/testblockcall.lua new file mode 100644 index 00000000..b0d7e76f --- /dev/null +++ b/service/testblockcall.lua @@ -0,0 +1,10 @@ +local skynet = require "skynet" + +skynet.start(function() + local ping = skynet.newservice("pingserver") + skynet.timeout(0,function() + print(skynet.call(ping,"lua","PING","ping")) + end) + + print(skynet.blockcall(ping,"lua","HELLO")) +end) diff --git a/skynet-src/skynet_mq.c b/skynet-src/skynet_mq.c index 7bb49dea..eb0de4bd 100644 --- a/skynet-src/skynet_mq.c +++ b/skynet-src/skynet_mq.c @@ -17,6 +17,7 @@ struct message_queue { int tail; int lock; int release; + int lock_session; int in_global; struct skynet_message *queue; }; @@ -89,6 +90,7 @@ skynet_mq_create(uint32_t handle) { q->lock = 0; q->in_global = 1; q->release = 0; + q->lock_session = 0; q->queue = malloc(sizeof(struct skynet_message) * q->cap); return q; @@ -128,39 +130,78 @@ skynet_mq_pop(struct message_queue *q, struct skynet_message *message) { return ret; } +static void +expand_queue(struct message_queue *q) { + struct skynet_message *new_queue = malloc(sizeof(struct skynet_message) * 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; +} + +static void +_pushhead(struct message_queue *q, struct skynet_message *message) { + int head = q->head - 1; + if (head < 0) { + head = q->cap - 1; + } + if (head == q->tail) { + expand_queue(q); + --q->tail; + head = q->cap - 1; + } + + q->queue[head] = *message; + q->head = head; + + // this api use in push a unlock message, so the in_global flags must be 1 , but the q is not exist in global queue. + assert(q->in_global); + skynet_globalmq_push(q); +} + void skynet_mq_push(struct message_queue *q, struct skynet_message *message) { + assert(message); LOCK(q) - - if (message) { + + if (q->lock_session !=0 && message->session == q->lock_session) { + _pushhead(q,message); + q->lock_session = 0; + } else { q->queue[q->tail] = *message; if (++ q->tail >= q->cap) { q->tail = 0; } if (q->head == q->tail) { - struct skynet_message *new_queue = malloc(sizeof(struct skynet_message) * q->cap * 2); - int i; - for (i=0;icap;i++) { - new_queue[i] = q->queue[(q->head + i) % q->cap]; + expand_queue(q); + } + + if (q->lock_session == 0) { + if (q->in_global == 0) { + q->in_global = 1; + skynet_globalmq_push(q); } - q->head = 0; - q->tail = q->cap; - q->cap *= 2; - - free(q->queue); - q->queue = new_queue; } } - - if (q->in_global == 0) { - q->in_global = 1; - skynet_globalmq_push(q); - } UNLOCK(q) } +void +skynet_mq_lock(struct message_queue *q, int session) { + LOCK(q) + assert(q->lock_session == 0); + q->lock_session = session; + UNLOCK(q) +} + void skynet_mq_init(int n) { struct global_queue *q = malloc(sizeof(*q)); @@ -181,6 +222,14 @@ skynet_mq_force_push(struct message_queue * queue) { skynet_globalmq_push(queue); } +void +skynet_mq_pushglobal(struct message_queue *queue) { + assert(queue->in_global); + if (queue->lock_session == 0) { + skynet_globalmq_push(queue); + } +} + void skynet_mq_mark_release(struct message_queue *q) { assert(q->release == 0); diff --git a/skynet-src/skynet_mq.h b/skynet-src/skynet_mq.h index c024250f..deadd1e7 100644 --- a/skynet-src/skynet_mq.h +++ b/skynet-src/skynet_mq.h @@ -23,7 +23,10 @@ uint32_t skynet_mq_handle(struct message_queue *); // 0 for success int skynet_mq_pop(struct message_queue *q, struct skynet_message *message); void skynet_mq_push(struct message_queue *q, struct skynet_message *message); +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); diff --git a/skynet-src/skynet_server.c b/skynet-src/skynet_server.c index 0d105fbc..a5ddd9a2 100644 --- a/skynet-src/skynet_server.c +++ b/skynet-src/skynet_server.c @@ -249,7 +249,7 @@ skynet_context_message_dispatch(void) { } assert(q == ctx->queue); - skynet_mq_force_push(q); + skynet_mq_pushglobal(q); skynet_context_release(ctx); return 0; @@ -326,6 +326,16 @@ skynet_command(struct skynet_context * context, const char * cmd , const char * return context->result; } + if (strcmp(cmd,"LOCK") == 0) { + if (context->init == 0) { + return NULL; + } + int session = strtol(param, NULL, 10); + assert(session); + skynet_mq_lock(context->queue, session); + return NULL; + } + if (strcmp(cmd,"REG") == 0) { if (param == NULL || param[0] == '\0') { sprintf(context->result, ":%x", context->handle);