erts: Add pid argument to enif_select

7 files changed, 64 insertions, 37 deletions

diff --git a/erts/doc/src/erl_nif.xml b/erts/doc/src/erl_nif.xml
index bd72bdb691..9800a530f2 100644
--- a/erts/doc/src/erl_nif.xml
+++ b/erts/doc/src/erl_nif.xml
@@ -2584,8 +2584,8 @@ enif_map_iterator_destroy(env, &amp;iter);</code>
 
     <func>
       <name><ret>enum ErlNifSelectReturn</ret>
-        <nametext>enif_select(ErlNifEnv* env, ErlNifEvent event,
-	                      enum ErlNifSelectFlags mode, void* obj, Eterm ref)</nametext>
+        <nametext>enif_select(ErlNifEnv* env, ErlNifEvent event, enum ErlNifSelectFlags mode,
+	void* obj, const ErlNifPid* pid, ERL_NIF_TERM ref)</nametext>
       </name>
       <fsummary>Manage subscription on IO event.</fsummary>
       <desc>
@@ -2599,11 +2599,12 @@ enif_map_iterator_destroy(env, &amp;iter);</code>
 	  <c>ERL_NIF_SELECT_READ</c>, <c>ERL_NIF_SELECT_WRITE</c> or a bitwise
 	  OR combination to wait for both. It can also be <c>ERL_NIF_SELECT_STOP</c>
 	  which is described further below. When a read or write event is triggerred,
-	  a notification message like this is sent to the Erlang process that called
-	  <c>enif_select</c>:</p>
+	  a notification message like this is sent to the process identified by
+	  <c>pid</c>:</p>
 	<code type="none"><c>{select, Obj, Ref, ready_input | ready_output}</c></code>
 	<p><c>ready_input</c> or <c>ready_output</c> indicates if the event object
 	  is ready for reading or writing.</p>
+	<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>.
 	  The purpose of the resource objects is as a container of the event object
@@ -2616,7 +2617,8 @@ enif_map_iterator_destroy(env, &amp;iter);</code>
 	  then a reference created just before the <c>receive</c> will exploit a runtime
 	  optimization that bypasses all earlier received messages in the queue.</p>
 	<p>The notifications are one-shot only. To receive further notifications of the same
-	  type (read or write), repeated calls to <c>enif_select</c> must be made.</p>
+	  type (read or write), repeated calls to <c>enif_select</c> must be made
+	  after receiving each notification.</p>
 	<p>Use <c>ERL_NIF_SELECT_STOP</c> as <c>mode</c> in order to safely
 	  close an event object that has been passed to <c>enif_select</c>. The
 	  <seealso marker="#ErlNifResourceStop"><c>stop</c></seealso> callback
