Distribution: add support for rpc from other nodes

Signed-off-by: Paul Guyot <pguyot@kallisys.net>

Author: pguyot

libs/estdlib/src/CMakeLists.txt

@@ -31,6 +31,7 @@ set(ERLANG_MODULES
     crypto
     dist_util
     erl_epmd
+    erpc
     erts_debug
     ets
     gen_event

src/libAtomVM/defaultatoms.def

@@ -178,3 +178,11 @@ X(INET_ATOM, "\x4", "inet")
 X(TIMEOUT_ATOM, "\x7", "timeout")
 
 X(DIST_DATA_ATOM, "\x9", "dist_data")
+
+X(REQUEST_ATOM, "\x7", "request")
+X(REPLY_TAG_ATOM, "\x9", "reply_tag")
+X(SPAWN_REPLY_ATOM, "\xB", "spawn_reply")
+X(REPLY_ATOM, "\x5", "reply")
+X(YES_ATOM, "\x3", "yes")
+X(NO_ATOM, "\x2", "no")
+X(ERROR_ONLY_ATOM, "\xA", "error_only")

src/libAtomVM/dist_nifs.c

@@ -70,6 +70,11 @@ enum
     OPERATION_ALIAS_SEND_TT = 34,
 };
 
+enum {
+    SPAWN_REPLY_FLAGS_LINK_CREATED = 1,
+    SPAWN_REPLY_FLAGS_MONITOR_CREATED = 2,
+};
+
 struct DistributionPacket
 {
     struct ListHead head;
@@ -318,6 +323,37 @@ static term nif_erlang_dist_ctrl_get_data(Context *ctx, int argc, term argv[])
     return result;
 }
 
+term dist_monitor(struct DistConnection *conn_obj, term from_pid, term target_proc, term monitor_ref, Context *ctx)
+{
+    int target_process_id = 0;
+    if (term_is_local_pid(target_proc)) {
+        target_process_id = term_to_local_process_id(target_proc);
+    } else if (term_is_atom(target_proc)) {
+        target_process_id = globalcontext_get_registered_process(ctx->global, term_to_atom_index(target_proc));
+    } else {
+        RAISE_ERROR(BADARG_ATOM);
+    }
+    struct RemoteMonitor *monitor = malloc(sizeof(struct RemoteMonitor));
+    monitor->target_proc = target_proc;
+    monitor->pid_number = term_get_external_pid_process_id(from_pid);
+    monitor->pid_serial = term_get_external_pid_serial(from_pid);
+    monitor->ref_len = term_get_external_reference_len(monitor_ref);
+    memcpy(monitor->ref_words, term_get_external_reference_words(monitor_ref), sizeof(uint32_t) * monitor->ref_len);
+    if (target_process_id) {
+        synclist_append(&conn_obj->remote_monitors, &monitor->head);
+        ErlNifPid target_process_pid = target_process_id;
+        if (UNLIKELY(enif_monitor_process(erl_nif_env_from_context(ctx), conn_obj, &target_process_pid, &monitor->process_monitor) != 0)) {
+            synclist_remove(&conn_obj->remote_monitors, &monitor->head);
+            dist_enqueue_monitor_exit_message(monitor, NOPROC_ATOM, conn_obj, ctx->global);
+            free(monitor);
+        }
+    } else {
+        dist_enqueue_monitor_exit_message(monitor, NOPROC_ATOM, conn_obj, ctx->global);
+        free(monitor);
+    }
+    return OK_ATOM;
+}
+
 static term nif_erlang_dist_ctrl_put_data(Context *ctx, int argc, term argv[])
 {
     UNUSED(argc);
@@ -369,32 +405,7 @@ static term nif_erlang_dist_ctrl_put_data(Context *ctx, int argc, term argv[])
             term from_pid = term_get_tuple_element(control, 1);
             term target_proc = term_get_tuple_element(control, 2);
             term monitor_ref = term_get_tuple_element(control, 3);
-            int target_process_id = 0;
-            if (term_is_local_pid(target_proc)) {
-                target_process_id = term_to_local_process_id(target_proc);
-            } else if (term_is_atom(target_proc)) {
-                target_process_id = globalcontext_get_registered_process(ctx->global, term_to_atom_index(target_proc));
-            } else {
-                RAISE_ERROR(BADARG_ATOM);
-            }
-            struct RemoteMonitor *monitor = malloc(sizeof(struct RemoteMonitor));
-            monitor->target_proc = target_proc;
-            monitor->pid_number = term_get_external_pid_process_id(from_pid);
-            monitor->pid_serial = term_get_external_pid_serial(from_pid);
-            monitor->ref_len = term_get_external_reference_len(monitor_ref);
-            memcpy(monitor->ref_words, term_get_external_reference_words(monitor_ref), sizeof(uint32_t) * monitor->ref_len);
-            if (target_process_id) {
-                synclist_append(&conn_obj->remote_monitors, &monitor->head);
-                ErlNifPid target_process_pid = target_process_id;
-                if (UNLIKELY(enif_monitor_process(erl_nif_env_from_context(ctx), conn_obj, &target_process_pid, &monitor->process_monitor) != 0)) {
-                    synclist_remove(&conn_obj->remote_monitors, &monitor->head);
-                    dist_enqueue_monitor_exit_message(monitor, NOPROC_ATOM, conn_obj, ctx->global);
-                    free(monitor);
-                }
-            } else {
-                dist_enqueue_monitor_exit_message(monitor, NOPROC_ATOM, conn_obj, ctx->global);
-                free(monitor);
-            }
+            dist_monitor(conn_obj, from_pid, target_proc, monitor_ref, ctx);
 
             break;
         }
