From c1e6737711f696e3980dee05a9a47625d74287e8 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E4=BA=91=E9=A3=8E?= Date: Sun, 23 Feb 2014 17:13:58 +0800 Subject: [PATCH] timing remote call --- Makefile | 2 +- lualib-src/lua-skynet.c | 67 ++++++++++++- lualib-src/timingqueue.c | 197 ++++++++++++++++++++++++++++++++++++++ lualib-src/timingqueue.h | 13 +++ lualib/skynet.lua | 40 +++++++- service-src/service_lua.h | 1 + service/launcher.lua | 16 ++++ 7 files changed, 331 insertions(+), 5 deletions(-) create mode 100644 lualib-src/timingqueue.c create mode 100644 lualib-src/timingqueue.h diff --git a/Makefile b/Makefile index 01bf2802..018f11d5 100644 --- a/Makefile +++ b/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 diff --git a/lualib-src/lua-skynet.c b/lualib-src/lua-skynet.c index 1b086078..6688c9c7 100644 --- a/lualib-src/lua-skynet.c +++ b/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); diff --git a/lualib-src/timingqueue.c b/lualib-src/timingqueue.c new file mode 100644 index 00000000..766d7cf5 --- /dev/null +++ b/lualib-src/timingqueue.c @@ -0,0 +1,197 @@ +#include "timingqueue.h" + +#include +#include +#include + +#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;icap;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;icap-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;icap-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;in;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; + } +} diff --git a/lualib-src/timingqueue.h b/lualib-src/timingqueue.h new file mode 100644 index 00000000..98d34e27 --- /dev/null +++ b/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 diff --git a/lualib/skynet.lua b/lualib/skynet.lua index 72688aa3..953d04bb 100644 --- a/lualib/skynet.lua +++ b/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 diff --git a/service-src/service_lua.h b/service-src/service_lua.h index 7c9b8dde..d864ad3e 100644 --- a/service-src/service_lua.h +++ b/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); }; diff --git a/service/launcher.lua b/service/launcher.lua index 72ee768e..34acda89 100644 --- a/service/launcher.lua +++ b/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)