Add support for encoded local pids in external terms

Also add partial support for external pids.

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

Author: pguyot

CHANGELOG.md

@@ -9,6 +9,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 ### Added
 - Added a limited implementation of the OTP `ets` interface
 - Added `code:all_loaded/0` and `code:all_available/0`
+- Added support for external pids and encoded pids in external terms
 
 ## [0.6.6] - Unreleased
 

src/libAtomVM/ets_hashtable.c

@@ -285,7 +285,7 @@ static uint32_t hash_term_incr(term t, int32_t h, GlobalContext *global)
         return hash_integer(t, h, global);
     } else if (term_is_float(t)) {
         return hash_float(t, h, global);
-    } else if (term_is_pid(t)) {
+    } else if (term_is_local_pid(t)) {
         return hash_pid(t, h, global);
     } else if (term_is_reference(t)) {
         return hash_reference(t, h, global);

src/libAtomVM/externalterm.c

@@ -29,13 +29,17 @@
 #include <stdlib.h>
 
 #include "bitstring.h"
+#include "defaultatoms.h"
+#include "term.h"
 #include "unicode.h"
 #include "utils.h"
 
 #define NEW_FLOAT_EXT 70
+#define NEW_PID_EXT 88
 #define SMALL_INTEGER_EXT 97
 #define INTEGER_EXT 98
 #define ATOM_EXT 100
+#define PID_EXT 103
 #define SMALL_TUPLE_EXT 104
 #define LARGE_TUPLE_EXT 105
 #define NIL_EXT 106
@@ -390,6 +394,33 @@ static int serialize_term(uint8_t *buf, term t, GlobalContext *glb)
             k += serialize_term(IS_NULL_PTR(buf) ? NULL : buf + k, mfa, glb);
         }
         return k;
+    } else if (term_is_local_pid(t)) {
+        if (!IS_NULL_PTR(buf)) {
+            buf[0] = NEW_PID_EXT;
+        }
+        size_t k = 1;
+        term node_name = glb->node_name;
+        uint32_t creation = node_name == NONODE_AT_NOHOST_ATOM ? 0 : glb->creation;
+        k += serialize_term(IS_NULL_PTR(buf) ? NULL : buf + k, node_name, glb);
+        if (!IS_NULL_PTR(buf)) {
+            WRITE_32_UNALIGNED(buf + k, term_to_local_process_id(t));
+            WRITE_32_UNALIGNED(buf + k + 4, 0); // serial
+            WRITE_32_UNALIGNED(buf + k + 8, creation);
+        }
+        return k + 12;
+    } else if (term_is_external_pid(t)) {
+        if (!IS_NULL_PTR(buf)) {
+            buf[0] = NEW_PID_EXT;
+        }
+        size_t k = 1;
+        term node = term_to_external_node(t);
+        k += serialize_term(IS_NULL_PTR(buf) ? NULL : buf + k, node, glb);
+        if (!IS_NULL_PTR(buf)) {
+            WRITE_32_UNALIGNED(buf + k, term_to_external_pid_process_id(t));
+            WRITE_32_UNALIGNED(buf + k + 4, term_to_external_pid_serial(t));
+            WRITE_32_UNALIGNED(buf + k + 8, term_to_external_node_creation(t));
+        }
+        return k + 12;
     } else {
         fprintf(stderr, "Unknown external term type: %" TERM_U_FMT "\n", t);
         AVM_ABORT();
@@ -659,6 +690,32 @@ static term parse_external_terms(const uint8_t *external_term_buf, size_t *eterm
             return term_from_atom_index(global_atom_id);
         }
 
+        case NEW_PID_EXT: {
+            size_t node_size;
+            term node = parse_external_terms(external_term_buf + 1, &node_size, copy, heap, glb);
+            if (UNLIKELY(!term_is_atom(node))) {
+                return term_invalid_term();
+            }
+            uint32_t number = READ_32_UNALIGNED(external_term_buf + node_size + 1);
+            uint32_t serial = READ_32_UNALIGNED(external_term_buf + node_size + 5);
+            uint32_t creation = READ_32_UNALIGNED(external_term_buf + node_size + 9);
+            *eterm_size = node_size + 13;
+            if (node != NONODE_AT_NOHOST_ATOM) {
+                term this_node = glb->node_name;
+                uint32_t this_creation = this_node == NONODE_AT_NOHOST_ATOM ? 0 : glb->creation;
+                if (node == this_node && creation == this_creation) {
+                    return term_from_local_process_id(number);
+                } else {
+                    return term_from_external_process_id(node, number, serial, creation, heap);
+                }
+            } else {
+                if (UNLIKELY(serial != 0 || creation != 0)) {
+                    return term_invalid_term();
+                }
+                return term_from_local_process_id(number);
+            }
+        }
+
         default:
             return term_invalid_term();
     }
