Fix leak with disterl monitor by name

Also add `monitored_by` in `process_info/2` to test this fix

Signed-off-by: Paul Guyot <[email protected]>

Author: pguyot

CHANGELOG.md

@@ -53,6 +53,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - Added `binary:replace/3`, `binary:replace/4`
 - Added `binary:match/2` and `binary:match/3`
 - Added `supervisor:which_children/1`
+- Added `monitored_by` in `process_info/2`
 
 ### Changed
 
@@ -72,6 +73,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - Support for zero count in `lists:duplicate/2`.
 - packbeam: fix memory leak preventing building with address sanitizer
 - Fixed a bug where empty atom could not be created on some platforms, thus breaking receiving a message for a registered process from an OTP node.
+- Fix a memory leak in distribution when a BEAM node would monitor a process by name.
 
 ## [0.6.7] - Unreleased
 

libs/estdlib/src/erlang.erl

@@ -178,6 +178,9 @@
     | port()
     | atom().
 
+% Current type until we make these references
+-type resource() :: binary().
+
 %%-----------------------------------------------------------------------------
 %% @param   Time time in milliseconds after which to send the timeout message.
 %% @param   Dest Pid or server name to which to send the timeout message.
@@ -255,6 +258,7 @@ send_after(Time, Dest, Msg) ->
 %%      <li><b>message_queue_len</b> the number of messages enqueued for the process (integer)</li>
 %%      <li><b>memory</b> the estimated total number of bytes in use by the process (integer)</li>
 %%      <li><b>links</b> the list of linked processes</li>
+%%      <li><b>monitored_by</b> the list of processes, NIF resources or ports that monitor the process</li>
 %% </ul>
 %% Specifying an unsupported term or atom raises a bad_arg error.
 %%
@@ -267,7 +271,8 @@ send_after(Time, Dest, Msg) ->
     (Pid :: pid(), stack_size) -> {stack_size, non_neg_integer()};
     (Pid :: pid(), message_queue_len) -> {message_queue_len, non_neg_integer()};
     (Pid :: pid(), memory) -> {memory, non_neg_integer()};
-    (Pid :: pid(), links) -> {links, [pid()]}.
+    (Pid :: pid(), links) -> {links, [pid()]};
+    (Pid :: pid(), monitored_by) -> {monitored_by, [pid() | resource() | port()]}.
 process_info(_Pid, _Key) ->
     erlang:nif_error(undefined).
 

src/libAtomVM/context.c

@@ -492,6 +492,19 @@ bool context_get_process_info(Context *ctx, term *out, size_t *term_size, term a
             }
             break;
         }
