From 50cb7c24f061fd3d7df5970d8202f47c470a4047 Mon Sep 17 00:00:00 2001 From: Rickard Green Date: Wed, 15 Aug 2012 01:28:55 +0200 Subject: Use ptab functionality also for ports --- erts/emulator/beam/atom.names | 1 + erts/emulator/beam/beam_emu.c | 5 +- erts/emulator/beam/bif.c | 100 +-- erts/emulator/beam/break.c | 9 +- erts/emulator/beam/dist.c | 25 +- erts/emulator/beam/dist.h | 2 +- erts/emulator/beam/erl_alloc.c | 5 +- erts/emulator/beam/erl_alloc.h | 2 + erts/emulator/beam/erl_alloc.types | 6 +- erts/emulator/beam/erl_async.c | 9 + erts/emulator/beam/erl_bif_ddll.c | 45 +- erts/emulator/beam/erl_bif_info.c | 30 +- erts/emulator/beam/erl_bif_port.c | 78 ++- erts/emulator/beam/erl_bif_trace.c | 59 +- erts/emulator/beam/erl_db_util.c | 11 +- erts/emulator/beam/erl_driver.h | 2 +- erts/emulator/beam/erl_lock_check.c | 17 +- erts/emulator/beam/erl_lock_check.h | 1 + erts/emulator/beam/erl_monitors.c | 7 +- erts/emulator/beam/erl_node_container_utils.h | 11 +- erts/emulator/beam/erl_node_tables.c | 36 +- erts/emulator/beam/erl_node_tables.h | 3 +- erts/emulator/beam/erl_port.h | 421 +++++++++---- erts/emulator/beam/erl_port_task.c | 100 ++- erts/emulator/beam/erl_process.c | 61 +- erts/emulator/beam/erl_process.h | 12 +- erts/emulator/beam/erl_ptab.c | 20 +- erts/emulator/beam/erl_ptab.h | 4 - erts/emulator/beam/erl_sys_driver.h | 1 - erts/emulator/beam/erl_trace.c | 8 +- erts/emulator/beam/global.h | 13 +- erts/emulator/beam/io.c | 876 +++++++++++--------------- erts/emulator/beam/register.c | 24 +- erts/emulator/beam/register.h | 19 +- 34 files changed, 1018 insertions(+), 1005 deletions(-) (limited to 'erts/emulator/beam') diff --git a/erts/emulator/beam/atom.names b/erts/emulator/beam/atom.names index 201719507f..d6d8633f6f 100644 --- a/erts/emulator/beam/atom.names +++ b/erts/emulator/beam/atom.names @@ -419,6 +419,7 @@ atom pid atom port atom ports atom port_count +atom port_limit atom print atom priority atom private diff --git a/erts/emulator/beam/beam_emu.c b/erts/emulator/beam/beam_emu.c index 3fb2fbd01c..d8ccbacc26 100644 --- a/erts/emulator/beam/beam_emu.c +++ b/erts/emulator/beam/beam_emu.c @@ -1660,6 +1660,7 @@ void process_main(void) reg[0] = r(0); result = erl_send(c_p, r(0), x(1)); PreFetch(0, next); + ERTS_VERIFY_UNUSED_TEMP_ALLOC(c_p); ERTS_SMP_REQ_PROC_MAIN_LOCK(c_p); PROCESS_MAIN_CHK_LOCKS(c_p); if (c_p->mbuf || MSO(c_p).overhead >= BIN_VHEAP_SZ(c_p)) { @@ -2636,6 +2637,7 @@ void process_main(void) reg[0] = r(0); result = (*bf)(c_p, reg, I); ASSERT(!ERTS_PROC_IS_EXITING(c_p) || is_non_value(result)); + ERTS_VERIFY_UNUSED_TEMP_ALLOC(c_p); ERTS_HOLE_CHECK(c_p); ERTS_SMP_REQ_PROC_MAIN_LOCK(c_p); PROCESS_MAIN_CHK_LOCKS(c_p); @@ -3370,7 +3372,6 @@ void process_main(void) PROCESS_MAIN_CHK_LOCKS(c_p); bif_nif_arity = I[-1]; ERTS_SMP_UNREQ_PROC_MAIN_LOCK(c_p); - ERTS_VERIFY_UNUSED_TEMP_ALLOC(c_p); ASSERT(!ERTS_PROC_IS_EXITING(c_p)); { @@ -3415,7 +3416,6 @@ void process_main(void) bif_nif_arity = I[-1]; ASSERT(bif_nif_arity <= 3); ERTS_SMP_UNREQ_PROC_MAIN_LOCK(c_p); - ERTS_VERIFY_UNUSED_TEMP_ALLOC(c_p); reg[0] = r(0); { Eterm (*bf)(Process*, Eterm*, BeamInstr*) = vbf; @@ -4646,7 +4646,6 @@ void process_main(void) ERTS_SMP_UNREQ_PROC_MAIN_LOCK(c_p); flags = erts_call_trace(c_p, ep->code, ep->match_prog_set, reg, 0, &ERTS_TRACER_PROC(c_p)); - ERTS_VERIFY_UNUSED_TEMP_ALLOC(c_p); ERTS_SMP_REQ_PROC_MAIN_LOCK(c_p); PROCESS_MAIN_CHK_LOCKS(c_p); ASSERT(!ERTS_PROC_IS_EXITING(c_p)); diff --git a/erts/emulator/beam/bif.c b/erts/emulator/beam/bif.c index 84513c1b0e..5ea441ad81 100644 --- a/erts/emulator/beam/bif.c +++ b/erts/emulator/beam/bif.c @@ -37,6 +37,8 @@ #include "erl_db_util.h" #include "register.h" #include "erl_thr_progress.h" +#define ERTS_PTAB_WANT_BIF_IMPL__ +#include "erl_ptab.h" static Export* flush_monitor_message_trap = NULL; static Export* set_cpu_topology_trap = NULL; @@ -155,7 +157,10 @@ BIF_RETTYPE link_1(BIF_ALIST_1) } if (is_internal_port(BIF_ARG_1)) { - Port *pt = erts_id2port(BIF_ARG_1, BIF_P, ERTS_PROC_LOCK_MAIN); + Port *pt = erts_id2port_sflgs(BIF_ARG_1, + BIF_P, + ERTS_PROC_LOCK_MAIN, + ERTS_PORT_SFLGS_INVALID_LOOKUP); if (!pt) { goto res_no_proc; } @@ -167,7 +172,7 @@ BIF_RETTYPE link_1(BIF_ALIST_1) /* else: already linked */ erts_smp_proc_unlock(BIF_P, ERTS_PROC_LOCK_LINK); - erts_smp_port_unlock(pt); + erts_port_release(pt); BIF_RET(am_true); } else if (is_external_port(BIF_ARG_1) @@ -948,7 +953,7 @@ BIF_RETTYPE unlink_1(BIF_ALIST_1) #ifdef ERTS_SMP if (ERTS_PROC_PENDING_EXIT(BIF_P)) { if (pt) - erts_smp_port_unlock(pt); + erts_port_release(pt); goto handle_pending_exit; } #endif @@ -959,7 +964,7 @@ BIF_RETTYPE unlink_1(BIF_ALIST_1) if (pt) { rl = erts_remove_link(&ERTS_P_LINKS(pt), BIF_P->common.id); - erts_smp_port_unlock(pt); + erts_port_release(pt); if (rl) erts_destroy_link(rl); } @@ -1341,7 +1346,7 @@ BIF_RETTYPE exit_2(BIF_ALIST_2) if (is_internal_port(BIF_ARG_1)) { Port *prt; erts_smp_proc_unlock(BIF_P, ERTS_PROC_LOCK_MAIN); - prt = erts_id2port(BIF_ARG_1, NULL, 0); + prt = erts_id2port(BIF_ARG_1); if (prt) { erts_do_exit_port(prt, BIF_P->common.id, BIF_ARG_2); erts_port_release(prt); @@ -1894,7 +1899,10 @@ do_send(Process *p, Eterm to, Eterm msg, int suspend) { if (erts_system_profile_flags.runnable_procs && erts_system_profile_flags.exclusive) { profile_runnable_proc(p, am_inactive); } - pt = erts_id2port(to, p, ERTS_PROC_LOCK_MAIN); + pt = erts_id2port_sflgs(to, + p, + ERTS_PROC_LOCK_MAIN, + ERTS_PORT_SFLGS_INVALID_LOOKUP); port_common: ERTS_SMP_LC_ASSERT(!pt || erts_lc_is_port_locked(pt)); @@ -1908,7 +1916,7 @@ do_send(Process *p, Eterm to, Eterm msg, int suspend) { profile_runnable_port(pt, am_active); } - state = erts_smp_atomic32_read_nob(&pt->state); + state = erts_atomic32_read_nob(&pt->state); /* XXX let port_command handle the busy stuff !!! */ if (state & ERTS_PORT_SFLG_PORT_BUSY) { if (suspend) { @@ -3510,75 +3518,25 @@ BIF_RETTYPE garbage_collect_message_area_0(BIF_ALIST_0) #endif } + /**********************************************************************/ -/* Return a list of active ports */ +/* + * The erlang:processes/0 BIF. + */ -BIF_RETTYPE ports_0(BIF_ALIST_0) +BIF_RETTYPE processes_0(BIF_ALIST_0) { - Eterm res = NIL; - Eterm* port_buf = erts_alloc(ERTS_ALC_T_TMP, - sizeof(Eterm)*erts_max_ports); - Eterm* pp = port_buf; - Eterm* dead_ports; - int alive, dead; - Uint32 next_ss; - int i; - - /* To get a consistent snapshot... - * We add alive ports from start of the buffer - * while dying ports are added from the other end by the killing threads. - */ - - erts_smp_mtx_lock(&ports_snapshot_mtx); /* One snapshot at a time */ - - erts_smp_atomic_set_nob(&erts_dead_ports_ptr, - (erts_aint_t) (port_buf + erts_max_ports)); - - next_ss = erts_smp_atomic32_inc_read_relb(&erts_ports_snapshot); - - for (i = erts_max_ports-1; i >= 0; i--) { - Port* prt = &erts_port[i]; - erts_aint32_t state = erts_smp_atomic32_read_acqb(&prt->state); - if (!(state & ERTS_PORT_SFLGS_DEAD)) { - erts_smp_port_minor_lock(prt); - state = erts_smp_atomic32_read_nob(&prt->state); - if (!(state & ERTS_PORT_SFLGS_DEAD) && prt->snapshot != next_ss) { - ASSERT(prt->snapshot == next_ss - 1); - *pp++ = prt->common.id; - prt->snapshot = next_ss; /* Consumed by this snapshot */ - } - erts_smp_port_minor_unlock(prt); - } - } - - dead_ports = (Eterm*)erts_smp_atomic_xchg_nob(&erts_dead_ports_ptr, - (erts_aint_t) NULL); - erts_smp_mtx_unlock(&ports_snapshot_mtx); - - ASSERT(pp <= dead_ports); - - alive = pp - port_buf; - dead = port_buf + erts_max_ports - dead_ports; - - ASSERT((alive+dead) <= erts_max_ports); - - if (alive+dead > 0) { - erts_aint_t i; - Eterm *hp = HAlloc(BIF_P, (alive+dead)*2); - - for (i = 0; i < alive; i++) { - res = CONS(hp, port_buf[i], res); - hp += 2; - } - for (i = 0; i < dead; i++) { - res = CONS(hp, dead_ports[i], res); - hp += 2; - } - } + return erts_ptab_list(BIF_P, &erts_proc); +} - erts_free(ERTS_ALC_T_TMP, port_buf); +/**********************************************************************/ +/* + * The erlang:ports/0 BIF. + */ - BIF_RET(res); +BIF_RETTYPE ports_0(BIF_ALIST_0) +{ + return erts_ptab_list(BIF_P, &erts_port); } /**********************************************************************/ diff --git a/erts/emulator/beam/break.c b/erts/emulator/beam/break.c index fba734623a..ad05316b39 100644 --- a/erts/emulator/beam/break.c +++ b/erts/emulator/beam/break.c @@ -61,9 +61,12 @@ extern char* erts_system_version[]; static void port_info(int to, void *to_arg) { - int i; - for (i = 0; i < erts_max_ports; i++) - print_port_info(to, to_arg, i); + int i, max = erts_ptab_max(&erts_port); + for (i = 0; i < max; i++) { + Port *p = erts_pix2port(i); + if (p) + print_port_info(p, to, to_arg); + } } void diff --git a/erts/emulator/beam/dist.c b/erts/emulator/beam/dist.c index 7aef26aa21..13715034cb 100644 --- a/erts/emulator/beam/dist.c +++ b/erts/emulator/beam/dist.c @@ -144,7 +144,7 @@ create_cache(DistEntry *dep) ERTS_SMP_LC_ASSERT( is_internal_port(dep->cid) - && erts_lc_is_port_locked(&erts_port[internal_port_index(dep->cid)])); + && erts_lc_is_port_locked(erts_port_lookup_raw(dep->cid))); ASSERT(!dep->cache); dep->cache = cp = (ErtsAtomCache*) erts_alloc(ERTS_ALC_T_DCACHE, @@ -421,9 +421,9 @@ int erts_do_net_exits(DistEntry *dep, Eterm reason) ASSERT(is_internal_port(prt_id)); erts_smp_rwmtx_rwunlock(&erts_dist_table_rwmtx); - prt = erts_id2port(prt_id, NULL, 0); + prt = erts_id2port(prt_id); if (prt) { - ASSERT(erts_smp_atomic32_read_nob(&prt->state) + ASSERT(erts_atomic32_read_nob(&prt->state) & ERTS_PORT_SFLG_DISTRIBUTION); ASSERT(prt->dist_entry); /* will call do_net_exists !!! */ @@ -455,7 +455,7 @@ int erts_do_net_exits(DistEntry *dep, Eterm reason) erts_smp_de_rwlock(dep); ERTS_SMP_LC_ASSERT(is_internal_port(dep->cid) - && erts_lc_is_port_locked(&erts_port[internal_port_index(dep->cid)])); + && erts_lc_is_port_locked(erts_port_lookup_raw(dep->cid))); if (erts_port_task_is_scheduled(&dep->dist_cmd)) erts_port_task_abort(dep->cid, &dep->dist_cmd); @@ -1957,7 +1957,7 @@ erts_dist_command(Port *prt, int reds_limit) dep->finalized_out_queue.first = NULL; dep->finalized_out_queue.last = NULL; - state = erts_smp_atomic32_read_nob(&prt->state); + state = erts_atomic32_read_nob(&prt->state); if (reds > reds_limit) goto preempted; @@ -1978,7 +1978,7 @@ erts_dist_command(Port *prt, int reds_limit) obufsize += size_obuf(fob); foq.first = foq.first->next; free_dist_obuf(fob); - state = erts_smp_atomic32_read_nob(&prt->state); + state = erts_atomic32_read_nob(&prt->state); preempt = reds > reds_limit || (state & ERTS_PORT_SFLGS_DEAD); if (state & ERTS_PORT_SFLG_PORT_BUSY) break; @@ -2060,7 +2060,7 @@ erts_dist_command(Port *prt, int reds_limit) obufsize += size_obuf(fob); oq.first = oq.first->next; free_dist_obuf(fob); - state = erts_smp_atomic32_read_nob(&prt->state); + state = erts_atomic32_read_nob(&prt->state); preempt = reds > reds_limit || (state & ERTS_PORT_SFLGS_DEAD); if ((state & ERTS_PORT_SFLG_PORT_BUSY) && oq.first && !preempt) goto finalize_only; @@ -2580,10 +2580,13 @@ BIF_RETTYPE setnode_3(BIF_ALIST_3) else if (!dep) goto system_limit; /* Should never happen!!! */ - pp = erts_id2port(BIF_ARG_2, BIF_P, ERTS_PROC_LOCK_MAIN); + pp = erts_id2port_sflgs(BIF_ARG_2, + BIF_P, + ERTS_PROC_LOCK_MAIN, + ERTS_PORT_SFLGS_INVALID_LOOKUP); erts_smp_de_rwlock(dep); - if (!pp || (erts_smp_atomic32_read_nob(&pp->state) + if (!pp || (erts_atomic32_read_nob(&pp->state) & ERTS_PORT_SFLG_EXITING)) goto badarg; @@ -2613,7 +2616,7 @@ BIF_RETTYPE setnode_3(BIF_ALIST_3) if (pp->dist_entry || is_not_nil(dep->cid)) goto badarg; - erts_smp_atomic32_read_bor_nob(&pp->state, ERTS_PORT_SFLG_DISTRIBUTION); + erts_atomic32_read_bor_nob(&pp->state, ERTS_PORT_SFLG_DISTRIBUTION); pp->dist_entry = dep; @@ -2658,7 +2661,7 @@ BIF_RETTYPE setnode_3(BIF_ALIST_3) } if (pp) - erts_smp_port_unlock(pp); + erts_port_release(pp); return ret; diff --git a/erts/emulator/beam/dist.h b/erts/emulator/beam/dist.h index 98c069da8a..7de8786c4a 100644 --- a/erts/emulator/beam/dist.h +++ b/erts/emulator/beam/dist.h @@ -187,7 +187,7 @@ void erts_schedule_dist_command(Port *prt, DistEntry *dist_entry) if (prt) { ERTS_SMP_LC_ASSERT(erts_lc_is_port_locked(prt)); - ASSERT((erts_smp_atomic32_read_nob(&prt->state) + ASSERT((erts_atomic32_read_nob(&prt->state) & ERTS_PORT_SFLGS_DEAD) == 0); ASSERT(prt->dist_entry); diff --git a/erts/emulator/beam/erl_alloc.c b/erts/emulator/beam/erl_alloc.c index d0c0a13400..e8a9c11034 100644 --- a/erts/emulator/beam/erl_alloc.c +++ b/erts/emulator/beam/erl_alloc.c @@ -2273,6 +2273,7 @@ erts_allocated_areas(int *print_to_p, void *print_to_arg, void *proc) int i, length; Uint reserved_atom_space, atom_space; int max_processes = erts_ptab_max(&erts_proc); + int max_ports = erts_ptab_max(&erts_port); if (proc) { ERTS_SMP_LC_ASSERT(ERTS_PROC_LOCK_MAIN @@ -2303,8 +2304,8 @@ erts_allocated_areas(int *print_to_p, void *print_to_arg, void *proc) values[i].arity = 2; values[i].name = "static"; - values[i].ui[0] = - erts_max_ports*sizeof(Port) /* Port table */ + values[i].ui[0] = + max_ports*sizeof(Port) /* Port table */ + erts_timer_wheel_memory_size() /* Timer wheel */ #ifdef SYS_TMP_BUF_SIZE + SYS_TMP_BUF_SIZE /* tmp_buf in sys on vxworks & ose */ diff --git a/erts/emulator/beam/erl_alloc.h b/erts/emulator/beam/erl_alloc.h index e475f9d8a2..ba5ec9c367 100644 --- a/erts/emulator/beam/erl_alloc.h +++ b/erts/emulator/beam/erl_alloc.h @@ -267,6 +267,8 @@ typedef void (*erts_alloc_verify_func_t)(Allctr_t *); erts_alloc_verify_func_t erts_alloc_get_verify_unused_temp_alloc(Allctr_t **allctr); +#define ERTS_ALC_DATA_ALIGN_SIZE(SZ) \ + (((((SZ) - 1) / 8) + 1) * 8) #define ERTS_ALC_CACHE_LINE_ALIGN_SIZE(SZ) \ (((((SZ) - 1) / ERTS_CACHE_LINE_SIZE) + 1) * ERTS_CACHE_LINE_SIZE) diff --git a/erts/emulator/beam/erl_alloc.types b/erts/emulator/beam/erl_alloc.types index 4ad0c41b50..92ba15214e 100644 --- a/erts/emulator/beam/erl_alloc.types +++ b/erts/emulator/beam/erl_alloc.types @@ -146,6 +146,7 @@ class SYSTEM system_data type SBMBC SBMBC SYSTEM small_block_mbc type PROC FIXED_SIZE PROCESSES proc +type PORT DRIVER SYSTEM port type ATOM LONG_LIVED ATOM atom_entry type MODULE LONG_LIVED CODE module_entry type REG_PROC STANDARD PROCESSES reg_proc @@ -188,7 +189,7 @@ type PORT_TABLE LONG_LIVED SYSTEM port_tab type TIMER_WHEEL LONG_LIVED SYSTEM timer_wheel type DRV DRIVER SYSTEM drv_internal type DRV_BINARY BINARY BINARIES drv_binary -type DRIVER STANDARD SYSTEM driver +type DRIVER DRIVER SYSTEM driver type NIF DRIVER SYSTEM nif_internal type BINARY BINARY BINARIES binary type NBIF_TABLE SYSTEM SYSTEM nbif_tab @@ -196,7 +197,6 @@ type ARG_REG STANDARD PROCESSES arg_reg type PROC_DICT STANDARD PROCESSES proc_dict type CALLS_BUF STANDARD PROCESSES calls_buf type BPD STANDARD SYSTEM bpd -type PORT_NAME STANDARD SYSTEM port_name type LINEBUF STANDARD SYSTEM line_buf type IOQ STANDARD SYSTEM io_queue type BITS_BUF STANDARD SYSTEM bits_buf @@ -236,7 +236,7 @@ type PORT_TASK SHORT_LIVED SYSTEM port_task type PORT_TASKQ SHORT_LIVED SYSTEM port_task_queue type MISC_OP_LIST SHORT_LIVED SYSTEM misc_op_list type PORT_NAMES SHORT_LIVED SYSTEM port_names -type PORT_DATA_LOCK STANDARD SYSTEM port_data_lock +type PORT_DATA_LOCK DRIVER SYSTEM port_data_lock type NODES_MON STANDARD PROCESSES nodes_monitor type PTAB_LIST_DEL SHORT_LIVED PROCESSES ptab_list_deleted_el type PTAB_LIST_CNKI SHORT_LIVED PROCESSES ptab_list_chunk_info diff --git a/erts/emulator/beam/erl_async.c b/erts/emulator/beam/erl_async.c index 8d3bea82c1..d8adaedc91 100644 --- a/erts/emulator/beam/erl_async.c +++ b/erts/emulator/beam/erl_async.c @@ -375,10 +375,15 @@ static ERTS_INLINE ErtsAsync *async_get(ErtsThrQ_t *q, static ERTS_INLINE void call_async_ready(ErtsAsync *a) { +#if ERTS_USE_ASYNC_READY_Q Port *p = erts_id2port_sflgs(a->port, NULL, 0, ERTS_PORT_SFLGS_INVALID_DRIVER_LOOKUP); +#else + Port *p = erts_thr_id2port_sflgs(a->port, + ERTS_PORT_SFLGS_INVALID_DRIVER_LOOKUP); +#endif if (!p) { if (a->async_free) a->async_free(a->async_data); @@ -388,7 +393,11 @@ static ERTS_INLINE void call_async_ready(ErtsAsync *a) if (a->async_free) a->async_free(a->async_data); } +#if ERTS_USE_ASYNC_READY_Q erts_port_release(p); +#else + erts_thr_port_release(p); +#endif } if (a->hndl) erts_ddll_dereference_driver(a->hndl); diff --git a/erts/emulator/beam/erl_bif_ddll.c b/erts/emulator/beam/erl_bif_ddll.c index a120a3e5f2..e18b1f0159 100644 --- a/erts/emulator/beam/erl_bif_ddll.c +++ b/erts/emulator/beam/erl_bif_ddll.c @@ -117,48 +117,27 @@ static void ddll_no_more_references(void *vdh); static void kill_ports_driver_unloaded(DE_Handle *dh) { - int ix; + int ix, max = erts_ptab_max(&erts_port); - for (ix = 0; ix < erts_max_ports; ix++) { - Port* prt = &erts_port[ix]; + for (ix = 0; ix < max; ix++) { erts_aint32_t state; - state = erts_smp_atomic32_read_acqb(&prt->state); - if (state & FREE_PORT_FLAGS) + Port* prt = erts_pix2port(ix); + if (!prt) continue; - erts_smp_port_minor_lock(prt); + ERTS_SMP_DATA_DEPENDENCY_READ_MEMORY_BARRIER; - if (prt->drv_ptr->handle != dh) { - erts_smp_port_minor_unlock(prt); + state = erts_atomic32_read_nob(&prt->state); + if (state & FREE_PORT_FLAGS) continue; - } - erts_smp_atomic32_inc_nob(&prt->common.refc); - erts_smp_port_minor_unlock(prt); - - state = erts_smp_atomic32_read_acqb(&prt->state); - if (!(state & FREE_PORT_FLAGS)) { -#if DDLL_SMP - if (state & ERTS_PORT_SFLG_INITIALIZING) { - /* Extremely rare busy wait */ - do { - ERTS_SPIN_BODY; - state = erts_smp_atomic32_read_acqb(&prt->state); - } while (state & ERTS_PORT_SFLG_INITIALIZING); - } + erts_smp_port_lock(prt); - erts_smp_mtx_lock(prt->lock); + state = erts_atomic32_read_nob(&prt->state); + if (!(state & ERTS_PORT_SFLGS_DEAD) && prt->drv_ptr->handle == dh) + driver_failure_atom((ErlDrvPort) prt, "driver_unloaded"); - if (prt->drv_ptr->handle == dh) { - state = erts_smp_atomic32_read_acqb(&prt->state); - if (!(state & ERTS_PORT_SFLGS_DEAD)) - driver_failure_atom(ix, "driver_unloaded"); - } -#else - driver_failure_atom(ix, "driver_unloaded"); -#endif - erts_port_release(prt); - } + erts_port_release(prt); } } diff --git a/erts/emulator/beam/erl_bif_info.c b/erts/emulator/beam/erl_bif_info.c index 847a0c33e9..0586a89041 100644 --- a/erts/emulator/beam/erl_bif_info.c +++ b/erts/emulator/beam/erl_bif_info.c @@ -2189,6 +2189,10 @@ BIF_RETTYPE system_info_1(BIF_ALIST_1) BIF_RET(make_small(erts_ptab_count(&erts_proc))); } else if (BIF_ARG_1 == am_process_limit) { BIF_RET(make_small(erts_ptab_max(&erts_proc))); + } else if (BIF_ARG_1 == am_port_count) { + BIF_RET(make_small(erts_ptab_count(&erts_port))); + } else if (BIF_ARG_1 == am_port_limit) { + BIF_RET(make_small(erts_ptab_max(&erts_port))); } else if (BIF_ARG_1 == am_info || BIF_ARG_1 == am_procs || BIF_ARG_1 == am_loaded @@ -2840,7 +2844,10 @@ static BIF_RETTYPE port_info(Process* p, Eterm portid, Eterm item) int count; if (is_internal_port(portid)) - prt = erts_id2port(portid, p, ERTS_PROC_LOCK_MAIN); + prt = erts_id2port_sflgs(portid, + p, + ERTS_PROC_LOCK_MAIN, + ERTS_PORT_SFLGS_INVALID_LOOKUP); else if (is_atom(portid)) erts_whereis_name(p, ERTS_PROC_LOCK_MAIN, portid, NULL, 0, 0, &prt); @@ -2975,7 +2982,7 @@ static BIF_RETTYPE port_info(Process* p, Eterm portid, Eterm item) #ifndef ERTS_SMP res = am_false; #else - if (erts_smp_atomic32_read_nob(&prt->state) + if (erts_atomic32_read_nob(&prt->state) & ERTS_PORT_SFLG_PORT_SPECIFIC_LOCK) { DECL_AM(port_level); ASSERT(prt->drv_ptr->flags @@ -2999,7 +3006,7 @@ static BIF_RETTYPE port_info(Process* p, Eterm portid, Eterm item) done: - erts_smp_port_unlock(prt); + erts_port_release(prt); return ret; } @@ -3357,9 +3364,8 @@ BIF_RETTYPE erts_debug_get_internal_state_1(BIF_ALIST_1) Eterm res; if (ERTS_IS_ATOM_STR("next_pid", BIF_ARG_1)) res = erts_ptab_test_next_id(&erts_proc, 0, 0); - else { - res = erts_test_next_port(0, 0); - } + else + res = erts_ptab_test_next_id(&erts_port, 0, 0); if (res < 0) BIF_RET(am_false); BIF_RET(erts_make_integer(res, BIF_P)); @@ -3466,11 +3472,14 @@ BIF_RETTYPE erts_debug_get_internal_state_1(BIF_ALIST_1) } else if(is_internal_port(tp[2])) { Eterm res; - Port *p = erts_id2port(tp[2], BIF_P, ERTS_PROC_LOCK_MAIN); + Port *p = erts_id2port_sflgs(tp[2], + BIF_P, + ERTS_PROC_LOCK_MAIN, + ERTS_PORT_SFLGS_INVALID_LOOKUP); if(!p) BIF_RET(am_undefined); res = make_link_list(BIF_P, ERTS_P_LINKS(p), NIL); - erts_smp_port_unlock(p); + erts_port_release(p); BIF_RET(res); } else if(is_node_name_atom(tp[2])) { @@ -3714,9 +3723,8 @@ BIF_RETTYPE erts_debug_set_internal_state_2(BIF_ALIST_2) if (ERTS_IS_ATOM_STR("next_pid", BIF_ARG_1)) res = erts_ptab_test_next_id(&erts_proc, 1, next); - else { - res = erts_test_next_port(1, next); - } + else + res = erts_ptab_test_next_id(&erts_port, 1, next); if (res < 0) BIF_RET(am_false); BIF_RET(erts_make_integer(res, BIF_P)); diff --git a/erts/emulator/beam/erl_bif_port.c b/erts/emulator/beam/erl_bif_port.c index 2b7760f556..31151dc9f5 100644 --- a/erts/emulator/beam/erl_bif_port.c +++ b/erts/emulator/beam/erl_bif_port.c @@ -42,7 +42,7 @@ #include "erl_bits.h" #include "dtrace-wrapper.h" -static int open_port(Process* p, Eterm name, Eterm settings, int *err_nump); +static Port *open_port(Process* p, Eterm name, Eterm settings, int *err_typep, int *err_nump); static byte* convert_environment(Process* p, Eterm env); static char **convert_args(Eterm); static void free_args(char **); @@ -54,16 +54,17 @@ port_call(Process* p, Eterm arg1, Eterm arg2, Eterm arg3); BIF_RETTYPE open_port_2(BIF_ALIST_2) { - int port_num; - Eterm port_val; + Port *port; + Eterm port_id; char *str; - int err_num; + int err_type, err_num; - if ((port_num = open_port(BIF_P, BIF_ARG_1, BIF_ARG_2, &err_num)) < 0) { - if (port_num == -3) { + port = open_port(BIF_P, BIF_ARG_1, BIF_ARG_2, &err_type, &err_num); + if (!port) { + if (err_type == -3) { ASSERT(err_num == BADARG || err_num == SYSTEM_LIMIT); BIF_ERROR(BIF_P, err_num); - } else if (port_num == -2) { + } else if (err_type == -2) { str = erl_errno_id(err_num); } else { str = "einval"; @@ -74,15 +75,15 @@ BIF_RETTYPE open_port_2(BIF_ALIST_2) erts_smp_proc_lock(BIF_P, ERTS_PROC_LOCK_LINK); - port_val = erts_port[port_num].common.id; - erts_add_link(&ERTS_P_LINKS(&erts_port[port_num]), LINK_PID, BIF_P->common.id); - erts_add_link(&ERTS_P_LINKS(BIF_P), LINK_PID, port_val); + port_id = port->common.id; + erts_add_link(&ERTS_P_LINKS(port), LINK_PID, BIF_P->common.id); + erts_add_link(&ERTS_P_LINKS(BIF_P), LINK_PID, port_id); erts_smp_proc_unlock(BIF_P, ERTS_PROC_LOCK_LINK); - erts_port_release(&erts_port[port_num]); + erts_port_release(port); - BIF_RET(port_val); + BIF_RET(port_id); } /**************************************************************************** @@ -112,7 +113,10 @@ id_or_name2port(Process *c_p, Eterm id) { Port *port; if (is_not_atom(id)) - port = erts_id2port(id, c_p, ERTS_PROC_LOCK_MAIN); + port = erts_id2port_sflgs(id, + c_p, + ERTS_PROC_LOCK_MAIN, + ERTS_PORT_SFLGS_INVALID_LOOKUP); else erts_whereis_name(c_p, ERTS_PROC_LOCK_MAIN, id, NULL, 0, 0, &port); return port; @@ -164,7 +168,7 @@ do_port_command(Process *BIF_P, Eterm arg1, Eterm arg2, Eterm arg3, ERTS_BIF_PREP_ERROR(res, BIF_P, EXC_NOTSUP); } else if (!(flags & ERTS_PORT_COMMAND_FLAG_FORCE) - && (erts_smp_atomic32_read_nob(&p->state) & ERTS_PORT_SFLG_PORT_BUSY)) { + && (erts_atomic32_read_nob(&p->state) & ERTS_PORT_SFLG_PORT_BUSY)) { if (flags & ERTS_PORT_COMMAND_FLAG_NOSUSPEND) { ERTS_BIF_PREP_RET(res, am_false); } @@ -538,7 +542,7 @@ BIF_RETTYPE port_connect_2(BIF_ALIST_2) rp = erts_pid2proc(BIF_P, ERTS_PROC_LOCK_MAIN, pid, ERTS_PROC_LOCK_LINK); if (!rp) { - erts_smp_port_unlock(prt); + erts_port_release(prt); ERTS_SMP_ASSERT_IS_NOT_EXITING(BIF_P); goto error; } @@ -549,7 +553,7 @@ BIF_RETTYPE port_connect_2(BIF_ALIST_2) erts_smp_proc_unlock(rp, ERTS_PROC_LOCK_LINK); prt->connected = pid; /* internal pid */ - erts_smp_port_unlock(prt); + erts_port_release(prt); #ifdef USE_VM_PROBES if (DTRACE_ENABLED(port_connect)) { DTRACE_CHARBUF(process_str, DTRACE_TERM_BUF_SIZE); @@ -591,7 +595,7 @@ BIF_RETTYPE port_set_data_2(BIF_ALIST_2) hp = bp->mem; prt->data = copy_struct(data, size, &hp, &bp->off_heap); } - erts_smp_port_unlock(prt); + erts_port_release(prt); BIF_RET(am_true); } @@ -612,7 +616,7 @@ BIF_RETTYPE port_get_data_1(BIF_ALIST_1) Eterm* hp = HAlloc(BIF_P, prt->bp->used_size); res = copy_struct(prt->data, prt->bp->used_size, &hp, &MSO(BIF_P)); } - erts_smp_port_unlock(prt); + erts_port_release(prt); BIF_RET(res); } @@ -625,11 +629,10 @@ BIF_RETTYPE port_get_data_1(BIF_ALIST_1) * either BADARG or SYSTEM_LIMIT). */ -static int -open_port(Process* p, Eterm name, Eterm settings, int *err_nump) +static Port * +open_port(Process* p, Eterm name, Eterm settings, int *err_typep, int *err_nump) { -#define OPEN_PORT_ERROR(VAL) do { port_num = (VAL); goto do_return; } while (0) - int i, port_num; + int i; Eterm option; Uint arity; Eterm* tp; @@ -641,6 +644,7 @@ open_port(Process* p, Eterm name, Eterm settings, int *err_nump) Eterm edir = NIL; byte dir[MAXPATHLEN]; erts_aint32_t sflgs = 0; + Port *port; /* These are the defaults */ opts.packet_bytes = 0; @@ -923,38 +927,40 @@ open_port(Process* p, Eterm name, Eterm settings, int *err_nump) erts_smp_proc_unlock(p, ERTS_PROC_LOCK_MAIN); - port_num = erts_open_driver(driver, p->common.id, name_buf, &opts, err_nump); + port = erts_open_driver(driver, p->common.id, name_buf, &opts, err_typep, err_nump); #ifdef USE_VM_PROBES - if (port_num >= 0 && DTRACE_ENABLED(port_open)) { + if (port && DTRACE_ENABLED(port_open)) { DTRACE_CHARBUF(process_str, DTRACE_TERM_BUF_SIZE); DTRACE_CHARBUF(port_str, DTRACE_TERM_BUF_SIZE); dtrace_proc_str(p, process_str); - erts_snprintf(port_str, sizeof(port_str), "%T", erts_port[port_num].id); + erts_snprintf(port_str, sizeof(port_str), "%T", port->common.id); DTRACE3(port_open, process_str, name_buf, port_str); } #endif erts_smp_proc_lock(p, ERTS_PROC_LOCK_MAIN); - if (port_num < 0) { - DEBUGF(("open_driver returned %d(%d)\n", port_num, *err_nump)); + if (!port) { + DEBUGF(("open_driver returned (%d:%d)\n", + err_typep ? *err_typep : 4711, + err_nump ? *err_nump : 4711)); if (IS_TRACED_FL(p, F_TRACE_SCHED_PROCS)) { trace_virtual_sched(p, am_in); } - OPEN_PORT_ERROR(port_num); + goto do_return; } if (IS_TRACED_FL(p, F_TRACE_SCHED_PROCS)) { trace_virtual_sched(p, am_in); } - if (linebuf && erts_port[port_num].linebuf == NULL){ - erts_port[port_num].linebuf = allocate_linebuf(linebuf); + if (linebuf && port->linebuf == NULL){ + port->linebuf = allocate_linebuf(linebuf); sflgs |= ERTS_PORT_SFLG_LINEBUF_IO; } if (sflgs) - erts_smp_atomic32_read_bor_relb(&erts_port[port_num].state, sflgs); + erts_atomic32_read_bor_relb(&port->state, sflgs); do_return: if (name_buf) @@ -965,13 +971,15 @@ open_port(Process* p, Eterm name, Eterm settings, int *err_nump) if (opts.wd && opts.wd != ((char *)dir)) { erts_free(ERTS_ALC_T_TMP, (void *) opts.wd); } - return port_num; + return port; badarg: - *err_nump = BADARG; - OPEN_PORT_ERROR(-3); + if (err_typep) + *err_typep = -3; + if (err_nump) + *err_nump = BADARG; + port = NULL; goto do_return; -#undef OPEN_PORT_ERROR } /* Arguments can be given i unicode and as raw binaries, convert filename is used to convert */ diff --git a/erts/emulator/beam/erl_bif_trace.c b/erts/emulator/beam/erl_bif_trace.c index d94b05fc25..60d5d039ed 100644 --- a/erts/emulator/beam/erl_bif_trace.c +++ b/erts/emulator/beam/erl_bif_trace.c @@ -151,14 +151,12 @@ trace_pattern(Process* p, Eterm MFA, Eterm Pattern, Eterm flaglist) } } else if (is_internal_port(meta_tracer_pid)) { Port *meta_tracer_port; - meta_tracer_proc = NULL; - if (internal_port_index(meta_tracer_pid) >= erts_max_ports) + meta_tracer_proc = NULL; + meta_tracer_port = (erts_port_lookup( + meta_tracer_pid, + ERTS_PORT_SFLGS_INVALID_TRACER_LOOKUP)); + if (!meta_tracer_port) goto error; - meta_tracer_port = - &erts_port[internal_port_index(meta_tracer_pid)]; - if (INVALID_TRACER_PORT(meta_tracer_port, meta_tracer_pid)) { - goto error; - } } else { goto error; } @@ -479,14 +477,14 @@ Eterm trace_3(BIF_ALIST_3) ? ERTS_PROC_LOCKS_ALL_MINOR : ERTS_PROC_LOCKS_ALL)); } else if (is_internal_port(tracer)) { - Port *tracer_port = erts_id2port(tracer, p, ERTS_PROC_LOCK_MAIN); - if (!erts_is_valid_tracer_port(tracer)) { - if (tracer_port) - erts_smp_port_unlock(tracer_port); + Port *tracer_port = erts_id2port_sflgs(tracer, + p, + ERTS_PROC_LOCK_MAIN, + ERTS_PORT_SFLGS_INVALID_TRACER_LOOKUP); + if (!tracer_port) goto error; - } ERTS_TRACE_FLAGS(tracer_port) |= F_TRACER; - erts_smp_port_unlock(tracer_port); + erts_port_release(tracer_port); } else goto error; @@ -519,12 +517,15 @@ Eterm trace_3(BIF_ALIST_3) if (pid_spec == tracer) goto error; - tracee_port = erts_id2port(pid_spec, p, ERTS_PROC_LOCK_MAIN); + tracee_port = erts_id2port_sflgs(pid_spec, + p, + ERTS_PROC_LOCK_MAIN, + ERTS_PORT_SFLGS_INVALID_LOOKUP); if (!tracee_port) goto error; if (tracer != NIL && port_already_traced(p, tracee_port, tracer)) { - erts_smp_port_unlock(tracee_port); + erts_port_release(tracee_port); goto already_traced; } @@ -538,7 +539,7 @@ Eterm trace_3(BIF_ALIST_3) else if (tracer != NIL) ERTS_TRACER_PROC(tracee_port) = tracer; - erts_smp_port_unlock(tracee_port); + erts_port_release(tracee_port); matches = 1; } else if (is_pid(pid_spec)) { @@ -677,15 +678,21 @@ Eterm trace_3(BIF_ALIST_3) } } if (ports || mods) { + int max = erts_ptab_max(&erts_port); /* tracing of ports */ - for (i = 0; i < erts_max_ports; i++) { - Port *tracee_port = &erts_port[i]; + for (i = 0; i < max; i++) { erts_aint32_t state; - state = erts_smp_atomic32_read_nob(&tracee_port->state); - if (state & ERTS_PORT_SFLGS_DEAD) continue; + Port *tracee_port = erts_pix2port(i); + if (!tracee_port) + continue; + state = erts_atomic32_read_nob(&tracee_port->state); + if (state & ERTS_PORT_SFLGS_DEAD) + continue; if (tracer != NIL) { - if (tracee_port->common.id == tracer) continue; - if (port_already_traced(NULL, tracee_port, tracer)) continue; + if (tracee_port->common.id == tracer) + continue; + if (port_already_traced(NULL, tracee_port, tracer)) + continue; } if (on) ERTS_TRACE_FLAGS(tracee_port) |= mask; @@ -2147,9 +2154,11 @@ BIF_RETTYPE system_profile_2(BIF_ALIST_2) if (!profiler_p) goto error; } else if (is_internal_port(profiler)) { - if (internal_port_index(profiler) >= erts_max_ports) goto error; - profiler_port = &erts_port[internal_port_index(profiler)]; - if (INVALID_TRACER_PORT(profiler_port, profiler)) goto error; + profiler_port = (erts_port_lookup( + profiler, + ERTS_PORT_SFLGS_INVALID_TRACER_LOOKUP)); + if (!profiler_port) + goto error; } else { goto error; } diff --git a/erts/emulator/beam/erl_db_util.c b/erts/emulator/beam/erl_db_util.c index d89190e3c1..67c2d0eb7f 100644 --- a/erts/emulator/beam/erl_db_util.c +++ b/erts/emulator/beam/erl_db_util.c @@ -182,12 +182,13 @@ set_match_trace(Process *tracee_p, Eterm fail_term, Eterm tracer, } } else if (is_internal_port(tracer)) { Port *tracer_port = - erts_id2port(tracer, tracee_p, ERTS_PROC_LOCKS_ALL); + erts_id2port_sflgs(tracer, + tracee_p, + ERTS_PROC_LOCKS_ALL, + ERTS_PORT_SFLGS_INVALID_TRACER_LOOKUP); if (tracer_port) { - if (! INVALID_TRACER_PORT(tracer_port, tracer)) { - ret = set_tracee_flags(tracee_p, tracer, d_flags, e_flags); - } - erts_smp_port_unlock(tracer_port); + ret = set_tracee_flags(tracee_p, tracer, d_flags, e_flags); + erts_port_release(tracer_port); } } else { ASSERT(is_nil(tracer)); diff --git a/erts/emulator/beam/erl_driver.h b/erts/emulator/beam/erl_driver.h index 1ae9a211d7..1dc43419fe 100644 --- a/erts/emulator/beam/erl_driver.h +++ b/erts/emulator/beam/erl_driver.h @@ -210,8 +210,8 @@ typedef struct erl_drv_binary { typedef struct _erl_drv_data* ErlDrvData; /* Data to be used by the driver itself. */ #ifndef ERL_SYS_DRV typedef struct _erl_drv_event* ErlDrvEvent; /* An event to be selected on. */ -typedef struct _erl_drv_port* ErlDrvPort; /* A port descriptor. */ #endif +typedef struct _erl_drv_port* ErlDrvPort; /* A port descriptor. */ typedef struct _erl_drv_port* ErlDrvThreadData; /* Thread data. */ #if !defined(__WIN32__) && !defined(_WIN32) && !defined(_WIN32_) && !defined(USE_SELECT) diff --git a/erts/emulator/beam/erl_lock_check.c b/erts/emulator/beam/erl_lock_check.c index 1db7b27412..06203a706d 100644 --- a/erts/emulator/beam/erl_lock_check.c +++ b/erts/emulator/beam/erl_lock_check.c @@ -111,9 +111,6 @@ static erts_lc_lock_order_t erts_lock_order[] = { { "efile_drv", "address" }, #if defined(ENABLE_CHILD_WAITER_THREAD) || defined(ERTS_SMP) { "child_status", NULL }, -#endif -#ifdef __WIN32__ - { "sys_driver_data_lock", NULL }, #endif { "drv_ev_state_grow", NULL, }, { "drv_ev_state", "address" }, @@ -159,7 +156,7 @@ static erts_lc_lock_order_t erts_lock_order[] = { { "port_taskq_pre_alloc_lock", "address" }, { "proclist_pre_alloc_lock", "address" }, { "port_tasks_lock", NULL }, - { "get_free_port", NULL }, + { "port_table", NULL }, { "port_state", "address" }, { "xports_list_pre_alloc_lock", "address" }, { "inet_buffer_stack_lock", NULL }, @@ -248,6 +245,7 @@ typedef struct { typedef struct erts_lc_locked_locks_t_ erts_lc_locked_locks_t; struct erts_lc_locked_locks_t_ { char *thread_name; + int emu_thread; erts_tid_t tid; erts_lc_locked_locks_t *next; erts_lc_locked_locks_t *prev; @@ -365,6 +363,7 @@ create_locked_locks(char *thread_name) if (!l_lcks->thread_name) lc_abort(); + l_lcks->emu_thread = 0; l_lcks->tid = erts_thr_self(); l_lcks->required.first = NULL; l_lcks->required.last = NULL; @@ -672,7 +671,7 @@ erts_lc_set_thread_name(char *thread_name) { erts_lc_locked_locks_t *l_lcks = get_my_locked_locks(); if (!l_lcks) - (void) create_locked_locks(thread_name); + l_lcks = create_locked_locks(thread_name); else { ASSERT(l_lcks->thread_name); free((void *) l_lcks->thread_name); @@ -680,6 +679,14 @@ erts_lc_set_thread_name(char *thread_name) if (!l_lcks->thread_name) lc_abort(); } + l_lcks->emu_thread = 1; +} + +int +erts_lc_is_emu_thr(void) +{ + erts_lc_locked_locks_t *l_lcks = get_my_locked_locks(); + return l_lcks->emu_thread; } int diff --git a/erts/emulator/beam/erl_lock_check.h b/erts/emulator/beam/erl_lock_check.h index b67f36fa06..e76bcf556e 100644 --- a/erts/emulator/beam/erl_lock_check.h +++ b/erts/emulator/beam/erl_lock_check.h @@ -102,6 +102,7 @@ void erts_lc_unrequire_lock_flg(erts_lc_lock_t *lck, Uint16 op_flags); void erts_lc_require_lock(erts_lc_lock_t *lck); void erts_lc_unrequire_lock(erts_lc_lock_t *lck); +int erts_lc_is_emu_thr(void); #define ERTS_LC_ASSERT(A) \ ((void) ((A) ? 1 : erts_lc_assert_failed(__FILE__, __LINE__, #A))) diff --git a/erts/emulator/beam/erl_monitors.c b/erts/emulator/beam/erl_monitors.c index 450aad975d..63175c44d6 100644 --- a/erts/emulator/beam/erl_monitors.c +++ b/erts/emulator/beam/erl_monitors.c @@ -985,12 +985,15 @@ Eterm erts_debug_dump_links_1(BIF_ALIST_1) Process *rp; DistEntry *dep; if (is_internal_port(pid)) { - Port *rport = erts_id2port(pid, p, ERTS_PROC_LOCK_MAIN); + Port *rport = erts_id2port_sflgs(pid, + p, + ERTS_PROC_LOCK_MAIN, + ERTS_PORT_SFLGS_INVALID_LOOKUP); if (rport) { erts_printf("Dumping port links----------------------\n"); erts_dump_links(ERTS_P_LINKS(rport), 0); erts_printf("Links dumped----------------------------\n"); - erts_smp_port_unlock(rport); + erts_port_release(rport); BIF_RET(am_true); } else { BIF_ERROR(p,BADARG); diff --git a/erts/emulator/beam/erl_node_container_utils.h b/erts/emulator/beam/erl_node_container_utils.h index 62fb277eef..7796cbfc2b 100644 --- a/erts/emulator/beam/erl_node_container_utils.h +++ b/erts/emulator/beam/erl_node_container_utils.h @@ -197,8 +197,10 @@ extern ErtsPTab erts_proc; * Ports * \* */ -#define internal_port_index(x) (internal_port_data((x)) \ - & erts_port_tab_index_mask) +extern ErtsPTab erts_port; + +#define internal_port_index(x) erts_ptab_data2ix(&erts_port, \ + internal_port_data((x))) #define internal_port_node_name(x) (internal_port_node((x))->sysname) #define external_port_node_name(x) (external_port_node((x))->sysname) @@ -237,7 +239,7 @@ extern ErtsPTab erts_proc; #define is_not_port(x) (!is_port(x)) /* Highest port-ID part in a term of type Port - Not necessarily the same as the variable erts_max_ports + Not necessarily the same as current maximum port table size which defines the maximum number of simultaneous Ports in the Erlang node. ERTS_MAX_PORTS is a hard upper limit. */ @@ -249,6 +251,9 @@ extern ErtsPTab erts_proc; #define ERTS_R9_PORTS_BITS (_PORT_R9_NUM_SIZE) #define ERTS_PORTS_BITS (_PORT_NUM_SIZE) + +#define ERTS_INVALID_PORT make_internal_port(ERTS_MAX_PORT_DATA) + /* * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * *\ * Refs * \* */ diff --git a/erts/emulator/beam/erl_node_tables.c b/erts/emulator/beam/erl_node_tables.c index 3b400ce0de..67f413ef28 100644 --- a/erts/emulator/beam/erl_node_tables.c +++ b/erts/emulator/beam/erl_node_tables.c @@ -1274,7 +1274,7 @@ setup_reference_table(void) ErlHeapFragment *hfp; DistEntry *dep; HashInfo hi; - int i, max_processes; + int i, max; DeclareTmpHeapNoproc(heap,3); inserted_bins = NULL; @@ -1309,9 +1309,9 @@ setup_reference_table(void) #endif UnUseTmpHeapNoproc(3); - max_processes = erts_ptab_max(&erts_proc); + max = erts_ptab_max(&erts_proc); /* Insert all processes */ - for (i = 0; i < max_processes; i++) { + for (i = 0; i < max; i++) { Process *proc = erts_pix2proc(i); if (proc) { ErlMessage *msg; @@ -1383,25 +1383,33 @@ setup_reference_table(void) #endif /* Insert all ports */ - for (i = 0; i < erts_max_ports; i++) { - erts_aint32_t state = erts_smp_atomic32_read_nob(&erts_port[i].state); + max = erts_ptab_max(&erts_port); + for (i = 0; i < max; i++) { + erts_aint32_t state; + Port *prt; + + prt = erts_pix2port(i); + if (!prt) + continue; + + state = erts_atomic32_read_nob(&prt->state); if (state & ERTS_PORT_SFLGS_DEAD) continue; /* Insert links */ - if (ERTS_P_LINKS(&erts_port[i])) - insert_links(ERTS_P_LINKS(&erts_port[i]), erts_port[i].common.id); + if (ERTS_P_LINKS(prt)) + insert_links(ERTS_P_LINKS(prt), prt->common.id); /* Insert monitors */ - if (ERTS_P_MONITORS(&erts_port[i])) - insert_monitors(ERTS_P_MONITORS(&erts_port[i]), erts_port[i].common.id); + if (ERTS_P_MONITORS(prt)) + insert_monitors(ERTS_P_MONITORS(prt), prt->common.id); /* Insert port data */ - for(hfp = erts_port[i].bp; hfp; hfp = hfp->next) - insert_offheap(&(hfp->off_heap), HEAP_REF, erts_port[i].common.id); + for(hfp = prt->bp; hfp; hfp = hfp->next) + insert_offheap(&(hfp->off_heap), HEAP_REF, prt->common.id); /* Insert controller */ - if (erts_port[i].dist_entry) - insert_dist_entry(erts_port[i].dist_entry, + if (prt->dist_entry) + insert_dist_entry(prt->dist_entry, CTRL_REF, - erts_port[i].common.id, + prt->common.id, 0); } diff --git a/erts/emulator/beam/erl_node_tables.h b/erts/emulator/beam/erl_node_tables.h index 4a015bdef9..3a74888522 100644 --- a/erts/emulator/beam/erl_node_tables.h +++ b/erts/emulator/beam/erl_node_tables.h @@ -100,7 +100,6 @@ typedef struct { */ struct erl_link; -struct port; typedef struct dist_entry_ { HashBucket hash_bucket; /* Hash bucket */ @@ -141,7 +140,7 @@ typedef struct dist_entry_ { erts_smp_atomic_t dist_cmd_scheduled; ErtsPortTaskHandle dist_cmd; - Uint (*send)(struct port *prt, ErtsDistOutputBuf *obuf); + Uint (*send)(Port *prt, ErtsDistOutputBuf *obuf); struct cache* cache; /* The atom cache */ } DistEntry; diff --git a/erts/emulator/beam/erl_port.h b/erts/emulator/beam/erl_port.h index 1a5bc636e2..fb34a6da2d 100644 --- a/erts/emulator/beam/erl_port.h +++ b/erts/emulator/beam/erl_port.h @@ -17,15 +17,21 @@ * %CopyrightEnd% */ -#ifndef ERL_PORT_H__ +#ifndef ERL_PORT_TYPE__ +#define ERL_PORT_TYPE__ +typedef struct _erl_drv_port Port; +#endif + +#if !defined(ERL_PORT_H__) && !defined(ERL_PORT_GET_PORT_TYPE_ONLY__) #define ERL_PORT_H__ -typedef struct port Port; #include "erl_port_task.h" #include "erl_ptab.h" +#include "erl_thr_progress.h" typedef struct erts_driver_t_ erts_driver_t; +#define ERTS_INVALID_ERL_DRV_PORT ((ErlDrvPort) (SWord) -1) #define SMALL_IO_QUEUE 5 /* Number of fixed elements */ typedef struct { @@ -90,16 +96,18 @@ typedef struct ErtsXPortsList_ ErtsXPortsList; * */ -struct port { +struct _erl_drv_port { ErtsPTabElementCommon common; /* *Need* to be first in struct */ ErtsPortTaskSched sched; ErtsPortTaskHandle timeout_task; #ifdef ERTS_SMP - erts_smp_mtx_t *lock; + erts_mtx_t *lock; ErtsXPortsList *xports; erts_smp_atomic_t run_queue; - erts_smp_spinlock_t state_lck; /* protects: id, status, snapshot */ +#else + erts_atomic32_t refc; + int cleanup; #endif Eterm connected; /* A connected process */ Eterm caller; /* Current caller. */ @@ -116,15 +124,20 @@ struct port { ErtsProcList *suspended; /* List of suspended processes. */ LineBuf *linebuf; /* Buffer to hold data not ready for process to get (line oriented I/O)*/ - erts_smp_atomic32_t state; /* Status and type flags */ + erts_atomic32_t state; /* Status and type flags */ int control_flags; /* Flags for port_control() */ - erts_aint32_t snapshot; /* Next snapshot that port should be part of */ ErlDrvPDL port_data_lock; ErtsPrtSD *psd; /* Port specific data */ }; +struct erl_drv_port_data_lock { + erts_mtx_t mtx; + erts_atomic_t refc; + Port *prt; +}; + ERTS_GLB_INLINE ErtsRunQueue *erts_port_runq(Port *prt); #if ERTS_GLB_INLINE_INCL_FUNC_DEF @@ -183,35 +196,6 @@ erts_prtsd_set(Port *prt, int ix, void *data) #endif -/* arrays that get malloced at startup */ -extern Port* erts_port; - -extern Uint erts_max_ports; -extern Uint erts_port_tab_index_mask; -extern erts_smp_atomic32_t erts_ports_snapshot; -extern erts_smp_atomic_t erts_dead_ports_ptr; - -ERTS_GLB_INLINE void erts_may_save_closed_port(Port *prt); - -#if ERTS_GLB_INLINE_INCL_FUNC_DEF - -ERTS_GLB_INLINE void erts_may_save_closed_port(Port *prt) -{ - ERTS_SMP_LC_ASSERT(erts_smp_lc_spinlock_is_locked(&prt->state_lck)); - if (prt->snapshot != erts_smp_atomic32_read_acqb(&erts_ports_snapshot)) { - /* Dead ports are added from the end of the snapshot buffer */ - Eterm* tombstone; - tombstone = (Eterm*) erts_smp_atomic_add_read_nob(&erts_dead_ports_ptr, - -(erts_aint_t)sizeof(Eterm)); - ASSERT(tombstone+1 != NULL); - ASSERT(prt->snapshot == erts_smp_atomic32_read_nob(&erts_ports_snapshot) - 1); - *tombstone = prt->common.id; - } - /*else no ongoing snapshot or port was already included or created after snapshot */ -} - -#endif - extern erts_smp_atomic_t erts_bytes_out; /* no bytes written out */ extern erts_smp_atomic_t erts_bytes_in; /* no bytes sent into the system */ @@ -263,8 +247,11 @@ extern erts_smp_atomic_t erts_bytes_in; /* no bytes sent into the system */ | ERTS_PORT_SFLG_PORT_BUSY \ | ERTS_PORT_SFLG_DISTRIBUTION) - +void print_port_info(Port *, int, void *); +void erts_port_free(Port *); +#ifndef ERTS_SMP void erts_port_cleanup(Port *); +#endif void erts_fire_port_monitor(Port *prt, Eterm ref); #ifdef ERTS_SMP void erts_smp_xports_unlock(Port *); @@ -274,8 +261,9 @@ void erts_smp_xports_unlock(Port *); int erts_lc_is_port_locked(Port *); #endif -ERTS_GLB_INLINE void erts_smp_port_minor_lock(Port*); -ERTS_GLB_INLINE void erts_smp_port_minor_unlock(Port*); +ERTS_GLB_INLINE void erts_port_inc_refc(Port *prt); +ERTS_GLB_INLINE void erts_port_dec_refc(Port *prt); +ERTS_GLB_INLINE void erts_port_add_refc(Port *prt, Sint32 add_refc); ERTS_GLB_INLINE int erts_smp_port_trylock(Port *prt); ERTS_GLB_INLINE void erts_smp_port_lock(Port *prt); @@ -283,64 +271,71 @@ ERTS_GLB_INLINE void erts_smp_port_unlock(Port *prt); #if ERTS_GLB_INLINE_INCL_FUNC_DEF -ERTS_GLB_INLINE void -erts_smp_port_minor_lock(Port* prt) +ERTS_GLB_INLINE void erts_port_inc_refc(Port *prt) { #ifdef ERTS_SMP - erts_smp_spin_lock(&prt->state_lck); + erts_ptab_inc_refc(&prt->common); +#else + erts_atomic32_inc_nob(&prt->refc); #endif } -ERTS_GLB_INLINE void -erts_smp_port_minor_unlock(Port *prt) +ERTS_GLB_INLINE void erts_port_dec_refc(Port *prt) { #ifdef ERTS_SMP - erts_smp_spin_unlock(&prt->state_lck); + int referred = erts_ptab_dec_test_refc(&prt->common); + if (!referred) + erts_port_free(prt); +#else + int refc = erts_atomic32_dec_read_nob(&prt->refc); + if (refc == 0) + erts_port_free(prt); #endif } +ERTS_GLB_INLINE void erts_port_add_refc(Port *prt, Sint32 add_refc) +{ +#ifdef ERTS_SMP + int referred = erts_ptab_add_test_refc(&prt->common, add_refc); + if (!referred) + erts_port_free(prt); +#else + int refc = erts_atomic32_add_read_nob(&prt->refc, add_refc); + if (refc == 0) + erts_port_free(prt); +#endif +} ERTS_GLB_INLINE int erts_smp_port_trylock(Port *prt) { - int res; - - ASSERT(erts_smp_atomic32_read_nob(&prt->common.refc) > 0); - erts_smp_atomic32_inc_nob(&prt->common.refc); - #ifdef ERTS_SMP - res = erts_smp_mtx_trylock(prt->lock); - if (res == EBUSY) { - erts_smp_atomic32_dec_nob(&prt->common.refc); - } + /* *Need* to be a managed thread */ + ERTS_SMP_LC_ASSERT(erts_thr_progress_is_managed_thread()); + return erts_mtx_trylock(prt->lock); #else - res = 0; + return 0; #endif - - return res; } ERTS_GLB_INLINE void erts_smp_port_lock(Port *prt) { - ASSERT(erts_smp_atomic32_read_nob(&prt->common.refc) > 0); - erts_smp_atomic32_inc_nob(&prt->common.refc); #ifdef ERTS_SMP - erts_smp_mtx_lock(prt->lock); + /* *Need* to be a managed thread */ + ERTS_SMP_LC_ASSERT(erts_thr_progress_is_managed_thread()); + erts_mtx_lock(prt->lock); #endif } ERTS_GLB_INLINE void erts_smp_port_unlock(Port *prt) { - erts_aint32_t refc; #ifdef ERTS_SMP - erts_smp_mtx_unlock(prt->lock); + /* *Need* to be a managed thread */ + ERTS_SMP_LC_ASSERT(erts_thr_progress_is_managed_thread()); + erts_mtx_unlock(prt->lock); #endif - refc = erts_smp_atomic32_dec_read_nob(&prt->common.refc); - ASSERT(refc >= 0); - if (refc == 0) - erts_port_cleanup(prt); } #endif /* #if ERTS_GLB_INLINE_INCL_FUNC_DEF */ @@ -348,7 +343,7 @@ erts_smp_port_unlock(Port *prt) #define ERTS_INVALID_PORT_OPT(PP, ID, FLGS) \ (!(PP) \ - || (erts_smp_atomic32_read_nob(&(PP)->state) & (FLGS)) \ + || (erts_atomic32_read_nob(&(PP)->state) & (FLGS)) \ || (PP)->common.id != (ID)) /* port lookup */ @@ -365,125 +360,301 @@ erts_smp_port_unlock(Port *prt) #define ERTS_PORT_SCHED_ID(P, ID) \ ((Uint) (UWord) erts_prtsd_set((P), ERTS_PSD_SCHED_ID, (void *) (UWord) (ID))) +extern const Port erts_invalid_port; +#define ERTS_PORT_LOCK_BUSY ((Port *) &erts_invalid_port) + #ifdef ERTS_SMP Port *erts_de2port(DistEntry *, Process *, ErtsProcLocks); #endif -#define erts_id2port(ID, P, PL) \ - erts_id2port_sflgs((ID), (P), (PL), ERTS_PORT_SFLGS_INVALID_LOOKUP) - -ERTS_GLB_INLINE Port*erts_id2port_sflgs(Eterm, Process *, ErtsProcLocks, Uint32); +ERTS_GLB_INLINE Port *erts_pix2port(int); +ERTS_GLB_INLINE Port *erts_port_lookup_raw(Eterm); +ERTS_GLB_INLINE Port *erts_port_lookup(Eterm, Uint32); +ERTS_GLB_INLINE Port*erts_id2port(Eterm id); +ERTS_GLB_INLINE Port *erts_id2port_sflgs(Eterm, Process *, ErtsProcLocks, Uint32); ERTS_GLB_INLINE void erts_port_release(Port *); -ERTS_GLB_INLINE Port*erts_drvport2port(ErlDrvPort, erts_aint32_t *); -ERTS_GLB_INLINE Port*erts_drvportid2port(Eterm); -ERTS_GLB_INLINE Uint32 erts_portid2status(Eterm id); -ERTS_GLB_INLINE int erts_is_port_alive(Eterm id); -ERTS_GLB_INLINE int erts_is_valid_tracer_port(Eterm id); +#ifdef ERTS_SMP +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_drvport2port_raw(ErlDrvPort drvport); +ERTS_GLB_INLINE Port *erts_drvport2port(ErlDrvPort, erts_aint32_t *); +ERTS_GLB_INLINE Port *erts_drvportid2port(Eterm); +ERTS_GLB_INLINE Eterm erts_drvport2id(ErlDrvPort); +ERTS_GLB_INLINE Uint32 erts_portid2status(Eterm); +ERTS_GLB_INLINE int erts_is_port_alive(Eterm); +ERTS_GLB_INLINE int erts_is_valid_tracer_port(Eterm); #if ERTS_GLB_INLINE_INCL_FUNC_DEF +ERTS_GLB_INLINE Port *erts_pix2port(int ix) +{ + Port *prt; + ASSERT(0 <= ix && ix < erts_ptab_max(&erts_port)); + prt = (Port *) erts_ptab_pix2intptr_nob(&erts_port, ix); + return prt == ERTS_PORT_LOCK_BUSY ? NULL : prt; +} + +ERTS_GLB_INLINE Port * +erts_port_lookup_raw(Eterm id) +{ + Port *prt; + + ERTS_SMP_LC_ASSERT(erts_thr_progress_lc_is_delaying()); + + if (is_not_internal_port(id)) + return NULL; + + prt = (Port *) erts_ptab_pix2intptr_ddrb(&erts_port, + internal_port_index(id)); + return prt && prt->common.id == id ? prt : NULL; +} + +ERTS_GLB_INLINE Port * +erts_port_lookup(Eterm id, Uint32 invalid_sflgs) +{ + Port *prt = erts_port_lookup_raw(id); + erts_aint32_t state = erts_atomic32_read_nob(&prt->state); + return (state & invalid_sflgs) ? NULL : prt; +} + + ERTS_GLB_INLINE Port* -erts_id2port_sflgs(Eterm id, Process *c_p, ErtsProcLocks c_p_locks, Uint32 sflgs) +erts_id2port(Eterm id) +{ + erts_aint32_t state; + Port *prt; + + /* Only allowed to be called from managed threads */ + ERTS_SMP_LC_ASSERT(erts_thr_progress_is_managed_thread()); + + if (is_not_internal_port(id)) + return NULL; + + prt = (Port *) erts_ptab_pix2intptr_ddrb(&erts_port, + internal_port_index(id)); + + if (!prt || prt->common.id != id) + return NULL; + + erts_smp_port_lock(prt); + state = erts_atomic32_read_nob(&prt->state); + if (state & ERTS_PORT_SFLGS_INVALID_LOOKUP) { + erts_smp_port_unlock(prt); + return NULL; + } + + return prt; +} + + +ERTS_GLB_INLINE Port* +erts_id2port_sflgs(Eterm id, + Process *c_p, ErtsProcLocks c_p_locks, + Uint32 invalid_sflgs) { #ifdef ERTS_SMP int no_proc_locks = !c_p || !c_p_locks; #endif + erts_aint32_t state; Port *prt; + /* Only allowed to be called from managed threads */ + ERTS_SMP_LC_ASSERT(erts_thr_progress_is_managed_thread()); + if (is_not_internal_port(id)) return NULL; - prt = &erts_port[internal_port_index(id)]; + prt = (Port *) erts_ptab_pix2intptr_ddrb(&erts_port, + internal_port_index(id)); - erts_smp_port_minor_lock(prt); - if (ERTS_INVALID_PORT_OPT(prt, id, sflgs)) { - erts_smp_port_minor_unlock(prt); - prt = NULL; + if (!prt || prt->common.id != id) + return NULL; + +#ifdef ERTS_SMP + if (no_proc_locks) + erts_smp_port_lock(prt); + else if (erts_smp_port_trylock(prt) == EBUSY) { + /* Unlock process locks, and acquire locks in lock order... */ + erts_smp_proc_unlock(c_p, c_p_locks); + erts_smp_port_lock(prt); + erts_smp_proc_lock(c_p, c_p_locks); + } +#endif + state = erts_atomic32_read_nob(&prt->state); + if (state & invalid_sflgs) { +#ifdef ERTS_SMP + erts_smp_port_unlock(prt); +#endif + return NULL; } - else { - erts_smp_atomic32_inc_nob(&prt->common.refc); - erts_smp_port_minor_unlock(prt); + return prt; +} + +ERTS_GLB_INLINE void +erts_port_release(Port *prt) +{ + /* Only allowed to be called from managed threads */ + ERTS_SMP_LC_ASSERT(erts_thr_progress_is_managed_thread()); #ifdef ERTS_SMP - if (no_proc_locks) - erts_smp_mtx_lock(prt->lock); - else if (erts_smp_mtx_trylock(prt->lock) == EBUSY) { - /* Unlock process locks, and acquire locks in lock order... */ - erts_smp_proc_unlock(c_p, c_p_locks); - erts_smp_mtx_lock(prt->lock); - erts_smp_proc_lock(c_p, c_p_locks); + erts_smp_port_unlock(prt); +#else + if (prt->cleanup) { + prt->cleanup = 0; + erts_port_cleanup(prt); + } +#endif +} + +#ifdef ERTS_SMP + +/* + * erts_thr_id2port_sflgs() and erts_thr_port_release() can + * be used by unmanaged threads in the SMP case. + */ +ERTS_GLB_INLINE Port * +erts_thr_id2port_sflgs(Eterm id, Uint32 invalid_sflgs) +{ + Port *prt; + ErtsThrPrgrDelayHandle dhndl; + + if (is_not_internal_port(id)) + return NULL; + + dhndl = erts_thr_progress_unmanaged_delay(); + + prt = (Port *) erts_ptab_pix2intptr_ddrb(&erts_port, + internal_port_index(id)); + + if (!prt || prt->common.id != id) { + erts_thr_progress_unmanaged_continue(dhndl); + prt = NULL; + } + else { + erts_aint32_t state; + if (dhndl != ERTS_THR_PRGR_DHANDLE_MANAGED) { + erts_port_inc_refc(prt); + erts_thr_progress_unmanaged_continue(dhndl); } - /* The id may not have changed... */ - ERTS_SMP_LC_ASSERT(prt->common.id == id); - /* ... but state may have... */ - if (erts_smp_atomic32_read_nob(&prt->state) & sflgs) { - erts_smp_port_unlock(prt); /* Also decrements common.refc... */ + erts_mtx_lock(prt->lock); + state = erts_atomic32_read_nob(&prt->state); + if (state & invalid_sflgs) { + erts_mtx_unlock(prt->lock); + if (dhndl != ERTS_THR_PRGR_DHANDLE_MANAGED) + erts_port_dec_refc(prt); prt = NULL; } -#endif - } return prt; } ERTS_GLB_INLINE void -erts_port_release(Port *prt) +erts_thr_port_release(Port *prt) { - erts_smp_port_unlock(prt); + erts_mtx_unlock(prt->lock); +#ifdef ERTS_SMP + if (!erts_thr_progress_is_managed_thread()) + erts_port_dec_refc(prt); +#endif +} + +#endif + +ERTS_GLB_INLINE Port* +erts_thr_drvport2port_raw(ErlDrvPort drvport) +{ +#if ERTS_ENABLE_LOCK_CHECK + int emu_thread = erts_lc_is_emu_thr(); +#endif + if (drvport == ERTS_INVALID_ERL_DRV_PORT) + return NULL; + else { + Port *prt = (Port *) drvport; +#if ERTS_ENABLE_LOCK_CHECK + if (emu_thread) { + ERTS_SMP_LC_ASSERT(erts_lc_is_port_locked(prt)); + ERTS_LC_ASSERT(!prt->port_data_lock + || erts_lc_mtx_is_locked(&prt->port_data_lock->mtx)); + } + else { + ERTS_LC_ASSERT(prt->port_data_lock); + ERTS_LC_ASSERT(erts_lc_mtx_is_locked(&prt->port_data_lock->mtx)); + } +#endif + return prt; + } +} + +ERTS_GLB_INLINE Port* +erts_drvport2port_raw(ErlDrvPort drvport) +{ + ERTS_LC_ASSERT(erts_lc_is_emu_thr()); + if (drvport == ERTS_INVALID_ERL_DRV_PORT) + return NULL; + else { + Port *prt = (Port *) drvport; + ERTS_SMP_LC_ASSERT(erts_lc_is_port_locked(prt)); + return prt; + } } ERTS_GLB_INLINE Port* erts_drvport2port(ErlDrvPort drvport, erts_aint32_t *statep) { - int ix = (int) drvport; + Port *prt = erts_drvport2port_raw(drvport); erts_aint32_t state; - if (ix < 0 || erts_max_ports <= ix) + if (!prt) return NULL; - state = erts_smp_atomic32_read_nob(&erts_port[ix].state); + state = erts_atomic32_read_nob(&prt->state); if (state & ERTS_PORT_SFLGS_INVALID_DRIVER_LOOKUP) return NULL; - ERTS_SMP_LC_ASSERT(erts_lc_is_port_locked(&erts_port[ix])); if (statep) *statep = state; - return &erts_port[ix]; + return prt; } ERTS_GLB_INLINE Port* erts_drvportid2port(Eterm id) { - int ix; + Port *prt; erts_aint32_t state; if (is_not_internal_port(id)) return NULL; - ix = (int) internal_port_index(id); - if (erts_max_ports <= ix) + prt = (Port *) erts_ptab_pix2intptr_nob(&erts_port, + internal_port_index(id)); + if (!prt) return NULL; - state = erts_smp_atomic32_read_nob(&erts_port[ix].state); - if (state & ERTS_PORT_SFLGS_INVALID_DRIVER_LOOKUP) + ERTS_SMP_LC_ASSERT(erts_lc_is_port_locked(prt)); + if (prt->common.id != id) return NULL; - if (erts_port[ix].common.id != id) + state = erts_atomic32_read_nob(&prt->state); + if (state & ERTS_PORT_SFLGS_INVALID_DRIVER_LOOKUP) return NULL; - ERTS_SMP_LC_ASSERT(erts_lc_is_port_locked(&erts_port[ix])); - return &erts_port[ix]; + return prt; +} + +ERTS_GLB_INLINE Eterm +erts_drvport2id(ErlDrvPort drvport) +{ + Port *prt = erts_drvport2port_raw(drvport); + if (!prt) + return am_undefined; + else + return prt->common.id; } ERTS_GLB_INLINE Uint32 erts_portid2status(Eterm id) { - if (is_not_internal_port(id)) + Port *prt = erts_port_lookup_raw(id); + if (prt) + return (Uint32) erts_atomic32_read_acqb(&prt->state); + else return ERTS_PORT_SFLG_INVALID; - else { - erts_aint32_t state; - int ix = internal_port_index(id); - if (erts_max_ports <= ix) - return ERTS_PORT_SFLG_INVALID; - state = erts_smp_atomic32_read_ddrb(&erts_port[ix].state); - if (erts_port[ix].common.id != id) - return ERTS_PORT_SFLG_INVALID; - return state; - } } ERTS_GLB_INLINE int diff --git a/erts/emulator/beam/erl_port_task.c b/erts/emulator/beam/erl_port_task.c index 6eb8c4869c..112d27c94e 100644 --- a/erts/emulator/beam/erl_port_task.c +++ b/erts/emulator/beam/erl_port_task.c @@ -51,12 +51,9 @@ #define ERTS_PORT_TASK_INVALID_PORT(P, ID) \ - ((erts_smp_atomic32_read_acqb(&(P)->state) & ERTS_PORT_SFLGS_DEAD) \ + ((erts_atomic32_read_acqb(&(P)->state) & ERTS_PORT_SFLGS_DEAD) \ || (P)->common.id != (ID)) -#define ERTS_PORT_IS_IN_RUNQ(RQ, P) \ - ((P)->sched.next || (P)->sched.prev || (RQ)->ports.start == (P)) - #ifdef USE_VM_PROBES #define DTRACE_DRIVER(PROBE_NAME, PP) \ if (DTRACE_ENABLED(driver_ready_input)) { \ @@ -76,7 +73,6 @@ erts_smp_atomic_t erts_port_task_outstanding_io_tasks; struct ErtsPortTaskQueue_ { ErtsPortTask *first; ErtsPortTask *last; - Port *port; }; struct ErtsPortTask_ { @@ -275,7 +271,6 @@ port_taskq_init(ErtsPortTaskQueue *ptqp, Port *pp) if (ptqp) { ptqp->first = NULL; ptqp->last = NULL; - ptqp->port = pp; } return ptqp; } @@ -429,7 +424,10 @@ erts_port_task_abort(Eterm id, ErtsPortTaskHandle *pthp) ErtsPortTask *ptp; Port *pp; - pp = &erts_port[internal_port_index(id)]; + pp = erts_port_lookup_raw(id); + if (!pp) + return 1; + runq = erts_port_runq(pp); if (!runq) return 1; @@ -512,7 +510,10 @@ erts_port_task_schedule(Eterm id, ptp = port_task_alloc(); ASSERT(is_internal_port(id)); - pp = &erts_port[internal_port_index(id)]; + pp = erts_port_lookup_raw(id); + if (!pp) + return -1; + runq = erts_port_runq(pp); if (!runq || ERTS_PORT_TASK_INVALID_PORT(pp, id)) { @@ -616,7 +617,7 @@ erts_port_task_free_port(Port *pp) ErtsPortTaskQueue *ptqp; ERTS_SMP_LC_ASSERT(erts_lc_is_port_locked(pp)); - ASSERT(!(erts_smp_atomic32_read_nob(&pp->state) & ERTS_PORT_SFLGS_DEAD)); + ASSERT(!(erts_atomic32_read_nob(&pp->state) & ERTS_PORT_SFLGS_DEAD)); runq = erts_port_runq(pp); ASSERT(runq); ERTS_PT_CHK_PRES_PORTQ(runq, pp); @@ -627,14 +628,10 @@ erts_port_task_free_port(Port *pp) ErtsPortTask *ptp; enqueue_free: ptp = port_task_alloc(); - erts_smp_port_minor_lock(pp); - erts_smp_atomic32_read_bset_relb(&pp->state, - (ERTS_PORT_SFLG_CLOSING - | ERTS_PORT_SFLG_FREE_SCHEDULED), - ERTS_PORT_SFLG_FREE_SCHEDULED); - erts_may_save_closed_port(pp); - erts_smp_port_minor_unlock(pp); - ERTS_LC_ASSERT(erts_smp_atomic32_read_nob(&pp->common.refc) > 1); + erts_atomic32_read_bset_relb(&pp->state, + (ERTS_PORT_SFLG_CLOSING + | ERTS_PORT_SFLG_FREE_SCHEDULED), + ERTS_PORT_SFLG_FREE_SCHEDULED); ptp->type = ERTS_PORT_TASK_FREE; ptp->event = (ErlDrvEvent) -1; ptp->event_data = NULL; @@ -651,20 +648,18 @@ erts_port_task_free_port(Port *pp) goto enqueue_free; } ASSERT(!pp->sched.taskq); - erts_smp_port_minor_lock(pp); - erts_smp_atomic32_read_bset_relb(&pp->state, - (ERTS_PORT_SFLG_CLOSING - | ERTS_PORT_SFLG_FREE_SCHEDULED), - ERTS_PORT_SFLG_FREE_SCHEDULED); - erts_may_save_closed_port(pp); - erts_smp_port_minor_unlock(pp); - erts_smp_atomic32_dec_nob(&pp->common.refc); /* Not alive */ - ERTS_LC_ASSERT(erts_smp_atomic32_read_nob(&pp->common.refc) > 0); /* Lock */ + erts_atomic32_read_bset_relb(&pp->state, + (ERTS_PORT_SFLG_CLOSING + | ERTS_PORT_SFLG_FREE_SCHEDULED), + ERTS_PORT_SFLG_FREE_SCHEDULED); handle_remaining_tasks(runq, pp); /* May release runq lock */ ASSERT(!pp->sched.exe_taskq && (!ptqp || !ptqp->first)); pp->sched.taskq = NULL; ERTS_PT_CHK_PRES_PORTQ(runq, pp); erts_smp_runq_unlock(runq); +#ifndef ERTS_SMP + pp->cleanup = 1; +#endif } } @@ -703,14 +698,21 @@ erts_port_task_execute(ErtsRunQueue *runq, Port **curr_port_pp) goto done; } + if (erts_smp_port_trylock(pp) == EBUSY) { + erts_smp_runq_unlock(runq); + erts_smp_port_lock(pp); + erts_smp_runq_lock(runq); + } + ASSERT(pp->sched.in_runq); pp->sched.in_runq = 0; + if (!pp->sched.taskq) { if (erts_system_profile_flags.runnable_ports) profile_runnable_port(pp, am_inactive); res = (erts_smp_atomic_read_nob(&erts_port_task_outstanding_io_tasks) != (erts_aint_t) 0); - goto done; + goto release_port_done; } *curr_port_pp = pp; @@ -721,12 +723,6 @@ erts_port_task_execute(ErtsRunQueue *runq, Port **curr_port_pp) ASSERT(!pp->sched.exe_taskq); pp->sched.exe_taskq = ptqp; - - if (erts_smp_port_trylock(pp) == EBUSY) { - erts_smp_runq_unlock(runq); - erts_smp_port_lock(pp); - erts_smp_runq_lock(runq); - } if (erts_sched_stat.enabled) { ErtsSchedulerData *esdp = erts_get_scheduler_data(); @@ -771,31 +767,31 @@ erts_port_task_execute(ErtsRunQueue *runq, Port **curr_port_pp) erts_smp_runq_lock(runq); erts_unblock_fpe(fpe_was_unmasked); - ASSERT(erts_smp_atomic32_read_nob(&pp->state) + ASSERT(erts_atomic32_read_nob(&pp->state) & ERTS_PORT_SFLG_FREE_SCHEDULED); if (ptqp->first || (pp->sched.taskq && pp->sched.taskq->first)) handle_remaining_tasks(runq, pp); ASSERT(!ptqp->first && (!pp->sched.taskq || !pp->sched.taskq->first)); - erts_smp_atomic32_dec_nob(&pp->common.refc); /* Not alive */ - ERTS_LC_ASSERT(erts_smp_atomic32_read_nob(&pp->common.refc) > 0); /* Lock */ port_task_free(ptp); if (pp->sched.taskq) port_taskq_free(pp->sched.taskq); pp->sched.taskq = NULL; - +#ifndef ERTS_SMP + pp->cleanup = 1; +#endif goto tasks_done; case ERTS_PORT_TASK_TIMEOUT: reds += ERTS_PORT_REDS_TIMEOUT; - if (!(erts_smp_atomic32_read_nob(&pp->state) & ERTS_PORT_SFLGS_DEAD)) { + if (!(erts_atomic32_read_nob(&pp->state) & ERTS_PORT_SFLGS_DEAD)) { DTRACE_DRIVER(driver_timeout, pp); (*pp->drv_ptr->timeout)((ErlDrvData) pp->drv_data); } break; case ERTS_PORT_TASK_INPUT: reds += ERTS_PORT_REDS_INPUT; - ASSERT((erts_smp_atomic32_read_nob(&pp->state) + ASSERT((erts_atomic32_read_nob(&pp->state) & ERTS_PORT_SFLGS_DEAD) == 0); DTRACE_DRIVER(driver_ready_input, pp); /* NOTE some windows drivers use ->ready_input for input and output */ @@ -804,7 +800,7 @@ erts_port_task_execute(ErtsRunQueue *runq, Port **curr_port_pp) break; case ERTS_PORT_TASK_OUTPUT: reds += ERTS_PORT_REDS_OUTPUT; - ASSERT((erts_smp_atomic32_read_nob(&pp->state) + ASSERT((erts_atomic32_read_nob(&pp->state) & ERTS_PORT_SFLGS_DEAD) == 0); DTRACE_DRIVER(driver_ready_output, pp); (*pp->drv_ptr->ready_output)((ErlDrvData) pp->drv_data, ptp->event); @@ -812,7 +808,7 @@ erts_port_task_execute(ErtsRunQueue *runq, Port **curr_port_pp) break; case ERTS_PORT_TASK_EVENT: reds += ERTS_PORT_REDS_EVENT; - ASSERT((erts_smp_atomic32_read_nob(&pp->state) + ASSERT((erts_atomic32_read_nob(&pp->state) & ERTS_PORT_SFLGS_DEAD) == 0); DTRACE_DRIVER(driver_event, pp); (*pp->drv_ptr->event)((ErlDrvData) pp->drv_data, ptp->event, ptp->event_data); @@ -828,7 +824,7 @@ erts_port_task_execute(ErtsRunQueue *runq, Port **curr_port_pp) break; } - if ((erts_smp_atomic32_read_nob(&pp->state) & ERTS_PORT_SFLG_CLOSING) + if ((erts_atomic32_read_nob(&pp->state) & ERTS_PORT_SFLG_CLOSING) && erts_is_port_ioq_empty(pp)) { reds += ERTS_PORT_REDS_TERMINATE; erts_terminate_port(pp); @@ -879,7 +875,7 @@ erts_port_task_execute(ErtsRunQueue *runq, Port **curr_port_pp) ErtsRunQueue *xrunq; #endif - ASSERT(!(erts_smp_atomic32_read_nob(&pp->state) & ERTS_PORT_SFLGS_DEAD)); + ASSERT(!(erts_atomic32_read_nob(&pp->state) & ERTS_PORT_SFLGS_DEAD)); ASSERT(pp->sched.taskq->first); #ifdef ERTS_SMP @@ -922,23 +918,9 @@ erts_port_task_execute(ErtsRunQueue *runq, Port **curr_port_pp) if (IS_TRACED_FL(pp, F_TRACE_SCHED_PORTS)) { trace_sched_ports(pp, am_out); } -#ifndef ERTS_SMP + +release_port_done: erts_port_release(pp); -#else - { - erts_aint32_t refc; - erts_smp_mtx_unlock(pp->lock); - refc = erts_smp_atomic32_dec_read_nob(&pp->common.refc); - ASSERT(refc >= 0); - if (refc == 0) { - erts_smp_runq_unlock(runq); - erts_port_cleanup(pp); /* Might aquire runq lock */ - erts_smp_runq_lock(runq); - res = (erts_smp_atomic_read_nob(&erts_port_task_outstanding_io_tasks) - != (erts_aint_t) 0); - } - } -#endif done: ERTS_SMP_LC_ASSERT(erts_smp_lc_runq_is_locked(runq)); diff --git a/erts/emulator/beam/erl_process.c b/erts/emulator/beam/erl_process.c index f9ff3c4bd5..e6fe4ccdc1 100644 --- a/erts/emulator/beam/erl_process.c +++ b/erts/emulator/beam/erl_process.c @@ -43,7 +43,6 @@ #include "erl_thr_queue.h" #include "erl_async.h" #include "dtrace-wrapper.h" -#define ERTS_PTAB_WANT_BIF_IMPL__ #include "erl_ptab.h" #define ERTS_RUNQ_CHECK_BALANCE_REDS_PER_SCHED (2000*CONTEXT_REDS) @@ -1482,37 +1481,32 @@ handle_reap_ports(ErtsAuxWorkData *awdp, erts_aint32_t aux_work) unset_aux_work_flags(awdp->ssi, ERTS_SSI_AUX_WORK_REAP_PORTS); awdp->esdp->run_queue->halt_in_progress = 1; if (erts_smp_atomic32_dec_read_acqb(&erts_halt_progress) == 0) { - int i; + int i, max = erts_ptab_max(&erts_port); erts_smp_atomic32_set_nob(&erts_halt_progress, 1); - for (i = 0; i < erts_max_ports; i++) { - Port *prt = &erts_port[i]; - erts_aint32_t state = erts_smp_atomic32_read_acqb(&prt->state); + for (i = 0; i < max; i++) { + erts_aint32_t state; + Port *prt = erts_pix2port(i); + if (!prt) + continue; + state = erts_atomic32_read_acqb(&prt->state); if (state & (ERTS_PORT_SFLGS_INVALID_DRIVER_LOOKUP | ERTS_PORT_SFLG_HALT)) continue; - erts_smp_port_minor_lock(prt); + /* We need to set the halt flag - get the port lock */ -#ifdef ERTS_SMP - erts_smp_atomic32_inc_nob(&prt->common.refc); -#endif - erts_smp_port_minor_unlock(prt); -#ifdef ERTS_SMP - erts_smp_mtx_lock(prt->lock); -#endif - state = erts_smp_atomic32_read_acqb(&prt->state); - if (state & (ERTS_PORT_SFLGS_INVALID_DRIVER_LOOKUP - | ERTS_PORT_SFLG_HALT)) { - erts_port_release(prt); - continue; - } - state = erts_smp_atomic32_read_bor_relb(&prt->state, + + erts_smp_port_lock(prt); + + state = erts_atomic32_read_nob(&prt->state); + if (!(state & (ERTS_PORT_SFLGS_INVALID_DRIVER_LOOKUP + | ERTS_PORT_SFLG_HALT))) { + state = erts_atomic32_read_bor_relb(&prt->state, ERTS_PORT_SFLG_HALT); - erts_smp_atomic32_inc_nob(&erts_halt_progress); - if (state & (ERTS_PORT_SFLG_EXITING|ERTS_PORT_SFLG_CLOSING)) { - erts_port_release(prt); - continue; + erts_smp_atomic32_inc_nob(&erts_halt_progress); + if (!(state & (ERTS_PORT_SFLG_EXITING|ERTS_PORT_SFLG_CLOSING))) + erts_do_exit_port(prt, prt->common.id, am_killed); } - erts_do_exit_port(prt, prt->common.id, am_killed); + erts_port_release(prt); } if (erts_smp_atomic32_dec_read_nob(&erts_halt_progress) == 0) { @@ -4304,6 +4298,11 @@ erts_init_scheduling(int no_schedulers, int no_schedulers_online) erts_smp_atomic32_init_relb(&erts_halt_progress, -1); erts_halt_code = 0; + +#if !defined(ERTS_SMP) && defined(ERTS_ENABLE_LOCK_CHECK) + erts_lc_set_thread_name("scheduler 1"); +#endif + } ErtsRunQueue * @@ -8112,7 +8111,7 @@ static void doit_exit_monitor(ErtsMonitor *mon, void *vpcontext) ASSERT(mon->type == MON_TARGET); ASSERT(is_pid(mon->pid) || is_internal_port(mon->pid)); if (is_internal_port(mon->pid)) { - Port *prt = erts_id2port(mon->pid, NULL, 0); + Port *prt = erts_id2port(mon->pid); if (prt == NULL) { goto done; } @@ -8199,7 +8198,7 @@ static void doit_exit_link(ErtsLink *lnk, void *vpcontext) switch(lnk->type) { case LINK_PID: if(is_internal_port(item)) { - Port *prt = erts_id2port(item, NULL, 0); + Port *prt = erts_id2port(item); if (prt) { rlnk = erts_remove_link(&ERTS_P_LINKS(prt), p->common.id); if (rlnk) @@ -8757,14 +8756,6 @@ stack_element_dump(int to, void *to_arg, Process* p, Eterm* sp, int yreg) return yreg; } -/* - * The processes/0 BIF. - */ -BIF_RETTYPE processes_0(BIF_ALIST_0) -{ - return erts_ptab_list(BIF_P, &erts_proc); -} - /* * A nice system halt closing all open port goes as follows: * 1) This function schedules the aux work ERTS_SSI_AUX_WORK_REAP_PORTS diff --git a/erts/emulator/beam/erl_process.h b/erts/emulator/beam/erl_process.h index 6f882c5e57..f5b7921820 100644 --- a/erts/emulator/beam/erl_process.h +++ b/erts/emulator/beam/erl_process.h @@ -42,6 +42,9 @@ typedef struct process Process; #include "erl_process_lock.h" /* Only pull out important types... */ #undef ERTS_PROCESS_LOCK_ONLY_PROC_LOCK_TYPE__ +#define ERL_PORT_GET_PORT_TYPE_ONLY__ +#include "erl_port.h" +#undef ERL_PORT_GET_PORT_TYPE_ONLY__ #include "erl_vm.h" #include "erl_smp.h" #include "erl_message.h" @@ -66,7 +69,6 @@ typedef struct process Process; #undef ERL_THR_PROGRESS_TSD_TYPE_ONLY struct ErtsNodesMonitor_; -struct port; #define ERTS_MAX_NO_OF_SCHEDULERS 1024 @@ -378,8 +380,8 @@ struct ErtsRunQueue_ { struct { ErtsRunQueueInfo info; - struct port *start; - struct port *end; + Port *start; + Port *end; } ports; }; @@ -476,7 +478,7 @@ struct ErtsSchedulerData_ { ErtsSchedulerSleepInfo *ssi; Process *current_process; Uint no; /* Scheduler number */ - struct port *current_port; + Port *current_port; ErtsRunQueue *run_queue; int virtual_reds; int cpu_id; /* >= 0 when bound */ @@ -1219,7 +1221,7 @@ Eterm erts_sched_stat_term(Process *p, int total); void erts_free_proc(Process *); -void erts_suspend(Process*, ErtsProcLocks, struct port*); +void erts_suspend(Process*, ErtsProcLocks, Port*); void erts_resume(Process*, ErtsProcLocks); int erts_resume_processes(ErtsProcList *); diff --git a/erts/emulator/beam/erl_ptab.c b/erts/emulator/beam/erl_ptab.c index 8195a350fb..8cab1de9da 100644 --- a/erts/emulator/beam/erl_ptab.c +++ b/erts/emulator/beam/erl_ptab.c @@ -660,11 +660,9 @@ erts_ptab_delete_element(ErtsPTab *ptab, erts_ptab_rwunlock(ptab); } -#ifdef ERTS_SMP erts_schedule_thr_prgr_later_op(ptab->r.o.release_element, (void *) ptab_el, &ptab_el->u.release); -#endif } /* @@ -1151,16 +1149,18 @@ ptab_list_bif_engine(Process *c_p, Eterm *res_accp, Binary *mbp) conses = ptlbdp->pid_ix; } - hp = HAlloc(c_p, conses*2); - ERTS_PTAB_LIST_DBG_SAVE_HEAP_ALLOC(ptlbdp, hp, conses*2); + if (conses) { + hp = HAlloc(c_p, conses*2); + ERTS_PTAB_LIST_DBG_SAVE_HEAP_ALLOC(ptlbdp, hp, conses*2); - for (ix = ptlbdp->pid_ix - 1; ix >= min_ix; ix--) { - ERTS_PTAB_LIST_ASSERT(erts_ptab_is_valid_id(ptlbdp->pid[ix])); - res = CONS(hp, ptlbdp->pid[ix], res); - hp += 2; - } + for (ix = ptlbdp->pid_ix - 1; ix >= min_ix; ix--) { + ERTS_PTAB_LIST_ASSERT(erts_ptab_is_valid_id(ptlbdp->pid[ix])); + res = CONS(hp, ptlbdp->pid[ix], res); + hp += 2; + } - ERTS_PTAB_LIST_DBG_VERIFY_HEAP_ALLOC_USED(ptlbdp, hp); + ERTS_PTAB_LIST_DBG_VERIFY_HEAP_ALLOC_USED(ptlbdp, hp); + } ptlbdp->pid_ix = min_ix; if (min_ix == 0) diff --git a/erts/emulator/beam/erl_ptab.h b/erts/emulator/beam/erl_ptab.h index b65db330e5..b2ae829d62 100644 --- a/erts/emulator/beam/erl_ptab.h +++ b/erts/emulator/beam/erl_ptab.h @@ -53,8 +53,6 @@ typedef struct { Eterm id; #ifdef ERTS_SMP erts_atomic32_t refc; -#else - erts_smp_atomic32_t refc; /* Temporary solution during dev; to be removed! */ #endif Eterm tracer_proc; Uint trace_flags; @@ -73,9 +71,7 @@ typedef struct { } alive; /* --- While being released --- */ -#ifdef ERTS_SMP ErtsThrPrgrLaterOp release; -#endif } u; } ErtsPTabElementCommon; diff --git a/erts/emulator/beam/erl_sys_driver.h b/erts/emulator/beam/erl_sys_driver.h index d429d0ce96..b991a2840c 100644 --- a/erts/emulator/beam/erl_sys_driver.h +++ b/erts/emulator/beam/erl_sys_driver.h @@ -31,7 +31,6 @@ #define ERL_SYS_DRV typedef long ErlDrvEvent; /* An event to be selected on. */ -typedef long ErlDrvPort; /* A port descriptor. */ /* typedef struct _SysDriverOpts SysDriverOpts; defined in sys.h */ diff --git a/erts/emulator/beam/erl_trace.c b/erts/emulator/beam/erl_trace.c index dd2a121165..991e573c14 100644 --- a/erts/emulator/beam/erl_trace.c +++ b/erts/emulator/beam/erl_trace.c @@ -3442,10 +3442,8 @@ sys_msg_dispatcher_func(void *unused) goto queue_proc_msg; } else if (is_internal_port(receiver)) { - port = erts_id2port_sflgs(receiver, - NULL, - 0, - ERTS_PORT_SFLGS_INVALID_TRACER_LOOKUP); + port = erts_thr_id2port_sflgs(receiver, + ERTS_PORT_SFLGS_INVALID_TRACER_LOOKUP); if (!port) goto failure; else { @@ -3459,7 +3457,7 @@ sys_msg_dispatcher_func(void *unused) #ifdef DEBUG_PRINTOUTS erts_fprintf(stderr, "delivered\n"); #endif - erts_port_release(port); + erts_thr_port_release(port); if (smqp->bp) free_message_buffer(smqp->bp); } diff --git a/erts/emulator/beam/global.h b/erts/emulator/beam/global.h index d93ac723a4..ae31ab54b5 100644 --- a/erts/emulator/beam/global.h +++ b/erts/emulator/beam/global.h @@ -884,11 +884,6 @@ int erts_global_garbage_collect(Process*, int, Eterm*, int); /* io.c */ -struct erl_drv_port_data_lock { - erts_mtx_t mtx; - erts_atomic_t refc; -}; - typedef struct { char *name; char *driver_name; @@ -901,22 +896,18 @@ typedef struct { int erts_add_driver_entry(ErlDrvEntry *drv, DE_Handle *handle, int driver_list_locked); void erts_destroy_driver(erts_driver_t *drv); void erts_wake_process_later(Port*, Process*); -int erts_open_driver(erts_driver_t*, Eterm, char*, SysDriverOpts*, int *); +Port *erts_open_driver(erts_driver_t*, Eterm, char*, SysDriverOpts*, int *, int *); int erts_is_port_ioq_empty(Port *); void erts_terminate_port(Port *); -void close_port(Eterm); void init_io(void); -void cleanup_io(void); void erts_do_exit_port(Port *, Eterm, Eterm); void erts_port_command(Process *, Eterm, Port *, Eterm); Eterm erts_port_control(Process*, Port*, Uint, Eterm); int erts_write_to_port(Eterm caller_id, Port *p, Eterm list); -void print_port_info(int, void *, int); void erts_raw_port_command(Port*, byte*, Uint); -void driver_report_exit(int, int); +void driver_report_exit(ErlDrvPort, int); LineBuf* allocate_linebuf(int); int async_ready(Port *, void*); -Sint erts_test_next_port(int, Uint); ErtsPortNames *erts_get_port_names(Eterm); void erts_free_port_names(ErtsPortNames *); Uint erts_port_ioq_size(Port *pp); diff --git a/erts/emulator/beam/io.c b/erts/emulator/beam/io.c index b772783f64..4dcec356a9 100644 --- a/erts/emulator/beam/io.c +++ b/erts/emulator/beam/io.c @@ -57,15 +57,14 @@ static erts_smp_tsd_key_t driver_list_lock_status_key; /*stop recursive locks wh static erts_smp_tsd_key_t driver_list_last_error_key; /* Save last DDLL error on a per thread basis (for BC interfaces) */ -Port* erts_port; /* The port table */ +ErtsPTab erts_port erts_align_attribute(ERTS_CACHE_LINE_SIZE); /* The port table */ erts_smp_atomic_t erts_bytes_out; /* No bytes sent out of the system */ erts_smp_atomic_t erts_bytes_in; /* No bytes gotten into the system */ -Uint erts_max_ports; -Uint erts_port_tab_index_mask; - const ErlDrvTermData driver_term_nil = (ErlDrvTermData)NIL; +const Port erts_invalid_port = {{ERTS_INVALID_PORT}}; + erts_driver_t vanilla_driver; erts_driver_t spawn_driver; erts_driver_t fd_driver; @@ -89,36 +88,12 @@ static void driver_monitor_unlock_pdl(Port *p); static ERTS_INLINE ErlIOQueue* drvport2ioq(ErlDrvPort drvport) { - int ix = (int) drvport; - erts_aint32_t state; - - if (ix < 0 || erts_max_ports <= ix) - return NULL; - - state = erts_smp_atomic32_read_nob(&erts_port[ix].state); - -#ifdef ERTS_ENABLE_LOCK_CHECK - - if (erts_get_scheduler_data()) { - ERTS_SMP_LC_ASSERT(erts_lc_is_port_locked(&erts_port[ix])); - ERTS_LC_ASSERT(!erts_port[ix].port_data_lock - || erts_lc_mtx_is_locked( - &erts_port[ix].port_data_lock->mtx)); - } - else { - ERTS_LC_ASSERT((state & ERTS_PORT_SFLGS_INVALID_DRIVER_LOOKUP) - || erts_port[ix].port_data_lock); - ERTS_LC_ASSERT(!erts_port[ix].port_data_lock - || erts_lc_mtx_is_locked( - &erts_port[ix].port_data_lock->mtx)); - } - -#endif - + Port *prt = erts_thr_drvport2port_raw(drvport); + erts_aint32_t state = erts_atomic32_read_nob(&prt->state); if (state & ERTS_PORT_SFLGS_INVALID_DRIVER_LOOKUP) return NULL; else - return &erts_port[ix].ioq; + return &prt->ioq; } static ERTS_INLINE int @@ -196,27 +171,13 @@ typedef struct line_buf_context { dtrace_port_str((PORT), port_str); #endif -/* The 'number' field in a port now has two parts: the lowest bits - contain the index in the port table, and the higher bits are a counter - which is incremented each time we look for a free port and start from - the beginning of the table. erts_max_ports is the number of file descriptors, - rounded up to a power of 2. - To get the index from a port, use the macro 'internal_port_index'; - 'port_number' returns the whole number field. -*/ - -static erts_smp_spinlock_t get_free_port_lck; -static Uint last_port_num; -static Uint port_num_mask; -erts_smp_atomic32_t erts_ports_snapshot; /* Identifies the _next_ snapshot (not the ongoing) */ - - static ERTS_INLINE void kill_port(Port *pp) { ERTS_SMP_LC_ASSERT(erts_lc_is_port_locked(pp)); + erts_ptab_delete_element(&erts_port, &pp->common); /* Time of death */ erts_port_task_free_port(pp); - ASSERT(erts_smp_atomic32_read_nob(&pp->state) & ERTS_PORT_SFLGS_DEAD); + /* In non-smp case the port structure may have been deallocated now */ } #ifdef ERTS_SMP @@ -227,149 +188,194 @@ erts_lc_is_port_locked(Port *prt) { if (!prt) return 0; + ERTS_SMP_LC_ASSERT(prt->lock); return erts_smp_lc_mtx_is_locked(prt->lock); } #endif #endif /* #ifdef ERTS_SMP */ -static int -get_free_port(void) -{ - Uint num; - Uint tries = erts_max_ports; - Port* port; - - erts_smp_spin_lock(&get_free_port_lck); - num = last_port_num + 1; - for (;; ++num) { - erts_aint32_t act; - - port = &erts_port[num & erts_port_tab_index_mask]; - - act = erts_smp_atomic32_read_nob(&port->state); - - while (act & ERTS_PORT_SFLG_FREE) { - erts_aint32_t exp = act; - act = erts_smp_atomic32_cmpxchg_relb(&port->state, - ERTS_PORT_SFLG_INITIALIZING, - exp); - if (act == exp) { - last_port_num = num; - erts_smp_spin_unlock(&get_free_port_lck); - ERTS_LC_ASSERT(erts_smp_atomic32_read_nob(&port->common.refc) == 0); - erts_smp_atomic32_set_nob(&port->common.refc, 2); /* Port alive + lock */ - return num & port_num_mask; - } - } +static void initq(Port* prt); - if (--tries == 0) { - erts_smp_spin_unlock(&get_free_port_lck); - return -1; - } - } +static void insert_port_struct(void *vprt, Eterm data) +{ + Port *prt = (Port *) vprt; + Eterm id = make_internal_port(data); +#ifdef ERTS_SMP + ASSERT(prt->drv_ptr && prt->lock); + /* + * We are breaking lock order in the port specific locking + * case. This is, however, safe since the lock has not been + * published, yet. + */ + if (!prt->drv_ptr->lock) + erts_mtx_init_locked_x(prt->lock, "port_lock", id); +#endif + prt->common.id = id; + erts_atomic32_init_relb(&prt->state, ERTS_PORT_SFLG_INITIALIZING); } -/* - * erts_test_next_port() is only used for testing. - */ -Sint -erts_test_next_port(int set, Uint next) +static Port *create_port(char *name, + erts_driver_t *driver, + erts_mtx_t *driver_lock, + Eterm pid, + int *enop) { - Uint i, num; - Sint res = -1; + Port *prt; + char *p; + size_t port_size, size; + erts_aint32_t state = ERTS_PORT_SFLG_CONNECTED; +#ifdef DEBUG + /* Make sure the debug flags survives until port is freed */ + state |= ERTS_PORT_SFLG_PORT_DEBUG; +#endif - erts_smp_spin_lock(&get_free_port_lck); - if (set) { - last_port_num = (next - 1) & port_num_mask; +#ifdef ERTS_SMP + if (!driver_lock) { + /* Align size for mutex following port struct */ + port_size = size = ERTS_ALC_DATA_ALIGN_SIZE(sizeof(Port)); + size += sizeof(erts_mtx_t); } - num = last_port_num + 1; - - for (i=0; i < erts_max_ports && res<0; ++i, ++num) { - - Port* port = &erts_port[num & erts_port_tab_index_mask]; + else +#endif + port_size = size = sizeof(Port); - erts_aint32_t state = erts_smp_atomic32_read_nob(&port->state); + size += sys_strlen(name) + 1; - if (state & ERTS_PORT_SFLG_FREE) { - last_port_num = num - 1; - res = num & port_num_mask; - } + p = erts_alloc_fnf(ERTS_ALC_T_PORT, size); + if (!p) { + if (enop) + *enop = ENOMEM; + return NULL; } - erts_smp_spin_unlock(&get_free_port_lck); - return res; -} -static void port_cleanup(Port *prt); + prt = (Port *) p; + p += port_size; #ifdef ERTS_SMP + if (driver_lock) { + prt->lock = driver_lock; + erts_mtx_lock(driver_lock); + } + else { + prt->lock = (erts_mtx_t *) p; + p += sizeof(erts_mtx_t); + state |= ERTS_PORT_SFLG_PORT_SPECIFIC_LOCK; + } + erts_smp_atomic_set_nob(&prt->run_queue, + (erts_aint_t) erts_get_runq_current(NULL)); + prt->xports = NULL; +#else + erts_atomic32_init_nob(&prt->refc, 1); + prt->cleanup = 0; +#endif + + prt->name = p; + sys_strcpy(p, name); + prt->drv_ptr = driver; + ERTS_P_LINKS(prt) = NULL; + ERTS_P_MONITORS(prt) = NULL; + prt->linebuf = NULL; + prt->bp = NULL; + prt->suspended = NULL; + prt->data = am_undefined; + prt->port_data_lock = NULL; + prt->control_flags = 0; + prt->bytes_in = 0; + prt->bytes_out = 0; + prt->dist_entry = NULL; + prt->connected = pid; + prt->common.u.alive.reg = NULL; +#ifdef ERTS_SMP + prt->common.u.alive.ptimer = NULL; +#else + sys_memset(&prt->common.u.alive.tm, 0, sizeof(ErlTimer)); +#endif + erts_port_task_handle_init(&prt->timeout_task); + erts_port_task_init_sched(&prt->sched); + prt->psd = NULL; + prt->drv_data = (SWord) 0; -static void -sched_port_cleanup(void *vprt) -{ - Port *prt = (Port *) vprt; - erts_smp_mtx_lock(prt->lock); - port_cleanup(prt); -} + /* Set default tracing */ + erts_get_default_tracing(&ERTS_TRACE_FLAGS(prt), &ERTS_TRACER_PROC(prt)); + + ASSERT(((char *) prt) == ((char *) &prt->common)); + if (!erts_ptab_new_element(&erts_port, + &prt->common, + (void *) prt, + insert_port_struct)) { +#ifdef ERTS_SMP + if (driver_lock) + erts_mtx_unlock(driver_lock); #endif + if (enop) + *enop = 0; + return NULL; + } + + ASSERT(prt == (Port *) (erts_ptab_pix2intptr_nob( + &erts_port, + internal_port_index(prt->common.id)))); + initq(prt); + + ERTS_SMP_LC_ASSERT(erts_lc_is_port_locked(prt)); + + erts_atomic32_set_relb(&prt->state, state); + return prt; +} + +#ifndef ERTS_SMP void erts_port_cleanup(Port *prt) { -#ifdef ERTS_SMP - if (erts_smp_mtx_trylock(prt->lock) == EBUSY) - erts_schedule_misc_op(sched_port_cleanup, (void *) prt); - else -#endif - port_cleanup(prt); + if (prt->drv_ptr && prt->drv_ptr->handle) + erts_ddll_dereference_driver(prt->drv_ptr->handle); + prt->drv_ptr = NULL; + erts_port_dec_refc(prt); } +#endif void -port_cleanup(Port *prt) +erts_port_free(Port *prt) { -#ifdef ERTS_SMP - erts_smp_mtx_t *mtx; -#endif #if defined(ERTS_SMP) || defined(ERTS_ENABLE_LOCK_CHECK) - erts_aint32_t state = erts_smp_atomic32_read_nob(&prt->state); + erts_aint32_t state = erts_atomic32_read_nob(&prt->state); #endif - erts_driver_t *driver; - - ERTS_SMP_LC_ASSERT(erts_lc_is_port_locked(prt)); - driver = prt->drv_ptr; - prt->drv_ptr = NULL; - ASSERT(driver); - - ERTS_LC_ASSERT(state & ERTS_PORT_SFLG_FREE_SCHEDULED); - ERTS_LC_ASSERT(state & ERTS_PORT_SFLG_PORT_DEBUG); + ERTS_LC_ASSERT(state & (ERTS_PORT_SFLG_FREE_SCHEDULED + | ERTS_PORT_SFLG_INITIALIZING)); + ASSERT(state & ERTS_PORT_SFLG_PORT_DEBUG); ERTS_LC_ASSERT(!(state & ERTS_PORT_SFLG_FREE)); - ERTS_LC_ASSERT(erts_smp_atomic32_read_nob(&prt->common.refc) == 0); - #ifdef ERTS_SMP - mtx = prt->lock; - ASSERT(mtx); - - prt->lock = NULL; - - erts_smp_mtx_unlock(mtx); + ERTS_LC_ASSERT(erts_atomic32_read_nob(&prt->common.refc) == 0); +#else + ERTS_LC_ASSERT(erts_atomic32_read_nob(&prt->refc) == 0); #endif - erts_smp_atomic32_set_relb(&prt->state, ERTS_PORT_SFLG_FREE); - #ifdef ERTS_SMP - if (state & ERTS_PORT_SFLG_PORT_SPECIFIC_LOCK) { - erts_smp_mtx_destroy(mtx); - erts_free(ERTS_ALC_T_PORT_LOCK, mtx); - } -#endif + ASSERT(prt->lock); + if (state & ERTS_PORT_SFLG_PORT_SPECIFIC_LOCK) + erts_mtx_destroy(prt->lock); - if (driver->handle) - erts_ddll_dereference_driver(driver->handle); + /* + * We cannot dereference a driver using driver + * locking until here in smp case. Otherwise, + * the driver lock may still be in use by others. + * + * In the non-smp case we cannot do it here since + * this function may be called by non-scheduler + * threads. This is done in erts_port_cleanup() + * in the non-smp case. + */ + if (prt->drv_ptr->handle) + erts_ddll_dereference_driver(prt->drv_ptr->handle); +#endif + erts_atomic32_set_nob(&prt->state, ERTS_PORT_SFLG_FREE); + erts_free(ERTS_ALC_T_PORT, prt); } - /* ** Initialize v_start to point to the small fixed vector. ** Once (reallocated) we never reset the pointer to the small vector @@ -416,73 +422,7 @@ static void stopq(Port* prt) if (prt->port_data_lock) { driver_pdl_unlock(prt->port_data_lock); driver_pdl_dec_refc(prt->port_data_lock); - prt->port_data_lock = NULL; - } -} - - - -static void -setup_port(Port* prt, Eterm pid, erts_driver_t *driver, - ErlDrvData drv_data, char *name, erts_aint32_t xstate) -{ - ErtsRunQueue *runq = erts_get_runq_current(NULL); - char *new_name, *old_name; -#ifdef DEBUG - /* Make sure the debug flags survives until port is freed */ - xstate |= ERTS_PORT_SFLG_PORT_DEBUG; -#endif - ASSERT(runq); - ERTS_SMP_LC_ASSERT(erts_lc_is_port_locked(prt)); - - - new_name = (char*) erts_alloc(ERTS_ALC_T_PORT_NAME, sys_strlen(name)+1); - sys_strcpy(new_name, name); - erts_smp_runq_lock(runq); - prt->snapshot = erts_smp_atomic32_read_nob(&erts_ports_snapshot); - old_name = prt->name; - prt->name = new_name; -#ifdef ERTS_SMP - erts_smp_atomic_set_nob(&prt->run_queue, (erts_aint_t) runq); -#endif - ASSERT(!prt->drv_ptr); - prt->drv_ptr = driver; - erts_smp_atomic32_set_relb(&prt->state, - ERTS_PORT_SFLG_CONNECTED | xstate); - erts_smp_runq_unlock(runq); -#ifdef ERTS_SMP - ASSERT(!prt->xports); -#endif - if (old_name) { - erts_free(ERTS_ALC_T_PORT_NAME, (void *) old_name); } - - prt->control_flags = 0; - prt->connected = pid; - prt->drv_data = (SWord) drv_data; - prt->bytes_in = 0; - prt->bytes_out = 0; - prt->dist_entry = NULL; - prt->common.u.alive.reg = NULL; -#ifdef ERTS_SMP - prt->common.u.alive.ptimer = NULL; -#else - sys_memset(&prt->common.u.alive.tm, 0, sizeof(ErlTimer)); -#endif - erts_port_task_handle_init(&prt->timeout_task); - prt->suspended = NULL; - sys_strcpy(prt->name, name); - ERTS_P_LINKS(prt) = NULL; - ERTS_P_MONITORS(prt) = NULL; - prt->linebuf = NULL; - prt->bp = NULL; - prt->data = am_undefined; - /* Set default tracing */ - erts_get_default_tracing(&ERTS_TRACE_FLAGS(prt), &ERTS_TRACER_PROC(prt)); - - prt->psd = NULL; - - initq(prt); } void @@ -493,7 +433,7 @@ erts_wake_process_later(Port *prt, Process *process) ERTS_SMP_LC_ASSERT(erts_lc_is_port_locked(prt)); - if (erts_smp_atomic32_read_nob(&prt->state) & ERTS_PORT_SFLGS_DEAD) + if (erts_atomic32_read_nob(&prt->state) & ERTS_PORT_SFLGS_DEAD) return; for (p = &(prt->suspended); *p != NULL; p = &((*p)->next)) @@ -513,36 +453,34 @@ erts_wake_process_later(Port *prt, Process *process) (*error_number_ptr must contain either BADARG or SYSTEM_LIMIT). The driver start function must obey the same conventions. */ -int +Port * erts_open_driver(erts_driver_t* driver, /* Pointer to driver. */ Eterm pid, /* Current process. */ char* name, /* Driver name. */ SysDriverOpts* opts, /* Options. */ - int *error_number_ptr) /* errno in case -2 is returned */ + int *error_type_ptr, /* error type */ + int *error_number_ptr) /* errno in case of error type -2 */ { - int port_num; - int port_ix; + +#undef ERTS_OPEN_DRIVER_RET +#define ERTS_OPEN_DRIVER_RET(Prt, EType, ENo) \ + do { \ + if (error_type_ptr) \ + *error_type_ptr = (EType); \ + if (error_number_ptr) \ + *error_number_ptr = (ENo); \ + return (Prt); \ + } while (0) + ErlDrvData drv_data = 0; - erts_aint32_t xstate = 0; Port *port; int fpe_was_unmasked; - - if (error_number_ptr) - *error_number_ptr = 0; + int error_type, error_number; + int port_errno = 0; + erts_mtx_t *driver_lock = NULL; ERTS_SMP_CHK_NO_PROC_LOCKS; - if ((port_num = get_free_port()) < 0) { - if (error_number_ptr) { - *error_number_ptr = SYSTEM_LIMIT; - } - return -3; - } - - port_ix = port_num & erts_port_tab_index_mask; - port = &erts_port[port_ix]; - port->common.id = make_internal_port(port_num); - erts_smp_mtx_lock(&erts_driver_list_lock); if (!driver) { for (driver = driver_list; driver; driver = driver->next) { @@ -551,9 +489,7 @@ erts_open_driver(erts_driver_t* driver, /* Pointer to driver. */ } if (!driver) { erts_smp_mtx_unlock(&erts_driver_list_lock); - if (error_number_ptr) - *error_number_ptr = BADARG; - return -3; + ERTS_OPEN_DRIVER_RET(NULL, -3, BADARG); } } if (driver == &spawn_driver) { @@ -598,32 +534,11 @@ erts_open_driver(erts_driver_t* driver, /* Pointer to driver. */ if (driver == NULL || (driver != &spawn_driver && opts->exit_status)) { erts_smp_mtx_unlock(&erts_driver_list_lock); - if (error_number_ptr) { - *error_number_ptr = BADARG; - } - /* Need to mark the port as free again */ - ERTS_LC_ASSERT(erts_smp_atomic32_read_nob(&port->common.refc) == 2); - erts_smp_atomic32_set_nob(&port->common.refc, 0); - erts_smp_atomic32_set_relb(&port->state, ERTS_PORT_SFLG_FREE); - return -3; + ERTS_OPEN_DRIVER_RET(NULL, -3, BADARG); } - /* - * We'll set up the port before calling the start function, - * to allow message sending and setting timers in the start function. - */ - #ifdef ERTS_SMP - ASSERT(!port->lock); - port->lock = driver->lock; - if (!port->lock) { - port->lock = erts_alloc(ERTS_ALC_T_PORT_LOCK, - sizeof(erts_smp_mtx_t)); - erts_smp_mtx_init_x(port->lock, - "port_lock", - port->common.id); - xstate |= ERTS_PORT_SFLG_PORT_SPECIFIC_LOCK; - } + driver_lock = driver->lock; #endif if (driver->handle != NULL) { @@ -632,18 +547,32 @@ erts_open_driver(erts_driver_t* driver, /* Pointer to driver. */ } erts_smp_mtx_unlock(&erts_driver_list_lock); -#ifdef ERTS_SMP - erts_smp_mtx_lock(port->lock); -#endif + /* + * We'll set up the port before calling the start function, + * to allow message sending and setting timers in the start function. + */ - setup_port(port, pid, driver, drv_data, name, xstate); + port = create_port(name, driver, driver_lock, pid, &port_errno); + if (!port) { + if (driver->handle) { + erts_smp_mtx_lock(&erts_driver_list_lock); + erts_ddll_decrement_port_count(driver->handle); + erts_smp_mtx_unlock(&erts_driver_list_lock); + erts_ddll_dereference_driver(driver->handle); + } + if (port_errno) + ERTS_OPEN_DRIVER_RET(NULL, -2, port_errno); + else + ERTS_OPEN_DRIVER_RET(NULL, -3, SYSTEM_LIMIT); + } if (IS_TRACED_FL(port, F_TRACE_PORTS)) { trace_port_open(port, pid, am_atom_put(port->name, strlen(port->name))); } - + + error_number = error_type = 0; if (driver->start) { if (IS_TRACED_FL(port, F_TRACE_SCHED_PORTS)) { trace_sched_ports_where(port, am_in, am_start); @@ -656,15 +585,28 @@ erts_open_driver(erts_driver_t* driver, /* Pointer to driver. */ } #endif fpe_was_unmasked = erts_block_fpe(); - drv_data = (*driver->start)((ErlDrvPort)(port_ix), - name, opts); + drv_data = (*driver->start)((ErlDrvPort) port, name, opts); + if (((SWord) drv_data) == -1) + error_type = -1; + else if (((SWord) drv_data) == -2) { + /* + * We need to save errno quickly after the + * call to the 'start' callback before + * something else modify it. + */ + error_type = -2; + error_number = errno; + } + else if (((SWord) drv_data) == -3) { + error_type = -3; + error_number = BADARG; + } + erts_unblock_fpe(fpe_was_unmasked); port->caller = NIL; if (IS_TRACED_FL(port, F_TRACE_SCHED_PORTS)) { trace_sched_ports_where(port, am_out, am_start); } - if (error_number_ptr && ((SWord) drv_data) == (SWord) -2) - *error_number_ptr = errno; #ifdef ERTS_SMP if (port->xports) erts_smp_xports_unlock(port); @@ -672,15 +614,7 @@ erts_open_driver(erts_driver_t* driver, /* Pointer to driver. */ #endif } - if (((SWord)drv_data) == -1 || - ((SWord)drv_data) == -2 || - ((SWord)drv_data) == -3) { - int res = (int) ((SWord) drv_data); - - if (res == -3 && error_number_ptr) { - *error_number_ptr = BADARG; - } - + if (error_type) { /* * Must clean up the port. */ @@ -690,7 +624,6 @@ erts_open_driver(erts_driver_t* driver, /* Pointer to driver. */ erts_cancel_timer(&(port->common.u.alive.tm)); #endif stopq(port); - kill_port(port); if (port->linebuf != NULL) { erts_free(ERTS_ALC_T_LINEBUF, (void *) port->linebuf); @@ -701,11 +634,14 @@ erts_open_driver(erts_driver_t* driver, /* Pointer to driver. */ erts_ddll_decrement_port_count(driver->handle); erts_smp_mtx_unlock(&erts_driver_list_lock); } + kill_port(port); erts_port_release(port); - return res; + ERTS_OPEN_DRIVER_RET(NULL, error_type, error_number); } - port->drv_data = (SWord) drv_data; - return port_ix; + port->drv_data = (UWord) drv_data; + ERTS_OPEN_DRIVER_RET(port, 0, 0); + +#undef ERTS_OPEN_DRIVER_RET } #ifdef ERTS_SMP @@ -734,15 +670,21 @@ driver_create_port(ErlDrvPort creator_port_ix, /* Creating port */ Port* port; erts_driver_t *driver; Process *rp; - int port_num; - Eterm port_id; - erts_aint32_t xstate = 0; + erts_mtx_t *driver_lock = NULL; ERTS_SMP_CHK_NO_PROC_LOCKS; + /* Need to be called from a scheduler thread */ + if (!erts_get_scheduler_id()) + return ERTS_INVALID_ERL_DRV_PORT; + creator_port = erts_drvport2port(creator_port_ix, NULL); if (!creator_port) - return (ErlDrvTermData) -1; + return ERTS_INVALID_ERL_DRV_PORT; + + rp = erts_proc_lookup(pid); + if (!rp) + return ERTS_INVALID_ERL_DRV_PORT; ERTS_SMP_LC_ASSERT(erts_lc_is_port_locked(creator_port)); @@ -750,55 +692,61 @@ driver_create_port(ErlDrvPort creator_port_ix, /* Creating port */ erts_smp_mtx_lock(&erts_driver_list_lock); if (!erts_ddll_driver_ok(driver->handle)) { erts_smp_mtx_unlock(&erts_driver_list_lock); - return (ErlDrvTermData) -1; + return ERTS_INVALID_ERL_DRV_PORT; } - rp = erts_pid2proc(NULL, 0, pid, ERTS_PROC_LOCK_LINK); - if (!rp) { - erts_smp_mtx_unlock(&erts_driver_list_lock); - return (ErlDrvTermData) -1; /* pid does not exist */ + if (driver->handle != NULL) { + erts_ddll_increment_port_count(driver->handle); + erts_ddll_reference_referenced_driver(driver->handle); + } + +#ifdef ERTS_SMP + driver_lock = driver->lock; +#endif + + erts_smp_mtx_unlock(&erts_driver_list_lock); + + port = create_port(name, driver, driver_lock, pid, NULL); + if (!port) { + if (driver->handle) { + erts_smp_mtx_lock(&erts_driver_list_lock); + erts_ddll_decrement_port_count(driver->handle); + erts_smp_mtx_unlock(&erts_driver_list_lock); + erts_ddll_dereference_driver(driver->handle); + } + return ERTS_INVALID_ERL_DRV_PORT; } - if ((port_num = get_free_port()) < 0) { - errno = SYSTEM_LIMIT; + ERTS_SMP_LC_ASSERT(erts_lc_is_port_locked(port)); + + erts_smp_proc_lock(rp, ERTS_PROC_LOCK_LINK); + if (ERTS_PROC_IS_EXITING(rp)) { erts_smp_proc_unlock(rp, ERTS_PROC_LOCK_LINK); - erts_smp_mtx_unlock(&erts_driver_list_lock); - return (ErlDrvTermData) -1; + if (driver->handle) { + erts_smp_mtx_lock(&erts_driver_list_lock); + erts_ddll_decrement_port_count(driver->handle); + erts_smp_mtx_unlock(&erts_driver_list_lock); + } + kill_port(port); + erts_port_release(port); + return ERTS_INVALID_ERL_DRV_PORT; } - port_id = make_internal_port(port_num); - port = &erts_port[port_num & erts_port_tab_index_mask]; + erts_add_link(&ERTS_P_LINKS(port), LINK_PID, pid); + erts_add_link(&ERTS_P_LINKS(rp), LINK_PID, port->common.id); + erts_smp_proc_unlock(rp, ERTS_PROC_LOCK_LINK); #ifdef ERTS_SMP - ASSERT(!port->lock); - port->lock = driver->lock; - if (!port->lock) { + if (!driver_lock) { ErtsXPortsList *xplp = xports_list_alloc(); xplp->port = port; xplp->next = creator_port->xports; creator_port->xports = xplp; - port->lock = erts_alloc(ERTS_ALC_T_PORT_LOCK, - sizeof(erts_smp_mtx_t)); - erts_smp_mtx_init_locked_x(port->lock, "port_lock", port_id); - xstate |= ERTS_PORT_SFLG_PORT_SPECIFIC_LOCK; } - #endif - if (driver->handle != NULL) { - erts_ddll_increment_port_count(driver->handle); - erts_ddll_reference_referenced_driver(driver->handle); - } - erts_smp_mtx_unlock(&erts_driver_list_lock); + port->drv_data = (UWord) drv_data; - ERTS_SMP_LC_ASSERT(erts_lc_is_port_locked(port)); - - setup_port(port, pid, driver, drv_data, name, xstate); - port->common.id = port_id; - - erts_add_link(&ERTS_P_LINKS(port), LINK_PID, pid); - erts_add_link(&ERTS_P_LINKS(rp), LINK_PID, port_id); - erts_smp_proc_unlock(rp, ERTS_PROC_LOCK_LINK); - return port_num & erts_port_tab_index_mask; + return (ErlDrvPort) port; } #ifdef ERTS_SMP @@ -1282,15 +1230,22 @@ int erts_write_to_port(Eterm caller_id, Port *p, Eterm list) } } +#ifdef ERTS_SMP +static void +release_port(void *vport) +{ + erts_port_dec_refc((Port *) vport); +} +#endif + /* initialize the port array */ void init_io(void) { - int i; ErlDrvEntry** dp; char maxports[21]; /* enough for any 64-bit integer */ size_t maxportssize = sizeof(maxports); Uint ports_bits = ERTS_PORTS_BITS; - Sint port_extra_shift; + int port_tab_size; #ifdef ERTS_SMP init_xports_list_alloc(); @@ -1299,66 +1254,39 @@ void init_io(void) pdl_init(); if (erts_sys_getenv("ERL_MAX_PORTS", maxports, &maxportssize) == 0) - erts_max_ports = atoi(maxports); + port_tab_size = atoi(maxports); else - erts_max_ports = sys_max_files(); + port_tab_size = sys_max_files(); - if (erts_max_ports > ERTS_MAX_PORTS) - erts_max_ports = ERTS_MAX_PORTS; - if (erts_max_ports < 1024) - erts_max_ports = 1024; + if (port_tab_size > ERTS_MAX_PORTS) + port_tab_size = ERTS_MAX_PORTS; + if (port_tab_size < 1024) + port_tab_size = 1024; if (erts_use_r9_pids_ports) { ports_bits = ERTS_R9_PORTS_BITS; - if (erts_max_ports > ERTS_MAX_R9_PORTS) - erts_max_ports = ERTS_MAX_R9_PORTS; + if (port_tab_size > ERTS_MAX_R9_PORTS) + port_tab_size = ERTS_MAX_R9_PORTS; } - port_extra_shift = erts_fit_in_bits_int32(erts_max_ports - 1); - port_num_mask = (1 << ports_bits) - 1; - - erts_port_tab_index_mask = ~(~((Uint) 0) << port_extra_shift); - erts_max_ports = 1 << port_extra_shift; - erts_smp_mtx_init(&erts_driver_list_lock,"driver_list"); driver_list = NULL; erts_smp_tsd_key_create(&driver_list_lock_status_key); erts_smp_tsd_key_create(&driver_list_last_error_key); - if (erts_max_ports * sizeof(Port) <= erts_max_ports) { - /* More memory needed than the whole address space. */ - erts_alloc_enomem(ERTS_ALC_T_PORT_TABLE, ~((Uint) 0)); - } - - erts_port = (Port *) erts_alloc(ERTS_ALC_T_PORT_TABLE, - erts_max_ports * sizeof(Port)); - - erts_smp_atomic_init_nob(&erts_bytes_out, 0); - erts_smp_atomic_init_nob(&erts_bytes_in, 0); - - for (i = 0; i < erts_max_ports; i++) { - erts_port_task_init_sched(&erts_port[i].sched); - erts_smp_atomic32_init_nob(&erts_port[i].common.refc, 0); + erts_ptab_init_table(&erts_port, + ERTS_ALC_T_PORT_TABLE, #ifdef ERTS_SMP - erts_port[i].lock = NULL; - erts_port[i].xports = NULL; - erts_smp_spinlock_init_x(&erts_port[i].state_lck, "port_state", make_small(i)); + release_port, +#else + NULL, #endif - ERTS_TRACER_PROC(&erts_port[i]) = NIL; - ERTS_TRACE_FLAGS(&erts_port[i]) = 0; - - erts_port[i].drv_ptr = NULL; - erts_smp_atomic32_init_nob(&erts_port[i].state, ERTS_PORT_SFLG_FREE); - erts_port[i].name = NULL; - ERTS_P_LINKS(&erts_port[i]) = NULL; - ERTS_P_MONITORS(&erts_port[i]) = NULL; - erts_port[i].linebuf = NULL; - erts_port[i].port_data_lock = NULL; - } + (ErtsPTabElementCommon *) &erts_invalid_port.common, + port_tab_size, + "port_table"); - erts_smp_atomic32_init_nob(&erts_ports_snapshot, (erts_aint32_t) 0); - last_port_num = 0; - erts_smp_spinlock_init(&get_free_port_lck, "get_free_port"); + erts_smp_atomic_init_nob(&erts_bytes_out, 0); + erts_smp_atomic_init_nob(&erts_bytes_in, 0); sys_init_io(); @@ -1769,7 +1697,7 @@ deliver_vec_message(Port* prt, /* Port */ if (!rp) return; - state = erts_smp_atomic32_read_nob(&prt->state); + state = erts_atomic32_read_nob(&prt->state); /* * Calculate the exact number of heap words needed. */ @@ -1907,7 +1835,7 @@ static void flush_port(Port *p) ASSERT(!p->xports); #endif } - if ((erts_smp_atomic32_read_nob(&p->state) & ERTS_PORT_SFLGS_DEAD) == 0 + if ((erts_atomic32_read_nob(&p->state) & ERTS_PORT_SFLGS_DEAD) == 0 && is_port_ioq_empty(p)) { terminate_port(p); } @@ -1929,8 +1857,8 @@ terminate_port(Port *prt) ASSERT(!ERTS_P_MONITORS(prt)); /* state may be altered by kill_port() below */ - state = erts_smp_atomic32_read_band_nob(&prt->state, - ~ERTS_PORT_SFLG_SEND_CLOSED); + state = erts_atomic32_read_band_nob(&prt->state, + ~ERTS_PORT_SFLG_SEND_CLOSED); if (state & ERTS_PORT_SFLG_SEND_CLOSED) { send_closed_port_id = prt->common.id; connected_id = prt->connected; @@ -1981,6 +1909,8 @@ terminate_port(Port *prt) if (prt->psd) erts_free(ERTS_ALC_T_PRTSD, prt->psd); + ASSERT(prt->dist_entry == NULL); + kill_port(prt); /* @@ -1989,13 +1919,11 @@ terminate_port(Port *prt) */ if ((state & ERTS_PORT_SFLG_HALT) && (erts_smp_atomic32_dec_read_nob(&erts_halt_progress) == 0)) { - erts_smp_port_unlock(prt); /* We will exit and never return */ + erts_port_release(prt); /* We will exit and never return */ erl_exit_flush_async(erts_halt_code, ""); } if (is_internal_port(send_closed_port_id)) deliver_result(send_closed_port_id, connected_id, am_closed); - - ASSERT(prt->dist_entry == NULL); } void @@ -2130,7 +2058,7 @@ erts_do_exit_port(Port *p, Eterm from, Eterm reason) } #endif - state = erts_smp_atomic32_read_nob(&p->state); + state = erts_atomic32_read_nob(&p->state); if ((state & (ERTS_PORT_SFLGS_DEAD | ERTS_PORT_SFLG_EXITING | ERTS_PORT_SFLG_IMMORTAL)) @@ -2149,12 +2077,12 @@ erts_do_exit_port(Port *p, Eterm from, Eterm reason) * Setting the port to not busy here, frees the list of pending * processes and makes them runnable. */ - set_busy_port((ErlDrvPort)internal_port_index(p->common.id), 0); + set_busy_port((ErlDrvPort) p, 0); if (p->common.u.alive.reg != NULL) (void) erts_unregister_name(NULL, 0, p, p->common.u.alive.reg->name); - state = erts_smp_atomic32_read_bor_relb(&p->state, ERTS_PORT_SFLG_EXITING); + state = erts_atomic32_read_bor_relb(&p->state, ERTS_PORT_SFLG_EXITING); { SweepContext sc = {p->common.id, rreason}; @@ -2174,16 +2102,16 @@ erts_do_exit_port(Port *p, Eterm from, Eterm reason) erts_do_net_exits(p->dist_entry, rreason); erts_deref_dist_entry(p->dist_entry); p->dist_entry = NULL; - erts_smp_atomic32_read_band_relb(&p->state, - ~ERTS_PORT_SFLG_DISTRIBUTION); + erts_atomic32_read_band_relb(&p->state, + ~ERTS_PORT_SFLG_DISTRIBUTION); } if ((reason != am_kill) && !is_port_ioq_empty(p)) { /* must turn exiting flag off */ - erts_smp_atomic32_read_bset_relb(&p->state, - (ERTS_PORT_SFLG_EXITING - | ERTS_PORT_SFLG_CLOSING), - ERTS_PORT_SFLG_CLOSING); + erts_atomic32_read_bset_relb(&p->state, + (ERTS_PORT_SFLG_EXITING + | ERTS_PORT_SFLG_CLOSING), + ERTS_PORT_SFLG_CLOSING); flush_port(p); } else { @@ -2232,8 +2160,8 @@ void erts_port_command(Process *proc, if ((pid = port->connected) == tp[1]) { /* PID must be connected */ if (tp[2] == am_close) { - erts_smp_atomic32_read_bor_relb(&port->state, - ERTS_PORT_SFLG_SEND_CLOSED); + erts_atomic32_read_bor_relb(&port->state, + ERTS_PORT_SFLG_SEND_CLOSED); erts_do_exit_port(port, pid, am_normal); #ifdef USE_VM_PROBES @@ -2456,16 +2384,15 @@ static void prt_one_lnk(ErtsLink *lnk, void *vprtd) } void -print_port_info(int to, void *arg, int i) +print_port_info(Port *p, int to, void *arg) { - Port* p = &erts_port[i]; - erts_aint32_t state = erts_smp_atomic32_read_nob(&p->state); + erts_aint32_t state = erts_atomic32_read_nob(&p->state); if (state & ERTS_PORT_SFLGS_DEAD) return; erts_print(to, arg, "=port:%T\n", p->common.id); - erts_print(to, arg, "Slot: %d\n", i); + erts_print(to, arg, "Slot: %d\n", internal_port_index(p->common.id)); if (state & ERTS_PORT_SFLG_CONNECTED) { erts_print(to, arg, "Connected: %T", p->connected); erts_print(to, arg, "\n"); @@ -2505,43 +2432,46 @@ print_port_info(int to, void *arg, int i) void set_busy_port(ErlDrvPort port_num, int on) { + Port *prt; #ifdef USE_VM_PROBES DTRACE_CHARBUF(port_str, 16); #endif ERTS_SMP_CHK_NO_PROC_LOCKS; - ERTS_SMP_LC_ASSERT(erts_lc_is_port_locked(&erts_port[port_num])); + prt = erts_drvport2port_raw(port_num); + if (!prt) + return; if (on) { - erts_smp_atomic32_read_bor_relb(&erts_port[port_num].state, - ERTS_PORT_SFLG_PORT_BUSY); + erts_atomic32_read_bor_relb(&prt->state, + ERTS_PORT_SFLG_PORT_BUSY); #ifdef USE_VM_PROBES if (DTRACE_ENABLED(port_busy)) { erts_snprintf(port_str, sizeof(port_str), - "%T", erts_port[port_num].id); + "%T", prt->id); DTRACE1(port_busy, port_str); } #endif } else { - ErtsProcList* plp = erts_port[port_num].suspended; - erts_smp_atomic32_read_band_relb(&erts_port[port_num].state, - ~ERTS_PORT_SFLG_PORT_BUSY); - erts_port[port_num].suspended = NULL; + ErtsProcList* plp = prt->suspended; + erts_atomic32_read_band_relb(&prt->state, + ~ERTS_PORT_SFLG_PORT_BUSY); + prt->suspended = NULL; #ifdef USE_VM_PROBES if (DTRACE_ENABLED(port_not_busy)) { erts_snprintf(port_str, sizeof(port_str), - "%T", erts_port[port_num].id); + "%T", prt->id); DTRACE1(port_not_busy, port_str); } #endif - if (erts_port[port_num].dist_entry) { + if (prt->dist_entry) { /* * Processes suspended on distribution ports are * normally queued on the dist entry. */ - erts_dist_port_not_busy(&erts_port[port_num]); + erts_dist_port_not_busy(prt); } /* @@ -2587,10 +2517,9 @@ set_busy_port(ErlDrvPort port_num, int on) void set_port_control_flags(ErlDrvPort port_num, int flags) { - - ERTS_SMP_LC_ASSERT(erts_lc_is_port_locked(&erts_port[port_num])); - - erts_port[port_num].control_flags = flags; + Port *prt = erts_drvport2port_raw(port_num); + if (prt) + prt->control_flags = flags; } int get_port_flags(ErlDrvPort ix) @@ -2649,7 +2578,7 @@ int async_ready(Port *p, void* data) ERTS_SMP_CHK_NO_PROC_LOCKS; if (p) { - erts_aint32_t state = erts_smp_atomic32_read_nob(&p->state); + erts_aint32_t state = erts_atomic32_read_nob(&p->state); ERTS_SMP_LC_ASSERT(erts_lc_is_port_locked(p)); ASSERT(!(state & ERTS_PORT_SFLGS_DEAD)); if (p->drv_ptr->ready_async != NULL) { @@ -2697,7 +2626,7 @@ erts_stale_drv_select(Eterm port, int deselect) { char *type; - ErlDrvPort drv_port = internal_port_index(port); + ErlDrvPort drv_port = (ErlDrvPort) erts_port_lookup_raw(port); ErtsPortNames *pnp = erts_get_port_names(port); erts_dsprintf_buf_t *dsbufp; @@ -2737,16 +2666,16 @@ erts_stale_drv_select(Eterm port, ErtsPortNames * erts_get_port_names(Eterm id) { + Port *prt = erts_port_lookup_raw(id); ErtsPortNames *pnp; ASSERT(is_nil(id) || is_internal_port(id)); - - if (is_not_internal_port(id)) { + + if (!prt) { pnp = erts_alloc(ERTS_ALC_T_PORT_NAMES, sizeof(ErtsPortNames)); pnp->name = NULL; pnp->driver_name = NULL; } else { - Port* prt = &erts_port[internal_port_index(id)]; int do_realloc = 1; int len = -1; size_t pnp_len = sizeof(ErtsPortNames); @@ -2762,17 +2691,10 @@ erts_get_port_names(Eterm id) pnp_len = sizeof(ErtsPortNames) + len; pnp = erts_alloc(ERTS_ALC_T_PORT_NAMES, pnp_len); } - erts_smp_port_minor_lock(prt); - if (id != prt->common.id) { - len = nlen = 0; - name = driver_name = NULL; - } - else { - name = prt->name; - len = nlen = name ? sys_strlen(name) + 1 : 0; - driver_name = (prt->drv_ptr ? prt->drv_ptr->name : NULL); - len += driver_name ? sys_strlen(driver_name) + 1 : 0; - } + name = prt->name; + len = nlen = name ? sys_strlen(name) + 1 : 0; + driver_name = (prt->drv_ptr ? prt->drv_ptr->name : NULL); + len += driver_name ? sys_strlen(driver_name) + 1 : 0; if (len <= pnp_len - sizeof(ErtsPortNames)) { if (!name) pnp->name = NULL; @@ -2790,7 +2712,6 @@ erts_get_port_names(Eterm id) } do_realloc = 0; } - erts_smp_port_minor_unlock(prt); } while (do_realloc); } return pnp; @@ -2827,7 +2748,7 @@ ErlDrvTermData driver_mk_term_nil(void) return driver_term_nil; } -void driver_report_exit(int ix, int status) +void driver_report_exit(ErlDrvPort ix, int status) { Port* prt = erts_drvport2port(ix, NULL); Eterm* hp; @@ -2868,27 +2789,6 @@ void driver_report_exit(int ix, int status) erts_smp_proc_dec_refc(rp); } - -static ERTS_INLINE int -deliver_term_check_port(ErlDrvPort drvport) -{ - int res; - int ix = (int) drvport; - if (ix < 0 || erts_max_ports <= ix) - res = -1; /* invalid */ - else { - Port* prt = &erts_port[ix]; - erts_aint32_t state = erts_smp_atomic32_read_nob(&prt->state); - if (!(state & ERTS_PORT_SFLGS_INVALID_LOOKUP)) - res = 1; /* ok */ - else if (state & ERTS_PORT_SFLG_CLOSING) - res = 0; /* closing */ - else - res = -1; /* invalid (dead) */ - } - return res; -} - #define ERTS_B2T_STATES_DEF_STATES_SZ 5 #define ERTS_B2T_STATES_DEF_STATES_INC 100 @@ -2976,10 +2876,7 @@ cleanup_b2t_states(struct b2t_states__ *b2tsp) */ static int -driver_deliver_term(ErlDrvPort port, - Eterm to, - ErlDrvTermData* data, - int len) +driver_deliver_term(Eterm to, ErlDrvTermData* data, int len) { #define ERTS_DDT_FAIL do { res = -1; goto done; } while (0) Uint need = 0; @@ -3181,11 +3078,8 @@ driver_deliver_term(ErlDrvPort port, b2t.ix = 0; /* - * The term is OK. Go ahead and validate the port and process. + * The term is OK. Go ahead and validate the process. */ - res = deliver_term_check_port(port); - if (res <= 0) - goto done; /* * Increase refc on proc if done from a non-scheduler thread. @@ -3449,25 +3343,29 @@ driver_deliver_term(ErlDrvPort port, #undef ERTS_DDT_FAIL } - int -driver_output_term(ErlDrvPort ix, ErlDrvTermData* data, int len) +driver_output_term(ErlDrvPort drvport, ErlDrvTermData* data, int len) { - Port* prt = erts_drvport2port(ix, NULL); - + erts_aint32_t state; + Port* prt = erts_drvport2port(drvport, &state); + if (!prt) + return -1; /* invalid (dead) */ ERTS_SMP_CHK_NO_PROC_LOCKS; ERTS_SMP_LC_ASSERT(erts_lc_is_port_locked(prt)); - - if (prt == NULL) - return -1; - return driver_deliver_term(ix, prt->connected, data, len); + if (!(state & ERTS_PORT_SFLGS_INVALID_LOOKUP)) + return driver_deliver_term(prt->connected, data, len); + else if (state & ERTS_PORT_SFLG_CLOSING) + return 0; /* closing */ + else + return -1; /* invalid (dead) */ } int driver_send_term(ErlDrvPort ix, ErlDrvTermData to, ErlDrvTermData* data, int len) { - return driver_deliver_term(ix, to, data, len); + /* driver_send_term() assume port is ok... */ + return driver_deliver_term(to, data, len); } @@ -3817,6 +3715,7 @@ static ERTS_INLINE void pdl_destroy(ErlDrvPDL pdl) { ERTS_LC_ASSERT(driver_pdl_get_refc(pdl) == 0); erts_mtx_destroy(&pdl->mtx); + erts_port_dec_refc(pdl->prt); erts_free(ERTS_ALC_T_PORT_DATA_LOCK, pdl); } @@ -3861,6 +3760,8 @@ driver_pdl_create(ErlDrvPort dp) sizeof(struct erl_drv_port_data_lock)); erts_mtx_init(&pdl->mtx, "port_data_lock"); pdl_init_refc(pdl); + erts_port_inc_refc(pp); + pdl->prt = pp; pp->port_data_lock = pdl; #ifdef HARDDEBUG erts_fprintf(stderr, "driver_pdl_create(%T) -> 0x%08X\r\n",pp->common.id,(unsigned) pdl); @@ -4317,7 +4218,7 @@ int driver_set_timer(ErlDrvPort ix, unsigned long t) if (prt == NULL) return -1; - ERTS_SMP_LC_ASSERT(erts_lc_is_port_locked(prt)); + if (prt->drv_ptr->timeout == NULL) return -1; drv_cancel_timer(prt); @@ -4424,7 +4325,7 @@ static int do_driver_monitor_process(Port *prt, /* * This can be called from a non scheduler thread iff a port_data_lock exists */ -int driver_monitor_process(ErlDrvPort port, +int driver_monitor_process(ErlDrvPort drvport, ErlDrvTermData process, ErlDrvMonitor *monitor) { @@ -4434,15 +4335,12 @@ int driver_monitor_process(ErlDrvPort port, #if !HEAP_ON_C_STACK || (defined(ERTS_SMP) && defined(ERTS_ENABLE_LOCK_CHECK)) ErtsSchedulerData *sched = erts_get_scheduler_data(); #endif - int ix = (int) port; - if (ix < 0 || erts_max_ports <= ix) { - return -1; - } - prt = &erts_port[ix]; + + prt = erts_thr_drvport2port_raw(drvport); DRV_MONITOR_LOCK_PDL(prt); - state = erts_smp_atomic32_read_nob(&erts_port[ix].state); + state = erts_atomic32_read_nob(&prt->state); if (state & ERTS_PORT_SFLGS_INVALID_DRIVER_LOOKUP) { DRV_MONITOR_UNLOCK_PDL(prt); @@ -4509,7 +4407,7 @@ static int do_driver_demonitor_process(Port *prt, Eterm *buf, return 0; } -int driver_demonitor_process(ErlDrvPort port, +int driver_demonitor_process(ErlDrvPort drvport, const ErlDrvMonitor *monitor) { Port *prt; @@ -4518,15 +4416,12 @@ int driver_demonitor_process(ErlDrvPort port, #if !HEAP_ON_C_STACK || (defined(ERTS_SMP) && defined(ERTS_ENABLE_LOCK_CHECK)) ErtsSchedulerData *sched = erts_get_scheduler_data(); #endif - int ix = (int) port; - if (ix < 0 || erts_max_ports <= ix) { - return -1; - } - prt = &erts_port[ix]; + + prt = erts_thr_drvport2port_raw(drvport); DRV_MONITOR_LOCK_PDL(prt); - state = erts_smp_atomic32_read_nob(&erts_port[ix].state); + state = erts_atomic32_read_nob(&prt->state); if (state & ERTS_PORT_SFLGS_INVALID_DRIVER_LOOKUP) { DRV_MONITOR_UNLOCK_PDL(prt); @@ -4575,7 +4470,7 @@ static ErlDrvTermData do_driver_get_monitored_process(Port *prt, Eterm *buf, } -ErlDrvTermData driver_get_monitored_process(ErlDrvPort port, +ErlDrvTermData driver_get_monitored_process(ErlDrvPort drvport, const ErlDrvMonitor *monitor) { Port *prt; @@ -4584,15 +4479,12 @@ ErlDrvTermData driver_get_monitored_process(ErlDrvPort port, #if !HEAP_ON_C_STACK || (defined(ERTS_SMP) && defined(ERTS_ENABLE_LOCK_CHECK)) ErtsSchedulerData *sched = erts_get_scheduler_data(); #endif - int ix = (int) port; - if (ix < 0 || erts_max_ports <= ix) { - return driver_term_nil; - } - prt = &erts_port[ix]; + + prt = erts_thr_drvport2port_raw(drvport); DRV_MONITOR_LOCK_PDL(prt); - state = erts_smp_atomic32_read_nob(&erts_port[ix].state); + state = erts_atomic32_read_nob(&prt->state); if (state & ERTS_PORT_SFLGS_INVALID_DRIVER_LOOKUP) { DRV_MONITOR_UNLOCK_PDL(prt); return driver_term_nil; @@ -4711,8 +4603,6 @@ int driver_exit(ErlDrvPort ix, int err) if (prt == NULL) return -1; - ERTS_SMP_LC_ASSERT(erts_lc_is_port_locked(prt)); - rp = erts_pid2proc(NULL, 0, prt->connected, ERTS_PROC_LOCK_LINK); if (rp) { rlnk = erts_remove_link(&ERTS_P_LINKS(rp),prt->common.id); @@ -5007,7 +4897,7 @@ no_event_callback(ErlDrvData drv_data, ErlDrvEvent event, ErlDrvEventData event_ { Port *prt = get_current_port(); report_missing_drv_callback(prt, "Event", "event()"); - driver_event((ErlDrvPort) internal_port_index(prt->common.id), event, NULL); + driver_event((ErlDrvPort) prt, event, NULL); } static void @@ -5015,7 +4905,7 @@ no_ready_input_callback(ErlDrvData drv_data, ErlDrvEvent event) { Port *prt = get_current_port(); report_missing_drv_callback(prt, "Input", "ready_input()"); - driver_select((ErlDrvPort) internal_port_index(prt->common.id), event, + driver_select((ErlDrvPort) prt, event, (ERL_DRV_READ | ERL_DRV_USE_NO_CALLBACK), 0); } @@ -5024,7 +4914,7 @@ no_ready_output_callback(ErlDrvData drv_data, ErlDrvEvent event) { Port *prt = get_current_port(); report_missing_drv_callback(prt, "Output", "ready_output()"); - driver_select((ErlDrvPort) internal_port_index(prt->common.id), event, + driver_select((ErlDrvPort) prt, event, (ERL_DRV_WRITE | ERL_DRV_USE_NO_CALLBACK), 0); } @@ -5059,13 +4949,13 @@ init_driver(erts_driver_t *drv, ErlDrvEntry *de, DE_Handle *handle) drv->lock = NULL; else { drv->lock = erts_alloc(ERTS_ALC_T_DRIVER_LOCK, - sizeof(erts_smp_mtx_t)); - erts_smp_mtx_init_x(drv->lock, - "driver_lock", + sizeof(erts_mtx_t)); + erts_mtx_init_x(drv->lock, + "driver_lock", #if defined(ERTS_ENABLE_LOCK_CHECK) || defined(ERTS_ENABLE_LOCK_COUNT) - am_atom_put(drv->name, sys_strlen(drv->name)) + am_atom_put(drv->name, sys_strlen(drv->name)) #else - NIL + NIL #endif ); } diff --git a/erts/emulator/beam/register.c b/erts/emulator/beam/register.c index feae745749..757e2800e6 100644 --- a/erts/emulator/beam/register.c +++ b/erts/emulator/beam/register.c @@ -182,7 +182,7 @@ int erts_register_name(Process *c_p, Eterm name, Eterm id) return res; erts_smp_proc_unlock(c_p, ERTS_PROC_LOCK_MAIN); if (is_internal_port(id)) { - port = erts_id2port(id, NULL, 0); + port = erts_id2port(id); if (!port) goto done; } @@ -241,7 +241,7 @@ int erts_register_name(Process *c_p, Eterm name, Eterm id) done: reg_write_unlock(); if (port) - erts_smp_port_unlock(port); + erts_port_release(port); if (c_p != proc) { if (proc) erts_smp_proc_unlock(proc, ERTS_PROC_LOCK_MAIN); @@ -397,16 +397,14 @@ erts_whereis_name(Process *c_p, if (!rp || !rp->pt) *port = NULL; else { -#ifndef ERTS_SMP - erts_smp_atomic32_inc_nob(&rp->pt->common.refc); -#else +#ifdef ERTS_SMP if (pending_port == rp->pt) pending_port = NULL; else { if (pending_port) { /* Ahh! Registered port changed while reg lock was unlocked... */ - erts_smp_port_unlock(pending_port); + erts_port_release(pending_port); pending_port = NULL; } @@ -418,7 +416,7 @@ erts_whereis_name(Process *c_p, current_c_p_locks = 0; } reg_read_unlock(); - pending_port = erts_id2port(id, NULL, 0); + pending_port = erts_id2port(id); goto restart; } } @@ -432,7 +430,7 @@ erts_whereis_name(Process *c_p, if (c_p && !current_c_p_locks) erts_smp_proc_lock(c_p, c_p_locks); if (pending_port) - erts_smp_port_unlock(pending_port); + erts_port_release(pending_port); #endif reg_read_unlock(); @@ -506,12 +504,10 @@ int erts_unregister_name(Process *c_p, if ((rp = (RegProc*) hash_get(&process_reg, (void*) &r)) != NULL) { if (rp->pt) { if (port != rp->pt) { -#ifndef ERTS_SMP - erts_smp_atomic32_inc_nob(&rp->pt->common.refc); -#else +#ifdef ERTS_SMP if (port) { ASSERT(port != c_prt); - erts_smp_port_unlock(port); + erts_port_release(port); port = NULL; } @@ -523,7 +519,7 @@ int erts_unregister_name(Process *c_p, current_c_p_locks = 0; } reg_write_unlock(); - port = erts_id2port(id, NULL, 0); + port = erts_id2port(id); goto restart; } #endif @@ -569,7 +565,7 @@ int erts_unregister_name(Process *c_p, reg_write_unlock(); if (c_prt != port) { if (port) { - erts_smp_port_unlock(port); + erts_port_release(port); } if (c_prt) { erts_smp_port_lock(c_prt); diff --git a/erts/emulator/beam/register.h b/erts/emulator/beam/register.h index 38e8cfbf28..7170463375 100644 --- a/erts/emulator/beam/register.h +++ b/erts/emulator/beam/register.h @@ -24,26 +24,19 @@ #ifndef __REGPROC_H__ #define __REGPROC_H__ -#ifndef __SYS_H__ #include "sys.h" -#endif - -#ifndef __HASH_H__ #include "hash.h" -#endif - -#ifndef __PROCESS_H__ #include "erl_process.h" -#endif - -struct port; +#define ERL_PORT_GET_PORT_TYPE_ONLY__ +#include "erl_port.h" +#undef ERL_PORT_GET_PORT_TYPE_ONLY__ typedef struct reg_proc { HashBucket bucket; /* MUST BE LOCATED AT TOP OF STRUCT!!! */ Process *p; /* The process registered (only one of this and 'pt' is non-NULL */ - struct port *pt; /* The port registered */ + Port *pt; /* The port registered */ Eterm name; /* Atom name */ } RegProc; @@ -55,12 +48,12 @@ int erts_register_name(Process *, Eterm, Eterm); Eterm erts_whereis_name_to_id(Process *, Eterm); void erts_whereis_name(Process *, ErtsProcLocks, Eterm, Process**, ErtsProcLocks, int, - struct port**); + Port**); Process *erts_whereis_process(Process *, ErtsProcLocks, Eterm, ErtsProcLocks, int); -int erts_unregister_name(Process *, ErtsProcLocks, struct port *, Eterm); +int erts_unregister_name(Process *, ErtsProcLocks, Port *, Eterm); #endif -- cgit v1.2.3