diff --git a/connection/main.c b/connection/main.c index d5f5f468..5b9d4ad8 100644 --- a/connection/main.c +++ b/connection/main.c @@ -17,7 +17,7 @@ struct connection { int fd; - char * addr; + uint32_t address; }; struct connection_server { @@ -40,10 +40,6 @@ connection_release(struct connection_server * server) { if (server->pool) { connection_deletepool(server->pool); } - int i; - for (i=0;imax_connection;i++) { - free(server->conn[i].addr); - } free(server->conn); free(server); } @@ -61,7 +57,7 @@ _expand(struct connection_server * server) { } static void -_add(struct connection_server * server, int fd , char * addr) { +_add(struct connection_server * server, int fd , uint32_t address) { ++server->current_connection; if (server->current_connection > server->max_connection) { _expand(server); @@ -69,9 +65,9 @@ _add(struct connection_server * server, int fd , char * addr) { int i; for (i=0;imax_connection;i++) { struct connection * c = &server->conn[i]; - if (c->addr == NULL) { + if (c->address == 0) { c->fd = fd; - c->addr = addr; + c->address = address; int err = connection_add(server->pool, fd , c); assert(err == 0); return; @@ -86,8 +82,7 @@ _del(struct connection_server * server, int fd) { for (i=0;imax_connection;i++) { struct connection * c = &server->conn[i]; if (c->fd == fd) { - free(c->addr); - c->addr = NULL; + c->address = 0; c->fd = 0; connection_del(server->pool, fd); return; @@ -117,15 +112,15 @@ _poll(struct connection_server * server) { if (size == 0) { connection_del(server->pool, c->fd); free(buffer); - skynet_send(server->ctx, NULL, c->addr, SESSION_CLIENT, NULL, 0, DONTCOPY); + skynet_send(server->ctx, 0, c->address, SESSION_CLIENT, NULL, 0, DONTCOPY); } else { - skynet_send(server->ctx, NULL, c->addr, SESSION_CLIENT, buffer, size, DONTCOPY); + skynet_send(server->ctx, 0, c->address, SESSION_CLIENT, buffer, size, DONTCOPY); } } } static int -_main(struct skynet_context * ctx, void * ud, int session, const char * uid, const void * msg, size_t sz) { +_main(struct skynet_context * ctx, void * ud, int session, uint32_t source, const void * msg, size_t sz) { if (msg == NULL) { _poll(ud); return 0; @@ -135,24 +130,29 @@ _main(struct skynet_context * ctx, void * ud, int session, const char * uid, con char * endptr; int fd = strtol(param, &endptr, 10); if (endptr == NULL) { - skynet_error(ctx, "[connection] Invalid ADD command from %s (session = %d)", uid, session); + skynet_error(ctx, "[connection] Invalid ADD command from %x (session = %d)", source, session); return 0; } int addr_sz = sz - (endptr - (char *)msg); - char * addr = malloc(addr_sz); + char addr [addr_sz]; memcpy(addr, endptr+1, addr_sz-1); addr[addr_sz-1] = '\0'; - _add(ud, fd, addr); + uint32_t address = strtoul(addr, NULL, 16); + if (address != 0) { + skynet_error(ctx, "[connection] Invalid ADD command from %x (session = %d)", source, session); + return 0; + } + _add(ud, fd, address); } else if (memcmp(msg, "DEL ", 4)==0) { char * endptr; int fd = strtol(param, &endptr, 10); if (endptr == NULL) { - skynet_error(ctx, "[connection] Invalid DEL command from %s (session = %d)", uid, session); + skynet_error(ctx, "[connection] Invalid DEL command from %x (session = %d)", source, session); return 0; } _del(ud, fd); } else { - skynet_error(ctx, "[connection] Invalid command from %s (session = %d)", uid, session); + skynet_error(ctx, "[connection] Invalid command from %x (session = %d)", source, session); } return 0; diff --git a/gate/main.c b/gate/main.c index 39f9b92f..f262a2e8 100644 --- a/gate/main.c +++ b/gate/main.c @@ -11,16 +11,16 @@ #include struct connection { - char * agent; - char * client; + uint32_t agent; + uint32_t client; int connection_id; int uid; }; struct gate { struct mread_pool * pool; - const char * watchdog; - const char * broker; + uint32_t watchdog; + uint32_t broker; int id_index; int cap; int max_connection; @@ -31,10 +31,7 @@ struct gate { struct gate * gate_create(void) { struct gate * g = malloc(sizeof(*g)); - g->pool = NULL; - g->max_connection = 0; - g->agent = NULL; - g->broker = NULL; + memset(g,0,sizeof(*g)); return g; } @@ -58,16 +55,10 @@ _parm(char *msg, int sz, int command_sz) { } static void -_forward_agent(struct gate * g, int id, const char * agentaddr, const char *clientaddr) { +_forward_agent(struct gate * g, int id, uint32_t agentaddr, uint32_t clientaddr) { struct connection * agent = _id_to_agent(g,id); - if (agent->agent) { - free(agent->agent); - } - agent->agent = strdup(agentaddr); - if (agent->client) { - free(agent->client); - } - agent->client = strdup(clientaddr); + agent->agent = agentaddr; + agent->client = clientaddr; } static void @@ -105,12 +96,14 @@ _ctrl(struct skynet_context * ctx, struct gate * g, const void * msg, int sz) { if (client == NULL) { return; } - _forward_agent(g, id, agent, client); + uint32_t agent_handle = strtoul(agent+1, NULL, 16); + uint32_t client_handle = strtoul(client+1, NULL, 16); + _forward_agent(g, id, agent_handle, client_handle); return; } if (memcmp(command,"broker",i)==0) { _parm(tmp, sz, i); - g->broker = strdup(command); + g->broker = skynet_queryname(ctx, command); return; } if (memcmp(command,"start",i) == 0) { @@ -122,7 +115,7 @@ _ctrl(struct skynet_context * ctx, struct gate * g, const void * msg, int sz) { static void _report(struct gate *g, struct skynet_context * ctx, const char * data, ...) { - if (g->watchdog == NULL) { + if (g->watchdog == 0) { return; } va_list ap; @@ -131,13 +124,13 @@ _report(struct gate *g, struct skynet_context * ctx, const char * data, ...) { int n = vsnprintf(tmp, sizeof(tmp), data, ap); va_end(ap); - skynet_send(ctx, NULL, g->watchdog, 0, tmp, n, 0); + skynet_send(ctx, 0, g->watchdog, 0, tmp, n, 0); } static void _forward(struct skynet_context * ctx,struct gate *g, int uid, void * data, size_t len) { if (g->broker) { - skynet_send(ctx, NULL, g->broker, SESSION_CLIENT, data, len, 0); + skynet_send(ctx, 0, g->broker, SESSION_CLIENT, data, len, 0); return; } struct connection * agent = _id_to_agent(g,uid); @@ -148,7 +141,7 @@ _forward(struct skynet_context * ctx,struct gate *g, int uid, void * data, size_ char * tmp = malloc(len + 32); int n = snprintf(tmp,len+32,"%d data ",uid); memcpy(tmp+n,data,len); - skynet_send(ctx, NULL, g->watchdog, 0, tmp, len + n, DONTCOPY); + skynet_send(ctx, 0, g->watchdog, 0, tmp, len + n, DONTCOPY); } } @@ -175,14 +168,11 @@ _remove_id(struct gate *g, int uid) { struct connection * conn = _id_to_agent(g,uid); assert(conn->uid == uid); conn->uid = 0; - if (conn->agent) { - free(conn->agent); - conn->agent = NULL; - } + conn->agent = 0; } static int -_cb(struct skynet_context * ctx, void * ud, int session, const char * uid, const void * msg, size_t sz) { +_cb(struct skynet_context * ctx, void * ud, int session, uint32_t source, const void * msg, size_t sz) { struct gate *g = ud; if (msg) { _ctrl(ctx, g , msg , (int)sz); @@ -265,10 +255,15 @@ gate_init(struct gate *g , struct skynet_context * ctx, char * parm) { return 1; } if (watchdog[0] == '!') { - g->watchdog = NULL; + g->watchdog = 0; } else { - g->watchdog = strdup(watchdog); + g->watchdog = skynet_queryname(ctx, watchdog); + if (g->watchdog == 0) { + skynet_error(ctx, "Invalid watchdog %s",watchdog); + return 1; + } } + g->pool = pool; int cap = 1; while (cap < max) { diff --git a/gate/mread.c b/gate/mread.c index be70c23d..0645ec37 100644 --- a/gate/mread.c +++ b/gate/mread.c @@ -110,7 +110,7 @@ mread_create(uint32_t addr, int port , int max , int buffer_size) { my_addr.sin_family = AF_INET; my_addr.sin_port = htons(port); my_addr.sin_addr.s_addr = addr; - printf("MREAD bind %s:%u\n",inet_ntoa(my_addr.sin_addr),ntohs(my_addr.sin_port)); +// printf("MREAD bind %s:%u\n",inet_ntoa(my_addr.sin_addr),ntohs(my_addr.sin_port)); if (bind(listen_fd, (struct sockaddr *)&my_addr, sizeof(struct sockaddr)) == -1) { close(listen_fd); return NULL; diff --git a/lualib-src/lua-remoteobj.c b/lualib-src/lua-remoteobj.c index 58707c8f..5a551173 100644 --- a/lualib-src/lua-remoteobj.c +++ b/lualib-src/lua-remoteobj.c @@ -78,27 +78,6 @@ _query(struct remote_objects *r, int handle) { return 0; } -static uint32_t -_getaddr(lua_State *L, int index) { - size_t sz; - const char * addr = luaL_checklstring(L,index,&sz); - if (addr[0] != ':') { - luaL_error(L, "Invalid address %s",addr); - } - return strtoul(addr+1, NULL, 16); -} - -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'; -} - static int lbind(lua_State *L) { uint32_t addr = lua_tounsigned(L,lua_upvalueindex(1)); @@ -107,9 +86,7 @@ lbind(lua_State *L) { if (old == 0) { luaL_error(L, "handle %d is not exist", handle); } - char tmp[10]; - _id_to_hex(tmp, old); - lua_pushlstring(L,tmp,9); + lua_pushnumber(L, old); return 1; } @@ -138,9 +115,7 @@ lquery(lua_State *L) { if (addr == 0) { return 0; } - char tmp[10]; - _id_to_hex(tmp, addr); - lua_pushlstring(L,tmp,9); + lua_pushnumber(L, addr); return 1; } @@ -153,7 +128,7 @@ remoteobj_init(lua_State *L) { } } - uint32_t address = _getaddr(L, -1); + uint32_t address = luaL_checkunsigned(L, -1); lua_pushcfunction(L, lquery); lua_pushnumber(L, address); diff --git a/lualib-src/lua-skynet.c b/lualib-src/lua-skynet.c index 41a9d051..6b6f767d 100644 --- a/lualib-src/lua-skynet.c +++ b/lualib-src/lua-skynet.c @@ -8,7 +8,7 @@ #include static int -_cb(struct skynet_context * context, void * ud, int session, const char * addr, const void * msg, size_t sz) { +_cb(struct skynet_context * context, void * ud, int session, uint32_t source, const void * msg, size_t sz) { lua_State *L = ud; int trace = 1; int top = lua_gettop(L); @@ -21,17 +21,17 @@ _cb(struct skynet_context * context, void * ud, int session, const char * addr, int r; if (msg == NULL) { - if (addr == NULL) { + if (source == 0) { lua_pushinteger(L, session); r = lua_pcall(L, 1, 0 , trace); } else { lua_pushinteger(L, session); - lua_pushstring(L, addr); + lua_pushnumber(L, source); r = lua_pcall(L, 2, 0 , trace); } } else { lua_pushinteger(L, session); - lua_pushstring(L, addr); + lua_pushnumber(L, source); lua_pushlightuserdata(L,(void *)msg); lua_pushinteger(L,sz); r = lua_pcall(L, 4, 0 , trace); @@ -41,16 +41,16 @@ _cb(struct skynet_context * context, void * ud, int session, const char * addr, const char * self = skynet_command(context, "REG", NULL); switch (r) { case LUA_ERRRUN: - skynet_error(context, "lua call [%s to %s : %d msgsz = %d] error : %s", addr , self, session, sz, lua_tostring(L,-1)); + skynet_error(context, "lua call [%x to %s : %d msgsz = %d] error : %s", source , self, session, sz, lua_tostring(L,-1)); break; case LUA_ERRMEM: - skynet_error(context, "lua memory error : [%s to %s : %d]", addr , self, session); + skynet_error(context, "lua memory error : [%x to %s : %d]", source , self, session); break; case LUA_ERRERR: - skynet_error(context, "lua error in error : [%s to %s : %d]", addr , self, session); + skynet_error(context, "lua error in error : [%x to %s : %d]", source , self, session); break; case LUA_ERRGCMM: - skynet_error(context, "lua gc error : [%s to %s : %d]", addr , self, session); + skynet_error(context, "lua gc error : [%x to %s : %d]", source , self, session); break; }; @@ -98,37 +98,87 @@ _command(lua_State *L) { static int _genid(lua_State *L) { struct skynet_context * context = lua_touserdata(L, lua_upvalueindex(1)); - int session = skynet_send(context, NULL, NULL, -1, NULL, 0 , 0); + int session = skynet_send(context, 0, 0, -1, NULL, 0 , 0); lua_pushinteger(L, session); return 1; } +static int +_sendname(lua_State *L, struct skynet_context * context, const char * dest) { + int session = 0; + int index = 0; + + if (lua_type(L,2) == LUA_TNUMBER) { + session = lua_tointeger(L,2); + ++index; + } + if (lua_gettop(L) == index + 1) { + session = skynet_sendname(context, dest, session , NULL, 0, 0); + } else { + int type = lua_type(L,index+2); + if (type == LUA_TSTRING) { + size_t len = 0; + void * msg = (void *)lua_tolstring(L,index+2,&len); + session = skynet_sendname(context, dest, session , msg, len, 0); + } else if (type == LUA_TNIL) { + session = skynet_sendname(context, dest, session , NULL, 0, 0); + } else { + luaL_checktype(L,index+2, LUA_TLIGHTUSERDATA); + void * msg = lua_touserdata(L,index+2); + int size = luaL_checkinteger(L,index+3); + session = skynet_sendname(context, dest, session, msg, size, DONTCOPY); + } + } + lua_pushinteger(L,session); + return 1; +} + static int _send(lua_State *L) { struct skynet_context * context = lua_touserdata(L, lua_upvalueindex(1)); int session = 0; - int index = 0; - const char * dest = luaL_checkstring(L,1); + int addr_type = lua_type(L,1); + uint32_t dest = 0; + switch(addr_type) { + case LUA_TNUMBER: + dest = lua_tounsigned(L,1); + break; + case LUA_TSTRING: { + const char * addrname = lua_tostring(L,1); + if (addrname[0] == '.' || addrname[0] == ':') { + dest = skynet_queryname(context, addrname); + if (dest == 0) { + luaL_error(L, "Invalid name %s", addrname); + } + } else { + return _sendname(L, context, addrname); + } + break; + } + default: + return luaL_error(L, "address must be number or string, got %s",lua_typename(L,addr_type)); + } + int index = 0; if (lua_type(L,2) == LUA_TNUMBER) { session = lua_tointeger(L,2); ++index; } if (lua_gettop(L) == index + 1) { - session = skynet_send(context, NULL, dest, session , NULL, 0, 0); + session = skynet_send(context, 0, dest, session , NULL, 0, 0); } else { int type = lua_type(L,index+2); if (type == LUA_TSTRING) { size_t len = 0; void * msg = (void *)lua_tolstring(L,index+2,&len); - session = skynet_send(context, NULL, dest, session , msg, len, 0); + session = skynet_send(context, 0, dest, session , msg, len, 0); } else if (type == LUA_TNIL) { - session = skynet_send(context, NULL, dest, session , NULL, 0, 0); + session = skynet_send(context, 0, dest, session , NULL, 0, 0); } else { luaL_checktype(L,index+2, LUA_TLIGHTUSERDATA); void * msg = lua_touserdata(L,index+2); int size = luaL_checkinteger(L,index+3); - session = skynet_send(context, NULL, dest, session, msg, size, DONTCOPY); + session = skynet_send(context, 0, dest, session, msg, size, DONTCOPY); } } if (session < 0) { @@ -141,8 +191,8 @@ _send(lua_State *L) { static int _redirect(lua_State *L) { struct skynet_context * context = lua_touserdata(L, lua_upvalueindex(1)); - const char * dest = luaL_checkstring(L,1); - const char * source = luaL_checkstring(L,2); + uint32_t dest = luaL_checkunsigned(L,1); + uint32_t source = luaL_checkunsigned(L,2); int session = luaL_checkinteger(L,3); if (lua_gettop(L) == 3) { diff --git a/lualib/skynet.lua b/lualib/skynet.lua index 248d63d5..f1def36a 100644 --- a/lualib/skynet.lua +++ b/lualib/skynet.lua @@ -85,19 +85,31 @@ function skynet.yield() end function skynet.register(name) - return c.command("REG", name) + c.command("REG", name) end function skynet.name(name, handle) c.command("NAME", name .. " " .. handle) end +local function string_to_handle(str) + return tonumber("0x" .. string.sub(str , 2)) +end + +local self_handle function skynet.self() - return c.command("REG") + if self_handle then + return self_handle + end + self_handle = string_to_handle(c.command("REG")) + return self_handle end function skynet.launch(...) - return c.command("LAUNCH", table.concat({...}," ")) + local addr = c.command("LAUNCH", table.concat({...}," ")) + if addr then + return string_to_handle(addr) + end end function skynet.now() @@ -232,6 +244,10 @@ function skynet.query_group(handle) return c.command("GROUP","QUERY " .. tostring(handle)) end +function skynet.address(addr) + return string.format(":%x",addr) +end + ------ remote object -------- do diff --git a/service-src/service_broker.c b/service-src/service_broker.c index 56bbffe4..86c47c1e 100644 --- a/service-src/service_broker.c +++ b/service-src/service_broker.c @@ -10,12 +10,13 @@ struct worker { int init; - char * name; + uint32_t address; }; struct broker { int init; int id; + uint32_t launcher; char * name; struct worker w[DEFAULT_NUMBER]; }; @@ -29,41 +30,42 @@ broker_create(void) { void broker_release(struct broker * b) { - int i; - for (i=0;iw[i].name); - } free(b->name); free(b); } static void -_init(struct broker *b, int session, const char * msg, size_t sz) { +_init(struct broker *b, int session, uint32_t address) { assert(session > 0 && session <= DEFAULT_NUMBER); - assert(msg); int id = session - 1; assert(b->w[id].init == 0); - b->w[id].name = malloc(sz+1); - memcpy(b->w[id].name, msg, sz); - b->w[id].name[sz] = '\0'; + b->w[id].address = address; b->w[id].init = 1; ++b->init; } static void _forward(struct broker *b, struct skynet_context * context) { - skynet_forward(context, b->w[b->id].name); + skynet_forward(context, b->w[b->id].address); b->id = (b->id + 1) % DEFAULT_NUMBER; } static int -_cb(struct skynet_context * context, void * ud, int session, const char * addr, const void * msg, size_t sz) { +_cb(struct skynet_context * context, void * ud, int session, uint32_t source, const void * msg, size_t sz) { struct broker * b = ud; if (b->init < DEFAULT_NUMBER) { - _init(b, session, msg, sz); + if (source != b->launcher) + return 0; + assert(sz == 9); + char addr[10]; + memcpy(addr, msg, 9); + addr[9] = '\0'; + uint32_t address = strtoul(addr+1, NULL, 16); + assert(address != 0); + _init(b, session, address); if (b->init == DEFAULT_NUMBER) { skynet_command(context, "REG", b->name); - skynet_send(context, NULL, LAUNCHER, 0, NULL, 0, 0); + skynet_send(context, 0, b->launcher, 0, NULL, 0, 0); } } else { _forward(b, context); @@ -75,6 +77,12 @@ _cb(struct skynet_context * context, void * ud, int session, const char * addr, int broker_init(struct broker *b, struct skynet_context *ctx, const char * args) { + b->launcher = skynet_queryname(ctx, LAUNCHER); + if (b->launcher == 0) { + skynet_error(ctx, "Can't query %s", LAUNCHER); + return 1; + } + char * service = strchr(args,' '); if (service == NULL) { return 1; @@ -92,7 +100,7 @@ broker_init(struct broker *b, struct skynet_context *ctx, const char * args) { if (len == 0) return 1; for (i=0;ilauncher , -1, service , len, 0); assert(id > 0 && id <= DEFAULT_NUMBER); } diff --git a/service-src/service_client.c b/service-src/service_client.c index b11d5198..b4b6c929 100644 --- a/service-src/service_client.c +++ b/service-src/service_client.c @@ -9,7 +9,7 @@ #include static int -_cb(struct skynet_context * context, void * ud, int session, const char * addr, const void * msg, size_t sz) { +_cb(struct skynet_context * context, void * ud, int session, uint32_t source, const void * msg, size_t sz) { assert(sz <= 65535); int fd = (int)(intptr_t)ud; @@ -31,7 +31,7 @@ _cb(struct skynet_context * context, void * ud, int session, const char * addr, } } if (err < 0) { - skynet_error(context, "Client socket error : Drop message from %s session = %d", addr, session); + skynet_error(context, "Client socket error : Drop message from %x session = %d", source, session); return 0; } assert(err == sz +2); diff --git a/service-src/service_harbor.c b/service-src/service_harbor.c index 71a710a0..38184acb 100644 --- a/service-src/service_harbor.c +++ b/service-src/service_harbor.c @@ -291,17 +291,22 @@ _send_package(int fd, const void * buffer, size_t sz) { static int _send_remote(int fd, const char * buffer, size_t sz, struct remote_message_header * cookie) { - struct iovec part[2]; - part[0].iov_base = (char *)buffer; - part[0].iov_len = sz; + uint16_t sz_header = htons(sz+sizeof(*cookie)); + struct iovec part[3]; + + part[0].iov_base = &sz_header; + part[0].iov_len = 2; + + part[1].iov_base = (char *)buffer; + part[1].iov_len = sz; uint32_t header[3]; _header_to_message(cookie, header); - part[1].iov_base = header; - part[1].iov_len = sizeof(header); + part[2].iov_base = header; + part[2].iov_len = sizeof(header); for (;;) { - int err = writev(fd, part, 2); + int err = writev(fd, part, 3); if (err < 0) { switch (errno) { case EAGAIN: @@ -309,7 +314,7 @@ _send_remote(int fd, const char * buffer, size_t sz, struct remote_message_heade continue; } } - if (err != sz+sizeof(*cookie)) { + if (err != sz+sizeof(*cookie)+2) { return 1; } return 0; @@ -405,28 +410,13 @@ _request_master(struct harbor *h, struct skynet_context * context, const char na n bytes string (name) */ -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'; -} - static int _remote_send_handle(struct harbor *h, struct skynet_context * context, uint32_t source, uint32_t destination, int session, const char * msg, size_t sz) { int harbor_id = destination >> HANDLE_REMOTE_SHIFT; assert(harbor_id != 0); if (harbor_id == h->id) { // local message - char srcstr[10]; - char desstr[10]; - _id_to_hex(srcstr, source); - _id_to_hex(desstr, destination); - skynet_send(context, srcstr, desstr , session, (void *)msg, sz, DONTCOPY); + skynet_send(context, source, destination , session, (void *)msg, sz, DONTCOPY); return 1; } @@ -491,7 +481,7 @@ _report_local_address(struct harbor *h, struct skynet_context * context, const c } static int -_mainloop(struct skynet_context * context, void * ud, int session, const char * addr, const void * msg, size_t sz) { +_mainloop(struct skynet_context * context, void * ud, int session, uint32_t source, const void * msg, size_t sz) { struct harbor * h = ud; if (session == SESSION_CLIENT) { const char * cookie = msg; @@ -517,11 +507,7 @@ _mainloop(struct skynet_context * context, void * ud, int session, const char * _update_remote_name(h, context, msg, header.destination); } } else { - char srcstr[10]; - char desstr[10]; - _id_to_hex(srcstr, header.source); - _id_to_hex(desstr, header.destination); - skynet_send(context, srcstr, desstr, (int)header.session, (void *)msg, sz-12, DONTCOPY); + skynet_send(context, header.source, header.destination, (int)header.session, (void *)msg, sz-12, DONTCOPY); return 1; } } else { @@ -531,13 +517,12 @@ _mainloop(struct skynet_context * context, void * ud, int session, const char * return 0; } assert(sz == sizeof(*rmsg)); - uint32_t source_handle = strtoul(addr+1, NULL, 16); if (rmsg->destination.handle == 0) { - if (_remote_send_name(h, context, source_handle , rmsg->destination.name, session, rmsg->message, rmsg->sz)) { + if (_remote_send_name(h, context, source , rmsg->destination.name, session, rmsg->message, rmsg->sz)) { return 0; } } else { - if (_remote_send_handle(h, context, source_handle , rmsg->destination.handle, session, rmsg->message, rmsg->sz)) { + if (_remote_send_handle(h, context, source , rmsg->destination.handle, session, rmsg->message, rmsg->sz)) { return 0; } } @@ -563,7 +548,6 @@ harbor_init(struct harbor *h, struct skynet_context *ctx, const char * args) { h->master_addr = strdup(master_addr); h->master_fd = master_fd; - const char * self_addr = skynet_command(ctx, "REG", NULL); char tmp[128]; sprintf(tmp,"gate ! %s %d 0",local_addr,REMOTE_MAX); const char * gate_addr = skynet_command(ctx, "LAUNCH", tmp); @@ -571,9 +555,15 @@ harbor_init(struct harbor *h, struct skynet_context *ctx, const char * args) { skynet_error(ctx, "Harbor : launch gate failed"); return 1; } + uint32_t gate = strtoul(gate_addr+1 , NULL, 16); + if (gate == 0) { + skynet_error(ctx, "Harbor : launch gate invalid %s", gate_addr); + return 1; + } + const char * self_addr = skynet_command(ctx, "REG", NULL); int n = sprintf(tmp,"broker %s",self_addr); - skynet_send(ctx, NULL, gate_addr, 0, tmp, n, 0); - skynet_send(ctx, NULL, gate_addr, 0, "start", 5, 0); + skynet_send(ctx, 0, gate, 0, tmp, n, 0); + skynet_send(ctx, 0, gate, 0, "start", 5, 0); h->id = harbor_id; skynet_callback(ctx, h, _mainloop); diff --git a/service-src/service_master.c b/service-src/service_master.c index d9dca149..28b0a852 100644 --- a/service-src/service_master.c +++ b/service-src/service_master.c @@ -231,7 +231,7 @@ _update_address(struct skynet_context * context, struct master *m, int harbor_id */ static int -_mainloop(struct skynet_context * context, void * ud, int session, const char * addr, const void * msg, size_t sz) { +_mainloop(struct skynet_context * context, void * ud, int session, uint32_t source, const void * msg, size_t sz) { struct master *m = ud; assert(session == SESSION_CLIENT); uint32_t handle = 0; @@ -256,15 +256,20 @@ int master_init(struct master *m, struct skynet_context *ctx, const char * args) { char tmp[strlen(args) + 32]; sprintf(tmp,"gate ! %s %d 0",args,REMOTE_MAX); - const char * self_addr = skynet_command(ctx, "REG", NULL); const char * gate_addr = skynet_command(ctx, "LAUNCH", tmp); if (gate_addr == NULL) { skynet_error(ctx, "Master : launch gate failed"); return 1; } + uint32_t gate = strtoul(gate_addr+1, NULL, 16); + if (gate == 0) { + skynet_error(ctx, "Master : launch gate invalid %s", gate_addr); + return 1; + } + const char * self_addr = skynet_command(ctx, "REG", NULL); int n = sprintf(tmp,"broker %s",self_addr); - skynet_send(ctx, NULL, gate_addr, 0, tmp, n, 0); - skynet_send(ctx, NULL, gate_addr, 0, "start", 5, 0); + skynet_send(ctx, 0, gate, 0, tmp, n, 0); + skynet_send(ctx, 0, gate, 0, "start", 5, 0); skynet_callback(ctx, m, _mainloop); diff --git a/service-src/service_multicast.c b/service-src/service_multicast.c index 3d185871..a2155935 100644 --- a/service-src/service_multicast.c +++ b/service-src/service_multicast.c @@ -16,9 +16,8 @@ multicast_release(struct skynet_multicast_group *g) { } static int -_maincb(struct skynet_context * context, void * ud, int session, const char * addr, const void * msg, size_t sz) { +_maincb(struct skynet_context * context, void * ud, int session, uint32_t source, 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; diff --git a/service/globallog.lua b/service/globallog.lua index a01e5e51..f506b6e4 100644 --- a/service/globallog.lua +++ b/service/globallog.lua @@ -2,7 +2,7 @@ local skynet = require "skynet" skynet.dispatch(function(msg, sz , session , from) local message = skynet.tostring(msg,sz) - print("[GLOBALLOG]", from,message) + print("[GLOBALLOG]", skynet.address(from),message) end) skynet.register "LOG" \ No newline at end of file diff --git a/service/main.lua b/service/main.lua index 30a96e8b..00847b46 100644 --- a/service/main.lua +++ b/service/main.lua @@ -5,20 +5,12 @@ skynet.dispatch() skynet.start(function() print("Server start") local lualog = skynet.launch("snlua","lualog") - print("lualog",lualog) local launcher = skynet.launch("snlua","launcher") - print("launcher", launcher) local remoteroot = skynet.launch("snlua","remote_root") - print("remoteroot", remoteroot) local console = skynet.launch("snlua","console") - print("console",console) local watchdog = skynet.launch("snlua","watchdog","8888 4 0") - print("watchdog",watchdog) local db = skynet.launch("snlua","simpledb") - print("simpledb",db) local connection = skynet.launch("connection","256") - print("connection",connection) local redis = skynet.launch("snlua","redis-mgr") - print("redis",redis) skynet.exit() end) diff --git a/service/main_log.lua b/service/main_log.lua index 976e0f49..408f7323 100644 --- a/service/main_log.lua +++ b/service/main_log.lua @@ -2,7 +2,6 @@ local skynet = require "skynet" print("Log server start") -local log = skynet.launch("snlua","globallog") -print("log",log) +skynet.launch("snlua","globallog") skynet.exit() diff --git a/service/simpledb.lua b/service/simpledb.lua index 28229285..9a8be1d1 100644 --- a/service/simpledb.lua +++ b/service/simpledb.lua @@ -15,7 +15,7 @@ end skynet.dispatch(function(msg, sz , session, from) local message = skynet.tostring(msg,sz) - print("simpledb",message, from, session) + print("simpledb",message, skynet.address(from), session) local cmd, key , value = string.match(message, "(%w+) (%w+) ?(.*)") local f = command[cmd] if f then diff --git a/service/watchdog.lua b/service/watchdog.lua index 29bf728f..707d3d56 100644 --- a/service/watchdog.lua +++ b/service/watchdog.lua @@ -10,12 +10,10 @@ function command:open(parm) fd = tonumber(fd) print("agent open",self,string.format("%d %d %s",self,fd,addr)) local client = skynet.launch("client",fd) - print("client",client) - local agent = skynet.launch("snlua","agent",client) - print("watchdog launch agent client:",agent,client) + local agent = skynet.launch("snlua","agent",skynet.address(client)) if agent then agent_all[self] = { agent , client } - skynet.send(gate, "forward ".. self .. " " .. agent .. " " .. client) + skynet.send(gate, "forward ".. self .. " " .. skynet.address(agent) .. " " .. skynet.address(client)) end end @@ -50,8 +48,7 @@ skynet.dispatch(function(msg, sz, session, address) end) skynet.start(function() - gate = skynet.launch("gate" , skynet.self(), port, max_agent, buffer) - print("gate = ", gate) + gate = skynet.launch("gate" , skynet.address(skynet.self()), port, max_agent, buffer) skynet.send(gate,"start") skynet.register(".watchdog") end) diff --git a/skynet-src/skynet.h b/skynet-src/skynet.h index 58392679..48c2fffa 100644 --- a/skynet-src/skynet.h +++ b/skynet-src/skynet.h @@ -13,11 +13,13 @@ struct skynet_context; void skynet_error(struct skynet_context * context, const char *msg, ...); const char * skynet_command(struct skynet_context * context, const char * cmd , const char * parm); -int skynet_send(struct skynet_context * context, const char * source, const char * addr , int session, void * msg, size_t sz, int flags); +uint32_t skynet_queryname(struct skynet_context * context, const char * name); +int skynet_send(struct skynet_context * context, uint32_t source, uint32_t destination , int session, void * msg, size_t sz, int flags); +int skynet_sendname(struct skynet_context * context, const char * destination , int session, void * msg, size_t sz, int flags); -void skynet_forward(struct skynet_context *, const char * destination); +void skynet_forward(struct skynet_context *, uint32_t destination); -typedef int (*skynet_cb)(struct skynet_context * context, void *ud, int session, const char * addr , const void * msg, size_t sz); +typedef int (*skynet_cb)(struct skynet_context * context, void *ud, int session, uint32_t source , const void * msg, size_t sz); void skynet_callback(struct skynet_context * context, void *ud, skynet_cb cb); #endif diff --git a/skynet-src/skynet_logger.c b/skynet-src/skynet_logger.c index 4df320be..56607eb7 100644 --- a/skynet-src/skynet_logger.c +++ b/skynet-src/skynet_logger.c @@ -2,6 +2,7 @@ #include #include +#include struct logger { FILE * handle; @@ -25,9 +26,9 @@ logger_release(struct logger * inst) { } static int -_logger(struct skynet_context * context, void *ud, int session, const char * uid, const void * msg, size_t sz) { +_logger(struct skynet_context * context, void *ud, int session, uint32_t source, const void * msg, size_t sz) { struct logger * inst = ud; - fprintf(inst->handle, "[%s] ",uid); + fprintf(inst->handle, "[%x] ",source); fwrite(msg, sz , 1, inst->handle); fprintf(inst->handle, "\n"); diff --git a/skynet-src/skynet_multicast.c b/skynet-src/skynet_multicast.c index 749cedae..9f462b1d 100644 --- a/skynet-src/skynet_multicast.c +++ b/skynet-src/skynet_multicast.c @@ -10,27 +10,16 @@ struct skynet_multicast_message { int ref; const void * msg; size_t sz; - char source[10]; + uint32_t source; }; -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); + mc->source = source; return mc; } diff --git a/skynet-src/skynet_multicast.h b/skynet-src/skynet_multicast.h index 17c8c5ce..1519688b 100644 --- a/skynet-src/skynet_multicast.h +++ b/skynet-src/skynet_multicast.h @@ -2,12 +2,13 @@ #define SKYNET_MULTICAST_H #include +#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); +typedef void (*skynet_multicast_func)(void *ud, uint32_t 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); diff --git a/skynet-src/skynet_server.c b/skynet-src/skynet_server.c index ac150636..7aaacbdb 100644 --- a/skynet-src/skynet_server.c +++ b/skynet-src/skynet_server.c @@ -38,14 +38,12 @@ struct skynet_context { struct skynet_module * mod; uint32_t handle; int ref; - char handle_name[10]; char result[32]; void * cb_ud; skynet_cb cb; int session_id; int init; uint32_t forward; - char forward_address[GLOBALNAME_LENGTH]; struct message_queue *queue; CHECKCALLING_DECL @@ -83,11 +81,8 @@ skynet_context_new(const char * name, const char *param) { ctx->session_id = 0; ctx->forward = 0; - ctx->forward_address[0] = '\0'; ctx->init = 0; ctx->handle = skynet_handle_register(ctx); - char * uid = ctx->handle_name; - _id_to_hex(uid, ctx->handle); struct message_queue * queue = ctx->queue = skynet_mq_create(ctx->handle); // init function maybe use ctx->handle, so it must init at last @@ -178,21 +173,11 @@ _forwarding(struct skynet_context *ctx, struct skynet_message *msg) { } return 1; } - if (ctx->forward_address[0]) { - struct remote_message * rmsg = malloc(sizeof(*rmsg)); - memcpy(rmsg->destination.name, ctx->forward_address,GLOBALNAME_LENGTH); - rmsg->destination.handle = 0; - rmsg->message = msg->data; - rmsg->sz = msg->sz; - skynet_harbor_send(rmsg, msg->source, msg->session); - ctx->forward_address[0] = '\0'; - return 1; - } return 0; } static void -_mc(void *ud, const char * source, const void * msg, size_t sz) { +_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); } @@ -202,14 +187,12 @@ _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); + ctx->cb(ctx, ctx->cb_ud, msg->session, 0, 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); - int reserve = ctx->cb(ctx, ctx->cb_ud, msg->session, tmp, msg->data, msg->sz); + int reserve = ctx->cb(ctx, ctx->cb_ud, msg->session, msg->source, msg->data, msg->sz); reserve |= _forwarding(ctx, msg); if (!reserve) { free(msg->data); @@ -292,6 +275,18 @@ _group_command(struct skynet_context * ctx, const char * cmd, int handle) { return NULL; } +uint32_t +skynet_queryname(struct skynet_context * context, const char * name) { + switch(name[0]) { + case ':': + return strtoul(name+1,NULL,16); + case '.': + return skynet_handle_findname(name + 1); + } + skynet_error(context, "Don't support query global name %s",name); + return 0; +} + const char * skynet_command(struct skynet_context * context, const char * cmd , const char * param) { if (strcmp(cmd,"TIMEOUT") == 0) { @@ -305,15 +300,10 @@ skynet_command(struct skynet_context * context, const char * cmd , const char * return context->result; } - if (strcmp(cmd,"NOW") == 0) { - uint32_t ti = skynet_gettime(); - sprintf(context->result,"%u",ti); - return context->result; - } - if (strcmp(cmd,"REG") == 0) { if (param == NULL || param[0] == '\0') { - return context->handle_name; + sprintf(context->result, ":%x", context->handle); + return context->result; } else if (param[0] == '.') { return skynet_handle_namehandle(context->handle, param + 1); } else { @@ -349,6 +339,12 @@ skynet_command(struct skynet_context * context, const char * cmd , const char * return NULL; } + if (strcmp(cmd,"NOW") == 0) { + uint32_t ti = skynet_gettime(); + sprintf(context->result,"%u",ti); + return context->result; + } + if (strcmp(cmd,"EXIT") == 0) { skynet_handle_retire(context->handle); return NULL; @@ -382,6 +378,7 @@ skynet_command(struct skynet_context * context, const char * cmd , const char * return NULL; } else { _id_to_hex(context->result, inst->handle); + printf("launch %s : %x\n",param, inst->handle); return context->result; } } @@ -428,38 +425,62 @@ skynet_command(struct skynet_context * context, const char * cmd , const char * } void -skynet_forward(struct skynet_context * context, const char * addr) { - uint32_t des = 0; - assert(context->forward == 0 && context->forward_address[0] == '\0'); - if (addr[0] == ':') { - des = strtol(addr+1, NULL, 16); - assert(des != 0); - } else if (addr[0] == '.') { - des = skynet_handle_findname(addr + 1); - if (des == 0) { - skynet_error(context, "Drop message forward %s", addr); - return; - } - } else { - _copy_name(context->forward_address,addr); - return; - } - context->forward = des; +skynet_forward(struct skynet_context * context, uint32_t destination) { + assert(context->forward == 0); + context->forward = destination; } int -skynet_send(struct skynet_context * context, const char * source, const char * addr , int session, void * data, size_t sz, int flags) { +skynet_send(struct skynet_context * context, uint32_t source, uint32_t destination , int session, void * data, size_t sz, int flags) { int session_id = session; - uint32_t source_handle; - if (source == NULL) { - source_handle = context->handle; + if (source == 0) { + source = context->handle; if (session < 0) { session = skynet_context_newsession(context); session_id = - session; } + } + + char * msg; + if ((flags & DONTCOPY) || data == NULL) { + msg = data; } else { - assert (source[0] == ':'); - source_handle = strtoul(source+1, NULL, 16); + msg = malloc(sz+1); + memcpy(msg, data, sz); + msg[sz] = '\0'; + } + if (destination == 0) { + return session; + } + if (skynet_harbor_message_isremote(destination)) { + struct remote_message * rmsg = malloc(sizeof(*rmsg)); + rmsg->destination.handle = destination; + rmsg->message = msg; + rmsg->sz = sz; + skynet_harbor_send(rmsg, source, session_id); + } else { + struct skynet_message smsg; + smsg.source = source; + smsg.session = session_id; + smsg.data = msg; + smsg.sz = sz; + + if (skynet_context_push(destination, &smsg)) { + free(msg); + skynet_error(NULL, "Drop message from %x to %x (size=%d)", source, destination, (int)sz); + return -1; + } + } + return session; +} + +int +skynet_sendname(struct skynet_context * context, const char * addr , int session, void * data, size_t sz, int flags) { + int session_id = session; + uint32_t source_handle = context->handle; + if (session < 0) { + session = skynet_context_newsession(context); + session_id = - session; } char * msg;