Factorize EventListeners

Both generic_unix and esp32 platforms featured event listeners.
Factorize code adding a new header that is included by both, listeners.h.
stm32 platform currently does not implement any listener.

Also fix a potential concurrency bug in esp32 drivers that removed listeners
without locking the list.

Signed-off-by: Paul Guyot <[email protected]>

Author: pguyot

src/libAtomVM/CMakeLists.txt

@@ -39,6 +39,7 @@ set(HEADER_FILES
     interop.h
     list.h
     linkedlist.h
+    listeners.h
     mailbox.h
     memory.h
     module.h

src/libAtomVM/globalcontext.c

@@ -75,6 +75,7 @@ GlobalContext *globalcontext_new()
     synclist_init(&glb->refc_binaries);
     synclist_init(&glb->processes_table);
     synclist_init(&glb->registered_processes);
+    synclist_init(&glb->listeners);
 
     glb->last_process_id = 0;
 
@@ -175,8 +176,13 @@ COLD_FUNC void globalcontext_destroy(GlobalContext *glb)
         refc_binary_destroy(refc, glb);
     }
     synclist_destroy(&glb->refc_binaries);
-
     synclist_destroy(&glb->avmpack_data);
+    struct ListHead *listeners = synclist_nolock(&glb->listeners);
+    MUTABLE_LIST_FOR_EACH (item, tmp, listeners) {
+        sys_listener_destroy(item);
+    }
+    synclist_destroy(&glb->listeners);
+
     free(glb);
 }
 

src/libAtomVM/globalcontext.h

@@ -75,6 +75,7 @@ struct GlobalContext
     struct SyncList refc_binaries;
     struct SyncList processes_table;
     struct SyncList registered_processes;
+    struct SyncList listeners;
 
     int32_t last_process_id;
 

src/libAtomVM/listeners.h

@@ -0,0 +1,151 @@
+/*
+ * This file is part of AtomVM.
+ *
+ * Copyright 2023 Paul Guyot <[email protected]>
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ * SPDX-License-Identifier: Apache-2.0 OR LGPL-2.1-or-later
+ */
+
+/**
+ * @file listeners.h
+ * @brief Common code for port listeners.
+ *
+ * @details This header defines convenient common functions to implement
+ * listeners, and should be included in platform's `sys.c`.
+ *
+ * Before including this file, define listener_event_t which represent a
+ * selectable event, as well as EventListener, which should have a list head
+ * member called `listeners_list_head` and a handler member called `handler`.
+ *
+ * On a platform using select(3) with file descriptors, this typically is done
+ * by creating a `platform_sys.h` header with:
+ * ```
+ * #include "sys.h"
+ *
+ * typedef int listener_event_t;
+ *
+ * struct EventListener
+ * {
+ *    struct ListHead listeners_list_head;
+ *    event_handler_t handler;
+ *    listener_event_t fd;
+ * };
+ * ```
+ *
+ * and by including `platform_sys.h` header in `sys.c` before `listeners.h`.
+ */
+
+#include <stdbool.h>
+
+/**
+ * @brief Add an event listener to the set of polled events.
+ *
+ * @details This function must be implemented and will typically access the
+ * platform data from `glb` and add the event to the set. It is called by
+ * `process_listener_handler` when a handler returns a new listener. It can be
+ * called by `sys_register_listener`. It may just set a dirty flag.
+ *
+ * @param listener the listener to add to polling set
+ * @param glb the global context
+ */
+static void event_listener_add_to_polling_set(struct EventListener *listener, GlobalContext *glb);
+
+/**
+ * @brief Remove an event from the set of polled events.
+ *
+ * @details This function must be implemented and will typically access the
+ * platform data from `glb` and remove the event to the set. It is called by
+ * `process_listener_handler` when a handler returns NULL or a new listener. It
+ * can be called by `sys_unregister_listener`. It may just set a dirty flag.
+ *
+ * Compared to `event_listener_add_to_polling_set`, the event listener may no
+ * longer exist if it was freed by the handler.
+ *
+ * @param event the listener event to remove from polling set
+ * @param glb the global context
+ */
+static void listener_event_remove_from_polling_set(listener_event_t event, GlobalContext *glb);
+
+/**
+ * @brief Determiner if an event is a listener's event.
+ *
+ * @param listener the listener to test
+ * @param event the event to test
+ * @return true if event is the listener's event
+ */
+static bool event_listener_is_event(EventListener *listener, listener_event_t event);
+
+/**
+ * @brief Process listener handlers, optionally in advancing order, especially useful with poll(2) which returns fd in the provided order.
+ *
+ * @param glb the global context
+ * @param current_event the selected event
+ * @param listeners the list of listeners (locked for writing)
+ * @param item_ptr the current cursor or NULL to search in items
+ * @param previous_ptr the previous cursor (ignored and can be NULL if item_ptr is NULL).
+ * @return true if the current_event was found
+ */
+static inline bool process_listener_handler(GlobalContext *glb, listener_event_t current_event, struct ListHead *listeners, struct ListHead **item_ptr, struct ListHead **previous_ptr)
+{
+    bool result = false;
+    struct ListHead *item;
+    struct ListHead *previous;
+    if (item_ptr) {
+        item = *item_ptr;
+        previous = *previous_ptr;
+    } else {
+        item = listeners->next;
+        previous = listeners;
+    }
+
+    while (item != listeners) {
+        struct ListHead *next = item->next;
+        EventListener *listener = GET_LIST_ENTRY(item, EventListener, listeners_list_head);
+        if (event_listener_is_event(listener, current_event)) {
+            EventListener *new_listener = listener->handler(glb, listener);
+            if (new_listener == NULL) {
+                listener_event_remove_from_polling_set(current_event, glb);
+                previous->next = next;
+                next->prev = previous;
+                item = next;
+            } else if (new_listener != listener) {
+                listener_event_remove_from_polling_set(current_event, glb);
+                event_listener_add_to_polling_set(new_listener, glb);
+                // Replace listener with new_listener in the list
+                // listener was freed by handler.
+                previous->next = &new_listener->listeners_list_head;
+                next->prev = &new_listener->listeners_list_head;
+                new_listener->listeners_list_head.prev = previous;
+                new_listener->listeners_list_head.next = next;
+                item = &new_listener->listeners_list_head;
+            }
+            result = true;
+            break;
+        }
+        previous = item;
+        item = next;
+    }
+    if (item_ptr) {
+        *previous_ptr = previous;
+        *item_ptr = item;
+    }
+    return result;
+}
+
+void sys_listener_destroy(struct ListHead *item)
+{
+    EventListener *listener = GET_LIST_ENTRY(item, EventListener, listeners_list_head);
+    free(listener);
+}

