add testresponse

release v0.6.1

HISTORY.md

@@ -1,3 +1,43 @@
+v0.6.2 (2014-9-1)
+-----------
+* bugfix: only skynet.call response PTYPE_ERROR
+
+v0.6.1 (2014-8-25)
+-----------
+* bugfix: datacenter.wakeup
+* change struct msg name to avoid conflict in mac
+* improve seri library
+
+v0.6.0 (2014-8-18)
+-----------
+* add sharedata
+* bugfix: service exit before init would not report back
+* add skynet.response and check multicall skynet.ret
+* skynet.newservice throw error when lanuch faild
+* Don't check imported function in snax.hotfix
+* snax service add change SERVICE_PATH and add it to package.path
+* skynet.redirect support string address
+* bugfix: skynet.harbor.link may block
+* add skynet.harbor.queryname to query globalname
+* add cluster.proxy 
+* add DEBUG command exit (send a message to lua service by DEBUG)
+* add DEBUG command run (debug_console command inject)
+* bugfix : socketchannel connect once
+* bugfix : mongo driver
+
+v0.5.2 (2014-8-11)
+-----------
+* Bugfix : httpd request
+* Bugifx : http chunked mode
+* Add : httpc
+* timer support more than 497 days
+
+v0.5.1 (2014-8-4)
+-----------
+* Bugfix : http module
+* Bugfix : multicast local channel delete
+* Bugfix : socket.read(fd)
+
 v0.5.0 (2014-7-28)
 -----------
 * skynet.exit will quit service immediately.

Makefile

@@ -43,7 +43,7 @@ jemalloc : $(MALLOC_STATICLIB)
 CSERVICE = snlua logger gate harbor
 LUA_CLIB = skynet socketdriver int64 bson mongo md5 netpack \
   cjson clientsocket memory profile multicast \
-  cluster crypt 
+  cluster crypt sharedata stm
 
 SKYNET_SRC = skynet_main.c skynet_handle.c skynet_module.c skynet_mq.c \
   skynet_server.c skynet_start.c skynet_timer.c skynet_error.c \
@@ -113,6 +113,12 @@ $(LUA_CLIB_PATH)/cluster.so : lualib-src/lua-cluster.c | $(LUA_CLIB_PATH)
 $(LUA_CLIB_PATH)/crypt.so : lualib-src/lua-crypt.c | $(LUA_CLIB_PATH)
 	$(CC) $(CFLAGS) $(SHARED) $^ -o $@ 
 
+$(LUA_CLIB_PATH)/sharedata.so : lualib-src/lua-sharedata.c | $(LUA_CLIB_PATH)
+	$(CC) $(CFLAGS) $(SHARED) $^ -o $@ 
+
+$(LUA_CLIB_PATH)/stm.so : lualib-src/lua-stm.c | $(LUA_CLIB_PATH)
+	$(CC) $(CFLAGS) $(SHARED) -Iskynet-src $^ -o $@ 
+
 clean :
 	rm -f $(SKYNET_BUILD_PATH)/skynet $(CSERVICE_PATH)/*.so $(LUA_CLIB_PATH)/*.so
 

README.md

@@ -34,8 +34,6 @@ Each lua file only load once and cache it in memory during skynet start . so if
 
 You can also use the offical lua version , edit the makefile by yourself .
 
-## Blog (in Chinese)
+## How To (in Chinese)
 
-* http://blog.codingnow.com/2012/09/the_design_of_skynet.html
+* Read Wiki https://github.com/cloudwu/skynet/wiki
-* http://blog.codingnow.com/2012/08/skynet.html
-* http://blog.codingnow.com/eo/skynet/

examples/cluster1.lua

@@ -2,8 +2,9 @@ local skynet = require "skynet"
 local cluster = require "cluster"
 
 skynet.start(function()
-	skynet.newservice("simpledb")
+	local sdb = skynet.newservice("simpledb")
-	print(skynet.call("SIMPLEDB", "lua", "SET", "a", "foobar"))
+	skynet.name(".simpledb", sdb)
-	print(skynet.call("SIMPLEDB", "lua", "GET", "a"))
+	print(skynet.call(".simpledb", "lua", "SET", "a", "foobar"))
+	print(skynet.call(".simpledb", "lua", "GET", "a"))
 	cluster.open "db"
 end)

examples/cluster2.lua

@@ -2,5 +2,7 @@ local skynet = require "skynet"
 local cluster = require "cluster"
 
 skynet.start(function()
-	print(cluster.call("db", "SIMPLEDB", "GET", "a"))
+	local proxy = cluster.proxy("db", ".simpledb")
+	print(skynet.call(proxy, "lua", "GET", "a"))
+	print(cluster.call("db", ".simpledb", "GET", "a"))
 end)

examples/globallog.lua

@@ -4,5 +4,6 @@ skynet.start(function()
 	skynet.dispatch("lua", function(session, address, ...)
 		print("[GLOBALLOG]", skynet.address(address), ...)
 	end)
+	skynet.register ".log"
 	skynet.register "LOG"
 end)

examples/share.lua

@@ -0,0 +1,72 @@
+local skynet = require "skynet"
+local sharedata = require "sharedata"
+
+local mode = ...
+
+if mode == "host" then
+
+skynet.start(function()
+	skynet.error("new foobar")
+	sharedata.new("foobar", { a=1, b= { "hello",  "world" } })
+
+	skynet.fork(function()
+		skynet.sleep(200)	-- sleep 2s
+		skynet.error("update foobar a = 2")
+		sharedata.update("foobar", { a =2 })
+		skynet.sleep(200)	-- sleep 2s
+		skynet.error("update foobar a = 3")
+		sharedata.update("foobar", { a = 3, b = { "change" } })
+		skynet.sleep(100)
+		skynet.error("delete foobar")
+		sharedata.delete "foobar"
+	end)
+end)
+
+else
+
+
+skynet.start(function()
+	skynet.newservice(SERVICE_NAME, "host")
+
+	local obj = sharedata.query "foobar"
+
+	local b = obj.b
+	skynet.error(string.format("a=%d", obj.a))
+
+	for k,v in ipairs(b) do
+		skynet.error(string.format("b[%d]=%s", k,v))
+	end
+
+	for i = 1, 5 do
+		skynet.sleep(100)
+		skynet.error("second " ..i)
+		for k,v in pairs(obj) do
+			skynet.error(string.format("%s = %s", k , tostring(v)))
+		end
+	end
+
+	local ok, err = pcall(function()
+		local tmp = { b[1], b[2] }	-- b is invalid , so pcall should failed
+	end)
+
+	if not ok then
+		skynet.error(err)
+	end
+
+	-- obj. b is not the same with local b
+	for k,v in ipairs(obj.b) do
+		skynet.error(string.format("b[%d] = %s", k , tostring(v)))
+	end
+
+	collectgarbage()
+	skynet.error("sleep")
+	skynet.sleep(100)
+	b = nil
+	collectgarbage()
+	skynet.error("sleep")
+	skynet.sleep(100)
+
+	skynet.exit()
+end)
+
+end

examples/simpleweb.lua

@@ -39,6 +39,11 @@ skynet.start(function()
 						table.insert(tmp, string.format("query: %s= %s", k,v))
 					end
 				end
+				table.insert(tmp, "-----header----")
+				for k,v in pairs(header) do
+					table.insert(tmp, string.format("%s = %s",k,v))
+				end
+				table.insert(tmp, "-----body----\n" .. body)
 				response(id, code, table.concat(tmp,"\n"))
 			end
 		else

lualib-src/lua-cluster.c

@@ -203,7 +203,7 @@ lunpackresponse(lua_State *L) {
 }
 
 int
-luaopen_cluster_c(lua_State *L) {
+luaopen_cluster_core(lua_State *L) {
 	luaL_Reg l[] = {
 		{ "packrequest", lpackrequest },
 		{ "unpackrequest", lunpackrequest },

lualib-src/lua-crypt.c

@@ -589,11 +589,11 @@ read64(lua_State *L, uint32_t xx[2], uint32_t yy[2]) {
 	size_t sz = 0;
 	const uint8_t *x = (const uint8_t *)luaL_checklstring(L, 1, &sz);
 	if (sz != 8) {
-		luaL_error(L, "Invalid hmac x");
+		luaL_error(L, "Invalid uint64 x");
 	}
 	const uint8_t *y = (const uint8_t *)luaL_checklstring(L, 2, &sz);
 	if (sz != 8) {
-		luaL_error(L, "Invalid hmac y");
+		luaL_error(L, "Invalid uint64 y");
 	}
 	xx[0] = x[0] | x[1]<<8 | x[2]<<16 | x[3]<<24;
 	xx[1] = x[4] | x[5]<<8 | x[6]<<16 | x[7]<<24;
@@ -703,7 +703,7 @@ ldhexchange(lua_State *L) {
 	size_t sz = 0;
 	const uint8_t *x = (const uint8_t *)luaL_checklstring(L, 1, &sz);
 	if (sz != 8) {
-		luaL_error(L, "Invalid hmac x");
+		luaL_error(L, "Invalid dh uint64 key");
 	}
 	uint32_t xx[2];
 	xx[0] = x[0] | x[1]<<8 | x[2]<<16 | x[3]<<24;

lualib-src/lua-mongo.c

@@ -507,8 +507,8 @@ op_insert(lua_State *L) {
 		for (i=1;i<=s;i++) {
 			lua_rawgeti(L,3,i);
 			document doc = lua_touserdata(L,-1);
+			lua_pop(L,1);	// must call lua_pop before luaL_addlstring, because addlstring may change stack top
 			luaL_addlstring(&b, (const char *)doc, get_length(doc));
-			lua_pop(L,1);
 		}
 	}
 

lualib-src/lua-multicast.c

@@ -149,7 +149,7 @@ mc_nextid(lua_State *L) {
 }
 
 int
-luaopen_multicast_c(lua_State *L) {
+luaopen_multicast_core(lua_State *L) {
 	luaL_Reg l[] = {
 		{ "pack", mc_packlocal },
 		{ "unpack", mc_unpacklocal },

lualib-src/lua-seri.c

@@ -1,5 +1,5 @@
 /*
-  https://github.com/cloudwu/lua-serialize
+	modify from https://github.com/cloudwu/lua-serialize
  */
 
 #include "skynet_malloc.h"
