From 4a0265020f211a753d06a130a0f99c78f7515b40 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?=E4=BA=91=E9=A3=8E?= <cloudwu@gmail.com>
Date: Tue, 14 Aug 2012 22:37:30 +0800
Subject: [PATCH] redesign connection module

---
 Makefile                   |  10 +-
 connection/Makefile        |  10 -
 connection/connection.c    | 378 ++-----------------------------------
 connection/connection.h    |  13 +-
 connection/lua-socket.c    | 186 ++++++++++++++++++
 connection/main.c          | 304 ++++++++++-------------------
 connection/test.c          |  63 -------
 lualib-src/lua-skynet.c    |   9 +
 lualib/skynet.lua          |   9 +-
 lualib/socket.lua          |  46 +++++
 service/redis-cli.lua      | 138 +++++++++-----
 skynet-src/skynet_harbor.c |   2 +-
 skynet-src/skynet_server.c |   3 +
 13 files changed, 466 insertions(+), 705 deletions(-)
 delete mode 100644 connection/Makefile
 create mode 100644 connection/lua-socket.c
 delete mode 100644 connection/test.c
 create mode 100644 lualib/socket.lua

diff --git a/Makefile b/Makefile
index 062ce266..cb63c9a8 100644
--- a/Makefile
+++ b/Makefile
@@ -7,9 +7,10 @@ service/logger.so \
 lualib/skynet.so \
 service/gate.so \
 service/client.so \
-service/connection.so \
 service/broker.so \
+service/connection.so \
 client \
+lualib/socket.so \
 lualib/lpeg.so \
 lualib/protobuf.so \
 lualib/int64.so \
@@ -36,7 +37,7 @@ service/snlua.so : service-src/service_lua.c
 	gcc -Wall -g -fPIC --shared $^ -o $@ -Iskynet-src
 
 service/gate.so : gate/mread.c gate/ringbuffer.c gate/main.c
-	gcc -Wall -g -fPIC --shared -o $@ $^ -I. -Igate -Iskynet-src
+	gcc -Wall -g -fPIC --shared -o $@ $^ -Igate -Iskynet-src
 
 lualib/skynet.so : lualib-src/lua-skynet.c lua-serialize/serialize.c
 	gcc -Wall -g -fPIC --shared $^ -o $@ -Iskynet-src -Ilua-serialize
@@ -45,7 +46,10 @@ service/client.so : service-src/service_client.c
 	gcc -Wall -g -fPIC --shared $^ -o $@ -Iskynet-src
 
 service/connection.so : connection/connection.c connection/main.c
-	gcc -Wall -g -fPIC --shared -o $@ $^ -I. -Iconnection -Iskynet-src
+	gcc -Wall -g -fPIC --shared -o $@ $^ -Iconnection -Iskynet-src
+
+lualib/socket.so : connection/lua-socket.c
+	gcc -Wall -g -fPIC --shared -o $@ $^ -Iconnection -Iskynet-src
 
 service/broker.so : service-src/service_broker.c
 	gcc -Wall -g -fPIC --shared $^ -o $@ -Iskynet-src
diff --git a/connection/Makefile b/connection/Makefile
deleted file mode 100644
index 437bd1a6..00000000
--- a/connection/Makefile
+++ /dev/null
@@ -1,10 +0,0 @@
-all : test 
-
-test : connection.c test.c
-	gcc -g -Wall -o $@ $^
-
-clean :
-	rm test 
-
-
-	
\ No newline at end of file
diff --git a/connection/connection.c b/connection/connection.c
index 3be7dd1e..f0f3d989 100644
--- a/connection/connection.c
+++ b/connection/connection.c
@@ -2,259 +2,55 @@
 
 #include <stdio.h>
 #include <sys/epoll.h>
-#include <sys/types.h>
-#include <sys/socket.h>
-#include <arpa/inet.h>
-#include <unistd.h>
 #include <errno.h>
-
-#include <stdio.h>
 #include <stdlib.h>
-#include <string.h>
-#include <assert.h>
+#include <unistd.h>
 
 #define EPOLLQUEUE 32
-#define SEPLEN 8
-#define DEFAULT_LINE 64
-
-struct connection {
-	int handle;
-	int fd;
-	short in_epoll;
-	short read_complete;
-	int readbuffer_cap;
-	int read_request;
-	int read_sz;
-	int yield_sz;
-	char sep[SEPLEN];
-	char * readbuffer;
-};
 
 struct connection_pool {
-	int size;
-	struct connection * conn;
 	int epoll_fd;
-	int handle_index;
 	int queue_len;
 	int queue_head;
 	struct epoll_event ev[EPOLLQUEUE];
 };
 
-struct connection_pool *
+struct connection_pool * 
 connection_newpool(int max) {
 	int epoll_fd = epoll_create(max);
 	if (epoll_fd == -1) {
 		return NULL;
 	}
 	struct connection_pool * pool = malloc(sizeof(*pool));
-	pool->size = max;
-	pool->conn = malloc(max * sizeof(struct connection));
-	memset(pool->conn, 0, max * sizeof(struct connection));
 	pool->epoll_fd = epoll_fd;
-	pool->handle_index =0;
 	pool->queue_len = 0;
 	pool->queue_head = 0;
 
 	return pool;
 }
 