src/libAtomVM/sys.h

@@ -22,7 +22,9 @@
  * @file sys.h
  * @brief Platform specific functions.
  *
- * @details This header defines platform dependent functions, that mostly deals with events.
+ * @details This header defines platform dependent functions, that mostly deals
+ * with events. Some functions can be implemented by using functions defined
+ * and implemented in `listeners.h` header.
  */
 
 #ifndef _SYS_H_
@@ -46,18 +48,97 @@ enum
 };
 
 /**
- * @brief Poll events (from drivers), with a timeout (in ms), or until
- * `sys_signal` is called.
+ * @brief Event listener
+ *
+ * An event listener structure should be defined by the platform. Event
+ * listeners belong to the `GlobalContext.listeners` synchronized list.
+ */
+typedef struct EventListener EventListener;
+
+/**
+ * @brief Event handlers (for ports)
+ *
+ * @details The event handler is called from the scheduler thread but outside
+ * any process. It can send messages to processes using `globalcontext_send_message`
+ * function.
+ *
+ * Result of this callback alters the list of handlers which is locked for
+ * writing when it is called. It can:
+ * - return `listener`, in which case the list is not modified
+ * - return `NULL`, in which case the entry is removed. The callback is
+ * responsible for freeing the listener.
+ * - return another listener, in which case the current listener is replaced
+ * by the other listener. The callback is responsible for freeing the previous
+ * listener if it is no longer needed.
+ *
+ * Appending a listener is also possible by altering the list head.
+ *
+ * This callback is defined for platforms using `listeners.h` header and can be
+ * ignored by others.
+ *
+ * @param glb global context
+ * @param listener the current listener
+ * @return NULL if the current listener should be removed, listener if it
+ * should be kept or another listener if it should be replaced.
+ */
+typedef EventListener *(*event_handler_t)(GlobalContext *glb, EventListener *listener);
+
+/**
+ * @brief Poll events (from drivers and select events), with a timeout (in ms),
+ * or until `sys_signal` is called.
  *
  * @details Depending on platforms, check all open file descriptors/queues and
  * call drivers that should send messages to contexts (which will unblock them).
  * With SMP builds, this function can be called from any scheduler.
  *
+ * If selectable events are supported on the platform, this function should also:
+ * - call `select_event_destroy` on select events that have close set to 1
+ * - include the set of ErlNifEvent that are marked for read or write in the
+ * select set, and if they are selected, call `select_event_notify` to send
+ * the notification.
+ *
+ * `select_event_count_and_destroy_closed` defined in resources.h can be used
+ * to process closed select events.
+ *
  * @param glb the global context.
  * @param timeout_ms the number of ms to wait, `SYS_POLL_EVENTS_WAIT_FOREVER` to wait forever.
  */
 void sys_poll_events(GlobalContext *glb, int timeout_ms);
 
