Merge PR-2084 from sverker/erts/enif_select-custom-msg OTP-15349

Custom message format for enif_select

11 files changed, 271 insertions, 151 deletions

diff --git a/erts/doc/src/erl_nif.xml b/erts/doc/src/erl_nif.xml
index 3fe6e00d57..34042cb4de 100644
--- a/erts/doc/src/erl_nif.xml
+++ b/erts/doc/src/erl_nif.xml
@@ -3063,6 +3063,11 @@ enif_map_iterator_destroy(env, &amp;iter);</code>
 	<code type="none">{select, Obj, Ref, ready_input | ready_output}</code>
 	<p><c>ready_input</c> or <c>ready_output</c> indicates if the event object
 	  is ready for reading or writing.</p>
+	<note><p>For complete control over the message format use the newer functions
+	  <seealso marker="#enif_select_read"><c>enif_select_read</c></seealso> or
+	  <seealso marker="#enif_select_write"><c>enif_select_write</c></seealso>
+	  introduced in erts-11.0 (OTP-22.0).</p>
+	</note>
 	<p>Argument <c>pid</c> may be <c>NULL</c> to indicate the calling process.</p>
 	<p>Argument <c>obj</c> is a resource object obtained from
           <seealso marker="#enif_alloc_resource"><c>enif_alloc_resource</c></seealso>.
@@ -3157,6 +3162,34 @@ if (retval &amp; ERL_NIF_SELECT_STOP_CALLED) {
     </func>
 
     <func>
+      <name since="OTP 22.0"><ret>int</ret>
+      <nametext>enif_select_read(ErlNifEnv* env, ErlNifEvent event, void* obj,
+      const ErlNifPid* pid, ERL_NIF_TERM msg, ErlNifEnv* msg_env)</nametext>
+      </name>
+      <name since="OTP 22.0"><ret>int</ret>
+      <nametext>enif_select_write(ErlNifEnv* env, ErlNifEvent event, void* obj,
+      const ErlNifPid* pid, ERL_NIF_TERM msg, ErlNifEnv* msg_env)</nametext>
+      </name>
+      <fsummary>Manage subscription on IO event.</fsummary>
+      <desc>
+	<p>These are variants of <seealso marker="#enif_select">enif_select</seealso>
+	where you can supply your own message term <c>msg</c> that will be sent to
+	the process instead of the predefined tuple <c>{select,_,_,_}.</c></p>
+        <p>Argument <c>msg_env</c> must either be <c>NULL</c> or the environment of
+	<c>msg</c> allocated with <seealso marker="#enif_alloc_env">
+	<c>enif_alloc_env</c></seealso>. If argument <c>msg_env</c> is
+	<c>NULL</c> the term <c>msg</c> will be copied, otherwise both
+	<c>msg</c> and <c>msg_env</c> will be invalidated by a successful call
+	to <c>enif_select_read</c> or <c>enif_select_write</c>.</p>
+	<p>Apart from the message format <c>enif_select_read</c> and
+	<c>enif_select_write</c> behaves exactly the same as <seealso
+	marker="#enif_select">enif_select</seealso> with argument <c>mode</c> as
+	either <c>ERL_NIF_SELECT_READ</c> or <c>ERL_NIF_SELECT_WRITE</c>. To
+	cancel or close events use <seealso marker="#enif_select">enif_select</seealso>.</p>
+      </desc>
+    </func>
+
+    <func>
       <name since="OTP R14B"><ret>ErlNifPid *</ret>
         <nametext>enif_self(ErlNifEnv* caller_env, ErlNifPid* pid)</nametext>
       </name>
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/beam/erl_nif.c b/erts/emulator/beam/erl_nif.c
index 7339aa8874..a48d0391f6 100644
--- a/erts/emulator/beam/erl_nif.c
+++ b/erts/emulator/beam/erl_nif.c
@@ -707,6 +707,29 @@ error:
     return reds;
 }
 
+/** @brief Create a message with the content of process independent \c msg_env.
+ *  Invalidates \c msg_env.
+ */
+ErtsMessage* erts_create_message_from_nif_env(ErlNifEnv* msg_env)
+{
+    struct enif_msg_environment_t* menv = (struct enif_msg_environment_t*)msg_env;
+    ErtsMessage* mp;
+
+    flush_env(msg_env);
+    mp = erts_alloc_message(0, NULL);
+    mp->data.heap_frag = menv->env.heap_frag;
+    ASSERT(mp->data.heap_frag == MBUF(&menv->phony_proc));
+    if (mp->data.heap_frag != NULL) {
+        /* Move all offheap's from phony proc to the first fragment.
+           Quick and dirty... */
+        ASSERT(!is_offheap(&mp->data.heap_frag->off_heap));
+        mp->data.heap_frag->off_heap = MSO(&menv->phony_proc);
+        clear_offheap(&MSO(&menv->phony_proc));
+        menv->env.heap_frag = NULL;
+        MBUF(&menv->phony_proc) = NULL;
+    }
+    return mp;
+}
 
 int enif_send(ErlNifEnv* env, const ErlNifPid* to_pid,
 	      ErlNifEnv* msg_env, ERL_NIF_TERM msg)
