From c1c032eab4a0e2e98ba5ac150e14e73ccde85e2e Mon Sep 17 00:00:00 2001
From: Cloud Wu <cloudwu@gmail.com>
Date: Wed, 14 Oct 2015 20:12:30 +0800
Subject: [PATCH] bugfix: socketchannel order mode may blocked. (used by redis
 driver)

---
 lualib/dns.lua           |  2 +-
 lualib/skynet.lua        |  4 +--
 lualib/socket.lua        |  6 ++--
 lualib/socketchannel.lua | 74 ++++++++++++++++++++++------------------
 4 files changed, 46 insertions(+), 40 deletions(-)

diff --git a/lualib/dns.lua b/lualib/dns.lua
index 35e38084..e349699f 100644
--- a/lualib/dns.lua
+++ b/lualib/dns.lua
@@ -265,7 +265,7 @@ local function suspend(tid, name, qtype)
 		co = coroutine.running(),
 	}
 	request_pool[tid] = req
-	skynet.wait()
+	skynet.wait(req.co)
 	local answers = request_pool[tid].answers
 	request_pool[tid] = nil
 	assert(answers, "no ip")
diff --git a/lualib/skynet.lua b/lualib/skynet.lua
index 554d8ff8..c20efaa2 100644
--- a/lualib/skynet.lua
+++ b/lualib/skynet.lua
@@ -275,10 +275,10 @@ function skynet.yield()
 	return skynet.sleep(0)
 end
 
-function skynet.wait()
+function skynet.wait(co)
 	local session = c.genid()
 	local ret, msg = coroutine_yield("SLEEP", session)
-	local co = coroutine.running()
+	co = co or coroutine.running()
 	sleep_session[co] = nil
 	session_id_coroutine[session] = nil
 end
diff --git a/lualib/socket.lua b/lualib/socket.lua
index 7f1d603b..58fe2678 100644
--- a/lualib/socket.lua
+++ b/lualib/socket.lua
@@ -30,7 +30,7 @@ end
 local function suspend(s)
 	assert(not s.co)
 	s.co = coroutine.running()
-	skynet.wait()
+	skynet.wait(s.co)
 	-- wakeup closing corouting every time suspend,
 	-- because socket.close() will wait last socket buffer operation before clear the buffer.
 	if s.closing then
@@ -232,7 +232,7 @@ function socket.close(id)
 			-- wait reading coroutine read the buffer.
 			assert(not s.closing)
 			s.closing = coroutine.running()
-			skynet.wait()
+			skynet.wait(s.closing)
 		else
 			suspend(s)
 		end
@@ -361,7 +361,7 @@ function socket.lock(id)
 	else
 		local co = coroutine.running()
 		table.insert(lock_set, co)
-		skynet.wait()
+		skynet.wait(co)
 	end
 end
 
diff --git a/lualib/socketchannel.lua b/lualib/socketchannel.lua
index 9ff5d451..547b9f87 100644
--- a/lualib/socketchannel.lua
+++ b/lualib/socketchannel.lua
@@ -115,8 +115,17 @@ local function dispatch_by_session(self)
 	end
 end
 
+local wait_response
+
 local function pop_response(self)
-	return table.remove(self.__request, 1), table.remove(self.__thread, 1)
+	while true do
+		local func,co = table.remove(self.__request, 1), table.remove(self.__thread, 1)
+		if func then
+			return func, co
+		end
+		wait_response = coroutine.running()
+		skynet.wait(wait_response)
+	end
 end
 
 local function push_response(self, response, co)
@@ -127,46 +136,43 @@ local function push_response(self, response, co)
 		-- response is a function, push it to __request
 		table.insert(self.__request, response)
 		table.insert(self.__thread, co)
+		if wait_response then
+			skynet.wakeup(wait_response)
+			wait_response = nil
+		end
 	end
 end
 
 local function dispatch_by_order(self)
 	while self.__sock do
 		local func, co = pop_response(self)
-		if func == nil then
-			if not socket.block(self.__sock[1]) then
-				close_channel_socket(self)
-				wakeup_all(self)
+		local ok, result_ok, result_data, padding = pcall(func, self.__sock)
+		if ok then
+			if padding and result_ok then
+				-- if padding is true, wait for next result_data
+				-- self.__result_data[co] is a table
+				local result = self.__result_data[co] or {}
+				self.__result_data[co] = result
+				table.insert(result, result_data)
+			else
+				self.__result[co] = result_ok
+				if result_ok and self.__result_data[co] then
+					table.insert(self.__result_data[co], result_data)
+				else
+					self.__result_data[co] = result_data
+				end
+				skynet.wakeup(co)
 			end
 		else
-			local ok, result_ok, result_data, padding = pcall(func, self.__sock)
-			if ok then
-				if padding and result_ok then
-					-- if padding is true, wait for next result_data
-					-- self.__result_data[co] is a table
-					local result = self.__result_data[co] or {}
-					self.__result_data[co] = result
-					table.insert(result, result_data)
-				else
-					self.__result[co] = result_ok
-					if result_ok and self.__result_data[co] then
-						table.insert(self.__result_data[co], result_data)
-					else
-						self.__result_data[co] = result_data
-					end
-					skynet.wakeup(co)
-				end
-			else
-				close_channel_socket(self)
-				local errmsg
-				if result_ok ~= socket_error then
-					errmsg = result_ok
-				end
-				self.__result[co] = socket_error
-				self.__result_data[co] = errmsg
-				skynet.wakeup(co)
-				wakeup_all(self, errmsg)
+			close_channel_socket(self)
+			local errmsg
+			if result_ok ~= socket_error then
+				errmsg = result_ok
 			end
+			self.__result[co] = socket_error
+			self.__result_data[co] = errmsg
+			skynet.wakeup(co)
+			wakeup_all(self, errmsg)
 		end
 	end
 end
@@ -288,7 +294,7 @@ local function block_connect(self, once)
 		-- connecting in other coroutine
 		local co = coroutine.running()
 		table.insert(self.__connecting, co)
-		skynet.wait()
+		skynet.wait(co)
 	else
 		self.__connecting[1] = true
 		try_connect(self, once)
@@ -319,7 +325,7 @@ end
 local function wait_for_response(self, response)
 	local co = coroutine.running()
 	push_response(self, response, co)
-	skynet.wait()
+	skynet.wait(co)
 
 	local result = self.__result[co]
 	self.__result[co] = nil