Forward port changes from v0.6 release branch

Merge #1635 "implement link protocol with unlink id and acknowledgement" from
release-0.6.

Author: bettio

CHANGELOG.md

@@ -102,6 +102,7 @@ memory error
 - Fixed segfault when calling `lists:reverse/1` (#1600)
 - Fixed nif_atomvm_posix_read GC bug
 - Fixed `erlang:is_number/1` function, now returns true also for floats
+- Fixed unlink protocol and add support for `link/1` on ports
 
 ### Changed
 

src/libAtomVM/context.c

@@ -149,18 +149,45 @@ struct Context
 typedef struct Context Context;
 #endif
 
-#define CONTEXT_MONITOR_RESOURCE_TAG 0x2
-#define CONTEXT_MONITOR_MONITORED_PID_TAG 0x3
-#define CONTEXT_MONITOR_MONITORING_PID_TAG 0x1
+enum ContextMonitorType
+{
+    CONTEXT_MONITOR_LINK_LOCAL,
+    CONTEXT_MONITOR_MONITORING_LOCAL,
+    CONTEXT_MONITOR_MONITORED_LOCAL,
+    CONTEXT_MONITOR_RESOURCE,
+};
+
+#define UNLINK_ID_LINK_ACTIVE 0x0
 
 /**
  * @brief A regular monitor or a half link.
  */
 struct Monitor
 {
     struct ListHead monitor_list_head;
-    uint64_t ref_ticks; // 0 for links
-    term monitor_obj; // pid for links, CONTEXT_MONITOR_*_TAG for monitors
+    enum ContextMonitorType monitor_type;
+};
+
+struct LinkLocalMonitor
+{
+    struct Monitor monitor;
+    uint64_t unlink_id;
+    term link_local_process_id;
+};
+
+struct MonitorLocalMonitor
+{
+    struct Monitor monitor;
+    uint64_t ref_ticks;
+    term monitor_obj;
+};
+
+// The other half is called ResourceMonitor and is a linked list of resources
+struct ResourceContextMonitor
+{
+    struct Monitor monitor;
+    uint64_t ref_ticks;
+    void *resource_obj;
 };
 
 struct ExtendedRegister
@@ -363,9 +390,10 @@ void context_process_kill_signal(Context *ctx, struct TermSignal *signal);
  * @brief Process a process info request signal.
  *
  * @param ctx the context being executed
- * @param signal the kill message
+ * @param signal the process info signal
+ * @param process_table_locked whether process table is already locked
  */
-void context_process_process_info_request_signal(Context *ctx, struct BuiltInAtomRequestSignal *signal);
+void context_process_process_info_request_signal(Context *ctx, struct BuiltInAtomRequestSignal *signal, bool process_table_locked);
 
 /**
  * @brief Process a trap answer signal.
@@ -385,6 +413,23 @@ 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 a link exit signal.
+ *
+ * @param ctx the context being executed
+ * @param signal the signal with the exit info tuple
+ * @return true if the process is trapping exit and info tuple was enqueued as a message;
+ */
+bool context_process_link_exit_signal(Context *ctx, struct TermSignal *signal);
+
+/**
+ * @brief Process a monitor down signal.
+ *
+ * @param ctx the context being executed
+ * @param signal the signal with the down info tuple
+ */
+void context_process_monitor_down_signal(Context *ctx, struct TermSignal *signal);
+
 /**
  * @brief Get process information.
  *
@@ -421,30 +466,53 @@ struct Monitor *monitor_new(term monitor_pid, uint64_t ref_ticks, bool is_monito
  *
  * @param resource resource object
  * @param ref_ticks reference associated with the monitor
- * @param process_id process being monitored
  * @return the allocated resource monitor or NULL if allocation failed
  */
 struct Monitor *monitor_resource_monitor_new(void *resource, uint64_t ref_ticks);
 
 /**
  * @brief Half-unlink process to another process
- * @details Called within the process only. For the other end of the
- * link, an UnlinkSignal is sent that calls this function.
+ * @details If process is found, an unlink id is generated and the link is
+ * deactivated.
  *
  * @param ctx the context being executed
- * @param monitor_pid process to unlink from
- * @return 0 on success
+ * @param link_pid process to unlink from
+ * @param unlink_id on output, unlink id to send to the target process
+ * @return true if process was found
  */
-void context_unlink(Context *ctx, term monitor_pid);
+bool context_set_unlink_id(Context *ctx, term link_pid, uint64_t *unlink_id);
 
 /**
- * @brief Destroy a monitor on a process.
- * @details Called within the process only. This function is called from
- * DemonitorSignal.
+ * @brief Half-unlink process to another process
+ * @details Called within the process only when an UnlinkID signal is received.
+ * If link is found, remove it and sends an UnlinkIDAck signal to the linked
+ * process.
+ *
+ * @param ctx the context being executed
+ * @param link_pid process to unlink from
+ * @param unlink_id unlink id from the signal
+ * @param process_table_locked whether process table is already locked
+ */
+void context_ack_unlink(Context *ctx, term link_pid, uint64_t unlink_id, bool process_table_locked);
+
+/**
+ * @brief Half-unlink process to another process
+ * @details Called within the process only when an UnlinkIDAck signal is received.
+ * If link is found and matches, remove it.
  *
  * @param ctx the context being executed
+ * @param link_pid process to unlink from
+ * @param unlink_id unlink id from the signal
+ */
+void context_unlink_ack(Context *ctx, term link_pid, uint64_t unlink_id);
+
+/**
+ * @brief Destroy a monitor on a process (monitoring, monitored or resource)
+ * @details Called within the process only. This function is called from
+ * DemonitorSignal as well as demonitor nif on monitoring process.
+ *
+ * @param ctx the context being executed (monitoring or monitored)
  * @param ref_ticks reference of the monitor to remove
- * @return 0 on success
  */
 void context_demonitor(Context *ctx, uint64_t ref_ticks);
 
@@ -467,9 +535,12 @@ term context_get_monitor_pid(Context *ctx, uint64_t ref_ticks, bool *is_monitori
  * only exist once.
  *
  * @param ctx the context being executed
- * @param new_monitor monitor object to add
+ * @param new_monitor monitor object to add (ownership belongs to context
+ * afterwards)
+ * @return true if the monitor was added, false if it already existed and
+ * new_monitor waw freed.
  */
-void context_add_monitor(Context *ctx, struct Monitor *new_monitor);
+bool context_add_monitor(Context *ctx, struct Monitor *new_monitor);
 
 #ifdef __cplusplus
 }

src/libAtomVM/context.h

@@ -698,27 +698,3 @@ Module *globalcontext_get_module_by_index(GlobalContext *global, int index)
     SMP_RWLOCK_UNLOCK(global->modules_lock);
     return result;
 }
