Merge pull request atomvm#729 from pguyot/w31/monitor-posix_select-s

Add monitor to posix_select

Monitor selecting process to avoid leaks.
Add tests

These changes are made under both the "Apache 2.0" and the "GNU Lesser General
Public License 2.1 or later" license terms (dual license).

SPDX-License-Identifier: Apache-2.0 OR LGPL-2.1-or-later

Author: bettio

src/libAtomVM/posix_nifs.c

@@ -35,6 +35,7 @@
 
 #include "defaultatoms.h"
 #include "erl_nif_priv.h"
+#include "globalcontext.h"
 #include "interop.h"
 #include "nifs.h"
 #include "posix_nifs.h"
@@ -115,7 +116,8 @@ term posix_errno_to_term(int err, GlobalContext *glb)
 struct PosixFd
 {
     int fd;
-    bool select;
+    int32_t selecting_process_id;
+    ErlNifMonitor selecting_process_monitor;
 };
 
 static void posix_fd_dtor(ErlNifEnv *caller_env, void *obj)
@@ -131,18 +133,32 @@ static void posix_fd_dtor(ErlNifEnv *caller_env, void *obj)
 
 static void posix_fd_stop(ErlNifEnv *caller_env, void *obj, ErlNifEvent event, int is_direct_call)
 {
-    UNUSED(caller_env);
     UNUSED(event);
     UNUSED(is_direct_call);
 
     struct PosixFd *fd_obj = (struct PosixFd *) obj;
-    fd_obj->select = false;
+    if (fd_obj->selecting_process_id != INVALID_PROCESS_ID) {
+        enif_demonitor_process(caller_env, fd_obj, &fd_obj->selecting_process_monitor);
+        fd_obj->selecting_process_id = INVALID_PROCESS_ID;
+    }
+}
+
+static void posix_fd_down(ErlNifEnv *caller_env, void *obj, ErlNifPid *pid, ErlNifMonitor *mon)
+{
+    UNUSED(pid);
+    UNUSED(mon);
+    struct PosixFd *fd_obj = (struct PosixFd *) obj;
+    if (fd_obj->selecting_process_id != INVALID_PROCESS_ID) {
+        fd_obj->selecting_process_id = INVALID_PROCESS_ID;
+        enif_select(caller_env, fd_obj->fd, ERL_NIF_SELECT_STOP, fd_obj, NULL, term_nil());
+    }
 }
 
 const ErlNifResourceTypeInit posix_fd_resource_type_init = {
-    .members = 2,
+    .members = 3,
     .dtor = posix_fd_dtor,
     .stop = posix_fd_stop,
+    .down = posix_fd_down,
 };
 
 #define O_EXEC_ATOM_STR ATOM_STR("\x6", "o_exec")
@@ -270,7 +286,7 @@ static term nif_atomvm_posix_open(Context *ctx, int argc, term argv[])
             RAISE_ERROR(OUT_OF_MEMORY_ATOM);
         }
         fd_obj->fd = fd;
-        fd_obj->select = false;
+        fd_obj->selecting_process_id = INVALID_PROCESS_ID;
         if (UNLIKELY(memory_ensure_free_opt(ctx, TUPLE_SIZE(2) + TERM_BOXED_RESOURCE_SIZE, MEMORY_CAN_SHRINK) != MEMORY_GC_OK)) {
             RAISE_ERROR(OUT_OF_MEMORY_ATOM);
         }
