From 619514492966ba6b7c6e6fee807329c81c8bf7a8 Mon Sep 17 00:00:00 2001 From: Sverker Eriksson Date: Thu, 6 Dec 2018 20:35:59 +0100 Subject: erts: Add ERL_NIF_SELECT_CUSTOM_MSG --- erts/emulator/beam/erl_drv_nif.h | 3 +- erts/emulator/beam/erl_message.c | 4 +- erts/emulator/sys/common/erl_check_io.c | 183 ++++++++++++++++++-------------- erts/emulator/sys/common/erl_check_io.h | 3 +- erts/emulator/test/nif_SUITE.erl | 70 ++++++------ 5 files changed, 146 insertions(+), 117 deletions(-) diff --git a/erts/emulator/beam/erl_drv_nif.h b/erts/emulator/beam/erl_drv_nif.h index 9ef7c39d41..a5ecbfff06 100644 --- a/erts/emulator/beam/erl_drv_nif.h +++ b/erts/emulator/beam/erl_drv_nif.h @@ -54,7 +54,8 @@ enum ErlNifSelectFlags { ERL_NIF_SELECT_READ = (1 << 0), ERL_NIF_SELECT_WRITE = (1 << 1), ERL_NIF_SELECT_STOP = (1 << 2), - ERL_NIF_SELECT_CANCEL = (1 << 3) + ERL_NIF_SELECT_CANCEL = (1 << 3), + ERL_NIF_SELECT_CUSTOM_MSG= (1 << 4) }; /* diff --git a/erts/emulator/beam/erl_message.c b/erts/emulator/beam/erl_message.c index a3274d7443..942bec84cf 100644 --- a/erts/emulator/beam/erl_message.c +++ b/erts/emulator/beam/erl_message.c @@ -532,9 +532,7 @@ erts_try_alloc_message_on_heap(Process *pp, if ((*psp) & ERTS_PSFLGS_VOLATILE_HEAP) goto in_message_fragment; - else if ( - *plp & ERTS_PROC_LOCK_MAIN - ) { + else if (*plp & ERTS_PROC_LOCK_MAIN) { try_on_heap: if (((*psp) & ERTS_PSFLGS_VOLATILE_HEAP) || (pp->flags & F_DISABLE_GC) diff --git a/erts/emulator/sys/common/erl_check_io.c b/erts/emulator/sys/common/erl_check_io.c index b4609007c9..487b9cba40 100644 --- a/erts/emulator/sys/common/erl_check_io.c +++ b/erts/emulator/sys/common/erl_check_io.c @@ -591,6 +591,90 @@ abort_tasks(ErtsDrvEventState *state, int mode) } } +static void prepare_select_msg(struct erts_nif_select_event* e, + enum ErlNifSelectFlags mode, + Eterm recipient, + ErtsResource* resource, + Eterm msg, + Eterm event_atom) +{ + ErtsMessage* mp; + Eterm* hp; + Eterm* hp_start; + Uint hsz; + + if (is_not_nil(e->pid)) { + ASSERT(e->mp); + erts_cleanup_messages(e->mp); + } + + if (mode & ERL_NIF_SELECT_CUSTOM_MSG) { + hsz = size_object(msg); + mp = erts_alloc_message(hsz, &hp); + hp_start = hp; + ERL_MESSAGE_TERM(mp) = copy_struct(msg, hsz, &hp, &mp->hfrag.off_heap); + } + else { + ErtsBinary* bin; + Eterm resource_term, ref_term, tuple; + + /* {select, Resource, Ref, EventAtom} */ + hsz = 5 + ERTS_MAGIC_REF_THING_SIZE; + if (is_internal_ref(msg)) + hsz += ERTS_REF_THING_SIZE; + else + ASSERT(is_immed(msg)); + + mp = erts_alloc_message(hsz, &hp); + hp_start = hp; + + bin = ERTS_MAGIC_BIN_FROM_UNALIGNED_DATA(resource); + resource_term = erts_mk_magic_ref(&hp, &mp->hfrag.off_heap, &bin->binary); + if (is_internal_ref(msg)) { + Uint32* refn = internal_ref_numbers(msg); + write_ref_thing(hp, refn[0], refn[1], refn[2]); + ref_term = make_internal_ref(hp); + hp += ERTS_REF_THING_SIZE; + } + else { + ASSERT(is_immed(msg)); + ref_term = msg; + } + tuple = TUPLE4(hp, am_select, resource_term, ref_term, event_atom); + hp += 5; + ERL_MESSAGE_TERM(mp) = tuple; + } + ASSERT(hp == hp_start + hsz); (void)hp_start; + + ASSERT(is_not_nil(recipient)); + e->pid = recipient; + e->mp = mp; +} + +static ERTS_INLINE void send_select_msg(struct erts_nif_select_event* e) +{ + Process* rp = erts_proc_lookup(e->pid); + + ASSERT(is_internal_pid(e->pid)); + if (!rp) { + erts_cleanup_messages(e->mp); + return; + } + + erts_queue_message(rp, 0, e->mp, ERL_MESSAGE_TERM(e->mp), am_system); +} + +static void clear_select_event(struct erts_nif_select_event* e) +{ + if (is_not_nil(e->pid)) { + /* Discard unsent message */ + ASSERT(e->mp); + erts_cleanup_messages(e->mp); + e->mp = NULL; + e->pid = NIL; + } +} + static void deselect(ErtsDrvEventState *state, int mode) { @@ -621,8 +705,8 @@ deselect(ErtsDrvEventState *state, int mode) erts_io_control(state, ERTS_POLL_OP_DEL, 0); switch (state->type) { case ERTS_EV_TYPE_NIF: - state->driver.nif->in.pid = NIL; - state->driver.nif->out.pid = NIL; + clear_select_event(&state->driver.nif->in); + clear_select_event(&state->driver.nif->out); enif_release_resource(state->driver.stop.resource->data); state->driver.stop.resource = NULL; break; @@ -948,7 +1032,7 @@ enif_select(ErlNifEnv* env, enum ErlNifSelectFlags mode, void* obj, const ErlNifPid* pid, - Eterm ref) + Eterm msg) { int on; ErtsResource* resource = DATA_TO_RESOURCE(obj); @@ -966,7 +1050,7 @@ enif_select(ErlNifEnv* env, #ifdef ERTS_SYS_CONTINOUS_FD_NUMBERS if (!grow_drv_ev_state(fd)) { - if (fd > 0) nif_select_large_fd_error(fd, mode, resource, ref); + if (fd > 0) nif_select_large_fd_error(fd, mode, resource, msg); return INT_MIN | ERL_NIF_SELECT_INVALID_EVENT; } #endif @@ -1012,21 +1096,21 @@ enif_select(ErlNifEnv* env, * Changing process and/or ref is ok (I think?). */ if (state->driver.stop.resource != resource) - nif_select_steal(state, ERL_DRV_READ | ERL_DRV_WRITE, resource, ref); + nif_select_steal(state, ERL_DRV_READ | ERL_DRV_WRITE, resource, msg); break; case ERTS_EV_TYPE_DRV_SEL: - nif_select_steal(state, mode, resource, ref); + nif_select_steal(state, mode, resource, msg); break; case ERTS_EV_TYPE_STOP_USE: { erts_dsprintf_buf_t *dsbufp = erts_create_logger_dsbuf(); - print_nif_select_op(dsbufp, fd, mode, resource, ref); + print_nif_select_op(dsbufp, fd, mode, resource, msg); steal_pending_stop_use(dsbufp, ERTS_INVALID_ERL_DRV_PORT, state, mode, on); ASSERT(state->type == ERTS_EV_TYPE_NONE); break; } case ERTS_EV_TYPE_STOP_NIF: { erts_dsprintf_buf_t *dsbufp = erts_create_logger_dsbuf(); - print_nif_select_op(dsbufp, fd, mode, resource, ref); + print_nif_select_op(dsbufp, fd, mode, resource, msg); steal_pending_stop_nif(dsbufp, resource, state, mode, on); if (state->type == ERTS_EV_TYPE_STOP_NIF) { ret = ERL_NIF_SELECT_STOP_SCHEDULED; /* ?? */ @@ -1082,7 +1166,6 @@ enif_select(ErlNifEnv* env, if (on) { const Eterm recipient = pid ? pid->pid : env->proc->common.id; - Uint32* refn; if (!state->driver.nif) state->driver.nif = alloc_nif_select_data(); if (state->type == ERTS_EV_TYPE_NONE) { @@ -1093,28 +1176,12 @@ enif_select(ErlNifEnv* env, ASSERT(state->type == ERTS_EV_TYPE_NIF); ASSERT(state->driver.stop.resource == resource); if (mode & ERL_DRV_READ) { - state->driver.nif->in.pid = recipient; - if (is_immed(ref)) { - state->driver.nif->in.immed = ref; - } else { - ASSERT(is_internal_ref(ref)); - refn = internal_ref_numbers(ref); - state->driver.nif->in.immed = THE_NON_VALUE; - sys_memcpy(state->driver.nif->in.refn, refn, - sizeof(state->driver.nif->in.refn)); - } + prepare_select_msg(&state->driver.nif->in, mode, recipient, + resource, msg, am_ready_input); } if (mode & ERL_DRV_WRITE) { - state->driver.nif->out.pid = recipient; - if (is_immed(ref)) { - state->driver.nif->out.immed = ref; - } else { - ASSERT(is_internal_ref(ref)); - refn = internal_ref_numbers(ref); - state->driver.nif->out.immed = THE_NON_VALUE; - sys_memcpy(state->driver.nif->out.refn, refn, - sizeof(state->driver.nif->out.refn)); - } + prepare_select_msg(&state->driver.nif->out, mode, recipient, + resource, msg, am_ready_output); } ret = 0; } @@ -1123,12 +1190,12 @@ enif_select(ErlNifEnv* env, if (state->type == ERTS_EV_TYPE_NIF) { if (mode & ERL_NIF_SELECT_READ && is_not_nil(state->driver.nif->in.pid)) { - state->driver.nif->in.pid = NIL; + clear_select_event(&state->driver.nif->in); ret |= ERL_NIF_SELECT_READ_CANCELLED; } if (mode & ERL_NIF_SELECT_WRITE && is_not_nil(state->driver.nif->out.pid)) { - state->driver.nif->out.pid = NIL; + clear_select_event(&state->driver.nif->out); ret |= ERL_NIF_SELECT_WRITE_CANCELLED; } } @@ -1545,53 +1612,6 @@ oready(Eterm id, ErtsDrvEventState *state) } } -static ERTS_INLINE void -send_event_tuple(struct erts_nif_select_event* e, ErtsResource* resource, - Eterm event_atom) -{ - Process* rp = erts_proc_lookup(e->pid); - ErtsProcLocks rp_locks = 0; - ErtsMessage* mp; - ErlOffHeap* ohp; - ErtsBinary* bin; - Eterm* hp; - Uint hsz; - Eterm resource_term, ref_term, tuple; - - if (!rp) { - return; - } - - bin = ERTS_MAGIC_BIN_FROM_UNALIGNED_DATA(resource); - - /* {select, Resource, Ref, EventAtom} */ - if (is_value(e->immed)) { - hsz = 5 + ERTS_MAGIC_REF_THING_SIZE; - } - else { - hsz = 5 + ERTS_MAGIC_REF_THING_SIZE + ERTS_REF_THING_SIZE; - } - - mp = erts_alloc_message_heap(rp, &rp_locks, hsz, &hp, &ohp); - - resource_term = erts_mk_magic_ref(&hp, ohp, &bin->binary); - if (is_value(e->immed)) { - ASSERT(is_immed(e->immed)); - ref_term = e->immed; - } - else { - write_ref_thing(hp, e->refn[0], e->refn[1], e->refn[2]); - ref_term = make_internal_ref(hp); - hp += ERTS_REF_THING_SIZE; - } - tuple = TUPLE4(hp, am_select, resource_term, ref_term, event_atom); - - erts_queue_message(rp, rp_locks, mp, tuple, am_system); - - if (rp_locks) - erts_proc_unlock(rp, rp_locks); -} - static void bad_fd_in_pollset(ErtsDrvEventState *, Eterm inport, Eterm outport); void @@ -1765,7 +1785,6 @@ erts_check_io(ErtsPollThread *psi, ErtsMonotonicTime timeout_time) case ERTS_EV_TYPE_NIF: { /* Requested via enif_select()... */ struct erts_nif_select_event in = {NIL}; struct erts_nif_select_event out = {NIL}; - ErtsResource* resource = NULL; if (revents & (ERTS_POLL_EV_IN|ERTS_POLL_EV_OUT)) { if (revents & ERTS_POLL_EV_OUT) { @@ -1773,6 +1792,7 @@ erts_check_io(ErtsPollThread *psi, ErtsMonotonicTime timeout_time) out = state->driver.nif->out; resource = state->driver.stop.resource; state->driver.nif->out.pid = NIL; + state->driver.nif->out.mp = NULL; } } if (revents & ERTS_POLL_EV_IN) { @@ -1780,6 +1800,7 @@ erts_check_io(ErtsPollThread *psi, ErtsMonotonicTime timeout_time) in = state->driver.nif->in; resource = state->driver.stop.resource; state->driver.nif->in.pid = NIL; + state->driver.nif->in.mp = NULL; } } state->events &= ~revents; @@ -1792,10 +1813,10 @@ erts_check_io(ErtsPollThread *psi, ErtsMonotonicTime timeout_time) erts_mtx_unlock(fd_mtx(fd)); if (is_not_nil(in.pid)) { - send_event_tuple(&in, resource, am_ready_input); + send_select_msg(&in); } if (is_not_nil(out.pid)) { - send_event_tuple(&out, resource, am_ready_output); + send_select_msg(&out); } continue; } diff --git a/erts/emulator/sys/common/erl_check_io.h b/erts/emulator/sys/common/erl_check_io.h index 31182be5ec..0f3fc4f7a2 100644 --- a/erts/emulator/sys/common/erl_check_io.h +++ b/erts/emulator/sys/common/erl_check_io.h @@ -138,8 +138,7 @@ typedef struct { struct erts_nif_select_event { Eterm pid; - Eterm immed; - Uint32 refn[ERTS_REF_NUMBERS]; + ErtsMessage *mp; }; typedef struct { diff --git a/erts/emulator/test/nif_SUITE.erl b/erts/emulator/test/nif_SUITE.erl index edad62a9fb..8de6d256c5 100644 --- a/erts/emulator/test/nif_SUITE.erl +++ b/erts/emulator/test/nif_SUITE.erl @@ -486,6 +486,7 @@ t_on_load(Config) when is_list(Config) -> -define(ERL_NIF_SELECT_WRITE, (1 bsl 1)). -define(ERL_NIF_SELECT_STOP, (1 bsl 2)). -define(ERL_NIF_SELECT_CANCEL, (1 bsl 3)). +-define(ERL_NIF_SELECT_CUSTOM_MSG, (1 bsl 4)). -define(ERL_NIF_SELECT_STOP_CALLED, (1 bsl 0)). -define(ERL_NIF_SELECT_STOP_SCHEDULED, (1 bsl 1)). @@ -496,33 +497,38 @@ t_on_load(Config) when is_list(Config) -> select(Config) when is_list(Config) -> ensure_lib_loaded(Config), + select_do(0, make_ref(), make_ref()), - Ref = make_ref(), - Ref2 = make_ref(), + RefBin = list_to_binary(lists:duplicate(100, $x)), + select_do(?ERL_NIF_SELECT_CUSTOM_MSG, + small, {a, tuple, with, "some", RefBin}), + ok. + +select_do(Flag, Ref, Ref2) -> {{R, R_ptr}, {W, W_ptr}} = pipe_nif(), ok = write_nif(W, <<"hej">>), <<"hej">> = read_nif(R, 3), %% Wait for read eagain = read_nif(R, 3), - 0 = select_nif(R,?ERL_NIF_SELECT_READ,R,null,Ref), + 0 = select_nif(R,?ERL_NIF_SELECT_READ bor Flag, R,null,Ref), [] = flush(0), ok = write_nif(W, <<"hej">>), - [{select, R, Ref, ready_input}] = flush(), - 0 = select_nif(R,?ERL_NIF_SELECT_READ,R,self(),Ref2), - [{select, R, Ref2, ready_input}] = flush(), + receive_ready(R, Ref, ready_input), + 0 = select_nif(R,?ERL_NIF_SELECT_READ bor Flag,R,self(),Ref2), + receive_ready(R, Ref2, ready_input), Papa = self(), Pid = spawn_link(fun() -> - [{select, R, Ref, ready_input}] = flush(), + receive_ready(R, Ref, ready_input), Papa ! {self(), done} end), - 0 = select_nif(R,?ERL_NIF_SELECT_READ,R,Pid,Ref), + 0 = select_nif(R, ?ERL_NIF_SELECT_READ bor Flag, R, Pid, Ref), {Pid, done} = receive_any(1000), %% Cancel read 0 = select_nif(R,?ERL_NIF_SELECT_READ bor ?ERL_NIF_SELECT_CANCEL,R,null,Ref), <<"hej">> = read_nif(R, 3), - 0 = select_nif(R,?ERL_NIF_SELECT_READ,R,null,Ref), + 0 = select_nif(R, ?ERL_NIF_SELECT_READ bor Flag, R, null, Ref), ?ERL_NIF_SELECT_READ_CANCELLED = select_nif(R,?ERL_NIF_SELECT_READ bor ?ERL_NIF_SELECT_CANCEL,R,null,Ref), ok = write_nif(W, <<"hej again">>), @@ -531,65 +537,63 @@ select(Config) when is_list(Config) -> %% Wait for write Written = write_full(W, $a), - 0 = select_nif(W,?ERL_NIF_SELECT_WRITE,W,self(),Ref), + 0 = select_nif(W, ?ERL_NIF_SELECT_WRITE bor Flag, W, self(), Ref), [] = flush(0), Written = read_nif(R,byte_size(Written)), - [{select, W, Ref, ready_output}] = flush(), + receive_ready(W, Ref, ready_output), %% Cancel write - 0 = select_nif(W,?ERL_NIF_SELECT_WRITE bor ?ERL_NIF_SELECT_CANCEL,W,null,Ref), + 0 = select_nif(W, ?ERL_NIF_SELECT_WRITE bor ?ERL_NIF_SELECT_CANCEL, W, null, Ref), Written2 = write_full(W, $b), - 0 = select_nif(W,?ERL_NIF_SELECT_WRITE,W,null,Ref), + 0 = select_nif(W, ?ERL_NIF_SELECT_WRITE bor Flag, W, null, Ref), ?ERL_NIF_SELECT_WRITE_CANCELLED = - select_nif(W,?ERL_NIF_SELECT_WRITE bor ?ERL_NIF_SELECT_CANCEL,W,null,Ref), + select_nif(W, ?ERL_NIF_SELECT_WRITE bor ?ERL_NIF_SELECT_CANCEL, W, null, Ref), Written2 = read_nif(R,byte_size(Written2)), [] = flush(0), %% Close write and wait for EOF eagain = read_nif(R, 1), - check_stop_ret(select_nif(W,?ERL_NIF_SELECT_STOP,W,null,Ref)), + check_stop_ret(select_nif(W, ?ERL_NIF_SELECT_STOP, W, null, Ref)), [{fd_resource_stop, W_ptr, _}] = flush(), {1, {W_ptr,_}} = last_fd_stop_call(), true = is_closed_nif(W), [] = flush(0), - 0 = select_nif(R,?ERL_NIF_SELECT_READ,R,self(),Ref), - [{select, R, Ref, ready_input}] = flush(), + 0 = select_nif(R, ?ERL_NIF_SELECT_READ bor Flag, R, self(), Ref), + receive_ready(R, Ref, ready_input), eof = read_nif(R,1), - check_stop_ret(select_nif(R,?ERL_NIF_SELECT_STOP,R,null,Ref)), + check_stop_ret(select_nif(R, ?ERL_NIF_SELECT_STOP, R, null, Ref)), [{fd_resource_stop, R_ptr, _}] = flush(), {1, {R_ptr,_}} = last_fd_stop_call(), true = is_closed_nif(R), - select_2(Config). + select_2(Flag, Ref, Ref2). -select_2(Config) -> +select_2(Flag, Ref1, Ref2) -> erlang:garbage_collect(), {_,_,2} = last_resource_dtor_call(), - Ref1 = make_ref(), - Ref2 = make_ref(), {{R, R_ptr}, {W, W_ptr}} = pipe_nif(), %% Change ref eagain = read_nif(R, 1), - 0 = select_nif(R,?ERL_NIF_SELECT_READ,R,null,Ref1), - 0 = select_nif(R,?ERL_NIF_SELECT_READ,R,self(),Ref2), + 0 = select_nif(R, ?ERL_NIF_SELECT_READ bor Flag, R, null, Ref1), + 0 = select_nif(R, ?ERL_NIF_SELECT_READ bor Flag, R, self(), Ref2), [] = flush(0), ok = write_nif(W, <<"hej">>), - [{select, R, Ref2, ready_input}] = flush(), + receive_ready(R, Ref2, ready_input), <<"hej">> = read_nif(R, 3), %% Change pid eagain = read_nif(R, 1), - 0 = select_nif(R,?ERL_NIF_SELECT_READ,R,null,Ref1), + 0 = select_nif(R, ?ERL_NIF_SELECT_READ bor Flag, R, null, Ref1), Papa = self(), spawn_link(fun() -> - 0 = select_nif(R,?ERL_NIF_SELECT_READ,R,null,Ref1), + 0 = select_nif(R, ?ERL_NIF_SELECT_READ bor Flag, R, null, Ref1), [] = flush(0), Papa ! sync, - [{select, R, Ref1, ready_input}] = flush(), + receive_ready(R, Ref1, ready_input), <<"hej">> = read_nif(R, 3), Papa ! done end), @@ -609,13 +613,19 @@ select_2(Config) -> {1, {W_ptr,1}} = last_fd_stop_call(), true = is_closed_nif(W), - select_3(Config). + select_3(). -select_3(_Config) -> +select_3() -> erlang:garbage_collect(), {_,_,2} = last_resource_dtor_call(), ok. +receive_ready(R, Ref, IOatom) when is_reference(Ref) -> + [{select, R, Ref, IOatom}] = flush(); +receive_ready(_, Msg, _) -> + [Got] = flush(), + {true,_,_} = {Got=:=Msg, Got, Msg}. + %% @doc The stealing child process for the select_steal test. Duplicates given %% W/RFds and runs select on them to steal select_steal_child_process(Parent, RFd) -> -- cgit v1.2.3