diff options
Diffstat (limited to 'erts/emulator/beam')
| -rw-r--r-- | erts/emulator/beam/bif.tab | 6 | ||||
| -rw-r--r-- | erts/emulator/beam/erl_alloc.types | 1 | ||||
| -rwxr-xr-x | erts/emulator/beam/erl_bif_info.c | 10 | ||||
| -rw-r--r-- | erts/emulator/beam/erl_bif_port.c | 173 | ||||
| -rw-r--r-- | erts/emulator/beam/erl_node_tables.c | 6 | ||||
| -rw-r--r-- | erts/emulator/beam/erl_port.h | 22 | ||||
| -rw-r--r-- | erts/emulator/beam/erl_port_task.c | 28 | ||||
| -rw-r--r-- | erts/emulator/beam/erl_port_task.h | 2 | ||||
| -rw-r--r-- | erts/emulator/beam/erl_process.c | 2 | ||||
| -rw-r--r-- | erts/emulator/beam/erl_process.h | 1 | ||||
| -rw-r--r-- | erts/emulator/beam/io.c | 315 |
11 files changed, 268 insertions, 298 deletions
diff --git a/erts/emulator/beam/bif.tab b/erts/emulator/beam/bif.tab index 8bc994c8c3..dc8e9101de 100644 --- a/erts/emulator/beam/bif.tab +++ b/erts/emulator/beam/bif.tab @@ -154,8 +154,10 @@ bif erts_internal:port_command/3 bif erts_internal:port_control/3 bif erts_internal:port_close/1 bif erts_internal:port_connect/2 -bif erts_internal:port_set_data/2 -bif erts_internal:port_get_data/1 + +# inet_db support +bif erlang:port_set_data/2 +bif erlang:port_get_data/1 # Tracing & debugging. bif erlang:trace_pattern/2 diff --git a/erts/emulator/beam/erl_alloc.types b/erts/emulator/beam/erl_alloc.types index 5e3615ccc2..5a92ab7f24 100644 --- a/erts/emulator/beam/erl_alloc.types +++ b/erts/emulator/beam/erl_alloc.types @@ -272,6 +272,7 @@ type CODE_IX_LOCK_Q SHORT_LIVED SYSTEM code_ix_lock_q type PROC_INTERVAL LONG_LIVED SYSTEM process_interval type BUSY_CALLER_TAB SHORT_LIVED SYSTEM busy_caller_table type BUSY_CALLER SHORT_LIVED SYSTEM busy_caller +type PORT_DATA_HEAP STANDARD SYSTEM port_data_heap +if threads_no_smp # Need thread safe allocs, but std_alloc and fix_alloc are not; diff --git a/erts/emulator/beam/erl_bif_info.c b/erts/emulator/beam/erl_bif_info.c index f65396df55..54eefe8d12 100755 --- a/erts/emulator/beam/erl_bif_info.c +++ b/erts/emulator/beam/erl_bif_info.c @@ -114,6 +114,9 @@ static char erts_system_version[] = ("Erlang " ERLANG_OTP_RELEASE #ifdef VALGRIND " [valgrind-compiled]" #endif +#ifdef ERTS_FRMPTR + " [frame-pointer]" +#endif #ifdef USE_DTRACE " [dtrace]" #endif @@ -2042,6 +2045,9 @@ BIF_RETTYPE system_info_1(BIF_ALIST_1) #elif defined(ERTS_ENABLE_LOCK_COUNT) ERTS_DECL_AM(lcnt); BIF_RET(AM_lcnt); +#elif defined(ERTS_FRMPTR) + ERTS_DECL_AM(frmptr); + BIF_RET(AM_frmptr); #else BIF_RET(am_opt); #endif @@ -2807,12 +2813,10 @@ erts_bld_port_info(Eterm **hpp, ErlOffHeap *ohp, Uint *szp, Port *prt, Eterm ite included though). */ Uint size = 0; - ErlHeapFragment* bp; erts_doforall_links(ERTS_P_LINKS(prt), &erts_one_link_size, &size); - for (bp = prt->bp; bp; bp = bp->next) - size += sizeof(ErlHeapFragment) + (bp->alloc_size - 1)*sizeof(Eterm); + size += erts_port_data_size(prt); if (prt->linebuf) size += sizeof(LineBuf) + prt->linebuf->ovsiz; diff --git a/erts/emulator/beam/erl_bif_port.c b/erts/emulator/beam/erl_bif_port.c index 44fa41c7b6..109c54fd7f 100644 --- a/erts/emulator/beam/erl_bif_port.c +++ b/erts/emulator/beam/erl_bif_port.c @@ -423,57 +423,164 @@ BIF_RETTYPE erts_internal_port_info_2(BIF_ALIST_2) } } +/* + * The erlang:port_set_data()/erlang:port_get_data() operations should + * be viewed as operations on a table (inet_db) with data values + * associated with port identifier keys. That is, these operations are + * *not* signals to/from ports. + */ + +#if (TAG_PRIMARY_IMMED1 & 0x3) == 0 +# error "erlang:port_set_data()/erlang:port_get_data() needs to be rewritten!" +#endif + +typedef struct { + ErtsThrPrgrLaterOp later_op; + Uint hsize; + Eterm data; + ErlOffHeap off_heap; + Eterm heap[1]; +} ErtsPortDataHeap; -BIF_RETTYPE erts_internal_port_set_data_2(BIF_ALIST_2) +static void +free_port_data_heap(void *vpdhp) { - Eterm ref; + erts_cleanup_offheap(&((ErtsPortDataHeap *) vpdhp)->off_heap); + erts_free(ERTS_ALC_T_PORT_DATA_HEAP, vpdhp); +} + +static ERTS_INLINE void +cleanup_old_port_data(erts_aint_t data) +{ + if ((data & 0x3) != 0) { + ASSERT(is_immed((Eterm) data)); + } + else { +#ifdef ERTS_SMP + ErtsPortDataHeap *pdhp = (ErtsPortDataHeap *) data; + size_t size; + ERTS_SMP_DATA_DEPENDENCY_READ_MEMORY_BARRIER; + size = sizeof(ErtsPortDataHeap) + pdhp->hsize*(sizeof(Eterm) - 1); + erts_schedule_thr_prgr_later_cleanup_op(free_port_data_heap, + (void *) pdhp, + &pdhp->later_op, + size); +#else + free_port_data_heap((void *) data); +#endif + } +} + +void +erts_init_port_data(Port *prt) +{ + erts_smp_atomic_init_nob(&prt->data, (erts_aint_t) am_undefined); +} + +void +erts_cleanup_port_data(Port *prt) +{ + ASSERT(erts_atomic32_read_nob(&prt->state) & ERTS_PORT_SFLGS_INVALID_LOOKUP); + cleanup_old_port_data(erts_smp_atomic_read_nob(&prt->data)); + erts_smp_atomic_set_nob(&prt->data, (erts_aint_t) THE_NON_VALUE); +} + +Uint +erts_port_data_size(Port *prt) +{ + erts_aint_t data = erts_smp_atomic_read_ddrb(&prt->data); + + if ((data & 0x3) != 0) { + ASSERT(is_immed((Eterm) (UWord) data)); + return (Uint) 0; + } + else { + ErtsPortDataHeap *pdhp = (ErtsPortDataHeap *) data; + return (Uint) sizeof(ErtsPortDataHeap) + pdhp->hsize*(sizeof(Eterm)-1); + } +} + +ErlOffHeap * +erts_port_data_offheap(Port *prt) +{ + erts_aint_t data = erts_smp_atomic_read_ddrb(&prt->data); + + if ((data & 0x3) != 0) { + ASSERT(is_immed((Eterm) (UWord) data)); + return NULL; + } + else { + ErtsPortDataHeap *pdhp = (ErtsPortDataHeap *) data; + return &pdhp->off_heap; + } +} + +BIF_RETTYPE port_set_data_2(BIF_ALIST_2) +{ + /* + * This is not a signal. See comment above. + */ + erts_aint_t data; Port* prt; prt = lookup_port(BIF_P, BIF_ARG_1); if (!prt) - BIF_RET(am_badarg); + BIF_ERROR(BIF_P, BADARG); - switch (erts_port_set_data(BIF_P, prt, BIF_ARG_2, &ref)) { - case ERTS_PORT_OP_CALLER_EXIT: - case ERTS_PORT_OP_BADARG: - case ERTS_PORT_OP_DROPPED: - BIF_RET(am_badarg); - case ERTS_PORT_OP_SCHEDULED: - ASSERT(is_internal_ref(ref)); - BIF_RET(ref); - case ERTS_PORT_OP_DONE: - BIF_RET(am_true); - default: - ERTS_INTERNAL_ERROR("Unexpected erts_port_set_data() result"); - BIF_RET(am_internal_error); + if (is_immed(BIF_ARG_2)) { + data = (erts_aint_t) BIF_ARG_2; + ASSERT((data & 0x3) != 0); } + else { + ErtsPortDataHeap *pdhp; + Uint hsize; + Eterm *hp; + + hsize = size_object(BIF_ARG_2); + pdhp = erts_alloc(ERTS_ALC_T_PORT_DATA_HEAP, + sizeof(ErtsPortDataHeap) + hsize*(sizeof(Eterm)-1)); + hp = &pdhp->heap[0]; + pdhp->off_heap.first = NULL; + pdhp->off_heap.overhead = 0; + pdhp->data = copy_struct(BIF_ARG_2, hsize, &hp, &pdhp->off_heap); + data = (erts_aint_t) pdhp; + ASSERT((data & 0x3) == 0); + } + + data = erts_smp_atomic_xchg_wb(&prt->data, data); + + cleanup_old_port_data(data); + + BIF_RET(am_true); } -BIF_RETTYPE erts_internal_port_get_data_1(BIF_ALIST_1) +BIF_RETTYPE port_get_data_1(BIF_ALIST_1) { - Eterm retval; + /* + * This is not a signal. See comment above. + */ + Eterm res; + erts_aint_t data; Port* prt; prt = lookup_port(BIF_P, BIF_ARG_1); if (!prt) - BIF_RET(am_badarg); + BIF_ERROR(BIF_P, BADARG); - switch (erts_port_get_data(BIF_P, prt, &retval)) { - case ERTS_PORT_OP_CALLER_EXIT: - case ERTS_PORT_OP_BADARG: - case ERTS_PORT_OP_DROPPED: - BIF_RET(am_badarg); - case ERTS_PORT_OP_SCHEDULED: - ASSERT(is_internal_ref(retval)); - BIF_RET(retval); - case ERTS_PORT_OP_DONE: - ASSERT(is_not_internal_ref(retval)); - BIF_RET(retval); - default: - ERTS_INTERNAL_ERROR("Unexpected erts_port_get_data() result"); - BIF_RET(am_internal_error); + data = erts_smp_atomic_read_ddrb(&prt->data); + + if ((data & 0x3) != 0) { + res = (Eterm) (UWord) data; + ASSERT(is_immed(res)); + } + else { + ErtsPortDataHeap *pdhp = (ErtsPortDataHeap *) data; + Eterm *hp = HAlloc(BIF_P, pdhp->hsize); + res = copy_struct(pdhp->data, pdhp->hsize, &hp, &MSO(BIF_P)); } + + BIF_RET(res); } /* diff --git a/erts/emulator/beam/erl_node_tables.c b/erts/emulator/beam/erl_node_tables.c index ebfba065d1..e688e55c88 100644 --- a/erts/emulator/beam/erl_node_tables.c +++ b/erts/emulator/beam/erl_node_tables.c @@ -1371,6 +1371,7 @@ setup_reference_table(void) /* Insert all ports */ max = erts_ptab_max(&erts_port); for (i = 0; i < max; i++) { + ErlOffHeap *ohp; erts_aint32_t state; Port *prt; @@ -1389,8 +1390,9 @@ setup_reference_table(void) if (ERTS_P_MONITORS(prt)) insert_monitors(ERTS_P_MONITORS(prt), prt->common.id); /* Insert port data */ - for(hfp = prt->bp; hfp; hfp = hfp->next) - insert_offheap(&(hfp->off_heap), HEAP_REF, prt->common.id); + ohp = erts_port_data_offheap(prt); + if (ohp) + insert_offheap(ohp, HEAP_REF, prt->common.id); /* Insert controller */ if (prt->dist_entry) insert_dist_entry(prt->dist_entry, diff --git a/erts/emulator/beam/erl_port.h b/erts/emulator/beam/erl_port.h index 377aa72ed5..ad3f104a68 100644 --- a/erts/emulator/beam/erl_port.h +++ b/erts/emulator/beam/erl_port.h @@ -167,8 +167,7 @@ struct _erl_drv_port { #endif erts_atomic_t connected; /* A connected process */ Eterm caller; /* Current caller. */ - Eterm data; /* Data associated with port. */ - ErlHeapFragment* bp; /* Heap fragment holding data (NULL if imm data). */ + erts_smp_atomic_t data; /* Data associated with port. */ Uint bytes_in; /* Number of bytes read */ Uint bytes_out; /* Number of bytes written */ @@ -189,6 +188,12 @@ struct _erl_drv_port { int reds; /* Only used while executing driver callbacks */ }; + +void erts_init_port_data(Port *); +void erts_cleanup_port_data(Port *); +Uint erts_port_data_size(Port *); +ErlOffHeap *erts_port_data_offheap(Port *); + #define ERTS_PORT_GET_CONNECTED(PRT) \ ((Eterm) erts_atomic_read_nob(&(PRT)->connected)) #define ERTS_PORT_SET_CONNECTED(PRT, PID) \ @@ -326,8 +331,6 @@ extern erts_smp_atomic_t erts_bytes_in; /* no bytes sent into the system */ #define ERTS_PORT_REDS_CONTROL (CONTEXT_REDS/100) #define ERTS_PORT_REDS_CALL (CONTEXT_REDS/50) #define ERTS_PORT_REDS_INFO (CONTEXT_REDS/100) -#define ERTS_PORT_REDS_SET_DATA (CONTEXT_REDS/100) -#define ERTS_PORT_REDS_GET_DATA (CONTEXT_REDS/100) #define ERTS_PORT_REDS_TERMINATE (CONTEXT_REDS/50) void print_port_info(Port *, int, void *); @@ -794,8 +797,6 @@ struct binary; #define ERTS_P2P_SIG_TYPE_INFO 7 #define ERTS_P2P_SIG_TYPE_LINK 8 #define ERTS_P2P_SIG_TYPE_UNLINK 9 -#define ERTS_P2P_SIG_TYPE_SET_DATA 10 -#define ERTS_P2P_SIG_TYPE_GET_DATA 11 #define ERTS_P2P_SIG_TYPE_BITS 4 #define ERTS_P2P_SIG_TYPE_MASK \ @@ -810,6 +811,7 @@ struct binary; #define ERTS_P2P_SIG_DATA_FLG_BAD_OUTPUT ERTS_P2P_SIG_DATA_FLG(4) #define ERTS_P2P_SIG_DATA_FLG_BROKEN_LINK ERTS_P2P_SIG_DATA_FLG(5) #define ERTS_P2P_SIG_DATA_FLG_SCHED ERTS_P2P_SIG_DATA_FLG(6) +#define ERTS_P2P_SIG_DATA_FLG_ASYNC ERTS_P2P_SIG_DATA_FLG(7) struct ErtsProc2PortSigData_ { int flags; @@ -856,10 +858,6 @@ struct ErtsProc2PortSigData_ { struct { Eterm from; } unlink; - struct { - ErlHeapFragment *bp; - Eterm data; - } set_data; } u; } ; @@ -919,6 +917,7 @@ erts_schedule_proc2port_signal(Process *, Eterm *, ErtsProc2PortSigData *, int, + ErtsPortTaskHandle *, ErtsProc2PortSigCallback); int erts_deliver_port_exit(Port *, Eterm, Eterm, int); @@ -932,6 +931,7 @@ int erts_deliver_port_exit(Port *, Eterm, Eterm, int); #define ERTS_PORT_SIG_FLG_BROKEN_LINK ERTS_P2P_SIG_DATA_FLG_BROKEN_LINK #define ERTS_PORT_SIG_FLG_BAD_OUTPUT ERTS_P2P_SIG_DATA_FLG_BAD_OUTPUT #define ERTS_PORT_SIG_FLG_FORCE_SCHED ERTS_P2P_SIG_DATA_FLG_SCHED +#define ERTS_PORT_SIG_FLG_ASYNC ERTS_P2P_SIG_DATA_FLG_ASYNC /* ERTS_PORT_SIG_FLG_FORCE_IMM_CALL only when crash dumping... */ #define ERTS_PORT_SIG_FLG_FORCE_IMM_CALL ERTS_P2P_SIG_DATA_FLG_BAD_OUTPUT @@ -953,7 +953,5 @@ ErtsPortOpResult erts_port_unlink(Process *, Port *, Eterm, Eterm *); ErtsPortOpResult erts_port_control(Process *, Port *, unsigned int, Eterm, Eterm *); ErtsPortOpResult erts_port_call(Process *, Port *, unsigned int, Eterm, Eterm *); ErtsPortOpResult erts_port_info(Process *, Port *, Eterm, Eterm *); -ErtsPortOpResult erts_port_set_data(Process *, Port *, Eterm, Eterm *); -ErtsPortOpResult erts_port_get_data(Process *, Port *, Eterm *); #endif diff --git a/erts/emulator/beam/erl_port_task.c b/erts/emulator/beam/erl_port_task.c index 0ed08bee01..53cb01a8c6 100644 --- a/erts/emulator/beam/erl_port_task.c +++ b/erts/emulator/beam/erl_port_task.c @@ -552,6 +552,19 @@ set_handle(ErtsPortTask *ptp, ErtsPortTaskHandle *pthp) } } +static ERTS_INLINE void +set_tmp_handle(ErtsPortTask *ptp, ErtsPortTaskHandle *pthp) +{ + ptp->u.alive.handle = NULL; + if (pthp) { + /* + * IMPORTANT! Task either need to be aborted, or task handle + * need to be detached before thread progress has been made. + */ + erts_smp_atomic_set_relb(pthp, (erts_aint_t) ptp); + } +} + /* * Busy port queue management @@ -1182,6 +1195,13 @@ erl_drv_consume_timeslice(ErlDrvPort dprt, int percent) return 1; } +void +erts_port_task_tmp_handle_detach(ErtsPortTaskHandle *pthp) +{ + ERTS_SMP_LC_ASSERT(erts_thr_progress_lc_is_delaying()); + reset_port_task_handle(pthp); +} + /* * Abort a scheduled task. */ @@ -1206,7 +1226,7 @@ erts_port_task_abort(ErtsPortTaskHandle *pthp) ERTS_SMP_READ_MEMORY_BARRIER; old_state = erts_smp_atomic32_read_nob(&ptp->state); if (old_state == ERTS_PT_STATE_SCHEDULED) { - ASSERT(saved_pthp == pthp); + ASSERT(!saved_pthp || saved_pthp == pthp); } #endif @@ -1227,9 +1247,6 @@ erts_port_task_abort(ErtsPortTaskHandle *pthp) &erts_port_task_outstanding_io_tasks) > 0); erts_smp_atomic_dec_relb(&erts_port_task_outstanding_io_tasks); break; - case ERTS_PORT_TASK_PROC_SIG: - ERTS_INTERNAL_ERROR("Aborted process to port signal"); - break; default: break; } @@ -1404,7 +1421,6 @@ erts_port_task_schedule(Eterm id, } case ERTS_PORT_TASK_PROC_SIG: { va_list argp; - ASSERT(!pthp); va_start(argp, type); sigdp = va_arg(argp, ErtsProc2PortSigData *); ptp = p2p_sig_data_to_task(sigdp); @@ -1412,7 +1428,7 @@ erts_port_task_schedule(Eterm id, ptp->u.alive.flags |= va_arg(argp, int); va_end(argp); if (!(ptp->u.alive.flags & ERTS_PT_FLG_NOSUSPEND)) - set_handle(ptp, pthp); + set_tmp_handle(ptp, pthp); else { ns_pthlp = erts_alloc(ERTS_ALC_T_PT_HNDL_LIST, sizeof(ErtsPortTaskHandleList)); diff --git a/erts/emulator/beam/erl_port_task.h b/erts/emulator/beam/erl_port_task.h index ae6cd69ae2..d35a15c27b 100644 --- a/erts/emulator/beam/erl_port_task.h +++ b/erts/emulator/beam/erl_port_task.h @@ -243,7 +243,9 @@ int erts_port_task_execute(ErtsRunQueue *, Port **); void erts_port_task_init(void); #endif +void erts_port_task_tmp_handle_detach(ErtsPortTaskHandle *); int erts_port_task_abort(ErtsPortTaskHandle *); + void erts_port_task_abort_nosuspend_tasks(Port *); int erts_port_task_schedule(Eterm, diff --git a/erts/emulator/beam/erl_process.c b/erts/emulator/beam/erl_process.c index 5d9c5af55f..7415a5721f 100644 --- a/erts/emulator/beam/erl_process.c +++ b/erts/emulator/beam/erl_process.c @@ -4670,6 +4670,8 @@ erts_init_scheduling(int no_schedulers, int no_schedulers_online) esdp->reductions = 0; init_sched_wall_time(&esdp->sched_wall_time); + + erts_port_task_handle_init(&esdp->nosuspend_port_task_handle); } init_misc_aux_work(); diff --git a/erts/emulator/beam/erl_process.h b/erts/emulator/beam/erl_process.h index c40a6f86e4..5a1f6bbe8d 100644 --- a/erts/emulator/beam/erl_process.h +++ b/erts/emulator/beam/erl_process.h @@ -507,6 +507,7 @@ struct ErtsSchedulerData_ { Uint64 reductions; ErtsSchedWallTime sched_wall_time; + ErtsPortTaskHandle nosuspend_port_task_handle; #ifdef ERTS_DO_VERIFY_UNUSED_TEMP_ALLOC erts_alloc_verify_func_t verify_unused_temp_alloc; diff --git a/erts/emulator/beam/io.c b/erts/emulator/beam/io.c index f93b599632..b6b7b47bd6 100644 --- a/erts/emulator/beam/io.c +++ b/erts/emulator/beam/io.c @@ -364,9 +364,8 @@ static Port *create_port(char *name, ERTS_P_LINKS(prt) = NULL; ERTS_P_MONITORS(prt) = NULL; prt->linebuf = NULL; - prt->bp = NULL; prt->suspended = NULL; - prt->data = am_undefined; + erts_init_port_data(prt); prt->port_data_lock = NULL; prt->control_flags = 0; prt->bytes_in = 0; @@ -1442,6 +1441,7 @@ erts_schedule_proc2port_signal(Process *c_p, Eterm *refp, ErtsProc2PortSigData *sigdp, int task_flags, + ErtsPortTaskHandle *pthp, ErtsProc2PortSigCallback callback) { int sched_res; @@ -1491,7 +1491,7 @@ erts_schedule_proc2port_signal(Process *c_p, /* Schedule port close call for later execution... */ sched_res = erts_port_task_schedule(prt->common.id, - NULL, + pthp, ERTS_PORT_TASK_PROC_SIG, sigdp, callback, @@ -1629,6 +1629,7 @@ bad_port_signal(Process *c_p, refp, sigdp, 0, + NULL, port_badsig); } @@ -1838,8 +1839,11 @@ erts_port_output(Process *c_p, ErlIOVec *evp = NULL; char *buf = NULL; int force_immediate_call = (flags & ERTS_PORT_SIG_FLG_FORCE_IMM_CALL); + int async_nosuspend; + ErtsPortTaskHandle *ns_pthp; ASSERT((flags & ~(ERTS_PORT_SIG_FLG_BANG_OP + | ERTS_PORT_SIG_FLG_ASYNC | ERTS_PORT_SIG_FLG_NOSUSPEND | ERTS_PORT_SIG_FLG_FORCE | ERTS_PORT_SIG_FLG_FORCE_IMM_CALL)) == 0); @@ -1861,6 +1865,12 @@ erts_port_output(Process *c_p, ? ERTS_PORT_OP_DROPPED : ERTS_PORT_OP_BUSY); + async_nosuspend = ((flags & (ERTS_PORT_SIG_FLG_ASYNC + | ERTS_PORT_SIG_FLG_NOSUSPEND + | ERTS_PORT_SIG_FLG_FORCE)) + == (ERTS_PORT_SIG_FLG_ASYNC + | ERTS_PORT_SIG_FLG_NOSUSPEND)); + try_call = (force_immediate_call /* crash dumping */ || !(sched_flags & (invalid_flags | ERTS_PTS_FLGS_FORCE_SCHEDULE_OP))); @@ -1995,6 +2005,15 @@ erts_port_output(Process *c_p, return ERTS_PORT_OP_DONE; case ERTS_TRY_IMM_DRV_CALL_INVALID_SCHED_FLAGS: sched_flags = try_call_state.sched_flags; + if (async_nosuspend + && (sched_flags & (busy_flgs|ERTS_PTS_FLG_EXIT))) { + driver_free_binary(cbin); + if (evp != &ev) + erts_free(ERTS_ALC_T_TMP, evp); + return ((sched_flags & ERTS_PTS_FLG_EXIT) + ? ERTS_PORT_OP_DROPPED + : ERTS_PORT_OP_BUSY); + } case ERTS_TRY_IMM_DRV_CALL_BUSY_LOCK: /* Schedule outputv() call instead... */ break; @@ -2142,6 +2161,13 @@ erts_port_output(Process *c_p, return ERTS_PORT_OP_DONE; case ERTS_TRY_IMM_DRV_CALL_INVALID_SCHED_FLAGS: sched_flags = try_call_state.sched_flags; + if (async_nosuspend + && (sched_flags & (busy_flgs|ERTS_PTS_FLG_EXIT))) { + erts_free(ERTS_ALC_T_TMP, buf); + return ((sched_flags & ERTS_PTS_FLG_EXIT) + ? ERTS_PORT_OP_DROPPED + : ERTS_PORT_OP_BUSY); + } case ERTS_TRY_IMM_DRV_CALL_BUSY_LOCK: /* Schedule outputv() call instead... */ break; @@ -2163,20 +2189,33 @@ erts_port_output(Process *c_p, task_flags = ERTS_PT_FLG_WAIT_BUSY; sigdp->flags |= flags; + ns_pthp = NULL; if (flags & (ERTS_P2P_SIG_DATA_FLG_FORCE|ERTS_P2P_SIG_DATA_FLG_NOSUSPEND)) { task_flags = 0; if (flags & ERTS_P2P_SIG_DATA_FLG_FORCE) sigdp->flags &= ~ERTS_P2P_SIG_DATA_FLG_NOSUSPEND; + else if (async_nosuspend) { + ErtsSchedulerData *esdp = (c_p + ? ERTS_PROC_GET_SCHDATA(c_p) + : erts_get_scheduler_data()); + ASSERT(esdp); + ns_pthp = &esdp->nosuspend_port_task_handle; + sigdp->flags &= ~ERTS_P2P_SIG_DATA_FLG_NOSUSPEND; + } else if (flags & ERTS_P2P_SIG_DATA_FLG_NOSUSPEND) task_flags = ERTS_PT_FLG_NOSUSPEND; } + ASSERT(ns_pthp || !async_nosuspend); + ASSERT(async_nosuspend || !ns_pthp); + res = erts_schedule_proc2port_signal(c_p, prt, c_p ? c_p->common.id : ERTS_INVALID_PID, refp, sigdp, task_flags, + ns_pthp, port_sig_callback); if (res != ERTS_PORT_OP_SCHEDULED) { @@ -2187,9 +2226,23 @@ erts_port_output(Process *c_p, return res; } - if (!(sched_flags & ERTS_PTS_FLG_EXIT) && (sched_flags & busy_flgs)) - return ERTS_PORT_OP_BUSY_SCHEDULED; - + if (!(flags & ERTS_PORT_SIG_FLG_FORCE)) { + sched_flags = erts_smp_atomic32_read_acqb(&prt->sched.flags); + if (!(sched_flags & ERTS_PTS_FLG_BUSY_PORT)) { + if (async_nosuspend) + erts_port_task_tmp_handle_detach(ns_pthp); + } + else { + if (!async_nosuspend) + return ERTS_PORT_OP_BUSY_SCHEDULED; + else { + if (erts_port_task_abort(ns_pthp) == 0) + return ERTS_PORT_OP_BUSY; + else + erts_port_task_tmp_handle_detach(ns_pthp); + } + } + } return res; bad_value: @@ -2285,6 +2338,7 @@ erts_port_exit(Process *c_p, ErlHeapFragment *bp = NULL; ASSERT((flags & ~(ERTS_PORT_SIG_FLG_BANG_OP + | ERTS_PORT_SIG_FLG_ASYNC | ERTS_PORT_SIG_FLG_BROKEN_LINK | ERTS_PORT_SIG_FLG_FORCE_SCHED)) == 0); @@ -2345,6 +2399,7 @@ erts_port_exit(Process *c_p, refp, sigdp, 0, + NULL, port_sig_exit); if (res == ERTS_PORT_OP_DROPPED) { @@ -2460,7 +2515,8 @@ erts_port_connect(Process *c_p, !refp, am_connect); - ASSERT((flags & ~ERTS_PORT_SIG_FLG_BANG_OP) == 0); + ASSERT((flags & ~(ERTS_PORT_SIG_FLG_BANG_OP + | ERTS_PORT_SIG_FLG_ASYNC)) == 0); if (is_not_internal_pid(connect)) connect_id = NIL; /* Fail in op (for signal order) */ @@ -2499,6 +2555,7 @@ erts_port_connect(Process *c_p, refp, sigdp, 0, + NULL, port_sig_connect); } @@ -2555,6 +2612,7 @@ erts_port_unlink(Process *c_p, Port *prt, Eterm from, Eterm *refp) refp, sigdp, 0, + NULL, port_sig_unlink); } @@ -2644,6 +2702,7 @@ erts_port_link(Process *c_p, Port *prt, Eterm to, Eterm *refp) refp, sigdp, 0, + NULL, port_sig_link); } @@ -3371,11 +3430,8 @@ terminate_port(Port *prt) erts_free(ERTS_ALC_T_LINEBUF, (void *) prt->linebuf); prt->linebuf = NULL; } - if (prt->bp != NULL) { - free_message_buffer(prt->bp); - prt->bp = NULL; - prt->data = am_undefined; - } + + erts_cleanup_port_data(prt); if (prt->psd) erts_free(ERTS_ALC_T_PRTSD, prt->psd); @@ -3625,6 +3681,10 @@ erts_port_command(Process *c_p, ASSERT(port); flags |= ERTS_PORT_SIG_FLG_BANG_OP; + if (!erts_port_synchronous_ops) { + flags |= ERTS_PORT_SIG_FLG_ASYNC; + refp = NULL; + } if (is_tuple_arity(command, 2)) { Eterm cntd; @@ -3632,21 +3692,14 @@ erts_port_command(Process *c_p, cntd = tp[1]; if (is_internal_pid(cntd)) { if (tp[2] == am_close) { - if (!erts_port_synchronous_ops) - refp = NULL; flags &= ~ERTS_PORT_SIG_FLG_NOSUSPEND; return erts_port_exit(c_p, flags, port, cntd, am_normal, refp); } else if (is_tuple_arity(tp[2], 2)) { tp = tuple_val(tp[2]); if (tp[1] == am_command) { - if (!(flags & ERTS_PORT_SIG_FLG_NOSUSPEND) - && !erts_port_synchronous_ops) - refp = NULL; return erts_port_output(c_p, flags, port, cntd, tp[2], refp); } else if (tp[1] == am_connect) { - if (!erts_port_synchronous_ops) - refp = NULL; flags &= ~ERTS_PORT_SIG_FLG_NOSUSPEND; return erts_port_connect(c_p, flags, port, cntd, tp[2], refp); } @@ -3655,8 +3708,6 @@ erts_port_command(Process *c_p, } /* badsig */ - if (!erts_port_synchronous_ops) - refp = NULL; flags &= ~ERTS_PORT_SIG_FLG_NOSUSPEND; return bad_port_signal(c_p, flags, port, c_p->common.id, refp, am_command); } @@ -4053,6 +4104,7 @@ erts_port_control(Process* c_p, retvalp, sigdp, 0, + NULL, port_sig_control); if (res != ERTS_PORT_OP_SCHEDULED) { cleanup_scheduled_control(binp, bufp); @@ -4333,6 +4385,7 @@ erts_port_call(Process* c_p, retvalp, sigdp, 0, + NULL, port_sig_call); if (res != ERTS_PORT_OP_SCHEDULED) { cleanup_scheduled_call(bufp); @@ -4499,228 +4552,10 @@ erts_port_info(Process* c_p, retvalp, sigdp, 0, + NULL, port_sig_info); } -static int -port_sig_set_data(Port *prt, - erts_aint32_t state, - int op, - ErtsProc2PortSigData *sigdp) -{ - ASSERT(sigdp->flags & ERTS_P2P_SIG_DATA_FLG_REPLY); - - if (op == ERTS_PROC2PORT_SIG_EXEC) { - if (prt->bp) - free_message_buffer(prt->bp); - prt->bp = sigdp->u.set_data.bp; - prt->data = sigdp->u.set_data.data; - port_sched_op_reply(sigdp->caller, sigdp->ref, am_true); - } - else { - if (sigdp->u.set_data.bp) - free_message_buffer(sigdp->u.set_data.bp); - port_sched_op_reply(sigdp->caller, sigdp->ref, am_badarg); - } - return ERTS_PORT_REDS_SET_DATA; -} - -ErtsPortOpResult -erts_port_set_data(Process* c_p, - Port *prt, - Eterm data, - Eterm *refp) -{ - ErtsPortOpResult res; - Eterm set_data; - ErlHeapFragment *bp; - ErtsProc2PortSigData *sigdp; - ErtsTryImmDrvCallResult try_call_res; - ErtsTryImmDrvCallState try_call_state - = ERTS_INIT_TRY_IMM_DRV_CALL_STATE( - c_p, - prt, - ERTS_PORT_SFLGS_INVALID_LOOKUP, - 0, - !refp, - am_set_data); - - if (is_immed(data)) { - set_data = data; - bp = NULL; - } - else { - Eterm *hp; - Uint sz = size_object(data); - bp = new_message_buffer(sz); - hp = bp->mem; - set_data = copy_struct(data, sz, &hp, &bp->off_heap); - } - - try_call_res = try_imm_drv_call(&try_call_state); - switch (try_call_res) { - case ERTS_TRY_IMM_DRV_CALL_OK: - if (prt->bp) - free_message_buffer(prt->bp); - prt->bp = bp; - prt->data = set_data; - finalize_imm_drv_call(&try_call_state); - BUMP_REDS(c_p, ERTS_PORT_REDS_SET_DATA); - return ERTS_PORT_OP_DONE; - case ERTS_TRY_IMM_DRV_CALL_INVALID_PORT: - return ERTS_PORT_OP_DROPPED; - case ERTS_TRY_IMM_DRV_CALL_INVALID_SCHED_FLAGS: - case ERTS_TRY_IMM_DRV_CALL_BUSY_LOCK: - /* Schedule call instead... */ - break; - } - - sigdp = erts_port_task_alloc_p2p_sig_data(); - sigdp->flags = ERTS_P2P_SIG_TYPE_SET_DATA; - sigdp->u.set_data.data = set_data; - sigdp->u.set_data.bp = bp; - - res = erts_schedule_proc2port_signal(c_p, - prt, - c_p->common.id, - refp, - sigdp, - 0, - port_sig_set_data); - if (res != ERTS_PORT_OP_SCHEDULED && bp) - free_message_buffer(bp); - return res; -} - -static int -port_sig_get_data(Port *prt, - erts_aint32_t state, - int op, - ErtsProc2PortSigData *sigdp) -{ - ASSERT(sigdp->flags & ERTS_P2P_SIG_DATA_FLG_REPLY); - if (op != ERTS_PROC2PORT_SIG_EXEC) - port_sched_op_reply(sigdp->caller, sigdp->ref, am_badarg); - else { - Process *rp; - ErtsProcLocks rp_locks = 0; - - rp = erts_proc_lookup_raw(sigdp->caller); - if (rp) { - Uint hsz; - Eterm *hp, *hp_start; - Eterm data, msg; - ErlHeapFragment *bp; - ErlOffHeap *ohp; - - hsz = ERTS_QUEUE_PORT_SCHED_OP_REPLY_SIZE; - hsz += 3; - if (prt->bp) - hsz += prt->bp->used_size; - - hp_start = hp = erts_alloc_message_heap(hsz, - &bp, - &ohp, - rp, - &rp_locks); - - if (is_immed(prt->data)) - data = prt->data; - else - data = copy_struct(prt->data, - prt->bp->used_size, - &hp, - &bp->off_heap); - - - - msg = TUPLE2(hp, am_ok, data); - hp += 3; - - queue_port_sched_op_reply(rp, - &rp_locks, - hp_start, - hp, - hsz, - bp, - sigdp->ref, - msg); - if (rp_locks) - erts_smp_proc_unlock(rp, rp_locks); - } - } - return ERTS_PORT_REDS_GET_DATA; -} - -ErtsPortOpResult -erts_port_get_data(Process* c_p, - Port *prt, - Eterm *retvalp) -{ - ErtsProc2PortSigData *sigdp; - ErtsTryImmDrvCallResult try_call_res; - ErtsTryImmDrvCallState try_call_state - = ERTS_INIT_TRY_IMM_DRV_CALL_STATE( - c_p, - prt, - ERTS_PORT_SFLGS_INVALID_LOOKUP, - 0, - 0, - am_get_data); - - try_call_res = try_imm_drv_call(&try_call_state); - switch (try_call_res) { - case ERTS_TRY_IMM_DRV_CALL_OK: { - Eterm *hp; - Eterm data; - ErlHeapFragment *bp; - Uint sz; - if (is_immed(prt->data)) { - bp = NULL; - data = prt->data; - } - else { - bp = new_message_buffer(prt->bp->used_size); - data = copy_struct(prt->data, - prt->bp->used_size, - &hp, - &bp->off_heap); - } - finalize_imm_drv_call(&try_call_state); - if (is_immed(data)) - sz = 0; - else - sz = bp->used_size; - - hp = HAlloc(c_p, sz + 3); - if (is_not_immed(data)) { - data = copy_struct(data, bp->used_size, &hp, &MSO(c_p)); - free_message_buffer(bp); - } - *retvalp = TUPLE2(hp, am_ok, data); - BUMP_REDS(c_p, ERTS_PORT_REDS_GET_DATA); - return ERTS_PORT_OP_DONE; - } - case ERTS_TRY_IMM_DRV_CALL_INVALID_PORT: - return ERTS_PORT_OP_DROPPED; - case ERTS_TRY_IMM_DRV_CALL_INVALID_SCHED_FLAGS: - case ERTS_TRY_IMM_DRV_CALL_BUSY_LOCK: - /* Schedule call instead... */ - break; - } - - sigdp = erts_port_task_alloc_p2p_sig_data(); - sigdp->flags = ERTS_P2P_SIG_TYPE_GET_DATA; - - return erts_schedule_proc2port_signal(c_p, - prt, - c_p->common.id, - retvalp, - sigdp, - 0, - port_sig_get_data); -} - typedef struct { int to; void *arg; |