multicast support

Makefile

@@ -18,6 +18,7 @@ lualib/lpeg.so \
 lualib/protobuf.so \
 lualib/int64.so \
 service/master.so \
+service/multicast.so \
 service/harbor.so
 
 skynet : \
@@ -30,9 +31,14 @@ skynet-src/skynet_start.c \
 skynet-src/skynet_timer.c \
 skynet-src/skynet_error.c \
 skynet-src/skynet_harbor.c \
+skynet-src/skynet_multicast.c \
+skynet-src/skynet_group.c \
 skynet-src/skynet_env.c
 	gcc $(CFLAGS) -Wl,-E -o $@ $^ -Iskynet-src -lpthread -ldl -lrt -Wl,-E -llua -lm
 
+service/multicast.so : service-src/service_multicast.c
+	gcc $(CFLAGS) $(SHARED) $^ -o $@ -Iskynet-src
+
 service/master.so : service-src/service_master.c
 	gcc $(CFLAGS) $(SHARED) $^ -o $@ -Iskynet-src
 

lualib/skynet.lua

@@ -216,6 +216,22 @@ function skynet.newservice(name, ...)
 	return handle
 end
 
+function skynet.enter_group(handle)
+	c.command("GROUP", "ENTER " .. tostring(handle))
+end
+
+function skynet.leave_group(handle)
+	c.command("GROUP", "LEAVE " .. tostring(handle))
+end
+
+function skynet.clear_group(handle)
+	c.command("GROUP", "CLEAR " .. tostring(handle))
+end
+
+function skynet.query_group(handle)
+	return c.command("GROUP","QUERY " .. tostring(handle))
+end
+
 ------ remote object --------
 
 do

service-src/service_multicast.c

@@ -0,0 +1,60 @@
+#include "skynet.h"
+#include "skynet_multicast.h"
+
+#include <stdlib.h>
+#include <string.h>
+#include <stdio.h>
+
+struct skynet_multicast_group *
+multicast_create() {
+	return skynet_multicast_newgroup();
+}
+
+void
+multicast_release(struct skynet_multicast_group *g) {
+	skynet_multicast_deletegroup(g);
+}
+
+static int
+_maincb(struct skynet_context * context, void * ud, int session, const char * addr, const void * msg, size_t sz) {
+	struct skynet_multicast_group *g = ud;
+	uint32_t source = strtoul(addr+1, NULL, 16);
+	if (source == 0) {
+		char cmd = '\0';
+		uint32_t handle = 0;
+		sscanf(msg,"%c %x",&cmd,&handle);
+		if (handle == 0) {
+			skynet_error(context, "Invalid handle %s",msg);
+			return 0;
+		}
+		switch (cmd) {
+		case 'E':
+			skynet_multicast_entergroup(g, handle);
+			break;
+		case 'L':
+			skynet_multicast_leavegroup(g, handle);
+			break;
+		case 'C':
+			skynet_command(context, "EXIT", NULL);
+			break;
+		default:
+			skynet_error(context, "Invalid command %s",msg);
+			break;
+		}
+		return 0;		
+	} else {
+		struct skynet_multicast_message * mc = skynet_multicast_create(msg, sz, source);
+		skynet_multicast_castgroup(context, g, mc);
+		return 1;
+	}
+}
+
+int
+multicast_init(struct skynet_multicast_group *g, struct skynet_context *ctx, const char * args) {
+	skynet_callback(ctx, g, _maincb);
+
+	return 0;
+}
+
+
+

service/testgroup.lua

@@ -0,0 +1,12 @@
+local skynet = require "skynet"
+
+skynet.dispatch()
+
+skynet.start(function()
+	local group = skynet.query_group(1)
+	for i=1,10 do
+		skynet.newservice("testgroup_c", tostring(i))
+	end
+	skynet.send(group,"Hello World")
+	skynet.exit()
+end)

service/testgroup_c.lua

