diff --git a/examples/cluster2.lua b/examples/cluster2.lua index 84648567..22c9cf7f 100644 --- a/examples/cluster2.lua +++ b/examples/cluster2.lua @@ -11,9 +11,11 @@ skynet.start(function() print(skynet.call(proxy, "lua", "SET", largekey, largevalue)) local v = skynet.call(proxy, "lua", "GET", largekey) assert(largevalue == v) + skynet.send(proxy, "lua", "PING", "proxy") print(cluster.call("db", sdb, "GET", "a")) print(cluster.call("db2", sdb, "GET", "b")) + cluster.send("db2", sdb, "PING", "db2:longstring" .. largevalue) -- test snax service local pingserver = cluster.snax("db", "pingserver") diff --git a/examples/simpledb.lua b/examples/simpledb.lua index d073dfe6..af6bac5e 100644 --- a/examples/simpledb.lua +++ b/examples/simpledb.lua @@ -16,7 +16,17 @@ end skynet.start(function() skynet.dispatch("lua", function(session, address, cmd, ...) - local f = command[string.upper(cmd)] + cmd = cmd:upper() + if cmd == "PING" then + assert(session == 0) + local str = (...) + if #str > 20 then + str = str:sub(1,20) .. "...(" .. #str .. ")" + end + skynet.error(string.format("%s ping %s", skynet.address(address), str)) + return + end + local f = command[cmd] if f then skynet.ret(skynet.pack(f(...))) else diff --git a/lualib-src/lua-cluster.c b/lualib-src/lua-cluster.c index 37696562..ce54df45 100644 --- a/lualib-src/lua-cluster.c +++ b/lualib-src/lua-cluster.c @@ -46,7 +46,7 @@ fill_header(lua_State *L, uint8_t *buf, int sz) { PADDING msg(sz) size > 0x8000 and address is id DWORD 13 - BYTE 1 ; multireq + BYTE 1 ; multireq , 0x41: multi push DWORD addr DWORD session DWORD sz @@ -60,7 +60,7 @@ fill_header(lua_State *L, uint8_t *buf, int sz) { PADDING msg(sz) size > 0x8000 and address is string DWORD 10 + namelen - BYTE 0x81 + BYTE 0x81 ; 0xc1 : multi push BYTE namelen STRING name DWORD session @@ -73,14 +73,14 @@ fill_header(lua_State *L, uint8_t *buf, int sz) { PADDING msgpart(sz) */ static int -packreq_number(lua_State *L, int session, void * msg, uint32_t sz) { +packreq_number(lua_State *L, int session, void * msg, uint32_t sz, int is_push) { uint32_t addr = (uint32_t)lua_tointeger(L,1); uint8_t buf[TEMP_LENGTH]; 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); + fill_uint32(buf+7, is_push ? 0 : (uint32_t)session); memcpy(buf+11,msg,sz); lua_pushlstring(L, (const char *)buf, sz+11); @@ -88,7 +88,7 @@ packreq_number(lua_State *L, int session, void * msg, uint32_t sz) { } else { int part = (sz - 1) / MULTI_PART + 1; fill_header(L, buf, 13); - buf[2] = 1; + buf[2] = is_push ? 0x41 : 1; // multi push or request fill_uint32(buf+3, addr); fill_uint32(buf+7, (uint32_t)session); fill_uint32(buf+11, sz); @@ -98,7 +98,7 @@ packreq_number(lua_State *L, int session, void * msg, uint32_t sz) { } static int -packreq_string(lua_State *L, int session, void * msg, uint32_t sz) { +packreq_string(lua_State *L, int session, void * msg, uint32_t sz, int is_push) { size_t namelen = 0; const char *name = lua_tolstring(L, 1, &namelen); if (name == NULL || namelen < 1 || namelen > 255) { @@ -112,7 +112,7 @@ packreq_string(lua_State *L, int session, void * msg, uint32_t sz) { buf[2] = 0x80; buf[3] = (uint8_t)namelen; memcpy(buf+4, name, namelen); - fill_uint32(buf+4+namelen, (uint32_t)session); + fill_uint32(buf+4+namelen, is_push ? 0 : (uint32_t)session); memcpy(buf+8+namelen,msg,sz); lua_pushlstring(L, (const char *)buf, sz+8+namelen); @@ -120,7 +120,7 @@ packreq_string(lua_State *L, int session, void * msg, uint32_t sz) { } else { int part = (sz - 1) / MULTI_PART + 1; fill_header(L, buf, 10+namelen); - buf[2] = 0x81; + buf[2] = is_push ? 0xc1 : 0x81; // multi push or request buf[3] = (uint8_t)namelen; memcpy(buf+4, name, namelen); fill_uint32(buf+4+namelen, (uint32_t)session); @@ -157,7 +157,7 @@ packreq_multi(lua_State *L, int session, void * msg, uint32_t sz) { } static int -lpackrequest(lua_State *L) { +packrequest(lua_State *L, int is_push) { void *msg = lua_touserdata(L,3); if (msg == NULL) { return luaL_error(L, "Invalid request message"); @@ -171,9 +171,9 @@ lpackrequest(lua_State *L) { int addr_type = lua_type(L,1); int multipak; if (addr_type == LUA_TNUMBER) { - multipak = packreq_number(L, session, msg, sz); + multipak = packreq_number(L, session, msg, sz, is_push); } else { - multipak = packreq_string(L, session, msg, sz); + multipak = packreq_string(L, session, msg, sz, is_push); } int current_session = session; if (++session < 0) { @@ -191,6 +191,16 @@ lpackrequest(lua_State *L) { } } +static int +lpackrequest(lua_State *L) { + return packrequest(L, 0); +} + +static int +lpackpush(lua_State *L) { + return packrequest(L, 1); +} + /* string packed message return @@ -215,12 +225,17 @@ unpackreq_number(lua_State *L, const uint8_t * buf, int sz) { lua_pushinteger(L, address); lua_pushinteger(L, session); lua_pushlstring(L, (const char *)buf+9, sz-9); + if (session == 0) { + lua_pushnil(L); + lua_pushboolean(L,1); // is_push, no reponse + return 5; + } return 3; } static int -unpackmreq_number(lua_State *L, const uint8_t * buf, int sz) { +unpackmreq_number(lua_State *L, const uint8_t * buf, int sz, int is_push) { if (sz != 13) { return luaL_error(L, "Invalid cluster message size %d (multi req must be 13)", sz); } @@ -231,8 +246,9 @@ unpackmreq_number(lua_State *L, const uint8_t * buf, int sz) { lua_pushinteger(L, session); lua_pushinteger(L, size); lua_pushboolean(L, 1); // padding multi part + lua_pushboolean(L, is_push); - return 4; + return 5; } static int @@ -263,12 +279,17 @@ unpackreq_string(lua_State *L, const uint8_t * buf, int sz) { uint32_t session = unpack_uint32(buf + namesz + 2); lua_pushinteger(L, (uint32_t)session); lua_pushlstring(L, (const char *)buf+2+namesz+4, sz - namesz - 6); + if (session == 0) { + lua_pushnil(L); + lua_pushboolean(L,1); // is_push, no reponse + return 5; + } return 3; } static int -unpackmreq_string(lua_State *L, const uint8_t * buf, int sz) { +unpackmreq_string(lua_State *L, const uint8_t * buf, int sz, int is_push) { if (sz < 2) { return luaL_error(L, "Invalid cluster message (size=%d)", sz); } @@ -282,8 +303,9 @@ unpackmreq_string(lua_State *L, const uint8_t * buf, int sz) { lua_pushinteger(L, session); lua_pushinteger(L, size); lua_pushboolean(L, 1); // padding multipart + lua_pushboolean(L, is_push); - return 4; + return 5; } static int @@ -295,14 +317,18 @@ lunpackrequest(lua_State *L) { case 0: return unpackreq_number(L, (const uint8_t *)msg, sz); case 1: - return unpackmreq_number(L, (const uint8_t *)msg, sz); + return unpackmreq_number(L, (const uint8_t *)msg, sz, 0); // request + case '\x41': + return unpackmreq_number(L, (const uint8_t *)msg, sz, 1); // push 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); + return unpackmreq_string(L, (const uint8_t *)msg, sz, 0 ); // request + case '\xc1': + return unpackmreq_string(L, (const uint8_t *)msg, sz, 1 ); // push default: return luaL_error(L, "Invalid req package type %d", msg[0]); } @@ -481,6 +507,7 @@ LUAMOD_API int luaopen_cluster_core(lua_State *L) { luaL_Reg l[] = { { "packrequest", lpackrequest }, + { "packpush", lpackpush }, { "unpackrequest", lunpackrequest }, { "packresponse", lpackresponse }, { "unpackresponse", lunpackresponse }, diff --git a/lualib/cluster.lua b/lualib/cluster.lua index 3b180753..14dee2ac 100644 --- a/lualib/cluster.lua +++ b/lualib/cluster.lua @@ -9,8 +9,8 @@ function cluster.call(node, address, ...) end function cluster.send(node, address, ...) - -- skynet.pack(...) will free by cluster.core.packrequest - skynet.send(clusterd, "lua", "req", node, address, skynet.pack(...)) + -- push is the same with req, but no response + skynet.send(clusterd, "lua", "push", node, address, skynet.pack(...)) end function cluster.open(port) diff --git a/lualib/skynet.lua b/lualib/skynet.lua index a40eff36..49c7e9b4 100644 --- a/lualib/skynet.lua +++ b/lualib/skynet.lua @@ -362,6 +362,11 @@ function skynet.send(addr, typename, ...) return c.send(addr, p.id, 0 , p.pack(...)) end +function skynet.rawsend(addr, typename, msg, sz) + local p = proto[typename] + return c.send(addr, p.id, 0 , msg, sz) +end + skynet.genid = assert(c.genid) skynet.redirect = function(dest,source,typename,...) diff --git a/service/clusterd.lua b/service/clusterd.lua index 20c43ee7..911b27d1 100644 --- a/service/clusterd.lua +++ b/service/clusterd.lua @@ -91,6 +91,21 @@ function command.req(...) end end +function command.push(source, node, addr, msg, sz) + local session = node_session[node] or 1 + local request, new_session, padding = cluster.packpush(addr, session, msg, sz) + if padding then -- is multi push + node_session[node] = new_session + end + + -- node_channel[node] may yield or throw error + local c = node_channel[node] + + c:request(request, nil, padding) + + -- notice: push may fail where the channel is disconnected or broken. +end + local proxy = {} function command.proxy(source, node, name) @@ -121,9 +136,9 @@ local large_request = {} function command.socket(source, subcmd, fd, msg) if subcmd == "data" then local sz - local addr, session, msg, padding = cluster.unpackrequest(msg) + local addr, session, msg, padding, is_push = cluster.unpackrequest(msg) if padding then - local req = large_request[session] or { addr = addr } + local req = large_request[session] or { addr = addr , is_push = is_push } large_request[session] = req table.insert(req, msg) return @@ -134,6 +149,7 @@ function command.socket(source, subcmd, fd, msg) table.insert(req, msg) msg,sz = cluster.concat(req) addr = req.addr + is_push = req.is_push end if not msg then local response = cluster.packresponse(session, false, "Invalid large req") @@ -152,6 +168,9 @@ function command.socket(source, subcmd, fd, msg) ok = false msg = "name not found" end + elseif is_push then + skynet.rawsend(addr, "lua", msg, sz) + return -- no response else ok , msg, sz = pcall(skynet.rawcall, addr, "lua", msg, sz) end diff --git a/service/clusterproxy.lua b/service/clusterproxy.lua index e003da2b..1e55cff6 100644 --- a/service/clusterproxy.lua +++ b/service/clusterproxy.lua @@ -23,6 +23,10 @@ skynet.forward_type( forward_map ,function() address = n end skynet.dispatch("system", function (session, source, msg, sz) - skynet.ret(skynet.rawcall(clusterd, "lua", skynet.pack("req", node, address, msg, sz))) + if session == 0 then + skynet.send(clusterd, "lua", "push", node, address, msg, sz) + else + skynet.ret(skynet.rawcall(clusterd, "lua", skynet.pack("req", node, address, msg, sz))) + end end) end)