@@ -35,14 +35,13 @@ struct block {
 
 struct write_block {
 	struct block * head;
-	int len;
 	struct block * current;
+	int len;
 	int ptr;
 };
 
 struct read_block {
 	char * buffer;
-	struct block * current;
 	int len;
 	int ptr;
 };
@@ -78,38 +77,17 @@ _again:
 
 static void
 wb_init(struct write_block *wb , struct block *b) {
-	if (b==NULL) {
+	wb->head = b;
-		wb->head = blk_alloc();
+	assert(b->next == NULL);
-		wb->len = 0;
+	wb->len = 0;
-		wb->current = wb->head;
+	wb->current = wb->head;
-		wb->ptr = 0;
+	wb->ptr = 0;
-		wb_push(wb, &wb->len, sizeof(wb->len));
-	} else {
-		wb->head = b;
-		int * plen = (int *)b->buffer;
-		int sz = *plen;
-		wb->len = sz;
-		while (b->next) {
-			sz -= BLOCK_SIZE;		
-			b = b->next;
-		}
-		wb->current = b;
-		wb->ptr = sz;
-	}
-}
-
-static struct block *
-wb_close(struct write_block *b) {
-	b->current = b->head;
-	b->ptr = 0;
-	wb_push(b, &b->len, sizeof(b->len));
-	b->current = NULL;
-	return b->head;
 }
 
 static void
 wb_free(struct write_block *wb) {
 	struct block *blk = wb->head;
+	blk = blk->next;	// the first block is on stack
 	while (blk) {
 		struct block * next = blk->next;
 		skynet_free(blk);
@@ -124,7 +102,6 @@ wb_free(struct write_block *wb) {
 static void
 rball_init(struct read_block * rb, char * buffer, int size) {
 	rb->buffer = buffer;
-	rb->current = NULL;
 	rb->len = size;
 	rb->ptr = 0;
 }
@@ -135,63 +112,10 @@ rb_read(struct read_block *rb, void *buffer, int sz) {
 		return NULL;
 	}
 
-	if (rb->buffer) {
+	int ptr = rb->ptr;
-		int ptr = rb->ptr;
+	rb->ptr += sz;
-		rb->ptr += sz;
+	rb->len -= sz;
-		rb->len -= sz;
+	return rb->buffer + ptr;
-		return rb->buffer + ptr;
-	}
-
-	if (rb->ptr == BLOCK_SIZE) {
-		struct block * next = rb->current->next;
-		skynet_free(rb->current);
-		rb->current = next;
-		rb->ptr = 0;
-	}
-
-	int copy = BLOCK_SIZE - rb->ptr;
-
-	if (sz <= copy) {
-		void * ret = rb->current->buffer + rb->ptr;
-		rb->ptr += sz;
-		rb->len -= sz;
-		return ret;
-	}
-
-	char * tmp = buffer;
-
-	memcpy(tmp, rb->current->buffer + rb->ptr,  copy);
-	sz -= copy;
-	tmp += copy;
-	rb->len -= copy;
-
-	for (;;) {
-		struct block * next = rb->current->next;
-		skynet_free(rb->current);
-		rb->current = next;
-
-		if (sz < BLOCK_SIZE) {
-			memcpy(tmp, rb->current->buffer, sz);
-			rb->ptr = sz;
-			rb->len -= sz;
-			return buffer;
-		}
-		memcpy(tmp, rb->current->buffer, BLOCK_SIZE);
-		sz -= BLOCK_SIZE;
-		tmp += BLOCK_SIZE;
-		rb->len -= BLOCK_SIZE;
-	}
-}
-
-static void
-rb_close(struct read_block *rb) {
-	while (rb->current) {
-		struct block * next = rb->current->next;
-		skynet_free(rb->current);
-		rb->current = next;
-	}
-	rb->len = 0;
-	rb->ptr = 0;
 }
 
 static inline void
@@ -271,7 +195,7 @@ wb_string(struct write_block *wb, const char *str, int len) {
 	}
 }
 
-static void _pack_one(lua_State *L, struct write_block *b, int index, int depth);
+static void pack_one(lua_State *L, struct write_block *b, int index, int depth);
 
 static int
 wb_table_array(lua_State *L, struct write_block * wb, int index, int depth) {
@@ -288,7 +212,7 @@ wb_table_array(lua_State *L, struct write_block * wb, int index, int depth) {
 	int i;
 	for (i=1;i<=array_size;i++) {
 		lua_rawgeti(L,index,i);
-		_pack_one(L, wb, -1, depth);
+		pack_one(L, wb, -1, depth);
 		lua_pop(L,1);
 	}
 
@@ -307,8 +231,8 @@ wb_table_hash(lua_State *L, struct write_block * wb, int index, int depth, int a
 				continue;
 			}
 		}
-		_pack_one(L,wb,-2,depth);
+		pack_one(L,wb,-2,depth);
-		_pack_one(L,wb,-1,depth);
+		pack_one(L,wb,-1,depth);
 		lua_pop(L, 1);
 	}
 	wb_nil(wb);
@@ -324,7 +248,7 @@ wb_table(lua_State *L, struct write_block *wb, int index, int depth) {
 }
 
 static void
-_pack_one(lua_State *L, struct write_block *b, int index, int depth) {
+pack_one(lua_State *L, struct write_block *b, int index, int depth) {
 	if (depth > MAX_DEPTH) {
 		wb_free(b);
 		luaL_error(L, "serialize can't pack too depth table");
@@ -370,27 +294,24 @@ _pack_one(lua_State *L, struct write_block *b, int index, int depth) {
 }
 
 static void
-_pack_from(lua_State *L, struct write_block *b, int from) {
+pack_from(lua_State *L, struct write_block *b, int from) {
 	int n = lua_gettop(L) - from;
 	int i;
 	for (i=1;i<=n;i++) {
-		_pack_one(L, b , from + i, 0);
+		pack_one(L, b , from + i, 0);
 	}
 }
 
 static inline void
-__invalid_stream(lua_State *L, struct read_block *rb, int line) {
+invalid_stream_line(lua_State *L, struct read_block *rb, int line) {
 	int len = rb->len;
-	if (rb->buffer == NULL) {
-		rb_close(rb);
-	}
 	luaL_error(L, "Invalid serialize stream %d (line:%d)", len, line);
 }
 
-#define _invalid_stream(L,rb) __invalid_stream(L,rb,__LINE__)
+#define invalid_stream(L,rb) invalid_stream_line(L,rb,__LINE__)
 
 static int
-_get_integer(lua_State *L, struct read_block *rb, int cookie) {
+get_integer(lua_State *L, struct read_block *rb, int cookie) {
 	switch (cookie) {
 	case 0:
 		return 0;
@@ -398,93 +319,93 @@ _get_integer(lua_State *L, struct read_block *rb, int cookie) {
 		uint8_t n = 0;
 		uint8_t * pn = rb_read(rb,&n,1);
 		if (pn == NULL)
-			_invalid_stream(L,rb);
+			invalid_stream(L,rb);
 		return *pn;
 	}
 	case 2: {
 		uint16_t n = 0;
 		uint16_t * pn = rb_read(rb,&n,2);
 		if (pn == NULL)
-			_invalid_stream(L,rb);
+			invalid_stream(L,rb);
 		return *pn;
 	}
 	case 4: {
 		int n = 0;
 		int * pn = rb_read(rb,&n,4);
 		if (pn == NULL)
-			_invalid_stream(L,rb);
+			invalid_stream(L,rb);
 		return *pn;
 	}
 	default:
-		_invalid_stream(L,rb);
+		invalid_stream(L,rb);
 		return 0;
 	}
 }
 
 static double
-_get_number(lua_State *L, struct read_block *rb, int cookie) {
+get_number(lua_State *L, struct read_block *rb, int cookie) {
 	if (cookie == 8) {
 		double n = 0;
 		double * pn = rb_read(rb,&n,8);
 		if (pn == NULL)
-			_invalid_stream(L,rb);
+			invalid_stream(L,rb);
 		return *pn;
 	} else {
-		return (double)_get_integer(L,rb,cookie);
+		return (double)get_integer(L,rb,cookie);
 	}
 }
 
 static void *
-_get_pointer(lua_State *L, struct read_block *rb) {
+get_pointer(lua_State *L, struct read_block *rb) {
 	void * userdata = 0;
 	void ** v = (void **)rb_read(rb,&userdata,sizeof(userdata));
 	if (v == NULL) {
-		_invalid_stream(L,rb);
+		invalid_stream(L,rb);
 	}
 	return *v;
 }
 
 static void
-_get_buffer(lua_State *L, struct read_block *rb, int len) {
+get_buffer(lua_State *L, struct read_block *rb, int len) {
 	char tmp[len];
 	char * p = rb_read(rb,tmp,len);
 	if (p == NULL) {
-		_invalid_stream(L,rb);
+		invalid_stream(L,rb);
 	}
 	lua_pushlstring(L,p,len);
 }
 
-static void _unpack_one(lua_State *L, struct read_block *rb);
+static void unpack_one(lua_State *L, struct read_block *rb);
 
 static void
-_unpack_table(lua_State *L, struct read_block *rb, int array_size) {
+unpack_table(lua_State *L, struct read_block *rb, int array_size) {
 	if (array_size == MAX_COOKIE-1) {
 		uint8_t type = 0;
 		uint8_t *t = rb_read(rb, &type, 1);
 		if (t==NULL || (*t & 7) != TYPE_NUMBER) {
-			_invalid_stream(L,rb);
+			invalid_stream(L,rb);
 		}
-		array_size = (int)_get_number(L,rb,*t >> 3);
+		array_size = (int)get_number(L,rb,*t >> 3);
 	}
 	lua_createtable(L,array_size,0);
 	int i;
 	for (i=1;i<=array_size;i++) {
-		_unpack_one(L,rb);
+		unpack_one(L,rb);
 		lua_rawseti(L,-2,i);
 	}
 	for (;;) {
-		_unpack_one(L,rb);
+		unpack_one(L,rb);
 		if (lua_isnil(L,-1)) {
 			lua_pop(L,1);
 			return;
 		}
-		_unpack_one(L,rb);
+		unpack_one(L,rb);
 		lua_rawset(L,-3);
 	}
 }
 
 static void
-_push_value(lua_State *L, struct read_block *rb, int type, int cookie) {
+push_value(lua_State *L, struct read_block *rb, int type, int cookie) {
 	switch(type) {
 	case TYPE_NIL:
 		lua_pushnil(L);
@@ -493,82 +414,69 @@ _push_value(lua_State *L, struct read_block *rb, int type, int cookie) {
 		lua_pushboolean(L,cookie);
 		break;
 	case TYPE_NUMBER:
-		lua_pushnumber(L,_get_number(L,rb,cookie));
+		lua_pushnumber(L,get_number(L,rb,cookie));
 		break;
 	case TYPE_USERDATA:
-		lua_pushlightuserdata(L,_get_pointer(L,rb));
+		lua_pushlightuserdata(L,get_pointer(L,rb));
 		break;
 	case TYPE_SHORT_STRING:
-		_get_buffer(L,rb,cookie);
+		get_buffer(L,rb,cookie);
 		break;
 	case TYPE_LONG_STRING: {
 		uint32_t len;
 		if (cookie == 2) {
 			uint16_t *plen = rb_read(rb, &len, 2);
 			if (plen == NULL) {
-				_invalid_stream(L,rb);
+				invalid_stream(L,rb);
 			}
-			_get_buffer(L,rb,(int)*plen);
+			get_buffer(L,rb,(int)*plen);
 		} else {
 			if (cookie != 4) {
-				_invalid_stream(L,rb);
+				invalid_stream(L,rb);
 			}
 			uint32_t *plen = rb_read(rb, &len, 4);
 			if (plen == NULL) {
-				_invalid_stream(L,rb);
+				invalid_stream(L,rb);
 			}
-			_get_buffer(L,rb,(int)*plen);
+			get_buffer(L,rb,(int)*plen);
 		}
 		break;
 	}
 	case TYPE_TABLE: {
-		_unpack_table(L,rb,cookie);
+		unpack_table(L,rb,cookie);
 		break;
 	}
 	default: {
-		_invalid_stream(L,rb);
+		invalid_stream(L,rb);
 		break;
 	}
 	}
 }
 
 static void
-_unpack_one(lua_State *L, struct read_block *rb) {
+unpack_one(lua_State *L, struct read_block *rb) {
 	uint8_t type = 0;
 	uint8_t *t = rb_read(rb, &type, 1);
 	if (t==NULL) {
-		_invalid_stream(L, rb);
+		invalid_stream(L, rb);
 	}
-	_push_value(L, rb, *t & 0x7, *t>>3);
+	push_value(L, rb, *t & 0x7, *t>>3);
 }
 
 static void
-_seri(lua_State *L, struct block *b) {
+seri(lua_State *L, struct block *b, int len) {
-	uint32_t len = 0;
-	memcpy(&len, b->buffer ,sizeof(len));
-
-	len -= 4;
 	uint8_t * buffer = skynet_malloc(len);
 	uint8_t * ptr = buffer;
 	int sz = len;
-	if (len < BLOCK_SIZE - 4) {
+	while(len>0) {
-		memcpy(ptr, b->buffer+4, len);
+		if (len >= BLOCK_SIZE) {
-	} else {
+			memcpy(ptr, b->buffer, BLOCK_SIZE);
-		memcpy(ptr, b->buffer+4, BLOCK_SIZE-4);
+			ptr += BLOCK_SIZE;
-		ptr += BLOCK_SIZE-4;
+			len -= BLOCK_SIZE;
-		len -= BLOCK_SIZE-4;
-		b = b->next;
-
-		while(len>0) {
-			if (len >= BLOCK_SIZE) {
-				memcpy(ptr, b->buffer, BLOCK_SIZE);
-				ptr += BLOCK_SIZE;
-				len -= BLOCK_SIZE;
-			} else {
-				memcpy(ptr, b->buffer, len);
-				break;
-			}
 			b = b->next;
+		} else {
+			memcpy(ptr, b->buffer, len);
+			break;
 		}
 	}
 	
@@ -611,7 +519,7 @@ _luaseri_unpack(lua_State *L) {
 		uint8_t *t = rb_read(&rb, &type, 1);
 		if (t==NULL)
 			break;
-		_push_value(L, &rb, *t & 0x7, *t>>3);
+		push_value(L, &rb, *t & 0x7, *t>>3);
 	}
 
 	// Need not free buffer
@@ -621,17 +529,15 @@ _luaseri_unpack(lua_State *L) {
 
 int
 _luaseri_pack(lua_State *L) {
+	struct block temp;
+	temp.next = NULL;
 	struct write_block wb;
-	wb_init(&wb, NULL);
+	wb_init(&wb, &temp);
-	_pack_from(L,&wb,0);
+	pack_from(L,&wb,0);
-	struct block * b = wb_close(&wb);
+	assert(wb.head == &temp);
-	_seri(L,b);
+	seri(L, &temp, wb.len);
 
-	while (b) {
+	wb_free(&wb);
-		struct block * next = b->next;
-		skynet_free(b);
-		b = next;
-	}
 
 	return 2;
 }

lualib-src/lua-sharedata.c

@@ -0,0 +1,789 @@
+// build: gcc -O2 -Wall --shared -o conf.so luaconf.c 
+
+#include <lua.h>
+#include <lauxlib.h>
+#include <stdint.h>
+#include <stdlib.h>
+#include <string.h>
+#include <assert.h>
+
+#define KEYTYPE_INTEGER 0
+#define KEYTYPE_STRING 1
+
+#define VALUETYPE_NIL 0
+#define VALUETYPE_NUMBER 1
+#define VALUETYPE_STRING 2
+#define VALUETYPE_BOOLEAN 3
+#define VALUETYPE_TABLE 4
+
+struct table;
+
+union value {
+	lua_Number n;
+	struct table * tbl;
+	int string;
+	int boolean;
+};
+
+struct node {
+	union value v;
+	int key;	// integer key or index of string table
+	int next;	// next slot index
+	uint32_t keyhash;
+	uint8_t keytype;	// key type must be integer or string
+	uint8_t valuetype;	// value type can be number/string/boolean/table
+	uint8_t nocolliding;	// 0 means colliding slot
+};
+
+struct table;
+
+struct state {
+	int dirty;
+	int ref;
+	struct table * root;
+};
+
+struct table {
+	int sizearray;
+	int sizehash;
+	uint8_t *arraytype;
+	union value * array;
+	struct node * hash;
+	lua_State * L;
+};
+
+struct context {
+	lua_State * L;
+	struct table * tbl;
+	int string_index;
+};
+
+struct ctrl {
+	struct table * root;
+	struct table * update;
+};
+
+static int
+countsize(lua_State *L, int sizearray) {
+	int n = 0;
+	lua_pushnil(L);
+	while (lua_next(L, 1) != 0) {
+		int type = lua_type(L, -2);
+		++n;
+		if (type == LUA_TNUMBER) {
+			lua_Number key = lua_tonumber(L, -2);
+			int nkey = (int)key;
+			if ((lua_Number)nkey != key) {
+				luaL_error(L, "Invalid key %f", key);
+			}
+			if (nkey > 0 && nkey <= sizearray) {
+				--n;
+			}
+		} else if (type != LUA_TSTRING && type != LUA_TTABLE) {
+			luaL_error(L, "Invalid key type %s", lua_typename(L, type));
+		}
+		lua_pop(L, 1);
+	}
+
+	return n;
+}
+
+static uint32_t
+calchash(const char * str, size_t l) {
+	uint32_t h = (uint32_t)l;
+	size_t l1;
+	size_t step = (l >> 5) + 1;
+	for (l1 = l; l1 >= step; l1 -= step) {
+		h = h ^ ((h<<5) + (h>>2) + (uint8_t)(str[l1 - 1]));
+	}
+	return h;
+}
+
+static int
+stringindex(struct context *ctx, const char * str, size_t sz) {
+	lua_State *L = ctx->L;
+	lua_pushlstring(L, str, sz);
+	lua_pushvalue(L, -1);
+	lua_rawget(L, 1);
+	int index;
+	// stringmap(1) str index
+	if (lua_isnil(L, -1)) {
+		index = ++ctx->string_index;
+		lua_pop(L, 1);
+		lua_pushinteger(L, index);
+		lua_rawset(L, 1);
+	} else {
+		index = lua_tointeger(L, -1);
+		lua_pop(L, 2);
+	}
+	return index;
+}
+
+static int convtable(lua_State *L);
+
+static void
+setvalue(struct context * ctx, lua_State *L, int index, struct node *n) {
+	int vt = lua_type(L, index);
+	switch(vt) {
+	case LUA_TNIL:
+		n->valuetype = VALUETYPE_NIL;
+		break;
+	case LUA_TNUMBER:
+		n->v.n = lua_tonumber(L, index);
+		n->valuetype = VALUETYPE_NUMBER;
+		break;
+	case LUA_TSTRING: {
+		size_t sz = 0;
+		const char * str = lua_tolstring(L, index, &sz);
+		n->v.string = stringindex(ctx, str, sz);
+		n->valuetype = VALUETYPE_STRING;
+		break;
+	}
+	case LUA_TBOOLEAN:
+		n->v.boolean = lua_toboolean(L, index);
+		n->valuetype = VALUETYPE_BOOLEAN;
+		break;
+	case LUA_TTABLE: {
+		struct table *tbl = ctx->tbl;
+		ctx->tbl = (struct table *)malloc(sizeof(struct table));
+		if (ctx->tbl == NULL) {
+			ctx->tbl = tbl;
+			luaL_error(L, "memory error");
+			// never get here
+		}
+		memset(ctx->tbl, 0, sizeof(struct table));
+		int absidx = lua_absindex(L, index);
+
+		lua_pushcfunction(L, convtable);
+		lua_pushvalue(L, absidx);
+		lua_pushlightuserdata(L, ctx);
+
+		lua_call(L, 2, 0);
+
+		n->v.tbl = ctx->tbl;
+		n->valuetype = VALUETYPE_TABLE;
+
+		ctx->tbl = tbl;
+
+		break;
+	}
+	default:
+		luaL_error(L, "Unsupport value type %s", lua_typename(L, vt));
+		break;
+	}
+}
+
+static void
+setarray(struct context *ctx, lua_State *L, int index, int key) {
+	struct node n;
+	setvalue(ctx, L, index, &n);
+	struct table *tbl = ctx->tbl;
+	--key;	// base 0
+	tbl->arraytype[key] = n.valuetype;
+	tbl->array[key] = n.v;
+}
+
+static int
+ishashkey(struct context * ctx, lua_State *L, int index, int *key, uint32_t *keyhash, int *keytype) {
+	int sizearray = ctx->tbl->sizearray;
+	int kt = lua_type(L, index);
+	if (kt == LUA_TNUMBER) {
+		*key = lua_tointeger(L, index);
+		if (*key > 0 && *key <= sizearray) {
+			return 0;
+		}
+		*keyhash = (uint32_t)*key;
+		*keytype = KEYTYPE_INTEGER;
+	} else {
+		size_t sz = 0;
+		const char * s = lua_tolstring(L, index, &sz);
+		*keyhash = calchash(s, sz);
+		*key = stringindex(ctx, s, sz);
+		*keytype = KEYTYPE_STRING;
+	}
+	return 1;
+}
+
+static void
+fillnocolliding(lua_State *L, struct context *ctx) {
+	struct table * tbl = ctx->tbl;
+	lua_pushnil(L);
+	while (lua_next(L, 1) != 0) {
+		int key;
+		int keytype;
+		uint32_t keyhash;
+		if (!ishashkey(ctx, L, -2, &key, &keyhash, &keytype)) {
+			setarray(ctx, L, -1, key);
+		} else {
+			struct node * n = &tbl->hash[keyhash % tbl->sizehash];
+			if (n->valuetype == VALUETYPE_NIL) {
+				n->key = key;
+				n->keytype = keytype;
+				n->keyhash = keyhash;
+				n->next = -1;
+				n->nocolliding = 1;
+				setvalue(ctx, L, -1, n);	// set n->v , n->valuetype
+			}
+		}
+		lua_pop(L,1);
+	}
+}
+
+static void
+fillcolliding(lua_State *L, struct context *ctx) {
+	struct table * tbl = ctx->tbl;
+	int sizehash = tbl->sizehash;
+	int emptyslot = 0;
+	int i;
+	lua_pushnil(L);
+	while (lua_next(L, 1) != 0) {
+		int key;
+		int keytype;
+		uint32_t keyhash;
+		if (ishashkey(ctx, L, -2, &key, &keyhash, &keytype)) {
+			struct node * mainpos = &tbl->hash[keyhash % tbl->sizehash];
+			if (!(mainpos->keytype == keytype && mainpos->key == key)) {
+				// the key has not insert
+				struct node * n = NULL;
+				for (i=emptyslot;i<sizehash;i++) {
+					if (tbl->hash[i].valuetype == VALUETYPE_NIL) {
+						n = &tbl->hash[i];
+						break;
+					}
+				}
+				assert(n);
+				n->next = mainpos->next;
+				mainpos->next = n - tbl->hash;
+				mainpos->nocolliding = 0;
+				n->key = key;
+				n->keytype = keytype;
+				n->keyhash = keyhash;
+				n->nocolliding = 0;
+				setvalue(ctx, L, -1, n);	// set n->v , n->valuetype
+			}
+		}
+		lua_pop(L,1);
+	}
+}
+
+// table need convert
+// struct context * ctx
+static int
+convtable(lua_State *L) {
+	int i;
+	struct context *ctx = lua_touserdata(L,2);
+	struct table *tbl = ctx->tbl;
+
+	tbl->L = ctx->L;
+
+	int sizearray = lua_rawlen(L, 1);
+	if (sizearray) {
+		tbl->arraytype = (uint8_t *)malloc(sizearray * sizeof(uint8_t));
+		if (tbl->arraytype == NULL) {
+			goto memerror;
+		}
+		for (i=0;i<sizearray;i++) {
+			tbl->arraytype[i] = VALUETYPE_NIL;
+		}
+		tbl->array = (union value *)malloc(sizearray * sizeof(union value));
+		if (tbl->array == NULL) {
+			goto memerror;
+		}
+		tbl->sizearray = sizearray;
+	}
+	int sizehash = countsize(L, sizearray);
+	if (sizehash) {
+		tbl->hash = (struct node *)malloc(sizehash * sizeof(struct node));
+		if (tbl->hash == NULL) {
+			goto memerror;
+		}
+		for (i=0;i<sizehash;i++) {
+			tbl->hash[i].valuetype = VALUETYPE_NIL;
+			tbl->hash[i].nocolliding = 0;
+		}
+		tbl->sizehash = sizehash;
+
+		fillnocolliding(L, ctx);
+		fillcolliding(L, ctx);
+	} else {
+		int i;
+		for (i=1;i<=sizearray;i++) {
+			lua_rawgeti(L, 1, i);
+			setarray(ctx, L, -1, i);
+			lua_pop(L,1);
+		}
+	}
+
+	return 0;
+memerror:
+	return luaL_error(L, "memory error");
+}
+
+static void
+delete_tbl(struct table *tbl) {
+	int i;
+	for (i=0;i<tbl->sizearray;i++) {
+		if (tbl->arraytype[i] == VALUETYPE_TABLE) {
+			delete_tbl(tbl->array[i].tbl);
+		}
+	}
+	for (i=0;i<tbl->sizehash;i++) {
+		if (tbl->hash[i].valuetype == VALUETYPE_TABLE) {
+			delete_tbl(tbl->hash[i].v.tbl);
+		}
+	}
+	free(tbl->arraytype);
+	free(tbl->array);
+	free(tbl->hash);
+	free(tbl);
+}
+
+static int
+pconv(lua_State *L) {
+	struct context *ctx = lua_touserdata(L,1);
+	lua_State * pL = lua_touserdata(L, 2);
+	int ret;
+
+	lua_settop(L, 0);
+
+	// init L (may throw memory error)
+	// create a table for string map
+	lua_newtable(L);
+
+	lua_pushcfunction(pL, convtable);
+	lua_pushvalue(pL,1);
+	lua_pushlightuserdata(pL, ctx);
+
+	ret = lua_pcall(pL, 2, 0, 0);
+	if (ret != LUA_OK) {
+		size_t sz = 0;
+		const char * error = lua_tolstring(pL, -1, &sz);
+		lua_pushlstring(L, error, sz);
+		lua_error(L);
+		// never get here
+	}
+
+	luaL_checkstack(L, ctx->string_index + 3, NULL);
+	lua_settop(L,1);
+
+	return 1;
+}
+
+static void
+convert_stringmap(struct context *ctx, struct table *tbl) {
+	lua_State *L = ctx->L;
+	lua_settop(L, ctx->string_index + 1);
+	lua_pushvalue(L, 1);
+	struct state * s = lua_newuserdata(L, sizeof(*s));
+	s->dirty = 0;
+	s->ref = 0;
+	s->root = tbl;
+	lua_replace(L, 1);
+	lua_replace(L, -2);
+
+	lua_pushnil(L);
+	// ... stringmap nil
+	while (lua_next(L, -2) != 0) {
+		int idx = lua_tointeger(L, -1);
+		lua_pop(L, 1);
+		lua_pushvalue(L, -1);
+		lua_replace(L, idx);
+	}
+
+	lua_pop(L, 1);
+
+	lua_gc(L, LUA_GCCOLLECT, 0);
+}
+
+static int
+lnewconf(lua_State *L) {
+	int ret;
+	struct context ctx;
+	struct table * tbl = NULL;
+	luaL_checktype(L,1,LUA_TTABLE);
+	ctx.L = luaL_newstate();
+	ctx.tbl = NULL;
+	ctx.string_index = 1;	// 1 reserved for dirty flag
+	if (ctx.L == NULL) {
+		lua_pushliteral(L, "memory error");
+		goto error;
+	}
+	tbl = (struct table *)malloc(sizeof(struct table));
+	if (tbl == NULL) {
+		// lua_pushliteral may fail because of memory error, close first.
+		lua_close(ctx.L);
+		ctx.L = NULL;
+		lua_pushliteral(L, "memory error");
+		goto error;
+	}
+	memset(tbl, 0, sizeof(struct table));
+	ctx.tbl = tbl;
+
+	lua_pushcfunction(ctx.L, pconv);
+	lua_pushlightuserdata(ctx.L , &ctx);
+	lua_pushlightuserdata(ctx.L , L);
+
+	ret = lua_pcall(ctx.L, 2, 1, 0);
+
+	if (ret != LUA_OK) {
+		size_t sz = 0;
+		const char * error = lua_tolstring(ctx.L, -1, &sz);
+		lua_pushlstring(L, error, sz);
+		goto error;
+	}
+
+	convert_stringmap(&ctx, tbl);
+
+	lua_pushlightuserdata(L, tbl);	
+
+	return 1;
+error:
+	if (ctx.L) {
+		lua_close(ctx.L);
+	}
+	if (tbl) {
+		delete_tbl(tbl);
+	}
+	lua_error(L);
+	return -1;
+}
+
+static struct table *
+get_table(lua_State *L, int index) {
+	struct table *tbl = lua_touserdata(L,index);
+	if (tbl == NULL) {
+		luaL_error(L, "Need a conf object");
+	}
+	return tbl;
+}
+
+static int
+ldeleteconf(lua_State *L) {
+	struct table *tbl = get_table(L,1);
+	lua_close(tbl->L);
+	delete_tbl(tbl);
+	return 0;
+}
+
+static void
+pushvalue(lua_State *L, lua_State *sL, uint8_t vt, union value *v) {
+	switch(vt) {
+	case VALUETYPE_NUMBER:
+		lua_pushnumber(L, v->n);
+		break;
+	case VALUETYPE_STRING: {
+		size_t sz = 0;
+		const char *str = lua_tolstring(sL, v->string, &sz);
+		lua_pushlstring(L, str, sz);
+		break;
+	}
+	case VALUETYPE_BOOLEAN:
+		lua_pushboolean(L, v->boolean);
+		break;
+	case VALUETYPE_TABLE:
+		lua_pushlightuserdata(L, v->tbl);
+		break;
+	default:
+		lua_pushnil(L);
+		break;
+	}
+}
+
+static struct node *
+lookup_key(struct table *tbl, uint32_t keyhash, int key, int keytype, const char *str, size_t sz) {
+	if (tbl->sizehash == 0)
+		return NULL;
+	struct node *n = &tbl->hash[keyhash % tbl->sizehash];
+	if (keyhash != n->keyhash && n->nocolliding)
+		return NULL;
+	for (;;) {
+		if (keyhash == n->keyhash) {
+			if (n->keytype == KEYTYPE_INTEGER) {
+				if (keytype == KEYTYPE_INTEGER && n->key == key) {
+					return n;
+				}
+			} else {
+				// n->keytype == KEYTYPE_STRING
+				if (keytype == KEYTYPE_STRING) {
+					size_t sz2 = 0;
+					const char * str2 = lua_tolstring(tbl->L, n->key, &sz2);
+					if (sz == sz2 && memcmp(str,str2,sz) == 0) {
+						return n;
+					}
+				}
+			}
+		}
+		if (n->next < 0) {
+			return NULL;
+		}
+		n = &tbl->hash[n->next];		
+	}
+}
+
+static int
+lindexconf(lua_State *L) {
+	struct table *tbl = get_table(L,1);
+	int kt = lua_type(L,2);
+	uint32_t keyhash;
+	int key = 0;
+	int keytype;
+	size_t sz = 0;
+	const char * str = NULL;
+	if (kt == LUA_TNUMBER) {
+		lua_Number k = lua_tonumber(L, 2);
+		key = (int)k;
+		if ((lua_Number)key != k) {
+			return luaL_error(L, "Invalid key %f", k);
+		}
+		if (key > 0 && key <= tbl->sizearray) {
+			--key;
+			pushvalue(L, tbl->L, tbl->arraytype[key], &tbl->array[key]);
+			return 1;
+		} 
+		keytype = KEYTYPE_INTEGER;
+		keyhash = (uint32_t)key;
+	} else {
+		str = luaL_checklstring(L, 2, &sz);
+		keyhash = calchash(str, sz);
+		keytype = KEYTYPE_STRING;
+	}
+
+	struct node *n = lookup_key(tbl, keyhash, key, keytype, str, sz);
+	if (n) {
+		pushvalue(L, tbl->L, n->valuetype, &n->v);
+		return 1;
+	} else {
+		return 0;
+	}
+}
+
+static void
+pushkey(lua_State *L, lua_State *sL, struct node *n) {
+	if (n->keytype == KEYTYPE_INTEGER) {
+		lua_pushinteger(L, n->key);
+	} else {
+		size_t sz = 0;
+		const char * str = lua_tolstring(sL, n->key, &sz);
+		lua_pushlstring(L, str, sz);
+	}
+}
+
+static int
+pushfirsthash(lua_State *L, struct table * tbl) {
+	if (tbl->sizehash) {
+		pushkey(L, tbl->L, &tbl->hash[0]);
+		return 1;
+	} else {
+		return 0;
+	}
+}
+
+static int
+lnextkey(lua_State *L) {
+	struct table *tbl = get_table(L,1);
+	if (lua_isnoneornil(L,2)) {
+		if (tbl->sizearray > 0) {
+			int i;
+			for (i=0;i<tbl->sizearray;i++) {
+				if (tbl->arraytype[i] != VALUETYPE_NIL) {
+					lua_pushinteger(L, i+1);
+					return 1;
+				}
+			}
+		}
+		return pushfirsthash(L, tbl);
+	}
+	int kt = lua_type(L,2);
+	uint32_t keyhash;
+	int key = 0;
+	int keytype;
+	size_t sz=0;
+	const char *str = NULL;
+	int sizearray = tbl->sizearray;
+	if (kt == LUA_TNUMBER) {
+		lua_Number k = lua_tonumber(L,2);
+		key = (int)k;
+		if ((lua_Number)key != k) {
+			return 0;
+		}
+		if (key > 0 && key <= sizearray) {
+			int i;
+			for (i=key;i<sizearray;i++) {
+				if (tbl->arraytype[i] != VALUETYPE_NIL) {
+					lua_pushinteger(L, i+1);
+					return 1;
+				}
+			}
+			return pushfirsthash(L, tbl);
+		}
+		keyhash = (uint32_t)key;
+		keytype = KEYTYPE_INTEGER;
+	} else {
+		str = luaL_checklstring(L, 2, &sz);
+		keyhash = calchash(str, sz);
+		keytype = KEYTYPE_STRING;
+	}
+
+	struct node *n = lookup_key(tbl, keyhash, key, keytype, str, sz);
+	if (n) {
+		++n;
+		int index = n-tbl->hash;
+		if (index == tbl->sizehash) {
+			return 0;
+		}
+		pushkey(L, tbl->L, n);
+		return 1;
+	} else {
+		return 0;
+	}
+}
+
+static int
+llen(lua_State *L) {
+	struct table *tbl = get_table(L,1);
+	lua_pushinteger(L, tbl->sizearray);
+	return 1;
+}
+
+static int
+lhashlen(lua_State *L) {
+	struct table *tbl = get_table(L,1);
+	lua_pushinteger(L, tbl->sizehash);
+	return 1;
+}
+
+static int
+releaseobj(lua_State *L) {
+	struct ctrl *c = lua_touserdata(L, 1);
+	struct table *tbl = c->root;
+	struct state *s = lua_touserdata(tbl->L, 1);
+	__sync_fetch_and_sub(&s->ref, 1);
+	c->root = NULL;
+	c->update = NULL;
+
+	return 0;
+}
+
+static int
+lboxconf(lua_State *L) {
+	struct table * tbl = get_table(L,1);	
+	struct state * s = lua_touserdata(tbl->L, 1);
+	__sync_fetch_and_add(&s->ref, 1);
+
+	struct ctrl * c = lua_newuserdata(L, sizeof(*c));
+	c->root = tbl;
+	c->update = NULL;
+	if (luaL_newmetatable(L, "confctrl")) {
+		lua_pushcfunction(L, releaseobj);
+		lua_setfield(L, -2, "__gc");
+	}
+	lua_setmetatable(L, -2);
+
+	return 1;
+}
+
+static int
+lmarkdirty(lua_State *L) {
+	struct table *tbl = get_table(L,1);
+	struct state * s = lua_touserdata(tbl->L, 1);
+	s->dirty = 1;
+	return 0;
+}
+
+static int
+lisdirty(lua_State *L) {
+	struct table *tbl = get_table(L,1);
+	struct state * s = lua_touserdata(tbl->L, 1);
+	int d = s->dirty;
+	lua_pushboolean(L, d);
+	
+	return 1;
+}
+
+static int
+lgetref(lua_State *L) {
+	struct table *tbl = get_table(L,1);
+	struct state * s = lua_touserdata(tbl->L, 1);
+	lua_pushinteger(L , s->ref);
+
+	return 1;
+}
+
+static int
+lincref(lua_State *L) {
+	struct table *tbl = get_table(L,1);
+	struct state * s = lua_touserdata(tbl->L, 1);
+	int ref = __sync_add_and_fetch(&s->ref, 1);
+	lua_pushinteger(L , ref);
+
+	return 1;
+}
+
+static int
+ldecref(lua_State *L) {
+	struct table *tbl = get_table(L,1);
+	struct state * s = lua_touserdata(tbl->L, 1);
+	int ref = __sync_sub_and_fetch(&s->ref, 1);
+	lua_pushinteger(L , ref);
+
+	return 1;
+}
+
+static int
+lneedupdate(lua_State *L) {
+	struct ctrl * c = lua_touserdata(L, 1);
+	if (c->update) {
+		lua_pushlightuserdata(L, c->update);
+		lua_getuservalue(L, 1);
+		return 2;
+	}
+	return 0;
+}
+
+static int
+lupdate(lua_State *L) {
+	luaL_checktype(L, 1, LUA_TUSERDATA);
+	luaL_checktype(L, 2, LUA_TLIGHTUSERDATA);
+	luaL_checktype(L, 3, LUA_TTABLE);
+	struct ctrl * c= lua_touserdata(L, 1);
+	struct table *n = lua_touserdata(L, 2);
+	if (c->update) {
+		return luaL_error(L, "can't update more than once");
+	}
+	if (c->root == n) {
+		return luaL_error(L, "You should update a new object");
+	}
+	lua_settop(L, 3);
+	lua_setuservalue(L, 1);
+	c->update = n;
+
+	return 0;
+}
+
+int
+luaopen_sharedata_core(lua_State *L) {
+	luaL_Reg l[] = {
+		// used by host
+		{ "new", lnewconf },
+		{ "delete", ldeleteconf },
+		{ "markdirty", lmarkdirty },
+		{ "getref", lgetref },
+		{ "incref", lincref },
+		{ "decref", ldecref },
+
+		// used by client
+		{ "box", lboxconf },
+		{ "index", lindexconf },
+		{ "nextkey", lnextkey },
+		{ "len", llen },
+		{ "hashlen", lhashlen },
+		{ "isdirty", lisdirty },
+		{ "needupdate", lneedupdate },
+		{ "update", lupdate },
+		{ NULL, NULL },
+	};
+	luaL_checkversion(L);
+	luaL_newlib(L, l);
+
+	return 1;
+}

lualib-src/lua-skynet.c

@@ -73,10 +73,17 @@ _cb(struct skynet_context * context, void * ud, int type, int session, uint32_t
 	return 0;
 }
 
+static int
+forward_cb(struct skynet_context * context, void * ud, int type, int session, uint32_t source, const void * msg, size_t sz) {
+	_cb(context, ud, type, session, source, msg, sz);
+	// don't delete msg in forward mode.
+	return 1;
+}
+
 static int
 _callback(lua_State *L) {
 	struct skynet_context * context = lua_touserdata(L, lua_upvalueindex(1));
-
+	int forward = lua_toboolean(L, 2);
 	luaL_checktype(L,1,LUA_TFUNCTION);
 	lua_settop(L,1);
 	lua_rawsetp(L, LUA_REGISTRYINDEX, _cb);
@@ -84,7 +91,11 @@ _callback(lua_State *L) {
 	lua_rawgeti(L, LUA_REGISTRYINDEX, LUA_RIDX_MAINTHREAD);
 	lua_State *gL = lua_tothread(L,-1);
 
-	skynet_callback(context, gL, _cb);
+	if (forward) {
+		skynet_callback(context, gL, forward_cb);
+	} else {
+		skynet_callback(context, gL, _cb);
+	}
 
 	return 0;
 }
@@ -115,44 +126,13 @@ _genid(lua_State *L) {
 	return 1;
 }
 
-// copy from _send
+static const char *
-
+get_dest_string(lua_State *L, int index) {
-static int
+	const char * dest_string = lua_tostring(L, index);
-_sendname(lua_State *L, struct skynet_context * context, const char * dest) {
+	if (dest_string == NULL) {
-	int type = luaL_checkinteger(L, 2);
+		luaL_error(L, "dest address type (%s) must be a string or number.", lua_typename(L, lua_type(L,index)));
-	int session = 0;
-	if (lua_isnil(L,3)) {
-		type |= PTYPE_TAG_ALLOCSESSION;
-	} else {
-		session = luaL_checkinteger(L,3);
 	}
-
+	return dest_string;
-	int mtype = lua_type(L,4);
-	switch (mtype) {
-	case LUA_TSTRING: {
-		size_t len = 0;
-		void * msg = (void *)lua_tolstring(L,4,&len);
-		session = skynet_sendname(context, dest, type, session , msg, len);
-		break;
-	}
-	case LUA_TNIL :
-		session = skynet_sendname(context, dest, type, session , NULL, 0);
-		break;
-	case LUA_TLIGHTUSERDATA: {
-		luaL_checktype(L, 4, LUA_TLIGHTUSERDATA);
-		void * msg = lua_touserdata(L,4);
-		int size = luaL_checkinteger(L,5);
-		session = skynet_sendname(context, dest, type | PTYPE_TAG_DONTCOPY, session, msg, size);
-		break;
-	}
-	default:
-		luaL_error(L, "skynet.send invalid param %s", lua_typename(L,lua_type(L,4)));
-	}
-	if (session < 0) {
-		return 0;
-	}
-	lua_pushinteger(L,session);
-	return 1;
 }
 
 /*
@@ -167,28 +147,10 @@ _sendname(lua_State *L, struct skynet_context * context, const char * dest) {
 static int
 _send(lua_State *L) {
 	struct skynet_context * context = lua_touserdata(L, lua_upvalueindex(1));
-	int addr_type = lua_type(L,1);
+	uint32_t dest = lua_tounsigned(L, 1);
-	uint32_t dest = 0;
+	const char * dest_string = NULL;
-	switch(addr_type) {
+	if (dest == 0) {
-	case LUA_TNUMBER:
+		dest_string = get_dest_string(L, 1);
-		dest = lua_tounsigned(L,1);
-		break;
-	case LUA_TSTRING: {
-		const char * addrname = lua_tostring(L,1);
-		if (addrname[0] == '.' || addrname[0] == ':') {
-			dest = skynet_queryname(context, addrname);
-			if (dest == 0) {
-				luaL_error(L, "Invalid name %s", addrname);
-			}
-		} else if ('0' <= addrname[0] && addrname[0] <= '9') {
-			luaL_error(L, "Invalid name %s: must not start with a digit", addrname);
-		} else {
-			return _sendname(L, context, addrname);
-		}
-		break;
-	}
-	default:
-		return luaL_error(L, "address must be number or string, got %s",lua_typename(L,addr_type));
 	}
 
 	int type = luaL_checkinteger(L, 2);
@@ -207,13 +169,21 @@ _send(lua_State *L) {
 		if (len == 0) {
 			msg = NULL;
 		}
-		session = skynet_send(context, 0, dest, type, session , msg, len);
+		if (dest_string) {
+			session = skynet_sendname(context, 0, dest_string, type, session , msg, len);
+		} else {
+			session = skynet_send(context, 0, dest, type, session , msg, len);
+		}
 		break;
 	}
 	case LUA_TLIGHTUSERDATA: {
 		void * msg = lua_touserdata(L,4);
 		int size = luaL_checkinteger(L,5);
-		session = skynet_send(context, 0, dest, type | PTYPE_TAG_DONTCOPY, session, msg, size);
+		if (dest_string) {
+			session = skynet_sendname(context, 0, dest_string, type | PTYPE_TAG_DONTCOPY, session, msg, size);
+		} else {
+			session = skynet_send(context, 0, dest, type | PTYPE_TAG_DONTCOPY, session, msg, size);
+		}
 		break;
 	}
 	default:
@@ -231,7 +201,11 @@ _send(lua_State *L) {
 static int
 _redirect(lua_State *L) {
 	struct skynet_context * context = lua_touserdata(L, lua_upvalueindex(1));
-	uint32_t dest = luaL_checkunsigned(L,1);
+	uint32_t dest = lua_tounsigned(L,1);
+	const char * dest_string = NULL;
+	if (dest == 0) {
+		dest_string = get_dest_string(L, 1);
+	}
 	uint32_t source = luaL_checkunsigned(L,2);
 	int type = luaL_checkinteger(L,3);
 	int session = luaL_checkinteger(L,4);
@@ -244,13 +218,21 @@ _redirect(lua_State *L) {
 		if (len == 0) {
 			msg = NULL;
 		}
-		session = skynet_send(context, source, dest, type, session , msg, len);
+		if (dest_string) {
+			session = skynet_sendname(context, source, dest_string, type, session , msg, len);
+		} else {
+			session = skynet_send(context, source, dest, type, session , msg, len);
+		}
 		break;
 	}
 	case LUA_TLIGHTUSERDATA: {
 		void * msg = lua_touserdata(L,5);
 		int size = luaL_checkinteger(L,6);
-		session = skynet_send(context, source, dest, type | PTYPE_TAG_DONTCOPY, session, msg, size);
+		if (dest_string) {
+			session = skynet_sendname(context, source, dest_string, type | PTYPE_TAG_DONTCOPY, session, msg, size);
+		} else {
+			session = skynet_send(context, source, dest, type | PTYPE_TAG_DONTCOPY, session, msg, size);
+		}
 		break;
 	}
 	default:
@@ -299,8 +281,28 @@ lpackstring(lua_State *L) {
 	return 1;
 }
 
+static int
+ltrash(lua_State *L) {
+	int t = lua_type(L,1);
+	switch (t) {
+	case LUA_TSTRING: {
+		break;
+	}
+	case LUA_TLIGHTUSERDATA: {
+		void * msg = lua_touserdata(L,1);
+		luaL_checkinteger(L,2);
+		skynet_free(msg);
+		break;
+	}
+	default:
+		luaL_error(L, "skynet.trash invalid param %s", lua_typename(L,t));
+	}
+
+	return 0;
+}
+
 int
-luaopen_skynet_c(lua_State *L) {
+luaopen_skynet_core(lua_State *L) {
 	luaL_checkversion(L);
 	
 	luaL_Reg l[] = {
@@ -314,6 +316,7 @@ luaopen_skynet_c(lua_State *L) {
 		{ "pack", _luaseri_pack },
 		{ "unpack", _luaseri_unpack },
 		{ "packstring", lpackstring },
+		{ "trash" , ltrash },
 		{ "callback", _callback },
 		{ NULL, NULL },
 	};

lualib-src/lua-socket.c

@@ -246,6 +246,7 @@ lclearbuffer(lua_State *L) {
 	while(sb->head) {
 		return_free_node(L,2,sb);
 	}
+	sb->size = 0;
 	return 0;
 }
 
@@ -264,6 +265,7 @@ lreadall(lua_State *L) {
 		return_free_node(L,2,sb);
 	}
 	luaL_pushresult(&b);
+	sb->size = 0;
 	return 1;
 }
 

lualib-src/lua-stm.c

@@ -0,0 +1,244 @@
+#include <lua.h>
+#include <lauxlib.h>
+#include <stdlib.h>
+#include <stdint.h>
+#include <assert.h>
+
+#include "rwlock.h"
+#include "skynet_malloc.h"
+
+struct stm_object {
+	struct rwlock lock;
+	int reference;
+	struct stm_copy * copy;
+};
+
+struct stm_copy {
+	int reference;
+	uint32_t sz;
+	void * msg;
+};
+
+// msg should alloc by skynet_malloc 
+static struct stm_copy *
+stm_newcopy(void * msg, int32_t sz) {
+	struct stm_copy * copy = skynet_malloc(sizeof(*copy));
+	copy->reference = 1;
+	copy->sz = sz;
+	copy->msg = msg;
+
+	return copy;
+}
+
+static struct stm_object *
+stm_new(void * msg, int32_t sz) {
+	struct stm_object * obj = skynet_malloc(sizeof(*obj));
+	rwlock_init(&obj->lock);
+	obj->reference = 1;
+	obj->copy = stm_newcopy(msg, sz);
+
+	return obj;
+}
+
+static void
+stm_releasecopy(struct stm_copy *copy) {
+	if (copy == NULL)
+		return;
+	if (__sync_sub_and_fetch(&copy->reference, 1) == 0) {
+		skynet_free(copy->msg);
+		skynet_free(copy);
+	}
+}
+
+static void
+stm_release(struct stm_object *obj) {
+	assert(obj->copy);
+	rwlock_wlock(&obj->lock);
+	// writer release the stm object, so release the last copy .
+	stm_releasecopy(obj->copy);
+	obj->copy = NULL;
+	if (--obj->reference > 0) {
+		// stm object grab by readers, reset the copy to NULL.
+		rwlock_wunlock(&obj->lock);
+		return;
+	}
+	// no one grab the stm object, no need to unlock wlock.
+	skynet_free(obj);
+}
+
+static void
+stm_releasereader(struct stm_object *obj) {
+	rwlock_rlock(&obj->lock);
+	if (__sync_sub_and_fetch(&obj->reference,1) == 0) {
+		// last reader, no writer. so no need to unlock
+		assert(obj->copy == NULL);
+		skynet_free(obj);
+		return;
+	}
+	rwlock_runlock(&obj->lock);
+}
+
+static void
+stm_grab(struct stm_object *obj) {
+	rwlock_rlock(&obj->lock);
+	int ref = __sync_fetch_and_add(&obj->reference,1);
+	rwlock_runlock(&obj->lock);
+	assert(ref > 0);
+}
+
+static struct stm_copy *
+stm_copy(struct stm_object *obj) {
+	rwlock_rlock(&obj->lock);
+	struct stm_copy * ret = obj->copy;
+	if (ret) {
+		int ref = __sync_fetch_and_add(&ret->reference,1);
+		assert(ref > 0);
+	}
+	rwlock_runlock(&obj->lock);
+	
+	return ret;
+}
+
+static void
+stm_update(struct stm_object *obj, void *msg, int32_t sz) {
+	struct stm_copy *copy = stm_newcopy(msg, sz);
+	rwlock_wlock(&obj->lock);
+	struct stm_copy *oldcopy = obj->copy;
+	obj->copy = copy;
+	rwlock_wunlock(&obj->lock);
+
+	stm_releasecopy(oldcopy);
+}
+
+// lua binding
+
+struct boxstm {
+	struct stm_object * obj;
+};
+
+static int
+lcopy(lua_State *L) {
+	struct boxstm * box = lua_touserdata(L, 1);
+	stm_grab(box->obj);
+	lua_pushlightuserdata(L, box->obj);
+	return 1;
+}
+
+static int
+lnewwriter(lua_State *L) {
+	void * msg = lua_touserdata(L, 1);
+	uint32_t sz = luaL_checkunsigned(L, 2);
+	struct boxstm * box = lua_newuserdata(L, sizeof(*box));
+	box->obj = stm_new(msg,sz);
+	lua_pushvalue(L, lua_upvalueindex(1));
+	lua_setmetatable(L, -2);
+
+	return 1;
+}
+
+static int
+ldeletewriter(lua_State *L) {
+	struct boxstm * box = lua_touserdata(L, 1);
+	stm_release(box->obj);
+	box->obj = NULL;
+
+	return 0;
+}
+
+static int
+lupdate(lua_State *L) {
+	struct boxstm * box = lua_touserdata(L, 1);
+	void * msg = lua_touserdata(L, 2);
+	uint32_t sz = luaL_checkunsigned(L, 3);
+	stm_update(box->obj, msg, sz);
+
+	return 0;
+}
+
+struct boxreader {
+	struct stm_object *obj;
+	struct stm_copy *lastcopy;
+};
+
+static int
+lnewreader(lua_State *L) {
+	struct boxreader * box = lua_newuserdata(L, sizeof(*box));
+	box->obj = lua_touserdata(L, 1);
+	box->lastcopy = NULL;
+	lua_pushvalue(L, lua_upvalueindex(1));
+	lua_setmetatable(L, -2);
+
+	return 1;
+}
+
+static int
+ldeletereader(lua_State *L) {
+	struct boxreader * box = lua_touserdata(L, 1);
+	stm_releasereader(box->obj);
+	box->obj = NULL;
+	stm_releasecopy(box->lastcopy);
+	box->lastcopy = NULL;
+
+	return 0;
+}
+
+static int
+lread(lua_State *L) {
+	struct boxreader * box = lua_touserdata(L, 1);
+	luaL_checktype(L, 2, LUA_TFUNCTION);
+	struct stm_copy * copy = stm_copy(box->obj);
+	if (copy == box->lastcopy) {
+		// not update
+		stm_releasecopy(copy);
+		lua_pushboolean(L, 0);
+		return 1;
+	}
+
+	stm_releasecopy(box->lastcopy);
+	box->lastcopy = copy;
+	if (copy) {
+		lua_settop(L, 2);
+		lua_pushlightuserdata(L, copy->msg);
+		lua_pushunsigned(L, copy->sz);
+		lua_call(L, 2, LUA_MULTRET);
+		lua_pushboolean(L, 1);
+		lua_replace(L, 1);
+		return lua_gettop(L);
+	} else {
+		lua_pushboolean(L, 0);
+		return 1;
+	}
+}
+
+int
+luaopen_stm(lua_State *L) {
+	luaL_checkversion(L);
+	lua_createtable(L, 0, 3);
+
+	lua_pushcfunction(L, lcopy);
+	lua_setfield(L, -2, "copy");
+
+	luaL_Reg writer[] = {
+		{ "new", lnewwriter },
+		{ NULL, NULL },
+	};
+	lua_createtable(L, 0, 2);
+	lua_pushcfunction(L, ldeletewriter),
+	lua_setfield(L, -2, "__gc");
+	lua_pushcfunction(L, lupdate),
+	lua_setfield(L, -2, "__call");
+	luaL_setfuncs(L, writer, 1);
+
+	luaL_Reg reader[] = {
+		{ "newcopy", lnewreader },
+		{ NULL, NULL },
+	};
+	lua_createtable(L, 0, 2);
+	lua_pushcfunction(L, ldeletereader),
+	lua_setfield(L, -2, "__gc");
+	lua_pushcfunction(L, lread),
+	lua_setfield(L, -2, "__call");
+	luaL_setfuncs(L, reader, 1);
+
+	return 1;
+}

lualib/cluster.lua

@@ -4,7 +4,7 @@ local clusterd
 local cluster = {}
 
 function cluster.call(node, address, ...)
-	-- skynet.pack(...) will free by cluster.c.packrequest
+	-- skynet.pack(...) will free by cluster.core.packrequest
 	return skynet.call(clusterd, "lua", "req", node, address, skynet.pack(...))
 end
 
@@ -20,6 +20,10 @@ function cluster.reload()
 	skynet.call(clusterd, "lua", "reload")
 end
 
+function cluster.proxy(node, name)
+	return skynet.call(clusterd, "lua", "proxy", node, name)
+end
+
 skynet.init(function()
 	clusterd = skynet.uniqueservice("clusterd")
 end)

lualib/http/httpc.lua

@@ -0,0 +1,114 @@
+local socket = require "http.sockethelper"
+local url = require "http.url"
+local internal = require "http.internal"
+local string = string
+local table = table
+
+local httpc = {}
+
+local function request(fd, method, host, url, recvheader, header, content)
+	local read = socket.readfunc(fd)
+	local write = socket.writefunc(fd)
+	local header_content = ""
+	if header then
+		for k,v in pairs(header) do
+			header_content = string.format("%s%s:%s\r\n", header_content, k, v)
+		end
+	end
+
+	if content then
+		local data = string.format("%s %s HTTP/1.1\r\nhost:%s\r\ncontent-length:%d\r\n%s\r\n%s", method, url, host, #content, header_content, content)
+		write(data)
+	else
+		local request_header = string.format("%s %s HTTP/1.1\r\nhost:%s\r\ncontent-length:0\r\n%s\r\n", method, url, host, header_content)
+		write(request_header)
+	end
+
+	local tmpline = {}
+	local body = internal.recvheader(read, tmpline, "")
+	if not body then
+		error(socket.socket_error)
+	end
+
+	local statusline = tmpline[1]
+	local code, info = statusline:match "HTTP/[%d%.]+%s+([%d]+)%s+(.*)$"
+	code = assert(tonumber(code))
+
+	local header = internal.parseheader(tmpline,2,recvheader or {})
+	if not header then
+		error("Invalid HTTP response header")
+	end
+
+	local length = header["content-length"]
+	if length then
+		length = tonumber(length)
+	end
+	local mode = header["transfer-encoding"]
+	if mode then
+		if mode ~= "identity" and mode ~= "chunked" then
+			error ("Unsupport transfer-encoding")
+		end
+	end
+
+	if mode == "chunked" then
+		body, header = internal.recvchunkedbody(read, nil, header, body)
+		if not body then
+			error("Invalid response body")
+		end
+	else
+		-- identity mode
+		if length then
+			if #body >= length then
+				body = body:sub(1,length)
+			else
+				local padding = read(length - #body)
+				body = body .. padding
+			end
+		else
+			body = nil
+		end
+	end
+
+	return code, body
+end
+
+function httpc.request(method, host, url, recvheader, header, content)
+	local hostname, port = host:match"([^:]+):?(%d*)$"
+	if port == "" then
+		port = 80
+	else
+		port = tonumber(port)
+	end
+	local fd = socket.connect(hostname, port)
+	local ok , statuscode, body = pcall(request, fd,method, host, url, recvheader, header, content)
+	if ok then
+		return statuscode, body
+	else
+		socket.close(fd)
+		error(statuscode)
+	end
+end
+
+function httpc.get(...)
+	return httpc.request("GET", ...)
+end
+
+local function escape(s)
+	return (string.gsub(s, "([^A-Za-z0-9_])", function(c)
+		return string.format("%%%02X", string.byte(c))
+	end))
+end
+
+function httpc.post(host, url, form, recvheader)
+	local header = {
+		["content-type"] = "application/x-www-form-urlencoded"
+	}
+	local body = {}
+	for k,v in pairs(form) do
+		table.insert(body, string.format("%s=%s",escape(k),escape(v)))
+	end
+
+	return httpc.request("POST", host, url, recvheader, header, table.concat(body , "&"))
+end
+
+return httpc

lualib/http/httpd.lua

@@ -1,3 +1,5 @@
+local internal = require "http.internal"
+
 local table = table
 
 local httpd = {}
@@ -45,140 +47,9 @@ local http_status_msg = {
 	[505] = "HTTP Version not supported",
 }
 
-local function recvheader(readbytes, limit, lines, header)
-	local result
-	local e = header:find("\r\n\r\n", 1, true)
-	if e then
-		result = header:sub(e+4)
-	else
-		while true do
-			local bytes = readbytes()
-			header = header .. bytes
-			if #header > limit then
-				return
-			end
-			e = header:find("\r\n\r\n", -#bytes-3, true)
-			if e then
-				result = header:sub(e+4)
-				break
-			end
-		end
-	end
-	local idx = 1
-	for v in header:gmatch("(.-)\r\n") do
-		if v == "" then
-			break
-		end
-		lines[idx] = v
-		idx = idx + 1
-	end
-	for i = idx, #lines do
-		lines[i] = nil
-	end
-	return result
-end
-
-local function parseheader(lines, from, header)
-	local name, value
-	for i=from,#lines do
-		local line = lines[i]
-		if line:byte(1) == 9 then	-- tab, append last line
-			if name == nil then
-				return
-			end
-			header[name] = header[name] .. line:sub(2)
-		else
-			name, value = line:match "^(.-):%s*(.*)"
-			if name == nil or value == nil then
-				return
-			end
-			name = name:lower()
-			if header[name] then
-				header[name] = header[name] .. ", " .. value
-			else
-				header[name] = value
-			end
-		end
-	end
-	return header
-end
-
-local function chunksize(readbytes, body)
-	while true do
-		if #body > 128 then
-			return
-		end
-		body = body .. readbytes()
-		local f,e = body:find("\r\n",1,true)
-		if f then
-			return tonumber(body:sub(1,f-1),16), body:sub(e+1)
-		end
-	end
-end
-
-local function readcrln(readbytes, body)
-	if #body > 2 then
-		if body:sub(1,2) ~= "\r\n" then
-			return
-		end
-		return body:sub(3)
-	else
-		body = body .. readbytes(2-#body)
-		if body ~= "\r\n" then
-			return
-		end
-		return ""
-	end
-end
-
-local tmpline = {}
-
-local function recvchunkedbody(readbytes, bodylimit, header, body)
-	local result = ""
-	local size = 0
-
-	while true do
-		local sz
-		sz , body = chunksize(readbytes, body)
-		if not sz then
-			return
-		end
-		if sz == 0 then
-			break
-		end
-		size = size + sz
-		if bodylimit and size > bodylimit then
-			return
-		end
-		if #body >= sz then
-			result = result .. body:sub(1,sz)
-			body = body:sub(sz+1)
-		else
-			result = result .. body .. readbytes(sz - #body)
-			body = ""
-		end
-		body = readcrln(readbytes, body)
-		if not body then
-			return
-		end
-	end
-
-	body = readcrln(readbytes, body)
-	if not body then
-		return
-	end
-	body = recvheader(readbytes, 8192, tmpline, body)
-	if not body then
-		return
-	end
-
-	header = parseheader(tmpline,1,header)
-
-	return result, header
-end
-
 local function readall(readbytes, bodylimit)
-	local body = recvheader(readbytes, 8192, tmpline, "")
+	local tmpline = {}
+	local body = internal.recvheader(readbytes, tmpline, "")
 	if not body then
 		return 413	-- Request Entity Too Large
 	end
@@ -189,7 +60,7 @@ local function readall(readbytes, bodylimit)
 	if httpver < 1.0 or httpver > 1.1 then
 		return 505	-- HTTP Version not supported
 	end
-	local header = parseheader(tmpline,2,{})
+	local header = internal.parseheader(tmpline,2,{})
 	if not header then
 		return 400	-- Bad request
 	end
@@ -199,13 +70,13 @@ local function readall(readbytes, bodylimit)
 	end
 	local mode = header["transfer-encoding"]
 	if mode then
-		if mode ~= "identity" or mode ~= "chunked" then
+		if mode ~= "identity" and mode ~= "chunked" then
 			return 501	-- Not Implemented
 		end
 	end
 
 	if mode == "chunked" then
-		body, header = recvchunkedbody(readbytes, bodylimit, header, body)
+		body, header = internal.recvchunkedbody(readbytes, bodylimit, header, body)
 		if not body then
 			return 413
 		end

lualib/http/internal.lua

@@ -0,0 +1,135 @@
+local M = {}
+
+local LIMIT = 8192
+
+local function chunksize(readbytes, body)
+	while true do
+		if #body > 128 then
+			return
+		end
+		body = body .. readbytes()
+		local f,e = body:find("\r\n",1,true)
+		if f then
+			return tonumber(body:sub(1,f-1),16), body:sub(e+1)
+		end
+	end
+end
+
+local function readcrln(readbytes, body)
+	if #body >= 2 then
+		if body:sub(1,2) ~= "\r\n" then
+			return
+		end
+		return body:sub(3)
+	else
+		body = body .. readbytes(2-#body)
+		if body ~= "\r\n" then
+			return
+		end
+		return ""
+	end
+end
+
+function M.recvheader(readbytes, lines, header)
+	if #header >= 2 then
+		if header:find "^\r\n" then
+			return header:sub(3)
+		end
+	end
+	local result
+	local e = header:find("\r\n\r\n", 1, true)
+	if e then
+		result = header:sub(e+4)
+	else
+		while true do
+			local bytes = readbytes()
+			header = header .. bytes
+			if #header > LIMIT then
+				return
+			end
+			e = header:find("\r\n\r\n", -#bytes-3, true)
+			if e then
+				result = header:sub(e+4)
+				break
+			end
+			if header:find "^\r\n" then
+				return header:sub(3)
+			end
+		end
+	end
+	for v in header:gmatch("(.-)\r\n") do
+		if v == "" then
+			break
+		end
+		table.insert(lines, v)
+	end
+	return result
+end
+
+function M.parseheader(lines, from, header)
+	local name, value
+	for i=from,#lines do
+		local line = lines[i]
+		if line:byte(1) == 9 then	-- tab, append last line
+			if name == nil then
+				return
+			end
+			header[name] = header[name] .. line:sub(2)
+		else
+			name, value = line:match "^(.-):%s*(.*)"
+			if name == nil or value == nil then
+				return
+			end
+			name = name:lower()
+			if header[name] then
+				header[name] = header[name] .. ", " .. value
+			else
+				header[name] = value
+			end
+		end
+	end
+	return header
+end
+
+function M.recvchunkedbody(readbytes, bodylimit, header, body)
+	local result = ""
+	local size = 0
+
+	while true do
+		local sz
+		sz , body = chunksize(readbytes, body)
+		if not sz then
+			return
+		end
+		if sz == 0 then
+			break
+		end
+		size = size + sz
+		if bodylimit and size > bodylimit then
+			return
+		end
+		if #body >= sz then
+			result = result .. body:sub(1,sz)
+			body = body:sub(sz+1)
+		else
+			result = result .. body .. readbytes(sz - #body)
+			body = ""
+		end
+		body = readcrln(readbytes, body)
+		if not body then
+			return
+		end
+	end
+
+	local tmpline = {}
+	body = M.recvheader(readbytes, tmpline, body)
+	if not body then
+		return
+	end
+
+	header = M.parseheader(tmpline,1,header)
+
+	return result, header
+end
+
+return M

lualib/http/sockethelper.lua

@@ -28,4 +28,16 @@ function sockethelper.writefunc(fd)
 	end
 end
 
+function sockethelper.connect(host, port)
+	local fd = socket.open(host, port)
+	if fd then
+		return fd
+	end
+	error(socket_error)
+end
+
+function sockethelper.close(fd)
+	socket.close(fd)
+end
+
 return sockethelper

lualib/mqueue.lua

@@ -1,7 +1,7 @@
 -- This is a deprecated module, use skynet.queue instead.
 
 local skynet = require "skynet"
-local c = require "skynet.c"
+local c = require "skynet.core"
 
 local mqueue = {}
 local init_once

lualib/multicast.lua

@@ -1,5 +1,5 @@
 local skynet = require "skynet"
-local mc = require "multicast.c"
+local mc = require "multicast.core"
 
 local multicastd
 local multicast = {}

lualib/sharedata.lua

@@ -0,0 +1,43 @@
+local skynet = require "skynet"
+local sd = require "sharedata.corelib"
+
+local service
+
+skynet.init(function()
+	service = skynet.uniqueservice "sharedatad"
+end)
+
+local sharedata = {}
+
+local function monitor(name, obj, cobj)
+	local newobj = cobj
+	while true do
+		newobj = skynet.call(service, "lua", "monitor", name, newobj)
+		if newobj == nil then
+			break
+		end
+		sd.update(obj, newobj)
+	end
+end
+
+function sharedata.query(name)
+	local obj = skynet.call(service, "lua", "query", name)
+	local r = sd.box(obj)
+	skynet.send(service, "lua", "confirm" , obj)
+	skynet.fork(monitor,name, r, obj)
+	return r
+end
+
+function sharedata.new(name, v)
+	skynet.call(service, "lua", "new", name, v)
+end
+
+function sharedata.update(name, v)
+	skynet.call(service, "lua", "update", name, v)
+end
+
+function sharedata.delete(name)
+	skynet.call(service, "lua", "delete", name)
+end
+
+return sharedata

lualib/sharedata/corelib.lua

@@ -0,0 +1,134 @@
+local core = require "sharedata.core"
+local type = type
+local next = next
+local rawget = rawget
+
+local conf = {}
+
+conf.host = {
+	new = core.new,
+	delete = core.delete,
+	getref = core.getref,
+	markdirty = core.markdirty,
+	incref = core.incref,
+	decref = core.decref,
+}
+
+local meta = {}
+
+local isdirty = core.isdirty
+local index = core.index
+local needupdate = core.needupdate
+local len = core.len
+
+local function findroot(self)
+	while self.__parent do
+		self = self.__parent
+	end
+	return self
+end
+
+local function update(root, cobj, gcobj)
+	root.__obj = cobj
+	root.__gcobj = gcobj
+	-- don't use pairs
+	for k,v in next, root do
+		if type(v)=="table" and k~="__parent" then
+			local pointer = index(cobj, k)
+			if type(pointer) == "userdata" then
+				update(v, pointer, gcobj)
+			else
+				root[k] = nil
+			end
+		end
+	end
+end
+
+local function genkey(self)
+	local key = tostring(self.__key)
+	while self.__parent do
+		self = self.__parent
+		key = self.__key .. "." .. key
+	end
+	return key
+end
+
+function meta:__index(key)
+	local obj = self.__obj
+	if isdirty(obj) then
+		local newobj, newtbl = needupdate(self.__gcobj)
+		if newobj then
+			local newgcobj = newtbl.__gcobj
+			-- todo: update
+			local root = findroot(self)
+			update(root, newobj, newgcobj)
+			if obj == self.__obj then
+				error ("The key [" .. genkey(self) .. "] doesn't exist after update")
+			end
+		end
+	end
+	local v = index(self.__obj, key)
+	if type(v) == "userdata" then
+		local r = setmetatable({
+			__obj = v,
+			__gcobj = self.__gcobj,
+			__parent = self,
+			__key = key,
+		}, meta)
+		self[key] = r
+		return r
+	else
+		return v
+	end
+end
+
+function meta:__len()
+	return len(self.__obj)
+end
+
+local function conf_ipairs(self, index)
+	local obj = self.__obj
+	index = index + 1
+	local value =  rawget(self, index)
+	if value then
+		return index, value
+	end
+	local sz = len(obj)
+	if sz < index then
+		return
+	end
+	return index, self[index]
+end
+
+function meta:__ipairs()
+	return conf_ipairs, self, 0
+end
+
+function meta:__pairs()
+	return conf.next, self, nil
+end
+
+function conf.next(obj, key)
+	local nextkey = core.nextkey(obj.__obj, key)
+	if nextkey then
+		return nextkey, obj[nextkey]
+	end
+end
+
+function conf.box(obj)
+	local gcobj = core.box(obj)
+	return setmetatable({
+		__parent = false,
+		__obj = obj,
+		__gcobj = gcobj,
+		__key = "",
+	} , meta)
+end
+
+function conf.update(self, pointer)
+	local cobj = self.__obj
+	assert(isdirty(cobj), "Obly dirty object can be update")
+	core.update(self.__gcobj, pointer, { __gcobj = core.box(pointer) })
+end
+
+return conf

lualib/skynet.lua

@@ -1,4 +1,4 @@
-local c = require "skynet.c"
+local c = require "skynet.core"
 local tostring = tostring
 local tonumber = tonumber
 local coroutine = coroutine
@@ -43,12 +43,14 @@ end
 local session_id_coroutine = {}
 local session_coroutine_id = {}
 local session_coroutine_address = {}
+local session_response = {}
 
 local wakeup_session = {}
 local sleep_session = {}
 
 local watching_service = {}
 local watching_session = {}
+local dead_service = {}
 local error_queue = {}
 
 -- suspend is function
@@ -73,7 +75,9 @@ local function _error_dispatch(error_session, error_source)
 	if error_session == 0 then
 		-- service is down
 		--  Don't remove from watching_service , because user may call dead service
-		watching_service[error_source] = false
+		if watching_service[error_source] then
+			dead_service[error_source] = true
+		end
 		for session, srv in pairs(watching_session) do
 			if srv == error_source then
 				table.insert(error_queue, session)
@@ -95,6 +99,7 @@ local coroutine_yield = coroutine.yield
 local function co_create(f)
 	local co = table.remove(coroutine_pool)
 	if co == nil then
+		local print = print
 		co = coroutine.create(function(...)
 			f(...)
 			while true do
@@ -122,12 +127,25 @@ local function dispatch_wakeup()
 	end
 end
 
+local function release_watching(address)
+	local ref = watching_service[address]
+	if ref then
+		ref = ref - 1
+		if ref > 0 then
+			watching_service[address] = ref
+		else
+			watching_service[address] = nil
+		end
+	end
+end
+
 -- suspend is local function
 function suspend(co, result, command, param, size)
 	if not result then
 		local session = session_coroutine_id[co]
 		local addr = session_coroutine_address[co]
-		if session then
+		if session ~= 0 then
+			-- only call response error
 			c.send(addr, skynet.PTYPE_ERROR, session, "")
 		end
 		session_coroutine_id[co] = nil
@@ -142,15 +160,67 @@ function suspend(co, result, command, param, size)
 	elseif command == "RETURN" then
 		local co_session = session_coroutine_id[co]
 		local co_address = session_coroutine_address[co]
-		if param == nil then
+		if param == nil or session_response[co] then
 			error(debug.traceback(co))
 		end
-		c.send(co_address, skynet.PTYPE_RESPONSE, co_session, param, size)
+		session_response[co] = true
-		return suspend(co, coroutine.resume(co))
+		local ret
+		if not dead_service[co_address] then
+			ret = c.send(co_address, skynet.PTYPE_RESPONSE, co_session, param, size) >= 0
+		elseif size == nil then
+			c.trash(param, size)
+			ret = false
+		end
+		return suspend(co, coroutine.resume(co, ret))
+	elseif command == "RESPONSE" then
+		local co_session = session_coroutine_id[co]
+		local co_address = session_coroutine_address[co]
+		if session_response[co] then
+			error(debug.traceback(co))
+		end
+		local f = param
+		local function response(ok, ...)
+			if ok == "TEST" then
+				if dead_service[co_address] then
+					release_watching(co_address)
+					f = false
+					return false
+				else
+					return true
+				end
+			end
+			if not f then
+				if f == false then
+					f = nil
+					return false
+				end
+				error "Can't response more than once"
+			end
+
+			local ret
+			if not dead_service[co_address] then
+				if ok then
+					ret = c.send(co_address, skynet.PTYPE_RESPONSE, co_session, f(...)) >=0
+				else
+					ret = c.send(co_address, skynet.PTYPE_ERROR, co_session, "") >=0
+				end
+			else
+				ret = false
+			end
+			release_watching(co_address)
+			f = nil
+			return ret
+		end
+		watching_service[co_address] = watching_service[co_address] + 1
+		session_response[co] = response
+		return suspend(co, coroutine.resume(co, response))
 	elseif command == "EXIT" then
 		-- coroutine exit
+		local address = session_coroutine_address[co]
+		release_watching(address)
 		session_coroutine_id[co] = nil
 		session_coroutine_address[co] = nil
+		session_response[co] = nil
 	elseif command == "QUIT" then
 		-- service exit
 		return
@@ -259,13 +329,21 @@ end
 
 function skynet.exit()
 	skynet.send(".launcher","lua","REMOVE",skynet.self())
+	-- report the sources that call me
 	for co, session in pairs(session_coroutine_id) do
 		local address = session_coroutine_address[co]
-		local self = skynet.self()
 		if session~=0 and address then
-			skynet.redirect(address, self, "error", session, "")
+			c.redirect(address, 0, skynet.PTYPE_ERROR, session, "")
 		end
 	end
+	-- report the sources I call but haven't return
+	local tmp = {}
+	for session, address in pairs(watching_session) do
+		tmp[address] = true
+	end
+	for address in pairs(tmp) do
+		c.redirect(address, 0, skynet.PTYPE_ERROR, 0, "")
+	end
 	c.command("EXIT")
 	-- quit service
 	coroutine_yield "QUIT"
@@ -294,9 +372,6 @@ end
 
 function skynet.send(addr, typename, ...)
 	local p = proto[typename]
-	if watching_service[addr] == false then
-		error("Service is dead")
-	end
 	return c.send(addr, p.id, 0 , p.pack(...))
 end
 
@@ -321,9 +396,6 @@ end
 
 function skynet.call(addr, typename, ...)
 	local p = proto[typename]
-	if watching_service[addr] == false then
-		error("Service is dead")
-	end
 	local session = c.send(addr, p.id , nil , p.pack(...))
 	if session == nil then
 		error("call to invalid address " .. skynet.address(addr))
@@ -333,16 +405,18 @@ end
 
 function skynet.rawcall(addr, typename, msg, sz)
 	local p = proto[typename]
-	if watching_service[addr] == false then
-		error("Service is dead")
-	end
 	local session = assert(c.send(addr, p.id , nil , msg, sz), "call to invalid address")
 	return yield_call(addr, session)
 end
 
 function skynet.ret(msg, sz)
 	msg = msg or ""
-	coroutine_yield("RETURN", msg, sz)
+	return coroutine_yield("RETURN", msg, sz)
+end
+
+function skynet.response(pack)
+	pack = pack or skynet.pack
+	return coroutine_yield("RESPONSE", pack)
 end
 
 function skynet.retpack(...)
@@ -412,6 +486,12 @@ local function raw_dispatch_message(prototype, msg, sz, session, source, ...)
 		local p = assert(proto[prototype], prototype)
 		local f = p.dispatch
 		if f then
+			local ref = watching_service[source]
+			if ref then
+				watching_service[source] = ref + 1
+			else
+				watching_service[source] = 1
+			end
 			local co = co_create(f)
 			session_coroutine_id[co] = session
 			session_coroutine_address[co] = source
@@ -470,7 +550,7 @@ end
 
 function skynet.address(addr)
 	if type(addr) == "number" then
-		return string.format(":%x",addr)
+		return string.format(":%08x",addr)
 	else
 		return tostring(addr)
 	end
@@ -570,6 +650,21 @@ function skynet.filter(f ,start_func)
 	end)
 end
 
+function skynet.forward_type(map, start_func)
+	c.callback(function(ptype, msg, sz, ...)
+		local prototype = map[ptype]
+		if prototype then
+			dispatch_message(prototype, msg, sz, ...)
+		else
+			dispatch_message(ptype, msg, sz, ...)
+			c.trash(msg, sz)
+		end
+	end, true)
+	skynet.timeout(0, function()
+		init_service(start_func)
+	end)
+end
+
 function skynet.endless()
 	return c.command("ENDLESS")~=nil
 end
@@ -607,6 +702,6 @@ end
 
 -- Inject internal debug framework
 local debug = require "skynet.debug"
-debug(skynet)
+debug(skynet, dispatch_message)
 
 return skynet

lualib/skynet/debug.lua

@@ -1,4 +1,8 @@
-return function (skynet)
+local io = io
+local table = table
+local debug = debug
+
+return function (skynet, dispatch_func)
 
 local internal_info_func
 
@@ -39,6 +43,17 @@ function dbgcmd.INFO()
 	end
 end
 
+function dbgcmd.EXIT()
+	skynet.exit()
+end
+
+function dbgcmd.RUN(source, filename)
+	local inject = require "skynet.inject"
+	local output = inject(source, filename , dispatch_func, skynet.register_protocol)
+	collectgarbage "collect"
+	skynet.ret(skynet.pack(table.concat(output, "\n")))
+end
+
 local function _debug_dispatch(session, address, cmd, ...)
 	local f = dbgcmd[cmd]
 	assert(f, cmd)
@@ -53,4 +68,4 @@ skynet.register_protocol {
 	dispatch = _debug_dispatch,
 }
 
-end
+end

lualib/skynet/harbor.lua

@@ -4,15 +4,19 @@ local harbor = {}
 
 function harbor.globalname(name, handle)
 	handle = handle or skynet.self()
-	skynet.send(".slave", "lua", "REGISTER", name, handle)
+	skynet.send(".cslave", "lua", "REGISTER", name, handle)
+end
+
+function harbor.queryname(name)
+	return skynet.call(".cslave", "lua", "QUERYNAME", name)
 end
 
 function harbor.link(id)
-	skynet.call(".slave", "lua", "LINK", id)
+	skynet.call(".cslave", "lua", "LINK", id)
 end
 
 function harbor.connect(id)
-	skynet.call(".slave", "lua", "CONNECT", id)
+	skynet.call(".cslave", "lua", "CONNECT", id)
 end
 
 return harbor

lualib/skynet/inject.lua

@@ -0,0 +1,65 @@
+local function getupvaluetable(u, func, unique)
+	i = 1
+	while true do
+		local name, value = debug.getupvalue(func, i)
+		if name == nil then
+			return
+		end
+		local t = type(value)
+		if t == "table" then
+			u[name] = value
+		elseif t == "function" then
+			if not unique[value] then
+				unique[value] = true
+				getupvaluetable(u, value, unique)
+			end
+		end
+		i=i+1
+	end
+end
+
+return function(source, filename , ...)
+	if filename then
+		filename = "@" .. filename
+	else
+		filename = "=(load)"
+	end
+	local output = {}
+
+	local function print(...)
+		local value = { ... }
+		for k,v in ipairs(value) do
+			value[k] = tostring(v)
+		end
+		table.insert(output, table.concat(value, "\t"))
+	end
+	local u = {}
+	local unique = {}
+	local funcs = { ... }
+	for k, func in ipairs(funcs) do
+		getupvaluetable(u, func, unique)
+	end
+	local p = {}
+	local proto = u.proto
+	if proto then
+		for k,v in pairs(proto) do
+			local name, dispatch = v.name, v.dispatch
+			if name and dispatch then
+				local pp = {}
+				p[name] = pp
+				getupvaluetable(pp, dispatch, unique)
+			end
+		end
+	end
+	local env = setmetatable( { print = print , _U = u, _P = p}, { __index = _ENV })
+	local func, err = load(source, filename, "bt", env)
+	if not func then
+		return { err }
+	end
+	local ok, err = xpcall(func, debug.traceback)
+	if not ok then
+		table.insert(output, err)
+	end
+
+	return output
+end

lualib/snax/hotfix.lua

@@ -3,7 +3,21 @@ local io = io
 
 local hotfix = {}
 
-local function collect_uv(f , uv)
+local function envid(f)
+	local i = 1
+	while true do
+		local name, value = debug.getupvalue(f, i)
+		if name == nil then
+			return
+		end
+		if name == "_ENV" then
+			return debug.upvalueid(f, i)
+		end
+		i = i + 1
+	end
+end
+
+local function collect_uv(f , uv, env)
 	local i = 1
 	while true do
 		local name, value = debug.getupvalue(f, i)
@@ -18,7 +32,9 @@ local function collect_uv(f , uv)
 			uv[name] = { func = f, index = i, id = id }
 
 			if type(value) == "function" then
-				collect_uv(value, uv)
+				if envid(value) == env then
+					collect_uv(value, uv, env)
+				end
 			end
 		end
 
@@ -30,7 +46,7 @@ local function collect_all_uv(funcs)
 	local global = {}
 	for _, v in pairs(funcs) do
 		if v[4] then
-			collect_uv(v[4], global)
+			collect_uv(v[4], global, envid(v[4]))
 		end
 	end
 

lualib/snax/interface.lua

@@ -61,6 +61,8 @@ return function (name , G, loader)
 
 	setmetatable(G,	{ __index = env , __newindex = init_system })
 
+	local pattern
+
 	do
 		local path = skynet.getenv "snax"
 
@@ -70,6 +72,7 @@ return function (name , G, loader)
 			local filename = string.gsub(pat, "?", name)
 			local f , err = loader(filename, "bt", G)
 			if f then
+				pattern = pat
 				mainfunc = f
 				break
 			else
@@ -90,5 +93,5 @@ return function (name , G, loader)
 		G[k] = v
 	end
 
-	return func
+	return func, pattern
 end

lualib/snax/loginserver.lua

@@ -143,16 +143,11 @@ local function launch_master(conf)
 	local balance = 1
 
 	skynet.dispatch("lua", function(_,source,command, ...)
-		if command == "register_slave" then
+		skynet.ret(skynet.pack(conf.command_handler(command, ...)))
-			table.insert(slave, source)
-			skynet.ret(skynet.pack(nil))
-		else
-			skynet.ret(skynet.pack(conf.command_handler(command, ...)))
-		end
 	end)
 
 	for i=1,instance do
-		skynet.newservice(SERVICE_NAME)
+		table.insert(slave, skynet.newservice(SERVICE_NAME))
 	end
 
 	skynet.error(string.format("login server listen at : %s %d", host, port))
@@ -178,7 +173,6 @@ local function login(conf)
 	skynet.start(function()
 		local loginmaster = skynet.localname(name)
 		if loginmaster then
-			skynet.call(loginmaster, "lua", "register_slave")
 			local auth_handler = assert(conf.auth_handler)
 			launch_master = nil
 			conf = nil

lualib/socket.lua

@@ -145,6 +145,8 @@ local function connect(id, func)
 	suspend(s)
 	if s.connected then
 		return id
+	else
+		socket_pool[id] = nil
 	end
 end
 

lualib/socketchannel.lua

@@ -218,9 +218,9 @@ local function try_connect(self , once)
 			if not once then
 				skynet.error("socket: connect to", self.__host, self.__port)
 			end
-			return
+			return true
 		elseif once then
-			error(string.format("Connect to %s:%d failed", self.__host, self.__port))
+			return false
 		end
 		if t > 1000 then
 			skynet.error("socket: try to reconnect", self.__host, self.__port)
@@ -233,7 +233,7 @@ local function try_connect(self , once)
 	end
 end
 
-local function block_connect(self, once)
+local function check_connection(self)
 	if self.__sock then
 		local authco = self.__authcoroutine
 		if not authco then
@@ -247,26 +247,36 @@ local function block_connect(self, once)
 	if self.__closed then
 		return false
 	end
+end
+
+local function block_connect(self, once)
+	local r = check_connection(self)
+	if r ~= nil then
+		return r
+	end
 
 	if #self.__connecting > 0 then
 		-- connecting in other coroutine
 		local co = coroutine.running()
 		table.insert(self.__connecting, co)
 		skynet.wait()
-		-- check connection again
+	else
-		return block_connect(self, once)
+		self.__connecting[1] = true
-	end
+		try_connect(self, once)
-	self.__connecting[1] = true
+		self.__connecting[1] = nil
-	try_connect(self, once)
+		for i=2, #self.__connecting do
-	self.__connecting[1] = nil
+			local co = self.__connecting[i]
-	for i=2, #self.__connecting do
+			self.__connecting[i] = nil
-		local co = self.__connecting[i]
+			skynet.wakeup(co)
-		self.__connecting[i] = nil
+		end
-		skynet.wakeup(co)
 	end
 
-	-- check again
+	r = check_connection(self)
-	return block_connect(self, once)
+	if r == nil then
+		error(string.format("Connect to %s:%d failed", self.__host, self.__port))
+	else
+		return r
+	end
 end
 
 function channel:connect(once)
@@ -296,7 +306,7 @@ local function wait_for_response(self, response)
 end
 
 function channel:request(request, response)
-	assert(block_connect(self))
+	assert(block_connect(self, true))	-- connect once
 
 	if not socket.write(self.__sock[1], request) then
 		close_channel_socket(self)

service-src/service_harbor.c

@@ -36,17 +36,17 @@ struct remote_message_header {
 	uint32_t session;
 };
 
-struct msg {
+struct harbor_msg {
 	struct remote_message_header header;
 	void * buffer;
 	size_t size;
 };
 
-struct msg_queue {
+struct harbor_msg_queue {
 	int size;
 	int head;
 	int tail;
-	struct msg * data;
+	struct harbor_msg * data;
 };
 
 struct keyvalue {
@@ -54,7 +54,7 @@ struct keyvalue {
 	char key[GLOBALNAME_LENGTH];
 	uint32_t hash;
 	uint32_t value;
-	struct msg_queue * queue;
+	struct harbor_msg_queue * queue;
 };
 
 struct hashmap {
@@ -69,7 +69,7 @@ struct hashmap {
 
 struct slave {
 	int fd;
-	struct msg_queue *queue;
+	struct harbor_msg_queue *queue;
 	int status;
 	int length;
 	int read;
@@ -88,11 +88,11 @@ struct harbor {
 // hash table
 
 static void
-push_queue_msg(struct msg_queue * queue, struct msg * m) {
+push_queue_msg(struct harbor_msg_queue * queue, struct harbor_msg * m) {
 	// If there is only 1 free slot which is reserved to distinguish full/empty
 	// of circular buffer, expand it.
 	if (((queue->tail + 1) % queue->size) == queue->head) {
-		struct msg * new_buffer = skynet_malloc(queue->size * 2 * sizeof(struct msg));
+		struct harbor_msg * new_buffer = skynet_malloc(queue->size * 2 * sizeof(struct harbor_msg));
 		int i;
 		for (i=0;i<queue->size-1;i++) {
 			new_buffer[i] = queue->data[(i+queue->head) % queue->size];
@@ -103,46 +103,46 @@ push_queue_msg(struct msg_queue * queue, struct msg * m) {
 		queue->tail = queue->size - 1;
 		queue->size *= 2;
 	}
-	struct msg * slot = &queue->data[queue->tail];
+	struct harbor_msg * slot = &queue->data[queue->tail];
 	*slot = *m;
 	queue->tail = (queue->tail + 1) % queue->size;
 }
 
 static void
-push_queue(struct msg_queue * queue, void * buffer, size_t sz, struct remote_message_header * header) {
+push_queue(struct harbor_msg_queue * queue, void * buffer, size_t sz, struct remote_message_header * header) {
-	struct msg m;
+	struct harbor_msg m;
 	m.header = *header;
 	m.buffer = buffer;
 	m.size = sz;
 	push_queue_msg(queue, &m);
 }
 
-static struct msg *
+static struct harbor_msg *
-pop_queue(struct msg_queue * queue) {
+pop_queue(struct harbor_msg_queue * queue) {
 	if (queue->head == queue->tail) {
 		return NULL;
 	}
-	struct msg * slot = &queue->data[queue->head];
+	struct harbor_msg * slot = &queue->data[queue->head];
 	queue->head = (queue->head + 1) % queue->size;
 	return slot;
 }
 
-static struct msg_queue *
+static struct harbor_msg_queue *
 new_queue() {
-	struct msg_queue * queue = skynet_malloc(sizeof(*queue));
+	struct harbor_msg_queue * queue = skynet_malloc(sizeof(*queue));
 	queue->size = DEFAULT_QUEUE_SIZE;
 	queue->head = 0;
 	queue->tail = 0;
-	queue->data = skynet_malloc(DEFAULT_QUEUE_SIZE * sizeof(struct msg));
+	queue->data = skynet_malloc(DEFAULT_QUEUE_SIZE * sizeof(struct harbor_msg));
 
 	return queue;
 }
 
 static void
-release_queue(struct msg_queue *queue) {
+release_queue(struct harbor_msg_queue *queue) {
 	if (queue == NULL)
 		return;
-	struct msg * m;
+	struct harbor_msg * m;
 	while ((m=pop_queue(queue)) != NULL) {
 		skynet_free(m->buffer);
 	}
@@ -334,7 +334,7 @@ send_remote(struct skynet_context * ctx, int fd, const char * buffer, size_t sz,
 
 static void
 dispatch_name_queue(struct harbor *h, struct keyvalue * node) {
-	struct msg_queue * queue = node->queue;
+	struct harbor_msg_queue * queue = node->queue;
 	uint32_t handle = node->value;
 	int harbor_id = handle >> HANDLE_REMOTE_SHIFT;
 	assert(harbor_id != 0);
@@ -352,7 +352,7 @@ dispatch_name_queue(struct harbor *h, struct keyvalue * node) {
 				s->queue = node->queue;
 				node->queue = NULL;
 			} else {
-				struct msg * m;
+				struct harbor_msg * m;
 				while ((m = pop_queue(queue))!=NULL) {
 					push_queue_msg(s->queue, m);
 				}
@@ -360,7 +360,7 @@ dispatch_name_queue(struct harbor *h, struct keyvalue * node) {
 		}
 		return;
 	}
-	struct msg * m;
+	struct harbor_msg * m;
 	while ((m = pop_queue(queue)) != NULL) {
 		m->header.destination |= (handle & HANDLE_MASK);
 		send_remote(context, fd, m->buffer, m->size, &m->header);
@@ -373,11 +373,11 @@ dispatch_queue(struct harbor *h, int id) {
 	int fd = s->fd;
 	assert(fd != 0);
 
-	struct msg_queue *queue = s->queue;
+	struct harbor_msg_queue *queue = s->queue;
 	if (queue == NULL)
 		return;
 
-	struct msg * m;
+	struct harbor_msg * m;
 	while ((m = pop_queue(queue)) != NULL) {
 		send_remote(h->ctx, fd, m->buffer, m->size, &m->header);
 	}

service-src/service_snlua.c

@@ -55,7 +55,7 @@ traceback (lua_State *L) {
 static void
 _report_launcher_error(struct skynet_context *ctx) {
 	// sizeof "ERROR" == 5
-	skynet_sendname(ctx, ".launcher", PTYPE_TEXT, 0, "ERROR", 5);
+	skynet_sendname(ctx, 0, ".launcher", PTYPE_TEXT, 0, "ERROR", 5);
 }
 
 static const char *

service/bootstrap.lua

@@ -13,31 +13,31 @@ skynet.start(function()
 		standalone = true
 		skynet.setenv("standalone", "true")
 
-		local slave = skynet.newservice "cdummy"
+		local ok, slave = pcall(skynet.newservice, "cdummy")
-		if slave == nil then
+		if not ok then
 			skynet.abort()
 		end
-		skynet.name(".slave", slave)
+		skynet.name(".cslave", slave)
 
 	else
 		if standalone then
-			if not skynet.newservice "cmaster" then
+			if not pcall(skynet.newservice,"cmaster") then
 				skynet.abort()
 			end
 		end
 
-		local slave = skynet.newservice "cslave"
+		local ok, slave = pcall(skynet.newservice, "cslave")
-		if slave == nil then
+		if not ok then
 			skynet.abort()
 		end
-		skynet.name(".slave", slave)
+		skynet.name(".cslave", slave)
 	end
 
 	if standalone then
-		local datacenter = assert(skynet.newservice "datacenterd")
+		local datacenter = skynet.newservice "datacenterd"
 		skynet.name("DATACENTER", datacenter)
 	end
-	assert(skynet.newservice "service_mgr")
+	skynet.newservice "service_mgr"
-	assert(skynet.newservice(skynet.getenv "start" or "main"))
+	pcall(skynet.newservice,skynet.getenv "start" or "main")
 	skynet.exit()
 end)

service/cdummy.lua

@@ -1,6 +1,7 @@
 local skynet = require "skynet"
 
 local globalname = {}
+local queryname = {}
 local harbor = {}
 
 skynet.register_protocol {
@@ -17,12 +18,43 @@ skynet.register_protocol {
 	unpack = skynet.tostring,
 }
 
+local function response_name(name)
+	local address = globalname[name]
+	if queryname[name] then
+		local tmp = queryname[name]
+		queryname[name] = nil
+		for _,resp in ipairs(tmp) do
+			resp(true, address)
+		end
+	end
+end
+
 function harbor.REGISTER(name, handle)
 	assert(globalname[name] == nil)
 	globalname[name] = handle
+	response_name(name)
 	skynet.redirect(harbor_service, handle, "harbor", 0, "N " .. name)
 end
 
+function harbor.QUERYNAME(name)
+	if name:byte() == 46 then	-- "." , local name
+		skynet.ret(skynet.pack(skynet.localname(name)))
+		return
+	end
+	local result = globalname[name]
+	if result then
+		skynet.ret(skynet.pack(result))
+		return
+	end
+	local queue = queryname[name]
+	if queue == nil then
+		queue = { skynet.response() }
+		queryname[name] = queue
+	else
+		table.insert(queue, skynet.response())
+	end
+end
+
 function harbor.LINK(id)
 	skynet.ret()
 end

service/clusterd.lua

@@ -1,7 +1,7 @@
 local skynet = require "skynet"
 local sc = require "socketchannel"
 local socket = require "socket"
-local cluster = require "cluster.c"
+local cluster = require "cluster.core"
 
 local config_name = skynet.getenv "cluster"
 local node_address = {}
@@ -51,13 +51,34 @@ function command.listen(source, addr, port)
 	skynet.ret(skynet.pack(nil))
 end
 
-function command.req(source, node, addr, msg, sz)
+local function send_request(source, node, addr, msg, sz)
 	local request
 	local c = node_channel[node]
 	local session = node_session[node]
 	-- msg is a local pointer, cluster.packrequest will free it
 	request, node_session[node] = cluster.packrequest(addr, session , msg, sz)
-	skynet.ret(c:request(request, session))
+
+	return c:request(request, session)
+end
+
+function command.req(...)
+	local ok, msg, sz = pcall(send_request, ...)
+	if ok then
+		skynet.ret(msg, sz)
+	else
+		skynet.error(msg)
+		skynet.response()(false)
+	end
+end
+
+local proxy = {}
+
+function command.proxy(source, node, name)
+	local fullname = node .. "." .. name
+	if proxy[fullname] == nil then
+		proxy[fullname] = skynet.newservice("clusterproxy", node, name)
+	end
+	skynet.ret(skynet.pack(proxy[fullname]))
 end
 
 local request_fd = {}

service/clusterproxy.lua

@@ -0,0 +1,27 @@
+local skynet = require "skynet"
+local cluster = require "cluster"
+
+local node, address = ...
+
+skynet.register_protocol {
+	name = "system",
+	id = skynet.PTYPE_SYSTEM,
+	unpack = function (...) return ... end,
+}
+
+local forward_map = {
+	[skynet.PTYPE_LUA] = skynet.PTYPE_SYSTEM,
+	[skynet.PTYPE_RESPONSE] = skynet.PTYPE_RESPONSE,	-- don't free response message
+}
+
+skynet.forward_type( forward_map ,function()
+	local clusterd = skynet.uniqueservice("clusterd")
+	local n = tonumber(address)
+	if n then
+		address = n
+	end
+	skynet.dispatch("system", function (session, source, msg, sz)
+		local m,s = skynet.rawcall(clusterd, "lua", skynet.pack("req", node, address, msg, sz))
+		skynet.ret(skynet.rawcall(clusterd, "lua", skynet.pack("req", node, address, msg, sz)))
+	end)
+end)

service/console.lua

@@ -7,10 +7,7 @@ local function console_main_loop()
 	while true do
 		local cmdline = socket.readline(stdin, "\n")
 		if cmdline ~= "" then
-			local handle = skynet.newservice(cmdline)
+			pcall(skynet.newservice,cmdline)
-			if handle == nil then
-				print("Launch error:",cmdline)
-			end
 		end
 	end
 	socket.unlock(stdin)

service/cslave.lua

@@ -1,9 +1,11 @@
 local skynet = require "skynet"
 local socket = require "socket"
+local table = table
 
 local slaves = {}
 local connect_queue = {}
 local globalname = {}
+local queryname = {}
 local harbor = {}
 local harbor_service
 local monitor = {}
@@ -28,7 +30,7 @@ local function monitor_clear(id)
 	if v then
 		monitor[id] = nil
 		for _, v in ipairs(v) do
-			skynet.redirect(v.address, 0, "response", v.session, "")
+			v(true)
 		end
 	end
 end
@@ -60,6 +62,17 @@ local function ready()
 	end
 end
 
+local function response_name(name)
+	local address = globalname[name]
+	if queryname[name] then
+		local tmp = queryname[name]
+		queryname[name] = nil
+		for _,resp in ipairs(tmp) do
+			resp(true, address)
+		end
+	end
+end
+
 local function monitor_master(master_fd)
 	while true do
 		local ok, t, id_name, address = pcall(read_package,master_fd)
@@ -72,6 +85,7 @@ local function monitor_master(master_fd)
 				end
 			elseif t == 'N' then
 				globalname[id_name] = address
+				response_name(id_name)
 				if connect_queue == nil then
 					skynet.redirect(harbor_service, address, "harbor", 0, "N " .. id_name)
 				end
@@ -79,6 +93,7 @@ local function monitor_master(master_fd)
 				local fd = slaves[id_name]
 				slaves[id_name] = false
 				if fd then
+					monitor_clear(id_name)
 					socket.close(fd)
 				end
 			end
@@ -129,14 +144,14 @@ local function monitor_harbor(master_fd)
 	return function(session, source, command)
 		local t = string.sub(command, 1, 1)
 		local arg = string.sub(command, 3)
-		if t == "Q" then
+		if t == 'Q' then
 			-- query name
 			if globalname[arg] then
 				skynet.redirect(harbor_service, globalname[arg], "harbor", 0, "N " .. arg)
 			else
 				socket.write(master_fd, pack_package("Q", arg))
 			end
-		elseif t == "D" then
+		elseif t == 'D' then
 			-- harbor down
 			local id = tonumber(arg)
 			if slaves[id] then
@@ -149,46 +164,66 @@ local function monitor_harbor(master_fd)
 	end
 end
 
-function harbor.REGISTER(_,_, fd, name, handle)
+function harbor.REGISTER(fd, name, handle)
 	assert(globalname[name] == nil)
 	globalname[name] = handle
+	response_name(name)
 	socket.write(fd, pack_package("R", name, handle))
 	skynet.redirect(harbor_service, handle, "harbor", 0, "N " .. name)
 end
 
-function harbor.LINK(session, source, fd, id)
+function harbor.LINK(fd, id)
 	if slaves[id] then
 		if monitor[id] == nil then
 			monitor[id] = {}
 		end
-		table.insert(monitor[id], { address = source, session = session })
+		table.insert(monitor[id], skynet.response())
 	else
 		skynet.ret()
 	end
 end
 
-function harbor.CONNECT(session, source, fd, id)
+function harbor.CONNECT(fd, id)
 	if not slaves[id] then
 		if monitor[id] == nil then
 			monitor[id] = {}
 		end
-		table.insert(monitor[id], { address = source, session = session })
+		table.insert(monitor[id], skynet.response())
 	else
 		skynet.ret()
 	end
 end
 
+function harbor.QUERYNAME(fd, name)
+	if name:byte() == 46 then	-- "." , local name
+		skynet.ret(skynet.pack(skynet.localname(name)))
+		return
+	end
+	local result = globalname[name]
+	if result then
+		skynet.ret(skynet.pack(result))
+		return
+	end
+	local queue = queryname[name]
+	if queue == nil then
+		queue = { skynet.response() }
+		queryname[name] = queue
+	else
+		table.insert(queue, skynet.response())
+	end
+end
+
 skynet.start(function()
 	local master_addr = skynet.getenv "master"
 	local harbor_id = tonumber(skynet.getenv "harbor")
 	local slave_address = assert(skynet.getenv "address")
 	local slave_fd = socket.listen(slave_address)
 	skynet.error("slave connect to master " .. tostring(master_addr))
-	local master_fd = socket.open(master_addr)
+	local master_fd = assert(socket.open(master_addr), "Can't connect to master")
 
-	skynet.dispatch("lua", function (session,source,command,...)
+	skynet.dispatch("lua", function (_,_,command,...)
 		local f = assert(harbor[command])
-		f(session, source, master_fd, ...)
+		f(master_fd, ...)
 	end)
 	skynet.dispatch("text", monitor_harbor(master_fd))
 

service/datacenterd.lua

@@ -33,7 +33,7 @@ local function update(db, key, value, ...)
 	end
 end
 
-local function wakeup(db, key1, key2, value, ...)
+local function wakeup(db, key1, key2, ...)
 	if key1 == nil then
 		return
 	end
@@ -43,19 +43,17 @@ local function wakeup(db, key1, key2, value, ...)
 	end
 	if q[mode] == "queue" then
 		db[key1] = nil
-		if value then
+		if key2 then
 			-- throw error because can't wake up a branch
-			for _,v in ipairs(q) do
+			for _,response in ipairs(q) do
-				local session = v[1]
+				response(false)
-				local source = v[2]
-				skynet.redirect(source, 0, "error", session, "")
 			end
 		else
 			return q
 		end
 	else
 		-- it's branch
-		return wakeup(q , key2, value, ...)
+		return wakeup(q , key2, ...)
 	end
 end
 
@@ -66,15 +64,13 @@ function command.UPDATE(...)
 	end
 	local q = wakeup(wait_queue, ...)
 	if q then
-		for _, v in ipairs(q) do
+		for _, response in ipairs(q) do
-			local session = v[1]
+			response(true,value)
-			local source = v[2]
-			skynet.redirect(source, 0, "response", session, skynet.pack(value))
 		end
 	end
 end
 
-local function waitfor(session, source, db, key1, key2, ...)
+local function waitfor(db, key1, key2, ...)
 	if key2 == nil then
 		-- push queue
 		local q = db[key1]
@@ -84,7 +80,7 @@ local function waitfor(session, source, db, key1, key2, ...)
 		else
 			assert(q[mode] == "queue")
 		end
-		table.insert(q, { session, source })
+		table.insert(q, skynet.response())
 	else
 		local q = db[key1]
 		if q == nil then
@@ -93,18 +89,18 @@ local function waitfor(session, source, db, key1, key2, ...)
 		else
 			assert(q[mode] == "branch")
 		end
-		return waitfor(session, source, q, key2, ...)
+		return waitfor(q, key2, ...)
 	end
 end
 
 skynet.start(function()
-	skynet.dispatch("lua", function (session, source, cmd, ...)
+	skynet.dispatch("lua", function (_, _, cmd, ...)
 		if cmd == "WAIT" then
 			local ret = command.QUERY(...)
 			if ret then
 				skynet.ret(skynet.pack(ret))
 			else
-				waitfor(session, source, wait_queue, ...)
+				waitfor(wait_queue, ...)
 			end
 		else
 			local f = assert(command[cmd])

service/debug_console.lua

@@ -36,6 +36,7 @@ local function dump_list(print, list)
 	for _,v in ipairs(index) do
 		dump_line(print, v, list[v])
 	end
+	print("OK")
 end
 
 local function split_cmdline(cmdline)
@@ -92,6 +93,9 @@ skynet.start(function()
 	socket.start(listen_socket , function(id, addr)
 		local function print(...)
 			local t = { ... }
+			for k,v in ipairs(t) do
+				t[k] = tostring(v)
+			end
 			socket.write(id, table.concat(t,"\t"))
 			socket.write(id, "\n")
 		end
@@ -106,6 +110,7 @@ function COMMAND.help()
 		list = "List all the service",
 		stat = "Dump all stats",
 		info = "Info address : get service infomation",
+		exit = "exit address : kill a lua service",
 		kill = "kill address : kill service",
 		mem = "mem : show memory status",
 		gc = "gc : force every lua service do garbage collect",
@@ -114,6 +119,7 @@ function COMMAND.help()
 		clearcache = "clear lua code cache",
 		service = "List unique service",
 		task = "task address : show service task detail",
+		inject = "inject address luascript.lua",
 	}
 end
 
@@ -122,8 +128,8 @@ function COMMAND.clearcache()
 end
 
 function COMMAND.start(...)
-	local addr = skynet.newservice(...)
+	local ok, addr = pcall(skynet.newservice, ...)
-	if addr then
+	if ok then
 		return { [skynet.address(addr)] = ... }
 	else
 		return "Failed"
@@ -131,8 +137,8 @@ function COMMAND.start(...)
 end
 
 function COMMAND.snax(...)
-	local s = snax.newservice(...)
+	local ok, s = pcall(snax.newservice, ...)
-	if s then
+	if ok then
 		local addr = s.handle
 		return { [skynet.address(addr)] = ... }
 	else
@@ -143,3 +149,35 @@ end
 function COMMAND.service()
 	return skynet.call("SERVICE", "lua", "LIST")
 end
+
+local function adjust_address(address)
+	if address:sub(1,1) ~= ":" then
+		address = bit32.replace( tonumber("0x" .. address), skynet.harbor(skynet.self()), 24, 8)
+	end
+	return address
+end
+
+function COMMAND.exit(address)
+	skynet.send(adjust_address(address), "debug", "EXIT")
+end
+
+function COMMAND.inject(address, filename)
+	address = adjust_address(address)
+	local f = io.open(filename, "rb")
+	if not f then
+		return "Can't open " .. filename
+	end
+	local source = f:read "*a"
+	f:close()
+	return skynet.call(address, "debug", "RUN", source, filename)
+end
+
+function COMMAND.task(address)
+	address = adjust_address(address)
+	return skynet.call(address,"debug","TASK")
+end
+
+function COMMAND.info(address)
+	address = adjust_address(address)
+	return skynet.call(address,"debug","INFO")
+end

service/gate.lua

@@ -31,7 +31,7 @@ end
 function handler.connect(fd, addr)
 	local c = {
 		fd = fd,
-		ip = msg,
+		ip = addr,
 	}
 	connection[fd] = c
 	skynet.send(watchdog, "lua", "socket", "open", fd, addr)

service/launcher.lua

@@ -28,27 +28,7 @@ function command.STAT()
 	return list
 end
 
-function command.INFO(_, _, handle)
+function command.KILL(_, handle)
-	handle = handle_to_address(handle)
-	if services[handle] == nil then
-		return
-	else
-		local result = skynet.call(handle,"debug","INFO")
-		return result
-	end
-end
-
-function command.TASK(_, _, handle)
-	handle = handle_to_address(handle)
-	if services[handle] == nil then
-		return
-	else
-		local result = skynet.call(handle,"debug","TASK")
-		return result
-	end
-end
-
-function command.KILL(_, _, handle)
 	handle = handle_to_address(handle)
 	skynet.kill(handle)
 	local ret = { [skynet.address(handle)] = tostring(services[handle]) }
@@ -72,18 +52,29 @@ function command.GC()
 	return command.MEM()
 end
 
-function command.REMOVE(_,_, handle)
+function command.REMOVE(_, handle)
 	services[handle] = nil
+	local response = instance[handle]
+	if response then
+		-- instance is dead
+		response(false)
+		instance[handle] = nil
+	end
+
 	-- don't return (skynet.ret) because the handle may exit
 	return NORET
 end
 
-function command.LAUNCH(address, session, service, ...)
+local function return_string(str)
+	return str
+end
+
+function command.LAUNCH(_, service, ...)
 	local param = table.concat({...}, " ")
 	local inst = skynet.launch(service, param)
 	if inst then
 		services[inst] = service .. " " .. param
-		instance[inst] = { session = session, address = address }
+		instance[inst] = skynet.response(return_string)
 	else
 		skynet.ret("")	-- launch failed
 	end
@@ -93,9 +84,9 @@ end
 function command.ERROR(address)
 	-- see serivce-src/service_lua.c
 	-- init failed
-	local reply = instance[address]
+	local response = instance[address]
-	if reply then
+	if response then
-		skynet.redirect(reply.address , 0, "response", reply.session, "")
+		response(false)
 		instance[address] = nil
 	end
 	services[address] = nil
@@ -104,9 +95,9 @@ end
 
 function command.LAUNCHOK(address)
 	-- init notice
-	local reply = instance[address]
+	local response = instance[address]
-	if reply then
+	if response then
-		skynet.redirect(reply.address , 0, "response", reply.session, skynet.address(address))
+		response(true, skynet.address(address))
 		instance[address] = nil
 	end
 
@@ -127,7 +118,7 @@ skynet.register_protocol {
 		else
 			-- launch request
 			local service, param = string.match(cmd,"([^ ]+) (.*)")
-			command.LAUNCH(address, session, service, param)
+			command.LAUNCH(_, service, param)
 		end
 	end,
 }
@@ -136,7 +127,7 @@ skynet.dispatch("lua", function(session, address, cmd , ...)
 	cmd = string.upper(cmd)
 	local f = command[cmd]
 	if f then
-		local ret = f(address, session, ...)
+		local ret = f(address, ...)
 		if ret ~= NORET then
 			skynet.ret(skynet.pack(ret))
 		end

service/multicastd.lua

@@ -1,5 +1,5 @@
 local skynet = require "skynet"
-local mc = require "multicast.c"
+local mc = require "multicast.core"
 local datacenter = require "datacenter"
 
 local harbor_id = skynet.harbor(skynet.self())
@@ -50,8 +50,10 @@ function command.DEL(source, c)
 	channel[c] = nil
 	channel_n[c] = nil
 	channel_remote[c] = nil
-	for node in pairs(remote) do
+	if remote then
-		skynet.send(node_address[node], "lua", "DELR", c)
+		for node in pairs(remote) do
+			skynet.send(node_address[node], "lua", "DELR", c)
+		end
 	end
 	return NORET
 end

service/service_mgr.lua

@@ -117,7 +117,6 @@ local function register_global()
 
 	function cmd.REPORT(m)
 		mgr[m] = true
-		skynet.watch(m)
 	end
 
 	local function add_list(all, m)

service/sharedatad.lua

@@ -0,0 +1,118 @@
+local skynet = require "skynet"
+local sharedata = require "sharedata.corelib"
+local table = table
+
+local NORET = {}
+local pool = {}
+local objmap = {}
+
+local function newobj(name, tbl)
+	assert(pool[name] == nil)
+	local cobj = sharedata.host.new(tbl)
+	sharedata.host.incref(cobj)
+	local v = { value = tbl , obj = cobj, watch = {} }
+	objmap[cobj] = v
+	pool[name] = v
+end
+
+local function collectobj()
+	while true do
+		skynet.sleep(600 * 100)	-- sleep 10 min
+		collectgarbage()
+		for obj, v in pairs(objmap) do
+			if v == true then
+				if sharedata.host.getref(obj) <= 0  then
+					objmap[obj] = nil
+					sharedata.host.delete(obj)
+				end
+			end
+		end
+	end
+end
+
+local CMD = {}
+
+function CMD.new(name, t)
+	local dt = type(t)
+	local value
+	if dt == "table" then
+		value = t
+	elseif dt == "string" then
+		value = {}
+		local f = load(t, "=" .. name, "t", env)
+		f()
+	elseif dt == "nil" then
+		value = {}
+	else
+		error ("Unknown data type " .. dt)
+	end
+	newobj(name, value)
+end
+
+function CMD.delete(name)
+	local v = assert(pool[name])
+	pool[name] = nil
+	assert(objmap[v.obj])
+	objmap[v.obj] = true
+	sharedata.host.decref(v.obj)
+	for _,response in ipairs(v.watch) do
+		response(true)
+	end
+end
+
+function CMD.query(name)
+	local v = assert(pool[name])
+	local obj = v.obj
+	sharedata.host.incref(obj)
+	return v.obj
+end
+
+function CMD.confirm(cobj)
+	if objmap[cobj] then
+		sharedata.host.decref(cobj)
+	end
+	return NORET
+end
+
+function CMD.update(name, t)
+	local v = pool[name]
+	local watch, oldcobj
+	if v then
+		watch = v.watch
+		oldcobj = v.obj
+		objmap[oldcobj] = true
+		sharedata.host.decref(oldcobj)
+		pool[name] = nil
+	end
+	CMD.new(name, t)
+	local newobj = pool[name].obj
+	if watch then
+		sharedata.host.markdirty(oldcobj)
+		for _,response in ipairs(watch) do
+			response(true, newobj)
+		end
+	end
+end
+
+function CMD.monitor(name, obj)
+	local v = assert(pool[name])
+	if obj ~= v.obj then
+		return v.obj
+	end
+
+	table.insert(v.watch, skynet.response())
+
+	return NORET
+end
+
+skynet.start(function()
+	skynet.fork(collectobj)
+	skynet.dispatch("lua", function (session, source ,cmd, ...)
+		local f = assert(CMD[cmd])
+		local r = f(...)
+		if r ~= NORET then
+			skynet.ret(skynet.pack(r))
+		end
+	end)
+end)
+

service/snaxd.lua

@@ -1,10 +1,17 @@
 local skynet = require "skynet"
-local c = require "skynet.c"
+local c = require "skynet.core"
 local snax_interface = require "snax.interface"
 local profile = require "profile"
 local snax = require "snax"
 
-local func = snax_interface(tostring(...), _ENV)
+local snax_name = tostring(...)
+local func, pattern = snax_interface(snax_name, _ENV)
+local snax_path = pattern:sub(1,pattern:find("?", 1, true)-1) .. snax_name ..  "/"
+package.path = snax_path .. "?.lua;" .. package.path
+
+SERVICE_NAME = snax_name
+SERVICE_PATH = snax_path
+
 local mode
 local thread_id
 local message_queue = {}

skynet-src/skynet.h

@@ -30,7 +30,7 @@ void skynet_error(struct skynet_context * context, const char *msg, ...);
 const char * skynet_command(struct skynet_context * context, const char * cmd , const char * parm);
 uint32_t skynet_queryname(struct skynet_context * context, const char * name);
 int skynet_send(struct skynet_context * context, uint32_t source, uint32_t destination , int type, int session, void * msg, size_t sz);
-int skynet_sendname(struct skynet_context * context, const char * destination , int type, int session, void * msg, size_t sz);
+int skynet_sendname(struct skynet_context * context, uint32_t source, const char * destination , int type, int session, void * msg, size_t sz);
 
 int skynet_isremote(struct skynet_context *, uint32_t handle, int * harbor);
 

skynet-src/skynet_handle.c

@@ -49,7 +49,6 @@ skynet_handle_register(struct skynet_context *ctx) {
 				rwlock_wunlock(&s->lock);
 
 				handle |= s->harbor;
-				skynet_context_init(ctx, handle);
 				return handle;
 			}
 		}
@@ -67,8 +66,9 @@ skynet_handle_register(struct skynet_context *ctx) {
 	}
 }
 
-void
+int
 skynet_handle_retire(uint32_t handle) {
+	int ret = 0;
 	struct handle_storage *s = H;
 
 	rwlock_wlock(&s->lock);
@@ -79,6 +79,7 @@ skynet_handle_retire(uint32_t handle) {
 	if (ctx != NULL && skynet_context_handle(ctx) == handle) {
 		skynet_context_release(ctx);
 		s->slot[hash] = NULL;
+		ret = 1;
 		int i;
 		int j=0, n=s->name_count;
 		for (i=0; i<n; ++i) {
@@ -94,6 +95,8 @@ skynet_handle_retire(uint32_t handle) {
 	}
 
 	rwlock_wunlock(&s->lock);
+
+	return ret;
 }
 
 void 
@@ -105,10 +108,14 @@ skynet_handle_retireall() {
 		for (i=0;i<s->slot_size;i++) {
 			rwlock_rlock(&s->lock);
 			struct skynet_context * ctx = s->slot[i];
+			uint32_t handle = 0;
+			if (ctx)
+				handle = skynet_context_handle(ctx);
 			rwlock_runlock(&s->lock);
-			if (ctx != NULL) {
+			if (handle != 0) {
-				++n;
+				if (skynet_handle_retire(handle)) {
-				skynet_handle_retire(skynet_context_handle(ctx));
+					++n;
+				}
 			}
 		}
 		if (n==0)

skynet-src/skynet_handle.h

@@ -8,7 +8,7 @@
 struct skynet_context;
 
 uint32_t skynet_handle_register(struct skynet_context *);
-void skynet_handle_retire(uint32_t handle);
+int skynet_handle_retire(uint32_t handle);
 struct skynet_context * skynet_handle_grab(uint32_t handle);
 void skynet_handle_retireall();
 

skynet-src/skynet_main.c

@@ -89,7 +89,6 @@ static const char * load_config = "\
 	local code = assert(f:read \'*a\')\
 	local function getenv(name) return assert(os.getenv(name), name) end\
 	code = string.gsub(code, \'%$([%w_%d]+)\', getenv)\
-	print(code)\
 	f:close()\
 	local result = {}\
 	assert(load(code,\'=(load)\',\'t\',result))()\
@@ -98,9 +97,13 @@ static const char * load_config = "\
 
 int
 main(int argc, char *argv[]) {
-	const char * config_file = "config";
+	const char * config_file = NULL ;
 	if (argc > 1) {
 		config_file = argv[1];
+	} else {
+		fprintf(stderr, "Need a config file. Please read skynet wiki : https://github.com/cloudwu/skynet/wiki/Config\n"
+			"usage: skynet configfilename\n");
+		return 1;
 	}
 	skynet_globalinit();
 	skynet_env_init();

skynet-src/skynet_server.c

@@ -132,6 +132,8 @@ skynet_context_new(const char * name, const char *param) {
 
 	ctx->init = false;
 	ctx->endless = false;
+	// Should set to 0 first to avoid skynet_handle_retireall get an uninitialized handle
+	ctx->handle = 0;	
 	ctx->handle = skynet_handle_register(ctx);
 	struct message_queue * queue = ctx->queue = skynet_mq_create(ctx->handle);
 	// init function maybe use ctx->handle, so it must init at last
@@ -230,7 +232,7 @@ _dispatch_message(struct skynet_context *ctx, struct skynet_message *msg) {
 	size_t sz = msg->sz & HANDLE_MASK;
 	if (!ctx->cb(ctx, ctx->cb_ud, type, msg->session, msg->source, msg->data, sz)) {
 		skynet_free(msg->data);
-	}
+	} 
 	CHECKCALLING_END(ctx)
 }
 
@@ -612,8 +614,10 @@ skynet_send(struct skynet_context * context, uint32_t source, uint32_t destinati
 }
 
 int
-skynet_sendname(struct skynet_context * context, const char * addr , int type, int session, void * data, size_t sz) {
+skynet_sendname(struct skynet_context * context, uint32_t source, const char * addr , int type, int session, void * data, size_t sz) {
-	uint32_t source = context->handle;
+	if (source == 0) {
+		source = context->handle;
+	}
 	uint32_t des = 0;
 	if (addr[0] == ':') {
 		des = strtoul(addr+1, NULL, 16);
@@ -646,11 +650,6 @@ skynet_context_handle(struct skynet_context *ctx) {
 	return ctx->handle;
 }
 
-void 
-skynet_context_init(struct skynet_context *ctx, uint32_t handle) {
-	ctx->handle = handle;
-}
-
 void 
 skynet_callback(struct skynet_context * context, void *ud, skynet_cb cb) {
 	context->cb = cb;

skynet-src/skynet_server.h

@@ -12,7 +12,6 @@ struct skynet_context * skynet_context_new(const char * name, const char * parm)
 void skynet_context_grab(struct skynet_context *);
 struct skynet_context * skynet_context_release(struct skynet_context *);
 uint32_t skynet_context_handle(struct skynet_context *);
-void skynet_context_init(struct skynet_context *, uint32_t handle);
 int skynet_context_push(uint32_t handle, struct skynet_message *message);
 void skynet_context_send(struct skynet_context * context, void * msg, size_t sz, uint32_t source, int type, int session);
 int skynet_context_newsession(struct skynet_context *);

skynet-src/skynet_timer.c

@@ -9,6 +9,7 @@
 #include <assert.h>
 #include <string.h>
 #include <stdlib.h>
+#include <stdint.h>
 
 #if defined(__APPLE__)
 #include <sys/time.h>
@@ -33,7 +34,7 @@ struct timer_event {
 
 struct timer_node {
 	struct timer_node *next;
-	int expire;
+	uint32_t expire;
 };
 
 struct link_list {
@@ -43,9 +44,9 @@ struct link_list {
 
 struct timer {
 	struct link_list near[TIME_NEAR];
-	struct link_list t[4][TIME_LEVEL-1];
+	struct link_list t[4][TIME_LEVEL];
 	int lock;
-	int time;
+	uint32_t time;
 	uint32_t current;
 	uint32_t starttime;
 	uint64_t current_point;
@@ -72,21 +73,22 @@ link(struct link_list *list,struct timer_node *node) {
 
 static void
 add_node(struct timer *T,struct timer_node *node) {
-	int time=node->expire;
+	uint32_t time=node->expire;
-	int current_time=T->time;
+	uint32_t current_time=T->time;
 	
 	if ((time|TIME_NEAR_MASK)==(current_time|TIME_NEAR_MASK)) {
 		link(&T->near[time&TIME_NEAR_MASK],node);
 	} else {
 		int i;
-		int mask=TIME_NEAR << TIME_LEVEL_SHIFT;
+		uint32_t mask=TIME_NEAR << TIME_LEVEL_SHIFT;
 		for (i=0;i<3;i++) {
 			if ((time|(mask-1))==(current_time|(mask-1))) {
 				break;
 			}
 			mask <<= TIME_LEVEL_SHIFT;
 		}
-		link(&T->t[i][((time>>(TIME_NEAR_SHIFT + i*TIME_LEVEL_SHIFT)) & TIME_LEVEL_MASK)-1],node);	
+
+		link(&T->t[i][((time>>(TIME_NEAR_SHIFT + i*TIME_LEVEL_SHIFT)) & TIME_LEVEL_MASK)],node);	
 	}
 }
 
@@ -103,29 +105,38 @@ timer_add(struct timer *T,void *arg,size_t sz,int time) {
 	UNLOCK(T);
 }
 
+static void
+move_list(struct timer *T, int level, int idx) {
+	struct timer_node *current = link_clear(&T->t[level][idx]);
+	while (current) {
+		struct timer_node *temp=current->next;
+		add_node(T,current);
+		current=temp;
+	}
+}
+
 static void
 timer_shift(struct timer *T) {
 	LOCK(T);
 	int mask = TIME_NEAR;
-	int time = (++T->time) >> TIME_NEAR_SHIFT;
+	uint32_t ct = ++T->time;
-	int i=0;
+	if (ct == 0) {
-	
+		move_list(T, 3, 0);
-	while ((T->time & (mask-1))==0) {
+	} else {
-		int idx=time & TIME_LEVEL_MASK;
+		uint32_t time = ct >> TIME_NEAR_SHIFT;
-		if (idx!=0) {
+		int i=0;
-			--idx;
+
-			struct timer_node *current = link_clear(&T->t[i][idx]);
+		while ((ct & (mask-1))==0) {
-			while (current) {
+			int idx=time & TIME_LEVEL_MASK;
-				struct timer_node *temp=current->next;
+			if (idx!=0) {
-				add_node(T,current);
+				move_list(T, i, idx);
-				current=temp;
+				break;				
 			}
-			break;				
+			mask <<= TIME_LEVEL_SHIFT;
+			time >>= TIME_LEVEL_SHIFT;
+			++i;
 		}
-		mask <<= TIME_LEVEL_SHIFT;
+	}
-		time >>= TIME_LEVEL_SHIFT;
-		++i;
-	}	
 	UNLOCK(T);
 }
 
@@ -187,7 +198,7 @@ timer_create_timer() {
 	}
 
 	for (i=0;i<4;i++) {
-		for (j=0;j<TIME_LEVEL-1;j++) {
+		for (j=0;j<TIME_LEVEL;j++) {
 			link_clear(&r->t[i][j]);
 		}
 	}
@@ -265,6 +276,7 @@ skynet_updatetime(void) {
 	uint64_t cp = gettime();
 	if(cp < TI->current_point) {
 		skynet_error(NULL, "time diff error: change from %lld to %lld", cp, TI->current_point);
+		TI->current_point = cp;
 	} else if (cp != TI->current_point) {
 		uint32_t diff = (uint32_t)(cp - TI->current_point);
 		TI->current_point = cp;

test/testblockcall.lua

@@ -1,10 +0,0 @@
-local skynet = require "skynet"
-
-skynet.start(function()
-	local ping = skynet.newservice("pingserver")
-	skynet.timeout(0,function()
-		print(skynet.call(ping,"lua","PING","ping"))
-	end)
-
-	print(skynet.blockcall(ping,"lua","HELLO"))
-end)

test/testdeadcall.lua

@@ -11,6 +11,15 @@ skynet.start(function()
 	end)
 end)
 
+elseif mode == "dead" then
+
+skynet.start(function()
+	skynet.dispatch("lua", function (...)
+		skynet.sleep(100)
+		print("return", skynet.ret "")
+	end)
+end)
+
 else
 
 	skynet.start(function()
@@ -20,6 +29,9 @@ else
 			skynet.call(test,"lua", "dead call")
 		end))
 
-		skynet.exit()
+		local dead = skynet.newservice(SERVICE_NAME, "dead")	-- launch self in dead mode
+
+		skynet.timeout(0, skynet.exit)	-- exit after a while, so the call never return
+		skynet.call(dead, "lua", "whould not return")
 	end)
-end
+end

test/testecho.lua

@@ -0,0 +1,44 @@
+local skynet = require "skynet"
+
+local mode = ...
+
+if mode == "slave" then
+
+skynet.start(function()
+	skynet.dispatch("lua", function(_,_, ...)
+		skynet.ret(skynet.pack(...))
+	end)
+end)
+
+else
+
+skynet.start(function()
+	local slave = skynet.newservice(SERVICE_NAME, "slave")
+	local n = 100000
+	local start = skynet.now()
+	print("call salve", n, "times in queue")
+	for i=1,n do
+		skynet.call(slave, "lua")
+	end
+	print("qps = ", n/ (skynet.now() - start) * 100)
+
+	start = skynet.now()
+
+	local worker = 10
+	local task = n/worker
+	print("call salve", n, "times in parallel, worker = ", worker)
+
+	for i=1,worker do
+		skynet.fork(function()
+			for i=1,task do
+				skynet.call(slave, "lua")
+			end
+			worker = worker -1
+			if worker == 0 then
+				print("qps = ", n/ (skynet.now() - start) * 100)
+			end
+		end)
+	end
+end)
+
+end

test/testharborlink.lua

@@ -6,6 +6,7 @@ skynet.start(function()
 	print("run skynet examples/config_log please")
 	harbor.connect(2)
 	print("harbor 2 connected")
+	print("LOG =", skynet.address(harbor.queryname "LOG"))
 	harbor.link(2)
 	print("disconnected")
 end)

test/testhttp.lua

@@ -0,0 +1,16 @@
+local skynet = require "skynet"
+local httpc = require "http.httpc"
+
+skynet.start(function()
+	print("GET baidu.com")
+	local header = {}
+	local status, body = httpc.get("baidu.com", "/", header)
+	print("[header] =====>")
+	for k,v in pairs(header) do
+		print(k,v)
+	end
+	print("[body] =====>", status)
+	print(body)
+
+	skynet.exit()
+end)

test/testmulticast2.lua

@@ -4,9 +4,13 @@ local mc = require "multicast"
 
 skynet.start(function()
 	print("remote start")
-	skynet.monitor("simplemonitor", true)
 	local console = skynet.newservice("console")
 	local channel = dc.get "MCCHANNEL"
+	if channel then
+		print("remote channel", channel)
+	else
+		print("create local channel")
+	end
 	for i=1,10 do
 		local sub = skynet.newservice("testmulticast", "sub")
 		skynet.call(sub, "lua", "init", channel)

test/testresponse.lua

@@ -0,0 +1,44 @@
+local skynet = require "skynet"
+
+local mode = ...
+
+if mode == "TICK" then
+-- this service whould response the request every 1s.
+
+local response_queue = {}
+
+local function response()
+	while true do
+		skynet.sleep(100)	-- sleep 1s
+		for k,v in ipairs(response_queue) do
+			v(true, skynet.now())		-- true means succ, false means error
+			response_queue[k] = nil
+		end
+	end
+end
+
+skynet.start(function()
+	skynet.fork(response)
+	skynet.dispatch("lua", function()
+		table.insert(response_queue, skynet.response())
+	end)
+end)
+
+else
+
+local function request(tick, i)
+	print(i, "call", skynet.now())
+	print(i, "response", skynet.call(tick, "lua"))
+	print(i, "end", skynet.now())
+end
+
+skynet.start(function()
+	local tick = skynet.newservice(SERVICE_NAME, "TICK")
+
+	for i=1,5 do
+		skynet.fork(request, tick, i)
+		skynet.sleep(10)
+	end
+end)
+
+end

test/teststm.lua

@@ -0,0 +1,36 @@
+local skynet = require "skynet"
+local stm = require "stm"
+
+local mode = ...
+
+if mode == "slave" then
+
+skynet.start(function()
+	skynet.dispatch("lua", function (_,_, obj)
+		local obj = stm.newcopy(obj)
+		print("read:", obj(skynet.unpack))
+		skynet.ret()
+		skynet.error("sleep and read")
+		for i=1,10 do
+			skynet.sleep(10)
+			print("read:", obj(skynet.unpack))
+		end
+		skynet.exit()
+	end)
+end)
+
+else
+
+skynet.start(function()
+	local slave = skynet.newservice(SERVICE_NAME, "slave")
+	local obj = stm.new(skynet.pack(1,2,3,4,5))
+	local copy = stm.copy(obj)
+	skynet.call(slave, "lua", copy)
+	for i=1,5 do
+		skynet.sleep(20)
+		print("write", i)
+		obj(skynet.pack("hello world", i))
+	end
+ 	skynet.exit()
+end)
+end