From 255753a5774abc33a2c8ce2cb8e63cdfa720aab5 Mon Sep 17 00:00:00 2001 From: Cloud Wu Date: Sun, 23 Mar 2014 16:05:36 +0800 Subject: [PATCH 1/6] introduce socket.channel, and rewrite redis/mongo driver --- lualib-src/lua-mongo.c | 19 --- lualib/mongo.lua | 124 ++++-------------- lualib/redis.lua | 272 ++++++++++++++------------------------ lualib/socket.lua | 291 +++++++++++++++++++++++++++++++++++++++-- service/testredis.lua | 32 +++-- 5 files changed, 419 insertions(+), 319 deletions(-) diff --git a/lualib-src/lua-mongo.c b/lualib-src/lua-mongo.c index c0cde6bc..fd35fa27 100644 --- a/lualib-src/lua-mongo.c +++ b/lualib-src/lua-mongo.c @@ -521,24 +521,6 @@ reply_length(lua_State *L) { return 1; } -static int -copy_result(lua_State *L) { - if (lua_type(L,2) == LUA_TNIL) - return 0; - int n1 = lua_rawlen(L,1); - int n2 = lua_rawlen(L,2); - int i; - for (i=0;i= 0 then + local data = fd:read(bytes + 2) + -- bulk[i] = nil when bytes < 0 + bulk[i] = string.sub(data,1,-3) + end + end + return true, bulk +end + +redcmd[36] = function(fd, data) -- '$' + local bytes = tonumber(data) + if bytes < 0 then + return true,nil + end + local firstline = fd:read(bytes+2) + return true,string.sub(firstline,1,-3) +end + +redcmd[43] = function(fd, data) -- '+' + return true,data +end + +redcmd[45] = function(fd, data) -- '-' + return false,data +end + +redcmd[58] = function(fd, data) -- ':' + -- todo: return string later + return true, tonumber(data) +end + +local function read_response(fd) + local result = fd:readline "\r\n" + local firstchar = string.byte(result) + local data = string.sub(result,2) + return redcmd[firstchar](fd,data) +end +------------------- + +local function redis_login(auth, db) + if auth == nil and db == nil then + return + end + return function(so) + if auth then + so:request("AUTH "..auth.."\r\n", read_response) + end + if db then + so:request("SELECT "..db.."\r\n", read_response) + end + end +end + +function redis.connect(dbname) + local db_conf = name[dbname] + local channel = socket.channel { + host = db_conf.host, + port = db_conf.port or 6379, + auth = redis_login(db_conf.auth, db_conf.db), + } + return setmetatable( { channel }, meta ) end function command:disconnect() - socket.close(self.__handle) + self[1]:close() setmetatable(self, nil) end @@ -58,140 +117,32 @@ local function compose_message(msg) return cmd end -local redcmd = {} - -redcmd[42] = function(fd, data) -- '*' - local n = tonumber(data) - if n < 0 then - return true, nil - end - local bulk = {} - for i = 1,n do - local line = readline(fd,"\r\n") - local bytes = tonumber(string.sub(line,2)) - if bytes >= 0 then - local data = readbytes(fd, bytes + 2) - -- bulk[i] = nil when bytes < 0 - bulk[i] = string.sub(data,1,-3) - end - end - return true, bulk -end - -redcmd[36] = function(fd, data) -- '$' - local bytes = tonumber(data) - if bytes < 0 then - return true,nil - end - local firstline = readbytes(fd, bytes+2) - return true,string.sub(firstline,1,-3) -end - -redcmd[43] = function(fd, data) -- '+' - return true,data -end - -redcmd[45] = function(fd, data) -- '-' - return false,data -end - -redcmd[58] = function(fd, data) -- ':' - -- todo: return string later - return true, tonumber(data) -end - -local function read_response(fd) - local result = readline(fd, "\r\n") - local firstchar = string.byte(result) - local data = string.sub(result,2) - return redcmd[firstchar](fd,data) -end - setmetatable(command, { __index = function(t,k) local cmd = string.upper(k) local f = function (self, ...) - local fd = self.__handle - if self.__mode then - socket.write(fd, compose_message { cmd, ... }) - self.__batch = self.__batch + 1 - else - socket.lock(fd) - socket.write(fd, compose_message { cmd, ... }) - local ok, ret = read_response(fd) - socket.unlock(fd) - assert(ok, ret) - return ret - end + return self[1]:request(compose_message { cmd, ... }, read_response) end t[k] = f return f end}) +local function read_boolean(so) + local ok, result = read_response(so) + return ok, result ~= 0 +end + function command:exists(key) - assert(not self.__mode, "exists can't used in batch mode") - local fd = self.__handle - socket.lock(fd) - socket.write(fd, compose_message { "EXISTS", key }) - local ok, exists = read_response(fd) - socket.unlock(fd) - assert(ok, exists) - return exists ~= 0 + local fd = self[1] + return fd:request(compose_message { "EXISTS", key }, read_boolean) end function command:sismember(key, value) - assert(not self.__mode, "sismember can't used in batch mode") - local fd = self.__handle - socket.lock(fd) - socket.write(fd, compose_message { "SISMEMBER", key, value }) - local ok, ismember = read_response(fd) - socket.unlock(fd) - assert(ok, ismember) - return ismember ~= 0 -end - -function command:batch(mode) - if mode == "end" then - local fd = self.__handle - if self.__mode == "read" then - local allok = true - local allret = {} - for i = 1, self.__batch do - local ok, ret = read_response(fd) - allok = allok and ok - allret[i] = ret - end - self.__mode = false - socket.unlock(self.__handle) - assert(allok, "batch read failed") - return allret - else - local allok = true - for i = 1, self.__batch do - local ok = read_response(fd) - allok = allok and ok - end - self.__mode = false - socket.unlock(self.__handle) - return allok - end - else - assert(mode == "read" or mode == "write") - socket.lock(self.__handle) - self.__mode = mode - self.__batch = 0 - end -end - -function command:multi() - local fd = self.__handle - socket.lock(fd) - self.__mode = "multi" - self.__batch = 0 - socket.write(fd, "MULTI\r\n") + local fd = self[1] + return fd:request(compose_message { "SISMEMBER", key, value }, read_boolean) end local function read_exec(fd) - local result = readline(fd, "\r\n") + local result = fd:readline "\r\n" local firstchar = string.byte(result) local data = string.sub(result,2) if firstchar ~= 42 then @@ -203,57 +154,24 @@ local function read_exec(fd) local err = nil for i = 1,n do local ok, r = read_response(fd) - result[i] = r - if err then - err[i] = ok + if ok then + result[i] = r else - if ok == false then + if not err then err = {} - for j = 1, i-1 do - err[j] = true - end - err[i] = false end + table.insert(err, i .. ":" .. r) end end - return result, err + if err then + return false, table.concat(err,",") + else + return true, result + end end function command:exec() - if self.__mode ~= "multi" then - error "call multi first" - end - local fd = self.__handle - socket.write(fd, "EXEC\r\n") - local allok = true - for i = 0, self.__batch do - local ok, queue = read_response(fd) - allok = allok and ok - end - if not allok then - self.__mode = false - socket.unlock(fd) - error "Queue command error" - end - - local result, err = read_exec(fd) - - self.__mode = false - socket.unlock(fd) - - if not result then - error(err) - elseif err then - local errmsg = "" - for k,v in ipairs(err) do - if v == false then - errmsg = errmsg .. k .. ":" .. result[k] - end - end - error(errmsg) - end - - return result + return self[1]:request( "EXEC\r\n" , read_exec) end return redis diff --git a/lualib/socket.lua b/lualib/socket.lua index dd716a02..db2a9912 100644 --- a/lualib/socket.lua +++ b/lualib/socket.lua @@ -93,11 +93,14 @@ end -- SKYNET_SOCKET_TYPE_ERROR = 5 socket_message[5] = function(id) - print("error on ", id) local s = socket_pool[id] if s == nil then + print("socket: error on unknown", id) return end + if s.connected then + print("socket: error on", id) + end s.connected = false wakeup(s) end @@ -146,6 +149,18 @@ function socket.start(id, func) return connect(id, func) end +local function clear_socket(id) + local s = socket_pool[id] + if s then + if s.buffer then + driver.clear(s.buffer,buffer_pool) + end + if s.connected then + driver.close(id) + end + end +end + function socket.close(id) local s = socket_pool[id] if s == nil then @@ -163,10 +178,9 @@ function socket.close(id) else suspend(s) end + s.connected = false end - if s.buffer then - driver.clear(s.buffer,buffer_pool) - end + clear_socket(id) assert(s.lock_set == nil or next(s.lock_set) == nil) socket_pool[id] = nil end @@ -235,6 +249,17 @@ function socket.readline(id, sep) end end +function socket.block(id) + local s = socket_pool[id] + if not s or not s.connected then + return false + end + assert(not s.read_required) + s.read_required = 0 + suspend(s) + return s.connected +end + socket.write = assert(driver.send) function socket.invalid(id) @@ -251,10 +276,10 @@ function socket.lock(id) lock_set = {} s.lock = lock_set end - local co = coroutine.running() if #lock_set == 0 then - lock_set[1] = co + lock_set[1] = true else + local co = coroutine.running() table.insert(lock_set, co) skynet.wait() end @@ -263,12 +288,9 @@ end function socket.unlock(id) local s = socket_pool[id] assert(s) - local lock_set = s.lock - assert(lock_set) - local co = coroutine.running() - assert(lock_set[1] == co) + local lock_set = assert(s.lock) table.remove(lock_set,1) - co = lock_set[1] + local co = lock_set[1] if co then skynet.wakeup(co) end @@ -284,4 +306,251 @@ function socket.abandon(id) socket_pool[id] = nil end +-- channel support auto reconnect , and capture socket error in request/response transaction +-- { host = "", port = , auth = function(so) , response = function(so) session, data } + +local channel = {} +local channel_socket = {} +local channel_meta = { __index = channel } +local channel_socket_meta = { + __index = channel_socket, + __gc = function(cs) + local fd = cs[1] + cs[1] = false + if fd then + clear_socket(fd) + end + end +} +local socket_error = channel_socket -- alias for error object + +function socket.channel(desc) + local c = { + __host = assert(desc.host), + __port = assert(desc.port), + __auth = desc.auth, + __response = desc.response, + __request = {}, -- request seq { response func or session } + __thread = {}, -- coroutine seq or session->coroutine map + __result = {}, -- response result { coroutine -> result } + __result_data = {}, + __connecting = {}, + __sock = false, + __closed = false, + } + + return setmetatable(c, channel_meta) +end + +local function close_channel_socket(self) + if self.__sock then + local so = self.__sock + self.__sock = false + socket.close(so[1]) + end +end + +local function wakeup_all(self, errmsg) + 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 + +local function dispatch_response(self) + local response = self.__response + if response then + -- response() return session + while self.__sock do + local ok , session, result_ok, result_data = pcall(response, self.__sock) + if ok then + local co = self.__thread[session] + self.__thread[session] = nil + if co then + self.__result[co] = result_ok + self.__result_data[co] = result_data + skynet.wakeup(co) + else + print("socket: unknown session :", session) + end + else + close_channel_socket(self) + local errormsg + if session ~= socket_error then + errormsg = session + end + for k,co in pairs(self.__thread) do + -- throw error (errormsg) + self.__thread[k] = nil + self.__result[co] = socket_error + self.__result_data[co] = errormsg + skynet.wakeup(co) + end + end + end + else + -- pop response function from __request + while self.__sock do + local func = table.remove(self.__request, 1) + 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 = pcall(func, self.__sock) + if ok then + local co = table.remove(self.__thread, 1) + self.__result[co] = result_ok + self.__result_data[co] = result_data + skynet.wakeup(co) + else + close_channel_socket(self) + local errmsg + if result ~= socket_error then + errmsg = result_ok + end + wakeup_all(self, errmsg) + end + end + end + end +end + +local function try_connect(self) + assert(not self.__sock) + local fd = socket.open(self.__host, self.__port) + if not fd then + return + end + self.__sock = setmetatable( {fd} , channel_socket_meta ) + skynet.fork(dispatch_response, self) + + if self.__auth then + local ok , message = pcall(self.__auth, self) + if not ok then + close_channel_socket(self) + if message ~= socket_error then + error(message) + end + end + end +end + +function channel:connect() + if self.__sock then + return true + end + if self.__closed then + self.__closed = false + end + if #self.__connecting > 0 then + -- connecting in other coroutine + local co = coroutine.running() + table.insert(self.__connecting, co) + skynet.wait() + else + self.__connecting[1] = true + try_connect(self) + self.__connecting[1] = nil + for i=2, #self.__connecting do + local co = self.__connecting[i] + self.__connecting[i] = nil + skynet.wakeup(co) + end + end + + if self.__sock then + return true + else + return false + end +end + +local function reconnect_channel(self) + local t = 100 + while not self.__closed do + if self:connect() then + return + end + if t > 1000 then + print("socket: try to reconnect", self.__host, self.__port) + skynet.sleep(t) + t = 0 + else + skynet.sleep(t) + end + t = t + 100 + end +end + +function channel:request(request, response) + if not self.__sock then + assert(not self.__closed) + reconnect_channel(self) + end + + if not socket.write(self.__sock[1], request) then + return self:request(request, response) + end + + if response == nil then + -- no response + return + end + + local co = coroutine.running() + + if self.__response then + -- response is session + self.__thread[response] = co + else + -- response is a function, push it to __request + table.insert(self.__request, response) + table.insert(self.__thread, co) + end + skynet.wait() + + local result = self.__result[co] + self.__result[co] = nil + local result_data = self.__result_data[co] + self.__result_data[co] = nil + + if result == socket_error then + if result_data then + print("socket: dispatch", request, result_data) + end + return self:request(request, response) + else + assert(result, result_data) + return result_data + end +end + +function channel:close() + if not self.__closed then + self.__closed = true + close_channel_socket(self) + end +end + +channel_meta.__gc = channel.close + +local function wrapper_socket_function(f) + return function(self, ...) + local result = f(self[1], ...) + if not result then + error(socket_error) + else + return result + end + end +end + +channel_socket.read = wrapper_socket_function(socket.read) +channel_socket.readline = wrapper_socket_function(socket.readline) + return socket diff --git a/service/testredis.lua b/service/testredis.lua index 0b5c1e76..70f2e2a1 100644 --- a/service/testredis.lua +++ b/service/testredis.lua @@ -3,34 +3,32 @@ local redis = require "redis" skynet.start(function() local db = redis.connect "main" - print(db:select(0)) - db:batch "write" -- ignore results - db:del "C" - db:set("A", "hello") - db:set("B", "world") - db:sadd("C", "one") - print(db:batch "end") + db:del "C" + db:set("A", "hello") + db:set("B", "world") + db:sadd("C", "one") - db:batch "read" - db:get("A") - db:get("B") - local r = db:batch "end" -- return all results in a table - for k,v in pairs(r) do - print(k,v) - end + print(db:get("A")) + print(db:get("B")) - db:batch "write" db:del "D" for i=1,1000 do db:hset("D",i,i) end - db:batch "end" local r = db:hvals "D" for k,v in pairs(r) do print(k,v) end - +--[[ + db:multi() + db:get "A" + db:get "B" + local t = db:exec() + for k,v in ipairs(t) do + print("Exec", v) + end +]] print(db:exists "A") print(db:get "A") print(db:set("A","hello world")) From 76498502df2a49d0287c75a600692785dbaeb878 Mon Sep 17 00:00:00 2001 From: Cloud Wu Date: Sun, 23 Mar 2014 16:35:56 +0800 Subject: [PATCH 2/6] bugfix: use so:close instead of socket.close --- lualib/mongo.lua | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/lualib/mongo.lua b/lualib/mongo.lua index 00ce4858..ca117d98 100644 --- a/lualib/mongo.lua +++ b/lualib/mongo.lua @@ -105,7 +105,7 @@ function mongo_client:disconnect() if self.__sock then local so = self.__sock self.__sock = false - socket.close(so) + so:close() end end From ffbc22e088e4d917343334eecf03309debf8e695 Mon Sep 17 00:00:00 2001 From: Cloud Wu Date: Sun, 23 Mar 2014 19:46:21 +0800 Subject: [PATCH 3/6] add socket.response to support redis watch mode --- lualib/redis.lua | 139 +++++++++++++++++++++++++++++------------- lualib/socket.lua | 31 +++++++++- service/testredis.lua | 28 +++++++-- 3 files changed, 149 insertions(+), 49 deletions(-) diff --git a/lualib/redis.lua b/lualib/redis.lua index efa0cd20..7142728d 100644 --- a/lualib/redis.lua +++ b/lualib/redis.lua @@ -19,24 +19,6 @@ local meta = { ---------- redis response local redcmd = {} -redcmd[42] = function(fd, data) -- '*' - local n = tonumber(data) - if n < 0 then - return true, nil - end - local bulk = {} - for i = 1,n do - local line = fd:readline "\r\n" - local bytes = tonumber(string.sub(line,2)) - if bytes >= 0 then - local data = fd:read(bytes + 2) - -- bulk[i] = nil when bytes < 0 - bulk[i] = string.sub(data,1,-3) - end - end - return true, bulk -end - redcmd[36] = function(fd, data) -- '$' local bytes = tonumber(data) if bytes < 0 then @@ -65,6 +47,25 @@ local function read_response(fd) local data = string.sub(result,2) return redcmd[firstchar](fd,data) end + +redcmd[42] = function(fd, data) -- '*' + local n = tonumber(data) + if n < 0 then + return true, nil + end + local bulk = {} + local noerr = true + for i = 1,n do + local ok, v = read_response(fd) + if ok then + bulk[i] = v + else + noerr = false + end + end + return noerr, bulk +end + ------------------- local function redis_login(auth, db) @@ -141,37 +142,87 @@ function command:sismember(key, value) return fd:request(compose_message { "SISMEMBER", key, value }, read_boolean) end -local function read_exec(fd) - local result = fd:readline "\r\n" - local firstchar = string.byte(result) - local data = string.sub(result,2) - if firstchar ~= 42 then - return false, data - end +--- watch mode - local n = tonumber(data) - local result = {} - local err = nil - for i = 1,n do - local ok, r = read_response(fd) - if ok then - result[i] = r - else - if not err then - err = {} - end - table.insert(err, i .. ":" .. r) +local watch = {} + +local watchmeta = { + __index = watch, + __gc = function(self) + self.__sock:close() + end, +} + +local function watch_login(obj, auth) + return function(so) + if auth then + so:request("AUTH "..auth.."\r\n", read_response) + end + for k in pairs(obj.__psubscribe) do + so:request(compose_message { "PSUBSCRIBE", k }) + end + for k in pairs(obj.__subscribe) do + so:request(compose_message { "SUBSCRIBE", k }) end - end - if err then - return false, table.concat(err,",") - else - return true, result end end -function command:exec() - return self[1]:request( "EXEC\r\n" , read_exec) +function redis.watch(dbname) + local db_conf = name[dbname] + local obj = { + __subscribe = {}, + __psubscribe = {}, + } + local channel = socket.channel { + host = db_conf.host, + port = db_conf.port or 6379, + auth = watch_login(obj, db_conf.auth), + } + obj.__sock = channel + + return setmetatable( obj, watchmeta ) +end + +function watch:disconnect() + self.__sock:close() + setmetatable(self, nil) +end + +local function watch_func( name ) + local NAME = string.upper(name) + watch[name] = function(self, ...) + local so = self.__sock + for i = 1, select("#", ...) do + local v = select(i, ...) + so:request(compose_message { NAME, v }) + end + end +end + +watch_func "subscribe" +watch_func "psubscribe" +watch_func "unsubscribe" +watch_func "punsubscribe" + +function watch:message() + local so = self.__sock + while true do + local ret = so:response(read_response) + local type , channel, data , data2 = ret[1], ret[2], ret[3], ret[4] + if type == "message" then + return data, channel + elseif type == "pmessage" then + return data2, data, channel + elseif type == "subscribe" then + self.__subscribe[channel] = true + elseif type == "psubscribe" then + self.__psubscribe[channel] = true + elseif type == "unsubscribe" then + self.__subscribe[channel] = nil + elseif type == "punsubscribe" then + self.__psubscribe[channel] = nil + end + end end return redis diff --git a/lualib/socket.lua b/lualib/socket.lua index db2a9912..7f3dad67 100644 --- a/lualib/socket.lua +++ b/lualib/socket.lua @@ -366,7 +366,7 @@ local function dispatch_response(self) -- response() return session while self.__sock do local ok , session, result_ok, result_data = pcall(response, self.__sock) - if ok then + if ok and session then local co = self.__thread[session] self.__thread[session] = nil if co then @@ -530,6 +530,35 @@ function channel:request(request, response) end end +function channel:response(response) + if not self.__sock then + assert(not self.__closed) + reconnect_channel(self) + end + assert(type(response) == "function") + + local co = coroutine.running() + table.insert(self.__request, response) + table.insert(self.__thread, co) + + skynet.wait() + + local result = self.__result[co] + self.__result[co] = nil + local result_data = self.__result_data[co] + self.__result_data[co] = nil + + if result == socket_error then + if result_data then + print("socket: dispatch", request, result_data) + end + return self:response(response) + else + assert(result, result_data) + return result_data + end +end + function channel:close() if not self.__closed then self.__closed = true diff --git a/service/testredis.lua b/service/testredis.lua index 70f2e2a1..5da19729 100644 --- a/service/testredis.lua +++ b/service/testredis.lua @@ -1,8 +1,19 @@ local skynet = require "skynet" local redis = require "redis" +local function watching() + local w = redis.watch "main" + w:subscribe "foo" + w:psubscribe "hello.*" + while true do + print("Watch", w:message()) + end +end + skynet.start(function() + skynet.fork(watching) local db = redis.connect "main" + db:del "C" db:set("A", "hello") db:set("B", "world") @@ -12,7 +23,7 @@ skynet.start(function() print(db:get("B")) db:del "D" - for i=1,1000 do + for i=1,10 do db:hset("D",i,i) end local r = db:hvals "D" @@ -20,7 +31,6 @@ skynet.start(function() print(k,v) end ---[[ db:multi() db:get "A" db:get "B" @@ -28,14 +38,24 @@ skynet.start(function() for k,v in ipairs(t) do print("Exec", v) end -]] + print(db:exists "A") print(db:get "A") print(db:set("A","hello world")) print(db:get("A")) print(db:sismember("C","one")) print(db:sismember("C","two")) + + print("===========publish============") + + for i=1,10 do + db:publish("foo", i) + end + for i=11,20 do + db:publish("hello.foo", i) + end + db:disconnect() - skynet.exit() +-- skynet.exit() end) From 6c98bb2e19f66292d15dfb03d60deec378795582 Mon Sep 17 00:00:00 2001 From: Cloud Wu Date: Mon, 24 Mar 2014 11:14:39 +0800 Subject: [PATCH 4/6] bugfix: auth before connected --- lualib/redis.lua | 4 ++ lualib/socket.lua | 106 ++++++++++++++++++++++++++-------------------- redisconf | 2 +- 3 files changed, 66 insertions(+), 46 deletions(-) diff --git a/lualib/redis.lua b/lualib/redis.lua index 7142728d..1f162daa 100644 --- a/lualib/redis.lua +++ b/lualib/redis.lua @@ -89,6 +89,8 @@ function redis.connect(dbname) port = db_conf.port or 6379, auth = redis_login(db_conf.auth, db_conf.db), } + -- try connect first + channel:connect() return setmetatable( { channel }, meta ) end @@ -180,6 +182,8 @@ function redis.watch(dbname) } obj.__sock = channel + -- try connect first + channel:connect() return setmetatable( obj, watchmeta ) end diff --git a/lualib/socket.lua b/lualib/socket.lua index 7f3dad67..e67fbc12 100644 --- a/lualib/socket.lua +++ b/lualib/socket.lua @@ -337,6 +337,7 @@ function socket.channel(desc) __connecting = {}, __sock = false, __closed = false, + __authcoroutine = false, } return setmetatable(c, channel_meta) @@ -346,7 +347,8 @@ local function close_channel_socket(self) if self.__sock then local so = self.__sock self.__sock = false - socket.close(so[1]) + -- never raise error + pcall(socket.close,so[1]) end end @@ -420,12 +422,13 @@ local function dispatch_response(self) end end -local function try_connect(self) - assert(not self.__sock) +local function connect_once(self) + assert(not self.__sock and not self.__authcoroutine) local fd = socket.open(self.__host, self.__port) if not fd then - return + return false end + self.__authcoroutine = coroutine.running() self.__sock = setmetatable( {fd} , channel_socket_meta ) skynet.fork(dispatch_response, self) @@ -434,46 +437,20 @@ local function try_connect(self) if not ok then close_channel_socket(self) if message ~= socket_error then - error(message) + print("socket: auth failed", message) end end + self.__authcoroutine = false + return ok end + + return true end -function channel:connect() - if self.__sock then - return true - end - if self.__closed then - self.__closed = false - end - if #self.__connecting > 0 then - -- connecting in other coroutine - local co = coroutine.running() - table.insert(self.__connecting, co) - skynet.wait() - else - self.__connecting[1] = true - try_connect(self) - self.__connecting[1] = nil - for i=2, #self.__connecting do - local co = self.__connecting[i] - self.__connecting[i] = nil - skynet.wakeup(co) - end - end - - if self.__sock then - return true - else - return false - end -end - -local function reconnect_channel(self) +local function try_connect(self) local t = 100 while not self.__closed do - if self:connect() then + if connect_once(self) then return end if t > 1000 then @@ -487,11 +464,52 @@ local function reconnect_channel(self) end end -function channel:request(request, response) - if not self.__sock then - assert(not self.__closed) - reconnect_channel(self) +local function block_connect(self) + if self.__sock then + local authco = self.__authcoroutine + if not authco then + return true + end + if authco == coroutine.running() then + -- authing + return true + end end + if self.__closed then + return false + end + if #self.__connecting > 0 then + -- connecting in other coroutine + local co = coroutine.running() + table.insert(self.__connecting, co) + skynet.wait() + -- check connection again + return block_connect(self) + end + + self.__connecting[1] = true + try_connect(self) + self.__connecting[1] = nil + for i=2, #self.__connecting do + local co = self.__connecting[i] + self.__connecting[i] = nil + skynet.wakeup(co) + end + + -- check again + return block_connect(self) +end + +function channel:connect() + if self.__closed then + self.__closed = false + end + + return block_connect(self) +end + +function channel:request(request, response) + assert(block_connect(self)) if not socket.write(self.__sock[1], request) then return self:request(request, response) @@ -531,10 +549,8 @@ function channel:request(request, response) end function channel:response(response) - if not self.__sock then - assert(not self.__closed) - reconnect_channel(self) - end + assert(block_connect(self)) + assert(type(response) == "function") local co = coroutine.running() diff --git a/redisconf b/redisconf index 1600a38d..6cad7264 100644 --- a/redisconf +++ b/redisconf @@ -1,2 +1,2 @@ -main = { host = "127.0.0.1" , port = 6379 } +main = { host = "127.0.0.1" , port = 6379 , db = 0 } From 1052b510adce672880640c911274afd1d7547866 Mon Sep 17 00:00:00 2001 From: Cloud Wu Date: Mon, 24 Mar 2014 16:12:02 +0800 Subject: [PATCH 5/6] socketchannel is an independent mod now, channel:request will throw channel.error when disconnected. --- lualib/mongo.lua | 5 +- lualib/redis.lua | 5 +- lualib/skynet.lua | 8 +- lualib/socket.lua | 296 +------------------------------------- lualib/socketchannel.lua | 300 +++++++++++++++++++++++++++++++++++++++ 5 files changed, 312 insertions(+), 302 deletions(-) create mode 100644 lualib/socketchannel.lua diff --git a/lualib/mongo.lua b/lualib/mongo.lua index ca117d98..eb8d1c24 100644 --- a/lualib/mongo.lua +++ b/lualib/mongo.lua @@ -1,5 +1,6 @@ local bson = require "bson" local socket = require "socket" +local socketchannel = require "socketchannel" local skynet = require "skynet" local driver = require "mongo.driver" local rawget = rawget @@ -78,7 +79,7 @@ end function mongo.client( obj ) obj.port = obj.port or 27017 obj.__id = 0 - obj.__sock = socket.channel { + obj.__sock = socketchannel.channel { host = obj.host, port = obj.port, response = dispatch_reply, @@ -285,4 +286,4 @@ function mongo_cursor:close() end end -return mongo \ No newline at end of file +return mongo diff --git a/lualib/redis.lua b/lualib/redis.lua index 1f162daa..fc7f52a1 100644 --- a/lualib/redis.lua +++ b/lualib/redis.lua @@ -1,5 +1,6 @@ local skynet = require "skynet" local socket = require "socket" +local socketchannel = require "socketchannel" local config = require "config" local redis_conf = skynet.getenv "redis" local name = config (redis_conf) @@ -84,7 +85,7 @@ end function redis.connect(dbname) local db_conf = name[dbname] - local channel = socket.channel { + local channel = socketchannel.channel { host = db_conf.host, port = db_conf.port or 6379, auth = redis_login(db_conf.auth, db_conf.db), @@ -175,7 +176,7 @@ function redis.watch(dbname) __subscribe = {}, __psubscribe = {}, } - local channel = socket.channel { + local channel = socketchannel.channel { host = db_conf.host, port = db_conf.port or 6379, auth = watch_login(obj, db_conf.auth), diff --git a/lualib/skynet.lua b/lualib/skynet.lua index 4652ce3c..a96aba06 100644 --- a/lualib/skynet.lua +++ b/lualib/skynet.lua @@ -158,7 +158,7 @@ function suspend(co, result, command, param, size) end session_coroutine_id[co] = nil session_coroutine_address[co] = nil - error(debug.traceback(co,command)) + error(debug.traceback(co,tostring(command))) end if command == "CALL" then c.trace_register(trace_handle, param) @@ -444,13 +444,13 @@ local function dispatch_message(...) if not fork_succ then if succ then succ = false - err = fork_err + err = tostring(fork_err) else - err = err .. "\n" .. fork_err + err = tostring(err) .. "\n" .. tostring(fork_err) end end end - assert(succ, err) + assert(succ, tostring(err)) end function skynet.newservice(name, ...) diff --git a/lualib/socket.lua b/lualib/socket.lua index e67fbc12..24ec2fdf 100644 --- a/lualib/socket.lua +++ b/lualib/socket.lua @@ -149,7 +149,7 @@ function socket.start(id, func) return connect(id, func) end -local function clear_socket(id) +function socket.shutdown(id) local s = socket_pool[id] if s then if s.buffer then @@ -180,7 +180,7 @@ function socket.close(id) end s.connected = false end - clear_socket(id) + socket.shutdown(id) assert(s.lock_set == nil or next(s.lock_set) == nil) socket_pool[id] = nil end @@ -306,296 +306,4 @@ function socket.abandon(id) socket_pool[id] = nil end --- channel support auto reconnect , and capture socket error in request/response transaction --- { host = "", port = , auth = function(so) , response = function(so) session, data } - -local channel = {} -local channel_socket = {} -local channel_meta = { __index = channel } -local channel_socket_meta = { - __index = channel_socket, - __gc = function(cs) - local fd = cs[1] - cs[1] = false - if fd then - clear_socket(fd) - end - end -} -local socket_error = channel_socket -- alias for error object - -function socket.channel(desc) - local c = { - __host = assert(desc.host), - __port = assert(desc.port), - __auth = desc.auth, - __response = desc.response, - __request = {}, -- request seq { response func or session } - __thread = {}, -- coroutine seq or session->coroutine map - __result = {}, -- response result { coroutine -> result } - __result_data = {}, - __connecting = {}, - __sock = false, - __closed = false, - __authcoroutine = false, - } - - return setmetatable(c, channel_meta) -end - -local function close_channel_socket(self) - if self.__sock then - local so = self.__sock - self.__sock = false - -- never raise error - pcall(socket.close,so[1]) - end -end - -local function wakeup_all(self, errmsg) - 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 - -local function dispatch_response(self) - local response = self.__response - if response then - -- response() return session - while self.__sock do - local ok , session, result_ok, result_data = pcall(response, self.__sock) - if ok and session then - local co = self.__thread[session] - self.__thread[session] = nil - if co then - self.__result[co] = result_ok - self.__result_data[co] = result_data - skynet.wakeup(co) - else - print("socket: unknown session :", session) - end - else - close_channel_socket(self) - local errormsg - if session ~= socket_error then - errormsg = session - end - for k,co in pairs(self.__thread) do - -- throw error (errormsg) - self.__thread[k] = nil - self.__result[co] = socket_error - self.__result_data[co] = errormsg - skynet.wakeup(co) - end - end - end - else - -- pop response function from __request - while self.__sock do - local func = table.remove(self.__request, 1) - 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 = pcall(func, self.__sock) - if ok then - local co = table.remove(self.__thread, 1) - self.__result[co] = result_ok - self.__result_data[co] = result_data - skynet.wakeup(co) - else - close_channel_socket(self) - local errmsg - if result ~= socket_error then - errmsg = result_ok - end - wakeup_all(self, errmsg) - end - end - end - end -end - -local function connect_once(self) - assert(not self.__sock and not self.__authcoroutine) - local fd = socket.open(self.__host, self.__port) - if not fd then - return false - end - self.__authcoroutine = coroutine.running() - self.__sock = setmetatable( {fd} , channel_socket_meta ) - skynet.fork(dispatch_response, self) - - if self.__auth then - local ok , message = pcall(self.__auth, self) - if not ok then - close_channel_socket(self) - if message ~= socket_error then - print("socket: auth failed", message) - end - end - self.__authcoroutine = false - return ok - end - - return true -end - -local function try_connect(self) - local t = 100 - while not self.__closed do - if connect_once(self) then - return - end - if t > 1000 then - print("socket: try to reconnect", self.__host, self.__port) - skynet.sleep(t) - t = 0 - else - skynet.sleep(t) - end - t = t + 100 - end -end - -local function block_connect(self) - if self.__sock then - local authco = self.__authcoroutine - if not authco then - return true - end - if authco == coroutine.running() then - -- authing - return true - end - end - if self.__closed then - return false - end - if #self.__connecting > 0 then - -- connecting in other coroutine - local co = coroutine.running() - table.insert(self.__connecting, co) - skynet.wait() - -- check connection again - return block_connect(self) - end - - self.__connecting[1] = true - try_connect(self) - self.__connecting[1] = nil - for i=2, #self.__connecting do - local co = self.__connecting[i] - self.__connecting[i] = nil - skynet.wakeup(co) - end - - -- check again - return block_connect(self) -end - -function channel:connect() - if self.__closed then - self.__closed = false - end - - return block_connect(self) -end - -function channel:request(request, response) - assert(block_connect(self)) - - if not socket.write(self.__sock[1], request) then - return self:request(request, response) - end - - if response == nil then - -- no response - return - end - - local co = coroutine.running() - - if self.__response then - -- response is session - self.__thread[response] = co - else - -- response is a function, push it to __request - table.insert(self.__request, response) - table.insert(self.__thread, co) - end - skynet.wait() - - local result = self.__result[co] - self.__result[co] = nil - local result_data = self.__result_data[co] - self.__result_data[co] = nil - - if result == socket_error then - if result_data then - print("socket: dispatch", request, result_data) - end - return self:request(request, response) - else - assert(result, result_data) - return result_data - end -end - -function channel:response(response) - assert(block_connect(self)) - - assert(type(response) == "function") - - local co = coroutine.running() - table.insert(self.__request, response) - table.insert(self.__thread, co) - - skynet.wait() - - local result = self.__result[co] - self.__result[co] = nil - local result_data = self.__result_data[co] - self.__result_data[co] = nil - - if result == socket_error then - if result_data then - print("socket: dispatch", request, result_data) - end - return self:response(response) - else - assert(result, result_data) - return result_data - end -end - -function channel:close() - if not self.__closed then - self.__closed = true - close_channel_socket(self) - end -end - -channel_meta.__gc = channel.close - -local function wrapper_socket_function(f) - return function(self, ...) - local result = f(self[1], ...) - if not result then - error(socket_error) - else - return result - end - end -end - -channel_socket.read = wrapper_socket_function(socket.read) -channel_socket.readline = wrapper_socket_function(socket.readline) - return socket diff --git a/lualib/socketchannel.lua b/lualib/socketchannel.lua new file mode 100644 index 00000000..974d6dae --- /dev/null +++ b/lualib/socketchannel.lua @@ -0,0 +1,300 @@ +local skynet = require "skynet" +local socket = require "socket" + +-- channel support auto reconnect , and capture socket error in request/response transaction +-- { host = "", port = , auth = function(so) , response = function(so) session, data } + +local socket_channel = {} +local channel = {} +local channel_socket = {} +local channel_meta = { __index = channel } +local channel_socket_meta = { + __index = channel_socket, + __gc = function(cs) + local fd = cs[1] + cs[1] = false + if fd then + socket.shutdown(fd) + end + end +} + +local socket_error = setmetatable({}, {__tostring = function() return "[Error: socket]" end }) -- alias for error object +socket_channel.error = socket_error + +function socket_channel.channel(desc) + local c = { + __host = assert(desc.host), + __port = assert(desc.port), + __auth = desc.auth, + __response = desc.response, + __request = {}, -- request seq { response func or session } + __thread = {}, -- coroutine seq or session->coroutine map + __result = {}, -- response result { coroutine -> result } + __result_data = {}, + __connecting = {}, + __sock = false, + __closed = false, + __authcoroutine = false, + } + + return setmetatable(c, channel_meta) +end + +local function close_channel_socket(self) + if self.__sock then + local so = self.__sock + self.__sock = false + -- never raise error + pcall(socket.close,so[1]) + end +end + +local function wakeup_all(self, errmsg) + 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 + +local function dispatch_response(self) + local response = self.__response + if response then + -- response() return session + while self.__sock do + local ok , session, result_ok, result_data = pcall(response, self.__sock) + if ok and session then + local co = self.__thread[session] + self.__thread[session] = nil + if co then + self.__result[co] = result_ok + self.__result_data[co] = result_data + skynet.wakeup(co) + else + print("socket: unknown session :", session) + end + else + close_channel_socket(self) + local errormsg + if session ~= socket_error then + errormsg = session + end + for k,co in pairs(self.__thread) do + -- throw error (errormsg) + self.__thread[k] = nil + self.__result[co] = socket_error + self.__result_data[co] = errormsg + skynet.wakeup(co) + end + end + end + else + -- pop response function from __request + while self.__sock do + local func = table.remove(self.__request, 1) + 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 = pcall(func, self.__sock) + if ok then + local co = table.remove(self.__thread, 1) + self.__result[co] = result_ok + self.__result_data[co] = result_data + skynet.wakeup(co) + else + close_channel_socket(self) + local errmsg + if result ~= socket_error then + errmsg = result_ok + end + wakeup_all(self, errmsg) + end + end + end + end +end + +local function connect_once(self) + assert(not self.__sock and not self.__authcoroutine) + local fd = socket.open(self.__host, self.__port) + if not fd then + return false + end + self.__authcoroutine = coroutine.running() + self.__sock = setmetatable( {fd} , channel_socket_meta ) + skynet.fork(dispatch_response, self) + + if self.__auth then + local ok , message = pcall(self.__auth, self) + if not ok then + close_channel_socket(self) + if message ~= socket_error then + print("socket: auth failed", message) + end + end + self.__authcoroutine = false + return ok + end + + self.__authcoroutine = false + return true +end + +local function try_connect(self) + local t = 100 + while not self.__closed do + if connect_once(self) then + return + end + if t > 1000 then + print("socket: try to reconnect", self.__host, self.__port) + skynet.sleep(t) + t = 0 + else + skynet.sleep(t) + end + t = t + 100 + end +end + +local function block_connect(self) + if self.__sock then + local authco = self.__authcoroutine + if not authco then + return true + end + if authco == coroutine.running() then + -- authing + return true + end + end + if self.__closed then + return false + end + if #self.__connecting > 0 then + -- connecting in other coroutine + local co = coroutine.running() + table.insert(self.__connecting, co) + skynet.wait() + -- check connection again + return block_connect(self) + end + + self.__connecting[1] = true + try_connect(self) + self.__connecting[1] = nil + for i=2, #self.__connecting do + local co = self.__connecting[i] + self.__connecting[i] = nil + skynet.wakeup(co) + end + + -- check again + return block_connect(self) +end + +function channel:connect() + if self.__closed then + self.__closed = false + end + + return block_connect(self) +end + +function channel:request(request, response) + assert(block_connect(self)) + + if not socket.write(self.__sock[1], request) then + error(socket_error) + end + + if response == nil then + -- no response + return + end + + local co = coroutine.running() + + if self.__response then + -- response is session + self.__thread[response] = co + else + -- response is a function, push it to __request + table.insert(self.__request, response) + table.insert(self.__thread, co) + end + skynet.wait() + + local result = self.__result[co] + self.__result[co] = nil + local result_data = self.__result_data[co] + self.__result_data[co] = nil + + if result == socket_error then + if result_data then + print("socket: dispatch", request, result_data) + end + error(socket_error) + else + assert(result, result_data) + return result_data + end +end + +function channel:response(response) + assert(block_connect(self)) + + assert(type(response) == "function") + + local co = coroutine.running() + table.insert(self.__request, response) + table.insert(self.__thread, co) + + skynet.wait() + + local result = self.__result[co] + self.__result[co] = nil + local result_data = self.__result_data[co] + self.__result_data[co] = nil + + if result == socket_error then + if result_data then + print("socket: dispatch", request, result_data) + end + error(socket_error) + else + assert(result, result_data) + return result_data + end +end + +function channel:close() + if not self.__closed then + self.__closed = true + close_channel_socket(self) + end +end + +channel_meta.__gc = channel.close + +local function wrapper_socket_function(f) + return function(self, ...) + local result = f(self[1], ...) + if not result then + error(socket_error) + else + return result + end + end +end + +channel_socket.read = wrapper_socket_function(socket.read) +channel_socket.readline = wrapper_socket_function(socket.readline) + +return socket_channel From 11464a3373ea3e3f7fc9aee934544d4b1e1bd96e Mon Sep 17 00:00:00 2001 From: Cloud Wu Date: Mon, 24 Mar 2014 17:12:32 +0800 Subject: [PATCH 6/6] bugfix: mongo driver, reply result --- lualib/mongo.lua | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/lualib/mongo.lua b/lualib/mongo.lua index eb8d1c24..ae46fc83 100644 --- a/lualib/mongo.lua +++ b/lualib/mongo.lua @@ -67,8 +67,8 @@ local collection_meta = { local function dispatch_reply(so) local len_reply = so:read(4) local reply = so:read(driver.length(len_reply)) - local result = {} - local succ, reply_id, document, cursor_id, startfrom = driver.reply(reply, result) + local result = { result = {} } + local succ, reply_id, document, cursor_id, startfrom = driver.reply(reply, result.result) result.document = document result.cursor_id = cursor_id result.startfrom = startfrom @@ -237,8 +237,8 @@ function mongo_cursor:hasNext() if ok then if doc then - self.__document = doc - self.__data = data + self.__document = result.result + self.__data = result.data self.__ptr = 1 self.__cursor = cursor return true