diff options
Diffstat (limited to 'erts/emulator/beam/io.c')
| -rw-r--r-- | erts/emulator/beam/io.c | 2067 |
1 files changed, 1490 insertions, 577 deletions
diff --git a/erts/emulator/beam/io.c b/erts/emulator/beam/io.c index 6c8bfa27d8..77dbe92241 100644 --- a/erts/emulator/beam/io.c +++ b/erts/emulator/beam/io.c @@ -1,18 +1,19 @@ /* * %CopyrightBegin% * - * Copyright Ericsson AB 1996-2014. All Rights Reserved. + * Copyright Ericsson AB 1996-2016. All Rights Reserved. * - * The contents of this file are subject to the Erlang Public License, - * Version 1.1, (the "License"); you may not use this file except in - * compliance with the License. You should have received a copy of the - * Erlang Public License along with this software. If not, it can be - * retrieved online at http://www.erlang.org/. + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at * - * Software distributed under the License is distributed on an "AS IS" - * basis, WITHOUT WARRANTY OF ANY KIND, either express or implied. See - * the License for the specific language governing rights and limitations - * under the License. + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. * * %CopyrightEnd% */ @@ -46,13 +47,18 @@ #define ERTS_WANT_EXTERNAL_TAGS #include "external.h" #include "dtrace-wrapper.h" +#include "lttng-wrapper.h" #include "erl_map.h" +#include "erl_bif_unique.h" +#include "erl_hl_timer.h" +#include "erl_time.h" extern ErlDrvEntry fd_driver_entry; -#ifndef __OSE__ extern ErlDrvEntry vanilla_driver_entry; -#endif extern ErlDrvEntry spawn_driver_entry; +#ifndef __WIN32__ +extern ErlDrvEntry forker_driver_entry; +#endif extern ErlDrvEntry *driver_tab[]; /* table of static drivers, only used during initialization */ erts_driver_t *driver_list; /* List of all drivers, static and dynamic. */ @@ -63,8 +69,6 @@ static erts_smp_tsd_key_t driver_list_last_error_key; /* Save last DDLL error o per thread basis (for BC interfaces) */ 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 */ const ErlDrvTermData driver_term_nil = (ErlDrvTermData)NIL; @@ -72,16 +76,23 @@ const Port erts_invalid_port = {{ERTS_INVALID_PORT}}; erts_driver_t vanilla_driver; erts_driver_t spawn_driver; +#ifndef __WIN32__ +erts_driver_t forker_driver; +#endif erts_driver_t fd_driver; int erts_port_synchronous_ops = 0; int erts_port_schedule_all_ops = 0; int erts_port_parallelism = 0; -static void deliver_result(Eterm sender, Eterm pid, Eterm res); +static erts_atomic64_t bytes_in; +static erts_atomic64_t bytes_out; + +static void deliver_result(Port *p, Eterm sender, Eterm pid, Eterm res); static int init_driver(erts_driver_t *, ErlDrvEntry *, DE_Handle *); static void terminate_port(Port *p); static void pdl_init(void); +static int driver_failure_term(ErlDrvPort ix, Eterm term, int eof); #ifdef ERTS_SMP static void driver_monitor_lock_pdl(Port *p); static void driver_monitor_unlock_pdl(Port *p); @@ -201,6 +212,7 @@ static ERTS_INLINE void kill_port(Port *pp) { ERTS_SMP_LC_ASSERT(erts_lc_is_port_locked(pp)); + ERTS_TRACER_CLEAR(&ERTS_TRACER(pp)); erts_ptab_delete_element(&erts_port, &pp->common); /* Time of death */ erts_port_task_free_port(pp); /* In non-smp case the port structure may have been deallocated now */ @@ -303,12 +315,9 @@ static Port *create_port(char *name, size_t port_size, busy_port_queue_size, size; erts_aint32_t state = ERTS_PORT_SFLG_CONNECTED; erts_aint32_t x_pts_flgs = 0; -#ifdef DEBUG - /* Make sure the debug flags survives until port is freed */ - state |= ERTS_PORT_SFLG_PORT_DEBUG; -#endif #ifdef ERTS_SMP + ErtsRunQueue *runq; if (!driver_lock) { /* Align size for mutex following port struct */ port_size = size = ERTS_ALC_DATA_ALIGN_SIZE(sizeof(Port)); @@ -318,6 +327,12 @@ static Port *create_port(char *name, #endif port_size = size = ERTS_ALC_DATA_ALIGN_SIZE(sizeof(Port)); +#ifdef DEBUG + /* Make sure the debug flags survives until port is freed */ + state |= ERTS_PORT_SFLG_PORT_DEBUG; +#endif + + busy_port_queue_size = ((driver->flags & ERL_DRV_FLAG_NO_BUSY_MSGQ) ? 0 @@ -353,8 +368,12 @@ static Port *create_port(char *name, 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)); + if (erts_get_scheduler_data()) + runq = erts_get_runq_current(NULL); + else + runq = ERTS_RUNQ_IX(0); + erts_smp_atomic_set_nob(&prt->run_queue, (erts_aint_t) runq); + prt->xports = NULL; #else erts_atomic32_init_nob(&prt->refc, 1); @@ -378,20 +397,17 @@ static Port *create_port(char *name, prt->dist_entry = NULL; ERTS_PORT_INIT_CONNECTED(prt, 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_PTMR_INIT(prt); erts_port_task_handle_init(&prt->timeout_task); - prt->psd = NULL; + erts_smp_atomic_init_nob(&prt->psd, (erts_aint_t) NULL); + prt->async_open_port = NULL; prt->drv_data = (SWord) 0; prt->os_pid = -1; /* Set default tracing */ - erts_get_default_tracing(&ERTS_TRACE_FLAGS(prt), &ERTS_TRACER_PROC(prt)); + erts_get_default_port_tracing(&ERTS_TRACE_FLAGS(prt), &ERTS_TRACER(prt)); - ASSERT(((char *) prt) == ((char *) &prt->common)); + ERTS_CT_ASSERT(offsetof(Port,common) == 0); #if !ERTS_PORT_INIT_INSTR_NEED_ID /* @@ -462,14 +478,15 @@ erts_port_free(Port *prt) | ERTS_PORT_SFLG_FREE)); ASSERT(state & ERTS_PORT_SFLG_PORT_DEBUG); -#ifdef ERTS_SMP - ERTS_LC_ASSERT(erts_atomic32_read_nob(&prt->common.refc) == 0); -#else - ERTS_LC_ASSERT(erts_atomic32_read_nob(&prt->refc) == 0); -#endif + ERTS_LC_ASSERT(erts_atomic_read_nob(&prt->common.refc.atmc) == 0); erts_port_task_fini_sched(&prt->sched); + if (prt->async_open_port) { + erts_free(ERTS_ALC_T_PRTSD, prt->async_open_port); + prt->async_open_port = NULL; + } + #ifdef ERTS_SMP ASSERT(prt->lock); if (state & ERTS_PORT_SFLG_PORT_SPECIFIC_LOCK) @@ -691,8 +708,9 @@ erts_open_driver(erts_driver_t* driver, /* Pointer to driver. */ error_number = error_type = 0; if (driver->start) { + ERTS_MSACC_PUSH_STATE_M(); if (IS_TRACED_FL(port, F_TRACE_SCHED_PORTS)) { - trace_sched_ports_where(port, am_in, am_start); + trace_sched_ports_where(port, am_in, am_open); } port->caller = pid; #ifdef USE_VM_PROBES @@ -701,6 +719,19 @@ erts_open_driver(erts_driver_t* driver, /* Pointer to driver. */ DTRACE3(driver_start, process_str, driver->name, port_str); } #endif + + ERTS_MSACC_SET_STATE_CACHED_M(ERTS_MSACC_STATE_PORT); + +#ifdef USE_LTTNG_VM_TRACEPOINTS + if (LTTNG_ENABLED(driver_start)) { + lttng_decl_portbuf(port_str); + lttng_decl_procbuf(proc_str); + lttng_pid_to_str(pid, proc_str); + lttng_port_to_str(port, port_str); + LTTNG3(driver_start, proc_str, driver->name, port_str); + } +#endif + fpe_was_unmasked = erts_block_fpe(); drv_data = (*driver->start)(ERTS_Port2ErlDrvPort(port), name, opts); if (((SWord) drv_data) == -1) @@ -720,9 +751,10 @@ erts_open_driver(erts_driver_t* driver, /* Pointer to driver. */ } erts_unblock_fpe(fpe_was_unmasked); + ERTS_MSACC_POP_STATE_M(); port->caller = NIL; if (IS_TRACED_FL(port, F_TRACE_SCHED_PORTS)) { - trace_sched_ports_where(port, am_out, am_start); + trace_sched_ports_where(port, am_out, am_open); } #ifdef ERTS_SMP if (port->xports) @@ -735,11 +767,7 @@ erts_open_driver(erts_driver_t* driver, /* Pointer to driver. */ /* * Must clean up the port. */ -#ifdef ERTS_SMP - erts_cancel_smp_ptimer(port->common.u.alive.ptimer); -#else - erts_cancel_timer(&(port->common.u.alive.tm)); -#endif + erts_cancel_port_timer(port); stopq(port); if (port->linebuf != NULL) { erts_free(ERTS_ALC_T_LINEBUF, @@ -910,18 +938,32 @@ int erts_port_handle_xports(Port *prt) ** -2 on type error */ -#define SET_VEC(iov, bv, bin, ptr, len, vlen) do { \ - (iov)->iov_base = (ptr); \ - (iov)->iov_len = (len); \ - if (sizeof((iov)->iov_len) < sizeof(len) \ - /* Check if (len) overflowed (iov)->iov_len */ \ - && (iov)->iov_len != (len)) { \ - goto L_overflow; \ - } \ - *(bv)++ = (bin); \ - (iov)++; \ - (vlen)++; \ -} while(0) +#ifdef DEBUG +#define MAX_SYSIOVEC_IOVLEN (1ull << (32 - 1)) +#else +#define MAX_SYSIOVEC_IOVLEN (1ull << (sizeof(((SysIOVec*)0)->iov_len) * 8 - 1)) +#endif + +static ERTS_INLINE void +io_list_to_vec_set_vec(SysIOVec **iov, ErlDrvBinary ***binv, + ErlDrvBinary *bin, byte *ptr, Uint len, + int *vlen) +{ + while (len > MAX_SYSIOVEC_IOVLEN) { + (*iov)->iov_base = ptr; + (*iov)->iov_len = MAX_SYSIOVEC_IOVLEN; + ptr += MAX_SYSIOVEC_IOVLEN; + len -= MAX_SYSIOVEC_IOVLEN; + (*iov)++; + (*vlen)++; + *(*binv)++ = bin; + } + (*iov)->iov_base = ptr; + (*iov)->iov_len = len; + *(*binv)++ = bin; + (*iov)++; + (*vlen)++; +} static int io_list_to_vec(Eterm obj, /* io-list */ @@ -932,11 +974,11 @@ io_list_to_vec(Eterm obj, /* io-list */ { DECLARE_ESTACK(s); Eterm* objp; - char *buf = cbin->orig_bytes; + byte *buf = (byte*)cbin->orig_bytes; Uint len = cbin->orig_size; Uint csize = 0; int vlen = 0; - char* cptr = buf; + byte* cptr = buf; goto L_jump_start; /* avoid push */ @@ -1004,15 +1046,17 @@ io_list_to_vec(Eterm obj, /* io-list */ len -= size; } else { if (csize != 0) { - SET_VEC(iov, binv, cbin, cptr, csize, vlen); + io_list_to_vec_set_vec(&iov, &binv, cbin, + cptr, csize, &vlen); cptr = buf; csize = 0; } if (pb->flags) { erts_emasculate_writable_binary(pb); } - SET_VEC(iov, binv, Binary2ErlDrvBinary(pb->val), - pb->bytes+offset, size, vlen); + io_list_to_vec_set_vec( + &iov, &binv, Binary2ErlDrvBinary(pb->val), + pb->bytes+offset, size, &vlen); } } else { ErlHeapBin* hb = (ErlHeapBin *) bptr; @@ -1032,7 +1076,7 @@ io_list_to_vec(Eterm obj, /* io-list */ } if (csize != 0) { - SET_VEC(iov, binv, cbin, cptr, csize, vlen); + io_list_to_vec_set_vec(&iov, &binv, cbin, cptr, csize, &vlen); } DESTROY_ESTACK(s); @@ -1058,10 +1102,13 @@ do { \ if (_bitsize != 0) goto L_type_error; \ if (thing_subtag(*binary_val(_real)) == REFC_BINARY_SUBTAG && \ _bitoffs == 0) { \ - b_size += _size; \ + b_size += _size; \ if (b_size < _size) goto L_overflow_error; \ in_clist = 0; \ - v_size++; \ + v_size++; \ + /* If iov_len is smaller then Uint we split the binary into*/ \ + /* multiple smaller (2GB) elements in the iolist.*/ \ + v_size += _size / MAX_SYSIOVEC_IOVLEN; \ if (_size >= ERL_SMALL_IO_BIN_LIMIT) { \ p_in_clist = 0; \ p_v_size++; \ @@ -1113,7 +1160,7 @@ io_list_vec_len(Eterm obj, int* vsize, Uint* csize, Uint p_v_size = 0; Uint p_c_size = 0; Uint p_in_clist = 0; - Uint total; /* Uint due to halfword emulator */ + Uint total; goto L_jump_start; /* avoid a push */ @@ -1213,7 +1260,7 @@ typedef struct { /* * Try doing an immediate driver callback call from a process. If * this fail, the operation should be scheduled in the normal case... - * + * Returns: ok to do the call, or error (lock busy, does not exist, etc) */ static ERTS_INLINE ErtsTryImmDrvCallResult try_imm_drv_call(ErtsTryImmDrvCallState *sp) @@ -1273,7 +1320,7 @@ try_imm_drv_call(ErtsTryImmDrvCallState *sp) reds_left_in = CONTEXT_REDS/10; else { if (IS_TRACED_FL(c_p, F_TRACE_SCHED_PROCS)) - trace_virtual_sched(c_p, am_out); + trace_sched(c_p, ERTS_PROC_LOCK_MAIN, am_out); /* * No status lock held while sending runnable * proc trace messages. It is however not needed @@ -1357,7 +1404,7 @@ finalize_imm_drv_call(ErtsTryImmDrvCallState *sp) } if (IS_TRACED_FL(c_p, F_TRACE_SCHED_PROCS)) - trace_virtual_sched(c_p, am_in); + trace_sched(c_p, ERTS_PROC_LOCK_MAIN, am_in); /* * No status lock held while sending runnable * proc trace messages. It is however not needed @@ -1404,52 +1451,35 @@ finalize_force_imm_drv_call(ErtsTryImmDrvCallState *sp) static ERTS_INLINE void queue_port_sched_op_reply(Process *rp, - ErtsProcLocks *rp_locksp, - Eterm *hp_start, - Eterm *hp, - Uint h_size, - ErlHeapFragment* bp, + ErtsProcLocks rp_locks, + ErtsHeapFactory* factory, Uint32 *ref_num, - Eterm msg) + Eterm msg, + Port* prt) { - Eterm ref = make_internal_ref(hp); + Eterm* hp = erts_produce_heap(factory, ERTS_QUEUE_PORT_SCHED_OP_REPLY_SIZE, 0); + Eterm ref; + + ref= make_internal_ref(hp); write_ref_thing(hp, ref_num[0], ref_num[1], ref_num[2]); hp += REF_THING_SIZE; msg = TUPLE2(hp, ref, msg); - hp += 3; - if (!bp) { - HRelease(rp, hp_start + h_size, hp); - } - else { - Uint used_h_size = hp - hp_start; - ASSERT(h_size >= used_h_size); - if (h_size > used_h_size) - bp = erts_resize_message_buffer(bp, used_h_size, &msg, 1); - } + erts_factory_trim_and_close(factory, &msg, 1); - erts_queue_message(rp, - rp_locksp, - bp, - msg, - NIL -#ifdef USE_VM_PROBES - , NIL -#endif - ); + erts_queue_message(rp, rp_locks, factory->message, msg, + prt ? prt->common.id : am_undefined); } static void -port_sched_op_reply(Eterm to, Uint32 *ref_num, Eterm msg) +port_sched_op_reply(Eterm to, Uint32 *ref_num, Eterm msg, Port* prt) { Process *rp = erts_proc_lookup_raw(to); if (rp) { - ErlOffHeap *ohp; - ErlHeapFragment* bp; + ErtsHeapFactory factory; Eterm msg_copy; Uint hsz, msg_sz; - Eterm *hp, *hp_start; ErtsProcLocks rp_locks = 0; hsz = ERTS_QUEUE_PORT_SCHED_OP_REPLY_SIZE; @@ -1460,24 +1490,20 @@ port_sched_op_reply(Eterm to, Uint32 *ref_num, Eterm msg) hsz += msg_sz; } - hp_start = hp = erts_alloc_message_heap(hsz, - &bp, - &ohp, - rp, - &rp_locks); - if (is_immed(msg)) - msg_copy = msg; - else - msg_copy = copy_struct(msg, msg_sz, &hp, ohp); + (void) erts_factory_message_create(&factory, rp, + &rp_locks, hsz); + msg_copy = (is_immed(msg) + ? msg + : copy_struct(msg, msg_sz, + &factory.hp, + factory.off_heap)); queue_port_sched_op_reply(rp, - &rp_locks, - hp_start, - hp, - hsz, - bp, + rp_locks, + &factory, ref_num, - msg_copy); + msg_copy, + prt); if (rp_locks) erts_smp_proc_unlock(rp, rp_locks); @@ -1532,9 +1558,8 @@ erts_schedule_proc2port_signal(Process *c_p, ERTS_SMP_MSGQ_MV_INQ2PRIVQ(c_p); c_p->msg.save = c_p->msg.last; - erts_smp_proc_unlock(c_p, - (ERTS_PROC_LOCK_MAIN - | ERTS_PROC_LOCKS_MSG_RECEIVE)); + erts_smp_proc_unlock(c_p, (ERTS_PROC_LOCKS_MSG_RECEIVE + | ERTS_PROC_LOCK_MAIN)); } @@ -1552,20 +1577,49 @@ erts_schedule_proc2port_signal(Process *c_p, erts_smp_proc_lock(c_p, ERTS_PROC_LOCK_MAIN); if (sched_res != 0) { - if (refp) + if (refp) { + /* + * We need to restore the message queue save + * pointer to the beginning of the message queue + * since the caller now wont wait for a message + * containing the reference created above... + */ + ASSERT(c_p); + erts_smp_proc_lock(c_p, ERTS_PROC_LOCKS_MSG_RECEIVE); + JOIN_MESSAGE(c_p); + erts_smp_proc_unlock(c_p, ERTS_PROC_LOCKS_MSG_RECEIVE); *refp = NIL; + } return ERTS_PORT_OP_DROPPED; } return ERTS_PORT_OP_SCHEDULED; } -static ERTS_INLINE void -send_badsig(Port *prt) +static int +erts_schedule_port2port_signal(Eterm port_num, ErtsProc2PortSigData *sigdp, + int task_flags, + ErtsProc2PortSigCallback callback) { + Port *prt = erts_port_lookup_raw(port_num); + + if (!prt) + return -1; + + sigdp->caller = ERTS_INVALID_PID; + + return erts_port_task_schedule(prt->common.id, + NULL, + ERTS_PORT_TASK_PROC_SIG, + sigdp, + callback, + task_flags); +} + +static ERTS_INLINE void +send_badsig(Port *prt) { ErtsProcLocks rp_locks = ERTS_PROC_LOCKS_XSIG_SEND; Process* rp; Eterm connected = ERTS_PORT_GET_CONNECTED(prt); - ERTS_SMP_CHK_NO_PROC_LOCKS; ERTS_LC_ASSERT(erts_get_scheduler_id()); @@ -1585,15 +1639,13 @@ send_badsig(Port *prt) 0); if (rp_locks) erts_smp_proc_unlock(rp, rp_locks); - } -} + } /* exit sent */ +} /* send_badsig */ static void -badsig_received(int bang_op, - Port *prt, +badsig_received(int bang_op, Port *prt, erts_aint32_t state, - int bad_output_value) -{ + int bad_output_value) { /* * if (bang_op) * we are part of a "Prt ! Something" operation @@ -1609,30 +1661,28 @@ badsig_received(int bang_op, } if (bang_op) send_badsig(prt); - } -} + } /* not invalid */ +} /* behaved accordingly */ static int -port_badsig(Port *prt, erts_aint32_t state, int op, ErtsProc2PortSigData *sigdp) -{ +port_badsig(Port *prt, erts_aint32_t state, int op, + ErtsProc2PortSigData *sigdp) { if (op == ERTS_PROC2PORT_SIG_EXEC) badsig_received(sigdp->flags & ERTS_P2P_SIG_DATA_FLG_BANG_OP, prt, state, sigdp->flags & ERTS_P2P_SIG_DATA_FLG_BAD_OUTPUT); if (sigdp->flags & ERTS_P2P_SIG_DATA_FLG_REPLY) - port_sched_op_reply(sigdp->caller, sigdp->ref, am_badarg); + port_sched_op_reply(sigdp->caller, sigdp->ref, am_badarg, prt); return ERTS_PORT_REDS_BADSIG; -} - - -/* - * bad_port_signal() will +} /* port_badsig */ +/* bad_port_signal() will * - preserve signal order of signals. * - send a 'badsig' exit signal to connected process if 'from' is an * internal pid and the port is alive when the bad signal reaches * it. */ + static ErtsPortOpResult bad_port_signal(Process *c_p, int flags, @@ -1707,24 +1757,44 @@ call_driver_outputv(int bang_op, if (bang_op && from != ERTS_PORT_GET_CONNECTED(prt)) send_badsig(prt); else { + ErtsSchedulerData *esdp = erts_get_scheduler_data(); ErlDrvSizeT size = evp->size; + ERTS_MSACC_PUSH_AND_SET_STATE_M(ERTS_MSACC_STATE_PORT); ERTS_SMP_LC_ASSERT(erts_lc_is_port_locked(prt) || ERTS_IS_CRASH_DUMPING); + + if (IS_TRACED_FL(prt, F_TRACE_RECEIVE)) + trace_port_receive(prt, caller, am_commandv, evp); + #ifdef USE_VM_PROBES if (DTRACE_ENABLED(driver_outputv)) { DTRACE_FORMAT_COMMON_PID_AND_PORT(caller, prt); DTRACE4(driver_outputv, process_str, port_str, prt->name, size); } #endif +#ifdef USE_LTTNG_VM_TRACEPOINTS + if (LTTNG_ENABLED(driver_outputv)) { + lttng_decl_portbuf(port_str); + lttng_decl_procbuf(proc_str); + lttng_pid_to_str(caller, proc_str); + lttng_port_to_str(prt, port_str); + LTTNG4(driver_outputv, proc_str, port_str, prt->name, size); + } +#endif prt->caller = caller; (*drv->outputv)((ErlDrvData) prt->drv_data, evp); prt->caller = NIL; prt->bytes_out += size; - erts_smp_atomic_add_nob(&erts_bytes_out, size); + if (esdp) + esdp->io.out += (Uint64) size; + else + erts_atomic64_add_nob(&bytes_out, (erts_aint64_t) size); + + ERTS_MSACC_POP_STATE_M(); } } @@ -1738,7 +1808,6 @@ cleanup_scheduled_outputv(ErlIOVec *ev, ErlDrvBinary *cbinp) driver_free_binary(ev->binv[i]); if (cbinp) driver_free_binary(cbinp); - erts_free(ERTS_ALC_T_DRV_CMD_DATA, ev); } static int @@ -1773,7 +1842,7 @@ port_sig_outputv(Port *prt, erts_aint32_t state, int op, ErtsProc2PortSigData *s } if (sigdp->flags & ERTS_P2P_SIG_DATA_FLG_REPLY) - port_sched_op_reply(sigdp->caller, sigdp->ref, reply); + port_sched_op_reply(sigdp->caller, sigdp->ref, reply, prt); cleanup_scheduled_outputv(sigdp->u.outputv.evp, sigdp->u.outputv.cbinp); @@ -1804,7 +1873,8 @@ call_driver_output(int bang_op, if (bang_op && from != ERTS_PORT_GET_CONNECTED(prt)) send_badsig(prt); else { - + ErtsSchedulerData *esdp = erts_get_scheduler_data(); + ERTS_MSACC_PUSH_AND_SET_STATE_M(ERTS_MSACC_STATE_PORT); ERTS_SMP_LC_ASSERT(erts_lc_is_port_locked(prt) || ERTS_IS_CRASH_DUMPING); @@ -1814,13 +1884,30 @@ call_driver_output(int bang_op, DTRACE4(driver_output, process_str, port_str, prt->name, size); } #endif +#ifdef USE_LTTNG_VM_TRACEPOINTS + if (LTTNG_ENABLED(driver_output)) { + lttng_decl_portbuf(port_str); + lttng_decl_procbuf(proc_str); + lttng_pid_to_str(caller, proc_str); + lttng_port_to_str(prt, port_str); + LTTNG4(driver_output, proc_str, port_str, prt->name, size); + } +#endif + + if (IS_TRACED_FL(prt, F_TRACE_RECEIVE)) + trace_port_receive(prt, caller, am_command, bufp, size); prt->caller = caller; (*drv->output)((ErlDrvData) prt->drv_data, bufp, size); prt->caller = NIL; prt->bytes_out += size; - erts_smp_atomic_add_nob(&erts_bytes_out, size); + if (esdp) + esdp->io.out += (Uint64) size; + else + erts_atomic64_add_nob(&bytes_out, (erts_aint64_t) size); + + ERTS_MSACC_POP_STATE_M(); } } @@ -1863,13 +1950,195 @@ port_sig_output(Port *prt, erts_aint32_t state, int op, ErtsProc2PortSigData *si } if (sigdp->flags & ERTS_P2P_SIG_DATA_FLG_REPLY) - port_sched_op_reply(sigdp->caller, sigdp->ref, reply); + port_sched_op_reply(sigdp->caller, sigdp->ref, reply, prt); cleanup_scheduled_output(sigdp->u.output.bufp); return ERTS_PORT_REDS_CMD_OUTPUT; } + +/* + * This erts_port_output will always create a port task. + * The call is treated as a port_command call, i.e. no + * badsig i generated if the input in invalid. However + * an error_logger message is generated. + */ +int +erts_port_output_async(Port *prt, Eterm from, Eterm list) +{ + + ErtsPortOpResult res; + ErtsProc2PortSigData *sigdp; + erts_driver_t *drv = prt->drv_ptr; + size_t size; + int task_flags; + ErtsProc2PortSigCallback port_sig_callback; + ErlDrvBinary *cbin = NULL; + ErlIOVec *evp = NULL; + char *buf = NULL; + ErtsPortTaskHandle *ns_pthp; + + if (drv->outputv) { + ErlIOVec ev; + SysIOVec* ivp; + ErlDrvBinary** bvp; + int vsize; + Uint csize; + Uint pvsize; + Uint pcsize; + size_t iov_offset, binv_offset, alloc_size; + Uint blimit = 0; + char *ptr; + int i; + + Eterm* bptr = NULL; + Uint offset; + + if (is_binary(list)) { + /* We optimize for when we get a procbin without offset */ + Eterm real_bin; + int bitoffs; + int bitsize; + ERTS_GET_REAL_BIN(list, real_bin, offset, bitoffs, bitsize); + bptr = binary_val(real_bin); + if (*bptr == HEADER_PROC_BIN && bitoffs == 0) { + size = binary_size(list); + vsize = 1; + } else + bptr = NULL; + } + + if (!bptr) { + if (io_list_vec_len(list, &vsize, &csize, &pvsize, &pcsize, &size)) + goto bad_value; + + /* To pack or not to pack (small binaries) ...? */ + if (vsize >= SMALL_WRITE_VEC) { + /* Do pack */ + vsize = pvsize + 1; + csize = pcsize; + blimit = ERL_SMALL_IO_BIN_LIMIT; + } + cbin = driver_alloc_binary(csize); + if (!cbin) + erts_alloc_enomem(ERTS_ALC_T_DRV_BINARY, ERTS_SIZEOF_Binary(csize)); + } + + + iov_offset = ERTS_ALC_DATA_ALIGN_SIZE(sizeof(ErlIOVec)); + binv_offset = iov_offset; + binv_offset += ERTS_ALC_DATA_ALIGN_SIZE((vsize+1)*sizeof(SysIOVec)); + alloc_size = binv_offset; + alloc_size += (vsize+1)*sizeof(ErlDrvBinary *); + + sigdp = erts_port_task_alloc_p2p_sig_data_extra(alloc_size, (void**)&ptr); + + evp = (ErlIOVec *) ptr; + ivp = evp->iov = (SysIOVec *) (ptr + iov_offset); + bvp = evp->binv = (ErlDrvBinary **) (ptr + binv_offset); + + ivp[0].iov_base = NULL; + ivp[0].iov_len = 0; + bvp[0] = NULL; + + if (bptr) { + ProcBin* pb = (ProcBin *) bptr; + + ivp[1].iov_base = pb->bytes+offset; + ivp[1].iov_len = size; + bvp[1] = Binary2ErlDrvBinary(pb->val); + + evp->vsize = 1; + } else { + + evp->vsize = io_list_to_vec(list, ivp+1, bvp+1, cbin, blimit); + if (evp->vsize < 0) { + if (evp != &ev) + erts_free(ERTS_ALC_T_DRV_CMD_DATA, evp); + driver_free_binary(cbin); + goto bad_value; + } + } +#if 0 + /* This assertion may say something useful, but it can + be falsified during the emulator test suites. */ + ASSERT(evp->vsize == vsize); +#endif + evp->vsize++; + evp->size = size; /* total size */ + + /* Need to increase refc on all binaries */ + for (i = 1; i < evp->vsize; i++) + if (bvp[i]) + driver_binary_inc_refc(bvp[i]); + + sigdp->flags = ERTS_P2P_SIG_TYPE_OUTPUTV; + sigdp->u.outputv.from = from; + sigdp->u.outputv.evp = evp; + sigdp->u.outputv.cbinp = cbin; + port_sig_callback = port_sig_outputv; + } else { + ErlDrvSizeT ERTS_DECLARE_DUMMY(r); + + /* + * Apperently there exist code that write 1 byte to + * much in buffer. Where it resides I don't know, but + * we can live with one byte extra allocated... + */ + + if (erts_iolist_size(list, &size)) + goto bad_value; + + buf = erts_alloc(ERTS_ALC_T_DRV_CMD_DATA, size + 1); + + r = erts_iolist_to_buf(list, buf, size); + ASSERT(ERTS_IOLIST_TO_BUF_SUCCEEDED(r)); + + sigdp = erts_port_task_alloc_p2p_sig_data(); + sigdp->flags = ERTS_P2P_SIG_TYPE_OUTPUT; + sigdp->u.output.from = from; + sigdp->u.output.bufp = buf; + sigdp->u.output.size = size; + port_sig_callback = port_sig_output; + } + sigdp->flags = 0; + ns_pthp = NULL; + task_flags = 0; + + res = erts_schedule_proc2port_signal(NULL, + prt, + ERTS_INVALID_PID, + NULL, + sigdp, + task_flags, + ns_pthp, + port_sig_callback); + + if (res != ERTS_PORT_OP_SCHEDULED) { + if (drv->outputv) + cleanup_scheduled_outputv(evp, cbin); + else + cleanup_scheduled_output(buf); + return 1; + } + return 1; + +bad_value: + + /* + * We call badsig directly here as this function is called with + * the main lock of the calling process still held. + * At the moment this operation is always not a bang_op, so + * only an error_logger message should be generated, no badsig. + */ + + badsig_received(0, prt, erts_atomic32_read_nob(&prt->state), 1); + + return 0; + +} + ErtsPortOpResult erts_port_output(Process *c_p, int flags, @@ -1879,7 +2148,7 @@ erts_port_output(Process *c_p, Eterm *refp) { ErtsPortOpResult res; - ErtsProc2PortSigData *sigdp; + ErtsProc2PortSigData *sigdp = NULL; erts_driver_t *drv = prt->drv_ptr; size_t size; int try_call; @@ -1932,7 +2201,6 @@ erts_port_output(Process *c_p, DTRACE4(port_command, process_str, port_str, prt->name, "command"); } #endif - if (drv->outputv) { ErlIOVec ev; SysIOVec iv[SMALL_WRITE_VEC]; @@ -1961,10 +2229,13 @@ erts_port_output(Process *c_p, evp = &ev; } else { - char *ptr = erts_alloc((try_call - ? ERTS_ALC_T_TMP - : ERTS_ALC_T_DRV_CMD_DATA), alloc_size); - + char *ptr; + if (try_call) { + ptr = erts_alloc(ERTS_ALC_T_TMP, alloc_size); + } else { + sigdp = erts_port_task_alloc_p2p_sig_data_extra( + alloc_size, (void**)&ptr); + } evp = (ErlIOVec *) ptr; ivp = evp->iov = (SysIOVec *) (ptr + iov_offset); bvp = evp->binv = (ErlDrvBinary **) (ptr + binv_offset); @@ -1993,9 +2264,12 @@ erts_port_output(Process *c_p, bvp[0] = NULL; evp->vsize = io_list_to_vec(list, ivp+1, bvp+1, cbin, blimit); if (evp->vsize < 0) { - if (evp != &ev) - erts_free(try_call ? ERTS_ALC_T_TMP : ERTS_ALC_T_DRV_CMD_DATA, - evp); + if (evp != &ev) { + if (try_call) + erts_free(ERTS_ALC_T_TMP, evp); + else + erts_port_task_free_p2p_sig_data(sigdp); + } driver_free_binary(cbin); goto bad_value; } @@ -2047,8 +2321,10 @@ erts_port_output(Process *c_p, /* Fall through... */ case ERTS_TRY_IMM_DRV_CALL_INVALID_PORT: driver_free_binary(cbin); - if (evp != &ev) + if (evp != &ev) { + ASSERT(!sigdp); erts_free(ERTS_ALC_T_TMP, evp); + } if (try_call_res != ERTS_TRY_IMM_DRV_CALL_OK) return ERTS_PORT_OP_DROPPED; if (c_p) @@ -2059,8 +2335,10 @@ erts_port_output(Process *c_p, if (async_nosuspend && (sched_flags & (busy_flgs|ERTS_PTS_FLG_EXIT))) { driver_free_binary(cbin); - if (evp != &ev) + if (evp != &ev) { + ASSERT(!sigdp); erts_free(ERTS_ALC_T_TMP, evp); + } return ((sched_flags & ERTS_PTS_FLG_EXIT) ? ERTS_PORT_OP_DROPPED : ERTS_PORT_OP_BUSY); @@ -2075,9 +2353,16 @@ erts_port_output(Process *c_p, if (bvp[i]) driver_binary_inc_refc(bvp[i]); - new_evp = erts_alloc(ERTS_ALC_T_DRV_CMD_DATA, alloc_size); + /* The port task and iovec is allocated in the + same structure as an optimization. This + is especially important in erts_port_output_async + of when !try_call */ + ASSERT(sigdp == NULL); + sigdp = erts_port_task_alloc_p2p_sig_data_extra( + alloc_size, (void**)&new_evp); if (evp != &ev) { + /* Copy from TMP alloc to port task */ sys_memcpy((void *) new_evp, (void *) evp, alloc_size); new_evp->iov = (SysIOVec *) (((char *) new_evp) + iov_offset); @@ -2125,7 +2410,6 @@ erts_port_output(Process *c_p, evp = new_evp; } - sigdp = erts_port_task_alloc_p2p_sig_data(); sigdp->flags = ERTS_P2P_SIG_TYPE_OUTPUTV; sigdp->u.outputv.from = from; sigdp->u.outputv.evp = evp; @@ -2247,7 +2531,7 @@ erts_port_output(Process *c_p, sigdp->flags &= ~ERTS_P2P_SIG_DATA_FLG_NOSUSPEND; else if (async_nosuspend) { ErtsSchedulerData *esdp = (c_p - ? ERTS_PROC_GET_SCHDATA(c_p) + ? erts_proc_sched_data(c_p) : erts_get_scheduler_data()); ASSERT(esdp); ns_pthp = &esdp->nosuspend_port_task_handle; @@ -2334,7 +2618,10 @@ call_deliver_port_exit(int bang_op, return ERTS_PORT_OP_DROPPED; } - if (!erts_deliver_port_exit(prt, from, reason, bang_op)) + if (IS_TRACED_FL(prt, F_TRACE_RECEIVE)) + trace_port_receive(prt, from, am_close); + + if (!erts_deliver_port_exit(prt, from, reason, bang_op, broken_link)) return ERTS_PORT_OP_DROPPED; #ifdef USE_VM_PROBES @@ -2371,7 +2658,7 @@ port_sig_exit(Port *prt, if (sigdp->u.exit.bp) free_message_buffer(sigdp->u.exit.bp); if (sigdp->flags & ERTS_P2P_SIG_DATA_FLG_REPLY) - port_sched_op_reply(sigdp->caller, sigdp->ref, msg); + port_sched_op_reply(sigdp->caller, sigdp->ref, msg, prt); return ERTS_PORT_REDS_EXIT; } @@ -2393,6 +2680,11 @@ erts_port_exit(Process *c_p, | ERTS_PORT_SIG_FLG_BROKEN_LINK | ERTS_PORT_SIG_FLG_FORCE_SCHED)) == 0); +#ifndef __WIN32__ + if (prt->drv_ptr == &forker_driver) + return ERTS_PORT_OP_DROPPED; +#endif + if (!(flags & ERTS_PORT_SIG_FLG_FORCE_SCHED)) { ErtsTryImmDrvCallState try_call_state = ERTS_INIT_TRY_IMM_DRV_CALL_STATE(c_p, @@ -2400,13 +2692,13 @@ erts_port_exit(Process *c_p, ERTS_PORT_SFLGS_INVALID_LOOKUP, 0, !refp, - am_exit); + am_close); switch (try_imm_drv_call(&try_call_state)) { case ERTS_TRY_IMM_DRV_CALL_OK: { res = call_deliver_port_exit(flags & ERTS_PORT_SIG_FLG_BANG_OP, - from, + c_p ? c_p->common.id : from, prt, try_call_state.state, reason, @@ -2485,8 +2777,11 @@ set_port_connected(int bang_op, return ERTS_PORT_OP_DROPPED; } + if (IS_TRACED_FL(prt, F_TRACE_RECEIVE)) + trace_port_receive(prt, from, am_connect, connect); + ERTS_PORT_SET_CONNECTED(prt, connect); - deliver_result(prt->common.id, from, am_connected); + deliver_result(prt, prt->common.id, from, am_connected); #ifdef USE_VM_PROBES if(DTRACE_ENABLED(port_command)) { @@ -2508,10 +2803,22 @@ set_port_connected(int bang_op, erts_add_link(&ERTS_P_LINKS(rp), LINK_PID, prt->common.id); erts_add_link(&ERTS_P_LINKS(prt), LINK_PID, connect); + if (IS_TRACED_FL(rp, F_TRACE_PROCS)) + trace_proc(NULL, 0, rp, am_getting_linked, prt->common.id); + ERTS_PORT_SET_CONNECTED(prt, connect); erts_smp_proc_unlock(rp, ERTS_PROC_LOCK_LINK); + if (IS_TRACED_FL(prt, F_TRACE_PORTS)) + trace_port(prt, am_getting_linked, connect); + if (IS_TRACED_FL(prt, F_TRACE_RECEIVE)) + trace_port_receive(prt, from, am_connect, connect); + if (IS_TRACED_FL(prt, F_TRACE_SEND)) { + Eterm hp[3]; + trace_port_send(prt, from, TUPLE2(hp, prt->common.id, am_connected), 1); + } + #ifdef USE_VM_PROBES if (DTRACE_ENABLED(port_connect)) { DTRACE_CHARBUF(process_str, DTRACE_TERM_BUF_SIZE); @@ -2544,7 +2851,7 @@ port_sig_connect(Port *prt, erts_aint32_t state, int op, ErtsProc2PortSigData *s msg = am_true; } if (sigdp->flags & ERTS_P2P_SIG_DATA_FLG_REPLY) - port_sched_op_reply(sigdp->caller, sigdp->ref, msg); + port_sched_op_reply(sigdp->caller, sigdp->ref, msg, prt); return ERTS_PORT_REDS_CONNECT; } @@ -2614,8 +2921,11 @@ static void port_unlink(Port *prt, Eterm from) { ErtsLink *lnk = erts_remove_link(&ERTS_P_LINKS(prt), from); - if (lnk) + if (lnk) { + if (IS_TRACED_FL(prt, F_TRACE_PORTS)) + trace_port(prt, am_getting_unlinked, from); erts_destroy_link(lnk); + } } static int @@ -2624,7 +2934,7 @@ port_sig_unlink(Port *prt, erts_aint32_t state, int op, ErtsProc2PortSigData *si if (op == ERTS_PROC2PORT_SIG_EXEC) port_unlink(prt, sigdp->u.unlink.from); if (sigdp->flags & ERTS_P2P_SIG_DATA_FLG_REPLY) - port_sched_op_reply(sigdp->caller, sigdp->ref, am_true); + port_sched_op_reply(sigdp->caller, sigdp->ref, am_true, prt); return ERTS_PORT_REDS_UNLINK; } @@ -2685,10 +2995,10 @@ port_link_failure(Eterm port_id, Eterm linker) NIL, NULL, 0); - if (xres >= 0 && IS_TRACED_FL(rp, F_TRACE_PROCS)) { + if (xres >= 0) { /* We didn't exit the process and it is traced */ if (IS_TRACED_FL(rp, F_TRACE_PROCS)) - trace_proc(NULL, rp, am_getting_unlinked, port_id); + trace_proc(NULL, 0, rp, am_getting_unlinked, port_id); } if (rp_locks) erts_smp_proc_unlock(rp, rp_locks); @@ -2699,10 +3009,15 @@ port_link_failure(Eterm port_id, Eterm linker) static void port_link(Port *prt, erts_aint32_t state, Eterm to) { - if (!(state & ERTS_PORT_SFLGS_INVALID_LOOKUP)) + if (IS_TRACED_FL(prt, F_TRACE_PORTS)) + trace_port(prt, am_getting_linked, to); + if (!(state & ERTS_PORT_SFLGS_INVALID_LOOKUP)) { erts_add_link(&ERTS_P_LINKS(prt), LINK_PID, to); - else + } else { port_link_failure(prt->common.id, to); + if (IS_TRACED_FL(prt, F_TRACE_PORTS)) + trace_port(prt, am_unlink, to); + } } static int @@ -2710,10 +3025,11 @@ port_sig_link(Port *prt, erts_aint32_t state, int op, ErtsProc2PortSigData *sigd { if (op == ERTS_PROC2PORT_SIG_EXEC) port_link(prt, state, sigdp->u.link.to); - else + else { port_link_failure(sigdp->u.link.port, sigdp->u.link.to); + } if (sigdp->flags & ERTS_P2P_SIG_DATA_FLG_REPLY) - port_sched_op_reply(sigdp->caller, sigdp->ref, am_true); + port_sched_op_reply(sigdp->caller, sigdp->ref, am_true, prt); return ERTS_PORT_REDS_LINK; } @@ -2757,6 +3073,317 @@ erts_port_link(Process *c_p, Port *prt, Eterm to, Eterm *refp) port_sig_link); } +static void +port_monitor_failure(Eterm port_id, Eterm origin, Eterm ref_DOWN) +{ + Process *origin_p; + ErtsProcLocks p_locks = ERTS_PROC_LOCK_LINK; + ASSERT(is_internal_pid(origin)); + + origin_p = erts_pid2proc(NULL, 0, origin, p_locks); + if (! origin_p) { return; } + + /* Send the DOWN message immediately. Ref is made on the fly because + * caller has never seen it yet. */ + erts_queue_monitor_message(origin_p, &p_locks, ref_DOWN, + am_port, port_id, am_noproc); + erts_smp_proc_unlock(origin_p, p_locks); +} + +/* Origin wants to monitor port Prt. State contains possible error, which has + * happened just before. Name is either NIL or an atom, if user monitors + * a port by name. Ref is premade reference that will be returned to user */ +static void +port_monitor(Port *prt, erts_aint32_t state, Eterm origin, + Eterm name, Eterm ref) +{ + Eterm name_or_nil = is_atom(name) ? name : NIL; + + ASSERT(is_pid(origin)); + ASSERT(is_atom(name) || is_port(name) || name == NIL); + ASSERT(is_internal_ref(ref)); + + if (!(state & ERTS_PORT_SFLGS_INVALID_LOOKUP)) { + ErtsProcLocks p_locks = ERTS_PROC_LOCK_LINK; + + Process *origin_p = erts_pid2proc(NULL, 0, origin, p_locks); + if (! origin_p) { + goto failure; + } + erts_add_monitor(&ERTS_P_MONITORS(origin_p), MON_ORIGIN, ref, + prt->common.id, name_or_nil); + erts_add_monitor(&ERTS_P_MONITORS(prt), MON_TARGET, ref, + origin, name_or_nil); + + erts_smp_proc_unlock(origin_p, p_locks); + } else { +failure: + port_monitor_failure(prt->common.id, origin, ref); + } +} + +static int +port_sig_monitor(Port *prt, erts_aint32_t state, int op, + ErtsProc2PortSigData *sigdp) +{ + Eterm hp[REF_THING_SIZE]; + Eterm ref = make_internal_ref(&hp); + write_ref_thing(hp, sigdp->ref[0], sigdp->ref[1], sigdp->ref[2]); + + if (op == ERTS_PROC2PORT_SIG_EXEC) { + /* erts_add_monitor call inside port_monitor will copy ref from hp */ + port_monitor(prt, state, + sigdp->u.monitor.origin, + sigdp->u.monitor.name, + ref); + } else { + port_monitor_failure(sigdp->u.monitor.name, + sigdp->u.monitor.origin, + ref); + } + if (sigdp->flags & ERTS_P2P_SIG_DATA_FLG_REPLY) { + port_sched_op_reply(sigdp->caller, sigdp->ref, am_true, prt); + } + return ERTS_PORT_REDS_MONITOR; +} + +/* Creates monitor between Origin and Target. Ref must be initialized to + * a reference (ref may be rewritten to be used to serve additionally as a + * signal id). Name is atom if user monitors port by name or NIL */ +ErtsPortOpResult +erts_port_monitor(Process *origin, Port *port, Eterm name, Eterm *refp) +{ + ErtsProc2PortSigData *sigdp; + ErtsTryImmDrvCallState try_call_state + = ERTS_INIT_TRY_IMM_DRV_CALL_STATE( + origin, port, ERTS_PORT_SFLGS_INVALID_LOOKUP, + 0, + 0, /* trap_ref is always set so !trap_ref always is false */ + am_monitor); + + ASSERT(origin); + ASSERT(port); + ASSERT(is_atom(name) || is_port(name)); + ASSERT(refp); + + switch (try_imm_drv_call(&try_call_state)) { + case ERTS_TRY_IMM_DRV_CALL_OK: + port_monitor(port, try_call_state.state, origin->common.id, name, *refp); + finalize_imm_drv_call(&try_call_state); + BUMP_REDS(origin, ERTS_PORT_REDS_MONITOR); + return ERTS_PORT_OP_DONE; + case ERTS_TRY_IMM_DRV_CALL_INVALID_PORT: + return ERTS_PORT_OP_BADARG; + default: + break; /* Schedule call instead... */ + } + + sigdp = erts_port_task_alloc_p2p_sig_data(); + sigdp->flags = ERTS_P2P_SIG_TYPE_MONITOR; + sigdp->u.monitor.origin = origin->common.id; + sigdp->u.monitor.name = name; /* either named monitor, or port id */ + + /* Ref contents will be initialized here */ + return erts_schedule_proc2port_signal(origin, port, origin->common.id, + refp, sigdp, 0, NULL, + port_sig_monitor); +} + +static void +port_demonitor_failure(Eterm port_id, Eterm origin, Eterm ref) +{ + Process *origin_p; + ErtsProcLocks rp_locks = ERTS_PROC_LOCK_LINK; + ErtsMonitor *mon1; + ASSERT(is_internal_pid(origin)); + + origin_p = erts_pid2proc(NULL, 0, origin, rp_locks); + if (! origin_p) { return; } + + /* do not send any DOWN messages, drop monitors on process */ + mon1 = erts_remove_monitor(&ERTS_P_MONITORS(origin_p), ref); + if (mon1 != NULL) { + erts_destroy_monitor(mon1); + } + + erts_smp_proc_unlock(origin_p, rp_locks); +} + +/* Origin wants to demonitor port Prt. State contains possible error, which has + * happened just before. Ref is reference to monitor */ +static void +port_demonitor(Port *port, erts_aint32_t state, Eterm origin, Eterm ref) +{ + ASSERT(port); + ASSERT(is_pid(origin)); + ASSERT(is_internal_ref(ref)); + + if (!(state & ERTS_PORT_SFLGS_INVALID_LOOKUP)) { + ErtsProcLocks p_locks = ERTS_PROC_LOCK_LINK; + Process *origin_p = erts_pid2proc(NULL, 0, origin, p_locks); + if (origin_p) { + ErtsMonitor *mon1 = erts_remove_monitor(&ERTS_P_MONITORS(origin_p), + ref); + if (mon1 != NULL) { + erts_destroy_monitor(mon1); + } + } + if (1) { + ErtsMonitor *mon2 = erts_remove_monitor(&ERTS_P_MONITORS(port), + ref); + if (mon2 != NULL) { + erts_destroy_monitor(mon2); + } + } + if (origin_p) { /* when origin is dying, it won't be found */ + erts_smp_proc_unlock(origin_p, p_locks); + } + } else { + port_demonitor_failure(port->common.id, origin, ref); + } +} + +static int +port_sig_demonitor(Port *prt, erts_aint32_t state, int op, + ErtsProc2PortSigData *sigdp) +{ + Eterm hp[REF_THING_SIZE]; + Eterm ref = make_internal_ref(&hp); + write_ref_thing(hp, sigdp->u.demonitor.ref[0], + sigdp->u.demonitor.ref[1], + sigdp->u.demonitor.ref[2]); + if (op == ERTS_PROC2PORT_SIG_EXEC) { + port_demonitor(prt, state, sigdp->u.demonitor.origin, ref); + } else { + port_demonitor_failure(sigdp->u.demonitor.name, + sigdp->u.demonitor.origin, + ref); + } + if (sigdp->flags & ERTS_P2P_SIG_DATA_FLG_REPLY) { + port_sched_op_reply(sigdp->caller, sigdp->ref, am_true, prt); + } + return ERTS_PORT_REDS_DEMONITOR; +} + +/* Removes monitor between origin and target, identified by ref. + * Mode defines normal or relaxed demonitor rules (process is at death) */ +ErtsPortOpResult erts_port_demonitor(Process *origin, ErtsDemonitorMode mode, + Port *target, Eterm ref, + Eterm *trap_ref) +{ + Process *c_p = mode == ERTS_PORT_DEMONITOR_NORMAL ? origin : NULL; + ErtsProc2PortSigData *sigdp; + ErtsTryImmDrvCallState try_call_state + = ERTS_INIT_TRY_IMM_DRV_CALL_STATE( + c_p, + target, ERTS_PORT_SFLGS_INVALID_LOOKUP, + 0, + !trap_ref, + am_demonitor); + + ASSERT(origin); + ASSERT(target); + ASSERT(is_internal_ref(ref)); + + switch (try_imm_drv_call(&try_call_state)) { + case ERTS_TRY_IMM_DRV_CALL_OK: + port_demonitor(target, try_call_state.state, origin->common.id, ref); + finalize_imm_drv_call(&try_call_state); + if (mode == ERTS_PORT_DEMONITOR_NORMAL) { + BUMP_REDS(origin, ERTS_PORT_REDS_DEMONITOR); + } + return ERTS_PORT_OP_DONE; + case ERTS_TRY_IMM_DRV_CALL_INVALID_PORT: + return ERTS_PORT_OP_BADARG; + default: + break; /* Schedule call instead... */ + } + + sigdp = erts_port_task_alloc_p2p_sig_data(); + sigdp->flags = ERTS_P2P_SIG_TYPE_DEMONITOR; + sigdp->u.demonitor.origin = origin->common.id; + sigdp->u.demonitor.name = target->common.id; + { + RefThing *reft = ref_thing_ptr(ref); + /* Start from 1 skip ref arity */ + sys_memcpy(sigdp->u.demonitor.ref, + internal_thing_ref_numbers(reft), + sizeof(sigdp->u.demonitor.ref)); + } + + /* Ref contents will be initialized here */ + return erts_schedule_proc2port_signal(c_p, target, origin->common.id, + trap_ref, sigdp, 0, NULL, + port_sig_demonitor); +} + +static void +init_ack_send_reply(Port *port, Eterm resp) +{ + + if (!is_internal_port(resp)) { + Process *rp = erts_proc_lookup_raw(port->async_open_port->to); + erts_smp_proc_lock(rp, ERTS_PROC_LOCK_LINK); + erts_remove_link(&ERTS_P_LINKS(port), port->async_open_port->to); + erts_remove_link(&ERTS_P_LINKS(rp), port->common.id); + erts_smp_proc_unlock(rp, ERTS_PROC_LOCK_LINK); + } + port_sched_op_reply(port->async_open_port->to, + port->async_open_port->ref, + resp, + port); + + erts_free(ERTS_ALC_T_PRTSD, port->async_open_port); + port->async_open_port = NULL; +} + +void +erl_drv_init_ack(ErlDrvPort ix, ErlDrvData res) { + Port *port = erts_drvport2port(ix); + SWord err_type = (SWord)res; + Eterm resp; + + if (port == ERTS_INVALID_ERL_DRV_PORT && port->async_open_port) + return; + + if (port->async_open_port) { + switch(err_type) { + case -3: + resp = am_badarg; + break; + case -2: { + char *str = erl_errno_id(errno); + resp = erts_atom_put((byte *) str, strlen(str), + ERTS_ATOM_ENC_LATIN1, 1); + break; + } + case -1: + resp = am_einval; + break; + default: + resp = port->common.id; + break; + } + + init_ack_send_reply(port, resp); + + if (err_type == -1 || err_type == -2 || err_type == -3) + driver_failure_term(ix, am_normal, 0); + port->drv_data = err_type; + } +} + +void +erl_drv_set_os_pid(ErlDrvPort ix, ErlDrvSInt pid) { + Port *port = erts_drvport2port(ix); + + if (port == ERTS_INVALID_ERL_DRV_PORT) + return; + + port->os_pid = (SWord)pid; + +} + void erts_init_io(int port_tab_size, int port_tab_size_ignore_files, int legacy_port_tab) @@ -2767,6 +3394,9 @@ void erts_init_io(int port_tab_size, drv_list_rwmtx_opts.type = ERTS_SMP_RWMTX_TYPE_EXTREMELY_FREQUENT_READ; drv_list_rwmtx_opts.lived = ERTS_SMP_RWMTX_LONG_LIVED; + erts_atomic64_init_nob(&bytes_in, 0); + erts_atomic64_init_nob(&bytes_out, 0); + common_element_size = ERTS_ALC_DATA_ALIGN_SIZE(sizeof(Port)); common_element_size += ERTS_ALC_DATA_ALIGN_SIZE(sizeof(ErtsPortTaskBusyPortQ)); common_element_size += 10; /* name */ @@ -2805,10 +3435,8 @@ void erts_init_io(int port_tab_size, port_tab_size, common_element_size, /* Doesn't need to be excact */ "port_table", - legacy_port_tab); - - erts_smp_atomic_init_nob(&erts_bytes_out, 0); - erts_smp_atomic_init_nob(&erts_bytes_in, 0); + legacy_port_tab, + 1); sys_init_io(); @@ -2816,10 +3444,11 @@ void erts_init_io(int port_tab_size, erts_smp_rwmtx_rwlock(&erts_driver_list_lock); init_driver(&fd_driver, &fd_driver_entry, NULL); -#ifndef __OSE__ init_driver(&vanilla_driver, &vanilla_driver_entry, NULL); -#endif init_driver(&spawn_driver, &spawn_driver_entry, NULL); +#ifndef __WIN32__ + init_driver(&forker_driver, &forker_driver_entry, NULL); +#endif erts_init_static_drivers(); for (dp = driver_tab; *dp != NULL; dp++) erts_add_driver_entry(*dp, NULL, 1); @@ -2829,7 +3458,6 @@ void erts_init_io(int port_tab_size, } #if defined(ERTS_ENABLE_LOCK_COUNT) && defined(ERTS_SMP) - static ERTS_INLINE void lcnt_enable_drv_lock_count(erts_driver_t *dp, int enable) { if (dp->lock) { @@ -2869,25 +3497,29 @@ static ERTS_INLINE void lcnt_enable_port_lock_count(Port *prt, int enable) } } -void erts_lcnt_enable_io_lock_count(int enable) -{ +void erts_lcnt_enable_io_lock_count(int enable) { erts_driver_t *dp; - int i, max = erts_ptab_max(&erts_port); + int ix, max = erts_ptab_max(&erts_port); + Port *prt; - for (i = 0; i < max; i++) { - Port *prt = erts_pix2port(i); - if (prt) + for (ix = 0; ix < max; ix++) { + if ((prt = erts_pix2port(ix)) != NULL) { lcnt_enable_port_lock_count(prt, enable); - } + } + } /* for all ports */ lcnt_enable_drv_lock_count(&vanilla_driver, enable); lcnt_enable_drv_lock_count(&spawn_driver, enable); +#ifndef __WIN32__ + lcnt_enable_drv_lock_count(&forker_driver, enable); +#endif lcnt_enable_drv_lock_count(&fd_driver, enable); - for (dp = driver_list; dp; dp = dp->next) + /* enable lock counting in all drivers */ + for (dp = driver_list; dp; dp = dp->next) { lcnt_enable_drv_lock_count(dp, enable); -} -#endif - + } +} /* enable/disable lock counting of ports */ +#endif /* defined(ERTS_ENABLE_LOCK_COUNT) && defined(ERTS_SMP) */ /* * Buffering of data when using line oriented I/O on ports */ @@ -3060,7 +3692,7 @@ static int read_linebuf(LineBufContext *bp) } static void -deliver_result(Eterm sender, Eterm pid, Eterm res) +deliver_result(Port *prt, Eterm sender, Eterm pid, Eterm res) { Process *rp; ErtsProcLocks rp_locks = 0; @@ -3068,33 +3700,40 @@ deliver_result(Eterm sender, Eterm pid, Eterm res) ERTS_SMP_CHK_NO_PROC_LOCKS; + ASSERT(!prt || prt->common.id == sender); +#ifdef ERTS_SMP + ASSERT(!prt || erts_lc_is_port_locked(prt)); +#endif + ASSERT(is_internal_port(sender) && is_internal_pid(pid)); rp = (scheduler ? erts_proc_lookup(pid) - : erts_pid2proc_opt(NULL, 0, pid, 0, ERTS_P2P_FLG_SMP_INC_REFC)); + : erts_pid2proc_opt(NULL, 0, pid, 0, ERTS_P2P_FLG_INC_REFC)); + + if (prt && IS_TRACED_FL(prt, F_TRACE_SEND)) { + Eterm hp[3]; + trace_port_send(prt, pid, TUPLE2(hp, sender, res), !!rp); + } if (rp) { Eterm tuple; - ErlHeapFragment *bp; + ErtsMessage *mp; ErlOffHeap *ohp; Eterm* hp; Uint sz_res; sz_res = size_object(res); - hp = erts_alloc_message_heap(sz_res + 3, &bp, &ohp, rp, &rp_locks); + mp = erts_alloc_message_heap(rp, &rp_locks, + sz_res + 3, &hp, &ohp); res = copy_struct(res, sz_res, &hp, ohp); tuple = TUPLE2(hp, sender, res); - erts_queue_message(rp, &rp_locks, bp, tuple, NIL -#ifdef USE_VM_PROBES - , NIL -#endif - ); + erts_queue_message(rp, rp_locks, mp, tuple, sender); if (rp_locks) erts_smp_proc_unlock(rp, rp_locks); if (!scheduler) - erts_smp_proc_dec_refc(rp); + erts_proc_dec_refc(rp); } } @@ -3117,10 +3756,11 @@ static void deliver_read_message(Port* prt, erts_aint32_t state, Eterm to, Eterm tuple; Process* rp; Eterm* hp; - ErlHeapFragment *bp; + ErtsMessage *mp; ErlOffHeap *ohp; ErtsProcLocks rp_locks = 0; int scheduler = erts_get_scheduler_id() != 0; + int trace_send = IS_TRACED_FL(prt, F_TRACE_SEND); ERTS_SMP_LC_ASSERT(erts_lc_is_port_locked(prt)); ERTS_SMP_CHK_NO_PROC_LOCKS; @@ -3138,12 +3778,12 @@ static void deliver_read_message(Port* prt, erts_aint32_t state, Eterm to, rp = (scheduler ? erts_proc_lookup(to) - : erts_pid2proc_opt(NULL, 0, to, 0, ERTS_P2P_FLG_SMP_INC_REFC)); + : erts_pid2proc_opt(NULL, 0, to, 0, ERTS_P2P_FLG_INC_REFC)); if (!rp) return; - hp = erts_alloc_message_heap(need, &bp, &ohp, rp, &rp_locks); + mp = erts_alloc_message_heap(trace_send ? NULL : rp, &rp_locks, need, &hp, &ohp); listp = NIL; if ((state & ERTS_PORT_SFLG_BINARY_IO) == 0) { @@ -3153,8 +3793,6 @@ static void deliver_read_message(Port* prt, erts_aint32_t state, Eterm to, Binary* bptr; bptr = erts_bin_nrml_alloc(len); - bptr->flags = 0; - bptr->orig_size = len; erts_refc_init(&bptr->refc, 1); sys_memcpy(bptr->orig_bytes, buf, len); @@ -3187,15 +3825,15 @@ static void deliver_read_message(Port* prt, erts_aint32_t state, Eterm to, tuple = TUPLE2(hp, prt->common.id, tuple); hp += 3; - erts_queue_message(rp, &rp_locks, bp, tuple, am_undefined -#ifdef USE_VM_PROBES - , NIL -#endif - ); + if (trace_send) + trace_port_send(prt, to, tuple, 1); + + ERL_MESSAGE_TOKEN(mp) = am_undefined; + erts_queue_message(rp, rp_locks, mp, tuple, prt->common.id); if (rp_locks) erts_smp_proc_unlock(rp, rp_locks); if (!scheduler) - erts_smp_proc_dec_refc(rp); + erts_proc_dec_refc(rp); } /* @@ -3265,11 +3903,12 @@ deliver_vec_message(Port* prt, /* Port */ Eterm tuple; Process* rp; Eterm* hp; - ErlHeapFragment *bp; + ErtsMessage *mp; ErlOffHeap *ohp; ErtsProcLocks rp_locks = 0; int scheduler = erts_get_scheduler_id() != 0; erts_aint32_t state; + int trace_send = IS_TRACED_FL(prt, F_TRACE_SEND); ERTS_SMP_LC_ASSERT(erts_lc_is_port_locked(prt)); ERTS_SMP_CHK_NO_PROC_LOCKS; @@ -3281,7 +3920,7 @@ deliver_vec_message(Port* prt, /* Port */ rp = (scheduler ? erts_proc_lookup(to) - : erts_pid2proc_opt(NULL, 0, to, 0, ERTS_P2P_FLG_SMP_INC_REFC)); + : erts_pid2proc_opt(NULL, 0, to, 0, ERTS_P2P_FLG_INC_REFC)); if (!rp) return; @@ -3297,7 +3936,7 @@ deliver_vec_message(Port* prt, /* Port */ need += (hlen+csize)*2; } - hp = erts_alloc_message_heap(need, &bp, &ohp, rp, &rp_locks); + mp = erts_alloc_message_heap(trace_send ? NULL : rp, &rp_locks, need, &hp, &ohp); listp = NIL; iov += vsize; @@ -3358,14 +3997,14 @@ deliver_vec_message(Port* prt, /* Port */ tuple = TUPLE2(hp, prt->common.id, tuple); hp += 3; - erts_queue_message(rp, &rp_locks, bp, tuple, am_undefined -#ifdef USE_VM_PROBES - , NIL -#endif - ); + if (IS_TRACED_FL(prt, F_TRACE_SEND)) + trace_port_send(prt, to, tuple, 1); + + ERL_MESSAGE_TOKEN(mp) = am_undefined; + erts_queue_message(rp, rp_locks, mp, tuple, prt->common.id); erts_smp_proc_unlock(rp, rp_locks); if (!scheduler) - erts_smp_proc_dec_refc(rp); + erts_proc_dec_refc(rp); } @@ -3402,18 +4041,32 @@ static void flush_port(Port *p) ERTS_SMP_LC_ASSERT(erts_lc_is_port_locked(p)); if (p->drv_ptr->flush != NULL) { + ERTS_MSACC_PUSH_STATE_M(); #ifdef USE_VM_PROBES if (DTRACE_ENABLED(driver_flush)) { DTRACE_FORMAT_COMMON_PID_AND_PORT(ERTS_PORT_GET_CONNECTED(p), p) DTRACE3(driver_flush, process_str, port_str, p->name); } #endif +#ifdef USE_LTTNG_VM_TRACEPOINTS + if (LTTNG_ENABLED(driver_flush)) { + lttng_decl_portbuf(port_str); + lttng_decl_procbuf(proc_str); + lttng_pid_to_str(ERTS_PORT_GET_CONNECTED(p), proc_str); + lttng_port_to_str(p, port_str); + LTTNG3(driver_flush, proc_str, port_str, p->name); + } +#endif + + if (IS_TRACED_FL(p, F_TRACE_SCHED_PORTS)) { trace_sched_ports_where(p, am_in, am_flush); } + ERTS_MSACC_SET_STATE_CACHED_M(ERTS_MSACC_STATE_PORT); fpe_was_unmasked = erts_block_fpe(); (*p->drv_ptr->flush)((ErlDrvData)p->drv_data); erts_unblock_fpe(fpe_was_unmasked); + ERTS_MSACC_POP_STATE_M(); if (IS_TRACED_FL(p, F_TRACE_SCHED_PORTS)) { trace_sched_ports_where(p, am_out, am_flush); } @@ -3437,6 +4090,7 @@ terminate_port(Port *prt) Eterm connected_id = NIL /* Initialize to silence compiler */; erts_driver_t *drv; erts_aint32_t state; + ErtsPrtSD *psd; ERTS_SMP_CHK_NO_PROC_LOCKS; ERTS_SMP_LC_ASSERT(erts_lc_is_port_locked(prt)); @@ -3455,29 +4109,43 @@ terminate_port(Port *prt) send_closed_port_id = NIL; } -#ifdef ERTS_SMP - erts_cancel_smp_ptimer(prt->common.u.alive.ptimer); -#else - erts_cancel_timer(&prt->common.u.alive.tm); -#endif + if (ERTS_PTMR_IS_SET(prt)) + erts_cancel_port_timer(prt); drv = prt->drv_ptr; if ((drv != NULL) && (drv->stop != NULL)) { int fpe_was_unmasked = erts_block_fpe(); + ERTS_MSACC_PUSH_AND_SET_STATE_M(ERTS_MSACC_STATE_PORT); #ifdef USE_VM_PROBES if (DTRACE_ENABLED(driver_stop)) { DTRACE_FORMAT_COMMON_PID_AND_PORT(connected_id, prt) DTRACE3(driver_stop, process_str, drv->name, port_str); } #endif +#ifdef USE_LTTNG_VM_TRACEPOINTS + if (LTTNG_ENABLED(driver_stop)) { + lttng_decl_portbuf(port_str); + lttng_decl_procbuf(proc_str); + lttng_pid_to_str(connected_id, proc_str); + lttng_port_to_str(prt, port_str); + LTTNG3(driver_stop, proc_str, port_str, drv->name); + } +#endif + (*drv->stop)((ErlDrvData)prt->drv_data); erts_unblock_fpe(fpe_was_unmasked); + ERTS_MSACC_POP_STATE_M(); #ifdef ERTS_SMP if (prt->xports) erts_port_handle_xports(prt); ASSERT(!prt->xports); #endif } + + if (is_internal_port(send_closed_port_id) + && IS_TRACED_FL(prt, F_TRACE_SEND)) + trace_port_send(prt, connected_id, am_closed, 1); + if(drv->handle != NULL) { erts_smp_rwmtx_rlock(&erts_driver_list_lock); erts_ddll_decrement_port_count(drv->handle); @@ -3491,8 +4159,9 @@ terminate_port(Port *prt) erts_cleanup_port_data(prt); - if (prt->psd) - erts_free(ERTS_ALC_T_PRTSD, prt->psd); + psd = (ErtsPrtSD *) erts_smp_atomic_read_nob(&prt->psd); + if (psd) + erts_free(ERTS_ALC_T_PRTSD, psd); ASSERT(prt->dist_entry == NULL); @@ -3505,10 +4174,10 @@ terminate_port(Port *prt) if ((state & ERTS_PORT_SFLG_HALT) && (erts_smp_atomic32_dec_read_nob(&erts_halt_progress) == 0)) { erts_port_release(prt); /* We will exit and never return */ - erl_exit_flush_async(erts_halt_code, ""); + erts_flush_async_exit(erts_halt_code, ""); } if (is_internal_port(send_closed_port_id)) - deliver_result(send_closed_port_id, connected_id, am_closed); + deliver_result(NULL, send_closed_port_id, connected_id, am_closed); } void @@ -3517,23 +4186,30 @@ erts_terminate_port(Port *pp) terminate_port(pp); } +static void port_fire_one_monitor(ErtsMonitor *mon, void *ctx0); static void sweep_one_monitor(ErtsMonitor *mon, void *vpsc) { - ErtsMonitor *rmon; - Process *rp; + switch (mon->type) { + case MON_ORIGIN: { + ErtsMonitor *rmon; + Process *rp; - ASSERT(mon->type == MON_ORIGIN); - ASSERT(is_internal_pid(mon->pid)); - rp = erts_pid2proc(NULL, 0, mon->pid, ERTS_PROC_LOCK_LINK); - if (!rp) { - goto done; - } - rmon = erts_remove_monitor(&ERTS_P_MONITORS(rp), mon->ref); - erts_smp_proc_unlock(rp, ERTS_PROC_LOCK_LINK); - if (rmon == NULL) { - goto done; + ASSERT(is_internal_pid(mon->pid)); + rp = erts_pid2proc(NULL, 0, mon->pid, ERTS_PROC_LOCK_LINK); + if (!rp) { + goto done; + } + rmon = erts_remove_monitor(&ERTS_P_MONITORS(rp), mon->ref); + erts_smp_proc_unlock(rp, ERTS_PROC_LOCK_LINK); + if (rmon == NULL) { + goto done; + } + erts_destroy_monitor(rmon); + } break; + case MON_TARGET: { + port_fire_one_monitor(mon, vpsc); /* forward call */ + } break; } - erts_destroy_monitor(rmon); done: erts_destroy_monitor(mon); } @@ -3541,7 +4217,7 @@ static void sweep_one_monitor(ErtsMonitor *mon, void *vpsc) typedef struct { - Eterm port; + Port *port; Eterm reason; } SweepContext; @@ -3550,10 +4226,13 @@ static void sweep_one_link(ErtsLink *lnk, void *vpsc) SweepContext *psc = vpsc; DistEntry *dep; Process *rp; - + Eterm port_id = psc->port->common.id; ASSERT(lnk->type == LINK_PID); - + + if (IS_TRACED_FL(psc->port, F_TRACE_PORTS)) + trace_port(psc->port, am_unlink, lnk->pid); + if (is_external_pid(lnk->pid)) { dep = external_pid_dist_entry(lnk->pid); if(dep != erts_this_dist_entry) { @@ -3566,9 +4245,9 @@ static void sweep_one_link(ErtsLink *lnk, void *vpsc) case ERTS_DSIG_PREP_NOT_CONNECTED: break; case ERTS_DSIG_PREP_CONNECTED: - erts_remove_dist_link(&dld, psc->port, lnk->pid, dep); + erts_remove_dist_link(&dld, port_id, lnk->pid, dep); erts_destroy_dist_link(&dld); - code = erts_dsig_send_exit(&dsd, psc->port, lnk->pid, + code = erts_dsig_send_exit(&dsd, port_id, lnk->pid, psc->reason); ASSERT(code == ERTS_DSIG_SEND_OK); break; @@ -3582,23 +4261,25 @@ static void sweep_one_link(ErtsLink *lnk, void *vpsc) ASSERT(is_internal_pid(lnk->pid)); rp = erts_pid2proc(NULL, 0, lnk->pid, rp_locks); if (rp) { - ErtsLink *rlnk = erts_remove_link(&ERTS_P_LINKS(rp), psc->port); + ErtsLink *rlnk = erts_remove_link(&ERTS_P_LINKS(rp), port_id); if (rlnk) { int xres = erts_send_exit_signal(NULL, - psc->port, + port_id, rp, &rp_locks, psc->reason, NIL, NULL, 0); - if (xres >= 0 && IS_TRACED_FL(rp, F_TRACE_PROCS)) { + if (xres >= 0) { + if (rp_locks & ERTS_PROC_LOCKS_XSIG_SEND) { + erts_smp_proc_unlock(rp, ERTS_PROC_LOCKS_XSIG_SEND); + rp_locks &= ~ERTS_PROC_LOCKS_XSIG_SEND; + } /* We didn't exit the process and it is traced */ - if (IS_TRACED_FL(rp, F_TRACE_PROCS)) { - trace_proc(NULL, rp, am_getting_unlinked, - psc->port); - } + if (IS_TRACED_FL(rp, F_TRACE_PROCS)) + trace_proc(NULL, 0, rp, am_getting_unlinked, port_id); } erts_destroy_link(rlnk); } @@ -3609,6 +4290,43 @@ static void sweep_one_link(ErtsLink *lnk, void *vpsc) erts_destroy_link(lnk); } +static void +port_fire_one_monitor(ErtsMonitor *mon, void *ctx0) +{ + Process *origin; + ErtsProcLocks origin_locks; + + if (mon->type != MON_TARGET || ! is_pid(mon->pid)) { + return; + } + /* + * Proceed here if someone monitors us, we (port) are the target and + * origin is some process + */ + origin_locks = ERTS_PROC_LOCKS_MSG_SEND | ERTS_PROC_LOCK_LINK; + + origin = erts_pid2proc(NULL, 0, mon->pid, origin_locks); + if (origin) { + DeclareTmpHeapNoproc(lhp,3); + SweepContext *ctx = (SweepContext *)ctx0; + ErtsMonitor *rmon; + Eterm watched = (is_atom(mon->name) + ? TUPLE2(lhp, mon->name, erts_this_dist_entry->sysname) + : ctx->port->common.id); + + erts_queue_monitor_message(origin, &origin_locks, mon->ref, am_port, + watched, ctx->reason); + UnUseTmpHeapNoproc(3); + + rmon = erts_remove_monitor(&ERTS_P_MONITORS(origin), mon->ref); + erts_smp_proc_unlock(origin, origin_locks); + + if (rmon) { + erts_destroy_monitor(rmon); + } + } +} + /* 'from' is sending 'this_port' an exit signal, (this_port must be internal). * If reason is normal we don't do anything, *unless* from is our connected * process in which case we close the port. Any other reason kills the port. @@ -3620,91 +4338,95 @@ static void sweep_one_link(ErtsLink *lnk, void *vpsc) */ int -erts_deliver_port_exit(Port *p, Eterm from, Eterm reason, int send_closed) +erts_deliver_port_exit(Port *prt, Eterm from, Eterm reason, int send_closed, + int drop_normal) { ErtsLink *lnk; - Eterm rreason; + Eterm modified_reason; erts_aint32_t state, set_state_flags; ERTS_SMP_CHK_NO_PROC_LOCKS; - ERTS_SMP_LC_ASSERT(erts_lc_is_port_locked(p)); + ERTS_SMP_LC_ASSERT(erts_lc_is_port_locked(prt)); - rreason = (reason == am_kill) ? am_killed : reason; + modified_reason = (reason == am_kill) ? am_killed : reason; #ifdef USE_VM_PROBES if (DTRACE_ENABLED(port_exit)) { DTRACE_CHARBUF(from_str, DTRACE_TERM_BUF_SIZE); DTRACE_CHARBUF(port_str, DTRACE_TERM_BUF_SIZE); - DTRACE_CHARBUF(rreason_str, 64); + DTRACE_CHARBUF(reason_str, 64); erts_snprintf(from_str, sizeof(DTRACE_CHARBUF_NAME(from_str)), "%T", from); - dtrace_port_str(p, port_str); - erts_snprintf(rreason_str, sizeof(DTRACE_CHARBUF_NAME(rreason_str)), "%T", rreason); - DTRACE4(port_exit, from_str, port_str, p->name, rreason_str); + dtrace_port_str(prt, port_str); + erts_snprintf(reason_str, sizeof(DTRACE_CHARBUF_NAME(reason_str)), "%T", + modified_reason); + DTRACE4(port_exit, from_str, port_str, prt->name, reason_str); } #endif - state = erts_atomic32_read_nob(&p->state); + state = erts_atomic32_read_nob(&prt->state); if (state & (ERTS_PORT_SFLGS_DEAD | ERTS_PORT_SFLG_EXITING | ERTS_PORT_SFLG_CLOSING)) return 0; - if (reason == am_normal && from != ERTS_PORT_GET_CONNECTED(p) && from != p->common.id) + if (reason == am_normal && from != ERTS_PORT_GET_CONNECTED(prt) + && from != prt->common.id && drop_normal) { return 0; + } set_state_flags = ERTS_PORT_SFLG_EXITING; if (send_closed) set_state_flags |= ERTS_PORT_SFLG_SEND_CLOSED; - erts_port_task_sched_enter_exiting_state(&p->sched); + erts_port_task_sched_enter_exiting_state(&prt->sched); - state = erts_atomic32_read_bor_mb(&p->state, set_state_flags); + state = erts_atomic32_read_bor_mb(&prt->state, set_state_flags); state |= set_state_flags; - if (IS_TRACED_FL(p, F_TRACE_PORTS)) { - trace_port(p, am_closed, reason); - } + if (IS_TRACED_FL(prt, F_TRACE_PORTS)) + trace_port(prt, am_closed, reason); - erts_trace_check_exiting(p->common.id); + erts_trace_check_exiting(prt->common.id); - set_busy_port(ERTS_Port2ErlDrvPort(p), 0); + set_busy_port(ERTS_Port2ErlDrvPort(prt), 0); - if (p->common.u.alive.reg != NULL) - (void) erts_unregister_name(NULL, 0, p, p->common.u.alive.reg->name); + if (prt->common.u.alive.reg != NULL) + (void) erts_unregister_name(NULL, 0, prt, prt->common.u.alive.reg->name); { - SweepContext sc = {p->common.id, rreason}; - lnk = ERTS_P_LINKS(p); - ERTS_P_LINKS(p) = NULL; + SweepContext sc = {prt, modified_reason}; + lnk = ERTS_P_LINKS(prt); + ERTS_P_LINKS(prt) = NULL; erts_sweep_links(lnk, &sweep_one_link, &sc); } - DRV_MONITOR_LOCK_PDL(p); + DRV_MONITOR_LOCK_PDL(prt); { - ErtsMonitor *moni = ERTS_P_MONITORS(p); - ERTS_P_MONITORS(p) = NULL; - erts_sweep_monitors(moni, &sweep_one_monitor, NULL); + SweepContext ctx = {prt, modified_reason}; + ErtsMonitor *moni = ERTS_P_MONITORS(prt); + ERTS_P_MONITORS(prt) = NULL; + erts_sweep_monitors(moni, &sweep_one_monitor, &ctx); } - DRV_MONITOR_UNLOCK_PDL(p); + DRV_MONITOR_UNLOCK_PDL(prt); - if ((state & ERTS_PORT_SFLG_DISTRIBUTION) && p->dist_entry) { - erts_do_net_exits(p->dist_entry, rreason); - erts_deref_dist_entry(p->dist_entry); - p->dist_entry = NULL; - erts_atomic32_read_band_relb(&p->state, + if ((state & ERTS_PORT_SFLG_DISTRIBUTION) && prt->dist_entry) { + erts_do_net_exits(prt->dist_entry, modified_reason); + erts_deref_dist_entry(prt->dist_entry); + prt->dist_entry = NULL; + erts_atomic32_read_band_relb(&prt->state, ~ERTS_PORT_SFLG_DISTRIBUTION); } - if ((reason != am_kill) && !is_port_ioq_empty(p)) { + if ((reason != am_kill) && !is_port_ioq_empty(prt)) { /* must turn exiting flag off */ - erts_atomic32_read_bset_relb(&p->state, + erts_atomic32_read_bset_relb(&prt->state, (ERTS_PORT_SFLG_EXITING | ERTS_PORT_SFLG_CLOSING), ERTS_PORT_SFLG_CLOSING); - flush_port(p); + flush_port(prt); } else { - terminate_port(p); + terminate_port(prt); } return 1; @@ -3782,6 +4504,7 @@ call_driver_control(Eterm caller, ErlDrvSizeT *from_size) { ErlDrvSSizeT cres; + ERTS_MSACC_PUSH_STATE_M(); if (!prt->drv_ptr->control) return ERTS_PORT_OP_BADARG; @@ -3796,6 +4519,21 @@ call_driver_control(Eterm caller, } #endif + if (IS_TRACED_FL(prt, F_TRACE_RECEIVE)) + trace_port_receive(prt, caller, am_control, command, bufp, size); + + ERTS_MSACC_SET_STATE_CACHED_M(ERTS_MSACC_STATE_PORT); + +#ifdef USE_LTTNG_VM_TRACEPOINTS + if (LTTNG_ENABLED(driver_control)) { + lttng_decl_procbuf(proc_str); + lttng_decl_portbuf(port_str); + lttng_pid_to_str(caller, proc_str); + lttng_port_to_str(prt, port_str); + LTTNG5(driver_control, proc_str, port_str, prt->name, command, size); + } +#endif + prt->caller = caller; cres = prt->drv_ptr->control((ErlDrvData) prt->drv_data, command, @@ -3805,9 +4543,14 @@ call_driver_control(Eterm caller, *from_size); prt->caller = NIL; + ERTS_MSACC_POP_STATE_M(); + if (cres < 0) return ERTS_PORT_OP_BADARG; + if (IS_TRACED_FL(prt, F_TRACE_SEND)) + trace_port_send_binary(prt, caller, am_control, *resp_bufp, cres); + *from_size = (ErlDrvSizeT) cres; return ERTS_PORT_OP_DONE; @@ -3856,7 +4599,6 @@ write_port_control_result(int control_flags, ErlDrvSizeT resp_size, char *pre_alloc_buf, Eterm **hpp, - ErlHeapFragment *bp, ErlOffHeap *ohp) { Eterm res; @@ -3930,48 +4672,40 @@ port_sig_control(Port *prt, if (res == ERTS_PORT_OP_DONE) { Eterm msg; - Eterm *hp, *hp_start; - ErlHeapFragment *bp; - ErlOffHeap *ohp; + ErtsHeapFactory factory; Process *rp; ErtsProcLocks rp_locks = 0; - Uint hsz; + Uint hsz, rsz; int control_flags; - rp = erts_proc_lookup_raw(sigdp->caller); + rp = sigdp->caller == ERTS_INVALID_PID ? NULL : erts_proc_lookup_raw(sigdp->caller); if (!rp) goto done; control_flags = prt->control_flags; - hsz = ERTS_QUEUE_PORT_SCHED_OP_REPLY_SIZE; - hsz += port_control_result_size(control_flags, + rsz = port_control_result_size(control_flags, resp_bufp, &resp_size, &resp_buf[0]); + hsz = rsz + ERTS_QUEUE_PORT_SCHED_OP_REPLY_SIZE; - hp_start = hp = erts_alloc_message_heap(hsz, - &bp, - &ohp, - rp, - &rp_locks); + + (void) erts_factory_message_create(&factory, rp, + &rp_locks, hsz); msg = write_port_control_result(control_flags, resp_bufp, resp_size, &resp_buf[0], - &hp, - bp, - ohp); - + &factory.hp, + factory.off_heap); queue_port_sched_op_reply(rp, - &rp_locks, - hp_start, - hp, - hsz, - bp, + rp_locks, + &factory, sigdp->ref, - msg); + msg, + prt); if (rp_locks) erts_smp_proc_unlock(rp, rp_locks); @@ -3981,7 +4715,8 @@ port_sig_control(Port *prt, /* failure */ - port_sched_op_reply(sigdp->caller, sigdp->ref, am_badarg); + if (sigdp->caller != ERTS_INVALID_PID) + port_sched_op_reply(sigdp->caller, sigdp->ref, am_badarg, prt); done: @@ -3991,6 +4726,23 @@ done: return ERTS_PORT_REDS_CONTROL; } +/* + * This is an asynchronous control call. I.e. it will not return anything + * to the caller. + */ +int +erl_drv_port_control(Eterm port_num, char cmd, char* buff, ErlDrvSizeT size) +{ + ErtsProc2PortSigData *sigdp = erts_port_task_alloc_p2p_sig_data(); + + sigdp->flags = ERTS_P2P_SIG_TYPE_CONTROL | ERTS_P2P_SIG_DATA_FLG_REPLY; + sigdp->u.control.binp = NULL; + sigdp->u.control.command = cmd; + sigdp->u.control.bufp = buff; + sigdp->u.control.size = size; + + return erts_schedule_port2port_signal(port_num, sigdp, 0, port_sig_control); +} ErtsPortOpResult erts_port_control(Process* c_p, @@ -4107,7 +4859,6 @@ erts_port_control(Process* c_p, resp_size, &resp_buf[0], &hp, - NULL, &c_p->off_heap); BUMP_REDS(c_p, ERTS_PORT_REDS_CONTROL); return ERTS_PORT_OP_DONE; @@ -4129,10 +4880,10 @@ erts_port_control(Process* c_p, binp = NULL; if (is_binary(data) && binary_bitoffset(data) == 0) { - Eterm *ebinp = binary_val_rel(data, NULL); + Eterm *ebinp = binary_val(data); ASSERT(!tmp_alloced); if (*ebinp == HEADER_SUB_BIN) - ebinp = binary_val_rel(((ErlSubBin *) ebinp)->orig, NULL); + ebinp = binary_val(((ErlSubBin *) ebinp)->orig); if (*ebinp != HEADER_PROC_BIN) copy = 1; else { @@ -4185,6 +4936,7 @@ call_driver_call(Eterm caller, unsigned *ret_flagsp) { ErlDrvSSizeT cres; + ERTS_MSACC_PUSH_STATE_M(); if (!prt->drv_ptr->call) return ERTS_PORT_OP_BADARG; @@ -4199,6 +4951,20 @@ call_driver_call(Eterm caller, DTRACE5(driver_call, process_str, port_str, prt->name, command, size); } #endif +#ifdef USE_LTTNG_VM_TRACEPOINTS + if (LTTNG_ENABLED(driver_call)) { + lttng_decl_procbuf(proc_str); + lttng_decl_portbuf(port_str); + lttng_pid_to_str(caller,proc_str); + lttng_port_to_str(prt, port_str); + LTTNG5(driver_call, proc_str, port_str, prt->name, command, size); + } +#endif + + if (IS_TRACED_FL(prt, F_TRACE_RECEIVE)) + trace_port_receive(prt, caller, am_call, command, bufp, size); + + ERTS_MSACC_SET_STATE_CACHED_M(ERTS_MSACC_STATE_PORT); prt->caller = caller; cres = prt->drv_ptr->call((ErlDrvData) prt->drv_data, @@ -4210,10 +4976,15 @@ call_driver_call(Eterm caller, ret_flagsp); prt->caller = NIL; + ERTS_MSACC_POP_STATE_M(); + if (cres <= 0 || ((byte) (*resp_bufp)[0]) != VERSION_MAGIC) return ERTS_PORT_OP_BADARG; + if (IS_TRACED_FL(prt, F_TRACE_SEND)) + trace_port_send_binary(prt, caller, am_call, *resp_bufp, cres); + *from_size = (ErlDrvSizeT) cres; return ERTS_PORT_OP_DONE; @@ -4256,8 +5027,6 @@ port_sig_call(Port *prt, if (res == ERTS_PORT_OP_DONE) { Eterm msg; Eterm *hp; - ErlHeapFragment *bp; - ErlOffHeap *ohp; Process *rp; ErtsProcLocks rp_locks = 0; Sint hsz; @@ -4268,45 +5037,39 @@ port_sig_call(Port *prt, hsz = erts_decode_ext_size((byte *) resp_bufp, resp_size); if (hsz >= 0) { - Eterm *hp_start; + ErtsHeapFactory factory; byte *endp; hsz += 3; /* ok tuple */ hsz += ERTS_QUEUE_PORT_SCHED_OP_REPLY_SIZE; - hp_start = hp = erts_alloc_message_heap(hsz, - &bp, - &ohp, - rp, - &rp_locks); + (void) erts_factory_message_create(&factory, rp, &rp_locks, hsz); endp = (byte *) resp_bufp; - msg = erts_decode_ext(&hp, ohp, &endp); + msg = erts_decode_ext(&factory, &endp, 0); if (is_value(msg)) { + hp = erts_produce_heap(&factory, + 3, + ERTS_QUEUE_PORT_SCHED_OP_REPLY_SIZE); msg = TUPLE2(hp, am_ok, msg); - hp += 3; queue_port_sched_op_reply(rp, - &rp_locks, - hp_start, - hp, - hsz, - bp, + rp_locks, + &factory, sigdp->ref, - msg); + msg, + prt); if (rp_locks) erts_smp_proc_unlock(rp, rp_locks); goto done; } - if (bp) - free_message_buffer(bp); if (rp_locks) erts_smp_proc_unlock(rp, rp_locks); } } } - port_sched_op_reply(sigdp->caller, sigdp->ref, am_badarg); + port_sched_op_reply(sigdp->caller, sigdp->ref, am_badarg, prt); done: @@ -4376,10 +5139,11 @@ erts_port_call(Process* c_p, try_call_res = try_imm_drv_call(&try_call_state); switch (try_call_res) { case ERTS_TRY_IMM_DRV_CALL_OK: { - Eterm *hp, *hp_end; + ErtsHeapFactory factory; Sint hsz; unsigned ret_flags = 0U; Eterm term; + Eterm* hp; res = call_driver_call(c_p->common.id, prt, @@ -4399,15 +5163,14 @@ erts_port_call(Process* c_p, if (hsz < 0) return ERTS_PORT_OP_BADARG; hsz += 3; - hp = HAlloc(c_p, hsz); - hp_end = hp + hsz; + erts_factory_proc_prealloc_init(&factory, c_p, hsz); endp = (byte *) resp_bufp; - term = erts_decode_ext(&hp, &MSO(c_p), &endp); + term = erts_decode_ext(&factory, &endp, 0); if (term == THE_NON_VALUE) return ERTS_PORT_OP_BADARG; + hp = erts_produce_heap(&factory,3,0); *retvalp = TUPLE2(hp, am_ok, term); - hp += 3; - HRelease(c_p, hp_end, hp); + erts_factory_close(&factory); if (resp_bufp != &resp_buf[0] && !(ret_flags & DRIVER_CALL_KEEP_BUFFER)) driver_free(resp_bufp); @@ -4521,7 +5284,7 @@ port_sig_info(Port *prt, { ASSERT(sigdp->flags & ERTS_P2P_SIG_DATA_FLG_REPLY); if (op != ERTS_PROC2PORT_SIG_EXEC) - port_sched_op_reply(sigdp->caller, sigdp->ref, am_undefined); + port_sched_op_reply(sigdp->caller, sigdp->ref, am_undefined, prt); else { Eterm *hp, *hp_start; Uint hsz; @@ -4542,14 +5305,16 @@ port_sig_info(Port *prt, prt, sigdp->u.info.item); if (is_value(value)) { + ErtsHeapFactory factory; + ErtsMessage *mp = erts_alloc_message(0, NULL); + mp->data.heap_frag = bp; + erts_factory_selfcontained_message_init(&factory, mp, hp); queue_port_sched_op_reply(rp, - &rp_locks, - hp_start, - hp, - hsz, - bp, + rp_locks, + &factory, sigdp->ref, - value); + value, + prt); } if (rp_locks) erts_smp_proc_unlock(rp, rp_locks); @@ -4618,6 +5383,103 @@ erts_port_info(Process* c_p, } typedef struct { + Uint sched_id; + Eterm pid; + Uint32 refn[ERTS_REF_NUMBERS]; + erts_smp_atomic32_t refc; +} ErtsIOBytesReq; + +static void +reply_io_bytes(void *vreq) +{ + ErtsIOBytesReq *req = (ErtsIOBytesReq *) vreq; + Process *rp; + + rp = erts_proc_lookup(req->pid); + if (rp) { + ErlOffHeap *ohp; + ErtsMessage *mp; + ErtsProcLocks rp_locks; + Eterm ref, msg, ein, eout, *hp; + Uint64 in, out; + Uint hsz; + ErtsSchedulerData *esdp = erts_get_scheduler_data(); + Uint sched_id = esdp->no; + in = esdp->io.in; + out = esdp->io.out; + if (req->sched_id != sched_id) + rp_locks = 0; + else { + in += (Uint64) erts_atomic64_read_nob(&bytes_in); + out += (Uint64) erts_atomic64_read_nob(&bytes_out); + rp_locks = ERTS_PROC_LOCK_MAIN; + } + + hsz = 5 /* 4-tuple */ + REF_THING_SIZE; + + erts_bld_uint64(NULL, &hsz, in); + erts_bld_uint64(NULL, &hsz, out); + + mp = erts_alloc_message_heap(rp, &rp_locks, hsz, &hp, &ohp); + + ref = make_internal_ref(hp); + write_ref_thing(hp, req->refn[0], req->refn[1], req->refn[2]); + hp += REF_THING_SIZE; + + ein = erts_bld_uint64(&hp, NULL, in); + eout = erts_bld_uint64(&hp, NULL, out); + + msg = TUPLE4(hp, ref, make_small(sched_id), ein, eout); + + erts_queue_message(rp, rp_locks, mp, msg, am_system); + + if (req->sched_id == sched_id) + rp_locks &= ~ERTS_PROC_LOCK_MAIN; + if (rp_locks) + erts_smp_proc_unlock(rp, rp_locks); + } + + if (erts_smp_atomic32_dec_read_nob(&req->refc) == 0) + erts_free(ERTS_ALC_T_IOB_REQ, req); +} + +Eterm +erts_request_io_bytes(Process *c_p) +{ + Uint *hp; + Eterm ref; + Uint32 *refn; + ErtsSchedulerData *esdp = erts_proc_sched_data(c_p); + ErtsIOBytesReq *req = erts_alloc(ERTS_ALC_T_IOB_REQ, + sizeof(ErtsIOBytesReq)); + + hp = HAlloc(c_p, REF_THING_SIZE); + ref = erts_sched_make_ref_in_buffer(esdp, hp); + refn = internal_ref_numbers(ref); + + req->sched_id = esdp->no; + req->pid = c_p->common.id; + req->refn[0] = refn[0]; + req->refn[1] = refn[1]; + req->refn[2] = refn[2]; + erts_smp_atomic32_init_nob(&req->refc, + (erts_aint32_t) erts_no_schedulers); + +#ifdef ERTS_SMP + if (erts_no_schedulers > 1) + erts_schedule_multi_misc_aux_work(1, + erts_no_schedulers, + reply_io_bytes, + (void *) req); +#endif + + reply_io_bytes((void *) req); + + return ref; +} + + +typedef struct { int to; void *arg; } prt_one_lnk_data; @@ -4675,6 +5537,10 @@ print_port_info(Port *p, int to, void *arg) erts_print(to, arg, "Port is a file: %s\n",p->name); } else if (p->drv_ptr == &spawn_driver) { erts_print(to, arg, "Port controls external process: %s\n",p->name); +#ifndef __WIN32__ + } else if (p->drv_ptr == &forker_driver) { + erts_print(to, arg, "Port controls forker process: %s\n",p->name); +#endif } else { erts_print(to, arg, "Port controls linked-in driver: %s\n",p->name); } @@ -4821,12 +5687,13 @@ int get_port_flags(ErlDrvPort ix) void erts_raw_port_command(Port* p, byte* buf, Uint len) { int fpe_was_unmasked; + ERTS_MSACC_PUSH_STATE_M(); ERTS_SMP_CHK_NO_PROC_LOCKS; ERTS_SMP_LC_ASSERT(erts_lc_is_port_locked(p)); if (len > (Uint) INT_MAX) - erl_exit(ERTS_ABORT_EXIT, + erts_exit(ERTS_ABORT_EXIT, "Absurdly large data buffer (%beu bytes) passed to" "output callback of %s driver.\n", len, @@ -4841,9 +5708,11 @@ void erts_raw_port_command(Port* p, byte* buf, Uint len) DTRACE4(driver_output, "-raw-", port_str, p->name, len); } #endif + ERTS_MSACC_SET_STATE_CACHED_M(ERTS_MSACC_STATE_PORT); fpe_was_unmasked = erts_block_fpe(); (*p->drv_ptr->output)((ErlDrvData)p->drv_data, (char*) buf, (int) len); erts_unblock_fpe(fpe_was_unmasked); + ERTS_MSACC_POP_STATE_M(); } int async_ready(Port *p, void* data) @@ -4855,14 +5724,25 @@ int async_ready(Port *p, void* data) if (p) { ERTS_SMP_LC_ASSERT(erts_lc_is_port_locked(p)); if (p->drv_ptr->ready_async != NULL) { + ERTS_MSACC_PUSH_AND_SET_STATE_M(ERTS_MSACC_STATE_PORT); #ifdef USE_VM_PROBES if (DTRACE_ENABLED(driver_ready_async)) { DTRACE_FORMAT_COMMON_PID_AND_PORT(ERTS_PORT_GET_CONNECTED(p), p) DTRACE3(driver_ready_async, process_str, port_str, p->name); } #endif +#ifdef USE_LTTNG_VM_TRACEPOINTS + if (LTTNG_ENABLED(driver_ready_async)) { + lttng_decl_portbuf(port_str); + lttng_decl_procbuf(proc_str); + lttng_pid_to_str(ERTS_PORT_GET_CONNECTED(p), proc_str); + lttng_port_to_str(p, port_str); + LTTNG3(driver_ready_async, proc_str, port_str, p->name); + } +#endif (*p->drv_ptr->ready_async)((ErlDrvData)p->drv_data, data); need_free = 0; + ERTS_MSACC_POP_STATE_M(); } erts_port_driver_callback_epilogue(p, NULL); @@ -5006,24 +5886,6 @@ erts_free_port_names(ErtsPortNames *pnp) erts_free(ERTS_ALC_T_PORT_NAMES, pnp); } -static void schedule_port_timeout(Port *p) -{ - /* - * Scheduling of port timeouts can be done without port locking, but - * since the task handle is stored in the port structure and the ptimer - * structure is protected by the port lock we require the port to be - * locked for now... - * - * TODO: Implement scheduling of port timeouts without locking - * the port. - * /Rickard - */ - ERTS_SMP_LC_ASSERT(erts_lc_is_port_locked(p)); - erts_port_task_schedule(p->common.id, - &p->timeout_task, - ERTS_PORT_TASK_TIMEOUT); -} - ErlDrvTermData driver_mk_term_nil(void) { return driver_term_nil; @@ -5032,14 +5894,15 @@ ErlDrvTermData driver_mk_term_nil(void) void driver_report_exit(ErlDrvPort ix, int status) { Eterm* hp; + ErlOffHeap *ohp; Eterm tuple; Process *rp; Eterm pid; - ErlHeapFragment *bp = NULL; - ErlOffHeap *ohp; + ErtsMessage *mp; ErtsProcLocks rp_locks = 0; int scheduler = erts_get_scheduler_id() != 0; Port* prt = erts_drvport2port(ix); + int trace_send = IS_TRACED_FL(prt, F_TRACE_SEND); if (prt == ERTS_INVALID_ERL_DRV_PORT) return; @@ -5052,25 +5915,25 @@ void driver_report_exit(ErlDrvPort ix, int status) rp = (scheduler ? erts_proc_lookup(pid) - : erts_pid2proc_opt(NULL, 0, pid, 0, ERTS_P2P_FLG_SMP_INC_REFC)); + : erts_pid2proc_opt(NULL, 0, pid, 0, ERTS_P2P_FLG_INC_REFC)); if (!rp) return; - hp = erts_alloc_message_heap(3+3, &bp, &ohp, rp, &rp_locks); + mp = erts_alloc_message_heap(trace_send ? NULL : rp, &rp_locks, 3+3, &hp, &ohp); tuple = TUPLE2(hp, am_exit_status, make_small(status)); hp += 3; tuple = TUPLE2(hp, prt->common.id, tuple); - erts_queue_message(rp, &rp_locks, bp, tuple, am_undefined -#ifdef USE_VM_PROBES - , NIL -#endif - ); + if (IS_TRACED_FL(prt, F_TRACE_SEND)) + trace_port_send(prt, pid, tuple, 1); + + ERL_MESSAGE_TOKEN(mp) = am_undefined; + erts_queue_message(rp, rp_locks, mp, tuple, prt->common.id); erts_smp_proc_unlock(rp, rp_locks); if (!scheduler) - erts_smp_proc_dec_refc(rp); + erts_proc_dec_refc(rp); } #define ERTS_B2T_STATES_DEF_STATES_SZ 5 @@ -5160,24 +6023,28 @@ cleanup_b2t_states(struct b2t_states__ *b2tsp) */ static int -driver_deliver_term(Eterm to, ErlDrvTermData* data, int len) +driver_deliver_term(Port *prt, Eterm to, ErlDrvTermData* data, int len) { +#define HEAP_EXTRA 200 #define ERTS_DDT_FAIL do { res = -1; goto done; } while (0) Uint need = 0; int depth = 0; - int res; - Eterm *hp = NULL, *hp_start = NULL, *hp_end = NULL; + int res = 0; ErlDrvTermData* ptr; ErlDrvTermData* ptr_end; DECLARE_ESTACK(stack); - Eterm mess = NIL; /* keeps compiler happy */ + Eterm mess; Process* rp = NULL; - ErlHeapFragment *bp = NULL; - ErlOffHeap *ohp; + ErtsHeapFactory factory; ErtsProcLocks rp_locks = 0; struct b2t_states__ b2t; - int scheduler = 1; /* Silence erroneous warning... */ + int scheduler; + ErtsSchedulerData *esdp = erts_get_scheduler_data(); + + ERTS_UNDEF(mess,NIL); + ERTS_UNDEF(scheduler,1); + factory.mode = FACTORY_CLOSED; init_b2t_states(&b2t); /* @@ -5210,25 +6077,17 @@ driver_deliver_term(Eterm to, ErlDrvTermData* data, int len) break; case ERL_DRV_INT: /* signed int argument */ ERTS_DDT_CHK_ENOUGH_ARGS(1); -#if HALFWORD_HEAP - erts_bld_sint64(NULL, &need, (Sint64)ptr[0]); -#else /* check for bignum */ if (!IS_SSMALL((Sint)ptr[0])) need += BIG_UINT_HEAP_SIZE; /* use small_to_big */ -#endif ptr++; depth++; break; case ERL_DRV_UINT: /* unsigned int argument */ ERTS_DDT_CHK_ENOUGH_ARGS(1); -#if HALFWORD_HEAP - erts_bld_uint64(NULL, &need, (Uint64)ptr[0]); -#else /* check for bignum */ if (!IS_USMALL(0, (Uint)ptr[0])) need += BIG_UINT_HEAP_SIZE; /* use small_to_big */ -#endif ptr++; depth++; break; @@ -5341,9 +6200,10 @@ driver_deliver_term(Eterm to, ErlDrvTermData* data, int len) #ifdef DEBUG b2t.org_ext[b2t.ix] = ext; #endif - hsz = erts_binary2term_prepare(&b2t.state[b2t.ix++], ext, size); + hsz = erts_binary2term_prepare(&b2t.state[b2t.ix], ext, size); if (hsz < 0) ERTS_DDT_FAIL; /* Invalid data */ + b2t.state[b2t.ix++].heap_size = hsz; need += hsz; ptr += 2; depth++; @@ -5352,7 +6212,11 @@ driver_deliver_term(Eterm to, ErlDrvTermData* data, int len) case ERL_DRV_MAP: { /* int */ ERTS_DDT_CHK_ENOUGH_ARGS(1); if ((int) ptr[0] < 0) ERTS_DDT_FAIL; - need += MAP_HEADER_SIZE + 1 + 2*ptr[0]; + if (ptr[0] > MAP_SMALL_MAP_LIMIT) { + need += HASHMAP_ESTIMATED_HEAP_SIZE(ptr[0]); + } else { + need += MAP_HEADER_FLATMAP_SZ + 1 + 2*ptr[0]; + } depth -= 2*ptr[0]; if (depth < 0) ERTS_DDT_FAIL; ptr++; @@ -5382,14 +6246,28 @@ driver_deliver_term(Eterm to, ErlDrvTermData* data, int len) scheduler = erts_get_scheduler_id() != 0; rp = (scheduler ? erts_proc_lookup(to) - : erts_pid2proc_opt(NULL, 0, to, 0, ERTS_P2P_FLG_SMP_INC_REFC)); + : erts_pid2proc_opt(NULL, 0, to, 0, ERTS_P2P_FLG_INC_REFC)); if (!rp) { - res = 0; - goto done; - } + if (!prt || !IS_TRACED_FL(prt, F_TRACE_SEND)) + goto done; + if (!erts_is_tracer_proc_enabled_send(NULL, 0, &prt->common)) + goto done; - hp_start = hp = erts_alloc_message_heap(need, &bp, &ohp, rp, &rp_locks); - hp_end = hp + need; + res = -2; + + /* We allocate a temporary heap to be used to create + the message that may be sent using tracing */ + erts_factory_tmp_init(&factory, erts_alloc(ERTS_ALC_T_DRIVER, need*sizeof(Eterm)), + need, ERTS_ALC_T_DRIVER); + + } else { + /* We force the creation of a heap fragment (rp == NULL) when send + tracing so that we don't have the main lock of the process while + tracing */ + Process *trace_rp = prt && IS_TRACED_FL(prt, F_TRACE_SEND) ? NULL : rp; + (void) erts_factory_message_create(&factory, trace_rp, &rp_locks, need); + res = 1; + } /* * Interpret the instructions and build the term. @@ -5409,40 +6287,36 @@ driver_deliver_term(Eterm to, ErlDrvTermData* data, int len) break; case ERL_DRV_INT: /* signed int argument */ -#if HALFWORD_HEAP - mess = erts_bld_sint64(&hp, NULL, (Sint64)ptr[0]); -#else + erts_reserve_heap(&factory, BIG_UINT_HEAP_SIZE); if (IS_SSMALL((Sint)ptr[0])) mess = make_small((Sint)ptr[0]); else { - mess = small_to_big((Sint)ptr[0], hp); - hp += BIG_UINT_HEAP_SIZE; + mess = small_to_big((Sint)ptr[0], factory.hp); + factory.hp += BIG_UINT_HEAP_SIZE; } -#endif ptr++; break; case ERL_DRV_UINT: /* unsigned int argument */ -#if HALFWORD_HEAP - mess = erts_bld_uint64(&hp, NULL, (Uint64)ptr[0]); -#else + erts_reserve_heap(&factory, BIG_UINT_HEAP_SIZE); if (IS_USMALL(0, (Uint)ptr[0])) mess = make_small((Uint)ptr[0]); else { - mess = uint_to_big((Uint)ptr[0], hp); - hp += BIG_UINT_HEAP_SIZE; + mess = uint_to_big((Uint)ptr[0], factory.hp); + factory.hp += BIG_UINT_HEAP_SIZE; } -#endif ptr++; break; case ERL_DRV_INT64: /* pointer to unsigned 64-bit int argument */ - mess = erts_bld_sint64(&hp, NULL, *((Sint64 *) ptr[0])); + erts_reserve_heap(&factory, BIG_NEED_FOR_BITS(64)); + mess = erts_bld_sint64(&factory.hp, NULL, *((Sint64 *) ptr[0])); ptr++; break; case ERL_DRV_UINT64: /* pointer to unsigned 64-bit int argument */ - mess = erts_bld_uint64(&hp, NULL, *((Uint64 *) ptr[0])); + erts_reserve_heap(&factory, BIG_NEED_FOR_BITS(64)); + mess = erts_bld_uint64(&factory.hp, NULL, *((Uint64 *) ptr[0])); ptr++; break; @@ -5456,11 +6330,14 @@ driver_deliver_term(Eterm to, ErlDrvTermData* data, int len) Uint size = ptr[1]; Uint offset = ptr[2]; - erts_smp_atomic_add_nob(&erts_bytes_in, (erts_aint_t) size); + if (esdp) + esdp->io.in += (Uint64) size; + else + erts_atomic64_add_nob(&bytes_in, (erts_aint64_t) size); if (size <= ERL_ONHEAP_BIN_LIMIT) { - ErlHeapBin* hbp = (ErlHeapBin *) hp; - hp += heap_bin_size(size); + ErlHeapBin* hbp = (ErlHeapBin *) erts_produce_heap(&factory, + heap_bin_size(size), HEAP_EXTRA); hbp->thing_word = header_heap_bin(size); hbp->size = size; if (size > 0) { @@ -5469,18 +6346,18 @@ driver_deliver_term(Eterm to, ErlDrvTermData* data, int len) mess = make_binary(hbp); } else { - ProcBin* pb = (ProcBin *) hp; + ProcBin* pb = (ProcBin *) erts_produce_heap(&factory, + PROC_BIN_SIZE, HEAP_EXTRA); driver_binary_inc_refc(b); /* caller will free binary */ pb->thing_word = HEADER_PROC_BIN; pb->size = size; - pb->next = ohp->first; - ohp->first = (struct erl_off_heap_header*)pb; + pb->next = factory.off_heap->first; + factory.off_heap->first = (struct erl_off_heap_header*)pb; pb->val = ErlDrvBinary2Binary(b); pb->bytes = ((byte*) b->orig_bytes) + offset; pb->flags = 0; mess = make_binary(pb); - hp += PROC_BIN_SIZE; - OH_OVERHEAD(ohp, pb->size / sizeof(Eterm)); + OH_OVERHEAD(factory.off_heap, pb->size / sizeof(Eterm)); } ptr += 3; break; @@ -5490,11 +6367,15 @@ driver_deliver_term(Eterm to, ErlDrvTermData* data, int len) byte *bufp = (byte *) ptr[0]; Uint size = (Uint) ptr[1]; - erts_smp_atomic_add_nob(&erts_bytes_in, (erts_aint_t) size); + if (esdp) + esdp->io.in += (Uint64) size; + else + erts_atomic64_add_nob(&bytes_in, (erts_aint64_t) size); if (size <= ERL_ONHEAP_BIN_LIMIT) { - ErlHeapBin* hbp = (ErlHeapBin *) hp; - hp += heap_bin_size(size); + ErlHeapBin* hbp = (ErlHeapBin *) erts_produce_heap(&factory, + heap_bin_size(size), + HEAP_EXTRA); hbp->thing_word = header_heap_bin(size); hbp->size = size; if (size > 0) { @@ -5507,20 +6388,18 @@ driver_deliver_term(Eterm to, ErlDrvTermData* data, int len) ProcBin* pbp; Binary* bp = erts_bin_nrml_alloc(size); ASSERT(bufp); - bp->flags = 0; - bp->orig_size = (SWord) size; erts_refc_init(&bp->refc, 1); sys_memcpy((void *) bp->orig_bytes, (void *) bufp, size); - pbp = (ProcBin *) hp; - hp += PROC_BIN_SIZE; + pbp = (ProcBin *) erts_produce_heap(&factory, + PROC_BIN_SIZE, HEAP_EXTRA); pbp->thing_word = HEADER_PROC_BIN; pbp->size = size; - pbp->next = ohp->first; - ohp->first = (struct erl_off_heap_header*)pbp; + pbp->next = factory.off_heap->first; + factory.off_heap->first = (struct erl_off_heap_header*)pbp; pbp->val = bp; pbp->bytes = (byte*) bp->orig_bytes; pbp->flags = 0; - OH_OVERHEAD(ohp, pbp->size / sizeof(Eterm)); + OH_OVERHEAD(factory.off_heap, pbp->size / sizeof(Eterm)); mess = make_binary(pbp); } ptr += 2; @@ -5528,14 +6407,19 @@ driver_deliver_term(Eterm to, ErlDrvTermData* data, int len) } case ERL_DRV_STRING: /* char*, length */ - erts_smp_atomic_add_nob(&erts_bytes_in, (erts_aint_t) ptr[1]); - mess = buf_to_intlist(&hp, (char*)ptr[0], ptr[1], NIL); + if (esdp) + esdp->io.in += (Uint64) ptr[1]; + else + erts_atomic64_add_nob(&bytes_in, (erts_aint64_t) ptr[1]); + erts_reserve_heap(&factory, 2*ptr[1]); + mess = buf_to_intlist(&factory.hp, (char*)ptr[0], ptr[1], NIL); ptr += 2; break; case ERL_DRV_STRING_CONS: /* char*, length */ mess = ESTACK_POP(stack); - mess = buf_to_intlist(&hp, (char*)ptr[0], ptr[1], mess); + erts_reserve_heap(&factory, 2*ptr[1]); + mess = buf_to_intlist(&factory.hp, (char*)ptr[0], ptr[1], mess); ptr += 2; break; @@ -5544,11 +6428,12 @@ driver_deliver_term(Eterm to, ErlDrvTermData* data, int len) mess = ESTACK_POP(stack); i--; + erts_reserve_heap(&factory, 2*i); while(i > 0) { Eterm hd = ESTACK_POP(stack); - mess = CONS(hp, hd, mess); - hp += 2; + mess = CONS(factory.hp, hd, mess); + factory.hp += 2; i--; } ptr++; @@ -5557,13 +6442,12 @@ driver_deliver_term(Eterm to, ErlDrvTermData* data, int len) case ERL_DRV_TUPLE: { /* int */ int size = (int)ptr[0]; - Eterm* tp = hp; + Eterm* tp = erts_produce_heap(&factory, size+1, HEAP_EXTRA); *tp = make_arityval(size); mess = make_tuple(tp); tp += size; /* point at last element */ - hp = tp+1; /* advance "heap" pointer */ while(size--) { *tp-- = ESTACK_POP(stack); @@ -5579,18 +6463,22 @@ driver_deliver_term(Eterm to, ErlDrvTermData* data, int len) case ERL_DRV_FLOAT: { /* double * */ FloatDef f; + Eterm* fp = erts_produce_heap(&factory, FLOAT_SIZE_OBJECT, HEAP_EXTRA); - mess = make_float(hp); + mess = make_float(fp); f.fd = *((double *) ptr[0]); - PUT_DOUBLE(f, hp); - hp += FLOAT_SIZE_OBJECT; + if (!erts_isfinite(f.fd)) + ERTS_DDT_FAIL; + PUT_DOUBLE(f, fp); ptr++; break; } case ERL_DRV_EXT2TERM: /* char *ext, int size */ ASSERT(b2t.org_ext[b2t.ix] == (byte *) ptr[0]); - mess = erts_binary2term_create(&b2t.state[b2t.ix++], &hp, ohp); + + erts_reserve_heap(&factory, b2t.state[b2t.ix].heap_size); + mess = erts_binary2term_create(&b2t.state[b2t.ix++], &factory); if (mess == THE_NON_VALUE) ERTS_DDT_FAIL; ptr += 2; @@ -5598,31 +6486,43 @@ driver_deliver_term(Eterm to, ErlDrvTermData* data, int len) case ERL_DRV_MAP: { /* int */ int size = (int)ptr[0]; - Eterm* tp = hp; - Eterm* vp; - map_t *mp; - - *tp = make_arityval(size); - - hp += 1 + size; - mp = (map_t*)hp; - mp->thing_word = MAP_HEADER; - mp->size = size; - mp->keys = make_tuple(tp); - mess = make_map(mp); - - hp += MAP_HEADER_SIZE + size; /* advance "heap" pointer */ - - tp += size; /* point at last key */ - vp = hp - 1; /* point at last value */ - - while(size--) { - *vp-- = ESTACK_POP(stack); - *tp-- = ESTACK_POP(stack); - } - if (!erts_validate_and_sort_map(mp)) - ERTS_DDT_FAIL; - ptr++; + if (size > MAP_SMALL_MAP_LIMIT) { + int ix = 2*size; + Eterm* leafs; + + erts_produce_heap(&factory, ix, HEAP_EXTRA); + leafs = factory.hp; + while(ix--) { *--leafs = ESTACK_POP(stack); } + + mess = erts_hashmap_from_array(&factory, leafs, size, 1); + if (is_non_value(mess)) + ERTS_DDT_FAIL; + } else { + Eterm* vp; + flatmap_t *mp; + Eterm* tp = erts_produce_heap(&factory, + 2*size + 1 + MAP_HEADER_FLATMAP_SZ, + HEAP_EXTRA); + + *tp = make_arityval(size); + + mp = (flatmap_t*) (tp + 1 + size); + mp->thing_word = MAP_HEADER_FLATMAP; + mp->size = size; + mp->keys = make_tuple(tp); + mess = make_flatmap(mp); + + tp += size; /* point at last key */ + vp = factory.hp - 1; /* point at last value */ + + while(size--) { + *vp-- = ESTACK_POP(stack); + *tp-- = ESTACK_POP(stack); + } + if (!erts_validate_and_sort_flatmap(mp)) + ERTS_DDT_FAIL; + } + ptr++; break; } @@ -5630,52 +6530,62 @@ driver_deliver_term(Eterm to, ErlDrvTermData* data, int len) ESTACK_PUSH(stack, mess); } - res = 1; - done: if (res > 0) { + Eterm from = am_undefined; mess = ESTACK_POP(stack); /* get resulting value */ - if (bp) - bp = erts_resize_message_buffer(bp, hp - hp_start, &mess, 1); - else { - ASSERT(hp); - HRelease(rp, hp_end, hp); + erts_factory_trim_and_close(&factory, &mess, 1); + + if (prt) { + if (IS_TRACED_FL(prt, F_TRACE_SEND)) { + trace_port_send(prt, to, mess, 1); + } + from = prt->common.id; } + /* send message */ - erts_queue_message(rp, &rp_locks, bp, mess, am_undefined -#ifdef USE_VM_PROBES - , NIL -#endif - ); + ERL_MESSAGE_TOKEN(factory.message) = am_undefined; + erts_queue_message(rp, rp_locks, factory.message, mess, from); + } + else if (res == -2) { + /* this clause only happens when we were requested to + generate a send trace, but the process to send to + did not exist any more */ + mess = ESTACK_POP(stack); /* get resulting value */ + + trace_port_send(prt, to, mess, 0); + + erts_factory_trim_and_close(&factory, &mess, 1); + erts_free(ERTS_ALC_T_DRIVER, factory.hp_start); + res = 0; } else { if (b2t.ix > b2t.used) b2t.used = b2t.ix; for (b2t.ix = 0; b2t.ix < b2t.used; b2t.ix++) erts_binary2term_abort(&b2t.state[b2t.ix]); - if (bp) - free_message_buffer(bp); - else if (hp) { - HRelease(rp, hp_end, hp); - } + if (factory.mode != FACTORY_CLOSED) { + ERL_MESSAGE_TERM(factory.message) = am_undefined; + erts_factory_undo(&factory); + } } -#ifdef ERTS_SMP if (rp) { if (rp_locks) erts_smp_proc_unlock(rp, rp_locks); if (!scheduler) - erts_smp_proc_dec_refc(rp); + erts_proc_dec_refc(rp); } -#endif cleanup_b2t_states(&b2t); DESTROY_ESTACK(stack); return res; #undef ERTS_DDT_FAIL +#undef HEAP_EXTRA } static ERTS_INLINE int -deliver_term_check_port(ErlDrvTermData port_id, Eterm *connected_p) +deliver_term_check_port(ErlDrvTermData port_id, Eterm *connected_p, + Port **trace_prt) { #ifdef ERTS_SMP ErtsThrPrgrDelayHandle dhndl = erts_thr_progress_unmanaged_delay(); @@ -5713,8 +6623,10 @@ done: ETHR_MEMBAR(ETHR_LoadLoad|ETHR_LoadStore); } else #endif - ERTS_SMP_LC_ASSERT(res != 1 || erts_lc_is_port_locked(prt)); - + if (res == 1) { + ERTS_SMP_LC_ASSERT(erts_lc_is_port_locked(prt)); + *trace_prt = prt; + } return res; } @@ -5722,10 +6634,11 @@ int erl_drv_output_term(ErlDrvTermData port_id, ErlDrvTermData* data, int len) { /* May be called from arbitrary thread */ Eterm connected = NIL; /* Shut up faulty warning... */ - int res = deliver_term_check_port(port_id, &connected); + Port *prt = NULL; + int res = deliver_term_check_port(port_id, &connected, &prt); if (res <= 0) return res; - return driver_deliver_term(connected, data, len); + return driver_deliver_term(prt, connected, data, len); } /* @@ -5749,7 +6662,7 @@ driver_output_term(ErlDrvPort drvport, ErlDrvTermData* data, int len) if (state & ERTS_PORT_SFLG_CLOSING) return 0; - return driver_deliver_term(ERTS_PORT_GET_CONNECTED(prt), data, len); + return driver_deliver_term(prt, ERTS_PORT_GET_CONNECTED(prt), data, len); } int erl_drv_send_term(ErlDrvTermData port_id, @@ -5758,10 +6671,11 @@ int erl_drv_send_term(ErlDrvTermData port_id, int len) { /* May be called from arbitrary thread */ - int res = deliver_term_check_port(port_id, NULL); + Port *prt = NULL; + int res = deliver_term_check_port(port_id, NULL, &prt); if (res <= 0) return res; - return driver_deliver_term(to, data, len); + return driver_deliver_term(prt, to, data, len); } /* @@ -5780,20 +6694,21 @@ driver_send_term(ErlDrvPort drvport, * to make this access safe without using a less efficient * internal data representation for ErlDrvPort. */ + Port* prt = NULL; ERTS_SMP_CHK_NO_PROC_LOCKS; #ifdef ERTS_SMP if (erts_thr_progress_is_managed_thread()) #endif { erts_aint32_t state; - Port* prt = erts_drvport2port_state(drvport, &state); + prt = erts_drvport2port_state(drvport, &state); if (prt == ERTS_INVALID_ERL_DRV_PORT) return -1; /* invalid (dead) */ ERTS_SMP_LC_ASSERT(erts_lc_is_port_locked(prt)); if (state & ERTS_PORT_SFLG_CLOSING) return 0; } - return driver_deliver_term(to, data, len); + return driver_deliver_term(prt, to, data, len); } @@ -5807,6 +6722,7 @@ int driver_output_binary(ErlDrvPort ix, char* hbuf, ErlDrvSizeT hlen, { erts_aint32_t state; Port* prt = erts_drvport2port_state(ix, &state); + ErtsSchedulerData *esdp = erts_get_scheduler_data(); ERTS_SMP_CHK_NO_PROC_LOCKS; @@ -5817,7 +6733,10 @@ int driver_output_binary(ErlDrvPort ix, char* hbuf, ErlDrvSizeT hlen, return 0; prt->bytes_in += (hlen + len); - erts_smp_atomic_add_nob(&erts_bytes_in, (erts_aint_t) (hlen + len)); + if (esdp) + esdp->io.in += (Uint64) (hlen + len); + else + erts_atomic64_add_nob(&bytes_in, (erts_aint64_t) (hlen + len)); if (state & ERTS_PORT_SFLG_DISTRIBUTION) { return erts_net_message(prt, prt->dist_entry, @@ -5842,6 +6761,7 @@ int driver_output2(ErlDrvPort ix, char* hbuf, ErlDrvSizeT hlen, { erts_aint32_t state; Port* prt = erts_drvport2port_state(ix, &state); + ErtsSchedulerData *esdp = erts_get_scheduler_data(); ERTS_SMP_CHK_NO_PROC_LOCKS; @@ -5853,7 +6773,10 @@ int driver_output2(ErlDrvPort ix, char* hbuf, ErlDrvSizeT hlen, return 0; prt->bytes_in += (hlen + len); - erts_smp_atomic_add_nob(&erts_bytes_in, (erts_aint_t) (hlen + len)); + if (esdp) + esdp->io.in += (Uint64) (hlen + len); + else + erts_atomic64_add_nob(&bytes_in, (erts_aint64_t) (hlen + len)); if (state & ERTS_PORT_SFLG_DISTRIBUTION) { if (len == 0) return erts_net_message(prt, @@ -5893,6 +6816,7 @@ int driver_outputv(ErlDrvPort ix, char* hbuf, ErlDrvSizeT hlen, ErlDrvBinary** binv; Port* prt; erts_aint32_t state; + ErtsSchedulerData *esdp = erts_get_scheduler_data(); ERTS_SMP_CHK_NO_PROC_LOCKS; @@ -5931,7 +6855,10 @@ int driver_outputv(ErlDrvPort ix, char* hbuf, ErlDrvSizeT hlen, /* XXX handle distribution !!! */ prt->bytes_in += (hlen + size); - erts_smp_atomic_add_nob(&erts_bytes_in, (erts_aint_t) (hlen + size)); + if (esdp) + esdp->io.in += (Uint64) (hlen + size); + else + erts_atomic64_add_nob(&bytes_in, (erts_aint64_t) (hlen + size)); deliver_vec_message(prt, ERTS_PORT_GET_CONNECTED(prt), hbuf, hlen, binv, iov, n, size); return 0; @@ -6007,9 +6934,7 @@ driver_alloc_binary(ErlDrvSizeT size) bin = erts_bin_drv_alloc_fnf((Uint) size); if (!bin) return NULL; /* The driver write must take action */ - bin->flags = BIN_FLAG_DRV; erts_refc_init(&bin->refc, 1); - bin->orig_size = (SWord) size; return Binary2ErlDrvBinary(bin); } @@ -6039,7 +6964,6 @@ ErlDrvBinary* driver_realloc_binary(ErlDrvBinary* bin, ErlDrvSizeT size) if (!newbin) return NULL; - newbin->orig_size = size; return Binary2ErlDrvBinary(newbin); } @@ -6623,18 +7547,6 @@ int driver_pushq(ErlDrvPort ix, char* buffer, ErlDrvSizeT len) return code; } -static ERTS_INLINE void -drv_cancel_timer(Port *prt) -{ -#ifdef ERTS_SMP - erts_cancel_smp_ptimer(prt->common.u.alive.ptimer); -#else - erts_cancel_timer(&prt->common.u.alive.tm); -#endif - if (erts_port_task_is_scheduled(&prt->timeout_task)) - erts_port_task_abort(&prt->timeout_task); -} - int driver_set_timer(ErlDrvPort ix, unsigned long t) { Port* prt = erts_drvport2port(ix); @@ -6646,19 +7558,8 @@ int driver_set_timer(ErlDrvPort ix, unsigned long t) if (prt->drv_ptr->timeout == NULL) return -1; - drv_cancel_timer(prt); -#ifdef ERTS_SMP - erts_create_smp_ptimer(&prt->common.u.alive.ptimer, - prt->common.id, - (ErlTimeoutProc) schedule_port_timeout, - t); -#else - erts_set_timer(&prt->common.u.alive.tm, - (ErlTimeoutProc) schedule_port_timeout, - NULL, - prt, - t); -#endif + + erts_set_port_timer(prt, (Sint64) t); return 0; } @@ -6668,28 +7569,28 @@ int driver_cancel_timer(ErlDrvPort ix) if (prt == ERTS_INVALID_ERL_DRV_PORT) return -1; ERTS_SMP_LC_ASSERT(erts_lc_is_port_locked(prt)); - drv_cancel_timer(prt); + erts_cancel_port_timer(prt); return 0; } - int driver_read_timer(ErlDrvPort ix, unsigned long* t) { Port* prt = erts_drvport2port(ix); + Sint64 left; ERTS_SMP_CHK_NO_PROC_LOCKS; if (prt == ERTS_INVALID_ERL_DRV_PORT) return -1; ERTS_SMP_LC_ASSERT(erts_lc_is_port_locked(prt)); -#ifdef ERTS_SMP - *t = (prt->common.u.alive.ptimer - ? erts_time_left(&prt->common.u.alive.ptimer->timer.tm) - : 0); -#else - *t = erts_time_left(&prt->common.u.alive.tm); -#endif + + left = erts_read_port_timer(prt); + if (left < 0) + left = 0; + + *t = (unsigned long) left; + return 0; } @@ -6709,11 +7610,33 @@ driver_get_now(ErlDrvNowData *now_data) return 0; } +ErlDrvTime +erl_drv_monotonic_time(ErlDrvTimeUnit time_unit) +{ + return (ErlDrvTime) erts_napi_monotonic_time((int) time_unit); +} + +ErlDrvTime +erl_drv_time_offset(ErlDrvTimeUnit time_unit) +{ + return (ErlDrvTime) erts_napi_time_offset((int) time_unit); +} + +ErlDrvTime +erl_drv_convert_time_unit(ErlDrvTime val, + ErlDrvTimeUnit from, + ErlDrvTimeUnit to) +{ + return (ErlDrvTime) erts_napi_convert_time_unit((ErtsMonotonicTime) val, + (int) from, + (int) to); +} + static void ref_to_driver_monitor(Eterm ref, ErlDrvMonitor *mon) { RefThing *refp; ASSERT(is_internal_ref(ref)); - ASSERT(sizeof(RefThing) <= sizeof(ErlDrvMonitor)); + ERTS_CT_ASSERT(sizeof(RefThing) <= sizeof(ErlDrvMonitor)); refp = ref_thing_ptr(ref); memset(mon,0,sizeof(ErlDrvMonitor)); memcpy(mon,refp,sizeof(RefThing)); @@ -6756,7 +7679,7 @@ int driver_monitor_process(ErlDrvPort drvport, { Port *prt; int ret; -#if !HEAP_ON_C_STACK || (defined(ERTS_SMP) && defined(ERTS_ENABLE_LOCK_CHECK)) +#if defined(ERTS_SMP) && defined(ERTS_ENABLE_LOCK_CHECK) ErtsSchedulerData *sched = erts_get_scheduler_data(); #endif @@ -6767,16 +7690,6 @@ int driver_monitor_process(ErlDrvPort drvport, /* Now (in SMP) we should have either the port lock (if we have a scheduler) or the port data lock (if we're a driver thread) */ ERTS_SMP_LC_ASSERT((sched != NULL || prt->port_data_lock)); - -#if !HEAP_ON_C_STACK - if (!sched) { - /* Need a separate allocation for the ref :( */ - Eterm *buf = erts_alloc(ERTS_ALC_T_TEMP_TERM, - sizeof(Eterm)*REF_THING_SIZE); - ret = do_driver_monitor_process(prt,buf,process,monitor); - erts_free(ERTS_ALC_T_TEMP_TERM,buf); - } else -#endif { DeclareTmpHeapNoproc(buf,REF_THING_SIZE); UseTmpHeapNoproc(REF_THING_SIZE); @@ -6829,7 +7742,7 @@ int driver_demonitor_process(ErlDrvPort drvport, { Port *prt; int ret; -#if !HEAP_ON_C_STACK || (defined(ERTS_SMP) && defined(ERTS_ENABLE_LOCK_CHECK)) +#if defined(ERTS_SMP) && defined(ERTS_ENABLE_LOCK_CHECK) ErtsSchedulerData *sched = erts_get_scheduler_data(); #endif @@ -6840,15 +7753,6 @@ int driver_demonitor_process(ErlDrvPort drvport, /* Now we should have either the port lock (if we have a scheduler) or the port data lock (if we're a driver thread) */ ERTS_SMP_LC_ASSERT((sched != NULL || prt->port_data_lock)); -#if !HEAP_ON_C_STACK - if (!sched) { - /* Need a separate allocation for the ref :( */ - Eterm *buf = erts_alloc(ERTS_ALC_T_TEMP_TERM, - sizeof(Eterm)*REF_THING_SIZE); - ret = do_driver_demonitor_process(prt,buf,monitor); - erts_free(ERTS_ALC_T_TEMP_TERM,buf); - } else -#endif { DeclareTmpHeapNoproc(buf,REF_THING_SIZE); UseTmpHeapNoproc(REF_THING_SIZE); @@ -6884,7 +7788,7 @@ ErlDrvTermData driver_get_monitored_process(ErlDrvPort drvport, { Port *prt; ErlDrvTermData ret; -#if !HEAP_ON_C_STACK || (defined(ERTS_SMP) && defined(ERTS_ENABLE_LOCK_CHECK)) +#if defined(ERTS_SMP) && defined(ERTS_ENABLE_LOCK_CHECK) ErtsSchedulerData *sched = erts_get_scheduler_data(); #endif @@ -6895,16 +7799,6 @@ ErlDrvTermData driver_get_monitored_process(ErlDrvPort drvport, /* Now we should have either the port lock (if we have a scheduler) or the port data lock (if we're a driver thread) */ ERTS_SMP_LC_ASSERT((sched != NULL || prt->port_data_lock)); - -#if !HEAP_ON_C_STACK - if (!sched) { - /* Need a separate allocation for the ref :( */ - Eterm *buf = erts_alloc(ERTS_ALC_T_TEMP_TERM, - sizeof(Eterm)*REF_THING_SIZE); - ret = do_driver_get_monitored_process(prt,buf,monitor); - erts_free(ERTS_ALC_T_TEMP_TERM,buf); - } else -#endif { DeclareTmpHeapNoproc(buf,REF_THING_SIZE); UseTmpHeapNoproc(REF_THING_SIZE); @@ -6928,6 +7822,7 @@ void erts_fire_port_monitor(Port *prt, Eterm ref) void (*callback)(ErlDrvData drv_data, ErlDrvMonitor *monitor); ErlDrvMonitor drv_monitor; int fpe_was_unmasked; + ERTS_MSACC_PUSH_STATE_M(); ERTS_SMP_LC_ASSERT(erts_lc_is_port_locked(prt)); ASSERT(prt->drv_ptr != NULL); @@ -6939,6 +7834,7 @@ void erts_fire_port_monitor(Port *prt, Eterm ref) callback = prt->drv_ptr->process_exit; ASSERT(callback != NULL); ref_to_driver_monitor(ref,&drv_monitor); + ERTS_MSACC_SET_STATE_CACHED_M(ERTS_MSACC_STATE_PORT); DRV_MONITOR_UNLOCK_PDL(prt); #ifdef USE_VM_PROBES if (DTRACE_ENABLED(driver_process_exit)) { @@ -6946,10 +7842,20 @@ void erts_fire_port_monitor(Port *prt, Eterm ref) DTRACE3(driver_process_exit, process_str, port_str, prt->name); } #endif +#ifdef USE_LTTNG_VM_TRACEPOINTS + if (LTTNG_ENABLED(driver_process_exit)) { + lttng_decl_portbuf(port_str); + lttng_decl_procbuf(proc_str); + lttng_pid_to_str(ERTS_PORT_GET_CONNECTED(prt), proc_str); + lttng_port_to_str(prt, port_str); + LTTNG3(driver_process_exit, proc_str, port_str, prt->name); + } +#endif fpe_was_unmasked = erts_block_fpe(); (*callback)((ErlDrvData) (prt->drv_data), &drv_monitor); erts_unblock_fpe(fpe_was_unmasked); DRV_MONITOR_LOCK_PDL(prt); + ERTS_MSACC_POP_STATE_M(); /* remove monitor *after* callback */ rmon = erts_remove_monitor(&ERTS_P_MONITORS(prt), ref); DRV_MONITOR_UNLOCK_PDL(prt); @@ -6970,12 +7876,15 @@ driver_failure_term(ErlDrvPort ix, Eterm term, int eof) if (prt == ERTS_INVALID_ERL_DRV_PORT) return -1; ERTS_SMP_LC_ASSERT(erts_lc_is_port_locked(prt)); + + if (prt->async_open_port) + init_ack_send_reply(prt, prt->common.id); if (eof) flush_linebuf_messages(prt, state); if (state & ERTS_PORT_SFLG_CLOSING) { terminate_port(prt); } else if (eof && (state & ERTS_PORT_SFLG_SOFT_EOF)) { - deliver_result(prt->common.id, ERTS_PORT_GET_CONNECTED(prt), am_eof); + deliver_result(prt, prt->common.id, ERTS_PORT_GET_CONNECTED(prt), am_eof); } else { /* XXX UGLY WORK AROUND, Let erts_deliver_port_exit() terminate the port */ if (prt->port_data_lock) @@ -6983,7 +7892,7 @@ driver_failure_term(ErlDrvPort ix, Eterm term, int eof) prt->ioq.size = 0; if (prt->port_data_lock) driver_pdl_unlock(prt->port_data_lock); - erts_deliver_port_exit(prt, prt->common.id, eof ? am_normal : term, 0); + erts_deliver_port_exit(prt, prt->common.id, eof ? am_normal : term, 0, 0); } return 0; } @@ -7236,7 +8145,7 @@ driver_system_info(ErlDrvSysInfo *sip, size_t si_size) * of ErlDrvSysInfo (introduced in driver version 1.0). */ if (!sip || si_size < ERL_DRV_SYS_INFO_SIZE(smp_support)) - erl_exit(1, + erts_exit(ERTS_ERROR_EXIT, "driver_system_info(%p, %ld) called with invalid arguments\n", sip, si_size); @@ -7287,11 +8196,13 @@ driver_system_info(ErlDrvSysInfo *sip, size_t si_size) * (driver version 3.1, NIF version 2.7) */ if (si_size >= ERL_DRV_SYS_INFO_SIZE(dirty_scheduler_support)) { -#if defined(ERL_NIF_DIRTY_SCHEDULER_SUPPORT) && defined(USE_THREADS) - sip->dirty_scheduler_support = 1; + sip->dirty_scheduler_support = +#ifdef ERTS_DIRTY_SCHEDULERS + 1 #else - sip->dirty_scheduler_support = 0; + 0 #endif + ; } } @@ -7419,6 +8330,8 @@ init_driver(erts_driver_t *drv, ErlDrvEntry *de, DE_Handle *handle) int fpe_was_unmasked = erts_block_fpe(); DTRACE4(driver_init, drv->name, drv->version.major, drv->version.minor, drv->flags); + LTTNG4(driver_init, drv->name, drv->version.major, drv->version.minor, + drv->flags); res = (*de->init)(); erts_unblock_fpe(fpe_was_unmasked); return res; @@ -7543,15 +8456,15 @@ int null_func(void) } int -erl_drv_putenv(char *key, char *value) +erl_drv_putenv(const char *key, char *value) { - return erts_sys_putenv_raw(key, value); + return erts_sys_putenv_raw((char*)key, value); } int -erl_drv_getenv(char *key, char *value, size_t *value_size) +erl_drv_getenv(const char *key, char *value, size_t *value_size) { - return erts_sys_getenv_raw(key, value, value_size); + return erts_sys_getenv_raw((char*)key, value, value_size); } /* get heart_port |