lihao/skynet
1
0
Fork 0
mirror of https://github.com/cloudwu/skynet.git synced 2026-07-24 20:23:06 +00:00
Code Issues Packages Projects Releases Wiki Activity

change skynet_send api (use uint32_t instead of string)

Browse Source
This commit is contained in:
云风
2012-08-30 13:09:34 +08:00
parent e26fbd99d6
commit 1791becd09
21 changed files with 282 additions and 242 deletions
Download Patch File Download Diff File Expand all files Collapse all files

36
connection/main.c
View File

@@ -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;i<server->max_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;i<server->max_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;i<server->max_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;

55
gate/main.c
View File

@@ -11,16 +11,16 @@
#include <stdarg.h>
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) {

2
gate/mread.c
View File

@@ -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;

31
lualib-src/lua-remoteobj.c
View File

@@ -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);

84
lualib-src/lua-skynet.c
View File

@@ -8,7 +8,7 @@
#include <assert.h>
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) {

22
lualib/skynet.lua
View File

@@ -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

38
service-src/service_broker.c
View File

@@ -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;i<DEFAULT_NUMBER;i++) {
free(b->w[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;i<DEFAULT_NUMBER;i++) {
int id = skynet_send(ctx, NULL, LAUNCHER , -1, service , len, 0);
int id = skynet_send(ctx, 0, b->launcher , -1, service , len, 0);
assert(id > 0 && id <= DEFAULT_NUMBER);
}

4
service-src/service_client.c
View File

@@ -9,7 +9,7 @@
#include <string.h>
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);

60
service-src/service_harbor.c
View File

@@ -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);

13
service-src/service_master.c
View File

@@ -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);

3
service-src/service_multicast.c
View File

@@ -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;

2
service/globallog.lua
View File

@@ -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"

8
service/main.lua
View File

@@ -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)

3
service/main_log.lua
View File

@@ -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()

2
service/simpledb.lua
View File

@@ -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

9
service/watchdog.lua
View File

@@ -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)

8
skynet-src/skynet.h
View File

@@ -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

5
skynet-src/skynet_logger.c
View File

@@ -2,6 +2,7 @@
#include <stdio.h>
#include <stdlib.h>
#include <stdint.h>
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");

15
skynet-src/skynet_multicast.c
View File

@@ -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;
}

3
skynet-src/skynet_multicast.h
View File

@@ -2,12 +2,13 @@
#define SKYNET_MULTICAST_H
#include <stddef.h>
#include <stdint.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);
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);

121
skynet-src/skynet_server.c
View File

@@ -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;