lihao/skynet
1
0
Fork 0
mirror of https://github.com/cloudwu/skynet.git synced 2026-07-22 02:53:09 +00:00
Code Issues Packages Projects Releases Wiki Activity

skynet support session in message

Browse Source
This commit is contained in:
云风
2012-08-05 18:05:17 +08:00
parent 71702c47b3
commit c0e4bb462c
14 changed files with 102 additions and 56 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
agent.lua
View File

@@ -2,6 +2,6 @@ local skynet = require "skynet"
print("agent",...)
skynet.callback(function(addr, msg)
print("[agent]",addr,msg)
skynet.callback(function(session, addr, msg)
print("[agent]",session,addr,msg)
end)

4
console.lua
View File

@@ -8,7 +8,7 @@ skynet.callback(function()
else
print("Lauch:",handle)
end
skynet.command("TIMEOUT","0:0")
skynet.command("TIMEOUT", 0, "0")
end)
skynet.command("TIMEOUT","0:0")
skynet.command("TIMEOUT", 0 ,"0")

14
gate/main.c
View File

@@ -112,7 +112,7 @@ _report(struct skynet_context * ctx, const char * data, ...) {
tmp[n] = '\0';
}
skynet_send(ctx, WATCHDOG, strdup(tmp), n);
skynet_send(ctx, WATCHDOG, 0, strdup(tmp), n);
}
static void
@@ -121,7 +121,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, agent->agent ? agent->agent : WATCHDOG, tmp, len + n);
skynet_send(ctx, agent->agent ? agent->agent : WATCHDOG, 0, tmp, len + n);
}
static int
@@ -163,7 +163,7 @@ _remove_id(struct gate *g, int uid) {
}
static void
_cb(struct skynet_context * ctx, void * ud, const char * uid, const void * msg, size_t sz) {
_cb(struct skynet_context * ctx, void * ud, int session, const char * uid, const void * msg, size_t sz) {
struct gate *g = ud;
if (msg) {
_ctrl(ctx, g , msg , (int)sz);
@@ -172,7 +172,7 @@ _cb(struct skynet_context * ctx, void * ud, const char * uid, const void * msg,
struct mread_pool * m = g->pool;
int connection_id = mread_poll(m,100); // timeout : 100ms
if (connection_id < 0) {
skynet_command(ctx, "TIMEOUT","1:0");
skynet_command(ctx, "TIMEOUT", 0, "1");
} else {
int id = g->map[connection_id].uid;
if (id == 0) {
@@ -203,7 +203,7 @@ _cb(struct skynet_context * ctx, void * ud, const char * uid, const void * msg,
_forward(ctx, g, id, data, *plen);
mread_yield(m);
_break:
skynet_command(ctx, "TIMEOUT","0:0");
skynet_command(ctx, "TIMEOUT",0,"0");
}
}
@@ -243,8 +243,8 @@ gate_init(struct gate *g , struct skynet_context * ctx, char * parm) {
}
skynet_callback(ctx,g,_cb);
skynet_command(ctx,"REG","gate");
skynet_command(ctx,"TIMEOUT","0:0");
skynet_command(ctx,"REG",0,"gate");
skynet_command(ctx,"TIMEOUT",0, "0");
return 0;
}

4
globallog.lua
View File

@@ -1,7 +1,7 @@
local skynet = require "skynet"
skynet.callback(function(from , message)
print("[GLOBALLOG]",from,message)
skynet.callback(function(session, from , message)
print("[GLOBALLOG]",session, from,message)
end)
skynet.command("REG","LOG")

49
lua-skynet.c
View File

@@ -6,17 +6,18 @@
#include <string.h>
static void
_cb(struct skynet_context * context, void * ud, const char * addr, const void * msg, size_t sz_session) {
_cb(struct skynet_context * context, void * ud, int session, const char * addr, const void * msg, size_t sz) {
lua_State *L = ud;
lua_rawgetp(L, LUA_REGISTRYINDEX, _cb);
int r;
if (msg == NULL) {
lua_pushinteger(L, (int)sz_session);
lua_pushinteger(L, session);
r = lua_pcall(L, 1, 0 , 0);
} else {
lua_pushinteger(L, session);
lua_pushstring(L, addr);
lua_pushlstring(L, msg, sz_session);
r = lua_pcall(L, 2, 0 , 0);
lua_pushlstring(L, msg, sz);
r = lua_pcall(L, 3, 0 , 0);
}
if (r == LUA_OK)
return;
@@ -45,13 +46,24 @@ static int
_command(lua_State *L) {
struct skynet_context * context = lua_touserdata(L, lua_upvalueindex(1));
const char * cmd = luaL_checkstring(L,1);
const char * parm = NULL;
int session = 0;
const char * result;
if (lua_gettop(L) == 2) {
const char * parm = luaL_checkstring(L,2);
result = skynet_command(context, cmd, parm);
} else {
result = skynet_command(context, cmd, NULL);
int top = lua_gettop(L);
if (top == 2) {
if (lua_type(L,2) == LUA_TNUMBER) {
session = lua_tointeger(L,2);
} else {
parm = luaL_checkstring(L,2);
}
} else if (top == 3) {
session = lua_tointeger(L,2);
parm = luaL_checkstring(L,3);
} else if (top != 1) {
luaL_error(L, "skynet.command support only 3 parms (%d)",top);
}
result = skynet_command(context, cmd, session, parm);
if (result) {
lua_pushstring(L, result);
return 1;
@@ -62,21 +74,28 @@ _command(lua_State *L) {
static int
_send(lua_State *L) {
struct skynet_context * context = lua_touserdata(L, lua_upvalueindex(1));
int session = -1;
int index = 0;
const char * dest = luaL_checkstring(L,1);
int type = lua_type(L,2);
if (lua_type(L,2) == LUA_TNUMBER) {
session = lua_tointeger(L,2);
++index;
}
int type = lua_type(L,index+2);
if (type == LUA_TSTRING) {
size_t len = 0;
void * msg = (void *)lua_tolstring(L,2,&len);
void * msg = (void *)lua_tolstring(L,index+2,&len);
void * message = malloc(len);
memcpy(message, msg, len);
skynet_send(context, dest, message, len);
skynet_send(context, dest, session , message, len);
} else {
void * msg = lua_touserdata(L,2);
void * msg = lua_touserdata(L,index+2);
if (msg == NULL) {
return luaL_error(L, "skynet.send need userdata or string (%s)", lua_typename(L,type));
}
int size = luaL_checkinteger(L,3);
skynet_send(context, dest, msg, size);
int size = luaL_checkinteger(L,index+3);
skynet_send(context, dest, session, msg, size);
}
return 0;
}

6
skynet.h
View File

@@ -6,10 +6,10 @@
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);
void skynet_send(struct skynet_context * context, const char * addr , void * msg, size_t sz_session);
const char * skynet_command(struct skynet_context * context, const char * cmd , int session, const char * parm);
void skynet_send(struct skynet_context * context, const char * addr , int session, void * msg, size_t sz);
typedef void (*skynet_cb)(struct skynet_context * context, void *ud, const char * uid , const void * msg, size_t sz_session);
typedef void (*skynet_cb)(struct skynet_context * context, void *ud, int session, const char * addr , const void * msg, size_t sz);
void skynet_callback(struct skynet_context * context, void *ud, skynet_cb cb);
#endif

1
skynet_error.c
View File

@@ -38,6 +38,7 @@ skynet_error(struct skynet_context * context, const char *msg, ...) {
} else {
smsg.source = skynet_context_handle(context);
}
smsg.session = 0;
smsg.data = strdup(tmp);
smsg.sz = len;
skynet_context_push(logger, &smsg);

11
skynet_harbor.c
View File

@@ -32,6 +32,7 @@ struct hashmap {
struct remote_header {
uint32_t source;
uint32_t destination;
uint32_t session;
};
struct remote {
@@ -61,6 +62,7 @@ static inline void
buffer_to_remote_header(uint8_t *buffer, struct remote_header *header) {
header->source = buffer[0] | buffer[1] << 8 | buffer[2] << 16 | buffer[3] << 24;
header->destination = buffer[4] | buffer[5] << 8 | buffer[6] << 16 | buffer[7] << 24;
header->session = buffer[8] | buffer[9] << 8 | buffer[10] << 16 | buffer[11] << 24;
}
static inline void
@@ -73,6 +75,10 @@ remote_header_to_buffer(struct remote_header *header, uint8_t *buffer) {
buffer[5] = (header->destination >>8) & 0xff;
buffer[6] = (header->destination >>16) & 0xff;
buffer[7] = (header->destination >>24) & 0xff;
buffer[8] = (header->session) & 0xff;
buffer[9] = (header->session >>8) & 0xff;
buffer[10] = (header->session >>16) & 0xff;
buffer[11] = (header->session >>24) & 0xff;
}
static uint32_t
@@ -247,6 +253,7 @@ _register_name(const char *name, uint32_t addr) {
message.destination = addr;
message.message = msg;
skynet_remotemq_push(Z->queue, &message);
send_notice();
}
} else {
while (!skynet_mq_pop(queue, &msg)) {
@@ -424,7 +431,7 @@ remote_socket_send(void * socket, struct skynet_remote_message *msg) {
rh.source = msg->message.source;
rh.destination = msg->destination;
zmq_msg_t part;
zmq_msg_init_size(&part,8);
zmq_msg_init_size(&part,sizeof(struct remote_header));
uint8_t * buffer = zmq_msg_data(&part);
remote_header_to_buffer(&rh,buffer);
zmq_send(socket, &part, ZMQ_SNDMORE);
@@ -447,7 +454,7 @@ _remote_recv() {
int rc = zmq_recv(Z->zmq_local,&header,0);
_report_zmq_error(rc);
size_t s = zmq_msg_size(&header);
if (s!=8) {
if (s!=sizeof(struct remote_header)) {
// s should be 0
if (s>0) {
char tmp[s+1];

4
skynet_logger.c
View File

@@ -25,7 +25,7 @@ logger_release(struct logger * inst) {
}
static void
_logger(struct skynet_context * context, void *ud, const char * uid, const void * msg, size_t sz) {
_logger(struct skynet_context * context, void *ud, int session, const char * uid, const void * msg, size_t sz) {
struct logger * inst = ud;
fprintf(inst->handle, "[%s] ",uid);
fwrite(msg, sz , 1, inst->handle);
@@ -45,7 +45,7 @@ logger_init(struct logger * inst, struct skynet_context *ctx, const char * parm)
}
if (inst->handle) {
skynet_callback(ctx, inst, _logger);
skynet_command(ctx, "REG", ".logger");
skynet_command(ctx, "REG", 0, ".logger");
return 0;
}
return 1;

1
skynet_mq.h
View File

@@ -6,6 +6,7 @@
struct skynet_message {
uint32_t source;
int session;
void * data;
size_t sz;
};

34
skynet_server.c
View File

@@ -23,6 +23,7 @@ struct skynet_context {
char result[32];
void * cb_ud;
skynet_cb cb;
int session_id;
int in_global_queue;
struct message_queue *queue;
};
@@ -54,6 +55,7 @@ skynet_context_new(const char * name, const char *parm) {
ctx->cb = NULL;
ctx->cb_ud = NULL;
ctx->in_global_queue = 0;
ctx->session_id = 0;
char * uid = ctx->handle_name;
uid[0] = ':';
_id_to_hex(uid+1, ctx->handle);
@@ -72,6 +74,17 @@ skynet_context_new(const char * name, const char *parm) {
}
}
static int
_new_session(struct skynet_context *ctx) {
int session = ++ctx->session_id;
if (session < 0) {
ctx->session_id = 1;
return 1;
}
return session;
}
void
skynet_context_grab(struct skynet_context *ctx) {
__sync_add_and_fetch(&ctx->ref,1);
@@ -98,17 +111,17 @@ skynet_context_release(struct skynet_context *ctx) {
static void
_dispatch_message(struct skynet_context *ctx, struct skynet_message *msg) {
if (msg->source == SKYNET_SYSTEM_TIMER) {
ctx->cb(ctx, ctx->cb_ud, NULL, msg->data, msg->sz);
ctx->cb(ctx, ctx->cb_ud, msg->session, NULL, msg->data, msg->sz);
} else {
char tmp[10];
tmp[0] = ':';
_id_to_hex(tmp+1, msg->source);
if (skynet_harbor_message_isremote(msg->source)) {
void * data = skynet_harbor_message_open(msg);
ctx->cb(ctx, ctx->cb_ud, tmp, data, msg->sz);
ctx->cb(ctx, ctx->cb_ud, msg->session, tmp, data, msg->sz);
skynet_harbor_message_close(msg);
} else {
ctx->cb(ctx, ctx->cb_ud, tmp, msg->data, msg->sz);
ctx->cb(ctx, ctx->cb_ud, msg->session, tmp, msg->data, msg->sz);
}
free(msg->data);
@@ -171,13 +184,10 @@ skynet_context_message_dispatch(void) {
}
const char *
skynet_command(struct skynet_context * context, const char * cmd , const char * parm) {
skynet_command(struct skynet_context * context, const char * cmd , int session, const char * parm) {
if (strcmp(cmd,"TIMEOUT") == 0) {
char * session_ptr = NULL;
int ti = strtol(parm, &session_ptr, 10);
char sep = session_ptr[0];
assert(sep == ':');
int session = strtol(session_ptr+1, NULL, 10);
skynet_timeout(context->handle, ti, session);
return NULL;
}
@@ -226,7 +236,10 @@ skynet_command(struct skynet_context * context, const char * cmd , const char *
}
void
skynet_send(struct skynet_context * context, const char * addr , void * msg, size_t sz) {
skynet_send(struct skynet_context * context, const char * addr , int session, void * msg, size_t sz) {
if (session < 0) {
session = _new_session(context);
}
uint32_t des = 0;
if (addr[0] == ':') {
des = strtol(addr+1, NULL, 16);
@@ -240,6 +253,7 @@ skynet_send(struct skynet_context * context, const char * addr , void * msg, siz
} else {
struct skynet_message smsg;
smsg.source = context->handle;
smsg.session = session;
smsg.data = msg;
smsg.sz = sz;
skynet_harbor_send(addr, 0, &smsg);
@@ -250,6 +264,7 @@ skynet_send(struct skynet_context * context, const char * addr , void * msg, siz
struct skynet_message smsg;
smsg.source = context->handle;
smsg.session = session;
smsg.data = msg;
smsg.sz = sz;
@@ -285,6 +300,9 @@ skynet_context_push(uint32_t handle, struct skynet_message *message) {
if (ctx == NULL) {
return -1;
}
if (message->session < 0) {
message->session = _new_session(ctx);
}
skynet_mq_push(ctx->queue, message);
if (__sync_lock_test_and_set(&ctx->in_global_queue,1) == 0) {
skynet_globalmq_push(ctx->queue);

6
skynet_timer.c
View File

@@ -110,8 +110,9 @@ timer_execute(struct timer *T)
struct timer_event * event = (struct timer_event *)(current+1);
struct skynet_message message;
message.source = SKYNET_SYSTEM_TIMER;
message.session = event->session;
message.data = NULL;
message.sz = (size_t) event->session;
message.sz = 0;
skynet_context_push(event->handle, &message);
@@ -175,8 +176,9 @@ skynet_timeout(uint32_t handle, int time, int session) {
if (time == 0) {
struct skynet_message message;
message.source = SKYNET_SYSTEM_TIMER;
message.session = session;
message.data = NULL;
message.sz = (size_t) session;
message.sz = 0;
if (skynet_context_push(handle, &message)) {
return;

10
testtimer.lua
View File

@@ -1,10 +1,8 @@
local skynet = require "skynet"
skynet.callback(function(addr,content)
if content == nil then
print("sn:",addr)
skynet.command("TIMEOUT","100:"..tostring(addr+1))
end
skynet.callback(function(session,addr,content)
print("sn:",session)
skynet.command("TIMEOUT", -1 , "100")
end)
skynet.command("TIMEOUT","0:0")
skynet.command("TIMEOUT",0,"0")

10
watchdog.lua
View File

@@ -5,22 +5,22 @@ local command = {}
function command:open(parm)
local fd,addr = string.match(parm,"(%d+) ([^%s]+)")
fd = tonumber(fd)
skynet.send("LOG",string.format("%d %d %s",self,fd,addr))
skynet.send("LOG", 0, string.format("%d %d %s",self,fd,addr))
local agent = skynet.command("LAUNCH","snlua agent.lua ".. self)
if agent then
skynet.send("gate","forward ".. self .. " " .. agent)
skynet.send("gate",0, "forward ".. self .. " " .. agent)
end
end
function command:close()
skynet.send("LOG",string.format("close %d",self))
skynet.send("LOG",0,string.format("close %d",self))
end
function command:data(data)
skynet.send("LOG",string.format("data %d size=%d",self,#data))
skynet.send("LOG",0,string.format("data %d size=%d",self,#data))
end
skynet.callback(function(from , message)
skynet.callback(function(session, from , message)
local id, cmd , parm = string.match(message, "(%d+) (%w+) ?(.*)")
id = tonumber(id)
local f = command[cmd]