aboutsummaryrefslogtreecommitdiffstats
path: root/erts/emulator/beam
diff options
context:
space:
mode:
Diffstat (limited to 'erts/emulator/beam')
-rw-r--r--erts/emulator/beam/erl_port.h6
-rw-r--r--erts/emulator/beam/erl_port_task.c66
-rw-r--r--erts/emulator/beam/io.c18
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) {
generated by cgit v1.2.3 (git 2.25.1) at 2024-12-22 17:14:24 +0000