hotfix the bug introduce by last release

clear coroutine_pool when gc

HISTORY.md

@@ -1,3 +1,21 @@
+v0.7.4 (2014-10-13)
+-----------
+* Bugfix : clear coroutine pool when GC
+* hotfix : A bug introduce by 0.7.3 
+
+v0.7.3 (2014-10-13)
+-----------
+* Add some logs (warning) when overload
+* Bugfix: crash on exit
+
+v0.7.2 (2014-9-29)
+-----------
+* Bugfix : datacenter.wait
+* Bugfix : error in forker coroutine
+* Add skynet.term
+* Accept socket report port
+* sharedata can be update more than once
+
 v0.7.1 (2014-9-22)
 -----------
 * bugfix: wakeup sleep should return BREAK

examples/main_log.lua

@@ -1,9 +1,16 @@
 local skynet = require "skynet"
+local harbor = require "skynet.harbor"
+
+local function monitor_master()
+	harbor.linkmaster()
+	print("master is down")
+	skynet.exit()
+end
 
 skynet.start(function()
 	print("Log server start")
 	skynet.monitor "simplemonitor"
 	local log = skynet.newservice("globallog")
-	skynet.exit()
+	skynet.fork(monitor_master)
 end)
 

examples/watchdog.lua

@@ -10,6 +10,7 @@ local gate
 local agent = {}
 
 function SOCKET.open(fd, addr)
+	skynet.error("New client from : " .. addr)
 	agent[fd] = skynet.newservice("agent")
 	skynet.call(agent[fd], "lua", "start", gate, fd, proto)
 end

lualib-src/lua-sharedata.c

