Merge pull request atomvm#606 from pguyot/w22/factorize-event-listeners

Factorize event listeners

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.

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

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;
 };