Merge pull request #1555 from pguyot/w08/fix-link-leak

Use signals for link/unlink/monitor/demonitor

Fix concurrency issues and memory leaks related to a poor usage of list of
monitors that was accessed by other processes (fix #1384)
Fix semantics of monitor where self cannot monitor itself and a monitor cannot
be removed from another process.

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

@@ -56,6 +56,7 @@ and a race condition in otp_socket code
 - Fixed a bug where calling repeatedly `process_info` on a stopped process could cause an out of
 memory error
 - Fixed possible concurrency problems in ESP32 UART driver
+- Fixed concurrency and memory leak related to links and monitors
 
 ### Changed
 

src/libAtomVM/context.c

@@ -52,7 +52,7 @@
 #define DEFAULT_STACK_SIZE 8
 #define BYTES_PER_TERM (TERM_BITS / 8)
 
-static struct ResourceMonitor *context_monitors_handle_terminate(Context *ctx);
+static struct Monitor *context_monitors_handle_terminate(Context *ctx);
 static void destroy_extended_registers(Context *ctx, unsigned int live);
 
 Context *context_new(GlobalContext *glb)
@@ -133,27 +133,54 @@ void context_destroy(Context *ctx)
     // Ensure process is not registered
     globalcontext_maybe_unregister_process_id(ctx->global, ctx->process_id);
 
+    // Process any link/unlink/monitor/demonitor signal that arrived recently
+    // Also process ProcessInfoRequestSignal so caller isn't trapped waiting
+    MailboxMessage *signal_message = mailbox_process_outer_list(&ctx->mailbox);
+    while (signal_message) {
+        if (signal_message->type == ProcessInfoRequestSignal) {
+            struct BuiltInAtomRequestSignal *request_signal
+                = CONTAINER_OF(signal_message, struct BuiltInAtomRequestSignal, base);
+            context_process_process_info_request_signal(ctx, request_signal);
+        } else if (signal_message->type == MonitorSignal) {
+            struct MonitorPointerSignal *monitor_signal
+                = CONTAINER_OF(signal_message, struct MonitorPointerSignal, base);
+            context_add_monitor(ctx, monitor_signal->monitor);
+        } else if (signal_message->type == UnlinkSignal) {
+            struct ImmediateSignal *immediate_signal
+                = CONTAINER_OF(signal_message, struct ImmediateSignal, base);
+            context_unlink(ctx, immediate_signal->immediate);
+        } else if (signal_message->type == DemonitorSignal) {
+            struct RefSignal *ref_signal
+                = CONTAINER_OF(signal_message, struct RefSignal, base);
+            context_demonitor(ctx, ref_signal->ref_ticks);
+        }
+        MailboxMessage *next = signal_message->next;
+        mailbox_message_dispose(signal_message, &ctx->heap);
+        signal_message = next;
+    }
+
     // When monitor message is sent, process is no longer in the table
     // and is no longer registered either.
-    struct ResourceMonitor *resource_monitor = context_monitors_handle_terminate(ctx);
+    struct Monitor *resource_monitors = context_monitors_handle_terminate(ctx);
 
     synclist_unlock(&ctx->global->processes_table);
 
     // Eventually call resource monitors handlers after the processes table was unlocked
     // The monitors were removed from the list of monitors.
-    if (resource_monitor) {
+    if (resource_monitors) {
         ErlNifEnv env;
         erl_nif_env_partial_init_from_globalcontext(&env, ctx->global);
 
         struct ListHead monitors;
-        list_prepend(&resource_monitor->base.monitor_list_head, &monitors);
+        list_prepend(&resource_monitors->monitor_list_head, &monitors);
 
         struct ListHead *item;
         struct ListHead *tmp;
         MUTABLE_LIST_FOR_EACH (item, tmp, &monitors) {
             struct Monitor *monitor = GET_LIST_ENTRY(item, struct Monitor, monitor_list_head);
             void *resource = term_to_term_ptr(monitor->monitor_obj);
             struct RefcBinary *refc = refc_binary_from_data(resource);
+            resource_type_demonitor(refc->resource_type, monitor->ref_ticks);
             refc->resource_type->down(&env, resource, &ctx->process_id, &monitor->ref_ticks);
             free(monitor);
         }
@@ -199,18 +226,18 @@ void context_process_process_info_request_signal(Context *ctx, struct BuiltInAto
         if (context_get_process_info(ctx, NULL, &term_size, signal->atom, NULL)) {
             Heap heap;
             if (UNLIKELY(memory_init_heap(&heap, term_size) != MEMORY_GC_OK)) {
-                mailbox_send_built_in_atom_signal(target, TrapExceptionSignal, OUT_OF_MEMORY_ATOM);
+                mailbox_send_immediate_signal(target, TrapExceptionSignal, OUT_OF_MEMORY_ATOM);
             } else {
                 term ret;
                 if (context_get_process_info(ctx, &ret, NULL, signal->atom, &heap)) {
                     mailbox_send_term_signal(target, TrapAnswerSignal, ret);
                 } else {
-                    mailbox_send_built_in_atom_signal(target, TrapExceptionSignal, ret);
+                    mailbox_send_immediate_signal(target, TrapExceptionSignal, ret);
                 }
                 memory_destroy_heap(&heap, ctx->global);
             }
         } else {
-            mailbox_send_built_in_atom_signal(target, TrapExceptionSignal, BADARG_ATOM);
+            mailbox_send_immediate_signal(target, TrapExceptionSignal, BADARG_ATOM);
         }
         globalcontext_get_process_unlock(ctx->global, target);
     } // else: sender died
@@ -375,60 +402,53 @@ bool context_get_process_info(Context *ctx, term *out, size_t *term_size, term a
     return true;
 }
 
-static struct ResourceMonitor *context_monitors_handle_terminate(Context *ctx)
+static struct Monitor *context_monitors_handle_terminate(Context *ctx)
 {
     GlobalContext *glb = ctx->global;
     struct ListHead *item;
     struct ListHead *tmp;
-    struct ResourceMonitor *result = NULL;
+    struct Monitor *result = NULL;
     MUTABLE_LIST_FOR_EACH (item, tmp, &ctx->monitors_head) {
         struct Monitor *monitor = GET_LIST_ENTRY(item, struct Monitor, monitor_list_head);
-        if (monitor->ref_ticks && term_is_boxed(monitor->monitor_obj)) {
-            // Resource monitor
-            struct ResourceMonitor *resource_monitor = CONTAINER_OF(monitor, struct ResourceMonitor, base);
-            // remove the monitor from the list of the resource
-            list_remove(&resource_monitor->resource_list_head);
-            list_init(&resource_monitor->resource_list_head);
+        if (monitor->ref_ticks && ((monitor->monitor_obj & 0x3) == CONTEXT_MONITOR_RESOURCE_TAG)) {
             // remove it from the list we are iterating on
             if (result == NULL) {
-                list_init(&resource_monitor->base.monitor_list_head);
-                result = resource_monitor;
+                list_init(&monitor->monitor_list_head);
+                result = monitor;
             } else {
-                list_append(&result->base.monitor_list_head, &resource_monitor->base.monitor_list_head);
+                list_append(&result->monitor_list_head, &monitor->monitor_list_head);
             }
-        } else {
+        } else if (monitor->ref_ticks == 0 || ((monitor->monitor_obj & 0x3) == CONTEXT_MONITOR_MONITORED_PID_TAG)) {
+            // term_to_local_process_id and monitor->monitor_obj >> 4 are identical here.
             int local_process_id = term_to_local_process_id(monitor->monitor_obj);
             Context *target = globalcontext_get_process_nolock(glb, local_process_id);
             if (IS_NULL_PTR(target)) {
-                // TODO: we should scan for existing monitors when a context is destroyed
-                // otherwise memory might be wasted for long living processes
+                // TODO: assess whether this can happen
                 free(monitor);
                 continue;
             }
 
-            if (monitor->ref_ticks == 0 && (ctx->exit_reason != NORMAL_ATOM || target->trap_exit)) {
+            if (monitor->ref_ticks == 0) {
+                term exited_pid = term_from_local_process_id(ctx->process_id);
+                mailbox_send_immediate_signal(target, UnlinkSignal, exited_pid);
                 if (target->trap_exit) {
                     if (UNLIKELY(memory_ensure_free(ctx, TUPLE_SIZE(3)) != MEMORY_GC_OK)) {
                         // TODO: handle out of memory here
                         fprintf(stderr, "Cannot handle out of memory.\n");
                         globalcontext_get_process_unlock(glb, target);
                         AVM_ABORT();
                     }
-
-                    term exited_pid = term_from_local_process_id(ctx->process_id);
-                    // Process table should be locked before context_unlink is
-                    // called. This is done in calling function context_destroy.
-                    context_unlink(target, exited_pid);
                     // Prepare the message on ctx's heap which will be freed afterwards.
                     term info_tuple = term_alloc_tuple(3, &ctx->heap);
                     term_put_tuple_element(info_tuple, 0, EXIT_ATOM);
                     term_put_tuple_element(info_tuple, 1, exited_pid);
                     term_put_tuple_element(info_tuple, 2, ctx->exit_reason);
                     mailbox_send(target, info_tuple);
-                } else {
+                } else if (ctx->exit_reason != NORMAL_ATOM) {
                     mailbox_send_term_signal(target, KillSignal, ctx->exit_reason);
                 }
-            } else if (monitor->ref_ticks) {
+            } else {
+                mailbox_send_ref_signal(target, DemonitorSignal, monitor->ref_ticks);
                 int required_terms = REF_SIZE + TUPLE_SIZE(5);
                 if (UNLIKELY(memory_ensure_free(ctx, required_terms) != MEMORY_GC_OK)) {
                     // TODO: handle out of memory here
@@ -454,63 +474,71 @@ static struct ResourceMonitor *context_monitors_handle_terminate(Context *ctx)
                 mailbox_send(target, info_tuple);
             }
             free(monitor);
+        } else {
+            // We are the monitoring process.
+            int local_process_id = monitor->monitor_obj >> 4;
+            Context *target = globalcontext_get_process_nolock(glb, local_process_id);
+            mailbox_send_ref_signal(target, DemonitorSignal, monitor->ref_ticks);
+            free(monitor);
         }
     }
     return result;
 }
 
-int context_link(Context *ctx, term link_pid)
+struct Monitor *monitor_link_new(term link_pid)
 {
-    struct ListHead *item;
-    struct Monitor *monitor;
-    LIST_FOR_EACH (item, &ctx->monitors_head) {
-        monitor = GET_LIST_ENTRY(item, struct Monitor, monitor_list_head);
-        if ((monitor->monitor_obj == link_pid) && (monitor->ref_ticks == 0)) {
-            return 0;
-        }
-    }
-    monitor = malloc(sizeof(struct Monitor));
+    struct Monitor *monitor = malloc(sizeof(struct Monitor));
     if (IS_NULL_PTR(monitor)) {
-        return -1;
+        return NULL;
     }
     monitor->monitor_obj = link_pid;
     monitor->ref_ticks = 0;
-    list_append(&ctx->monitors_head, &monitor->monitor_list_head);
 
-    return 0;
+    return monitor;
 }
 
-uint64_t context_monitor(Context *ctx, term monitor_pid)
+struct Monitor *monitor_new(term monitor_pid, uint64_t ref_ticks, bool is_monitoring)
 {
-    uint64_t ref_ticks = globalcontext_get_ref_ticks(ctx->global);
-
     struct Monitor *monitor = malloc(sizeof(struct Monitor));
     if (IS_NULL_PTR(monitor)) {
-        return 0;
+        return NULL;
+    }
+    int32_t local_process_id = term_to_local_process_id(monitor_pid);
+    if (is_monitoring) {
+        monitor->monitor_obj = (local_process_id << 4) | CONTEXT_MONITOR_MONITORING_PID_TAG;
+    } else {
+        monitor->monitor_obj = (local_process_id << 4) | CONTEXT_MONITOR_MONITORED_PID_TAG;
     }
-    monitor->monitor_obj = monitor_pid;
     monitor->ref_ticks = ref_ticks;
-    list_append(&ctx->monitors_head, &monitor->monitor_list_head);
 
-    return ref_ticks;
+    return monitor;
 }
 
-struct ResourceMonitor *context_resource_monitor(Context *ctx, void *resource)
+struct Monitor *monitor_resource_monitor_new(void *resource, uint64_t ref_ticks)
 {
-    uint64_t ref_ticks = globalcontext_get_ref_ticks(ctx->global);
-
-    struct ResourceMonitor *monitor = malloc(sizeof(struct ResourceMonitor));
+    struct Monitor *monitor = malloc(sizeof(struct Monitor));
     if (IS_NULL_PTR(monitor)) {
         return NULL;
     }
-    // Not really boxed, but sufficient to distinguish from pids
-    monitor->base.monitor_obj = ((term) resource) | TERM_BOXED_VALUE_TAG;
-    monitor->base.ref_ticks = ref_ticks;
-    list_append(&ctx->monitors_head, &monitor->base.monitor_list_head);
+    monitor->monitor_obj = ((term) resource) | CONTEXT_MONITOR_RESOURCE_TAG;
+    monitor->ref_ticks = ref_ticks;
 
     return monitor;
 }
 
+void context_add_monitor(Context *ctx, struct Monitor *new_monitor)
+{
+    struct ListHead *item;
+    LIST_FOR_EACH (item, &ctx->monitors_head) {
+        struct Monitor *existing = GET_LIST_ENTRY(item, struct Monitor, monitor_list_head);
+        if ((existing->monitor_obj == new_monitor->monitor_obj) && (existing->ref_ticks == new_monitor->ref_ticks)) {
+            free(new_monitor);
+            return;
+        }
+    }
+    list_append(&ctx->monitors_head, &new_monitor->monitor_list_head);
+}
+
 void context_unlink(Context *ctx, term link_pid)
 {
     struct ListHead *item;
@@ -523,3 +551,36 @@ void context_unlink(Context *ctx, term link_pid)
         }
     }
 }
+
+void context_demonitor(Context *ctx, uint64_t ref_ticks)
+{
+    struct ListHead *item;
+    LIST_FOR_EACH (item, &ctx->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);
+            return;
+        }
+    }
+}
+
+term context_get_monitor_pid(Context *ctx, uint64_t ref_ticks, bool *is_monitoring)
+{
+    struct ListHead *item;
+    LIST_FOR_EACH (item, &ctx->monitors_head) {
+        struct Monitor *monitor = GET_LIST_ENTRY(item, struct Monitor, monitor_list_head);
+        if (monitor->ref_ticks == ref_ticks) {
+            if ((monitor->monitor_obj & 0x3) == CONTEXT_MONITOR_MONITORED_PID_TAG) {
+                *is_monitoring = false;
+                return term_from_local_process_id(monitor->monitor_obj >> 4);
+            } else if ((monitor->monitor_obj & 0x3) == CONTEXT_MONITOR_MONITORING_PID_TAG) {
+                *is_monitoring = true;
+                return term_from_local_process_id(monitor->monitor_obj >> 4);
+            } else {
+                return term_invalid_term();
+            }
+        }
+    }
+    return term_invalid_term();
+}