mirror of
https://github.com/cloudwu/skynet.git
synced 2026-07-23 19:43:09 +00:00
remove skynet.watch, move simplemonitor to examples
This commit is contained in:
@@ -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")
|
||||
|
||||
@@ -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
|
||||
@@ -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
|
||||
|
||||
@@ -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;
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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;
|
||||
}
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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;
|
||||
}
|
||||
}
|
||||
|
||||
27
test/testdeadcall.lua
Normal file
27
test/testdeadcall.lua
Normal file
@@ -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
|
||||
Reference in New Issue
Block a user