@@ -421,6 +432,52 @@ static term nif_erlang_dist_ctrl_put_data(Context *ctx, int argc, term argv[])
             synclist_unlock(&conn_obj->remote_monitors);
             break;
         }
+        case OPERATION_SPAWN_REQUEST: {
+            if (UNLIKELY(arity != 6)) {
+                RAISE_ERROR(BADARG_ATOM);
+            }
+            term roots[4];
+            roots[0] = argv[0];
+            roots[1] = argv[1];
+            roots[2] = control;
+            term arglist = externalterm_to_term_with_roots(data + 1 + bytes_read, binary_len - 1 - bytes_read, ctx, ExternalTermCopy, &bytes_read, 3, roots);
+            roots[3] = arglist;
+            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)) {
+                RAISE_ERROR(OUT_OF_MEMORY_ATOM);
+            }
+            term mfa = term_get_tuple_element(control, 4);
+            if (UNLIKELY(!term_is_tuple(mfa) || term_get_tuple_arity(mfa) != 3)) {
+                RAISE_ERROR(BADARG_ATOM);
+            }
+            if (UNLIKELY(!term_is_list(arglist))) {
+                RAISE_ERROR(BADARG_ATOM);
+            }
+            term reqid = term_get_tuple_element(control, 1);
+            term from = term_get_tuple_element(control, 2);
+            if (UNLIKELY(!term_is_pid(from))) {
+                RAISE_ERROR(BADARG_ATOM);
+            }
+            // term groupleader = term_get_tuple_element(control, 3);
+            term options = term_get_tuple_element(control, 5);
+
+            term request_tuple = term_alloc_tuple(4, &ctx->heap);
+            term_put_tuple_element(request_tuple, 0, argv[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 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);
+            term spawn_opts = term_list_prepend(request_opt, term_nil(), &ctx->heap);
+
+            // reuse roots for args
+            roots[0] = term_get_tuple_element(mfa, 0);
+            roots[1] = term_get_tuple_element(mfa, 1);
+            roots[2] = arglist;
+            roots[3] = spawn_opts;
+            nif_erlang_spawn_opt(ctx, 4, roots);
+            break;
+        }
         default:
             printf("Unknown distribution protocol operation id %d\n", (int) term_to_int(operation));
             RAISE_ERROR(BADARG_ATOM);
