bugfix: socketchannel order mode may blocked. (used by redis driver)

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

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

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
 

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,18 +136,16 @@ 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)
-			end
-		else
 		local ok, result_ok, result_data, padding = pcall(func, self.__sock)
 		if ok then
 			if padding and result_ok then
@@ -168,7 +175,6 @@ local function dispatch_by_order(self)
 			wakeup_all(self, errmsg)
 		end
 	end
-	end
 end
 
 local function dispatch_function(self)
@@ -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