hotfix the bug introduce by last release

clear coroutine_pool when gc

HISTORY.md

@@ -1,3 +1,13 @@
+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

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
@@ -703,8 +709,15 @@ 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,7 +49,7 @@ 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

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>

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) {
-			q->head = 0;
+		int head = q->head;
+		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);

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