From a700abcdb381270b7033ddc9fe81e7541b0350e2 Mon Sep 17 00:00:00 2001 From: Cloud Wu Date: Tue, 6 May 2014 14:57:01 +0800 Subject: [PATCH] remove skynet.watch, move simplemonitor to examples --- examples/main.lua | 1 - {service => examples}/simplemonitor.lua | 2 +- lualib/skynet.lua | 28 ++++++------------------- service-src/service_harbor.c | 5 ++--- skynet-src/skynet.h | 4 ++-- skynet-src/skynet_mq.c | 14 ++++--------- skynet-src/skynet_mq.h | 2 +- skynet-src/skynet_server.c | 26 ++++++++++++++++------- test/testdeadcall.lua | 27 ++++++++++++++++++++++++ 9 files changed, 62 insertions(+), 47 deletions(-) rename {service => examples}/simplemonitor.lua (93%) create mode 100644 test/testdeadcall.lua diff --git a/examples/main.lua b/examples/main.lua index 7731bdab..8019c2bf 100644 --- a/examples/main.lua +++ b/examples/main.lua @@ -4,7 +4,6 @@ local max_client = 64 skynet.start(function() print("Server start") - skynet.monitor "simplemonitor" local console = skynet.newservice("console") skynet.newservice("debug_console",8000) skynet.newservice("simpledb") diff --git a/service/simplemonitor.lua b/examples/simplemonitor.lua similarity index 93% rename from service/simplemonitor.lua rename to examples/simplemonitor.lua index 39a7c17c..1528bb50 100644 --- a/service/simplemonitor.lua +++ b/examples/simplemonitor.lua @@ -12,7 +12,7 @@ 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 diff --git a/lualib/skynet.lua b/lualib/skynet.lua index c3b38df9..1d936052 100644 --- a/lualib/skynet.lua +++ b/lualib/skynet.lua @@ -69,13 +69,13 @@ local function dispatch_error_queue() end end -local function _error_dispatch(error_session, monitor, service) - if service then +local function _error_dispatch(error_session, error_source) + 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 @@ -87,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 = {} @@ -289,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 @@ -478,8 +463,7 @@ do REG { name = "error", id = skynet.PTYPE_ERROR, - pack = skynet.pack, - unpack = skynet.unpack, + unpack = function(...) return ... end, dispatch = _error_dispatch, } end diff --git a/service-src/service_harbor.c b/service-src/service_harbor.c index 4c5fe48b..ead12d72 100644 --- a/service-src/service_harbor.c +++ b/service-src/service_harbor.c @@ -402,9 +402,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) { - skynet_send(context, destination, source, PTYPE_RESERVED_ERROR, session, NULL, 0); - } + // report the destination is dead + 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; diff --git a/skynet-src/skynet.h b/skynet-src/skynet.h index ce82b7ac..0df5ca9c 100644 --- a/skynet-src/skynet.h +++ b/skynet-src/skynet.h @@ -8,13 +8,13 @@ #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 -#define PTYPE_RESERVED_ERROR 7 +#define PTYPE_ERROR 7 // read lualib/skynet.lua lualib/mqueue.lua #define PTYPE_RESERVED_QUEUE 8 #define PTYPE_RESERVED_DEBUG 9 diff --git a/skynet-src/skynet_mq.c b/skynet-src/skynet_mq.c index 8d235388..035880b0 100644 --- a/skynet-src/skynet_mq.c +++ b/skynet-src/skynet_mq.c @@ -162,7 +162,7 @@ void skynet_mq_push(struct message_queue *q, struct skynet_message *message) { assert(message); LOCK(q) - + q->queue[q->tail] = *message; if (++ q->tail >= q->cap) { q->tail = 0; @@ -216,30 +216,24 @@ skynet_mq_mark_release(struct message_queue *q) { UNLOCK(q) } -static int +static void _drop_queue(struct message_queue *q, message_drop drop_func, void *ud) { struct skynet_message msg; - int s = 0; while(!skynet_mq_pop(q, &msg)) { - ++s; drop_func(&msg, ud); } _release(q); - return s; } -int +void 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_func, ud); + _drop_queue(q, drop_func, ud); } else { skynet_mq_force_push(q); UNLOCK(q) } - - return ret; } diff --git a/skynet-src/skynet_mq.h b/skynet-src/skynet_mq.h index be30018c..5ac582a1 100644 --- a/skynet-src/skynet_mq.h +++ b/skynet-src/skynet_mq.h @@ -20,7 +20,7 @@ void skynet_mq_mark_release(struct message_queue *q); typedef void (*message_drop)(struct skynet_message *, void *); -int skynet_mq_release(struct message_queue *q, message_drop drop_func, void *ud); +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 diff --git a/skynet-src/skynet_server.c b/skynet-src/skynet_server.c index 08fdc016..95947b67 100644 --- a/skynet-src/skynet_server.c +++ b/skynet-src/skynet_server.c @@ -90,9 +90,18 @@ _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 * @@ -137,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_mq_release(queue, drop_message, NULL); + skynet_handle_retire(handle); + struct drop_t d = { handle }; + skynet_mq_release(queue, drop_message, &d); return NULL; } } @@ -227,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, drop_message, NULL); - if (s>0) { - skynet_error(NULL, "Drop message queue %x (%d messages)", handle,s); - } + struct drop_t d = { handle }; + skynet_mq_release(q, drop_message, &d); return 0; } @@ -518,7 +527,10 @@ 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)); + if (destination) { + // don't report the message to 0 (system service) + skynet_error(NULL, "Drop message from %x to %x (type=%d)(size=%d)", source, destination, type&0xff, (int)(sz & HANDLE_MASK)); + } return -1; } } diff --git a/test/testdeadcall.lua b/test/testdeadcall.lua new file mode 100644 index 00000000..d348a411 --- /dev/null +++ b/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 \ No newline at end of file