From b33d4387155502338b3d2914427d84667592f2eb Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E4=BA=91=E9=A3=8E?= Date: Mon, 6 Aug 2012 17:04:06 +0800 Subject: [PATCH] simple rpc support --- agent.lua | 4 +- console.lua | 24 ++++++------ globallog.lua | 4 +- lua-skynet.c | 55 ++++++++++++++++++-------- main.lua | 8 ++-- main_log.lua | 7 +++- service_lua.c | 16 +++++++- simpledb.lua | 22 +++++++++++ skynet.h | 3 +- skynet.lua | 100 ++++++++++++++++++++++++++++++++++++++++++++++++ skynet_harbor.c | 12 ++---- skynet_server.c | 32 ++++++++-------- skynet_server.h | 1 + skynet_timer.c | 16 +++++++- skynet_timer.h | 2 +- testdb.lua | 12 ++++++ testtimer.lua | 15 +++++--- watchdog.lua | 6 +-- 18 files changed, 266 insertions(+), 73 deletions(-) create mode 100644 simpledb.lua create mode 100644 skynet.lua create mode 100644 testdb.lua diff --git a/agent.lua b/agent.lua index 02f625d9..386a06ef 100644 --- a/agent.lua +++ b/agent.lua @@ -2,6 +2,6 @@ local skynet = require "skynet" print("agent",...) -skynet.callback(function(session, addr, msg) - print("[agent]",session,addr,msg) +skynet.dispatch(function(msg , addr) + print("[agent]",addr,msg) end) diff --git a/console.lua b/console.lua index 15d26aea..a9d1fea1 100644 --- a/console.lua +++ b/console.lua @@ -1,14 +1,16 @@ local skynet = require "skynet" -skynet.callback(function() - local cmd = io.read() - local handle = skynet.command("LAUNCH",cmd) - if handle == nil then - print("Launch error:",cmd) - else - print("Lauch:",handle) - end - skynet.command("TIMEOUT", 0, "0") -end) +skynet.dispatch() -skynet.command("TIMEOUT", 0 ,"0") +skynet.start(function() + while true do + local cmd = io.read() + local handle = skynet.launch(cmd) + if handle == nil then + print("Launch error:",cmd) + else + print("Lauch:",handle) + end + skynet.yield() + end +end) diff --git a/globallog.lua b/globallog.lua index e023faee..e2f4bfbb 100644 --- a/globallog.lua +++ b/globallog.lua @@ -1,7 +1,7 @@ local skynet = require "skynet" -skynet.callback(function(session, from , message) +skynet.dispatch(function(message, from, session) print("[GLOBALLOG]",session, from,message) end) -skynet.command("REG","LOG") \ No newline at end of file +skynet.register "LOG" \ No newline at end of file diff --git a/lua-skynet.c b/lua-skynet.c index 0bbc99ca..404d6d67 100644 --- a/lua-skynet.c +++ b/lua-skynet.c @@ -5,19 +5,32 @@ #include #include +static int +traceback (lua_State *L) { + const char *msg = lua_tostring(L, 1); + if (msg) + luaL_traceback(L, L, msg, 1); + else if (!lua_isnoneornil(L, 1)) { /* is there an error object? */ + if (!luaL_callmeta(L, 1, "__tostring")) /* try its 'tostring' metamethod */ + lua_pushliteral(L, "(no error message)"); + } + return 1; +} + static void _cb(struct skynet_context * context, void * ud, int session, const char * addr, const void * msg, size_t sz) { lua_State *L = ud; + lua_pushcfunction(L, traceback); lua_rawgetp(L, LUA_REGISTRYINDEX, _cb); int r; if (msg == NULL) { lua_pushinteger(L, session); - r = lua_pcall(L, 1, 0 , 0); + r = lua_pcall(L, 1, 0 , -3); } else { lua_pushinteger(L, session); lua_pushstring(L, addr); lua_pushlstring(L, msg, sz); - r = lua_pcall(L, 3, 0 , 0); + r = lua_pcall(L, 3, 0 , -5); } if (r == LUA_OK) return; @@ -82,22 +95,32 @@ _send(lua_State *L) { session = lua_tointeger(L,2); ++index; } - int type = lua_type(L,index+2); - if (type == LUA_TSTRING) { - size_t len = 0; - void * msg = (void *)lua_tolstring(L,index+2,&len); - void * message = malloc(len); - memcpy(message, msg, len); - skynet_send(context, dest, session , message, len); + if (lua_gettop(L) == index + 1) { + session = skynet_send(context, dest, session , NULL, 0); } else { - void * msg = lua_touserdata(L,index+2); - if (msg == NULL) { - return luaL_error(L, "skynet.send need userdata or string (%s)", lua_typename(L,type)); + int type = lua_type(L,index+2); + if (type == LUA_TSTRING) { + size_t len = 0; + void * msg = (void *)lua_tolstring(L,index+2,&len); + void * message = malloc(len); + memcpy(message, msg, len); + session = skynet_send(context, dest, session , message, len); + } else if (type == LUA_TNIL) { + session = skynet_send(context, dest, session , NULL, 0); + } else { + void * msg = lua_touserdata(L,index+2); + if (msg == NULL) { + return luaL_error(L, "skynet.send need userdata or string (%s)", lua_typename(L,type)); + } + int size = luaL_checkinteger(L,index+3); + session = skynet_send(context, dest, session, msg, size); } - int size = luaL_checkinteger(L,index+3); - skynet_send(context, dest, session, msg, size); } - return 0; + if (session < 0) { + return luaL_error(L, "skynet.send drop the message to %s", dest); + } + lua_pushinteger(L,session); + return 1; } static int @@ -108,7 +131,7 @@ _error(lua_State *L) { } int -luaopen_skynet(lua_State *L) { +luaopen_skynet_c(lua_State *L) { luaL_checkversion(L); luaL_Reg l[] = { { "send" , _send }, diff --git a/main.lua b/main.lua index 0fa03f15..5e026338 100644 --- a/main.lua +++ b/main.lua @@ -2,11 +2,11 @@ local skynet = require "skynet" print("Server start") -local console = skynet.command("LAUNCH","snlua console.lua") +local console = skynet.launch("snlua","console.lua") print("console",console) -local watchdog = skynet.command("LAUNCH","snlua watchdog.lua") +local watchdog = skynet.launch("snlua","watchdog.lua") print("watchdog",watchdog) -local gate = skynet.command("LAUNCH","gate 8888 4 0") +local gate = skynet.launch("gate","8888 4 0") print("gate",gate) -skynet.command("EXIT") +skynet.exit() diff --git a/main_log.lua b/main_log.lua index e16754d2..55e987d3 100644 --- a/main_log.lua +++ b/main_log.lua @@ -2,7 +2,10 @@ local skynet = require "skynet" print("Log server start") -local log = skynet.command("LAUNCH","snlua globallog.lua") +local log = skynet.launch("snlua","globallog.lua") print("log",log) -skynet.command("EXIT") +local db = skynet.launch("snlua","simpledb.lua") +print("simpledb",db) + +skynet.exit() diff --git a/service_lua.c b/service_lua.c index 6423f7b5..6858bc6d 100644 --- a/service_lua.c +++ b/service_lua.c @@ -19,6 +19,18 @@ _load(lua_State *L, char ** filename) { return r != LUA_OK; } +static int +traceback (lua_State *L) { + const char *msg = lua_tostring(L, 1); + if (msg) + luaL_traceback(L, L, msg, 1); + else if (!lua_isnoneornil(L, 1)) { /* is there an error object? */ + if (!luaL_callmeta(L, 1, "__tostring")) /* try its 'tostring' metamethod */ + lua_pushliteral(L, "(no error message)"); + } + return 1; +} + int snlua_init(lua_State *L, struct skynet_context *ctx, const char * args) { lua_gc(L, LUA_GCSTOP, 0); @@ -31,6 +43,8 @@ snlua_init(lua_State *L, struct skynet_context *ctx, const char * args) { char *parm = tmp; strcpy(parm,args); + lua_pushcfunction(L, traceback); + const char * filename = parm; int r = _load(L, &parm); if (r) { @@ -45,7 +59,7 @@ snlua_init(lua_State *L, struct skynet_context *ctx, const char * args) { ++n; } } - r = lua_pcall(L,n,0,0); + r = lua_pcall(L,n,0,-n-2); switch (r) { case LUA_OK: return 0; diff --git a/simpledb.lua b/simpledb.lua new file mode 100644 index 00000000..0ca9fb39 --- /dev/null +++ b/simpledb.lua @@ -0,0 +1,22 @@ +local skynet = require "skynet" +local db = {} + +local command = {} + +function command.GET(key) + skynet.ret(db[key]) +end + +function command.SET(key, value) + local last = db[key] + db[key] = value + skynet.ret(last) +end + +skynet.dispatch(function(message, from, session) + print("simpledb",message, from, session) + local cmd, key , value = string.match(message, "(%w+) (%w+) ?(.*)") + command[cmd](key,value) +end) + +skynet.register "SIMPLEDB" diff --git a/skynet.h b/skynet.h index 0d9a8591..a6bd72ad 100644 --- a/skynet.h +++ b/skynet.h @@ -2,12 +2,13 @@ #define SKYNET_H #include +#include struct skynet_context; void skynet_error(struct skynet_context * context, const char *msg, ...); const char * skynet_command(struct skynet_context * context, const char * cmd , int session, const char * parm); -void skynet_send(struct skynet_context * context, const char * addr , int session, void * msg, size_t sz); +int skynet_send(struct skynet_context * context, const char * addr , int session, void * msg, size_t sz); typedef void (*skynet_cb)(struct skynet_context * context, void *ud, int session, const char * addr , const void * msg, size_t sz); void skynet_callback(struct skynet_context * context, void *ud, skynet_cb cb); diff --git a/skynet.lua b/skynet.lua new file mode 100644 index 00000000..e2d4d838 --- /dev/null +++ b/skynet.lua @@ -0,0 +1,100 @@ +local c = require "skynet.c" +local tostring = tostring +local tonumber = tonumber +local coroutine = coroutine +local assert = assert + +local skynet = {} +local session_id_coroutine = {} +local session_coroutine_id = {} +local session_coroutine_address = {} + +local function suspend(co, result, command, param) + assert(result, command) + if command == "CALL" or command == "SLEEP" then + session_id_coroutine[param] = co + elseif command == "RETURN" then + local co_session = session_coroutine_id[co] + local co_address = session_coroutine_address[co] + c.send(co_address, co_session, param) + else + assert(command == nil, command) + session_coroutine_id[co] = nil + session_coroutine_address[co] = nil + end +end + +function skynet.timeout(ti, func, ...) + local session = c.command("TIMEOUT",tostring(ti)) + assert(session) + session = tonumber(session) + local co = coroutine.create(func) + assert(session_id_coroutine[session] == nil) + session_id_coroutine[session] = co + suspend(co, coroutine.resume(co, ...)) +end + +function skynet.yield() + local session = c.command("TIMEOUT","0") + coroutine.yield("SLEEP", tonumber(session)) +end + +function skynet.sleep(ti) + local session = c.command("TIMEOUT",tostring(ti)) + assert(session) + coroutine.yield("SLEEP", tonumber(session)) +end + +function skynet.register(name) + return c.command("REG", name) +end + +function skynet.self() + return c.command("REG") +end + +function skynet.launch(...) + return c.command("LAUNCH", table.concat({...}," ")) +end + +function skynet.now() + return tonumber(c.command("NOW")) +end + +function skynet.exit() + c.command("EXIT") +end + +skynet.send = c.send + +function skynet.call(addr, message) + local session = c.send(addr, -1, message) + return coroutine.yield("CALL", session) +end + +function skynet.ret(message) + coroutine.yield("RETURN", message) +end + +function skynet.dispatch(f) + c.callback(function(session, address , message) + local co = session_id_coroutine[session] + if co == nil then + co = coroutine.create(f) + session_coroutine_id[co] = session + session_coroutine_address[co] = address + suspend(co, coroutine.resume(co, message, address, session)) + else + session_id_coroutine[session] = nil + suspend(co, coroutine.resume(co, message)) + end + end) +end + +function skynet.start(f) + c.command("TIMEOUT",0,"0") + local co = coroutine.create(f) + session_id_coroutine[0] = co +end + +return skynet diff --git a/skynet_harbor.c b/skynet_harbor.c index dd35c6f2..f5fe2a1d 100644 --- a/skynet_harbor.c +++ b/skynet_harbor.c @@ -162,17 +162,11 @@ skynet_harbor_send(const char *name, uint32_t destination, struct skynet_message assert(destination!=0); int remote_id = destination >> HANDLE_REMOTE_SHIFT; assert(remote_id > 0 && remote_id <= REMOTE_MAX); - --remote_id; - struct remote * r = &Z->remote[remote_id]; struct skynet_remote_message message; message.destination = destination; message.message = *msg; - if (r->socket) { - skynet_remotemq_push(Z->queue, &message); - send_notice(); - } else { - skynet_remotemq_push(r->queue, &message); - } + skynet_remotemq_push(Z->queue, &message); + send_notice(); } else { _lock(); struct keyvalue * node = _hash_search(Z->map, name); @@ -454,6 +448,7 @@ remote_socket_send(void * socket, struct skynet_remote_message *msg) { struct remote_header rh; rh.source = msg->message.source; rh.destination = msg->destination; + rh.session = msg->message.session; zmq_msg_t part; zmq_msg_init_size(&part,sizeof(struct remote_header)); uint8_t * buffer = zmq_msg_data(&part); @@ -500,6 +495,7 @@ _remote_recv() { _report_zmq_error(rc); struct skynet_message msg; + msg.session = (int)rh.session; msg.source = rh.source; msg.data = data; msg.sz = zmq_msg_size(data); diff --git a/skynet_server.c b/skynet_server.c index c91d7712..ff112e68 100644 --- a/skynet_server.c +++ b/skynet_server.c @@ -74,8 +74,8 @@ skynet_context_new(const char * name, const char *parm) { } } -static int -_new_session(struct skynet_context *ctx) { +int +skynet_context_newsession(struct skynet_context *ctx) { int session = ++ctx->session_id; if (session < 0) { ctx->session_id = 1; @@ -151,7 +151,7 @@ skynet_context_message_dispatch(void) { struct skynet_context * ctx = skynet_handle_grab(handle); if (ctx == NULL) { - skynet_error(NULL, "Drop message queue %u ", handle); + skynet_error(NULL, "Drop message queue %x ", handle); _drop_queue(q); return 0; } @@ -177,7 +177,7 @@ skynet_context_message_dispatch(void) { skynet_harbor_message_close(&msg); } free(msg.data); - skynet_error(NULL, "Drop message from %u to %u without callback , size = %d",msg.source, handle, (int)msg.sz); + skynet_error(NULL, "Drop message from %x to %x without callback , size = %d",msg.source, handle, (int)msg.sz); } else { _dispatch_message(ctx, &msg); } @@ -196,8 +196,11 @@ skynet_command(struct skynet_context * context, const char * cmd , int session, if (strcmp(cmd,"TIMEOUT") == 0) { char * session_ptr = NULL; int ti = strtol(parm, &session_ptr, 10); - skynet_timeout(context->handle, ti, session); - return NULL; + session = skynet_timeout(context->handle, ti, session); + if (session < 0) + return NULL; + sprintf(context->result, "%d", session); + return context->result; } if (strcmp(cmd,"NOW") == 0) { @@ -243,10 +246,10 @@ skynet_command(struct skynet_context * context, const char * cmd , int session, return NULL; } -void +int skynet_send(struct skynet_context * context, const char * addr , int session, void * msg, size_t sz) { if (session < 0) { - session = _new_session(context); + session = skynet_context_newsession(context); } uint32_t des = 0; if (addr[0] == ':') { @@ -256,7 +259,7 @@ skynet_send(struct skynet_context * context, const char * addr , int session, vo if (des == 0) { free(msg); skynet_error(context, "Drop message to %s, size = %d", addr, (int)sz); - return; + return session; } } else { struct skynet_message smsg; @@ -265,7 +268,7 @@ skynet_send(struct skynet_context * context, const char * addr , int session, vo smsg.data = msg; smsg.sz = sz; skynet_harbor_send(addr, 0, &smsg); - return; + return session; } assert(des > 0); @@ -280,9 +283,10 @@ skynet_send(struct skynet_context * context, const char * addr , int session, vo skynet_harbor_send(NULL, des, &smsg); } else if (skynet_context_push(des, &smsg)) { free(msg); - skynet_error(NULL, "Drop message from %u to %s (size=%d)", smsg.source, addr, (int)sz); - return; + skynet_error(NULL, "Drop message from %x to %s (size=%d)", smsg.source, addr, (int)sz); + return -1; } + return session; } uint32_t @@ -308,9 +312,7 @@ skynet_context_push(uint32_t handle, struct skynet_message *message) { if (ctx == NULL) { return -1; } - if (message->session < 0) { - message->session = _new_session(ctx); - } + assert(message->session >= 0); skynet_mq_push(ctx->queue, message); if (__sync_lock_test_and_set(&ctx->in_global_queue,1) == 0) { skynet_globalmq_push(ctx->queue); diff --git a/skynet_server.h b/skynet_server.h index 85da2621..95ad8a86 100644 --- a/skynet_server.h +++ b/skynet_server.h @@ -12,6 +12,7 @@ struct skynet_context * skynet_context_release(struct skynet_context *); uint32_t skynet_context_handle(struct skynet_context *); void skynet_context_init(struct skynet_context *, uint32_t handle); int skynet_context_push(uint32_t handle, struct skynet_message *message); +int skynet_context_newsession(struct skynet_context *); int skynet_context_message_dispatch(void); // return 1 when block #endif diff --git a/skynet_timer.c b/skynet_timer.c index 0050ccf0..3beadbb7 100644 --- a/skynet_timer.c +++ b/skynet_timer.c @@ -1,6 +1,7 @@ #include "skynet_timer.h" #include "skynet_mq.h" #include "skynet_server.h" +#include "skynet_handle.h" #include #include @@ -171,8 +172,17 @@ timer_create_timer() return r; } -void +int skynet_timeout(uint32_t handle, int time, int session) { + if (session < 0) { + struct skynet_context * ctx = skynet_handle_grab(handle); + if (ctx == NULL) { + return -1; + } + session = skynet_context_newsession(ctx); + skynet_context_release(ctx); + } + if (time == 0) { struct skynet_message message; message.source = SKYNET_SYSTEM_TIMER; @@ -181,7 +191,7 @@ skynet_timeout(uint32_t handle, int time, int session) { message.sz = 0; if (skynet_context_push(handle, &message)) { - return; + return -1; } } else { struct timer_event event; @@ -189,6 +199,8 @@ skynet_timeout(uint32_t handle, int time, int session) { event.session = session; timer_add(TI, &event, sizeof(event), time); } + + return session; } static uint32_t diff --git a/skynet_timer.h b/skynet_timer.h index f750e2b8..b9f085c8 100644 --- a/skynet_timer.h +++ b/skynet_timer.h @@ -5,7 +5,7 @@ #include -void skynet_timeout(uint32_t handle, int time, int session); +int skynet_timeout(uint32_t handle, int time, int session); void skynet_updatetime(void); uint32_t skynet_gettime(void); diff --git a/testdb.lua b/testdb.lua new file mode 100644 index 00000000..d561984b --- /dev/null +++ b/testdb.lua @@ -0,0 +1,12 @@ +local skynet = require "skynet" + +-- register a dummy callback function +skynet.dispatch() + +skynet.start(function() + local result = skynet.call("SIMPLEDB","SET foobar hello") + print(result) + result = skynet.call("SIMPLEDB","GET foobar") + print(result) + skynet.exit() +end) diff --git a/testtimer.lua b/testtimer.lua index 418c8c7a..0f6fc004 100644 --- a/testtimer.lua +++ b/testtimer.lua @@ -1,8 +1,13 @@ local skynet = require "skynet" -skynet.callback(function(session,addr,content) - print("sn:",session) - skynet.command("TIMEOUT", -1 , "100") -end) +-- register a dummy callback function +skynet.dispatch() -skynet.command("TIMEOUT",0,"0") \ No newline at end of file +skynet.start(function() + for i = 1, 10 do + print(i) + skynet.sleep(100) + end + skynet.exit() + print("Test timer exit") +end) diff --git a/watchdog.lua b/watchdog.lua index f93b3bee..e45e5b3e 100644 --- a/watchdog.lua +++ b/watchdog.lua @@ -6,7 +6,7 @@ function command:open(parm) local fd,addr = string.match(parm,"(%d+) ([^%s]+)") fd = tonumber(fd) skynet.send("LOG", 0, string.format("%d %d %s",self,fd,addr)) - local agent = skynet.command("LAUNCH","snlua agent.lua ".. self) + local agent = skynet.launch("snlua","agent.lua",self) if agent then skynet.send("gate",0, "forward ".. self .. " " .. agent) end @@ -20,7 +20,7 @@ function command:data(data) skynet.send("LOG",0,string.format("data %d size=%d",self,#data)) end -skynet.callback(function(session, from , message) +skynet.dispatch(function(message) local id, cmd , parm = string.match(message, "(%d+) (%w+) ?(.*)") id = tonumber(id) local f = command[cmd] @@ -31,4 +31,4 @@ skynet.callback(function(session, from , message) end end) -skynet.command("REG",".watchdog") \ No newline at end of file +skynet.register ".watchdog" \ No newline at end of file