erlang进程监控的实现原理

Erlang支持Monitor和Link两种监控进程的方式，使得所有进程可以连成一个整体。当某个进程出错退出时，监控进程会收到该进程退出的消息通知。有了这些特点，使用Erlang建立一个简单，并且健壮的系统就不是什么难事。前面有文章分析了两种方式的用法，这里分析下monitor和link的实现。

源码分析

monitor 和link实现有点类似，下面以monitor为例做说明（erlang版本R16B02）

erlang:monitor/2的实现

// bif.c 实现 erlang:monitor/2
BIF_RETTYPE monitor_2(BIF_ALIST_2)
{
    Eterm target = BIF_ARG_2;
    BIF_RETTYPE ret;
    DistEntry  *dep = NULL; 
    int deref_de = 0;

    /* 目前只支持 erlang:monitor(process, Target) */
    if (BIF_ARG_1 != am_process) {
	goto error;
    }

    if (is_internal_pid(target)) { // 如果是本节点进程
    local_pid:
	ret = local_pid_monitor(BIF_P, target); // 处理本节点进程
    } else if (is_external_pid(target)) { // 如果是其他节点进程
	dep = external_pid_dist_entry(target);
	if (dep == erts_this_dist_entry) // 如果进程归属于本节点，跳到本节点进程处理
	    goto local_pid;
	ret = remote_monitor(BIF_P, BIF_ARG_1, BIF_ARG_2, dep, target, 0); // 处理其他节点进程
    } else if (is_atom(target)) { // Target是atom处理
	ret = local_name_monitor(BIF_P, target);
    } else if (is_tuple(target)) { // Target是tuple处理
	Eterm *tp = tuple_val(target);
	Eterm remote_node;
	Eterm name;
	if (arityval(*tp) != 2) 
	    goto error;
	remote_node = tp[2];
	name = tp[1];
	if (!is_atom(remote_node) || !is_atom(name)) {
	    goto error;
	}
	if (!erts_is_alive && remote_node != am_Noname) {
	    goto error; /* Remote monitor from (this) undistributed node */
	}
	dep = erts_sysname_to_connected_dist_entry(remote_node);
	if (dep == erts_this_dist_entry) {
	    deref_de = 1;
	    ret = local_name_monitor(BIF_P, name);
	} else {
	    if (dep)
		deref_de = 1;
	    ret = remote_monitor(BIF_P, BIF_ARG_1, BIF_ARG_2, dep, name, 1);
	}
    } else {
    error:
	ERTS_BIF_PREP_ERROR(ret, BIF_P, BADARG);
    }
    if (deref_de) {
	deref_de = 0;
	erts_deref_dist_entry(dep);
    }

    return ret;
}

现在，看下本节点进程的监控处理：

// bif.c 实现本地节点进程监控处理
static BIF_RETTYPE local_pid_monitor(Process *p, Eterm target)
{
    BIF_RETTYPE ret;
    Eterm mon_ref;
    Process *rp;
    ErtsProcLocks p_locks = ERTS_PROC_LOCK_MAIN|ERTS_PROC_LOCK_LINK;

    mon_ref = erts_make_ref(p);
    ERTS_BIF_PREP_RET(ret, mon_ref);
    if (target == p->common.id) { // 如果进程监控自己
	return ret;
    }

    erts_smp_proc_lock(p, ERTS_PROC_LOCK_LINK); // 锁住进程link操作，避免进程监控数据被脏写
    rp = erts_pid2proc_opt(p, p_locks,
			   target, ERTS_PROC_LOCK_LINK,  // 同样是link锁
			   ERTS_P2P_FLG_ALLOW_OTHER_X);
    if (!rp) {
	erts_smp_proc_unlock(p, ERTS_PROC_LOCK_LINK);
	p_locks &= ~ERTS_PROC_LOCK_LINK;
	erts_queue_monitor_message(p, &p_locks,
				   mon_ref, am_process, target, am_noproc);
    }
    else {
	ASSERT(rp != p);

	// 当前进程添加监控数据
	erts_add_monitor(&ERTS_P_MONITORS(p), MON_ORIGIN, mon_ref, target, NIL); 
	// 目标进程添加被监控数据
	erts_add_monitor(&ERTS_P_MONITORS(rp), MON_TARGET, mon_ref, p->common.id, NIL); 

	erts_smp_proc_unlock(rp, ERTS_PROC_LOCK_LINK);
    }

    erts_smp_proc_unlock(p, p_locks & ~ERTS_PROC_LOCK_MAIN);

    return ret;
}

实际上，这里只是修改进程的监控数据，监控者和被监控者两份数据。

