Merge branch 'rickard/port-optimizations/OTP-10336' into rickard/r16/port-optimizations/OTP-10336

* rickard/port-optimizations/OTP-10336:
  Change annotate level for emacs-22 in cerl
  Update etp-commands
  Add documentation on communication in Erlang
  Add support for busy port message queue
  Add driver callback epilogue
  Implement true asynchronous signaling between processes and ports
  Add erl_drv_[send|output]_term
  Move busy port flag
  Use rwlock for driver list
  Optimize management of port tasks
  Improve configuration of process and port tables
  Remove R9 compatibility features
  Use ptab functionality also for ports
  Prepare for use of ptab functionality also for ports
  Atomic port state
  Generalize process table implementation
  Implement functionality for delaying thread progress from unmanaged threads

Conflicts:
	erts/doc/src/erl_driver.xml
	erts/doc/src/erlang.xml
	erts/emulator/beam/beam_bif_load.c
	erts/emulator/beam/beam_bp.c
	erts/emulator/beam/beam_emu.c
	erts/emulator/beam/bif.c
	erts/emulator/beam/copy.c
	erts/emulator/beam/erl_alloc.c
	erts/emulator/beam/erl_alloc.types
	erts/emulator/beam/erl_bif_info.c
	erts/emulator/beam/erl_bif_port.c
	erts/emulator/beam/erl_bif_trace.c
	erts/emulator/beam/erl_init.c
	erts/emulator/beam/erl_message.c
	erts/emulator/beam/erl_port_task.c
	erts/emulator/beam/erl_process.c
	erts/emulator/beam/erl_process.h
	erts/emulator/beam/erl_process_lock.c
	erts/emulator/beam/erl_trace.c
	erts/emulator/beam/export.h
	erts/emulator/beam/global.h
	erts/emulator/beam/io.c
	erts/emulator/sys/unix/sys.c
	erts/emulator/sys/vxworks/sys.c
	erts/emulator/test/port_SUITE.erl
	erts/etc/unix/cerl.src
	erts/preloaded/ebin/erlang.beam
	erts/preloaded/ebin/prim_inet.beam
	erts/preloaded/src/prim_inet.erl
	lib/hipe/cerl/erl_bif_types.erl
	lib/kernel/doc/src/inet.xml
	lib/kernel/src/inet.erl

1 files changed, 225 insertions, 60 deletions

diff --git a/erts/emulator/beam/erl_thr_progress.c b/erts/emulator/beam/erl_thr_progress.c
index 88524bdd4c..3b2f95d43b 100644
--- a/erts/emulator/beam/erl_thr_progress.c
+++ b/erts/emulator/beam/erl_thr_progress.c
@@ -96,17 +96,14 @@
 
 #define ERTS_THR_PRGR_LFLG_BLOCK	(((erts_aint32_t) 1) << 31)
 #define ERTS_THR_PRGR_LFLG_NO_LEADER	(((erts_aint32_t) 1) << 30)
-#define ERTS_THR_PRGR_LFLG_ACTIVE_MASK	(~(ERTS_THR_PRGR_LFLG_NO_LEADER \
-					   | ERTS_THR_PRGR_LFLG_BLOCK))
+#define ERTS_THR_PRGR_LFLG_WAITING_UM	(((erts_aint32_t) 1) << 29)
+#define ERTS_THR_PRGR_LFLG_ACTIVE_MASK	(~(ERTS_THR_PRGR_LFLG_NO_LEADER	\
+					   | ERTS_THR_PRGR_LFLG_BLOCK	\
+					   | ERTS_THR_PRGR_LFLG_WAITING_UM))
 
-#define ERTS_THR_PRGR_LFLGS_ACTIVE(LFLGS) \
+#define ERTS_THR_PRGR_LFLGS_ACTIVE(LFLGS)				\
     ((LFLGS) & ERTS_THR_PRGR_LFLG_ACTIVE_MASK)
 
