Merge pull request #113 from cloudwu/dev

bugfix

HISTORY.md

@@ -0,0 +1,40 @@
+v0.2.1 (2014-5-19)
+-----------
+* Bugfix: check all the events already read after socket close
+* Bugfix: socket data in gate service 
+* Bugfix: boundary problem in harbor service
+* Bugfix: stdin handle is 0
+
+v0.2.0 (2014-5-12)
+-----------
+
+* Rewrite malloc hook , use `pthread_getspecific` instead of `__thread` to get current service handle.
+* Optimize global unique service query, rewrite `service_mgr` .
+* Add some snax api, snax.uniqueservice (etc.) , use independent protocol `PTYPE_SNAX` .
+* Add bootstrap lua script , remove some code in C .
+* Use a lua loader to load lua service code (and set the lua environment), remove some code in C.
+* Support preload a file before each lua serivce start.
+* Add datacenter serivce.
+* Add multicast api.
+* Remove skynet.blockcall , simplify the implement of message queue.
+* When dropping message queue (at service exit) , dispatcher will post an error back to the source of each message.
+* Remove skynet.watch , monitor is not necessary for watching skynet.call . so simplemonitor.lua is move to examples.
+* Remove the limit of global queue size (64K actived service limit before).
+* Refactoring `skynet_command`.
+
+v0.1.1 (2014-4-28)
+------------------
+
+* Socket channel should clear request queue when reconnect.
+* Fix the issue that socket close may block the coroutine.
+* Fix the issue that jemalloc api may crash on macosx (disable jemalloc on macosx).
+
+v0.1.0 (2014-4-23)
+------------------
+
+* First release version.
+
+First public version (2012-8-1)
+------------------
+
+* Make skynet from a private project to public.

Makefile

@@ -5,7 +5,7 @@ CSERVICE_PATH ?= cservice
 
 SKYNET_BUILD_PATH ?= .
 
-CFLAGS = -g -O2 -Wall -I$(LUA_INC) $(MYCFLAGS)
+CFLAGS = -g -O2 -Wall -I$(LUA_INC) $(MYCFLAGS) 
 
 # lua
 
@@ -22,8 +22,11 @@ JEMALLOC_STATICLIB := 3rd/jemalloc/lib/libjemalloc_pic.a
 JEMALLOC_INC := 3rd/jemalloc/include/jemalloc
 
 all : jemalloc
+	
 .PHONY : jemalloc
 
+MALLOC_STATICLIB := $(JEMALLOC_STATICLIB)
+
 $(JEMALLOC_STATICLIB) : 3rd/jemalloc/Makefile
 	cd 3rd/jemalloc && $(MAKE) CC=$(CC) 
 
@@ -33,12 +36,12 @@ $(JEMALLOC_STATICLIB) : 3rd/jemalloc/Makefile
 3rd/jemalloc/Makefile : | 3rd/jemalloc/autogen.sh
 	cd 3rd/jemalloc && ./autogen.sh --with-jemalloc-prefix=je_ --disable-valgrind
 
-jemalloc : $(JEMALLOC_STATICLIB)
+jemalloc : $(MALLOC_STATICLIB)
 
 # skynet
 
 CSERVICE = snlua logger gate master harbor
-LUA_CLIB = skynet socketdriver int64 bson mongo md5 netpack cjson clientsocket memory profile
+LUA_CLIB = skynet socketdriver int64 bson mongo md5 netpack cjson clientsocket memory profile multicast
 
 SKYNET_SRC = skynet_main.c skynet_handle.c skynet_module.c skynet_mq.c \
   skynet_server.c skynet_start.c skynet_timer.c skynet_error.c \
@@ -50,8 +53,8 @@ all : \
   $(foreach v, $(CSERVICE), $(CSERVICE_PATH)/$(v).so) \
   $(foreach v, $(LUA_CLIB), $(LUA_CLIB_PATH)/$(v).so) 
 
-$(SKYNET_BUILD_PATH)/skynet : $(foreach v, $(SKYNET_SRC), skynet-src/$(v)) $(LUA_LIB) $(JEMALLOC_STATICLIB)
+$(SKYNET_BUILD_PATH)/skynet : $(foreach v, $(SKYNET_SRC), skynet-src/$(v)) $(LUA_LIB) $(MALLOC_STATICLIB)
-	$(CC) $(CFLAGS) -o $@ $^ -Iskynet-src -I$(JEMALLOC_INC) $(LDFLAGS) $(EXPORT) $(LIBS)
+	$(CC) $(CFLAGS) -o $@ $^ -Iskynet-src -I$(JEMALLOC_INC) $(LDFLAGS) $(EXPORT) $(SKYNET_LIBS) $(SKYNET_DEFINES)
 
 $(LUA_CLIB_PATH) :
 	mkdir $(LUA_CLIB_PATH)
@@ -88,7 +91,7 @@ $(LUA_CLIB_PATH)/netpack.so : lualib-src/lua-netpack.c | $(LUA_CLIB_PATH)
 	$(CC) $(CFLAGS) $(SHARED) $^ -Iskynet-src -o $@ 
 
 $(LUA_CLIB_PATH)/cjson.so : | $(LUA_CLIB_PATH)
-	cd 3rd/lua-cjson && $(MAKE) LUA_INCLUDE_DIR=../../$(LUA_INC) CC=$(CC) CJSON_LDFLAGS="$(SHARED)" && cp cjson.so ../../$@
+	cd 3rd/lua-cjson && $(MAKE) LUA_INCLUDE_DIR=../../$(LUA_INC) CC=$(CC) CJSON_LDFLAGS="$(SHARED)" && cd ../.. && cp 3rd/lua-cjson/cjson.so $@
 
 $(LUA_CLIB_PATH)/clientsocket.so : lualib-src/lua-clientsocket.c | $(LUA_CLIB_PATH)
 	$(CC) $(CFLAGS) $(SHARED) $^ -o $@ -lpthread
@@ -99,6 +102,9 @@ $(LUA_CLIB_PATH)/memory.so : lualib-src/lua-memory.c | $(LUA_CLIB_PATH)
 $(LUA_CLIB_PATH)/profile.so : lualib-src/lua-profile.c | $(LUA_CLIB_PATH)
 	$(CC) $(CFLAGS) $(SHARED) $^ -o $@ 
 
+$(LUA_CLIB_PATH)/multicast.so : lualib-src/lua-multicast.c | $(LUA_CLIB_PATH)
+	$(CC) $(CFLAGS) $(SHARED) -Iskynet-src $^ -o $@ 
+
 clean :
 	rm -f $(SKYNET_BUILD_PATH)/skynet $(CSERVICE_PATH)/*.so $(LUA_CLIB_PATH)/*.so
 

examples/config

@@ -4,8 +4,11 @@ logger = nil
 harbor = 1
 address = "127.0.0.1:2526"
 master = "127.0.0.1:2013"
-start = "main"
+start = "main"	-- main script
+bootstrap = "snlua bootstrap"	-- The service for bootstrap
 standalone = "0.0.0.0:2013"
 luaservice = root.."service/?.lua;"..root.."test/?.lua;"..root.."examples/?.lua"
+lualoader = "lualib/loader.lua"
+-- preload = "./examples/preload.lua"	-- run preload.lua before every lua service run
 snax = root.."examples/?.lua;"..root.."test/?.lua"
 cpath = root.."cservice/?.so"

examples/config.mc

@@ -0,0 +1,14 @@
+root = "./"
+thread = 8
+logger = nil
+harbor = 2
+address = "127.0.0.1:2527"
+master = "127.0.0.1:2013"
+start = "testmulticast2"	-- main script
+bootstrap = "snlua bootstrap"	-- The service for bootstrap
+--standalone = "0.0.0.0:2013"
+luaservice = root.."service/?.lua;"..root.."test/?.lua;"..root.."examples/?.lua"
+lualoader = "lualib/loader.lua"
+-- preload = "./examples/preload.lua"	-- run preload.lua before every lua service run
+snax = root.."examples/?.lua;"..root.."test/?.lua"
+cpath = root.."cservice/?.so"

examples/config_log

@@ -7,3 +7,4 @@ address = "127.0.0.1:2527"
 master = "127.0.0.1:2013"
 start = "main_log"
 luaservice ="./service/?.lua;./test/?.lua;./examples/?.lua"
+snax = "./examples/?.lua;./test/?.lua"

examples/main.lua

@@ -4,10 +4,8 @@ local max_client = 64
 
 skynet.start(function()
 	print("Server start")
-	local service = skynet.newservice("service_mgr")
-	skynet.monitor "simplemonitor"
 	local console = skynet.newservice("console")
---	skynet.newservice("debug_console",8000)
+	skynet.newservice("debug_console",8000)
 	skynet.newservice("simpledb")
 	local watchdog = skynet.newservice("watchdog")
 	skynet.call(watchdog, "lua", "start", {

examples/preload.lua

@@ -0,0 +1,5 @@
+-- This file will execute before every lua service start
+-- See config
+
+print("PRELOAD", ...)
+

examples/simplemonitor.lua

@@ -12,11 +12,10 @@ skynet.register_protocol {
 		local w = service_map[address]
 		if w then
 			for watcher in pairs(w) do
-				skynet.send(watcher, "error", address)
+				skynet.redirect(watcher, address, "error", "")
 			end
 			service_map[address] = false
 		end
-		print(string.format("[:%x] exit", address))
 	end
 }
 

lualib-src/lua-mongo.c

@@ -55,7 +55,7 @@ little_endian(uint32_t v) {
 typedef void * document;
 
 static inline uint32_t
-get_length(const document buffer) {
+get_length(document buffer) {
 	union {
 		uint32_t v;
 		uint8_t b[4];
@@ -243,7 +243,7 @@ op_reply(lua_State *L) {
 			lua_pushlightuserdata(L, (void *)doc);
 			lua_rawseti(L, 2, i);
 
-			int32_t doc_len = get_length((const document)doc);
+			int32_t doc_len = get_length((document)doc);
 
 			doc += doc_len;
 			sz -= doc_len;

lualib-src/lua-multicast.c

@@ -0,0 +1,167 @@
+#include "skynet.h"
+
+#include <lua.h>
+#include <lauxlib.h>
+#include <stdint.h>
+#include <string.h>
+
+struct mc_package {
+	int reference;
+	uint32_t size;
+	void *data;
+};
+
+static int
+pack(lua_State *L, void *data, size_t size) {
+	struct mc_package * pack = skynet_malloc(sizeof(struct mc_package));
+	pack->reference = 0;
+	pack->size = (uint32_t)size;
+	pack->data = data;
+	struct mc_package ** ret = skynet_malloc(sizeof(*ret));
+	*ret = pack;
+	lua_pushlightuserdata(L, ret);
+	lua_pushinteger(L, sizeof(ret));
+	return 2;
+}
+
+/*
+	lightuserdata
+	integer size
+
+	return lightuserdata, sizeof(struct mc_package *)
+ */
+static int
+mc_packlocal(lua_State *L) {
+	void * data = lua_touserdata(L, 1);
+	size_t size = luaL_checkunsigned(L, 2);
+	if (size != (uint32_t)size) {
+		return luaL_error(L, "Size should be 32bit integer");
+	}
+	return pack(L, data, size);
+}
+
+/*
+	lightuserdata
+	integer size
+
+	return lightuserdata, sizeof(struct mc_package *)
+ */
+static int
+mc_packremote(lua_State *L) {
+	void * data = lua_touserdata(L, 1);
+	size_t size = luaL_checkunsigned(L, 2);
+	if (size != (uint32_t)size) {
+		return luaL_error(L, "Size should be 32bit integer");
+	}
+	void * msg = skynet_malloc(size);
+	memcpy(msg, data, size);
+	return pack(L, msg, size);
+}
+
+static int
+mc_packstring(lua_State *L) {
+	size_t size;
+	const char * msg = luaL_checklstring(L, 1, &size);
+	if (size != (uint32_t)size) {
+		return luaL_error(L, "string is too long");
+	}
+	void * data = skynet_malloc(size);
+	memcpy(data, msg, size);
+	return pack(L, data, size);
+}
+
+/*
+	lightuserdata struct mc_package **
+	integer size (must be sizeof(struct mc_package *)
+
+	return package, lightuserdata, size
+ */
+static int
+mc_unpacklocal(lua_State *L) {
+	struct mc_package ** pack = lua_touserdata(L,1);
+	int sz = luaL_checkinteger(L,2);
+	if (sz != sizeof(*pack)) {
+		return luaL_error(L, "Invalid multicast package size %d", sz);
+	}
+	lua_settop(L, 1);
+	lua_pushlightuserdata(L, (*pack)->data);
+	lua_pushunsigned(L, (*pack)->size);
+	return 3;
+}
+
+/*
+	lightuserdata struct mc_package **
+	integer reference
+ */
+static int
+mc_bindrefer(lua_State *L) {
+	struct mc_package ** pack = lua_touserdata(L,1);
+	int ref = luaL_checkinteger(L,2);
+	if ((*pack)->reference != 0) {
+		return luaL_error(L, "Can't bind a multicast package more than once");
+	}
+	(*pack)->reference = ref;
+
+	return 0;
+}
+
+/*
+	lightuserdata struct mc_package **
+ */
+static int
+mc_closelocal(lua_State *L) {
+	struct mc_package **ptr = lua_touserdata(L,1);
+	struct mc_package *pack = *ptr;
+
+	int ref = __sync_sub_and_fetch(&pack->reference, 1);
+	if (ref <= 0) {
+		skynet_free(pack->data);
+		skynet_free(pack);
+		if (ref < 0) {
+			return luaL_error(L, "Invalid multicast package reference %d", ref);
+		}
+	}
+
+	return 0;
+}
+
+/*
+	lightuserdata struct mc_package **
+	return lightuserdata/size
+ */
+static int
+mc_remote(lua_State *L) {
+	struct mc_package **ptr = lua_touserdata(L,1);
+	struct mc_package *pack = *ptr;
+	lua_pushlightuserdata(L, pack->data);
+	lua_pushunsigned(L, pack->size);
+	skynet_free(pack);
+	return 2;
+}
+
+static int
+mc_nextid(lua_State *L) {
+	uint32_t id = luaL_checkunsigned(L, 1);
+	id += 256;
+	lua_pushunsigned(L, id);
+
+	return 1;
+}
+
+int
+luaopen_multicast_c(lua_State *L) {
+	luaL_Reg l[] = {
+		{ "pack", mc_packlocal },
+		{ "unpack", mc_unpacklocal },
+		{ "bind", mc_bindrefer },
+		{ "close", mc_closelocal },
+		{ "remote", mc_remote },
+		{ "packstring", mc_packstring },
+		{ "packremote", mc_packremote },
+		{ "nextid", mc_nextid },
+		{ NULL, NULL },
+	};
+	luaL_checkversion(L);
+	luaL_newlib(L,l);
+	return 1;
+}

lualib-src/lua-skynet.c

@@ -1,6 +1,5 @@
 #include "skynet.h"
 #include "lua-seri.h"
-#include "service_lua.h"
 
 #define KNRM  "\x1B[0m"
 #define KRED  "\x1B[31m"
@@ -11,18 +10,36 @@
 #include <string.h>
 #include <assert.h>
 
