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/examples/config.userlog b/examples/config.userlog index 97bef6ef..dcd47b7b 100644 --- a/examples/config.userlog +++ 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 diff --git a/lualib-src/lua-cluster.c b/lualib-src/lua-cluster.c index e9d9e142..40cc453e 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] = 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 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/mysql.lua b/lualib/mysql.lua index a83a0986..305e900b 100755 --- a/lualib/mysql.lua +++ b/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 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-src/service_harbor.c b/service-src/service_harbor.c index 5940d62c..90d8bcc6 100644 --- a/service-src/service_harbor.c +++ b/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); 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 diff --git a/skynet-src/skynet_error.c b/skynet-src/skynet_error.c index b9c53efd..962ea597 100644 --- a/skynet-src/skynet_error.c +++ b/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); } diff --git a/skynet-src/skynet_handle.h b/skynet-src/skynet_handle.h index e067eed0..293b6571 100644 --- a/skynet-src/skynet_handle.h +++ b/skynet-src/skynet_handle.h @@ -3,7 +3,9 @@ #include -#include "skynet_harbor.h" +// reserve high 8 bits for remote id +#define HANDLE_MASK 0xffffff +#define HANDLE_REMOTE_SHIFT 24 struct skynet_context; diff --git a/skynet-src/skynet_harbor.c b/skynet-src/skynet_harbor.c index 379fce80..5e39ca62 100644 --- a/skynet-src/skynet_harbor.c +++ b/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 #include @@ -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); } diff --git a/skynet-src/skynet_harbor.h b/skynet-src/skynet_harbor.h index 62982455..97a76c20 100644 --- a/skynet-src/skynet_harbor.h +++ b/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; diff --git a/skynet-src/skynet_mq.h b/skynet-src/skynet_mq.h index 17178b4f..3721e32b 100644 --- a/skynet-src/skynet_mq.h +++ b/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); diff --git a/skynet-src/skynet_server.c b/skynet-src/skynet_server.c index ea1aadcf..52331193 100644 --- a/skynet-src/skynet_server.c +++ b/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); } diff --git a/skynet-src/skynet_socket.c b/skynet-src/skynet_socket.c index 3f2c9a04..2ea26c0a 100644 --- a/skynet-src/skynet_socket.c +++ b/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 diff --git a/skynet-src/skynet_start.c b/skynet-src/skynet_start.c index 4123c624..5345c76c 100644 --- a/skynet-src/skynet_start.c +++ b/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 #include diff --git a/skynet-src/skynet_timer.c b/skynet-src/skynet_timer.c index f7d5a1ff..f23114dc 100644 --- a/skynet-src/skynet_timer.c +++ b/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;