broker & launcher

Makefile

@@ -1,4 +1,4 @@
-all : skynet snlua.so logger.so skynet.so gate.so client.so connection.so client skynet-master
+all : skynet snlua.so logger.so skynet.so gate.so client.so connection.so broker.so client skynet-master
 
 skynet : skynet_main.c skynet_handle.c skynet_module.c skynet_mq.c skynet_server.c skynet_start.c skynet_timer.c skynet_error.c skynet_harbor.c master/master.c
 	gcc -Wall -g -Wl,-E -o $@ $^ -Imaster -lpthread -ldl -lrt -Wl,-E -llua -lm -lzmq
@@ -21,6 +21,9 @@ client.so : service_client.c
 connection.so : connection/connection.c connection/main.c
 	gcc -Wall -g -fPIC --shared -o $@ $^ -I. -Iconnection
 
+broker.so : service_broker.c
+	gcc -Wall -g -fPIC --shared $^ -o $@
+
 client : client.c
 	gcc -Wall -g $^ -o $@ -lpthread
 

console.lua

@@ -9,7 +9,7 @@ skynet.start(function()
 		if handle == nil then
 			print("Launch error:",cmd)
 		else
-			print("Lauch:",handle)
+			print("Launch:",handle,cmd)
 		end
 		skynet.yield()
 	end

launcher.lua

@@ -0,0 +1,26 @@
+local skynet = require "skynet"
+local string = string
+
+local instance = {}
+
+skynet.dispatch(function(message, session, address)
+	if session == 0 then
+		-- init notice
+		local reply = instance[address]
+		if reply then
+			skynet.send(reply[2] , reply[1], address)
+			instance[address] = nil
+		end
+	else
+		-- launch request
+		local service, param = string.match(message, "([^ ]+) (.*)")
+		local inst = skynet.launch(service, param)
+		if inst then
+			instance[inst] = { session, address }
+		else
+			skynet.send(address, session)
+		end
+	end
+end)
+
+skynet.register(".launcher")

lua-skynet.c

@@ -7,14 +7,13 @@
 
 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;
+	const char *msg = lua_tostring(L, 1);
+	if (msg) {
+		luaL_traceback(L, L, msg, 1);
+	} else {
+		lua_pushliteral(L, "(no error message)");
+	}
+	return 1;
 }
 
 static void
