Implementation of true asynchronous signaling between processes

Communication between Erlang processes has conceptually always been
performed through asynchronous signaling. The runtime system
implementation has however previously preformed most operation
synchronously. In a system with only one true thread of execution, this
is not problematic (often the opposite). In a system with multiple threads
of execution (as current runtime system implementation with SMP support)
it becomes problematic. This since it often involves locking of structures
when updating them which in turn cause resource contention. Utilizing
true asynchronous communication often avoids these resource contention
issues.

The case that triggered this change was contention on the link lock due
to frequent updates of the monitor trees during communication with a
frequently used server. The signal order delivery guarantees of the
language makes it hard to change the implementation of only some signals
to use true asynchronous signaling. Therefore the implementations
of (almost) all signals have been changed.

Currently the following signals have been implemented as true
asynchronous signals:
- Message signals
- Exit signals
- Monitor signals
- Demonitor signals
- Monitor triggered signals (DOWN, CHANGE, etc)
- Link signals
- Unlink signals
- Group leader signals

All of the above already defined as asynchronous signals in the
language. The implementation of messages signals was quite
asynchronous to begin with, but had quite strict delivery constraints
due to the ordering guarantees of signals between a pair of processes.

The previously used message queue partitioned into two halves has been
replaced by a more general signal queue partitioned into three parts
that service all kinds of signals. More details regarding the signal
queue can be found in comments in the erl_proc_sig_queue.h file.

The monitor and link implementations have also been completely replaced
in order to fit the new asynchronous signaling implementation as good
as possible. More details regarding the new monitor and link
implementations can be found in the erl_monitor_link.h file.

1 files changed, 111 insertions, 136 deletions