-#define ERTS_THR_PRGR_LFLGS_ALL_WAITING(LFLGS) \
-    (((LFLGS) & (ERTS_THR_PRGR_LFLG_NO_LEADER \
-		 |ERTS_THR_PRGR_LFLG_ACTIVE_MASK)) \
-     == ERTS_THR_PRGR_LFLG_NO_LEADER)
-
 /*
  * We use a 64-bit value for thread progress. By this wrapping of
  * the thread progress will more or less never occur.
@@ -262,6 +259,11 @@ typedef struct {
     erts_atomic32_t managed_count;
     erts_atomic32_t managed_id;
     erts_atomic32_t unmanaged_id;    
+    int chk_next_ix;
+    struct {
+	int waiting;
+	erts_atomic32_t current;
+    } umrefc_ix;
 } ErtsThrPrgrMiscData;
 
 typedef struct {
@@ -276,12 +278,18 @@ typedef union {
     char align__[ERTS_ALC_CACHE_LINE_ALIGN_SIZE(sizeof(ErtsThrPrgrElement))];
 } ErtsThrPrgrArray;
 
+typedef union {
+    erts_atomic_t refc;
+    char align__[ERTS_ALC_CACHE_LINE_ALIGN_SIZE(sizeof(erts_atomic_t))];
+} ErtsThrPrgrUnmanagedRefc;
+
 typedef struct {
     union {
 	ErtsThrPrgrMiscData data;
 	char align__[ERTS_ALC_CACHE_LINE_ALIGN_SIZE(
 		sizeof(ErtsThrPrgrMiscData))];
     } misc;
+    ErtsThrPrgrUnmanagedRefc umrefc[2];
     ErtsThrPrgrArray *thr;
     struct {
 	int no;
@@ -346,7 +354,9 @@ init_tmp_thr_prgr_data(ErtsThrPrgrData *tpd)
     tpd->is_managed = 0;
     tpd->is_blocking = 0;
     tpd->is_temporary = 1;
-
+#ifdef ERTS_ENABLE_LOCK_CHECK
+    tpd->is_delaying = 0;
+#endif
     erts_tsd_set(erts_thr_prgr_data_key__, (void *) tpd);
 }
 
@@ -461,6 +471,9 @@ erts_thr_progress_init(int no_schedulers, int managed, int unmanaged)
     erts_atomic32_init_nob(&intrnl->misc.data.managed_count, 0);
     erts_atomic32_init_nob(&intrnl->misc.data.managed_id, no_schedulers);
     erts_atomic32_init_nob(&intrnl->misc.data.unmanaged_id, -1);
+    intrnl->misc.data.chk_next_ix = 0;
+    intrnl->misc.data.umrefc_ix.waiting = -1;
+    erts_atomic32_init_nob(&intrnl->misc.data.umrefc_ix.current, 0);
 
     intrnl->thr = (ErtsThrPrgrArray *) ptr;
     ptr += thr_arr_sz;
@@ -547,6 +560,9 @@ erts_thr_progress_register_unmanaged_thread(ErtsThrPrgrCallbacks *callbacks)
     tpd->is_managed = 0;
     tpd->is_blocking = is_blocking;
     tpd->is_temporary = 0;
+#ifdef ERTS_ENABLE_LOCK_CHECK
+    tpd->is_delaying = 0;
+#endif
     ASSERT(tpd->id >= 0);
     if (tpd->id >= intrnl->unmanaged.no)
 	erl_exit(ERTS_ABORT_EXIT,
@@ -600,6 +616,9 @@ erts_thr_progress_register_managed_thread(ErtsSchedulerData *esdp,
     tpd->is_managed = 1;
     tpd->is_blocking = is_blocking;
     tpd->is_temporary = 0;
+#ifdef ERTS_ENABLE_LOCK_CHECK
+    tpd->is_delaying = 1;
+#endif
 
     init_wakeup_request_array(&tpd->wakeup_request[0]);
 
@@ -607,8 +626,8 @@ erts_thr_progress_register_managed_thread(ErtsSchedulerData *esdp,
 
     tpd->leader = 0;
     tpd->active = 1;
-    tpd->previous.local = 0;
-    tpd->previous.current = ERTS_THR_PRGR_VAL_WAITING;
+    tpd->confirmed = 0;
+    tpd->leader_state.current = ERTS_THR_PRGR_VAL_WAITING;
     erts_tsd_set(erts_thr_prgr_data_key__, (void *) tpd);
 
     erts_atomic32_inc_nob(&intrnl->misc.data.lflgs);
@@ -651,60 +670,113 @@ leader_update(ErtsThrPrgrData *tpd)
 	block_thread(tpd);
     }
     else {
+	ErtsThrPrgrVal current;
+	int ix, chk_next_ix, umrefc_ix, my_ix, no_managed, waiting_unmanaged;
 	erts_aint32_t lflgs;
 	ErtsThrPrgrVal next;
-	int ix, sz, make_progress;
+	erts_aint_t refc;
 
-	if (tpd->previous.current == ERTS_THR_PRGR_VAL_WAITING) {
-	    /* Took over as leader from another thread */
-	    tpd->previous.current = read_acqb(&erts_thr_prgr__.current);
-	    tpd->previous.next = tpd->previous.current;
-	    tpd->previous.next++;
-	    if (tpd->previous.next == ERTS_THR_PRGR_VAL_WAITING)
-		tpd->previous.next = 0;
-	}
+	my_ix = tpd->id;
 
