timing remote call

Makefile

@@ -80,7 +80,7 @@ service/gate.so : service-src/service_gate.c
 service/localcast.so : service-src/service_localcast.c
 	gcc $(CFLAGS) $(SHARED) $^ -o $@ -Iskynet-src
 
-luaclib/skynet.so : lualib-src/lua-skynet.c lualib-src/lua-seri.c lualib-src/trace_service.c | luaclib
+luaclib/skynet.so : lualib-src/lua-skynet.c lualib-src/lua-seri.c lualib-src/trace_service.c lualib-src/timingqueue.c | luaclib
 	gcc $(CFLAGS) $(SHARED) -Iluacompat $^ -o $@ -Iskynet-src -Iservice-src -Ilualib-src
 
 service/client.so : service-src/service_client.c

lualib-src/lua-skynet.c

@@ -2,6 +2,7 @@
 #include "trace_service.h"
 #include "lua-seri.h"
 #include "service_lua.h"
+#include "timingqueue.h"
 
 #include "luacompat52.h"
 
@@ -26,6 +27,7 @@ struct stat {
 	uint32_t ti_sec;
 	uint32_t ti_nsec;
 	struct trace_pool *trace;
+	struct tqueue * tq;
 	struct snlua *lua;
 };
 
@@ -64,6 +66,31 @@ _stat(lua_State *L) {
 	return 0;
 }
 