diff --git a/erts/emulator/beam/erl_gc.c b/erts/emulator/beam/erl_gc.c
index 1c64644efc..5da8e3eda5 100644
--- a/erts/emulator/beam/erl_gc.c
+++ b/erts/emulator/beam/erl_gc.c
@@ -1,7 +1,7 @@
 /*
  * %CopyrightBegin%
  *
- * Copyright Ericsson AB 2002-2017. All Rights Reserved.
+ * Copyright Ericsson AB 2002-2018. All Rights Reserved.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -43,6 +43,7 @@
 #include "erl_bif_unique.h"
 #include "dist.h"
 #include "erl_nfunc_sched.h"
+#include "erl_proc_sig_queue.h"
 
 #define ERTS_INACT_WR_PB_LEAVE_MUCH_LIMIT 1
 #define ERTS_INACT_WR_PB_LEAVE_MUCH_PERCENTAGE 20
@@ -154,7 +155,6 @@ static void offset_rootset(Process *p, Sint offs, char* area, Uint area_size,
 			   Eterm* objv, int nobj);
 static void offset_off_heap(Process* p, Sint offs, char* area, Uint area_size);
 static void offset_mqueue(Process *p, Sint offs, char* area, Uint area_size);
-static void move_msgq_to_heap(Process *p);
 static int reached_max_heap_size(Process *p, Uint total_heap_size,
                                  Uint extra_heap_size, Uint extra_old_heap_size);
 static void init_gc_info(ErtsGCInfo *gcip);
@@ -189,6 +189,21 @@ static struct {
     ErtsGCInfo info;
 } dirty_gc;
 
+static void move_msgs_to_heap(Process *c_p)
+{
+    Uint64 pre_oh, post_oh;
+
+    pre_oh = c_p->off_heap.overhead;
+    erts_proc_sig_move_msgs_to_heap(c_p);
+    post_oh = c_p->off_heap.overhead;
+
+    if (pre_oh != post_oh) {
+	/* Got new binaries; update bin vheap size... */
+        c_p->bin_vheap_sz = next_vheap_size(c_p, post_oh,
+                                            c_p->bin_vheap_sz);
+    }
+}
+
 static ERTS_INLINE int
 gc_cost(Uint gc_moved_live_words, Uint resize_moved_words)
 {
@@ -565,20 +580,26 @@ young_gen_usage(Process *p)
     hsz = p->mbuf_sz;
 
     if (p->flags & F_ON_HEAP_MSGQ) {
-	ErtsMessage *mp;
-	for (mp = p->msg.first; mp; mp = mp->next) {
-	    /*
-	     * We leave not yet decoded distribution messages
-	     * as they are in the queue since it is not
-	     * possible to determine a maximum size until
-	     * actual decoding. However, we use their estimated
-	     * size when calculating need, and by this making
-	     * it more likely that they will fit on the heap
-	     * when actually decoded.
-	     */
-	    if (mp->data.attached)
-		hsz += erts_msg_attached_data_size(mp);
-	}
+        ERTS_FOREACH_SIG_PRIVQS(
+            p, mp,
+            {
+                /*
+                 * We leave not yet decoded distribution messages
+                 * as they are in the queue since it is not
+                 * possible to determine a maximum size until
+                 * actual decoding. However, we use their estimated
+                 * size when calculating need, and by this making
+                 * it more likely that they will fit on the heap
+                 * when actually decoded.
+                 *
+                 * We however ignore off heap messages...
+                 */
+                if (ERTS_SIG_IS_MSG(mp)
+                    && mp->data.attached
+                    && mp->data.attached != ERTS_MSG_COMBINED_HFRAG) {
+                    hsz += erts_msg_attached_data_size(mp);
+                }
+            });
     }
 
     hsz += p->htop - p->heap;
@@ -621,7 +642,7 @@ check_for_possibly_long_gc(Process *p, Uint ygen_usage)
     sz = ygen_usage;
     sz += p->hend - p->stop;
     if (p->flags & F_ON_HEAP_MSGQ)
-        sz += p->msg.len;
+        sz += p->sig_qs.len;
     if (major)
 	sz += p->old_htop - p->old_heap;
 
@@ -761,13 +782,9 @@ do_major_collection:
        long_gc/large_gc triggers when this happens as process was
        killed before a GC could be done. */
     if (reds == -2) {
-        ErtsProcLocks locks = ERTS_PROC_LOCKS_ALL;
         int res;
 
-        erts_proc_lock(p, ERTS_PROC_LOCKS_ALL_MINOR);
-        erts_send_exit_signal(p, p->common.id, p, &locks,
-                              am_kill, NIL, NULL, 0);
-        erts_proc_unlock(p, locks & ERTS_PROC_LOCKS_ALL_MINOR);
+        erts_set_self_exiting(p, am_killed);
 
     delay_gc_after_start:
         /* erts_send_exit_signal looks for ERTS_PSFLG_GC, so
@@ -1375,7 +1392,7 @@ minor_collection(Process* p, ErlHeapFragment *live_hf_end,
 		 new_sz, objv, nobj);
 
 	if (p->flags & F_ON_HEAP_MSGQ)
-	    move_msgq_to_heap(p);
+	    move_msgs_to_heap(p);
 
 	new_mature = p->old_htop - prev_old_htop;
 
@@ -1760,7 +1777,7 @@ major_collection(Process* p, ErlHeapFragment *live_hf_end,
     HIGH_WATER(p) = HEAP_TOP(p);
 
     if (p->flags & F_ON_HEAP_MSGQ)
-	move_msgq_to_heap(p);
+	move_msgs_to_heap(p);
 
     ErtsGcQuickSanityCheck(p);
 
@@ -2332,9 +2349,9 @@ collect_live_heap_frags(Process* p, ErlHeapFragment *live_hf_end, Eterm* n_htop)
     return n_htop;
 }
 
-static ERTS_INLINE void
-copy_one_frag(Eterm** hpp, ErlOffHeap* off_heap,
-	      ErlHeapFragment *bp, Eterm *refs, int nrefs)
+void
+erts_copy_one_frag(Eterm** hpp, ErlOffHeap* off_heap,
+                   ErlHeapFragment *bp, Eterm *refs, int nrefs)
 {
     Uint sz;
     int i;
@@ -2440,61 +2457,6 @@ copy_one_frag(Eterm** hpp, ErlOffHeap* off_heap,
     bp->off_heap.first = NULL;
 }
 
-static void
-move_msgq_to_heap(Process *p)
-{
-    
-    ErtsMessage **mpp = &p->msg.first;
-    Uint64 pre_oh = MSO(p).overhead;
-
-    while (*mpp) {
-	ErtsMessage *mp = *mpp;
-
-	if (mp->data.attached) {
-	    ErlHeapFragment *bp;
-
-	    /*
-	     * We leave not yet decoded distribution messages
-	     * as they are in the queue since it is not
-	     * possible to determine a maximum size until
-	     * actual decoding...
-	     */
-	    if (is_value(ERL_MESSAGE_TERM(mp))) {
-
-                bp = erts_message_to_heap_frag(mp);
-
-		if (bp->next)
-		    erts_move_multi_frags(&p->htop, &p->off_heap, bp,
-					  mp->m, ERL_MESSAGE_REF_ARRAY_SZ, 0);
-		else
-		    copy_one_frag(&p->htop, &p->off_heap, bp,
-				  mp->m, ERL_MESSAGE_REF_ARRAY_SZ);
-
-		if (mp->data.attached != ERTS_MSG_COMBINED_HFRAG) {
-		    mp->data.heap_frag = NULL;
-		    free_message_buffer(bp);
-		}
-		else {
-		    ErtsMessage *new_mp = erts_alloc_message(0, NULL);
-		    sys_memcpy((void *) new_mp->m, (void *) mp->m,
-			       sizeof(Eterm)*ERL_MESSAGE_REF_ARRAY_SZ);
-		    erts_msgq_replace_msg_ref(&p->msg, new_mp, mpp);
-		    mp->next = NULL;
-		    erts_cleanup_messages(mp);
-		    mp = new_mp;
-		}
-	    }
-	}
-
-	mpp = &(*mpp)->next;
-    }
-
-    if (pre_oh != MSO(p).overhead) {
-	/* Got new binaries; update vheap size... */
-	BIN_VHEAP_SZ(p) = next_vheap_size(p, MSO(p).overhead, BIN_VHEAP_SZ(p));
-    }
-}
-
 static Uint
 setup_rootset(Process *p, Eterm *objv, int nobj, Rootset *rootset)
 {
@@ -2583,11 +2545,10 @@ setup_rootset(Process *p, Eterm *objv, int nobj, Rootset *rootset)
 	 * We do not have off heap message queue enabled, i.e. we
 	 * need to add message queue to rootset...
 	 */
-	ErtsMessage *mp;
 
 	/* Ensure large enough rootset... */
-	if (n + p->msg.len > rootset->size) {
-	    Uint new_size = n + p->msg.len;
+	if (n + p->sig_qs.len > rootset->size) {
+	    Uint new_size = n + p->sig_qs.len;
 	    ERTS_GC_ASSERT(roots == rootset->def);
 	    roots = erts_alloc(ERTS_ALC_T_ROOTSET,
 			       new_size*sizeof(Roots));
@@ -2595,18 +2556,19 @@ setup_rootset(Process *p, Eterm *objv, int nobj, Rootset *rootset)
 	    rootset->size = new_size;
 	}
 
-	for (mp = p->msg.first; mp; mp = mp->next) {
-
-	    if (!mp->data.attached) {
-		/*
-		 * Message may refer data on heap;
-		 * add it to rootset...
-		 */
-		roots[n].v = mp->m;
-		roots[n].sz = ERL_MESSAGE_REF_ARRAY_SZ;
-		n++;
-	    }
-	}
+        ERTS_FOREACH_SIG_PRIVQS(
+            p, mp,
+            {
+                if (ERTS_SIG_IS_INTERNAL_MSG(mp) && !mp->data.attached) {
+                    /*
+                     * Message may refer data on heap;
+                     * add it to rootset...
+                     */
+                    roots[n].v = mp->m;
+                    roots[n].sz = ERL_MESSAGE_REF_ARRAY_SZ;
+                    n++;
+                }
+            });
 	break;
     }
     }
