diff options
Diffstat (limited to 'erts/emulator/beam/erl_bif_port.c')
-rw-r--r-- | erts/emulator/beam/erl_bif_port.c | 867 |
1 files changed, 365 insertions, 502 deletions
diff --git a/erts/emulator/beam/erl_bif_port.c b/erts/emulator/beam/erl_bif_port.c index f9009166c0..81146e38d7 100644 --- a/erts/emulator/beam/erl_bif_port.c +++ b/erts/emulator/beam/erl_bif_port.c @@ -42,28 +42,26 @@ #include "erl_bits.h" #include "dtrace-wrapper.h" -static int open_port(Process* p, Eterm name, Eterm settings, int *err_nump); +static Port *open_port(Process* p, Eterm name, Eterm settings, int *err_typep, int *err_nump); static byte* convert_environment(Process* p, Eterm env); static char **convert_args(Eterm); static void free_args(char **); char *erts_default_arg0 = "default"; -static BIF_RETTYPE -port_call(Process* p, Eterm arg1, Eterm arg2, Eterm arg3); - BIF_RETTYPE open_port_2(BIF_ALIST_2) { - int port_num; - Eterm port_val; + Port *port; + Eterm port_id; char *str; - int err_num; + int err_type, err_num; - if ((port_num = open_port(BIF_P, BIF_ARG_1, BIF_ARG_2, &err_num)) < 0) { - if (port_num == -3) { + port = open_port(BIF_P, BIF_ARG_1, BIF_ARG_2, &err_type, &err_num); + if (!port) { + if (err_type == -3) { ASSERT(err_num == BADARG || err_num == SYSTEM_LIMIT); BIF_ERROR(BIF_P, err_num); - } else if (port_num == -2) { + } else if (err_type == -2) { str = erl_errno_id(err_num); } else { str = "einval"; @@ -74,546 +72,408 @@ BIF_RETTYPE open_port_2(BIF_ALIST_2) erts_smp_proc_lock(BIF_P, ERTS_PROC_LOCK_LINK); - port_val = erts_port[port_num].id; - erts_add_link(&(erts_port[port_num].nlinks), LINK_PID, BIF_P->id); - erts_add_link(&(BIF_P->nlinks), LINK_PID, port_val); + port_id = port->common.id; + erts_add_link(&ERTS_P_LINKS(port), LINK_PID, BIF_P->common.id); + erts_add_link(&ERTS_P_LINKS(BIF_P), LINK_PID, port_id); erts_smp_proc_unlock(BIF_P, ERTS_PROC_LOCK_LINK); - erts_port_release(&erts_port[port_num]); + erts_port_release(port); - BIF_RET(port_val); + BIF_RET(port_id); } -/**************************************************************************** - - PORT BIFS: - - port_command/2 -- replace Port ! {..., {command, Data}} - port_command(Port, Data) -> true - when port(Port), io-list(Data) - - port_control/3 -- new port_control(Port, Ctl, Data) -> Reply - port_control(Port, Ctl, Data) -> Reply - where integer(Ctl), io-list(Data), io-list(Reply) - - port_close/1 -- replace Port ! {..., close} - port_close(Port) -> true - when port(Port) - - port_connect/2 -- replace Port ! {..., {connect, Pid}} - port_connect(Port, Pid) - when port(Port), pid(Pid) - - ***************************************************************************/ - -static Port* -id_or_name2port(Process *c_p, Eterm id) +static ERTS_INLINE Port * +lookup_port(Process *c_p, Eterm id_or_name) { - Port *port; - if (is_not_atom(id)) - port = erts_id2port(id, c_p, ERTS_PROC_LOCK_MAIN); + /* TODO: Implement nicer lookup in register... */ + Eterm id; + if (is_atom(id_or_name)) + id = erts_whereis_name_to_id(c_p, id_or_name); else - erts_whereis_name(c_p, ERTS_PROC_LOCK_MAIN, id, NULL, 0, 0, &port); - return port; + id = id_or_name; + return erts_port_lookup(id, ERTS_PORT_SFLGS_INVALID_LOOKUP); } -#define ERTS_PORT_COMMAND_FLAG_FORCE (((Uint32) 1) << 0) -#define ERTS_PORT_COMMAND_FLAG_NOSUSPEND (((Uint32) 1) << 1) +/* + * erts_internal:port_command/3 is used by the + * erlang:port_command/2 and erlang:port_command/3 + * BIFs. + */ -static BIF_RETTYPE -do_port_command(Process *BIF_P, Eterm arg1, Eterm arg2, Eterm arg3, - Uint32 flags) +BIF_RETTYPE erts_internal_port_command_3(BIF_ALIST_3) { BIF_RETTYPE res; - Port *p; - - /* Trace sched out before lock check wait */ - if (IS_TRACED_FL(BIF_P, F_TRACE_SCHED_PROCS)) { - trace_virtual_sched(BIF_P, am_out); - } - - if (erts_system_profile_flags.runnable_procs && erts_system_profile_flags.exclusive) { - profile_runnable_proc(BIF_P, am_inactive); - } - - p = id_or_name2port(BIF_P, arg1); - if (!p) { - if (IS_TRACED_FL(BIF_P, F_TRACE_SCHED_PROCS)) { - trace_virtual_sched(BIF_P, am_in); + Port *prt; + int flags = 0; + Eterm ref; + + if (is_not_nil(BIF_ARG_3)) { + Eterm l = BIF_ARG_3; + while (is_list(l)) { + Eterm* cons = list_val(l); + Eterm car = CAR(cons); + if (car == am_force) + flags |= ERTS_PORT_SIG_FLG_FORCE; + else if (car == am_nosuspend) + flags |= ERTS_PORT_SIG_FLG_NOSUSPEND; + else + BIF_RET(am_badarg); + l = CDR(cons); } - if (erts_system_profile_flags.runnable_procs && erts_system_profile_flags.exclusive) { - profile_runnable_proc(BIF_P, am_active); - } - BIF_ERROR(BIF_P, BADARG); + if (!is_nil(l)) + BIF_RET(am_badarg); } - - /* Trace port in, id_or_name2port causes wait */ - if (IS_TRACED_FL(p, F_TRACE_SCHED_PORTS)) { - trace_sched_ports_where(p, am_in, am_command); - } - if (erts_system_profile_flags.runnable_ports && !erts_port_is_scheduled(p)) { - profile_runnable_port(p, am_active); + prt = lookup_port(BIF_P, BIF_ARG_1); + if (!prt) + BIF_RET(am_badarg); + + if (flags & ERTS_PORT_SIG_FLG_FORCE) { + if (!(prt->drv_ptr->flags & ERL_DRV_FLAG_SOFT_BUSY)) + BIF_RET(am_notsup); } - ERTS_BIF_PREP_RET(res, am_true); +#ifdef DEBUG + ref = NIL; +#endif - if ((flags & ERTS_PORT_COMMAND_FLAG_FORCE) - && !(p->drv_ptr->flags & ERL_DRV_FLAG_SOFT_BUSY)) { - ERTS_BIF_PREP_ERROR(res, BIF_P, EXC_NOTSUP); - } - else if (!(flags & ERTS_PORT_COMMAND_FLAG_FORCE) - && p->status & ERTS_PORT_SFLG_PORT_BUSY) { - if (flags & ERTS_PORT_COMMAND_FLAG_NOSUSPEND) { + switch (erts_port_output(BIF_P, flags, prt, prt->common.id, BIF_ARG_2, &ref)) { + case ERTS_PORT_OP_CALLER_EXIT: + case ERTS_PORT_OP_BADARG: + case ERTS_PORT_OP_DROPPED: + ERTS_BIF_PREP_RET(res, am_badarg); + break; + case ERTS_PORT_OP_BUSY: + ASSERT(!(flags & ERTS_PORT_SIG_FLG_FORCE)); + if (flags & ERTS_PORT_SIG_FLG_NOSUSPEND) ERTS_BIF_PREP_RET(res, am_false); - } else { - erts_suspend(BIF_P, ERTS_PROC_LOCK_MAIN, p); - if (erts_system_monitor_flags.busy_port) { - monitor_generic(BIF_P, am_busy_port, p->id); - } - ERTS_BIF_PREP_YIELD3(res, bif_export[BIF_port_command_3], BIF_P, - arg1, arg2, arg3); - } - } else { - int wres; - erts_smp_proc_unlock(BIF_P, ERTS_PROC_LOCK_MAIN); - ERTS_SMP_CHK_NO_PROC_LOCKS; - wres = erts_write_to_port(BIF_P->id, p, arg2); - erts_smp_proc_lock(BIF_P, ERTS_PROC_LOCK_MAIN); - if (wres != 0) { - ERTS_BIF_PREP_ERROR(res, BIF_P, BADARG); + erts_suspend(BIF_P, ERTS_PROC_LOCK_MAIN, prt); + ERTS_BIF_PREP_YIELD3(res, bif_export[BIF_erts_internal_port_command_3], + BIF_P, BIF_ARG_1, BIF_ARG_2, BIF_ARG_3); } - } - - if (IS_TRACED_FL(p, F_TRACE_SCHED_PORTS)) { - trace_sched_ports_where(p, am_out, am_command); - } - if (erts_system_profile_flags.runnable_ports && !erts_port_is_scheduled(p)) { - profile_runnable_port(p, am_inactive); + break; + case ERTS_PORT_OP_BUSY_SCHEDULED: + ASSERT(!(flags & ERTS_PORT_SIG_FLG_FORCE)); + /* Fall through... */ + case ERTS_PORT_OP_SCHEDULED: + ASSERT(is_internal_ref(ref)); + ERTS_BIF_PREP_RET(res, ref); + break; + case ERTS_PORT_OP_DONE: + ERTS_BIF_PREP_RET(res, am_true); + break; + default: + ERTS_INTERNAL_ERROR("Unexpected erts_port_output() result"); + break; } - erts_port_release(p); - /* Trace sched in after port release */ - if (IS_TRACED_FL(BIF_P, F_TRACE_SCHED_PROCS)) { - trace_virtual_sched(BIF_P, am_in); - } - if (erts_system_profile_flags.runnable_procs && erts_system_profile_flags.exclusive) { - profile_runnable_proc(BIF_P, am_active); - } - if (ERTS_PROC_IS_EXITING(BIF_P)) { KILL_CATCHES(BIF_P); /* Must exit */ ERTS_BIF_PREP_ERROR(res, BIF_P, EXC_ERROR); } - return res; -} -BIF_RETTYPE port_command_2(BIF_ALIST_2) -{ - return do_port_command(BIF_P, BIF_ARG_1, BIF_ARG_2, NIL, 0); + return res; } -BIF_RETTYPE port_command_3(BIF_ALIST_3) +BIF_RETTYPE erts_internal_port_call_3(BIF_ALIST_3) { - Eterm l = BIF_ARG_3; - Uint32 flags = 0; - while (is_list(l)) { - Eterm* cons = list_val(l); - Eterm car = CAR(cons); - if (car == am_force) { - flags |= ERTS_PORT_COMMAND_FLAG_FORCE; - } else if (car == am_nosuspend) { - flags |= ERTS_PORT_COMMAND_FLAG_NOSUSPEND; - } else { - BIF_ERROR(BIF_P, BADARG); - } - l = CDR(cons); - } - if(!is_nil(l)) { - BIF_ERROR(BIF_P, BADARG); + Port* prt; + Eterm retval; + Uint uint_op; + unsigned int op; + erts_aint32_t state; + + prt = lookup_port(BIF_P, BIF_ARG_1); + if (!prt) + BIF_RET(am_badarg); + + if (!term_to_Uint(BIF_ARG_2, &uint_op)) + BIF_RET(am_badarg); + + if (uint_op > (Uint) UINT_MAX) + BIF_RET(am_badarg); + + op = (unsigned int) uint_op; + + switch (erts_port_call(BIF_P, prt, op, BIF_ARG_3, &retval)) { + case ERTS_PORT_OP_CALLER_EXIT: + case ERTS_PORT_OP_DROPPED: + case ERTS_PORT_OP_BADARG: + retval = am_badarg; + break; + case ERTS_PORT_OP_SCHEDULED: + ASSERT(is_internal_ref(retval)); + break; + case ERTS_PORT_OP_DONE: + ASSERT(is_not_internal_ref(retval)); + break; + default: + ERTS_INTERNAL_ERROR("Unexpected erts_port_call() result"); + retval = am_internal_error; + break; } - return do_port_command(BIF_P, BIF_ARG_1, BIF_ARG_2, BIF_ARG_3, flags); -} -BIF_RETTYPE port_call_2(BIF_ALIST_2) -{ - return port_call(BIF_P,BIF_ARG_1, make_small(0), BIF_ARG_2); -} + state = erts_smp_atomic32_read_acqb(&BIF_P->state); + if (state & (ERTS_PSFLG_EXITING|ERTS_PSFLG_PENDING_EXIT)) { +#ifdef ERTS_SMP + if (state & ERTS_PSFLG_PENDING_EXIT) + erts_handle_pending_exit(BIF_P, ERTS_PROC_LOCK_MAIN); +#endif + ERTS_BIF_EXITED(BIF_P); + } -BIF_RETTYPE port_call_3(BIF_ALIST_3) -{ - return port_call(BIF_P, BIF_ARG_1, BIF_ARG_2, BIF_ARG_3); + BIF_RET(retval); } -static BIF_RETTYPE -port_call(Process* c_p, Eterm arg1, Eterm arg2, Eterm arg3) +BIF_RETTYPE erts_internal_port_control_3(BIF_ALIST_3) { - Uint op; - Port *p; - Uint size; - byte *bytes; - byte *endp; - ErlDrvSizeT real_size; - erts_driver_t *drv; - byte port_input[256]; /* Default input buffer to encode in */ - byte port_result[256]; /* Buffer for result from port. */ - byte* port_resp; /* Pointer to result buffer. */ - char *prc; - ErlDrvSSizeT ret; - Eterm res; - Sint result_size; - Eterm *hp; - Eterm *hp_end; - unsigned ret_flags = 0U; - int fpe_was_unmasked; - - bytes = &port_input[0]; - port_resp = port_result; - /* trace of port scheduling with virtual process descheduling - * lock wait - */ - if (IS_TRACED_FL(c_p, F_TRACE_SCHED_PROCS)) { - trace_virtual_sched(c_p, am_out); - } - - if (erts_system_profile_flags.runnable_procs && erts_system_profile_flags.exclusive) { - profile_runnable_proc(c_p, am_inactive); + Port* prt; + Eterm retval; + Uint uint_op; + unsigned int op; + erts_aint32_t state; + + prt = lookup_port(BIF_P, BIF_ARG_1); + if (!prt) + BIF_RET(am_badarg); + + if (!term_to_Uint(BIF_ARG_2, &uint_op)) + BIF_RET(am_badarg); + + if (uint_op > (Uint) UINT_MAX) + BIF_RET(am_badarg); + + op = (unsigned int) uint_op; + + switch (erts_port_control(BIF_P, prt, op, BIF_ARG_3, &retval)) { + case ERTS_PORT_OP_CALLER_EXIT: + case ERTS_PORT_OP_BADARG: + case ERTS_PORT_OP_DROPPED: + retval = am_badarg; + break; + case ERTS_PORT_OP_SCHEDULED: + ASSERT(is_internal_ref(retval)); + break; + case ERTS_PORT_OP_DONE: + ASSERT(is_not_internal_ref(retval)); + break; + default: + ERTS_INTERNAL_ERROR("Unexpected erts_port_control() result"); + retval = am_internal_error; + break; } - p = id_or_name2port(c_p, arg1); - if (!p) { - error: - if (port_resp != port_result && - !(ret_flags & DRIVER_CALL_KEEP_BUFFER)) { - driver_free(port_resp); - } - if (bytes != &port_input[0]) - erts_free(ERTS_ALC_T_PORT_CALL_BUF, bytes); - /* Need to virtual schedule in the process if there - * was an error. - */ - if (IS_TRACED_FL(c_p, F_TRACE_SCHED_PROCS)) { - trace_virtual_sched(c_p, am_in); - } - - if (erts_system_profile_flags.runnable_procs && erts_system_profile_flags.exclusive) { - profile_runnable_proc(c_p, am_active); - } - - if (p) - erts_port_release(p); + state = erts_smp_atomic32_read_acqb(&BIF_P->state); + if (state & (ERTS_PSFLG_EXITING|ERTS_PSFLG_PENDING_EXIT)) { #ifdef ERTS_SMP - ERTS_SMP_BIF_CHK_PENDING_EXIT(c_p, ERTS_PROC_LOCK_MAIN); -#else - ERTS_BIF_CHK_EXITED(c_p); + if (state & ERTS_PSFLG_PENDING_EXIT) + erts_handle_pending_exit(BIF_P, ERTS_PROC_LOCK_MAIN); #endif - BIF_ERROR(c_p, BADARG); - } - - if ((drv = p->drv_ptr) == NULL) { - goto error; - } - if (drv->call == NULL) { - goto error; - } - if (!term_to_Uint(arg2, &op)) { - goto error; - } - p->caller = c_p->id; - - /* Lock taken, virtual schedule of port */ - if (IS_TRACED_FL(p, F_TRACE_SCHED_PORTS)) { - trace_sched_ports_where(p, am_in, am_call); - } - - if (erts_system_profile_flags.runnable_ports && !erts_port_is_scheduled(p)) { - profile_runnable_port(p, am_active); + ERTS_BIF_EXITED(BIF_P); } - size = erts_encode_ext_size(arg3); - if (size > sizeof(port_input)) - bytes = erts_alloc(ERTS_ALC_T_PORT_CALL_BUF, size); - endp = bytes; - erts_encode_ext(arg3, &endp); + BIF_RET(retval); +} - real_size = endp - bytes; - if (real_size > size) { - erl_exit(1, "%s, line %d: buffer overflow: %d word(s)\n", - __FILE__, __LINE__, endp - (bytes + size)); - } - erts_smp_proc_unlock(c_p, ERTS_PROC_LOCK_MAIN); -#ifdef USE_VM_PROBES - if (DTRACE_ENABLED(driver_call)) { - DTRACE_CHARBUF(process_str, DTRACE_TERM_BUF_SIZE); - DTRACE_CHARBUF(port_str, DTRACE_TERM_BUF_SIZE); +/* + * erts_internal:port_close/1 is used by the + * erlang:port_close/1 BIF. + */ +BIF_RETTYPE erts_internal_port_close_1(BIF_ALIST_1) +{ + Eterm ref; + Port *prt; - dtrace_pid_str(p->connected, process_str); - dtrace_port_str(p, port_str); - DTRACE5(driver_call, process_str, port_str, p->name, op, real_size); - } -#endif - prc = (char *) port_resp; - fpe_was_unmasked = erts_block_fpe(); - ret = drv->call((ErlDrvData)p->drv_data, - (unsigned) op, - (char *) bytes, - (int) real_size, - &prc, - (int) sizeof(port_result), - &ret_flags); - erts_unblock_fpe(fpe_was_unmasked); - if (IS_TRACED_FL(p, F_TRACE_SCHED_PORTS)) { - trace_sched_ports_where(p, am_out, am_call); - } - - if (erts_system_profile_flags.runnable_ports && !erts_port_is_scheduled(p)) { - profile_runnable_port(p, am_inactive); - } - - port_resp = (byte *) prc; - p->caller = NIL; - erts_smp_proc_lock(c_p, ERTS_PROC_LOCK_MAIN); -#ifdef HARDDEBUG - { - ErlDrvSizeT z; - printf("real_size = %ld,%d, ret = %ld,%d\r\n", (unsigned long) real_size, - (int) real_size, (unsigned long)ret, (int) ret); - printf("["); - for(z = 0; z < real_size; ++z) { - printf("%d, ",(int) bytes[z]); - } - printf("]\r\n"); - printf("["); - for(z = 0; z < ret; ++z) { - printf("%d, ",(int) port_resp[z]); - } - printf("]\r\n"); - } -#endif - if (ret <= 0 || port_resp[0] != VERSION_MAGIC) { - /* Error or a binary without magic/ with wrong magic */ - goto error; - } - result_size = erts_decode_ext_size(port_resp, ret); - if (result_size < 0) { - goto error; - } - hp = HAlloc(c_p, result_size); - hp_end = hp + result_size; - endp = port_resp; - res = erts_decode_ext(&hp, &MSO(c_p), &endp); - if (res == THE_NON_VALUE) { - goto error; - } - HRelease(c_p, hp_end, hp); - if (port_resp != port_result && !(ret_flags & DRIVER_CALL_KEEP_BUFFER)) { - driver_free(port_resp); - } - if (bytes != &port_input[0]) - erts_free(ERTS_ALC_T_PORT_CALL_BUF, bytes); - if (p) - erts_port_release(p); -#ifdef ERTS_SMP - ERTS_SMP_BIF_CHK_PENDING_EXIT(c_p, ERTS_PROC_LOCK_MAIN); -#else - ERTS_BIF_CHK_EXITED(c_p); +#ifdef DEBUG + ref = NIL; #endif - if (IS_TRACED_FL(c_p, F_TRACE_SCHED_PROCS)) { - trace_virtual_sched(c_p, am_in); - } - if (erts_system_profile_flags.runnable_procs && erts_system_profile_flags.exclusive) { - profile_runnable_proc(c_p, am_active); + prt = lookup_port(BIF_P, BIF_ARG_1); + if (!prt) + BIF_RET(am_badarg); + + + switch (erts_port_exit(BIF_P, 0, prt, prt->common.id, am_normal, &ref)) { + case ERTS_PORT_OP_CALLER_EXIT: + case ERTS_PORT_OP_BADARG: + case ERTS_PORT_OP_DROPPED: + BIF_RET(am_badarg); + case ERTS_PORT_OP_SCHEDULED: + ASSERT(is_internal_ref(ref)); + BIF_RET(ref); + case ERTS_PORT_OP_DONE: + BIF_RET(am_true); + default: + ERTS_INTERNAL_ERROR("Unexpected erts_port_exit() result"); + BIF_RET(am_internal_error); } - - return res; } - -BIF_RETTYPE port_control_3(BIF_ALIST_3) -{ - Port* p; - Uint op; - Eterm res = THE_NON_VALUE; - - /* Virtual schedule out calling process before lock wait */ - if (IS_TRACED_FL(BIF_P, F_TRACE_SCHED_PROCS)) { - trace_virtual_sched(BIF_P, am_out); - } - if (erts_system_profile_flags.runnable_procs && erts_system_profile_flags.exclusive) { - profile_runnable_proc(BIF_P, am_inactive); - } +/* + * erts_internal:port_connect/2 is used by the + * erlang:port_connect/2 BIF. + */ +BIF_RETTYPE erts_internal_port_connect_2(BIF_ALIST_2) +{ + Eterm ref; + Port* prt; - p = id_or_name2port(BIF_P, BIF_ARG_1); - if (!p) { - /* Schedule the process before exiting */ - if (IS_TRACED_FL(BIF_P, F_TRACE_SCHED_PROCS)) { - trace_virtual_sched(BIF_P, am_in); - } - - if (erts_system_profile_flags.runnable_procs && erts_system_profile_flags.exclusive) { - profile_runnable_proc(BIF_P, am_active); - } - - BIF_ERROR(BIF_P, BADARG); - } + prt = lookup_port(BIF_P, BIF_ARG_1); + if (!prt) + BIF_RET(am_badarg); - /* Trace the port for scheduling in */ - if (IS_TRACED_FL(p, F_TRACE_SCHED_PORTS)) { - trace_sched_ports_where(p, am_in, am_control); - } - - if (erts_system_profile_flags.runnable_ports && !erts_port_is_scheduled(p)) { - profile_runnable_port(p, am_active); - } +#ifdef DEBUG + ref = NIL; +#endif - if (term_to_Uint(BIF_ARG_2, &op)) - res = erts_port_control(BIF_P, p, op, BIF_ARG_3); - - /* Trace the port for scheduling out */ - if (IS_TRACED_FL(p, F_TRACE_SCHED_PORTS)) { - trace_sched_ports_where(p, am_out, am_control); + switch (erts_port_connect(BIF_P, 0, prt, prt->common.id, BIF_ARG_2, &ref)) { + case ERTS_PORT_OP_CALLER_EXIT: + case ERTS_PORT_OP_BADARG: + case ERTS_PORT_OP_DROPPED: + BIF_RET(am_badarg); + case ERTS_PORT_OP_SCHEDULED: + ASSERT(is_internal_ref(ref)); + BIF_RET(ref); + break; + case ERTS_PORT_OP_DONE: + BIF_RET(am_true); + break; + default: + ERTS_INTERNAL_ERROR("Unexpected erts_port_connect() result"); + BIF_RET(am_internal_error); } +} - if (erts_system_profile_flags.runnable_ports && !erts_port_is_scheduled(p)) { - profile_runnable_port(p, am_inactive); - } +BIF_RETTYPE erts_internal_port_info_1(BIF_ALIST_1) +{ + Eterm retval; + Port* prt; - erts_port_release(p); -#ifdef ERTS_SMP - ERTS_SMP_BIF_CHK_PENDING_EXIT(BIF_P, ERTS_PROC_LOCK_MAIN); -#else - ERTS_BIF_CHK_EXITED(BIF_P); -#endif - - if (IS_TRACED_FL(BIF_P, F_TRACE_SCHED_PROCS)) { - trace_virtual_sched(BIF_P, am_in); + if (is_internal_port(BIF_ARG_1) || is_atom(BIF_ARG_1)) { + prt = lookup_port(BIF_P, BIF_ARG_1); + if (!prt) + BIF_RET(am_undefined); } - - if (erts_system_profile_flags.runnable_procs && erts_system_profile_flags.exclusive) { - profile_runnable_proc(BIF_P, am_active); + else if (is_external_port(BIF_ARG_1)) { + if (external_port_dist_entry(BIF_ARG_1) == erts_this_dist_entry) + BIF_RET(am_undefined); + else + BIF_RET(am_badarg); } - - if (is_non_value(res)) { - BIF_ERROR(BIF_P, BADARG); + else { + BIF_RET(am_badarg); + } + + switch (erts_port_info(BIF_P, prt, THE_NON_VALUE, &retval)) { + case ERTS_PORT_OP_CALLER_EXIT: + case ERTS_PORT_OP_BADARG: + BIF_RET(am_badarg); + case ERTS_PORT_OP_DROPPED: + BIF_RET(am_undefined); + case ERTS_PORT_OP_SCHEDULED: + ASSERT(is_internal_ref(retval)); + BIF_RET(retval); + case ERTS_PORT_OP_DONE: + ASSERT(is_not_internal_ref(retval)); + BIF_RET(retval); + default: + ERTS_INTERNAL_ERROR("Unexpected erts_port_info() result"); + BIF_RET(am_internal_error); } - BIF_RET(res); } -BIF_RETTYPE port_close_1(BIF_ALIST_1) -{ - Port* p; - erts_smp_proc_unlock(BIF_P, ERTS_PROC_LOCK_MAIN); - p = id_or_name2port(NULL, BIF_ARG_1); - if (!p) { - erts_smp_proc_lock(BIF_P, ERTS_PROC_LOCK_MAIN); - BIF_ERROR(BIF_P, BADARG); - } - erts_do_exit_port(p, p->connected, am_normal); - /* if !ERTS_SMP: since we terminate port with reason normal - we SHOULD never get an exit signal ourselves - */ - erts_port_release(p); - erts_smp_proc_lock(BIF_P, ERTS_PROC_LOCK_MAIN); - BIF_RET(am_true); -} -BIF_RETTYPE port_connect_2(BIF_ALIST_2) +BIF_RETTYPE erts_internal_port_info_2(BIF_ALIST_2) { + Eterm retval; Port* prt; - Process* rp; - Eterm pid = BIF_ARG_2; - if (is_not_internal_pid(pid)) { - error: - BIF_ERROR(BIF_P, BADARG); + if (is_internal_port(BIF_ARG_1) || is_atom(BIF_ARG_1)) { + prt = lookup_port(BIF_P, BIF_ARG_1); + if (!prt) + BIF_RET(am_undefined); } - prt = id_or_name2port(BIF_P, BIF_ARG_1); - if (!prt) { - goto error; + else if (is_external_port(BIF_ARG_1)) { + if (external_port_dist_entry(BIF_ARG_1) == erts_this_dist_entry) + BIF_RET(am_undefined); + else + BIF_RET(am_badarg); } - - rp = erts_pid2proc(BIF_P, ERTS_PROC_LOCK_MAIN, - pid, ERTS_PROC_LOCK_LINK); - if (!rp) { - erts_smp_port_unlock(prt); - ERTS_SMP_ASSERT_IS_NOT_EXITING(BIF_P); - goto error; - } - - erts_add_link(&(rp->nlinks), LINK_PID, prt->id); - erts_add_link(&(prt->nlinks), LINK_PID, pid); - - erts_smp_proc_unlock(rp, ERTS_PROC_LOCK_LINK); - - prt->connected = pid; /* internal pid */ - erts_smp_port_unlock(prt); -#ifdef USE_VM_PROBES - if (DTRACE_ENABLED(port_connect)) { - DTRACE_CHARBUF(process_str, DTRACE_TERM_BUF_SIZE); - DTRACE_CHARBUF(port_str, DTRACE_TERM_BUF_SIZE); - DTRACE_CHARBUF(newprocess_str, DTRACE_TERM_BUF_SIZE); - - dtrace_pid_str(prt->connected, process_str); - erts_snprintf(port_str, sizeof(port_str), "%T", prt->id); - dtrace_proc_str(rp, newprocess_str); - DTRACE4(port_connect, process_str, port_str, prt->name, newprocess_str); + else { + BIF_RET(am_badarg); + } + + switch (erts_port_info(BIF_P, prt, BIF_ARG_2, &retval)) { + case ERTS_PORT_OP_CALLER_EXIT: + case ERTS_PORT_OP_BADARG: + BIF_RET(am_badarg); + case ERTS_PORT_OP_DROPPED: + BIF_RET(am_undefined); + case ERTS_PORT_OP_SCHEDULED: + ASSERT(is_internal_ref(retval)); + BIF_RET(retval); + case ERTS_PORT_OP_DONE: + ASSERT(is_not_internal_ref(retval)); + BIF_RET(retval); + default: + ERTS_INTERNAL_ERROR("Unexpected erts_port_info() result"); + BIF_RET(am_internal_error); } -#endif - BIF_RET(am_true); } -BIF_RETTYPE port_set_data_2(BIF_ALIST_2) + +BIF_RETTYPE erts_internal_port_set_data_2(BIF_ALIST_2) { + Eterm ref; Port* prt; - Eterm portid = BIF_ARG_1; - Eterm data = BIF_ARG_2; - prt = id_or_name2port(BIF_P, portid); - if (!prt) { - BIF_ERROR(BIF_P, BADARG); - } - if (prt->bp != NULL) { - free_message_buffer(prt->bp); - prt->bp = NULL; - } - if (IS_CONST(data)) { - prt->data = data; - } else { - Uint size; - ErlHeapFragment* bp; - Eterm* hp; - - size = size_object(data); - prt->bp = bp = new_message_buffer(size); - hp = bp->mem; - prt->data = copy_struct(data, size, &hp, &bp->off_heap); + prt = lookup_port(BIF_P, BIF_ARG_1); + if (!prt) + BIF_RET(am_badarg); + + switch (erts_port_set_data(BIF_P, prt, BIF_ARG_2, &ref)) { + case ERTS_PORT_OP_CALLER_EXIT: + case ERTS_PORT_OP_BADARG: + case ERTS_PORT_OP_DROPPED: + BIF_RET(am_badarg); + case ERTS_PORT_OP_SCHEDULED: + ASSERT(is_internal_ref(ref)); + BIF_RET(ref); + case ERTS_PORT_OP_DONE: + BIF_RET(am_true); + default: + ERTS_INTERNAL_ERROR("Unexpected erts_port_set_data() result"); + BIF_RET(am_internal_error); } - erts_smp_port_unlock(prt); - BIF_RET(am_true); } -BIF_RETTYPE port_get_data_1(BIF_ALIST_1) +BIF_RETTYPE erts_internal_port_get_data_1(BIF_ALIST_1) { - BIF_RETTYPE res; + Eterm retval; Port* prt; - Eterm portid = BIF_ARG_1; - prt = id_or_name2port(BIF_P, portid); - if (!prt) { - BIF_ERROR(BIF_P, BADARG); - } - if (prt->bp == NULL) { /* MUST be CONST! */ - res = prt->data; - } else { - Eterm* hp = HAlloc(BIF_P, prt->bp->used_size); - res = copy_struct(prt->data, prt->bp->used_size, &hp, &MSO(BIF_P)); + prt = lookup_port(BIF_P, BIF_ARG_1); + if (!prt) + BIF_RET(am_badarg); + + switch (erts_port_get_data(BIF_P, prt, &retval)) { + case ERTS_PORT_OP_CALLER_EXIT: + case ERTS_PORT_OP_BADARG: + case ERTS_PORT_OP_DROPPED: + BIF_RET(am_badarg); + case ERTS_PORT_OP_SCHEDULED: + ASSERT(is_internal_ref(retval)); + BIF_RET(retval); + case ERTS_PORT_OP_DONE: + ASSERT(is_not_internal_ref(retval)); + BIF_RET(retval); + default: + ERTS_INTERNAL_ERROR("Unexpected erts_port_get_data() result"); + BIF_RET(am_internal_error); } - erts_smp_port_unlock(prt); - BIF_RET(res); } /* @@ -625,11 +485,10 @@ BIF_RETTYPE port_get_data_1(BIF_ALIST_1) * either BADARG or SYSTEM_LIMIT). */ -static int -open_port(Process* p, Eterm name, Eterm settings, int *err_nump) +static Port * +open_port(Process* p, Eterm name, Eterm settings, int *err_typep, int *err_nump) { -#define OPEN_PORT_ERROR(VAL) do { port_num = (VAL); goto do_return; } while (0) - int i, port_num; + int i; Eterm option; Uint arity; Eterm* tp; @@ -637,11 +496,11 @@ open_port(Process* p, Eterm name, Eterm settings, int *err_nump) erts_driver_t* driver; char* name_buf = NULL; SysDriverOpts opts; - int binary_io; - int soft_eof; Sint linebuf; Eterm edir = NIL; byte dir[MAXPATHLEN]; + erts_aint32_t sflgs = 0; + Port *port; /* These are the defaults */ opts.packet_bytes = 0; @@ -655,8 +514,7 @@ open_port(Process* p, Eterm name, Eterm settings, int *err_nump) opts.overlapped_io = 0; opts.spawn_type = ERTS_SPAWN_ANY; opts.argv = NULL; - binary_io = 0; - soft_eof = 0; + opts.parallelism = erts_port_parallelism; linebuf = 0; *err_nump = 0; @@ -734,6 +592,13 @@ open_port(Process* p, Eterm name, Eterm settings, int *err_nump) } } else if (option == am_cd) { edir = *tp; + } else if (option == am_parallelism) { + if (*tp == am_true) + opts.parallelism = 1; + else if (*tp == am_false) + opts.parallelism = 0; + else + goto badarg; } else { goto badarg; } @@ -748,13 +613,13 @@ open_port(Process* p, Eterm name, Eterm settings, int *err_nump) } else if (*nargs == am_nouse_stdio) { opts.use_stdio = 0; } else if (*nargs == am_binary) { - binary_io = 1; + sflgs |= ERTS_PORT_SFLG_BINARY_IO; } else if (*nargs == am_in) { opts.read_write |= DO_READ; } else if (*nargs == am_out) { opts.read_write |= DO_WRITE; } else if (*nargs == am_eof) { - soft_eof = 1; + sflgs |= ERTS_PORT_SFLG_SOFT_EOF; } else if (*nargs == am_hide) { opts.hide_window = 1; } else if (*nargs == am_exit_status) { @@ -902,9 +767,9 @@ open_port(Process* p, Eterm name, Eterm settings, int *err_nump) heap[2] = make_small(0); heap[3] = NIL; iolist = make_list(heap); - r = io_list_to_buf(iolist, (char*) dir, MAXPATHLEN); + r = erts_iolist_to_buf(iolist, (char*) dir, MAXPATHLEN); UnUseTmpHeap(4,p); - if (r < 0) { + if (ERTS_IOLIST_TO_BUF_FAILED(r)) { goto badarg; } opts.wd = (char *) dir; @@ -926,44 +791,40 @@ open_port(Process* p, Eterm name, Eterm settings, int *err_nump) erts_smp_proc_unlock(p, ERTS_PROC_LOCK_MAIN); - port_num = erts_open_driver(driver, p->id, name_buf, &opts, err_nump); + port = erts_open_driver(driver, p->common.id, name_buf, &opts, err_typep, err_nump); #ifdef USE_VM_PROBES - if (port_num >= 0 && DTRACE_ENABLED(port_open)) { + if (port && DTRACE_ENABLED(port_open)) { DTRACE_CHARBUF(process_str, DTRACE_TERM_BUF_SIZE); DTRACE_CHARBUF(port_str, DTRACE_TERM_BUF_SIZE); dtrace_proc_str(p, process_str); - erts_snprintf(port_str, sizeof(port_str), "%T", erts_port[port_num].id); + erts_snprintf(port_str, sizeof(port_str), "%T", port->common.id); DTRACE3(port_open, process_str, name_buf, port_str); } #endif erts_smp_proc_lock(p, ERTS_PROC_LOCK_MAIN); - if (port_num < 0) { - DEBUGF(("open_driver returned %d(%d)\n", port_num, *err_nump)); + if (!port) { + DEBUGF(("open_driver returned (%d:%d)\n", + err_typep ? *err_typep : 4711, + err_nump ? *err_nump : 4711)); if (IS_TRACED_FL(p, F_TRACE_SCHED_PROCS)) { trace_virtual_sched(p, am_in); } - OPEN_PORT_ERROR(port_num); + goto do_return; } if (IS_TRACED_FL(p, F_TRACE_SCHED_PROCS)) { trace_virtual_sched(p, am_in); } - if (binary_io) { - erts_port_status_bor_set(&erts_port[port_num], - ERTS_PORT_SFLG_BINARY_IO); - } - if (soft_eof) { - erts_port_status_bor_set(&erts_port[port_num], - ERTS_PORT_SFLG_SOFT_EOF); - } - if (linebuf && erts_port[port_num].linebuf == NULL){ - erts_port[port_num].linebuf = allocate_linebuf(linebuf); - erts_port_status_bor_set(&erts_port[port_num], - ERTS_PORT_SFLG_LINEBUF_IO); + if (linebuf && port->linebuf == NULL){ + port->linebuf = allocate_linebuf(linebuf); + sflgs |= ERTS_PORT_SFLG_LINEBUF_IO; } + + if (sflgs) + erts_atomic32_read_bor_relb(&port->state, sflgs); do_return: if (name_buf) @@ -974,13 +835,15 @@ open_port(Process* p, Eterm name, Eterm settings, int *err_nump) if (opts.wd && opts.wd != ((char *)dir)) { erts_free(ERTS_ALC_T_TMP, (void *) opts.wd); } - return port_num; + return port; badarg: - *err_nump = BADARG; - OPEN_PORT_ERROR(-3); + if (err_typep) + *err_typep = -3; + if (err_nump) + *err_nump = BADARG; + port = NULL; goto do_return; -#undef OPEN_PORT_ERROR } /* Arguments can be given i unicode and as raw binaries, convert filename is used to convert */ |