From 978e010e67f6efed9fb33ecf033ff23d85007771 Mon Sep 17 00:00:00 2001 From: Cloud Wu Date: Wed, 13 Aug 2014 17:49:30 +0800 Subject: [PATCH] add cluster.proxy and cluster.ncall --- HISTORY.md | 2 ++ examples/cluster2.lua | 5 ++++- examples/globallog.lua | 1 + examples/simpledb.lua | 1 + lualib-src/lua-skynet.c | 15 +++++++++++++-- lualib/cluster.lua | 19 ++++++++++++++++++- lualib/skynet.lua | 18 +++++++++++++++++- service/cdummy.lua | 6 +++++- service/clusterd.lua | 10 ++++++++++ service/cslave.lua | 9 +++++++-- skynet-src/skynet_server.c | 2 +- 11 files changed, 79 insertions(+), 9 deletions(-) diff --git a/HISTORY.md b/HISTORY.md index 6bf2437c..cb29fd78 100644 --- a/HISTORY.md +++ b/HISTORY.md @@ -7,6 +7,8 @@ Dev version * Don't check imported function in snax.hotfix * snax service add change SERVICE_PATH and add it to package.path * skynet.redirect support string address +* add skynet.harbor.queryname to query globalname +* add cluster.proxy and cluster.ncall v0.5.2 (2014-8-11) ----------- diff --git a/examples/cluster2.lua b/examples/cluster2.lua index a61ff5c4..d45605f8 100644 --- a/examples/cluster2.lua +++ b/examples/cluster2.lua @@ -2,5 +2,8 @@ local skynet = require "skynet" local cluster = require "cluster" skynet.start(function() - print(cluster.call("db", "SIMPLEDB", "GET", "a")) + local proxy = cluster.proxy("db", ".simpledb") + print(skynet.call(proxy, "lua", "GET", "a")) + print(cluster.ncall("db.simpledb","GET", "a")) + print(cluster.call("db", ".simpledb", "GET", "a")) end) diff --git a/examples/globallog.lua b/examples/globallog.lua index ffd0a0f3..2939756c 100644 --- a/examples/globallog.lua +++ b/examples/globallog.lua @@ -4,5 +4,6 @@ skynet.start(function() skynet.dispatch("lua", function(session, address, ...) print("[GLOBALLOG]", skynet.address(address), ...) end) + skynet.register ".log" skynet.register "LOG" end) diff --git a/examples/simpledb.lua b/examples/simpledb.lua index 3a7649e6..e70d2f86 100644 --- a/examples/simpledb.lua +++ b/examples/simpledb.lua @@ -22,5 +22,6 @@ skynet.start(function() error(string.format("Unknown command %s", tostring(cmd))) end end) + skynet.register ".simpledb" skynet.register "SIMPLEDB" end) diff --git a/lualib-src/lua-skynet.c b/lualib-src/lua-skynet.c index 7fe3bd02..644c1b2e 100644 --- a/lualib-src/lua-skynet.c +++ b/lualib-src/lua-skynet.c @@ -73,10 +73,17 @@ _cb(struct skynet_context * context, void * ud, int type, int session, uint32_t return 0; } +static int +forward_cb(struct skynet_context * context, void * ud, int type, int session, uint32_t source, const void * msg, size_t sz) { + _cb(context, ud, type, session, source, msg, sz); + // don't delete msg in forward mode. + return 1; +} + static int _callback(lua_State *L) { struct skynet_context * context = lua_touserdata(L, lua_upvalueindex(1)); - + int forward = lua_toboolean(L, 2); luaL_checktype(L,1,LUA_TFUNCTION); lua_settop(L,1); lua_rawsetp(L, LUA_REGISTRYINDEX, _cb); @@ -84,7 +91,11 @@ _callback(lua_State *L) { lua_rawgeti(L, LUA_REGISTRYINDEX, LUA_RIDX_MAINTHREAD); lua_State *gL = lua_tothread(L,-1); - skynet_callback(context, gL, _cb); + if (forward) { + skynet_callback(context, gL, forward_cb); + } else { + skynet_callback(context, gL, _cb); + } return 0; } diff --git a/lualib/cluster.lua b/lualib/cluster.lua index d53e6d9b..ec3f6abf 100644 --- a/lualib/cluster.lua +++ b/lualib/cluster.lua @@ -4,7 +4,7 @@ local clusterd local cluster = {} function cluster.call(node, address, ...) - -- skynet.pack(...) will free by cluster.c.packrequest + -- skynet.pack(...) will free by cluster.core.packrequest return skynet.call(clusterd, "lua", "req", node, address, skynet.pack(...)) end @@ -20,6 +20,23 @@ function cluster.reload() skynet.call(clusterd, "lua", "reload") end +function cluster.proxy(node, name) + return skynet.call(clusterd, "lua", "proxy", node, name) +end + +local namecache = {} + +function cluster.ncall(name, ...) + local s = namecache[name] + if not s then + local node , lname = name:match "(.-)(%..+)" + assert(node and lname) + s = cluster.proxy(node, lname) + namecache[name] = assert(s) + end + return skynet.call(s, "lua", ...) +end + skynet.init(function() clusterd = skynet.uniqueservice("clusterd") end) diff --git a/lualib/skynet.lua b/lualib/skynet.lua index e1dcdc1f..563b0f26 100644 --- a/lualib/skynet.lua +++ b/lualib/skynet.lua @@ -99,6 +99,7 @@ local coroutine_yield = coroutine.yield local function co_create(f) local co = table.remove(coroutine_pool) if co == nil then + local print = print co = coroutine.create(function(...) f(...) while true do @@ -548,7 +549,7 @@ end function skynet.address(addr) if type(addr) == "number" then - return string.format(":%x",addr) + return string.format(":%08x",addr) else return tostring(addr) end @@ -648,6 +649,21 @@ function skynet.filter(f ,start_func) end) end +function skynet.forward_type(map, start_func) + c.callback(function(ptype, msg, sz, ...) + local prototype = map[ptype] + if prototype then + dispatch_message(prototype, msg, sz, ...) + else + dispatch_message(ptype, msg, sz, ...) + c.trash(msg, sz) + end + end, true) + skynet.timeout(0, function() + init_service(start_func) + end) +end + function skynet.endless() return c.command("ENDLESS")~=nil end diff --git a/service/cdummy.lua b/service/cdummy.lua index 107c626c..52b70d7c 100644 --- a/service/cdummy.lua +++ b/service/cdummy.lua @@ -36,7 +36,11 @@ function harbor.REGISTER(name, handle) skynet.redirect(harbor_service, handle, "harbor", 0, "N " .. name) end -function harbor.QUERYNAME(fd, name) +function harbor.QUERYNAME(name) + if name:byte() == 46 then -- "." , local name + skynet.ret(skynet.pack(skynet.localname(name))) + return + end local result = globalname[name] if result then skynet.ret(skynet.pack(result)) diff --git a/service/clusterd.lua b/service/clusterd.lua index 3032784e..23fa5b0f 100644 --- a/service/clusterd.lua +++ b/service/clusterd.lua @@ -60,6 +60,16 @@ function command.req(source, node, addr, msg, sz) skynet.ret(c:request(request, session)) end +local proxy = {} + +function command.proxy(source, node, name) + local fullname = node .. "." .. name + if proxy[fullname] == nil then + proxy[fullname] = skynet.newservice("clusterproxy", node, name) + end + skynet.ret(skynet.pack(proxy[fullname])) +end + local request_fd = {} function command.socket(source, subcmd, fd, msg) diff --git a/service/cslave.lua b/service/cslave.lua index a75e9f36..fdbe9088 100644 --- a/service/cslave.lua +++ b/service/cslave.lua @@ -93,6 +93,7 @@ local function monitor_master(master_fd) local fd = slaves[id_name] slaves[id_name] = false if fd then + monitor_clear(id_name) socket.close(fd) end end @@ -143,14 +144,14 @@ local function monitor_harbor(master_fd) return function(session, source, command) local t = string.sub(command, 1, 1) local arg = string.sub(command, 3) - if t == "Q" then + if t == 'Q' then -- query name if globalname[arg] then skynet.redirect(harbor_service, globalname[arg], "harbor", 0, "N " .. arg) else socket.write(master_fd, pack_package("Q", arg)) end - elseif t == "D" then + elseif t == 'D' then -- harbor down local id = tonumber(arg) if slaves[id] then @@ -194,6 +195,10 @@ function harbor.CONNECT(fd, id) end function harbor.QUERYNAME(fd, name) + if name:byte() == 46 then -- "." , local name + skynet.ret(skynet.pack(skynet.localname(name))) + return + end local result = globalname[name] if result then skynet.ret(skynet.pack(result)) diff --git a/skynet-src/skynet_server.c b/skynet-src/skynet_server.c index fa426ac9..eaf6e14d 100644 --- a/skynet-src/skynet_server.c +++ b/skynet-src/skynet_server.c @@ -232,7 +232,7 @@ _dispatch_message(struct skynet_context *ctx, struct skynet_message *msg) { size_t sz = msg->sz & HANDLE_MASK; if (!ctx->cb(ctx, ctx->cb_ud, type, msg->session, msg->source, msg->data, sz)) { skynet_free(msg->data); - } + } CHECKCALLING_END(ctx) }