bugfix: socket channel reconnect

This commit is contained in:
Cloud Wu
2014-04-24 23:08:35 +08:00
parent b65757d8bf
commit f7783d3881
4 changed files with 49 additions and 46 deletions

View File

@@ -39,9 +39,7 @@ local client_meta = {
return "[mongo client : " .. self.host .. port_string .."]" return "[mongo client : " .. self.host .. port_string .."]"
end, end,
__gc = function(self) -- DO NOT need disconnect, because channel will shutdown during gc
self:disconnect()
end
} }
local mongo_db = {} local mongo_db = {}

View File

@@ -9,9 +9,7 @@ local redis = {}
local command = {} local command = {}
local meta = { local meta = {
__index = command, __index = command,
__gc = function(self) -- DO NOT close channel in __gc
self[1]:close()
end,
} }
---------- redis response ---------- redis response

View File

@@ -30,6 +30,11 @@ local function suspend(s)
assert(not s.co) assert(not s.co)
s.co = coroutine.running() s.co = coroutine.running()
skynet.wait() skynet.wait()
-- wakeup closing corouting every time suspend,
-- because socket.close() will wait last socket buffer operation before clear the buffer.
if s.closing then
skynet.wakeup(s.closing)
end
end end
-- read skynet_socket.h for these macro -- read skynet_socket.h for these macro
@@ -102,6 +107,7 @@ socket_message[5] = function(id)
print("socket: error on", id) print("socket: error on", id)
end end
s.connected = false s.connected = false
wakeup(s) wakeup(s)
end end
@@ -175,7 +181,8 @@ function socket.close(id)
-- notice: call socket.close in __gc should be carefully, -- notice: call socket.close in __gc should be carefully,
-- because skynet.wait never return in __gc, so driver.clear may not be called -- because skynet.wait never return in __gc, so driver.clear may not be called
if s.co then if s.co then
-- reading this socket on another coroutine -- reading this socket on another coroutine, so don't shutdown (clear the buffer) immediatel
-- wait reading coroutine read the buffer.
assert(not s.closing) assert(not s.closing)
s.closing = coroutine.running() s.closing = coroutine.running()
skynet.wait() skynet.wait()
@@ -189,13 +196,6 @@ function socket.close(id)
socket_pool[id] = nil socket_pool[id] = nil
end end
local function close_socket(s)
if s.closing then
skynet.wakeup(s.closing)
end
return driver.readall(s.buffer, buffer_pool)
end
function socket.read(id, sz) function socket.read(id, sz)
local s = socket_pool[id] local s = socket_pool[id]
assert(s) assert(s)
@@ -204,7 +204,7 @@ function socket.read(id, sz)
return ret return ret
end end
if not s.connected then if not s.connected then
return false, close_socket(s) return false, driver.readall(s.buffer, buffer_pool)
end end
assert(not s.read_required) assert(not s.read_required)
@@ -214,7 +214,7 @@ function socket.read(id, sz)
if ret then if ret then
return ret return ret
else else
return false, close_socket(s) return false, driver.readall(s.buffer, buffer_pool)
end end
end end
@@ -222,14 +222,14 @@ function socket.readall(id)
local s = socket_pool[id] local s = socket_pool[id]
assert(s) assert(s)
if not s.connected then if not s.connected then
local r = close_socket(s) local r = driver.readall(s.buffer, buffer_pool)
return r ~= "" and r return r ~= "" and r
end end
assert(not s.read_required) assert(not s.read_required)
s.read_required = true s.read_required = true
suspend(s) suspend(s)
assert(s.connected == false) assert(s.connected == false)
return close_socket(s) return driver.readall(s.buffer, buffer_pool)
end end
function socket.readline(id, sep) function socket.readline(id, sep)
@@ -241,7 +241,7 @@ function socket.readline(id, sep)
return ret return ret
end end
if not s.connected then if not s.connected then
return false, close_socket(s) return false, driver.readall(s.buffer, buffer_pool)
end end
assert(not s.read_required) assert(not s.read_required)
s.read_required = sep s.read_required = sep
@@ -249,7 +249,7 @@ function socket.readline(id, sep)
if s.connected then if s.connected then
return driver.readline(s.buffer, buffer_pool, sep) return driver.readline(s.buffer, buffer_pool, sep)
else else
return false, close_socket(s) return false, driver.readall(s.buffer, buffer_pool)
end end
end end
@@ -261,9 +261,6 @@ function socket.block(id)
assert(not s.read_required) assert(not s.read_required)
s.read_required = 0 s.read_required = 0
suspend(s) suspend(s)
if not s.connected and s.closing then
skynet.wakeup(s.closing)
end
return s.connected return s.connected
end end

