add skynet.blockcall

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)

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)

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)

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;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;
+}
+
+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;i<q->cap;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);

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

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