diff --git a/erts/emulator/beam/erl_nif_api_funcs.h b/erts/emulator/beam/erl_nif_api_funcs.h
index a9ed246962..1fc6842acc 100644
--- a/erts/emulator/beam/erl_nif_api_funcs.h
+++ b/erts/emulator/beam/erl_nif_api_funcs.h
@@ -175,7 +175,7 @@ ERL_NIF_API_FUNC_DECL(size_t, enif_binary_to_term, (ErlNifEnv *env, const unsign
 ERL_NIF_API_FUNC_DECL(int, enif_port_command, (ErlNifEnv *env, const ErlNifPort* to_port, ErlNifEnv *msg_env, ERL_NIF_TERM msg));
 ERL_NIF_API_FUNC_DECL(int,enif_thread_type,(void));
 ERL_NIF_API_FUNC_DECL(int,enif_snprintf,(char * buffer, size_t size, const char *format, ...));
-ERL_NIF_API_FUNC_DECL(int,enif_select,(ErlNifEnv* env, ErlNifEvent e, enum ErlNifSelectFlags flags, void* obj, ERL_NIF_TERM ref));
+ERL_NIF_API_FUNC_DECL(int,enif_select,(ErlNifEnv* env, ErlNifEvent e, enum ErlNifSelectFlags flags, void* obj, const ErlNifPid* pid, ERL_NIF_TERM ref));
 ERL_NIF_API_FUNC_DECL(ErlNifResourceType*,enif_open_resource_type_x,(ErlNifEnv*, const char* name_str, const ErlNifResourceTypeInit*, ErlNifResourceFlags flags, ErlNifResourceFlags* tried));
 ERL_NIF_API_FUNC_DECL(int, enif_monitor_process,(ErlNifEnv*,void* obj,const ErlNifPid*,ErlDrvMonitor *monitor));
 ERL_NIF_API_FUNC_DECL(int, enif_demonitor_process,(ErlNifEnv*,void* obj,const ErlDrvMonitor *monitor));
diff --git a/erts/emulator/sys/common/erl_check_io.c b/erts/emulator/sys/common/erl_check_io.c
index c8d958e72f..089f10fd8e 100644
--- a/erts/emulator/sys/common/erl_check_io.c
+++ b/erts/emulator/sys/common/erl_check_io.c
@@ -1203,10 +1203,10 @@ ERTS_CIO_EXPORT(enif_select)(ErlNifEnv* env,
                              ErlNifEvent e,
                              enum ErlNifSelectFlags mode,
                              void* obj,
+                             const ErlNifPid* pid,
                              Eterm ref)
 {
     int on;
-    const Eterm id = env->proc->common.id;
     ErtsResource* resource = DATA_TO_RESOURCE(obj);
     ErtsSysFdType fd = (ErtsSysFdType) e;
     ErtsPollEvents ctl_events = (ErtsPollEvents) 0;
@@ -1342,6 +1342,7 @@ ERTS_CIO_EXPORT(enif_select)(ErlNifEnv* env,
 
     state->events = new_events;
     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();
@@ -1353,7 +1354,7 @@ ERTS_CIO_EXPORT(enif_select)(ErlNifEnv* env,
         ASSERT(state->type == ERTS_EV_TYPE_NIF);
         ASSERT(state->driver.stop.resource == resource);
         if (ctl_events & ERTS_POLL_EV_IN) {
-            state->driver.nif->in.pid = id;
+            state->driver.nif->in.pid = recipient;
             if (is_immed(ref)) {
                 state->driver.nif->in.immed = ref;
             } else {
@@ -1367,7 +1368,7 @@ ERTS_CIO_EXPORT(enif_select)(ErlNifEnv* env,
             state->driver.nif->in.ddeselect_cnt = 0;
         }
         if (ctl_events & ERTS_POLL_EV_OUT) {
-            state->driver.nif->out.pid = id;
+            state->driver.nif->out.pid = recipient;
             if (is_immed(ref)) {
                 state->driver.nif->out.immed = ref;
             } else {
diff --git a/erts/emulator/sys/common/erl_check_io.h b/erts/emulator/sys/common/erl_check_io.h
index 41e2e9210a..4f9efeefcd 100644
--- a/erts/emulator/sys/common/erl_check_io.h
+++ b/erts/emulator/sys/common/erl_check_io.h
@@ -34,8 +34,8 @@
 
 int driver_select_kp(ErlDrvPort, ErlDrvEvent, int, int);
 int driver_select_nkp(ErlDrvPort, ErlDrvEvent, int, int);
-enum ErlNifSelectReturn enif_select_kp(ErlNifEnv*, ErlNifEvent, enum ErlNifSelectFlags, void*, Eterm);
-enum ErlNifSelectReturn enif_select_nkp(ErlNifEnv*, ErlNifEvent, enum ErlNifSelectFlags, void*, Eterm);
+enum ErlNifSelectReturn enif_select_kp(ErlNifEnv*, ErlNifEvent, enum ErlNifSelectFlags, void*, const ErlNifPid*, Eterm);
+enum ErlNifSelectReturn enif_select_nkp(ErlNifEnv*, ErlNifEvent, enum ErlNifSelectFlags, void*, const ErlNifPid*, Eterm);
 int driver_event_kp(ErlDrvPort, ErlDrvEvent, ErlDrvEventData);
 int driver_event_nkp(ErlDrvPort, ErlDrvEvent, ErlDrvEventData);
 Uint erts_check_io_size_kp(void);
diff --git a/erts/emulator/sys/unix/sys.c b/erts/emulator/sys/unix/sys.c
index a852550915..4843ee4ba2 100644
--- a/erts/emulator/sys/unix/sys.c
+++ b/erts/emulator/sys/unix/sys.c
@@ -161,7 +161,7 @@ int erts_use_kernel_poll = 0;
 
 struct {
     int (*select)(ErlDrvPort, ErlDrvEvent, int, int);
-    enum ErlNifSelectReturn (*enif_select)(ErlNifEnv*, ErlNifEvent, enum ErlNifSelectFlags, void*, Eterm);
+    enum ErlNifSelectReturn (*enif_select)(ErlNifEnv*, ErlNifEvent, enum ErlNifSelectFlags, void*, const ErlNifPid*, Eterm);
     int (*event)(ErlDrvPort, ErlDrvEvent, ErlDrvEventData);
     void (*check_io_as_interrupt)(void);
     void (*check_io_interrupt)(int);
@@ -186,9 +186,9 @@ driver_event(ErlDrvPort port, ErlDrvEvent event, ErlDrvEventData event_data)
 }
 
 int enif_select(ErlNifEnv* env, ErlNifEvent event,
-                enum ErlNifSelectFlags flags, void* obj, Eterm ref)
+                enum ErlNifSelectFlags flags, void* obj, const ErlNifPid* pid, Eterm ref)
 {
-    return (*io_func.enif_select)(env, event, flags, obj, ref);
+    return (*io_func.enif_select)(env, event, flags, obj, pid, ref);
 }
 
 
diff --git a/erts/emulator/test/nif_SUITE.erl b/erts/emulator/test/nif_SUITE.erl
index e3af9f7454..de26e73c3d 100644
--- a/erts/emulator/test/nif_SUITE.erl
+++ b/erts/emulator/test/nif_SUITE.erl
@@ -468,21 +468,31 @@ select(Config) when is_list(Config) ->
     ensure_lib_loaded(Config),
 
     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,Ref),
+    0 = select_nif(R,?ERL_NIF_SELECT_READ,R,null,Ref),
     [] = flush(),
     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(),
+    Papa = self(),
+    Pid = spawn_link(fun() ->
+                             [{select, R, Ref, ready_input}] = flush(),
+                             Papa ! {self(), done}
+                     end),
+    0 = select_nif(R,?ERL_NIF_SELECT_READ,R,Pid,Ref),
+    {Pid, done} = receive_any(1000),
     <<"hej">> = read_nif(R, 3),
 
     %% Wait for write
     Written = write_full(W, $a),
-    0 = select_nif(W,?ERL_NIF_SELECT_WRITE,W,Ref),
+    0 = select_nif(W,?ERL_NIF_SELECT_WRITE,W,self(),Ref),
     [] = flush(),
     Half = byte_size(Written) div 2,
     <<First:Half/binary,Second/binary>> = Written,
@@ -494,16 +504,16 @@ select(Config) when is_list(Config) ->
 
     %% Close write and wait for EOF
     eagain = read_nif(R, 1),
-    check_stop_ret(select_nif(W,?ERL_NIF_SELECT_STOP,W,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 = select_nif(R,?ERL_NIF_SELECT_READ,R,Ref),
+    0 = select_nif(R,?ERL_NIF_SELECT_READ,R,self(),Ref),
     [{select, R, Ref, ready_input}] = flush(),
     eof = read_nif(R,1),
 
-    check_stop_ret(select_nif(R,?ERL_NIF_SELECT_STOP,R,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),
@@ -520,8 +530,8 @@ select_2(Config) ->
 
     %% Change ref
     eagain = read_nif(R, 1),
-    0 = select_nif(R,?ERL_NIF_SELECT_READ,R,Ref1),
-    0 = select_nif(R,?ERL_NIF_SELECT_READ,R,Ref2),
+    0 = select_nif(R,?ERL_NIF_SELECT_READ,R,null,Ref1),
+    0 = select_nif(R,?ERL_NIF_SELECT_READ,R,self(),Ref2),
 
     [] = flush(),
     ok = write_nif(W, <<"hej">>),
@@ -530,35 +540,35 @@ select_2(Config) ->
 
     %% Change pid
     eagain = read_nif(R, 1),
-    0 = select_nif(R,?ERL_NIF_SELECT_READ,R,Ref1),
+    0 = select_nif(R,?ERL_NIF_SELECT_READ,R,null,Ref1),
     Papa = self(),
-    Pid2 = spawn_link(fun() ->
-                              0 = select_nif(R,?ERL_NIF_SELECT_READ,R,Ref1),
-                              [] = flush(),
-                              Papa ! sync,
-                              [{select, R, Ref1, ready_input}] = flush(),
-                              <<"hej">> = read_nif(R, 3),
-                              Papa ! done
-                      end),
+    spawn_link(fun() ->
+                       0 = select_nif(R,?ERL_NIF_SELECT_READ,R,null,Ref1),
+                       [] = flush(),
+                       Papa ! sync,
+                       [{select, R, Ref1, ready_input}] = flush(),
+                       <<"hej">> = read_nif(R, 3),
+                       Papa ! done
+               end),
     sync = receive_any(),
     ok = write_nif(W, <<"hej">>),
     done = receive_any(),
     [] = flush(),
 
-    check_stop_ret(select_nif(R,?ERL_NIF_SELECT_STOP,R,Ref1)),
+    check_stop_ret(select_nif(R,?ERL_NIF_SELECT_STOP,R,null,Ref1)),
     [{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,Ref1),
+    ?ERL_NIF_SELECT_STOP_CALLED = select_nif(W,?ERL_NIF_SELECT_STOP,W,null,Ref1),
     [{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(Config) ->
+select_3(_Config) ->
     erlang:garbage_collect(),
     {_,_,2} = last_resource_dtor_call(),
     ok.
@@ -2263,6 +2273,10 @@ call(Pid,Cmd) ->
 receive_any() ->
     receive M -> M end.	     
 
+receive_any(Timeout) ->
+    receive M -> M
+    after Timeout -> timeout end.
+
 flush() ->
     flush(10).
 flush(Timeout) ->
@@ -2635,7 +2649,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.
 pipe_nif() -> ?nif_stub.
 write_nif(_,_) -> ?nif_stub.
 read_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 ee925512d2..6cf02c6efe 100644
--- a/erts/emulator/test/nif_SUITE_data/nif_SUITE.c
+++ b/erts/emulator/test/nif_SUITE_data/nif_SUITE.c
@@ -75,6 +75,7 @@ static ERL_NIF_TERM atom_init;
 static ERL_NIF_TERM atom_stats;
 static ERL_NIF_TERM atom_done;
 static ERL_NIF_TERM atom_stop;
+static ERL_NIF_TERM atom_null;
 
 typedef struct
 {
@@ -242,6 +243,7 @@ static int load(ErlNifEnv* env, void** priv_data, ERL_NIF_TERM load_info)
     atom_stats = enif_make_atom(env,"stats");
     atom_done = enif_make_atom(env,"done");
     atom_stop = enif_make_atom(env,"stop");
+    atom_null = enif_make_atom(env,"null");
 
     *priv_data = data;
     return 0;
@@ -2119,6 +2121,7 @@ static ERL_NIF_TERM select_nif(ErlNifEnv* env, int argc, const ERL_NIF_TERM argv
     struct fd_resource* fdr;
     enum ErlNifSelectFlags mode;
     void* obj;
+    ErlNifPid nifpid, *pid = NULL;
     ERL_NIF_TERM ref;
     enum ErlNifSelectReturn retval;
 
@@ -2129,11 +2132,16 @@ static ERL_NIF_TERM select_nif(ErlNifEnv* env, int argc, const ERL_NIF_TERM argv
         return enif_make_badarg(env);
     }
 
-    ref = argv[3];
+    if (argv[3] != atom_null) {
+	if (!enif_get_local_pid(env, argv[3], &nifpid))
+	    return enif_make_badarg(env);
+	pid = &nifpid;
+    }
+    ref = argv[4];
 
     fdr->was_selected = 1;
     enif_self(env, &fdr->pid);
-    retval = enif_select(env, fdr->fd, mode, obj, ref);
+    retval = enif_select(env, fdr->fd, mode, obj, pid, ref);
 
     return enif_make_int(env, (int)retval);
 }
@@ -2160,7 +2168,9 @@ static ERL_NIF_TERM pipe_nif(ErlNifEnv* env, int argc, const ERL_NIF_TERM argv[]
     read_rsrc  = enif_alloc_resource(fd_resource_type, sizeof(struct fd_resource));
     write_rsrc = enif_alloc_resource(fd_resource_type, sizeof(struct fd_resource));
     read_rsrc->fd  = fds[0];
+    read_rsrc->was_selected = 0;
     write_rsrc->fd = fds[1];
+    write_rsrc->was_selected = 0;
     read_fd  = enif_make_resource(env, read_rsrc);
     write_fd = enif_make_resource(env, write_rsrc);
     enif_release_resource(read_rsrc);
@@ -2845,7 +2855,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", 4, select_nif},
+    {"select_nif", 5, select_nif},
     {"pipe_nif", 0, pipe_nif},
     {"write_nif", 2, write_nif},
     {"read_nif", 2, read_nif},