@@ -747,9 +747,6 @@ lupdate(lua_State *L) {
 	luaL_checktype(L, 3, LUA_TTABLE);
 	struct ctrl * c= lua_touserdata(L, 1);
 	struct table *n = lua_touserdata(L, 2);
-	if (c->update) {
-		return luaL_error(L, "can't update more than once");
-	}
 	if (c->root == n) {
 		return luaL_error(L, "You should update a new object");
 	}

lualib/http/internal.lua

@@ -4,14 +4,15 @@ local LIMIT = 8192
 
 local function chunksize(readbytes, body)
 	while true do
-		if #body > 128 then
-			return
-		end
-		body = body .. readbytes()
 		local f,e = body:find("\r\n",1,true)
 		if f then
 			return tonumber(body:sub(1,f-1),16), body:sub(e+1)
 		end
+		if #body > 128 then
+			-- pervent the attacker send very long stream without \r\n
+			return
+		end
+		body = body .. readbytes()
 	end
 end
 

lualib/sharedata/corelib.lua

@@ -59,15 +59,15 @@ function meta:__index(key)
 		local newobj, newtbl = needupdate(self.__gcobj)
 		if newobj then
 			local newgcobj = newtbl.__gcobj
-			-- todo: update
 			local root = findroot(self)
 			update(root, newobj, newgcobj)
 			if obj == self.__obj then
 				error ("The key [" .. genkey(self) .. "] doesn't exist after update")
 			end
+			obj = self.__obj
 		end
 	end
-	local v = index(self.__obj, key)
+	local v = index(obj, key)
 	if type(v) == "userdata" then
 		local r = setmetatable({
 			__obj = v,
@@ -127,7 +127,7 @@ end
 
 function conf.update(self, pointer)
 	local cobj = self.__obj
-	assert(isdirty(cobj), "Obly dirty object can be update")
+	assert(isdirty(cobj), "Only dirty object can be update")
 	core.update(self.__gcobj, pointer, { __gcobj = core.box(pointer) })
 end
 

lualib/skynet.lua

@@ -95,6 +95,7 @@ end
 
 local coroutine_pool = {}
 local coroutine_yield = coroutine.yield
+local coroutine_count = 0
 
 local function co_create(f)
 	local co = table.remove(coroutine_pool)
@@ -109,6 +110,11 @@ local function co_create(f)
 				f(coroutine_yield())
 			end
 		end)
+		coroutine_count = coroutine_count + 1
+		if coroutine_count > 1024 then
+			skynet.error("May overload, create 1024 task")
+			coroutine_count = 0
+		end
 	else
 		coroutine.resume(co, f)
 	end
@@ -143,13 +149,15 @@ end
 function suspend(co, result, command, param, size)
 	if not result then
 		local session = session_coroutine_id[co]
-		local addr = session_coroutine_address[co]
+		if session then -- coroutine may fork by others (session is nil)
-		if session ~= 0 then
+			local addr = session_coroutine_address[co]
-			-- only call response error
+			if session ~= 0 then
-			c.send(addr, skynet.PTYPE_ERROR, session, "")
+				-- only call response error
+				c.send(addr, skynet.PTYPE_ERROR, session, "")
+			end
+			session_coroutine_id[co] = nil
+			session_coroutine_address[co] = nil
 		end
-		session_coroutine_id[co] = nil
-		session_coroutine_address[co] = nil
 		error(debug.traceback(co,tostring(command)))
 	end
 	if command == "CALL" then
@@ -166,7 +174,7 @@ function suspend(co, result, command, param, size)
 		session_response[co] = true
 		local ret
 		if not dead_service[co_address] then
-			ret = c.send(co_address, skynet.PTYPE_RESPONSE, co_session, param, size) >= 0
+			ret = c.send(co_address, skynet.PTYPE_RESPONSE, co_session, param, size) ~= nil
 		elseif size == nil then
 			c.trash(param, size)
 			ret = false
@@ -200,9 +208,9 @@ function suspend(co, result, command, param, size)
 			local ret
 			if not dead_service[co_address] then
 				if ok then
-					ret = c.send(co_address, skynet.PTYPE_RESPONSE, co_session, f(...)) >=0
+					ret = c.send(co_address, skynet.PTYPE_RESPONSE, co_session, f(...)) ~= nil
 				else
-					ret = c.send(co_address, skynet.PTYPE_ERROR, co_session, "") >=0
+					ret = c.send(co_address, skynet.PTYPE_ERROR, co_session, "") ~= nil
 				end
 			else
 				ret = false
@@ -329,6 +337,7 @@ function skynet.time()
 end
 
 function skynet.exit()
+	fork_queue = {}	-- no fork coroutine can be execute after skynet.exit
 	skynet.send(".launcher","lua","REMOVE",skynet.self())
 	-- report the sources that call me
 	for co, session in pairs(session_coroutine_id) do
@@ -696,8 +705,19 @@ function skynet.task(ret)
 	return t
 end
 
+function skynet.term(service)
+	return _error_dispatch(0, service)
+end
+
+local function clear_pool()
+	coroutine_pool = {}
+end
+
 -- Inject internal debug framework
 local debug = require "skynet.debug"
-debug(skynet, dispatch_message)
+debug(skynet, {
+	dispatch = dispatch_message,
+	clear = clear_pool,
+})
 
 return skynet

lualib/skynet/debug.lua

@@ -2,7 +2,7 @@ local io = io
 local table = table
 local debug = debug
 
-return function (skynet, dispatch_func)
+return function (skynet, export)
 
 local internal_info_func
 
@@ -18,7 +18,7 @@ function dbgcmd.MEM()
 end
 
 function dbgcmd.GC()
-	coroutine_pool = {}
+	export.clear()
 	collectgarbage "collect"
 end
 
@@ -49,11 +49,15 @@ end
 
 function dbgcmd.RUN(source, filename)
 	local inject = require "skynet.inject"
-	local output = inject(source, filename , dispatch_func, skynet.register_protocol)
+	local output = inject(source, filename , export.dispatch, skynet.register_protocol)
 	collectgarbage "collect"
 	skynet.ret(skynet.pack(table.concat(output, "\n")))
 end
 
+function dbgcmd.TERM(service)
+	skynet.term(service)
+end
+
 local function _debug_dispatch(session, address, cmd, ...)
 	local f = dbgcmd[cmd]
 	assert(f, cmd)

lualib/skynet/harbor.lua

@@ -19,4 +19,8 @@ function harbor.connect(id)
 	skynet.call(".cslave", "lua", "CONNECT", id)
 end
 
+function harbor.linkmaster()
+	skynet.call(".cslave", "lua", "LINKMASTER")
+end
+
 return harbor

lualib/snax/gateserver.lua

@@ -37,6 +37,7 @@ function gateserver.start(handler)
 		local port = assert(conf.port)
 		maxclient = conf.maxclient or 1024
 		nodelay = conf.nodelay
+		skynet.error(string.format("Listen on %s:%d", address, port))
 		socket = socketdriver.listen(address, port)
 		socketdriver.start(socket)
 		if handler.open then

service-src/service_snlua.c

@@ -3,7 +3,6 @@
 #include <lua.h>
 #include <lualib.h>
 #include <lauxlib.h>
-#include <assert.h>
 
 #include <assert.h>
 #include <string.h>

service/cslave.lua

@@ -9,6 +9,7 @@ local queryname = {}
 local harbor = {}
 local harbor_service
 local monitor = {}
+local monitor_master_set = {}
 
 local function read_package(fd)
 	local sz = socket.read(fd, 1)
@@ -58,7 +59,7 @@ local function ready()
 		connect_slave(k,v)
 	end
 	for name,address in pairs(globalname) do
-		skynet.redirect(harbor_service, address, "harbor", "N " .. name)
+		skynet.redirect(harbor_service, address, "harbor", 0, "N " .. name)
 	end
 end
 
@@ -99,6 +100,9 @@ local function monitor_master(master_fd)
 			end
 		else
 			skynet.error("Master disconnect")
+			for _, v in ipairs(monitor_master_set) do
+				v(true)
+			end
 			socket.close(master_fd)
 			break
 		end
@@ -183,6 +187,10 @@ function harbor.LINK(fd, id)
 	end
 end
 
+function harbor.LINKMASTER()
+	table.insert(monitor_master_set, skynet.response())
+end
+
 function harbor.CONNECT(fd, id)
 	if not slaves[id] then
 		if monitor[id] == nil then

service/datacenterd.lua

@@ -33,7 +33,7 @@ local function update(db, key, value, ...)
 	end
 end
 
-local function wakeup(db, key1, key2, ...)
+local function wakeup(db, key1, ...)
 	if key1 == nil then
 		return
 	end
@@ -43,7 +43,7 @@ local function wakeup(db, key1, key2, ...)
 	end
 	if q[mode] == "queue" then
 		db[key1] = nil
-		if key2 then
+		if select("#", ...) ~= 1 then
 			-- throw error because can't wake up a branch
 			for _,response in ipairs(q) do
 				response(false)
@@ -53,7 +53,7 @@ local function wakeup(db, key1, key2, ...)
 		end
 	else
 		-- it's branch
-		return wakeup(q , key2, ...)
+		return wakeup(q , ...)
 	end
 end
 

service/sharedatad.lua

@@ -4,6 +4,7 @@ local table = table
 
 local NORET = {}
 local pool = {}
+local pool_count = {}
 local objmap = {}
 
 local function newobj(name, tbl)
@@ -13,6 +14,7 @@ local function newobj(name, tbl)
 	local v = { value = tbl , obj = cobj, watch = {} }
 	objmap[cobj] = v
 	pool[name] = v
+	pool_count[name] = { n = 0, threshold = 16 }
 end
 
 local function collectobj()
@@ -55,10 +57,11 @@ end
 function CMD.delete(name)
 	local v = assert(pool[name])
 	pool[name] = nil
+	pool_count[name] = nil
 	assert(objmap[v.obj])
 	objmap[v.obj] = true
 	sharedata.host.decref(v.obj)
-	for _,response in ipairs(v.watch) do
+	for _,response in pairs(v.watch) do
 		response(true)
 	end
 end
@@ -86,23 +89,42 @@ function CMD.update(name, t)
 		objmap[oldcobj] = true
 		sharedata.host.decref(oldcobj)
 		pool[name] = nil
+		pool_count[name] = nil
 	end
 	CMD.new(name, t)
 	local newobj = pool[name].obj
 	if watch then
 		sharedata.host.markdirty(oldcobj)
-		for _,response in ipairs(watch) do
+		for _,response in pairs(watch) do
 			response(true, newobj)
 		end
 	end
 end
 
+local function check_watch(queue)
+	local n = 0
+	for k,response in pairs(queue) do
+		if not response "TEST" then
+			queue[k] = nil
+			n = n + 1
+		end
+	end
+	return n
+end
+
 function CMD.monitor(name, obj)
 	local v = assert(pool[name])
 	if obj ~= v.obj then
 		return v.obj
 	end
 
+	local n = pool_count[name].n
+	pool_count[name].n = n + 1
+	if n > pool_count[name].threshold then
+		n = n - check_watch(v.watch)
+		pool_count[name].threshold = n * 2
+	end
+
 	table.insert(v.watch, skynet.response())
 
 	return NORET

skynet-src/skynet_harbor.c

@@ -31,5 +31,17 @@ skynet_harbor_init(int harbor) {
 
 void
 skynet_harbor_start(void *ctx) {
+	// the HARBOR must be reserved to ensure the pointer is valid.
+	// It will be released at last by calling skynet_harbor_exit
+	skynet_context_reserve(ctx);
 	REMOTE = ctx;
 }
+
+void
+skynet_harbor_exit() {
+	struct skynet_context * ctx = REMOTE;
+	REMOTE= NULL;
+	if (ctx) {
+		skynet_context_release(ctx);
+	}
+}

skynet-src/skynet_harbor.h

@@ -26,5 +26,6 @@ void skynet_harbor_send(struct remote_message *rmsg, uint32_t source, int sessio
 int skynet_harbor_message_isremote(uint32_t handle);
 void skynet_harbor_init(int harbor);
 void skynet_harbor_start(void * ctx);
+void skynet_harbor_exit();
 
 #endif

skynet-src/skynet_mq.c

@@ -15,6 +15,7 @@
 // 1 means mq is in global mq , or the message is dispatching.
 
 #define MQ_IN_GLOBAL 1
+#define MQ_OVERLOAD 1024
 
 struct message_queue {
 	uint32_t handle;
@@ -24,6 +25,8 @@ struct message_queue {
 	int lock;
 	int release;
 	int in_global;
+	int overload;
+	int overload_threshold;
 	struct skynet_message *queue;
 	struct message_queue *next;
 };
@@ -116,6 +119,8 @@ skynet_mq_create(uint32_t handle) {
 	// 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->overload = 0;
+	q->overload_threshold = MQ_OVERLOAD;
 	q->queue = skynet_malloc(sizeof(struct skynet_message) * q->cap);
 	q->next = NULL;
 
@@ -150,17 +155,42 @@ skynet_mq_length(struct message_queue *q) {
 	return tail + cap - head;
 }
 
+int
+skynet_mq_overload(struct message_queue *q) {
+	if (q->overload) {
+		int overload = q->overload;
+		q->overload = 0;
+		return overload;
+	} 
+	return 0;
+}
+
 int
 skynet_mq_pop(struct message_queue *q, struct skynet_message *message) {
 	int ret = 1;
 	LOCK(q)
 
 	if (q->head != q->tail) {
-		*message = q->queue[q->head];
+		*message = q->queue[q->head++];
 		ret = 0;
-		if ( ++ q->head >= q->cap) {
+		int head = q->head;
-			q->head = 0;
+		int tail = q->tail;
+		int cap = q->cap;
+
+		if (head >= cap) {
+			q->head = head = 0;
 		}
+		int length = tail - head;
+		if (length < 0) {
+			length += cap;
+		}
+		while (length > q->overload_threshold) {
+			q->overload = length;
+			q->overload_threshold *= 2;
+		}
+	} else {
+		// reset overload_threshold when queue is empty
+		q->overload_threshold = MQ_OVERLOAD;
 	}
 
 	if (ret) {

skynet-src/skynet_mq.h

@@ -30,6 +30,7 @@ void skynet_mq_push(struct message_queue *q, struct skynet_message *message);
 
 // return the length of message queue, for debug
 int skynet_mq_length(struct message_queue *q);
+int skynet_mq_overload(struct message_queue *q);
 
 void skynet_mq_init();
 

skynet-src/skynet_server.c

@@ -178,6 +178,14 @@ skynet_context_grab(struct skynet_context *ctx) {
 	__sync_add_and_fetch(&ctx->ref,1);
 }
 
+void
+skynet_context_reserve(struct skynet_context *ctx) {
+	skynet_context_grab(ctx);
+	// don't count the context reserved, because skynet abort (the worker threads terminate) only when the total context is 0 .
+	// the reserved context will be release at last.
+	context_dec();
+}
+
 static void 
 delete_context(struct skynet_context *ctx) {
 	if (ctx->logfile) {
@@ -283,6 +291,10 @@ skynet_context_message_dispatch(struct skynet_monitor *sm, struct message_queue
 			n = skynet_mq_length(q);
 			n >>= weight;
 		}
+		int overload = skynet_mq_overload(q);
+		if (overload) {
+			skynet_error(ctx, "May overload, message queue length = %d", overload);
+		}
 
 		skynet_monitor_trigger(sm, msg.source , handle);
 

skynet-src/skynet_server.h

@@ -10,6 +10,7 @@ struct skynet_monitor;
 
 struct skynet_context * skynet_context_new(const char * name, const char * parm);
 void skynet_context_grab(struct skynet_context *);
+void skynet_context_reserve(struct skynet_context *ctx);
 struct skynet_context * skynet_context_release(struct skynet_context *);
 uint32_t skynet_context_handle(struct skynet_context *);
 int skynet_context_push(uint32_t handle, struct skynet_message *message);

skynet-src/skynet_socket.c

@@ -36,10 +36,9 @@ forward_message(int type, bool padding, struct socket_message * result) {
 	int sz = sizeof(*sm);
 	if (padding) {
 		if (result->data) {
-			sz += strlen(result->data) + 1;
+			sz += strlen(result->data);
 		} else {
 			result->data = "";
-			sz += 1;
 		}
 	}
 	sm = (struct skynet_socket_message *)skynet_malloc(sz);

skynet-src/skynet_start.c

@@ -127,7 +127,6 @@ _worker(void *p) {
 	for (;;) {
 		q = skynet_context_message_dispatch(sm, q, weight);
 		if (q == NULL) {
-			CHECK_ABORT
 			if (pthread_mutex_lock(&m->mutex) == 0) {
 				++ m->sleep;
 				// "spurious wakeup" is harmless,
@@ -140,6 +139,7 @@ _worker(void *p) {
 				}
 			}
 		} 
+		CHECK_ABORT
 	}
 	return NULL;
 }
@@ -232,6 +232,9 @@ skynet_start(struct skynet_config * config) {
 	bootstrap(ctx, config->bootstrap);
 
 	_start(config->thread);
+
+	// harbor_exit may call socket send, so it should exit before socket_free
+	skynet_harbor_exit();
 	skynet_socket_free();
 	if (config->daemon) {
 		daemon_exit(config->daemon);

skynet-src/socket_server.c

@@ -335,7 +335,6 @@ open_socket(struct socket_server *ss, struct request_open * request, struct sock
 			sock = -1;
 			continue;
 		}
-		sp_nonblocking(sock);
 		break;
 	}
 
@@ -832,7 +831,10 @@ report_accept(struct socket_server *ss, struct socket *s, struct socket_message
 	result->data = NULL;
 
 	void * sin_addr = (u.s.sa_family == AF_INET) ? (void*)&u.v4.sin_addr : (void *)&u.v6.sin6_addr;
-	if (inet_ntop(u.s.sa_family, sin_addr, ss->buffer, sizeof(ss->buffer))) {
+	int sin_port = ntohs((u.s.sa_family == AF_INET) ? u.v4.sin_port : u.v6.sin6_port);
+	char tmp[INET6_ADDRSTRLEN];
+	if (inet_ntop(u.s.sa_family, sin_addr, tmp, sizeof(tmp))) {
+		snprintf(ss->buffer, sizeof(ss->buffer), "%s:%d", tmp, sin_port);
 		result->data = ss->buffer;
 	}
 

test/testoverload.lua

@@ -0,0 +1,44 @@
+local skynet = require "skynet"
+
+local mode = ...
+
+if mode == "slave" then
+
+local CMD = {}
+
+function CMD.sum(n)
+	skynet.error("for loop begin")
+	local s = 0
+	for i = 1, n do
+		s = s + i
+	end
+	skynet.error("for loop end")
+end
+
+function CMD.blackhole()
+end
+
+skynet.start(function()
+	skynet.dispatch("lua", function(_,_, cmd, ...)
+		local f = CMD[cmd]
+		f(...)
+	end)
+end)
+
+else
+
+skynet.start(function()
+	local slave = skynet.newservice(SERVICE_NAME, "slave")
+	for step = 1, 20 do
+		skynet.error("overload test ".. step)
+		for i = 1, 512 * step do
+			skynet.send(slave, "lua", "blackhole")
+		end
+		skynet.sleep(step)
+	end
+	local n = 1000000000
+	skynet.error(string.format("endless test n=%d", n))
+	skynet.send(slave, "lua", "sum", n)
+end)
+
+end

test/testterm.lua

@@ -0,0 +1,15 @@
+local skynet = require "skynet"
+
+local function term()
+	skynet.error("Sleep one second, and term the call to UNEXIST")
+	skynet.sleep(100)
+	local self = skynet.self()
+	skynet.send(skynet.self(), "debug", "TERM", "UNEXIST")
+end
+
+skynet.start(function()
+	skynet.fork(term)
+	skynet.error("call an unexist named service UNEXIST, may block")
+	pcall(skynet.call, "UNEXIST", "lua", "test")
+	skynet.error("unblock the unexisted service call")
+end)