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

* rickard/r16/port-optimizations/OTP-10336:
  Fix driver_monitor_process() ASSERT
  Fix scheduled port link operation
  Fix aborts of port tasks when terminating ports

3 files changed, 68 insertions, 22 deletions

diff --git a/erts/emulator/beam/erl_port.h b/erts/emulator/beam/erl_port.h
index f4d73e716a..fb07f3d5bc 100644
--- a/erts/emulator/beam/erl_port.h
+++ b/erts/emulator/beam/erl_port.h
@@ -430,7 +430,7 @@ ERTS_GLB_INLINE void erts_port_release(Port *);
 ERTS_GLB_INLINE Port *erts_thr_id2port_sflgs(Eterm id, Uint32 invalid_sflgs);
 ERTS_GLB_INLINE void erts_thr_port_release(Port *prt);
 #endif
-ERTS_GLB_INLINE Port *erts_thr_drvport2port_raw(ErlDrvPort);
+ERTS_GLB_INLINE Port *erts_thr_drvport2port_raw(ErlDrvPort, int);
 ERTS_GLB_INLINE Port *erts_drvport2port_raw(ErlDrvPort drvport);
 ERTS_GLB_INLINE Port *erts_drvport2port(ErlDrvPort, erts_aint32_t *);
 ERTS_GLB_INLINE Port *erts_drvportid2port(Eterm);
@@ -622,7 +622,7 @@ erts_thr_port_release(Port *prt)
 #endif
 
 ERTS_GLB_INLINE Port*
-erts_thr_drvport2port_raw(ErlDrvPort drvport)
+erts_thr_drvport2port_raw(ErlDrvPort drvport, int lock_pdl)
 {
 #if ERTS_ENABLE_LOCK_CHECK
     int emu_thread = erts_lc_is_emu_thr();
@@ -631,6 +631,8 @@ erts_thr_drvport2port_raw(ErlDrvPort drvport)
 	return NULL;
     else {
 	Port *prt = (Port *) drvport;
+	if (lock_pdl && prt->port_data_lock)
+	    driver_pdl_lock(prt->port_data_lock);
 #if ERTS_ENABLE_LOCK_CHECK
 	if (!ERTS_IS_CRASH_DUMPING) {
 	    if (emu_thread) {
diff --git a/erts/emulator/beam/erl_port_task.c b/erts/emulator/beam/erl_port_task.c
index b661c26036..8ceadcdb8c 100644
--- a/erts/emulator/beam/erl_port_task.c
+++ b/erts/emulator/beam/erl_port_task.c
@@ -120,7 +120,9 @@ struct ErtsPortTaskBusyCallerTable_ {
 };
 
 
-static void begin_port_cleanup(Port *pp, ErtsPortTask **execq);
+static void begin_port_cleanup(Port *pp,
+			       ErtsPortTask **execq,
+			       int *processing_busy_q_p);
 
 ERTS_SCHED_PREF_QUICK_ALLOC_IMPL(port_task,
 				 ErtsPortTask,
@@ -1525,7 +1527,7 @@ erts_port_task_free_port(Port *pp)
 	erts_resume_processes(suspended);
 
     if (!(flags & (ERTS_PTS_FLG_IN_RUNQ|ERTS_PTS_FLG_EXEC)))
-	begin_port_cleanup(pp, NULL);
+	begin_port_cleanup(pp, NULL, NULL);
 }
 
 /*
@@ -1681,8 +1683,7 @@ erts_port_task_execute(ErtsRunQueue *runq, Port **curr_port_pp)
     begin_handle_tasks:
 	if (state & ERTS_PORT_SFLG_FREE) {
 	    reds += ERTS_PORT_REDS_FREE;
-
-	    begin_port_cleanup(pp, &execq);
+	    begin_port_cleanup(pp, &execq, &processing_busy_q);
 
 	    break;
 	}
@@ -1773,22 +1774,66 @@ release_port(void *vport)
 #endif
 
 static void
-begin_port_cleanup(Port *pp, ErtsPortTask **execqp)
+begin_port_cleanup(Port *pp, ErtsPortTask **execqp, int *processing_busy_q_p)
 {
     int i, max;
-    ErtsPortTask *qs[2];
+    ErtsPortTaskBusyCallerTable *tabp;
+    ErtsPortTask *qs[3];
 
     ERTS_SMP_LC_ASSERT(erts_lc_is_port_locked(pp));
 
 
     /*
-     * Handle remaining tasks...
+     * Abort remaining tasks...
+     *
+     * We want to process queues in the following order in order
+     * to preserve signal ordering guarantees:
+     *  1. Local busy queue
+     *  2. Local queue
+     *  3. In queue
      */
 
     max = 0;
-    if (execqp && *execqp) {
-	qs[max++] = *execqp;
+    if (!execqp) {
+	if (pp->sched.taskq.local.busy.first)
+	    qs[max++] = pp->sched.taskq.local.busy.first;
+	if (pp->sched.taskq.local.first)
+	    qs[max++] = pp->sched.taskq.local.first;
+    }
+    else {
+	if (*processing_busy_q_p) {
+	    if (*execqp)
+		qs[max++] = *execqp;
+	    if (pp->sched.taskq.local.first)
+		qs[max++] = pp->sched.taskq.local.first;
+	}
+	else {
+	    if (pp->sched.taskq.local.busy.first)
+		qs[max++] = pp->sched.taskq.local.busy.first;
+	    if (*execqp)
+		qs[max++] = *execqp;
+	}
 	*execqp = NULL;
+	*processing_busy_q_p = 0;
+    }
+    pp->sched.taskq.local.busy.first = NULL;
+    pp->sched.taskq.local.busy.last = NULL;
+    pp->sched.taskq.local.first = NULL;
+    tabp = pp->sched.taskq.local.busy.table;
+    if (tabp) {
+	int bix;
+	for (bix = 0; bix < ERTS_PORT_TASK_BUSY_CALLER_TABLE_BUCKETS; bix++) {
+	    ErtsPortTaskBusyCaller *bcp = tabp->bucket[bix];
+	    while (bcp) {
+		ErtsPortTaskBusyCaller *free_bcp = bcp;
+		bcp = bcp->next;
+		if (free_bcp != &tabp->pre_alloc_busy_caller)
+		    erts_free(ERTS_ALC_T_BUSY_CALLER, free_bcp);
+	    }
+	}
+
+	busy_caller_table_free(tabp);
+	pp->sched.taskq.local.busy.table = NULL;
     }
 
     erts_port_task_sched_lock(&pp->sched);
@@ -1875,7 +1920,8 @@ begin_port_cleanup(Port *pp, ErtsPortTask **execqp)
     }
 
     erts_smp_atomic32_read_band_nob(&pp->sched.flags,
-				    ~ERTS_PTS_FLG_HAVE_TASKS);
+				    ~(ERTS_PTS_FLG_HAVE_BUSY_TASKS
+				      |ERTS_PTS_FLG_HAVE_TASKS));
 
     /*
      * Schedule cleanup of port structure...
diff --git a/erts/emulator/beam/io.c b/erts/emulator/beam/io.c
index be094862d4..e466f0e299 100644
--- a/erts/emulator/beam/io.c
+++ b/erts/emulator/beam/io.c
@@ -82,9 +82,11 @@ static void pdl_init(void);
 #ifdef ERTS_SMP
 static void driver_monitor_lock_pdl(Port *p);
 static void driver_monitor_unlock_pdl(Port *p);
+#define DRV_MONITOR_LOOKUP_PORT_LOCK_PDL(Port) erts_thr_drvport2port_raw((Port), 1)
 #define DRV_MONITOR_LOCK_PDL(Port) driver_monitor_lock_pdl(Port)
 #define DRV_MONITOR_UNLOCK_PDL(Port) driver_monitor_unlock_pdl(Port)
 #else
+#define DRV_MONITOR_LOOKUP_PORT_LOCK_PDL(Port) erts_thr_drvport2port_raw((Port), 0)
 #define DRV_MONITOR_LOCK_PDL(Port) /* nothing */
 #define DRV_MONITOR_UNLOCK_PDL(Port) /* nothing */
 #endif
@@ -95,7 +97,7 @@ static void driver_monitor_unlock_pdl(Port *p);
 static ERTS_INLINE ErlIOQueue*
 drvport2ioq(ErlDrvPort drvport)
 {
-    Port *prt = erts_thr_drvport2port_raw(drvport);
+    Port *prt = erts_thr_drvport2port_raw(drvport, 0);
     erts_aint32_t state = erts_atomic32_read_nob(&prt->state);
     if (state & ERTS_PORT_SFLGS_INVALID_DRIVER_LOOKUP)
 	return NULL;
@@ -2534,6 +2536,8 @@ port_link_failure(Eterm port_id, Eterm linker)
 		if (IS_TRACED_FL(rp, F_TRACE_PROCS))
 		    trace_proc(NULL, rp, am_getting_unlinked, port_id);
 	    }
+	    if (rp_locks)
+		erts_smp_proc_unlock(rp, rp_locks);
 	}
     }
 }
