From 3b523c25af0df45fbf68ab3cf50c0556f1d4e0a1 Mon Sep 17 00:00:00 2001 From: Rickard Green Date: Mon, 4 Jun 2012 20:47:08 +0200 Subject: Atomic port state --- erts/emulator/beam/erl_bif_port.c | 27 +++++++++------------------ 1 file changed, 9 insertions(+), 18 deletions(-) (limited to 'erts/emulator/beam/erl_bif_port.c') diff --git a/erts/emulator/beam/erl_bif_port.c b/erts/emulator/beam/erl_bif_port.c index 5525426824..1a5b05fcfc 100644 --- a/erts/emulator/beam/erl_bif_port.c +++ b/erts/emulator/beam/erl_bif_port.c @@ -164,7 +164,7 @@ do_port_command(Process *BIF_P, Eterm arg1, Eterm arg2, Eterm arg3, ERTS_BIF_PREP_ERROR(res, BIF_P, EXC_NOTSUP); } else if (!(flags & ERTS_PORT_COMMAND_FLAG_FORCE) - && p->status & ERTS_PORT_SFLG_PORT_BUSY) { + && (erts_smp_atomic32_read_nob(&p->state) & ERTS_PORT_SFLG_PORT_BUSY)) { if (flags & ERTS_PORT_COMMAND_FLAG_NOSUSPEND) { ERTS_BIF_PREP_RET(res, am_false); } @@ -637,11 +637,10 @@ 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; /* These are the defaults */ opts.packet_bytes = 0; @@ -655,8 +654,6 @@ 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; linebuf = 0; *err_nump = 0; @@ -748,13 +745,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) { @@ -951,19 +948,13 @@ open_port(Process* p, Eterm name, Eterm settings, int *err_nump) 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); + erts_port[port_num].linebuf = allocate_linebuf(linebuf); + sflgs |= ERTS_PORT_SFLG_LINEBUF_IO; } + + if (sflgs) + erts_smp_atomic32_read_bor_relb(&erts_port[port_num].state, sflgs); do_return: if (name_buf) -- cgit v1.2.3 From b434a3ab242dde66e23a72122474854f51a61eff Mon Sep 17 00:00:00 2001 From: Rickard Green Date: Wed, 8 Aug 2012 02:20:05 +0200 Subject: Prepare for use of ptab functionality also for ports --- erts/emulator/beam/erl_bif_port.c | 18 +++++++++--------- 1 file changed, 9 insertions(+), 9 deletions(-) (limited to 'erts/emulator/beam/erl_bif_port.c') diff --git a/erts/emulator/beam/erl_bif_port.c b/erts/emulator/beam/erl_bif_port.c index 1a5b05fcfc..2b7760f556 100644 --- a/erts/emulator/beam/erl_bif_port.c +++ b/erts/emulator/beam/erl_bif_port.c @@ -74,9 +74,9 @@ 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_val = erts_port[port_num].common.id; + erts_add_link(&ERTS_P_LINKS(&erts_port[port_num]), LINK_PID, BIF_P->common.id); + erts_add_link(&ERTS_P_LINKS(BIF_P), LINK_PID, port_val); erts_smp_proc_unlock(BIF_P, ERTS_PROC_LOCK_LINK); @@ -171,7 +171,7 @@ do_port_command(Process *BIF_P, Eterm arg1, Eterm arg2, Eterm arg3, 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); + monitor_generic(BIF_P, am_busy_port, p->common.id); } ERTS_BIF_PREP_YIELD3(res, bif_export[BIF_port_command_3], BIF_P, arg1, arg2, arg3); @@ -180,7 +180,7 @@ do_port_command(Process *BIF_P, Eterm arg1, Eterm arg2, Eterm arg3, 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); + wres = erts_write_to_port(BIF_P->common.id, p, arg2); erts_smp_proc_lock(BIF_P, ERTS_PROC_LOCK_MAIN); if (wres != 0) { ERTS_BIF_PREP_ERROR(res, BIF_P, BADARG); @@ -321,7 +321,7 @@ port_call(Process* c_p, Eterm arg1, Eterm arg2, Eterm arg3) if (!term_to_Uint(arg2, &op)) { goto error; } - p->caller = c_p->id; + p->caller = c_p->common.id; /* Lock taken, virtual schedule of port */ if (IS_TRACED_FL(p, F_TRACE_SCHED_PORTS)) { @@ -543,8 +543,8 @@ BIF_RETTYPE port_connect_2(BIF_ALIST_2) goto error; } - erts_add_link(&(rp->nlinks), LINK_PID, prt->id); - erts_add_link(&(prt->nlinks), LINK_PID, pid); + erts_add_link(&ERTS_P_LINKS(rp), LINK_PID, prt->common.id); + erts_add_link(&ERTS_P_LINKS(prt), LINK_PID, pid); erts_smp_proc_unlock(rp, ERTS_PROC_LOCK_LINK); @@ -923,7 +923,7 @@ 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_num = erts_open_driver(driver, p->common.id, name_buf, &opts, err_nump); #ifdef USE_VM_PROBES if (port_num >= 0 && DTRACE_ENABLED(port_open)) { DTRACE_CHARBUF(process_str, DTRACE_TERM_BUF_SIZE); -- cgit v1.2.3 From 50cb7c24f061fd3d7df5970d8202f47c470a4047 Mon Sep 17 00:00:00 2001 From: Rickard Green Date: Wed, 15 Aug 2012 01:28:55 +0200 Subject: Use ptab functionality also for ports --- erts/emulator/beam/erl_bif_port.c | 78 +++++++++++++++++++++------------------ 1 file changed, 43 insertions(+), 35 deletions(-) (limited to 'erts/emulator/beam/erl_bif_port.c') diff --git a/erts/emulator/beam/erl_bif_port.c b/erts/emulator/beam/erl_bif_port.c index 2b7760f556..31151dc9f5 100644 --- a/erts/emulator/beam/erl_bif_port.c +++ b/erts/emulator/beam/erl_bif_port.c @@ -42,7 +42,7 @@ #include "erl_bits.h" #include "dtrace-wrapper.h" -static int open_port(Process* p, Eterm name, Eterm settings, int *err_nump); +static Port *open_port(Process* p, Eterm name, Eterm settings, int *err_typep, int *err_nump); static byte* convert_environment(Process* p, Eterm env); static char **convert_args(Eterm); static void free_args(char **); @@ -54,16 +54,17 @@ port_call(Process* p, Eterm arg1, Eterm arg2, Eterm arg3); BIF_RETTYPE open_port_2(BIF_ALIST_2) { - int port_num; - Eterm port_val; + Port *port; + Eterm port_id; char *str; - int err_num; + int err_type, err_num; - if ((port_num = open_port(BIF_P, BIF_ARG_1, BIF_ARG_2, &err_num)) < 0) { - if (port_num == -3) { + port = open_port(BIF_P, BIF_ARG_1, BIF_ARG_2, &err_type, &err_num); + if (!port) { + if (err_type == -3) { ASSERT(err_num == BADARG || err_num == SYSTEM_LIMIT); BIF_ERROR(BIF_P, err_num); - } else if (port_num == -2) { + } else if (err_type == -2) { str = erl_errno_id(err_num); } else { str = "einval"; @@ -74,15 +75,15 @@ BIF_RETTYPE open_port_2(BIF_ALIST_2) erts_smp_proc_lock(BIF_P, ERTS_PROC_LOCK_LINK); - port_val = erts_port[port_num].common.id; - erts_add_link(&ERTS_P_LINKS(&erts_port[port_num]), LINK_PID, BIF_P->common.id); - erts_add_link(&ERTS_P_LINKS(BIF_P), LINK_PID, port_val); + port_id = port->common.id; + erts_add_link(&ERTS_P_LINKS(port), LINK_PID, BIF_P->common.id); + erts_add_link(&ERTS_P_LINKS(BIF_P), LINK_PID, port_id); erts_smp_proc_unlock(BIF_P, ERTS_PROC_LOCK_LINK); - erts_port_release(&erts_port[port_num]); + erts_port_release(port); - BIF_RET(port_val); + BIF_RET(port_id); } /**************************************************************************** @@ -112,7 +113,10 @@ id_or_name2port(Process *c_p, Eterm id) { Port *port; if (is_not_atom(id)) - port = erts_id2port(id, c_p, ERTS_PROC_LOCK_MAIN); + port = erts_id2port_sflgs(id, + c_p, + ERTS_PROC_LOCK_MAIN, + ERTS_PORT_SFLGS_INVALID_LOOKUP); else erts_whereis_name(c_p, ERTS_PROC_LOCK_MAIN, id, NULL, 0, 0, &port); return port; @@ -164,7 +168,7 @@ do_port_command(Process *BIF_P, Eterm arg1, Eterm arg2, Eterm arg3, ERTS_BIF_PREP_ERROR(res, BIF_P, EXC_NOTSUP); } else if (!(flags & ERTS_PORT_COMMAND_FLAG_FORCE) - && (erts_smp_atomic32_read_nob(&p->state) & ERTS_PORT_SFLG_PORT_BUSY)) { + && (erts_atomic32_read_nob(&p->state) & ERTS_PORT_SFLG_PORT_BUSY)) { if (flags & ERTS_PORT_COMMAND_FLAG_NOSUSPEND) { ERTS_BIF_PREP_RET(res, am_false); } @@ -538,7 +542,7 @@ BIF_RETTYPE port_connect_2(BIF_ALIST_2) rp = erts_pid2proc(BIF_P, ERTS_PROC_LOCK_MAIN, pid, ERTS_PROC_LOCK_LINK); if (!rp) { - erts_smp_port_unlock(prt); + erts_port_release(prt); ERTS_SMP_ASSERT_IS_NOT_EXITING(BIF_P); goto error; } @@ -549,7 +553,7 @@ BIF_RETTYPE port_connect_2(BIF_ALIST_2) erts_smp_proc_unlock(rp, ERTS_PROC_LOCK_LINK); prt->connected = pid; /* internal pid */ - erts_smp_port_unlock(prt); + erts_port_release(prt); #ifdef USE_VM_PROBES if (DTRACE_ENABLED(port_connect)) { DTRACE_CHARBUF(process_str, DTRACE_TERM_BUF_SIZE); @@ -591,7 +595,7 @@ BIF_RETTYPE port_set_data_2(BIF_ALIST_2) hp = bp->mem; prt->data = copy_struct(data, size, &hp, &bp->off_heap); } - erts_smp_port_unlock(prt); + erts_port_release(prt); BIF_RET(am_true); } @@ -612,7 +616,7 @@ BIF_RETTYPE port_get_data_1(BIF_ALIST_1) Eterm* hp = HAlloc(BIF_P, prt->bp->used_size); res = copy_struct(prt->data, prt->bp->used_size, &hp, &MSO(BIF_P)); } - erts_smp_port_unlock(prt); + erts_port_release(prt); BIF_RET(res); } @@ -625,11 +629,10 @@ BIF_RETTYPE port_get_data_1(BIF_ALIST_1) * either BADARG or SYSTEM_LIMIT). */ -static int -open_port(Process* p, Eterm name, Eterm settings, int *err_nump) +static Port * +open_port(Process* p, Eterm name, Eterm settings, int *err_typep, int *err_nump) { -#define OPEN_PORT_ERROR(VAL) do { port_num = (VAL); goto do_return; } while (0) - int i, port_num; + int i; Eterm option; Uint arity; Eterm* tp; @@ -641,6 +644,7 @@ open_port(Process* p, Eterm name, Eterm settings, int *err_nump) Eterm edir = NIL; byte dir[MAXPATHLEN]; erts_aint32_t sflgs = 0; + Port *port; /* These are the defaults */ opts.packet_bytes = 0; @@ -923,38 +927,40 @@ open_port(Process* p, Eterm name, Eterm settings, int *err_nump) erts_smp_proc_unlock(p, ERTS_PROC_LOCK_MAIN); - port_num = erts_open_driver(driver, p->common.id, name_buf, &opts, err_nump); + port = erts_open_driver(driver, p->common.id, name_buf, &opts, err_typep, err_nump); #ifdef USE_VM_PROBES - if (port_num >= 0 && DTRACE_ENABLED(port_open)) { + if (port && DTRACE_ENABLED(port_open)) { DTRACE_CHARBUF(process_str, DTRACE_TERM_BUF_SIZE); DTRACE_CHARBUF(port_str, DTRACE_TERM_BUF_SIZE); dtrace_proc_str(p, process_str); - erts_snprintf(port_str, sizeof(port_str), "%T", erts_port[port_num].id); + erts_snprintf(port_str, sizeof(port_str), "%T", port->common.id); DTRACE3(port_open, process_str, name_buf, port_str); } #endif erts_smp_proc_lock(p, ERTS_PROC_LOCK_MAIN); - if (port_num < 0) { - DEBUGF(("open_driver returned %d(%d)\n", port_num, *err_nump)); + if (!port) { + DEBUGF(("open_driver returned (%d:%d)\n", + err_typep ? *err_typep : 4711, + err_nump ? *err_nump : 4711)); if (IS_TRACED_FL(p, F_TRACE_SCHED_PROCS)) { trace_virtual_sched(p, am_in); } - OPEN_PORT_ERROR(port_num); + goto do_return; } if (IS_TRACED_FL(p, F_TRACE_SCHED_PROCS)) { trace_virtual_sched(p, am_in); } - if (linebuf && erts_port[port_num].linebuf == NULL){ - erts_port[port_num].linebuf = allocate_linebuf(linebuf); + if (linebuf && port->linebuf == NULL){ + port->linebuf = allocate_linebuf(linebuf); sflgs |= ERTS_PORT_SFLG_LINEBUF_IO; } if (sflgs) - erts_smp_atomic32_read_bor_relb(&erts_port[port_num].state, sflgs); + erts_atomic32_read_bor_relb(&port->state, sflgs); do_return: if (name_buf) @@ -965,13 +971,15 @@ open_port(Process* p, Eterm name, Eterm settings, int *err_nump) if (opts.wd && opts.wd != ((char *)dir)) { erts_free(ERTS_ALC_T_TMP, (void *) opts.wd); } - return port_num; + return port; badarg: - *err_nump = BADARG; - OPEN_PORT_ERROR(-3); + if (err_typep) + *err_typep = -3; + if (err_nump) + *err_nump = BADARG; + port = NULL; goto do_return; -#undef OPEN_PORT_ERROR } /* Arguments can be given i unicode and as raw binaries, convert filename is used to convert */ -- cgit v1.2.3 From 34fc6f243f8a462f4b2370a9fe5376df1ca08f1d Mon Sep 17 00:00:00 2001 From: Rickard Green Date: Fri, 14 Sep 2012 21:42:11 +0200 Subject: Move busy port flag --- erts/emulator/beam/erl_bif_port.c | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) (limited to 'erts/emulator/beam/erl_bif_port.c') diff --git a/erts/emulator/beam/erl_bif_port.c b/erts/emulator/beam/erl_bif_port.c index 31151dc9f5..477671f11f 100644 --- a/erts/emulator/beam/erl_bif_port.c +++ b/erts/emulator/beam/erl_bif_port.c @@ -168,7 +168,8 @@ do_port_command(Process *BIF_P, Eterm arg1, Eterm arg2, Eterm arg3, ERTS_BIF_PREP_ERROR(res, BIF_P, EXC_NOTSUP); } else if (!(flags & ERTS_PORT_COMMAND_FLAG_FORCE) - && (erts_atomic32_read_nob(&p->state) & ERTS_PORT_SFLG_PORT_BUSY)) { + && (erts_smp_atomic32_read_nob(&p->sched.flags) + & ERTS_PTS_FLG_BUSY)) { if (flags & ERTS_PORT_COMMAND_FLAG_NOSUSPEND) { ERTS_BIF_PREP_RET(res, am_false); } -- cgit v1.2.3 From 23c6f9e07a3cae7c05e55abd01ff798384241538 Mon Sep 17 00:00:00 2001 From: Rickard Green Date: Sun, 16 Sep 2012 02:45:32 +0200 Subject: Add erl_drv_[send|output]_term --- erts/emulator/beam/erl_bif_port.c | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) (limited to 'erts/emulator/beam/erl_bif_port.c') diff --git a/erts/emulator/beam/erl_bif_port.c b/erts/emulator/beam/erl_bif_port.c index 477671f11f..0d0733b8d0 100644 --- a/erts/emulator/beam/erl_bif_port.c +++ b/erts/emulator/beam/erl_bif_port.c @@ -354,7 +354,7 @@ port_call(Process* c_p, Eterm arg1, Eterm arg2, Eterm arg3) DTRACE_CHARBUF(process_str, DTRACE_TERM_BUF_SIZE); DTRACE_CHARBUF(port_str, DTRACE_TERM_BUF_SIZE); - dtrace_pid_str(p->connected, process_str); + dtrace_pid_str(ERTS_PORT_GET_CONNECTED(p), process_str); dtrace_port_str(p, port_str); DTRACE5(driver_call, process_str, port_str, p->name, op, real_size); } @@ -516,7 +516,7 @@ BIF_RETTYPE port_close_1(BIF_ALIST_1) erts_smp_proc_lock(BIF_P, ERTS_PROC_LOCK_MAIN); BIF_ERROR(BIF_P, BADARG); } - erts_do_exit_port(p, p->connected, am_normal); + erts_do_exit_port(p, ERTS_PORT_GET_CONNECTED(p), am_normal); /* if !ERTS_SMP: since we terminate port with reason normal we SHOULD never get an exit signal ourselves */ @@ -553,7 +553,7 @@ BIF_RETTYPE port_connect_2(BIF_ALIST_2) erts_smp_proc_unlock(rp, ERTS_PROC_LOCK_LINK); - prt->connected = pid; /* internal pid */ + ERTS_PORT_SET_CONNECTED_RELB(prt, pid); /* internal pid */ erts_port_release(prt); #ifdef USE_VM_PROBES if (DTRACE_ENABLED(port_connect)) { @@ -561,7 +561,7 @@ BIF_RETTYPE port_connect_2(BIF_ALIST_2) DTRACE_CHARBUF(port_str, DTRACE_TERM_BUF_SIZE); DTRACE_CHARBUF(newprocess_str, DTRACE_TERM_BUF_SIZE); - dtrace_pid_str(prt->connected, process_str); + dtrace_pid_str(pid, 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); -- cgit v1.2.3 From 6e01408aba71e26884c5db81b8e4fa89bd803576 Mon Sep 17 00:00:00 2001 From: Rickard Green Date: Fri, 21 Sep 2012 15:12:07 +0200 Subject: Implement true asynchronous signaling between processes and ports --- erts/emulator/beam/erl_bif_port.c | 791 ++++++++++++++++---------------------- 1 file changed, 327 insertions(+), 464 deletions(-) (limited to 'erts/emulator/beam/erl_bif_port.c') diff --git a/erts/emulator/beam/erl_bif_port.c b/erts/emulator/beam/erl_bif_port.c index 0d0733b8d0..97f147132e 100644 --- a/erts/emulator/beam/erl_bif_port.c +++ b/erts/emulator/beam/erl_bif_port.c @@ -49,9 +49,6 @@ 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) { Port *port; @@ -86,539 +83,397 @@ BIF_RETTYPE open_port_2(BIF_ALIST_2) 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_sflgs(id, - c_p, - ERTS_PROC_LOCK_MAIN, - ERTS_PORT_SFLGS_INVALID_LOOKUP); + /* 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) - && (erts_smp_atomic32_read_nob(&p->sched.flags) - & ERTS_PTS_FLG_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->common.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->common.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; /* To satisfy hybrid heap architecture */ - 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); + ERTS_BIF_EXITED(BIF_P); } - 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->common.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); - } - 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(ERTS_PORT_GET_CONNECTED(p), 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) + +/* + * erts_internal:port_connect/2 is used by the + * erlang:port_connect/2 BIF. + */ +BIF_RETTYPE erts_internal_port_connect_2(BIF_ALIST_2) { - 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); - } + Eterm ref; + Port* prt; - if (erts_system_profile_flags.runnable_procs && erts_system_profile_flags.exclusive) { - profile_runnable_proc(BIF_P, am_inactive); - } + prt = lookup_port(BIF_P, BIF_ARG_1); + if (!prt) + BIF_RET(am_badarg); - 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); - } - - /* 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, ERTS_PORT_GET_CONNECTED(p), 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); - } - prt = id_or_name2port(BIF_P, BIF_ARG_1); - if (!prt) { - goto error; + 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); } - - rp = erts_pid2proc(BIF_P, ERTS_PROC_LOCK_MAIN, - pid, ERTS_PROC_LOCK_LINK); - if (!rp) { - erts_port_release(prt); - ERTS_SMP_ASSERT_IS_NOT_EXITING(BIF_P); - 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); } - - erts_add_link(&ERTS_P_LINKS(rp), LINK_PID, prt->common.id); - erts_add_link(&ERTS_P_LINKS(prt), LINK_PID, pid); - - erts_smp_proc_unlock(rp, ERTS_PROC_LOCK_LINK); - - ERTS_PORT_SET_CONNECTED_RELB(prt, pid); /* internal pid */ - erts_port_release(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(pid, 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_port_release(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_port_release(prt); - BIF_RET(res); } /* @@ -659,6 +514,7 @@ open_port(Process* p, Eterm name, Eterm settings, int *err_typep, int *err_nump) opts.overlapped_io = 0; opts.spawn_type = ERTS_SPAWN_ANY; opts.argv = NULL; + opts.parallelism = erts_port_parallelism; linebuf = 0; *err_nump = 0; @@ -736,6 +592,13 @@ open_port(Process* p, Eterm name, Eterm settings, int *err_typep, 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; } @@ -904,9 +767,9 @@ open_port(Process* p, Eterm name, Eterm settings, int *err_typep, 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; @@ -1035,7 +898,7 @@ static byte* convert_environment(Process* p, Eterm env) Eterm* hp; Uint heap_size; int n; - Uint size; + ErlDrvSizeT size; byte* bytes; if ((n = list_length(env)) < 0) { @@ -1087,7 +950,7 @@ static byte* convert_environment(Process* p, Eterm env) * Put the result in a binary (no risk for a memory leak that way). */ (void) erts_new_heap_binary(p, NULL, size, &bytes); - io_list_to_buf(all, (char*)bytes, size); + erts_iolist_to_buf(all, (char*)bytes, size); done: erts_free(ERTS_ALC_T_TMP, temp_heap); -- cgit v1.2.3