@@ -948,6 +1005,33 @@ static int calculate_heap_usage(const uint8_t *external_term_buf, size_t remaini
             return 0;
         }
 
+        case NEW_PID_EXT: {
+            if (UNLIKELY(remaining < 1)) {
+                return INVALID_TERM_SIZE;
+            }
+            remaining -= 1;
+            int buf_pos = 1;
+            size_t heap_size = EXTERNAL_PID_SIZE;
+            size_t node_size = 0;
+            int u = calculate_heap_usage(external_term_buf + buf_pos, remaining, &node_size, copy);
+            if (UNLIKELY(u == INVALID_TERM_SIZE)) {
+                return INVALID_TERM_SIZE;
+            }
+            if (external_term_buf[1] == SMALL_ATOM_UTF8_EXT) {
+                // Check if it's non-distributed node, in which case it's always a local pid
+                if (external_term_buf[2] == strlen("nonode@nohost") && memcmp(external_term_buf + 3, "nonode@nohost", strlen("nonode@nohost")) == 0) {
+                    heap_size = 0;
+                }
+                // If this is our node, but we're distributed, we'll allocate more memory and may not use it.
+                // This way we're sure to not go out of bounds if distribution changes between now and when we deserialize
+            } else if (UNLIKELY(external_term_buf[1] != ATOM_EXT)) {
+                return INVALID_TERM_SIZE;
+            }
+            buf_pos += node_size;
+            *eterm_size = buf_pos + 12;
+            return heap_size + u;
+        }
+
         default:
             return INVALID_TERM_SIZE;
     }

src/libAtomVM/memory.c