+/**
+ * @brief Register a listener.
+ *
+ * @details This function is called by ports to register a listener which is a
+ * native alternative to select events. The actual definition of listeners
+ * is platform dependent.
+ *
+ * @param global the global context.
+ * @param listener the listener to register
+ */
+void sys_register_listener(GlobalContext *global, EventListener *listener);
+
+/**
+ * @brief Unregister a listener.
+ *
+ * @details This function is called by ports to unregister a listener which is
+ * a native alternative to select events. The actual definition of listeners
+ * is platform dependent.
+ *
+ * @param global the global context.
+ * @param listener the listener to unregister.
+ */
+void sys_unregister_listener(GlobalContext *global, EventListener *listener);
+
+/**
+ * @brief Free a listener
+ *
+ * @details This function is called when the global context is destroyed on
+ * every remaining listener. An implementation is available in `listeners.h`.
+ *
+ * @param item list item
+ */
+void sys_listener_destroy(struct ListHead *item);
+
 #ifndef AVM_NO_SMP
 
 /**

src/platforms/esp32/components/avm_builtins/gpio_driver.c

@@ -343,7 +343,6 @@ static bool gpiodriver_is_gpio_attached(struct GPIOData *gpio_data, int gpio_num
 static term gpiodriver_set_int(Context *ctx, int32_t target_pid, term cmd)
 {
     GlobalContext *glb = ctx->global;
-    struct ESP32PlatformData *platform = glb->platform_data;
 
     struct GPIOData *gpio_data = ctx->platform_data;
 
@@ -399,7 +398,7 @@ static term gpiodriver_set_int(Context *ctx, int32_t target_pid, term cmd)
     list_append(&gpio_data->gpio_listeners, &data->gpio_listener_list_head);
     data->gpio = gpio_num;
     data->target_local_pid = target_pid;
-    synclist_append(&platform->listeners, &data->listener.listeners_list_head);
+    sys_register_listener(glb, &data->listener);
     data->listener.sender = data;
     data->listener.handler = gpio_interrupt_callback;
 
@@ -426,7 +425,7 @@ static term gpiodriver_remove_int(Context *ctx, term cmd)
         struct GPIOListenerData *gpio_listener = GET_LIST_ENTRY(item, struct GPIOListenerData, gpio_listener_list_head);
         if (gpio_listener->gpio == gpio_num) {
             list_remove(&gpio_listener->gpio_listener_list_head);
-            list_remove(&gpio_listener->listener.listeners_list_head);
+            sys_unregister_listener(ctx->global, &gpio_listener->listener);
             free(gpio_listener);
 
             gpio_set_intr_type(gpio_num, GPIO_INTR_DISABLE);

src/platforms/esp32/components/avm_builtins/socket_driver.c

@@ -285,7 +285,7 @@ void socket_driver_init(GlobalContext *glb)
     struct ESP32PlatformData *platform = glb->platform_data;
     socket_listener->sender = netconn_events;
     socket_listener->handler = socket_events_handler;
-    synclist_append(&platform->listeners, &socket_listener->listeners_list_head);
+    sys_register_listener(glb, socket_listener);
 
     synclist_init(&platform->sockets);
     list_init(&platform->ready_connections);

src/platforms/esp32/components/avm_builtins/uart_driver.c

@@ -267,15 +267,13 @@ Context *uart_driver_create_port(GlobalContext *global, term opts)
 
     uart_set_pin(uart_num, tx_pin, rx_pin, rts_pin, cts_pin);
 
-    struct ESP32PlatformData *platform = global->platform_data;
-
     struct UARTData *uart_data = malloc(sizeof(struct UARTData));
     if (IS_NULL_PTR(uart_data)) {
         fprintf(stderr, "Failed to allocate memory: %s:%i.\n", __FILE__, __LINE__);
         AVM_ABORT();
     }
     uart_data->listener.handler = uart_interrupt_callback;
-    synclist_append(&platform->listeners, &uart_data->listener.listeners_list_head);
+    sys_register_listener(global, &uart_data->listener);
     uart_data->reader_process_pid = term_invalid_term();
     uart_data->reader_ref_ticks = 0;
     uart_data->uart_num = uart_num;
@@ -423,7 +421,7 @@ static void uart_driver_do_close(Context *ctx, term msg)
 
     int local_pid = term_to_local_process_id(pid);
 
-    list_remove(&uart_data->listener.listeners_list_head);
+    sys_unregister_listener(glb, &uart_data->listener);
 
     if (UNLIKELY(memory_ensure_free(ctx, TUPLE_SIZE(2) + REF_SIZE) != MEMORY_GC_OK)) {
         ESP_LOGE(TAG, "[uart_driver_do_close] Failed to allocate space for return value");

src/platforms/esp32/components/avm_sys/include/esp32_sys.h

@@ -26,6 +26,8 @@
 
 #include <time.h>
 
+#include "sys.h"
+
 #define REGISTER_PORT_DRIVER(NAME, INIT_CB, CREATE_CB)                \
     struct PortDriverDef NAME##_port_driver_def = {                   \
         .port_driver_name = #NAME,                                    \
@@ -61,21 +63,17 @@
 
 #define EVENT_DESCRIPTORS_COUNT 16
 
-typedef struct EventListener EventListener;
-
-typedef EventListener *(*event_handler_t)(GlobalContext *glb, EventListener *listener);
+typedef void *listener_event_t;
 
 struct EventListener
 {
     struct ListHead listeners_list_head;
-
     event_handler_t handler;
-    void *sender;
+    listener_event_t sender;
 };
 
 struct ESP32PlatformData
 {
-    struct SyncList listeners;
     struct SyncList sockets;
     struct ListHead ready_connections;
 };