-
-bool globalcontext_demonitor(GlobalContext *global, uint64_t ref_ticks)
-{
-    struct ListHead *pitem;
-
-    struct ListHead *processes_table_list = synclist_wrlock(&global->processes_table);
-    LIST_FOR_EACH (pitem, processes_table_list) {
-        Context *p = GET_LIST_ENTRY(pitem, Context, processes_table_head);
-
-        struct ListHead *item;
-        LIST_FOR_EACH (item, &p->monitors_head) {
-            struct Monitor *monitor = GET_LIST_ENTRY(item, struct Monitor, monitor_list_head);
-            if (monitor->ref_ticks == ref_ticks) {
-                list_remove(&monitor->monitor_list_head);
-                free(monitor);
-                synclist_unlock(&global->processes_table);
-                return true;
-            }
-        }
-    }
-
-    synclist_unlock(&global->processes_table);
-    return false;
-}

src/libAtomVM/globalcontext.c

@@ -491,17 +491,6 @@ Module *globalcontext_get_module(GlobalContext *global, AtomString module_name_a
  */
 Module *globalcontext_load_module_from_avm(GlobalContext *global, const char *module_name);
 
-/**
- * @brief remove a monitor
- *
- * @details iterate on the list of all processes and then on each monitor
- * to find a given monitor, and remove it
- * @param global the global context
- * @param ref_ticks the reference to the monitor
- * @return true if the monitor was found
- */
-bool globalcontext_demonitor(GlobalContext *global, uint64_t ref_ticks);
-
 #ifndef __cplusplus
 static inline uint64_t globalcontext_get_ref_ticks(GlobalContext *global)
 {

src/libAtomVM/globalcontext.h

@@ -96,7 +96,9 @@ void mailbox_message_dispose(MailboxMessage *m, Heap *heap)
             break;
         }
         case KillSignal:
-        case TrapAnswerSignal: {
+        case TrapAnswerSignal:
+        case LinkExitSignal:
+        case MonitorDownSignal: {
             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);
@@ -109,12 +111,17 @@ void mailbox_message_dispose(MailboxMessage *m, Heap *heap)
             free(request_signal);
             break;
         }
-        case TrapExceptionSignal:
-        case UnlinkSignal: {
+        case TrapExceptionSignal: {
             struct ImmediateSignal *immediate_signal = CONTAINER_OF(m, struct ImmediateSignal, base);
             free(immediate_signal);
             break;
         }
+        case UnlinkIDSignal:
+        case UnlinkIDAckSignal: {
+            struct ImmediateRefSignal *immediate_ref_signal = CONTAINER_OF(m, struct ImmediateRefSignal, base);
+            free(immediate_ref_signal);
+            break;
+        }
         case FlushMonitorSignal:
         case FlushInfoMonitorSignal:
         case DemonitorSignal: {
@@ -291,6 +298,20 @@ void mailbox_send_ref_signal(Context *c, enum MessageType type, uint64_t ref_tic
     mailbox_post_message(c, &ref_signal->base);
 }
 
+void mailbox_send_immediate_ref_signal(Context *c, enum MessageType type, term immediate, uint64_t ref_ticks)
+{
+    struct ImmediateRefSignal *immediate_ref_signal = malloc(sizeof(struct ImmediateRefSignal));
+    if (IS_NULL_PTR(immediate_ref_signal)) {
+        fprintf(stderr, "Failed to allocate memory: %s:%i.\n", __FILE__, __LINE__);
+        return;
+    }
+    immediate_ref_signal->base.type = type;
+    immediate_ref_signal->immediate = immediate;
+    immediate_ref_signal->ref_ticks = ref_ticks;
+
+    mailbox_post_message(c, &immediate_ref_signal->base);
+}
+
 void mailbox_send_monitor_signal(Context *c, enum MessageType type, struct Monitor *monitor)
 {
     struct MonitorPointerSignal *monitor_signal = malloc(sizeof(struct MonitorPointerSignal));

src/libAtomVM/mailbox.c

@@ -82,8 +82,11 @@ enum MessageType
     FlushMonitorSignal,
     FlushInfoMonitorSignal,
     MonitorSignal,
-    UnlinkSignal,
+    UnlinkIDSignal,
+    UnlinkIDAckSignal,
+    LinkExitSignal,
     DemonitorSignal,
+    MonitorDownSignal,
 };
 
 struct MailboxMessage
@@ -137,6 +140,14 @@ struct RefSignal
     uint64_t ref_ticks;
 };
 
+struct ImmediateRefSignal
+{
+    MailboxMessage base;
+
+    term immediate;
+    uint64_t ref_ticks;
+};
+
 struct MonitorPointerSignal
 {
     MailboxMessage base;
@@ -240,6 +251,16 @@ void mailbox_send_built_in_atom_request_signal(
  */
 void mailbox_send_ref_signal(Context *c, enum MessageType type, uint64_t ref_ticks);
 
+/**
+ * @brief Sends an immediate and ref signal to a certain mailbox.
+ *
+ * @param c the process context.
+ * @param type the type of the signal
+ * @param immediate the immediate term to send (atom or pid)
+ * @param ref_ticks the ref
+ */
+void mailbox_send_immediate_ref_signal(Context *c, enum MessageType type, term immediate, uint64_t ref_ticks);
+
 /**
  * @brief Sends a ref immediate signal to a certain mailbox.
  *

src/libAtomVM/mailbox.h

@@ -3879,19 +3879,22 @@ static term nif_erlang_link(Context *ctx, int argc, term argv[])
         RAISE_ERROR(OUT_OF_MEMORY_ATOM);
     }
 
-    term callee_pid = term_from_local_process_id(ctx->process_id);
-    struct Monitor *other_link = monitor_link_new(callee_pid);
+    term self_pid = term_from_local_process_id(ctx->process_id);
+    struct Monitor *other_link = monitor_link_new(self_pid);
     if (IS_NULL_PTR(other_link)) {
         globalcontext_get_process_unlock(ctx->global, target);
-        free(self_link);
         RAISE_ERROR(OUT_OF_MEMORY_ATOM);
     }
 
+    if (UNLIKELY(!context_add_monitor(ctx, self_link))) {
+        globalcontext_get_process_unlock(ctx->global, target);
+        free(other_link);
+        return TRUE_ATOM;
+    }
+
     mailbox_send_monitor_signal(target, MonitorSignal, other_link);
     globalcontext_get_process_unlock(ctx->global, target);
 
-    context_add_monitor(ctx, self_link);
-
     return TRUE_ATOM;
 }
 
@@ -3909,9 +3912,14 @@ static term nif_erlang_unlink(Context *ctx, int argc, term argv[])
         return TRUE_ATOM;
     }
 
-    context_unlink(ctx, term_from_local_process_id(target->process_id));
-    term callee_pid = term_from_local_process_id(ctx->process_id);
-    mailbox_send_immediate_signal(target, UnlinkSignal, callee_pid);
+    uint64_t unlink_id;
+    if (UNLIKELY(!context_set_unlink_id(ctx, argv[0], &unlink_id))) {
+        globalcontext_get_process_unlock(ctx->global, target);
+        return TRUE_ATOM;
+    }
+
+    term self_pid = term_from_local_process_id(ctx->process_id);
+    mailbox_send_immediate_ref_signal(target, UnlinkIDSignal, self_pid, unlink_id);
     globalcontext_get_process_unlock(ctx->global, target);
 
     return TRUE_ATOM;