add endless-loop monitor

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
 

skynet-src/skynet_monitor.c

@@ -0,0 +1,42 @@
+#include "skynet_monitor.h"
+#include "skynet.h"
+
+#include <stdlib.h>
+#include <string.h>
+
+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;
+	}
+}

skynet-src/skynet_monitor.h

@@ -0,0 +1,13 @@
+#ifndef SKYNET_MONITOR_H
+#define SKYNET_MONITOR_H
+
+#include <stdint.h>
+
+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

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 <string.h>
 #include <assert.h>
@@ -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;
 }
 

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

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 <pthread.h>
 #include <unistd.h>
 #include <assert.h>
 #include <stdio.h>
+#include <stdlib.h>
+
+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;i<n;i++) {
+			skynet_monitor_check(m->m[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<thread;i++) {
+		m.m[i] = skynet_monitor_new();
+	}
 
 	pthread_create(&pid[0], NULL, _timer, NULL);
+	pthread_create(&pid[1], NULL, _monitor, &m);
 
-	int i;
+	for (i=0;i<thread;i++) {
-
+		pthread_create(&pid[i+2], NULL, _worker, m.m[i]);
-	for (i=1;i<thread+1;i++) {
-		pthread_create(&pid[i], NULL, _worker, NULL);
 	}
 
-	for (i=0;i<thread+1;i++) {
+	for (i=0;i<thread+2;i++) {
 		pthread_join(pid[i], NULL); 
 	}
+
+	for (i=0;i<thread;i++) {
+		skynet_monitor_delete(m.m[i]);
+	}
+	free(m.m);
 }
 
 static int