来看下erts_add_monitor的实现：

// erl_monitors.c 实现进程增加监控信息
void erts_add_monitor(ErtsMonitor **root, Uint type, Eterm ref, Eterm pid, 
		      Eterm name)
{
    void *tstack[STACK_NEED];
    int tpos = 0;
    int dstack[STACK_NEED+1];
    int dpos = 1;
    int state = 0;
    ErtsMonitor **this = root;
    Sint c;
  
    dstack[0] = DIR_END;
    for (;;) {
	if (!*this) { /* Found our place */
	    state = 1;
	    *this = create_monitor(type,ref,pid,name);
	    break;
	} else if ((c = CMP_MON_REF(ref,(*this)->ref)) < 0) { 
	    /* go left */
	    dstack[dpos++] = DIR_LEFT;
	    tstack[tpos++] = this;
	    this = &((*this)->left);
	} else if (c > 0) { /* go right */
	    dstack[dpos++] = DIR_RIGHT;
	    tstack[tpos++] = this;
	    this = &((*this)->right);
	} else { /* Equal key is an error for monitors */
	    erl_exit(1,"Insertion of already present monitor!");
	    break;
	}
    }
    insertion_rotation(dstack, dpos, tstack, tpos, state);
}

再看下这个宏，取的就是进程结构的监控数据。就是说，每个进程都有一份监控数据，记录了监控和被监控信息，保存为AVL树结构。

#define ERTS_P_MONITORS(P)	((P)->common.u.alive.monitors)

进程监控的处理

前面分析，监控进程只是在被监控进程打个标记，那进程退出时是怎么处理的？

// erl_monitor.c 触发所有monitor（遍历 monitor 数据，执行 doit 函数回调）
void erts_sweep_monitors(ErtsMonitor *root, 
			 void (*doit)(ErtsMonitor *, void *),
			 void *context) 
{
    ErtsMonitor *tstack[STACK_NEED];
    int tpos = 0;
    int dstack[STACK_NEED+1];
    int dpos = 1;
    int dir;
    
    dstack[0] = DIR_END;

    for (;;) {
	if (root == NULL) {
	    if ((dir = dstack[dpos-1]) == DIR_END) {
		return;
	    }
	    if (dir == DIR_LEFT) {
		/* Still has DIR_RIGHT to do */
		dstack[dpos-1] = DIR_RIGHT;
		root = (tstack[tpos-1])->right;
	    } else {
		/* stacktop is an object to be deleted */
		(*doit)(tstack[--tpos],context); // 执行回调
		--dpos;
		root = NULL;
	    }
	} else {
	    dstack[dpos++] = DIR_LEFT;
	    tstack[tpos++] = root;
	    root = root->left;
	}
    }
}

什么时候会触发监控回调？

1.进程关闭

2.分布式端口关闭

以上都会触发监控回调，这里以进程关闭做说明：

// erl_process.c 进程关闭处理（有删节）
void erts_continue_exit_process(Process *p)
{
    //...
    mon = ERTS_P_MONITORS(p);
    lnk = ERTS_P_LINKS(p);

    //...

    if (lnk) { // link的处理
	DeclareTmpHeap(tmp_heap,4,p);
	Eterm exit_tuple;
	Uint exit_tuple_sz;
	Eterm* hp;
	UseTmpHeap(4,p);
	hp = &tmp_heap[0];
	exit_tuple = TUPLE3(hp, am_EXIT, p->common.id, reason);
	exit_tuple_sz = size_object(exit_tuple);
	{
	    ExitLinkContext context = {p, reason, exit_tuple, exit_tuple_sz};
	    erts_sweep_links(lnk, &doit_exit_link, &context);
	}
	UnUseTmpHeap(4,p);
    }

    { // monitor的处理
	ExitMonitorContext context = {reason, p};
	erts_sweep_monitors(mon,&doit_exit_monitor,&context); /* Allocates TmpHeap, but we
								 have none here */
    }

    //...
}

看下以上代码中回调函数的处理

