From 1c2654e117a854d5b43102dcb392102e633f1108 Mon Sep 17 00:00:00 2001 From: Cloud Wu Date: Wed, 30 Apr 2014 12:17:27 +0800 Subject: [PATCH] change multicast api --- lualib-src/lua-multicast.c | 10 ++++ lualib/multicast.lua | 93 ++++++++++++++++++++++++++------------ service/multicastd.lua | 76 +++++++++++++++++++++++++++---- test/testmulticast.lua | 18 ++++---- 4 files changed, 152 insertions(+), 45 deletions(-) diff --git a/lualib-src/lua-multicast.c b/lualib-src/lua-multicast.c index ff948575..25b9d936 100644 --- a/lualib-src/lua-multicast.c +++ b/lualib-src/lua-multicast.c @@ -139,6 +139,15 @@ mc_remote(lua_State *L) { return 2; } +static int +mc_nextid(lua_State *L) { + uint32_t id = luaL_checkunsigned(L, 1); + id += 256; + lua_pushunsigned(L, id); + + return 1; +} + int luaopen_multicast_c(lua_State *L) { luaL_Reg l[] = { @@ -149,6 +158,7 @@ luaopen_multicast_c(lua_State *L) { { "remote", mc_remote }, { "packstring", mc_packstring }, { "packremote", mc_packremote }, + { "nextid", mc_nextid }, { NULL, NULL }, }; luaL_checkversion(L); diff --git a/lualib/multicast.lua b/lualib/multicast.lua index ea2617d4..1664e844 100644 --- a/lualib/multicast.lua +++ b/lualib/multicast.lua @@ -1,10 +1,24 @@ local skynet = require "skynet" -local c = require "multicast.c" +local mc = require "multicast.c" local multicastd local multicast = {} local dispatch = {} +local chan = {} +local chan_meta = { + __index = chan, + __gc = function(self) + self:unsubscribe() + local c = self.channel + self.channel = nil + dispatch[c] = nil + end, + __tostring = function (self) + return string.format("[Multicast:%x]",self.channel) + end, +} + local function default_conf(conf) conf = conf or {} conf.pack = conf.pack or skynet.pack @@ -13,48 +27,69 @@ local function default_conf(conf) return conf end -function multicast.newchannel(conf) +function multicast.new(conf) assert(multicastd, "Init first") - local channel = skynet.call(multicastd, "lua", "NEW") - dispatch[channel] = default_conf(conf) - return channel + local self = {} + conf = conf or self + self.channel = conf.channel + if self.channel == nil then + self.channel = skynet.call(multicastd, "lua", "NEW") + end + self.__pack = conf.pack or skynet.pack + self.__unpack = conf.unpack or skynet.unpack + self.__dispatch = conf.dispatch + + return setmetatable(self, chan_meta) end -function multicast.bind(channel, conf) - assert(multicastd, "Init first") - assert(not dispatch[channel]) - dispatch[channel] = default_conf(conf) +function chan:delete() + local c = assert(self.channel) + skynet.send(multicastd, "lua", "DEL", c) + self.channel = nil + self.__subscribe = nil end -function multicast.publish(channel, ...) - local conf = assert(dispatch[channel]) - skynet.call(multicastd, "lua", "PUB", channel, c.pack(conf.pack(...))) +function chan:publish(...) + local c = assert(self.channel) + skynet.call(multicastd, "lua", "PUB", c, mc.pack(self.__pack(...))) end -function multicast.subscribe(channel, conf) - assert(multicastd, "Init first") - assert(conf.dispatch) - skynet.call(multicastd, "lua", "SUB", channel) - dispatch[channel] = default_conf(conf) +function chan:subscribe() + local c = assert(self.channel) + if self.__subscribe then + -- already subscribe + return + end + skynet.call(multicastd, "lua", "SUB", c) + self.__subscribe = true + dispatch[c] = self end -function multicast.unsubscribe(channel) - assert(multicastd, "Init first") - assert(dispatch[channel]) - dispatch[channel] = nil - skynet.call(multicastd, "lua", "USUB", channel) +function chan:unsubscribe() + if not self.__subscribe then + -- already unsubscribe + return + end + local c = assert(self.channel) + skynet.send(multicastd, "lua", "USUB", c) + self.__subscribe = nil end local function dispatch_subscribe(channel, source, pack, msg, sz) - local conf = dispatch[channel] - if not conf then - c.close(pack) + local self = dispatch[channel] + if not self then + mc.close(pack) error ("Unknown channel " .. channel) end - local ok, err = pcall(conf.dispatch, channel, source, conf.unpack(msg, sz)) - c.close(pack) - assert(ok, err) + if self.__subscribe then + local ok, err = pcall(self.__dispatch, self, source, self.__unpack(msg, sz)) + mc.close(pack) + assert(ok, err) + else + -- maybe unsubscribe first, but the message is send out. drop the message unneed + mc.close(pack) + end end local function init() @@ -62,7 +97,7 @@ local function init() skynet.register_protocol { name = "multicast", id = skynet.PTYPE_MULTICAST, - unpack = c.unpack, + unpack = mc.unpack, dispatch = dispatch_subscribe, } end diff --git a/service/multicastd.lua b/service/multicastd.lua index aed35190..d82c10fe 100644 --- a/service/multicastd.lua +++ b/service/multicastd.lua @@ -9,6 +9,7 @@ local channel = {} local channel_n = {} local channel_remote = {} local channel_id = harbor_id +local NORET = {} local function get_address(t, id) local v = assert(datacenter.get("multicast", id)) @@ -18,20 +19,58 @@ end local node_address = setmetatable({}, { __index = get_address }) +-- new LOCAL channel , The low 8bit is the same with harbor_id function command.NEW() + while channel[channel_id] do + channel_id = mc.nextid(channel_id) + end channel[channel_id] = {} channel_n[channel_id] = 0 local ret = channel_id - channel_id = channel_id + 256 + channel_id = mc.nextid(channel_id) return ret end +-- MUST call by the owner node of channel, delete a remote channel +function command.DELR(source, c) + channel[c] = nil + channel_n[c] = nil + return NORET +end + +-- delete a channel, if the channel is remote, forward the command to the owner node +-- otherwise, delete the channel, and call all the remote node, DELR +function command.DEL(source, c) + local node = c % 256 + if node ~= harbor_id then + skynet.send(node_address[node], "lua", "DEL", c) + return NORET + end + local remote = channel_remote[c] + channel[c] = nil + channel_n[c] = nil + channel_remote[c] = nil + for node in pairs(remote) do + skynet.send(node_address[node], "lua", "DELR", c) + end + return NORET +end + +-- forward multicast message to a node (channel id use the session field) local function remote_publish(node, channel, source, ...) skynet.redirect(node_address[node], source, "multicast", channel, ...) end +-- publish a message, for local node, use the message pointer (call mc.bind to add the reference) +-- for remote node, call remote_publish. (call mc.unpack and skynet.tostring to convert message pointer to string) local function publish(c , source, pack, size) - local group = assert(channel[c]) + local group = channel[c] + if group == nil then + -- dead channel, delete the pack + mc.bind(pack, 1) + mc.close(pack) + return + end mc.bind(pack, channel_n[c]) local msg = skynet.tostring(pack, size) for k in pairs(group) do @@ -56,6 +95,8 @@ skynet.register_protocol { dispatch = publish, } +-- publish a message, if the caller is remote, forward the message to the owner node (by remote_publish) +-- If the caller is local, call publish function command.PUB(source, c, pack, size) assert(skynet.harbor(source) == harbor_id) local node = c % 256 @@ -67,9 +108,17 @@ function command.PUB(source, c, pack, size) end end +-- the node (source) subscribe a channel +-- MUST call by channel owner node (assert source is not local and channel is create by self) +-- If channel is not exist, return true +-- Else set channel_remote[channel] true function command.SUBR(source, c) local node = skynet.harbor(source) - assert(node ~= harbor_id) + if not channel[c] then + -- channel none exist + return true + end + assert(node ~= harbor_id and c % 256 == harbor_id) local group = channel_remote[c] if group == nil then group = {} @@ -78,30 +127,37 @@ function command.SUBR(source, c) group[node] = true end +-- the service (source) subscribe a channel +-- If the channel is remote, node subscribe it by send a SUBR to the owner . function command.SUB(source, c) local node = c % 256 if node ~= harbor_id then -- remote group if channel[c] == nil then + if skynet.call(node_address[node], "lua", "SUBR", c) then + return + end 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 + local group = channel[c] + if group and not group[source] then channel_n[c] = channel_n[c] + 1 group[source] = true end end +-- MUST call by a node, unsubscribe a channel function command.USUBR(source, c) local node = skynet.harbor(source) assert(node ~= harbor_id) local group = assert(channel_remote[c]) group[node] = nil + return NORET end +-- Unsubscribe a channel, if the subscriber is empty and the channel is remote, send USUBR to the channel owner function command.USUB(source, c) local group = assert(channel[c]) if group[source] then @@ -113,16 +169,20 @@ function command.USUB(source, c) -- remote group channel[c] = nil channel_n[c] = nil - skynet.call(node_address[node], "lua", "USUBR", c) + skynet.send(node_address[node], "lua", "USUBR", c) end end end + return NORET end skynet.start(function() skynet.dispatch("lua", function(_,source, cmd, ...) local f = assert(command[cmd]) - skynet.ret(skynet.pack(f(source, ...))) + local result = f(source, ...) + if result ~= NORET then + skynet.ret(skynet.pack(result)) + end end) local self = skynet.self() local id = skynet.harbor(self) diff --git a/test/testmulticast.lua b/test/testmulticast.lua index 805a23e1..88f79ccb 100644 --- a/test/testmulticast.lua +++ b/test/testmulticast.lua @@ -9,11 +9,14 @@ if mode == "sub" then skynet.start(function() skynet.dispatch("lua", function (_,_, cmd, channel) assert(cmd == "init") - mc.subscribe(channel, { + local c = mc.new { + channel = channel , dispatch = function (channel, source, ...) - print(string.format("%s <=== %s (%d)",skynet.address(skynet.self()),skynet.address(source), channel), ...) + print(string.format("%s <=== %s %s",skynet.address(skynet.self()),skynet.address(source), channel), ...) end - }) + } + print(skynet.address(skynet.self()), "sub", c) + c:subscribe() skynet.ret(skynet.pack()) end) end) @@ -21,18 +24,17 @@ end) else skynet.start(function() - local channel = mc.newchannel() + local channel = mc.new() print("New channel", channel) for i=1,10 do local sub = skynet.newservice("testmulticast", "sub") - skynet.call(sub, "lua", "init", channel) + skynet.call(sub, "lua", "init", channel.channel) end - print("set channel", channel) - dc.set("CHANNEL", channel) + dc.set("MCCHANNEL", channel.channel) -- for multi node test print(skynet.address(skynet.self()), "===>", channel) - mc.publish(channel, "Hello World") + channel:publish("Hello World") end) end \ No newline at end of file