erts: Add ERL_NIF_SELECT_CUSTOM_MSG

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) ->