探讨erlang消息选择性接收和改进

讨论erlang消息选择性接收

When there is no way to match a given message, it is put in a save queue and the next message is tried. If the second message matches, the first message is put back on top of the mailbox to be retried later.

就是说，erlang消息匹配不上，就会把消息放到 Save Queue 的队列中，当匹配到了后再把消息放回进程信箱。以上是形象化的说法，如果是这样就必然存在消息复制成本，那VM到底是不是这样实现呢？

/* 
 * beam_emu.c process_main() 线程入口函数，实现VM调度 
 * 以下截取 i_loop_rec_fr 处理过程
 * 作用是从信箱取出一条消息放到 x(0) 寄存器；没消息则跳到 wait或者 wait_timeout指令
 */ 
 OpCase(i_loop_rec_fr):
 {
     BeamInstr *next;
     ErlMessage* msgp;

 loop_rec__:

     PROCESS_MAIN_CHK_LOCKS(c_p);

     // 取出“当前位置”的消息
     msgp = PEEK_MESSAGE(c_p);

     if (!msgp) { //如果消息不存在，尝试从SMP下public queue获取消息
#ifdef ERTS_SMP
	 erts_smp_proc_lock(c_p, ERTS_PROC_LOCKS_MSG_RECEIVE);
	 if (ERTS_PROC_PENDING_EXIT(c_p)) {
             // 如果进程准备退出，则不处理消息了
	     erts_smp_proc_unlock(c_p, ERTS_PROC_LOCKS_MSG_RECEIVE);
	     SWAPOUT;
	     goto do_schedule;  // 等待下一次调度
	 }
	 // SMP下把消息移到进程私有堆尾部（纯指针操作）
	 ERTS_SMP_MSGQ_MV_INQ2PRIVQ(c_p);
	 
	 // 再尝试取出“当前位置”的消息
	 msgp = PEEK_MESSAGE(c_p);
	 if (msgp)
	     erts_smp_proc_unlock(c_p, ERTS_PROC_LOCKS_MSG_RECEIVE);
	 else
#endif
	 {
             // 信箱没消息则跳到 wait或者 wait_timeout指令（实际上就是执行下一条执行）
	     SET_I((BeamInstr *) Arg(0));
	     Goto(*I);
	 }
     }
     // 解析分布式消息，把消息附加的数据复制到进程私有堆
     ErtsMoveMsgAttachmentIntoProc(msgp, c_p, E, HTOP, FCALLS,
				   {
				       SWAPOUT;
				       reg[0] = r(0);
				       PROCESS_MAIN_CHK_LOCKS(c_p);
				   },
				   {
				       ERTS_VERIFY_UNUSED_TEMP_ALLOC(c_p);
				       PROCESS_MAIN_CHK_LOCKS(c_p);
				       r(0) = reg[0];
				       SWAPIN;
				   });
     if (is_non_value(ERL_MESSAGE_TERM(msgp))) {
	 /*
	  * 如果消息损坏就移除（出现这种情况是分布式消息解码出现错误）
	  */
	 ASSERT(!msgp->data.attached);
	 UNLINK_MESSAGE(c_p, msgp); // 移除消息，侧重将“当前”位置指向下一条消息
	 free_message(msgp); // 销毁消息
	 goto loop_rec__; // 跳到上面继续
     }
     PreFetch(1, next); // 标记下一条指令位置
     r(0) = ERL_MESSAGE_TERM(msgp);
     NextPF(1, next);   // 执行下一条指令
 }

/* Get "current" message */
#define PEEK_MESSAGE(p)  (*(p)->msg.save)

#define UNLINK_MESSAGE(p,msgp) do {      ErlMessage* __mp = (msgp)->next;      *(p)->msg.save = __mp;      (p)->msg.len--;      if (__mp == NULL)          (p)->msg.last = (p)->msg.save;      (p)->msg.mark = 0; } while(0)

/* 
 * beam_emu.c process_main() 线程入口函数，实现VM调度 
 * 以下截取 remove_message  处理过程（已删除不必要的代码）
 * 作用是将消息从信箱队列中移除
 */ 
 OpCase(remove_message): {
     BeamInstr *next;
     ErlMessage* msgp;

     PROCESS_MAIN_CHK_LOCKS(c_p);

     PreFetch(0, next);
     msgp = PEEK_MESSAGE(c_p); // 取出当前的消息

     if (ERTS_PROC_GET_SAVED_CALLS_BUF(c_p)) {
	 save_calls(c_p, &exp_receive);
     }
     if (ERL_MESSAGE_TOKEN(msgp) == NIL) {
	     SEQ_TRACE_TOKEN(c_p) = NIL;
     } else if (ERL_MESSAGE_TOKEN(msgp) != am_undefined) {
	     // 追踪调试内容，可以忽略
	     Eterm msg;
	     SEQ_TRACE_TOKEN(c_p) = ERL_MESSAGE_TOKEN(msgp);
	     c_p->seq_trace_lastcnt = unsigned_val(SEQ_TRACE_TOKEN_SERIAL(c_p));
	     if (c_p->seq_trace_clock < unsigned_val(SEQ_TRACE_TOKEN_SERIAL(c_p))) {
		 c_p->seq_trace_clock = unsigned_val(SEQ_TRACE_TOKEN_SERIAL(c_p));
	     }
	     msg = ERL_MESSAGE_TERM(msgp);
	     seq_trace_output(SEQ_TRACE_TOKEN(c_p), msg, SEQ_TRACE_RECEIVE, 
			      c_p->common.id, c_p);
     }
     UNLINK_MESSAGE(c_p, msgp); // 移除消息，侧重队列长度-1
     JOIN_MESSAGE(c_p); // 重置“当前”位置，指向了队列第一条消息
     CANCEL_TIMER(c_p);
     free_message(msgp); // 销毁消息

     ERTS_VERIFY_UNUSED_TEMP_ALLOC(c_p);
     PROCESS_MAIN_CHK_LOCKS(c_p);

     NextPF(0, next); // 执行下一条指令
 }

/* Reset message save point (after receive match) */
#define JOIN_MESSAGE(p)      (p)->msg.save = &(p)->msg.first

/* 
 * beam_emu.c process_main() 线程入口函数，实现VM调度 
 * 以下截取 loop_rec_end_f   处理过程
 * 作用是继续取出最新的消息匹配
 */ 
 /*
 * Advance the save pointer to the next message (the current
 * message didn‘t match), then jump to the loop_rec instruction.
 */
 OpCase(loop_rec_end_f): {
     SET_I((BeamInstr *) Arg(0));
     SAVE_MESSAGE(c_p);  // “当前”位置指向下一个位置
     goto loop_rec__;  // 继续取出消息匹配
 }

/* Save current message */
#define SAVE_MESSAGE(p)      (p)->msg.save = &(*(p)->msg.save)->next