Fragmented young heap generation and off_heap_message_queue option

* The youngest generation of the heap can now consist of multiple
  blocks. Heap fragments and message fragments are added to the
  youngest generation when needed without triggering a GC. After
  a GC the youngest generation is contained in one single block.
* The off_heap_message_queue process flag has been added. When
  enabled all message data in the queue is kept off heap. When
  a message is selected from the queue, the message fragment (or
  heap fragment) containing the actual message is attached to the
  youngest generation. Messages stored off heap is not part of GC.

1 files changed, 103 insertions, 167 deletions

diff --git a/erts/emulator/beam/erl_bif_info.c b/erts/emulator/beam/erl_bif_info.c
index a684c81445..1eb106a551 100644
--- a/erts/emulator/beam/erl_bif_info.c
+++ b/erts/emulator/beam/erl_bif_info.c
@@ -589,6 +589,7 @@ static Eterm pi_args[] = {
     am_min_bin_vheap_size,
     am_current_location,
     am_current_stacktrace,
+    am_off_heap_message_queue
 };
 
 #define ERTS_PI_ARGS ((int) (sizeof(pi_args)/sizeof(Eterm)))
@@ -636,6 +637,7 @@ pi_arg2ix(Eterm arg)
     case am_min_bin_vheap_size:			return 28;
     case am_current_location:			return 29;
     case am_current_stacktrace:			return 30;
+    case am_off_heap_message_queue:		return 31;
     default:					return -1;
     }
 }
@@ -718,9 +720,10 @@ pi_pid2proc(Process *c_p, Eterm pid, ErtsProcLocks info_locks)
 
 
 
-BIF_RETTYPE
+static BIF_RETTYPE
 process_info_aux(Process *BIF_P,
 		 Process *rp,
+		 ErtsProcLocks rp_locks,
 		 Eterm rpid,
 		 Eterm item,
 		 int always_wrap);
@@ -811,10 +814,31 @@ process_info_list(Process *c_p, Eterm pid, Eterm list, int always_wrap,
 	*fail_type = ERTS_PI_FAIL_TYPE_AWAIT_EXIT;
 	goto done;
     }
-    else if (!(locks & ERTS_PROC_LOCK_STATUS)) {
-	erts_smp_proc_unlock(rp, ERTS_PROC_LOCK_STATUS);
+    else {
+	ErtsProcLocks unlock_locks = 0;
+
+	if (c_p == rp)
+	    locks |= ERTS_PROC_LOCK_MAIN;
+
+	if (!(locks & ERTS_PROC_LOCK_STATUS))
+	    unlock_locks |= ERTS_PROC_LOCK_STATUS;
+
+	if (locks & ERTS_PROC_LOCK_MSGQ) {
+	    /*
+	     * Move in queue into private queue and
+	     * release msgq lock, enabling others to
+	     * send messages to the process while it
+	     * is being inspected...
+	     */
+	    ASSERT(locks & ERTS_PROC_LOCK_MAIN);
+	    ERTS_SMP_MSGQ_MV_INQ2PRIVQ(rp);
+	    locks &= ~ERTS_PROC_LOCK_MSGQ;
+	    unlock_locks |= ERTS_PROC_LOCK_MSGQ;
+	}
+
+	if (unlock_locks)
+	    erts_smp_proc_unlock(rp, unlock_locks);
     }
-	
 
     /*
      * We always handle 'messages' first if it should be part
@@ -826,7 +850,7 @@ process_info_list(Process *c_p, Eterm pid, Eterm list, int always_wrap,
     if (want_messages) {
 	ix = pi_arg2ix(am_messages);
 	ASSERT(part_res[ix] == THE_NON_VALUE);
-	part_res[ix] = process_info_aux(c_p, rp, pid, am_messages, always_wrap);
+	part_res[ix] = process_info_aux(c_p, rp, locks, pid, am_messages, always_wrap);
 	ASSERT(part_res[ix] != THE_NON_VALUE);
     }
 
@@ -834,7 +858,7 @@ process_info_list(Process *c_p, Eterm pid, Eterm list, int always_wrap,
 	ix = res_elem_ix[res_elem_ix_ix];
 	if (part_res[ix] == THE_NON_VALUE) {
 	    arg = pi_ix2arg(ix);
-	    part_res[ix] = process_info_aux(c_p, rp, pid, arg, always_wrap);
+	    part_res[ix] = process_info_aux(c_p, rp, locks, pid, arg, always_wrap);
 	    ASSERT(part_res[ix] != THE_NON_VALUE);
 	}
     }
@@ -965,9 +989,31 @@ BIF_RETTYPE process_info_2(BIF_ALIST_2)
 	ERTS_BIF_AWAIT_X_DATA_TRAP(BIF_P, BIF_ARG_1, am_undefined);
     }
     else {
+	ErtsProcLocks unlock_locks = 0;
+
+	if (BIF_P == rp)
+	    info_locks |= ERTS_PROC_LOCK_MAIN;
+
 	if (!(info_locks & ERTS_PROC_LOCK_STATUS))
-	    erts_smp_proc_unlock(rp, ERTS_PROC_LOCK_STATUS);
-	res = process_info_aux(BIF_P, rp, pid, BIF_ARG_2, 0);
+	    unlock_locks |= ERTS_PROC_LOCK_STATUS;
+
+	if (info_locks & ERTS_PROC_LOCK_MSGQ) {
+	    /*
+	     * Move in queue into private queue and
+	     * release msgq lock, enabling others to
+	     * send messages to the process while it
+	     * is being inspected...
+	     */
+	    ASSERT(info_locks & ERTS_PROC_LOCK_MAIN);
+	    ERTS_SMP_MSGQ_MV_INQ2PRIVQ(rp);
+	    info_locks &= ~ERTS_PROC_LOCK_MSGQ;
+	    unlock_locks |= ERTS_PROC_LOCK_MSGQ;
+	}
+
+	if (unlock_locks)
+	    erts_smp_proc_unlock(rp, unlock_locks);
+
+	res = process_info_aux(BIF_P, rp, info_locks, pid, BIF_ARG_2, 0);
     }
     ASSERT(is_value(res));
 
