From b46ba5838c26d48414cf8d308dacb90a63cbfcab Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E4=BA=91=E9=A3=8E?= Date: Fri, 28 Sep 2012 16:43:04 +0800 Subject: [PATCH] add endless-loop monitor --- Makefile | 1 + skynet-src/skynet_monitor.c | 42 +++++++++++++++++++++++++++++++ skynet-src/skynet_monitor.h | 13 ++++++++++ skynet-src/skynet_server.c | 7 +++++- skynet-src/skynet_server.h | 3 ++- skynet-src/skynet_start.c | 49 +++++++++++++++++++++++++++++++------ 6 files changed, 106 insertions(+), 9 deletions(-) create mode 100644 skynet-src/skynet_monitor.c create mode 100644 skynet-src/skynet_monitor.h diff --git a/Makefile b/Makefile index 7092729e..0df05675 100644 --- a/Makefile +++ b/Makefile @@ -33,6 +33,7 @@ skynet : \ skynet-src/skynet_multicast.c \ skynet-src/skynet_group.c \ skynet-src/skynet_env.c \ + skynet-src/skynet_monitor.c \ luacompat/compat52.c gcc $(CFLAGS) -Iluacompat -Wl,-E -o $@ $^ -Iskynet-src -lpthread -ldl -lrt -Wl,-E $(LUALIB) -lm diff --git a/skynet-src/skynet_monitor.c b/skynet-src/skynet_monitor.c new file mode 100644 index 00000000..85c137cb --- /dev/null +++ b/skynet-src/skynet_monitor.c @@ -0,0 +1,42 @@ +#include "skynet_monitor.h" +#include "skynet.h" + +#include +#include + +struct skynet_monitor { + int version; + int check_version; + uint32_t source; + uint32_t destination; +}; + +struct skynet_monitor * +skynet_monitor_new() { + struct skynet_monitor * ret = malloc(sizeof(*ret)); + memset(ret, 0, sizeof(*ret)); + return ret; +} + +void +skynet_monitor_delete(struct skynet_monitor *sm) { + free(sm); +} + +void +skynet_monitor_trigger(struct skynet_monitor *sm, uint32_t source, uint32_t destination) { + sm->source = source; + sm->destination = destination; + __sync_fetch_and_add(&sm->version , 1); +} + +void +skynet_monitor_check(struct skynet_monitor *sm) { + if (sm->version == sm->check_version) { + if (sm->destination) { + skynet_error(NULL, "A message from [ :%08x ] to [ :%08x ] maybe in an endless loop", sm->source , sm->destination); + } + } else { + sm->check_version = sm->version; + } +} diff --git a/skynet-src/skynet_monitor.h b/skynet-src/skynet_monitor.h new file mode 100644 index 00000000..4fb592c8 --- /dev/null +++ b/skynet-src/skynet_monitor.h @@ -0,0 +1,13 @@ +#ifndef SKYNET_MONITOR_H +#define SKYNET_MONITOR_H + +#include + +struct skynet_monitor; + +struct skynet_monitor * skynet_monitor_new(); +void skynet_monitor_delete(struct skynet_monitor *); +void skynet_monitor_trigger(struct skynet_monitor *, uint32_t source, uint32_t destination); +void skynet_monitor_check(struct skynet_monitor *); + +#endif diff --git a/skynet-src/skynet_server.c b/skynet-src/skynet_server.c index d5a7bda3..04e85cd7 100644 --- a/skynet-src/skynet_server.c +++ b/skynet-src/skynet_server.c @@ -8,6 +8,7 @@ #include "skynet.h" #include "skynet_multicast.h" #include "skynet_group.h" +#include "skynet_monitor.h" #include #include @@ -220,7 +221,7 @@ _dispatch_message(struct skynet_context *ctx, struct skynet_message *msg) { } int -skynet_context_message_dispatch(void) { +skynet_context_message_dispatch(struct skynet_monitor *sm) { struct message_queue * q = skynet_globalmq_pop(); if (q==NULL) return 1; @@ -242,6 +243,8 @@ skynet_context_message_dispatch(void) { return 0; } + skynet_monitor_trigger(sm, msg.source , handle); + if (ctx->cb == NULL) { free(msg.data); skynet_error(NULL, "Drop message from %x to %x without callback , size = %d",msg.source, handle, (int)msg.sz); @@ -253,6 +256,8 @@ skynet_context_message_dispatch(void) { skynet_mq_pushglobal(q); skynet_context_release(ctx); + skynet_monitor_trigger(sm, 0,0); + return 0; } diff --git a/skynet-src/skynet_server.h b/skynet-src/skynet_server.h index 07275700..36cc0141 100644 --- a/skynet-src/skynet_server.h +++ b/skynet-src/skynet_server.h @@ -6,6 +6,7 @@ struct skynet_context; struct skynet_message; +struct skynet_monitor; struct skynet_context * skynet_context_new(const char * name, const char * parm); void skynet_context_grab(struct skynet_context *); @@ -16,6 +17,6 @@ void skynet_context_init(struct skynet_context *, uint32_t handle); int skynet_context_push(uint32_t handle, struct skynet_message *message); void skynet_context_send(struct skynet_context * context, void * msg, size_t sz, uint32_t source, int type, int session); int skynet_context_newsession(struct skynet_context *); -int skynet_context_message_dispatch(void); // return 1 when block +int skynet_context_message_dispatch(struct skynet_monitor *); // return 1 when block #endif diff --git a/skynet-src/skynet_start.c b/skynet-src/skynet_start.c index d0a6e1a6..20a03caa 100644 --- a/skynet-src/skynet_start.c +++ b/skynet-src/skynet_start.c @@ -6,11 +6,33 @@ #include "skynet_timer.h" #include "skynet_harbor.h" #include "skynet_group.h" +#include "skynet_monitor.h" #include #include #include #include +#include + +struct monitor { + int count; + struct skynet_monitor ** m; +}; + +static void * +_monitor(void *p) { + struct monitor * m = p; + int i; + int n = m->count; + for (;;) { + for (i=0;im[i]); + } + sleep(5); + } + + return NULL; +} static void * _timer(void *p) { @@ -23,8 +45,9 @@ _timer(void *p) { static void * _worker(void *p) { + struct skynet_monitor *sm = p; for (;;) { - if (skynet_context_message_dispatch()) { + if (skynet_context_message_dispatch(sm)) { usleep(1000); } } @@ -33,19 +56,31 @@ _worker(void *p) { static void _start(int thread) { - pthread_t pid[thread+1]; + pthread_t pid[thread+2]; + + struct monitor m; + m.count = thread; + m.m = malloc(thread * sizeof(struct skynet_monitor *)); + int i; + for (i=0;i