Release alpha 10

add pthread lock

3rd/lua/lfunc.c

@@ -135,6 +135,7 @@ static void freesharedproto (lua_State *L, SharedProto *f) {
   luaM_freearray(L, f->lineinfo, f->sizelineinfo);
   luaM_freearray(L, f->locvars, f->sizelocvars);
   luaM_freearray(L, f->upvalues, f->sizeupvalues);
+  luaM_free(L, f);
 }
 
 void luaF_freeproto (lua_State *L, Proto *f) {

HISTORY.md

@@ -1,10 +1,20 @@
+v1.0.0-alpha10 (2015-8-17)
+-----------
+* Remove the size limit of cluster RPC message.
+* Remove the size limit of local message.
+* Add cluster.query and clsuter.register.
+* Add an option of pthread mutex lock.
+* Add skynet.core.intcommand to optimize skynet.sleep etc.
+* Fix a memory leak bug in lua shared proto.
+* snax.msgserver use string instead of lightuserdata/size.
+* Remove some unused api in netpack.
+* Raise error when skynet.send to 0.
+
 v1.0.0-alpha9 (2015-8-10)
 -----------
 * Improve lua serialization , support pairs metamethod.
 * Bugfix : sproto (See commits log of sproto)
 * Add user log service support (In config)
-* Remove the size limit of cluster RPC message.
-* Remove the size limit of local message.
 * Other minor bugfix (See commits log)
 
 v1.0.0-alpha8 (2015-6-29)

Makefile

@@ -5,7 +5,8 @@ CSERVICE_PATH ?= cservice
 
 SKYNET_BUILD_PATH ?= .
 
-CFLAGS = -g -O2 -Wall -I$(LUA_INC) $(MYCFLAGS) 
+CFLAGS = -g -O2 -Wall -I$(LUA_INC) $(MYCFLAGS)
+# CFLAGS += -DUSE_PTHREAD_LOCK
 
 # lua
 
@@ -79,7 +80,7 @@ $(LUA_CLIB_PATH)/socketdriver.so : lualib-src/lua-socket.c | $(LUA_CLIB_PATH)
 	$(CC) $(CFLAGS) $(SHARED) $^ -o $@ -Iskynet-src -Iservice-src
 
 $(LUA_CLIB_PATH)/bson.so : lualib-src/lua-bson.c | $(LUA_CLIB_PATH)
-	$(CC) $(CFLAGS) $(SHARED) $^ -o $@ -Iskynet-src
+	$(CC) $(CFLAGS) $(SHARED) -Iskynet-src $^ -o $@ -Iskynet-src
 
 $(LUA_CLIB_PATH)/mongo.so : lualib-src/lua-mongo.c | $(LUA_CLIB_PATH)
 	$(CC) $(CFLAGS) $(SHARED) $^ -o $@ -Iskynet-src
@@ -109,7 +110,7 @@ $(LUA_CLIB_PATH)/crypt.so : lualib-src/lua-crypt.c lualib-src/lsha1.c | $(LUA_CL
 	$(CC) $(CFLAGS) $(SHARED) $^ -o $@ 
 
 $(LUA_CLIB_PATH)/sharedata.so : lualib-src/lua-sharedata.c | $(LUA_CLIB_PATH)
-	$(CC) $(CFLAGS) $(SHARED) $^ -o $@ 
+	$(CC) $(CFLAGS) $(SHARED) -Iskynet-src $^ -o $@ 
 
 $(LUA_CLIB_PATH)/stm.so : lualib-src/lua-stm.c | $(LUA_CLIB_PATH)
 	$(CC) $(CFLAGS) $(SHARED) -Iskynet-src $^ -o $@ 
@@ -124,7 +125,7 @@ $(LUA_CLIB_PATH)/mysqlaux.so : lualib-src/lua-mysqlaux.c | $(LUA_CLIB_PATH)
 	$(CC) $(CFLAGS) $(SHARED) $^ -o $@	
 
 $(LUA_CLIB_PATH)/debugchannel.so : lualib-src/lua-debugchannel.c | $(LUA_CLIB_PATH)
-	$(CC) $(CFLAGS) $(SHARED) $^ -o $@	
+	$(CC) $(CFLAGS) $(SHARED) -Iskynet-src $^ -o $@	
 
 clean :
 	rm -f $(SKYNET_BUILD_PATH)/skynet $(CSERVICE_PATH)/*.so $(LUA_CLIB_PATH)/*.so

examples/cluster1.lua

@@ -1,15 +1,17 @@
 local skynet = require "skynet"
 local cluster = require "cluster"
-require "skynet.manager"	-- import skynet.name
 local snax = require "snax"
 
 skynet.start(function()
 	local sdb = skynet.newservice("simpledb")
-	skynet.name(".simpledb", sdb)
-	print(skynet.call(".simpledb", "lua", "SET", "a", "foobar"))
-	print(skynet.call(".simpledb", "lua", "SET", "b", "foobar2"))
-	print(skynet.call(".simpledb", "lua", "GET", "a"))
-	print(skynet.call(".simpledb", "lua", "GET", "b"))
+	-- register name "sdb" for simpledb, you can use cluster.query() later.
+	-- See cluster2.lua
+	cluster.register("sdb", sdb)
+
+	print(skynet.call(sdb, "lua", "SET", "a", "foobar"))
+	print(skynet.call(sdb, "lua", "SET", "b", "foobar2"))
+	print(skynet.call(sdb, "lua", "GET", "a"))
+	print(skynet.call(sdb, "lua", "GET", "b"))
 	cluster.open "db"
 	cluster.open "db2"
 	-- unique snax service

examples/cluster2.lua

@@ -2,10 +2,18 @@ local skynet = require "skynet"
 local cluster = require "cluster"
 
 skynet.start(function()
-	local proxy = cluster.proxy("db", ".simpledb")
-	print(skynet.call(proxy, "lua", "GET", "a"))
-	print(cluster.call("db", ".simpledb", "GET", "a"))
-	print(cluster.call("db2", ".simpledb", "GET", "b"))
+	-- query name "sdb" of cluster db.
+	local sdb = cluster.query("db", "sdb")
+	print("db.sbd=",sdb)
+	local proxy = cluster.proxy("db", sdb)
+	local largekey = string.rep("X", 128*1024)
+	local largevalue = string.rep("R", 100 * 1024)
+	print(skynet.call(proxy, "lua", "SET", largekey, largevalue))
+	local v = skynet.call(proxy, "lua", "GET", largekey)
+	assert(largevalue == v)
+
+	print(cluster.call("db", sdb, "GET", "a"))
+	print(cluster.call("db2", sdb, "GET", "b"))
 
 	-- test snax service
 	local pingserver = cluster.snax("db", "pingserver")

examples/config.userlog

@@ -3,12 +3,9 @@ thread = 8
 logger = "userlog"
 logservice = "snlua"
 logpath = "."
-harbor = 1
-address = "127.0.0.1:2526"
-master = "127.0.0.1:2013"
+harbor = 0
 start = "main"	-- main script
 bootstrap = "snlua bootstrap"	-- The service for bootstrap
-standalone = "0.0.0.0:2013"
 luaservice = root.."service/?.lua;"..root.."test/?.lua;"..root.."examples/?.lua"
 lualoader = "lualib/loader.lua"
 -- preload = "./examples/preload.lua"	-- run preload.lua before every lua service run

examples/login/gated.lua

@@ -74,9 +74,9 @@ function server.disconnect_handler(username)
 end
 
 -- call by self (when recv a request from client)
-function server.request_handler(username, msg, sz)
+function server.request_handler(username, msg)
 	local u = username_map[username]
-	return skynet.tostring(skynet.rawcall(u.agent, "client", msg, sz))
+	return skynet.tostring(skynet.rawcall(u.agent, "client", msg))
 end
 
 -- call by self (when gate open)

examples/watchdog.lua

@@ -32,6 +32,11 @@ function SOCKET.error(fd, msg)
 	close_agent(fd)
 end
 
+function SOCKET.warning(fd, size)
+	-- size K bytes havn't send out in fd
+	print("socket warning", fd, size)
+end
+
 function SOCKET.data(fd, msg)
 end
 

lualib-src/lua-bson.c

@@ -8,6 +8,7 @@
 #include <stdlib.h>
 #include <string.h>
 #include <stdbool.h>
+#include "atomic.h"
 
 #define DEFAULT_CAP 64
 #define MAX_NUMBER 1024
@@ -1140,7 +1141,7 @@ lobjectid(lua_State *L) {
 	} else {
 		time_t ti = time(NULL);
 		// old_counter is a static var, use atom inc.
-		uint32_t id = __sync_fetch_and_add(&oid_counter,1);
+		uint32_t id = ATOM_FINC(&oid_counter);
 
 		oid[2] = (ti>>24) & 0xff;
 		oid[3] = (ti>>16) & 0xff;

lualib-src/lua-clientsocket.c

@@ -128,11 +128,8 @@ lusleep(lua_State *L) {
 
 #define QUEUE_SIZE 1024
 
-#define LOCK(q) while (__sync_lock_test_and_set(&(q)->lock,1)) {}
-#define UNLOCK(q) __sync_lock_release(&(q)->lock);
-
 struct queue {
-	int lock;
+	pthread_mutex_t lock;
 	int head;
 	int tail;
 	char * queue[QUEUE_SIZE];
@@ -153,7 +150,7 @@ readline_stdin(void * arg) {
 		memcpy(str, tmp, n);
 		str[n] = 0;
 
-		LOCK(q);
+		pthread_mutex_lock(&q->lock);
 		q->queue[q->tail] = str;
 
 		if (++q->tail >= QUEUE_SIZE) {
@@ -163,7 +160,7 @@ readline_stdin(void * arg) {
 			// queue overflow
 			exit(1);
 		}
-		UNLOCK(q);
+		pthread_mutex_unlock(&q->lock);
 	}
 	return NULL;
 }
@@ -171,16 +168,16 @@ readline_stdin(void * arg) {
 static int
 lreadstdin(lua_State *L) {
 	struct queue *q = lua_touserdata(L, lua_upvalueindex(1));
-	LOCK(q);
+	pthread_mutex_lock(&q->lock);
 	if (q->head == q->tail) {
-		UNLOCK(q);
+		pthread_mutex_unlock(&q->lock);
 		return 0;
 	}
 	char * str = q->queue[q->head];
 	if (++q->head >= QUEUE_SIZE) {
 		q->head = 0;
 	}
-	UNLOCK(q);
+	pthread_mutex_unlock(&q->lock);
 	lua_pushstring(L, str);
 	free(str);
 	return 1;
@@ -201,6 +198,7 @@ luaopen_clientsocket(lua_State *L) {
 
 	struct queue * q = lua_newuserdata(L, sizeof(*q));
 	memset(q, 0, sizeof(*q));
+	pthread_mutex_init(&q->lock, NULL);
 	lua_pushcclosure(L, lreadstdin, 1);
 	lua_setfield(L, -2, "readstdin");
 

lualib-src/lua-cluster.c

@@ -1,6 +1,7 @@
 #include <lua.h>
 #include <lauxlib.h>
 #include <string.h>
+#include <assert.h>
 
 #include "skynet.h"
 
@@ -15,7 +16,8 @@
 		uint32_t next_session
  */
 
-#define TEMP_LENGTH 0x10007
+#define TEMP_LENGTH 0x8200
+#define MULTI_PART 0x8000
 
 static void
 fill_uint32(uint8_t * buf, uint32_t n) {
@@ -26,30 +28,75 @@ fill_uint32(uint8_t * buf, uint32_t n) {
 }
 
 static void
-fill_header(lua_State *L, uint8_t *buf, int sz, void *msg) {
-	if (sz >= 0x10000) {
-		skynet_free(msg);
-		luaL_error(L, "request message is too long %d", sz);
-	}
+fill_header(lua_State *L, uint8_t *buf, int sz) {
+	assert(sz < 0x10000);
 	buf[0] = (sz >> 8) & 0xff;
 	buf[1] = sz & 0xff;
 }
 
-static void
-packreq_number(lua_State *L, int session, void * msg, size_t sz) {
+/*
+	The request package :
+	size <= 0x8000 (32K) and address is id
+		WORD sz+9
+		BYTE 0
+		DWORD addr
+		DWORD session
+		PADDING msg(sz)
+	size > 0x8000 and address is id
+		DWORD 13
+		BYTE 1	; multireq	
+		DWORD addr
+		DWORD session
+		DWORD sz
+
+	size <= 0x8000 (32K) and address is string
+		WORD sz+6+namelen
+		BYTE 0x80
+		BYTE namelen
+		STRING name
+		DWORD session
+		PADDING msg(sz)
+	size > 0x8000 and address is string
+		DWORD 10 + namelen
+		BYTE 0x81
+		BYTE namelen
+		STRING name
+		DWORD session
+		DWORD sz
+
+	multi req
+		WORD sz + 5
+		BYTE 2/3 ; 2:multipart, 3:multipart end
+		DWORD SESSION
+		PADDING msgpart(sz)
+ */
+static int
+packreq_number(lua_State *L, int session, void * msg, uint32_t sz) {
 	uint32_t addr = (uint32_t)lua_tointeger(L,1);
 	uint8_t buf[TEMP_LENGTH];
-	fill_header(L, buf, sz+9, msg);
-	buf[2] = 0;
-	fill_uint32(buf+3, addr);
-	fill_uint32(buf+7, (uint32_t)session);
-	memcpy(buf+11,msg,sz);
+	if (sz < MULTI_PART) {
+		fill_header(L, buf, sz+9);
+		buf[2] = 0;
+		fill_uint32(buf+3, addr);
+		fill_uint32(buf+7, (uint32_t)session);
+		memcpy(buf+11,msg,sz);
 
-	lua_pushlstring(L, (const char *)buf, sz+11);
+		lua_pushlstring(L, (const char *)buf, sz+11);
+		return 0;
+	} else {
+		int part = (sz - 1) / MULTI_PART + 1;
+		fill_header(L, buf, 13);
+		buf[2] = 1;
+		fill_uint32(buf+3, addr);
+		fill_uint32(buf+7, (uint32_t)session);
+		fill_uint32(buf+11, sz);
+		lua_pushlstring(L, (const char *)buf, 15);
+		return part;
+	}
 }
 
-static void
-packreq_string(lua_State *L, int session, void * msg, size_t sz) {
+static int
+packreq_string(lua_State *L, int session, void * msg, uint32_t sz) {
 	size_t namelen = 0;
 	const char *name = lua_tolstring(L, 1, &namelen);
 	if (name == NULL || namelen < 1 || namelen > 255) {
@@ -58,13 +105,53 @@ packreq_string(lua_State *L, int session, void * msg, size_t sz) {
 	}
 
 	uint8_t buf[TEMP_LENGTH];
-	fill_header(L, buf, sz+5+namelen, msg);
-	buf[2] = (uint8_t)namelen;
-	memcpy(buf+3, name, namelen);
-	fill_uint32(buf+3+namelen, (uint32_t)session);
-	memcpy(buf+7+namelen,msg,sz);
+	if (sz < MULTI_PART) {
+		fill_header(L, buf, sz+6+namelen);
+		buf[2] = 0x80;
+		buf[3] = (uint8_t)namelen;
+		memcpy(buf+4, name, namelen);
+		fill_uint32(buf+4+namelen, (uint32_t)session);
+		memcpy(buf+8+namelen,msg,sz);
 
-	lua_pushlstring(L, (const char *)buf, sz+7+namelen);
+		lua_pushlstring(L, (const char *)buf, sz+8+namelen);
+		return 0;
+	} else {
+		int part = (sz - 1) / MULTI_PART + 1;
+		fill_header(L, buf, 10+namelen);
+		buf[2] = 0x81;
+		buf[3] = (uint8_t)namelen;
+		memcpy(buf+4, name, namelen);
+		fill_uint32(buf+4+namelen, (uint32_t)session);
+		fill_uint32(buf+8+namelen, sz);
+
+		lua_pushlstring(L, (const char *)buf, 12+namelen);
+		return part;
+	}
+}
+
+static void
+packreq_multi(lua_State *L, int session, void * msg, uint32_t sz) {
+	uint8_t buf[TEMP_LENGTH];
+	int part = (sz - 1) / MULTI_PART + 1;
+	int i;
+	char *ptr = msg;
+	for (i=0;i<part;i++) {
+		uint32_t s;
+		if (sz > MULTI_PART) {
+			s = MULTI_PART;
+			buf[2] = 2;
+		} else {
+			s = sz;
+			buf[2] = 3;	// the last multi part
+		}
+		fill_header(L, buf, s+5);
+		fill_uint32(buf+3, (uint32_t)session);
+		memcpy(buf+7, ptr, s);
+		lua_pushlstring(L, (const char *)buf, s+7);
+		lua_rawseti(L, -2, i+1);
+		sz -= s;
+		ptr += s;
+	}
 }
 
 static int
@@ -73,24 +160,33 @@ lpackrequest(lua_State *L) {
 	if (msg == NULL) {
 		return luaL_error(L, "Invalid request message");
 	}
-	size_t sz = (size_t)luaL_checkinteger(L,4);
+	uint32_t sz = (uint32_t)luaL_checkinteger(L,4);
 	int session = luaL_checkinteger(L,2);
 	if (session <= 0) {
 		skynet_free(msg);
 		return luaL_error(L, "Invalid request session %d", session);
 	}
 	int addr_type = lua_type(L,1);
+	int multipak;
 	if (addr_type == LUA_TNUMBER) {
-		packreq_number(L, session, msg, sz);
+		multipak = packreq_number(L, session, msg, sz);
 	} else {
-		packreq_string(L, session, msg, sz);
+		multipak = packreq_string(L, session, msg, sz);
 	}
+	int current_session = session;
 	if (++session < 0) {
 		session = 1;
 	}
-	skynet_free(msg);
 	lua_pushinteger(L, session);
-	return 2;
+	if (multipak) {
+		lua_createtable(L, multipak, 0);
+		packreq_multi(L, current_session, msg, sz);
+		skynet_free(msg);
+		return 3;
+	} else {
+		skynet_free(msg);
+		return 2;
+	}
 }
 
 /*
@@ -99,6 +195,7 @@ lpackrequest(lua_State *L) {
 		uint32_t or string addr
 		int session
 		string msg
+		boolean padding
  */
 
 static inline uint32_t
@@ -107,44 +204,122 @@ unpack_uint32(const uint8_t * buf) {
 }
 
 static int
-unpackreq_number(lua_State *L, const uint8_t * buf, size_t sz) {
+unpackreq_number(lua_State *L, const uint8_t * buf, int sz) {
 	if (sz < 9) {
-		return luaL_error(L, "Invalid cluster message");
+		return luaL_error(L, "Invalid cluster message (size=%d)", sz);
 	}
 	uint32_t address = unpack_uint32(buf+1);
 	uint32_t session = unpack_uint32(buf+5);
-	lua_pushinteger(L, (uint32_t)address);
-	lua_pushinteger(L, (uint32_t)session);
+	lua_pushinteger(L, address);
+	lua_pushinteger(L, session);
 	lua_pushlstring(L, (const char *)buf+9, sz-9);
 
 	return 3;
 }
 
 static int
-unpackreq_string(lua_State *L, const uint8_t * buf, size_t sz) {
-	size_t namesz = buf[0];
-	if (sz < namesz + 5) {
-		return luaL_error(L, "Invalid cluster message");
+unpackmreq_number(lua_State *L, const uint8_t * buf, int sz) {
+	if (sz != 13) {
+		return luaL_error(L, "Invalid cluster message size %d (multi req must be 13)", sz);
 	}
-	lua_pushlstring(L, (const char *)buf+1, namesz);
-	uint32_t session = unpack_uint32(buf + namesz + 1);
+	uint32_t address = unpack_uint32(buf+1);
+	uint32_t session = unpack_uint32(buf+5);
+	uint32_t size = unpack_uint32(buf+9);
+	lua_pushinteger(L, address);
+	lua_pushinteger(L, session);
+	lua_pushinteger(L, size);
+	lua_pushboolean(L, 1);	// padding multi part
+
+	return 4;
+}
+
+static int
+unpackmreq_part(lua_State *L, const uint8_t * buf, int sz) {
+	if (sz < 5) {
+		return luaL_error(L, "Invalid cluster multi part message");
+	}
+	int padding = (buf[0] == 2);
+	uint32_t session = unpack_uint32(buf+1);
+	lua_pushboolean(L, 0);	// no address
+	lua_pushinteger(L, session);
+	lua_pushlstring(L, (const char *)buf+5, sz-5);
+	lua_pushboolean(L, padding);
+
+	return 4;
+}
+
+static int
+unpackreq_string(lua_State *L, const uint8_t * buf, int sz) {
+	if (sz < 2) {
+		return luaL_error(L, "Invalid cluster message (size=%d)", sz);
+	}
+	size_t namesz = buf[1];
+	if (sz < namesz + 6) {
+		return luaL_error(L, "Invalid cluster message (size=%d)", sz);
+	}
+	lua_pushlstring(L, (const char *)buf+2, namesz);
+	uint32_t session = unpack_uint32(buf + namesz + 2);
 	lua_pushinteger(L, (uint32_t)session);
-	lua_pushlstring(L, (const char *)buf+1+namesz+4, sz - namesz - 5);
+	lua_pushlstring(L, (const char *)buf+2+namesz+4, sz - namesz - 6);
 
 	return 3;
 }
 
+static int
+unpackmreq_string(lua_State *L, const uint8_t * buf, int sz) {
+	if (sz < 2) {
+		return luaL_error(L, "Invalid cluster message (size=%d)", sz);
+	}
+	size_t namesz = buf[1];
+	if (sz < namesz + 10) {
+		return luaL_error(L, "Invalid cluster message (size=%d)", sz);
+	}
+	lua_pushlstring(L, (const char *)buf+2, namesz);
+	uint32_t session = unpack_uint32(buf + namesz + 2);
+	uint32_t size = unpack_uint32(buf + namesz + 6);
+	lua_pushinteger(L, session);
+	lua_pushinteger(L, size);
+	lua_pushboolean(L, 1);	// padding multipart
+
+	return 4;
+}
+
 static int
 lunpackrequest(lua_State *L) {
-	size_t sz;
-	const char *msg = luaL_checklstring(L,1,&sz);
-	if (msg[0] == 0) {
+	size_t ssz;
+	const char *msg = luaL_checklstring(L,1,&ssz);
+	int sz = (int)ssz;
+	switch (msg[0]) {
+	case 0:
 		return unpackreq_number(L, (const uint8_t *)msg, sz);
-	} else {
+	case 1:
+		return unpackmreq_number(L, (const uint8_t *)msg, sz);
+	case 2:
+	case 3:
+		return unpackmreq_part(L, (const uint8_t *)msg, sz);
+	case '\x80':
 		return unpackreq_string(L, (const uint8_t *)msg, sz);
+	case '\x81':
+		return unpackmreq_string(L, (const uint8_t *)msg, sz);
+	default:
+		return luaL_error(L, "Invalid req package type %d", msg[0]);
 	}
 }
 
+/*
+	DWORD session
+	BYTE type
+		0: error
+		1: ok
+		2: multi begin
+		3: multi part
+		4: multi end
+	PADDING msg
+		type = 0, error msg
+		type = 1, msg
+		type = 2, DWORD size
+		type = 3/4, msg
+ */
 /*
 	int session
 	boolean ok
@@ -163,16 +338,56 @@ lpackresponse(lua_State *L) {
 	
 	if (lua_type(L,3) == LUA_TSTRING) {
 		msg = (void *)lua_tolstring(L, 3, &sz);
-		if (sz > 0x1000) {
-			sz = 0x1000;
-		}
 	} else {
 		msg = lua_touserdata(L,3);
 		sz = (size_t)luaL_checkinteger(L, 4);
 	}
 
+	if (!ok) {
+		if (sz > MULTI_PART) {
+			// truncate the error msg if too long
+			sz = MULTI_PART;
+		}
+	} else {
+		if (sz > MULTI_PART) {
+			// return 
+			int part = (sz - 1) / MULTI_PART + 1;
+			lua_createtable(L, part+1, 0);
+			uint8_t buf[TEMP_LENGTH];
+
+			// multi part begin
+			fill_header(L, buf, 9);
+			fill_uint32(buf+2, session);
+			buf[6] = 2;
+			fill_uint32(buf+7, (uint32_t)sz);
+			lua_pushlstring(L, (const char *)buf, 11);
+			lua_rawseti(L, -2, 1);
+
+			char * ptr = msg;
+			int i;
+			for (i=0;i<part;i++) {
+				int s;
+				if (sz > MULTI_PART) {
+					s = MULTI_PART;
+					buf[6] = 3;
+				} else {
+					s = sz;
+					buf[6] = 4;
+				}
+				fill_header(L, buf, s+5);
+				fill_uint32(buf+2, session);
+				memcpy(buf+7,ptr,s);
+				lua_pushlstring(L, (const char *)buf, s+7);
+				lua_rawseti(L, -2, i+2);
+				sz -= s;
+				ptr += s;
+			}
+			return 1;
+		}
+	}
+
 	uint8_t buf[TEMP_LENGTH];
-	fill_header(L, buf, sz+5, msg);
+	fill_header(L, buf, sz+5);
 	fill_uint32(buf+2, session);
 	buf[6] = ok;
 	memcpy(buf+7,msg,sz);
@@ -187,6 +402,7 @@ lpackresponse(lua_State *L) {
 	return integer session
 		boolean ok
 		string msg
+		boolean padding
  */
 static int
 lunpackresponse(lua_State *L) {
@@ -197,10 +413,66 @@ lunpackresponse(lua_State *L) {
 	}
 	uint32_t session = unpack_uint32((const uint8_t *)buf);
 	lua_pushinteger(L, (lua_Integer)session);
-	lua_pushboolean(L, buf[4]);
-	lua_pushlstring(L, buf+5, sz-5);
+	switch(buf[4]) {
+	case 0:	// error
+		lua_pushboolean(L, 0);
+		lua_pushlstring(L, buf+5, sz-5);
+		return 3;
+	case 1:	// ok
+	case 4:	// multi end
+		lua_pushboolean(L, 1);
+		lua_pushlstring(L, buf+5, sz-5);
+		return 3;
+	case 2:	// multi begin
+		if (sz != 9) {
+			return 0;
+		}
+		sz = unpack_uint32((const uint8_t *)buf+5);
+		lua_pushboolean(L, 1);
+		lua_pushinteger(L, sz);
+		lua_pushboolean(L, 1);
+		return 4;
+	case 3:	// multi part
+		lua_pushboolean(L, 1);
+		lua_pushlstring(L, buf+5, sz-5);
+		lua_pushboolean(L, 1);
+		return 4;
+	default:
+		return 0;
+	}
+}
 
-	return 3;
+static int
+lconcat(lua_State *L) {
+	if (!lua_istable(L,1))
+		return 0;
+	if (lua_geti(L,1,1) != LUA_TNUMBER)
+		return 0;
+	int sz = lua_tointeger(L,-1);
+	lua_pop(L,1);
+	char * buff = skynet_malloc(sz);
+	int idx = 2;
+	int offset = 0;
+	while(lua_geti(L,1,idx) == LUA_TSTRING) {
+		size_t s;
+		const char * str = lua_tolstring(L, -1, &s);
+		if (s+offset > sz) {
+			skynet_free(buff);
+			return 0;
+		}
+		memcpy(buff+offset, str, s);
+		lua_pop(L,1);
+		offset += s;
+		++idx;
+	}
+	if (offset != sz) {
+		skynet_free(buff);
+		return 0;
+	}
+	// buff/sz will send to other service, See clusterd.lua
+	lua_pushlightuserdata(L, buff);
+	lua_pushinteger(L, sz);
+	return 2;
 }
 
 int
@@ -210,6 +482,7 @@ luaopen_cluster_core(lua_State *L) {
 		{ "unpackrequest", lunpackrequest },
 		{ "packresponse", lpackresponse },
 		{ "unpackresponse", lunpackresponse },
+		{ "concat", lconcat },
 		{ NULL, NULL },
 	};
 	luaL_checkversion(L);

lualib-src/lua-debugchannel.c

@@ -4,10 +4,9 @@
 #include <unistd.h>
 #include <stdlib.h>
 #include <string.h>
+#include "spinlock.h"
 
 #define METANAME "debugchannel"
-#define LOCK(q) while (__sync_lock_test_and_set(&(q)->lock,1)) {}
-#define UNLOCK(q) __sync_lock_release(&(q)->lock);
 
 struct command {
 	struct command * next;
@@ -15,7 +14,7 @@ struct command {
 };
 
 struct channel {
-	int lock;
+	struct spinlock lock;
 	int ref;
 	struct command * head;
 	struct command * tail;
@@ -26,6 +25,7 @@ channel_new() {
 	struct channel * c = malloc(sizeof(*c));
 	memset(c, 0 , sizeof(*c));
 	c->ref = 1;
+	SPIN_INIT(c)
 
 	return c;
 }
@@ -33,21 +33,21 @@ channel_new() {
 static struct channel *
 channel_connect(struct channel *c) {
 	struct channel * ret = NULL;
-	LOCK(c)
+	SPIN_LOCK(c)
 	if (c->ref == 1) {
 		++c->ref;
 		ret = c;
 	}
-	UNLOCK(c)
+	SPIN_UNLOCK(c)
 	return ret;
 }
 
 static struct channel *
 channel_release(struct channel *c) {
-	LOCK(c)
+	SPIN_LOCK(c)
 	--c->ref;
 	if (c->ref > 0) {
-		UNLOCK(c)
+		SPIN_UNLOCK(c)
 		return c;
 	}
 	// never unlock while reference is 0
@@ -59,6 +59,8 @@ channel_release(struct channel *c) {
 		free(p);
 		p = next;
 	}
+	SPIN_UNLOCK(c)
+	SPIN_DESTROY(c)
 	free(c);
 	return NULL;
 }
@@ -67,9 +69,9 @@ channel_release(struct channel *c) {
 static struct command *
 channel_read(struct channel *c, double timeout) {
 	struct command * ret = NULL;
-	LOCK(c)
+	SPIN_LOCK(c)
 	if (c->head == NULL) {
-		UNLOCK(c)
+		SPIN_UNLOCK(c)
 		int ti = (int)(timeout * 100000);
 		usleep(ti);
 		return NULL;
@@ -79,7 +81,7 @@ channel_read(struct channel *c, double timeout) {
 	if (c->head == NULL) {
 		c->tail = NULL;
 	}
-	UNLOCK(c)
+	SPIN_UNLOCK(c)
 	
 	return ret;
 }
@@ -90,14 +92,14 @@ channel_write(struct channel *c, const char * s, size_t sz) {
 	cmd->sz = sz;
 	cmd->next = NULL;
 	memcpy(cmd+1, s, sz);
-	LOCK(c)
+	SPIN_LOCK(c)
 	if (c->tail == NULL) {
 		c->head = c->tail = cmd;
 	} else {
 		c->tail->next = cmd;
 		c->tail = cmd;
 	}
-	UNLOCK(c)
+	SPIN_UNLOCK(c)
 }
 
 struct channel_box {

lualib-src/lua-multicast.c

@@ -5,6 +5,8 @@
 #include <stdint.h>
 #include <string.h>
 
+#include "atomic.h"
+
 struct mc_package {
 	int reference;
 	uint32_t size;
@@ -116,7 +118,7 @@ static int
 mc_closelocal(lua_State *L) {
 	struct mc_package *pack = lua_touserdata(L,1);
 
-	int ref = __sync_sub_and_fetch(&pack->reference, 1);
+	int ref = ATOM_DEC(&pack->reference);
 	if (ref <= 0) {
 		skynet_free(pack->data);
 		skynet_free(pack);

lualib-src/lua-netpack.c

@@ -19,6 +19,7 @@
 #define TYPE_ERROR 3
 #define TYPE_OPEN 4
 #define TYPE_CLOSE 5
+#define TYPE_WARNING 6
 
 /*
 	Each package is uint16 + data , uint16 (serialized in big-endian) is the number of bytes comprising the data .
@@ -371,6 +372,11 @@ lfilter(lua_State *L) {
 		lua_pushinteger(L, message->id);
 		pushstring(L, buffer, size);
 		return 4;
+	case SKYNET_SOCKET_TYPE_WARNING:
+		lua_pushvalue(L, lua_upvalueindex(TYPE_WARNING));
+		lua_pushinteger(L, message->id);
+		lua_pushinteger(L, message->ud);
+		return 4;
 	default:
 		// never get here
 		return 1;
@@ -442,57 +448,6 @@ lpack(lua_State *L) {
 	return 2;
 }
 
-static int
-lpack_string(lua_State *L) {
-	uint8_t tmp[SMALLSTRING+2];
-	size_t len;
-	uint8_t *buffer;
-	const char * ptr = tolstring(L, &len, 1);
-	if (len > 0x10000) {
-		return luaL_error(L, "Invalid size (too long) of data : %d", (int)len);
-	}
-
-	if (len <= SMALLSTRING) {
-		buffer = tmp;
-	} else {
-		buffer = lua_newuserdata(L, len + 2);
-	}
-
-	write_size(buffer, len);
-	memcpy(buffer+2, ptr, len);
-	lua_pushlstring(L, (const char *)buffer, len+2);
-
-	return 1;
-}
-
-static int
-lpack_padding(lua_State *L) {
-	uint8_t tmp[SMALLSTRING+2];
-	size_t content_sz;
-	uint8_t *buffer;
-	const char * ptr = tolstring(L, &content_sz, 2);
-	size_t cookie_sz = 0;
-	const char * cookie = luaL_checklstring(L,1,&cookie_sz);
-	size_t len = cookie_sz + content_sz;
-
-	if (len > 0x10000) {
-		return luaL_error(L, "Invalid size (too long) of data : %d", (int)len);
-	}
-
-	if (len <= SMALLSTRING) {
-		buffer = tmp;
-	} else {
-		buffer = lua_newuserdata(L, len + 2);
-	}
-
-	write_size(buffer, len);
-	memcpy(buffer+2, ptr, content_sz);
-	memcpy(buffer+2+content_sz, cookie, cookie_sz);
-	lua_pushlstring(L, (const char *)buffer, len+2);
-
-	return 1;
-}
-
 static int
 ltostring(lua_State *L) {
 	void * ptr = lua_touserdata(L, 1);
@@ -500,19 +455,8 @@ ltostring(lua_State *L) {
 	if (ptr == NULL) {
 		lua_pushliteral(L, "");
 	} else {
-		if (lua_isnumber(L, 3)) {
-			int offset = lua_tointeger(L, 3);
-			if (offset < 0) {
-				return luaL_error(L, "Invalid offset %d", offset);
-			}
-			if (offset > size) {
-				offset = size;
-			}
-			lua_pushlstring(L, (const char *)ptr + offset, size-offset);
-		} else {
-			lua_pushlstring(L, (const char *)ptr, size);
-			skynet_free(ptr);
-		}
+		lua_pushlstring(L, (const char *)ptr, size);
+		skynet_free(ptr);
 	}
 	return 1;
 }
@@ -523,8 +467,6 @@ luaopen_netpack(lua_State *L) {
 	luaL_Reg l[] = {
 		{ "pop", lpop },
 		{ "pack", lpack },
-		{ "pack_string", lpack_string },
-		{ "pack_padding", lpack_padding },
 		{ "clear", lclear },
 		{ "tostring", ltostring },
 		{ NULL, NULL },
@@ -537,8 +479,9 @@ luaopen_netpack(lua_State *L) {
 	lua_pushliteral(L, "error");
 	lua_pushliteral(L, "open");
 	lua_pushliteral(L, "close");
+	lua_pushliteral(L, "warning");
 
-	lua_pushcclosure(L, lfilter, 5);
+	lua_pushcclosure(L, lfilter, 6);
 	lua_setfield(L, -2, "filter");
 
 	return 1;

lualib-src/lua-sharedata.c

@@ -4,6 +4,7 @@
 #include <stdlib.h>
 #include <string.h>
 #include <assert.h>
+#include "atomic.h"
 
 #define KEYTYPE_INTEGER 0
 #define KEYTYPE_STRING 1
@@ -663,7 +664,7 @@ 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);
+	ATOM_DEC(&s->ref);
 	c->root = NULL;
 	c->update = NULL;
 
@@ -674,7 +675,7 @@ 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);
+	ATOM_INC(&s->ref);
 
 	struct ctrl * c = lua_newuserdata(L, sizeof(*c));
 	c->root = tbl;
@@ -719,7 +720,7 @@ 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);
+	int ref = ATOM_INC(&s->ref);
 	lua_pushinteger(L , ref);
 
 	return 1;
@@ -729,7 +730,7 @@ 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);
+	int ref = ATOM_DEC(&s->ref);
 	lua_pushinteger(L , ref);
 
 	return 1;

lualib-src/lua-skynet.c

@@ -118,6 +118,28 @@ _command(lua_State *L) {
 	return 0;
 }
 
+static int
+_intcommand(lua_State *L) {
+	struct skynet_context * context = lua_touserdata(L, lua_upvalueindex(1));
+	const char * cmd = luaL_checkstring(L,1);
+	const char * result;
+	const char * parm = NULL;
+	char tmp[64];	// for integer parm
+	if (lua_gettop(L) == 2) {
+		int32_t n = (int32_t)luaL_checkinteger(L,2);
+		sprintf(tmp, "%d", n);
+		parm = tmp;
+	}
+
+	result = skynet_command(context, cmd, parm);
+	if (result) {
+		lua_Integer r = strtoll(result, NULL, 0);
+		lua_pushinteger(L, r);
+		return 1;
+	}
+	return 0;
+}
+
 static int
 _genid(lua_State *L) {
 	struct skynet_context * context = lua_touserdata(L, lua_upvalueindex(1));
@@ -150,6 +172,9 @@ _send(lua_State *L) {
 	uint32_t dest = (uint32_t)lua_tointeger(L, 1);
 	const char * dest_string = NULL;
 	if (dest == 0) {
+		if (lua_type(L,1) == LUA_TNUMBER) {
+			return luaL_error(L, "Invalid service address 0");
+		}
 		dest_string = get_dest_string(L, 1);
 	}
 
@@ -310,6 +335,7 @@ luaopen_skynet_core(lua_State *L) {
 		{ "genid", _genid },
 		{ "redirect", _redirect },
 		{ "command" , _command },
+		{ "intcommand", _intcommand },
 		{ "error", _error },
 		{ "tostring", _tostring },
 		{ "harbor", _harbor },

lualib-src/lua-stm.c

@@ -7,6 +7,7 @@
 
 #include "rwlock.h"
 #include "skynet_malloc.h"
+#include "atomic.h"
 
 struct stm_object {
 	struct rwlock lock;
@@ -45,7 +46,7 @@ static void
 stm_releasecopy(struct stm_copy *copy) {
 	if (copy == NULL)
 		return;
-	if (__sync_sub_and_fetch(&copy->reference, 1) == 0) {
+	if (ATOM_DEC(&copy->reference) == 0) {
 		skynet_free(copy->msg);
 		skynet_free(copy);
 	}
@@ -70,7 +71,7 @@ stm_release(struct stm_object *obj) {
 static void
 stm_releasereader(struct stm_object *obj) {
 	rwlock_rlock(&obj->lock);
-	if (__sync_sub_and_fetch(&obj->reference,1) == 0) {
+	if (ATOM_DEC(&obj->reference) == 0) {
 		// last reader, no writer. so no need to unlock
 		assert(obj->copy == NULL);
 		skynet_free(obj);
@@ -82,7 +83,7 @@ stm_releasereader(struct stm_object *obj) {
 static void
 stm_grab(struct stm_object *obj) {
 	rwlock_rlock(&obj->lock);
-	int ref = __sync_fetch_and_add(&obj->reference,1);
+	int ref = ATOM_FINC(&obj->reference);
 	rwlock_runlock(&obj->lock);
 	assert(ref > 0);
 }
@@ -92,7 +93,7 @@ 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);
+		int ref = ATOM_FINC(&ret->reference);
 		assert(ref > 0);
 	}
 	rwlock_runlock(&obj->lock);

lualib/cluster.lua

@@ -33,6 +33,16 @@ function cluster.snax(node, name, address)
 	return snax.bind(handle, name)
 end
 
+function cluster.register(name, addr)
+	assert(type(name) == "string")
+	assert(addr == nil or type(addr) == "number")
+	return skynet.call(clusterd, "lua", "register", name, addr)
+end
+
+function cluster.query(node, name)
+	return skynet.call(clusterd, "lua", "req", node, 0, skynet.pack(name))
+end
+
 skynet.init(function()
 	clusterd = skynet.uniqueservice("clusterd")
 end)

lualib/mysql.lua

@@ -386,7 +386,8 @@ end
 
 local function _recv_decode_packet_resp(self)
      return function(sock)
-        return true, _recv_packet(self,sock)
+		-- don't return more than 2 results
+        return true, (_recv_packet(self,sock))
     end
 end
 

lualib/skynet.lua

@@ -248,18 +248,16 @@ function suspend(co, result, command, param, size)
 end
 
 function skynet.timeout(ti, func)
-	local session = c.command("TIMEOUT",tostring(ti))
+	local session = c.intcommand("TIMEOUT",ti)
 	assert(session)
-	session = tonumber(session)
 	local co = co_create(func)
 	assert(session_id_coroutine[session] == nil)
 	session_id_coroutine[session] = co
 end
 
 function skynet.sleep(ti)
-	local session = c.command("TIMEOUT",tostring(ti))
+	local session = c.intcommand("TIMEOUT",ti)
 	assert(session)
-	session = tonumber(session)
 	local succ, ret = coroutine_yield("SLEEP", session)
 	sleep_session[coroutine.running()] = nil
 	if succ then
@@ -273,7 +271,7 @@ function skynet.sleep(ti)
 end
 
 function skynet.yield()
-	return skynet.sleep("0")
+	return skynet.sleep(0)
 end
 
 function skynet.wait()
@@ -301,11 +299,11 @@ function skynet.localname(name)
 end
 
 function skynet.now()
-	return tonumber(c.command("NOW"))
+	return c.intcommand("NOW")
 end
 
 function skynet.starttime()
-	return tonumber(c.command("STARTTIME"))
+	return c.intcommand("STARTTIME")
 end
 
 function skynet.time()
@@ -643,7 +641,7 @@ function skynet.endless()
 end
 
 function skynet.mqlen()
-	return tonumber(c.command "MQLEN")
+	return c.intcommand "MQLEN"
 end
 
 function skynet.task(ret)

lualib/snax/gateserver.lua

@@ -114,6 +114,12 @@ function gateserver.start(handler)
 		close_fd(fd)
 	end
 
+	function MSG.warning(fd, size)
+		if handler.warning then
+			handler.warning(fd, size)
+		end
+	end
+
 	skynet.register_protocol {
 		name = "socket",
 		id = skynet.PTYPE_SOCKET,	-- PTYPE_SOCKET = 6

lualib/snax/msgserver.lua

@@ -70,7 +70,7 @@ Config for server.start:
 	conf.login_handler(uid, secret) -> subid : the function when a new user login, alloc a subid for it. (may call by login server)
 	conf.logout_handler(uid, subid) : the functon when a user logout. (may call by agent)
 	conf.kick_handler(uid, subid) : the functon when a user logout. (may call by login server)
-	conf.request_handler(username, session, msg, sz) : the function when recv a new request.
+	conf.request_handler(username, session, msg) : the function when recv a new request.
 	conf.register_handler(servername) : call when gate open
 	conf.disconnect_handler(username) : call when a connection disconnect (afk)
 ]]
@@ -234,10 +234,10 @@ function server.start(conf)
 		end
 	end
 
-	local function do_request(fd, msg, sz)
+	local function do_request(fd, message)
 		local u = assert(connection[fd], "invalid fd")
-		local msg_sz = sz - 4
-		local session = netpack.tostring(msg, sz, msg_sz)
+		local session = string.unpack(">I4", message, -4)
+		message = message:sub(1,-5)
 		local p = u.response[session]
 		if p then
 			-- session can be reuse in the same connection
@@ -256,21 +256,20 @@ function server.start(conf)
 		if p == nil then
 			p = { fd }
 			u.response[session] = p
-			local ok, result = pcall(conf.request_handler, u.username, msg, msg_sz)
-			result = result or ""
+			local ok, result = pcall(conf.request_handler, u.username, message)
 			-- NOTICE: YIELD here, socket may close.
+			result = result or ""
 			if not ok then
 				skynet.error(result)
-				result = "\0" .. session
+				result = string.pack(">BI4", 0, session)
 			else
-				result = result .. '\1' .. session
+				result = result .. string.pack(">BI4", 1, session)
 			end
 
-			p[2] = netpack.pack_string(result)
+			p[2] = string.pack(">s2",result)
 			p[3] = u.version
 			p[4] = u.index
 		else
-			netpack.tostring(msg, sz) -- request before, so free msg
 			-- update version/index, change return fd.
 			-- resend response.
 			p[1] = fd
@@ -292,10 +291,11 @@ function server.start(conf)
 	end
 
 	local function request(fd, msg, sz)
-		local ok, err = pcall(do_request, fd, msg, sz)
+		local message = netpack.tostring(msg, sz)
+		local ok, err = pcall(do_request, fd, message)
 		-- not atomic, may yield
 		if not ok then
-			skynet.error(string.format("Invalid package %s : %s", err, netpack.tostring(msg, sz)))
+			skynet.error(string.format("Invalid package %s : %s", err, message))
 			if connection[fd] then
 				gateserver.closeclient(fd)
 			end

lualib/socket.lua

@@ -133,6 +133,25 @@ socket_message[6] = function(id, size, data, address)
 	s.callback(str, address)
 end
 
+local function default_warning(id, size)
+	local s = socket_pool[id]
+		local last = s.warningsize or 0
+		if last + 64 < size then	-- if size increase 64K
+			s.warningsize = size
+			skynet.error(string.format("WARNING: %d K bytes need to send out (fd = %d)", size, id))
+		end
+		s.warningsize = size
+end
+
+-- SKYNET_SOCKET_TYPE_WARNING
+socket_message[7] = function(id, size)
+	local s = socket_pool[id]
+	if s then
+		local warning = s.warning or default_warning
+		warning(id, size)
+	end
+end
+
 skynet.register_protocol {
 	name = "socket",
 	id = skynet.PTYPE_SOCKET,	-- PTYPE_SOCKET = 6
@@ -404,4 +423,10 @@ end
 socket.sendto = assert(driver.udp_send)
 socket.udp_address = assert(driver.udp_address)
 
+function socket.warning(id, callback)
+	local obj = socket_pool[id]
+	assert(obj)
+	obj.warning = callback
+end
+
 return socket

lualib/socketchannel.lua

@@ -81,15 +81,27 @@ local function dispatch_by_session(self)
 	local response = self.__response
 	-- response() return session
 	while self.__sock do
-		local ok , session, result_ok, result_data = pcall(response, self.__sock)
+		local ok , session, result_ok, result_data, padding = pcall(response, self.__sock)
 		if ok and session then
 			local co = self.__thread[session]
-			self.__thread[session] = nil
 			if co then
-				self.__result[co] = result_ok
-				self.__result_data[co] = result_data
-				skynet.wakeup(co)
+				if padding and result_ok then
+					-- If padding is true, append result_data to a table (self.__result_data[co])
+					local result = self.__result_data[co] or {}
+					self.__result_data[co] = result
+					table.insert(result, result_data)
+				else
+					self.__thread[session] = nil
+					self.__result[co] = result_ok
+					if result_ok and self.__result_data[co] then
+						table.insert(self.__result_data[co], result_data)
+					else
+						self.__result_data[co] = result_data
+					end
+					skynet.wakeup(co)
+				end
 			else
+				self.__thread[session] = nil
 				skynet.error("socket: unknown session :", session)
 			end
 		else
@@ -127,11 +139,23 @@ local function dispatch_by_order(self)
 				wakeup_all(self)
 			end
 		else
-			local ok, result_ok, result_data = pcall(func, self.__sock)
+			local ok, result_ok, result_data, padding = pcall(func, self.__sock)
 			if ok then
-				self.__result[co] = result_ok
-				self.__result_data[co] = result_data
-				skynet.wakeup(co)
+				if padding and result_ok then
+					-- if padding is true, wait for next result_data
+					-- self.__result_data[co] is a table
+					local result = self.__result_data[co] or {}
+					self.__result_data[co] = result
+					table.insert(result, result_data)
+				else
+					self.__result[co] = result_ok
+					if result_ok and self.__result_data[co] then
+						table.insert(self.__result_data[co], result_data)
+					else
+						self.__result_data[co] = result_data
+					end
+					skynet.wakeup(co)
+				end
 			else
 				close_channel_socket(self)
 				local errmsg
@@ -314,13 +338,27 @@ local function wait_for_response(self, response)
 	end
 end
 
-function channel:request(request, response)
-	assert(block_connect(self, true))	-- connect once
+local socket_write = socket.write
+local socket_lwrite = socket.lwrite
 
-	if not socket.write(self.__sock[1], request) then
-		close_channel_socket(self)
-		wakeup_all(self)
-		error(socket_error)
+function channel:request(request, response, padding)
+	assert(block_connect(self, true))	-- connect once
+	local fd = self.__sock[1]
+
+	if padding then
+		-- padding may be a table, to support multi part request
+		-- multi part request use low priority socket write
+		-- socket_lwrite returns nothing
+		socket_lwrite(fd , request)
+		for _,v in ipairs(padding) do
+			socket_lwrite(fd, v)
+		end
+	else
+		if not socket_write(fd , request) then
+			close_channel_socket(self)
+			wakeup_all(self)
+			error(socket_error)
+		end
 	end
 
 	if response == nil then

service-src/service_gate.c

@@ -262,6 +262,9 @@ dispatch_socket_message(struct gate *g, const struct skynet_socket_message * mes
 			skynet_socket_start(ctx, message->ud);
 		}
 		break;
+	case SKYNET_SOCKET_TYPE_WARNING:
+		skynet_error(ctx, "fd (%d) send buffer (%d)K", message->id, message->ud);
+		break;
 	}
 }
 

service-src/service_harbor.c

@@ -1,6 +1,7 @@
 #include "skynet.h"
 #include "skynet_harbor.h"
 #include "skynet_socket.h"
+#include "skynet_handle.h"
 
 /*
 	harbor listen the PTYPE_HARBOR (in text)
@@ -323,9 +324,13 @@ forward_local_messsage(struct harbor *h, void *msg, int sz) {
 
 static void
 send_remote(struct skynet_context * ctx, int fd, const char * buffer, size_t sz, struct remote_message_header * cookie) {
-	uint32_t sz_header = sz+sizeof(*cookie);
+	size_t sz_header = sz+sizeof(*cookie);
+	if (sz_header > UINT32_MAX) {
+		skynet_error(ctx, "remote message from :%08x to :%08x is too large.", cookie->source, cookie->destination);
+		return;
+	}
 	uint8_t * sendbuf = skynet_malloc(sz_header+4);
-	to_bigendian(sendbuf, sz_header);
+	to_bigendian(sendbuf, (uint32_t)sz_header);
 	memcpy(sendbuf+4, buffer, sz);
 	header_to_message(cookie, sendbuf+4+sz);
 
@@ -657,6 +662,13 @@ mainloop(struct skynet_context * context, void * ud, int type, int session, uint
 		case SKYNET_SOCKET_TYPE_CONNECT:
 			// fd forward to this service
 			break;
+		case SKYNET_SOCKET_TYPE_WARNING: {
+			int id = harbor_id(h, message->id);
+			if (id) {
+				skynet_error(context, "message havn't send to Harbor (%d) reach %d K", id, message->ud);
+			}
+			break;
+		}
 		default:
 			skynet_error(context, "recv invalid socket message type %d", type);
 			break;

service/clusterd.lua

@@ -11,7 +11,7 @@ local command = {}
 local function read_response(sock)
 	local sz = socket.header(sock:read(2))
 	local msg = sock:read(sz)
-	return cluster.unpackresponse(msg)	-- session, ok, data
+	return cluster.unpackresponse(msg)	-- session, ok, data, padding
 end
 
 local function open_channel(t, key)
@@ -64,19 +64,23 @@ end
 local function send_request(source, node, addr, msg, sz)
 	local session = node_session[node] or 1
 	-- msg is a local pointer, cluster.packrequest will free it
-	local request, new_session = cluster.packrequest(addr, session, msg, sz)
+	local request, new_session, padding = cluster.packrequest(addr, session, msg, sz)
 	node_session[node] = new_session
 
 	-- node_channel[node] may yield or throw error
 	local c = node_channel[node]
 
-	return c:request(request, session)
+	return c:request(request, session, padding)
 end
 
 function command.req(...)
 	local ok, msg, sz = pcall(send_request, ...)
 	if ok then
-		skynet.ret(msg, sz)
+		if type(msg) == "table" then
+			skynet.ret(cluster.concat(msg))
+		else
+			skynet.ret(msg)
+		end
 	else
 		skynet.error(msg)
 		skynet.response()(false)
@@ -93,23 +97,78 @@ function command.proxy(source, node, name)
 	skynet.ret(skynet.pack(proxy[fullname]))
 end
 
-local request_fd = {}
+local register_name = {}
+
+function command.register(source, name, addr)
+	assert(register_name[name] == nil)
+	addr = addr or source
+	local old_name = register_name[addr]
+	if old_name then
+		register_name[old_name] = nil
+	end
+	register_name[addr] = name
+	register_name[name] = addr
+	skynet.ret(nil)
+	skynet.error(string.format("Register [%s] :%08x", name, addr))
+end
+
+local large_request = {}
 
 function command.socket(source, subcmd, fd, msg)
 	if subcmd == "data" then
-		local addr, session, msg = cluster.unpackrequest(msg)
-		local ok , msg, sz = pcall(skynet.rawcall, addr, "lua", msg)
-		local response
+		local sz
+		local addr, session, msg, padding = cluster.unpackrequest(msg)
+		if padding then
+			local req = large_request[session] or { addr = addr }
+			large_request[session] = req
+			table.insert(req, msg)
+			return
+		else
+			local req = large_request[session]
+			if req then
+				large_request[session] = nil
+				table.insert(req, msg)
+				msg,sz = cluster.concat(req)
+				addr = req.addr
+			end
+			if not msg then
+				local response = cluster.packresponse(session, false, "Invalid large req")
+				socket.write(fd, response)
+				return
+			end
+		end
+		local ok, response
+		if addr == 0 then
+			local name = skynet.unpack(msg, sz)
+			local addr = register_name[name]
+			if addr then
+				ok = true
+				msg, sz = skynet.pack(addr)
+			else
+				ok = false
+				msg = "name not found"
+			end
+		else
+			ok , msg, sz = pcall(skynet.rawcall, addr, "lua", msg, sz)
+		end
 		if ok then
 			response = cluster.packresponse(session, true, msg, sz)
+			if type(response) == "table" then
+				for _, v in ipairs(response) do
+					socket.lwrite(fd, v)
+				end
+			else
+				socket.write(fd, response)
+			end
 		else
 			response = cluster.packresponse(session, false, msg)
+			socket.write(fd, response)
 		end
-		socket.write(fd, response)
 	elseif subcmd == "open" then
 		skynet.error(string.format("socket accept from %s", msg))
 		skynet.call(source, "lua", "accept", fd)
 	else
+		large_request = {}
 		skynet.error(string.format("socket %s %d : %s", subcmd, fd, msg))
 	end
 end

service/gate.lua

@@ -63,6 +63,10 @@ function handler.error(fd, msg)
 	skynet.send(watchdog, "lua", "socket", "error", fd, msg)
 end
 
+function handler.warning(fd, size)
+	skynet.send(watchdog, "lua", "socket", "warning", fd, size)
+end
+
 local CMD = {}
 
 function CMD.forward(source, fd, client, address)

skynet-src/atomic.h

@@ -0,0 +1,14 @@
+#ifndef SKYNET_ATOMIC_H
+#define SKYNET_ATOMIC_H
+
+#define ATOM_CAS(ptr, oval, nval) __sync_bool_compare_and_swap(ptr, oval, nval)
+#define ATOM_CAS_POINTER(ptr, oval, nval) __sync_bool_compare_and_swap(ptr, oval, nval)
+#define ATOM_INC(ptr) __sync_add_and_fetch(ptr, 1)
+#define ATOM_FINC(ptr) __sync_fetch_and_add(ptr, 1)
+#define ATOM_DEC(ptr) __sync_sub_and_fetch(ptr, 1)
+#define ATOM_FDEC(ptr) __sync_fetch_and_sub(ptr, 1)
+#define ATOM_ADD(ptr,n) __sync_add_and_fetch(ptr, n)
+#define ATOM_SUB(ptr,n) __sync_sub_and_fetch(ptr, n)
+#define ATOM_AND(ptr,n) __sync_and_and_fetch(ptr, n)
+
+#endif

skynet-src/malloc_hook.c

@@ -6,6 +6,7 @@
 
 #include "malloc_hook.h"
 #include "skynet.h"
+#include "atomic.h"
 
 static size_t _used_memory = 0;
 static size_t _memory_block = 0;
@@ -32,11 +33,11 @@ get_allocated_field(uint32_t handle) {
 	ssize_t old_alloc = data->allocated;
 	if(old_handle == 0 || old_alloc <= 0) {
 		// data->allocated may less than zero, because it may not count at start.
-		if(!__sync_bool_compare_and_swap(&data->handle, old_handle, handle)) {
+		if(!ATOM_CAS(&data->handle, old_handle, handle)) {
 			return 0;
 		}
 		if (old_alloc < 0) {
-			__sync_bool_compare_and_swap(&data->allocated, old_alloc, 0);
+			ATOM_CAS(&data->allocated, old_alloc, 0);
 		}
 	}
 	if(data->handle != handle) {
@@ -47,21 +48,21 @@ get_allocated_field(uint32_t handle) {
 
 inline static void 
 update_xmalloc_stat_alloc(uint32_t handle, size_t __n) {
-	__sync_add_and_fetch(&_used_memory, __n);
-	__sync_add_and_fetch(&_memory_block, 1); 
+	ATOM_ADD(&_used_memory, __n);
+	ATOM_INC(&_memory_block); 
 	ssize_t* allocated = get_allocated_field(handle);
 	if(allocated) {
-		__sync_add_and_fetch(allocated, __n);
+		ATOM_ADD(allocated, __n);
 	}
 }
 
 inline static void
 update_xmalloc_stat_free(uint32_t handle, size_t __n) {
-	__sync_sub_and_fetch(&_used_memory, __n);
-	__sync_sub_and_fetch(&_memory_block, 1);
+	ATOM_SUB(&_used_memory, __n);
+	ATOM_DEC(&_memory_block);
 	ssize_t* allocated = get_allocated_field(handle);
 	if(allocated) {
-		__sync_sub_and_fetch(allocated, __n);
+		ATOM_SUB(allocated, __n);
 	}
 }
 

skynet-src/rwlock.h

@@ -1,6 +1,8 @@
 #ifndef SKYNET_RWLOCK_H
 #define SKYNET_RWLOCK_H
 
+#ifndef USE_PTHREAD_LOCK
+
 struct rwlock {
 	int write;
 	int read;
@@ -45,4 +47,42 @@ rwlock_runlock(struct rwlock *lock) {
 	__sync_sub_and_fetch(&lock->read,1);
 }
 
+#else
+
+#include <pthread.h>
+
+// only for some platform doesn't have __sync_*
+// todo: check the result of pthread api
+
+struct rwlock {
+	pthread_rwlock_t lock;
+};
+
+static inline void
+rwlock_init(struct rwlock *lock) {
+	pthread_rwlock_init(&lock->lock, NULL);
+}
+
+static inline void
+rwlock_rlock(struct rwlock *lock) {
+	 pthread_rwlock_rdlock(&lock->lock);
+}
+
+static inline void
+rwlock_wlock(struct rwlock *lock) {
+	 pthread_rwlock_wrlock(&lock->lock);
+}
+
+static inline void
+rwlock_wunlock(struct rwlock *lock) {
+	pthread_rwlock_unlock(&lock->lock);
+}
+
+static inline void
+rwlock_runlock(struct rwlock *lock) {
+	pthread_rwlock_unlock(&lock->lock);
+}
+
+#endif
+
 #endif

skynet-src/skynet_env.c

@@ -1,5 +1,6 @@
 #include "skynet.h"
 #include "skynet_env.h"
+#include "spinlock.h"
 
 #include <lua.h>
 #include <lauxlib.h>
@@ -8,18 +9,15 @@
 #include <assert.h>
 
 struct skynet_env {
-	int lock;
+	struct spinlock lock;
 	lua_State *L;
 };
 
 static struct skynet_env *E = NULL;
 
-#define LOCK(q) while (__sync_lock_test_and_set(&(q)->lock,1)) {}
-#define UNLOCK(q) __sync_lock_release(&(q)->lock);
-
 const char * 
 skynet_getenv(const char *key) {
-	LOCK(E)
+	SPIN_LOCK(E)
 
 	lua_State *L = E->L;
 	
@@ -27,14 +25,14 @@ skynet_getenv(const char *key) {
 	const char * result = lua_tostring(L, -1);
 	lua_pop(L, 1);
 
-	UNLOCK(E)
+	SPIN_UNLOCK(E)
 
 	return result;
 }
 
 void 
 skynet_setenv(const char *key, const char *value) {
-	LOCK(E)
+	SPIN_LOCK(E)
 	
 	lua_State *L = E->L;
 	lua_getglobal(L, key);
@@ -43,12 +41,12 @@ skynet_setenv(const char *key, const char *value) {
 	lua_pushstring(L,value);
 	lua_setglobal(L,key);
 
-	UNLOCK(E)
+	SPIN_UNLOCK(E)
 }
 
 void
 skynet_env_init() {
 	E = skynet_malloc(sizeof(*E));
-	E->lock = 0;
+	SPIN_INIT(E)
 	E->L = luaL_newstate();
 }

skynet-src/skynet_error.c

@@ -54,7 +54,7 @@ skynet_error(struct skynet_context * context, const char *msg, ...) {
 	}
 	smsg.session = 0;
 	smsg.data = data;
-	smsg.sz = len | (PTYPE_TEXT << HANDLE_REMOTE_SHIFT);
+	smsg.sz = len | ((size_t)PTYPE_TEXT << MESSAGE_TYPE_SHIFT);
 	skynet_context_push(logger, &smsg);
 }
 

skynet-src/skynet_handle.h

@@ -3,7 +3,9 @@
 
 #include <stdint.h>
 
-#include "skynet_harbor.h"
+// reserve high 8 bits for remote id
+#define HANDLE_MASK 0xffffff
+#define HANDLE_REMOTE_SHIFT 24
 
 struct skynet_context;
 

skynet-src/skynet_harbor.c

@@ -1,6 +1,8 @@
 #include "skynet.h"
 #include "skynet_harbor.h"
 #include "skynet_server.h"
+#include "skynet_mq.h"
+#include "skynet_handle.h"
 
 #include <string.h>
 #include <stdio.h>
@@ -11,8 +13,8 @@ static unsigned int HARBOR = ~0;
 
 void 
 skynet_harbor_send(struct remote_message *rmsg, uint32_t source, int session) {
-	int type = rmsg->sz >> HANDLE_REMOTE_SHIFT;
-	rmsg->sz &= HANDLE_MASK;
+	int type = rmsg->sz >> MESSAGE_TYPE_SHIFT;
+	rmsg->sz &= MESSAGE_TYPE_MASK;
 	assert(type != PTYPE_SYSTEM && type != PTYPE_HARBOR && REMOTE);
 	skynet_context_send(REMOTE, rmsg, sizeof(*rmsg) , source, type , session);
 }

skynet-src/skynet_harbor.h

@@ -7,10 +7,6 @@
 #define GLOBALNAME_LENGTH 16
 #define REMOTE_MAX 256
 
-// reserve high 8 bits for remote id
-#define HANDLE_MASK 0xffffff
-#define HANDLE_REMOTE_SHIFT 24
-
 struct remote_name {
 	char name[GLOBALNAME_LENGTH];
 	uint32_t handle;

skynet-src/skynet_module.c

@@ -1,6 +1,7 @@
 #include "skynet.h"
 
 #include "skynet_module.h"
+#include "spinlock.h"
 
 #include <assert.h>
 #include <string.h>
@@ -13,7 +14,7 @@
 
 struct modules {
 	int count;
-	int lock;
+	struct spinlock lock;
 	const char * path;
 	struct skynet_module m[MAX_MODULE_TYPE];
 };
@@ -95,7 +96,7 @@ skynet_module_query(const char * name) {
 	if (result)
 		return result;
 
-	while(__sync_lock_test_and_set(&M->lock,1)) {}
+	SPIN_LOCK(M)
 
 	result = _query(name); // double check
 
@@ -114,21 +115,22 @@ skynet_module_query(const char * name) {
 		}
 	}
 
-	__sync_lock_release(&M->lock);
+	SPIN_UNLOCK(M)
 
 	return result;
 }
 
 void 
 skynet_module_insert(struct skynet_module *mod) {
-	while(__sync_lock_test_and_set(&M->lock,1)) {}
+	SPIN_LOCK(M)
 
 	struct skynet_module * m = _query(mod->name);
 	assert(m == NULL && M->count < MAX_MODULE_TYPE);
 	int index = M->count;
 	M->m[index] = *mod;
 	++M->count;
-	__sync_lock_release(&M->lock);
+
+	SPIN_UNLOCK(M)
 }
 
 void * 
@@ -164,7 +166,8 @@ skynet_module_init(const char *path) {
 	struct modules *m = skynet_malloc(sizeof(*m));
 	m->count = 0;
 	m->path = skynet_strdup(path);
-	m->lock = 0;
+
+	SPIN_INIT(m)
 
 	M = m;
 }

skynet-src/skynet_monitor.c

@@ -3,6 +3,7 @@
 #include "skynet_monitor.h"
 #include "skynet_server.h"
 #include "skynet.h"
+#include "atomic.h"
 
 #include <stdlib.h>
 #include <string.h>
@@ -30,7 +31,7 @@ void
 skynet_monitor_trigger(struct skynet_monitor *sm, uint32_t source, uint32_t destination) {
 	sm->source = source;
 	sm->destination = destination;
-	__sync_fetch_and_add(&sm->version , 1);
+	ATOM_INC(&sm->version);
 }
 
 void 

skynet-src/skynet_mq.c

@@ -1,6 +1,7 @@
 #include "skynet.h"
 #include "skynet_mq.h"
 #include "skynet_handle.h"
+#include "spinlock.h"
 
 #include <stdio.h>
 #include <stdlib.h>
@@ -18,11 +19,11 @@
 #define MQ_OVERLOAD 1024
 
 struct message_queue {
+	struct spinlock lock;
 	uint32_t handle;
 	int cap;
 	int head;
 	int tail;
-	int lock;
 	int release;
 	int in_global;
 	int overload;
@@ -34,19 +35,16 @@ struct message_queue {
 struct global_queue {
 	struct message_queue *head;
 	struct message_queue *tail;
-	int lock;
+	struct spinlock lock;
 };
 
 static struct global_queue *Q = NULL;
 
-#define LOCK(q) while (__sync_lock_test_and_set(&(q)->lock,1)) {}
-#define UNLOCK(q) __sync_lock_release(&(q)->lock);
-
 void 
 skynet_globalmq_push(struct message_queue * queue) {
 	struct global_queue *q= Q;
 
-	LOCK(q)
+	SPIN_LOCK(q)
 	assert(queue->next == NULL);
 	if(q->tail) {
 		q->tail->next = queue;
@@ -54,14 +52,14 @@ skynet_globalmq_push(struct message_queue * queue) {
 	} else {
 		q->head = q->tail = queue;
 	}
-	UNLOCK(q)
+	SPIN_UNLOCK(q)
 }
 
 struct message_queue * 
 skynet_globalmq_pop() {
 	struct global_queue *q = Q;
 
-	LOCK(q)
+	SPIN_LOCK(q)
 	struct message_queue *mq = q->head;
 	if(mq) {
 		q->head = mq->next;
@@ -71,7 +69,7 @@ skynet_globalmq_pop() {
 		}
 		mq->next = NULL;
 	}
-	UNLOCK(q)
+	SPIN_UNLOCK(q)
 
 	return mq;
 }
@@ -83,7 +81,7 @@ skynet_mq_create(uint32_t handle) {
 	q->cap = DEFAULT_QUEUE_SIZE;
 	q->head = 0;
 	q->tail = 0;
-	q->lock = 0;
+	SPIN_INIT(q)
 	// When the queue is create (always between service create and service init) ,
 	// set in_global flag to avoid push it to global queue .
 	// If the service init success, skynet_context_new will call skynet_mq_force_push to push it to global queue.
@@ -100,6 +98,7 @@ skynet_mq_create(uint32_t handle) {
 static void 
 _release(struct message_queue *q) {
 	assert(q->next == NULL);
+	SPIN_DESTROY(q)
 	skynet_free(q->queue);
 	skynet_free(q);
 }
@@ -113,11 +112,11 @@ int
 skynet_mq_length(struct message_queue *q) {
 	int head, tail,cap;
 
-	LOCK(q)
+	SPIN_LOCK(q)
 	head = q->head;
 	tail = q->tail;
 	cap = q->cap;
-	UNLOCK(q)
+	SPIN_UNLOCK(q)
 	
 	if (head <= tail) {
 		return tail - head;
@@ -138,7 +137,7 @@ skynet_mq_overload(struct message_queue *q) {
 int
 skynet_mq_pop(struct message_queue *q, struct skynet_message *message) {
 	int ret = 1;
-	LOCK(q)
+	SPIN_LOCK(q)
 
 	if (q->head != q->tail) {
 		*message = q->queue[q->head++];
@@ -167,7 +166,7 @@ skynet_mq_pop(struct message_queue *q, struct skynet_message *message) {
 		q->in_global = 0;
 	}
 	
-	UNLOCK(q)
+	SPIN_UNLOCK(q)
 
 	return ret;
 }
@@ -190,7 +189,7 @@ expand_queue(struct message_queue *q) {
 void 
 skynet_mq_push(struct message_queue *q, struct skynet_message *message) {
 	assert(message);
-	LOCK(q)
+	SPIN_LOCK(q)
 
 	q->queue[q->tail] = *message;
 	if (++ q->tail >= q->cap) {
@@ -206,25 +205,26 @@ skynet_mq_push(struct message_queue *q, struct skynet_message *message) {
 		skynet_globalmq_push(q);
 	}
 	
-	UNLOCK(q)
+	SPIN_UNLOCK(q)
 }
 
 void 
 skynet_mq_init() {
 	struct global_queue *q = skynet_malloc(sizeof(*q));
 	memset(q,0,sizeof(*q));
+	SPIN_INIT(q);
 	Q=q;
 }
 
 void 
 skynet_mq_mark_release(struct message_queue *q) {
-	LOCK(q)
+	SPIN_LOCK(q)
 	assert(q->release == 0);
 	q->release = 1;
 	if (q->in_global != MQ_IN_GLOBAL) {
 		skynet_globalmq_push(q);
 	}
-	UNLOCK(q)
+	SPIN_UNLOCK(q)
 }
 
 static void
@@ -238,13 +238,13 @@ _drop_queue(struct message_queue *q, message_drop drop_func, void *ud) {
 
 void 
 skynet_mq_release(struct message_queue *q, message_drop drop_func, void *ud) {
-	LOCK(q)
+	SPIN_LOCK(q)
 	
 	if (q->release) {
-		UNLOCK(q)
+		SPIN_UNLOCK(q)
 		_drop_queue(q, drop_func, ud);
 	} else {
 		skynet_globalmq_push(q);
-		UNLOCK(q)
+		SPIN_UNLOCK(q)
 	}
 }

skynet-src/skynet_mq.h

@@ -11,6 +11,10 @@ struct skynet_message {
 	size_t sz;
 };
 
+// type is encoding in skynet_message.sz high 8bit
+#define MESSAGE_TYPE_MASK (SIZE_MAX >> 8)
+#define MESSAGE_TYPE_SHIFT ((sizeof(size_t)-1) * 8)
+
 struct message_queue;
 
 void skynet_globalmq_push(struct message_queue * queue);

skynet-src/skynet_server.c

@@ -10,6 +10,8 @@
 #include "skynet_monitor.h"
 #include "skynet_imp.h"
 #include "skynet_log.h"
+#include "spinlock.h"
+#include "atomic.h"
 
 #include <pthread.h>
 
@@ -21,16 +23,18 @@
 
 #ifdef CALLING_CHECK
 
-#define CHECKCALLING_BEGIN(ctx) assert(__sync_lock_test_and_set(&ctx->calling,1) == 0);
-#define CHECKCALLING_END(ctx) __sync_lock_release(&ctx->calling);
-#define CHECKCALLING_INIT(ctx) ctx->calling = 0;
-#define CHECKCALLING_DECL int calling;
+#define CHECKCALLING_BEGIN(ctx) if (!(spinlock_trylock(&ctx->calling))) { assert(0); }
+#define CHECKCALLING_END(ctx) spinlock_unlock(&ctx->calling);
+#define CHECKCALLING_INIT(ctx) spinlock_init(&ctx->calling);
+#define CHECKCALLING_DESTROY(ctx) spinlock_destroy(&ctx->calling);
+#define CHECKCALLING_DECL struct spinlock calling;
 
 #else
 
 #define CHECKCALLING_BEGIN(ctx)
 #define CHECKCALLING_END(ctx)
 #define CHECKCALLING_INIT(ctx)
+#define CHECKCALLING_DESTROY(ctx)
 #define CHECKCALLING_DECL
 
 #endif
@@ -68,12 +72,12 @@ skynet_context_total() {
 
 static void
 context_inc() {
-	__sync_fetch_and_add(&G_NODE.total,1);
+	ATOM_INC(&G_NODE.total);
 }
 
 static void
 context_dec() {
-	__sync_fetch_and_sub(&G_NODE.total,1);
+	ATOM_DEC(&G_NODE.total);
 }
 
 uint32_t 
@@ -179,7 +183,7 @@ skynet_context_newsession(struct skynet_context *ctx) {
 
 void 
 skynet_context_grab(struct skynet_context *ctx) {
-	__sync_add_and_fetch(&ctx->ref,1);
+	ATOM_INC(&ctx->ref);
 }
 
 void
@@ -197,13 +201,14 @@ delete_context(struct skynet_context *ctx) {
 	}
 	skynet_module_instance_release(ctx->mod, ctx->instance);
 	skynet_mq_mark_release(ctx->queue);
+	CHECKCALLING_DESTROY(ctx)
 	skynet_free(ctx);
 	context_dec();
 }
 
 struct skynet_context * 
 skynet_context_release(struct skynet_context *ctx) {
-	if (__sync_sub_and_fetch(&ctx->ref,1) == 0) {
+	if (ATOM_DEC(&ctx->ref) == 0) {
 		delete_context(ctx);
 		return NULL;
 	}
@@ -246,8 +251,8 @@ dispatch_message(struct skynet_context *ctx, struct skynet_message *msg) {
 	assert(ctx->init);
 	CHECKCALLING_BEGIN(ctx)
 	pthread_setspecific(G_NODE.handle_key, (void *)(uintptr_t)(ctx->handle));
-	int type = msg->sz >> HANDLE_REMOTE_SHIFT;
-	size_t sz = msg->sz & HANDLE_MASK;
+	int type = msg->sz >> MESSAGE_TYPE_SHIFT;
+	size_t sz = msg->sz & MESSAGE_TYPE_MASK;
 	if (ctx->logfile) {
 		skynet_log_output(ctx->logfile, msg->source, type, msg->session, msg->data, sz);
 	}
@@ -560,7 +565,7 @@ cmd_logon(struct skynet_context * context, const char * param) {
 	if (lastf == NULL) {
 		f = skynet_log_open(context, handle);
 		if (f) {
-			if (!__sync_bool_compare_and_swap(&ctx->logfile, NULL, f)) {
+			if (!ATOM_CAS_POINTER(&ctx->logfile, NULL, f)) {
 				// logfile opens in other thread, close this one.
 				fclose(f);
 			}
@@ -581,7 +586,7 @@ cmd_logoff(struct skynet_context * context, const char * param) {
 	FILE * f = ctx->logfile;
 	if (f) {
 		// logfile may close in other thread
-		if (__sync_bool_compare_and_swap(&ctx->logfile, f, NULL)) {
+		if (ATOM_CAS_POINTER(&ctx->logfile, f, NULL)) {
 			skynet_log_close(context, f, handle);
 		}
 	}
@@ -662,13 +667,13 @@ _filter_args(struct skynet_context * context, int type, int *session, void ** da
 		*data = msg;
 	}
 
-	*sz |= type << HANDLE_REMOTE_SHIFT;
+	*sz |= (size_t)type << MESSAGE_TYPE_SHIFT;
 }
 
 int
 skynet_send(struct skynet_context * context, uint32_t source, uint32_t destination , int type, int session, void * data, size_t sz) {
-	if ((sz & HANDLE_MASK) != sz) {
-		skynet_error(context, "The message to %x is too large (sz = %lu)", destination, sz);
+	if ((sz & MESSAGE_TYPE_MASK) != sz) {
+		skynet_error(context, "The message to %x is too large", destination);
 		skynet_free(data);
 		return -1;
 	}
@@ -751,7 +756,7 @@ skynet_context_send(struct skynet_context * ctx, void * msg, size_t sz, uint32_t
 	smsg.source = source;
 	smsg.session = session;
 	smsg.data = msg;
-	smsg.sz = sz | type << HANDLE_REMOTE_SHIFT;
+	smsg.sz = sz | (size_t)type << MESSAGE_TYPE_SHIFT;
 
 	skynet_mq_push(ctx->queue, &smsg);
 }

skynet-src/skynet_socket.c

@@ -33,7 +33,7 @@ skynet_socket_free() {
 static void
 forward_message(int type, bool padding, struct socket_message * result) {
 	struct skynet_socket_message *sm;
-	int sz = sizeof(*sm);
+	size_t sz = sizeof(*sm);
 	if (padding) {
 		if (result->data) {
 			sz += strlen(result->data);
@@ -56,7 +56,7 @@ forward_message(int type, bool padding, struct socket_message * result) {
 	message.source = 0;
 	message.session = 0;
 	message.data = sm;
-	message.sz = sz | PTYPE_SOCKET << HANDLE_REMOTE_SHIFT;
+	message.sz = sz | ((size_t)PTYPE_SOCKET << MESSAGE_TYPE_SHIFT);
 	
 	if (skynet_context_push((uint32_t)result->opaque, &message)) {
 		// todo: report somewhere to close socket
@@ -109,10 +109,13 @@ check_wsz(struct skynet_context *ctx, int id, void *buffer, int64_t wsz) {
 	if (wsz < 0) {
 		return -1;
 	} else if (wsz > 1024 * 1024) {
-		int kb4 = wsz / 1024 / 4;
-		if (kb4 % 256 == 0) {
-			skynet_error(ctx, "%d Mb bytes on socket %d need to send out", (int)(wsz / (1024 * 1024)), id);
-		}
+		struct skynet_socket_message tmp;
+		tmp.type = SKYNET_SOCKET_TYPE_WARNING;
+		tmp.id = id;
+		tmp.ud = (int)(wsz / 1024);
+		tmp.buffer = NULL;
+		skynet_send(ctx, 0, skynet_context_handle(ctx), PTYPE_SOCKET, 0 , &tmp, sizeof(tmp));
+//		skynet_error(ctx, "%d Mb bytes on socket %d need to send out", (int)(wsz / (1024 * 1024)), id);
 	}
 	return 0;
 }

skynet-src/skynet_socket.h

@@ -9,6 +9,7 @@ struct skynet_context;
 #define SKYNET_SOCKET_TYPE_ACCEPT 4
 #define SKYNET_SOCKET_TYPE_ERROR 5
 #define SKYNET_SOCKET_TYPE_UDP 6
+#define SKYNET_SOCKET_TYPE_WARNING 7
 
 struct skynet_socket_message {
 	int type;

skynet-src/skynet_start.c

@@ -8,6 +8,7 @@
 #include "skynet_monitor.h"
 #include "skynet_socket.h"
 #include "skynet_daemon.h"
+#include "skynet_harbor.h"
 
 #include <pthread.h>
 #include <unistd.h>

skynet-src/skynet_timer.c

@@ -4,6 +4,7 @@
 #include "skynet_mq.h"
 #include "skynet_server.h"
 #include "skynet_handle.h"
+#include "spinlock.h"
 
 #include <time.h>
 #include <assert.h>
@@ -17,9 +18,6 @@
 
 typedef void (*timer_execute_func)(void *ud,void *arg);
 
-#define LOCK(q) while (__sync_lock_test_and_set(&(q)->lock,1)) {}
-#define UNLOCK(q) __sync_lock_release(&(q)->lock);
-
 #define TIME_NEAR_SHIFT 8
 #define TIME_NEAR (1 << TIME_NEAR_SHIFT)
 #define TIME_LEVEL_SHIFT 6
@@ -45,7 +43,7 @@ struct link_list {
 struct timer {
 	struct link_list near[TIME_NEAR];
 	struct link_list t[4][TIME_LEVEL];
-	int lock;
+	struct spinlock lock;
 	uint32_t time;
 	uint32_t current;
 	uint32_t starttime;
@@ -97,12 +95,12 @@ timer_add(struct timer *T,void *arg,size_t sz,int time) {
 	struct timer_node *node = (struct timer_node *)skynet_malloc(sizeof(*node)+sz);
 	memcpy(node+1,arg,sz);
 
-	LOCK(T);
+	SPIN_LOCK(T);
 
 		node->expire=time+T->time;
 		add_node(T,node);
 
-	UNLOCK(T);
+	SPIN_UNLOCK(T);
 }
 
 static void
@@ -146,7 +144,7 @@ dispatch_list(struct timer_node *current) {
 		message.source = 0;
 		message.session = event->session;
 		message.data = NULL;
-		message.sz = PTYPE_RESPONSE << HANDLE_REMOTE_SHIFT;
+		message.sz = (size_t)PTYPE_RESPONSE << MESSAGE_TYPE_SHIFT;
 
 		skynet_context_push(event->handle, &message);
 		
@@ -162,16 +160,16 @@ timer_execute(struct timer *T) {
 	
 	while (T->near[idx].head.next) {
 		struct timer_node *current = link_clear(&T->near[idx]);
-		UNLOCK(T);
+		SPIN_UNLOCK(T);
 		// dispatch_list don't need lock T
 		dispatch_list(current);
-		LOCK(T);
+		SPIN_LOCK(T);
 	}
 }
 
 static void 
 timer_update(struct timer *T) {
-	LOCK(T);
+	SPIN_LOCK(T);
 
 	// try to dispatch timeout 0 (rare condition)
 	timer_execute(T);
@@ -181,7 +179,7 @@ timer_update(struct timer *T) {
 
 	timer_execute(T);
 
-	UNLOCK(T);
+	SPIN_UNLOCK(T);
 }
 
 static struct timer *
@@ -201,7 +199,8 @@ timer_create_timer() {
 		}
 	}
 
-	r->lock = 0;
+	SPIN_INIT(r)
+
 	r->current = 0;
 
 	return r;
@@ -214,7 +213,7 @@ skynet_timeout(uint32_t handle, int time, int session) {
 		message.source = 0;
 		message.session = session;
 		message.data = NULL;
-		message.sz = PTYPE_RESPONSE << HANDLE_REMOTE_SHIFT;
+		message.sz = (size_t)PTYPE_RESPONSE << MESSAGE_TYPE_SHIFT;
 
 		if (skynet_context_push(handle, &message)) {
 			return -1;

skynet-src/socket_server.c

@@ -2,6 +2,7 @@
 
 #include "socket_server.h"
 #include "socket_poll.h"
+#include "atomic.h"
 
 #include <sys/types.h>
 #include <sys/socket.h>
@@ -237,13 +238,13 @@ static int
 reserve_id(struct socket_server *ss) {
 	int i;
 	for (i=0;i<MAX_SOCKET;i++) {
-		int id = __sync_add_and_fetch(&(ss->alloc_id), 1);
+		int id = ATOM_INC(&(ss->alloc_id));
 		if (id < 0) {
-			id = __sync_and_and_fetch(&(ss->alloc_id), 0x7fffffff);
+			id = ATOM_AND(&(ss->alloc_id), 0x7fffffff);
 		}
 		struct socket *s = &ss->slot[HASH_ID(id)];
 		if (s->type == SOCKET_TYPE_INVALID) {
-			if (__sync_bool_compare_and_swap(&s->type, SOCKET_TYPE_INVALID, SOCKET_TYPE_RESERVE)) {
+			if (ATOM_CAS(&s->type, SOCKET_TYPE_INVALID, SOCKET_TYPE_RESERVE)) {
 				s->id = id;
 				s->fd = -1;
 				return id;

skynet-src/spinlock.h

@@ -0,0 +1,77 @@
+#ifndef SKYNET_SPINLOCK_H
+#define SKYNET_SPINLOCK_H
+
+#define SPIN_INIT(q) spinlock_init(&(q)->lock);
+#define SPIN_LOCK(q) spinlock_lock(&(q)->lock);
+#define SPIN_UNLOCK(q) spinlock_unlock(&(q)->lock);
+#define SPIN_DESTROY(q) spinlock_destroy(&(q)->lock);
+
+#ifndef USE_PTHREAD_LOCK
+
+struct spinlock {
+	int lock;
+};
+
+static inline void
+spinlock_init(struct spinlock *lock) {
+	lock->lock = 0;
+}
+
+static inline void
+spinlock_lock(struct spinlock *lock) {
+	while (__sync_lock_test_and_set(&lock->lock,1)) {}
+}
+
+static inline int
+spinlock_trylock(struct spinlock *lock) {
+	return __sync_lock_test_and_set(&lock->lock,1) == 0;
+}
+
+static inline void
+spinlock_unlock(struct spinlock *lock) {
+	__sync_lock_release(&lock->lock);
+}
+
+static inline void
+spinlock_destroy(struct spinlock *lock) {
+}
+
+#else
+
+#include <pthread.h>
+
+// we use mutex instead of spinlock for some reason
+// you can also replace to pthread_spinlock
+
+struct spinlock {
+	pthread_mutex_t lock;
+};
+
+static inline void
+spinlock_init(struct spinlock *lock) {
+	pthread_mutex_init(&lock->lock, NULL);
+}
+
+static inline void
+spinlock_lock(struct spinlock *lock) {
+	pthread_mutex_lock(&lock->lock);
+}
+
+static inline int
+spinlock_trylock(struct spinlock *lock) {
+	return pthread_mutex_trylock(&lock->lock) == 0;
+}
+
+static inline void
+spinlock_unlock(struct spinlock *lock) {
+	pthread_mutex_unlock(&lock->lock);
+}
+
+static inline void
+spinlock_destroy(struct spinlock *lock) {
+	pthread_mutex_destroy(&lock->lock);
+}
+
+#endif
+
+#endif