Allow sending messages directly from driver (FreeRTOS) tasks

Introduce and document new functions to send messages from a driver task, typically a callback on a platform with FreeRTOS. While sending messages from such a task may work with SMP builds, it was not necessarily safe and it didn't work with non-SMP builds. Introduce the required synchronization primitives that try and acquire locks. Also fix ESP32 network driver to use these new functions as it was broken with non-SMP builds (e.g. ESP32C6). Signed-off-by: Paul Guyot <[email protected]>
pguyot · Mar 2, 2024 · 3764769 · 3764769
1 parent 1f78a79
commit 3764769
Show file tree

Hide file tree

Showing 35 changed files with 757 additions and 255 deletions.
diff --git a/CMakeLists.txt b/CMakeLists.txt
@@ -28,6 +28,7 @@ find_package(Elixir)
 
 option(AVM_DISABLE_FP "Disable floating point support." OFF)
 option(AVM_DISABLE_SMP "Disable SMP." OFF)
+option(AVM_DISABLE_TASK_DRIVER "Disable task driver support." OFF)
 option(AVM_USE_32BIT_FLOAT "Use 32 bit floats." OFF)
 option(AVM_VERBOSE_ABORT "Print module and line number on VM abort" OFF)
 option(AVM_RELEASE "Build an AtomVM release" OFF)

diff --git a/doc/src/atomvm-internals.md b/doc/src/atomvm-internals.md
@@ -144,6 +144,27 @@ If there already is one thread in `sys_poll_events`, other scheduler threads pic
 >:::{doxygenfunction} sys_signal
 ````
 
+## Tasks and synchronization mechanisms
+
+AtomVM SMP builds run on operating or runtime systems implementing tasks (FreeRTOS SMP on ESP32, Unix and WebAssembly) as well as on systems with no task implementation (Raspberry Pi Pico).
+
+On runtime systems with tasks, each scheduler thread is implemented as a task. On Pico, a scheduler thread runs on Core 0 and another one runs on Core 1, and they are effectively pinned to each core.
+
+For synchronization purposes, AtomVM uses mutexes, condition variables, RW locks, spinlocks and Atomics.
+
+Availability of RW Locks and atomics are verified at compile time using detection of symbols for RW Locks and `ATOMIC_*_LOCK_FREE` C11 macros for atomics.
+
+Mutexes and condition variables are provided by the SDK or the runtime system. If RW Locks are not available, AtomVM uses mutexes. Atomics are not available on Pico and are replaced by critical sections. Spinlocks are implemented by AtomVM on top of Atomics, or using mutexes on Pico.
+
+Importantly, locking synchronization mechanisms (mutexes, RW locks, spinlocks) are not interrupt-safe. Interrupt service routines must not try to lock as they could fail forever if interrupted code owns the lock. Atomics, including emulation on Pico, are interrupt-safe.
+
+Drivers can send messages from event callbacks typically called from FreeRTOS tasks using `globalcontext_send_message_from_task` function instead of `globalcontext_send_message`. This function tries to acquire required locks and if it fails, enqueues sent message in a queue, so it is later processed when the scheduler performs context switching. The function is undefined if option `AVM_DISABLE_TASK_DRIVER` is passed.
+
+````{only} html
+>:::{doxygenfunction} globalcontext_send_message
+>:::{doxygenfunction} globalcontext_send_message_from_task
+````
+
 ## Mailboxes and signals
 
 Erlang processes receive messages in a mailbox.  The mailbox is the interface with other processes.

diff --git a/src/libAtomVM/CMakeLists.txt b/src/libAtomVM/CMakeLists.txt
@@ -144,25 +144,35 @@ endif()
 
 if (AVM_DISABLE_SMP)
     target_compile_definitions(libAtomVM PUBLIC AVM_NO_SMP)
