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, 96 insertions, 10 deletions

diff --git a/erts/emulator/beam/erl_process.h b/erts/emulator/beam/erl_process.h
index 65422b8c15..c6376c0166 100644
--- a/erts/emulator/beam/erl_process.h
+++ b/erts/emulator/beam/erl_process.h
@@ -916,6 +916,7 @@ struct process {
     Eterm* stop;		/* Stack top */
     Eterm* heap;		/* Heap start */
     Eterm* hend;		/* Heap end */
+    Eterm* abandoned_heap;
     Uint heap_sz;		/* Size of heap in words */
     Uint min_heap_size;         /* Minimum size of heap (in words). */
     Uint min_vheap_size;        /* Minimum size of virtual heap (in words). */
@@ -1012,8 +1013,10 @@ struct process {
     Uint16 gen_gcs;		/* Number of (minor) generational GCs. */
     Uint16 max_gen_gcs;		/* Max minor gen GCs before fullsweep. */
     ErlOffHeap off_heap;	/* Off-heap data updated by copy_struct(). */
-    ErlHeapFragment* mbuf;	/* Pointer to message buffer list */
-    Uint mbuf_sz;		/* Size of all message buffers */
+    ErlHeapFragment* mbuf;	/* Pointer to heap fragment list */
+    ErlHeapFragment* live_hf_end;
+    ErtsMessage *msg_frag;	/* Pointer to message fragment list */
+    Uint mbuf_sz;		/* Total size of heap fragments and message fragments */
     ErtsPSD *psd;		/* Rarely used process specific data */
 
     Uint64 bin_vheap_sz;	/* Virtual heap block size for binaries */
@@ -1041,6 +1044,7 @@ struct process {
 #ifdef CHECK_FOR_HOLES
     Eterm* last_htop;		/* No need to scan the heap below this point. */
     ErlHeapFragment* last_mbuf;	/* No need to scan beyond this mbuf. */
+    ErlHeapFragment* heap_hfrag; /* Heap abandoned, htop now lives in this frag */
 #endif
 
 #ifdef DEBUG
@@ -1064,6 +1068,7 @@ extern const Process erts_invalid_process;
 do {						\
   (p)->last_htop = 0;				\
   (p)->last_mbuf = 0;				\
+  (p)->heap_hfrag = NULL;			\
 } while (0)
 
 # define ERTS_HOLE_CHECK(p) erts_check_for_holes((p))
@@ -1141,14 +1146,15 @@ void erts_check_for_holes(Process* p);
 #define ERTS_PSFLG_RUNNING_SYS		ERTS_PSFLG_BIT(15)
 #define ERTS_PSFLG_PROXY		ERTS_PSFLG_BIT(16)
 #define ERTS_PSFLG_DELAYED_SYS		ERTS_PSFLG_BIT(17)
+#define ERTS_PSFLG_OFF_HEAP_MSGQ	ERTS_PSFLG_BIT(18)
 #ifdef ERTS_DIRTY_SCHEDULERS
-#define ERTS_PSFLG_DIRTY_CPU_PROC	ERTS_PSFLG_BIT(18)
-#define ERTS_PSFLG_DIRTY_IO_PROC	ERTS_PSFLG_BIT(19)
-#define ERTS_PSFLG_DIRTY_CPU_PROC_IN_Q	ERTS_PSFLG_BIT(20)
-#define ERTS_PSFLG_DIRTY_IO_PROC_IN_Q	ERTS_PSFLG_BIT(21)
-#define ERTS_PSFLG_MAX  (ERTS_PSFLGS_ZERO_BIT_OFFSET + 22)
+#define ERTS_PSFLG_DIRTY_CPU_PROC	ERTS_PSFLG_BIT(19)
+#define ERTS_PSFLG_DIRTY_IO_PROC	ERTS_PSFLG_BIT(20)
+#define ERTS_PSFLG_DIRTY_CPU_PROC_IN_Q	ERTS_PSFLG_BIT(21)
+#define ERTS_PSFLG_DIRTY_IO_PROC_IN_Q	ERTS_PSFLG_BIT(22)
+#define ERTS_PSFLG_MAX  (ERTS_PSFLGS_ZERO_BIT_OFFSET + 23)
 #else
-#define ERTS_PSFLG_MAX  (ERTS_PSFLGS_ZERO_BIT_OFFSET + 18)
+#define ERTS_PSFLG_MAX  (ERTS_PSFLGS_ZERO_BIT_OFFSET + 19)
 #endif
 
 #define ERTS_PSFLGS_IN_PRQ_MASK 	(ERTS_PSFLG_IN_PRQ_MAX		\
@@ -1196,12 +1202,15 @@ void erts_check_for_holes(Process* p);
 #define SPO_USE_ARGS 2
 #define SPO_MONITOR 4
 #define SPO_SYSTEM_PROC 8
+#define SPO_OFF_HEAP_MSGQ 16
+
+extern int erts_default_spo_flags;
 
 /*
  * The following struct contains options for a process to be spawned.
  */
 typedef struct {
-    Uint flags;
+    int flags;
     int error_code;		/* Error code returned from create_process(). */
     Eterm mref;			/* Monitor ref returned (if SPO_MONITOR was given). */
 
@@ -1283,6 +1292,9 @@ extern struct erts_system_profile_flags_t erts_system_profile_flags;
 #define F_P2PNR_RESCHED      (1 <<  9) /* Process has been rescheduled via erts_pid2proc_not_running() */
 #define F_FORCE_GC           (1 << 10) /* Force gc at process in-scheduling */
 #define F_DISABLE_GC         (1 << 11) /* Disable GC */
+#define F_OFF_HEAP_MSGQ      (1 << 12) /* Off heap msg queue */
+#define F_OFF_HEAP_MSGQ_CHNG (1 << 13) /* Off heap msg queue changing */
+#define F_ABANDONED_HEAP_USE (1 << 14) /* Have usage of abandoned heap */
 
 /* process trace_flags */
 #define F_SENSITIVE          (1 << 0)
@@ -1616,6 +1628,7 @@ void erts_schedule_thr_prgr_later_cleanup_op(void (*)(void *),
 					     void *,
 					     ErtsThrPrgrLaterOp *,
 					     UWord);
+void erts_schedule_complete_off_heap_message_queue_change(Eterm pid);
 
 #if defined(ERTS_SMP) && defined(ERTS_ENABLE_LOCK_CHECK)
 int erts_dbg_check_halloc_lock(Process *p);
@@ -1743,7 +1756,7 @@ Uint erts_debug_nbalance(void);
 
 int erts_debug_wait_completed(Process *c_p, int flags);
 
-Uint erts_process_memory(Process *c_p);
+Uint erts_process_memory(Process *c_p, int incl_msg_inq);
 
 #ifdef ERTS_SMP
 #  define ERTS_GET_SCHEDULER_DATA_FROM_PROC(PROC) ((PROC)->scheduler_data)
@@ -2058,6 +2071,22 @@ ERTS_GLB_INLINE void erts_smp_xrunq_unlock(ErtsRunQueue *rq, ErtsRunQueue *xrq);
 ERTS_GLB_INLINE void erts_smp_runqs_lock(ErtsRunQueue *rq1, ErtsRunQueue *rq2);
 ERTS_GLB_INLINE void erts_smp_runqs_unlock(ErtsRunQueue *rq1, ErtsRunQueue *rq2);
 
+ERTS_GLB_INLINE ErtsMessage *erts_alloc_message_heap_state(Process *pp,
+							   erts_aint32_t *psp,
+							   ErtsProcLocks *plp,
+							   Uint sz,
+							   Eterm **hpp,
+							   ErlOffHeap **ohpp);
+ERTS_GLB_INLINE ErtsMessage *erts_alloc_message_heap(Process *pp,
+						     ErtsProcLocks *plp,
+						     Uint sz,
+						     Eterm **hpp,
+						     ErlOffHeap **ohpp);
+
+ERTS_GLB_INLINE void erts_shrink_message_heap(ErtsMessage **msgpp, Process *pp,
+					      Eterm *start_hp, Eterm *used_hp, Eterm *end_hp,
+					      Eterm *brefs, Uint brefs_size);
+
 #if ERTS_GLB_INLINE_INCL_FUNC_DEF
 
 ERTS_GLB_INLINE
@@ -2206,6 +2235,63 @@ erts_smp_runqs_unlock(ErtsRunQueue *rq1, ErtsRunQueue *rq2)
 #endif
 }
 
+ERTS_GLB_INLINE ErtsMessage *
+erts_alloc_message_heap_state(Process *pp,
+			      erts_aint32_t *psp,
+			      ErtsProcLocks *plp,
+			      Uint sz,
+			      Eterm **hpp,
+			      ErlOffHeap **ohpp)
+{
+    int on_heap;
+
+    if ((*psp) & ERTS_PSFLG_OFF_HEAP_MSGQ) {
+	ErtsMessage *mp = erts_alloc_message(sz, hpp);
+	*ohpp = sz == 0 ? NULL : &mp->hfrag.off_heap;
+	return mp;
+    }
+
+    return erts_try_alloc_message_on_heap(pp, psp, plp, sz, hpp, ohpp, &on_heap);
+}
+
+ERTS_GLB_INLINE ErtsMessage *
+erts_alloc_message_heap(Process *pp,
+			ErtsProcLocks *plp,
+			Uint sz,
+			Eterm **hpp,
+			ErlOffHeap **ohpp)
+{
+    erts_aint32_t state = erts_smp_atomic32_read_nob(&pp->state);
+    return erts_alloc_message_heap_state(pp, &state, plp, sz, hpp, ohpp);
+}
+
+ERTS_GLB_INLINE void
+erts_shrink_message_heap(ErtsMessage **msgpp, Process *pp,
+			 Eterm *start_hp, Eterm *used_hp, Eterm *end_hp,
+			 Eterm *brefs, Uint brefs_size)
+{
+    ASSERT(start_hp <= used_hp && used_hp <= end_hp);
+    if ((*msgpp)->data.attached == ERTS_MSG_COMBINED_HFRAG)
+	*msgpp = erts_shrink_message(*msgpp, used_hp - start_hp,
+				     brefs, brefs_size);
+    else if (!(*msgpp)->data.attached) {
+	ERTS_SMP_LC_ASSERT(ERTS_PROC_LOCK_MAIN
+			   & erts_proc_lc_my_proc_locks(pp));
+	HRelease(pp, end_hp, used_hp);
+    }
+    else {
+	ErlHeapFragment *hfrag = (*msgpp)->data.heap_frag;
+	if (start_hp != used_hp)
+	    hfrag = erts_resize_message_buffer(hfrag, used_hp - start_hp,
+					       brefs, brefs_size);
+	else {
+	    free_message_buffer(hfrag);
+	    hfrag = NULL;
+	}
+	(*msgpp)->data.heap_frag = hfrag;
+    }
+}
+
 #endif /* #if ERTS_GLB_INLINE_INCL_FUNC_DEF */
 
 ERTS_GLB_INLINE ErtsAtomCacheMap *erts_get_atom_cache_map(Process *c_p);