mirror of
https://github.com/cloudwu/skynet.git
synced 2026-07-25 12:43:09 +00:00
exit dispatch_by_order when close channel
This commit is contained in:
@@ -68,14 +68,21 @@ local function wakeup_all(self, errmsg)
|
|||||||
for i = 1, #self.__thread do
|
for i = 1, #self.__thread do
|
||||||
local co = self.__thread[i]
|
local co = self.__thread[i]
|
||||||
self.__thread[i] = nil
|
self.__thread[i] = nil
|
||||||
self.__result[co] = socket_error
|
if co then -- ignore the close signal
|
||||||
self.__result_data[co] = errmsg
|
self.__result[co] = socket_error
|
||||||
skynet.wakeup(co)
|
self.__result_data[co] = errmsg
|
||||||
|
skynet.wakeup(co)
|
||||||
|
end
|
||||||
end
|
end
|
||||||
end
|
end
|
||||||
end
|
end
|
||||||
|
|
||||||
|
local function exit_thread(self)
|
||||||
|
local co = coroutine.running()
|
||||||
|
if self.__dispatch_thread == co then
|
||||||
|
self.__dispatch_thread = nil
|
||||||
|
end
|
||||||
|
end
|
||||||
|
|
||||||
local function dispatch_by_session(self)
|
local function dispatch_by_session(self)
|
||||||
local response = self.__response
|
local response = self.__response
|
||||||
@@ -113,6 +120,7 @@ local function dispatch_by_session(self)
|
|||||||
wakeup_all(self, errormsg)
|
wakeup_all(self, errormsg)
|
||||||
end
|
end
|
||||||
end
|
end
|
||||||
|
exit_thread(self)
|
||||||
end
|
end
|
||||||
|
|
||||||
local function pop_response(self)
|
local function pop_response(self)
|
||||||
@@ -144,6 +152,11 @@ end
|
|||||||
local function dispatch_by_order(self)
|
local function dispatch_by_order(self)
|
||||||
while self.__sock do
|
while self.__sock do
|
||||||
local func, co = pop_response(self)
|
local func, co = pop_response(self)
|
||||||
|
if not co then
|
||||||
|
-- close signal
|
||||||
|
wakeup_all(self, errmsg)
|
||||||
|
break
|
||||||
|
end
|
||||||
local ok, result_ok, result_data, padding = pcall(func, self.__sock)
|
local ok, result_ok, result_data, padding = pcall(func, self.__sock)
|
||||||
if ok then
|
if ok then
|
||||||
if padding and result_ok then
|
if padding and result_ok then
|
||||||
@@ -173,6 +186,7 @@ local function dispatch_by_order(self)
|
|||||||
wakeup_all(self, errmsg)
|
wakeup_all(self, errmsg)
|
||||||
end
|
end
|
||||||
end
|
end
|
||||||
|
exit_thread(self)
|
||||||
end
|
end
|
||||||
|
|
||||||
local function dispatch_function(self)
|
local function dispatch_function(self)
|
||||||
@@ -221,7 +235,7 @@ local function connect_once(self)
|
|||||||
end
|
end
|
||||||
|
|
||||||
self.__sock = setmetatable( {fd} , channel_socket_meta )
|
self.__sock = setmetatable( {fd} , channel_socket_meta )
|
||||||
skynet.fork(dispatch_function(self), self)
|
self.__dispatch_thread = skynet.fork(dispatch_function(self), self)
|
||||||
|
|
||||||
if self.__auth then
|
if self.__auth then
|
||||||
self.__authcoroutine = coroutine.running()
|
self.__authcoroutine = coroutine.running()
|
||||||
@@ -385,8 +399,13 @@ end
|
|||||||
|
|
||||||
function channel:close()
|
function channel:close()
|
||||||
if not self.__closed then
|
if not self.__closed then
|
||||||
|
local thread = assert(self.__dispatch_thread)
|
||||||
self.__closed = true
|
self.__closed = true
|
||||||
close_channel_socket(self)
|
close_channel_socket(self)
|
||||||
|
if not self.__response and self.__dispatch_thread == thread then
|
||||||
|
-- dispatch by order, send close signal to dispatch thread
|
||||||
|
push_response(self, true, false) -- (true, false) is close signal
|
||||||
|
end
|
||||||
end
|
end
|
||||||
end
|
end
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user