+        case MONITORED_BY_ATOM: {
+            struct ListHead *item;
+            ret_size = TUPLE_SIZE(2);
+            LIST_FOR_EACH (item, &ctx->monitors_head) {
+                struct Monitor *monitor = GET_LIST_ENTRY(item, struct Monitor, monitor_list_head);
+                if (monitor->monitor_type == CONTEXT_MONITOR_MONITORED_LOCAL) {
+                    ret_size += CONS_SIZE;
+                } else if (monitor->monitor_type == CONTEXT_MONITOR_RESOURCE) {
+                    ret_size += CONS_SIZE + TERM_BOXED_RESOURCE_SIZE;
+                }
+            }
+            break;
+        }
         default:
             if (out != NULL) {
                 *out = BADARG_ATOM;
@@ -582,6 +595,24 @@ bool context_get_process_info(Context *ctx, term *out, size_t *term_size, term a
             term_put_tuple_element(ret, 1, list);
             break;
         }
+        // pids of monitoring processes / resources
+        case MONITORED_BY_ATOM: {
+            term_put_tuple_element(ret, 0, MONITORED_BY_ATOM);
+            term list = term_nil();
+            struct ListHead *item;
+            LIST_FOR_EACH (item, &ctx->monitors_head) {
+                struct Monitor *monitor = GET_LIST_ENTRY(item, struct Monitor, monitor_list_head);
+                if (monitor->monitor_type == CONTEXT_MONITOR_MONITORED_LOCAL) {
+                    struct MonitorLocalMonitor *monitored_monitor = CONTAINER_OF(monitor, struct MonitorLocalMonitor, monitor);
+                    list = term_list_prepend(monitored_monitor->monitor_obj, list, heap);
+                } else if (monitor->monitor_type == CONTEXT_MONITOR_RESOURCE) {
+                    struct ResourceContextMonitor *resource_monitor = CONTAINER_OF(monitor, struct ResourceContextMonitor, monitor);
+                    list = term_list_prepend(term_from_resource(resource_monitor->resource_obj, heap), list, heap);
+                }
+            }
+            term_put_tuple_element(ret, 1, list);
+            break;
+        }
 
         default:
             UNREACHABLE();

src/libAtomVM/defaultatoms.def

@@ -190,3 +190,5 @@ X(SCOPE_ATOM, "\x5", "scope")
 X(NOMATCH_ATOM, "\x7", "nomatch")
 
 X(INIT_ATOM, "\x4", "init")
+
+X(MONITORED_BY_ATOM, "\xC", "monitored_by")

src/libAtomVM/dist_nifs.c

@@ -414,12 +414,13 @@ static term nif_erlang_dist_ctrl_get_data(Context *ctx, int argc, term argv[])
 
 term dist_monitor(struct DistConnection *conn_obj, term from_pid, term target_proc, term monitor_ref, Context *ctx)
 {
-    if (term_is_atom(target_proc)) {
-        target_proc = globalcontext_get_registered_process(ctx->global, term_to_atom_index(target_proc));
-    }
     int target_process_id = 0;
-    if (term_is_local_pid(target_proc)) {
-        target_process_id = term_to_local_process_id(target_proc);
+    term target_process_pid = target_proc;
+    if (term_is_atom(target_process_pid)) {
+        target_process_pid = globalcontext_get_registered_process(ctx->global, term_to_atom_index(target_process_pid));
+    }
+    if (term_is_local_pid(target_process_pid)) {
+        target_process_id = term_to_local_process_id(target_process_pid);
     } else {
         RAISE_ERROR(BADARG_ATOM);
     }

tests/erlang_tests/test_process_info.erl

@@ -40,6 +40,11 @@ start() ->
     {links, [Self]} = process_info(Pid, links),
     unlink(Pid),
     {links, []} = process_info(Pid, links),
+    Monitor = monitor(process, Pid),
+    % Twice because of spawn_opt above
+    {monitored_by, [Self, Self]} = process_info(Pid, monitored_by),
+    demonitor(Monitor),
+    {monitored_by, [Self]} = process_info(Pid, monitored_by),
     Pid ! {Self, stop},
     _Accum =
         receive

tests/libs/estdlib/test_net_kernel.erl

@@ -170,11 +170,14 @@ test_autoconnect_to_beam(Platform) ->
                 "F = fun(G) ->"
                 " receive"
                 "   {Caller, ping} -> Caller ! {self(), pong}, G(G);"
-                "   {Caller, quit} -> Caller ! {self(), quit}"
-                "   after 5000 -> exit(timeout)"
-                " end "
-                "end, "
-                "F(F).\" -s init stop -noshell"
+                "   {Caller, net_adm_ping} -> Caller ! {self(), net_adm:ping('" ++
+                    atom_to_list(Node) ++
+                    "')}, G(G);"
+                    "   {Caller, quit} -> Caller ! {self(), quit}"
+                    "   after 5000 -> exit(timeout)"
+                    " end "
+                    "end, "
+                    "F(F).\" -s init stop -noshell"
             )
         end,
         [link, monitor]
@@ -202,6 +205,14 @@ test_autoconnect_to_beam(Platform) ->
             {OTPPid, pong} -> ok
         after 5000 -> timeout
         end,
+    OTPPid ! {self(), net_adm_ping},
+    ok =
+        receive
+            {OTPPid, pong} -> ok
+        after 5000 -> timeout
+        end,
+    % Ensure there is no leak
+    {monitored_by, []} = process_info(whereis(net_kernel), monitored_by),
     erlang:send(OTPPid, {self(), quit}),
     ok =
         receive