@@ -25,8 +24,14 @@ _cb(struct skynet_context * context, void * ud, int session, const char * addr,
 	lua_rawgetp(L, LUA_REGISTRYINDEX, _cb);
 	int r;
 	if (msg == NULL) {
-		lua_pushinteger(L, session);
-		r = lua_pcall(L, 1, 0 , trace);
+		if (addr == NULL) {
+			lua_pushinteger(L, session);
+			r = lua_pcall(L, 1, 0 , trace);
+		} else {
+			lua_pushinteger(L, session);
+			lua_pushstring(L, addr);
+			r = lua_pcall(L, 2, 0 , trace);
+		}
 	} else {
 		lua_pushinteger(L, session);
 		lua_pushstring(L, addr);
@@ -35,7 +40,24 @@ _cb(struct skynet_context * context, void * ud, int session, const char * addr,
 	}
 	if (r == LUA_OK) 
 		return;
-	skynet_error(context, "lua error %s", lua_tostring(L,-1));
+	const char * self = skynet_command(context, "REG", NULL);
+	switch (r) {
+	case LUA_ERRRUN:
+		printf("Lua_ERRRUN %p\n",L);
+		skynet_error(context, "lua call [%s to %s : %d] error : %s", addr , self, session, lua_tostring(L,-1));
+		break;
+	case LUA_ERRMEM:
+		skynet_error(context, "lua memory error : [%s to %s : %d]", addr , self, session);
+		break;
+	case LUA_ERRERR:
+		skynet_error(context, "lua error in error : [%s to %s : %d]", addr , self, session);
+		break;
+	case LUA_ERRGCMM:
+		skynet_error(context, "lua gc error : [%s to %s : %d]", addr , self, session);
+		break;
+	};
+
+	lua_pop(L,1);
 }
 
 static int

main.lua

@@ -1,7 +1,8 @@
 local skynet = require "skynet"
 
 print("Server start")
-
+local launcher = skynet.launch("snlua","launcher.lua")
+print("launcher", launcher)
 local console = skynet.launch("snlua","console.lua")
 print("console",console)
 local watchdog = skynet.launch("snlua","watchdog.lua")
@@ -10,5 +11,6 @@ local gate = skynet.launch("gate","8888 4 0")
 print("gate",gate)
 local db = skynet.launch("snlua","simpledb.lua")
 print("simpledb",db)
-
+local connection = skynet.launch("connection","256")
+print(connection)
 skynet.exit()

redis-cli.lua

@@ -0,0 +1,58 @@
+local skynet = require "skynet"
+local string = string
+local table = table
+local tonumber = tonumber
+local redis_server = ...
+local fd
+local write_fd
+local readline_fd
+local read_fd
+
+local function init_fd(fdstr)
+	fd = fdstr
+	write_fd = "WRITE "..fd.." "
+	readline_fd = "READLINE ".. fd .." \r\n"
+	read_fd = "READ " .. fd .. " "
+end
+
+skynet.dispatch(function(message, ...)
+	skynet.send(".connection", write_fd .. message)
+	local result = skynet.call(".connection", readline_fd)
+	local firstchar = string.byte(result)
+	if firstchar == 42 then	-- '*'
+		local n = tonumber(string.sub(result,2))
+		if n < 1 then
+			skynet.ret(result .. "\r\n")
+			return
+		end
+		local bulk = { result }
+		for i = 1,n do
+			local line = skynet.call(".connection", readline_fd)
+			table.insert(result, line)
+			local bytes = tonumber(string.sub(line,2) + 2)
+			table.insert(result, bytes)
+		end
+		table.insert(result,"")
+		skynet.ret(table.concat(result,"\r\n"))
+	elseif firstchar == 36 then -- '$'
+		local bytes = tonumber(string.sub(result,2))
+		if bytes < 0 then
+			skynet.ret(result .. "\r\n")
+			return
+		end
+		local firstline = skynet.call(".connection", read_fd .. (bytes + 2))
+		skynet.ret(result .. "\r\n" .. firstline)
+	else
+		skynet.ret(result .. "\r\n")
+	end
+end)
+
+skynet.start(function()
+	fd = skynet.call(".connection", "CONNECT " .. redis_server)
+	if fd == nil then
+		print("Connect to redis server error : ", redis_server)
+		skynet.exit()
+		return
+	end
+	init_fd(fd)
+end)

service_broker.c

@@ -0,0 +1,100 @@
+#include "skynet.h"
+
+#include <stdlib.h>
+#include <string.h>
+#include <assert.h>
+#include <stdio.h>
+
+#define DEFAULT_NUMBER 16
+#define LAUNCHER ".launcher"
+
+struct worker {
+	int init;
+	char * name;
+};
+
+struct broker {
+	int init;
+	int id;
+	char * name;
+	struct worker w[DEFAULT_NUMBER];
+};
+
+struct broker * 
+broker_create(void) {
+	struct broker *b = malloc(sizeof(*b));
+	memset(b,0,sizeof(*b));
+	return b;
+}
+
+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) {
+	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].init = 1;
+	++b->init;
+}
+
+static void
+_forward(struct broker *b, struct skynet_context * context) {
+	skynet_forward(context, b->w[b->id].name);
+	b->id = (b->id + 1) % DEFAULT_NUMBER;
+}
+
+static void
+_cb(struct skynet_context * context, void * ud, int session, const char * addr, const void * msg, size_t sz) {
+	struct broker * b = ud;
+	if (b->init < DEFAULT_NUMBER) {
+		_init(b, session, msg, sz);
+		if (b->init == DEFAULT_NUMBER) {
+			skynet_command(context, "REG", b->name);
+			skynet_send(context, LAUNCHER, 0, NULL, 0, 0);
+		}
+	} else {
+		_forward(b, context);
+	}
+}
+
+
+int
+broker_init(struct broker *b, struct skynet_context *ctx, const char * args) {
+	char * service = strchr(args,' ');
+	if (service == NULL) {
+		return 1;
+	}
+	int len = service - args;
+	if (len>0) {
+		b->name = malloc(len +1);
+		memcpy(b->name, args, len);
+		b->name[len] = '\0';
+	}
+	service++;
+
+	int i;
+	len = strlen(service);
+	if (len == 0)
+		return 1;
+	for (i=0;i<DEFAULT_NUMBER;i++) {
+		int id = skynet_send(ctx, LAUNCHER , -1, service , len, 0);
+		assert(id > 0 && id <= DEFAULT_NUMBER);
+	}
+
+	skynet_callback(ctx, b, _cb);
+
+	return 0;
+}

