From c0e4bb462cd437a0064f5991fe8000011f222472 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?=E4=BA=91=E9=A3=8E?= <cloudwu@gmail.com>
Date: Sun, 5 Aug 2012 18:05:17 +0800
Subject: [PATCH] skynet support session in message

---
 agent.lua       |  4 ++--
 console.lua     |  4 ++--
 gate/main.c     | 14 +++++++-------
 globallog.lua   |  4 ++--
 lua-skynet.c    | 49 ++++++++++++++++++++++++++++++++++---------------
 skynet.h        |  6 +++---
 skynet_error.c  |  1 +
 skynet_harbor.c | 11 +++++++++--
 skynet_logger.c |  4 ++--
 skynet_mq.h     |  1 +
 skynet_server.c | 34 ++++++++++++++++++++++++++--------
 skynet_timer.c  |  6 ++++--
 testtimer.lua   | 10 ++++------
 watchdog.lua    | 10 +++++-----
 14 files changed, 102 insertions(+), 56 deletions(-)

diff --git a/agent.lua b/agent.lua
index ed42070e..02f625d9 100644
--- a/agent.lua
+++ b/agent.lua
@@ -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)
diff --git a/console.lua b/console.lua
index 9931227e..15d26aea 100644
--- a/console.lua
+++ b/console.lua
@@ -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")
diff --git a/gate/main.c b/gate/main.c
index 41109cf8..9a75cdc9 100644
--- a/gate/main.c
+++ b/gate/main.c
@@ -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;
 }
diff --git a/globallog.lua b/globallog.lua
index 71f81cca..e023faee 100644
--- a/globallog.lua
+++ b/globallog.lua
@@ -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")
\ No newline at end of file
diff --git a/lua-skynet.c b/lua-skynet.c
index c7a78dfe..0bbc99ca 100644
--- a/lua-skynet.c
+++ b/lua-skynet.c
@@ -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;
 }
diff --git a/skynet.h b/skynet.h
index 12eb6d2b..0d9a8591 100644
--- a/skynet.h
+++ b/skynet.h
@@ -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
diff --git a/skynet_error.c b/skynet_error.c
index 3493d712..3d525c52 100644
--- a/skynet_error.c
+++ b/skynet_error.c
@@ -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);
diff --git a/skynet_harbor.c b/skynet_harbor.c
index 0438d857..9eaec98c 100644
--- a/skynet_harbor.c
+++ b/skynet_harbor.c
@@ -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];
diff --git a/skynet_logger.c b/skynet_logger.c
index 6919054b..0f11b91c 100644
--- a/skynet_logger.c
+++ b/skynet_logger.c
@@ -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;
diff --git a/skynet_mq.h b/skynet_mq.h
index a64ee205..acd222f1 100644
--- a/skynet_mq.h
+++ b/skynet_mq.h
@@ -6,6 +6,7 @@
 
 struct skynet_message {
 	uint32_t source;
+	int session;
 	void * data;
 	size_t sz;
 };
diff --git a/skynet_server.c b/skynet_server.c
index 8d8e64bd..95677821 100644
--- a/skynet_server.c
+++ b/skynet_server.c
@@ -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);
diff --git a/skynet_timer.c b/skynet_timer.c
index 37511a10..0050ccf0 100644
--- a/skynet_timer.c
+++ b/skynet_timer.c
@@ -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;
diff --git a/testtimer.lua b/testtimer.lua
index e95ee527..418c8c7a 100644
--- a/testtimer.lua
+++ b/testtimer.lua
@@ -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")
\ No newline at end of file
+skynet.command("TIMEOUT",0,"0")
\ No newline at end of file
diff --git a/watchdog.lua b/watchdog.lua
index 8d640f6f..f93b3bee 100644
--- a/watchdog.lua
+++ b/watchdog.lua
@@ -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]