+static inline double
+current_time_tick(struct stat *S) {
+	return (double)S->ti_sec + (double)S->ti_nsec / NANOSEC;
+}
+
+static struct stat *
+get_stat(lua_State *L) {
+	struct stat * S = lua_touserdata(L, lua_upvalueindex(2));
+	if (S == NULL) {
+		lua_rawgetp(L, LUA_REGISTRYINDEX, _stat);
+		S = lua_touserdata(L, -1);
+		lua_replace(L, lua_upvalueindex(2));
+	}
+
+	return S;
+}
+
+static inline void
+save_session(lua_State *L, int session) {
+	if (session > 0) {
+		struct stat * S = get_stat(L);
+		tqueue_push(S->tq, session, current_time_tick(S));
+	}
+}
+
 static int
 _cb(struct skynet_context * context, void * ud, int type, int session, uint32_t source, const void * msg, size_t sz) {
 	struct stat *S = ud;
@@ -71,6 +98,7 @@ _cb(struct skynet_context * context, void * ud, int type, int session, uint32_t
 	struct timespec ti;
 	_stat_begin(S, &ti);
 	int trace = 1;
+	int r;
 	int top = lua_gettop(L);
 	if (top == 1) {
 		lua_rawgetp(L, LUA_REGISTRYINDEX, _cb);
@@ -85,7 +113,18 @@ _cb(struct skynet_context * context, void * ud, int type, int session, uint32_t
 	lua_pushinteger(L, session);
 	lua_pushnumber(L, source);
 
-	int r = lua_pcall(L, 5, 0 , trace);
+	if (type == PTYPE_RESPONSE && session > 0) {
+		double t = tqueue_pop(S->tq, session);
+		if (t != 0) {
+			t = current_time_tick(S) - t;
+			lua_pushnumber(L, t);
+			r = lua_pcall(L, 6, 0 , trace);
+		} else {
+			r = lua_pcall(L, 5, 0 , trace);
+		}
+	} else {
+		r = lua_pcall(L, 5, 0 , trace);
+	}
 
 	_stat_end(S, &ti);
 
@@ -136,10 +175,24 @@ _cb(struct skynet_context * context, void * ud, int type, int session, uint32_t
 	return 0;
 }
 
+static int
+_timing(lua_State *L) {
+	int session = luaL_checkinteger(L,1);
+	struct stat * S = get_stat(L);
+	double t = tqueue_pop(S->tq, session);
+	if (t != 0) {
+		t = current_time_tick(S) - t;
+	}
+	lua_pushnumber(L, t);
+
+	return 1;
+}
+
 static int
 _delete_stat(lua_State *L) {
 	struct stat * S = lua_touserdata(L,1);
 	trace_release(S->trace);
+	tqueue_delete(S->tq);
 	return 0;
 }
 
@@ -162,6 +215,7 @@ _callback(lua_State *L) {
 	memset(S, 0, sizeof(*S));
 	S->L = gL;
 	S->trace = trace_create();
+	S->tq = tqueue_new();
 	S->lua = lua;
 
 	lua_createtable(L,0,1);
@@ -220,16 +274,19 @@ _sendname(lua_State *L, struct skynet_context * context, const char * dest) {
 		size_t len = 0;
 		void * msg = (void *)lua_tolstring(L,4,&len);
 		session = skynet_sendname(context, dest, type, session , msg, len);
+		save_session(L, session);
 		break;
 	}
 	case LUA_TNIL :
 		session = skynet_sendname(context, dest, type, session , NULL, 0);
+		save_session(L, session);
 		break;
 	case LUA_TLIGHTUSERDATA: {
 		luaL_checktype(L, 4, LUA_TLIGHTUSERDATA);
 		void * msg = lua_touserdata(L,4);
 		int size = luaL_checkinteger(L,5);
 		session = skynet_sendname(context, dest, type | PTYPE_TAG_DONTCOPY, session, msg, size);
+		save_session(L, session);
 		break;
 	}
 	default:
@@ -295,12 +352,14 @@ _send(lua_State *L) {
 			msg = NULL;
 		}
 		session = skynet_send(context, 0, dest, type, session , msg, len);
+		save_session(L, session);
 		break;
 	}
 	case LUA_TLIGHTUSERDATA: {
 		void * msg = lua_touserdata(L,4);
 		int size = luaL_checkinteger(L,5);
 		session = skynet_send(context, 0, dest, type | PTYPE_TAG_DONTCOPY, session, msg, size);
+		save_session(L, session);
 		break;
 	}
 	default:
@@ -332,12 +391,14 @@ _redirect(lua_State *L) {
 			msg = NULL;
 		}
 		session = skynet_send(context, source, dest, type, session , msg, len);
+		save_session(L, session);
 		break;
 	}
 	case LUA_TLIGHTUSERDATA: {
 		void * msg = lua_touserdata(L,5);
 		int size = luaL_checkinteger(L,6);
 		session = skynet_send(context, source, dest, type | PTYPE_TAG_DONTCOPY, session, msg, size);
+		save_session(L, session);
 		break;
 	}
 	default:
@@ -461,6 +522,7 @@ luaopen_skynet_c(lua_State *L) {
 	luaL_Reg l[] = {
 		{ "send" , _send },
 		{ "genid", _genid },
+		{ "timing", _timing },
 		{ "redirect", _redirect },
 		{ "forward", _forward },
 		{ "command" , _command },
@@ -501,7 +563,8 @@ luaopen_skynet_c(lua_State *L) {
 	lua_setfield(L, -2, "reload");
 
 	lua_pushlightuserdata(L, lua->ctx);
-	luaL_setfuncs(L,l,1);
+	lua_pushnil(L);
+	luaL_setfuncs(L,l,2);
 
 	luaL_setfuncs(L,l2,0);
 

lualib-src/timingqueue.c

@@ -0,0 +1,197 @@
+#include "timingqueue.h"
+
+#include <stdlib.h>
+#include <assert.h>
+#include <stdio.h>
+
+#define INIT_CAP 8
+
+struct session_time {
+	int session;
+	double time;
+};
+
+struct tqueue {
+	int cap;
+	int n;
+	int head;
+	int tail;
+	struct session_time * q;
+};
+
+struct tqueue *
+tqueue_new() {
+	struct tqueue * tq = malloc(sizeof(*tq));
+	tq->cap = INIT_CAP;
+	tq->head = 0;
+	tq->tail = 0;
+	tq->n = 0;
+	tq->q = malloc(sizeof(struct session_time) * tq->cap);
+	return tq;
+}
+
+void
+tqueue_delete(struct tqueue *tq) {
+	free(tq->q);
+	free(tq);
+}
+
+static struct session_time *
+expand_queue(struct tqueue *tq) {
+	if (tq->n == tq->cap) {
+		struct session_time * q = malloc(sizeof(*q) * tq->cap * 2);
+		int i;
+		for (i=0;i<tq->cap;i++) {
+			int index = tq->head + i;
+			if (index >= tq->cap) {
+				index -= tq->cap;
+			}
+			q[i] = tq->q[index];
+		}
+		tq->head = 0;
+		tq->tail = tq->cap;
+		tq->cap *= 2;
+		free(tq->q);
+		tq->q = q;
+		return &q[tq->tail-1];
+	} else {
+		int i;
+		for (i=0;i<tq->cap-1;i++) {
+			int index = tq->head + i;
+			if (index >= tq->cap) {
+				index -= tq->cap;
+			}
+			if (tq->q[index].session == 0)
+				break;
+		}
+		int p = i + tq->head;
+		if (p >= tq->cap)
+			p -= tq->cap;
+		for (++i;i<tq->cap-1;i++) {
+			int index = tq->head + i;
+			if (index >= tq->cap) {
+				index -= tq->cap;
+			}
+			if (tq->q[index].session == 0) {
+				continue;
+			}
+			tq->q[p] = tq->q[index];
+			++ p;
+			if (p >= tq->cap) {
+				p -= tq->cap;
+			}
+		}
+		tq->tail = p + 1;
+		if (tq->tail >= tq->cap) {
+			tq->tail -= tq->cap;
+		}
+		return &tq->q[p];
+	}
+}
+
+static inline struct session_time *
+last_one(struct tqueue * tq, struct session_time *current) {
+	--current;
+	if (current < tq->q) {
+		return &tq->q[tq->cap-1];
+	}
+	return current;
+}
+
+void
+tqueue_push(struct tqueue *tq, int session, double time) {
+	assert(session !=0);
+	++ tq->n;
+	if (tq->head == tq->tail) {
+		// queue is empty;
+		tq->head = 0;
+		tq->tail = 1;
+		tq->q[0].session = session;
+		tq->q[0].time = time;
+		return;
+	}
+	struct session_time * st = &tq->q[tq->tail++];
+	if (tq->tail >= tq->cap) {
+		tq->tail = 0;
+	}
+	if (tq->tail == tq->head) {
+		st = expand_queue(tq);
+	}
+	st->session = session;
+	st->time = time;
+
+	// session must great than last one
+	int i;
+	for (i=1;i<tq->n;i++) {
+		struct session_time *last = last_one(tq, st);
+		if (session > last->session)
+			return;
+		// swap st, last
+		struct session_time temp = *last;
+		*last = *st;
+		*st = temp;
+
+		st = last;
+	}
+}
+
+double
+tqueue_pop(struct tqueue *tq, int session) {
+	if (tq->head == tq->tail) {
+		return 0;
+	}
+	if (session == tq->q[tq->head].session) {
+		--tq->n;
+		double ret = tq->q[tq->head].time;
+		do {
+			++tq->head;
+			if (tq->head == tq->cap) {
+				tq->head = 0;
+			}
+		} while (tq->head != tq->tail && tq->q[tq->head].session == 0);
+		return ret;
+	}
+	// binary search session
+	int n = tq->tail - tq->head;
+	if (n < 0) {
+		n += tq->cap;
+	}
+	int begin = 1;
+	int end = n;
+	while (begin < end) {
+		int mid = (begin + end)/2;
+		int index = mid + tq->head;
+		if (index >= tq->cap) {
+			index -= tq->cap;
+		}
+		struct session_time *st = &tq->q[index];
+		if (st->session == session) {
+			--tq->n;
+			st->session = 0;
+			return st->time;
+		}
+		if (session > st->session) {
+			begin = mid + 1;
+		} else {
+			end = mid;
+		}
+	}
+	// not found
+	return 0;
+}
+
+void
+tqueue_dump(struct tqueue *tq) {
+	printf("cap = %d, head = %d, tail = %d, n = %d\n",
+		tq->cap, tq->head, tq->tail, tq->n);
+	int head = tq->head;
+	while (head!=tq->tail) {
+		struct session_time *st = &tq->q[head];
+		if (st->session != 0) {
+			printf("%d : %f\n", st->session, st->time);
+		}
+		++ head;
+		if (head >= tq->cap) 
+			head -= tq->cap;
+	}
+}

lualib-src/timingqueue.h

@@ -0,0 +1,13 @@
+#ifndef skynet_timing_queue_h
+#define skynet_timing_queue_h
+
+struct tqueue * tqueue_new();
+void tqueue_delete(struct tqueue *);
+
+void tqueue_push(struct tqueue *, int session, double time);
+double tqueue_pop(struct tqueue *, int session);
+
+// for debug
+void tqueue_dump(struct tqueue *tq);
+
+#endif

lualib/skynet.lua

@@ -46,6 +46,9 @@ local watching_service = {}
 local watching_session = {}
 local error_queue = {}
 
+-- timing remote call, turn off by default. Use skynet.timing_call() to turn on.
+local timing_call = nil
+
 -- suspend is function
 local suspend
 
@@ -384,9 +387,23 @@ function skynet.fork(func,...)
 	table.insert(fork_queue, co)
 end
 
-local function raw_dispatch_message(prototype, msg, sz, session, source, ...)
+local function timing(session, source, ti)
+	local t = timing_call[source]
+	if t == nil then
+		t = { n = 1, ti = 0 }
+		timing_call[source] = t
+	else
+		t.n = t.n + 1
+	end
+	t.ti = t.ti + ti
+end
+
+local function raw_dispatch_message(prototype, msg, sz, session, source, ti)
 	-- skynet.PTYPE_RESPONSE = 1, read skynet.h
 	if prototype == 1 then
+		if ti and timing_call then
+			timing(session, source, ti)
+		end
 		local co = session_id_coroutine[session]
 		if co == "BREAK" then
 			session_id_coroutine[session] = nil
@@ -404,7 +421,7 @@ local function raw_dispatch_message(prototype, msg, sz, session, source, ...)
 			local co = co_create(f)
 			session_coroutine_id[co] = session
 			session_coroutine_address[co] = source
-			suspend(co, coroutine.resume(co, session,source, p.unpack(msg,sz, ...)))
+			suspend(co, coroutine.resume(co, session,source, p.unpack(msg,sz)))
 		else
 			print("Unknown request :" , p.unpack(msg,sz))
 			error(string.format("Can't dispatch type %s : ", p.name))
@@ -538,6 +555,17 @@ function dbgcmd.INFO()
 	end
 end
 
+function dbgcmd.TIMING()
+	if timing_call then
+		skynet.ret(skynet.pack(timing_call))
+		-- turn off timing
+		timing_call = nil
+	else
+		-- turn on timing
+		timing_call = {}
+	end
+end
+
 function dbgcmd.RELOAD(...)
 	local cmd = table.concat({...}, " ")
 	c.reload(cmd)
@@ -697,4 +725,12 @@ function skynet.mqlen()
 	return tonumber(c.command "MQLEN")
 end
 
+function skynet.timing_call()
+	timing_call = {}
+end
+
+function skynet.timing_session(session)
+	return c.timing(session)
+end
+
 return skynet

service-src/service_lua.h

@@ -5,6 +5,7 @@ struct snlua {
 	lua_State * L;
 	const char * reload;
 	struct skynet_context * ctx;
+	struct tqueue * tq;
 	int (*init)(struct snlua *l, struct skynet_context *ctx, const char * args);
 };
 

service/launcher.lua

@@ -48,6 +48,22 @@ function command.INFO(handle)
 	end
 end
 
+function command.TIMING(handle)
+	handle = handle_to_address(handle)
+	if services[handle] == nil then
+		skynet.ret(skynet.pack(nil))
+	else
+		local r = skynet.call(handle,"debug","TIMING")
+		local result = {}
+		for k,v in pairs(r) do
+			local key = services[k] or string.format("[:%08x]", k)
+			v.avg = v.ti/v.n
+			result[key] = v
+		end
+		skynet.ret(skynet.pack(result))
+	end
+end
+
 function command.KILL(handle)
 	handle = handle_to_address(handle)
 	skynet.kill(handle)