From e7539a07b6648a11689c14bc2a4416ca4f60e1ca Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E4=BA=91=E9=A3=8E?= Date: Wed, 15 Aug 2012 11:42:32 +0800 Subject: [PATCH] reconnect --- connection/connection.c | 2 +- connection/lua-socket.c | 52 ++++++++++++++---------- connection/main.c | 2 + lualib/socket.lua | 18 ++++----- service/redis-cli.lua | 88 ++++++++++++++++++++++++++--------------- 5 files changed, 99 insertions(+), 63 deletions(-) diff --git a/connection/connection.c b/connection/connection.c index f0f3d989..bd6f8dd1 100644 --- a/connection/connection.c +++ b/connection/connection.c @@ -68,7 +68,7 @@ _read_queue(struct connection_pool * pool, int timeout) { void * connection_poll(struct connection_pool * pool, int timeout) { if (pool->queue_head >= pool->queue_len) { - if (_read_queue(pool, timeout) == -1) { + if (_read_queue(pool, timeout) <= 0) { return NULL; } } diff --git a/connection/lua-socket.c b/connection/lua-socket.c index 18f70dc8..27592e83 100644 --- a/connection/lua-socket.c +++ b/connection/lua-socket.c @@ -10,6 +10,7 @@ #include #include #include +#include #include #include @@ -36,6 +37,13 @@ _open(lua_State *L) { return 1; } +static int +_close(lua_State *L) { + int fd = luaL_checkinteger(L,1); + close(fd); + return 0; +} + static int _write(lua_State *L) { int fd = luaL_checkinteger(L,1); @@ -101,21 +109,36 @@ _push(lua_State *L) { buf = lua_touserdata(L,2); size = luaL_checkinteger(L,3); } - buffer->read = 0; - if (size + buffer->size > buffer->cap) { + int old_size = buffer->size - buffer->read; + if (size + old_size > buffer->cap) { if (buffer->cap == 0) { lua_rawgetp(L, LUA_REGISTRYINDEX, _delete); lua_setmetatable(L, 1); } - buffer->buffer = realloc(buffer->buffer, buffer->size + size); + if (buffer->read == 0) { + buffer->buffer = realloc(buffer->buffer, size+old_size); + memcpy(buffer->buffer + old_size , buf , size); + buffer->size += size; + } else { + char * new_buffer = malloc(size + old_size); + memcpy(new_buffer, buffer->buffer + buffer->read, old_size); + memcpy(new_buffer + old_size , buf, size); + free(buffer->buffer); + buffer->buffer = new_buffer; + buffer->read = 0; + buffer->size = old_size + size; + } + } else if (buffer->size + size > buffer->cap) { + memmove(buffer->buffer, buffer->buffer + buffer->read, old_size); + memcpy(buffer->buffer + old_size, buf, size); + buffer->read = 0; + buffer->size = old_size + size; + } else { memcpy(buffer->buffer + buffer->size, buf, size); buffer->size += size; - buffer->cap = buffer->size; - return 0; } - memcpy(buffer->buffer+buffer->size, buf, size); - buffer->size += size; + return 0; } @@ -150,29 +173,16 @@ _readline(lua_State *L) { return 0; } -static int -_yield(lua_State *L) { - luaL_checktype(L,1,LUA_TUSERDATA); - struct buffer * buffer = lua_touserdata(L,1); - if (buffer->read) { - int size = buffer->size - buffer->read; - memmove(buffer->buffer, buffer->buffer + buffer->read, size); - buffer->size -= buffer->read; - buffer->read = 0; - } - return 0; -} - int luaopen_socket_c(lua_State *L) { luaL_Reg l[] = { { "open", _open }, + { "close", _close }, { "write", _write }, { "new", _new }, { "push", _push }, { "read", _read }, { "readline", _readline }, - { "yield", _yield }, { NULL, NULL }, }; luaL_checkversion(L); diff --git a/connection/main.c b/connection/main.c index 9a377e9d..d7046ff6 100644 --- a/connection/main.c +++ b/connection/main.c @@ -5,6 +5,7 @@ #include #include +#include #include #include #include @@ -114,6 +115,7 @@ _poll(struct connection_server * server) { continue; } if (size == 0) { + connection_del(server->pool, c->fd); free(buffer); skynet_send(server->ctx, NULL, c->addr, 0x7fffffff, NULL, 0, DONTCOPY); } else { diff --git a/lualib/socket.lua b/lualib/socket.lua index 79a2d151..b64e82ed 100644 --- a/lualib/socket.lua +++ b/lualib/socket.lua @@ -9,14 +9,15 @@ function socket.connect(addr) local ip, port = string.match(addr,"([^:]+):(.+)") port = tonumber(port) fd = c.open(ip,port) + if fd == nil then + return true + end skynet.send(".connection","ADD "..fd.." "..skynet.self()) object = c.new() end function socket.push(msg,sz) - if msg == nil then - socket.close() - else + if msg then c.push(object, msg, sz) end end @@ -33,13 +34,12 @@ function socket.write(...) c.write(fd, ...) end -function socket.yield() - c.yield(object) -end - function socket.close() - skynet.send(".connection","DEL "..fd) - fd = nil + if fd then + c.close(fd) + skynet.send(".connection","DEL "..fd) + fd = nil + end end return socket diff --git a/service/redis-cli.lua b/service/redis-cli.lua index b9bc498c..72752bd2 100644 --- a/service/redis-cli.lua +++ b/service/redis-cli.lua @@ -24,13 +24,6 @@ local function select_db(id) assert(result and ok == "OK") end -local function init() - socket.connect(redis_server) - if redis_db then - select_db(redis_db) - end -end - local request_queue = { head = 1, tail = 1 } local function push_request_queue(reply) @@ -51,6 +44,26 @@ local function response(...) skynet.send(reply[2],reply[1],skynet.pack(...)) end +local function readline(sep) + while true do + local line = socket.readline(sep) + if line then + return line + end + coroutine.yield() + end +end + +local function readbytes(bytes) + while true do + local block = socket.read(bytes) + if block then + return block + end + coroutine.yield() + end +end + local redcmd = {} redcmd[42] = function(data) -- '*' @@ -61,15 +74,9 @@ redcmd[42] = function(data) -- '*' end local bulk = {} for i = 1,n do - local line = socket.readline "\r\n" - if line == nil then - return "BLOCK" - end + local line = readline "\r\n" local bytes = tonumber(string.sub(line,2) + 2) - local data = socket.read(bytes) - if data == nil then - return "BLOCK" - end + local data = readbytes(bytes) table.insert(bulk, string.sub(data,1,-3)) end response(true, bulk) @@ -81,10 +88,7 @@ redcmd[36] = function(data) -- '$' response(true,nil) return end - local firstline = socket.read(bytes+2) - if firstline == nil then - return "BLOCK" - end + local firstline = readbytes(bytes+2) response(true,string.sub(firstline,1,-3)) end @@ -101,31 +105,51 @@ redcmd[58] = function(data) -- ':' end local function split_package() - local result = socket.readline "\r\n" - if result == nil then - return + while true do + local result = readline "\r\n" + local firstchar = string.byte(result) + local data = string.sub(result,2) + local f = redcmd[firstchar] + assert(f) + f(data) end - local firstchar = string.byte(result) - local data = string.sub(result,2) - local f = redcmd[firstchar] - assert(f) - if f(data) then - return +end + +local function init() + while socket.connect(redis_server) do + print("Connect failed : "..redis_server) + skynet.sleep(1000) end - socket.yield() - return true + if redis_db then + select_db(redis_db) + end +end + +local split_co = coroutine.create(split_package) + +local function reconnect() + init() + for i = request_queue.head, request_queue.tail-1 do + local request = request_queue[i] + socket.write(request[3]) + end + split_co = coroutine.create(split_package) end skynet.filter( function(session, address , msg, sz) if session == 0x7fffffff then + if msg == nil then + skynet.timeout(0, reconnect) + return + end socket.push(msg,sz) - while split_package() do end + coroutine.resume(split_co) elseif session < 0 then local message = { skynet.unpack(msg,sz) } local cmd = compose_message(message) socket.write(cmd) - push_request_queue { -session , address } + push_request_queue { -session , address, cmd } else return session, address, msg , sz end