@@ -985,6 +1031,7 @@ BIF_RETTYPE process_info_2(BIF_ALIST_2)
 Eterm
 process_info_aux(Process *BIF_P,
 		 Process *rp,
+		 ErtsProcLocks rp_locks,
 		 Eterm rpid,
 		 Eterm item,
 		 int always_wrap)
@@ -1056,171 +1103,55 @@ process_info_aux(Process *BIF_P,
 	break;
 
     case am_messages: {
-	ErlMessage* mp;
-	int n;
 
-	ERTS_SMP_MSGQ_MV_INQ2PRIVQ(rp);
-	n = rp->msg.len;
-
-	if (n == 0 || ERTS_TRACE_FLAGS(rp) & F_SENSITIVE) {
+	if (rp->msg.len == 0 || ERTS_TRACE_FLAGS(rp) & F_SENSITIVE) {
 	    hp = HAlloc(BIF_P, 3);
 	} else {
-	    int remove_bad_messages = 0;
-	    struct {
-		Uint copy_struct_size;
-		ErlMessage* msgp;
-	    } *mq = erts_alloc(ERTS_ALC_T_TMP, n*sizeof(*mq));
-	    Sint i = 0;
-	    Uint heap_need = 3;
+	    ErtsMessageInfo *mip;
+	    Sint i;
+	    Uint heap_need;
+#ifdef DEBUG
 	    Eterm *hp_end;
+#endif
 
-	    for (mp = rp->msg.first; mp; mp = mp->next) {
-		heap_need += 2;
-		mq[i].msgp = mp;
-		if (rp != BIF_P) {
-		    Eterm msg = ERL_MESSAGE_TERM(mq[i].msgp);
-		    if (is_value(msg)) {
-			mq[i].copy_struct_size = (is_immed(msg)? 0 :
-						  size_object(msg));
-		    }
-		    else if (mq[i].msgp->data.attached) {
-			mq[i].copy_struct_size
-			    = erts_msg_attached_data_size(mq[i].msgp);
-		    }
-		    else {
-			/* Bad distribution message; ignore */
-			remove_bad_messages = 1;
-			mq[i].copy_struct_size = 0;
-		    }
-		    heap_need += mq[i].copy_struct_size;
-		}
-		else {
-		    mq[i].copy_struct_size = mp->data.attached ?
-                        erts_msg_attached_data_size(mp) : 0;
-		}
-		i++;
-	    }
+	    mip = erts_alloc(ERTS_ALC_T_TMP,
+			     rp->msg.len*sizeof(ErtsMessageInfo));
 
-            if (rp != BIF_P) {
-                hp = HAlloc(BIF_P, heap_need);
-                hp_end = hp + heap_need;
-                ASSERT(i == n);
-                for (i--; i >= 0; i--) {
-                    Eterm msg = ERL_MESSAGE_TERM(mq[i].msgp);
-		    if (is_value(msg)) {
-			if (mq[i].copy_struct_size)
-			    msg = copy_struct(msg,
-					      mq[i].copy_struct_size,
-					      &hp,
-					      &MSO(BIF_P));
-		    }
-		    else if (mq[i].msgp->data.attached) {
-			ErlHeapFragment *hfp;
-			/*
-			 * Decode it into a message buffer and attach it
-			 * to the message instead of the attached external
-			 * term.
-			 *
-			 * Note that we may not pass a process pointer
-			 * to erts_msg_distext2heap(), since it would then
-			 * try to alter locks on that process.
-			 */
-			msg = erts_msg_distext2heap(
-			    NULL, NULL, &hfp, &ERL_MESSAGE_TOKEN(mq[i].msgp),
-			    mq[i].msgp->data.dist_ext);
-
-			ERL_MESSAGE_TERM(mq[i].msgp) = msg;
-			mq[i].msgp->data.heap_frag = hfp;
-
-			if (is_non_value(msg)) {
-			    ASSERT(!mq[i].msgp->data.heap_frag);
-			    /* Bad distribution message; ignore */
-			    remove_bad_messages = 1;
-			    continue;
-			}
-			else {
-			    /* Make our copy of the message */
-			    ASSERT(size_object(msg) == erts_used_frag_sz(hfp));
-			    msg = copy_struct(msg,
-					      erts_used_frag_sz(hfp),
-					      &hp,
-					      &MSO(BIF_P));
-			}
-		    }
-		    else {
-			/* Bad distribution message; ignore */
-			remove_bad_messages = 1;
-			continue;
-		    }
-                    res = CONS(hp, msg, res);
-                    hp += 2;
-		}
-                HRelease(BIF_P, hp_end, hp+3);
-            }
-	    else {
-                for (i--; i >= 0; i--) {
-                    ErtsHeapFactory factory;
-                    Eterm msg = ERL_MESSAGE_TERM(mq[i].msgp);
-
-                    erts_factory_proc_prealloc_init(&factory, BIF_P,
-                                                    mq[i].copy_struct_size+2);
-		    if (mq[i].msgp->data.attached) {
-			/* Decode it on the heap */
-			erts_move_msg_attached_data_to_heap(&factory,
-							    mq[i].msgp);
-			msg = ERL_MESSAGE_TERM(mq[i].msgp);
-			ASSERT(!mq[i].msgp->data.attached);
-                    }
-                    if (is_value(msg)) {
-                        hp = erts_produce_heap(&factory, 2, 0);
-                        res = CONS(hp, msg, res);
-                    }
-                    else {
-                        /* Bad distribution message; ignore */
-                        remove_bad_messages = 1;
-                        continue;
-                    }
-                    erts_factory_close(&factory);
-		}
-                hp = HAlloc(BIF_P, 3);
-	    }
-	    erts_free(ERTS_ALC_T_TMP, mq);
-	    if (remove_bad_messages) {
-		ErlMessage **mpp;
-		/*
-		 * We need to remove bad distribution messages from
-		 * the queue, so that the value returned for
-		 * 'message_queue_len' is consistent with the value
-		 * returned for 'messages'.
-		 */
-		mpp = &rp->msg.first;
-		mp = rp->msg.first;
-		while (mp) {
-		    if (is_value(ERL_MESSAGE_TERM(mp))) {
-			mpp = &mp->next;
-			mp = mp->next;
-		    }
-		    else {
-			ErlMessage* bad_mp = mp;
-			ASSERT(!mp->data.attached);
-			if (rp->msg.save == &mp->next)
-			    rp->msg.save = mpp;
-			if (rp->msg.last == &mp->next)
-			    rp->msg.last = mpp;
-			*mpp = mp->next;
-			mp = mp->next;
-			rp->msg.len--;
-			free_message(bad_mp);
-		    }
-		}
+	    /*
+	     * Note that message queue may shrink when calling
+	     * erts_prep_msgq_for_inspection() since it removes
+	     * corrupt distribution messages.
+	     */
+	    heap_need = erts_prep_msgq_for_inspection(BIF_P, rp, rp_locks, mip);
+	    heap_need += 3; /* top 2-tuple */
+	    heap_need += rp->msg.len*2; /* Cons cells */
+
+	    hp = HAlloc(BIF_P, heap_need); /* heap_need is exact */
+#ifdef DEBUG
+	    hp_end = hp + heap_need;
+#endif
+
+	    /* Build list of messages... */
+	    for (i = rp->msg.len - 1, res = NIL; i >= 0; i--) {
+		Eterm msg = ERL_MESSAGE_TERM(mip[i].msgp);
+		Uint sz = mip[i].size;
+
+		if (sz != 0)
+		    msg = copy_struct(msg, sz, &hp, &BIF_P->off_heap);
+
+		res = CONS(hp, msg, res);
+		hp += 2;
 	    }
+
+	    ASSERT(hp_end == hp + 3);
+
+	    erts_free(ERTS_ALC_T_TMP, mip);
 	}
 	break;
     }
 
     case am_message_queue_len:
 	hp = HAlloc(BIF_P, 3);
-	ERTS_SMP_MSGQ_MV_INQ2PRIVQ(rp);
 	res = make_small(rp->msg.len);
 	break;
 
@@ -1408,7 +1339,7 @@ process_info_aux(Process *BIF_P,
     }
 
     case am_total_heap_size: {
-	ErlMessage *mp;
+	ErtsMessage *mp;
 	Uint total_heap_size;
 	Uint hsz = 3;
 
@@ -1418,8 +1349,6 @@ process_info_aux(Process *BIF_P,
 
 	total_heap_size += rp->mbuf_sz;
 
-	ERTS_SMP_MSGQ_MV_INQ2PRIVQ(rp);
-
 	for (mp = rp->msg.first; mp; mp = mp->next)
 	    if (mp->data.attached)
 		total_heap_size += erts_msg_attached_data_size(mp);
@@ -1441,7 +1370,7 @@ process_info_aux(Process *BIF_P,
 
     case am_memory: { /* Memory consumed in bytes */
 	Uint hsz = 3;
-	Uint size = erts_process_memory(rp);
+	Uint size = erts_process_memory(rp, 0);
 	(void) erts_bld_uint(NULL, &hsz, size);
 	hp = HAlloc(BIF_P, hsz);
 	res = erts_bld_uint(&hp, NULL, size);
@@ -1567,6 +1496,11 @@ process_info_aux(Process *BIF_P,
 	break;
     }
 
+    case am_off_heap_message_queue:
+	res = BIF_P->flags & F_OFF_HEAP_MSGQ ? am_true : am_false;
+	hp = HAlloc(BIF_P, 3);
+	break;
+
     default:
 	return THE_NON_VALUE; /* will produce badarg */
 
@@ -2728,6 +2662,10 @@ BIF_RETTYPE system_info_1(BIF_ALIST_1)
 	BIF_RET(am_true);
     }
 #endif
+    else if (BIF_ARG_1 == am_off_heap_message_queue) {
+	BIF_RET(erts_default_spo_flags & SPO_OFF_HEAP_MSGQ
+		? am_true : am_false);
+    }
     else if (ERTS_IS_ATOM_STR("compile_info",BIF_ARG_1)) {
 	Uint  sz;
 	Eterm res = NIL, tup, text;
@@ -3865,9 +3803,7 @@ BIF_RETTYPE erts_debug_set_internal_state_2(BIF_ALIST_2)
 		BIF_RET(am_false);
 	    }
 	    else {
-		FLAGS(rp) |= F_FORCE_GC;
-		if (BIF_P != rp)
-		    erts_smp_proc_unlock(rp, ERTS_PROC_LOCK_MAIN);
+		ERTS_FORCE_GC(BIF_P);
 		BIF_RET(am_true);
 	    }
 	}