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, 48 insertions, 20 deletions

diff --git a/erts/emulator/beam/erl_trace.c b/erts/emulator/beam/erl_trace.c
index e9dd96efc4..d02f1f7213 100644
--- a/erts/emulator/beam/erl_trace.c
+++ b/erts/emulator/beam/erl_trace.c
@@ -114,15 +114,10 @@ void erts_init_trace(void) {
 
 static Eterm system_seq_tracer;
 
-#ifdef ERTS_SMP
 #define ERTS_ALLOC_SYSMSG_HEAP(SZ, BPP, OHPP, UNUSED) \
   (*(BPP) = new_message_buffer((SZ)), \
    *(OHPP) = &(*(BPP))->off_heap, \
    (*(BPP))->mem)
-#else
-#define ERTS_ALLOC_SYSMSG_HEAP(SZ, BPP, OHPP, RPP) \
-  erts_alloc_message_heap((SZ), (BPP), (OHPP), (RPP), 0)
-#endif
 
 #ifdef ERTS_SMP
 #define ERTS_ENQ_TRACE_MSG(FPID, TPID, MSG, BP) \
@@ -131,8 +126,12 @@ do { \
     enqueue_sys_msg_unlocked(SYS_MSG_TYPE_TRACE, (FPID), (TPID), (MSG), (BP)); \
 } while(0)
 #else
-#define ERTS_ENQ_TRACE_MSG(FPID, TPROC, MSG, BP) \
-    erts_queue_message((TPROC), NULL, (BP), (MSG), NIL)
+#define ERTS_ENQ_TRACE_MSG(FPID, TPROC, MSG, BP)		\
+    do {							\
+	ErtsMessage *mp__ = erts_alloc_message(0, NULL);		\
+	mp__->data.heap_frag = (BP);				\
+	erts_queue_message((TPROC), NULL, mp__, (MSG), NIL);	\
+    } while (0)
 #endif
 
 /*
@@ -591,11 +590,9 @@ send_to_port(Process *c_p, Eterm message,
 static void 
 profile_send(Eterm from, Eterm message) {
     Uint sz = 0;
-    ErlHeapFragment *bp = NULL;
     Uint *hp = NULL;
     Eterm msg = NIL;
     Process *profile_p = NULL;
-    ErlOffHeap *off_heap = NULL;
 
     Eterm profiler = erts_get_system_profile();
 
@@ -621,6 +618,7 @@ profile_send(Eterm from, Eterm message) {
 	}
     	
     } else {
+	ErtsMessage *mp;
 	ASSERT(is_internal_pid(profiler));
         
 	profile_p = erts_proc_lookup(profiler);
@@ -629,10 +627,13 @@ profile_send(Eterm from, Eterm message) {
 	    return;
 
 	sz = size_object(message);
-	hp = erts_alloc_message_heap(sz, &bp, &off_heap, profile_p, 0);
-	msg = copy_struct(message, sz, &hp, &bp->off_heap);
-	
-        erts_queue_message(profile_p, NULL, bp, msg, NIL);
+	mp = erts_alloc_message(sz, &hp);
+	if (sz == 0)
+	    msg = message;
+	else
+	    msg = copy_struct(message, sz, &hp, &mp->hfrag.off_heap);
+
+        erts_queue_message(profile_p, NULL, mp, msg, NIL);
     }
 }
 
@@ -1233,7 +1234,11 @@ seq_trace_output_generic(Eterm token, Eterm msg, Uint type,
 	erts_smp_mtx_unlock(&smq_mtx);
 #else
         /* trace_token must be NIL here */
-	erts_queue_message(tracer, NULL, bp, mess, NIL);
+	{
+	    ErtsMessage *mp = erts_alloc_message(0, NULL);
+	    mp->data.heap_frag = bp;
+	    erts_queue_message(tracer, NULL, mp, mess, NIL);
+	}
 #endif
     }
 }
@@ -2308,7 +2313,11 @@ monitor_long_schedule_proc(Process *p, BeamInstr *in_fp, BeamInstr *out_fp, Uint
 #ifdef ERTS_SMP
     enqueue_sys_msg(SYS_MSG_TYPE_SYSMON, p->common.id, NIL, msg, bp);
 #else
-    erts_queue_message(monitor_p, NULL, bp, msg, NIL);
+    {
+	ErtsMessage *mp = erts_alloc_message(0, NULL);
+	mp->data.heap_frag = bp;
+	erts_queue_message(monitor_p, NULL, mp, msg, NIL);
+    }
 #endif
 }
 void 
@@ -2369,7 +2378,11 @@ monitor_long_schedule_port(Port *pp, ErtsPortTaskType type, Uint time)
 #ifdef ERTS_SMP
     enqueue_sys_msg(SYS_MSG_TYPE_SYSMON, pp->common.id, NIL, msg, bp);
 #else
-    erts_queue_message(monitor_p, NULL, bp, msg, NIL);
+    {
+	ErtsMessage *mp = erts_alloc_message(0, NULL);
+	mp->data.heap_frag = bp;
+	erts_queue_message(monitor_p, NULL, mp, msg, NIL);
+    }
 #endif
 }
 
@@ -2440,7 +2453,11 @@ monitor_long_gc(Process *p, Uint time) {
 #ifdef ERTS_SMP
     enqueue_sys_msg(SYS_MSG_TYPE_SYSMON, p->common.id, NIL, msg, bp);
 #else
-    erts_queue_message(monitor_p, NULL, bp, msg, NIL);
+    {
+	ErtsMessage *mp = erts_alloc_message(0, NULL);
+	mp->data.heap_frag = bp;
+	erts_queue_message(monitor_p, NULL, mp, msg, NIL);
+    }
 #endif
 }
 
@@ -2511,7 +2528,11 @@ monitor_large_heap(Process *p) {
 #ifdef ERTS_SMP
     enqueue_sys_msg(SYS_MSG_TYPE_SYSMON, p->common.id, NIL, msg, bp);
 #else
-    erts_queue_message(monitor_p, NULL, bp, msg, NIL);
+    {
+	ErtsMessage *mp = erts_alloc_message(0, NULL);
+	mp->data.heap_frag = bp;
+	erts_queue_message(monitor_p, NULL, mp, msg, NIL);
+    }
 #endif
 }
 
@@ -2539,7 +2560,11 @@ monitor_generic(Process *p, Eterm type, Eterm spec) {
 #ifdef ERTS_SMP
     enqueue_sys_msg(SYS_MSG_TYPE_SYSMON, p->common.id, NIL, msg, bp);
 #else
-    erts_queue_message(monitor_p, NULL, bp, msg, NIL);
+    {
+	ErtsMessage *mp = erts_alloc_message(0, NULL);
+	mp->data.heap_frag = bp;
+	erts_queue_message(monitor_p, NULL, mp, msg, NIL);
+    }
 #endif
 
 }
@@ -3331,8 +3356,11 @@ sys_msg_dispatcher_func(void *unused)
 		    goto failure;
 		}
 		else {
+		    ErtsMessage *mp;
 		queue_proc_msg:
-		    erts_queue_message(proc,&proc_locks,smqp->bp,smqp->msg,NIL);
+		    mp = erts_alloc_message(0, NULL);
+		    mp->data.heap_frag = smqp->bp;
+		    erts_queue_message(proc,&proc_locks,mp,smqp->msg,NIL);
 #ifdef DEBUG_PRINTOUTS
 		    erts_fprintf(stderr, "delivered\n");
 #endif