View File

@@ -51,18 +51,29 @@ local function close_channel_socket(self)
end end
local function wakeup_all(self, errmsg) local function wakeup_all(self, errmsg)
for i = 1, #self.__request do if self.__response then
self.__request[i] = nil for k,co in pairs(self.__thread) do
end self.__thread[k] = nil
for i = 1, #self.__thread do self.__result[co] = socket_error
local co = self.__thread[i] self.__result_data[co] = errmsg
self.__thread[i] = nil skynet.wakeup(co)
self.__result[co] = socket_error end
self.__result_data[co] = errmsg else
skynet.wakeup(co) for i = 1, #self.__request do
self.__request[i] = nil
end
for i = 1, #self.__thread do
local co = self.__thread[i]
self.__thread[i] = nil
self.__result[co] = socket_error
self.__result_data[co] = errmsg
skynet.wakeup(co)
end
end end
end end
local function dispatch_response(self) local function dispatch_response(self)
local response = self.__response local response = self.__response
if response then if response then
@@ -85,13 +96,7 @@ local function dispatch_response(self)
if session ~= socket_error then if session ~= socket_error then
errormsg = session errormsg = session
end end
for k,co in pairs(self.__thread) do wakeup_all(self, errormsg)
-- throw error (errormsg)
self.__thread[k] = nil
self.__result[co] = socket_error
self.__result_data[co] = errormsg
skynet.wakeup(co)
end
end end
end end
else else
@@ -153,6 +158,9 @@ local function try_connect(self , once)
local t = 100 local t = 100
while not self.__closed do while not self.__closed do
if connect_once(self) then if connect_once(self) then
if not once then
print("socket: connect to", self.__host, self.__port)
end
return return
elseif once then elseif once then
error(string.format("Connect to %s:%d failed", self.__host, self.__port)) error(string.format("Connect to %s:%d failed", self.__host, self.__port))
@@ -182,6 +190,7 @@ local function block_connect(self, once)
if self.__closed then if self.__closed then
return false return false
end end
if #self.__connecting > 0 then if #self.__connecting > 0 then
-- connecting in other coroutine -- connecting in other coroutine
local co = coroutine.running() local co = coroutine.running()
@@ -190,7 +199,6 @@ local function block_connect(self, once)
-- check connection again -- check connection again
return block_connect(self, once) return block_connect(self, once)
end end
self.__connecting[1] = true self.__connecting[1] = true
try_connect(self, once) try_connect(self, once)
self.__connecting[1] = nil self.__connecting[1] = nil
@@ -216,6 +224,8 @@ function channel:request(request, response)
assert(block_connect(self)) assert(block_connect(self))
if not socket.write(self.__sock[1], request) then if not socket.write(self.__sock[1], request) then
close_channel_socket(self)
wakeup_all(self)
error(socket_error) error(socket_error)
end end
@@ -242,9 +252,9 @@ function channel:request(request, response)
self.__result_data[co] = nil self.__result_data[co] = nil
if result == socket_error then if result == socket_error then
if result_data then -- if result_data then
print("socket: dispatch", request, result_data) -- print("socket: dispatch", request, result_data)
end -- end
error(socket_error) error(socket_error)
else else
assert(result, result_data) assert(result, result_data)
@@ -269,9 +279,9 @@ function channel:response(response)
self.__result_data[co] = nil self.__result_data[co] = nil
if result == socket_error then if result == socket_error then
if result_data then -- if result_data then
print("socket: dispatch", request, result_data) -- print("socket: dispatch", request, result_data)
end -- end
error(socket_error) error(socket_error)
else else
assert(result, result_data) assert(result, result_data)