local multicast support

Makefile

@@ -41,7 +41,7 @@ jemalloc : $(MALLOC_STATICLIB)
 # skynet
 
 CSERVICE = snlua logger gate master harbor
-LUA_CLIB = skynet socketdriver int64 bson mongo md5 netpack cjson clientsocket memory profile
+LUA_CLIB = skynet socketdriver int64 bson mongo md5 netpack cjson clientsocket memory profile multicast
 
 SKYNET_SRC = skynet_main.c skynet_handle.c skynet_module.c skynet_mq.c \
   skynet_server.c skynet_start.c skynet_timer.c skynet_error.c \
@@ -102,6 +102,9 @@ $(LUA_CLIB_PATH)/memory.so : lualib-src/lua-memory.c | $(LUA_CLIB_PATH)
 $(LUA_CLIB_PATH)/profile.so : lualib-src/lua-profile.c | $(LUA_CLIB_PATH)
 	$(CC) $(CFLAGS) $(SHARED) $^ -o $@ 
 
+$(LUA_CLIB_PATH)/multicast.so : lualib-src/lua-multicast.c | $(LUA_CLIB_PATH)
+	$(CC) $(CFLAGS) $(SHARED) -Iskynet-src $^ -o $@ 
+
 clean :
 	rm -f $(SKYNET_BUILD_PATH)/skynet $(CSERVICE_PATH)/*.so $(LUA_CLIB_PATH)/*.so
 

lualib-src/lua-multicast.c

@@ -0,0 +1,104 @@
+#include "skynet.h"
+
+#include <lua.h>
+#include <lauxlib.h>
+#include <stdint.h>
+
+struct mc_package {
+	int reference;
+	uint32_t size;
+	void *data;
+};
+
+/*
+	lightuserdata
+	integer size
+
+	return lightuserdata, sizeof(struct mc_package *)
+ */
+static int
+mc_packlocal(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");
+	}
+	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 struct mc_package **
+	integer size (must be sizeof(struct mc_package *)
+
+	return package, lightuserdata, size
+ */
+static int
+mc_unpacklocal(lua_State *L) {
+	struct mc_package ** pack = lua_touserdata(L,1);
+	int sz = luaL_checkinteger(L,2);
+	if (sz != sizeof(*pack)) {
+		return luaL_error(L, "Invalid multicast package size %d", sz);
+	}
+	lua_settop(L, 1);
+	lua_pushlightuserdata(L, (*pack)->data);
+	lua_pushunsigned(L, (*pack)->size);
+	return 3;
+}
+
+/*
+	lightuserdata struct mc_package **
+	integer reference
+ */
+static int
+mc_bindrefer(lua_State *L) {
+	struct mc_package ** pack = lua_touserdata(L,1);
+	int ref = luaL_checkinteger(L,2);
+	if ((*pack)->reference != 0) {
+		return luaL_error(L, "Can't bind a multicast package more than once");
+	}
+	(*pack)->reference = ref;
+
+	return 0;
+}
+
+/*
+	lightuserdata struct mc_package **
+ */
+static int
+mc_closelocal(lua_State *L) {
+	struct mc_package **ptr = lua_touserdata(L,1);
+	struct mc_package *pack = *ptr;
+
+	int ref = __sync_sub_and_fetch(&pack->reference, 1);
+	if (ref <= 0) {
+		skynet_free(pack->data);
+		skynet_free(pack);
+		if (ref < 0) {
+			return luaL_error(L, "Invalid multicast package reference %d", ref);
+		}
+	}
+
+	return 0;
+}
+
+int
+luaopen_multicast_c(lua_State *L) {
+	luaL_Reg l[] = {
+		{ "pack", mc_packlocal },
+		{ "unpack", mc_unpacklocal },
+		{ "bind", mc_bindrefer },
+		{ "close", mc_closelocal },
+		{ NULL, NULL },
+	};
+	luaL_checkversion(L);
+	luaL_newlib(L,l);
+	return 1;
+}

lualib/multicast.lua

@@ -0,0 +1,66 @@
+local skynet = require "skynet"
+local c = require "multicast.c"
+
+local multicastd
+local multicast = {}
+local dispatch = {}
+
+local function default_conf(conf)
+	conf = conf or {}
+	conf.pack = conf.pack or skynet.pack
+	conf.unpack = conf.unpack or skynet.unpack
+
+	return conf
+end
+
+function multicast.newchannel(conf)
+	assert(multicastd, "Init first")
+	local channel = skynet.call(multicastd, "lua", "NEW")
+	dispatch[channel] = default_conf(conf)
+	return channel
+end
+
+function multicast.publish(channel, ...)
+	local conf = assert(dispatch[channel])
+	skynet.call(multicastd, "lua", "PUB", channel, c.pack(conf.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)
+end
+
+function multicast.unsubscribe(channel)
+	assert(multicastd, "Init first")
+	assert(dispatch[channel])
+	dispatch[channel] = nil
+	skynet.call(multicastd, "lua", "USUB", channel)
+end
+
+local function dispatch_subscribe(channel, source, pack, msg, sz)
+	local conf = dispatch[channel]
+	if not conf then
+		c.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)
+end
+
+local function init()
+	multicastd = skynet.uniqueservice "multicastd"
+	skynet.register_protocol {
+		name = "multicast",
+		id = skynet.PTYPE_MULTICAST,
+		unpack = c.unpack,
+		dispatch = dispatch_subscribe,
+	}
+end
+
+skynet.init(init, "multicast")
+
+return multicast

lualib/skynet.lua

@@ -16,7 +16,7 @@ local skynet = {
 	-- read skynet.h
 	PTYPE_TEXT = 0,
 	PTYPE_RESPONSE = 1,
---	PTYPE_MULTICAST = 2, -- DEPRECATED
+	PTYPE_MULTICAST = 2,
 	PTYPE_CLIENT = 3,
 	PTYPE_SYSTEM = 4,
 	PTYPE_HARBOR = 5,

service/multicastd.lua

@@ -0,0 +1,52 @@
+local skynet = require "skynet"
+local mc = require "multicast.c"
+
+local command = {}
+local channel = {}
+local channel_n = {}
+local channel_id = skynet.harbor(skynet.self())
+
+function command.NEW()
+	channel[channel_id] = {}
+	channel_n[channel_id] = 0
+	local ret = channel_id
+	channel_id = channel_id + 256
+	return ret
+end
+
+function command.PUB(source, c, 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
+end
+
+function command.SUB(source, c)
+	local group = assert(channel[c])
+	if not group[source] then
+		channel_n[c] = channel_n[c] + 1
+		group[source] = true
+	end
+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
+	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])
+		skynet.ret(skynet.pack(f(source, ...)))
+	end)
+end)

test/testmulticast.lua

@@ -0,0 +1,32 @@
+local skynet = require "skynet"
+local mc = require "multicast"
+
+local mode = ...
+
+if mode == "sub" then
+
+skynet.start(function()
+	skynet.dispatch("lua", function (_,_, cmd, channel)
+		assert(cmd == "init")
+		mc.subscribe(channel, {
+			dispatch = function (channel, source, ...)
+				print(string.format("%s ===> %s (%d)",skynet.address(source), skynet.address(skynet.self()), channel), ...)
+			end
+		})
+	end)
+end)
+
+else
+
+skynet.start(function()
+	local channel = mc.newchannel()
+	print("New channel", channel)
+	for i=1,10 do
+		local sub = skynet.newservice("testmulticast", "sub")
+		skynet.send(sub, "lua", "init", channel)
+	end
+
+	mc.publish(channel, "Hello World")
+end)
+
+end