@@ -0,0 +1,12 @@
+local skynet = require "skynet"
+local id = ...
+
+skynet.dispatch(function (msg,sz)
+	print(id, skynet.tostring(msg,sz))
+end
+)
+
+skynet.start(function()
+	print("start",id)
+	skynet.enter_group(1)
+end)

skynet-src/skynet.h

@@ -6,6 +6,8 @@
 
 #define DONTCOPY 1
 #define SESSION_CLIENT 0x7fffffff
+#define SESSION_MULTICAST 0x7ffffffe
+#define SESSION_MAX 0x7ffffff0
 
 struct skynet_context;
 

skynet-src/skynet_group.c

@@ -0,0 +1,149 @@
+#include "skynet_group.h"
+#include "skynet_multicast.h"
+#include "skynet_server.h"
+#include "skynet.h"
+
+#include <string.h>
+#include <stdio.h>
+#include <assert.h>
+
+#define HASH_SIZE 1024
+
+struct group_node {
+	int handle;
+	struct skynet_context *ctx;
+	struct group_node * next;
+};
+
+struct group {
+	int lock;
+	struct group_node * node[HASH_SIZE];
+};
+
+struct group * _G = NULL;
+
+inline static void
+_lock(struct group *g) {
+	while (__sync_lock_test_and_set(&g->lock,1)) {}
+}
+
+inline static void
+_unlock(struct group *g) {
+	__sync_lock_release(&g->lock);
+}
+
+static struct skynet_context *
+_create_group(struct group * g, int handle) {
+	int hash = handle % HASH_SIZE;
+	struct skynet_context * inst = skynet_context_new("multicast",NULL);
+	assert(inst);
+	struct group_node * new_node = malloc(sizeof(struct group_node));
+	new_node->handle = handle;
+	new_node->ctx = inst;
+	new_node->next = g->node[hash];
+	g->node[hash] = new_node;
+
+	return inst;
+}
+
+uint32_t
+skynet_group_query(int handle) {
+	struct group *g = _G;
+	_lock(g);
+	
+	int hash = handle % HASH_SIZE;
+	struct group_node * node = g->node[hash];
+	while (node) {
+		if (node->handle == handle) {
+			struct skynet_context * ctx = node->ctx;
+			uint32_t addr = skynet_context_handle(ctx);
+			_unlock(g);
+			return addr;
+		}
+		node = node->next;
+	}
+	struct skynet_context * ctx = _create_group(g, handle);
+	uint32_t addr = skynet_context_handle(ctx);
+	_unlock(g);
+
+	return addr;
+}
+
+static void
+send_command(struct skynet_context *ctx, const char * cmd, uint32_t node) {
+	char * tmp = malloc(16);
+	int n = sprintf(tmp, "%s %x", cmd, node);
+	skynet_context_send(ctx, tmp, n+1 , 0, 0);
+}
+
+void 
+skynet_group_enter(int handle, uint32_t n) {
+	struct group *g = _G;
+	_lock(g);
+
+	int hash = handle % HASH_SIZE;
+	struct group_node * node = g->node[hash];
+	while (node) {
+		if (node->handle == handle) {
+			send_command(node->ctx, "E", n);
+			_unlock(g);
+			return;
+		}
+		node = node->next;
+	}
+	struct skynet_context * inst = _create_group(g, handle);
+
+	send_command(inst, "E", n);
+
+	_unlock(g);
+}
+
+void 
+skynet_group_leave(int handle, uint32_t n) {
+	struct group *g = _G;
+	_lock(g);
+
+	int hash = handle % HASH_SIZE;
+	struct group_node * node = g->node[hash];
+	while (node) {
+		if (node->handle == handle) {
+			send_command(node->ctx, "L", n);
+			break;
+		}
+		node = node->next;
+	}
+
+	_unlock(g);
+}
+
+void
+skynet_group_clear(int handle) {
+	struct group *g = _G;
+	_lock(g);
+
+	int hash = handle % HASH_SIZE;
+	struct group_node ** pnode = &g->node[hash];
+	while (*pnode) {
+		struct group_node * node = *pnode;
+		if (node->handle == handle) {
+			struct skynet_context * ctx = node->ctx;
+			
+			char * cmd = malloc(8);
+			int n = sprintf(cmd, "C");
+			skynet_context_send(ctx, cmd, n+1, 0 , 0);
+			*pnode = node->next;
+			free(node);
+			break;
+		}
+		pnode = &node->next;
+	}
+
+	_unlock(g);
+}
+
+void 
+skynet_group_init() {
+	struct group * g = malloc(sizeof(*g));
+	memset(g,0,sizeof(*g));
+	_G = g;
+}

