diff --git a/examples/cluster1.lua b/examples/cluster1.lua index 9ba8fcbf..d44899ba 100644 --- a/examples/cluster1.lua +++ b/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")) diff --git a/examples/cluster2.lua b/examples/cluster2.lua index 09e53076..0aef412c 100644 --- a/examples/cluster2.lua +++ b/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")) diff --git a/lualib-src/lua-cluster.c b/lualib-src/lua-cluster.c index e9d9e142..c558fb0f 100644 --- a/lualib-src/lua-cluster.c +++ 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 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 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); diff --git a/lualib/socketchannel.lua b/lualib/socketchannel.lua index 282e6283..9ff5d451 100644 --- a/lualib/socketchannel.lua +++ b/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 diff --git a/service/clusterd.lua b/service/clusterd.lua index fc8b4126..b3c74722 100644 --- a/service/clusterd.lua +++ b/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