-else()
-    include(CheckIncludeFile)
-    CHECK_INCLUDE_FILE(stdatomic.h STDATOMIC_INCLUDE)
-    if(HAVE_PLATFORM_SMP_H)
-        target_compile_definitions(libAtomVM PUBLIC HAVE_PLATFORM_SMP_H)
-    endif()
-    include(CheckCSourceCompiles)
-    check_c_source_compiles("
-        #include <stdatomic.h>
-        int main() {
-            _Static_assert(ATOMIC_POINTER_LOCK_FREE == 2, \"Expected ATOMIC_POINTER_LOCK_FREE to be equal to 2\");
-        }
-    " ATOMIC_POINTER_LOCK_FREE_IS_TWO)
-    if (NOT ATOMIC_POINTER_LOCK_FREE_IS_TWO AND NOT HAVE_PLATFORM_SMP_H)
-        if (NOT STDATOMIC_INCLUDE)
-            message(FATAL_ERROR "stdatomic.h cannot be found, you need to disable SMP on this platform or provide platform_smp.h and define HAVE_PLATFORM_SMP_H")
-        else()
-            message(FATAL_ERROR "Platform doesn't support atomic pointers, you need to disable SMP or provide platform_smp.h and define HAVE_PLATFORM_SMP_H")
-        endif()
+endif()
+if (AVM_DISABLE_TASK_DRIVER)
+    target_compile_definitions(libAtomVM PUBLIC AVM_NO_TASK_DRIVER)
+endif()
+
+if(HAVE_PLATFORM_SMP_H)
+    target_compile_definitions(libAtomVM PUBLIC HAVE_PLATFORM_SMP_H)
+endif()
+if(HAVE_PLATFORM_ATOMIC_H)
+    target_compile_definitions(libAtomVM PUBLIC HAVE_PLATFORM_ATOMIC_H)
+endif()
+
+include(CheckIncludeFile)
+CHECK_INCLUDE_FILE(stdatomic.h STDATOMIC_INCLUDE)
+include(CheckCSourceCompiles)
+check_c_source_compiles("
+    #include <stdatomic.h>
+    int main() {
+        _Static_assert(ATOMIC_POINTER_LOCK_FREE == 2, \"Expected ATOMIC_POINTER_LOCK_FREE to be equal to 2\");
+    }
+" ATOMIC_POINTER_LOCK_FREE_IS_TWO)
+if (ATOMIC_POINTER_LOCK_FREE_IS_TWO)
+    target_compile_definitions(libAtomVM PUBLIC HAVE_ATOMIC)
+endif()
+if (NOT ATOMIC_POINTER_LOCK_FREE_IS_TWO AND NOT (HAVE_PLATFORM_ATOMIC_H OR (AVM_DISABLE_SMP AND AVM_DISABLE_ISR)))
+    if (NOT STDATOMIC_INCLUDE)
+        message(FATAL_ERROR "stdatomic.h cannot be found, you need to provide platform_atomic.h and define HAVE_PLATFORM_ATOMIC_H or alternatively pass AVM_DISABLE_SMP and AVM_DISABLE_TASK_DRIVER")
+    else()
+        message(FATAL_ERROR "Platform doesn't support atomic pointers, you need to provide platform_atomic.h and define HAVE_PLATFORM_ATOMIC_H or alternatively pass AVM_DISABLE_SMP and AVM_DISABLE_TASK_DRIVER")
     endif()
 endif()
 

diff --git a/src/libAtomVM/context.c b/src/libAtomVM/context.c
@@ -35,6 +35,15 @@
 #include "term.h"
 #include "utils.h"
 
+#ifdef HAVE_PLATFORM_ATOMIC_H
+#include "platform_atomic.h"
+#else
+#if defined(HAVE_ATOMIC)
+#include <stdatomic.h>
+#define ATOMIC_COMPARE_EXCHANGE_WEAK_INT atomic_compare_exchange_weak
+#endif
+#endif
+
 #define IMPL_EXECUTE_LOOP
 #include "opcodesswitch.h"
 #undef IMPL_EXECUTE_LOOP
@@ -235,7 +244,7 @@ void context_update_flags(Context *ctx, int mask, int value) CLANG_THREAD_SANITI
     enum ContextFlags desired;
     do {
         desired = (expected & mask) | value;
-    } while (!ATOMIC_COMPARE_EXCHANGE_WEAK(&ctx->flags, &expected, desired));
+    } while (!ATOMIC_COMPARE_EXCHANGE_WEAK_INT(&ctx->flags, &expected, desired));
 #else
     ctx->flags = (ctx->flags & mask) | value;
 #endif

diff --git a/src/libAtomVM/globalcontext.c b/src/libAtomVM/globalcontext.c
@@ -30,30 +30,24 @@
 #include "defaultatoms.h"
 #include "erl_nif_priv.h"
 #include "list.h"
+#include "mailbox.h"
 #include "posix_nifs.h"
 #include "refc_binary.h"
 #include "resources.h"
+#include "scheduler.h"
+#include "smp.h"
 #include "synclist.h"
 #include "sys.h"
 #include "utils.h"
 #include "valueshashtable.h"
 
-#ifndef AVM_NO_SMP
-#define SMP_SPINLOCK_LOCK(spinlock) smp_spinlock_lock(spinlock)
-#define SMP_SPINLOCK_UNLOCK(spinlock) smp_spinlock_unlock(spinlock)
-#define SMP_MUTEX_LOCK(mutex) smp_mutex_lock(mutex)
-#define SMP_MUTEX_UNLOCK(mutex) smp_mutex_unlock(mutex)
-#define SMP_RWLOCK_RDLOCK(lock) smp_rwlock_rdlock(lock)
-#define SMP_RWLOCK_WRLOCK(lock) smp_rwlock_wrlock(lock)
-#define SMP_RWLOCK_UNLOCK(lock) smp_rwlock_unlock(lock)
+#ifdef HAVE_PLATFORM_ATOMIC_H
+#include "platform_atomic.h"
 #else
-#define SMP_SPINLOCK_LOCK(spinlock)
-#define SMP_SPINLOCK_UNLOCK(spinlock)
-#define SMP_MUTEX_LOCK(mutex)
-#define SMP_MUTEX_UNLOCK(mutex)
-#define SMP_RWLOCK_RDLOCK(lock)
-#define SMP_RWLOCK_WRLOCK(lock)
-#define SMP_RWLOCK_UNLOCK(lock)
+#if defined(HAVE_ATOMIC)
+#include <stdatomic.h>
+#define ATOMIC_COMPARE_EXCHANGE_WEAK_PTR atomic_compare_exchange_weak
+#endif
 #endif
 
 struct RegisteredProcess
@@ -75,6 +69,9 @@ GlobalContext *globalcontext_new()
     list_init(&glb->waiting_processes);
 #ifndef AVM_NO_SMP
     smp_spinlock_init(&glb->processes_spinlock);
+#endif
+#ifndef AVM_NO_TASK_DRIVER
+    glb->message_queue = NULL;
 #endif
     synclist_init(&glb->avmpack_data);
     synclist_init(&glb->refc_binaries);
@@ -272,6 +269,29 @@ Context *globalcontext_get_process_lock(GlobalContext *glb, int32_t process_id)
     return NULL;
 }
 
+#if !defined(AVM_NO_SMP) && !defined(AVM_NO_TASK_DRIVER)
+static bool globalcontext_get_process_trylock(GlobalContext *glb, int32_t process_id, Context **output)
+{
+    struct ListHead *item;
+    Context *p = NULL;
+
+    struct ListHead *processes_table_list = synclist_tryrdlock(&glb->processes_table);
+    if (processes_table_list == NULL) {
+        return false;
+    }
+    LIST_FOR_EACH (item, processes_table_list) {
+        p = GET_LIST_ENTRY(item, Context, processes_table_head);
+
+        if (p->process_id == process_id) {
+            *output = p;
+        }
+    }
+    synclist_unlock(&glb->processes_table);
+
+    return true;
+}
+#endif
+
 void globalcontext_get_process_unlock(GlobalContext *glb, Context *c)
 {
     if (c) {
@@ -303,6 +323,78 @@ void globalcontext_send_message_nolock(GlobalContext *glb, int32_t process_id, t
     }
 }
 
+#ifndef AVM_NO_TASK_DRIVER
+void globalcontext_send_message_from_task(GlobalContext *glb, int32_t process_id, enum MessageType type, term t)
+{
+    MailboxMessage *message = NULL;
+    bool postponed = false;
+#ifndef AVM_NO_SMP
+    Context *p = NULL;
+    if (globalcontext_get_process_trylock(glb, process_id, &p)) {
+        if (p) {
+            message = mailbox_message_create_from_term(type, t);
+            // Ensure we can acquire the spinlock
+            if (smp_spinlock_trylock(&glb->processes_spinlock)) {
+                // We can send the message.
+                mailbox_enqueue_message(p, message);
+                scheduler_signal_message_from_task(p);
+                smp_spinlock_unlock(&glb->processes_spinlock);
+            } else {
+                postponed = true;
+            }
+            globalcontext_get_process_unlock(glb, p);
+        }
+    } else {
+        postponed = true;
+    }
+#else
+    // Without SMP, we have no lock, so we must always enqueue.
+    postponed = true;
+#endif
+    if (postponed) {
+        if (message == NULL) {
+            message = mailbox_message_create_from_term(type, t);
+        }
+        struct MessageQueueItem *queued_item = malloc(sizeof(struct MessageQueueItem));
+        if (IS_NULL_PTR(queued_item)) {
+            fprintf(stderr, "Failed to allocate memory: %s:%i.\n", __FILE__, __LINE__);
+            AVM_ABORT();
+        }
+        queued_item->message = message;
+        queued_item->process_id = process_id;
+
+        struct MessageQueueItem *current_first = NULL;
+        do {
+            queued_item->next = current_first;
+        } while (!ATOMIC_COMPARE_EXCHANGE_WEAK_PTR(&glb->message_queue, &current_first, queued_item));
+        // Make sure the scheduler is busy
+        sys_signal(glb);
+    }
+}
+
+void globalcontext_process_message_queue(GlobalContext *glb)
+{
+    struct MessageQueueItem *current = glb->message_queue;
+    // Empty outer list using CAS
+    if (current) {
+        while (!ATOMIC_COMPARE_EXCHANGE_WEAK_PTR(&glb->message_queue, &current, NULL)) {
+        };
+        (void) synclist_rdlock(&glb->processes_table);
+        while (current) {
+            Context *context = globalcontext_get_process_nolock(glb, current->process_id);
+            if (context) {
+                mailbox_enqueue_message(context, current->message);
+                scheduler_signal_message(context);
+            }
+            struct MessageQueueItem *old = current;
+            current = old->next;
+            free(old);
+        }
+        synclist_unlock(&glb->processes_table);
+    }
+}
+#endif
+
 void globalcontext_init_process(GlobalContext *glb, Context *ctx)
 {
     ctx->global = glb;

diff --git a/src/libAtomVM/globalcontext.h b/src/libAtomVM/globalcontext.h
@@ -38,6 +38,7 @@ extern "C" {
 #include "atom_table.h"
 #include "erl_nif.h"
 #include "list.h"
+#include "mailbox.h"
 #include "smp.h"
 #include "synclist.h"
 #include "term.h"
@@ -64,6 +65,18 @@ typedef struct Module Module;
 typedef struct GlobalContext GlobalContext;
 #endif
 
+#ifndef TYPEDEF_MAILBOXMESSAGE
+#define TYPEDEF_MAILBOXMESSAGE
+typedef struct MailboxMessage MailboxMessage;
+#endif
+
+struct MessageQueueItem
+{
+    struct MessageQueueItem *next;
+    MailboxMessage *message;
+    int32_t process_id;
+};
+
 struct GlobalContext
 {
     struct ListHead ready_processes;
@@ -73,6 +86,10 @@ struct GlobalContext
     // when running native handlers.
 #ifndef AVM_NO_SMP
     SpinLock processes_spinlock;
+#endif
+#ifndef AVM_NO_TASK_DRIVER
+    // Queue of messages that could not be directly sent from a driver task
+    struct MessageQueueItem *ATOMIC message_queue;
 #endif
     struct SyncList refc_binaries;
     struct SyncList processes_table;
@@ -215,6 +232,33 @@ void globalcontext_send_message(GlobalContext *glb, int32_t process_id, term t);
  */
 void globalcontext_send_message_nolock(GlobalContext *glb, int32_t process_id, term t);
 
+#ifndef AVM_NO_TASK_DRIVER
+/**
+ * @brief Send a message to a process identified by its id. This variant is to
+ * be used from task drivers. It tries to acquire the necessary locks and if it
+ * fails, it enqueues the message which will be delivered on the next scheduler
+ * context switch.
+ *
+ * @details Safely send a message to the process, doing nothing if the process
+ * cannot be found.
+ *
+ * @param glb the global context (that owns the process table).
+ * @param process_id the target process id.
+ * @param type the type of message to send, can be NormalMessage or a signal
+ * @param t the message to send.
+ */
+void globalcontext_send_message_from_task(GlobalContext *glb, int32_t process_id, enum MessageType type, term t);
+
+/**
+ * @brief Process queue of message enqueued from ISR.
+ *
+ * @details This function is called from the scheduler.
+ *
+ * @param glb the global context (that owns the process table).
+ */
+void globalcontext_process_message_queue(GlobalContext *glb);
+#endif
+
 /**
  * @brief Initialize a new process, providing it with a process id.
  *