@@ -6737,9 +6741,7 @@ int driver_monitor_process(ErlDrvPort drvport,
     ErtsSchedulerData *sched = erts_get_scheduler_data();
 #endif
 
-    prt = erts_thr_drvport2port_raw(drvport);
-
-    DRV_MONITOR_LOCK_PDL(prt);
+    prt = DRV_MONITOR_LOOKUP_PORT_LOCK_PDL(drvport);
 
     state = erts_atomic32_read_nob(&prt->state);
 
@@ -6818,9 +6820,7 @@ int driver_demonitor_process(ErlDrvPort drvport,
     ErtsSchedulerData *sched = erts_get_scheduler_data();
 #endif
 
-    prt = erts_thr_drvport2port_raw(drvport);
-
-    DRV_MONITOR_LOCK_PDL(prt);
+    prt = DRV_MONITOR_LOOKUP_PORT_LOCK_PDL(drvport);
 
     state = erts_atomic32_read_nob(&prt->state);
 
@@ -6881,9 +6881,7 @@ ErlDrvTermData driver_get_monitored_process(ErlDrvPort drvport,
     ErtsSchedulerData *sched = erts_get_scheduler_data();
 #endif
 
-    prt = erts_thr_drvport2port_raw(drvport);
-
-    DRV_MONITOR_LOCK_PDL(prt);
+    prt = DRV_MONITOR_LOOKUP_PORT_LOCK_PDL(drvport);
 
     state = erts_atomic32_read_nob(&prt->state);
     if (state & ERTS_PORT_SFLGS_INVALID_DRIVER_LOOKUP) {