@@ -446,6 +503,24 @@ void dist_send_message(term external_pid, term payload, Context *ctx)
     synclist_unlock(&ctx->global->dist_connections);
 }
 
+void dist_spawn_reply(term req_id, term to_pid, bool link, bool monitor, term result, struct DistConnection *connection, GlobalContext *global)
+{
+    int flags =
+        (link ? SPAWN_REPLY_FLAGS_LINK_CREATED : 0)
+        | (monitor ? SPAWN_REPLY_FLAGS_MONITOR_CREATED : 0);
+    // allocate tuple
+    BEGIN_WITH_STACK_HEAP(TUPLE_SIZE(5), heap)
+    term control_message = term_alloc_tuple(5, &heap);
+    term_put_tuple_element(control_message, 0, term_from_int(OPERATION_SPAWN_REPLY));
+    term_put_tuple_element(control_message, 1, req_id);
+    term_put_tuple_element(control_message, 2, to_pid);
+    term_put_tuple_element(control_message, 3, term_from_int(flags));
+    term_put_tuple_element(control_message, 4, result);
+
+    dist_enqueue_message(control_message, term_invalid_term(), connection, global);
+    END_WITH_STACK_HEAP(heap, global)
+}
+
 const struct Nif setnode_3_nif = {
     .base.type = NIFFunctionType,
     .nif_ptr = nif_erlang_setnode_3

src/libAtomVM/dist_nifs.h

@@ -41,8 +41,35 @@ extern const struct Nif dist_ctrl_get_data_notification_nif;
 extern const struct Nif dist_ctrl_get_data_nif;
 extern const struct Nif dist_ctrl_put_data_nif;
 
+struct DistConnection;
+
 void dist_send_message(term external_pid, term payload, Context *ctx);
 
+/**
+ * @doc Setup a monitor on a local process for a distributed process.
+ * @end
+ * @param conn_obj object of the connection
+ * @param from_pid remote pid setting up the monitor
+ * @param target_proc atom (for registered process) or pid of the local
+ * process to monitor
+ * @param monitor_ref reference used for monitor
+ * @param ctx context for memory allocation
+ */
+term dist_monitor(struct DistConnection *conn_obj, term from_pid, term target_proc, term monitor_ref, Context *ctx);
+
+/**
+ * @doc Send a spawn reply signal to a node
+ * @end
+ * @param conn_obj object of the connection
+ * @param req_id reference identifying the request
+ * @param to_pid (remote) process id identifying the caller
+ * @param link if a link was created
+ * @param monitor if a monitor was created
+ * @param result pid of the spawned process or atom for an error
+ * @param ctx context for memory allocation
+ */
+void dist_spawn_reply(term req_id, term to_pid, bool link, bool monitor, term result, struct DistConnection *connection, GlobalContext *global);
+
 #ifdef __cplusplus
 }
 #endif

src/libAtomVM/nifs.c

@@ -41,6 +41,7 @@
 #include "defaultatoms.h"
 #include "dictionary.h"
 #include "dist_nifs.h"
+#include "erl_nif_priv.h"
 #include "ets.h"
 #include "externalterm.h"
 #include "globalcontext.h"
@@ -134,7 +135,7 @@ static term nif_erlang_register_2(Context *ctx, int argc, term argv[]);
 static term nif_erlang_unregister_1(Context *ctx, int argc, term argv[]);
 static term nif_erlang_send_2(Context *ctx, int argc, term argv[]);
 static term nif_erlang_setelement_3(Context *ctx, int argc, term argv[]);
-static term nif_erlang_spawn_opt(Context *ctx, int argc, term argv[]);
+// static term nif_erlang_spawn_opt(Context *ctx, int argc, term argv[]);
 static term nif_erlang_spawn_fun_opt(Context *ctx, int argc, term argv[]);
 static term nif_erlang_whereis_1(Context *ctx, int argc, term argv[]);
 static term nif_erlang_system_time_1(Context *ctx, int argc, term argv[]);