// erl_process.c 进程关闭监控处理
static void doit_exit_monitor(ErtsMonitor *mon, void *vpcontext)
{
    ExitMonitorContext *pcontext = vpcontext;
    DistEntry *dep;
    ErtsMonitor *rmon;
    Process *rp;

    if (mon->type == MON_ORIGIN) { //如果该进程有监控其他进程，删除其他进程的被监控信息
	/* We are monitoring someone else, we need to demonitor that one.. */
	if (is_atom(mon->pid)) { /* remote by name */
	    ASSERT(is_node_name_atom(mon->pid));
	    dep = erts_sysname_to_connected_dist_entry(mon->pid);
	    if (dep) { // 如果该进程监控远程节点的进程
		erts_smp_de_links_lock(dep);
		// 先删除DistEntry的监控信息
		rmon = erts_remove_monitor(&(dep->monitors), mon->ref);
		erts_smp_de_links_unlock(dep);
		if (rmon) { // 然后通知远程节点去掉被监控信息
		    ErtsDSigData dsd;
		    int code = erts_dsig_prepare(&dsd, dep, NULL,
						 ERTS_DSP_NO_LOCK, 0);
		    if (code == ERTS_DSIG_PREP_CONNECTED) {
			code = erts_dsig_send_demonitor(&dsd,
							rmon->pid,
							mon->name,
							mon->ref,
							1);
			ASSERT(code == ERTS_DSIG_SEND_OK);
		    }
		    erts_destroy_monitor(rmon);
		}
		erts_deref_dist_entry(dep);
	    }
	} else {
	    ASSERT(is_pid(mon->pid));
	    if (is_internal_pid(mon->pid)) { // 如果是本节点进程
		rp = erts_pid2proc(NULL, 0, mon->pid, ERTS_PROC_LOCK_LINK);
		if (!rp) {
		    goto done;
		}
		// 删除被监控进程的监控信息
		rmon = erts_remove_monitor(&ERTS_P_MONITORS(rp), mon->ref);
		erts_smp_proc_unlock(rp, ERTS_PROC_LOCK_LINK);
		if (rmon == NULL) {
		    goto done;
		}
		erts_destroy_monitor(rmon);
	    } else { /* remote by pid */
		ASSERT(is_external_pid(mon->pid));
		dep = external_pid_dist_entry(mon->pid);
		ASSERT(dep != NULL);
		if (dep) {
		    erts_smp_de_links_lock(dep);
			// 先删除DistEntry的监控信息
		    rmon = erts_remove_monitor(&(dep->monitors), mon->ref);
		    erts_smp_de_links_unlock(dep);
		    if (rmon) {// 然后通知远程节点去掉被监控信息
			ErtsDSigData dsd;
			int code = erts_dsig_prepare(&dsd, dep, NULL,
						     ERTS_DSP_NO_LOCK, 0);
			if (code == ERTS_DSIG_PREP_CONNECTED) {
			    code = erts_dsig_send_demonitor(&dsd,
							    rmon->pid,
							    mon->pid,
							    mon->ref,
							    1);
			    ASSERT(code == ERTS_DSIG_SEND_OK);
			}
			erts_destroy_monitor(rmon);
		    }
		}
	    }
	}
    } else { //如果有进程监控该进程，则通知监控进程
	ASSERT(mon->type == MON_TARGET);
	ASSERT(is_pid(mon->pid) || is_internal_port(mon->pid));
	if (is_internal_port(mon->pid)) { // 如果监控进程是本节点端口
	    Port *prt = erts_id2port(mon->pid);
	    if (prt == NULL) {
		goto done;
	    }
	    erts_fire_port_monitor(prt, mon->ref);
	    erts_port_release(prt); 
	} else if (is_internal_pid(mon->pid)) { // 如果监控进程是本节点进程
	    Eterm watched;
	    DeclareTmpHeapNoproc(lhp,3);
	    ErtsProcLocks rp_locks = (ERTS_PROC_LOCK_LINK
				      | ERTS_PROC_LOCKS_MSG_SEND);
	    rp = erts_pid2proc(NULL, 0, mon->pid, rp_locks);
	    if (rp == NULL) {
		goto done;
	    }
	    UseTmpHeapNoproc(3);
		// 先把监控进程的监控信息移除掉
	    rmon = erts_remove_monitor(&ERTS_P_MONITORS(rp), mon->ref);
	    if (rmon) {
		erts_destroy_monitor(rmon);
		watched = (is_atom(mon->name)
			   ? TUPLE2(lhp, mon->name, 
				    erts_this_dist_entry->sysname)
			   : pcontext->p->common.id);
		// 然后把进程关闭信息以消息通知监控进程 {'DOWN',Ref,process,Pid,Reason}
		erts_queue_monitor_message(rp, &rp_locks, mon->ref, am_process, 
					   watched, pcontext->reason);
	    }
	    UnUseTmpHeapNoproc(3);
	    /* else: demonitor while we exited, i.e. do nothing... */
	    erts_smp_proc_unlock(rp, rp_locks);
	} else { // 如果监控进程是远程节点进程
	    ASSERT(is_external_pid(mon->pid));    
	    dep = external_pid_dist_entry(mon->pid);
	    ASSERT(dep != NULL);
	    if (dep) {
		erts_smp_de_links_lock(dep);
		// 先删除DistEntry的监控信息
		rmon = erts_remove_monitor(&(dep->monitors), mon->ref);
		erts_smp_de_links_unlock(dep);
		if (rmon) {// 然后通知远程节点该进程退出消息
		    ErtsDSigData dsd;
		    int code = erts_dsig_prepare(&dsd, dep, NULL,
						 ERTS_DSP_NO_LOCK, 0);
		    if (code == ERTS_DSIG_PREP_CONNECTED) {
			code = erts_dsig_send_m_exit(&dsd,
						     mon->pid,
						     (rmon->name != NIL
						      ? rmon->name
						      : rmon->pid),
						     mon->ref,
						     pcontext->reason);
			ASSERT(code == ERTS_DSIG_SEND_OK);
		    }
		    erts_destroy_monitor(rmon);
		}
	    }
	}
    }
 done:
    /* As the monitors are previously removed from the process, 
       distribution operations will not cause monitors to disappear,
       we can safely delete it. */
       
    erts_destroy_monitor(mon);
}

