`
cwqcwk1
  • 浏览: 87401 次
文章分类
社区版块
存档分类
最新评论

erlang send剖析及参数意义

 
阅读更多
erlang send是一个很基础的消息发送函数,用于进程把一个消息发给另外一个进程。这个函数可以同时用于本地节点进程通信,或者和远程节点进程之间的通信。

前言

最近有同事遇到erlang:send导致消息堆积问题,这个引起了我的强烈关注。我也看了这块的代码,这里简单做个分享。

函数原型:
erlang:send(Dest, Msg, Options) -> Res
Options可以是以下2个:
nosuspend
If the sender would have to be suspended to do the send, nosuspend is returned instead.

noconnect
If the destination node would have to be auto-connected before doing the send, noconnect is returned instead.

字面上意思是说:
nosuspend 遇到会挂起进程的场合不挂起进程,直接返回nosuspend
noconnect 遇到远程节点没有连接时不会自动连接发送消息,直接返回noconnect

但是,erlang直接在文档说慎用nosuspend
Warning

As witherlang:send_nosuspend/2,3: Use with extreme care!

到底为什么这么说,而且返回nosuspend 时这个消息是发送出去了,还是没发送出去?

源码剖析

看了erlang代码,erlang:send是bif实现的,这里以R16B02做说明:
/**
 *   bif.c send_3()函数,实现erlang:send/3
 */
BIF_RETTYPE send_3(BIF_ALIST_3)
{
    Eterm ref;
    Process *p = BIF_P;
    Eterm to = BIF_ARG_1;
    Eterm msg = BIF_ARG_2;
    Eterm opts = BIF_ARG_3;

    int connect = !0; // 初始值设1,表示非0值
    int suspend = !0; // 同上
    Eterm l = opts;
    Sint result;
    
    while (is_list(l)) { //遍历参数列表
	if (CAR(list_val(l)) == am_noconnect) {
	    connect = 0; // 参数带 noconnect,则 connect 取值0
	} else if (CAR(list_val(l)) == am_nosuspend) {
	    suspend = 0; // 同上
	} else {
	    BIF_ERROR(p, BADARG);
	}
	l = CDR(list_val(l));
    }
    if(!is_nil(l)) {
	BIF_ERROR(p, BADARG);
    }

    // 调用 do_send 发送消息;result 大于0表示本次消息发送要扣除的reds,其他则表示错误码
    result = do_send(p, to, msg, suspend, &ref); 
    if (result > 0) {
	ERTS_VBUMP_REDS(p, result); // 扣除本次消息发送的 reds
	if (ERTS_IS_PROC_OUT_OF_REDS(p))
	    goto yield_return;
	BIF_RET(am_ok);
    }

    switch (result) {
    case 0:
	/* May need to yield even though we do not bump reds here... */
	if (ERTS_IS_PROC_OUT_OF_REDS(p))
	    goto yield_return;
	BIF_RET(am_ok); 
	break;

    // 遇到 SEND_TRAP 错误
    case SEND_TRAP: 
	if (connect) {
            // connect 不等于 0
	    BIF_TRAP3(dsend3_trap, p, to, msg, opts); 
	} else {
            // connect 等于 0,直接返回 noconnect
	    BIF_RET(am_noconnect);
	}
	break;
    // 遇到 SEND_YIELD 错误
    case SEND_YIELD: 
	if (suspend) {
            // suspend 不等于 0
	    ERTS_BIF_YIELD3(bif_export[BIF_send_3], p, to, msg, opts);
	} else {
            // suspend 等于 0,直接返回 nosuspend
	    BIF_RET(am_nosuspend);
	}
	break;

    // 遇到 SEND_YIELD_RETURN 错误
    case SEND_YIELD_RETURN: 
        // suspend 等于 0,直接返回 nosuspend
	if (!suspend)
	    BIF_RET(am_nosuspend);
    yield_return:
	ERTS_BIF_YIELD_RETURN(p, am_ok);
    case SEND_AWAIT_RESULT:
	ASSERT(is_internal_ref(ref));
	BIF_TRAP3(await_port_send_result_trap, p, ref, am_nosuspend, am_ok);
    case SEND_BADARG:
	BIF_ERROR(p, BADARG); 
	break;
    case SEND_USER_ERROR:
	BIF_ERROR(p, EXC_ERROR); 
	break;
    case SEND_INTERNAL_ERROR:
	BIF_ERROR(p, EXC_INTERNAL_ERROR);
	break;
    default:
	ASSERT(! "Illegal send result"); 
	break;
    }
    ASSERT(! "Can not arrive here");
    BIF_ERROR(p, BADARG);
}

再来看看 do_send() 函数:
/*
 * bif.c do_send()函数,实现发送到其他进程,或端口,或远程进程
 * 返回消息发送的reds,或错误码
 */
//以下几个是 do_send 可能返回的错误码
#define SEND_TRAP		(-1)
#define SEND_YIELD		(-2)
#define SEND_YIELD_RETURN	(-3)
#define SEND_BADARG		(-4)
#define SEND_USER_ERROR		(-5)
#define SEND_INTERNAL_ERROR	(-6)
#define SEND_AWAIT_RESULT	(-7)

Sint
do_send(Process *p, Eterm to, Eterm msg, int suspend, Eterm *refp) {
    Eterm portid;
    Port *pt;
    Process* rp;
    DistEntry *dep;
    Eterm* tp;

	// 如果目标进程在本地节点
    if (is_internal_pid(to)) {
	if (IS_TRACED(p))
	    trace_send(p, to, msg);
	if (ERTS_PROC_GET_SAVED_CALLS_BUF(p))
	    save_calls(p, &exp_send);

	rp = erts_proc_lookup_raw(to);	
	if (!rp)
	    return 0;
	// 找到这个进程则执行最后的 send_message

    } else if (is_external_pid(to)) {
	// 如果目标进程在远程节点
	dep = external_pid_dist_entry(to);
	if(dep == erts_this_dist_entry) {
	    erts_dsprintf_buf_t *dsbufp = erts_create_logger_dsbuf();
	    erts_dsprintf(dsbufp,
			  "Discarding message %T from %T to %T in an old "
			  "incarnation (%d) of this node (%d)\n",
			  msg,
			  p->common.id,
			  to,
			  external_pid_creation(to),
			  erts_this_node->creation);
	    erts_send_error_to_logger(p->group_leader, dsbufp);
	    return 0;
	}
	// 远程消息调用 remote_send 发送
	return remote_send(p, dep, to, to, msg, suspend);
    } else if (is_atom(to)) {
	// 如果传参是原子,尝试从进程表找到这个进程
	Eterm id = erts_whereis_name_to_id(p, to);

	rp = erts_proc_lookup(id);
	if (rp)
	    goto send_message;

	// 找不到这个进程,检查目标是不是端口
	pt = erts_port_lookup(id, ERTS_PORT_SFLGS_INVALID_LOOKUP);
	if (pt) {
	    portid = id;
	    goto port_common;
	}

	if (IS_TRACED(p))
	    trace_send(p, to, msg);
	if (ERTS_PROC_GET_SAVED_CALLS_BUF(p))
	    save_calls(p, &exp_send);
	
	return SEND_BADARG;
    } else if (is_external_port(to)
	       && (external_port_dist_entry(to)
		   == erts_this_dist_entry)) {
	erts_dsprintf_buf_t *dsbufp = erts_create_logger_dsbuf();
	erts_dsprintf(dsbufp,
		      "Discarding message %T from %T to %T in an old "
		      "incarnation (%d) of this node (%d)\n",
		      msg,
		      p->common.id,
		      to,
		      external_port_creation(to),
		      erts_this_node->creation);
	erts_send_error_to_logger(p->group_leader, dsbufp);
	return 0;
    } else if (is_internal_port(to)) {
	// 如果是本地端口
	int ret_val;
	portid = to;

	pt = erts_port_lookup(portid, ERTS_PORT_SFLGS_INVALID_LOOKUP);

      port_common:
	ret_val = 0;
        
	if (pt) {
	    int ps_flags = suspend ? 0 : ERTS_PORT_SIG_FLG_NOSUSPEND;
	    *refp = NIL;

		// 执行端口操作
	    switch (erts_port_command(p, ps_flags, pt, msg, refp)) {
	    case ERTS_PORT_OP_CALLER_EXIT:
		/* We are exiting... */
		return SEND_USER_ERROR;
	    case ERTS_PORT_OP_BUSY:
		/* Nothing has been sent */
		if (suspend)
		    erts_suspend(p, ERTS_PROC_LOCK_MAIN, pt);
		// 如果 nosuspend 返回 SEND_YIELD,消息还没发送
		return SEND_YIELD;
	    case ERTS_PORT_OP_BUSY_SCHEDULED:
		/* Message was sent */
		if (suspend) {
		    erts_suspend(p, ERTS_PROC_LOCK_MAIN, pt);
		    ret_val = SEND_YIELD_RETURN;
		    break;
		}
		// 这里没有break,如果 nosuspend 将执行下一步操作,消息已发送
		/* Fall through */
	    case ERTS_PORT_OP_SCHEDULED:
		if (is_not_nil(*refp)) {
		    ASSERT(is_internal_ref(*refp));
		    ret_val = SEND_AWAIT_RESULT;
		}
		break;
	    case ERTS_PORT_OP_DROPPED:
	    case ERTS_PORT_OP_BADARG:
	    case ERTS_PORT_OP_DONE:
		break;
	    default:
		ERTS_INTERNAL_ERROR("Unexpected erts_port_command() result");
		break;
	    }
	}
	
	if (IS_TRACED(p)) 	/* trace once only !! */
	    trace_send(p, portid, msg);
	if (ERTS_PROC_GET_SAVED_CALLS_BUF(p))
	    save_calls(p, &exp_send);
	
	if (SEQ_TRACE_TOKEN(p) != NIL
#ifdef USE_VM_PROBES
	    && SEQ_TRACE_TOKEN(p) != am_have_dt_utag
#endif
	    ) {
	    seq_trace_update_send(p);
	    seq_trace_output(SEQ_TRACE_TOKEN(p), msg, 
			     SEQ_TRACE_SEND, portid, p);
	}	    
	
	if (ERTS_PROC_IS_EXITING(p)) {
	    KILL_CATCHES(p); /* Must exit */
	    return SEND_USER_ERROR;
	}
	return ret_val;
    } else if (is_tuple(to)) { /* Remote send */
	// 如果to是原子,走到这里只有是发送远程消息的情况了
	int ret;
	tp = tuple_val(to);
	if (*tp != make_arityval(2))
	    return SEND_BADARG;
	if (is_not_atom(tp[1]) || is_not_atom(tp[2]))
	    return SEND_BADARG;
	
	/* sysname_to_connected_dist_entry will return NULL if there
	   is no dist_entry or the dist_entry has no port,
	   but remote_send() will handle that. */

	// 找到 dist_entry 就用本地进程消息或端口形式发送
	dep = erts_sysname_to_connected_dist_entry(tp[2]);

	if (dep == erts_this_dist_entry) {
	    Eterm id;
	    erts_deref_dist_entry(dep);
	    if (IS_TRACED(p))
		trace_send(p, to, msg);
	    if (ERTS_PROC_GET_SAVED_CALLS_BUF(p))
		save_calls(p, &exp_send);

	    id = erts_whereis_name_to_id(p, tp[1]);

	    rp = erts_proc_lookup_raw(id);
	    if (rp)
		goto send_message;
	    pt = erts_port_lookup(id, ERTS_PORT_SFLGS_INVALID_LOOKUP);
	    if (pt) {
		portid = id;
		goto port_common;
	    }
	    return 0;
	}

	// 找不到 dist_entry 就用 remote_send 发送
	ret = remote_send(p, dep, tp[1], to, msg, suspend);
	if (dep)
	    erts_deref_dist_entry(dep);
	return ret;
    } else {
	if (IS_TRACED(p)) /* XXX Is this really neccessary ??? */
	    trace_send(p, to, msg);
	if (ERTS_PROC_GET_SAVED_CALLS_BUF(p))
	    save_calls(p, &exp_send);
	return SEND_BADARG;
    }
    
	// 以下过程是处理本节点进程消息
 send_message: {
	ErtsProcLocks rp_locks = 0;
	Sint res;
#ifdef ERTS_SMP
	if (p == rp)
	    rp_locks |= ERTS_PROC_LOCK_MAIN;
#endif
	/* send to local process */
	res = erts_send_message(p, rp, &rp_locks, msg, 0);
	if (erts_use_sender_punish)
	    res *= 4;
	else
	    res = 0;
	erts_smp_proc_unlock(rp,
			     p == rp
			     ? (rp_locks & ~ERTS_PROC_LOCK_MAIN)
			     : rp_locks);
	return res;
    }
}
再看下本地进程消息处理erts_send_message() 函数:
/*
 * erl_message.c erts_send_message()函数 实现发送本地消息给进程
 */

Sint
erts_send_message(Process* sender,
		  Process* receiver,
		  ErtsProcLocks *receiver_locks,
		  Eterm message,
		  unsigned flags)
{
    Uint msize;
    ErlHeapFragment* bp = NULL;
    Eterm token = NIL;
    Sint res = 0;
#ifdef USE_VM_PROBES
    DTRACE_CHARBUF(sender_name, 64);
    DTRACE_CHARBUF(receiver_name, 64);
    Sint tok_label = 0;
    Sint tok_lastcnt = 0;
    Sint tok_serial = 0;
#endif
    BM_STOP_TIMER(system);
    BM_MESSAGE(message,sender,receiver);
    BM_START_TIMER(send);

 #ifdef USE_VM_PROBES
    *sender_name = *receiver_name = '\0';
   if (DTRACE_ENABLED(message_send)) {
        erts_snprintf(sender_name, sizeof(DTRACE_CHARBUF_NAME(sender_name)),
               "%T", sender->common.id);
        erts_snprintf(receiver_name, sizeof(DTRACE_CHARBUF_NAME(receiver_name)), 
               "%T", receiver->common.id);
    }
#endif
    if (SEQ_TRACE_TOKEN(sender) != NIL && !(flags & ERTS_SND_FLG_NO_SEQ_TRACE)) {
	// 发送进程打了跟踪标记 sequential_trace_token;后面处理进程跟踪过程
	// 见 http://www.erlang.org/doc/man/erlang.html#process_flag-3

    /*
	 * 篇幅有限,这里省略了部分无关代码
	 */

    } else if (sender == receiver) {
	// 进程发送消息给自己
	
	/* 如果进程正在关闭,则丢弃消息 */
#ifdef ERTS_SMP
	ErtsProcLocks need_locks = (~(*receiver_locks)
				    & (ERTS_PROC_LOCK_MSGQ
				       | ERTS_PROC_LOCK_STATUS));
	if (need_locks) {
	    *receiver_locks |= need_locks;
	    if (erts_smp_proc_trylock(receiver, need_locks) == EBUSY) {
		if (need_locks == ERTS_PROC_LOCK_MSGQ) {
		    erts_smp_proc_unlock(receiver, ERTS_PROC_LOCK_STATUS);
		    need_locks = ERTS_PROC_LOCK_MSGQ|ERTS_PROC_LOCK_STATUS;
		}
		erts_smp_proc_lock(receiver, need_locks);
	    }
	}
	if (!ERTS_PROC_PENDING_EXIT(receiver))
#endif
	{
	    ErlMessage* mp = message_alloc();

            DTRACE6(message_send, sender_name, receiver_name,
                    size_object(message), tok_label, tok_lastcnt, tok_serial);
	    mp->data.attached = NULL;
	    ERL_MESSAGE_TERM(mp) = message;
	    ERL_MESSAGE_TOKEN(mp) = NIL;
#ifdef USE_VM_PROBES
	    ERL_MESSAGE_DT_UTAG(mp) = NIL;
#endif
	    mp->next = NULL;

		// SMP下把消息移到进程私有堆尾部(纯指针操作)
	    ERTS_SMP_MSGQ_MV_INQ2PRIVQ(receiver);

		// 把消息追加到消息队列尾部(纯指针操作)
	    LINK_MESSAGE_PRIVQ(receiver, mp);

		// res 取进程消息队列长度
	    res = receiver->msg.len;

	    if (IS_TRACED_FL(receiver, F_TRACE_RECEIVE)) {
		trace_receive(receiver, message);
	    }
	}
        BM_SWAP_TIMER(send,system);
    } else {
	// 进程发送消息给别的进程
	
#ifdef ERTS_SMP
	ErlOffHeap *ohp;
        Eterm *hp;
	erts_aint32_t state;

	BM_SWAP_TIMER(send,size);
	msize = size_object(message);
	BM_SWAP_TIMER(size,send);
	
	// 接收者进程分配一个大小为 msize 的堆空间,用于存放这个消息
	hp = erts_alloc_message_heap_state(msize,
					   &bp,
					   &ohp,
					   receiver,
					   receiver_locks,
					   &state);
	BM_SWAP_TIMER(send,copy);
	
	// 复制消息到接受者进程,返回值 message 有可能是引用二进制 refc binary
	message = copy_struct(message, msize, &hp, ohp);
	BM_MESSAGE_COPIED(msz);
	BM_SWAP_TIMER(copy,send);
        DTRACE6(message_send, sender_name, receiver_name,
                msize, tok_label, tok_lastcnt, tok_serial);

	// res 取接收者进程消息队列长度
	res = queue_message(sender,
			    receiver,
			    receiver_locks,
			    &state,
			    bp,
			    message,
			    token
#ifdef USE_VM_PROBES
			    , NIL
#endif
	    );
        BM_SWAP_TIMER(send,system);
#else
	ErlMessage* mp = message_alloc();
        Eterm *hp;
        BM_SWAP_TIMER(send,size);
	msize = size_object(message);
        BM_SWAP_TIMER(size,send);
	
	// 检查接收者进程内存不足,执行GC
	if (receiver->stop - receiver->htop <= msize) {
            BM_SWAP_TIMER(send,system);
	    erts_garbage_collect(receiver, msize, receiver->arg_reg, receiver->arity);
            BM_SWAP_TIMER(system,send);
	}
	hp = receiver->htop;
	receiver->htop = hp + msize;
        BM_SWAP_TIMER(send,copy);
		
	// 处理引用二进制的数据(修改引用计数)
	message = copy_struct(message, msize, &hp, &receiver->off_heap);
	BM_MESSAGE_COPIED(msize);
        BM_SWAP_TIMER(copy,send);
        DTRACE6(message_send, sender_name, receiver_name,
                (uint32_t)msize, tok_label, tok_lastcnt, tok_serial);
	ERL_MESSAGE_TERM(mp) = message;
	ERL_MESSAGE_TOKEN(mp) = NIL;
#ifdef USE_VM_PROBES
	ERL_MESSAGE_DT_UTAG(mp) = NIL;
#endif
	mp->next = NULL;
	mp->data.attached = NULL;
	LINK_MESSAGE(receiver, mp);
	
	// res 取接收者进程消息队列长度
	res = receiver->msg.len;
	
	/* 将接收者进程添加到调度队列
	 * 接收者进程可能receive消息导致失去调度,在新消息到来时需要将进程加到调度队列。
	 */
	erts_proc_notify_new_message(receiver);

	if (IS_TRACED_FL(receiver, F_TRACE_RECEIVE)) {
	    trace_receive(receiver, message);
	}
        BM_SWAP_TIMER(send,system);
#endif /* #ifndef ERTS_SMP */
    }
   return res;
}
再来看看 remote_send() 函数:
/** 
 *   bif.c remote_send()函数,实现远程消息发送
 *   dist_entry是erlang分布式接口
 */
static Sint remote_send(Process *p, DistEntry *dep,
			Eterm to, Eterm full_to, Eterm msg, int suspend)
{
    Sint res;
    int code;
    ErtsDSigData dsd;

    ASSERT(is_atom(to) || is_external_pid(to));

	//检查dist_entry状态
    code = erts_dsig_prepare(&dsd, dep, p, ERTS_DSP_NO_LOCK, !suspend);
    switch (code) {
    case ERTS_DSIG_PREP_NOT_ALIVE:
    case ERTS_DSIG_PREP_NOT_CONNECTED: // 连接不存在直接返回 SEND_TRAP
	res = SEND_TRAP;
	break;
	
	// suspend 走这里,返回 SEND_YIELD
    case ERTS_DSIG_PREP_WOULD_SUSPEND:
	ASSERT(!suspend);
	res = SEND_YIELD;
	break;
	
	// nosuspend 在dist_entry连接非稳定状态时会走这里,强制把消息压入发送队列,返回 0,即发送完成
    case ERTS_DSIG_PREP_CONNECTED: {

	if (is_atom(to))
	    code = erts_dsig_send_reg_msg(&dsd, to, msg);
	else
	    code = erts_dsig_send_msg(&dsd, to, msg);
	/*
	 * Note that reductions have been bumped on calling
	 * process by erts_dsig_send_reg_msg() or
	 * erts_dsig_send_msg().
	 */
	if (code == ERTS_DSIG_SEND_YIELD)
	    res = SEND_YIELD_RETURN;
	else
	    res = 0;
	break;
    }
    default:
	ASSERT(! "Invalid dsig prepare result");
	res = SEND_INTERNAL_ERROR;
    }

    if (res >= 0) {
	if (IS_TRACED(p))
	    trace_send(p, full_to, msg);
	if (ERTS_PROC_GET_SAVED_CALLS_BUF(p))
	    save_calls(p, &exp_send);
    }

    return res;
}
其中,erts_dsig_send_msg()底层调用了 dsig_send,把消息放到发送队列,再由 port_task 工作线程负责把消息投递到其他节点。这里面篇幅较大,主要是dist.c,erl_port_task.c, erl_process.c,erl_node_tables.c, io.c 这几个模块,以后找时间再讲。


问题讨论

1、erlang:send 不带 nosuspend/ noconnect, 会导致消息堆积?
erlang被挂起时(也就是进程是{status,suspended}),就会导致当前这个进程消息堆积。
30> Pid = spawn(fun() -> receive M -> M end end).
<0.68.0>
31> process_info(Pid,messages).
{messages,[]}
32> erlang:suspend_process(Pid).
true
33> Pid ! hello.
hello
34> process_info(Pid,status).
{status,suspended}
35> process_info(Pid,messages).
{messages,[hello]}

2、为什么erlang文档说慎用 nosuspend ?
因为,当消息无法发送时,原本会导致发送者进程失去调度权,但是这种方法则会以返回值的方式通知调用者端口忙,不会抑制发送者进程。特别是在分布式环境下,当erlang:send 使用了nosuspend时,当端口繁忙堆积了很多消息时,还会强制把消息压入端口发送队列,如果端口一直处于不稳定的状态,就会导致消息不停的堆积,撑爆内存。所以,还是建议使用 noconnect,当连接异常时就会返回 noconnect ,程序这边再做异常处理

3、erlang:send返回nosuspend 时这个消息是发送出去了,还是没发送出去?
当返回nosuspend 时这个消息肯定是没发送出去。(文章这里最开始写错了,所以网上搜到这篇文章都是错的,建议还是少上那些复制网站)


扩展延伸

reductions
可以理解为 Erlang的基本调度计量单位,Erlang VM基于reduction来进行调度,用来保证调度实现的准实时性。进程的 reduction 值越高,得到的调度机会就越多。

trap_send
前面代码提到了SEND_TRAP 错误,也就是这里
    case SEND_TRAP:   
    if (connect) {  
        BIF_TRAP3(dsend3_trap, p, to, msg, opts);   
    } else {  
        BIF_RET(am_noconnect);  
    }  
    break; 
看下BIF_TRAP3的代码,实际是个宏,修改当前进程的'寄存器'信息,关键是设置 freason 为 TRAP
#define BIF_TRAP3(Trap_, p, A0, A1, A2) do {			\
      Eterm* reg = ERTS_PROC_GET_SCHDATA((p))->x_reg_array;	\
      (p)->arity = 3;						\
      reg[0] = (A0);						\
      reg[1] = (A1);						\
      reg[2] = (A2);						\
      (p)->i = (BeamInstr*) ((Trap_)->addressv[erts_active_code_ix()]); \
      (p)->freason = TRAP;					\
      return THE_NON_VALUE;					\
 } while(0)
现在看下这个过程是怎么工作的:
/*
 * beam_emu.c process_main() 线程入口函数,实现VM调度
 * 以下截取 bif 处理过程
 */
OpCase(call_bif_e):
    {
	Eterm (*bf)(Process*, Eterm*, BeamInstr*) = GET_BIF_ADDRESS(Arg(0)); // 根据参数获取bif实际执行函数
	Eterm result;
	BeamInstr *next;

	PRE_BIF_SWAPOUT(c_p);
	c_p->fcalls = FCALLS - 1;
	if (FCALLS <= 0) {
	   save_calls(c_p, (Export *) Arg(0));
	}
	PreFetch(1, next);
	ASSERT(!ERTS_PROC_IS_EXITING(c_p));
	reg[0] = r(0);
	result = (*bf)(c_p, reg, I); // 执行bif函数
	ASSERT(!ERTS_PROC_IS_EXITING(c_p) || is_non_value(result));
	ERTS_VERIFY_UNUSED_TEMP_ALLOC(c_p);
	ERTS_HOLE_CHECK(c_p);
	ERTS_SMP_REQ_PROC_MAIN_LOCK(c_p);
	PROCESS_MAIN_CHK_LOCKS(c_p);
	if (c_p->mbuf || MSO(c_p).overhead >= BIN_VHEAP_SZ(c_p)) {
	    Uint arity = ((Export *)Arg(0))->code[2];
	    result = erts_gc_after_bif_call(c_p, result, reg, arity);
	    E = c_p->stop;
	}
	HTOP = HEAP_TOP(c_p);
	FCALLS = c_p->fcalls;
	if (is_value(result)) {
	    r(0) = result;
	    CHECK_TERM(r(0));
	    NextPF(1, next);
	} else if (c_p->freason == TRAP) { // 当 freason 设置为 TRAP 时
	    SET_CP(c_p, I+2);
	    SET_I(c_p->i);
	    SWAPIN;
	    r(0) = reg[0];
	    Dispatch(); // 到这一步之后就会调用 dsend3_trap 指向的函数,涉及VM实现,这里不多讲了

这里也说下dsend3_trap 指向哪个函数:
/* dist.c erlang分布式上层实现函数
 */
static Export* trap_function(Eterm func, int arity)
{
    return erts_export_put(am_erlang, func, arity); // 从导出函数表获取函数地址
}

void init_dist(void)
{
    init_nodes_monitors();

    nodedown.reason = NIL;
    nodedown.bp = NULL;

    erts_smp_atomic_init_nob(&no_nodes, 0);
    erts_smp_atomic_init_nob(&no_caches, 0);

    /* Lookup/Install all references to trap functions */
    dsend2_trap = trap_function(am_dsend,2);
    dsend3_trap = trap_function(am_dsend,3);    // dsend3_trap 指向的函数就是 erlang:dsend/3
    /*    dsend_nosuspend_trap = trap_function(am_dsend_nosuspend,2);*/
    dlink_trap = trap_function(am_dlink,1);
    dunlink_trap = trap_function(am_dunlink,1);
    dmonitor_node_trap = trap_function(am_dmonitor_node,3);
    dgroup_leader_trap = trap_function(am_dgroup_leader,2);
    dexit_trap = trap_function(am_dexit, 2);
    dmonitor_p_trap = trap_function(am_dmonitor_p, 2);
}
为什么erlang要大费周章搞trap?erlang:send() bif化后又执行非bif函数(erlang:dsend/3)
erlang:send/3 只有在节点连接失败的情况下才会执行erlang:dsend/3,这个时候,当前进程就会失去CPU调度权,放到了下一次调度去执行这个函数。而且,这一来一回就扣除了不少reductions

时间不早了,有点困了。找时间再研究下reductions 的实际影响,怎么用来实现软实时

2015/1/25 修正erlang:send 返回nosuspend时消息没发送出去
2015/5/22 修正erts_proc_notify_new_message 的注释说明
参考:http://blog.csdn.net/mycwq/article/details/42845385

分享到:
评论

相关推荐

    erlang 程序设计 源码

    在分析Erlang源码时,重点关注以下方面: - **进程间的交互**:查看进程是如何创建、通信和结束的,理解消息传递的工作原理。 - **模块和函数**:理解模块的功能划分,以及函数如何协同工作完成任务。 - **错误处理...

    erlang的timer和实现机制

    `timer:apply_after/4`与`send_after/3,4`类似,但不是发送消息,而是应用指定的函数和参数。这可以用来在将来某个时刻执行一个特定的操作,无需手动创建和管理进程。 `timer:cancel_timer/1`接收一个定时器引用,...

    erlang server源码

    Erlang Server源码分析与详解 Erlang是一种面向并发的、函数式编程语言,以其在分布式系统、高可用性和容错性方面的优势而受到广泛关注。尤其在构建大规模聊天室服务器这样的实时通信系统中,Erlang的性能表现突出...

    erlang 入门练习

    通过分析`client.erl`,我们可以逐步了解Erlang编程的核心概念,并进一步深入学习这个强大的并发编程语言。在实际项目中,Erlang常用于构建大规模、高并发的系统,如网络服务器、分布式系统等。

    erlang程序设计(第二版)随书源码

    通过阅读和分析这些代码,我们可以了解Erlang的语法特性,如模式匹配、函数式编程思想、进程通信等核心概念。Erlang的并发模型基于轻量级进程(Lightweight Processes, LPs),这些进程之间的通信主要依靠消息传递,...

    erlang学习4.pdf

    本篇文档旨在深入解析Erlang的分布式调用机制及分布式TableServer的相关实现细节,帮助读者更全面地理解Erlang在分布式环境中的应用。 #### Erlang的分布式调用支持 在Erlang中,分布式处理是通过节点间的通信来...

    erlang学习3pdf

    ### Erlang学习:支持多个连接的TableServer及进程通信详解 #### 一、引言 在Erlang的学习过程中,理解和掌握如何构建能够处理多个连接的服务器是非常重要的一步。本篇文档将详细介绍如何通过Erlang语言实现一个...

    编写分布式的Erlang程序_陷阱和对策(中文版)

    熟悉Erlang的性能分析工具,如`instrument`模块和`timer`模块,可以帮助优化代码。 在《编写分布式的Erlang程序_陷阱和对策(中文版)_p37-svensson-cn.pdf》这本书中,可能会详细讨论上述的一些概念,并提供具体的...

    erlang_standard_snippets-源码.rar

    Erlang是一种面向并发、分布式计算的编程语言,由瑞典电信设备制造商Ericsson开发,主要用于构建高可用性、...通过分析和学习这个源码库,你可以深入了解Erlang语言的特性,提高在并发编程和分布式系统设计方面的技能。

    tcp_client:一个Erlang TCP客户端连接管理器

    通过理解以上知识点,你可以更好地理解和使用Erlang中的tcp_client模块,无论是创建自己的TCP客户端连接管理器,还是分析已有的实现。对于Erlang开发者来说,熟练掌握TCP客户端的实现和管理是提升系统性能和稳定性的...

    parallelAlgorithms:Erlang 中的并行算法

    **Erlang 并行算法概述** Erlang 是一种函数式编程语言,以其在并发处理和分布式系统方面的强大功能而闻名。Erlang 的设计初衷是为了构建容错性高、可扩展性强的实时系统。在 Erlang 中,实现并行算法主要依赖于其...

    clojang:用于ErlangOTP通信的Clojure API(基于jiface构建)

    然后,你可以使用`send`或`call`方法来发送消息给Erlang进程,并接收返回的结果。clojang也支持Erlang的模式匹配和错误处理,使得Clojure开发者能够充分利用Erlang的强项。 在标签中提到了“java”、“distributed-...

    e语言几个简单多线程写法

    **标题与描述解析** 标题"e语言几个简单多线程写法"表明本文将探讨如何在e语言(可能指的是Erlang或者一种类似的编程语言)中实现多线程编程...同时,阅读和分析开源Erlang项目中的多线程实现也是提升技能的有效方法。

    messager.zip

    《基于Erlang的messager聊天程序初探》 ...通过对该项目的分析和研究,开发者不仅能掌握Erlang的基础语法,还能理解其在聊天应用开发中的核心理念,从而为进一步构建复杂的分布式系统打下坚实基础。

    修正Socket Base Distribution一节实例代码的Bug

    在IT行业中,网络编程是不可或缺的一部分,...通过细心的分析和测试,我们可以确保这些Erlang Socket程序运行稳定,提供可靠的服务。在实际项目中,还应考虑性能优化、安全性以及扩展性等因素,以适应不断变化的需求。

    自己写一个tcp 通用服务器

    通过理解和分析这些源码,你可以学习到Erlang中如何编写TCP服务器和客户端,以及如何处理并发连接和数据通信。这对于提升网络编程技能,尤其是对Erlang环境下的并发编程有极大的帮助。你可以从这些代码中学习到如何...

    RabbitMq安装详细步骤

    本文将详细介绍 RabbitMQ 的安装步骤及常见问题解决办法,包括安装失败的原因分析、解决 VS 无法正常安装的问题、RabbitMQ 启动后服务自动关闭的 bug 修复等内容。 #### 二、安装准备 在安装 RabbitMQ 之前,需要...

    tsung_user_guide

    - **客户端配置**:指定参与测试的客户端数量及分布。 - **服务端配置**:设置被测服务的相关信息。 ##### 6.3 监控 - **性能监控**:设置监控项以收集性能数据。 - **资源监控**:监控CPU、内存等系统资源的使用...

    elang 程序设计 第二版 源代码

    源代码中会包含大量进程间的通信示例,如使用`send`和`receive`操作进行同步,以及如何利用`spawn`创建新进程。 3. **模式匹配和递归**:Elang的模式匹配允许在函数定义中对输入进行解构,简化了数据处理。同时,...

    A Calculus of Mobile Processes Part I and part II

    π-演算的理论基础对后来的编程语言设计产生了深远影响,例如,像Erlang、Scala和OCaml这样的函数式编程语言都受到了π-演算的启发,支持高效并发和分布式计算。此外,π-演算也是形式化验证和模型检查等软件工程...

Global site tag (gtag.js) - Google Analytics