diff --git a/Makefile b/Makefile index 7e061a36..e6a7326f 100644 --- a/Makefile +++ b/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 diff --git a/lualib/skynet.lua b/lualib/skynet.lua index dbe319f4..248d63d5 100644 --- a/lualib/skynet.lua +++ b/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 diff --git a/service-src/service_multicast.c b/service-src/service_multicast.c new file mode 100644 index 00000000..3d185871 --- /dev/null +++ b/service-src/service_multicast.c @@ -0,0 +1,60 @@ +#include "skynet.h" +#include "skynet_multicast.h" + +#include +#include +#include + +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; +} + + + diff --git a/service/testgroup.lua b/service/testgroup.lua new file mode 100644 index 00000000..a4e1235b --- /dev/null +++ b/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) diff --git a/service/testgroup_c.lua b/service/testgroup_c.lua new file mode 100644 index 00000000..741a3190 --- /dev/null +++ b/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) diff --git a/skynet-src/skynet.h b/skynet-src/skynet.h index 70e671de..58392679 100644 --- a/skynet-src/skynet.h +++ b/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; diff --git a/skynet-src/skynet_group.c b/skynet-src/skynet_group.c new file mode 100644 index 00000000..067e3c47 --- /dev/null +++ b/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 +#include +#include + +#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; +} diff --git a/skynet-src/skynet_group.h b/skynet-src/skynet_group.h new file mode 100644 index 00000000..b12daf8b --- /dev/null +++ b/skynet-src/skynet_group.h @@ -0,0 +1,13 @@ +#ifndef SKYNET_GROUP_H +#define SKYNET_GROUP_H + +#include + +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 diff --git a/skynet-src/skynet_mq.c b/skynet-src/skynet_mq.c index bdd3eae5..966b7eff 100644 --- a/skynet-src/skynet_mq.c +++ b/skynet-src/skynet_mq.c @@ -1,4 +1,6 @@ +#include "skynet.h" #include "skynet_mq.h" +#include "skynet_multicast.h" #include #include @@ -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; diff --git a/skynet-src/skynet_multicast.c b/skynet-src/skynet_multicast.c new file mode 100644 index 00000000..749cedae --- /dev/null +++ b/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 +#include + +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;inumber;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= 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;inumber;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; +} diff --git a/skynet-src/skynet_multicast.h b/skynet-src/skynet_multicast.h new file mode 100644 index 00000000..17c8c5ce --- /dev/null +++ b/skynet-src/skynet_multicast.h @@ -0,0 +1,22 @@ +#ifndef SKYNET_MULTICAST_H +#define SKYNET_MULTICAST_H + +#include + +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 diff --git a/skynet-src/skynet_server.c b/skynet-src/skynet_server.c index 64e7a2a2..ac150636 100644 --- a/skynet-src/skynet_server.c +++ b/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 #include @@ -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; } diff --git a/skynet-src/skynet_server.h b/skynet-src/skynet_server.h index 992ee1d2..5297c834 100644 --- a/skynet-src/skynet_server.h +++ b/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); diff --git a/skynet-src/skynet_start.c b/skynet-src/skynet_start.c index ef863095..a865355f 100644 --- a/skynet-src/skynet_start.c +++ b/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 #include @@ -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);