-	if (tpd->previous.local == tpd->previous.current) {
-	    ErtsThrPrgrVal val = tpd->previous.current + 1;
-	    if (val == ERTS_THR_PRGR_VAL_WAITING)
-		val = 0;
-	    tpd->previous.local = val;
-	    set_mb(&intrnl->thr[tpd->id].data.current, val);
+	if (tpd->leader_state.current == ERTS_THR_PRGR_VAL_WAITING) {
+	    /* Took over as leader from another thread */
+	    tpd->leader_state.current = read_nob(&erts_thr_prgr__.current);
+	    tpd->leader_state.next = tpd->leader_state.current;
+	    tpd->leader_state.next++;
+	    if (tpd->leader_state.next == ERTS_THR_PRGR_VAL_WAITING)
+		tpd->leader_state.next = 0;
+	    tpd->leader_state.chk_next_ix = intrnl->misc.data.chk_next_ix;
+	    tpd->leader_state.umrefc_ix.waiting = intrnl->misc.data.umrefc_ix.waiting;
+	    tpd->leader_state.umrefc_ix.current =
+		(int) erts_atomic32_read_nob(&intrnl->misc.data.umrefc_ix.current);
+
+	    if (tpd->confirmed == tpd->leader_state.current) {
+		ErtsThrPrgrVal val = tpd->leader_state.current + 1;
+		if (val == ERTS_THR_PRGR_VAL_WAITING)
+		    val = 0;
+		tpd->confirmed = val;
+		set_mb(&intrnl->thr[my_ix].data.current, val);
+	    }
 	}
 
-	next = tpd->previous.next;
 
-	make_progress = 1;
-	sz = intrnl->managed.no;
-	for (ix = 0; ix < sz; ix++) {
-	    ErtsThrPrgrVal tmp;
-	    tmp = read_nob(&intrnl->thr[ix].data.current);
-	    if (tmp != next && tmp != ERTS_THR_PRGR_VAL_WAITING) {
-		make_progress = 0;
-		ASSERT(erts_thr_progress_has_passed__(next, tmp));
-		break;
+	next = tpd->leader_state.next;
+
+	waiting_unmanaged = 0;
+	umrefc_ix = -1; /* Shut up annoying warning */
+
+	chk_next_ix = tpd->leader_state.chk_next_ix;
+	no_managed = intrnl->managed.no;
+	ASSERT(0 <= chk_next_ix && chk_next_ix <= no_managed);
+	/* Check manged threads */
+	if (chk_next_ix < no_managed) {
+	    for (ix = chk_next_ix; ix < no_managed; ix++) {
+		ErtsThrPrgrVal tmp;
+		if (ix == my_ix)
+		    continue;
+		tmp = read_nob(&intrnl->thr[ix].data.current);
+		if (tmp != next && tmp != ERTS_THR_PRGR_VAL_WAITING) {
+		    tpd->leader_state.chk_next_ix = ix;
+		    ASSERT(erts_thr_progress_has_passed__(next, tmp));
+		    goto done;
+		}
 	    }
 	}
 
-	if (make_progress) {
-	    ErtsThrPrgrVal current = next;
+	/* Check unmanged threads */
+	waiting_unmanaged = tpd->leader_state.umrefc_ix.waiting != -1;
+	umrefc_ix = (waiting_unmanaged
+		     ? tpd->leader_state.umrefc_ix.waiting
+		     : tpd->leader_state.umrefc_ix.current);
+	refc = erts_atomic_read_nob(&intrnl->umrefc[umrefc_ix].refc);
+	ASSERT(refc >= 0);
+	if (refc != 0) {
+	    int new_umrefc_ix;
+
+	    if (waiting_unmanaged)
+		goto done;
+
+	    new_umrefc_ix = (umrefc_ix + 1) & 0x1;
+	    tpd->leader_state.umrefc_ix.waiting = umrefc_ix;
+	    tpd->leader_state.chk_next_ix = no_managed;
+	    erts_atomic32_set_nob(&intrnl->misc.data.umrefc_ix.current,
+				  (erts_aint32_t) new_umrefc_ix);
+	    ETHR_MEMBAR(ETHR_StoreLoad);
+	    refc = erts_atomic_read_nob(&intrnl->umrefc[umrefc_ix].refc);
+	    ASSERT(refc >= 0);
+	    waiting_unmanaged = 1;
+	    if (refc != 0)
+		goto done;
+	}
+
+	/* Make progress */
+	current = next;
 
-	    next++;
-	    if (next == ERTS_THR_PRGR_VAL_WAITING)
-		next = 0;
+	next++;
+	if (next == ERTS_THR_PRGR_VAL_WAITING)
+	    next = 0;
 
-	    set_nob(&intrnl->thr[tpd->id].data.current, next);
-	    set_mb(&erts_thr_prgr__.current, current);
-	    tpd->previous.local = next;
-	    tpd->previous.next = next;
-	    tpd->previous.current = current;
+	set_nob(&intrnl->thr[my_ix].data.current, next);
+	set_mb(&erts_thr_prgr__.current, current);
+	tpd->confirmed = next;
+	tpd->leader_state.next = next;
+	tpd->leader_state.current = current;
 
 #if ERTS_THR_PRGR_PRINT_VAL
-	    if (current % 1000 == 0)
-		erts_fprintf(stderr, "%b64u\n", current);
+	if (current % 1000 == 0)
+	    erts_fprintf(stderr, "%b64u\n", current);
 #endif
-	    handle_wakeup_requests(current);
+	handle_wakeup_requests(current);
+
+	if (waiting_unmanaged) {
+	    waiting_unmanaged = 0;
+	    tpd->leader_state.umrefc_ix.waiting = -1;
+	    erts_atomic32_read_band_nob(&intrnl->misc.data.lflgs,
+					~ERTS_THR_PRGR_LFLG_WAITING_UM);
 	}
+	tpd->leader_state.chk_next_ix = 0;
+
+    done:
 
 	if (tpd->active) {
 	    lflgs = erts_atomic32_read_nob(&intrnl->misc.data.lflgs);
@@ -712,20 +784,44 @@ leader_update(ErtsThrPrgrData *tpd)
 		(void) block_thread(tpd);
 	}
 	else {
+	    int force_wakeup_check = 0;
+	    erts_aint32_t set_flags = ERTS_THR_PRGR_LFLG_NO_LEADER;
 	    tpd->leader = 0;
-	    tpd->previous.current = ERTS_THR_PRGR_VAL_WAITING;
+	    tpd->leader_state.current = ERTS_THR_PRGR_VAL_WAITING;
 #if ERTS_THR_PRGR_PRINT_LEADER
 	    erts_fprintf(stderr, "L <- %d\n", tpd->id);
 #endif
 
 	    ERTS_THR_PROGRESS_STATE_DEBUG_SET_LEADER(tpd->id, 0);
 
+	    if (waiting_unmanaged)
+		set_flags |= ERTS_THR_PRGR_LFLG_WAITING_UM;
+
 	    lflgs = erts_atomic32_read_bor_relb(&intrnl->misc.data.lflgs,
-						ERTS_THR_PRGR_LFLG_NO_LEADER);
+						set_flags);
+	    lflgs |= set_flags;
 	    if (lflgs & ERTS_THR_PRGR_LFLG_BLOCK)
 		lflgs = block_thread(tpd);
-	    if (ERTS_THR_PRGR_LFLGS_ACTIVE(lflgs) == 0 && got_sched_wakeups())
+
+	    if (waiting_unmanaged) {
+		/* Need to check umrefc again */
+		ETHR_MEMBAR(ETHR_StoreLoad);
+		refc = erts_atomic_read_nob(&intrnl->umrefc[umrefc_ix].refc);
+		if (refc == 0) {
+		    /* Need to force wakeup check */
+		    force_wakeup_check = 1;
+		}
+	    }
+
+	    if ((force_wakeup_check
+		 || ((lflgs & (ERTS_THR_PRGR_LFLG_NO_LEADER
+			       | ERTS_THR_PRGR_LFLG_WAITING_UM
+			       | ERTS_THR_PRGR_LFLG_ACTIVE_MASK))
+		     == ERTS_THR_PRGR_LFLG_NO_LEADER))
+		&& got_sched_wakeups()) {
+		/* Someone need to make progress */
 		wakeup_managed(0);
+	    }
 	}
     }
 
@@ -744,11 +840,11 @@ update(ErtsThrPrgrData *tpd)
 	erts_aint32_t lflgs;
 	res = 0;
 	val = read_acqb(&erts_thr_prgr__.current);
-	if (tpd->previous.local == val) {
+	if (tpd->confirmed == val) {
 	    val++;
 	    if (val == ERTS_THR_PRGR_VAL_WAITING)
 		val = 0;
-	    tpd->previous.local = val;
+	    tpd->confirmed = val;
 	    set_mb(&intrnl->thr[tpd->id].data.current, val);
 	}
 
@@ -801,12 +897,19 @@ erts_thr_progress_prepare_wait(ErtsSchedulerData *esdp)
 
     block_count_dec();
 
-    tpd->previous.local = ERTS_THR_PRGR_VAL_WAITING;
+    tpd->confirmed = ERTS_THR_PRGR_VAL_WAITING;
     set_mb(&intrnl->thr[tpd->id].data.current, ERTS_THR_PRGR_VAL_WAITING);
 
     lflgs = erts_atomic32_read_nob(&intrnl->misc.data.lflgs);
-    if (ERTS_THR_PRGR_LFLGS_ALL_WAITING(lflgs) && got_sched_wakeups())
-	wakeup_managed(0); /* Someone need to make progress */
+
+    if ((lflgs & (ERTS_THR_PRGR_LFLG_NO_LEADER
+		  | ERTS_THR_PRGR_LFLG_WAITING_UM
+		  | ERTS_THR_PRGR_LFLG_ACTIVE_MASK))
+	== ERTS_THR_PRGR_LFLG_NO_LEADER 
+	&& got_sched_wakeups()) {
+	/* Someone need to make progress */
+	wakeup_managed(0);
+    }
 }
 
 void
@@ -828,7 +931,7 @@ erts_thr_progress_finalize_wait(ErtsSchedulerData *esdp)
 	val++;
 	if (val == ERTS_THR_PRGR_VAL_WAITING)
 	    val = 0;
-	tpd->previous.local = val;
+	tpd->confirmed = val;
 	set_mb(&intrnl->thr[tpd->id].data.current, val);
 	val = read_acqb(&erts_thr_prgr__.current);
 	if (current == val)
@@ -875,6 +978,68 @@ erts_thr_progress_active(ErtsSchedulerData *esdp, int on)
 
 }
 
+static ERTS_INLINE void
+unmanaged_continue(ErtsThrPrgrDelayHandle handle)
+{
+    int umrefc_ix = (int) handle;
+    erts_aint_t refc;
+
+    ASSERT(umrefc_ix == 0 || umrefc_ix == 1);
+    refc = erts_atomic_dec_read_relb(&intrnl->umrefc[umrefc_ix].refc);
+    ASSERT(refc >= 0);
+    if (refc == 0) {
+	erts_aint_t lflgs;
+	ERTS_THR_READ_MEMORY_BARRIER;
+	lflgs = erts_atomic32_read_nob(&intrnl->misc.data.lflgs);
+	if ((lflgs & (ERTS_THR_PRGR_LFLG_NO_LEADER
+		      | ERTS_THR_PRGR_LFLG_WAITING_UM
+		      | ERTS_THR_PRGR_LFLG_ACTIVE_MASK))
+	    == (ERTS_THR_PRGR_LFLG_NO_LEADER|ERTS_THR_PRGR_LFLG_WAITING_UM)
+	    && got_sched_wakeups()) {
+	    /* Others waiting for us... */
+	    wakeup_managed(0);
+	}
+    }
+}
+
+void
+erts_thr_progress_unmanaged_continue__(ErtsThrPrgrDelayHandle handle)
+{
+#ifdef ERTS_ENABLE_LOCK_CHECK
+    ErtsThrPrgrData *tpd = perhaps_thr_prgr_data(NULL);
+    ERTS_LC_ASSERT(tpd && tpd->is_delaying);
+    tpd->is_delaying = 0;
+    return_tmp_thr_prgr_data(tpd);
+#endif
+    ASSERT(!erts_thr_progress_is_managed_thread());
+
+    unmanaged_continue(handle);
+}
+
+ErtsThrPrgrDelayHandle
+erts_thr_progress_unmanaged_delay__(void)
+{
+    int umrefc_ix;
+    ASSERT(!erts_thr_progress_is_managed_thread());
+    umrefc_ix = (int) erts_atomic32_read_acqb(&intrnl->misc.data.umrefc_ix.current);
+    while (1) {
+	int tmp_ix;
+	erts_atomic_inc_acqb(&intrnl->umrefc[umrefc_ix].refc);
+	tmp_ix = (int) erts_atomic32_read_acqb(&intrnl->misc.data.umrefc_ix.current);
+	if (tmp_ix == umrefc_ix)
+	    break;
+	unmanaged_continue(umrefc_ix);
+	umrefc_ix = tmp_ix;
+    }
+#ifdef ERTS_ENABLE_LOCK_CHECK
+    {
+	ErtsThrPrgrData *tpd = tmp_thr_prgr_data(NULL);
+	tpd->is_delaying = 1;
+    }
+#endif
+    return (ErtsThrPrgrDelayHandle) umrefc_ix;
+}
+
 static ERTS_INLINE int
 has_reached_wakeup(ErtsThrPrgrVal wakeup)
 {
@@ -931,7 +1096,7 @@ request_wakeup_managed(ErtsThrPrgrData *tpd, ErtsThrPrgrVal value)
      */
 
     ASSERT(tpd->is_managed);
-    ASSERT(tpd->previous.local != ERTS_THR_PRGR_VAL_WAITING);
+    ASSERT(tpd->confirmed != ERTS_THR_PRGR_VAL_WAITING);
 
     if (has_reached_wakeup(value)) {
 	wakeup_managed(tpd->id);
@@ -946,7 +1111,7 @@ request_wakeup_managed(ErtsThrPrgrData *tpd, ErtsThrPrgrVal value)
 					  tpd->wakeup_request[wix]));
 
 
-    if (tpd->previous.local == value) {
+    if (tpd->confirmed == value) {
 	/*
 	 * We have already confirmed this value. We need to request
 	 * wakeup for a value later than our latest confirmed value in