Merge pull request #1635 from pguyot/w16/fix-unlink-protocol

Implement link protocol with unlink id and acknowledgement

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

@@ -66,6 +66,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

@@ -646,27 +646,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

@@ -463,17 +463,6 @@ Module *globalcontext_get_module_by_index(GlobalContext *global, int index);
  */
 Module *globalcontext_get_module(GlobalContext *global, AtomString module_name_atom);
 
-/**
- * @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

@@ -3656,19 +3656,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;
 }
 
@@ -3686,9 +3689,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;