service_lua.c

@@ -21,14 +21,13 @@ _load(lua_State *L, char ** filename) {
 
 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;
+	const char *msg = lua_tostring(L, 1);
+	if (msg)
+		luaL_traceback(L, L, msg, 1);
+	else {
+		lua_pushliteral(L, "(no error message)");
+	}
+	return 1;
 }
 
 int
@@ -44,6 +43,7 @@ snlua_init(lua_State *L, struct skynet_context *ctx, const char * args) {
 	strcpy(parm,args);
 
 	lua_pushcfunction(L, traceback);
+	int traceback_index = lua_gettop(L);
 
 	const char * filename = parm;
 	int r = _load(L, &parm);
@@ -59,7 +59,7 @@ snlua_init(lua_State *L, struct skynet_context *ctx, const char * args) {
 			++n;
 		}
 	}
-	r = lua_pcall(L,n,0,-n-2);
+	r = lua_pcall(L,n,0,traceback_index);
 	switch (r) {
 	case LUA_OK:
 		return 0;
@@ -77,6 +77,8 @@ snlua_init(lua_State *L, struct skynet_context *ctx, const char * args) {
 		break;
 	};
 
+	lua_pop(L,1);
+
 	return 1;
 }
 

simpledb.lua

@@ -13,7 +13,7 @@ function command.SET(key, value)
 	skynet.ret(last)
 end
 
