Skynet源码之：监视器(10)

// config->thread 又是通过skynet_main.c 中读取启动配置获得的
// thread == config->thread
// 因此 thread 就是启动配置的数值

struct monitor *m = skynet_malloc(sizeof(*m)); // 申请内存
memset(m, 0, sizeof(*m)); // 把内存都置为0
m->count = thread; // 把开启的worker线程数记录在count
m->sleep = 0; // 初始值为0

m->m = skynet_malloc(thread * sizeof(struct skynet_monitor *)); // 很显然，老做法了
// 申请一段内存，实际就是一个数组，数组大小为需要监控的线程数量
// 数组的元素是什么呢？是指针
// 指针的类型是什么呢？指向 struct skynet_monitor 结构体的地址
int i;
for (i=0;i<thread;i++) {
	m->m[i] = skynet_monitor_new(); // 最后为每个数组分配一个 指向struct skynet_monitor 结构体的指针
}

struct monitor {
	int count;// 需要监控线程的数量
	struct skynet_monitor ** m; // 指向 存放struct skynet_monitor *指针 地址的指针，属于二级指针
    // 即m指向的地址，里面存放的是一个指针数组的开头
    // 而数组元素里面存放的指针，才实际指向 struct skynet_monitor 的地址
	pthread_cond_t cond;
	pthread_mutex_t mutex;
	int sleep;
	int quit;
};

struct skynet_monitor * 
skynet_monitor_new() {
	struct skynet_monitor * ret = skynet_malloc(sizeof(*ret)); // 申请一块内存
	memset(ret, 0, sizeof(*ret)); // 把所有值都设为 0 
	return ret;
}

struct skynet_monitor {
	ATOM_INT version; // 基础版本，0
	int check_version; // 检查版本，0
	uint32_t source; // 值为0
	uint32_t destination; // 值为0
};

void 
skynet_monitor_delete(struct skynet_monitor *sm) {
	skynet_free(sm); // free掉监控的线程所属的skynet_monitor就行了
}

void 
skynet_monitor_trigger(struct skynet_monitor *sm, uint32_t source, uint32_t destination) {
	sm->source = source; // 记录消息的源地址
	sm->destination = destination; // 记录消息的目的地址
	ATOM_FINC(&sm->version);
}

struct message_queue * 
skynet_context_message_dispatch(struct skynet_monitor *sm, struct message_queue *q, int weight) {
	
	// .....代码省略
	
	for (i=0;i<n;i++) {
		if (skynet_mq_pop(q,&msg)) {
			skynet_context_release(ctx);
			return skynet_globalmq_pop();
		} else if (i==0 && weight >= 0) {
			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);
		}
		
        // 注意看这段代码：在消息处理前执行
		// 把 source 置为消息的source，即消息的源地址服务
        // 把 destination 置为消息的目的地址，即本服务
        // 并且把 sm->version 基础版本的值+1
		skynet_monitor_trigger(sm, msg.source, handle);

        // 此时时间进入 消息处理时段
		if (ctx->cb == NULL) {
			skynet_free(msg.data);
		} else {
			dispatch_message(ctx, &msg);
		}
        // 注意看这段代码：在消息处理后执行
        // 而skynet_monitor_trigger(sm, 0, 0)的作用是
        // 把 source 置为0，destination置为0
        // 并且把 sm->version 基础版本的值+1
        // 为什么要这么做呢？接着往下看
		skynet_monitor_trigger(sm, 0, 0);
	}
    
    // .....代码省略
	
	return q;

}

// 调用接口
struct monitor *m;
create_thread(&pid[0], thread_monitor, m);

static void *
thread_monitor(void *p) {
	struct monitor * m = p;
	int i;
	int n = m->count; // 获得需要监控的线程数量
	skynet_initthread(THREAD_MONITOR);
	for (;;) { // 死循环检测
		CHECK_ABORT
		for (i=0;i<n;i++) {
			skynet_monitor_check(m->m[i]); // 调用检测函数，检查worker线程是否陷入死循环
		}
        // 这里等待5秒，也就是5秒之后再检查worker线程
        // 为什么不是直接 sleep(5) ？
        // CHECK_ABORT 又是什么？  
        // #define CHECK_ABORT if (skynet_context_total()==0) break; 即
        // if (skynet_context_total()==0) 
        // 	{
        //		break;
    	//	}
        // 也就是说，没有服务的时候，monitor就不用检测了，可以退出
        // 因此需要for来循环5秒，这样万一(skynet_context_total()==0)才能及时退出
        // 如果直接sleep(5)，会导致5秒之后才退出
		for (i=0;i<5;i++) {
			CHECK_ABORT
			sleep(1);
		}
	}

	return NULL;
}

void 
skynet_monitor_check(struct skynet_monitor *sm) {
	if (sm->version == sm->check_version) {// 版本没更新，出现问题了
		if (sm->destination) {
			skynet_context_endless(sm->destination); // 发现服务进入死循环，调用 服务的退出逻辑，具体代码看 服务模块
			skynet_error(NULL, "A message from [ :%08x ] to [ :%08x ] maybe in an endless loop (version = %d)",
                         sm->source , sm->destination, sm->version);
            // 特别注意看：A message from [ :%08x ] to [ :%08x ]
            // 如果真的打出这句话，这意味着第二个的 [ :%08x ] 服务才是出问题的服务
            // 第一个 [ :%08x ] 只是发出请求消息的服务
		}
	} else {
		sm->check_version = sm->version;
	}
}