1 files changed, 117 insertions, 121 deletions

diff --git a/erts/emulator/beam/erl_message.c b/erts/emulator/beam/erl_message.c
index 919567ab27..c98ca09f3f 100644
--- a/erts/emulator/beam/erl_message.c
+++ b/erts/emulator/beam/erl_message.c
@@ -296,36 +296,6 @@ erts_msg_distext2heap(Process *pp,
     return THE_NON_VALUE;
  }
 
-static ERTS_INLINE void
-notify_new_message(Process *receiver)
-{
-    ERTS_SMP_LC_ASSERT(ERTS_PROC_LOCK_STATUS
-		       & erts_proc_lc_my_proc_locks(receiver));
-
-    switch (receiver->status) {
-    case P_GARBING:
-	switch (receiver->gcstatus) {
-	case P_SUSPENDED:
-	    goto suspended;
-	case P_WAITING:
-	    goto waiting;
-	default:
-	    break;
-	}
-	break;
-    case P_SUSPENDED:
-    suspended:
-	receiver->rstatus = P_RUNABLE;
-	break;
-    case P_WAITING:
-    waiting:
-	erts_add_to_runq(receiver);
-	break;
-    default:
-	break;
-    }
-}
-
 void
 erts_queue_dist_message(Process *rcvr,
 			ErtsProcLocks *rcvr_locks,
@@ -339,7 +309,7 @@ erts_queue_dist_message(Process *rcvr,
     Sint tok_serial = 0;
 #endif
 #ifdef ERTS_SMP
-    ErtsProcLocks need_locks;
+    erts_aint_t state;
 #endif
 
     ERTS_SMP_LC_ASSERT(*rcvr_locks == erts_proc_lc_my_proc_locks(rcvr));
@@ -347,20 +317,21 @@ erts_queue_dist_message(Process *rcvr,
     mp = message_alloc();
 
 #ifdef ERTS_SMP
-    need_locks = ~(*rcvr_locks) & (ERTS_PROC_LOCK_MSGQ|ERTS_PROC_LOCK_STATUS);
-    if (need_locks) {
-	*rcvr_locks |= need_locks;
-	if (erts_smp_proc_trylock(rcvr, need_locks) == EBUSY) {
-	    if (need_locks == ERTS_PROC_LOCK_MSGQ) {
+    if (!(*rcvr_locks & ERTS_PROC_LOCK_MSGQ)) {
+	if (erts_smp_proc_trylock(rcvr, ERTS_PROC_LOCK_MSGQ) == EBUSY) {
+	    ErtsProcLocks need_locks = ERTS_PROC_LOCK_MSGQ;
+	    if (*rcvr_locks & ERTS_PROC_LOCK_STATUS) {
 		erts_smp_proc_unlock(rcvr, ERTS_PROC_LOCK_STATUS);
-		need_locks = (ERTS_PROC_LOCK_MSGQ
-			      | ERTS_PROC_LOCK_STATUS);
+		need_locks |= ERTS_PROC_LOCK_STATUS;
 	    }
 	    erts_smp_proc_lock(rcvr, need_locks);
 	}
     }
 
-    if (rcvr->is_exiting || ERTS_PROC_PENDING_EXIT(rcvr)) {
+    state = erts_smp_atomic32_read_acqb(&rcvr->state);
+    if (state & (ERTS_PSFLG_PENDING_EXIT|ERTS_PSFLG_EXITING)) {
+	if (!(*rcvr_locks & ERTS_PROC_LOCK_MSGQ))
+	    erts_smp_proc_unlock(rcvr, ERTS_PROC_LOCK_MSGQ);
 	/* Drop message if receiver is exiting or has a pending exit ... */
 	if (is_not_nil(token)) {
 	    ErlHeapFragment *heap_frag;
@@ -376,6 +347,8 @@ erts_queue_dist_message(Process *rcvr,
 	/* Ahh... need to decode it in order to trace it... */
 	ErlHeapFragment *mbuf;
 	Eterm msg;
+	if (!(*rcvr_locks & ERTS_PROC_LOCK_MSGQ))
+	    erts_smp_proc_unlock(rcvr, ERTS_PROC_LOCK_MSGQ);
 	message_free(mp);
 	msg = erts_msg_distext2heap(rcvr, rcvr_locks, &mbuf, &token, dist_ext);
 	if (is_value(msg))
@@ -437,26 +410,32 @@ erts_queue_dist_message(Process *rcvr,
 	mp->data.dist_ext = dist_ext;
 	LINK_MESSAGE(rcvr, mp);
 
-	notify_new_message(rcvr);
+	if (!(*rcvr_locks & ERTS_PROC_LOCK_MSGQ))
+	    erts_smp_proc_unlock(rcvr, ERTS_PROC_LOCK_MSGQ);
+
+	erts_proc_notify_new_message(rcvr);
     }
 }
 
 /* Add a message last in message queue */
-void
-erts_queue_message(Process* receiver,
-		   ErtsProcLocks *receiver_locks,
-		   ErlHeapFragment* bp,
-		   Eterm message,
-		   Eterm seq_trace_token
+static int
+queue_message(Process *c_p,
+	      Process* receiver,
+	      ErtsProcLocks *receiver_locks,
+	      erts_aint32_t *receiver_state,
+	      ErlHeapFragment* bp,
+	      Eterm message,
+	      Eterm seq_trace_token
 #ifdef USE_VM_PROBES
 		   , Eterm dt_utag
 #endif
-)
+    )
 {
     ErlMessage* mp;
-#ifdef ERTS_SMP
-    ErtsProcLocks need_locks;
-#else
+    int locked_msgq = 0;
+    erts_aint_t state;
+
+#ifndef ERTS_SMP
     ASSERT(bp != NULL || receiver->mbuf == NULL);
 #endif
 
@@ -464,31 +443,45 @@ erts_queue_message(Process* receiver,
 
     mp = message_alloc();
 
+    if (receiver_state)
+	state = *receiver_state;
+    else
+	state = erts_smp_atomic32_read_acqb(&receiver->state);
+
 #ifdef ERTS_SMP
-    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) {
+
+    if (state & (ERTS_PSFLG_EXITING|ERTS_PSFLG_PENDING_EXIT))
+	goto exiting;
+
+    if (!(*receiver_locks & ERTS_PROC_LOCK_MSGQ)) {
+	if (erts_smp_proc_trylock(receiver, ERTS_PROC_LOCK_MSGQ) == EBUSY) {
+	    ErtsProcLocks need_locks = ERTS_PROC_LOCK_MSGQ;
+	    if (*receiver_locks & ERTS_PROC_LOCK_STATUS) {
 		erts_smp_proc_unlock(receiver, ERTS_PROC_LOCK_STATUS);
-		need_locks = (ERTS_PROC_LOCK_MSGQ
-			      | ERTS_PROC_LOCK_STATUS);
+		need_locks |= ERTS_PROC_LOCK_STATUS;
 	    }
 	    erts_smp_proc_lock(receiver, need_locks);
 	}
+	locked_msgq = 1;
+	state = erts_smp_atomic32_read_nob(&receiver->state);
+	if (receiver_state)
+	    *receiver_state = state;
     }
 
-    if (receiver->is_exiting || ERTS_PROC_PENDING_EXIT(receiver)) {
-	/* Drop message if receiver is exiting or has a pending
-	 * exit ...
-	 */
+#endif
+
+    if (state & (ERTS_PSFLG_PENDING_EXIT|ERTS_PSFLG_EXITING)) {
+#ifdef ERTS_SMP
+    exiting:
+#endif
+	/* Drop message if receiver is exiting or has a pending exit... */
+	if (locked_msgq)
+	    erts_smp_proc_unlock(receiver, ERTS_PROC_LOCK_MSGQ);
 	if (bp)
 	    free_message_buffer(bp);
 	message_free(mp);
-	return;
+	return 0;
     }
-#endif
 
     ERL_MESSAGE_TERM(mp) = message;
     ERL_MESSAGE_TOKEN(mp) = seq_trace_token;
@@ -511,12 +504,11 @@ erts_queue_message(Process* receiver,
 	ERTS_SMP_MSGQ_MV_INQ2PRIVQ(receiver);
 	LINK_MESSAGE_PRIVQ(receiver, mp);
     }
-    else {
+    else
+#endif
+    {
 	LINK_MESSAGE(receiver, mp);
     }
-#else
-    LINK_MESSAGE(receiver, mp);
-#endif
 
 #ifdef USE_VM_PROBES
     if (DTRACE_ENABLED(message_queued)) {
@@ -536,15 +528,43 @@ erts_queue_message(Process* receiver,
                 tok_label, tok_lastcnt, tok_serial);
     }
 #endif
-    notify_new_message(receiver);
 
-    if (IS_TRACED_FL(receiver, F_TRACE_RECEIVE)) {
+    if (IS_TRACED_FL(receiver, F_TRACE_RECEIVE))
 	trace_receive(receiver, message);
-    }
+
+    if (locked_msgq)
+	erts_smp_proc_unlock(receiver, ERTS_PROC_LOCK_MSGQ);
+
+    erts_proc_notify_new_message(receiver);
 
 #ifndef ERTS_SMP
     ERTS_HOLE_CHECK(receiver);
 #endif
+    return 0;
+}
+
+void
+erts_queue_message(Process* receiver,
+		   ErtsProcLocks *receiver_locks,
+		   ErlHeapFragment* bp,
+		   Eterm message,
+		   Eterm seq_trace_token
+#ifdef USE_VM_PROBES
+		   , Eterm dt_utag
+#endif
+    )
+{
+    queue_message(NULL,
+		  receiver,
+		  receiver_locks,
+		  NULL,
+		  bp,
+		  message,
+		  seq_trace_token
+#ifdef USE_VM_PROBES
+		  , dt_utag
+#endif
+	);
 }
 
 void
@@ -576,9 +596,7 @@ erts_move_msg_mbuf_to_heap(Eterm** hpp, ErlOffHeap* off_heap, ErlMessage *msg)
 #endif
 
 #ifdef HARD_DEBUG
-    ProcBin *dbg_mso_start = off_heap->mso;
-    ErlFunThing *dbg_fun_start = off_heap->funs;
-    ExternalThing *dbg_external_start = off_heap->externals;
+    struct erl_off_heap_header* dbg_oh_start = off_heap->first;
     Eterm dbg_term, dbg_token;
     ErlHeapFragment *dbg_bp;
     Uint *dbg_hp, *dbg_thp_start;
@@ -752,48 +770,16 @@ copy_done:
 	int i, j;
 	ErlHeapFragment* frag;
 	{
-	    ProcBin *mso = off_heap->mso;
+	    struct erl_off_heap_header* dbg_oh = off_heap->first;
 	    i = j = 0;
-	    while (mso != dbg_mso_start) {
-		mso = mso->next;
+	    while (dbg_oh != dbg_oh_start) {
+		dbg_oh = dbg_oh->next;
 		i++;
 	    }
 	    for (frag=bp; frag; frag=frag->next) {
-		mso = frag->off_heap.mso;
-		while (mso) {
-		    mso = mso->next;
-		    j++;
-		}
-	    }
-	    ASSERT(i == j);
-	}
-	{
-	    ErlFunThing *fun = off_heap->funs;
-	    i = j = 0;
-	    while (fun != dbg_fun_start) {
-		fun = fun->next;
-		i++;
-	    }
-	    for (frag=bp; frag; frag=frag->next) {
-		fun = frag->off_heap.funs;
-		while (fun) {
-		    fun = fun->next;
-		    j++;
-		}
-	    }
-	    ASSERT(i == j);
-	}
-	{
-	    ExternalThing *external = off_heap->externals;
-	    i = j = 0;
-	    while (external != dbg_external_start) {
-		external = external->next;
-		i++;
-	    }
-	    for (frag=bp; frag; frag=frag->next) {
-		external = frag->off_heap.externals;
-		while (external) {
-		    external = external->next;
+		dbg_oh = frag->off_heap.first;
+		while (dbg_oh) {
+		    dbg_oh = dbg_oh->next;
 		    j++;
 		}
 	    }
@@ -1036,21 +1022,34 @@ erts_send_message(Process* sender,
 #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);
-	hp = erts_alloc_message_heap(msize,&bp,&ohp,receiver,receiver_locks);
+	hp = erts_alloc_message_heap_state(msize,
+					   &bp,
+					   &ohp,
+					   receiver,
+					   receiver_locks,
+					   &state);
 	BM_SWAP_TIMER(send,copy);
 	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);
-	erts_queue_message(receiver, receiver_locks, bp, message, token
+	queue_message(sender,
+		      receiver,
+		      receiver_locks,
+		      &state,
+		      bp,
+		      message,
+		      token
 #ifdef USE_VM_PROBES
-			   , NIL
+		      , NIL
 #endif
-			   );
+	    );
         BM_SWAP_TIMER(send,system);
 #else
 	ErlMessage* mp = message_alloc();
@@ -1081,11 +1080,8 @@ erts_send_message(Process* sender,
 	mp->data.attached = NULL;
 	LINK_MESSAGE(receiver, mp);
 
-	if (receiver->status == P_WAITING) {
-	    erts_add_to_runq(receiver);
-	} else if (receiver->status == P_SUSPENDED) {
-	    receiver->rstatus = P_RUNABLE;
-	}
+	erts_proc_notify_new_message(receiver);
+
 	if (IS_TRACED_FL(receiver, F_TRACE_RECEIVE)) {
 	    trace_receive(receiver, message);
 	}