@@ -476,7 +476,11 @@ unsigned long memory_estimate_usage(term t)
         } else if (term_is_nil(t)) {
             t = temp_stack_pop(&temp_stack);
 
-        } else if (term_is_pid(t)) {
+        } else if (term_is_local_pid(t)) {
+            t = temp_stack_pop(&temp_stack);
+
+        } else if (term_is_external_pid(t)) {
+            acc += EXTERNAL_PID_SIZE;
             t = temp_stack_pop(&temp_stack);
 
         } else if (term_is_nonempty_list(t)) {
@@ -587,7 +591,7 @@ static void memory_scan_and_copy(HeapFragment *old_fragment, term *mem_start, co
             TRACE("Found NIL (%" TERM_X_FMT ")\n", t);
             ptr++;
 
-        } else if (term_is_pid(t)) {
+        } else if (term_is_local_pid(t)) {
             TRACE("Found PID (%" TERM_X_FMT ")\n", t);
             ptr++;
 
@@ -620,6 +624,10 @@ static void memory_scan_and_copy(HeapFragment *old_fragment, term *mem_start, co
                     TRACE("- Found ref.\n");
                     break;
 
+                case TERM_BOXED_EXTERNAL_PID:
+                    TRACE("- Found external pid.\n");
+                    break;
+
                 case TERM_BOXED_FUN: {
                     int fun_size = term_get_size_from_boxed_header(t);
                     TRACE("- Found fun, size: %i.\n", fun_size);
@@ -740,6 +748,10 @@ static void memory_scan_and_rewrite(size_t count, term *terms, const term *old_s
                     ptr += term_get_size_from_boxed_header(t);
                     break;
 
+                case TERM_BOXED_EXTERNAL_PID:
+                    ptr += EXTERNAL_PID_SIZE - 1;
+                    break;
+
                 case TERM_BOXED_FUN:
                     // Skip header and module and process next terms
                     ptr++;
@@ -810,7 +822,7 @@ HOT_FUNC static term memory_shallow_copy_term(HeapFragment *old_fragment, term t
     } else if (term_is_nil(t)) {
         return t;
 
-    } else if (term_is_pid(t)) {
+    } else if (term_is_local_pid(t)) {
         return t;
 
     } else if (term_is_cp(t)) {

src/libAtomVM/nifs.c

@@ -996,7 +996,7 @@ static term nif_erlang_register_2(Context *ctx, int argc, term argv[])
     term reg_name_term = argv[0];
     VALIDATE_VALUE(reg_name_term, term_is_atom);
     term pid_or_port_term = argv[1];
-    VALIDATE_VALUE(pid_or_port_term, term_is_pid);
+    VALIDATE_VALUE(pid_or_port_term, term_is_local_pid);
 
     int atom_index = term_to_atom_index(reg_name_term);
     int32_t pid = term_to_local_process_id(pid_or_port_term);
@@ -1400,7 +1400,7 @@ static term nif_erlang_send_2(Context *ctx, int argc, term argv[])
     term target = argv[0];
     GlobalContext *glb = ctx->global;
 
-    if (term_is_pid(target)) {
+    if (term_is_local_pid(target)) {
         int32_t local_process_id = term_to_local_process_id(target);
 
         globalcontext_send_message(glb, local_process_id, argv[1]);
@@ -2741,7 +2741,7 @@ static term nif_erlang_process_flag(Context *ctx, int argc, term argv[])
         flag = argv[1];
         value = argv[2];
 
-        VALIDATE_VALUE(pid, term_is_pid);
+        VALIDATE_VALUE(pid, term_is_local_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)) {
@@ -3191,7 +3191,7 @@ static term nif_binary_split(Context *ctx, int argc, term argv[])
 
     if (num_segments == 1) {
         // not found
-        if (UNLIKELY(memory_ensure_free_with_roots(ctx, 2, 1, argv, MEMORY_CAN_SHRINK) != MEMORY_GC_OK)) {
+        if (UNLIKELY(memory_ensure_free_with_roots(ctx, LIST_SIZE(1, 0), 1, argv, MEMORY_CAN_SHRINK) != MEMORY_GC_OK)) {
             RAISE_ERROR(OUT_OF_MEMORY_ATOM);
         }
 
@@ -3443,11 +3443,11 @@ static term nif_erlang_pid_to_list(Context *ctx, int argc, term argv[])
 
     term t = argv[0];
     VALIDATE_VALUE(t, term_is_pid);
+    size_t max_len = term_is_external(t) ? EXTERNAL_PID_AS_CSTRING_LEN : LOCAL_PID_AS_CSTRING_LEN;
 
-    char buf[PID_AS_CSTRING_LEN];
-    int str_len = term_snprint(buf, PID_AS_CSTRING_LEN, t, ctx->global);
+    char buf[max_len];
+    int str_len = term_snprint(buf, max_len, t, ctx->global);
     if (UNLIKELY(str_len < 0)) {
-        // TODO: change to internal error or something like that
         RAISE_ERROR(OUT_OF_MEMORY_ATOM);
     }
 
@@ -3559,7 +3559,7 @@ static term nif_erlang_garbage_collect(Context *ctx, int argc, term argv[])
     } else {
         // argc == 1
         term t = argv[0];
-        VALIDATE_VALUE(t, term_is_pid);
+        VALIDATE_VALUE(t, term_is_local_pid);
 
         int local_id = term_to_local_process_id(t);
         Context *target = globalcontext_get_process_lock(ctx->global, local_id);
@@ -3602,7 +3602,7 @@ static term nif_erlang_exit(Context *ctx, int argc, term argv[])
         RAISE(LOWERCASE_EXIT_ATOM, reason);
     } else {
         term target_process = argv[0];
-        VALIDATE_VALUE(target_process, term_is_pid);
+        VALIDATE_VALUE(target_process, term_is_local_pid);
         term reason = argv[1];
         GlobalContext *glb = ctx->global;
         Context *target = globalcontext_get_process_lock(glb, term_to_local_process_id(target_process));
@@ -3715,7 +3715,7 @@ static term nif_erlang_monitor(Context *ctx, int argc, term argv[])
         RAISE_ERROR(BADARG_ATOM);
     }
 
-    VALIDATE_VALUE(target_pid, term_is_pid);
+    VALIDATE_VALUE(target_pid, term_is_local_pid);
 
     int local_process_id = term_to_local_process_id(target_pid);
     Context *target = globalcontext_get_process_lock(ctx->global, local_process_id);
@@ -3783,7 +3783,7 @@ static term nif_erlang_link(Context *ctx, int argc, term argv[])
 
     term target_pid = argv[0];
 
-    VALIDATE_VALUE(target_pid, term_is_pid);
+    VALIDATE_VALUE(target_pid, term_is_local_pid);
 
     int local_process_id = term_to_local_process_id(target_pid);
     Context *target = globalcontext_get_process_lock(ctx->global, local_process_id);
@@ -3814,7 +3814,7 @@ static term nif_erlang_unlink(Context *ctx, int argc, term argv[])
 
     term target_pid = argv[0];
 
-    VALIDATE_VALUE(target_pid, term_is_pid);
+    VALIDATE_VALUE(target_pid, term_is_local_pid);
 
     int local_process_id = term_to_local_process_id(target_pid);
     Context *target = globalcontext_get_process_lock(ctx->global, local_process_id);
@@ -3845,8 +3845,8 @@ static term nif_erlang_group_leader(Context *ctx, int argc, term argv[])
     } else {
         term leader = argv[0];
         term pid = argv[1];
-        VALIDATE_VALUE(pid, term_is_pid);
-        VALIDATE_VALUE(leader, term_is_pid);
+        VALIDATE_VALUE(pid, term_is_local_pid);
+        VALIDATE_VALUE(leader, term_is_local_pid);
 
         int local_process_id = term_to_local_process_id(pid);
         Context *target = globalcontext_get_process_lock(ctx->global, local_process_id);

src/libAtomVM/opcodesswitch.h

@@ -2384,7 +2384,7 @@ HOT_FUNC int scheduler_entry_point(GlobalContext *glb)
                 #ifdef IMPL_EXECUTE_LOOP
                     term recipient_term = x_regs[0];
                     int local_process_id;
-                    if (term_is_pid(recipient_term)) {
+                    if (term_is_local_pid(recipient_term)) {
                         local_process_id = term_to_local_process_id(recipient_term);
                     } else if (term_is_atom(recipient_term)) {
                         local_process_id = globalcontext_get_registered_process(ctx->global, term_to_atom_index(recipient_term));
@@ -2984,7 +2984,7 @@ HOT_FUNC int scheduler_entry_point(GlobalContext *glb)
                 #ifdef IMPL_EXECUTE_LOOP
                     TRACE("is_port/2, label=%i, arg1=%lx\n", label, arg1);
 
-                    if (term_is_pid(arg1)) {
+                    if (term_is_local_pid(arg1)) {
                         int local_process_id = term_to_local_process_id(arg1);
                         Context *target = globalcontext_get_process_lock(ctx->global, local_process_id);
                         bool is_port_driver = false;

src/libAtomVM/posix_nifs.c

@@ -427,7 +427,7 @@ static term nif_atomvm_posix_write(Context *ctx, int argc, term argv[])
 static term nif_atomvm_posix_select(Context *ctx, term argv[], enum ErlNifSelectFlags mode)
 {
     term process_pid_term = argv[1];
-    VALIDATE_VALUE(process_pid_term, term_is_pid);
+    VALIDATE_VALUE(process_pid_term, term_is_local_pid);
     int32_t process_pid = term_to_local_process_id(process_pid_term);
     term select_ref_term = argv[2];
     if (select_ref_term != UNDEFINED_ATOM) {

src/libAtomVM/term.c

@@ -189,10 +189,17 @@ int term_funprint(PrinterFun *fun, term t, const GlobalContext *global)
             ret += printed;
             return ret;
         }
-    } else if (term_is_pid(t)) {
+    } else if (term_is_local_pid(t)) {
         int32_t process_id = term_to_local_process_id(t);
         return fun->print(fun, "<0.%" PRIu32 ".0>", process_id);
 
+    } else if (term_is_external_pid(t)) {
+        uint32_t node_atom_index = term_to_atom_index(term_to_external_node(t));
+        uint32_t number = term_to_external_pid_process_id(t);
+        uint32_t serial = term_to_external_pid_serial(t);
+        // creation is not printed
+        return fun->print(fun, "<%" PRIu32 ".%" PRIu32 ".%" PRIu32 ">", node_atom_index, number, serial);
+
     } else if (term_is_function(t)) {
         const term *boxed_value = term_to_const_term_ptr(t);
 
@@ -664,11 +671,45 @@ TermCompareResult term_compare(term t, term other, TermCompareOpts opts, GlobalC
             result = (atom_cmp_result > 0) ? TermGreaterThan : TermLessThan;
             break;
 
-        } else if (term_is_pid(t) && term_is_pid(other)) {
+        } else if (term_is_external_pid(t) && term_is_external_pid(other)) {
+            term node = term_to_external_node(t);
+            term other_node = term_to_external_node(other);
+            if (node == other_node) {
+                uint32_t creation = term_to_external_node_creation(t);
+                uint32_t other_creation = term_to_external_node_creation(other);
+                if (creation == other_creation) {
+                    uint32_t serial = term_to_external_pid_serial(t);
+                    uint32_t other_serial = term_to_external_pid_serial(other);
+                    if (serial == other_serial) {
+                        uint32_t process_id = term_to_external_pid_process_id(t);
+                        uint32_t other_process_id = term_to_external_pid_process_id(other);
+                        if (process_id == other_process_id) {
+                            CMP_POP_AND_CONTINUE();
+                        } else {
+                            result = (process_id > other_process_id) ? TermGreaterThan : TermLessThan;
+                            break;
+                        }
+                    } else {
+                        result = (serial > other_serial) ? TermGreaterThan : TermLessThan;
+                        break;
+                    }
+                } else {
+                    result = (creation > other_creation) ? TermGreaterThan : TermLessThan;
+                    break;
+                }
+            } else {
+                result = (node > other_node) ? TermGreaterThan : TermLessThan;
+                break;
+            }
+        } else if (term_is_local_pid(t) && term_is_local_pid(other)) {
             //TODO: handle ports
             result = (t > other) ? TermGreaterThan : TermLessThan;
             break;
 
+        } else if (term_is_pid(t) && term_is_pid(other)) {
+            result = term_is_local_pid(other) ? TermGreaterThan : TermLessThan;
+            break;
+
         } else {
             result = (term_type_to_index(t) > term_type_to_index(other)) ? TermGreaterThan : TermLessThan;
             break;