+struct snlua {
+	lua_State * L;
+	struct skynet_context * ctx;
+	const char * preload;
+};
+
+static int 
+traceback (lua_State *L) {
+	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 int
 _cb(struct skynet_context * context, void * ud, int type, int session, uint32_t source, const void * msg, size_t sz) {
 	lua_State *L = ud;
 	int trace = 1;
 	int r;
 	int top = lua_gettop(L);
-	if (top == 1) {
+	if (top == 0) {
+		lua_pushcfunction(L, traceback);
 		lua_rawgetp(L, LUA_REGISTRYINDEX, _cb);
 	} else {
 		assert(top == 2);
-		lua_pushvalue(L,2);
 	}
+	lua_pushvalue(L,2);
 
 	lua_pushinteger(L, type);
 	lua_pushlightuserdata(L, (void *)msg);
@@ -290,15 +307,14 @@ luaopen_skynet_c(lua_State *L) {
 		{ NULL, NULL },
 	};
 
-	lua_getfield(L, LUA_REGISTRYINDEX, "skynet_lua");
+	luaL_newlibtable(L, l);
-	struct snlua *lua = lua_touserdata(L,-1);
+
-	if (lua == NULL || lua->ctx == NULL) {
+	lua_getfield(L, LUA_REGISTRYINDEX, "skynet_context");
+	struct skynet_context *ctx = lua_touserdata(L,-1);
+	if (ctx == NULL) {
 		return luaL_error(L, "Init skynet context first");
 	}
-	assert(lua->L == L);
 
-	luaL_newlibtable(L, l);
-	lua_pushlightuserdata(L, lua->ctx);
 	luaL_setfuncs(L,l,1);
 
 	return 1;

lualib-src/lua-socket.c

@@ -10,7 +10,6 @@
 #include <lauxlib.h>
 
 #include "skynet_socket.h"
-#include "service_lua.h"
 
 #define BACKLOG 32
 // 2 ** 12 == 4096
@@ -485,15 +484,12 @@ luaopen_socketdriver(lua_State *L) {
 		{ "start", lstart },
 		{ NULL, NULL },
 	};
-	lua_getfield(L, LUA_REGISTRYINDEX, "skynet_lua");
+	lua_getfield(L, LUA_REGISTRYINDEX, "skynet_context");
-	struct snlua *lua = lua_touserdata(L,-1);
+	struct skynet_context *ctx = lua_touserdata(L,-1);
-	if (lua == NULL || lua->ctx == NULL) {
+	if (ctx == NULL) {
 		return luaL_error(L, "Init skynet context first");
 	}
-	// assert(lua->L == L);
-	lua_pop(L,1);
 
-	lua_pushlightuserdata(L, lua->ctx);
 	luaL_setfuncs(L,l2,1);
 
 	return 1;

lualib/datacenter.lua

@@ -0,0 +1,14 @@
+local skynet = require "skynet"
+
+local datacenter = {}
+
+function datacenter.get(...)
+	return skynet.call("DATACENTER", "lua", "QUERY", ...)
+end
+
+function datacenter.set(...)
+	return skynet.call("DATACENTER", "lua", "UPDATE", ...)
+end
+
+return datacenter
+

lualib/loader.lua

@@ -0,0 +1,46 @@
+local args = {}
+for word in string.gmatch(..., "%S+") do
+	table.insert(args, word)
+end
+
+local main, pattern
+
+local err = {}
+for pat in string.gmatch(LUA_SERVICE, "([^;]+);*") do
+	local filename = string.gsub(pat, "?", args[1])
+	local f, msg = loadfile(filename)
+	if not f then
+		table.insert(err, msg)
+	else
+		pattern = pat
+		main = f
+		break
+	end
+end
+
+if not main then
+	error(table.concat(err, "\n"))
+end
+
+LUA_SERVICE = nil
+package.path , LUA_PATH = LUA_PATH
+package.cpath , LUA_CPATH = LUA_CPATH
+
+local service_path = string.match(pattern, "(.*/)[^/?]+$")
+
+if service_path then
+	service_path = string.gsub(service_path, "?", args[1])
+	package.path = service_path .. "?.lua;" .. package.path
+	SERVICE_PATH = service_path
+else
+	local p = string.match(pattern, "(.*/).+$")
+	SERVICE_PATH = p
+end
+
+if LUA_PRELOAD then
+	local f = assert(loadfile(LUA_PRELOAD))
+	f(table.unpack(args))
+	LUA_PRELOAD = nil
+end
+
+main(select(2, table.unpack(args)))

lualib/mongo.lua

@@ -39,9 +39,7 @@ local client_meta = {
 
 		return "[mongo client : " .. self.host .. port_string .."]"
 	end,
-	__gc = function(self)
+	-- DO NOT need disconnect, because channel will shutdown during gc
-		self:disconnect()
-	end
 }
 
 local mongo_db = {}

lualib/multicast.lua

@@ -0,0 +1,102 @@
+local skynet = require "skynet"
+local mc = require "multicast.c"
+
+local multicastd
+local multicast = {}
+local dispatch = setmetatable({} , {__mode = "kv" })
+
+local chan = {}
+local chan_meta = {
+	__index = chan,
+	__gc = unsubscribe,
+	__tostring = function (self)
+		return string.format("[Multicast:%x]",self.channel)
+	end,
+}
+
+local function default_conf(conf)
+	conf = conf or {}
+	conf.pack = conf.pack or skynet.pack
+	conf.unpack = conf.unpack or skynet.unpack
+
+	return conf
+end
+
+function multicast.new(conf)
+	assert(multicastd, "Init first")
+	local self = {}
+	conf = conf or self
+	self.channel = conf.channel
+	if self.channel == nil then
+		self.channel = skynet.call(multicastd, "lua", "NEW")
+	end
+	self.__pack = conf.pack or skynet.pack
+	self.__unpack = conf.unpack or skynet.unpack
+	self.__dispatch = conf.dispatch
+
+	return setmetatable(self, chan_meta)
+end
+
+function chan:delete()
+	local c = assert(self.channel)
+	skynet.send(multicastd, "lua", "DEL", c)
+	self.channel = nil
+	self.__subscribe = nil
+end
+
+function chan:publish(...)
+	local c = assert(self.channel)
+	skynet.call(multicastd, "lua", "PUB", c, mc.pack(self.__pack(...)))
+end
+
+function chan:subscribe()
+	local c = assert(self.channel)
+	if self.__subscribe then
+		-- already subscribe
+		return
+	end
+	skynet.call(multicastd, "lua", "SUB", c)
+	self.__subscribe = true
+	dispatch[c] = self
+end
+
+function chan:unsubscribe()
+	if not self.__subscribe then
+		-- already unsubscribe
+		return
+	end
+	local c = assert(self.channel)
+	skynet.send(multicastd, "lua", "USUB", c)
+	self.__subscribe = nil
+end
+
+local function dispatch_subscribe(channel, source, pack, msg, sz)
+	local self = dispatch[channel]
+	if not self then
+		mc.close(pack)
+		error ("Unknown channel " .. channel)
+	end
+
+	if self.__subscribe then
+		local ok, err = pcall(self.__dispatch, self, source, self.__unpack(msg, sz))
+		mc.close(pack)
+		assert(ok, err)
+	else
+		-- maybe unsubscribe first, but the message is send out. drop the message unneed
+		mc.close(pack)
+	end
+end
+
+local function init()
+	multicastd = skynet.uniqueservice "multicastd"
+	skynet.register_protocol {
+		name = "multicast",
+		id = skynet.PTYPE_MULTICAST,
+		unpack = mc.unpack,
+		dispatch = dispatch_subscribe,
+	}
+end
+
+skynet.init(init, "multicast")
+
+return multicast

lualib/redis.lua

@@ -9,9 +9,7 @@ local redis = {}
 local command = {}
 local meta = {
 	__index = command,
-	__gc = function(self)
+	-- DO NOT close channel in __gc
-		self[1]:close()
-	end,
 }
 
 ---------- redis response

lualib/skynet.lua

@@ -16,7 +16,7 @@ local skynet = {
 	-- read skynet.h
 	PTYPE_TEXT = 0,
 	PTYPE_RESPONSE = 1,
---	PTYPE_MULTICAST = 2, -- DEPRECATED
+	PTYPE_MULTICAST = 2,
 	PTYPE_CLIENT = 3,
 	PTYPE_SYSTEM = 4,
 	PTYPE_HARBOR = 5,
@@ -24,7 +24,8 @@ local skynet = {
 	PTYPE_ERROR = 7,
 	PTYPE_QUEUE = 8,
 	PTYPE_DEBUG = 9,
-	PTYPE_LUA = 10
+	PTYPE_LUA = 10,
+	PTYPE_SNAX = 11,
 }
 
 -- code cache
@@ -68,13 +69,13 @@ local function dispatch_error_queue()
 	end
 end
 
-local function _error_dispatch(error_session, monitor, service)
+local function _error_dispatch(error_session, error_source)
-	if service then
+	if error_session == 0 then
 		-- service is down
 		--  Don't remove from watching_service , because user may call dead service
-		watching_service[service] = false
+		watching_service[error_source] = false
 		for session, srv in pairs(watching_session) do
-			if srv == service then
+			if srv == error_source then
 				table.insert(error_queue, session)
 			end
 		end
@@ -86,21 +87,6 @@ local function _error_dispatch(error_session, monitor, service)
 	end
 end
 
-local watch_monitor
-
-function skynet.watch(service)
-	assert(type(service) == "number")
-	if watch_monitor == nil then
-		watch_monitor = string_to_handle(c.command("MONITOR"))
-		assert(watch_monitor, "Need a monitor")
-	end
-	if watching_service[service] == nil then
-		watching_service[service] = true
-		-- read lualib/simplemonitor.lua
-		assert(skynet.call(watch_monitor, "lua", "WATCH", service), "watch a dead service")
-	end
-end
-
 -- coroutine reuse
 
 local coroutine_pool = {}
@@ -209,7 +195,7 @@ function skynet.register(name)
 end
 
 function skynet.name(name, handle)
-	c.command("NAME", name .. " " .. handle)
+	c.command("NAME", name .. " " .. skynet.address(handle))
 end
 
 local self_handle
@@ -254,7 +240,12 @@ function skynet.kill(name)
 end
 
 function skynet.getenv(key)
-	return c.command("GETENV",key)
+	local ret = c.command("GETENV",key)
+	if ret == "" then
+		return
+	else
+		return ret
+	end
 end
 
 function skynet.setenv(key, value)
@@ -283,7 +274,7 @@ local function yield_call(service, session)
 	watching_session[session] = service
 	local succ, msg, sz = coroutine_yield("CALL", session)
 	watching_session[session] = nil
-	assert(succ, "Capture an error")
+	assert(succ, debug.traceback())
 	return msg,sz
 end
 
@@ -299,17 +290,6 @@ function skynet.call(addr, typename, ...)
 	return p.unpack(yield_call(addr, session))
 end
 
-function skynet.blockcall(addr, typename , ...)
-	local p = proto[typename]
-	c.command("LOCK")
-	local session = c.send(addr, p.id , nil , p.pack(...))
-	if session == nil then
-		c.command("UNLOCK")
-		error("call to invalid address " .. skynet.address(addr))
-	end
-	return p.unpack(yield_call(addr, session))
-end
-
 function skynet.rawcall(addr, typename, msg, sz)
 	local p = proto[typename]
 	local session = assert(c.send(addr, p.id , nil , msg, sz), "call to invalid address")
@@ -322,7 +302,7 @@ function skynet.ret(msg, sz)
 end
 
 function skynet.retpack(...)
-    return skynet.ret(skynet.pack(...))
+	return skynet.ret(skynet.pack(...))
 end
 
 function skynet.wakeup(co)
@@ -339,14 +319,14 @@ function skynet.dispatch(typename, func)
 end
 
 local function unknown_request(session, address, msg, sz)
-    print("Unknown request :" , c.tostring(msg,sz))
+	print("Unknown request :" , c.tostring(msg,sz))
 	error(string.format("Unknown session : %d from %x", session, address))
 end
 
 function skynet.dispatch_unknown_request(unknown)
-    local prev = unknown_request
+	local prev = unknown_request
-    unknown_request = unknown
+	unknown_request = unknown
-    return prev
+	return prev
 end
 
 local function unknown_response(session, address, msg, sz)
@@ -393,7 +373,7 @@ local function raw_dispatch_message(prototype, msg, sz, session, source, ...)
 			session_coroutine_address[co] = source
 			suspend(co, coroutine.resume(co, session,source, p.unpack(msg,sz, ...)))
 		else
-            unknown_request(session, source, msg, sz)
+			unknown_request(session, source, msg, sz)
 		end
 	end
 end
@@ -420,8 +400,7 @@ local function dispatch_message(...)
 end
 
 function skynet.newservice(name, ...)
-	local param =  table.concat({"snlua", name, ...}, " ")
+	local handle = skynet.tostring(skynet.rawcall(".launcher", "lua" , skynet.pack("LAUNCH", "snlua", name, ...)))
-	local handle = skynet.call(".launcher", "text" , param)
 	if handle == "" then
 		return nil
 	else
@@ -430,32 +409,18 @@ function skynet.newservice(name, ...)
 end
 
 function skynet.uniqueservice(global, ...)
-	local handle
 	if global == true then
-		handle = skynet.call("SERVICE", "lua", "LAUNCH", ...)
+		return assert(skynet.call(".service", "lua", "GLAUNCH", ...))
 	else
-		handle = skynet.call(".service", "lua", "LAUNCH", global, ...)
+		return assert(skynet.call(".service", "lua", "LAUNCH", global, ...))
 	end
-	assert(handle , "Unique service launch failed")
-	return handle
 end
 
 function skynet.queryservice(global, ...)
-	local handle
 	if global == true then
-		handle = skynet.call("SERVICE", "lua", "QUERY", ...)
+		return assert(skynet.call(".service", "lua", "GQUERY", ...))
 	else
-		handle = skynet.call(".service", "lua", "QUERY", global, ...)
+		return assert(skynet.call(".service", "lua", "QUERY", global, ...))
-	end
-	assert(handle , "Unique service query failed")
-	return handle
-end
-
-local function group_command(cmd, handle, address)
-	if address then
-		return string.format("%s %d :%x",cmd, handle, address)
-	else
-		return string.format("%s %d",cmd,handle)
 	end
 end
 
@@ -479,66 +444,10 @@ function skynet.error(...)
 	return c.error(table.concat(t, " "))
 end
 
------ debug
-
-local internal_info_func
-
-function skynet.info_func(func)
-	internal_info_func = func
-end
-
-local dbgcmd = {}
-
-function dbgcmd.MEM()
-	local kb, bytes = collectgarbage "count"
-	skynet.ret(skynet.pack(kb,bytes))
-end
-
-function dbgcmd.GC()
-	coroutine_pool = {}
-	collectgarbage "collect"
-end
-
-function dbgcmd.STAT()
-	local stat = {}
-	stat.mqlen = skynet.mqlen()
-	skynet.ret(skynet.pack(stat))
-end
-
-function dbgcmd.INFO()
-	if internal_info_func then
-		skynet.ret(skynet.pack(internal_info_func()))
-	else
-		skynet.ret(skynet.pack(nil))
-	end
-end
-
-local function _debug_dispatch(session, address, cmd, ...)
-	local f = dbgcmd[cmd]
-	assert(f, cmd)
-	f(...)
-end
-
 ----- register protocol
 do
 	local REG = skynet.register_protocol
 
-	REG {
-		name = "text",
-		id = skynet.PTYPE_TEXT,
-		pack = function (...)
-			local n = select ("#" , ...)
-			if n == 0 then
-				return ""
-			elseif n == 1 then
-				return tostring(...)
-			else
-				return table.concat({...}," ")
-			end
-		end,
-		unpack = c.tostring
-	}
-
 	REG {
 		name = "lua",
 		id = skynet.PTYPE_LUA,
@@ -551,19 +460,10 @@ do
 		id = skynet.PTYPE_RESPONSE,
 	}
 
-	REG {
-		name = "debug",
-		id = skynet.PTYPE_DEBUG,
-		pack = skynet.pack,
-		unpack = skynet.unpack,
-		dispatch = _debug_dispatch,
-	}
-
 	REG {
 		name = "error",
 		id = skynet.PTYPE_ERROR,
-		pack = skynet.pack,
+		unpack = function(...) return ... end,
-		unpack = skynet.unpack,
 		dispatch = _error_dispatch,
 	}
 end
@@ -603,10 +503,10 @@ local function init_service(start)
 	local ok, err = xpcall(init_template, debug.traceback, start)
 	if not ok then
 		print("init service failed:", err)
-		skynet.send(".launcher","text", "ERROR")
+		skynet.send(".launcher","lua", "ERROR")
 		skynet.exit()
 	else
-		skynet.send(".launcher","text", "")
+		skynet.send(".launcher","lua", "LAUNCHOK")
 	end
 end
 
@@ -649,4 +549,8 @@ function skynet.mqlen()
 	return tonumber(c.command "MQLEN")
 end
 
+-- Inject internal debug framework
+local debug = require "skynet.debug"
+debug(skynet)
+
 return skynet

lualib/skynet/debug.lua

@@ -0,0 +1,49 @@
+return function (skynet)
+
+local internal_info_func
+
+function skynet.info_func(func)
+	internal_info_func = func
+end
+
+local dbgcmd = {}
+
+function dbgcmd.MEM()
+	local kb, bytes = collectgarbage "count"
+	skynet.ret(skynet.pack(kb,bytes))
+end
+
+function dbgcmd.GC()
+	coroutine_pool = {}
+	collectgarbage "collect"
+end
+
+function dbgcmd.STAT()
+	local stat = {}
+	stat.mqlen = skynet.mqlen()
+	skynet.ret(skynet.pack(stat))
+end
+
+function dbgcmd.INFO()
+	if internal_info_func then
+		skynet.ret(skynet.pack(internal_info_func()))
+	else
+		skynet.ret(skynet.pack(nil))
+	end
+end
+
+local function _debug_dispatch(session, address, cmd, ...)
+	local f = dbgcmd[cmd]
+	assert(f, cmd)
+	f(...)
+end
+
+skynet.register_protocol {
+	name = "debug",
+	id = assert(skynet.PTYPE_DEBUG),
+	pack = assert(skynet.pack),
+	unpack = assert(skynet.unpack),
+	dispatch = _debug_dispatch,
+}
+
+end

lualib/snax.lua

@@ -1,11 +1,19 @@
 local skynet = require "skynet"
-local snax_interface = require "snax_interface"
+local snax_interface = require "snax.interface"
 
 local snax = {}
 local typeclass = {}
 
 local G = { require = function() end }
 
+skynet.register_protocol {
+	name = "snax",
+	id = skynet.PTYPE_SNAX,
+	pack = skynet.pack,
+	unpack = skynet.unpack,
+}
+
+
 function snax.interface(name)
 	if typeclass[name] then
 		return typeclass[name]
@@ -38,7 +46,7 @@ local function gen_post(type, handle)
 		__index = function( t, k )
 			local id = assert(type.accept[k] , string.format("post %s no exist", k))
 			return function(...)
-				skynet_send(handle, "lua", id, ...)
+				skynet_send(handle, "snax", id, ...)
 			end
 		end })
 end
@@ -48,7 +56,7 @@ local function gen_req(type, handle)
 		__index = function( t, k )
 			local id = assert(type.response[k] , string.format("request %s no exist", k))
 			return function(...)
-				return skynet_call(handle, "lua", id, ...)
+				return skynet_call(handle, "snax", id, ...)
 			end
 		end })
 end
@@ -64,16 +72,14 @@ end
 
 local handle_cache = setmetatable( {} , { __mode = "kv" } )
 
-function snax.newservice(name, ...)
+function snax.rawnewservice(name, ...)
 	local t = snax.interface(name)
 	local handle = skynet.newservice("snaxd", name)
 	assert(handle_cache[handle] == nil)
 	if t.system.init then
-		skynet.call(handle, "lua", t.system.init, ...)
+		skynet.call(handle, "snax", t.system.init, ...)
 	end
-	local ret = wrapper(handle, name, t)
+	return handle
-	handle_cache[handle] = ret
-	return ret
 end
 
 function snax.bind(handle, type)
@@ -88,9 +94,42 @@ function snax.bind(handle, type)
 	return ret
 end
 
+function snax.newservice(name, ...)
+	local handle = snax.rawnewservice(name, ...)
+	return snax.bind(handle, name)
+end
+
+local function service_name(global, name, ...)
+	if global == true then
+		return name
+	else
+		return global
+	end
+end
+
+function snax.uniqueservice(name, ...)
+	local handle = assert(skynet.call(".service", "lua", "LAUNCH", "snaxd", name, ...))
+	return snax.bind(handle, name)
+end
+
+function snax.globalservice(name, ...)
+	local handle = assert(skynet.call(".service", "lua", "GLAUNCH", "snaxd", name, ...))
+	return snax.bind(handle, name)
+end
+
+function snax.queryservice(name)
+	local handle = assert(skynet.call(".service", "lua", "QUERY", "snaxd", name))
+	return snax.bind(handle, name)
+end
+
+function snax.queryglobal(name)
+	local handle = assert(skynet.call(".service", "lua", "GQUERY", "snaxd", name))
+	return snax.bind(handle, name)
+end
+
 function snax.kill(obj, ...)
 	local t = snax.interface(obj.type)
-	skynet_call(obj.handle, "lua", t.system.exit, ...)
+	skynet_call(obj.handle, "snax", t.system.exit, ...)
 end
 
 local function test_result(ok, ...)
@@ -103,7 +142,7 @@ end
 
 function snax.hotfix(obj, source, ...)
 	local t = snax.interface(obj.type)
-	return test_result(skynet_call(obj.handle, "lua", t.system.hotfix, source, ...))
+	return test_result(skynet_call(obj.handle, "snax", t.system.hotfix, source, ...))
 end
 
 return snax

lualib/snax/hotfix.lua

@@ -1,4 +1,4 @@
-local si = require "snax_interface"
+local si = require "snax.interface"
 local io = io
 
 local hotfix = {}

lualib/socket.lua

@@ -30,6 +30,11 @@ local function suspend(s)
 	assert(not s.co)
 	s.co = coroutine.running()
 	skynet.wait()
+	-- wakeup closing corouting every time suspend,
+	-- because socket.close() will wait last socket buffer operation before clear the buffer.
+	if s.closing then
+		skynet.wakeup(s.closing)
+	end
 end
 
 -- read skynet_socket.h for these macro
@@ -37,7 +42,7 @@ end
 socket_message[1] = function(id, size, data)
 	local s = socket_pool[id]
 	if s == nil then
-		print("socket: drop package from " .. id)
+		skynet.error("socket: drop package from " .. id)
 		driver.drop(data, size)
 		return
 	end
@@ -95,13 +100,14 @@ end
 socket_message[5] = function(id)
 	local s = socket_pool[id]
 	if s == nil then
-		print("socket: error on unknown", id)
+		skynet.error("socket: error on unknown", id)
 		return
 	end
 	if s.connected then
-		print("socket: error on", id)
+		skynet.error("socket: error on", id)
 	end
 	s.connected = false
+
 	wakeup(s)
 end
 
@@ -145,7 +151,7 @@ function socket.bind(os_fd)
 end
 
 function socket.stdin()
-	return socket.bind(1)
+	return socket.bind(0)
 end
 
 function socket.start(id, func)
@@ -175,7 +181,8 @@ function socket.close(id)
 		-- notice: call socket.close in __gc should be carefully,
 		-- because skynet.wait never return in __gc, so driver.clear may not be called
 		if s.co then
-			-- reading this socket on another coroutine
+			-- reading this socket on another coroutine, so don't shutdown (clear the buffer) immediatel
+			-- wait reading coroutine read the buffer.
 			assert(not s.closing)
 			s.closing = coroutine.running()
 			skynet.wait()
@@ -189,13 +196,6 @@ function socket.close(id)
 	socket_pool[id] = nil
 end
 
-local function close_socket(s)
-	if s.closing then
-		skynet.wakeup(s.closing)
-	end
-	return driver.readall(s.buffer, buffer_pool)
-end
-
 function socket.read(id, sz)
 	local s = socket_pool[id]
 	assert(s)
@@ -204,7 +204,7 @@ function socket.read(id, sz)
 		return ret
 	end
 	if not s.connected then
-		return false, close_socket(s)
+		return false, driver.readall(s.buffer, buffer_pool)
 	end
 
 	assert(not s.read_required)
@@ -214,7 +214,7 @@ function socket.read(id, sz)
 	if ret then
 		return ret
 	else
-		return false, close_socket(s)
+		return false, driver.readall(s.buffer, buffer_pool)
 	end
 end
 
@@ -222,14 +222,14 @@ function socket.readall(id)
 	local s = socket_pool[id]
 	assert(s)
 	if not s.connected then
-		local r = close_socket(s)
+		local r = driver.readall(s.buffer, buffer_pool)
 		return r ~= "" and r
 	end
 	assert(not s.read_required)
 	s.read_required = true
 	suspend(s)
 	assert(s.connected == false)
-	return close_socket(s)
+	return driver.readall(s.buffer, buffer_pool)
 end
 
 function socket.readline(id, sep)
@@ -241,7 +241,7 @@ function socket.readline(id, sep)
 		return ret
 	end
 	if not s.connected then
-		return false, close_socket(s)
+		return false, driver.readall(s.buffer, buffer_pool)
 	end
 	assert(not s.read_required)
 	s.read_required = sep
@@ -249,7 +249,7 @@ function socket.readline(id, sep)
 	if s.connected then
 		return driver.readline(s.buffer, buffer_pool, sep)
 	else
-		return false, close_socket(s)
+		return false, driver.readall(s.buffer, buffer_pool)
 	end
 end
 
@@ -261,9 +261,6 @@ function socket.block(id)
 	assert(not s.read_required)
 	s.read_required = 0
 	suspend(s)
-	if not s.connected and s.closing then
-		skynet.wakeup(s.closing)
-	end
 	return s.connected
 end
 

lualib/socketchannel.lua

@@ -27,8 +27,8 @@ function socket_channel.channel(desc)
 		__host = assert(desc.host),
 		__port = assert(desc.port),
 		__auth = desc.auth,
-		__response = desc.response,
+		__response = desc.response,	-- It's for session mode
-		__request = {},	-- request seq { response func or session }
+		__request = {},	-- request seq { response func or session }	-- It's for order mode
 		__thread = {}, -- coroutine seq or session->coroutine map
 		__result = {}, -- response result { coroutine -> result }
 		__result_data = {},
@@ -51,72 +51,101 @@ local function close_channel_socket(self)
 end
 
 local function wakeup_all(self, errmsg)
-	for i = 1, #self.__thread do
+	if self.__response then
-		local co = self.__thread[i]
+		for k,co in pairs(self.__thread) do
-		self.__thread[i] = nil
+			self.__thread[k] = nil
-		self.__result[co] = socket_error
+			self.__result[co] = socket_error
-		self.__result_data[co] = errmsg
+			self.__result_data[co] = errmsg
-		skynet.wakeup(co)
+			skynet.wakeup(co)
+		end
+	else
+		for i = 1, #self.__request do
+			self.__request[i] = nil
+		end
+		for i = 1, #self.__thread do
+			local co = self.__thread[i]
+			self.__thread[i] = nil
+			self.__result[co] = socket_error
+			self.__result_data[co] = errmsg
+			skynet.wakeup(co)
+		end
 	end
 end
 
-local function dispatch_response(self)
+
+
+local function dispatch_by_session(self)
 	local response = self.__response
-	if response then
+	-- response() return session
-		-- response() return session
+	while self.__sock do
-		while self.__sock do
+		local ok , session, result_ok, result_data = pcall(response, self.__sock)
-			local ok , session, result_ok, result_data = pcall(response, self.__sock)
+		if ok and session then
-			if ok and session then
+			local co = self.__thread[session]
-				local co = self.__thread[session]
+			self.__thread[session] = nil
-				self.__thread[session] = nil
+			if co then
-				if co then
+				self.__result[co] = result_ok
-					self.__result[co] = result_ok
+				self.__result_data[co] = result_data
-					self.__result_data[co] = result_data
+				skynet.wakeup(co)
-					skynet.wakeup(co)
+			else
-				else
+				skynet.error("socket: unknown session :", session)
-					print("socket: unknown session :", session)
+			end
-				end
+		else
+			close_channel_socket(self)
+			local errormsg
+			if session ~= socket_error then
+				errormsg = session
+			end
+			wakeup_all(self, errormsg)
+		end
+	end
+end
+
+local function pop_response(self)
+	return table.remove(self.__request, 1), table.remove(self.__thread, 1)
+end
+
+local function push_response(self, response, co)
+	if self.__response then
+		-- response is session
+		self.__thread[response] = co
+	else
+		-- response is a function, push it to __request
+		table.insert(self.__request, response)
+		table.insert(self.__thread, co)
+	end
+end
+
+local function dispatch_by_order(self)
+	while self.__sock do
+		local func, co = pop_response(self)
+		if func == nil then
+			if not socket.block(self.__sock[1]) then
+				close_channel_socket(self)
+				wakeup_all(self)
+			end
+		else
+			local ok, result_ok, result_data = pcall(func, self.__sock)
+			if ok then
+				self.__result[co] = result_ok
+				self.__result_data[co] = result_data
+				skynet.wakeup(co)
 			else
 				close_channel_socket(self)
-				local errormsg
+				local errmsg
-				if session ~= socket_error then
+				if result ~= socket_error then
-					errormsg = session
+					errmsg = result_ok
-				end
-				for k,co in pairs(self.__thread) do
-					-- throw error (errormsg)
-					self.__thread[k] = nil
-					self.__result[co] = socket_error
-					self.__result_data[co] = errormsg
-					skynet.wakeup(co)
 				end
+				wakeup_all(self, errmsg)
 			end
 		end
+	end
+end
+
+local function dispatch_function(self)
+	if self.__response then
+		return dispatch_by_session
 	else
-		-- pop response function from __request
+		return dispatch_by_order
-		while self.__sock do
-			local func = table.remove(self.__request, 1)
-			if func == nil then
-				if not socket.block(self.__sock[1]) then
-					close_channel_socket(self)
-					wakeup_all(self)
-				end
-			else
-				local ok, result_ok, result_data = pcall(func, self.__sock)
-				if ok then
-					local co = table.remove(self.__thread, 1)
-					self.__result[co] = result_ok
-					self.__result_data[co] = result_data
-					skynet.wakeup(co)
-				else
-					close_channel_socket(self)
-					local errmsg
-					if result ~= socket_error then
-						errmsg = result_ok
-					end
-					wakeup_all(self, errmsg)
-				end
-			end
-		end
 	end
 end
 
@@ -128,14 +157,14 @@ local function connect_once(self)
 	end
 	self.__authcoroutine = coroutine.running()
 	self.__sock = setmetatable( {fd} , channel_socket_meta )
-	skynet.fork(dispatch_response, self)
+	skynet.fork(dispatch_function(self), self)
 
 	if self.__auth then
 		local ok , message = pcall(self.__auth, self)
 		if not ok then
 			close_channel_socket(self)
 			if message ~= socket_error then
-				print("socket: auth failed", message)
+				skynet.error("socket: auth failed", message)
 			end
 		end
 		self.__authcoroutine = false
@@ -150,12 +179,15 @@ local function try_connect(self , once)
 	local t = 100
 	while not self.__closed do
 		if connect_once(self) then
+			if not once then
+				skynet.error("socket: connect to", self.__host, self.__port)
+			end
 			return
 		elseif once then
 			error(string.format("Connect to %s:%d failed", self.__host, self.__port))
 		end
 		if t > 1000 then
-			print("socket: try to reconnect", self.__host, self.__port)
+			skynet.error("socket: try to reconnect", self.__host, self.__port)
 			skynet.sleep(t)
 			t = 0
 		else
@@ -179,6 +211,7 @@ local function block_connect(self, once)
 	if self.__closed then
 		return false
 	end
+
 	if #self.__connecting > 0 then
 		-- connecting in other coroutine
 		local co = coroutine.running()
@@ -187,7 +220,6 @@ local function block_connect(self, once)
 		-- check connection again
 		return block_connect(self, once)
 	end
-
 	self.__connecting[1] = true
 	try_connect(self, once)
 	self.__connecting[1] = nil
@@ -209,10 +241,30 @@ function channel:connect(once)
 	return block_connect(self, once)
 end
 
+local function wait_for_response(self, response)
+	local co = coroutine.running()
+	push_response(self, response, co)
+	skynet.wait()
+
+	local result = self.__result[co]
+	self.__result[co] = nil
+	local result_data = self.__result_data[co]
+	self.__result_data[co] = nil
+
+	if result == socket_error then
+		error(socket_error)
+	else
+		assert(result, result_data)
+		return result_data
+	end
+end
+
 function channel:request(request, response)
 	assert(block_connect(self))
 
 	if not socket.write(self.__sock[1], request) then
+		close_channel_socket(self)
+		wakeup_all(self)
 		error(socket_error)
 	end
 
@@ -221,59 +273,13 @@ function channel:request(request, response)
 		return
 	end
 
-	local co = coroutine.running()
+	return wait_for_response(self, response)
-
-	if self.__response then
-		-- response is session
-		self.__thread[response] = co
-	else
-		-- response is a function, push it to __request
-		table.insert(self.__request, response)
-		table.insert(self.__thread, co)
-	end
-	skynet.wait()
-
-	local result = self.__result[co]
-	self.__result[co] = nil
-	local result_data = self.__result_data[co]
-	self.__result_data[co] = nil
-
-	if result == socket_error then
-		if result_data then
-			print("socket: dispatch", request, result_data)
-		end
-		error(socket_error)
-	else
-		assert(result, result_data)
-		return result_data
-	end
 end
 
 function channel:response(response)
 	assert(block_connect(self))
 
-	assert(type(response) == "function")
+	return wait_for_response(self, response)
-
-	local co = coroutine.running()
-	table.insert(self.__request, response)
-	table.insert(self.__thread, co)
-
-	skynet.wait()
-
-	local result = self.__result[co]
-	self.__result[co] = nil
-	local result_data = self.__result_data[co]
-	self.__result_data[co] = nil
-
-	if result == socket_error then
-		if result_data then
-			print("socket: dispatch", request, result_data)
-		end
-		error(socket_error)
-	else
-		assert(result, result_data)
-		return result_data
-	end
 end
 
 function channel:close()

platform.mk

@@ -5,30 +5,36 @@ CC ?= gcc
 
 .PHONY : none $(PLATS) clean all cleanall
 
-ifneq ($(PLAT), none)
+#ifneq ($(PLAT), none)
 
 .PHONY : default
 
 default :
 	$(MAKE) $(PLAT)
 
-endif
+#endif
 
 none :
 	@echo "Please do 'make PLATFORM' where PLATFORM is one of these:"
 	@echo "   $(PLATS)"
 
-LIBS := -lpthread -lm
+SKYNET_LIBS := -lpthread -lm
 SHARED := -fPIC --shared
 EXPORT := -Wl,-E
 
-$(PLATS) : all 
-
 linux : PLAT = linux
 macosx : PLAT = macosx
 freebsd : PLAT = freebsd
 
-macosx linux : LIBS += -ldl
 macosx : SHARED := -fPIC -dynamiclib -Wl,-undefined,dynamic_lookup
 macosx : EXPORT :=
-linux freebsd : LIBS += -lrt
+macosx linux : SKYNET_LIBS += -ldl
+linux freebsd : SKYNET_LIBS += -lrt
+
+# Turn off jemalloc and malloc hook on macosx
+
+macosx : MALLOC_STATICLIB :=
+macosx : SKYNET_DEFINES :=-DNOUSE_JEMALLOC
+
+linux macosx freebsd :
+	$(MAKE) all PLAT=$@ SKYNET_LIBS="$(SKYNET_LIBS)" SHARED="$(SHARED)" EXPORT="$(EXPORT)" MALLOC_STATICLIB="$(MALLOC_STATICLIB)" SKYNET_DEFINES="$(SKYNET_DEFINES)"

service-src/service_harbor.c

@@ -47,6 +47,9 @@ struct remote_message_header {
 	uint32_t session;
 };
 
+// 12 is sizeof(struct remote_message_header)
+#define HEADER_COOKIE_LENGTH 12
+
 struct harbor {
 	struct skynet_context *ctx;
 	char * local_addr;
@@ -402,9 +405,8 @@ _remote_send_handle(struct harbor *h, uint32_t source, uint32_t destination, int
 		_send_remote(context, fd, msg,sz,&cookie);
 	} else {
 		// throw an error return to source
-		if (session != 0) {
+		// report the destination is dead
-			skynet_send(context, destination, source, PTYPE_RESERVED_ERROR, session, NULL, 0);
+		skynet_send(context, destination, source, PTYPE_ERROR, 0 , NULL, 0);
-		}
 		skynet_error(context, "Drop message to harbor %d from %x to %x (session = %d, msgsz = %d)",harbor_id, source, destination,session,(int)sz);
 	}
 	return 0;
@@ -503,23 +505,23 @@ _mainloop(struct skynet_context * context, void * ud, int type, int session, uin
 	case PTYPE_HARBOR: {
 		// remote message in
 		const char * cookie = msg;
-		cookie += sz - 12;
+		cookie += sz - HEADER_COOKIE_LENGTH;
 		struct remote_message_header header;
 		_message_to_header((const uint32_t *)cookie, &header);
 		if (header.source == 0) {
 			if (header.destination < REMOTE_MAX) {
 				// 1 byte harbor id (0~255)
 				// update remote harbor address
-				char ip [sz - 11];
+				char ip [sz - HEADER_COOKIE_LENGTH + 1];
-				memcpy(ip, msg, sz-12);
+				memcpy(ip, msg, sz-HEADER_COOKIE_LENGTH);
-				ip[sz-11] = '\0';
+				ip[sz-HEADER_COOKIE_LENGTH] = '\0';
 				_update_remote_address(h, header.destination, ip);
 			} else {
 				// update global name
-				if (sz - 12 > GLOBALNAME_LENGTH) {
+				if (sz - HEADER_COOKIE_LENGTH > GLOBALNAME_LENGTH) {
-					char name[sz-11];
+					char name[sz-HEADER_COOKIE_LENGTH+1];
-					memcpy(name, msg, sz-12);
+					memcpy(name, msg, sz-HEADER_COOKIE_LENGTH);
-					name[sz-11] = '\0';
+					name[sz-HEADER_COOKIE_LENGTH] = '\0';
 					skynet_error(context, "Global name is too long %s", name);
 				}
 				_update_remote_name(h, msg, header.destination);
@@ -528,7 +530,7 @@ _mainloop(struct skynet_context * context, void * ud, int type, int session, uin
 			uint32_t destination = header.destination;
 			int type = (destination >> HANDLE_REMOTE_SHIFT) | PTYPE_TAG_DONTCOPY;
 			destination = (destination & HANDLE_MASK) | ((uint32_t)h->id << HANDLE_REMOTE_SHIFT);
-			skynet_send(context, header.source, destination, type, (int)header.session, (void *)msg, sz-12);
+			skynet_send(context, header.source, destination, type, (int)header.session, (void *)msg, sz-HEADER_COOKIE_LENGTH);
 			return 1;
 		}
 		return 0;
@@ -593,6 +595,8 @@ harbor_init(struct harbor *h, struct skynet_context *ctx, const char * args) {
 		fprintf(stderr, "Harbor: Connect to master failed\n");
 		exit(1);
 	}
+	skynet_harbor_start(ctx);
+
 	h->local_addr = skynet_strdup(local_addr);
 
 	_launch_gate(ctx, local_addr);

service-src/service_lua.h

@@ -1,10 +0,0 @@
-#ifndef SKYNET_SERVICE_LUA_H
-#define SKYNET_SERVICE_LUA_H
-
-struct snlua {
-	lua_State * L;
-	struct skynet_context * ctx;
-	int (*init)(struct snlua *l, struct skynet_context *ctx, const char * args);
-};
-
-#endif

service-src/service_snlua.c

@@ -4,13 +4,17 @@
 #include <lualib.h>
 #include <lauxlib.h>
 #include <assert.h>
-#include "service_lua.h"
 
 #include <assert.h>
 #include <string.h>
 #include <stdlib.h>
 #include <stdio.h>
 
+struct snlua {
+	lua_State * L;
+	struct skynet_context * ctx;
+};
+
 // LUA_CACHELIB may defined in patched lua for shared proto
 #ifdef LUA_CACHELIB
 
@@ -30,121 +34,13 @@ codecache(lua_State *L) {
 		{ NULL, NULL },
 	};
 	luaL_newlib(L,l);
-  lua_getglobal(L, "loadfile");
+	lua_getglobal(L, "loadfile");
-  lua_setfield(L, -2, "loadfile");
+	lua_setfield(L, -2, "loadfile");
 	return 1;
 }
 
 #endif
 
-// time
-
-#include <time.h>
-#define NANOSEC 1000000000
-
-#if defined(__APPLE__)
-#include <mach/task.h>
-#include <mach/mach.h>
-#endif
-
-static void
-current_time(struct timespec *ti) {
-#if  !defined(__APPLE__)
-	clock_gettime(CLOCK_THREAD_CPUTIME_ID, ti);
-#else
-	struct task_thread_times_info aTaskInfo;
-	mach_msg_type_number_t aTaskInfoCount = TASK_THREAD_TIMES_INFO_COUNT;
-	assert(KERN_SUCCESS == task_info(mach_task_self(), TASK_THREAD_TIMES_INFO, (task_info_t )&aTaskInfo, &aTaskInfoCount));
-	ti->tv_sec = aTaskInfo.user_time.seconds;
-	ti->tv_nsec = aTaskInfo.user_time.microseconds * 1000;
-#endif
-}
-
-static double
-diff_time(struct timespec *ti) {
-	struct timespec end;
-	current_time(&end);
-	int diffsec = end.tv_sec - ti->tv_sec;
-	int diffnsec = end.tv_nsec - ti->tv_nsec;
-	if (diffnsec < 0) {
-		--diffsec;
-		diffnsec += NANOSEC;
-	}
-	return (double)diffsec + (double)diffnsec / NANOSEC;
-}
-
-static int
-_try_load(lua_State *L, const char * path, int pathlen, const char * name) {
-	int namelen = strlen(name);
-	char tmp[pathlen + namelen];
-	int i;
-	for (i=0;i<pathlen;i++) {
-		if (path[i] == '?')
-			break;
-		tmp[i] = path[i];
-	}
-	if (path[i] == '?') {
-		memcpy(tmp+i,name,namelen);
-		memcpy(tmp+i+namelen,path+i+1,pathlen - i -1);
-	} else {
-		fprintf(stderr,"snlua : Invalid lua service path\n");
-		exit(1);
-	}
-	tmp[namelen+pathlen-1] = '\0';
-	int r = luaL_loadfile(L,tmp);
-	if (r == LUA_OK) {
-		int i;
-		for (i=namelen+pathlen-2;i>=0;i--) {
-			if (tmp[i] == '/') {
-				lua_pushlstring(L,tmp,i+1);
-				lua_setglobal(L,"SERVICE_PATH");
-				break;
-			}
-		}
-		if (i<0) {
-			return 0;
-		}
-		lua_getglobal(L,"package");
-		lua_getfield(L,-1,"path");
-		luaL_Buffer b;
-		luaL_buffinit(L, &b);
-		luaL_addlstring(&b, tmp, i+1);
-		luaL_addstring(&b, "?.lua;");
-		luaL_addvalue(&b);
-		luaL_pushresult(&b);
-		lua_setfield(L,-2,"path");
-		lua_pop(L,1);
-		return 0;
-	} else if (r == LUA_ERRFILE) {
-		lua_pop(L,1);
-		return -1;
-	}
-	return 1;
-}
-
-static int
-_load(lua_State *L, char ** filename) {
-	const char * name = strsep(filename, " \r\n");
-	const char * path = skynet_command(NULL, "GETENV", "luaservice");
-	while (path[0]) {
-		int pathlen;
-		char * pathend = strchr(path,';');
-		if (pathend) {
-			pathlen = pathend - path;
-		} else {
-			pathlen = strlen(path);
-		}
-		int r = _try_load(L, path, pathlen, name);
-		if (r >=0) {
-			return r;
-		}
-		path+=pathlen;
-		if (path[0]==';')
-			++path;
-	}
-	return -1;
-}
-
 static int 
 traceback (lua_State *L) {
 	const char *msg = lua_tostring(L, 1);
@@ -156,80 +52,70 @@ traceback (lua_State *L) {
 	return 1;
 }
 
-static void
-_report_error(lua_State *L, struct skynet_context *ctx, const char *filename, int err) {
-	switch (err) {
-	case LUA_ERRRUN:
-		skynet_error(ctx, "lua do [%s] error : %s", filename, lua_tostring(L,-1));
-		break;
-	case LUA_ERRMEM:
-		skynet_error(ctx, "lua memory error : %s",filename);
-		break;
-	case LUA_ERRERR:
-		skynet_error(ctx, "lua message error : %s",filename);
-		break;
-	case LUA_ERRGCMM:
-		skynet_error(ctx, "lua gc error : %s",filename);
-		break;
-	};
-	lua_pop(L,1);
-}
-
 static void
 _report_launcher_error(struct skynet_context *ctx) {
 	// sizeof "ERROR" == 5
 	skynet_sendname(ctx, ".launcher", PTYPE_TEXT, 0, "ERROR", 5);
 }
 
+static const char *
+optstring(struct skynet_context *ctx, const char *key, const char * str) {
+	const char * ret = skynet_command(ctx, "GETENV", key);
+	if (ret == NULL) {
+		return str;
+	}
+	return ret;
+}
+
 static int
-_init(struct snlua *l, struct skynet_context *ctx, const char * args) {
+_init(struct snlua *l, struct skynet_context *ctx, const char * args, size_t sz) {
 	lua_State *L = l->L;
 	l->ctx = ctx;
 	lua_gc(L, LUA_GCSTOP, 0);
+	lua_pushboolean(L, 1);  /* signal for libraries to ignore env. vars. */
+	lua_setfield(L, LUA_REGISTRYINDEX, "LUA_NOENV");
 	luaL_openlibs(L);
-	lua_pushlightuserdata(L, l);
+	lua_pushlightuserdata(L, ctx);
-	lua_setfield(L, LUA_REGISTRYINDEX, "skynet_lua");
+	lua_setfield(L, LUA_REGISTRYINDEX, "skynet_context");
 	luaL_requiref(L, "skynet.codecache", codecache , 0);
 	lua_pop(L,1);
-	lua_gc(L, LUA_GCRESTART, 0);
 
-	char tmp[strlen(args)+1];
+	const char *path = optstring(ctx, "lua_path","./lualib/?.lua;./lualib/?/init.lua");
-	char *parm = tmp;
+	lua_pushstring(L, path);
-	strcpy(parm,args);
+	lua_setglobal(L, "LUA_PATH");
+	const char *cpath = optstring(ctx, "lua_cpath","./luaclib/?.so");
+	lua_pushstring(L, cpath);
+	lua_setglobal(L, "LUA_CPATH");
+	const char *service = optstring(ctx, "luaservice", "./service/?.lua");
+	lua_pushstring(L, service);
+	lua_setglobal(L, "LUA_SERVICE");
+	const char *preload = skynet_command(ctx, "GETENV", "preload");
+	lua_pushstring(L, preload);
+	lua_setglobal(L, "LUA_PRELOAD");
 
 	lua_pushcfunction(L, traceback);
-	int traceback_index = lua_gettop(L);
+	assert(lua_gettop(L) == 1);
-	assert(traceback_index == 1);
 
-	const char * filename = parm;
+	const char * loader = optstring(ctx, "lualoader", "./lualib/loader.lua");
-	int r = _load(L, &parm);
+
-	if (r != 0) {
+	int r = luaL_loadfile(L,loader);
-		if (r<0) {
+	if (r != LUA_OK) {
-			skynet_error(ctx, "lua parser [%s] load error", filename);
+		skynet_error(ctx, "Can't load %s : %s", loader, lua_tostring(L, -1));
-		} else {
-			skynet_error(ctx, "lua parser [%s] error : %s", filename, lua_tostring(L,-1));
-		}
 		_report_launcher_error(ctx);
 		return 1;
 	}
-	int n=0;
+	lua_pushlstring(L, args, sz);
-	while(parm) {
+	r = lua_pcall(L,1,0,1);
-		const char * arg = strsep(&parm, " \r\n");
+	if (r != LUA_OK) {
-		if (arg && arg[0]!='\0') {
+		skynet_error(ctx, "lua loader error : %s", lua_tostring(L, -1));
-			lua_pushstring(L, arg);
+		_report_launcher_error(ctx);
-			++n;
+		return 1;
-		}
 	}
-	r = lua_pcall(L,n,0,traceback_index);
+	lua_settop(L,0);
-	if (r == LUA_OK) {
+
-		r = lua_gc(L, LUA_GCCOLLECT, 0);
+	lua_gc(L, LUA_GCRESTART, 0);
-		if (r == LUA_OK) {
+
-			return 0;
+	return 0;
-		}
-	}
-	_report_error(L, ctx, filename, r);
-	_report_launcher_error(ctx);
-	return 1;
 }
 
 static int
@@ -237,12 +123,7 @@ _launch(struct skynet_context * context, void *ud, int type, int session, uint32
 	assert(type == 0 && session == 0);
 	struct snlua *l = ud;
 	skynet_callback(context, NULL, NULL);
-	struct timespec ti;
+	int err = _init(l, context, msg, sz);
-	current_time(&ti);
-	int err = _init(l, context, msg);
-	double t = diff_time(&ti);
-	lua_pushnumber(l->L, t);
-	lua_setfield(l->L, LUA_REGISTRYINDEX, "skynet_boottime");
 	if (err) {
 		skynet_command(context, "EXIT", NULL);
 	}
@@ -253,13 +134,13 @@ _launch(struct skynet_context * context, void *ud, int type, int session, uint32
 int
 snlua_init(struct snlua *l, struct skynet_context *ctx, const char * args) {
 	int sz = strlen(args);
-	char * tmp = skynet_malloc(sz+1);
+	char * tmp = skynet_malloc(sz);
-	memcpy(tmp, args, sz+1);
+	memcpy(tmp, args, sz);
 	skynet_callback(ctx, l , _launch);
 	const char * self = skynet_command(ctx, "REG", NULL);
 	uint32_t handle_id = strtoul(self+1, NULL, 16);
 	// it must be first message
-	skynet_send(ctx, 0, handle_id, PTYPE_TAG_DONTCOPY,0, tmp, sz+1);
+	skynet_send(ctx, 0, handle_id, PTYPE_TAG_DONTCOPY,0, tmp, sz);
 	return 0;
 }
 
@@ -268,7 +149,6 @@ snlua_create(void) {
 	struct snlua * l = skynet_malloc(sizeof(*l));
 	memset(l,0,sizeof(*l));
 	l->L = lua_newstate(skynet_lalloc, NULL);
-	l->init = _init;
 	return l;
 }
 

service/bootstrap.lua

@@ -0,0 +1,28 @@
+local skynet = require "skynet"
+
+skynet.start(function()
+	assert(skynet.launch("logger", skynet.getenv "logger"))
+
+	local standalone = skynet.getenv "standalone"
+	local master_addr = skynet.getenv "master"
+
+	if standalone then
+		assert(skynet.launch("master", master_addr))
+	end
+
+	local local_addr = skynet.getenv "address"
+	local harbor_id = skynet.getenv "harbor"
+
+	assert(skynet.launch("harbor",master_addr, local_addr, harbor_id))
+
+	local launcher = assert(skynet.launch("snlua","launcher"))
+	skynet.name(".launcher", launcher)
+
+	if standalone then
+		local datacenter = assert(skynet.newservice "datacenterd")
+		skynet.name("DATACENTER", datacenter)
+	end
+	assert(skynet.newservice "service_mgr")
+	assert(skynet.newservice(skynet.getenv "start" or "main"))
+	skynet.exit()
+end)

service/datacenterd.lua

@@ -0,0 +1,43 @@
+local skynet = require "skynet"
+
+local command = {}
+local database = {}
+
+local function query(db, key, ...)
+	if key == nil then
+		return db
+	else
+		return query(db[key], ...)
+	end
+end
+
+function command.QUERY(key, ...)
+	local d = database[key]
+	if d then
+		return query(d, ...)
+	end
+end
+
+local function update(db, key, value, ...)
+	if select("#",...) == 0 then
+		local ret = db[key]
+		db[key] = value
+		return ret
+	else
+		if db[key] == nil then
+			db[key] = {}
+		end
+		return update(db[key], value, ...)
+	end
+end
+
+function command.UPDATE(...)
+	return update(database, ...)
+end
+
+skynet.start(function()
+	skynet.dispatch("lua", function (_, source, cmd, ...)
+		local f = assert(command[cmd])
+		skynet.ret(skynet.pack(f(...)))
+	end)
+end)

service/debug_console.lua

@@ -1,6 +1,7 @@
 local skynet = require "skynet"
 local codecache = require "skynet.codecache"
 local socket = require "socket"
+local snax = require "snax"
 
 local port = tonumber(...)
 local COMMAND = {}
@@ -109,7 +110,9 @@ function COMMAND.help()
 		mem = "mem : show memory status",
 		gc = "gc : force every lua service do garbage collect",
 		start = "lanuch a new lua service",
+		snax = "lanuch a new snax service",
 		clearcache = "clear lua code cache",
+		service = "List unique service",
 	}
 end
 
@@ -126,4 +129,16 @@ function COMMAND.start(...)
 	end
 end
 
+function COMMAND.snax(...)
+	local s = snax.newservice(...)
+	if s then
+		local addr = s.handle
+		return { [skynet.address(addr)] = ... }
+	else
+		return "Failed"
+	end
+end
 
+function COMMAND.service()
+	return skynet.call("SERVICE", "lua", "LIST")
+end

service/gate.lua

@@ -85,7 +85,7 @@ function MSG.data(fd, msg, sz)
 	if agent then
 		skynet.redirect(agent, c.client, "client", 0, msg, sz)
 	else
-		skynet.send(watchdog, "lua", "socket", "data", netpack.tostring(msg, sz))
+		skynet.send(watchdog, "lua", "socket", "data", fd, netpack.tostring(msg, sz))
 	end
 end
 

service/launcher.lua

@@ -2,8 +2,8 @@ local skynet = require "skynet"
 local string = string
 
 local services = {}
-
 local command = {}
+local instance = {} -- for confirm (function command.LAUNCH / command.ERROR / command.LAUNCHOK)
 
 local function handle_to_address(handle)
 	return tonumber("0x" .. string.sub(handle , 2))
@@ -19,18 +19,6 @@ function command.LIST()
 	return list
 end
 
-function command.RELOAD(handle)
-	handle = handle_to_address(handle)
-	local cmd = string.match(services[handle], "snlua (.+)")
-	print(services[handle],cmd)
-	if cmd then
-		skynet.send(handle,"debug","RELOAD",cmd)
-		return {cmd}
-	else
-		return {"Support only snlua"}
-	end
-end
-
 function command.STAT()
 	local list = {}
 	for k,v in pairs(services) do
@@ -40,7 +28,7 @@ function command.STAT()
 	return list
 end
 
-function command.INFO(handle)
+function command.INFO(_, _, handle)
 	handle = handle_to_address(handle)
 	if services[handle] == nil then
 		return
@@ -50,23 +38,7 @@ function command.INFO(handle)
 	end
 end
 
-function command.TIMING(handle)
+function command.KILL(_, _, handle)
-	handle = handle_to_address(handle)
-	if services[handle] == nil then
-		return
-	else
-		local r = skynet.call(handle,"debug","TIMING")
-		local result = {}
-		for k,v in pairs(r) do
-			v.name = services[k]
-			v.avg = v.ti/v.n
-			result[skynet.address(k)] = v
-		end
-		return result
-	end
-end
-
-function command.KILL(handle)
 	handle = handle_to_address(handle)
 	skynet.kill(handle)
 	local ret = { [skynet.address(handle)] = tostring(services[handle]) }
@@ -90,48 +62,71 @@ function command.GC()
 	return command.MEM()
 end
 
-function command.REMOVE(handle)
+function command.REMOVE(_,_, handle)
 	services[handle] = nil
 	-- don't return (skynet.ret) because the handle may exit
 	return NORET
 end
 
-local instance = {}
+function command.LAUNCH(address, session, service, ...)
-
+	local param = table.concat({...}, " ")
-skynet.dispatch("text" , function(session, address , cmd)
+	local inst = skynet.launch(service, param)
-	if cmd == "" then
+	if inst then
-		-- init notice
+		services[inst] = service .. " " .. param
-		local reply = instance[address]
+		instance[inst] = { session = session, address = address }
-		if reply then
-			skynet.redirect(reply.address , 0, "response", reply.session, skynet.address(address))
-			instance[address] = nil
-		end
-	elseif cmd == "ERROR" then
-		-- see serivce-src/service_lua.c
-		-- init failed
-		local reply = instance[address]
-		if reply then
-			skynet.redirect(reply.address , 0, "response", reply.session, "")
-			instance[address] = nil
-		end
 	else
-		-- launch request
+		skynet.ret("")	-- launch failed
-		local service, param = string.match(cmd,"([^ ]+) (.*)")
-		local inst = skynet.launch(service, param)
-		if inst then
-			services[inst] = cmd
-			instance[inst] = { session = session, address = address }
-		else
-			skynet.ret("")
-		end
 	end
-end)
+	return NORET
+end
+
+function command.ERROR(address)
+	-- see serivce-src/service_lua.c
+	-- init failed
+	local reply = instance[address]
+	if reply then
+		skynet.redirect(reply.address , 0, "response", reply.session, "")
+		instance[address] = nil
+	end
+	services[address] = nil
+	return NORET
+end
+
+function command.LAUNCHOK(address)
+	-- init notice
+	local reply = instance[address]
+	if reply then
+		skynet.redirect(reply.address , 0, "response", reply.session, skynet.address(address))
+		instance[address] = nil
+	end
+
+	return NORET
+end
+
+-- for historical reasons, launcher support text command (for C service)
+
+skynet.register_protocol {
+	name = "text",
+	id = skynet.PTYPE_TEXT,
+	unpack = skynet.tostring,
+	dispatch = function(session, address , cmd)
+		if cmd == "" then
+			command.LAUNCHOK(address)
+		elseif cmd == "ERROR" then
+			command.ERROR(address)
+		else
+			-- launch request
+			local service, param = string.match(cmd,"([^ ]+) (.*)")
+			command.LAUNCH(address, session, service, param)
+		end
+	end,
+}
 
 skynet.dispatch("lua", function(session, address, cmd , ...)
 	cmd = string.upper(cmd)
 	local f = command[cmd]
 	if f then
-		local ret = f(...)
+		local ret = f(address, session, ...)
 		if ret ~= NORET then
 			skynet.ret(skynet.pack(ret))
 		end

service/multicastd.lua

@@ -0,0 +1,194 @@
+local skynet = require "skynet"
+local mc = require "multicast.c"
+local datacenter = require "datacenter"
+
+local harbor_id = skynet.harbor(skynet.self())
+
+local command = {}
+local channel = {}
+local channel_n = {}
+local channel_remote = {}
+local channel_id = harbor_id
+local NORET = {}
+
+local function get_address(t, id)
+	local v = assert(datacenter.get("multicast", id))
+	t[id] = v
+	return v
+end
+
+local node_address = setmetatable({}, { __index = get_address })
+
+-- new LOCAL channel , The low 8bit is the same with harbor_id
+function command.NEW()
+	while channel[channel_id] do
+		channel_id = mc.nextid(channel_id)
+	end
+	channel[channel_id] = {}
+	channel_n[channel_id] = 0
+	local ret = channel_id
+	channel_id = mc.nextid(channel_id)
+	return ret
+end
+
+-- MUST call by the owner node of channel, delete a remote channel
+function command.DELR(source, c)
+	channel[c] = nil
+	channel_n[c] = nil
+	return NORET
+end
+
+-- delete a channel, if the channel is remote, forward the command to the owner node
+-- otherwise, delete the channel, and call all the remote node, DELR
+function command.DEL(source, c)
+	local node = c % 256
+	if node ~= harbor_id then
+		skynet.send(node_address[node], "lua", "DEL", c)
+		return NORET
+	end
+	local remote = channel_remote[c]
+	channel[c] = nil
+	channel_n[c] = nil
+	channel_remote[c] = nil
+	for node in pairs(remote) do
+		skynet.send(node_address[node], "lua", "DELR", c)
+	end
+	return NORET
+end
+
+-- forward multicast message to a node (channel id use the session field)
+local function remote_publish(node, channel, source, ...)
+	skynet.redirect(node_address[node], source, "multicast", channel, ...)
+end
+
+-- publish a message, for local node, use the message pointer (call mc.bind to add the reference)
+-- for remote node, call remote_publish. (call mc.unpack and skynet.tostring to convert message pointer to string)
+local function publish(c , source, pack, size)
+	local group = channel[c]
+	if group == nil then
+		-- dead channel, delete the pack
+		mc.bind(pack, 1)
+		mc.close(pack)
+		return
+	end
+	mc.bind(pack, channel_n[c])
+	local msg = skynet.tostring(pack, size)
+	for k in pairs(group) do
+		skynet.redirect(k, source, "multicast", c , msg)
+	end
+	local remote = channel_remote[c]
+	if remote then
+		local _, msg, sz = mc.unpack(pack, size)
+		local msg = skynet.tostring(msg,sz)
+		for node in pairs(remote) do
+			remote_publish(node, c, source, msg)
+		end
+	end
+end
+
+skynet.register_protocol {
+	name = "multicast",
+	id = skynet.PTYPE_MULTICAST,
+	unpack = function(msg, sz)
+		return mc.packremote(msg, sz)
+	end,
+	dispatch = publish,
+}
+
+-- publish a message, if the caller is remote, forward the message to the owner node (by remote_publish)
+-- If the caller is local, call publish
+function command.PUB(source, c, pack, size)
+	assert(skynet.harbor(source) == harbor_id)
+	local node = c % 256
+	if node ~= harbor_id then
+		-- remote publish
+		remote_publish(node, c, source, mc.remote(pack))
+	else
+		publish(c, source, pack,size)
+	end
+end
+
+-- the node (source) subscribe a channel
+-- MUST call by channel owner node (assert source is not local and channel is create by self)
+-- If channel is not exist, return true
+-- Else set channel_remote[channel] true
+function command.SUBR(source, c)
+	local node = skynet.harbor(source)
+	if not channel[c] then
+		-- channel none exist
+		return true
+	end
+	assert(node ~= harbor_id and c % 256 == harbor_id)
+	local group = channel_remote[c]
+	if group == nil then
+		group = {}
+		channel_remote[c] = group
+	end
+	group[node] = true
+end
+
+-- the service (source) subscribe a channel
+-- If the channel is remote, node subscribe it by send a SUBR to the owner .
+function command.SUB(source, c)
+	local node = c % 256
+	if node ~= harbor_id then
+		-- remote group
+		if channel[c] == nil then
+			if skynet.call(node_address[node], "lua", "SUBR", c) then
+				return
+			end
+			if channel[c] == nil then
+				-- double check, because skynet.call whould yield, other SUB may occur.
+				channel[c] = {}
+				channel_n[c] = 0
+			end
+		end
+	end
+	local group = channel[c]
+	if group and not group[source] then
+		channel_n[c] = channel_n[c] + 1
+		group[source] = true
+	end
+end
+
+-- MUST call by a node, unsubscribe a channel
+function command.USUBR(source, c)
+	local node = skynet.harbor(source)
+	assert(node ~= harbor_id)
+	local group = assert(channel_remote[c])
+	group[node] = nil
+	return NORET
+end
+
+-- Unsubscribe a channel, if the subscriber is empty and the channel is remote, send USUBR to the channel owner
+function command.USUB(source, c)
+	local group = assert(channel[c])
+	if group[source] then
+		group[source] = nil
+		channel_n[c] = channel_n[c] - 1
+		if channel_n[c] == 0 then
+			local node = c % 256
+			if node ~= harbor_id then
+				-- remote group
+				channel[c] = nil
+				channel_n[c] = nil
+				skynet.send(node_address[node], "lua", "USUBR", c)
+			end
+		end
+	end
+	return NORET
+end
+
+skynet.start(function()
+	skynet.dispatch("lua", function(_,source, cmd, ...)
+		local f = assert(command[cmd])
+		local result = f(source, ...)
+		if result ~= NORET then
+			skynet.ret(skynet.pack(result))
+		end
+	end)
+	local self = skynet.self()
+	local id = skynet.harbor(self)
+	assert(datacenter.set("multicast", id, self) == nil)
+end)
+

service/service_mgr.lua

@@ -1,72 +1,191 @@
 local skynet = require "skynet"
+local snax = require "snax"
 
 local cmd = {}
 local service = {}
 
-function cmd.LAUNCH(service_name, ...)
+local function request(name, func, ...)
-	local s = service[service_name]
+	local ok, handle = pcall(func, ...)
-	if type(s) == "number" then
+	local s = service[name]
-		return s
+	assert(type(s) == "table")
+	if ok then
+		service[name] = handle
+	else
+		service[name] = tostring(handle)
 	end
 
-	if s == nil then
-		s = { launch = true }
-		service[service_name] = s
-	elseif s.launch then
-		assert(type(s) == "table")
-		local co = coroutine.running()
-		table.insert(s, co)
-		skynet.wait()
-		s = service[service_name]
-		assert(type(s) == "number")
-		return s
-	end
-
-	local handle = skynet.newservice(service_name, ...)
 	for _,v in ipairs(s) do
 		skynet.wakeup(v)
 	end
 
-	service[service_name] = handle
+	if ok then
-
+		return handle
-	return handle
+	else
+		error(tostring(handle))
+	end
 end
 
-function cmd.QUERY(service_name)
+local function waitfor(name , func, ...)
-	local s = service[service_name]
+	local s = service[name]
 	if type(s) == "number" then
 		return s
 	end
+	local co = coroutine.running()
+
 	if s == nil then
 		s = {}
-		service[service_name] = s
+		service[name] = s
+	elseif type(s) == "string" then
+		error(s)
 	end
+
 	assert(type(s) == "table")
-	local co = coroutine.running()
+
+	if not s.launch and func then
+		s.launch = true
+		return request(name, func, ...)
+	end
+
 	table.insert(s, co)
 	skynet.wait()
-	s = service[service_name]
+	s = service[name]
+	if type(s) == "string" then
+		error(s)
+	end
 	assert(type(s) == "number")
 	return s
 end
 
+local function read_name(service_name)
+	if string.byte(service_name) == 64 then -- '@'
+		return string.sub(service_name , 2)
+	else
+		return service_name
+	end
+end
+
+function cmd.LAUNCH(service_name, subname, ...)
+	local realname = read_name(service_name)
+
+	if realname == "snaxd" then
+		return waitfor(service_name.."."..subname, snax.rawnewservice, subname, ...)
+	else
+		return waitfor(service_name, skynet.newservice, realname, subname, ...)
+	end
+end
+
+function cmd.QUERY(service_name, subname)
+	local realname = read_name(service_name)
+
+	if realname == "snaxd" then
+		return waitfor(service_name.."."..subname)
+	else
+		return waitfor(service_name)
+	end
+end
+
+local function list_service()
+	local result = {}
+	for k,v in pairs(service) do
+		if type(v) == "string" then
+			v = "Error: " .. v
+		elseif type(v) == "table" then
+			v = "Querying"
+		else
+			v = skynet.address(v)
+		end
+
+		result[k] = v
+	end
+
+	return result
+end
+
+
+local function register_global()
+	function cmd.GLAUNCH(name, ...)
+		local global_name = "@" .. name
+		return cmd.LAUNCH(global_name, ...)
+	end
+
+	function cmd.GQUERY(name, ...)
+		local global_name = "@" .. name
+		return cmd.QUERY(global_name, ...)
+	end
+
+	local mgr = {}
+
+	function cmd.REPORT(m)
+		mgr[m] = true
+		skynet.watch(m)
+	end
+
+	local function add_list(all, m)
+		local harbor = "@" .. skynet.harbor(m)
+		local result = skynet.call(m, "lua", "LIST")
+		for k,v in pairs(result) do
+			all[k .. harbor] = v
+		end
+	end
+
+	function cmd.LIST()
+		local result = {}
+		for k in pairs(mgr) do
+			pcall(add_list, result, k)
+		end
+		local l = list_service()
+		for k, v in pairs(l) do
+			result[k] = v
+		end
+		return result
+	end
+end
+
+local function register_local()
+	function cmd.GLAUNCH(name, ...)
+		local global_name = "@" .. name
+		return waitfor(global_name, skynet.call, "SERVICE", "lua", "LAUNCH", global_name, ...)
+	end
+
+	function cmd.GQUERY(name, ...)
+		local global_name = "@" .. name
+		return waitfor(global_name, skynet.call, "SERVICE", "lua", "QUERY", global_name, ...)
+	end
+
+	function cmd.LIST()
+		return list_service()
+	end
+
+	skynet.call("SERVICE", "lua", "REPORT", skynet.self())
+end
+
 skynet.start(function()
-	skynet.dispatch("lua", function(session, address, command, service_name , ...)
+	skynet.dispatch("lua", function(session, address, command, ...)
 		local f = cmd[command]
 		if f == nil then
-			skynet.ret(skynet.pack(nil))
+			skynet.ret(skynet.pack(nil, "Invalid command " .. command))
 			return
 		end
 
-		local ok, r = pcall(f, service_name, ...)
+		local ok, r = pcall(f, ...)
+
 		if ok then
 			skynet.ret(skynet.pack(r))
 		else
-			skynet.ret(skynet.pack(nil))
+			skynet.ret(skynet.pack(nil, r))
 		end
 	end)
-	skynet.register(".service")
+	local handle = skynet.localname ".service"
+	if  handle ~= 0 then
+		skynet.error(".service is already register by ", skynet.address(handle))
+		skynet.exit()
+	else
+		skynet.register(".service")
+	end
 	if skynet.getenv "standalone" then
 		skynet.register("SERVICE")
+		register_global()
+	else
+		register_local()
 	end
 end)

service/snaxd.lua

@@ -1,7 +1,8 @@
 local skynet = require "skynet"
 local c = require "skynet.c"
-local snax_interface = require "snax_interface"
+local snax_interface = require "snax.interface"
 local profile = require "profile"
+local snax = require "snax"
 
 local func = snax_interface(tostring(...), _ENV)
 local mode
@@ -20,8 +21,10 @@ local function update_stat(name, ti)
 	t.time = t.time + ti
 end
 
+local traceback = debug.traceback
+
 local function do_func(f, msg)
-	return pcall(f, table.unpack(msg))
+	return xpcall(f, traceback, table.unpack(msg))
 end
 
 local function dispatch(f, ...)
@@ -41,7 +44,7 @@ local function message_dispatch()
 				if method[2] == "accept" then
 					-- no return
 					profile.start()
-					local ok, data = pcall(f, table.unpack(msg))
+					local ok, data = xpcall(f, traceback, table.unpack(msg))
 					local ti = profile.stop()
 					update_stat(method[3], ti)
 					if not ok then
@@ -49,7 +52,7 @@ local function message_dispatch()
 					end
 				else
 					profile.start()
-					local ok, data, size = pcall(dispatch, f, table.unpack(msg))
+					local ok, data, size = xpcall(dispatch, traceback, f, table.unpack(msg))
 					local ti = profile.stop()
 					update_stat(method[3], ti)
 					if ok then
@@ -88,9 +91,9 @@ local function timing( method, ... )
 	profile.start()
 	if method[2] == "accept" then
 		-- no return
-		err,msg = pcall(method[4], ...)
+		err,msg = xpcall(method[4], traceback, ...)
 	else
-		err,msg = pcall(return_f, method[4], ...)
+		err,msg = xpcall(return_f, traceback, method[4], ...)
 	end
 	local ti = profile.stop()
 	update_stat(method[3], ti)
@@ -98,13 +101,13 @@ local function timing( method, ... )
 end
 
 skynet.start(function()
-	skynet.dispatch("lua", function ( session , source , id, ...)
+	skynet.dispatch("snax", function ( session , source , id, ...)
 		local method = func[id]
 
 		if method[2] == "system" then
 			local command = method[3]
 			if command == "hotfix" then
-				local hotfix = require "snax_hotfix"
+				local hotfix = require "snax.hotfix"
 				skynet.ret(skynet.pack(hotfix(func, ...)))
 			elseif command == "init" then
 				assert(not init, "Already init")

skynet-src/malloc_hook.c

@@ -1,18 +1,16 @@
 #include <stdio.h>
 #include <string.h>
 #include <assert.h>
+#include <stdlib.h>
 
 #include "malloc_hook.h"
-#include "jemalloc.h"
-
 #include "skynet.h"
 
-
 static size_t _used_memory = 0;
 static size_t _memory_block = 0;
 typedef struct _mem_data {
-    uint32_t handle;
+	uint32_t handle;
-    ssize_t allocated;
+	ssize_t allocated;
 } mem_data;
 
 #define SLOT_SIZE 0x10000
@@ -20,13 +18,18 @@ typedef struct _mem_data {
 
 static mem_data mem_stats[SLOT_SIZE];
 
+
+#ifndef NOUSE_JEMALLOC
+
+#include "jemalloc.h"
+
 static ssize_t*
 get_allocated_field(uint32_t handle) {
-    int h = (int)(handle & (SLOT_SIZE - 1));
+	int h = (int)(handle & (SLOT_SIZE - 1));
-    mem_data *data = &mem_stats[h];
+	mem_data *data = &mem_stats[h];
 	uint32_t old_handle = data->handle;
 	ssize_t old_alloc = data->allocated;
-    if(old_handle == 0 || old_alloc <= 0) {
+	if(old_handle == 0 || old_alloc <= 0) {
 		// data->allocated may less than zero, because it may not count at start.
 		if(!__sync_bool_compare_and_swap(&data->handle, old_handle, handle)) {
 			return 0;
@@ -34,150 +37,174 @@ get_allocated_field(uint32_t handle) {
 		if (old_alloc < 0) {
 			__sync_bool_compare_and_swap(&data->allocated, old_alloc, 0);
 		}
-    }
+	}
-    if(data->handle != handle) {
+	if(data->handle != handle) {
-        return 0;
+		return 0;
-    }
+	}
-    return &data->allocated;
+	return &data->allocated;
 }
 
 inline static void 
 update_xmalloc_stat_alloc(uint32_t handle, size_t __n) {
-    __sync_add_and_fetch(&_used_memory, __n);
+	__sync_add_and_fetch(&_used_memory, __n);
-    __sync_add_and_fetch(&_memory_block, 1); 
+	__sync_add_and_fetch(&_memory_block, 1); 
-    ssize_t* allocated = get_allocated_field(handle);
+	ssize_t* allocated = get_allocated_field(handle);
-    if(allocated) {
+	if(allocated) {
-        __sync_add_and_fetch(allocated, __n);
+		__sync_add_and_fetch(allocated, __n);
-    }
+	}
 }
 
 inline static void
 update_xmalloc_stat_free(uint32_t handle, size_t __n) {
-    __sync_sub_and_fetch(&_used_memory, __n);
+	__sync_sub_and_fetch(&_used_memory, __n);
-    __sync_sub_and_fetch(&_memory_block, 1);
+	__sync_sub_and_fetch(&_memory_block, 1);
-    ssize_t* allocated = get_allocated_field(handle);
+	ssize_t* allocated = get_allocated_field(handle);
-    if(allocated) {
+	if(allocated) {
-        __sync_sub_and_fetch(allocated, __n);
+		__sync_sub_and_fetch(allocated, __n);
-    }
+	}
 }
 
 inline static void*
 fill_prefix(char* ptr) {
-    uint32_t handle = skynet_current_handle();
+	uint32_t handle = skynet_current_handle();
-    size_t size = je_malloc_usable_size(ptr);
+	size_t size = je_malloc_usable_size(ptr);
-    uint32_t *p = (uint32_t *)(ptr + size - sizeof(uint32_t));
+	uint32_t *p = (uint32_t *)(ptr + size - sizeof(uint32_t));
-    memcpy(p, &handle, sizeof(handle));
+	memcpy(p, &handle, sizeof(handle));
 
 	update_xmalloc_stat_alloc(handle, size);
-    return ptr;
+	return ptr;
 }
 
 inline static void*
 clean_prefix(char* ptr) {
-    size_t size = je_malloc_usable_size(ptr);
+	size_t size = je_malloc_usable_size(ptr);
-    uint32_t *p = (uint32_t *)(ptr + size - sizeof(uint32_t));
+	uint32_t *p = (uint32_t *)(ptr + size - sizeof(uint32_t));
-    uint32_t handle;
+	uint32_t handle;
-    memcpy(&handle, p, sizeof(handle));
+	memcpy(&handle, p, sizeof(handle));
-    update_xmalloc_stat_free(handle, size);
+	update_xmalloc_stat_free(handle, size);
-    return ptr;
+	return ptr;
 }
 
 static void malloc_oom(size_t size) {
-    fprintf(stderr, "xmalloc: Out of memory trying to allocate %zu bytes\n",
+	fprintf(stderr, "xmalloc: Out of memory trying to allocate %zu bytes\n",
-        size);
+		size);
-    fflush(stderr);
+	fflush(stderr);
-    abort();
+	abort();
 }
 
-size_t
+void 
-malloc_used_memory(void) {
+memory_info_dump(void) {
-    return _used_memory;
+	je_malloc_stats_print(0,0,0);
 }
 
-size_t
+size_t 
-malloc_memory_block(void) {
+mallctl_int64(const char* name, size_t* newval) {
-    return _memory_block;
+	size_t v = 0;
+	size_t len = sizeof(v);
+	if(newval) {
+		je_mallctl(name, &v, &len, newval, sizeof(size_t));
+	} else {
+		je_mallctl(name, &v, &len, NULL, 0);
+	}
+	// skynet_error(NULL, "name: %s, value: %zd\n", name, v);
+	return v;
 }
 
-void memory_info_dump(void) {
+int 
-    je_malloc_stats_print(0,0,0);
+mallctl_opt(const char* name, int* newval) {
+	int v = 0;
+	size_t len = sizeof(v);
+	if(newval) {
+		int ret = je_mallctl(name, &v, &len, newval, sizeof(int));
+		if(ret == 0) {
+			skynet_error(NULL, "set new value(%d) for (%s) succeed\n", *newval, name);
+		} else {
+			skynet_error(NULL, "set new value(%d) for (%s) failed: error -> %d\n", *newval, name, ret);
+		}
+	} else {
+		je_mallctl(name, &v, &len, NULL, 0);
+	}
+
+	return v;
 }
 
-size_t mallctl_int64(const char* name, size_t* newval) {
+// hook : malloc, realloc, free, calloc
-    size_t v = 0;
-    size_t len = sizeof(v);
-    if(newval) {
-        je_mallctl(name, &v, &len, newval, sizeof(size_t));
-    } else {
-        je_mallctl(name, &v, &len, NULL, 0);
-    }
-    // printf("name: %s, value: %zd\n", name, v);
-    return v;
-}
-
-int mallctl_opt(const char* name, int* newval) {
-    int    v   = 0;
-    size_t len = sizeof(v);
-    if(newval) {
-        int ret = je_mallctl(name, &v, &len, newval, sizeof(int));
-        if(ret == 0) {
-            printf("set new value(%d) for (%s) succeed\n", *newval, name);
-        } else {
-            printf("set new value(%d) for (%s) failed: error -> %d\n", *newval, name, ret);
-        }
-    } else {
-        je_mallctl(name, &v, &len, NULL, 0);
-    }
-
-    return v;
-}
-
-void
-dump_c_mem() {
-    int i;
-    size_t total = 0;
-    printf("dump all service mem:\n");
-    for(i=0; i<SLOT_SIZE; i++) {
-        mem_data* data = &mem_stats[i];
-        if(data->handle != 0 && data->allocated != 0) {
-            total += data->allocated;
-            printf("0x%x -> %zdkb\n", data->handle, data->allocated >> 10);
-        }
-    }
-    printf("+total: %zdkb\n",total >> 10);
-}
-
-// hook : malloc, realloc, memalign, free, calloc
 
 void *
 skynet_malloc(size_t size) {
-    void* ptr = je_malloc(size + PREFIX_SIZE);
+	void* ptr = je_malloc(size + PREFIX_SIZE);
-    if(!ptr) malloc_oom(size);
+	if(!ptr) malloc_oom(size);
-    return fill_prefix(ptr);
+	return fill_prefix(ptr);
 }
 
 void *
 skynet_realloc(void *ptr, size_t size) {
-    if (ptr == NULL) return skynet_malloc(size);
+	if (ptr == NULL) return skynet_malloc(size);
 
-    void* rawptr = clean_prefix(ptr);
+	void* rawptr = clean_prefix(ptr);
-    void *newptr = je_realloc(rawptr, size+PREFIX_SIZE);
+	void *newptr = je_realloc(rawptr, size+PREFIX_SIZE);
-    if(!newptr) malloc_oom(size);
+	if(!newptr) malloc_oom(size);
-    return fill_prefix(newptr);
+	return fill_prefix(newptr);
 }
 
 void
 skynet_free(void *ptr) {
-    if (ptr == NULL) return;
+	if (ptr == NULL) return;
-    void* rawptr = clean_prefix(ptr);
+	void* rawptr = clean_prefix(ptr);
-    je_free(rawptr);
+	je_free(rawptr);
 }
 
 void *
 skynet_calloc(size_t nmemb,size_t size) {
-    void* ptr = je_calloc(nmemb + ((PREFIX_SIZE+size-1)/size), size );
+	void* ptr = je_calloc(nmemb + ((PREFIX_SIZE+size-1)/size), size );
-    if(!ptr) malloc_oom(size);
+	if(!ptr) malloc_oom(size);
-    return fill_prefix(ptr);
+	return fill_prefix(ptr);
+}
+
+#else
+
+void 
+memory_info_dump(void) {
+	skynet_error(NULL, "No jemalloc");
+}
+
+size_t 
+mallctl_int64(const char* name, size_t* newval) {
+	skynet_error(NULL, "No jemalloc : mallctl_int64 %s.", name);
+	return 0;
+}
+
+int 
+mallctl_opt(const char* name, int* newval) {
+	skynet_error(NULL, "No jemalloc : mallctl_opt %s.", name);
+	return 0;
+}
+
+#endif
+
+size_t
+malloc_used_memory(void) {
+	return _used_memory;
+}
+
+size_t
+malloc_memory_block(void) {
+	return _memory_block;
+}
+
+void
+dump_c_mem() {
+	int i;
+	size_t total = 0;
+	skynet_error(NULL, "dump all service mem:");
+	for(i=0; i<SLOT_SIZE; i++) {
+		mem_data* data = &mem_stats[i];
+		if(data->handle != 0 && data->allocated != 0) {
+			total += data->allocated;
+			skynet_error(NULL, "0x%x -> %zdkb", data->handle, data->allocated >> 10);
+		}
+	}
+	skynet_error(NULL, "+total: %zdkb",total >> 10);
 }
 
 char *
@@ -197,4 +224,3 @@ skynet_lalloc(void *ud, void *ptr, size_t osize, size_t nsize) {
 		return skynet_realloc(ptr, nsize);
 	}
 }
-

skynet-src/skynet.h

@@ -8,17 +8,18 @@
 
 #define PTYPE_TEXT 0
 #define PTYPE_RESPONSE 1
-#define PTYPE_MULTICAST_DEPRECATED 2
+#define PTYPE_MULTICAST 2
 #define PTYPE_CLIENT 3
 #define PTYPE_SYSTEM 4
 #define PTYPE_HARBOR 5
 #define PTYPE_SOCKET 6
-// read lualib/skynet.lua lualib/simplemonitor.lua
+// read lualib/skynet.lua examples/simplemonitor.lua
-#define PTYPE_RESERVED_ERROR 7	
+#define PTYPE_ERROR 7	
-// read lualib/skynet.lua lualib/mqueue.lua
+// read lualib/skynet.lua lualib/mqueue.lua lualib/snax.lua
 #define PTYPE_RESERVED_QUEUE 8
 #define PTYPE_RESERVED_DEBUG 9
 #define PTYPE_RESERVED_LUA 10
+#define PTYPE_RESERVED_SNAX 11
 
 #define PTYPE_TAG_DONTCOPY 0x10000
 #define PTYPE_TAG_ALLOCSESSION 0x20000

skynet-src/skynet_handle.c

@@ -9,6 +9,7 @@
 #include <string.h>
 
 #define DEFAULT_SLOT_SIZE 4
+#define MAX_SLOT_SIZE 0x40000000
 
 struct handle_name {
 	char * name;
@@ -168,6 +169,7 @@ static void
 _insert_name_before(struct handle_storage *s, char *name, uint32_t handle, int before) {
 	if (s->name_count >= s->name_cap) {
 		s->name_cap *= 2;
+		assert(s->name_cap <= MAX_SLOT_SIZE);
 		struct handle_name * n = skynet_malloc(s->name_cap * sizeof(struct handle_name));
 		int i;
 		for (i=0;i<before;i++) {

skynet-src/skynet_harbor.c

@@ -34,6 +34,7 @@ skynet_harbor_register(struct remote_name *rname) {
 
 int 
 skynet_harbor_message_isremote(uint32_t handle) {
+	assert(HARBOR != 0);
 	int h = (handle & ~HANDLE_MASK);
 	return h != HARBOR && h !=0;
 }
@@ -43,16 +44,7 @@ skynet_harbor_init(int harbor) {
 	HARBOR = (unsigned int)harbor << HANDLE_REMOTE_SHIFT;
 }
 
-int
+void
-skynet_harbor_start(const char * master, const char *local) {
+skynet_harbor_start(void *ctx) {
-	size_t sz = strlen(master) + strlen(local) + 32;
+	REMOTE = ctx;
-	char args[sz];
-	sprintf(args, "%s %s %d",master,local,HARBOR >> HANDLE_REMOTE_SHIFT);
-	struct skynet_context * inst = skynet_context_new("harbor",args);
-	if (inst == NULL) {
-		return 1;
-	}
-	REMOTE = inst;
-
-	return 0;
 }

skynet-src/skynet_harbor.h

@@ -26,6 +26,6 @@ void skynet_harbor_send(struct remote_message *rmsg, uint32_t source, int sessio
 void skynet_harbor_register(struct remote_name *rname);
 int skynet_harbor_message_isremote(uint32_t handle);
 void skynet_harbor_init(int harbor);
-int skynet_harbor_start(const char * master, const char *local);
+void skynet_harbor_start(void * ctx);
 
 #endif

skynet-src/skynet_imp.h

@@ -4,14 +4,16 @@
 struct skynet_config {
 	int thread;
 	int harbor;
-	const char * logger;
 	const char * module_path;
-	const char * master;
+	const char * bootstrap;
-	const char * local;
-	const char * start;
-	const char * standalone;
 };
 
+#define THREAD_WORKER 0
+#define THREAD_MAIN 1
+#define THREAD_SOCKET 2
+#define THREAD_TIMER 3
+#define THREAD_MONITOR 4
+
 void skynet_start(struct skynet_config * config);
 
 #endif

skynet-src/skynet_main.c

@@ -2,6 +2,7 @@
 
 #include "skynet_imp.h"
 #include "skynet_env.h"
+#include "skynet_server.h"
 
 #include <stdio.h>
 #include <stdlib.h>
@@ -87,6 +88,7 @@ main(int argc, char *argv[]) {
 	if (argc > 1) {
 		config_file = argv[1];
 	}
+	skynet_globalinit();
 	skynet_env_init();
 
 	sigign();
@@ -111,24 +113,15 @@ main(int argc, char *argv[]) {
   printf("Skynet lua code cache enable\n");
 #endif
 
-	const char *path = optstring("lua_path","./lualib/?.lua;./lualib/?/init.lua");
-	setenv("LUA_PATH",path,1);
-	const char *cpath = optstring("lua_cpath","./luaclib/?.so");
-	setenv("LUA_CPATH",cpath,1);
-	optstring("luaservice","./service/?.lua");
-
 	config.thread =  optint("thread",8);
 	config.module_path = optstring("cpath","./cservice/?.so");
-	config.logger = optstring("logger",NULL);
 	config.harbor = optint("harbor", 1);
-	config.master = optstring("master","127.0.0.1:2012");
+	config.bootstrap = optstring("bootstrap","snlua bootstrap");
-	config.start = optstring("start","main.lua");
-	config.local = optstring("address","127.0.0.1:2525");
-	config.standalone = optstring("standalone",NULL);
 
 	lua_close(L);
 
 	skynet_start(&config);
+	skynet_globalexit();
 
 	printf("skynet exit\n");
 

skynet-src/skynet_malloc.h

@@ -3,14 +3,10 @@
 
 #include <stddef.h>
 
-#ifdef SKYNET_MALLOC_RENAME
+#define skynet_malloc malloc
-
+#define skynet_calloc calloc
-#define malloc skynet_malloc
+#define skynet_realloc realloc
-#define calloc skynet_calloc
+#define skynet_free free
-#define realloc skynet_realloc
-#define free skynet_free
-
-#endif
 
 void * skynet_malloc(size_t sz);
 void * skynet_calloc(size_t nmemb,size_t size);

skynet-src/skynet_mq.c

@@ -13,12 +13,8 @@
 
 // 0 means mq is not in global mq.
 // 1 means mq is in global mq , or the message is dispatching.
-// 2 means message is dispatching with locked session set.
-// 3 means mq is not in global mq, and locked session has been set.
 
 #define MQ_IN_GLOBAL 1
-#define MQ_DISPATCHING 2
-#define MQ_LOCKED 3
 
 struct message_queue {
 	uint32_t handle;
@@ -27,17 +23,19 @@ struct message_queue {
 	int tail;
 	int lock;
 	int release;
-	int lock_session;
 	int in_global;
 	struct skynet_message *queue;
+	struct message_queue *next;
 };
 
 struct global_queue {
 	uint32_t head;
 	uint32_t tail;
 	struct message_queue ** queue;
+	// We use a separated flag array to ensure the mq is pushed.
+	// See the comments below.
 	bool * flag;
-
+	struct message_queue *list;
 };
 
 static struct global_queue *Q = NULL;
@@ -51,7 +49,22 @@ static void
 skynet_globalmq_push(struct message_queue * queue) {
 	struct global_queue *q= Q;
 
+	if (q->flag[GP(q->tail)]) {
+		// The queue may full seldom, save queue in list
+		assert(queue->next == NULL);
+		struct message_queue * last;
+		do {
+			last = q->list;
+			queue->next = last;
+		} while(!__sync_bool_compare_and_swap(&q->list, last, queue));
+
+		return;
+	}
+
 	uint32_t tail = GP(__sync_fetch_and_add(&q->tail,1));
+	// The thread would suspend here, and the q->queue[tail] is last version ,
+	// but the queue tail is increased.
+	// So we set q->flag[tail] after changing q->queue[tail].
 	q->queue[tail] = queue;
 	__sync_synchronize();
 	q->flag[tail] = true;
@@ -61,11 +74,26 @@ struct message_queue *
 skynet_globalmq_pop() {
 	struct global_queue *q = Q;
 	uint32_t head =  q->head;
-	uint32_t head_ptr = GP(head);
+
-	if (head_ptr == GP(q->tail)) {
+	if (head == q->tail) {
+		// The queue is empty.
 		return NULL;
 	}
 
+	uint32_t head_ptr = GP(head);
+
+	struct message_queue * list = q->list;
+	if (list) {
+		// If q->list is not empty, try to load it back to the queue
+		struct message_queue *newhead = list->next;
+		if (__sync_bool_compare_and_swap(&q->list, list, newhead)) {
+			// try load list only once, if success , push it back to the queue.
+			list->next = NULL;
+			skynet_globalmq_push(list);
+		}
+	}
+
+	// Check the flag first, if the flag is false, the pushing may not complete.
 	if(!q->flag[head_ptr]) {
 		return NULL;
 	}
@@ -89,16 +117,20 @@ skynet_mq_create(uint32_t handle) {
 	q->head = 0;
 	q->tail = 0;
 	q->lock = 0;
+	// When the queue is create (always between service create and service init) ,
+	// set in_global flag to avoid push it to global queue .
+	// If the service init success, skynet_context_new will call skynet_mq_force_push to push it to global queue.
 	q->in_global = MQ_IN_GLOBAL;
 	q->release = 0;
-	q->lock_session = 0;
 	q->queue = skynet_malloc(sizeof(struct skynet_message) * q->cap);
+	q->next = NULL;
 
 	return q;
 }
 
 static void 
 _release(struct message_queue *q) {
+	assert(q->next == NULL);
 	skynet_free(q->queue);
 	skynet_free(q);
 }
@@ -161,82 +193,28 @@ expand_queue(struct message_queue *q) {
 	q->queue = new_queue;
 }
 
-static void
-_unlock(struct message_queue *q) {
-	// this api use in push a unlock message, so the in_global flags must not be 0 , 
-	// but the q is not exist in global queue.
-	if (q->in_global == MQ_LOCKED) {
-		skynet_globalmq_push(q);
-		q->in_global = MQ_IN_GLOBAL;
-	} else {
-		assert(q->in_global == MQ_DISPATCHING);
-	}
-	q->lock_session = 0;
-}
-
-static void 
-_pushhead(struct message_queue *q, struct skynet_message *message) {
-	int head = q->head - 1;
-	if (head < 0) {
-		head = q->cap - 1;
-	}
-	if (head == q->tail) {
-		expand_queue(q);
-		--q->tail;
-		head = q->cap - 1;
-	}
-
-	q->queue[head] = *message;
-	q->head = head;
-
-	_unlock(q);
-}
-
 void 
 skynet_mq_push(struct message_queue *q, struct skynet_message *message) {
 	assert(message);
 	LOCK(q)
-	
-	if (q->lock_session !=0 && message->session == q->lock_session) {
-		_pushhead(q,message);
-	} else {
-		q->queue[q->tail] = *message;
-		if (++ q->tail >= q->cap) {
-			q->tail = 0;
-		}
 
-		if (q->head == q->tail) {
+	q->queue[q->tail] = *message;
-			expand_queue(q);
+	if (++ q->tail >= q->cap) {
-		}
+		q->tail = 0;
+	}
 
-		if (q->lock_session == 0) {
+	if (q->head == q->tail) {
-			if (q->in_global == 0) {
+		expand_queue(q);
-				q->in_global = MQ_IN_GLOBAL;
+	}
-				skynet_globalmq_push(q);
+
-			}
+	if (q->in_global == 0) {
-		}
+		q->in_global = MQ_IN_GLOBAL;
+		skynet_globalmq_push(q);
 	}
 	
 	UNLOCK(q)
 }
 
-void
-skynet_mq_lock(struct message_queue *q, int session) {
-	LOCK(q)
-	assert(q->lock_session == 0);
-	assert(q->in_global == MQ_IN_GLOBAL);
-	q->in_global = MQ_DISPATCHING;
-	q->lock_session = session;
-	UNLOCK(q)
-}
-
-void
-skynet_mq_unlock(struct message_queue *q) {
-	LOCK(q)
-	_unlock(q);
-	UNLOCK(q)
-}
-
 void 
 skynet_mq_init() {
 	struct global_queue *q = skynet_malloc(sizeof(*q));
@@ -257,14 +235,8 @@ void
 skynet_mq_pushglobal(struct message_queue *queue) {
 	LOCK(queue)
 	assert(queue->in_global);
-	if (queue->in_global == MQ_DISPATCHING) {
+	skynet_globalmq_push(queue);
-		// lock message queue just now.
+	queue->in_global = MQ_IN_GLOBAL;
-		queue->in_global = MQ_LOCKED;
-	}
-	if (queue->lock_session == 0) {
-		skynet_globalmq_push(queue);
-		queue->in_global = MQ_IN_GLOBAL;
-	}
 	UNLOCK(queue)
 }
 
@@ -279,31 +251,24 @@ skynet_mq_mark_release(struct message_queue *q) {
 	UNLOCK(q)
 }
 
-static int
+static void
-_drop_queue(struct message_queue *q) {
+_drop_queue(struct message_queue *q, message_drop drop_func, void *ud) {
-	// todo: send message back to message source
 	struct skynet_message msg;
-	int s = 0;
 	while(!skynet_mq_pop(q, &msg)) {
-		++s;
+		drop_func(&msg, ud);
-		skynet_free(msg.data);
 	}
 	_release(q);
-	return s;
 }
 
-int 
+void 
-skynet_mq_release(struct message_queue *q) {
+skynet_mq_release(struct message_queue *q, message_drop drop_func, void *ud) {
-	int ret = 0;
 	LOCK(q)
 	
 	if (q->release) {
 		UNLOCK(q)
-		ret = _drop_queue(q);
+		_drop_queue(q, drop_func, ud);
 	} else {
 		skynet_mq_force_push(q);
 		UNLOCK(q)
 	}
-	
-	return ret;
 }

skynet-src/skynet_mq.h

@@ -17,14 +17,15 @@ struct message_queue * skynet_globalmq_pop(void);
 
 struct message_queue * skynet_mq_create(uint32_t handle);
 void skynet_mq_mark_release(struct message_queue *q);
-int skynet_mq_release(struct message_queue *q);
+
+typedef void (*message_drop)(struct skynet_message *, void *);
+
+void skynet_mq_release(struct message_queue *q, message_drop drop_func, void *ud);
 uint32_t skynet_mq_handle(struct message_queue *);
 
 // 0 for success
 int skynet_mq_pop(struct message_queue *q, struct skynet_message *message);
 void skynet_mq_push(struct message_queue *q, struct skynet_message *message);
-void skynet_mq_lock(struct message_queue *q, int session);
-void skynet_mq_unlock(struct message_queue *q);
 
 // return the length of message queue, for debug
 int skynet_mq_length(struct message_queue *q);

skynet-src/skynet_server.c

@@ -8,6 +8,9 @@
 #include "skynet_harbor.h"
 #include "skynet_env.h"
 #include "skynet_monitor.h"
+#include "skynet_imp.h"
+
+#include <pthread.h>
 
 #include <string.h>
 #include <assert.h>
@@ -50,10 +53,10 @@ struct skynet_context {
 struct skynet_node {
 	int total;
 	uint32_t monitor_exit;
+	pthread_key_t handle_key;
 };
 
-static struct skynet_node G_NODE = { 0,0 };
+static struct skynet_node G_NODE;
-static __thread uint32_t handle_tls = 0xffffffff;
 
 int 
 skynet_context_total() {
@@ -61,22 +64,23 @@ skynet_context_total() {
 }
 
 static void
-_context_inc() {
+context_inc() {
 	__sync_fetch_and_add(&G_NODE.total,1);
 }
 
 static void
-_context_dec() {
+context_dec() {
 	__sync_fetch_and_sub(&G_NODE.total,1);
 }
 
 uint32_t 
 skynet_current_handle(void) {
-	return handle_tls;
+	void * handle = pthread_getspecific(G_NODE.handle_key);
+	return (uint32_t)(uintptr_t)handle;
 }
 
 static void
-_id_to_hex(char * str, uint32_t id) {
+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] = ':';
@@ -86,6 +90,20 @@ _id_to_hex(char * str, uint32_t id) {
 	str[9] = '\0';
 }
 
+struct drop_t {
+	uint32_t handle;
+};
+
+static void
+drop_message(struct skynet_message *msg, void *ud) {
+	struct drop_t *d = ud;
+	skynet_free(msg->data);
+	uint32_t source = d->handle;
+	assert(source);
+	// report error to the message source
+	skynet_send(NULL, source, msg->source, PTYPE_ERROR, 0, NULL, 0);
+}
+
 struct skynet_context * 
 skynet_context_new(const char * name, const char *param) {
 	struct skynet_module * mod = skynet_module_query(name);
@@ -111,7 +129,7 @@ skynet_context_new(const char * name, const char *param) {
 	ctx->handle = skynet_handle_register(ctx);
 	struct message_queue * queue = ctx->queue = skynet_mq_create(ctx->handle);
 	// init function maybe use ctx->handle, so it must init at last
-	_context_inc();
+	context_inc();
 
 	CHECKCALLING_BEGIN(ctx)
 	int r = skynet_module_instance_init(mod, inst, ctx, param);
@@ -128,9 +146,11 @@ skynet_context_new(const char * name, const char *param) {
 		return ret;
 	} else {
 		skynet_error(ctx, "FAILED launch %s", name);
+		uint32_t handle = ctx->handle;
 		skynet_context_release(ctx);
-		skynet_handle_retire(ctx->handle);
+		skynet_handle_retire(handle);
-		skynet_mq_release(queue);
+		struct drop_t d = { handle };
+		skynet_mq_release(queue, drop_message, &d);
 		return NULL;
 	}
 }
@@ -148,17 +168,17 @@ skynet_context_grab(struct skynet_context *ctx) {
 }
 
 static void 
-_delete_context(struct skynet_context *ctx) {
+delete_context(struct skynet_context *ctx) {
 	skynet_module_instance_release(ctx->mod, ctx->instance);
 	skynet_mq_mark_release(ctx->queue);
 	skynet_free(ctx);
-	_context_dec();
+	context_dec();
 }
 
 struct skynet_context * 
 skynet_context_release(struct skynet_context *ctx) {
 	if (__sync_sub_and_fetch(&ctx->ref,1) == 0) {
-		_delete_context(ctx);
+		delete_context(ctx);
 		return NULL;
 	}
 	return ctx;
@@ -199,13 +219,12 @@ static void
 _dispatch_message(struct skynet_context *ctx, struct skynet_message *msg) {
 	assert(ctx->init);
 	CHECKCALLING_BEGIN(ctx)
-	handle_tls = ctx->handle;
+	pthread_setspecific(G_NODE.handle_key, (void *)(uintptr_t)(ctx->handle));
 	int type = msg->sz >> HANDLE_REMOTE_SHIFT;
 	size_t sz = msg->sz & HANDLE_MASK;
 	if (!ctx->cb(ctx, ctx->cb_ud, type, msg->session, msg->source, msg->data, sz)) {
 		skynet_free(msg->data);
 	}
-	handle_tls = 0xffffffff;
 	CHECKCALLING_END(ctx)
 }
 
@@ -219,10 +238,8 @@ skynet_context_message_dispatch(struct skynet_monitor *sm) {
 
 	struct skynet_context * ctx = skynet_handle_grab(handle);
 	if (ctx == NULL) {
-		int s = skynet_mq_release(q);
+		struct drop_t d = { handle };
-		if (s>0) {
+		skynet_mq_release(q, drop_message, &d);
-			skynet_error(NULL, "Drop message queue %x (%d messages)", handle,s);
-		}
 		return 0;
 	}
 
@@ -236,7 +253,6 @@ skynet_context_message_dispatch(struct skynet_monitor *sm) {
 
 	if (ctx->cb == NULL) {
 		skynet_free(msg.data);
-		skynet_error(NULL, "Drop message from %x to %x without callback , size = %d",msg.source, handle, (int)msg.sz);
 	} else {
 		_dispatch_message(ctx, &msg);
 	}
@@ -251,7 +267,7 @@ skynet_context_message_dispatch(struct skynet_monitor *sm) {
 }
 
 static void
-_copy_name(char name[GLOBALNAME_LENGTH], const char * addr) {
+copy_name(char name[GLOBALNAME_LENGTH], const char * addr) {
 	int i;
 	for (i=0;i<GLOBALNAME_LENGTH && addr[i];i++) {
 		name[i] = addr[i];
@@ -287,192 +303,225 @@ handle_exit(struct skynet_context * context, uint32_t handle) {
 	skynet_handle_retire(handle);
 }
 
-const char * 
+// skynet command
-skynet_command(struct skynet_context * context, const char * cmd , const char * param) {
+
-	if (strcmp(cmd,"TIMEOUT") == 0) {
+struct command_func {
-		char * session_ptr = NULL;
+	const char *name;
-		int ti = strtol(param, &session_ptr, 10);
+	const char * (*func)(struct skynet_context * context, const char * param);
-		int session = skynet_context_newsession(context);
+};
-		skynet_timeout(context->handle, ti, session);
+
-		sprintf(context->result, "%d", session);
+static const char *
+cmd_timeout(struct skynet_context * context, const char * param) {
+	char * session_ptr = NULL;
+	int ti = strtol(param, &session_ptr, 10);
+	int session = skynet_context_newsession(context);
+	skynet_timeout(context->handle, ti, session);
+	sprintf(context->result, "%d", session);
+	return context->result;
+}
+
+static const char *
+cmd_reg(struct skynet_context * context, const char * param) {
+	if (param == NULL || param[0] == '\0') {
+		sprintf(context->result, ":%x", context->handle);
+		return context->result;
+	} else if (param[0] == '.') {
+		return skynet_handle_namehandle(context->handle, param + 1);
+	} else {
+		assert(context->handle!=0);
+		struct remote_name *rname = skynet_malloc(sizeof(*rname));
+		copy_name(rname->name, param);
+		rname->handle = context->handle;
+		skynet_harbor_register(rname);
+		return NULL;
+	}
+}
+
+static const char *
+cmd_query(struct skynet_context * context, const char * param) {
+	if (param[0] == '.') {
+		uint32_t handle = skynet_handle_findname(param+1);
+		sprintf(context->result, ":%x", handle);
 		return context->result;
 	}
+	return NULL;
+}
 
-	if (strcmp(cmd,"LOCK") == 0) {
+static const char *
-		if (context->init == false) {
+cmd_name(struct skynet_context * context, const char * param) {
-			return NULL;
+	int size = strlen(param);
-		}
+	char name[size+1];
-		skynet_mq_lock(context->queue, context->session_id+1);
+	char handle[size+1];
+	sscanf(param,"%s %s",name,handle);
+	if (handle[0] != ':') {
 		return NULL;
 	}
-
+	uint32_t handle_id = strtoul(handle+1, NULL, 16);
-	if (strcmp(cmd,"UNLOCK") == 0) {
+	if (handle_id == 0) {
-		if (context->init == false) {
-			return NULL;
-		}
-		skynet_mq_unlock(context->queue);
 		return NULL;
 	}
-
+	if (name[0] == '.') {
-	if (strcmp(cmd,"REG") == 0) {
+		return skynet_handle_namehandle(handle_id, name + 1);
-		if (param == NULL || param[0] == '\0') {
+	} else {
-			sprintf(context->result, ":%x", context->handle);
+		struct remote_name *rname = skynet_malloc(sizeof(*rname));
-			return context->result;
+		copy_name(rname->name, name);
-		} else if (param[0] == '.') {
+		rname->handle = handle_id;
-			return skynet_handle_namehandle(context->handle, param + 1);
+		skynet_harbor_register(rname);
-		} else {
-			assert(context->handle!=0);
-			struct remote_name *rname = skynet_malloc(sizeof(*rname));
-			_copy_name(rname->name, param);
-			rname->handle = context->handle;
-			skynet_harbor_register(rname);
-			return NULL;
-		}
 	}
+	return NULL;
+}
 
-	if (strcmp(cmd,"QUERY") == 0) {
+static const char *
-		if (param[0] == '.') {
+cmd_now(struct skynet_context * context, const char * param) {
-			uint32_t handle = skynet_handle_findname(param+1);
+	uint32_t ti = skynet_gettime();
-			sprintf(context->result, ":%x", handle);
+	sprintf(context->result,"%u",ti);
+	return context->result;
+}
+
+static const char *
+cmd_exit(struct skynet_context * context, const char * param) {
+	handle_exit(context, 0);
+	return NULL;
+}
+
+static const char *
+cmd_kill(struct skynet_context * context, const char * param) {
+	uint32_t handle = 0;
+	if (param[0] == ':') {
+		handle = strtoul(param+1, NULL, 16);
+	} else if (param[0] == '.') {
+		handle = skynet_handle_findname(param+1);
+	} else {
+		skynet_error(context, "Can't kill %s",param);
+		// todo : kill global service
+	}
+	if (handle) {
+		handle_exit(context, handle);
+	}
+	return NULL;
+}
+
+static const char *
+cmd_launch(struct skynet_context * context, const char * param) {
+	size_t sz = strlen(param);
+	char tmp[sz+1];
+	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) {
+		return NULL;
+	} else {
+		id_to_hex(context->result, inst->handle);
+		return context->result;
+	}
+}
+
+static const char *
+cmd_getenv(struct skynet_context * context, const char * param) {
+	return skynet_getenv(param);
+}
+
+static const char *
+cmd_setenv(struct skynet_context * context, const char * param) {
+	size_t sz = strlen(param);
+	char key[sz+1];
+	int i;
+	for (i=0;param[i] != ' ' && param[i];i++) {
+		key[i] = param[i];
+	}
+	if (param[i] == '\0')
+		return NULL;
+
+	key[i] = '\0';
+	param += i+1;
+	
+	skynet_setenv(key,param);
+	return NULL;
+}
+
+static const char *
+cmd_starttime(struct skynet_context * context, const char * param) {
+	uint32_t sec = skynet_gettime_fixsec();
+	sprintf(context->result,"%u",sec);
+	return context->result;
+}
+
+static const char *
+cmd_endless(struct skynet_context * context, const char * param) {
+	if (context->endless) {
+		strcpy(context->result, "1");
+		context->endless = false;
+		return context->result;
+	}
+	return NULL;
+}
+
+static const char *
+cmd_abort(struct skynet_context * context, const char * param) {
+	skynet_handle_retireall();
+	return NULL;
+}
+
+static const char *
+cmd_monitor(struct skynet_context * context, const char * param) {
+	uint32_t handle=0;
+	if (param == NULL || param[0] == '\0') {
+		if (G_NODE.monitor_exit) {
+			// return current monitor serivce
+			sprintf(context->result, ":%x", G_NODE.monitor_exit);
 			return context->result;
 		}
 		return NULL;
-	}
+	} else {
-
-	if (strcmp(cmd,"NAME") == 0) {
-		int size = strlen(param);
-		char name[size+1];
-		char handle[size+1];
-		sscanf(param,"%s %s",name,handle);
-		if (handle[0] != ':') {
-			return NULL;
-		}
-		uint32_t handle_id = strtoul(handle+1, NULL, 16);
-		if (handle_id == 0) {
-			return NULL;
-		}
-		if (name[0] == '.') {
-			return skynet_handle_namehandle(handle_id, name + 1);
-		} else {
-			struct remote_name *rname = skynet_malloc(sizeof(*rname));
-			_copy_name(rname->name, name);
-			rname->handle = handle_id;
-			skynet_harbor_register(rname);
-		}
-		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) {
-		handle_exit(context, 0);
-		return NULL;
-	}
-
-	if (strcmp(cmd,"KILL") == 0) {
-		uint32_t handle = 0;
 		if (param[0] == ':') {
 			handle = strtoul(param+1, NULL, 16);
 		} else if (param[0] == '.') {
 			handle = skynet_handle_findname(param+1);
 		} else {
-			skynet_error(context, "Can't kill %s",param);
+			skynet_error(context, "Can't monitor %s",param);
-			// todo : kill global service
+			// todo : monitor global service
-		}
-		if (handle) {
-			handle_exit(context, handle);
-		}
-		return NULL;
-	}
-
-	if (strcmp(cmd,"LAUNCH") == 0) {
-		size_t sz = strlen(param);
-		char tmp[sz+1];
-		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) {
-			return NULL;
-		} else {
-			_id_to_hex(context->result, inst->handle);
-			return context->result;
 		}
 	}
+	G_NODE.monitor_exit = handle;
+	return NULL;
+}
 
-	if (strcmp(cmd,"GETENV") == 0) {
+static const char *
-		return skynet_getenv(param);
+cmd_mqlen(struct skynet_context * context, const char * param) {
-	}
+	int len = skynet_mq_length(context->queue);
+	sprintf(context->result, "%d", len);
+	return context->result;
+}
 
-	if (strcmp(cmd,"SETENV") == 0) {
+static struct command_func cmd_funcs[] = {
-		size_t sz = strlen(param);
+	{ "TIMEOUT", cmd_timeout },
-		char key[sz+1];
+	{ "REG", cmd_reg },
-		int i;
+	{ "QUERY", cmd_query },
-		for (i=0;param[i] != ' ' && param[i];i++) {
+	{ "NAME", cmd_name },
-			key[i] = param[i];
+	{ "NOW", cmd_now },
+	{ "EXIT", cmd_exit },
+	{ "KILL", cmd_kill },
+	{ "LAUNCH", cmd_launch },
+	{ "GETENV", cmd_getenv },
+	{ "SETENV", cmd_setenv },
+	{ "STARTTIME", cmd_starttime },
+	{ "ENDLESS", cmd_endless },
+	{ "ABORT", cmd_abort },
+	{ "MONITOR", cmd_monitor },
+	{ "MQLEN", cmd_mqlen },
+	{ NULL, NULL },
+};
+
+const char * 
+skynet_command(struct skynet_context * context, const char * cmd , const char * param) {
+	struct command_func * method = &cmd_funcs[0];
+	while(method->name) {
+		if (strcmp(cmd, method->name) == 0) {
+			return method->func(context, param);
 		}
-		if (param[i] == '\0')
+		++method;
-			return NULL;
-
-		key[i] = '\0';
-		param += i+1;
-		
-		skynet_setenv(key,param);
-		return NULL;
-	}
-
-	if (strcmp(cmd,"STARTTIME") == 0) {
-		uint32_t sec = skynet_gettime_fixsec();
-		sprintf(context->result,"%u",sec);
-		return context->result;
-	}
-
-	if (strcmp(cmd,"ENDLESS") == 0) {
-		if (context->endless) {
-			strcpy(context->result, "1");
-			context->endless = false;
-			return context->result;
-		}
-		return NULL;
-	}
-
-	if (strcmp(cmd,"ABORT") == 0) {
-		skynet_handle_retireall();
-		return NULL;
-	}
-
-	if (strcmp(cmd,"MONITOR") == 0) {
-		uint32_t handle=0;
-		if (param == NULL || param[0] == '\0') {
-			if (G_NODE.monitor_exit) {
-				// return current monitor serivce
-				sprintf(context->result, ":%x", G_NODE.monitor_exit);
-				return context->result;
-			}
-			return NULL;
-		} else {
-			if (param[0] == ':') {
-				handle = strtoul(param+1, NULL, 16);
-			} else if (param[0] == '.') {
-				handle = skynet_handle_findname(param+1);
-			} else {
-				skynet_error(context, "Can't monitor %s",param);
-				// todo : monitor global service
-			}
-		}
-		G_NODE.monitor_exit = handle;
-		return NULL;
-	}
-
-	if (strcmp(cmd, "MQLEN") == 0) {
-		int len = skynet_mq_length(context->queue);
-		sprintf(context->result, "%d", len);
-		return context->result;
 	}
 
 	return NULL;
@@ -526,7 +575,6 @@ skynet_send(struct skynet_context * context, uint32_t source, uint32_t destinati
 
 		if (skynet_context_push(destination, &smsg)) {
 			skynet_free(data);
-			skynet_error(NULL, "Drop message from %x to %x (type=%d)(size=%d)", source, destination, type&0xff, (int)(sz & HANDLE_MASK));
 			return -1;
 		}
 	}
@@ -543,16 +591,15 @@ skynet_sendname(struct skynet_context * context, const char * addr , int type, i
 		des = skynet_handle_findname(addr + 1);
 		if (des == 0) {
 			if (type & PTYPE_TAG_DONTCOPY) {
-  			skynet_free(data);
+				skynet_free(data);
-  		}
+			}
-			skynet_error(context, "Drop message to %s", addr);
 			return session;
 		}
 	} else {
 		_filter_args(context, type, &session, (void **)&data, &sz);
 
 		struct remote_message * rmsg = skynet_malloc(sizeof(*rmsg));
-		_copy_name(rmsg->destination.name, addr);
+		copy_name(rmsg->destination.name, addr);
 		rmsg->destination.handle = 0;
 		rmsg->message = data;
 		rmsg->sz = sz;
@@ -590,3 +637,27 @@ skynet_context_send(struct skynet_context * ctx, void * msg, size_t sz, uint32_t
 
 	skynet_mq_push(ctx->queue, &smsg);
 }
+
+void 
+skynet_globalinit(void) {
+	G_NODE.total = 0;
+	G_NODE.monitor_exit = 0;
+	if (pthread_key_create(&G_NODE.handle_key, NULL)) {
+		fprintf(stderr, "pthread_key_create failed");
+		exit(1);
+	}
+	// set mainthread's key
+	skynet_initthread(THREAD_MAIN);
+}
+
+void 
+skynet_globalexit(void) {
+	pthread_key_delete(G_NODE.handle_key);
+}
+
+void
+skynet_initthread(int m) {
+	uintptr_t v = (uint32_t)(-m);
+	pthread_setspecific(G_NODE.handle_key, (void *)v);
+}
+

skynet-src/skynet_server.h

@@ -21,4 +21,8 @@ int skynet_context_total();
 
 void skynet_context_endless(uint32_t handle);	// for monitor
 
+void skynet_globalinit(void);
+void skynet_globalexit(void);
+void skynet_initthread(int m);
+
 #endif

skynet-src/skynet_start.c

@@ -5,7 +5,6 @@
 #include "skynet_handle.h"
 #include "skynet_module.h"
 #include "skynet_timer.h"
-#include "skynet_harbor.h"
 #include "skynet_monitor.h"
 #include "skynet_socket.h"
 
@@ -50,6 +49,7 @@ wakeup(struct monitor *m, int busy) {
 static void *
 _socket(void *p) {
 	struct monitor * m = p;
+	skynet_initthread(THREAD_SOCKET);
 	for (;;) {
 		int r = skynet_socket_poll();
 		if (r==0)
@@ -81,6 +81,7 @@ _monitor(void *p) {
 	struct monitor * m = p;
 	int i;
 	int n = m->count;
+	skynet_initthread(THREAD_MONITOR);
 	for (;;) {
 		CHECK_ABORT
 		for (i=0;i<n;i++) {
@@ -98,6 +99,7 @@ _monitor(void *p) {
 static void *
 _timer(void *p) {
 	struct monitor * m = p;
+	skynet_initthread(THREAD_TIMER);
 	for (;;) {
 		skynet_updatetime();
 		CHECK_ABORT
@@ -117,6 +119,7 @@ _worker(void *p) {
 	int id = wp->id;
 	struct monitor *m = wp->m;
 	struct skynet_monitor *sm = m->m[id];
+	skynet_initthread(THREAD_WORKER);
 	for (;;) {
 		if (skynet_context_message_dispatch(sm)) {
 			CHECK_ABORT
@@ -177,12 +180,17 @@ _start(int thread) {
 	free_monitor(m);
 }
 
-static int
+static void
-_start_master(const char * master) {
+bootstrap(const char * cmdline) {
-	struct skynet_context *ctx = skynet_context_new("master", master);
+	int sz = strlen(cmdline);
-	if (ctx == NULL)
+	char name[sz+1];
-		return 1;
+	char args[sz+1];
-	return 0;	
+	sscanf(cmdline, "%s %s", name, args);
+	struct skynet_context *ctx = skynet_context_new(name, args);
+	if (ctx == NULL) {
+		skynet_error(NULL, "Bootstrap error : %s\n", cmdline);
+		exit(1);
+	}
 }
 
 void 
@@ -194,32 +202,8 @@ skynet_start(struct skynet_config * config) {
 	skynet_timer_init();
 	skynet_socket_init();
 
-	struct skynet_context *ctx;
+	bootstrap(config->bootstrap);
-	ctx = skynet_context_new("logger", config->logger);
-	if (ctx == NULL) {
-		fprintf(stderr,"launch logger error");
-		exit(1);
-	}
-
-	if (config->standalone) {
-		if (_start_master(config->standalone)) {
-			fprintf(stderr, "Init fail : mater");
-			return;
-		}
-	}
-	// harbor must be init first
-	if (skynet_harbor_start(config->master , config->local)) {
-		fprintf(stderr, "Init fail : no master");
-		return;
-	}
-
-	ctx = skynet_context_new("snlua", "launcher");
-	if (ctx) {
-		skynet_command(ctx, "REG", ".launcher");
-		ctx = skynet_context_new("snlua", config->start);
-	}
 
 	_start(config->thread);
 	skynet_socket_free();
 }
-

skynet-src/socket_server.c

@@ -815,6 +815,22 @@ report_accept(struct socket_server *ss, struct socket *s, struct socket_message
 	return 1;
 }
 
+static inline void 
+clear_closed_event(struct socket_server *ss, struct socket_message * result, int type) {
+	if (type == SOCKET_CLOSE || type == SOCKET_ERROR) {
+		int id = result->id;
+		int i;
+		for (i=ss->event_index; i<ss->event_n; i++) {
+			struct event *e = &ss->ev[i];
+			struct socket *s = e->s;
+			if (s) {
+				if (s->type == SOCKET_TYPE_INVALID && s->id == id) {
+					e->s = NULL;
+				}
+			}
+		}
+	}
+}
 
 // return type
 int 
@@ -823,9 +839,10 @@ socket_server_poll(struct socket_server *ss, struct socket_message * result, int
 		if (ss->checkctrl) {
 			if (has_cmd(ss)) {
 				int type = ctrl_cmd(ss, result);
-				if (type != -1)
+				if (type != -1) {
+					clear_closed_event(ss, result, type);
 					return type;
-				else
+				} else
 					continue;
 			} else {
 				ss->checkctrl = 0;
@@ -865,12 +882,14 @@ socket_server_poll(struct socket_server *ss, struct socket_message * result, int
 				int type = send_buffer(ss, s, result);
 				if (type == -1)
 					break;
+				clear_closed_event(ss, result, type);
 				return type;
 			}
 			if (e->read) {
 				int type = forward_message(ss, s, result);
 				if (type == -1)
 					break;
+				clear_closed_event(ss, result, type);
 				return type;
 			}
 			break;

test/pingqueue.lua

@@ -1,12 +0,0 @@
-local skynet = require "skynet"
-local mqueue = require "mqueue"
-
-skynet.start(function()
-	local id = 0
-	local pingserver = skynet.newservice "pingserver"
-	mqueue.register(function(str)
-		id = id + 1
-		str = string.format("id = %d , %s",id, str)
-		return skynet.call(pingserver, "lua", "PING", str)
-	end)
-end)

test/testdeadcall.lua

@@ -0,0 +1,27 @@
+local skynet = require "skynet"
+
+local mode = ...
+
+if mode == "test" then
+
+skynet.start(function()
+	skynet.dispatch("lua", function (...)
+		print("====>", ...)
+		skynet.exit()
+	end)
+end)
+
+else
+
+	skynet.start(function()
+		local test = skynet.newservice("testdeadcall", "test")	-- launch self in test mode
+
+		print(pcall(function()
+			skynet.send(test,"lua", "hello world")
+			skynet.send(test,"lua", "never get there")
+			skynet.call(test,"lua", "fake call")
+		end))
+
+		skynet.exit()
+	end)
+end

test/testlog.lua

@@ -1,9 +0,0 @@
-local skynet = require "skynet"
-local log = require "log"
-
-skynet.start(function()
-	log.Info("hello world")
-	skynet.exit()
-end)
-
-

test/testmulticast.lua

@@ -0,0 +1,40 @@
+local skynet = require "skynet"
+local mc = require "multicast"
+local dc = require "datacenter"
+
+local mode = ...
+
+if mode == "sub" then
+
+skynet.start(function()
+	skynet.dispatch("lua", function (_,_, cmd, channel)
+		assert(cmd == "init")
+		local c = mc.new {
+			channel = channel ,
+			dispatch = function (channel, source, ...)
+				print(string.format("%s <=== %s %s",skynet.address(skynet.self()),skynet.address(source), channel), ...)
+			end
+		}
+		print(skynet.address(skynet.self()), "sub", c)
+		c:subscribe()
+		skynet.ret(skynet.pack())
+	end)
+end)
+
+else
+
+skynet.start(function()
+	local channel = mc.new()
+	print("New channel", channel)
+	for i=1,10 do
+		local sub = skynet.newservice("testmulticast", "sub")
+		skynet.call(sub, "lua", "init", channel.channel)
+	end
+
+	dc.set("MCCHANNEL", channel.channel)	-- for multi node test
+
+	print(skynet.address(skynet.self()), "===>", channel)
+	channel:publish("Hello World")
+end)
+
+end

test/testmulticast2.lua

@@ -0,0 +1,24 @@
+local skynet = require "skynet"
+local dc = require "datacenter"
+local mc = require "multicast"
+
+skynet.start(function()
+	print("remote start")
+	skynet.monitor("simplemonitor", true)
+	local console = skynet.newservice("console")
+	local channel = dc.get "MCCHANNEL"
+	for i=1,10 do
+		local sub = skynet.newservice("testmulticast", "sub")
+		skynet.call(sub, "lua", "init", channel)
+	end
+	local c = mc.new {
+		channel = channel ,
+		dispatch = function(...) print("======>", ...) end,
+	}
+	c:subscribe()
+	c:publish("Remote message")
+	c:unsubscribe()
+	c:publish("Remote message2")
+	c:delete()
+	skynet.exit()
+end)

test/testping.lua

@@ -2,7 +2,7 @@ local skynet = require "skynet"
 local snax = require "snax"
 
 skynet.start(function()
-	local ps = snax.newservice ("pingserver", "hello world")
+	local ps = snax.uniqueservice ("pingserver", "hello world")
 	print(ps.req.ping("foobar"))
 	print(ps.post.hello())
 	print(pcall(ps.req.error))

test/testsocket.lua

@@ -37,7 +37,8 @@ else
 	end
 
 	skynet.start(function()
-		local id = socket.listen("127.0.0.1", 8000)
+		local id = socket.listen("127.0.0.1", 8001)
+		print("Listen socket :", "127.0.0.1", 8001)
 
 		socket.start(id , function(id, addr)
 			print("connect from " .. addr .. " " .. id)