@@ -803,20 +826,8 @@ int enif_send(ErlNifEnv* env, const ErlNifPid* to_pid,
             }
 #endif
         }
-        flush_env(msg_env);
-        mp = erts_alloc_message(0, NULL);
+        mp = erts_create_message_from_nif_env(msg_env);
         ERL_MESSAGE_TOKEN(mp) = token;
-        mp->data.heap_frag = menv->env.heap_frag;
-        ASSERT(mp->data.heap_frag == MBUF(&menv->phony_proc));
-        if (mp->data.heap_frag != NULL) {
-            /* Move all offheap's from phony proc to the first fragment.
-               Quick and dirty... */
-            ASSERT(!is_offheap(&mp->data.heap_frag->off_heap));
-            mp->data.heap_frag->off_heap = MSO(&menv->phony_proc);
-            clear_offheap(&MSO(&menv->phony_proc));
-            menv->env.heap_frag = NULL;
-            MBUF(&menv->phony_proc) = NULL;
-        }
     } else {
         erts_literal_area_t litarea;
 	ErlOffHeap *ohp;
diff --git a/erts/emulator/beam/erl_nif.h b/erts/emulator/beam/erl_nif.h
index 58a217c20b..3fd1a8fd4c 100644
--- a/erts/emulator/beam/erl_nif.h
+++ b/erts/emulator/beam/erl_nif.h
@@ -54,7 +54,7 @@
 ** 2.13: 20.1 add enif_ioq
 ** 2.14: 21.0 add enif_ioq_peek_head, enif_(mutex|cond|rwlock|thread)_name
 **                enif_vfprintf, enif_vsnprintf, enif_make_map_from_arrays
-** 2.15: 22.0 ERL_NIF_SELECT_CANCEL
+** 2.15: 22.0 ERL_NIF_SELECT_CANCEL, enif_select_(read|write)
 */
 #define ERL_NIF_MAJOR_VERSION 2
 #define ERL_NIF_MINOR_VERSION 15
diff --git a/erts/emulator/beam/erl_nif_api_funcs.h b/erts/emulator/beam/erl_nif_api_funcs.h
index 81f64f2390..129166562d 100644
--- a/erts/emulator/beam/erl_nif_api_funcs.h
+++ b/erts/emulator/beam/erl_nif_api_funcs.h
@@ -210,6 +210,9 @@ ERL_NIF_API_FUNC_DECL(int,enif_vsnprintf,(char*, size_t, const char *fmt, va_lis
 
 ERL_NIF_API_FUNC_DECL(int,enif_make_map_from_arrays,(ErlNifEnv *env, ERL_NIF_TERM keys[], ERL_NIF_TERM values[], size_t cnt, ERL_NIF_TERM *map_out));
 
+ERL_NIF_API_FUNC_DECL(int,enif_select_x,(ErlNifEnv* env, ErlNifEvent e, enum ErlNifSelectFlags flags, void* obj, const ErlNifPid* pid, ERL_NIF_TERM msg, ErlNifEnv* msg_env));
+
+
 /*
 ** ADD NEW ENTRIES HERE (before this comment) !!!
 */
@@ -392,6 +395,7 @@ ERL_NIF_API_FUNC_DECL(int,enif_make_map_from_arrays,(ErlNifEnv *env, ERL_NIF_TER
 #  define enif_vfprintf ERL_NIF_API_FUNC_MACRO(enif_vfprintf)
 #  define enif_vsnprintf ERL_NIF_API_FUNC_MACRO(enif_vsnprintf)
 #  define enif_make_map_from_arrays ERL_NIF_API_FUNC_MACRO(enif_make_map_from_arrays)
+#  define enif_select_x ERL_NIF_API_FUNC_MACRO(enif_select_x)
 
 /*
 ** ADD NEW ENTRIES HERE (before this comment)
@@ -623,6 +627,12 @@ static ERL_NIF_INLINE ERL_NIF_TERM enif_make_list9(ErlNifEnv* env,
 #ifndef enif_make_pid
 
 #  define enif_make_pid(ENV, PID) ((void)(ENV),(const ERL_NIF_TERM)((PID)->pid))
+#  define enif_select_read(ENV, E, OBJ, PID, MSG, MSG_ENV) \
+    enif_select_x(ENV, E, ERL_NIF_SELECT_READ | ERL_NIF_SELECT_CUSTOM_MSG, \
+                  OBJ, PID, MSG, MSG_ENV)
+#  define enif_select_write(ENV, E, OBJ, PID, MSG, MSG_ENV) \
+    enif_select_x(ENV, E, ERL_NIF_SELECT_WRITE | ERL_NIF_SELECT_CUSTOM_MSG, \
+                  OBJ, PID, MSG, MSG_ENV)
 
 #if SIZEOF_LONG == 8
 #  define enif_get_int64 enif_get_long
diff --git a/erts/emulator/beam/global.h b/erts/emulator/beam/global.h
index 77b5a3ca05..9eb7b58dbb 100644
--- a/erts/emulator/beam/global.h
+++ b/erts/emulator/beam/global.h
@@ -131,6 +131,7 @@ extern Eterm erts_nif_call_function(Process *p, Process *tracee,
 
 int erts_call_dirty_nif(ErtsSchedulerData *esdp, Process *c_p,
 			BeamInstr *I, Eterm *reg);
+ErtsMessage* erts_create_message_from_nif_env(ErlNifEnv* msg_env);
 
 
 /* Driver handle (wrapper for old plain handle) */
diff --git a/erts/emulator/sys/common/erl_check_io.c b/erts/emulator/sys/common/erl_check_io.c
index b4609007c9..304e0a5d0c 100644
--- a/erts/emulator/sys/common/erl_check_io.c
+++ b/erts/emulator/sys/common/erl_check_io.c
@@ -591,6 +591,96 @@ 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,
+                               ErlNifEnv* msg_env,
+                               Eterm event_atom)
+{
+    ErtsMessage* mp;
+    Eterm* hp;
+    Uint hsz;
+
+    if (is_not_nil(e->pid)) {
+        ASSERT(e->mp);
+        erts_cleanup_messages(e->mp);
+    }
+
+    if (mode & ERL_NIF_SELECT_CUSTOM_MSG) {
+        if (msg_env) {
+            mp = erts_create_message_from_nif_env(msg_env);
+            ERL_MESSAGE_TERM(mp) = msg;
+        }
+        else {
+            hsz = size_object(msg);
+            mp = erts_alloc_message(hsz, &hp);
+            ERL_MESSAGE_TERM(mp) = copy_struct(msg, hsz, &hp, &mp->hfrag.off_heap);
+        }
+    }
+    else {
+        ErtsBinary* bin;
+        Eterm resource_term, ref_term, tuple;
+        Eterm* hp_start;
+
+         /* {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 +711,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;
@@ -943,12 +1033,21 @@ done_unknown:
 }
 
 int
-enif_select(ErlNifEnv* env,
-            ErlNifEvent e,
-            enum ErlNifSelectFlags mode,
-            void* obj,
-            const ErlNifPid* pid,
-            Eterm ref)
+enif_select(ErlNifEnv* env, ErlNifEvent e, enum ErlNifSelectFlags mode,
+            void* obj, const ErlNifPid* pid, Eterm msg)
+{
+    return enif_select_x(env, e, mode, obj, pid, msg, NULL);
+}
+
+
+int
+enif_select_x(ErlNifEnv* env,
+              ErlNifEvent e,
+              enum ErlNifSelectFlags mode,
+              void* obj,
+              const ErlNifPid* pid,
+              Eterm msg,
+              ErlNifEnv* msg_env)
 {
     int on;
     ErtsResource* resource = DATA_TO_RESOURCE(obj);
@@ -966,7 +1065,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 +1111,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 +1181,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 +1191,13 @@ 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, msg_env, am_ready_input);
+            msg_env = NULL;
         }
         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, msg_env, am_ready_output);
         }
         ret = 0;
     }
@@ -1123,12 +1206,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 +1628,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 +1801,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 +1808,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 +1816,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 +1829,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..0d0930e124 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,100 +497,106 @@ 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(), null),
+
+    RefBin = list_to_binary(lists:duplicate(100, $x)),
+    [select_do(?ERL_NIF_SELECT_CUSTOM_MSG,
+               small, {a, tuple, with, "some", RefBin}, MSG_ENV)
+     || MSG_ENV <- [null, alloc_env]],
+    ok.
+
+select_do(Flag, Ref, Ref2, MSG_ENV) ->
+    io:format("select_do(~p, ~p, ~p)\n", [Ref, Ref2, MSG_ENV]),
 
-    Ref = make_ref(),
-    Ref2 = make_ref(),
     {{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,MSG_ENV),
     [] = 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,MSG_ENV),
+    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,MSG_ENV),
     {Pid, done} = receive_any(1000),
 
     %% Cancel read
-    0 = select_nif(R,?ERL_NIF_SELECT_READ bor ?ERL_NIF_SELECT_CANCEL,R,null,Ref),
+    0 = select_nif(R,?ERL_NIF_SELECT_READ bor ?ERL_NIF_SELECT_CANCEL,R,null,Ref,null),
     <<"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, MSG_ENV),
     ?ERL_NIF_SELECT_READ_CANCELLED =
-        select_nif(R,?ERL_NIF_SELECT_READ bor ?ERL_NIF_SELECT_CANCEL,R,null,Ref),
+        select_nif(R,?ERL_NIF_SELECT_READ bor ?ERL_NIF_SELECT_CANCEL,R,null,Ref,null),
     ok = write_nif(W, <<"hej again">>),
     [] = flush(0),
     <<"hej again">> = read_nif(R, 9),
 
     %% 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, MSG_ENV),
     [] = 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, null),
     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, MSG_ENV),
     ?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, null),
     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, null)),
     [{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, MSG_ENV),
+    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, null)),
     [{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, MSG_ENV).
 
-select_2(Config) ->
+select_2(Flag, Ref1, Ref2, MSG_ENV) ->
     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, MSG_ENV),
+    0 = select_nif(R, ?ERL_NIF_SELECT_READ bor Flag, R, self(), Ref2, MSG_ENV),
 
     [] = 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, MSG_ENV),
     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, MSG_ENV),
                        [] = flush(0),
                        Papa ! sync,
-                       [{select, R, Ref1, ready_input}] = flush(),
+                       receive_ready(R, Ref1, ready_input),
                        <<"hej">> = read_nif(R, 3),
                        Papa ! done
                end),
@@ -598,24 +605,30 @@ select_2(Config) ->
     done = receive_any(),
     [] = flush(0),
 
-    check_stop_ret(select_nif(R,?ERL_NIF_SELECT_STOP,R,null,Ref1)),
+    check_stop_ret(select_nif(R,?ERL_NIF_SELECT_STOP,R,null,Ref1, null)),
     [{fd_resource_stop, R_ptr, _}] = flush(),
     {1, {R_ptr,_}} = last_fd_stop_call(),
     true = is_closed_nif(R),
 
     %% Stop without previous read/write select
-    ?ERL_NIF_SELECT_STOP_CALLED = select_nif(W,?ERL_NIF_SELECT_STOP,W,null,Ref1),
+    ?ERL_NIF_SELECT_STOP_CALLED = select_nif(W,?ERL_NIF_SELECT_STOP,W,null,Ref1,null),
     [{fd_resource_stop, W_ptr, 1}] = flush(),
     {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) ->
@@ -624,7 +637,7 @@ select_steal_child_process(Parent, RFd) ->
     Ref2 = make_ref(),
 
     %% Try to select from the child pid (steal from parent)
-    ?assertEqual(0, select_nif(R2Fd, ?ERL_NIF_SELECT_READ, R2Fd, null, Ref2)),
+    ?assertEqual(0, select_nif(R2Fd, ?ERL_NIF_SELECT_READ, R2Fd, null, Ref2, null)),
     ?assertEqual([], flush(0)),
     ?assertEqual(eagain, read_nif(R2Fd, 1)),
 
@@ -632,7 +645,7 @@ select_steal_child_process(Parent, RFd) ->
     Parent ! {self(), stage1}, % signal parent to send the <<"stolen1">>
 
     %% Receive <<"stolen1">> via enif_select
-    ?assertEqual(0, select_nif(R2Fd, ?ERL_NIF_SELECT_READ, R2Fd, null, Ref2)),
+    ?assertEqual(0, select_nif(R2Fd, ?ERL_NIF_SELECT_READ, R2Fd, null, Ref2, null)),
     ?assertMatch([{select, R2Fd, Ref2, ready_input}], flush()),
     ?assertEqual(<<"stolen1">>, read_nif(R2Fd, 7)),
 
@@ -650,7 +663,7 @@ select_steal(Config) when is_list(Config) ->
     {{RFd, RPtr}, {WFd, WPtr}} = pipe_nif(),
 
     %% Bind the socket to current pid in enif_select
-    ?assertEqual(0, select_nif(RFd, ?ERL_NIF_SELECT_READ, RFd, null, Ref)),
+    ?assertEqual(0, select_nif(RFd, ?ERL_NIF_SELECT_READ, RFd, null, Ref, null)),
     ?assertEqual([], flush(0)),
 
     %% Spawn a process and do some stealing
@@ -664,15 +677,15 @@ select_steal(Config) when is_list(Config) ->
     ?assertMatch([{Pid, done}], flush(1)),  % synchronize with the child
 
     %% Try to select from the parent pid (steal back)
-    ?assertEqual(0, select_nif(RFd, ?ERL_NIF_SELECT_READ, RFd, Pid, Ref)),
+    ?assertEqual(0, select_nif(RFd, ?ERL_NIF_SELECT_READ, RFd, Pid, Ref, null)),
 
     %% Ensure that no data is hanging and close.
     %% Rfd is stolen at this point.
-    check_stop_ret(select_nif(WFd, ?ERL_NIF_SELECT_STOP, WFd, null, Ref)),
+    check_stop_ret(select_nif(WFd, ?ERL_NIF_SELECT_STOP, WFd, null, Ref, null)),
     ?assertMatch([{fd_resource_stop, WPtr, _}], flush()),
     {1, {WPtr, 1}} = last_fd_stop_call(),
 
-    check_stop_ret(select_nif(RFd, ?ERL_NIF_SELECT_STOP, RFd, null, Ref)),
+    check_stop_ret(select_nif(RFd, ?ERL_NIF_SELECT_STOP, RFd, null, Ref, null)),
     ?assertMatch([{fd_resource_stop, RPtr, _}], flush()),
     {1, {RPtr, _DirectCall}} = last_fd_stop_call(),
 
@@ -3396,7 +3409,7 @@ term_to_binary_nif(_, _) -> ?nif_stub.
 binary_to_term_nif(_, _, _) -> ?nif_stub.
 port_command_nif(_, _) -> ?nif_stub.
 format_term_nif(_,_) -> ?nif_stub.
-select_nif(_,_,_,_,_) -> ?nif_stub.
+select_nif(_,_,_,_,_,_) -> ?nif_stub.
 dupe_resource_nif(_) -> ?nif_stub.
 pipe_nif() -> ?nif_stub.
 write_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 f2ce6dbe67..21af4b05b3 100644
--- a/erts/emulator/test/nif_SUITE_data/nif_SUITE.c
+++ b/erts/emulator/test/nif_SUITE_data/nif_SUITE.c
@@ -2486,7 +2486,8 @@ static ERL_NIF_TERM select_nif(ErlNifEnv* env, int argc, const ERL_NIF_TERM argv
     enum ErlNifSelectFlags mode;
     void* obj;
     ErlNifPid nifpid, *pid = NULL;
-    ERL_NIF_TERM ref;
+    ERL_NIF_TERM ref_or_msg;
+    ErlNifEnv* msg_env = NULL;
     int retval;
 
     if (!get_fd(env, argv[0], &fdr)
@@ -2501,11 +2502,27 @@ static ERL_NIF_TERM select_nif(ErlNifEnv* env, int argc, const ERL_NIF_TERM argv
 	    return enif_make_badarg(env);
 	pid = &nifpid;
     }
-    ref = argv[4];
+    ref_or_msg = argv[4];
+    if (argv[5] != atom_null) {
+        msg_env = enif_alloc_env();
+        ref_or_msg = enif_make_copy(msg_env, ref_or_msg);
+    }
 
     fdr->was_selected = 1;
     enif_self(env, &fdr->pid);
-    retval = enif_select(env, fdr->fd, mode, obj, pid, ref);
+    switch (mode) {
+    case ERL_NIF_SELECT_CUSTOM_MSG | ERL_NIF_SELECT_READ:
+        retval = enif_select_read(env, fdr->fd, obj, pid, ref_or_msg, msg_env);
+        break;
+    case ERL_NIF_SELECT_CUSTOM_MSG | ERL_NIF_SELECT_WRITE:
+        retval = enif_select_write(env, fdr->fd, obj, pid, ref_or_msg, msg_env);
+        break;
+    default:
+        retval = enif_select(env, fdr->fd, mode, obj, pid, ref_or_msg);
+    }
+
+    if (msg_env)
+        enif_free_env(msg_env);
 
     return enif_make_int(env, retval);
 }
@@ -3565,7 +3582,7 @@ static ErlNifFunc nif_funcs[] =
     {"binary_to_term_nif", 3, binary_to_term},
     {"port_command_nif", 2, port_command},
     {"format_term_nif", 2, format_term},
-    {"select_nif", 5, select_nif},
+    {"select_nif", 6, select_nif},
 #ifndef __WIN32__
     {"pipe_nif", 0, pipe_nif},
     {"write_nif", 2, write_nif},