-void
-connection_deletepool(struct connection_pool *pool) {
-	int i;
-	for (i=0;i<pool->size;i++) {
-		struct connection * c = &pool->conn[i];
-		if (c->handle) {
-			close(c->fd);
-		}
-		if (c->readbuffer) {
-			free(c->readbuffer);
-		}
-	}
+void 
+connection_deletepool(struct connection_pool * pool) {
 	close(pool->epoll_fd);
-	free(pool->conn);
 	free(pool);
 }
 
-static struct connection *
-_new_connection(struct connection_pool *pool) {
-	int i;
-	for (i=1;i<=pool->size;i++) {
-		int handle = pool->handle_index + i;
-		struct connection *c = &pool->conn[handle % pool->size];
-		if (c->handle == 0) {
-			c->handle = handle;
-			c->in_epoll = 0;
-			c->read_complete = 0;
-			pool->handle_index = handle;
-			return c;
-		}
-	}
-	return NULL;
-}
-
 int 
-connection_open(struct connection_pool *pool, const char * address) {
-	char * sep = strchr(address,':');
-	if (sep == NULL)
-		return 0;
-	int len = sep-address;
-	char tmp[len+1];
-	memcpy(tmp, address, len);
-	tmp[len] = '\0';
-	int port = strtol(sep+1, NULL, 10);
+connection_add(struct connection_pool * pool, int fd, void *ud) {
+	struct epoll_event ev;
+	ev.events = EPOLLIN;
+	ev.data.ptr = ud;
 
-	struct sockaddr_in my_addr;
-	memset(&my_addr, 0, sizeof(struct sockaddr_in));
-	my_addr.sin_family = AF_INET;
-	my_addr.sin_port = htons(port);
-	my_addr.sin_addr.s_addr=inet_addr(tmp);
-
-	int fd = socket(AF_INET,SOCK_STREAM,0);
-	int r = connect(fd,(struct sockaddr *)&my_addr,sizeof(struct sockaddr_in));
-
-	if (r == -1) {
-		return 0;
+	if (epoll_ctl(pool->epoll_fd, EPOLL_CTL_ADD, fd, &ev) == -1) {
+		return 1;
 	}
-
-	struct connection * c = _new_connection(pool);
-	if (c == NULL) {
-		close(fd);
-		return 0;
-	}
-	c->fd = fd;
-	return c->handle;
-}
-
-static void
-_remove_connection(struct connection_pool * pool, struct connection *c) {
-	c->handle = 0;
-	if (c->in_epoll) {
-		epoll_ctl(pool->epoll_fd, EPOLL_CTL_DEL, c->fd , NULL);
-	}
-	close(c->fd);
-}
-
-struct connection *
-_get_connection(struct connection_pool *pool, int handle) {
-	assert(handle != 0);
-	struct connection * c = &pool->conn[handle % pool->size];
-	if (c->handle != handle)
-		return NULL;
-	return c;
+	return 0;
 }
 
 void 
-connection_close(struct connection_pool * pool, int handle) {
-	struct connection * c = _get_connection(pool, handle);
-	if (c) {
-		_remove_connection(pool, c);
-	}
-}
-
-static void
-_add_epoll(struct connection_pool * pool, struct connection *c) {
-	if (!c->in_epoll) {
-		struct epoll_event ev;
-		ev.events = EPOLLIN;
-		ev.data.ptr = c;
-		if (epoll_ctl(pool->epoll_fd, EPOLL_CTL_ADD, c->fd, &ev) == -1) {
-			close(c->fd);
-			c->handle = 0;
-			return;
-		}
-		c->in_epoll = 1;
-	}
-}
-
-static void
-_yield(struct connection *c) {
-	c->read_complete = 0;
-	int len = c->read_sz - c->yield_sz;
-	if (len == 0) {
-		c->read_sz = 0;
-		return;
-	}
-	assert(len > 0);
-	memmove(c->readbuffer, c->readbuffer + c->yield_sz, len);
-	c->read_sz = len;
-}
-
-void * 
-connection_read(struct connection_pool * pool, int handle, size_t sz) {
-	struct connection * c = _get_connection(pool, handle);
-	if (c == NULL)
-		return NULL;
-	if (c->read_complete) {
-		_yield(c);
-	}
-	if (c->readbuffer_cap < sz) {
-		int cap = DEFAULT_LINE;
-		while (cap <= sz) {
-			cap *=2;
-		}
-		c->readbuffer = realloc(c->readbuffer, cap);
-		c->readbuffer_cap = cap;
-	}
-
-	if (c->read_sz >= sz) {
-		c->yield_sz = sz;
-		c->read_complete = 1;
-		c->read_request = 0;
-		return c->readbuffer;
-	}
-
-	c->read_request = sz;
-	_add_epoll(pool,c);
-
-	return NULL;
-}
-
-static const char *
-_try_readline(struct connection_pool * pool, struct connection *c) {
-	int seplen = strlen(c->sep);
-	int i;
-	for (i=0;i<= c->read_sz - seplen;i++) {
-		if (memcmp(c->readbuffer+i,c->sep,seplen) == 0) {
-			c->readbuffer[i] = '\0';
-			c->read_request = i;
-			c->yield_sz = i + seplen;
-			c->read_complete = 1;
-			return c->readbuffer;
-		}
-	}
-	return NULL;
-}
-
-void * 
-connection_readline(struct connection_pool * pool, int handle, const char * sep, size_t *sz) {
-	struct connection * c = _get_connection(pool, handle);
-	if (c == NULL)
-		return NULL;
-
-	if (c->read_complete) {
-		_yield(c);
-	}
-	int i;
-	for (i=0;i<SEPLEN;i++) {
-		c->sep[i] = sep[i];
-		if (sep[i] == '\0')
-			break;
-	}
-	assert(i < SEPLEN);
-
-	const char * line = _try_readline(pool,c);
-	if (line) {
-		if (sz) {
-			*sz = c->read_request;
-		}
-		c->read_request = 0;
-		return c->readbuffer;
-	}
-
-	c->read_request = -1;
-	_add_epoll(pool,c);
-
-	return NULL;
+connection_del(struct connection_pool * pool, int fd) {
+	epoll_ctl(pool->epoll_fd, EPOLL_CTL_DEL, fd , NULL);
+	close(fd);
 }
 
 static int
@@ -269,155 +65,13 @@ _read_queue(struct connection_pool * pool, int timeout) {
 	return n;
 }
 
-inline static struct connection *
-_read_one(struct connection_pool * pool) {
-	if (pool->queue_head >= pool->queue_len) {
-		return NULL;
-	}
-	return pool->ev[pool->queue_head ++].data.ptr;
-}
-
-static int
-_read_buffer(struct connection_pool * pool, struct connection *c) {
-	int bytes = recv(c->fd, c->readbuffer + c->read_sz, c->readbuffer_cap - c->read_sz, MSG_DONTWAIT);
-	if (bytes > 0) {
-		c->read_sz += bytes;
-	}
-	if (bytes == 0) {
-		// closed
-		_remove_connection(pool, c);
-	}
-
-	return bytes;
-}
-
-static int
-_read_block(struct connection_pool * pool, struct connection *c, size_t * id) {
-	assert(c->read_complete == 0);
-	int need_read = c->read_request - c->read_sz;
-	if (need_read <= 0) {
-		c->yield_sz = c->read_request;
-		c->read_complete = 1;
-		return 1;
-	}
-
-	int handle = c->handle;
-	int bytes = _read_buffer(pool,c);
-	if (bytes > 0) {
-		if (c->read_request - c->read_sz <= 0) {
-			c->yield_sz = c->read_request;
-			c->read_complete = 1;
-			return 1;
-		} 
-	} else if (bytes == 0) {
-		if (id) {
-			*id = c - pool->conn;
-		}
-		return -handle;
-	}
-
-	return 0;
-}
-
-static int
-_read_line(struct connection_pool * pool, struct connection *c, size_t * id) {
-	assert(c->read_complete == 0);
-	if (c->readbuffer_cap == c->read_sz) {
-		int cap = DEFAULT_LINE;
-		while (cap <= c->read_sz) {
-			cap *=2;
-		}
-		c->readbuffer = realloc(c->readbuffer, cap);
-		c->readbuffer_cap = cap;
-	}
-	int handle = c->handle;
-	int bytes = _read_buffer(pool,c);
-	if (bytes == 0) {
-		if (id) {
-			*id = c - pool->conn;
-		}
-		return -handle;	// close
-	}
-	const char * line = _try_readline(pool,c);
-	if (line)
-		return 1;
-	return 0;
-}
-
-static int
-_read_request(struct connection_pool * pool, struct connection *c, size_t * id) {
-	if (c->read_request < 0) {
-		return _read_line(pool, c, id);
-	} else {
-		return _read_block(pool, c, id);
-	}
-}
-
 void * 
-connection_poll(struct connection_pool * pool, int timeout, int *phandle, size_t * id) {
+connection_poll(struct connection_pool * pool, int timeout) {
 	if (pool->queue_head >= pool->queue_len) {
 		if (_read_queue(pool, timeout) == -1) {
 			return NULL;
 		}
 	}
-
-	for (;;) {
-		struct connection * c = _read_one(pool);
-		if (c == NULL) {
-			*phandle = 0;
-			return NULL;
-		}
-		if (c->read_complete) {
-			assert(c->in_epoll);
-			epoll_ctl(pool->epoll_fd, EPOLL_CTL_DEL, c->fd , NULL);
-			c->in_epoll = 0;
-			continue;
-		}
-		int result = _read_request(pool, c, id);
-		switch (result) {
-		case 0:
-			// block
-			*phandle = 0;
-			break;
-		case 1:
-			*phandle = c->handle;
-			if (id) {
-				*id = c->read_request;
-			}
-			c->read_request = 0;
-			return c->readbuffer;
-		default:
-			// nagetive close
-			*phandle = -result;
-			return NULL;
-		}
-	}
+	return pool->ev[pool->queue_head ++].data.ptr;
 }
 
-void 
-connection_write(struct connection_pool * pool, int handle, const void * buffer, size_t sz) {
-	struct connection * c = _get_connection(pool, handle);
-	if (c) {
-		for (;;) {
-			int err = send(c->fd, buffer, sz, 0);
-			if (err < 0) {
-				switch (errno) {
-				case EAGAIN:
-				case EINTR:
-					continue;
-				}
-			}
-			assert(err == sz);
-			return;
-		}
-	}
-}
-
-int 
-connection_id(struct connection_pool * pool, int handle) {
-	struct connection * c = _get_connection(pool, handle);
-	if (c == NULL) {
-		return 0;
-	}
-	return c - pool->conn + 1;
-}
diff --git a/connection/connection.h b/connection/connection.h
index df13e1b3..d9236174 100644
--- a/connection/connection.h
+++ b/connection/connection.h
@@ -8,16 +8,9 @@ struct connection_pool;
 struct connection_pool * connection_newpool(int max);
 void connection_deletepool(struct connection_pool *);
 
-int connection_open(struct connection_pool *, const char * ipaddr);
-void connection_close(struct connection_pool *, int handle);
+int connection_add(struct connection_pool *, int fd, void *ud);
+void connection_del(struct connection_pool *, int fd);
 
-void * connection_read(struct connection_pool *, int handle, size_t sz);
-void * connection_readline(struct connection_pool *, int handle, const char * sep, size_t *sz);
-
-void * connection_poll(struct connection_pool *, int timeout, int *handle, size_t *sz);
-int connection_closed(struct connection_pool *, int handle);
-
-void connection_write(struct connection_pool *, int handle, const void * buffer, size_t sz);
-int connection_id(struct connection_pool *, int handle);
+void * connection_poll(struct connection_pool *, int timeout);
 
 #endif
diff --git a/connection/lua-socket.c b/connection/lua-socket.c
new file mode 100644
index 00000000..18f70dc8
--- /dev/null
+++ b/connection/lua-socket.c
@@ -0,0 +1,186 @@
+#include "skynet.h"
+
+#include <stdio.h>
+#include <stdlib.h>
+#include <string.h>
+#include <assert.h>
+
+#include <sys/types.h>
+#include <sys/socket.h>
+#include <arpa/inet.h>
+#include <arpa/inet.h>
+#include <errno.h>
+
+#include <lua.h>
+#include <lauxlib.h>
+
+static int
+_open(lua_State *L) {
+	const char * ip = luaL_checkstring(L,1);
+	int port = luaL_checkinteger(L,2);
+
+	struct sockaddr_in my_addr;
+	memset(&my_addr, 0, sizeof(struct sockaddr_in));
+	my_addr.sin_family = AF_INET;
+	my_addr.sin_port = htons(port);
+	my_addr.sin_addr.s_addr=inet_addr(ip);
+
+	int fd = socket(AF_INET,SOCK_STREAM,0);
+	int r = connect(fd,(struct sockaddr *)&my_addr,sizeof(struct sockaddr_in));
+
+	if (r == -1) {
+		return 0;
+	}
+
+	lua_pushinteger(L,fd);
+	return 1;
+}
+
+static int
+_write(lua_State *L) {
+	int fd = luaL_checkinteger(L,1);
+	int type = lua_type(L,2);
+	const char * buffer;
+	size_t sz;
+	if (type == LUA_TSTRING) {
+		buffer = lua_tolstring(L,2,&sz);
+	} else if (type == LUA_TLIGHTUSERDATA) {
+		buffer = lua_touserdata(L,2);
+		sz = luaL_checkinteger(L,3);
+	}
+	for (;;) {
+		int err = send(fd, buffer, sz, 0);
+		if (err < 0) {
+			switch (errno) {
+			case EAGAIN:
+			case EINTR:
+				continue;
+			}
+		}
+		assert(err == sz);
+		return 0;
+	}
+}
+
+struct buffer {
+	int cap;
+	int size;
+	char * buffer;
+	int read;
+};
+
+static int
+_new(lua_State *L) {
+	struct buffer * buffer = lua_newuserdata(L, sizeof(struct buffer));
+	buffer->cap = 0;
+	buffer->size = 0;
+	buffer->buffer = NULL;
+	buffer->read = 0;
+	return 1;
+}
+
+static int
+_delete(lua_State *L) {
+	struct buffer * buffer = lua_touserdata(L,1);
+	free(buffer->buffer);
+	buffer->buffer = NULL;
+	return 0;
+}
+
+static int
+_push(lua_State *L) {
+	luaL_checktype(L,1,LUA_TUSERDATA);
+	struct buffer * buffer = lua_touserdata(L,1);
+	int type = lua_type(L,2);
+	const char * buf;
+	size_t size;
+	if (type == LUA_TSTRING) {
+		buf = lua_tolstring(L,2,&size);
+	} else {
+		luaL_checktype(L,2,LUA_TLIGHTUSERDATA);
+		buf = lua_touserdata(L,2);
+		size = luaL_checkinteger(L,3);
+	}
+	buffer->read = 0;
+	if (size + buffer->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);
+		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;
+}
+
+static int
+_read(lua_State *L) {
+	luaL_checktype(L,1,LUA_TUSERDATA);
+	struct buffer * buffer = lua_touserdata(L,1);
+	int need = luaL_checkinteger(L,2);
+	int size = buffer->size - buffer->read;
+	if (need <= size) {
+		lua_pushlstring(L, buffer->buffer + buffer->read, need);
+		buffer->read += need;
+		return 1;
+	}
+	return 0;
+}
+
+static int
+_readline(lua_State *L) {
+	luaL_checktype(L,1,LUA_TUSERDATA);
+	struct buffer * buffer = lua_touserdata(L,1);
+	size_t sz;
+	const char * sep = luaL_checklstring(L,2,&sz);
+	int i;
+	for (i=buffer->read;i<=buffer->size - (int)sz;i++) {
+		if (memcmp(buffer->buffer+i,sep,sz) == 0) {
+			lua_pushlstring(L, buffer->buffer + buffer->read, i-buffer->read);
+			buffer->read = i+sz;
+			return 1;
+		}
+	}
+	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 },
+		{ "write", _write },
+		{ "new", _new },
+		{ "push", _push },
+		{ "read", _read },
+		{ "readline", _readline },
+		{ "yield", _yield },
+		{ NULL, NULL },
+	};
+	luaL_checkversion(L);
+	luaL_newlib(L,l);
+	lua_createtable(L,0,1);
+	lua_pushcfunction(L, _delete);
+	lua_setfield(L, -2, "__gc");
+	lua_rawsetp(L, LUA_REGISTRYINDEX, _delete);
+	return 1;
+}
+
diff --git a/connection/main.c b/connection/main.c
index 617dac3a..aa355ccf 100644
--- a/connection/main.c
+++ b/connection/main.c
@@ -1,30 +1,30 @@
 #include "connection.h"
 #include "skynet.h"
 
+
+#include <sys/types.h>
+#include <sys/socket.h>
+
 #include <assert.h>
 #include <stdio.h>
 #include <stdlib.h>
 #include <stdint.h>
 #include <string.h>
 
-struct reply {
-	int session;
-	uint32_t dest;
+#define DEFAULT_BUFFER_SIZE 1024
+#define DEFAULT_CONNECTION 16
+
+struct connection {
+	int fd;
+	char * addr;
 };
 
 struct connection_server {
-	int poll;
 	int max_connection;
+	int current_connection;
 	struct connection_pool *pool;
 	struct skynet_context *ctx;
-	struct reply * reply;
-};
-
-typedef void (*command_func)(struct connection_server *server, const char * param, size_t sz, int session, const char * reply);
-
-struct command {
-	const char * name;
-	command_func func;
+	struct connection * conn;
 };
 
 struct connection_server *
@@ -39,183 +39,86 @@ connection_release(struct connection_server * server) {
 	if (server->pool) {
 		connection_deletepool(server->pool);
 	}
-	free(server->reply);
+	int i;
+	for (i=0;i<server->max_connection;i++) {
+		free(server->conn[i].addr);
+	}
+	free(server->conn);
 	free(server);
 }
 
-static inline const char * 
-_command(const char * cmd, const char * msg, size_t sz) {
+static void
+_expand(struct connection_server * server) {
+	connection_deletepool(server->pool);
+	server->pool = connection_newpool(server->max_connection * 2);
 	int i;
-	for (i=0;i<sz;i++) {
-		if (cmd[i] == '\0') {
-			if (msg[i] != ' ')
-				return NULL;
-			return msg+i+1;
-		}
-		if (cmd[i] != msg[i])
-			return NULL;
+	for (i=0;i<server->max_connection;i++) {
+		struct connection * c = &server->conn[i];
+		connection_add(server->pool, c->fd , c);
 	}
-	return NULL;
+	server->max_connection *= 2;
 }
 
 static void
-_connect(struct connection_server *server, const char * ipaddr, size_t sz, int session, const char * reply) {
-	char tmp[sz+1];
-	memcpy(tmp, ipaddr, sz);
-	tmp[sz] = '\0';
-	int id = connection_open(server->pool, ipaddr);
-	if (id == 0) {
-		skynet_send(server->ctx, NULL, reply, session, NULL, 0, 0);
-		return;
+_add(struct connection_server * server, int fd , char * addr) {
+	++server->current_connection;
+	if (server->current_connection > server->max_connection) {
+		_expand(server);
 	}
-	char idstring[20];
-	int n = sprintf(idstring, "%d", id);
-	skynet_send(server->ctx, NULL, reply, session, idstring, n, 0);
-}
-
-static void
-_close(struct connection_server *server, const char * param, size_t sz, int session, const char * reply) {
-	int handle = strtol(param, NULL, 10);
-	if (handle <= 0) {
-		skynet_error(server->ctx, "[connection] Close invalid handle from %s", reply);
-		return;
-	}
-	connection_close(server->pool, handle);
-}
-
-static void
-_write(struct connection_server *server, const char * param, size_t sz, int session, const char * reply) {
-	char * endptr = NULL;
-	int handle = strtol(param, &endptr, 10);
-	if (handle <= 0 || endptr == NULL || *endptr !=' ') {
-		skynet_error(server->ctx, "[connection] Write invalid handle from %s", reply);
-		return;
-	}
-	++endptr;
-	sz -= endptr - param;
-	connection_write(server->pool, handle, endptr, sz);
-}
-
-static void
-_read(struct connection_server *server, const char * param, size_t sz, int session, const char * reply) {
-	char * endptr = NULL;
-	int handle = strtol(param, &endptr, 10);
-	if (handle <= 0 || endptr == NULL || *endptr !=' ') {
-		skynet_error(server->ctx, "[connection] Read invalid handle from %s", reply);
-		return;
-	}
-	int size = strtol(endptr+1, &endptr, 10);
-
-	if (size <= 0 || endptr == NULL) {
-		skynet_error(server->ctx, "[connection] Read invalid size (%d) from %s", size, reply);
-		return;
-	}
-	void * buffer = connection_read(server->pool, handle, size);
-	if (buffer == NULL) {
-		int id = connection_id(server->pool, handle);
-		if (id == 0) {
-			skynet_send(server->ctx, NULL, reply, session, NULL, 0, 0);
-			return;
-		}
-		--id;
-		assert(id < server->max_connection);
-		assert(reply[0] == ':');
-		server->reply[id].session = session;
-		server->reply[id].dest = strtoul(reply+1, NULL, 16); 
-		++server->poll;
-		if (server->poll == 1) {
-			skynet_command(server->ctx, "TIMEOUT","0");
-			return;
-		}
-	} else {
-		skynet_send(server->ctx, NULL, reply, session, buffer, size, 0);
-	}
-}
-
-static void
-_readline(struct connection_server *server, const char * param, size_t sz, int session, const char * reply) {
-	char * endptr = NULL;
-	int handle = strtol(param, &endptr, 10);
-	if (handle <= 0 || endptr == NULL || *endptr !=' ') {
-		skynet_error(server->ctx, "[connection] Readline invalid handle from %s", reply);
-		return;
-	}
-
-	sz -= endptr - param + 1;
-	if (sz < 1 || sz > 7) {
-		skynet_error(server->ctx, "[connection] Readline invalid sep (size = %d) from %s", sz, reply);
-		return;
-	}
-
-	char sep[sz+1];
-	memcpy(sep, endptr+1, sz);
-	sep[sz] = '\0';
-
-	void * buffer = connection_readline(server->pool, handle, sep, &sz);
-	if (buffer == NULL) {
-		int id = connection_id(server->pool, handle);
-		if (id == 0) {
-			skynet_send(server->ctx, NULL, reply, session, NULL, 0, 0);
-			return;
-		}
-		--id;
-		assert(id < server->max_connection);
-		assert(reply[0] == ':');
-		server->reply[id].session = session;
-		server->reply[id].dest = strtoul(reply+1, NULL, 16); 
-		++server->poll;
-		if (server->poll == 1) {
-			skynet_command(server->ctx, "TIMEOUT","0");
-			return;
-		}
-	} else {
-		skynet_send(server->ctx, NULL, reply, session, buffer, sz, 0);
-	}
-}
-
-static void
-_id_to_hex(char * str, uint32_t id) {
 	int i;
-	static char hex[16] = { '0','1','2','3','4','5','6','7','8','9','A','B','C','D','E','F' };
-	for (i=0;i<8;i++) {
-		str[i] = hex[(id >> ((7-i) * 4))&0xf];
+	for (i=0;i<server->max_connection;i++) {
+		struct connection * c = &server->conn[i];
+		if (c->addr == NULL) {
+			c->fd = fd;
+			c->addr = addr;
+			int err = connection_add(server->pool, fd , c);
+			assert(err == 0);
+			return;
+		}
 	}
-	str[8] = '\0';
+	assert(0);
 }
 
 static void
-_poll(struct connection_server *server) {
-	int handle = 0;
-	size_t sz = 0;
+_del(struct connection_server * server, int fd) {
+	int i;
+	for (i=0;i<server->max_connection;i++) {
+		struct connection * c = &server->conn[i];
+		if (c->fd == fd) {
+			free(c->addr);
+			c->addr = NULL;
+			c->fd = 0;
+			connection_del(server->pool, fd);
+			return;
+		}
+	}
+
+	skynet_error(server->ctx, "[connection] Delete invalid handle %d", fd);
+}
+
+static void
+_poll(struct connection_server * server) {
 	int timeout = 100;
-	char addr[10];
 	for (;;) {
-		void * buffer = connection_poll(server->pool, timeout, &handle, &sz);
+		struct connection * c = connection_poll(server->pool, timeout);
+		if (c==NULL) {
+			skynet_command(server->ctx,"TIMEOUT","1");
+			return;
+		}
 		timeout = 0;
-		if (buffer == NULL) {
-			if (handle) {
-				--server->poll;
-				int id = sz;
-				struct reply * r = &server->reply[id];
-				addr[0] = ':';
-				_id_to_hex(addr+1, id);
-				skynet_send(server->ctx, NULL, addr , r->session, NULL, 0, 0);
-			} else {
-				assert(server->poll >= 0);
-				if (server->poll > 0) {
-					skynet_command(server->ctx, "TIMEOUT","1");
-				}
-				return;
-			}
+
+		void * buffer = malloc(DEFAULT_BUFFER_SIZE);
+
+		int size = recv(c->fd, buffer, DEFAULT_BUFFER_SIZE, MSG_DONTWAIT);
+		if (size < 0) {
+			continue;
+		}
+		if (size == 0) {
+			printf("disconnect\n");
+			free(buffer);
+			skynet_send(server->ctx, NULL, c->addr, 0x7fffffff, NULL, 0, DONTCOPY);
 		} else {
-			--server->poll;
-			int id = connection_id(server->pool, handle);
-			assert(id > 0);
-			--id;
-			struct reply * r = &server->reply[id];
-			addr[0] = ':';
-			_id_to_hex(addr+1, r->dest);
-			skynet_send(server->ctx, NULL, addr, r->session, buffer, sz, 0);
+			skynet_send(server->ctx, NULL, c->addr, 0x7fffffff, buffer, size, DONTCOPY);
 		}
 	}
 }
@@ -223,52 +126,49 @@ _poll(struct connection_server *server) {
 static void
 _main(struct skynet_context * ctx, void * ud, int session, const char * uid, const void * msg, size_t sz) {
 	if (msg == NULL) {
-		assert(session >= 0);
 		_poll(ud);
 		return;
 	}
-	if (session > 0) {
-		skynet_error(ctx, "[connection] Invalid response (session = %d) from %s", session, uid);
-		return;
-	}
-	struct command cmd[] = {
-		{ "CONNECT", _connect },
-		{ "CLOSE" , _close },
-		{ "READ" , _read },
-		{ "READLINE", _readline },
-		{ "WRITE", _write },
-		{ NULL, NULL },
-	};
-
-	struct command * p = cmd;
-	while (p->name) {
-		const char * param = _command(p->name, msg, sz);
-		if (param) {
-			p->func(ud, param, sz - (param-(const char *)msg), -session, uid);
+	const char * param = (const char *)msg + 4;
+	if (memcmp(msg, "ADD ", 4)==0) {
+		char * endptr;
+		int fd = strtol(param, &endptr, 10);
+		if (endptr == NULL) {
+			skynet_error(ctx, "[connection] Invalid ADD command from %s (session = %d)", uid, session);
 			return;
 		}
-		++p;
+		int addr_sz = sz - (endptr - (char *)msg);
+		char * addr = malloc(addr_sz);
+		memcpy(addr, endptr+1, addr_sz-1);
+		addr[addr_sz] = '\0';
+		_add(ud, fd, addr);
+	} else if (memcmp(msg, "DEL ", 4)==0) {
+		char * endptr;
+		int fd = strtol(param, &endptr, 10);
+		if (endptr == NULL) {
+			skynet_error(ctx, "[connection] Invalid DEL command from %s (session = %d)", uid, session);
+			return;
+		}
+		_del(ud, fd);
+	} else {
+		skynet_error(ctx, "[connection] Invalid command from %s (session = %d)", uid, session);
 	}
-
-	skynet_error(ctx, "[connection] Invalid command from %s (session = %d)", uid, -session);
 }
 
 int
 connection_init(struct connection_server * server, struct skynet_context * ctx, char * param) {
-	int max = strtol(param, NULL, 10);
-	if (max <=0) {
-		return 1;
-	}
-	server->pool = connection_newpool(max);
+	server->pool = connection_newpool(DEFAULT_CONNECTION);
 	if (server->pool == NULL)
 		return 1;
-	server->max_connection = max;
-	server->reply = malloc(sizeof(struct reply) * max);
-	memset(server->reply, 0, sizeof(struct reply) * max);
+	server->max_connection = DEFAULT_CONNECTION;
+	server->current_connection = 0;
 	server->ctx = ctx;
-	server->poll = 0;
+	server->conn = malloc(server->max_connection * sizeof(struct connection));
+	memset(server->conn, 0, server->max_connection * sizeof(struct connection));
+
 	skynet_callback(ctx, server, _main);
 	skynet_command(ctx,"REG",".connection");
+	skynet_command(ctx,"TIMEOUT","0");
 	return 0;
 }
 
diff --git a/connection/test.c b/connection/test.c
deleted file mode 100644
index 8d325efc..00000000
--- a/connection/test.c
+++ /dev/null
@@ -1,63 +0,0 @@
-#include "connection.h"
-
-#include <stdio.h>
-#include <stdint.h>
-
-static void
-test(struct connection_pool *p, int id) {
-	int i=0;
-	while (i<5) {
-		int handle = id;
-		const char * line = connection_readline(p, handle, "\n",  NULL);
-		if (line == NULL) {
-			line = connection_poll(p,1000,&handle,NULL);
-		}
-		if (line) {
-			printf("%d %d: %s\n",i,handle, line);
-			connection_write(p,handle,"readline\n",9);
-			++i;
-		} else {
-			if (handle) {
-				printf("Close %d\n",handle);
-				return;
-			}
-		}
-	}
-	i=0;
-	while (i<5) {
-		int handle = id;
-		uint8_t * buffer = connection_read(p, handle, 8);
-		if (buffer == NULL) {
-			buffer = connection_poll(p,1000,&handle,NULL);
-		}
-		if (buffer) {
-			int j;
-			printf("%d %d: ",i,handle);
-			for (j=0;j<8;j++) {
-				printf("%02x ",buffer[j]);
-			}
-			printf("\n");
-			connection_write(p,handle,"readblock\n",10);
-			++i;
-		} else {
-			if (handle) {
-				printf("Close %d\n", handle);
-				return;
-			}
-		}
-	}
-}
-
-int
-main() {
-	struct connection_pool * p = connection_newpool(16);
-
-	int handle = connection_open(p, "127.0.0.1:8888");
-
-	test(p,handle);
-
-	connection_deletepool(p);
-	
-	return 0;
-};
-
diff --git a/lualib-src/lua-skynet.c b/lualib-src/lua-skynet.c
index 41cf5731..0454c5a6 100644
--- a/lualib-src/lua-skynet.c
+++ b/lualib-src/lua-skynet.c
@@ -97,6 +97,14 @@ _command(lua_State *L) {
 	return 0;
 }
 
+static int
+_genid(lua_State *L) {
+	struct skynet_context * context = lua_touserdata(L, lua_upvalueindex(1));
+	int session = skynet_send(context, NULL, NULL, -1, NULL, 0 , 0);
+	lua_pushinteger(L, session);
+	return 1;
+}
+
 static int
 _send(lua_State *L) {
 	struct skynet_context * context = lua_touserdata(L, lua_upvalueindex(1));
@@ -183,6 +191,7 @@ luaopen_skynet_c(lua_State *L) {
 	luaL_checkversion(L);
 	luaL_Reg l[] = {
 		{ "send" , _send },
+		{ "genid", _genid },
 		{ "redirect", _redirect },
 		{ "command" , _command },
 		{ "callback" , _callback },
diff --git a/lualib/skynet.lua b/lualib/skynet.lua
index 237f9faf..ba52bab8 100644
--- a/lualib/skynet.lua
+++ b/lualib/skynet.lua
@@ -18,10 +18,11 @@ local function suspend(co, result, command, param, size)
 		local co_address = session_coroutine_address[co]
 		c.send(co_address, co_session, param, size)
 		return suspend(co, coroutine.resume(co))
-	else
-		assert(command == nil, command)
+	elseif command == nil then
 		session_coroutine_id[co] = nil
 		session_coroutine_address[co] = nil
+	else
+		error("Unknown command : " .. command)
 	end
 end
 
@@ -99,6 +100,7 @@ function skynet.setenv(key, value)
 end
 
 skynet.send = assert(c.send)
+skynet.genid = assert(c.genid)
 skynet.redirect = assert(c.redirect)
 skynet.pack = assert(c.pack)
 skynet.tostring = assert(c.tostring)
@@ -126,6 +128,9 @@ end
 
 local function default_dispatch(f)
 	return function(session, address , msg, sz)
+		if session == nil then
+			return
+		end
 		if session <= 0 then
 			session = - session
 			co = coroutine.create(f)
diff --git a/lualib/socket.lua b/lualib/socket.lua
new file mode 100644
index 00000000..79a2d151
--- /dev/null
+++ b/lualib/socket.lua
@@ -0,0 +1,46 @@
+local skynet = require "skynet"
+local c = require "socket.c"
+
+local socket = {}
+local fd
+local object
+
+function socket.connect(addr)
+	local ip, port = string.match(addr,"([^:]+):(.+)")
+	port = tonumber(port)
+	fd = c.open(ip,port)
+	skynet.send(".connection","ADD "..fd.." "..skynet.self())
+	object = c.new()
+end
+
+function socket.push(msg,sz)
+	if msg == nil then
+		socket.close()
+	else
+		c.push(object, msg, sz)
+	end
+end
+
+function socket.read(bytes)
+	return c.read(object, bytes)
+end
+
+function socket.readline(sep)
+	return c.readline(object, sep)
+end
+
+function socket.write(...)
+	c.write(fd, ...)
+end
+
+function socket.yield()
+	c.yield(object)
+end
+
+function socket.close()
+	skynet.send(".connection","DEL "..fd)
+	fd = nil
+end
+
+return socket
+
diff --git a/service/redis-cli.lua b/service/redis-cli.lua
index 079237fe..b38640dc 100644
--- a/service/redis-cli.lua
+++ b/service/redis-cli.lua
@@ -1,34 +1,31 @@
 local skynet = require "skynet"
+local socket = require "socket"
 local string = string
 local table = table
 local tonumber = tonumber
 local ipairs = ipairs
 local unpack = unpack
 local redis_server = ...
-local fd
-local write_fd
-local readline_fd
-local read_fd
-local close_fd
 
-local function init_fd(fdstr)
-	fd = fdstr
-	write_fd = "WRITE "..fd.." "
-	readline_fd = "READLINE ".. fd .." \r\n"
-	read_fd = "READ " .. fd .. " "
-	close_fd = "CLOSE "..fd
-end
-
-local function init()
-	fd = skynet.call(".connection", "CONNECT " .. redis_server)
-	if fd == nil then
-		print("Connect to redis server error : ", redis_server)
-		skynet.exit()
-		return true
+local function read(bytes)
+	while true do
+		local result = socket.read(bytes)
+		if result then
+			return result
+		end
+		coroutine.yield()
 	end
-	init_fd(fd)
 end
 
+local function readline(sep)
+	while true do
+		local result = socket.readline(sep)
+		if result then
+			return result
+		end
+		coroutine.yield()
+	end
+end
 
 local function compose_message(msg)
 	local lines = { "*" .. #msg }
@@ -42,72 +39,109 @@ local function compose_message(msg)
 	return cmd
 end
 
+local function init()
+	socket.connect(redis_server)
+end
+
+local request_queue = { head = 1, tail = 1 }
+
+local function push_request_queue(reply)
+	request_queue[request_queue.tail] = reply
+	request_queue.tail = request_queue.tail + 1
+end
+
+local function pop_request_queue()
+	assert(request_queue.head < request_queue.tail)
+	local reply = request_queue[request_queue.head]
+	request_queue[request_queue.head] = nil
+	request_queue.head = request_queue.head + 1
+	return reply
+end
+
+local function response(...)
+	local reply = pop_request_queue()
+	skynet.send(reply[2],reply[1],skynet.pack(...))
+end
+
 local redcmd = {}
 
 redcmd[42] = function(data)	-- '*'
 	local n = tonumber(data)
 	if n < 1 then
-		skynet.ret(skynet.pack(true, nil))
+		response(true, nil)
 		return
 	end
 	local bulk = {}
 	for i = 1,n do
-		local line = skynet.call(".connection", readline_fd)
+		local line = socket.readline "\r\n"
+		if line == nil then
+			return "BLOCK"
+		end
 		local bytes = tonumber(string.sub(line,2) + 2)
-		local data = skynet.call(".connection", read_fd .. bytes)
+		local data = socket.read(bytes)
+		if data == nil then
+			return "BLOCK"
+		end
 		table.insert(bulk, string.sub(data,1,-3))
 	end
-	skynet.ret(skynet.pack(true,bulk))
+	response(true, bulk)
 end
 
 redcmd[36] = function(data) -- '$'
 	local bytes = tonumber(data)
 	if bytes < 0 then
-		skynet.ret(skynet.pack(true, nil))
+		response(true,nil)
 		return
 	end
-	local firstline = skynet.call(".connection", read_fd .. (bytes + 2))
-	skynet.ret(skynet.pack(true,string.sub(firstline,1,-3)))
+	local firstline = socket.read(bytes+2)
+	if firstline == nil then
+		return "BLOCK"
+	end
+	response(true,string.sub(firstline,1,-3))
 end
 
 redcmd[43] = function(data) -- '+'
-	skynet.ret(skynet.pack(true, data))
+	response(true,data)
 end
 
 redcmd[45] = function(data) -- '-'
-	skynet.ret(skynet.pack(false, data))
+	response(false,data)
 end
 
 redcmd[58] = function(data) -- ':'
-	skynet.ret(skynet.pack(true, tonumber(data)))
+	response(true, tonumber(data))
 end
 
-skynet.dispatch(function(msg, sz, session, address)
-	local message = { skynet.unpack(msg,sz) }
-	local write_cmd = write_fd .. compose_message(message)
-	local result
-	while true do
-		skynet.send(".connection", write_cmd )
-		result = skynet.call(".connection", readline_fd)
-		if result then
-			break
-		end
-		-- reconnect
-		if init() then
-			skynet.ret(skynet.pack(false , "Disconnected"))
-			return
-		end
+local function split_package()
+	local result = socket.readline "\r\n"
+	if result == nil then
+		return
 	end
 	local firstchar = string.byte(result)
 	local data = string.sub(result,2)
 	local f = redcmd[firstchar]
-	if f == nil then
-		skynet.ret(skynet.pack(false , "Invalid result"))
-		skynet.send(".connection", close_fd)
-		init()
-	else
-		f(data)
+	assert(f)
+	if f(data) then
+		return
 	end
-end)
+	socket.yield()
+	return true
+end
+
+skynet.filter(
+	function(session, address , msg, sz)
+		if session == 0x7fffffff then
+			socket.push(msg,sz)
+			while split_package() do end
+		elseif session < 0 then
+			local message = { skynet.unpack(msg,sz) }
+			local cmd = compose_message(message)
+			socket.write(cmd)
+			push_request_queue { -session , address }
+		else
+			return session, address, msg , sz
+		end
+	end
+)
 
 skynet.start(init)
diff --git a/skynet-src/skynet_harbor.c b/skynet-src/skynet_harbor.c
index ca91fba1..205e82d9 100644
--- a/skynet-src/skynet_harbor.c
+++ b/skynet-src/skynet_harbor.c
@@ -275,7 +275,7 @@ _remote_harbor_update(int harbor_id, const char * addr) {
 	void *socket = zmq_socket( Z->zmq_context, ZMQ_PUSH);
 	int rc = zmq_connect(socket, addr);
 	if (rc<0) {
-		skynet_error(NULL, "Can't connect to %d %s",harbor_id,addr);
+		skynet_error(NULL, "Can't connect to harbor %d %s",harbor_id,addr);
 		zmq_close(socket);
 		socket = NULL;
 	}
diff --git a/skynet-src/skynet_server.c b/skynet-src/skynet_server.c
index d876014c..391b5a96 100644
--- a/skynet-src/skynet_server.c
+++ b/skynet-src/skynet_server.c
@@ -399,6 +399,9 @@ skynet_send(struct skynet_context * context, const char * source, const char * a
 		session = skynet_context_newsession(context);
 		session_id = - session;
 	}
+	if (addr == NULL) {
+		return session;
+	}
 	uint32_t des = 0;
 	if (addr[0] == ':') {
 		des = strtoul(addr+1, NULL, 16);