From b5e10b8f9f7a3fc78dcbbcdab8e1e7a4dbbd0a2a Mon Sep 17 00:00:00 2001
From: Cloud Wu <cloudwu@gmail.com>
Date: Wed, 9 Jul 2014 15:26:34 +0800
Subject: [PATCH] add skynet.queue

---
 HISTORY.md              |  1 +
 lualib/mqueue.lua       |  2 ++
 lualib/skynet.lua       |  2 +-
 lualib/skynet/queue.lua | 33 +++++++++++++++++++++++++++++++++
 test/pingserver.lua     | 21 +++++++++++++++++++++
 test/testqueue.lua      | 18 ++++++++++++++++++
 6 files changed, 76 insertions(+), 1 deletion(-)
 create mode 100644 lualib/skynet/queue.lua
 create mode 100644 test/testqueue.lua

diff --git a/HISTORY.md b/HISTORY.md
index 076fe91d..7a2ecf4a 100644
--- a/HISTORY.md
+++ b/HISTORY.md
@@ -3,6 +3,7 @@ Dev version
 * Bugfix : invalid negative socket id 
 * Add optional TCP_NODELAY support
 * Add worker thread weight
+* Add skynet.queue
 
 v0.4.1 (2014-7-7)
 -----------
diff --git a/lualib/mqueue.lua b/lualib/mqueue.lua
index 11931200..f21921ca 100644
--- a/lualib/mqueue.lua
+++ b/lualib/mqueue.lua
@@ -1,3 +1,5 @@
+-- This is a deprecated module, use skynet.queue instead.
+
 local skynet = require "skynet"
 local c = require "skynet.c"
 
diff --git a/lualib/skynet.lua b/lualib/skynet.lua
index f9d934c3..7eba476c 100644
--- a/lualib/skynet.lua
+++ b/lualib/skynet.lua
@@ -22,7 +22,7 @@ local skynet = {
 	PTYPE_HARBOR = 5,
 	PTYPE_SOCKET = 6,
 	PTYPE_ERROR = 7,
-	PTYPE_QUEUE = 8,
+	PTYPE_QUEUE = 8,	-- use in deprecated mqueue, use skynet.queue instead
 	PTYPE_DEBUG = 9,
 	PTYPE_LUA = 10,
 	PTYPE_SNAX = 11,
diff --git a/lualib/skynet/queue.lua b/lualib/skynet/queue.lua
new file mode 100644
index 00000000..9f40e247
--- /dev/null
+++ b/lualib/skynet/queue.lua
@@ -0,0 +1,33 @@
+local skynet = require "skynet"
+local coroutine = coroutine
+local pcall = pcall
+local table = table
+
+function skynet.queue()
+	local current_thread
+	local ref = 0
+	local thread_queue = {}
+	return function(f, ...)
+		local thread = coroutine.running()
+		if ref == 0 then
+			current_thread = thread
+		elseif current_thread ~= thread then
+			table.insert(thread_queue, thread)
+			skynet.wait()
+			assert(ref == 0)
+		end
+		ref = ref + 1
+		local ok, err = pcall(f, ...)
+		ref = ref - 1
+		if ref == 0 then
+			current_thread = nil
+			local co = table.remove(thread_queue,1)
+			if co then
+				skynet.wakeup(co)
+			end
+		end
+		assert(ok,err)
+	end
+end
+
+return skynet.queue
\ No newline at end of file
diff --git a/test/pingserver.lua b/test/pingserver.lua
index 635cfeab..2ddefd7f 100644
--- a/test/pingserver.lua
+++ b/test/pingserver.lua
@@ -1,4 +1,5 @@
 local skynet = require "skynet"
+local queue = require "skynet.queue"
 
 local i = 0
 local hello = "hello"
@@ -8,9 +9,27 @@ function response.ping(hello)
 	return hello
 end
 
+-- response.sleep and accept.hello share one lock
+local lock
+
+function accept.sleep(queue, n)
+	if queue then
+		lock(
+		function()
+			print("queue=",queue, n)
+			skynet.sleep(n)
+		end)
+	else
+		print("queue=",queue, n)
+		skynet.sleep(n)
+	end
+end
+
 function accept.hello()
+	lock(function()
 	i = i + 1
 	print (i, hello)
+	end)
 end
 
 function response.error()
@@ -19,6 +38,8 @@ end
 
 function init( ... )
 	print ("ping server start:", ...)
+	-- init queue
+	lock = queue()
 
 -- You can return "queue" for queue service mode
 --	return "queue"
diff --git a/test/testqueue.lua b/test/testqueue.lua
new file mode 100644
index 00000000..2076531e
--- /dev/null
+++ b/test/testqueue.lua
@@ -0,0 +1,18 @@
+local skynet = require "skynet"
+local snax = require "snax"
+
+skynet.start(function()
+	local ps = snax.uniqueservice ("pingserver", "test queue")
+	for i=1, 10 do
+		ps.post.sleep(true,i*10)
+		ps.post.hello()
+	end
+	for i=1, 10 do
+		ps.post.sleep(false,i*10)
+		ps.post.hello()
+	end
+
+	skynet.exit()
+end)
+
+