-skynet.dispatch(function(message, from, session)
+skynet.dispatch(function(message, session, from)
 	print("simpledb",message, from, session)
 	local cmd, key , value = string.match(message, "(%w+) (%w+) ?(.*)")
 	local f = command[cmd]

skynet.h

@@ -12,6 +12,7 @@ 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 * addr , int session, void * msg, size_t sz, int flags);
 
+void skynet_forward(struct skynet_context *, const char * addr);
 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

@@ -99,7 +99,12 @@ end
 
 function skynet.start(f)
 	local session = c.command("TIMEOUT","0")
-	local co = coroutine.create(f)
+	local co = coroutine.create(
+		function(...)
+			f(...)
+			skynet.send(".launcher",0)
+		end
+	)
 	session_id_coroutine[tonumber(session)] = co
 end
 

skynet_main.c

@@ -58,7 +58,7 @@ main(int argc, char *argv[]) {
 		lua_close(L);
 		return 1;
 	} 
-	const char *path = optstring(L,"lua_path","./?.lua");
+	const char *path = optstring(L,"lua_path","./?.lua;./?/init.lua");
 	setenv("LUA_PATH",path,1);
 	const char *cpath = optstring(L,"lua_cpath","./?.so");
 	setenv("LUA_CPATH",cpath,1);

skynet_server.c

@@ -25,6 +25,9 @@ struct skynet_context {
 	skynet_cb cb;
 	int session_id;
 	int in_global_queue;
+	int init;
+	uint32_t forward;
+	char * forward_address;
 	struct message_queue *queue;
 };
 
@@ -39,7 +42,7 @@ _id_to_hex(char * str, uint32_t id) {
 }
 
 struct skynet_context * 
-skynet_context_new(const char * name, const char *parm) {
+skynet_context_new(const char * name, const char *param) {
 	struct skynet_module * mod = skynet_module_query(name);
 
 	if (mod == NULL)
@@ -54,20 +57,31 @@ skynet_context_new(const char * name, const char *parm) {
 	ctx->ref = 2;
 	ctx->cb = NULL;
 	ctx->cb_ud = NULL;
-	ctx->in_global_queue = 0;
+	// a trick, global loop can't be dispatch in init process
+	ctx->in_global_queue = 1;
+
+	ctx->forward = 0;
+	ctx->forward_address = NULL;
 	ctx->session_id = 0;
+	ctx->init = 0;
+	ctx->handle = skynet_handle_register(ctx);
 	char * uid = ctx->handle_name;
 	uid[0] = ':';
 	_id_to_hex(uid+1, ctx->handle);
-
-	ctx->handle = skynet_handle_register(ctx);
 	ctx->queue = skynet_mq_create(ctx->handle);
 	// init function maybe use ctx->handle, so it must init at last
 
-	int r = skynet_module_instance_init(mod, inst, ctx, parm);
+	int r = skynet_module_instance_init(mod, inst, ctx, param);
 	if (r == 0) {
-		return skynet_context_release(ctx);
+		struct skynet_context * ret = skynet_context_release(ctx);
+		if (ret) {
+			ctx->init = 1;
+			skynet_globalmq_push(ctx->queue);
+			return ret;
+		}
+		return NULL;
 	} else {
+		ctx->in_global_queue = 0;
 		skynet_context_release(ctx);
 		skynet_handle_retire(ctx->handle);
 		return NULL;
@@ -108,23 +122,55 @@ skynet_context_release(struct skynet_context *ctx) {
 	return ctx;
 }
 
+static int
+_forwarding(struct skynet_context *ctx, struct skynet_message *msg) {
+	if (ctx->forward) {
+		uint32_t des = ctx->forward;
+		ctx->forward = 0;
+		if (skynet_harbor_message_isremote(des)) {
+			skynet_harbor_send(NULL, des, msg);
+		} else {
+			if (skynet_context_push(des, msg)) {
+				free(msg->data);
+				skynet_error(NULL, "Drop message from %x forward to %x (size=%d)", msg->source, des, (int)msg->sz);
+				return 1;
+			}
+		}
+		return 1;
+	}
+	if (ctx->forward_address) {
+		skynet_harbor_send(ctx->forward_address, 0, msg);
+		free(ctx->forward_address);
+		ctx->forward_address = NULL;
+		return 1;
+	}
+	return 0;
+}
+
 static void
 _dispatch_message(struct skynet_context *ctx, struct skynet_message *msg) {
+	assert(ctx->init);
 	if (msg->source == SKYNET_SYSTEM_TIMER) {
 		ctx->cb(ctx, ctx->cb_ud, msg->session, NULL, msg->data, msg->sz);
 	} else {
 		char tmp[10];
 		tmp[0] = ':';
+		int not_delete;
 		_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, msg->session, tmp, data, msg->sz);
-			skynet_harbor_message_close(msg);
+			not_delete = _forwarding(ctx, msg);
+			if (!not_delete) {
+				skynet_harbor_message_close(msg);
+			}
 		} else {
 			ctx->cb(ctx, ctx->cb_ud, msg->session, tmp, msg->data, msg->sz);
+			not_delete = _forwarding(ctx, msg);
+		}
+		if (!not_delete) {
+			free(msg->data);
 		}
-
-		free(msg->data);
 	}
 }
 
@@ -197,10 +243,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 , const char * param) {
 	if (strcmp(cmd,"TIMEOUT") == 0) {
 		char * session_ptr = NULL;
-		int ti = strtol(parm, &session_ptr, 10);
+		int ti = strtol(param, &session_ptr, 10);
 		int session = skynet_context_newsession(context);
 		if (session < 0) 
 			return NULL;
@@ -216,13 +262,13 @@ skynet_command(struct skynet_context * context, const char * cmd , const char *
 	}
 
 	if (strcmp(cmd,"REG") == 0) {
-		if (parm == NULL || parm[0] == '\0') {
+		if (param == NULL || param[0] == '\0') {
 			return context->handle_name;
-		} else if (parm[0] == '.') {
-			return skynet_handle_namehandle(context->handle, parm + 1);
+		} else if (param[0] == '.') {
+			return skynet_handle_namehandle(context->handle, param + 1);
 		} else {
 			assert(context->handle!=0);
-			skynet_harbor_register(parm, context->handle);
+			skynet_harbor_register(param, context->handle);
 			return NULL;
 		}
 	}
@@ -234,10 +280,10 @@ skynet_command(struct skynet_context * context, const char * cmd , const char *
 
 	if (strcmp(cmd,"KILL") == 0) {
 		uint32_t handle = 0;
-		if (parm[0] == ':') {
-			handle = strtoul(parm+1, NULL, 16);
-		} else if (parm[0] == '.') {
-			handle = skynet_handle_findname(parm+1);
+		if (param[0] == ':') {
+			handle = strtoul(param+1, NULL, 16);
+		} else if (param[0] == '.') {
+			handle = skynet_handle_findname(param+1);
 		} else {
 			// todo : kill global service
 		}
@@ -248,14 +294,15 @@ skynet_command(struct skynet_context * context, const char * cmd , const char *
 	}
 
 	if (strcmp(cmd,"LAUNCH") == 0) {
-		size_t sz = strlen(parm);
+		size_t sz = strlen(param);
 		char tmp[sz+1];
-		strcpy(tmp,parm);
-		char * parm = tmp;
-		char * mod = strsep(&parm, " \t\r\n");
-		parm = strsep(&parm, "\r\n");
-		struct skynet_context * inst = skynet_context_new(mod,parm);
+		strcpy(tmp,param);
+		char * args = tmp;
+		char * mod = strsep(&args, " \t\r\n");
+		args = strsep(&args, "\r\n");
+		struct skynet_context * inst = skynet_context_new(mod,args);
 		if (inst == NULL) {
+			fprintf(stderr, "Launch %s %s failed\n",mod,args);
 			return NULL;
 		} else {
 			context->result[0] = ':';
@@ -267,6 +314,26 @@ skynet_command(struct skynet_context * context, const char * cmd , const char *
 	return NULL;
 }
 
+void 
+skynet_forward(struct skynet_context * context, const char * addr) {
+	uint32_t des = 0;
+	assert(context->forward == 0 && context->forward_address == NULL);
+	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 {
+		context->forward_address = strdup(addr);
+		return;
+	}
+	context->forward = des;
+}
+
 int
 skynet_send(struct skynet_context * context, const char * addr , int session, void * data, size_t sz, int flags) {
 	char * msg;

testredis.lua

@@ -0,0 +1,15 @@
+local skynet = require "skynet"
+
+skynet.dispatch()
+
+local command ="*2\r\n$3\r\nGET\r\n$1\r\nA\r\n"
+
+skynet.start(function()
+	local cli = skynet.call(".launcher","broker redis snlua redis-cli.lua 127.0.0.1:6379")
+	print("redis-cli:", cli)
+	assert(cli)
+	local result = skynet.call(cli, command)
+	print(result)
+	skynet.exit()
+end)
+