@@ -295,7 +311,7 @@ static term nif_atomvm_posix_close(Context *ctx, int argc, term argv[])
     }
     struct PosixFd *fd_obj = (struct PosixFd *) fd_obj_ptr;
     if (fd_obj->fd != CLOSED_FD) {
-        if (fd_obj->select) {
+        if (fd_obj->selecting_process_id != INVALID_PROCESS_ID) {
             fprintf(stderr, "Calling close on a selectable posix file, missing call to posix_select_stop?\n");
         }
         if (UNLIKELY(close(fd_obj->fd) < 0)) {
@@ -398,11 +414,26 @@ static term nif_atomvm_posix_select(Context *ctx, term argv[], enum ErlNifSelect
         RAISE_ERROR(BADARG_ATOM);
     }
     struct PosixFd *fd_obj = (struct PosixFd *) fd_obj_ptr;
-
-    if (UNLIKELY(enif_select(erl_nif_env_from_context(ctx), fd_obj->fd, mode, fd_obj, &process_pid, select_ref_term) < 0)) {
+    ErlNifEnv *env = erl_nif_env_from_context(ctx);
+    if (fd_obj->selecting_process_id != process_pid && fd_obj->selecting_process_id != INVALID_PROCESS_ID) {
+        if (UNLIKELY(enif_demonitor_process(env, fd_obj, &fd_obj->selecting_process_monitor) != 0)) {
+            RAISE_ERROR(BADARG_ATOM);
+        }
+        fd_obj->selecting_process_id = INVALID_PROCESS_ID;
+    }
+    // Monitor first as select is less likely to fail and it's less expensive to demonitor
+    // if select fails than to stop select if monitor fails
+    if (fd_obj->selecting_process_id != process_pid) {
+        if (UNLIKELY(enif_monitor_process(env, fd_obj, &process_pid, &fd_obj->selecting_process_monitor) != 0)) {
+            RAISE_ERROR(BADARG_ATOM);
+        }
+        fd_obj->selecting_process_id = process_pid;
+    }
+    if (UNLIKELY(enif_select(env, fd_obj->fd, mode, fd_obj, &process_pid, select_ref_term) < 0)) {
+        enif_demonitor_process(env, fd_obj, &fd_obj->selecting_process_monitor);
+        fd_obj->selecting_process_id = INVALID_PROCESS_ID;
         RAISE_ERROR(BADARG_ATOM);
     }
-    fd_obj->select = 1;
 
     return OK_ATOM;
 }

tests/libs/eavmlib/test_file.erl

@@ -29,6 +29,8 @@ test() ->
     ok = test_basic_file(),
     ok = test_fifo_select(HasSelect),
     ok = test_gc(HasSelect),
+    ok = test_crash_no_leak(HasSelect),
+    ok = test_select_with_gone_process(HasSelect),
     ok.
 
 test_basic_file() ->
@@ -83,7 +85,26 @@ test_fifo_select(_HasSelect) ->
             M2 -> {unexpected, M2}
         after 200 -> fail
         end,
-    ok = atomvm:posix_select_stop(RdFd),
+    Parent = self(),
+    Child = spawn(fun() ->
+        ChildRef = make_ref(),
+        ok = atomvm:posix_select_read(RdFd, self(), ChildRef),
+        ok =
+            receive
+                {select, RdFd, ChildRef, ready_input} -> ok;
+                M3 -> {unexpected, M3}
+            after 200 -> fail
+            end,
+        ok = atomvm:posix_select_stop(RdFd),
+        {ok, <<" World">>} = atomvm:posix_read(RdFd, 10),
+        Parent ! {self(), world}
+    end),
+    ok =
+        receive
+            {Child, world} -> ok;
+            M4 -> {unexpected, M4}
+        after 200 -> fail
+        end,
     ok =
         receive
             Message -> {unexpected, Message}
@@ -179,3 +200,70 @@ gc_loop(Path, File) ->
         {Caller, quit} ->
             Caller ! {self(), quit}
     end.
+
+% Test is based on the fact that `erlang:memory(binary)` count resources.
+test_crash_no_leak(false) ->
+    ok;
+test_crash_no_leak(true) ->
+    Before = erlang:memory(binary),
+    Path = "/tmp/atomvm.tmp." ++ integer_to_list(erlang:system_time(millisecond)),
+    Pid = spawn(fun() -> crash_leak(Path) end),
+    Ref = monitor(process, Pid),
+    ok =
+        receive
+            {'DOWN', Ref, process, Pid, _} -> ok
+        after 5000 -> timeout
+        end,
+    % Make sure sys_poll_events is called so selected resource is
+    % properly disposed. This also gives time to context_destroy to
+    % sweep mso after DOWN message is sent
+    receive
+    after 100 -> ok
+    end,
+    After = erlang:memory(binary),
+    0 = After - Before,
+    ok = atomvm:posix_unlink(Path),
+    ok.
+
+crash_leak(Path) ->
+    ok = setup_and_forget(Path),
+    % We don't really need to crash here
+    % the test consists in not calling select_stop
+    ok.
+
+setup_and_forget(Path) ->
+    {ok, Fd} = atomvm:posix_open(Path, [o_rdwr, o_creat], 8#644),
+    ok = atomvm:posix_select_write(Fd, self(), undefined),
+    ok.
+
+test_select_with_gone_process(false) ->
+    ok;
+test_select_with_gone_process(true) ->
+    Before = erlang:memory(binary),
+    Path = "/tmp/atomvm.tmp." ++ integer_to_list(erlang:system_time(millisecond)),
+    test_select_with_gone_process0(Path),
+    % GC so Fd's ref count is decremented
+    erlang:garbage_collect(),
+    After = erlang:memory(binary),
+    0 = After - Before,
+    ok = atomvm:posix_unlink(Path),
+    ok.
+
+test_select_with_gone_process0(Path) ->
+    GonePid = spawn(fun() -> ok end),
+    Ref = monitor(process, GonePid),
+    ok =
+        receive
+            {'DOWN', Ref, process, GonePid, _} -> ok
+        after 5000 -> timeout
+        end,
+    {ok, Fd} = atomvm:posix_open(Path, [o_rdwr, o_creat], 8#644),
+    ok =
+        try
+            atomvm:posix_select_write(Fd, GonePid, undefined),
+            fail
+        catch
+            error:badarg -> ok
+        end,
+    ok = atomvm:posix_close(Fd),
+    ok.