@@ -3117,51 +3079,59 @@ offset_off_heap(Process* p, Sint offs, char* area, Uint area_size)
     }
 }
 
+#ifndef USE_VM_PROBES
+#define ERTS_OFFSET_DT_UTAG(MP, A, ASZ, O)
+#else
+#define ERTS_OFFSET_DT_UTAG(MP, A, ASZ, O)                                      \
+    do {                                                                        \
+        Eterm utag__ = ERL_MESSAGE_DT_UTAG((MP));                               \
+        if (is_boxed(utag__) && ErtsInArea(ptr_val(utag__), (A), (ASZ))) {      \
+            ERL_MESSAGE_DT_UTAG((MP)) = offset_ptr(utag__, (O));                \
+        }                                                                       \
+    } while (0)
+#endif
+
+static ERTS_INLINE void
+offset_message(ErtsMessage *mp, Sint offs, char* area, Uint area_size)
+{
+    Eterm mesg = ERL_MESSAGE_TERM(mp);
+    if (ERTS_SIG_IS_MSG_TAG(mesg)) {
+        if (ERTS_SIG_IS_INTERNAL_MSG_TAG(mesg)) {
+            switch (primary_tag(mesg)) {
+            case TAG_PRIMARY_LIST:
+            case TAG_PRIMARY_BOXED:
+                if (ErtsInArea(ptr_val(mesg), area, area_size)) {
+                    ERL_MESSAGE_TERM(mp) = offset_ptr(mesg, offs);
+                }
+                break;
+            }
+        }
+        mesg = ERL_MESSAGE_TOKEN(mp);
+        if (is_boxed(mesg) && ErtsInArea(ptr_val(mesg), area, area_size)) {
+            ERL_MESSAGE_TOKEN(mp) = offset_ptr(mesg, offs);
+        }
+
+        ERTS_OFFSET_DT_UTAG(mp, area, area_size, offs);
+	
+        ASSERT((is_nil(ERL_MESSAGE_TOKEN(mp)) ||
+                is_tuple(ERL_MESSAGE_TOKEN(mp)) ||
+                is_atom(ERL_MESSAGE_TOKEN(mp))));
+    }
+}
+
 /*
  * Offset pointers in message queue.
  */
 static void
 offset_mqueue(Process *p, Sint offs, char* area, Uint area_size)
 {
-    ErtsMessage* mp = p->msg.first;
-
-    if ((p->flags & (F_OFF_HEAP_MSGQ|F_OFF_HEAP_MSGQ_CHNG)) != F_OFF_HEAP_MSGQ) {
-
-	while (mp != NULL) {
-	    Eterm mesg = ERL_MESSAGE_TERM(mp);
-	    if (is_value(mesg)) {
-		switch (primary_tag(mesg)) {
-		case TAG_PRIMARY_LIST:
-		case TAG_PRIMARY_BOXED:
-		    if (ErtsInArea(ptr_val(mesg), area, area_size)) {
-			ERL_MESSAGE_TERM(mp) = offset_ptr(mesg, offs);
-		    }
-		    break;
-		}
-	    }
-	    mesg = ERL_MESSAGE_TOKEN(mp);
-	    if (is_boxed(mesg) && ErtsInArea(ptr_val(mesg), area, area_size)) {
-		ERL_MESSAGE_TOKEN(mp) = offset_ptr(mesg, offs);
-	    }
-#ifdef USE_VM_PROBES
-	    mesg = ERL_MESSAGE_DT_UTAG(mp);
-	    if (is_boxed(mesg) && ErtsInArea(ptr_val(mesg), area, area_size)) {
-		ERL_MESSAGE_DT_UTAG(mp) = offset_ptr(mesg, offs);
-	    }
-#endif	
-	
-	    ASSERT((is_nil(ERL_MESSAGE_TOKEN(mp)) ||
-		    is_tuple(ERL_MESSAGE_TOKEN(mp)) ||
-		    is_atom(ERL_MESSAGE_TOKEN(mp))));
-	    mp = mp->next;
-	}
-
-    }
+    if ((p->flags & (F_OFF_HEAP_MSGQ|F_OFF_HEAP_MSGQ_CHNG)) != F_OFF_HEAP_MSGQ)
+        ERTS_FOREACH_SIG_PRIVQS(p, mp, offset_message(mp, offs, area, area_size));
 }
 
 static void ERTS_INLINE
 offset_one_rootset(Process *p, Sint offs, char* area, Uint area_size,
-	       Eterm* objv, int nobj)
+                   Eterm* objv, int nobj)
 {
     Eterm *v;
     Uint sz;
@@ -3378,7 +3348,6 @@ erts_process_gc_info(Process *p, Uint *sizep, Eterm **hpp,
     };
 
     Eterm res = THE_NON_VALUE;
-    ErtsMessage *mp;
 
     ERTS_CT_ASSERT(sizeof(values)/sizeof(*values) == sizeof(tags)/sizeof(*tags));
     ERTS_CT_ASSERT(sizeof(values)/sizeof(*values) == ERTS_PROCESS_GC_INFO_MAX_TERMS);
@@ -3397,9 +3366,15 @@ erts_process_gc_info(Process *p, Uint *sizep, Eterm **hpp,
            be the same as adding old_heap_block_size + heap_block_size
            + mbuf_size.
         */
-        for (mp = p->msg.first; mp; mp = mp->next)
-            if (mp->data.attached)
-                values[2] += erts_msg_attached_data_size(mp);
+        ERTS_FOREACH_SIG_PRIVQS(
+            p, mp,
+            {
+                if (ERTS_SIG_IS_MSG(mp)
+                    && mp->data.attached
+                    && mp->data.attached != ERTS_MSG_COMBINED_HFRAG) {
+                    values[2] += erts_msg_attached_data_size(mp);
+                }
+            });
     }
 
     res = erts_bld_atom_uword_2tup_list(hpp,