add skynet.queue

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

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"
 

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,

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

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"

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