跨节点进程监控的实现

前面说到本节点的处理，那跨节点进程监控是怎么实现的，有什么区别？

// bif.c 跨节点进程监控的处理
static BIF_RETTYPE remote_monitor(Process *p, Eterm bifarg1, Eterm bifarg2,
	       DistEntry *dep, Eterm target, int byname)
{
    ErtsDSigData dsd;
    BIF_RETTYPE ret;
    int code;

    erts_smp_proc_lock(p, ERTS_PROC_LOCK_LINK);
    code = erts_dsig_prepare(&dsd, dep, p, ERTS_DSP_RLOCK, 0); // 获取分布式端口的状态
    switch (code) {
    case ERTS_DSIG_PREP_NOT_ALIVE: // 端口还没激活使用，使用Trap处理
	/* Let the dmonitor_p trap handle it */
    case ERTS_DSIG_PREP_NOT_CONNECTED: // 端口未连接，使用Trap处理
	erts_smp_proc_unlock(p, ERTS_PROC_LOCK_LINK);
	ERTS_BIF_PREP_TRAP2(ret, dmonitor_p_trap, p, bifarg1, bifarg2); // 使用Trap处理，在下次调度时调用erlang:dmonitor_p/2
	break;
    case ERTS_DSIG_PREP_CONNECTED: // 端口已连接，可发送数据
	if (!(dep->flags & DFLAG_DIST_MONITOR)
	    || (byname && !(dep->flags & DFLAG_DIST_MONITOR_NAME))) {
	    erts_smp_de_runlock(dep);
	    erts_smp_proc_unlock(p, ERTS_PROC_LOCK_LINK);
	    ERTS_BIF_PREP_ERROR(ret, p, BADARG);
	}
	else {
	    Eterm p_trgt, p_name, d_name, mon_ref;

	    mon_ref = erts_make_ref(p);

	    if (byname) {
		p_trgt = dep->sysname;
		p_name = target;
		d_name = target;
	    }
	    else {
		p_trgt = target;
		p_name = NIL;
		d_name = NIL;
	    }

	    erts_smp_de_links_lock(dep);
		// 当前进程添加监控数据 
	    erts_add_monitor(&ERTS_P_MONITORS(p), MON_ORIGIN, mon_ref, p_trgt,
			     p_name);
		// DistEntry添加被监控数据 
	    erts_add_monitor(&(dep->monitors), MON_TARGET, mon_ref, p->common.id,
			     d_name);

	    erts_smp_de_links_unlock(dep);
	    erts_smp_de_runlock(dep);
	    erts_smp_proc_unlock(p, ERTS_PROC_LOCK_LINK);

		// 发监控消息到远程节点
	    code = erts_dsig_send_monitor(&dsd, p->common.id, target, mon_ref);
	    if (code == ERTS_DSIG_SEND_YIELD)
		ERTS_BIF_PREP_YIELD_RETURN(ret, p, mon_ref);
	    else
		ERTS_BIF_PREP_RET(ret, mon_ref);
	}
	break;
    default: // 其他端口状态，如端口将被挂起
	ASSERT(! "Invalid dsig prepare result");
	ERTS_BIF_PREP_ERROR(ret, p, EXC_INTERNAL_ERROR);
	break;
    }

    return ret;
}

接着，看下发消息给远程节点的处理。

