2 files changed, 383 insertions, 11 deletions

diff --git a/erts/emulator/test/nif_SUITE.erl b/erts/emulator/test/nif_SUITE.erl
index 36d512e388..27276e6646 100644
--- a/erts/emulator/test/nif_SUITE.erl
+++ b/erts/emulator/test/nif_SUITE.erl
@@ -31,6 +31,7 @@
 	 init_per_testcase/2, end_per_testcase/2,
          basic/1, reload_error/1, upgrade/1, heap_frag/1,
          t_on_load/1,
+         select/1,
          hipe/1,
 	 types/1, many_args/1, binaries/1, get_string/1, get_atom/1,
 	 maps/1,
@@ -64,6 +65,7 @@ all() ->
     [{group, G} || G <- api_groups()]
         ++
     [reload_error, heap_frag, types, many_args,
+     select,
      hipe,
      binaries, get_string, get_atom, maps, api_macros, from_array,
      iolist_as_binary, resource, resource_binary,
@@ -434,6 +436,129 @@ t_on_load(Config) when is_list(Config) ->
     verify_tmpmem(TmpMem),
     ok.
 
+-define(ERL_NIF_SELECT_READ, (1 bsl 0)).
+-define(ERL_NIF_SELECT_WRITE, (1 bsl 1)).
+-define(ERL_NIF_SELECT_STOP, (1 bsl 2)).
+
+-define(ERL_NIF_SELECT_ERROR, (1 bsl 0)).
+-define(ERL_NIF_SELECT_STOP_CALLED, (1 bsl 1)).
+-define(ERL_NIF_SELECT_STOP_SCHEDULED, (1 bsl 2)).
+-define(ERL_NIF_SELECT_INVALID_EVENT, (1 bsl 3)).
+-define(ERL_NIF_SELECT_FAILED, (1 bsl 4)).
+
+
+select(Config) when is_list(Config) ->
+    ensure_lib_loaded(Config),
+
+    Ref = 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),
+    [] = flush(),
+    ok = write_nif(W, <<"hej">>),
+    [{select, R, Ref, ready_input}] = flush(),
+    <<"hej">> = read_nif(R, 3),
+
+    %% Wait for write
+    Written = write_full(W, $a),
+    0 = select_nif(W,?ERL_NIF_SELECT_WRITE,W,Ref),
+    [] = flush(),
+    Half = byte_size(Written) div 2,
+    <<First:Half/binary,Second/binary>> = Written,
+    First = read_nif(R,Half),
+    [{select, W, Ref, ready_output}] = flush(),
+    Third = write_full(W, $A),
+    Half2 = byte_size(Second),
+    <<Second:Half2/binary, Third/binary>> = read_nif(R, byte_size(Written)),
+
+    %% Close write and wait for EOF
+    eagain = read_nif(R, 1),
+    check_stop_ret(select_nif(W,?ERL_NIF_SELECT_STOP,W,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),
+    [{select, R, Ref, ready_input}] = flush(),
+    eof = read_nif(R,1),
+
+    check_stop_ret(select_nif(R,?ERL_NIF_SELECT_STOP,R,Ref)),
+    [{fd_resource_stop, R_ptr, _}] = flush(),
+    {1, {R_ptr,_}} = last_fd_stop_call(),
+    true = is_closed_nif(R),
+
+    select_2(Config).
+
+select_2(Config) ->
+    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,Ref1),
+    0 = select_nif(R,?ERL_NIF_SELECT_READ,R,Ref2),
+
+    [] = flush(),
+    ok = write_nif(W, <<"hej">>),
+    [{select, R, Ref2, ready_input}] = flush(),
+    <<"hej">> = read_nif(R, 3),
+
+    %% Change pid
+    eagain = read_nif(R, 1),
+    0 = select_nif(R,?ERL_NIF_SELECT_READ,R,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),
+    sync = receive_any(),
+    ok = write_nif(W, <<"hej">>),
+    done = receive_any(),
+    [] = flush(),
+
+    check_stop_ret(select_nif(R,?ERL_NIF_SELECT_STOP,R,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),
+    [{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) ->
+    erlang:garbage_collect(),
+    {_,_,2} = last_resource_dtor_call(),
+    ok.
+
+check_stop_ret(?ERL_NIF_SELECT_STOP_CALLED) -> ok;    
+check_stop_ret(?ERL_NIF_SELECT_STOP_SCHEDULED) -> ok.
+
+write_full(W, C) ->
+    write_full(W, C, <<>>).
+write_full(W, C, Acc) ->
+    case write_nif(W, <<C>>) of
+        ok ->
+            write_full(W, (C+1) band 255, <<Acc/binary, C>>);
+        {eagain,0} ->
+            Acc
+    end.
+
 hipe(Config) when is_list(Config) ->
     Data = proplists:get_value(data_dir, Config),
     Priv = proplists:get_value(priv_dir, Config),
@@ -1910,6 +2035,15 @@ call(Pid,Cmd) ->
 receive_any() ->
     receive M -> M end.	     
 
+flush() ->
+    flush(10).
+flush(Timeout) ->
+    receive M ->
+            [M | flush(Timeout)]
+    after Timeout ->
+            []
+    end.
+
 repeat(0, _, Arg) ->
     Arg;
 repeat(N, Fun, Arg0) ->
@@ -2273,6 +2407,12 @@ 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.
+pipe_nif() -> ?nif_stub.
+write_nif(_,_) -> ?nif_stub.
+read_nif(_,_) -> ?nif_stub.
+is_closed_nif(_) -> ?nif_stub.
+last_fd_stop_call() -> ?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 2c93891852..c4f9611ec8 100644
--- a/erts/emulator/test/nif_SUITE_data/nif_SUITE.c
+++ b/erts/emulator/test/nif_SUITE_data/nif_SUITE.c
@@ -25,6 +25,8 @@
 #include <limits.h>
 #ifndef __WIN32__
 #include <unistd.h>
+#include <fcntl.h>
+#include <errno.h>
 #endif
 
 #include "nif_mod.h"
@@ -42,7 +44,10 @@ static ERL_NIF_TERM atom_second;
 static ERL_NIF_TERM atom_millisecond;
 static ERL_NIF_TERM atom_microsecond;
 static ERL_NIF_TERM atom_nanosecond;
-
+static ERL_NIF_TERM atom_eagain;
+static ERL_NIF_TERM atom_eof;
+static ERL_NIF_TERM atom_error;
+static ERL_NIF_TERM atom_fd_resource_stop;
 
 typedef struct
 {
@@ -102,6 +107,20 @@ struct binary_resource {
     unsigned size;
 };
 
+static ErlNifResourceType* fd_resource_type;
+static void fd_resource_dtor(ErlNifEnv* env, void* obj);
+static void fd_resource_stop(ErlNifEnv* env, void* obj, ErlNifEvent, int);
+static ErlNifResourceTypeInit fd_rt_init = {
+    fd_resource_dtor,
+    fd_resource_stop
+};
+struct fd_resource {
+    int fd;
+    int was_selected;
+    ErlNifPid pid;
+};
+
+
 static int get_pointer(ErlNifEnv* env, ERL_NIF_TERM term, void** pp)
 {
     ErlNifBinary bin;
@@ -116,7 +135,6 @@ static ERL_NIF_TERM make_pointer(ErlNifEnv* env, void* p)
 {
     void** bin_data;
     ERL_NIF_TERM res;
-    ADD_CALL("get_priv_data_ptr");
     bin_data = (void**)enif_make_new_binary(env, sizeof(void*), &res);
     *bin_data = p;
     return res;
@@ -144,6 +162,9 @@ static int load(ErlNifEnv* env, void** priv_data, ERL_NIF_TERM load_info)
     msgenv_resource_type =  enif_open_resource_type(env,NULL,"nif_SUITE.msgenv",
 						    msgenv_dtor,
 						    ERL_NIF_RT_CREATE, NULL);
+    fd_resource_type =  enif_open_resource_type_x(env, "nif_SUITE.fd",
+                                                  &fd_rt_init,
+                                                  ERL_NIF_RT_CREATE, NULL);
     atom_false = enif_make_atom(env,"false");
     atom_true = enif_make_atom(env,"true");
     atom_self = enif_make_atom(env,"self");
@@ -154,6 +175,10 @@ static int load(ErlNifEnv* env, void** priv_data, ERL_NIF_TERM load_info)
     atom_millisecond = enif_make_atom(env,"millisecond");
     atom_microsecond = enif_make_atom(env,"microsecond");
     atom_nanosecond = enif_make_atom(env,"nanosecond");
+    atom_eagain = enif_make_atom(env, "eagain");
+    atom_eof = enif_make_atom(env, "eof");
+    atom_error = enif_make_atom(env, "error");
+    atom_fd_resource_stop = enif_make_atom(env, "fd_resource_stop");
 
     *priv_data = data;
     return 0;
@@ -389,8 +414,7 @@ static ERL_NIF_TERM type_test(ErlNifEnv* env, int argc, const ERL_NIF_TERM argv[
     ErlNifSInt64 sint64;
     ErlNifUInt64 uint64;
     double d;
-    ERL_NIF_TERM atom, ref1, ref2, term;
-    size_t len;
+    ERL_NIF_TERM atom, ref1, ref2;
 
     sint = INT_MIN;
     do {
@@ -1024,6 +1048,7 @@ struct make_term_info
 {
     ErlNifEnv* caller_env;
     ErlNifEnv* dst_env;
+    int dst_env_valid;
     ERL_NIF_TERM reuse[MAKE_TERM_REUSE_LEN];
     unsigned reuse_push;
     unsigned reuse_pull;
@@ -1053,6 +1078,7 @@ static ERL_NIF_TERM pull_term(struct make_term_info* mti)
 	mti->reuse_push < MAKE_TERM_REUSE_LEN) {
 	mti->reuse_pull = 0;
 	if (mti->reuse_push == 0) {
+            assert(mti->dst_env_valid);
 	    mti->reuse[0] = enif_make_list(mti->dst_env, 0);
 	}
     }
@@ -1106,10 +1132,6 @@ static ERL_NIF_TERM make_term_string(struct make_term_info* mti, int n)
 {
     return enif_make_string(mti->dst_env, "Hello!", ERL_NIF_LATIN1);
 }
-static ERL_NIF_TERM make_term_ref(struct make_term_info* mti, int n)
-{
-    return enif_make_ref(mti->dst_env);
-}
 static ERL_NIF_TERM make_term_sub_binary(struct make_term_info* mti, int n)
 {
     ERL_NIF_TERM orig;
@@ -1217,7 +1239,6 @@ static Make_term_Func* make_funcs[] = {
     make_term_atom,
     make_term_existing_atom,
     make_term_string,
-    //make_term_ref,
     make_term_sub_binary,
     make_term_uint,
     make_term_long,
@@ -1241,6 +1262,7 @@ static unsigned num_of_make_funcs()
 static int make_term_n(struct make_term_info* mti, int n, ERL_NIF_TERM* res)
 {
     if (n < num_of_make_funcs()) {
+        assert(mti->dst_env_valid);
 	*res = make_funcs[n](mti, n);
 	push_term(mti, *res);
 	return 1;
@@ -1257,6 +1279,7 @@ static ERL_NIF_TERM make_blob(ErlNifEnv* caller_env, ErlNifEnv* dst_env,
     struct make_term_info mti;
     mti.caller_env = caller_env;
     mti.dst_env = dst_env;
+    mti.dst_env_valid = 1;
     mti.reuse_push = 0;
     mti.reuse_pull = 0;
     mti.resource_type = priv->rt_arr[0].t;
@@ -1297,6 +1320,7 @@ static ERL_NIF_TERM alloc_msgenv(ErlNifEnv* env, int argc, const ERL_NIF_TERM ar
 						       sizeof(*mti));
     mti->caller_env = NULL;
     mti->dst_env = enif_alloc_env();
+    mti->dst_env_valid = 1;
     mti->reuse_push = 0;
     mti->reuse_pull = 0;
     mti->resource_type = priv->rt_arr[0].t;
@@ -1328,6 +1352,7 @@ static ERL_NIF_TERM clear_msgenv(ErlNifEnv* env, int argc, const ERL_NIF_TERM ar
 	return enif_make_badarg(env);
     }
     enif_clear_env(mti.p->dst_env);
+    mti.p->dst_env_valid = 1;
     mti.p->reuse_pull = 0;
     mti.p->reuse_push = 0;
     mti.p->blob = enif_make_list(mti.p->dst_env, 0);
@@ -1362,6 +1387,8 @@ static ERL_NIF_TERM send_blob(ErlNifEnv* env, int argc, const ERL_NIF_TERM argv[
     }
     copy = enif_make_copy(env, mti.p->blob);
     res = enif_send(env, &to, mti.p->dst_env, mti.p->blob);
+    if (res)
+        mti.p->dst_env_valid = 0;
     return enif_make_tuple3(env, atom_ok, enif_make_int(env,res), copy);
 }
 
@@ -1369,7 +1396,6 @@ static ERL_NIF_TERM send3_blob(ErlNifEnv* env, int argc, const ERL_NIF_TERM argv
 {
     mti_t mti;
     ErlNifPid to;
-    ERL_NIF_TERM copy;
     int res;
     if (!enif_get_resource(env, argv[0], msgenv_resource_type, &mti.vp)
 	|| !enif_get_local_pid(env, argv[1], &to)) {
@@ -1379,6 +1405,8 @@ static ERL_NIF_TERM send3_blob(ErlNifEnv* env, int argc, const ERL_NIF_TERM argv
 				   enif_make_copy(mti.p->dst_env, argv[2]),
 				   mti.p->blob);
     res = enif_send(env, &to, mti.p->dst_env, mti.p->blob);
+    if (res)
+        mti.p->dst_env_valid = 0;
     return enif_make_int(env,res);
 }
 
@@ -1395,6 +1423,8 @@ void* threaded_sender(void *arg)
     mti.p->send_it = 0;
     enif_mutex_unlock(mti.p->mtx);
     mti.p->send_res = enif_send(NULL, &mti.p->to_pid, mti.p->dst_env, mti.p->blob);
+    if (mti.p->send_res)
+        mti.p->dst_env_valid = 0;
     return NULL;
 }
 
@@ -2010,6 +2040,202 @@ static ERL_NIF_TERM format_term(ErlNifEnv* env, int argc, const ERL_NIF_TERM arg
 }
 
 
+static int get_fd(ErlNifEnv* env, ERL_NIF_TERM term, struct fd_resource** rsrc)
+{
+    if (!enif_get_resource(env, term, fd_resource_type, (void**)rsrc)) {
+        return 0;
+    }
+    return 1;
+}
+
+static ERL_NIF_TERM select_nif(ErlNifEnv* env, int argc, const ERL_NIF_TERM argv[])
+{
+    struct fd_resource* fdr;
+    enum ErlNifSelectFlags mode;
+    void* obj;
+    ERL_NIF_TERM ref;
+    enum ErlNifSelectReturn retval;
+
+    if (!get_fd(env, argv[0], &fdr)
+        || !enif_get_uint(env, argv[1], &mode)
+        || !enif_get_resource(env, argv[2], fd_resource_type, &obj))
+    {
+        return enif_make_badarg(env);
+    }
+
+    ref = argv[3];
+
+    fdr->was_selected = 1;
+    enif_self(env, &fdr->pid);
+    retval = enif_select(env, fdr->fd, mode, obj, ref);
+
+    return enif_make_int(env, (int)retval);
+}
+
+static ERL_NIF_TERM pipe_nif(ErlNifEnv* env, int argc, const ERL_NIF_TERM argv[])
+{
+    struct fd_resource* read_rsrc;
+    struct fd_resource* write_rsrc;
+    ERL_NIF_TERM read_fd, write_fd;
+    int fds[2], flags;
+
+    if (pipe(fds) < 0)
+        return enif_make_string(env, "pipe failed", ERL_NIF_LATIN1);
+
+    if ((flags = fcntl(fds[0], F_GETFL, 0)) < 0
+        || fcntl(fds[0], F_SETFL, flags|O_NONBLOCK) < 0
+        || (flags = fcntl(fds[1], F_GETFL, 0)) < 0
+        || fcntl(fds[1], F_SETFL, flags|O_NONBLOCK) < 0) {
+        close(fds[0]);
+        close(fds[1]);
+        return enif_make_string(env, "fcntl failed on pipe", ERL_NIF_LATIN1);
+    }
+
+    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];
+    write_rsrc->fd = fds[1];
+    read_fd  = enif_make_resource(env, read_rsrc);
+    write_fd = enif_make_resource(env, write_rsrc);
+    enif_release_resource(read_rsrc);
+    enif_release_resource(write_rsrc);
+
+    return enif_make_tuple2(env,
+               enif_make_tuple2(env, read_fd, make_pointer(env, read_rsrc)),
+               enif_make_tuple2(env, write_fd, make_pointer(env, write_rsrc)));
+}
+
+static ERL_NIF_TERM write_nif(ErlNifEnv* env, int argc, const ERL_NIF_TERM argv[])
+{
+    struct fd_resource* fdr;
+    ErlNifBinary bin;
+    int n, written = 0;
+
+    if (!get_fd(env, argv[0], &fdr)
+        || !enif_inspect_binary(env, argv[1], &bin))
+        return enif_make_badarg(env);
+
+    for (;;) {
+        n = write(fdr->fd, bin.data + written, bin.size - written);
+        if (n >= 0) {
+            written += n;
+            if (written == bin.size) {
+                return atom_ok;
+            }
+        }
+        else if (errno == EAGAIN) {
+            return enif_make_tuple2(env, atom_eagain, enif_make_int(env, written));
+        }
+        else if (errno == EINTR) {
+            continue;
+        }
+        else {
+            return enif_make_tuple2(env, atom_error, enif_make_int(env, errno));
+        }
+    }
+}
+
+static ERL_NIF_TERM read_nif(ErlNifEnv* env, int argc, const ERL_NIF_TERM argv[])
+{
+    struct fd_resource* fdr;
+    unsigned char* buf;
+    int n, count;
+    ERL_NIF_TERM res;
+
+    if (!get_fd(env, argv[0], &fdr)
+        || !enif_get_int(env, argv[1], &count) || count < 1)
+        return enif_make_badarg(env);
+
+    buf = enif_make_new_binary(env, count, &res);
+
+    for (;;) {
+        n = read(fdr->fd, buf, count);
+        if (n > 0) {
+            if (n < count) {
+                res = enif_make_sub_binary(env, res, 0, n);
+            }
+            return res;
+        }
+        else if (n == 0) {
+            return atom_eof;
+        }
+        else if (errno == EAGAIN) {
+            return atom_eagain;
+        }
+        else if (errno == EINTR) {
+            continue;
+        }
+        else {
+            return enif_make_tuple2(env, atom_error, enif_make_int(env, errno));
+        }
+    }
+}
+
+static ERL_NIF_TERM is_closed_nif(ErlNifEnv* env, int argc, const ERL_NIF_TERM argv[])
+{
+    struct fd_resource* fdr;
+
+    if (!get_fd(env, argv[0], &fdr))
+        return enif_make_badarg(env);
+
+    return fdr->fd < 0 ? atom_true : atom_false;
+}
+
+static void fd_resource_dtor(ErlNifEnv* env, void* obj)
+{
+    struct fd_resource* fdr = (struct fd_resource*)obj;
+    resource_dtor(env, obj);
+    if (fdr->fd >= 0) {
+        assert(!fdr->was_selected);
+        close(fdr->fd);
+    }
+}
+
+static struct {
+    void* obj;
+    int was_direct_call;
+}last_fd_stop;
+int fd_stop_cnt = 0;
+
+static void fd_resource_stop(ErlNifEnv* env, void* obj, ErlNifEvent fd,
+                             int is_direct_call)
+{
+    struct fd_resource* fdr = (struct fd_resource*)obj;
+    assert(fd == fdr->fd);
+    assert(fd >= 0);
+
+    last_fd_stop.obj = obj;
+    last_fd_stop.was_direct_call = is_direct_call;
+    fd_stop_cnt++;
+
+    close(fd);
+    fdr->fd = -1;   /* thread safety ? */
+    fdr->was_selected = 0;
+
+    {
+        ErlNifEnv* msg_env = enif_alloc_env();
+        ERL_NIF_TERM msg;
+        msg = enif_make_tuple3(msg_env,
+                               atom_fd_resource_stop,
+                               make_pointer(msg_env, obj),
+                               enif_make_int(msg_env, is_direct_call));
+
+        enif_send(env, &fdr->pid, msg_env, msg);
+        enif_free_env(msg_env);
+    }
+}
+
+static ERL_NIF_TERM last_fd_stop_call(ErlNifEnv* env, int argc, const ERL_NIF_TERM argv[])
+{
+    ERL_NIF_TERM last, ret;
+    last = enif_make_tuple2(env, make_pointer(env, last_fd_stop.obj),
+                            enif_make_int(env, last_fd_stop.was_direct_call));
+    ret = enif_make_tuple2(env, enif_make_int(env, fd_stop_cnt), last);
+    fd_stop_cnt = 0;
+    return ret;
+}
+
+
 static ErlNifFunc nif_funcs[] =
 {
     {"lib_version", 0, lib_version},
@@ -2086,7 +2312,13 @@ static ErlNifFunc nif_funcs[] =
     {"term_to_binary_nif", 2, term_to_binary},
     {"binary_to_term_nif", 3, binary_to_term},
     {"port_command_nif", 2, port_command},
-    {"format_term_nif", 2, format_term}
+    {"format_term_nif", 2, format_term},
+    {"select_nif", 4, select_nif},
+    {"pipe_nif", 0, pipe_nif},
+    {"write_nif", 2, write_nif},
+    {"read_nif", 2, read_nif},
+    {"is_closed_nif", 1, is_closed_nif},
+    {"last_fd_stop_call", 0, last_fd_stop_call}
 };
 
 ERL_NIF_INIT(nif_SUITE,nif_funcs,load,NULL,upgrade,unload)