diff options
| -rw-r--r-- | erts/doc/src/erl_nif.xml | 32 | ||||
| -rw-r--r-- | erts/emulator/beam/erl_nif.c | 23 | ||||
| -rw-r--r-- | erts/emulator/beam/erl_nif_api_funcs.h | 2 | ||||
| -rw-r--r-- | erts/emulator/beam/erl_port.h | 2 | ||||
| -rw-r--r-- | erts/emulator/beam/erl_port_task.c | 30 | ||||
| -rw-r--r-- | erts/emulator/beam/erl_port_task.h | 2 | ||||
| -rw-r--r-- | erts/emulator/beam/io.c | 223 | ||||
| -rw-r--r-- | erts/emulator/test/nif_SUITE.erl | 36 | ||||
| -rw-r--r-- | erts/emulator/test/nif_SUITE_data/nif_SUITE.c | 15 |
9 files changed, 346 insertions, 19 deletions
diff --git a/erts/doc/src/erl_nif.xml b/erts/doc/src/erl_nif.xml index 47d84bb813..81b6eed24a 100644 --- a/erts/doc/src/erl_nif.xml +++ b/erts/doc/src/erl_nif.xml @@ -1464,6 +1464,38 @@ enif_map_iterator_destroy(env, &iter); and <seealso marker="#upgrade">upgrade</seealso>.</p> </desc> </func> + <func><name><ret>int</ret><nametext>enif_port_command(ErlNifEnv* env, const ErlNifPort* to_port, ErlNifEnv *msg_env, ERL_NIF_TERM msg)</nametext></name> + <fsummary>Send a port_command to to_port</fsummary> + <desc> + <p>This function works the same as <seealso marker="erlang#port_command-2">erlang:port_command/2</seealso> + except that it is always completely asynchronous. This call may return false + if it detects that the port is already dead, otherwise it will return true. + </p> + <taglist> + <tag><c>env</c></tag> + <item>The environment of the calling process. May not be NULL.</item> + <tag><c>*to_port</c></tag> + <item>The port id of the receiving port. The port id should refer to a + port on the local node.</item> + <tag><c>msg_env</c></tag> + <item>The environment of the message term. Can be a process + independent environment allocated with + <seealso marker="#enif_alloc_env">enif_alloc_env</seealso> or NULL.</item> + <tag><c>msg</c></tag> + <item>The message term to send. The same limitations apply as on the + payload to <seealso marker="erlang#port_command-2">erlang:port_command/2</seealso>.</item> + </taglist> + <p>Using a <c>msg_env</c> of NULL is an optimization which groups together + calls to <c>enif_alloc_env</c>, <c>enif_make_copy</c>, <c>enif_port_command</c> + and <c>enif_free_env</c> into one call. This optimization is only usefull + when a majority of the terms are to be copied from <c>env</c> to the <c>msg_env</c>.</p> + <p>The call may return false if it detects that the command failed for some reason. Otherwise true is returned.</p> + <p>See also:</p> + <list> + <item><seealso marker="#enif_get_local_port"><c>enif_get_local_port</c></seealso></item> + </list> + </desc> + </func> <func><name><ret>void *</ret><nametext>enif_priv_data(ErlNifEnv* env)</nametext></name> <fsummary>Get the private data of a NIF library</fsummary> <desc><p>Return the pointer to the private data that was set by <c>load</c>, diff --git a/erts/emulator/beam/erl_nif.c b/erts/emulator/beam/erl_nif.c index 21bf752bb6..6ed89780b4 100644 --- a/erts/emulator/beam/erl_nif.c +++ b/erts/emulator/beam/erl_nif.c @@ -375,6 +375,29 @@ int enif_send(ErlNifEnv* env, const ErlNifPid* to_pid, return 1; } +int +enif_port_command(ErlNifEnv *env, const ErlNifPort* to_port, + ErlNifEnv *msg_env, ERL_NIF_TERM msg) +{ + + ErtsSchedulerData *esdp = erts_get_scheduler_data(); + int scheduler = esdp ? esdp->no : 0; + Port *prt; + + if (scheduler == 0 || !env) + return 0; + + prt = erts_port_lookup(to_port->port_id, + (erts_port_synchronous_ops + ? ERTS_PORT_SFLGS_INVALID_DRIVER_LOOKUP + : ERTS_PORT_SFLGS_INVALID_LOOKUP)); + + if (!prt) + return 0; + + return erts_port_output_async(prt, env->proc->common.id, msg); +} + ERL_NIF_TERM enif_make_copy(ErlNifEnv* dst_env, ERL_NIF_TERM src_term) { Uint sz; diff --git a/erts/emulator/beam/erl_nif_api_funcs.h b/erts/emulator/beam/erl_nif_api_funcs.h index 0333e95331..32afb53bfc 100644 --- a/erts/emulator/beam/erl_nif_api_funcs.h +++ b/erts/emulator/beam/erl_nif_api_funcs.h @@ -171,6 +171,7 @@ ERL_NIF_API_FUNC_DECL(int, enif_is_port_alive, (ErlNifEnv *env, ErlNifPort *port ERL_NIF_API_FUNC_DECL(int, enif_get_local_port, (ErlNifEnv* env, ERL_NIF_TERM, ErlNifPort* port_id)); ERL_NIF_API_FUNC_DECL(int, enif_term_to_binary, (ErlNifEnv *env, ERL_NIF_TERM term, ErlNifBinary *bin)); ERL_NIF_API_FUNC_DECL(int, enif_binary_to_term, (ErlNifEnv *env, ErlNifBinary *bin, ERL_NIF_TERM *term)); +ERL_NIF_API_FUNC_DECL(int, enif_port_command, (ErlNifEnv *env, const ErlNifPort* to_port, ErlNifEnv *msg_env, ERL_NIF_TERM msg)); /* ** ADD NEW ENTRIES HERE (before this comment) !!! @@ -334,6 +335,7 @@ ERL_NIF_API_FUNC_DECL(int,enif_is_on_dirty_scheduler,(ErlNifEnv*)); # define enif_get_local_port ERL_NIF_API_FUNC_MACRO(enif_get_local_port) # define enif_term_to_binary ERL_NIF_API_FUNC_MACRO(enif_term_to_binary) # define enif_binary_to_term ERL_NIF_API_FUNC_MACRO(enif_binary_to_term) +# define enif_port_command ERL_NIF_API_FUNC_MACRO(enif_port_command) /* ** ADD NEW ENTRIES HERE (before this comment) diff --git a/erts/emulator/beam/erl_port.h b/erts/emulator/beam/erl_port.h index fa97707a87..dbc6ead216 100644 --- a/erts/emulator/beam/erl_port.h +++ b/erts/emulator/beam/erl_port.h @@ -949,6 +949,8 @@ ErtsPortOpResult erts_port_control(Process *, Port *, unsigned int, Eterm, Eterm ErtsPortOpResult erts_port_call(Process *, Port *, unsigned int, Eterm, Eterm *); ErtsPortOpResult erts_port_info(Process *, Port *, Eterm, Eterm *); +int erts_port_output_async(Port *, Eterm, Eterm); + /* * Signals from ports to ports. Used by sys drivers. */ diff --git a/erts/emulator/beam/erl_port_task.c b/erts/emulator/beam/erl_port_task.c index e85395e062..b200344af5 100644 --- a/erts/emulator/beam/erl_port_task.c +++ b/erts/emulator/beam/erl_port_task.c @@ -180,10 +180,9 @@ p2p_sig_data_to_task(ErtsProc2PortSigData *sigdp) return ptp; } -ErtsProc2PortSigData * -erts_port_task_alloc_p2p_sig_data(void) +static ERTS_INLINE ErtsProc2PortSigData * +p2p_sig_data_init(ErtsPortTask *ptp) { - ErtsPortTask *ptp = port_task_alloc(); ptp->type = ERTS_PORT_TASK_PROC_SIG; ptp->u.alive.flags = ERTS_PT_FLG_SIG_DEP; @@ -194,6 +193,31 @@ erts_port_task_alloc_p2p_sig_data(void) return &ptp->u.alive.td.psig.data; } +ErtsProc2PortSigData * +erts_port_task_alloc_p2p_sig_data(void) +{ + ErtsPortTask *ptp = port_task_alloc(); + + return p2p_sig_data_init(ptp); +} + +ErtsProc2PortSigData * +erts_port_task_alloc_p2p_sig_data_extra(size_t extra, void **extra_ptr) +{ + ErtsPortTask *ptp = erts_alloc(ERTS_ALC_T_PORT_TASK, + sizeof(ErtsPortTask) + extra); + + *extra_ptr = ptp+1; + + return p2p_sig_data_init(ptp); +} + +void +erts_port_task_free_p2p_sig_data(ErtsProc2PortSigData *sigdp) +{ + schedule_port_task_free(p2p_sig_data_to_task(sigdp)); +} + static ERTS_INLINE Eterm task_caller(ErtsPortTask *ptp) { diff --git a/erts/emulator/beam/erl_port_task.h b/erts/emulator/beam/erl_port_task.h index 335f7a77d5..56cef4e352 100644 --- a/erts/emulator/beam/erl_port_task.h +++ b/erts/emulator/beam/erl_port_task.h @@ -269,6 +269,8 @@ int erts_port_task_schedule(Eterm, void erts_port_task_free_port(Port *); int erts_port_is_scheduled(Port *); ErtsProc2PortSigData *erts_port_task_alloc_p2p_sig_data(void); +ErtsProc2PortSigData *erts_port_task_alloc_p2p_sig_data_extra(size_t extra, void **extra_ptr); +void erts_port_task_free_p2p_sig_data(ErtsProc2PortSigData *sigdp); #ifdef ERTS_SMP void erts_enqueue_port(ErtsRunQueue *rq, Port *pp); diff --git a/erts/emulator/beam/io.c b/erts/emulator/beam/io.c index 797e4cdadc..093334aab7 100644 --- a/erts/emulator/beam/io.c +++ b/erts/emulator/beam/io.c @@ -1749,7 +1749,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 @@ -1887,6 +1886,188 @@ port_sig_output(Port *prt, erts_aint32_t state, int op, ErtsProc2PortSigData *si 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, @@ -1896,7 +2077,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; @@ -1978,10 +2159,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); @@ -2010,9 +2194,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; } @@ -2064,8 +2251,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) @@ -2076,8 +2265,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); @@ -2092,9 +2283,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); @@ -2142,7 +2340,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; diff --git a/erts/emulator/test/nif_SUITE.erl b/erts/emulator/test/nif_SUITE.erl index af25af9eeb..bbd523b4ef 100644 --- a/erts/emulator/test/nif_SUITE.erl +++ b/erts/emulator/test/nif_SUITE.erl @@ -45,7 +45,8 @@ nif_monotonic_time/1, nif_time_offset/1, nif_convert_time_unit/1, nif_now_time/1, nif_cpu_time/1, nif_unique_integer/1, nif_is_process_alive/1, nif_is_port_alive/1, - nif_term_to_binary/1, nif_binary_to_term/1 + nif_term_to_binary/1, nif_binary_to_term/1, + nif_port_command/1 ]). -export([many_args_100/100]). @@ -79,7 +80,8 @@ all() -> nif_monotonic_time, nif_time_offset, nif_convert_time_unit, nif_now_time, nif_cpu_time, nif_unique_integer, nif_is_process_alive, nif_is_port_alive, - nif_term_to_binary, nif_binary_to_term + nif_term_to_binary, nif_binary_to_term, + nif_port_command ]. init_per_testcase(_Case, Config) -> @@ -1977,6 +1979,35 @@ nif_binary_to_term(Config) -> true = binary_to_term_nif(Bin, self()), receive T -> ok end. +nif_port_command(Config) -> + ensure_lib_loaded(Config), + + Port = open_port({spawn,"/bin/sh -s unix:cmd"},[stderr_to_stdout,eof]), + true = port_command_nif(Port, "echo hello\n"), + receive {Port,{data,"hello\n"}} -> ok + after 1000 -> ct:fail(timeout) end, + + RefcBin = lists:flatten([lists:duplicate(100, "hello"),"\n"]), + true = port_command_nif(Port, iolist_to_binary(["echo ",RefcBin])), + receive {Port,{data,RefcBin}} -> ok + after 1000 -> ct:fail(timeout) end, + + %% Test that invalid arguments correctly returns + %% badarg and that the port survives. + {'EXIT', {badarg, _}} = (catch port_command_nif(Port, [ok])), + + IoList = [lists:duplicate(100,<<"hello">>),"\n"], + true = port_command_nif(Port, ["echo ",IoList]), + FlatIoList = binary_to_list(iolist_to_binary(IoList)), + receive {Port,{data,FlatIoList}} -> ok + after 1000 -> ct:fail(timeout) end, + + port_close(Port), + + {'EXIT', {badarg, _}} = (catch port_command_nif(Port, "echo hello\n")), + + ok. + %% The NIFs: lib_version() -> undefined. call_history() -> ?nif_stub. @@ -2039,6 +2070,7 @@ is_process_alive_nif(_) -> ?nif_stub. is_port_alive_nif(_) -> ?nif_stub. term_to_binary_nif(_, _) -> ?nif_stub. binary_to_term_nif(_, _) -> ?nif_stub. +port_command_nif(_, _) -> ?nif_stub. %% maps is_map_nif(_) -> ?nif_stub. diff --git a/erts/emulator/test/nif_SUITE_data/nif_SUITE.c b/erts/emulator/test/nif_SUITE_data/nif_SUITE.c index 9db7614405..317f440257 100644 --- a/erts/emulator/test/nif_SUITE_data/nif_SUITE.c +++ b/erts/emulator/test/nif_SUITE_data/nif_SUITE.c @@ -2079,6 +2079,18 @@ static ERL_NIF_TERM binary_to_term(ErlNifEnv* env, int argc, const ERL_NIF_TERM } } +static ERL_NIF_TERM port_command(ErlNifEnv* env, int argc, const ERL_NIF_TERM argv[]) +{ + ErlNifPort port; + + if (!enif_get_local_port(env, argv[0], &port)) + return enif_make_badarg(env); + + if (!enif_port_command(env, &port, NULL, argv[1])) + return enif_make_badarg(env); + return atom_true; +} + static ErlNifFunc nif_funcs[] = { {"lib_version", 0, lib_version}, @@ -2158,7 +2170,8 @@ static ErlNifFunc nif_funcs[] = {"is_process_alive_nif", 1, is_process_alive}, {"is_port_alive_nif", 1, is_port_alive}, {"term_to_binary_nif", 2, term_to_binary}, - {"binary_to_term_nif", 2, binary_to_term} + {"binary_to_term_nif", 2, binary_to_term}, + {"port_command_nif", 2, port_command} }; ERL_NIF_INIT(nif_SUITE,nif_funcs,load,reload,upgrade,unload) |