@@ -1222,6 +1223,7 @@ static term do_spawn(Context *ctx, Context *new_ctx, term opts_term)
     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);
 
     if (min_heap_size_term != term_nil()) {
         if (UNLIKELY(!term_is_integer(min_heap_size_term))) {
@@ -1303,6 +1305,34 @@ static term do_spawn(Context *ctx, Context *new_ctx, term opts_term)
         term_put_tuple_element(pid_ref_tuple, 1, ref);
 
         return pid_ref_tuple;
+    } else if (UNLIKELY(request_term != term_nil())) {
+        // Handling of spawn_request
+        // spawn_request requires that the reply is enqueued before
+        // any message from the spawned process
+
+        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);
+        monitor_term = interop_proplist_get_value(request_opts, MONITOR_ATOM);
+        // link_term = interop_proplist_get_value(request_opts, LINK_ATOM);
+
+        void *rsrc_obj_ptr;
+        if (UNLIKELY(!enif_get_resource(erl_nif_env_from_context(ctx), dhandle, ctx->global->dist_connection_resource_type, &rsrc_obj_ptr))) {
+            RAISE_ERROR(BADARG_ATOM);
+        }
+        struct DistConnection *conn_obj = (struct DistConnection *) rsrc_obj_ptr;
+
+        dist_spawn_reply(request_ref, request_from, false, monitor_term != term_nil(), new_pid, conn_obj, ctx->global);
+
+        // Also setup monitor, if any.
+        if (monitor_term != term_nil()) {
+            printf("setting up the monitor\n");
+            dist_monitor(conn_obj, request_from, new_pid, request_ref, ctx);
+        }
+
+        scheduler_init_ready(new_ctx);
+        return new_pid;
     } else {
         scheduler_init_ready(new_ctx);
         return new_pid;
@@ -1360,7 +1390,7 @@ static term nif_erlang_spawn_fun_opt(Context *ctx, int argc, term argv[])
     return do_spawn(ctx, new_ctx, opts_term);
 }
 
-static term nif_erlang_spawn_opt(Context *ctx, int argc, term argv[])
+term nif_erlang_spawn_opt(Context *ctx, int argc, term argv[])
 {
     UNUSED(argc);
 

src/libAtomVM/nifs.h

@@ -48,6 +48,8 @@ extern "C" {
 
 const struct Nif *nifs_get(AtomString module, AtomString function, int arity);
 
+// spawn opt is used by distribution nifs
+term nif_erlang_spawn_opt(Context *ctx, int argc, term argv[]);
 #ifdef __cplusplus
 }
 #endif

tests/libs/estdlib/test_net_kernel.erl

@@ -33,6 +33,7 @@ test() ->
             ok = setup(Platform),
             ok = test_ping_from_beam(Platform),
             ok = test_fail_with_wrong_cookie(Platform),
+            ok = test_rpc_from_beam(Platform),
             ok;
         false ->
             io:format("~s: skipped\n", [?MODULE]),
@@ -92,6 +93,19 @@ test_fail_with_wrong_cookie(Platform) ->
     net_kernel:stop(),
     ok.
 
+test_rpc_from_beam(Platform) ->
+    {ok, _NetKernelPid} = net_kernel:start(atomvm, #{name_domain => shortnames}),
+    Node = node(),
+    erlang:set_cookie('AtomVM'),
+    Result = execute_command(
+        Platform,
+        "erl -sname otp -setcookie AtomVM -eval \"R = rpc:call('" ++ atom_to_list(Node) ++
+            "', erlang, system_info, [machine]), erlang:display(R).\" -s init stop -noshell"
+    ),
+    true = Result =:= lists:flatten(io_lib:format("~p\r\n", [Platform])),
+    net_kernel:stop(),
+    ok.
+
 % On AtomVM, we need to start kernel.
 setup("BEAM") ->
     ok;