Merge pull request #320 from cloudwu/multipart

Multipart

examples/cluster1.lua

@@ -6,6 +6,7 @@ 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"))

examples/cluster2.lua

@@ -3,7 +3,12 @@ local cluster = require "cluster"
 
 skynet.start(function()
 	local proxy = cluster.proxy("db", ".simpledb")
-	print(skynet.call(proxy, "lua", "GET", "a"))
+	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", ".simpledb", "GET", "a"))
 	print(cluster.call("db2", ".simpledb", "GET", "b"))
 

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

lualib-src/lua-cluster.c

@@ -15,7 +15,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) {
@@ -35,21 +36,69 @@ fill_header(lua_State *L, uint8_t *buf, int sz, void *msg) {
 	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, msg);
+		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, msg);
+		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 +107,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, msg);
+		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, msg);
+		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, msg);
+		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 +162,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 +197,7 @@ lpackrequest(lua_State *L) {
 		uint32_t or string addr
 		int session
 		string msg
+		boolean padding
  */
 
 static inline uint32_t
@@ -107,44 +206,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 != 15) {
+		return luaL_error(L, "Invalid cluster message size %d (multi req must be 15)", 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,14 +340,54 @@ 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, msg);
+			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, msg);
+				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_uint32(buf+2, session);
@@ -187,6 +404,7 @@ lpackresponse(lua_State *L) {
 	return integer session
 		boolean ok
 		string msg
+		boolean padding
  */
 static int
 lunpackresponse(lua_State *L) {
@@ -197,10 +415,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 +484,7 @@ luaopen_cluster_core(lua_State *L) {
 		{ "unpackrequest", lunpackrequest },
 		{ "packresponse", lpackresponse },
 		{ "unpackresponse", lunpackresponse },
+		{ "concat", lconcat },
 		{ NULL, NULL },
 	};
 	luaL_checkversion(L);

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/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_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);
 

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,51 @@ function command.proxy(source, node, name)
 	skynet.ret(skynet.pack(proxy[fullname]))
 end
 
-local request_fd = {}
+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 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 , msg, sz = pcall(skynet.rawcall, addr, "lua", msg, sz)
 		local response
 		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

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_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

@@ -246,8 +246,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);
 	}
@@ -662,13 +662,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 +751,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

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

@@ -146,7 +146,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);
 		
@@ -214,7 +214,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;