skynet-src/skynet_group.h

@@ -0,0 +1,13 @@
+#ifndef SKYNET_GROUP_H
+#define SKYNET_GROUP_H
+
+#include <stdint.h>
+
+uint32_t skynet_group_query(int handle);
+void skynet_group_enter(int handle, uint32_t node);
+void skynet_group_leave(int handle, uint32_t node);
+void skynet_group_clear(int handle);
+
+void skynet_group_init();
+
+#endif

skynet-src/skynet_mq.c

@@ -1,4 +1,6 @@
+#include "skynet.h"
 #include "skynet_mq.h"
+#include "skynet_multicast.h"
 
 #include <stdio.h>
 #include <stdlib.h>
@@ -28,11 +30,6 @@ struct global_queue {
 
 static struct global_queue *Q = NULL;
 
-static inline void
-_lock_global_queue() {
-	while (__sync_lock_test_and_set(&Q->lock,1)) {}
-}
-
 #define LOCK(q) while (__sync_lock_test_and_set(&(q)->lock,1)) {}
 #define UNLOCK(q) __sync_lock_release(&(q)->lock);
 
@@ -196,7 +193,12 @@ _drop_queue(struct message_queue *q) {
 	int s = 0;
 	while(!skynet_mq_pop(q, &msg)) {
 		++s;
-		free(msg.data);
+		if (msg.session == SESSION_MULTICAST) {
+			assert(msg.sz == 0);
+			skynet_multicast_dispatch((struct skynet_multicast_message *)msg.data, NULL, NULL);
+		} else {
+			free(msg.data);
+		}
 	}
 	_release(q);
 	return s;

skynet-src/skynet_multicast.c

@@ -0,0 +1,242 @@
+#include "skynet.h"
+#include "skynet_multicast.h"
+#include "skynet_server.h"
+#include "skynet_handle.h"
+
+#include <stdlib.h>
+#include <string.h>
+
+struct skynet_multicast_message {
+	int ref;
+	const void * msg;
+	size_t sz;
+	char source[10];
+};
+
+static void
+_id_to_hex(char * str, uint32_t id) {
+	int i;
+	static char hex[16] = { '0','1','2','3','4','5','6','7','8','9','A','B','C','D','E','F' };
+	str[0] = ':';
+	for (i=0;i<8;i++) {
+		str[i+1] = hex[(id >> ((7-i) * 4))&0xf];
+	}
+	str[9] = '\0';
+}
+
+struct skynet_multicast_message * 
+skynet_multicast_create(const void * msg, size_t sz, uint32_t source) {
+	struct skynet_multicast_message * mc = malloc(sizeof(*mc));
+	mc->ref = 0;
+	mc->msg = msg;
+	mc->sz = sz;
+	_id_to_hex(mc->source, source);
+	return mc;
+}
+
+void 
+skynet_multicast_copy(struct skynet_multicast_message *mc, int copy) {
+	__sync_fetch_and_add(&mc->ref, copy);
+}
+
+void 
+skynet_multicast_dispatch(struct skynet_multicast_message * msg, void * ud, skynet_multicast_func func) {
+	if (func) {
+		func(ud, msg->source, msg->msg, msg->sz);
+	}
+	int ref = __sync_sub_and_fetch(&msg->ref, 1);
+	if (ref == 0) {
+		free((void *)msg->msg);
+		free(msg);
+	}
+}
+
+struct array {
+	int cap;
+	int number;
+	uint32_t *data;
+};
+
+struct pair {
+	uint32_t handle;
+	struct skynet_context * ctx;
+};
+
+struct skynet_multicast_group {
+	struct array enter_queue;
+	struct array leave_queue;
+	int cap;
+	int number;
+	struct pair * data;
+};
+
+struct skynet_multicast_group * 
+skynet_multicast_newgroup() {
+	struct skynet_multicast_group * g = malloc(sizeof(*g));
+	memset(g,0,sizeof(*g));
+	return g;
+}
+
+void 
+skynet_multicast_deletegroup(struct skynet_multicast_group * g) {
+	int i;
+	for (i=0;i<g->number;i++) {
+		if (g->data[i].ctx) {
+			skynet_context_release(g->data[i].ctx);
+		}
+	}
+	free(g->data);
+	free(g->enter_queue.data);
+	free(g->leave_queue.data);
+	free(g);
+}
+
+static void
+push_array(struct array * a, uint32_t v) {
+	if (a->number >= a->cap) {
+		a->cap *= 2;
+		if (a->cap == 0) {
+			a->cap = 4;
+		}
+		a->data = realloc(a->data, a->cap * sizeof(uint32_t));
+	}
+	a->data[a->number++] = v;
+}
+
+void 
+skynet_multicast_entergroup(struct skynet_multicast_group * group, uint32_t handle) {
+	push_array(&group->enter_queue, handle);
+}
+
+void 
+skynet_multicast_leavegroup(struct skynet_multicast_group * group, uint32_t handle) {
+	push_array(&group->leave_queue, handle);
+}
+
+
+static int
+compar_uint(const void *a, const void *b) {
+	const uint32_t * aa = a;
+	const uint32_t * bb = b;
+	return (int)(*aa - *bb);
+}
+
+static void
+combine_queue(struct skynet_context * from, struct skynet_multicast_group * group) {
+	qsort(group->enter_queue.data, group->enter_queue.number, sizeof(uint32_t), compar_uint);
+	qsort(group->leave_queue.data, group->leave_queue.number, sizeof(uint32_t), compar_uint);
+	int i;
+	int enter = group->enter_queue.number;
+	uint32_t last = 0;
+
+	int new_size = group->number + enter;
+	if (new_size > group->cap) {
+		group->data = realloc(group->data, new_size * sizeof(struct pair));
+		group->cap = new_size;
+	}
+
+	// combine enter queue
+	int old_index = group->number - 1;
+	int new_index = new_size - 1;
+	for (i= enter - 1;i >=0 ; i--) {
+		uint32_t handle = group->enter_queue.data[i];
+		if (handle == last)
+			continue;
+		last = handle;
+		struct skynet_context * ctx = skynet_handle_grab(handle);
+		if (ctx == NULL)
+			continue;
+		if (old_index < 0) {
+			group->data[new_index].handle = handle;
+			group->data[new_index].ctx = ctx;
+		} else {
+			struct pair * p = &group->data[old_index];
+			if (handle == p->handle)
+				continue;
+			if (handle > p->handle) {
+				group->data[new_index].handle = handle;
+				group->data[new_index].ctx = ctx;
+			} else {
+				group->data[new_index] = group->data[old_index];
+				--old_index;
+				last = 0;
+				++i;
+			}
+		}
+		--new_index;
+	}
+	while (old_index >= 0) {
+		group->data[new_index] = group->data[old_index];
+		--old_index;
+		--new_index;
+	}
+	group->enter_queue.number = 0;
+
+	// remove leave queue
+	old_index = new_index + 1;
+	new_index = 0;
+
+	int count = new_size - old_index;
+
+	int leave = group->leave_queue.number;
+	for (i=0;i<leave;i++) {
+		if (old_index >= new_size) {
+			count = 0;
+			break;
+		}
+		uint32_t handle = group->leave_queue.data[i];
+		struct pair * p = &group->data[old_index];
+		if (handle == p->handle) {
+			--count;
+			++old_index;
+			if (p->ctx) {
+				skynet_context_release(p->ctx);
+			}
+		} else if ( handle > p->handle) {
+			group->data[new_index] = group->data[old_index];
+			++new_index;
+			++old_index;
+			--i;
+		} else {
+			skynet_error(from, "Try to remove a none exist handle : %x", handle);
+		}
+	}
+	while (new_index < count) {
+		group->data[new_index] = group->data[old_index];
+		++new_index;
+		++old_index;
+	}
+
+	group->leave_queue.number = 0;
+
+	group->number = new_index;
+}
+
+int 
+skynet_multicast_castgroup(struct skynet_context * from, struct skynet_multicast_group * group, struct skynet_multicast_message *msg) {
+	combine_queue(from, group);
+	if (group->number == 0) {
+		skynet_multicast_dispatch(msg, NULL, NULL);
+		return 0;
+	}
+	uint32_t source = skynet_context_handle(from);
+	skynet_multicast_copy(msg, group->number);
+	int i;
+	int release = 0;
+	for (i=0;i<group->number;i++) {
+		struct pair * p = &group->data[i];
+		skynet_context_send(p->ctx, msg, 0 , source, SESSION_MULTICAST);
+		int ref = skynet_context_ref(p->ctx);
+		if (ref == 1) {
+			skynet_context_release(p->ctx);
+			struct skynet_context * ctx = skynet_handle_grab(p->handle);
+			if (ctx == NULL) {
+				p->ctx = NULL;
+				skynet_multicast_leavegroup(group, p->handle);
+				++release;
+			}
+		}
+	}
+	
+	return group->number - release;
+}

skynet-src/skynet_multicast.h

@@ -0,0 +1,22 @@
+#ifndef SKYNET_MULTICAST_H
+#define SKYNET_MULTICAST_H
+
+#include <stddef.h>
+
+struct skynet_multicast_message;
+struct skynet_multicast_group;
+struct skynet_context;
+
+typedef void (*skynet_multicast_func)(void *ud, const char * source, const void * msg, size_t sz);
+
+struct skynet_multicast_message * skynet_multicast_create(const void * msg, size_t sz, uint32_t source);
+void skynet_multicast_copy(struct skynet_multicast_message *, int copy);
+void skynet_multicast_dispatch(struct skynet_multicast_message * msg, void * ud, skynet_multicast_func func);
+
+struct skynet_multicast_group * skynet_multicast_newgroup();
+void skynet_multicast_deletegroup(struct skynet_multicast_group * group);
+void skynet_multicast_entergroup(struct skynet_multicast_group * group, uint32_t handle);
+void skynet_multicast_leavegroup(struct skynet_multicast_group * group, uint32_t handle);
+int skynet_multicast_castgroup(struct skynet_context * from, struct skynet_multicast_group * group, struct skynet_multicast_message *msg);
+
+#endif

skynet-src/skynet_server.c

@@ -6,6 +6,8 @@
 #include "skynet_harbor.h"
 #include "skynet_env.h"
 #include "skynet.h"
+#include "skynet_multicast.h"
+#include "skynet_group.h"
 
 #include <string.h>
 #include <assert.h>
@@ -109,7 +111,7 @@ skynet_context_new(const char * name, const char *param) {
 int
 skynet_context_newsession(struct skynet_context *ctx) {
 	int session = ++ctx->session_id;
-	if (session >= SESSION_CLIENT) {
+	if (session >= SESSION_MAX) {
 		ctx->session_id = 1;
 		return 1;
 	}
@@ -138,6 +140,12 @@ skynet_context_release(struct skynet_context *ctx) {
 	return ctx;
 }
 
+int 
+skynet_context_ref(struct skynet_context *ctx) {
+	return ctx->ref;
+}
+
+
 int
 skynet_context_push(uint32_t handle, struct skynet_message *message) {
 	struct skynet_context * ctx = skynet_handle_grab(handle);
@@ -183,12 +191,21 @@ _forwarding(struct skynet_context *ctx, struct skynet_message *msg) {
 	return 0;
 }
 
+static void
+_mc(void *ud, const char * source, const void * msg, size_t sz) {
+	struct skynet_context * ctx = ud;
+	ctx->cb(ctx, ctx->cb_ud, 0, source, msg, sz);
+}
+
 static void
 _dispatch_message(struct skynet_context *ctx, struct skynet_message *msg) {
 	assert(ctx->init);
 	CHECKCALLING_BEGIN(ctx)
 	if (msg->source == SKYNET_SYSTEM_TIMER) {
 		ctx->cb(ctx, ctx->cb_ud, msg->session, NULL, msg->data, msg->sz);
+	} else if (msg->session == SESSION_MULTICAST) {
+		assert(msg->sz == 0);
+		skynet_multicast_dispatch((struct skynet_multicast_message *)msg->data, ctx, _mc);
 	} else {
 		char tmp[10];
 		_id_to_hex(tmp, msg->source);
@@ -249,6 +266,32 @@ _copy_name(char name[GLOBALNAME_LENGTH], const char * addr) {
 	}
 }
 
+static const char *
+_group_command(struct skynet_context * ctx, const char * cmd, int handle) {
+	uint32_t self = ctx->handle;
+	if (strcmp(cmd, "ENTER") == 0) {
+		skynet_group_enter(handle, self);
+		return NULL;
+	}
+	if (strcmp(cmd, "LEAVE") == 0) {
+		skynet_group_leave(handle, self);
+		return NULL;
+	}
+	if (strcmp(cmd, "QUERY") == 0) {
+		uint32_t addr = skynet_group_query(handle);
+		if (addr == 0) {
+			return NULL;
+		}
+		_id_to_hex(ctx->result, addr);
+		return ctx->result;
+	}
+	if (strcmp(cmd, "CLEAR") == 0) {
+		skynet_group_clear(handle);
+		return NULL;
+	}
+	return NULL;
+}
+
 const char * 
 skynet_command(struct skynet_context * context, const char * cmd , const char * param) {
 	if (strcmp(cmd,"TIMEOUT") == 0) {
@@ -370,6 +413,17 @@ skynet_command(struct skynet_context * context, const char * cmd , const char *
 		return context->result;
 	}
 
+	if (strcmp(cmd,"GROUP") == 0) {
+		int sz = strlen(param);
+		char tmp[sz+1];
+		strcpy(tmp,param);
+		tmp[sz] = '\0';
+		char cmd[sz+1];
+		int handle=0;
+		sscanf(tmp, "%s %d",cmd,&handle);
+		return _group_command(context, cmd, handle);
+	}
+
 	return NULL;
 }
 

skynet-src/skynet_server.h

@@ -10,6 +10,7 @@ struct skynet_message;
 struct skynet_context * skynet_context_new(const char * name, const char * parm);
 void skynet_context_grab(struct skynet_context *);
 struct skynet_context * skynet_context_release(struct skynet_context *);
+int skynet_context_ref(struct skynet_context *);
 uint32_t skynet_context_handle(struct skynet_context *);
 void skynet_context_init(struct skynet_context *, uint32_t handle);
 int skynet_context_push(uint32_t handle, struct skynet_message *message);

skynet-src/skynet_start.c

@@ -5,6 +5,7 @@
 #include "skynet_module.h"
 #include "skynet_timer.h"
 #include "skynet_harbor.h"
+#include "skynet_group.h"
 
 #include <pthread.h>
 #include <unistd.h>
@@ -55,6 +56,7 @@ _start_master(const char * master) {
 
 void 
 skynet_start(struct skynet_config * config) {
+	skynet_group_init();
 	skynet_harbor_init(config->harbor);
 	skynet_handle_init(config->harbor);
 	skynet_mq_init(config->mqueue_size);