cluster rpc support large package

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"))
 

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] = 0x80;
+		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/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/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