From ed53a29e367b7cf1904f05f2c4c570952612580d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E4=BA=91=E9=A3=8E?= Date: Wed, 8 Aug 2012 20:46:57 +0800 Subject: [PATCH] broker & launcher --- Makefile | 5 +- console.lua | 2 +- launcher.lua | 26 +++++++++++ lua-skynet.c | 44 +++++++++++++----- main.lua | 6 ++- redis-cli.lua | 58 +++++++++++++++++++++++ service_broker.c | 100 ++++++++++++++++++++++++++++++++++++++++ service_lua.c | 20 ++++---- simpledb.lua | 2 +- skynet.h | 1 + skynet.lua | 7 ++- skynet_main.c | 2 +- skynet_server.c | 117 +++++++++++++++++++++++++++++++++++++---------- testredis.lua | 15 ++++++ 14 files changed, 353 insertions(+), 52 deletions(-) create mode 100644 launcher.lua create mode 100644 redis-cli.lua create mode 100644 service_broker.c create mode 100644 testredis.lua diff --git a/Makefile b/Makefile index d89a8b0c..38843622 100644 --- a/Makefile +++ b/Makefile @@ -1,4 +1,4 @@ -all : skynet snlua.so logger.so skynet.so gate.so client.so connection.so client skynet-master +all : skynet snlua.so logger.so skynet.so gate.so client.so connection.so broker.so client skynet-master skynet : skynet_main.c skynet_handle.c skynet_module.c skynet_mq.c skynet_server.c skynet_start.c skynet_timer.c skynet_error.c skynet_harbor.c master/master.c gcc -Wall -g -Wl,-E -o $@ $^ -Imaster -lpthread -ldl -lrt -Wl,-E -llua -lm -lzmq @@ -21,6 +21,9 @@ client.so : service_client.c connection.so : connection/connection.c connection/main.c gcc -Wall -g -fPIC --shared -o $@ $^ -I. -Iconnection +broker.so : service_broker.c + gcc -Wall -g -fPIC --shared $^ -o $@ + client : client.c gcc -Wall -g $^ -o $@ -lpthread diff --git a/console.lua b/console.lua index a9d1fea1..63846d5c 100644 --- a/console.lua +++ b/console.lua @@ -9,7 +9,7 @@ skynet.start(function() if handle == nil then print("Launch error:",cmd) else - print("Lauch:",handle) + print("Launch:",handle,cmd) end skynet.yield() end diff --git a/launcher.lua b/launcher.lua new file mode 100644 index 00000000..2faf4720 --- /dev/null +++ b/launcher.lua @@ -0,0 +1,26 @@ +local skynet = require "skynet" +local string = string + +local instance = {} + +skynet.dispatch(function(message, session, address) + if session == 0 then + -- init notice + local reply = instance[address] + if reply then + skynet.send(reply[2] , reply[1], address) + instance[address] = nil + end + else + -- launch request + local service, param = string.match(message, "([^ ]+) (.*)") + local inst = skynet.launch(service, param) + if inst then + instance[inst] = { session, address } + else + skynet.send(address, session) + end + end +end) + +skynet.register(".launcher") \ No newline at end of file diff --git a/lua-skynet.c b/lua-skynet.c index c3e8838a..2a34f045 100644 --- a/lua-skynet.c +++ b/lua-skynet.c @@ -7,14 +7,13 @@ 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; + const char *msg = lua_tostring(L, 1); + if (msg) { + luaL_traceback(L, L, msg, 1); + } else { + lua_pushliteral(L, "(no error message)"); + } + return 1; } static void @@ -25,8 +24,14 @@ _cb(struct skynet_context * context, void * ud, int session, const char * addr, lua_rawgetp(L, LUA_REGISTRYINDEX, _cb); int r; if (msg == NULL) { - lua_pushinteger(L, session); - r = lua_pcall(L, 1, 0 , trace); + if (addr == NULL) { + lua_pushinteger(L, session); + r = lua_pcall(L, 1, 0 , trace); + } else { + lua_pushinteger(L, session); + lua_pushstring(L, addr); + r = lua_pcall(L, 2, 0 , trace); + } } else { lua_pushinteger(L, session); lua_pushstring(L, addr); @@ -35,7 +40,24 @@ _cb(struct skynet_context * context, void * ud, int session, const char * addr, } if (r == LUA_OK) return; - skynet_error(context, "lua error %s", lua_tostring(L,-1)); + const char * self = skynet_command(context, "REG", NULL); + switch (r) { + case LUA_ERRRUN: + printf("Lua_ERRRUN %p\n",L); + skynet_error(context, "lua call [%s to %s : %d] error : %s", addr , self, session, lua_tostring(L,-1)); + break; + case LUA_ERRMEM: + skynet_error(context, "lua memory error : [%s to %s : %d]", addr , self, session); + break; + case LUA_ERRERR: + skynet_error(context, "lua error in error : [%s to %s : %d]", addr , self, session); + break; + case LUA_ERRGCMM: + skynet_error(context, "lua gc error : [%s to %s : %d]", addr , self, session); + break; + }; + + lua_pop(L,1); } static int diff --git a/main.lua b/main.lua index 5d633b06..43e790ce 100644 --- a/main.lua +++ b/main.lua @@ -1,7 +1,8 @@ local skynet = require "skynet" print("Server start") - +local launcher = skynet.launch("snlua","launcher.lua") +print("launcher", launcher) local console = skynet.launch("snlua","console.lua") print("console",console) local watchdog = skynet.launch("snlua","watchdog.lua") @@ -10,5 +11,6 @@ local gate = skynet.launch("gate","8888 4 0") print("gate",gate) local db = skynet.launch("snlua","simpledb.lua") print("simpledb",db) - +local connection = skynet.launch("connection","256") +print(connection) skynet.exit() diff --git a/redis-cli.lua b/redis-cli.lua new file mode 100644 index 00000000..f8b3755c --- /dev/null +++ b/redis-cli.lua @@ -0,0 +1,58 @@ +local skynet = require "skynet" +local string = string +local table = table +local tonumber = tonumber +local redis_server = ... +local fd +local write_fd +local readline_fd +local read_fd + +local function init_fd(fdstr) + fd = fdstr + write_fd = "WRITE "..fd.." " + readline_fd = "READLINE ".. fd .." \r\n" + read_fd = "READ " .. fd .. " " +end + +skynet.dispatch(function(message, ...) + skynet.send(".connection", write_fd .. message) + local result = skynet.call(".connection", readline_fd) + local firstchar = string.byte(result) + if firstchar == 42 then -- '*' + local n = tonumber(string.sub(result,2)) + if n < 1 then + skynet.ret(result .. "\r\n") + return + end + local bulk = { result } + for i = 1,n do + local line = skynet.call(".connection", readline_fd) + table.insert(result, line) + local bytes = tonumber(string.sub(line,2) + 2) + table.insert(result, bytes) + end + table.insert(result,"") + skynet.ret(table.concat(result,"\r\n")) + elseif firstchar == 36 then -- '$' + local bytes = tonumber(string.sub(result,2)) + if bytes < 0 then + skynet.ret(result .. "\r\n") + return + end + local firstline = skynet.call(".connection", read_fd .. (bytes + 2)) + skynet.ret(result .. "\r\n" .. firstline) + else + skynet.ret(result .. "\r\n") + end +end) + +skynet.start(function() + fd = skynet.call(".connection", "CONNECT " .. redis_server) + if fd == nil then + print("Connect to redis server error : ", redis_server) + skynet.exit() + return + end + init_fd(fd) +end) diff --git a/service_broker.c b/service_broker.c new file mode 100644 index 00000000..a6c8f9a1 --- /dev/null +++ b/service_broker.c @@ -0,0 +1,100 @@ +#include "skynet.h" + +#include +#include +#include +#include + +#define DEFAULT_NUMBER 16 +#define LAUNCHER ".launcher" + +struct worker { + int init; + char * name; +}; + +struct broker { + int init; + int id; + char * name; + struct worker w[DEFAULT_NUMBER]; +}; + +struct broker * +broker_create(void) { + struct broker *b = malloc(sizeof(*b)); + memset(b,0,sizeof(*b)); + return b; +} + +void +broker_release(struct broker * b) { + int i; + for (i=0;iw[i].name); + } + free(b->name); + free(b); +} + +static void +_init(struct broker *b, int session, const char * msg, size_t sz) { + assert(session > 0 && session <= DEFAULT_NUMBER); + assert(msg); + int id = session - 1; + assert(b->w[id].init == 0); + b->w[id].name = malloc(sz+1); + memcpy(b->w[id].name, msg, sz); + b->w[id].name[sz] = '\0'; + b->w[id].init = 1; + ++b->init; +} + +static void +_forward(struct broker *b, struct skynet_context * context) { + skynet_forward(context, b->w[b->id].name); + b->id = (b->id + 1) % DEFAULT_NUMBER; +} + +static void +_cb(struct skynet_context * context, void * ud, int session, const char * addr, const void * msg, size_t sz) { + struct broker * b = ud; + if (b->init < DEFAULT_NUMBER) { + _init(b, session, msg, sz); + if (b->init == DEFAULT_NUMBER) { + skynet_command(context, "REG", b->name); + skynet_send(context, LAUNCHER, 0, NULL, 0, 0); + } + } else { + _forward(b, context); + } +} + + +int +broker_init(struct broker *b, struct skynet_context *ctx, const char * args) { + char * service = strchr(args,' '); + if (service == NULL) { + return 1; + } + int len = service - args; + if (len>0) { + b->name = malloc(len +1); + memcpy(b->name, args, len); + b->name[len] = '\0'; + } + service++; + + int i; + len = strlen(service); + if (len == 0) + return 1; + for (i=0;i 0 && id <= DEFAULT_NUMBER); + } + + skynet_callback(ctx, b, _cb); + + return 0; +} diff --git a/service_lua.c b/service_lua.c index 6858bc6d..3ac9c83e 100644 --- a/service_lua.c +++ b/service_lua.c @@ -21,14 +21,13 @@ _load(lua_State *L, char ** filename) { 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; + const char *msg = lua_tostring(L, 1); + if (msg) + luaL_traceback(L, L, msg, 1); + else { + lua_pushliteral(L, "(no error message)"); + } + return 1; } int @@ -44,6 +43,7 @@ snlua_init(lua_State *L, struct skynet_context *ctx, const char * args) { strcpy(parm,args); lua_pushcfunction(L, traceback); + int traceback_index = lua_gettop(L); const char * filename = parm; int r = _load(L, &parm); @@ -59,7 +59,7 @@ snlua_init(lua_State *L, struct skynet_context *ctx, const char * args) { ++n; } } - r = lua_pcall(L,n,0,-n-2); + r = lua_pcall(L,n,0,traceback_index); switch (r) { case LUA_OK: return 0; @@ -77,6 +77,8 @@ snlua_init(lua_State *L, struct skynet_context *ctx, const char * args) { break; }; + lua_pop(L,1); + return 1; } diff --git a/simpledb.lua b/simpledb.lua index 3568990b..c82fe639 100644 --- a/simpledb.lua +++ b/simpledb.lua @@ -13,7 +13,7 @@ function command.SET(key, value) skynet.ret(last) end -skynet.dispatch(function(message, from, session) +skynet.dispatch(function(message, session, from) print("simpledb",message, from, session) local cmd, key , value = string.match(message, "(%w+) (%w+) ?(.*)") local f = command[cmd] diff --git a/skynet.h b/skynet.h index 8bf39d0b..f74c3083 100644 --- a/skynet.h +++ b/skynet.h @@ -12,6 +12,7 @@ void skynet_error(struct skynet_context * context, const char *msg, ...); const char * skynet_command(struct skynet_context * context, const char * cmd , const char * parm); int skynet_send(struct skynet_context * context, const char * addr , int session, void * msg, size_t sz, int flags); +void skynet_forward(struct skynet_context *, const char * addr); 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 index 8f18bda3..d8f79f1d 100644 --- a/skynet.lua +++ b/skynet.lua @@ -99,7 +99,12 @@ end function skynet.start(f) local session = c.command("TIMEOUT","0") - local co = coroutine.create(f) + local co = coroutine.create( + function(...) + f(...) + skynet.send(".launcher",0) + end + ) session_id_coroutine[tonumber(session)] = co end diff --git a/skynet_main.c b/skynet_main.c index da15bb7a..027685dc 100644 --- a/skynet_main.c +++ b/skynet_main.c @@ -58,7 +58,7 @@ main(int argc, char *argv[]) { lua_close(L); return 1; } - const char *path = optstring(L,"lua_path","./?.lua"); + const char *path = optstring(L,"lua_path","./?.lua;./?/init.lua"); setenv("LUA_PATH",path,1); const char *cpath = optstring(L,"lua_cpath","./?.so"); setenv("LUA_CPATH",cpath,1); diff --git a/skynet_server.c b/skynet_server.c index f1ba3899..f502817f 100644 --- a/skynet_server.c +++ b/skynet_server.c @@ -25,6 +25,9 @@ struct skynet_context { skynet_cb cb; int session_id; int in_global_queue; + int init; + uint32_t forward; + char * forward_address; struct message_queue *queue; }; @@ -39,7 +42,7 @@ _id_to_hex(char * str, uint32_t id) { } struct skynet_context * -skynet_context_new(const char * name, const char *parm) { +skynet_context_new(const char * name, const char *param) { struct skynet_module * mod = skynet_module_query(name); if (mod == NULL) @@ -54,20 +57,31 @@ skynet_context_new(const char * name, const char *parm) { ctx->ref = 2; ctx->cb = NULL; ctx->cb_ud = NULL; - ctx->in_global_queue = 0; + // a trick, global loop can't be dispatch in init process + ctx->in_global_queue = 1; + + ctx->forward = 0; + ctx->forward_address = NULL; ctx->session_id = 0; + ctx->init = 0; + ctx->handle = skynet_handle_register(ctx); char * uid = ctx->handle_name; uid[0] = ':'; _id_to_hex(uid+1, ctx->handle); - - ctx->handle = skynet_handle_register(ctx); ctx->queue = skynet_mq_create(ctx->handle); // init function maybe use ctx->handle, so it must init at last - int r = skynet_module_instance_init(mod, inst, ctx, parm); + int r = skynet_module_instance_init(mod, inst, ctx, param); if (r == 0) { - return skynet_context_release(ctx); + struct skynet_context * ret = skynet_context_release(ctx); + if (ret) { + ctx->init = 1; + skynet_globalmq_push(ctx->queue); + return ret; + } + return NULL; } else { + ctx->in_global_queue = 0; skynet_context_release(ctx); skynet_handle_retire(ctx->handle); return NULL; @@ -108,23 +122,55 @@ skynet_context_release(struct skynet_context *ctx) { return ctx; } +static int +_forwarding(struct skynet_context *ctx, struct skynet_message *msg) { + if (ctx->forward) { + uint32_t des = ctx->forward; + ctx->forward = 0; + if (skynet_harbor_message_isremote(des)) { + skynet_harbor_send(NULL, des, msg); + } else { + if (skynet_context_push(des, msg)) { + free(msg->data); + skynet_error(NULL, "Drop message from %x forward to %x (size=%d)", msg->source, des, (int)msg->sz); + return 1; + } + } + return 1; + } + if (ctx->forward_address) { + skynet_harbor_send(ctx->forward_address, 0, msg); + free(ctx->forward_address); + ctx->forward_address = NULL; + return 1; + } + return 0; +} + static void _dispatch_message(struct skynet_context *ctx, struct skynet_message *msg) { + assert(ctx->init); if (msg->source == SKYNET_SYSTEM_TIMER) { ctx->cb(ctx, ctx->cb_ud, msg->session, NULL, msg->data, msg->sz); } else { char tmp[10]; tmp[0] = ':'; + int not_delete; _id_to_hex(tmp+1, msg->source); if (skynet_harbor_message_isremote(msg->source)) { void * data = skynet_harbor_message_open(msg); ctx->cb(ctx, ctx->cb_ud, msg->session, tmp, data, msg->sz); - skynet_harbor_message_close(msg); + not_delete = _forwarding(ctx, msg); + if (!not_delete) { + skynet_harbor_message_close(msg); + } } else { ctx->cb(ctx, ctx->cb_ud, msg->session, tmp, msg->data, msg->sz); + not_delete = _forwarding(ctx, msg); + } + if (!not_delete) { + free(msg->data); } - - free(msg->data); } } @@ -197,10 +243,10 @@ skynet_context_message_dispatch(void) { } const char * -skynet_command(struct skynet_context * context, const char * cmd , const char * parm) { +skynet_command(struct skynet_context * context, const char * cmd , const char * param) { if (strcmp(cmd,"TIMEOUT") == 0) { char * session_ptr = NULL; - int ti = strtol(parm, &session_ptr, 10); + int ti = strtol(param, &session_ptr, 10); int session = skynet_context_newsession(context); if (session < 0) return NULL; @@ -216,13 +262,13 @@ skynet_command(struct skynet_context * context, const char * cmd , const char * } if (strcmp(cmd,"REG") == 0) { - if (parm == NULL || parm[0] == '\0') { + if (param == NULL || param[0] == '\0') { return context->handle_name; - } else if (parm[0] == '.') { - return skynet_handle_namehandle(context->handle, parm + 1); + } else if (param[0] == '.') { + return skynet_handle_namehandle(context->handle, param + 1); } else { assert(context->handle!=0); - skynet_harbor_register(parm, context->handle); + skynet_harbor_register(param, context->handle); return NULL; } } @@ -234,10 +280,10 @@ skynet_command(struct skynet_context * context, const char * cmd , const char * if (strcmp(cmd,"KILL") == 0) { uint32_t handle = 0; - if (parm[0] == ':') { - handle = strtoul(parm+1, NULL, 16); - } else if (parm[0] == '.') { - handle = skynet_handle_findname(parm+1); + if (param[0] == ':') { + handle = strtoul(param+1, NULL, 16); + } else if (param[0] == '.') { + handle = skynet_handle_findname(param+1); } else { // todo : kill global service } @@ -248,14 +294,15 @@ skynet_command(struct skynet_context * context, const char * cmd , const char * } if (strcmp(cmd,"LAUNCH") == 0) { - size_t sz = strlen(parm); + size_t sz = strlen(param); char tmp[sz+1]; - strcpy(tmp,parm); - char * parm = tmp; - char * mod = strsep(&parm, " \t\r\n"); - parm = strsep(&parm, "\r\n"); - struct skynet_context * inst = skynet_context_new(mod,parm); + strcpy(tmp,param); + char * args = tmp; + char * mod = strsep(&args, " \t\r\n"); + args = strsep(&args, "\r\n"); + struct skynet_context * inst = skynet_context_new(mod,args); if (inst == NULL) { + fprintf(stderr, "Launch %s %s failed\n",mod,args); return NULL; } else { context->result[0] = ':'; @@ -267,6 +314,26 @@ skynet_command(struct skynet_context * context, const char * cmd , const char * return NULL; } +void +skynet_forward(struct skynet_context * context, const char * addr) { + uint32_t des = 0; + assert(context->forward == 0 && context->forward_address == NULL); + if (addr[0] == ':') { + des = strtol(addr+1, NULL, 16); + assert(des != 0); + } else if (addr[0] == '.') { + des = skynet_handle_findname(addr + 1); + if (des == 0) { + skynet_error(context, "Drop message forward %s", addr); + return; + } + } else { + context->forward_address = strdup(addr); + return; + } + context->forward = des; +} + int skynet_send(struct skynet_context * context, const char * addr , int session, void * data, size_t sz, int flags) { char * msg; diff --git a/testredis.lua b/testredis.lua new file mode 100644 index 00000000..ab098afd --- /dev/null +++ b/testredis.lua @@ -0,0 +1,15 @@ +local skynet = require "skynet" + +skynet.dispatch() + +local command ="*2\r\n$3\r\nGET\r\n$1\r\nA\r\n" + +skynet.start(function() + local cli = skynet.call(".launcher","broker redis snlua redis-cli.lua 127.0.0.1:6379") + print("redis-cli:", cli) + assert(cli) + local result = skynet.call(cli, command) + print(result) + skynet.exit() +end) +