simple rpc support

agent.lua

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

console.lua

@@ -1,14 +1,16 @@
 local skynet = require "skynet"
 
-skynet.callback(function()
-	local cmd = io.read()
-	local handle = skynet.command("LAUNCH",cmd)
-	if handle == nil then
-		print("Launch error:",cmd)
-	else
-		print("Lauch:",handle)
-	end
-	skynet.command("TIMEOUT", 0, "0")
-end)
+skynet.dispatch()
 
-skynet.command("TIMEOUT", 0 ,"0")
+skynet.start(function()
+	while true do
+		local cmd = io.read()
+		local handle = skynet.launch(cmd)
+		if handle == nil then
+			print("Launch error:",cmd)
+		else
+			print("Lauch:",handle)
+		end
+		skynet.yield()
+	end
+end)

globallog.lua

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

lua-skynet.c

@@ -5,19 +5,32 @@
 #include <stdlib.h>
 #include <string.h>
 
+static int 
+traceback (lua_State *L) {
+  const char *msg = lua_tostring(L, 1);
+  if (msg)
+    luaL_traceback(L, L, msg, 1);
+  else if (!lua_isnoneornil(L, 1)) {  /* is there an error object? */
+    if (!luaL_callmeta(L, 1, "__tostring"))  /* try its 'tostring' metamethod */
+      lua_pushliteral(L, "(no error message)");
+  }
+  return 1;
+}
+
 static void
 _cb(struct skynet_context * context, void * ud, int session, const char * addr, const void * msg, size_t sz) {
 	lua_State *L = ud;
+	lua_pushcfunction(L, traceback);
 	lua_rawgetp(L, LUA_REGISTRYINDEX, _cb);
 	int r;
 	if (msg == NULL) {
 		lua_pushinteger(L, session);
-		r = lua_pcall(L, 1, 0 , 0);
+		r = lua_pcall(L, 1, 0 , -3);
 	} else {
 		lua_pushinteger(L, session);
 		lua_pushstring(L, addr);
 		lua_pushlstring(L, msg, sz);
-		r = lua_pcall(L, 3, 0 , 0);
+		r = lua_pcall(L, 3, 0 , -5);
 	}
 	if (r == LUA_OK) 
 		return;
@@ -82,22 +95,32 @@ _send(lua_State *L) {
 		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,index+2,&len);
-		void * message = malloc(len);
-		memcpy(message, msg, len);
-		skynet_send(context, dest, session , message, len);
+	if (lua_gettop(L) == index + 1) {
+		session = skynet_send(context, dest, session , NULL, 0);
 	} else {
-		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 type = lua_type(L,index+2);
+		if (type == LUA_TSTRING) {
+			size_t len = 0;
+			void * msg = (void *)lua_tolstring(L,index+2,&len);
+			void * message = malloc(len);
+			memcpy(message, msg, len);
+			session = skynet_send(context, dest, session , message, len);
+		} else if (type == LUA_TNIL) {
+			session = skynet_send(context, dest, session , NULL, 0);
+		} else {
+			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,index+3);
+			session = skynet_send(context, dest, session, msg, size);
 		}
-		int size = luaL_checkinteger(L,index+3);
-		skynet_send(context, dest, session, msg, size);
 	}
-	return 0;
+	if (session < 0) {
+		return luaL_error(L, "skynet.send drop the message to %s", dest);
+	}
+	lua_pushinteger(L,session);
+	return 1;
 }
 
 static int
@@ -108,7 +131,7 @@ _error(lua_State *L) {
 }
 
 int
