From 5ce65055d2eabd1827e904b9a8d0efc45d45fbac Mon Sep 17 00:00:00 2001 From: Cloud Wu Date: Tue, 29 Apr 2014 21:56:58 +0800 Subject: [PATCH] multicast support remote publish/subscribe --- examples/main.lua | 2 + lualib-src/lua-multicast.c | 71 ++++++++++++++++++++++++++---- service/bootstrap.lua | 2 +- service/datacenterd.lua | 2 +- service/multicastd.lua | 88 +++++++++++++++++++++++++++++++++++--- test/testmulticast.lua | 7 ++- 6 files changed, 153 insertions(+), 19 deletions(-) diff --git a/examples/main.lua b/examples/main.lua index 7731bdab..a9a2143b 100644 --- a/examples/main.lua +++ b/examples/main.lua @@ -14,5 +14,7 @@ skynet.start(function() maxclient = max_client, }) + skynet.newservice("testmulticast") + skynet.exit() end) diff --git a/lualib-src/lua-multicast.c b/lualib-src/lua-multicast.c index 16e469de..ff948575 100644 --- a/lualib-src/lua-multicast.c +++ b/lualib-src/lua-multicast.c @@ -3,6 +3,7 @@ #include #include #include +#include struct mc_package { int reference; @@ -10,6 +11,19 @@ struct mc_package { void *data; }; +static int +pack(lua_State *L, void *data, size_t size) { + struct mc_package * pack = skynet_malloc(sizeof(struct mc_package)); + pack->reference = 0; + pack->size = (uint32_t)size; + pack->data = data; + struct mc_package ** ret = skynet_malloc(sizeof(*ret)); + *ret = pack; + lua_pushlightuserdata(L, ret); + lua_pushinteger(L, sizeof(ret)); + return 2; +} + /* lightuserdata integer size @@ -23,15 +37,37 @@ mc_packlocal(lua_State *L) { if (size != (uint32_t)size) { return luaL_error(L, "Size should be 32bit integer"); } - struct mc_package * pack = skynet_malloc(sizeof(struct mc_package)); - pack->reference = 0; - pack->size = (uint32_t)size; - pack->data = data; - struct mc_package ** ret = skynet_malloc(sizeof(*ret)); - *ret = pack; - lua_pushlightuserdata(L, ret); - lua_pushinteger(L, sizeof(ret)); - return 2; + return pack(L, data, size); +} + +/* + lightuserdata + integer size + + return lightuserdata, sizeof(struct mc_package *) + */ +static int +mc_packremote(lua_State *L) { + void * data = lua_touserdata(L, 1); + size_t size = luaL_checkunsigned(L, 2); + if (size != (uint32_t)size) { + return luaL_error(L, "Size should be 32bit integer"); + } + void * msg = skynet_malloc(size); + memcpy(msg, data, size); + return pack(L, msg, size); +} + +static int +mc_packstring(lua_State *L) { + size_t size; + const char * msg = luaL_checklstring(L, 1, &size); + if (size != (uint32_t)size) { + return luaL_error(L, "string is too long"); + } + void * data = skynet_malloc(size); + memcpy(data, msg, size); + return pack(L, data, size); } /* @@ -89,6 +125,20 @@ mc_closelocal(lua_State *L) { return 0; } +/* + lightuserdata struct mc_package ** + return lightuserdata/size + */ +static int +mc_remote(lua_State *L) { + struct mc_package **ptr = lua_touserdata(L,1); + struct mc_package *pack = *ptr; + lua_pushlightuserdata(L, pack->data); + lua_pushunsigned(L, pack->size); + skynet_free(pack); + return 2; +} + int luaopen_multicast_c(lua_State *L) { luaL_Reg l[] = { @@ -96,6 +146,9 @@ luaopen_multicast_c(lua_State *L) { { "unpack", mc_unpacklocal }, { "bind", mc_bindrefer }, { "close", mc_closelocal }, + { "remote", mc_remote }, + { "packstring", mc_packstring }, + { "packremote", mc_packremote }, { NULL, NULL }, }; luaL_checkversion(L); diff --git a/service/bootstrap.lua b/service/bootstrap.lua index c928b90a..7924b7b2 100644 --- a/service/bootstrap.lua +++ b/service/bootstrap.lua @@ -7,8 +7,8 @@ skynet.start(function() if skynet.getenv "standalone" then local datacenter = assert(skynet.newservice "datacenterd") skynet.name("DATACENTER", datacenter) - local smgr = assert(skynet.newservice "service_mgr") end + assert(skynet.newservice "service_mgr") skynet.uniqueservice("multicastd") assert(skynet.newservice(skynet.getenv "start" or "main")) skynet.exit() diff --git a/service/datacenterd.lua b/service/datacenterd.lua index 999200fa..ad431ae2 100644 --- a/service/datacenterd.lua +++ b/service/datacenterd.lua @@ -38,6 +38,6 @@ end skynet.start(function() skynet.dispatch("lua", function (_, source, cmd, ...) local f = assert(command[cmd]) - skynet.ret(skynet.pack(f(source, ...))) + skynet.ret(skynet.pack(f(...))) end) end) diff --git a/service/multicastd.lua b/service/multicastd.lua index 2afce758..aed35190 100644 --- a/service/multicastd.lua +++ b/service/multicastd.lua @@ -2,10 +2,21 @@ local skynet = require "skynet" local mc = require "multicast.c" local datacenter = require "datacenter" +local harbor_id = skynet.harbor(skynet.self()) + local command = {} local channel = {} local channel_n = {} -local channel_id = skynet.harbor(skynet.self()) +local channel_remote = {} +local channel_id = harbor_id + +local function get_address(t, id) + local v = assert(datacenter.get("multicast", id)) + t[id] = v + return v +end + +local node_address = setmetatable({}, { __index = get_address }) function command.NEW() channel[channel_id] = {} @@ -15,16 +26,68 @@ function command.NEW() return ret end -function command.PUB(source, c, pack, size) +local function remote_publish(node, channel, source, ...) + skynet.redirect(node_address[node], source, "multicast", channel, ...) +end + +local function publish(c , source, pack, size) local group = assert(channel[c]) mc.bind(pack, channel_n[c]) local msg = skynet.tostring(pack, size) for k in pairs(group) do skynet.redirect(k, source, "multicast", c , msg) end + local remote = channel_remote[c] + if remote then + local _, msg, sz = mc.unpack(pack, size) + local msg = skynet.tostring(msg,sz) + for node in pairs(remote) do + remote_publish(node, c, source, msg) + end + end +end + +skynet.register_protocol { + name = "multicast", + id = skynet.PTYPE_MULTICAST, + unpack = function(msg, sz) + return mc.packremote(msg, sz) + end, + dispatch = publish, +} + +function command.PUB(source, c, pack, size) + assert(skynet.harbor(source) == harbor_id) + local node = c % 256 + if node ~= harbor_id then + -- remote publish + remote_publish(node, c, source, mc.remote(pack)) + else + publish(c, source, pack,size) + end +end + +function command.SUBR(source, c) + local node = skynet.harbor(source) + assert(node ~= harbor_id) + local group = channel_remote[c] + if group == nil then + group = {} + channel_remote[c] = group + end + group[node] = true end function command.SUB(source, c) + local node = c % 256 + if node ~= harbor_id then + -- remote group + if channel[c] == nil then + channel[c] = {} + channel_n[c] = 0 + skynet.call(node_address[node], "lua", "SUBR", c) + end + end local group = assert(channel[c]) if not group[source] then channel_n[c] = channel_n[c] + 1 @@ -32,19 +95,30 @@ function command.SUB(source, c) end end +function command.USUBR(source, c) + local node = skynet.harbor(source) + assert(node ~= harbor_id) + local group = assert(channel_remote[c]) + group[node] = nil +end + function command.USUB(source, c) local group = assert(channel[c]) if group[source] then group[source] = nil channel_n[c] = channel_n[c] - 1 + if channel_n[c] == 0 then + local node = c % 256 + if node ~= harbor_id then + -- remote group + channel[c] = nil + channel_n[c] = nil + skynet.call(node_address[node], "lua", "USUBR", c) + end + end end end -skynet.register_protocol { - name = "multicast", - id = skynet.PTYPE_MULTICAST, -} - skynet.start(function() skynet.dispatch("lua", function(_,source, cmd, ...) local f = assert(command[cmd]) diff --git a/test/testmulticast.lua b/test/testmulticast.lua index 127da25f..805a23e1 100644 --- a/test/testmulticast.lua +++ b/test/testmulticast.lua @@ -1,5 +1,6 @@ local skynet = require "skynet" local mc = require "multicast" +local dc = require "datacenter" local mode = ... @@ -13,6 +14,7 @@ skynet.start(function() print(string.format("%s <=== %s (%d)",skynet.address(skynet.self()),skynet.address(source), channel), ...) end }) + skynet.ret(skynet.pack()) end) end) @@ -23,9 +25,12 @@ skynet.start(function() print("New channel", channel) for i=1,10 do local sub = skynet.newservice("testmulticast", "sub") - skynet.send(sub, "lua", "init", channel) + skynet.call(sub, "lua", "init", channel) end + print("set channel", channel) + dc.set("CHANNEL", channel) + print(skynet.address(skynet.self()), "===>", channel) mc.publish(channel, "Hello World") end)