// dist.c 发监控消息到远程节点
int erts_dsig_send_monitor(ErtsDSigData *dsdp, Eterm watcher, Eterm watched,
		       Eterm ref)
{
    Eterm ctl;
    DeclareTmpHeapNoproc(ctl_heap,5);
    int res;

    UseTmpHeapNoproc(5);
    ctl = TUPLE4(&ctl_heap[0],
		 make_small(DOP_MONITOR_P),
		 watcher, watched, ref);

	// 构造消息{DOP_MONITOR_P, LocalPid, RemotePidOrName, Ref} 发给远程节点
    res = dsig_send(dsdp, ctl, THE_NON_VALUE, 0);
    UnUseTmpHeapNoproc(5);
    return res;
}

看下远程接收到这个消息后的处理。

// dist.c 处理其他节点发来的消息（有删节）
int erts_net_message(Port *prt,
		     DistEntry *dep,
		     byte *hbuf,
		     ErlDrvSizeT hlen,
		     byte *buf,
		     ErlDrvSizeT len)
{
    // ...
    switch (type = unsigned_val(tuple[1])) {
    // ...
	
	// 处理 {DOP_MONITOR_P, Remote pid, local pid or name, ref}
    case DOP_MONITOR_P: {
	/* A remote process wants to monitor us, we get:
	   {DOP_MONITOR_P, Remote pid, local pid or name, ref} */
	Eterm name;
	
	if (tuple_arity != 4) {
	    goto invalid_message;
	}

	watcher = tuple[2];
	watched = tuple[3];  /* local proc to monitor */
	ref     = tuple[4];

	if (is_not_ref(ref)) {
	    goto invalid_message;
	}

	if (is_atom(watched)) {
	    name = watched;
	    rp = erts_whereis_process(NULL, 0,
				      watched, ERTS_PROC_LOCK_LINK,
				      ERTS_P2P_FLG_ALLOW_OTHER_X);
	}
	else {
	    name = NIL;
	    rp = erts_pid2proc_opt(NULL, 0,
				   watched, ERTS_PROC_LOCK_LINK,
				   ERTS_P2P_FLG_ALLOW_OTHER_X);
	}

	if (!rp) { // 如果被监控进程不存在，则回复进程退出消息
	    ErtsDSigData dsd;
	    int code;
	    code = erts_dsig_prepare(&dsd, dep, NULL, ERTS_DSP_NO_LOCK, 0);
	    if (code == ERTS_DSIG_PREP_CONNECTED) {
		code = erts_dsig_send_m_exit(&dsd, watcher, watched, ref,
					     am_noproc);
		ASSERT(code == ERTS_DSIG_SEND_OK);
	    }
	}
	else {
	    if (is_atom(watched))
		watched = rp->common.id;
	    erts_smp_de_links_lock(dep);
		// DistEntry添加监控数据
	    erts_add_monitor(&(dep->monitors), MON_ORIGIN, ref, watched, name);
		// 进程添加被监控数据
	    erts_add_monitor(&ERTS_P_MONITORS(rp), MON_TARGET, ref, watcher, name);
	    erts_smp_de_links_unlock(dep);
	    erts_smp_proc_unlock(rp, ERTS_PROC_LOCK_LINK);
	}

	break;
    }

    //...
}

对比本节点和跨节点的处理

本节点进程监控处理如下：（进程X监控进程Y）

/**********************************************************************
 *        Process X          Process Y
 *       +-------------+    +-------------+
 * Type: | MON_ORIGIN  |    | MON_TARGET  |
 *       +-------------+    +-------------+
 * Pid:  | Pid(Y)      |    | Pid(X)      |
 *       +-------------+    +-------------+
 **********************************************************************/

跨节点的处理：（节点A的进程X监控节点B的进程Y）

/**********************************************************************
 *                    Node A              |           Node B
 *       ---------------------------------+----------------------------------
 *        Process X (@A)   Distentry @A       Distentry @B     Process Y (@B)
 *                         for node B         for node A       
 *       +-------------+  +-------------+    +-------------+  +-------------+
 * Type: | MON_ORIGIN  |  | MON_TARGET  |    | MON_ORIGIN  |  | MON_TARGET  |
 *       +-------------+  +-------------+    +-------------+  +-------------+
 * Pid:  | Atom(node B)|  | Pid(X)      |    | Pid(Y)      |  | Pid(X)      |
 *       +-------------+  +-------------+    +-------------+  +-------------+
 **********************************************************************/

对比就是多了一步DistEntry的处理，这是由跨节点网络的不稳定性决定的。远程进程出现异常，可能是进程挂了，也有可能是节点连接出问题。当远程节点出现异常，就要触发这个节点关联进程的处理。

小结

从上面的分析可以了解，进程监控实际只是在被监控进程打个标记，然后在这个被监控进程出现异常时处理所有监控进程。

参考：http://blog.csdn.net/mycwq/article/details/46961489