Merge pull request #1790 from pguyot/w32/fix-oom-disterl

Fix leak with disterl monitor by name

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

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