cluster trace

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"))
+	skynet.fork(function()
-	print(cluster.call("db2", "@sdb", "GET", "b"))
+		skynet.trace()
-	cluster.send("db2", "@sdb", "PING", "db2:longstring" .. largevalue)
+		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()

lualib-src/lua-cluster.c

@@ -4,6 +4,7 @@
 #include <lauxlib.h>
 #include <string.h>
 #include <assert.h>
+#include <unistd.h>
 
 #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);

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
+		if param then
-		c.trace(param, "trace")
+			session_coroutine_tracetag[co] = param
-		return suspend(co, coroutine_resume(co))
+			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

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

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