reconnect

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;
 		}
 	}

connection/lua-socket.c

@@ -10,6 +10,7 @@
 #include <arpa/inet.h>
 #include <arpa/inet.h>
 #include <errno.h>
+#include <unistd.h>
 
 #include <lua.h>
 #include <lauxlib.h>
@@ -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);

connection/main.c

@@ -5,6 +5,7 @@
 #include <sys/types.h>
 #include <sys/socket.h>
 
+#include <unistd.h>
 #include <assert.h>
 #include <stdio.h>
 #include <stdlib.h>
@@ -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 {

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

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