diff --git a/config_log b/config_log index f4e4e0f4..3a5f51aa 100644 --- a/config_log +++ b/config_log @@ -1,4 +1,4 @@ -thread = 2 +thread = 8 mqueue = 256 cpath = "./service/?.so" logger = nil diff --git a/gate/main.c b/gate/main.c index f262a2e8..1c1fa3a9 100644 --- a/gate/main.c +++ b/gate/main.c @@ -135,7 +135,7 @@ _forward(struct skynet_context * ctx,struct gate *g, int uid, void * data, size_ } struct connection * agent = _id_to_agent(g,uid); if (agent->agent) { - // todo: client package has not session , send 0x7fffffff + // client package has not session , send 0x7fffffff skynet_send(ctx, agent->client, agent->agent, SESSION_CLIENT, data, len, 0); } else if (g->watchdog) { char * tmp = malloc(len + 32); @@ -202,6 +202,7 @@ _cb(struct skynet_context * ctx, void * ud, int session, uint32_t source, const } // big-endian uint16_t len = plen[0] << 8 | plen[1]; + void * data = mread_pull(m, len); if (data == NULL) { if (mread_closed(m)) { diff --git a/gate/mread.c b/gate/mread.c index 0645ec37..f795750a 100644 --- a/gate/mread.c +++ b/gate/mread.c @@ -276,7 +276,7 @@ mread_poll(struct mread_pool * self , int timeout) { socklen_t len = sizeof(struct sockaddr_in); int client_fd = accept(self->listen_fd , (struct sockaddr *)&remote_addr , &len); if (client_fd >= 0) { -// printf("MREAD connect %s:%u (fd=%d)\n",inet_ntoa(remote_addr.sin_addr),ntohs(remote_addr.sin_port), client_fd); +// printf("MREAD(%p) connect %s:%u (fd=%d)\n",self, inet_ntoa(remote_addr.sin_addr),ntohs(remote_addr.sin_port), client_fd); struct socket * s = _add_client(self, client_fd); if (s) { self->active = -1; @@ -390,6 +390,14 @@ mread_pull(struct mread_pool * self , int size) { for (;;) { int bytes = recv(s->fd, buffer, rd, MSG_DONTWAIT); if (bytes > 0) { +/* + printf("recv: [%d] ",s->fd); + int i; + for (i=0;iactive, s , blk); diff --git a/lualib/mcgroup.lua b/lualib/mcgroup.lua new file mode 100644 index 00000000..d7c09321 --- /dev/null +++ b/lualib/mcgroup.lua @@ -0,0 +1,30 @@ +local skynet = require "skynet" + +local SERVICE = "GROUPMGR" + +local group = {} + +function group.create() + return skynet.call(SERVICE, skynet.unpack, skynet.pack("NEW" , skynet.self())) +end + +function group.address(id) + local send_id = id * 2 + 1 + return skynet.query_group(send_id) +end + +function group.enter(id, handle) + handle = handle or skynet.self() + skynet.send(SERVICE, 0 , skynet.pack("ENTER", handle, id)) +end + +function group.leave(id, handle) + handle = handle or skynet.self() + skynet.send(SERVICE, 0 , skynet.pack("LEAVE", handle, id)) +end + +function group.release(id) + skynet.send(SERVICE, 0 , skynet.pack("CLEAR", skynet.self(), id)) +end + +return group \ No newline at end of file diff --git a/lualib/skynet.lua b/lualib/skynet.lua index 496b0daa..01a98c02 100644 --- a/lualib/skynet.lua +++ b/lualib/skynet.lua @@ -248,7 +248,7 @@ function skynet.clear_group(handle) end function skynet.query_group(handle) - return c.command("GROUP","QUERY " .. tostring(handle)) + return string_to_handle(c.command("GROUP","QUERY " .. tostring(handle))) end function skynet.address(addr) diff --git a/redisconf b/redisconf new file mode 100644 index 00000000..971fa540 --- /dev/null +++ b/redisconf @@ -0,0 +1,2 @@ +main = "127.0.0.1:6379" + diff --git a/service-src/service_harbor.c b/service-src/service_harbor.c index 38184acb..6bc7ead8 100644 --- a/service-src/service_harbor.c +++ b/service-src/service_harbor.c @@ -442,6 +442,19 @@ _remote_send_handle(struct harbor *h, struct skynet_context * context, uint32_t return 0; } +static void +_remote_register_name(struct harbor *h, struct skynet_context * context, const char name[GLOBALNAME_LENGTH], uint32_t handle) { + int i; + for (i=0;imap, name); @@ -457,23 +470,14 @@ _remote_send_name(struct harbor *h, struct skynet_context * context, uint32_t so header.destination = 0; header.session = (uint32_t)session; _push_queue(node->queue, msg, sz, &header); + // 0 for request + _remote_register_name(h, context, name, 0); return 1; } else { return _remote_send_handle(h, context, source, node->value, session, msg, sz); } } -static void -_remote_register_name(struct harbor *h, struct skynet_context * context, const char name[GLOBALNAME_LENGTH], uint32_t handle) { - int i; - for (i=0;iremote_fd[harbor_id] = fd; _broadcast(context, m, addr, sz, harbor_id); + + int i; + for (i=1;iremote_addr[i]; + if (addr == NULL) { + continue; + } + _send_to(fd, addr, strlen(addr), i); + } } diff --git a/service-src/service_tunnel.c b/service-src/service_tunnel.c new file mode 100644 index 00000000..9fb6a92d --- /dev/null +++ b/service-src/service_tunnel.c @@ -0,0 +1,22 @@ +#include "skynet.h" +#include + +static int +_cb(struct skynet_context * context, void * ud, int session, uint32_t source, const void * msg, size_t sz) { + uint32_t dest = (uint32_t)(uintptr_t)ud; + skynet_forward(context, dest); + return 0; +} + +int +tunnel_init(void * dummy, struct skynet_context *ctx, const char * args) { + uint32_t dest = skynet_queryname(ctx, args); + if (dest == 0) { + skynet_error(ctx, "Can't create tunnel to %s",args); + return 1; + } + + skynet_callback(ctx, (void*)(intptr_t)dest, _cb); + + return 0; +} diff --git a/service/console.lua b/service/console.lua index 63846d5c..70e22b12 100644 --- a/service/console.lua +++ b/service/console.lua @@ -8,8 +8,6 @@ skynet.start(function() local handle = skynet.launch(cmd) if handle == nil then print("Launch error:",cmd) - else - print("Launch:",handle,cmd) end skynet.yield() end diff --git a/service/group_agent.lua b/service/group_agent.lua new file mode 100644 index 00000000..3dcfa84b --- /dev/null +++ b/service/group_agent.lua @@ -0,0 +1,58 @@ +local skynet = require "skynet" + +local command = {} +local group = {} + +function command.CREATE(id) + assert(group[id] == nil) + group[id] = {} + local recv_id = id * 2 + local addr = skynet.query_group(recv_id) + skynet.ret(skynet.pack(addr)) +end + +function command.AGENT(id, handle) + local tunnel = skynet.launch("tunnel", skynet.address(handle)) + assert(tunnel) + local send_id = id * 2 + 1 + skynet.enter_group(send_id, tunnel) + table.insert(group[id] , tunnel) +end + +function command.ENTER(id, handle) + local recv_id = id * 2 + local send_id = id * 2 + 1 + skynet.enter_group(recv_id, handle) + skynet.enter_group(send_id, handle) +end + +function command.LEAVE(id, handle) + local recv_id = id * 2 + local send_id = id * 2 + 1 + skynet.leave_group(recv_id, handle) + skynet.leave_group(send_id, handle) +end + +function command.CLEAR(id) + local g = group[id] + assert(g) + local recv_id = id * 2 + local send_id = id * 2 + 1 + skynet.clear_group(recv_id) + skynet.clear_group(send_id) + group[id] = nil + for _,v in ipairs(g) do + skynet.kill(v) + end +end + +skynet.dispatch(function (msg,sz) + local cmd , id , handle = skynet.unpack(msg,sz) + local f = command[cmd] + assert(f, cmd) + f(id,handle) +end) + +skynet.start(function() + skynet.send("GROUPMGR" , 0, skynet.pack("MASTER", skynet.self())) +end) diff --git a/service/group_mgr.lua b/service/group_mgr.lua new file mode 100644 index 00000000..34e2e68c --- /dev/null +++ b/service/group_mgr.lua @@ -0,0 +1,69 @@ +local skynet = require "skynet" + +local id = 0 +local harbor_ctrl = {} +local multicast = {} +local command = {} + +function command.MASTER(address, harbor) + harbor_ctrl[harbor] = address +end + +local function create_group_in(harbor, id) + local local_ctrl = assert(harbor_ctrl[harbor]) + local g = multicast[id] + local local_mc = skynet.call(local_ctrl, skynet.unpack, skynet.pack("CREATE", id)) + for _,mc in pairs(g) do + skynet.send(local_ctrl, 0, skynet.pack("AGENT", id, mc)) + end + for harbor_id,ctrl in pairs(harbor_ctrl) do + if harbor_id ~= harbor then + skynet.send(ctrl,0, skynet.pack("AGENT", id, local_mc)) + end + end + g[harbor] = local_mc +end + +function command.NEW() + id = id + 1 + multicast[id] = {} + skynet.ret(skynet.pack(id)) +end + +function command.ENTER(address, harbor, id) + local g = multicast[id] + assert(g,id) + if g[harbor] == nil then + create_group_in(harbor, id) + end + local local_ctrl = assert(harbor_ctrl[harbor]) + + skynet.send(local_ctrl, 0, skynet.pack("ENTER", id, address)) +end + +function command.LEAVE(address, harbor, id) + local g = multicast[id] + assert(g,id) + local local_ctrl = assert(harbor_ctrl[harbor]) + skynet.send(local_ctrl, 0, skynet.pack("LEAVE", id, address)) +end + +function command.DELETE(_,_, id) + local g = multicast[id] + assert(g,id) + multicast[id] = nil + + for harbor_id,_ in pairs(g) do + skynet.send(harbor_ctrl[harbor_id],0, skynet.pack("CLEAR", id)) + end +end + +skynet.dispatch(function (msg,sz) + local cmd , address , param = skynet.unpack(msg,sz) + local f = command[cmd] + assert(f, cmd, param) + local harbor = skynet.harbor(address) + f(address, harbor, param) +end) + +skynet.register("GROUPMGR") \ No newline at end of file diff --git a/service/main.lua b/service/main.lua index 00847b46..a1951b8f 100644 --- a/service/main.lua +++ b/service/main.lua @@ -6,11 +6,14 @@ skynet.start(function() print("Server start") local lualog = skynet.launch("snlua","lualog") local launcher = skynet.launch("snlua","launcher") + local group_mgr = skynet.launch("snlua", "group_mgr") + local group_agent = skynet.launch("snlua", "group_agent") local remoteroot = skynet.launch("snlua","remote_root") local console = skynet.launch("snlua","console") local watchdog = skynet.launch("snlua","watchdog","8888 4 0") local db = skynet.launch("snlua","simpledb") local connection = skynet.launch("connection","256") local redis = skynet.launch("snlua","redis-mgr") +-- skynet.launch("snlua","testgroup") skynet.exit() end) diff --git a/service/main_log.lua b/service/main_log.lua index 408f7323..4b98a112 100644 --- a/service/main_log.lua +++ b/service/main_log.lua @@ -1,7 +1,15 @@ local skynet = require "skynet" -print("Log server start") +skynet.dispatch() -skynet.launch("snlua","globallog") +skynet.start(function() + print("Log server start") + local lualog = skynet.launch("snlua","lualog") + local launcher = skynet.launch("snlua","launcher") + local group_agent = skynet.launch("snlua", "group_agent") + local console = skynet.launch("snlua","console") + local log = skynet.launch("snlua","globallog") +-- skynet.launch("snlua","testgroup_c 11 1") + skynet.exit() +end) -skynet.exit() diff --git a/service/testgroup.lua b/service/testgroup.lua index 05e1d69a..9d4cb557 100644 --- a/service/testgroup.lua +++ b/service/testgroup.lua @@ -1,13 +1,24 @@ local skynet = require "skynet" +local group = require "mcgroup" skynet.dispatch() skynet.start(function() - local group = skynet.query_group(1) + local gid = group.create() + local gaddr = group.address(gid) + print("=== Create Group ===",gid,skynet.address(gaddr)) for i=1,10 do local address = skynet.newservice("testgroup_c", tostring(i)) - skynet.enter_group(1 , address) + group.enter(gid , address) end - skynet.send(group,"Hello World") - skynet.exit() + skynet.sleep(1000) + skynet.send(gaddr,"Hello World") + +-- local group = skynet.query_group(1) +-- for i=1,10 do +-- local address = skynet.newservice("testgroup_c", tostring(i)) +-- skynet.enter_group(1 , address) +-- end +-- skynet.send(group,"Hello World") +-- skynet.exit() end) diff --git a/service/testgroup_c.lua b/service/testgroup_c.lua index 6a729883..3407c1a2 100644 --- a/service/testgroup_c.lua +++ b/service/testgroup_c.lua @@ -1,11 +1,14 @@ local skynet = require "skynet" -local id = ... +local group = require "mcgroup" +local id, g = ... skynet.dispatch(function (msg,sz) - print(id, skynet.tostring(msg,sz)) + print("===>",id, skynet.tostring(msg,sz)) end ) skynet.start(function() - print("start",id) + if g then + group.enter(tonumber(g)) + end end) diff --git a/skynet-src/skynet_server.c b/skynet-src/skynet_server.c index 61d2d5af..91f13fa3 100644 --- a/skynet-src/skynet_server.c +++ b/skynet-src/skynet_server.c @@ -17,6 +17,8 @@ #define BLACKHOLE "blackhole" #define DEFAULT_MESSAGE_QUEUE 16 +#define CALLING_CHECK + #ifdef CALLING_CHECK #define CHECKCALLING_BEGIN(ctx) assert(__sync_lock_test_and_set(&ctx->calling,1) == 0); @@ -162,24 +164,28 @@ skynet_isremote(struct skynet_context * ctx, uint32_t handle, int * harbor) { return ret; } +static void +_send_message(uint32_t des, struct skynet_message *msg) { + if (skynet_harbor_message_isremote(des)) { + struct remote_message * rmsg = malloc(sizeof(*rmsg)); + rmsg->destination.handle = des; + rmsg->message = msg->data; + rmsg->sz = msg->sz; + skynet_harbor_send(rmsg, msg->source, msg->session); + } else { + if (skynet_context_push(des, msg)) { + free(msg->data); + skynet_error(NULL, "Drop message from %x forward to %x (size=%d)", msg->source, des, (int)msg->sz); + } + } +} + static int _forwarding(struct skynet_context *ctx, struct skynet_message *msg) { if (ctx->forward) { uint32_t des = ctx->forward; ctx->forward = 0; - if (skynet_harbor_message_isremote(des)) { - struct remote_message * rmsg = malloc(sizeof(*rmsg)); - rmsg->destination.handle = des; - rmsg->message = msg->data; - rmsg->sz = msg->sz; - skynet_harbor_send(rmsg, msg->source, msg->session); - } else { - if (skynet_context_push(des, msg)) { - free(msg->data); - skynet_error(NULL, "Drop message from %x forward to %x (size=%d)", msg->source, des, (int)msg->sz); - return 1; - } - } + _send_message(des, msg); return 1; } return 0; @@ -189,6 +195,17 @@ static void _mc(void *ud, uint32_t source, const void * msg, size_t sz) { struct skynet_context * ctx = ud; ctx->cb(ctx, ctx->cb_ud, 0, source, msg, sz); + if (ctx->forward) { + uint32_t des = ctx->forward; + ctx->forward = 0; + struct skynet_message message; + message.source = source; + message.session = 0; + message.data = malloc(sz); + memcpy(message.data, msg, sz); + message.sz = sz; + _send_message(des, &message); + } } static void