From 70c224c3e7c020e555395a271b50ae39306a6f5a Mon Sep 17 00:00:00 2001 From: Cloud Wu Date: Sat, 14 Apr 2018 16:11:26 +0800 Subject: [PATCH 1/5] use clusteragent to dispatch cluster request --- examples/cluster2.lua | 2 +- lualib-src/lua-cluster.c | 13 +++-- service/clusteragent.lua | 92 +++++++++++++++++++++++++++++++++++ service/clusterd.lua | 101 ++++++++++++++------------------------- 4 files changed, 138 insertions(+), 70 deletions(-) create mode 100644 service/clusteragent.lua diff --git a/examples/cluster2.lua b/examples/cluster2.lua index 3b7a2c02..d249e7ac 100644 --- a/examples/cluster2.lua +++ b/examples/cluster2.lua @@ -8,7 +8,7 @@ skynet.start(function() local proxy = cluster.proxy("db", sdb) local largekey = string.rep("X", 128*1024) local largevalue = string.rep("R", 100 * 1024) - print(skynet.call(proxy, "lua", "SET", largekey, largevalue)) + skynet.call(proxy, "lua", "SET", largekey, largevalue) local v = skynet.call(proxy, "lua", "GET", largekey) assert(largevalue == v) skynet.send(proxy, "lua", "PING", "proxy") diff --git a/lualib-src/lua-cluster.c b/lualib-src/lua-cluster.c index 3c55ac62..e4e2f57f 100644 --- a/lualib-src/lua-cluster.c +++ b/lualib-src/lua-cluster.c @@ -312,9 +312,16 @@ unpackmreq_string(lua_State *L, const uint8_t * buf, int sz, int is_push) { static int lunpackrequest(lua_State *L) { - size_t ssz; - const char *msg = luaL_checklstring(L,1,&ssz); - int sz = (int)ssz; + int sz; + const char *msg; + if (lua_type(L, 1) == LUA_TLIGHTUSERDATA) { + msg = (const char *)lua_touserdata(L, 1); + sz = luaL_checkinteger(L, 2); + } else { + size_t ssz; + msg = luaL_checklstring(L,1,&ssz); + sz = (int)ssz; + } switch (msg[0]) { case 0: return unpackreq_number(L, (const uint8_t *)msg, sz); diff --git a/service/clusteragent.lua b/service/clusteragent.lua new file mode 100644 index 00000000..d297dbb7 --- /dev/null +++ b/service/clusteragent.lua @@ -0,0 +1,92 @@ +local skynet = require "skynet" +local sc = require "skynet.socketchannel" +local socket = require "skynet.socket" +local cluster = require "skynet.cluster.core" + +local clusterd, gate, fd = ... +clusterd = tonumber(clusterd) +gate = tonumber(gate) +fd = tonumber(fd) + +local large_request = {} +local register_name = {} + +local function dispatch_request(_,_,addr, session, msg, padding, is_push) + local sz + if padding then + local req = large_request[session] or { addr = addr , is_push = is_push } + 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 + is_push = req.is_push + end + if not msg then + local response = cluster.packresponse(session, false, "Invalid large req") + socket.write(fd, response) + return + end + end + local ok, response + if addr == 0 then + local name = skynet.unpack(msg, sz) + local addr = register_name[name] + if addr == nil then + addr = skynet.call(clusterd, "lua", "queryname", name) + register_name[name] = addr + end + if addr then + ok = true + msg, sz = skynet.pack(addr) + else + 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 + 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 +end + +skynet.start(function() + skynet.register_protocol { + name = "client", + id = skynet.PTYPE_CLIENT, + unpack = cluster.unpackrequest, + dispatch = dispatch_request, + } + -- fd can write, but don't read fd, the data package will forward from gate though client protocol. + skynet.call(gate, "lua", "forward", fd) + + skynet.dispatch("lua", function(_,source, cmd, ...) + if cmd == "exit" then + socket.close(fd) + skynet.exit() + elseif cmd == "namechange" then + register_name = {} + else + skynet.error(string.format("Invalid command %s from %s", cmd, skynet.address(source))) + end + end) +end) diff --git a/service/clusterd.lua b/service/clusterd.lua index 3196820a..381e6ad1 100644 --- a/service/clusterd.lua +++ b/service/clusterd.lua @@ -156,14 +156,24 @@ function command.proxy(source, node, name) skynet.ret(skynet.pack(proxy[fullname])) end +local cluster_agent = {} -- fd:service local register_name = {} +local function clearnamecache() + for fd, service in pairs(cluster_agent) do + if type(service) == "number" then + skynet.send(service, "lua", "namechange") + end + end +end + function command.register(source, name, addr) assert(register_name[name] == nil) addr = addr or source local old_name = register_name[addr] if old_name then register_name[old_name] = nil + clearnamecache() end register_name[addr] = name register_name[name] = addr @@ -171,76 +181,35 @@ function command.register(source, name, addr) skynet.error(string.format("Register [%s] :%08x", name, addr)) end -local large_request = {} +function command.queryname(source, name) + skynet.ret(skynet.pack(register_name[name])) +end function command.socket(source, subcmd, fd, msg) - if subcmd == "data" then - local sz - local addr, session, msg, padding, is_push = cluster.unpackrequest(msg) - if padding then - local requests = large_request[fd] - if requests == nil then - requests = {} - large_request[fd] = requests - end - local req = requests[session] or { addr = addr , is_push = is_push } - requests[session] = req - table.insert(req, msg) - return - else - local requests = large_request[fd] - if requests then - local req = requests[session] - if req then - requests[session] = nil - table.insert(req, msg) - msg,sz = cluster.concat(req) - addr = req.addr - is_push = req.is_push - end - end - if not msg then - local response = cluster.packresponse(session, false, "Invalid large req") - socket.write(fd, response) - return - end - end - local ok, response - if addr == 0 then - local name = skynet.unpack(msg, sz) - local addr = register_name[name] - if addr then - ok = true - msg, sz = skynet.pack(addr) - else - 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 - 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 - elseif subcmd == "open" then + if subcmd == "open" then skynet.error(string.format("socket accept from %s", msg)) - skynet.call(source, "lua", "accept", fd) + -- new cluster agent + cluster_agent[fd] = false + local agent = skynet.newservice("clusteragent", skynet.self(), source, fd) + local closed = cluster_agent[fd] + cluster_agent[fd] = agent + if closed then + skynet.send(agent, "lua", "exit") + cluster_agent[fd] = nil + end else - large_request[fd] = nil - skynet.error(string.format("socket %s %d %s", subcmd, fd, msg or "")) + if subcmd == "close" or subcmd == "error" then + -- close cluster agent + local agent = cluster_agent[fd] + if type(agent) == "boolean" then + cluster_agent[fd] = true + else + skynet.send(agent, "lua", "exit") + cluster_agent[fd] = nil + end + else + skynet.error(string.format("socket %s %d %s", subcmd, fd, msg or "")) + end end end From 4b88f68ab85ff21ef9856272c10f0844ec66e706 Mon Sep 17 00:00:00 2001 From: Cloud Wu Date: Sat, 14 Apr 2018 18:14:44 +0800 Subject: [PATCH 2/5] Improve cluster package module --- lualib-src/lua-cluster.c | 62 +++++++++++++++++++++++++++++++++------- service/clusteragent.lua | 8 +++--- skynet-src/malloc_hook.c | 2 +- 3 files changed, 56 insertions(+), 16 deletions(-) diff --git a/lualib-src/lua-cluster.c b/lualib-src/lua-cluster.c index e4e2f57f..8ea8ad08 100644 --- a/lualib-src/lua-cluster.c +++ b/lualib-src/lua-cluster.c @@ -208,8 +208,10 @@ lpackpush(lua_State *L) { return uint32_t or string addr int session - string msg + lightuserdata msg + int sz boolean padding + boolean is_push */ static inline uint32_t @@ -217,6 +219,14 @@ unpack_uint32(const uint8_t * buf) { return buf[0] | buf[1]<<8 | buf[2]<<16 | buf[3]<<24; } +static void +return_buffer(lua_State *L, const char * buffer, int sz) { + void * ptr = skynet_malloc(sz); + memcpy(ptr, buffer, sz); + lua_pushlightuserdata(L, ptr); + lua_pushinteger(L, sz); +} + static int unpackreq_number(lua_State *L, const uint8_t * buf, int sz) { if (sz < 9) { @@ -226,14 +236,15 @@ unpackreq_number(lua_State *L, const uint8_t * buf, int sz) { uint32_t session = unpack_uint32(buf+5); lua_pushinteger(L, address); lua_pushinteger(L, session); - lua_pushlstring(L, (const char *)buf+9, sz-9); + + return_buffer(L, (const char *)buf+9, sz-9); if (session == 0) { lua_pushnil(L); lua_pushboolean(L,1); // is_push, no reponse - return 5; + return 6; } - return 3; + return 4; } static int @@ -246,11 +257,12 @@ unpackmreq_number(lua_State *L, const uint8_t * buf, int sz, int is_push) { uint32_t size = unpack_uint32(buf+9); lua_pushinteger(L, address); lua_pushinteger(L, session); + lua_pushnil(L); lua_pushinteger(L, size); lua_pushboolean(L, 1); // padding multi part lua_pushboolean(L, is_push); - return 5; + return 6; } static int @@ -262,10 +274,10 @@ unpackmreq_part(lua_State *L, const uint8_t * buf, int sz) { 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); + return_buffer(L, (const char *)buf+5, sz-5); lua_pushboolean(L, padding); - return 4; + return 5; } static int @@ -280,14 +292,14 @@ unpackreq_string(lua_State *L, const uint8_t * buf, int 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+2+namesz+4, sz - namesz - 6); + return_buffer(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 6; } - return 3; + return 4; } static int @@ -303,11 +315,12 @@ unpackmreq_string(lua_State *L, const uint8_t * buf, int sz, int is_push) { uint32_t session = unpack_uint32(buf + namesz + 2); uint32_t size = unpack_uint32(buf + namesz + 6); lua_pushinteger(L, session); + lua_pushnil(L); lua_pushinteger(L, size); lua_pushboolean(L, 1); // padding multipart lua_pushboolean(L, is_push); - return 5; + return 6; } static int @@ -481,6 +494,32 @@ lunpackresponse(lua_State *L) { } } +/* + table + pointer + sz + + push (pointer/sz) as string into table, and free pointer + */ +static int +lappend(lua_State *L) { + luaL_checktype(L, 1, LUA_TTABLE); + int n = lua_rawlen(L, 1); + if (lua_isnil(L, 2)) { + lua_settop(L, 3); + lua_seti(L, 1, n + 1); + return 0; + } + void * buffer = lua_touserdata(L, 2); + if (buffer == NULL) + return luaL_error(L, "Need lightuserdata"); + int sz = luaL_checkinteger(L, 3); + lua_pushlstring(L, (const char *)buffer, sz); + skynet_free((void *)buffer); + lua_seti(L, 1, n+1); + return 0; +} + static int lconcat(lua_State *L) { if (!lua_istable(L,1)) @@ -522,6 +561,7 @@ luaopen_skynet_cluster_core(lua_State *L) { { "unpackrequest", lunpackrequest }, { "packresponse", lpackresponse }, { "unpackresponse", lunpackresponse }, + { "append", lappend }, { "concat", lconcat }, { NULL, NULL }, }; diff --git a/service/clusteragent.lua b/service/clusteragent.lua index d297dbb7..be201552 100644 --- a/service/clusteragent.lua +++ b/service/clusteragent.lua @@ -11,18 +11,17 @@ fd = tonumber(fd) local large_request = {} local register_name = {} -local function dispatch_request(_,_,addr, session, msg, padding, is_push) - local sz +local function dispatch_request(_,_,addr, session, msg, sz, padding, is_push) if padding then local req = large_request[session] or { addr = addr , is_push = is_push } large_request[session] = req - table.insert(req, msg) + cluster.append(req, msg, sz) return else local req = large_request[session] if req then large_request[session] = nil - table.insert(req, msg) + cluster.append(req, msg, sz) msg,sz = cluster.concat(req) addr = req.addr is_push = req.is_push @@ -36,6 +35,7 @@ local function dispatch_request(_,_,addr, session, msg, padding, is_push) local ok, response if addr == 0 then local name = skynet.unpack(msg, sz) + skynet.trash(msg, sz) local addr = register_name[name] if addr == nil then addr = skynet.call(clusterd, "lua", "queryname", name) diff --git a/skynet-src/malloc_hook.c b/skynet-src/malloc_hook.c index 25c3491b..7174f0a9 100644 --- a/skynet-src/malloc_hook.c +++ b/skynet-src/malloc_hook.c @@ -246,7 +246,7 @@ dump_c_mem() { struct mem_data* data = &mem_stats[i]; if(data->handle != 0 && data->allocated != 0) { total += data->allocated; - skynet_error(NULL, "0x%x -> %zdkb", data->handle, data->allocated >> 10); + skynet_error(NULL, ":%08x -> %zdkb %db", data->handle, data->allocated >> 10, (int)(data->allocated % 1024)); } } skynet_error(NULL, "+total: %zdkb",total >> 10); From 5a3d2fdac7c4b7fd9554838a7572505f368b2fc6 Mon Sep 17 00:00:00 2001 From: Cloud Wu Date: Sat, 14 Apr 2018 20:30:47 +0800 Subject: [PATCH 3/5] add addresscommand --- lualib-src/lua-skynet.c | 33 +++++++++++++++++++++++++++++++++ lualib/skynet.lua | 12 ++---------- 2 files changed, 35 insertions(+), 10 deletions(-) diff --git a/lualib-src/lua-skynet.c b/lualib-src/lua-skynet.c index daab81ed..c12d337f 100644 --- a/lualib-src/lua-skynet.c +++ b/lualib-src/lua-skynet.c @@ -120,6 +120,38 @@ lcommand(lua_State *L) { return 0; } +static int +laddresscommand(lua_State *L) { + struct skynet_context * context = lua_touserdata(L, lua_upvalueindex(1)); + const char * cmd = luaL_checkstring(L,1); + const char * result; + const char * parm = NULL; + if (lua_gettop(L) == 2) { + parm = luaL_checkstring(L,2); + } + result = skynet_command(context, cmd, parm); + if (result && result[0] == ':') { + int i; + uint32_t addr = 0; + for (i=1;result[i];i++) { + int c = result[i]; + if (c>='0' && c<='9') { + c = c - '0'; + } else if (c>='a' && c<='f') { + c = c - 'a' + 10; + } else if (c>='A' && c<='F') { + c = c - 'A' + 10; + } else { + return 0; + } + addr = addr * 16 + c; + } + lua_pushinteger(L, addr); + return 1; + } + return 0; +} + static int lintcommand(lua_State *L) { struct skynet_context * context = lua_touserdata(L, lua_upvalueindex(1)); @@ -356,6 +388,7 @@ luaopen_skynet_core(lua_State *L) { { "redirect", lredirect }, { "command" , lcommand }, { "intcommand", lintcommand }, + { "addresscommand", laddresscommand }, { "error", lerror }, { "tostring", ltostring }, { "harbor", lharbor }, diff --git a/lualib/skynet.lua b/lualib/skynet.lua index 0b16de0b..cf91d81d 100644 --- a/lualib/skynet.lua +++ b/lualib/skynet.lua @@ -295,20 +295,12 @@ function skynet.wait(co) session_id_coroutine[session] = nil end -local self_handle function skynet.self() - if self_handle then - return self_handle - end - self_handle = string_to_handle(c.command("REG")) - return self_handle + return c.addresscommand "REG" end function skynet.localname(name) - local addr = c.command("QUERY", name) - if addr then - return string_to_handle(addr) - end + return c.addresscommand("QUERY", name) end skynet.now = c.now From 311d7c8f6a6a1c04434d4dc49602a95817bb557e Mon Sep 17 00:00:00 2001 From: Cloud Wu Date: Mon, 16 Apr 2018 12:56:34 +0800 Subject: [PATCH 4/5] remove unused sz --- service/clusterd.lua | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/service/clusterd.lua b/service/clusterd.lua index 381e6ad1..0aae9ff5 100644 --- a/service/clusterd.lua +++ b/service/clusterd.lua @@ -118,7 +118,7 @@ local function send_request(source, node, addr, msg, sz) end function command.req(...) - local ok, msg, sz = pcall(send_request, ...) + local ok, msg = pcall(send_request, ...) if ok then if type(msg) == "table" then skynet.ret(cluster.concat(msg)) From 2bfba716ef4187f13111d736831750c4e7b40b93 Mon Sep 17 00:00:00 2001 From: Cloud Wu Date: Wed, 18 Apr 2018 19:29:41 +0800 Subject: [PATCH 5/5] forward fd to agent --- service-src/service_gate.c | 11 ++++++++--- 1 file changed, 8 insertions(+), 3 deletions(-) diff --git a/service-src/service_gate.c b/service-src/service_gate.c index b8da7441..baeb1754 100644 --- a/service-src/service_gate.c +++ b/service-src/service_gate.c @@ -168,21 +168,26 @@ _report(struct gate * g, const char * data, ...) { static void _forward(struct gate *g, struct connection * c, int size) { struct skynet_context * ctx = g->ctx; + int fd = c->id; + if (fd <= 0) { + // socket error + return; + } if (g->broker) { void * temp = skynet_malloc(size); databuffer_read(&c->buffer,&g->mp,temp, size); - skynet_send(ctx, 0, g->broker, g->client_tag | PTYPE_TAG_DONTCOPY, 1, temp, size); + skynet_send(ctx, 0, g->broker, g->client_tag | PTYPE_TAG_DONTCOPY, fd, temp, size); return; } if (c->agent) { void * temp = skynet_malloc(size); databuffer_read(&c->buffer,&g->mp,temp, size); - skynet_send(ctx, c->client, c->agent, g->client_tag | PTYPE_TAG_DONTCOPY, 1 , temp, size); + skynet_send(ctx, c->client, c->agent, g->client_tag | PTYPE_TAG_DONTCOPY, fd , temp, size); } else if (g->watchdog) { char * tmp = skynet_malloc(size + 32); int n = snprintf(tmp,32,"%d data ",c->id); databuffer_read(&c->buffer,&g->mp,tmp+n,size); - skynet_send(ctx, 0, g->watchdog, PTYPE_TEXT | PTYPE_TAG_DONTCOPY, 1, tmp, size + n); + skynet_send(ctx, 0, g->watchdog, PTYPE_TEXT | PTYPE_TAG_DONTCOPY, fd, tmp, size + n); } }