-luaopen_skynet(lua_State *L) {
+luaopen_skynet_c(lua_State *L) {
 	luaL_checkversion(L);
 	luaL_Reg l[] = {
 		{ "send" , _send },

main.lua

@@ -2,11 +2,11 @@ local skynet = require "skynet"
 
 print("Server start")
 
-local console = skynet.command("LAUNCH","snlua console.lua")
+local console = skynet.launch("snlua","console.lua")
 print("console",console)
-local watchdog = skynet.command("LAUNCH","snlua watchdog.lua")
+local watchdog = skynet.launch("snlua","watchdog.lua")
 print("watchdog",watchdog)
-local gate = skynet.command("LAUNCH","gate 8888 4 0")
+local gate = skynet.launch("gate","8888 4 0")
 print("gate",gate)
 
-skynet.command("EXIT")
+skynet.exit()

main_log.lua

@@ -2,7 +2,10 @@ local skynet = require "skynet"
 
 print("Log server start")
 
-local log = skynet.command("LAUNCH","snlua globallog.lua")
+local log = skynet.launch("snlua","globallog.lua")
 print("log",log)
 
-skynet.command("EXIT")
+local db = skynet.launch("snlua","simpledb.lua")
+print("simpledb",db)
+
+skynet.exit()

service_lua.c

@@ -19,6 +19,18 @@ _load(lua_State *L, char ** filename) {
 	return r != LUA_OK;
 }
 
+static int 
+traceback (lua_State *L) {
+  const char *msg = lua_tostring(L, 1);
+  if (msg)
+    luaL_traceback(L, L, msg, 1);
+  else if (!lua_isnoneornil(L, 1)) {  /* is there an error object? */
+    if (!luaL_callmeta(L, 1, "__tostring"))  /* try its 'tostring' metamethod */
+      lua_pushliteral(L, "(no error message)");
+  }
+  return 1;
+}
+
 int
 snlua_init(lua_State *L, struct skynet_context *ctx, const char * args) {
 	lua_gc(L, LUA_GCSTOP, 0);
@@ -31,6 +43,8 @@ snlua_init(lua_State *L, struct skynet_context *ctx, const char * args) {
 	char *parm = tmp;
 	strcpy(parm,args);
 
+	lua_pushcfunction(L, traceback);
+
 	const char * filename = parm;
 	int r = _load(L, &parm);
 	if (r) {
@@ -45,7 +59,7 @@ snlua_init(lua_State *L, struct skynet_context *ctx, const char * args) {
 			++n;
 		}
 	}
-	r = lua_pcall(L,n,0,0);
+	r = lua_pcall(L,n,0,-n-2);
 	switch (r) {
 	case LUA_OK:
 		return 0;

simpledb.lua

@@ -0,0 +1,22 @@
+local skynet = require "skynet"
+local db = {}
+
+local command = {}
+
+function command.GET(key)
+	skynet.ret(db[key])
+end
+
+function command.SET(key, value)
+	local last = db[key]
+	db[key] = value
+	skynet.ret(last)
+end
+
+skynet.dispatch(function(message, from, session)
+	print("simpledb",message, from, session)
+	local cmd, key , value = string.match(message, "(%w+) (%w+) ?(.*)")
+	command[cmd](key,value)
+end)
+
+skynet.register "SIMPLEDB"

skynet.h

@@ -2,12 +2,13 @@
 #define SKYNET_H
 
 #include <stddef.h>
+#include <stdint.h>
 
 struct skynet_context;
 
 void skynet_error(struct skynet_context * context, const char *msg, ...);
 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);
+int 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, int session, const char * addr , const void * msg, size_t sz);
 void skynet_callback(struct skynet_context * context, void *ud, skynet_cb cb);

skynet.lua

@@ -0,0 +1,100 @@
+local c = require "skynet.c"
+local tostring = tostring
+local tonumber = tonumber
+local coroutine = coroutine
+local assert = assert
+
+local skynet = {}
+local session_id_coroutine = {}
+local session_coroutine_id = {}
+local session_coroutine_address = {}
+
+local function suspend(co, result, command, param)
+	assert(result, command)
+	if command == "CALL" or command == "SLEEP" then
+		session_id_coroutine[param] = co
+	elseif command == "RETURN" then
+		local co_session = session_coroutine_id[co]
+		local co_address = session_coroutine_address[co]
+		c.send(co_address, co_session, param)
+	else
+		assert(command == nil, command)
+		session_coroutine_id[co] = nil
+		session_coroutine_address[co] = nil
+	end
+end
+
+function skynet.timeout(ti, func, ...)
+	local session = c.command("TIMEOUT",tostring(ti))
+	assert(session)
+	session = tonumber(session)
+	local co = coroutine.create(func)
+	assert(session_id_coroutine[session] == nil)
+	session_id_coroutine[session] = co
+	suspend(co, coroutine.resume(co, ...))
+end
+
+function skynet.yield()
+	local session = c.command("TIMEOUT","0")
+	coroutine.yield("SLEEP", tonumber(session))
+end
+
+function skynet.sleep(ti)
+	local session = c.command("TIMEOUT",tostring(ti))
+	assert(session)
+	coroutine.yield("SLEEP", tonumber(session))
+end
+
+function skynet.register(name)
+	return c.command("REG", name)
+end
+
+function skynet.self()
+	return c.command("REG")
+end
+
+function skynet.launch(...)
+	return c.command("LAUNCH", table.concat({...}," "))
+end
+
+function skynet.now()
+	return tonumber(c.command("NOW"))
+end
+
+function skynet.exit()
+	c.command("EXIT")
+end
+
+skynet.send = c.send
+
+function skynet.call(addr, message)
+	local session = c.send(addr, -1, message)
+	return coroutine.yield("CALL", session)
+end
+
+function skynet.ret(message)
+	coroutine.yield("RETURN", message)
+end
+
+function skynet.dispatch(f)
+	c.callback(function(session, address , message)
+		local co = session_id_coroutine[session]
+		if co == nil then
+			co = coroutine.create(f)
+			session_coroutine_id[co] = session
+			session_coroutine_address[co] = address
+			suspend(co, coroutine.resume(co, message, address, session))
+		else
+			session_id_coroutine[session] = nil
+			suspend(co, coroutine.resume(co, message))
+		end
+	end)
+end
+
+function skynet.start(f)
+	c.command("TIMEOUT",0,"0")
+	local co = coroutine.create(f)
+	session_id_coroutine[0] = co
+end
+
+return skynet

skynet_harbor.c

@@ -162,17 +162,11 @@ skynet_harbor_send(const char *name, uint32_t destination, struct skynet_message
 		assert(destination!=0);
 		int remote_id = destination >> HANDLE_REMOTE_SHIFT;
 		assert(remote_id > 0 && remote_id <= REMOTE_MAX);
-		--remote_id;
-		struct remote * r = &Z->remote[remote_id];
 		struct skynet_remote_message message;
 		message.destination = destination;
 		message.message = *msg;
-		if (r->socket) {
-			skynet_remotemq_push(Z->queue, &message);
-			send_notice();
-		} else {
-			skynet_remotemq_push(r->queue, &message);
-		}
+		skynet_remotemq_push(Z->queue, &message);
+		send_notice();
 	} else {
 		_lock();
 		struct keyvalue * node = _hash_search(Z->map, name);
@@ -454,6 +448,7 @@ remote_socket_send(void * socket, struct skynet_remote_message *msg) {
 	struct remote_header rh;
 	rh.source = msg->message.source;
 	rh.destination = msg->destination;
+	rh.session = msg->message.session;
 	zmq_msg_t part;
 	zmq_msg_init_size(&part,sizeof(struct remote_header));
 	uint8_t * buffer = zmq_msg_data(&part);
@@ -500,6 +495,7 @@ _remote_recv() {
 	_report_zmq_error(rc);
 
 	struct skynet_message msg;
+	msg.session = (int)rh.session;
 	msg.source = rh.source;
 	msg.data = data;
 	msg.sz = zmq_msg_size(data);

skynet_server.c

@@ -74,8 +74,8 @@ skynet_context_new(const char * name, const char *parm) {
 	}
 }
 
-static int
-_new_session(struct skynet_context *ctx) {
+int
+skynet_context_newsession(struct skynet_context *ctx) {
 	int session = ++ctx->session_id;
 	if (session < 0) {
 		ctx->session_id = 1;
@@ -151,7 +151,7 @@ skynet_context_message_dispatch(void) {
 
 	struct skynet_context * ctx = skynet_handle_grab(handle);
 	if (ctx == NULL) {
-		skynet_error(NULL, "Drop message queue %u ", handle);
+		skynet_error(NULL, "Drop message queue %x ", handle);
 		_drop_queue(q);
 		return 0;
 	}
@@ -177,7 +177,7 @@ skynet_context_message_dispatch(void) {
 			skynet_harbor_message_close(&msg);
 		}
 		free(msg.data);
-		skynet_error(NULL, "Drop message from %u to %u without callback , size = %d",msg.source, handle, (int)msg.sz);
+		skynet_error(NULL, "Drop message from %x to %x without callback , size = %d",msg.source, handle, (int)msg.sz);
 	} else {
 		_dispatch_message(ctx, &msg);
 	}
@@ -196,8 +196,11 @@ skynet_command(struct skynet_context * context, const char * cmd , int session,
 	if (strcmp(cmd,"TIMEOUT") == 0) {
 		char * session_ptr = NULL;
 		int ti = strtol(parm, &session_ptr, 10);
-		skynet_timeout(context->handle, ti, session);
-		return NULL;
+		session = skynet_timeout(context->handle, ti, session);
+		if (session < 0)
+			return NULL;
+		sprintf(context->result, "%d", session);
+		return context->result;
 	}
 
 	if (strcmp(cmd,"NOW") == 0) {
@@ -243,10 +246,10 @@ skynet_command(struct skynet_context * context, const char * cmd , int session,
 	return NULL;
 }
 
-void 
+int
 skynet_send(struct skynet_context * context, const char * addr , int session, void * msg, size_t sz) {
 	if (session < 0) {
-		session = _new_session(context);
+		session = skynet_context_newsession(context);
 	}
 	uint32_t des = 0;
 	if (addr[0] == ':') {
@@ -256,7 +259,7 @@ skynet_send(struct skynet_context * context, const char * addr , int session, vo
 		if (des == 0) {
 			free(msg);
 			skynet_error(context, "Drop message to %s, size = %d", addr, (int)sz);
-			return;
+			return session;
 		}
 	} else {
 		struct skynet_message smsg;
@@ -265,7 +268,7 @@ skynet_send(struct skynet_context * context, const char * addr , int session, vo
 		smsg.data = msg;
 		smsg.sz = sz;
 		skynet_harbor_send(addr, 0, &smsg);
-		return;
+		return session;
 	}
 
 	assert(des > 0);
@@ -280,9 +283,10 @@ skynet_send(struct skynet_context * context, const char * addr , int session, vo
 		skynet_harbor_send(NULL, des, &smsg);
 	} else if (skynet_context_push(des, &smsg)) {
 		free(msg);
-		skynet_error(NULL, "Drop message from %u to %s (size=%d)", smsg.source, addr, (int)sz);
-		return;
+		skynet_error(NULL, "Drop message from %x to %s (size=%d)", smsg.source, addr, (int)sz);
+		return -1;
 	}
+	return session;
 }
 
 uint32_t 
@@ -308,9 +312,7 @@ skynet_context_push(uint32_t handle, struct skynet_message *message) {
 	if (ctx == NULL) {
 		return -1;
 	}
-	if (message->session < 0) {
-		message->session = _new_session(ctx);
-	}
+	assert(message->session >= 0);
 	skynet_mq_push(ctx->queue, message);
 	if (__sync_lock_test_and_set(&ctx->in_global_queue,1) == 0) {
 		skynet_globalmq_push(ctx->queue);

skynet_server.h

@@ -12,6 +12,7 @@ struct skynet_context * skynet_context_release(struct skynet_context *);
 uint32_t skynet_context_handle(struct skynet_context *);
 void skynet_context_init(struct skynet_context *, uint32_t handle);
 int skynet_context_push(uint32_t handle, struct skynet_message *message);
+int skynet_context_newsession(struct skynet_context *);
 int skynet_context_message_dispatch(void);	// return 1 when block
 
 #endif

skynet_timer.c

@@ -1,6 +1,7 @@
 #include "skynet_timer.h"
 #include "skynet_mq.h"
 #include "skynet_server.h"
+#include "skynet_handle.h"
 
 #include <time.h>
 #include <assert.h>
@@ -171,8 +172,17 @@ timer_create_timer()
 	return r;
 }
 
-void 
+int
 skynet_timeout(uint32_t handle, int time, int session) {
+	if (session < 0) {
+		struct skynet_context * ctx = skynet_handle_grab(handle);
+		if (ctx == NULL) {
+			return -1;
+		}
+		session = skynet_context_newsession(ctx);
+		skynet_context_release(ctx);
+	}
+
 	if (time == 0) {
 		struct skynet_message message;
 		message.source = SKYNET_SYSTEM_TIMER;
@@ -181,7 +191,7 @@ skynet_timeout(uint32_t handle, int time, int session) {
 		message.sz = 0;
 
 		if (skynet_context_push(handle, &message)) {
-			return;
+			return -1;
 		}
 	} else {
 		struct timer_event event;
@@ -189,6 +199,8 @@ skynet_timeout(uint32_t handle, int time, int session) {
 		event.session = session;
 		timer_add(TI, &event, sizeof(event), time);
 	}
+
+	return session;
 }
 
 static uint32_t

skynet_timer.h

@@ -5,7 +5,7 @@
 
 #include <stdint.h>
 
-void skynet_timeout(uint32_t handle, int time, int session);
+int skynet_timeout(uint32_t handle, int time, int session);
 void skynet_updatetime(void);
 uint32_t skynet_gettime(void);
 

testdb.lua

@@ -0,0 +1,12 @@
+local skynet = require "skynet"
+
+-- register a dummy callback function
+skynet.dispatch()
+
+skynet.start(function()
+	local result = skynet.call("SIMPLEDB","SET foobar hello")
+	print(result)
+	result = skynet.call("SIMPLEDB","GET foobar")
+	print(result)
+	skynet.exit()
+end)

testtimer.lua

@@ -1,8 +1,13 @@
 local skynet = require "skynet"
 
-skynet.callback(function(session,addr,content)
-	print("sn:",session)
-	skynet.command("TIMEOUT", -1 , "100")
-end)
+-- register a dummy callback function
+skynet.dispatch()
 
-skynet.command("TIMEOUT",0,"0")
+skynet.start(function()
+	for i = 1, 10 do
+		print(i)
+		skynet.sleep(100)
+	end
+	skynet.exit()
+	print("Test timer exit")
+end)

watchdog.lua

@@ -6,7 +6,7 @@ function command:open(parm)
 	local fd,addr = string.match(parm,"(%d+) ([^%s]+)")
 	fd = tonumber(fd)
 	skynet.send("LOG", 0, string.format("%d %d %s",self,fd,addr))
-	local agent = skynet.command("LAUNCH","snlua agent.lua ".. self)
+	local agent = skynet.launch("snlua","agent.lua",self)
 	if agent then
 		skynet.send("gate",0, "forward ".. self .. " " .. agent)
 	end
@@ -20,7 +20,7 @@ function command:data(data)
 	skynet.send("LOG",0,string.format("data %d size=%d",self,#data))
 end
 
-skynet.callback(function(session, from , message)
+skynet.dispatch(function(message)
 	local id, cmd , parm = string.match(message, "(%d+) (%w+) ?(.*)")
 	id = tonumber(id)
 	local f = command[cmd]
@@ -31,4 +31,4 @@ skynet.callback(function(session, from , message)
 	end
 end)
 
-skynet.command("REG",".watchdog")
+skynet.register ".watchdog"