Merge pull request #1480 from pguyot/w04/erlang-distribution-04

Distribution: Add support for external pid as group leader

Continuation of #1468

- Store group leader pid on the heap if it's an external pid
- Introduce a specific signal to change group leader to an external pid as we
cannot touch heap from another process
- Add simple test based on rpc and BEAM

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

src/libAtomVM/context.c

@@ -217,6 +217,17 @@ bool context_process_signal_trap_answer(Context *ctx, struct TermSignal *signal)
     return true;
 }
 
+bool context_process_signal_set_group_leader(Context *ctx, struct TermSignal *signal)
+{
+    size_t leader_term_size = memory_estimate_usage(signal->signal_term);
+    ctx->group_leader = UNDEFINED_ATOM;
+    if (UNLIKELY(memory_ensure_free_opt(ctx, leader_term_size, MEMORY_CAN_SHRINK) != MEMORY_GC_OK)) {
+        return false;
+    }
+    ctx->group_leader = memory_copy_term_tree(&ctx->heap, signal->signal_term);
+    return true;
+}
+
 void context_process_flush_monitor_signal(Context *ctx, uint64_t ref_ticks, bool info)
 {
     context_update_flags(ctx, ~Trap, NoFlags);

src/libAtomVM/context.h

@@ -393,6 +393,15 @@ bool context_process_signal_trap_answer(Context *ctx, struct TermSignal *signal)
  */
 void context_process_flush_monitor_signal(Context *ctx, uint64_t ref_ticks, bool info);
 
+/**
+ * @brief Process set group leader signal
+ *
+ * @param ctx the context being executed
+ * @param signal the message with the group leader term
+ * @return \c true if successful, \c false in case of memory error
+ */
+bool context_process_signal_set_group_leader(Context *ctx, struct TermSignal *signal);
+
 /**
  * @brief Get process information.
  *

src/libAtomVM/dist_nifs.c

@@ -496,10 +496,10 @@ static term nif_erlang_dist_ctrl_put_data(Context *ctx, int argc, term argv[])
             }
             term roots[4];
             roots[0] = argv[0];
-            roots[1] = argv[1];
+            roots[1] = argv[1]; // dist handle, ensure it's not garbage collected until we return
             roots[2] = control;
             roots[3] = externalterm_to_term_with_roots(data + 1 + bytes_read, binary_len - 1 - bytes_read, ctx, ExternalTermCopy, &bytes_read, 3, roots);
-            if (UNLIKELY(memory_ensure_free_with_roots(ctx, LIST_SIZE(1, TUPLE_SIZE(2) + TUPLE_SIZE(4)), 4, roots, MEMORY_CAN_SHRINK) != MEMORY_GC_OK)) {
+            if (UNLIKELY(memory_ensure_free_with_roots(ctx, LIST_SIZE(1, TUPLE_SIZE(2) + TUPLE_SIZE(5)), 4, roots, MEMORY_CAN_SHRINK) != MEMORY_GC_OK)) {
                 RAISE_ERROR(OUT_OF_MEMORY_ATOM);
             }
             control = roots[2];
@@ -516,15 +516,18 @@ static term nif_erlang_dist_ctrl_put_data(Context *ctx, int argc, term argv[])
             if (UNLIKELY(!term_is_pid(from))) {
                 RAISE_ERROR(BADARG_ATOM);
             }
-            // term groupleader = term_get_tuple_element(control, 3);
-            // TODO: handle groupleader which is an externalpid
+            term groupleader = term_get_tuple_element(control, 3);
+            if (UNLIKELY(!term_is_pid(groupleader))) {
+                RAISE_ERROR(BADARG_ATOM);
+            }
             term options = term_get_tuple_element(control, 5);
 
-            term request_tuple = term_alloc_tuple(4, &ctx->heap);
+            term request_tuple = term_alloc_tuple(5, &ctx->heap);
             term_put_tuple_element(request_tuple, 0, roots[0]);
             term_put_tuple_element(request_tuple, 1, reqid);
             term_put_tuple_element(request_tuple, 2, from);
-            term_put_tuple_element(request_tuple, 3, options);
+            term_put_tuple_element(request_tuple, 3, groupleader);
+            term_put_tuple_element(request_tuple, 4, options);
             term request_opt = term_alloc_tuple(2, &ctx->heap);
             term_put_tuple_element(request_opt, 0, REQUEST_ATOM);
             term_put_tuple_element(request_opt, 1, request_tuple);

src/libAtomVM/mailbox.c

@@ -96,7 +96,8 @@ void mailbox_message_dispose(MailboxMessage *m, Heap *heap)
             break;
         }
         case KillSignal:
-        case TrapAnswerSignal: {
+        case TrapAnswerSignal:
+        case SetGroupLeaderSignal: {
             struct TermSignal *term_signal = CONTAINER_OF(m, struct TermSignal, base);
             term mso_list = term_signal->storage[STORAGE_MSO_LIST_INDEX];
             HeapFragment *fragment = mailbox_message_to_heap_fragment(term_signal, term_signal->heap_end);

src/libAtomVM/mailbox.h

@@ -89,6 +89,7 @@ enum MessageType
     TrapExceptionSignal,
     FlushMonitorSignal,
     FlushInfoMonitorSignal,
+    SetGroupLeaderSignal,
 };
 
 struct MailboxMessage

src/libAtomVM/memory.c

@@ -300,6 +300,9 @@ static enum MemoryGCResult memory_gc(Context *ctx, size_t new_size, size_t num_r
     TRACE("- Running copy GC on exit reason\n");
     ctx->exit_reason = memory_shallow_copy_term(old_root_fragment, ctx->exit_reason, &ctx->heap.heap_ptr, true);
 
+    TRACE("- Running copy GC on group leader\n");
+    ctx->group_leader = memory_shallow_copy_term(old_root_fragment, ctx->group_leader, &ctx->heap.heap_ptr, true);
+
     TRACE("- Running copy GC on provided roots\n");
     for (size_t i = 0; i < num_roots; i++) {
         roots[i] = memory_shallow_copy_term(old_root_fragment, roots[i], &ctx->heap.heap_ptr, 1);
@@ -373,6 +376,8 @@ static enum MemoryGCResult memory_shrink(Context *ctx, size_t new_size, size_t n
     }
     // ...exit_reason
     memory_scan_and_rewrite(1, &ctx->exit_reason, old_heap_root, old_end, delta, true);
+    // ...group_leader
+    memory_scan_and_rewrite(1, &ctx->group_leader, old_heap_root, old_end, delta, true);
     // ...and MSO list.
     term *mso_ptr = &ctx->heap.root->mso_list;
     while (!term_is_nil(*mso_ptr)) {

src/libAtomVM/nifs.c

@@ -1211,14 +1211,21 @@ static NativeHandlerResult process_console_mailbox(Context *ctx)
 
 // Common handling of spawn/1, spawn/3, spawn_opt/2, spawn_opt/4
 // opts_term is [] for spawn/1,3
-static term do_spawn(Context *ctx, Context *new_ctx, term opts_term)
+static term do_spawn(Context *ctx, Context *new_ctx, size_t arity, size_t n_freeze, term opts_term)
 {
     term min_heap_size_term = interop_proplist_get_value(opts_term, MIN_HEAP_SIZE_ATOM);
     term max_heap_size_term = interop_proplist_get_value(opts_term, MAX_HEAP_SIZE_ATOM);
     term link_term = interop_proplist_get_value(opts_term, LINK_ATOM);
     term monitor_term = interop_proplist_get_value(opts_term, MONITOR_ATOM);
     term heap_growth_strategy = interop_proplist_get_value_default(opts_term, ATOMVM_HEAP_GROWTH_ATOM, BOUNDED_FREE_ATOM);
     term request_term = interop_proplist_get_value(opts_term, REQUEST_ATOM);
+    term group_leader;
+
+    if (UNLIKELY(request_term != term_nil())) {
+        group_leader = term_get_tuple_element(request_term, 3);
+    } else {
+        group_leader = ctx->group_leader;
+    }
 
     if (min_heap_size_term != term_nil()) {
         if (UNLIKELY(!term_is_integer(min_heap_size_term))) {
@@ -1245,6 +1252,21 @@ static term do_spawn(Context *ctx, Context *new_ctx, term opts_term)
         }
     }
 
+    int size = 0;
+    for (uint32_t i = 0; i < n_freeze; i++) {
+        size += memory_estimate_usage(new_ctx->x[i + arity - n_freeze]);
+    }
+    size += memory_estimate_usage(group_leader);
+    if (UNLIKELY(memory_ensure_free_opt(new_ctx, size, MEMORY_CAN_SHRINK) != MEMORY_GC_OK)) {
+        //TODO: new process should be terminated, however a new pid is returned anyway
+        fprintf(stderr, "Unable to allocate sufficient memory to spawn process.\n");
+        AVM_ABORT();
+    }
+    new_ctx->group_leader = memory_copy_term_tree(&new_ctx->heap, group_leader);
+    for (uint32_t i = 0; i < arity; i++) {
+        new_ctx->x[i] = memory_copy_term_tree(&new_ctx->heap, new_ctx->x[i]);
+    }
+
     switch (heap_growth_strategy) {
         case BOUNDED_FREE_ATOM:
             new_ctx->heap_growth_strategy = BoundedFreeHeapGrowth;
@@ -1308,7 +1330,7 @@ static term do_spawn(Context *ctx, Context *new_ctx, term opts_term)
         term dhandle = term_get_tuple_element(request_term, 0);
         term request_ref = term_get_tuple_element(request_term, 1);
         term request_from = term_get_tuple_element(request_term, 2);
-        term request_opts = term_get_tuple_element(request_term, 3);
+        term request_opts = term_get_tuple_element(request_term, 4);
         monitor_term = interop_proplist_get_value(request_opts, MONITOR_ATOM);
         // TODO handle link with external nodes
         // link_term = interop_proplist_get_value(request_opts, LINK_ATOM);
@@ -1344,7 +1366,6 @@ static term nif_erlang_spawn_fun_opt(Context *ctx, int argc, term argv[])
     VALIDATE_VALUE(opts_term, term_is_list);
 
     Context *new_ctx = context_new(ctx->global);
-    new_ctx->group_leader = ctx->group_leader;
 
     const term *boxed_value = term_to_const_term_ptr(fun_term);
 
@@ -1365,24 +1386,15 @@ static term nif_erlang_spawn_fun_opt(Context *ctx, int argc, term argv[])
 
     // TODO: new process should fail with badarity if arity != 0
 
-    int size = 0;
     for (uint32_t i = 0; i < n_freeze; i++) {
-        size += memory_estimate_usage(boxed_value[i + 3]);
-    }
-    if (UNLIKELY(memory_ensure_free_opt(new_ctx, size, MEMORY_CAN_SHRINK) != MEMORY_GC_OK)) {
-        //TODO: new process should be terminated, however a new pid is returned anyway
-        fprintf(stderr, "Unable to allocate sufficient memory to spawn process.\n");
-        AVM_ABORT();
-    }
-    for (uint32_t i = 0; i < n_freeze; i++) {
-        new_ctx->x[i + arity - n_freeze] = memory_copy_term_tree(&new_ctx->heap, boxed_value[i + 3]);
+        new_ctx->x[i + arity - n_freeze] = boxed_value[i + 3];
     }
 
     new_ctx->saved_module = fun_module;
     new_ctx->saved_ip = fun_module->labels[label];
     new_ctx->cp = module_address(fun_module->module_index, fun_module->end_instruction_ii);
 
-    return do_spawn(ctx, new_ctx, opts_term);
+    return do_spawn(ctx, new_ctx, arity, n_freeze, opts_term);
 }
 
 term nif_erlang_spawn_opt(Context *ctx, int argc, term argv[])
@@ -1399,7 +1411,6 @@ term nif_erlang_spawn_opt(Context *ctx, int argc, term argv[])
     VALIDATE_VALUE(opts_term, term_is_list);
 
     Context *new_ctx = context_new(ctx->global);
-    new_ctx->group_leader = ctx->group_leader;
 
     AtomString module_string = globalcontext_atomstring_from_term(ctx->global, argv[0]);
     AtomString function_string = globalcontext_atomstring_from_term(ctx->global, argv[1]);
@@ -1439,14 +1450,8 @@ term nif_erlang_spawn_opt(Context *ctx, int argc, term argv[])
         new_ctx->min_heap_size = min_heap_size;
     }
 
-    avm_int_t size = memory_estimate_usage(args_term);
-    if (UNLIKELY(memory_ensure_free_opt(new_ctx, size, MEMORY_CAN_SHRINK) != MEMORY_GC_OK)) {
-        // Context was not scheduled yet, we can destroy it.
-        context_destroy(new_ctx);
-        RAISE_ERROR(OUT_OF_MEMORY_ATOM);
-    }
     while (term_is_nonempty_list(args_term)) {
-        new_ctx->x[reg_index] = memory_copy_term_tree(&new_ctx->heap, term_get_list_head(args_term));
+        new_ctx->x[reg_index] = term_get_list_head(args_term);
         reg_index++;
 
         args_term = term_get_list_tail(args_term);
@@ -1456,7 +1461,7 @@ term nif_erlang_spawn_opt(Context *ctx, int argc, term argv[])
         }
     }
 
-    return do_spawn(ctx, new_ctx, opts_term);
+    return do_spawn(ctx, new_ctx, reg_index, reg_index, opts_term);
 }
 
 static term nif_erlang_send_2(Context *ctx, int argc, term argv[])
@@ -4003,15 +4008,21 @@ static term nif_erlang_group_leader(Context *ctx, int argc, term argv[])
         term leader = argv[0];
         term pid = argv[1];
         VALIDATE_VALUE(pid, term_is_local_pid);
-        VALIDATE_VALUE(leader, term_is_local_pid);
+        VALIDATE_VALUE(leader, term_is_pid);
 
         int local_process_id = term_to_local_process_id(pid);
         Context *target = globalcontext_get_process_lock(ctx->global, local_process_id);
         if (IS_NULL_PTR(target)) {
             RAISE_ERROR(BADARG_ATOM);
         }
 
-        target->group_leader = leader;
+        if (term_is_local_pid(leader)) {
+            // We cannot put leader term on the heap
+            mailbox_send_term_signal(target, SetGroupLeaderSignal, leader);
+        } else {
+            target->group_leader = leader;
+        }
+
         globalcontext_get_process_unlock(ctx->global, target);
         return TRUE_ATOM;
     }

src/libAtomVM/opcodesswitch.h

@@ -1049,6 +1049,15 @@ static void destroy_extended_registers(Context *ctx, unsigned int live)
                     context_process_flush_monitor_signal(ctx, flush_signal->ref_ticks, info);   \
                     break;                                                                      \
                 }                                                                               \
+                case SetGroupLeaderSignal: {                                                    \
+                    struct TermSignal *group_leader                                             \
+                        = CONTAINER_OF(signal_message, struct TermSignal, base);                \
+                    if (UNLIKELY(!context_process_signal_set_group_leader(ctx, group_leader))) { \
+                        SET_ERROR(OUT_OF_MEMORY_ATOM);                                          \
+                        next_label = &&handle_error;                                            \
+                    }                                                                           \
+                    break;                                                                      \
+                }                                                                               \
                 case NormalMessage: {                                                           \
                     UNREACHABLE();                                                              \
                 }                                                                               \

tests/libs/estdlib/test_net_kernel.erl

@@ -37,6 +37,7 @@ test() ->
             ok = test_rpc_loop_from_beam(Platform),
             ok = test_autoconnect_fail(Platform),
             ok = test_autoconnect_to_beam(Platform),
+            ok = test_groupleader(Platform),
             ok;
         false ->
             io:format("~s: skipped\n", [?MODULE]),
@@ -156,21 +157,24 @@ test_autoconnect_to_beam(Platform) ->
     {ok, _NetKernelPid} = net_kernel_start(Platform, atomvm),
     Node = node(),
     erlang:set_cookie(Node, 'AtomVM'),
-    spawn_link(fun() ->
-        execute_command(
-            Platform,
-            "erl -sname otp -setcookie AtomVM -eval \""
-            "register(beam, self()),"
-            "F = fun(G) ->"
-            " receive"
-            "   {Caller, ping} -> Caller ! {self(), pong}, G(G);"
-            "   {Caller, quit} -> Caller ! {self(), quit}"
-            "   after 5000 -> timeout"
-            " end "
-            "end, "
-            "F(F).\" -s init stop -noshell"
-        )
-    end),
+    {Pid, MonitorRef} = spawn_opt(
+        fun() ->
+            [] = execute_command(
+                Platform,
+                "erl -sname otp -setcookie AtomVM -eval \""
+                "register(beam, self()),"
+                "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"
+            )
+        end,
+        [link, monitor]
+    ),
     % Wait sufficiently for beam to be up, without connecting to it since
     % that's part of the test
     timer:sleep(1000),
@@ -200,6 +204,79 @@ test_autoconnect_to_beam(Platform) ->
             {OTPPid, quit} -> ok
         after 5000 -> timeout
         end,
+    normal =
+        receive
+            {'DOWN', MonitorRef, process, Pid, Reason} -> Reason
+        after 5000 -> timeout
+        end,
+    net_kernel:stop(),
+    ok.
+
+test_groupleader(Platform) ->
+    {ok, _NetKernelPid} = net_kernel_start(Platform, atomvm),
+    Node = node(),
+    erlang:set_cookie(Node, 'AtomVM'),
+    register(atomvm, self()),
+    Parent = self(),
+    {Pid, MonitorRef} = spawn_opt(
+        fun() ->
+            Result = execute_command(
+                Platform,
+                "erl -sname otp -setcookie AtomVM -eval \""
+                "{atomvm, '" ++ atom_to_list(Node) ++
+                    "'} ! {beam, self()}, "
+                    "F = fun(G) ->"
+                    " receive"
+                    "   {Caller, apply, M, F, A} -> Result = apply(M, F, A), Caller ! {self(), Result}, G(G);"
+                    "   {Caller, quit} -> Caller ! {self(), quit}"
+                    "   after 5000 -> exit(timeout)"
+                    " end "
+                    "end, "
+                    "F(F).\" -s init stop -noshell"
+            ),
+            Parent ! {io_result, Result}
+        end,
+        [link, monitor]
+    ),
+    BeamMainPid =
+        receive
+            {beam, BeamMainPid0} ->
+                BeamMainPid0;
+            {io_result, Result0} ->
+                io:format("~s\n", [Result0]),
+                exit(timeout)
+        after 5000 -> exit(timeout)
+        end,
+    BeamMainPid ! {self(), apply, rpc, call, [Node, io, format, ["hello group leader"]]},
+    ok =
+        receive
+            {BeamMainPid, Result} ->
+                Result;
+            {io_result, Result1} ->
+                io:format("~s\n", [Result1]),
+                exit(timeout)
+        after 5000 -> exit(timeout)
+        end,
+    BeamMainPid ! {self(), quit},
+    ok =
+        receive
+            {BeamMainPid, quit} ->
+                ok;
+            {io_result, Result2} ->
+                io:format("~s\n", [Result2]),
+                exit(timeout)
+        after 5000 -> timeout
+        end,
+    "hello group leader" =
+        receive
+            {io_result, IOResult} -> IOResult
+        after 5000 -> timeout
+        end,
+    normal =
+        receive
+            {'DOWN', MonitorRef, process, Pid, Reason} -> Reason
+        after 5000 -> timeout
+        end,
     net_kernel:stop(),
     ok.