From 9035fbe96b73c1691a4cfba9ce69516864064651 Mon Sep 17 00:00:00 2001 From: Cloud Wu Date: Mon, 28 May 2018 16:39:04 +0800 Subject: [PATCH] cluster trace --- examples/cluster2.lua | 9 ++++++--- lualib-src/lua-cluster.c | 43 ++++++++++++++++++++++++++++++++++++++++ lualib/skynet.lua | 25 ++++++++++++++++++++--- service/clusteragent.lua | 19 ++++++++++++++++-- service/clusterd.lua | 6 ++++++ 5 files changed, 94 insertions(+), 8 deletions(-) diff --git a/examples/cluster2.lua b/examples/cluster2.lua index f50389e9..8e965cbe 100644 --- a/examples/cluster2.lua +++ b/examples/cluster2.lua @@ -10,9 +10,12 @@ skynet.start(function() 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) + skynet.fork(function() + skynet.trace() + print(cluster.call("db", "@sdb", "GET", "a")) + print(cluster.call("db2", "@sdb", "GET", "b")) + cluster.send("db2", "@sdb", "PING", "db2:longstring" .. largevalue) + end) -- test snax service skynet.timeout(300,function() diff --git a/lualib-src/lua-cluster.c b/lualib-src/lua-cluster.c index 4f1f0092..47c6592d 100644 --- a/lualib-src/lua-cluster.c +++ b/lualib-src/lua-cluster.c @@ -4,6 +4,7 @@ #include #include #include +#include #include "skynet.h" @@ -73,6 +74,11 @@ fill_header(lua_State *L, uint8_t *buf, int sz) { BYTE 2/3 ; 2:multipart, 3:multipart end DWORD SESSION PADDING msgpart(sz) + + trace + WORD stringsz + 1 + BYTE 4 + STRING tag */ static int packreq_number(lua_State *L, int session, void * msg, uint32_t sz, int is_push) { @@ -203,6 +209,21 @@ lpackpush(lua_State *L) { return packrequest(L, 1); } +static int +lpacktrace(lua_State *L) { + size_t sz; + const char * tag = luaL_checklstring(L, 1, &sz); + if (sz > 0x8000) { + return luaL_error(L, "trace tag is too long : %d", (int) sz); + } + uint8_t buf[TEMP_LENGTH]; + buf[2] = 4; + fill_header(L, buf, sz+1); + memcpy(buf+3, tag, sz); + lua_pushlstring(L, (const char *)buf, sz+3); + return 1; +} + /* string packed message return @@ -280,6 +301,12 @@ unpackmreq_part(lua_State *L, const uint8_t * buf, int sz) { return 5; } +static int +unpacktrace(lua_State *L, const char * buf, int sz) { + lua_pushlstring(L, buf + 1, sz - 1); + return 1; +} + static int unpackreq_string(lua_State *L, const uint8_t * buf, int sz) { if (sz < 2) { @@ -345,6 +372,8 @@ lunpackrequest(lua_State *L) { case 2: case 3: return unpackmreq_part(L, (const uint8_t *)msg, sz); + case 4: + return unpacktrace(L, msg, sz); case '\x80': return unpackreq_string(L, (const uint8_t *)msg, sz); case '\x81': @@ -563,17 +592,31 @@ lisname(lua_State *L) { return 0; } +static int +lnodename(lua_State *L) { + pid_t pid = getpid(); + char hostname[256]; + if (gethostname(hostname, sizeof(hostname))==0) { + lua_pushfstring(L, "%s%d", hostname, (int)pid); + } else { + lua_pushfstring(L, "noname%d", (int)pid); + } + return 1; +} + LUAMOD_API int luaopen_skynet_cluster_core(lua_State *L) { luaL_Reg l[] = { { "packrequest", lpackrequest }, { "packpush", lpackpush }, + { "packtrace", lpacktrace }, { "unpackrequest", lunpackrequest }, { "packresponse", lpackresponse }, { "unpackresponse", lunpackresponse }, { "append", lappend }, { "concat", lconcat }, { "isname", lisname }, + { "nodename", lnodename }, { NULL, NULL }, }; luaL_checkversion(L); diff --git a/lualib/skynet.lua b/lualib/skynet.lua index 6cdfe11b..e64df834 100644 --- a/lualib/skynet.lua +++ b/lualib/skynet.lua @@ -285,9 +285,13 @@ function suspend(co, result, command, param, param2) session_coroutine_id[co] = nil return suspend(co, coroutine_resume(co)) elseif command == "TRACE" then - session_coroutine_tracetag[co] = param - c.trace(param, "trace") - return suspend(co, coroutine_resume(co)) + if param then + session_coroutine_tracetag[co] = param + c.trace(param, "trace on") + else + param = session_coroutine_tracetag[co] + end + return suspend(co, coroutine_resume(co, param)) elseif command == nil then -- debug trace return @@ -351,6 +355,10 @@ function skynet.trace() coroutine_yield("TRACE", string.format(":%08x-%d",skynet.self(), traceid)) end +function skynet.tracetag() + return coroutine_yield "TRACE" +end + local starttime function skynet.starttime() @@ -448,6 +456,16 @@ function skynet.rawcall(addr, typename, msg, sz) return yield_call(addr, session) end +function skynet.tracecall(tag, addr, typename, msg, sz) + c.trace(tag, "trace begin") + c.send(addr, skynet.PTYPE_TRACE, 0, tag) + local p = proto[typename] + local session = assert(c.send(addr, p.id , nil , msg, sz), "call to invalid address") + local msg, sz = yield_call(addr, session) + c.trace(tag, "trace end") + return msg, sz +end + function skynet.ret(msg, sz) msg = msg or "" return coroutine_yield("RETURN", msg, sz) @@ -624,6 +642,7 @@ function skynet.harbor(addr) end skynet.error = c.error +skynet.tracelog = c.trace ----- register protocol do diff --git a/service/clusteragent.lua b/service/clusteragent.lua index 0a644517..04b14eab 100644 --- a/service/clusteragent.lua +++ b/service/clusteragent.lua @@ -22,16 +22,25 @@ setmetatable(register_name, { __index = end }) +local tracetag + local function dispatch_request(_,_,addr, session, msg, sz, padding, is_push) ignoreret() -- session is fd, don't call skynet.ret + if session == nil then + -- trace + tracetag = addr + return + end if padding then - local req = large_request[session] or { addr = addr , is_push = is_push } + local req = large_request[session] or { addr = addr , is_push = is_push, tracetag = tracetag } + tracetag = nil large_request[session] = req cluster.append(req, msg, sz) return else local req = large_request[session] if req then + tracetag = req.tracetag large_request[session] = nil cluster.append(req, msg, sz) msg,sz = cluster.concat(req) @@ -39,6 +48,7 @@ local function dispatch_request(_,_,addr, session, msg, sz, padding, is_push) is_push = req.is_push end if not msg then + tracetag = nil local response = cluster.packresponse(session, false, "Invalid large req") socket.write(fd, response) return @@ -65,7 +75,12 @@ local function dispatch_request(_,_,addr, session, msg, sz, padding, is_push) skynet.rawsend(addr, "lua", msg, sz) return -- no response else - ok , msg, sz = pcall(skynet.rawcall, addr, "lua", msg, sz) + if tracetag then + ok , msg, sz = pcall(skynet.tracecall, tracetag, addr, "lua", msg, sz) + tracetag = nil + else + ok , msg, sz = pcall(skynet.rawcall, addr, "lua", msg, sz) + end end else ok = false diff --git a/service/clusterd.lua b/service/clusterd.lua index ca600561..c6945ff3 100644 --- a/service/clusterd.lua +++ b/service/clusterd.lua @@ -8,6 +8,7 @@ local node_address = {} local node_session = {} local command = {} local config = {} +local nodename = cluster.nodename() local function read_response(sock) local sz = socket.header(sock:read(2)) @@ -128,6 +129,11 @@ local function send_request(source, node, addr, msg, sz) -- node_channel[node] may yield or throw error local c = node_channel[node] + local tracetag = skynet.tracetag() + if tracetag then + skynet.tracelog(tracetag, string.format("cluster %s-%s(%d)", nodename, node, session)) + c:request(cluster.packtrace(string.format("%s-%s(%d)%s", nodename, node, session, tracetag))) + end return c:request(request, session, padding) end