diff --git a/lualib/skynet.lua b/lualib/skynet.lua index 11daa1e8..03699f95 100644 --- a/lualib/skynet.lua +++ b/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 diff --git a/lualib/snax/gateserver.lua b/lualib/snax/gateserver.lua index bc5f6547..6cce5266 100644 --- a/lualib/snax/gateserver.lua +++ b/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 diff --git a/service-src/service_snlua.c b/service-src/service_snlua.c index 50530e33..0063c8c6 100644 --- a/service-src/service_snlua.c +++ b/service-src/service_snlua.c @@ -3,7 +3,6 @@ #include #include #include -#include #include #include diff --git a/skynet-src/skynet_harbor.c b/skynet-src/skynet_harbor.c index 48814c83..379fce80 100644 --- a/skynet-src/skynet_harbor.c +++ b/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); + } +} diff --git a/skynet-src/skynet_harbor.h b/skynet-src/skynet_harbor.h index b699f625..62982455 100644 --- a/skynet-src/skynet_harbor.h +++ b/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 diff --git a/skynet-src/skynet_mq.c b/skynet-src/skynet_mq.c index 027c4dc0..8cc8f9da 100644 --- a/skynet-src/skynet_mq.c +++ b/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) { diff --git a/skynet-src/skynet_mq.h b/skynet-src/skynet_mq.h index df1a9998..17178b4f 100644 --- a/skynet-src/skynet_mq.h +++ b/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(); diff --git a/skynet-src/skynet_server.c b/skynet-src/skynet_server.c index bdf0dcc1..0a3a6b00 100644 --- a/skynet-src/skynet_server.c +++ b/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); diff --git a/skynet-src/skynet_server.h b/skynet-src/skynet_server.h index 27b6b768..92e125b0 100644 --- a/skynet-src/skynet_server.h +++ b/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); diff --git a/skynet-src/skynet_start.c b/skynet-src/skynet_start.c index bef06b4c..5f511d29 100644 --- a/skynet-src/skynet_start.c +++ b/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); diff --git a/test/testoverload.lua b/test/testoverload.lua new file mode 100644 index 00000000..10fe39a8 --- /dev/null +++ b/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