[FL-3833] Furi: event loop (#3675)

* Furi: epoll prototype

* Gui: simplify view_dispatcher custom event processing

* Furi: add missing critical sections to epoll

* Furi: add epoll unit tests, fully implement level processing for in and out events

* Furi: properly trigger epoll item event on adding mq, update tests.

* Unit tests: cleanup defines

* Furi: protect epoll from modification in callback

* Furi: rename epoll into event_loop, cleanup api naming

* Sync API Symbols

* Furi: add event loop contract and link api, port mq to new api, cleanup code

* Format Sources

* Furi: cleanup mq and event loop code

* Furi: remove unused staff from message queue

* ApiSymbols: remove event loop from public APIs.

* Fix furi unit tests

---------

Co-authored-by: Georgii Surkov <georgii.surkov@outlook.com>
Co-authored-by: Georgii Surkov <37121527+gsurkov@users.noreply.github.com>

furi/core/base.h

@@ -2,6 +2,7 @@
 
 #include <stdint.h>
 #include <stdbool.h>
+#include <stddef.h>
 #include <furi_config.h>
 
 #ifdef __cplusplus

furi/core/event_loop.c

@@ -0,0 +1,352 @@
+#include "event_loop_i.h"
+#include "message_queue_i.h"
+
+#include "check.h"
+#include "thread.h"
+
+#include <m-bptree.h>
+#include <m-i-list.h>
+
+#include <FreeRTOS.h>
+#include <task.h>
+
+struct FuriEventLoopItem {
+    // Source
+    FuriEventLoop* owner;
+
+    // Tracking item
+    const FuriEventLoopContract* contract;
+    void* object;
+    FuriEventLoopEvent event;
+
+    // Callback and context
+    FuriEventLoopMessageQueueCallback callback;
+    void* callback_context;
+
+    // Waiting list
+    ILIST_INTERFACE(WaitingList, struct FuriEventLoopItem);
+};
+
+ILIST_DEF(WaitingList, FuriEventLoopItem, M_POD_OPLIST)
+
+static FuriEventLoopItem* furi_event_loop_item_alloc(
+    FuriEventLoop* owner,
+    const FuriEventLoopContract* contract,
+    void* object,
+    FuriEventLoopEvent event);
+
+static void furi_event_loop_item_free(FuriEventLoopItem* instance);
+
+static void furi_event_loop_item_set_callback(
+    FuriEventLoopItem* instance,
+    FuriEventLoopMessageQueueCallback callback,
+    void* callback_context);
+
+static void furi_event_loop_item_notify(FuriEventLoopItem* instance);
+
+/* Event Loop RB tree */
+#define FURI_EVENT_LOOP_TREE_RANK (4)
+
+BPTREE_DEF2( // NOLINT
+    FuriEventLoopTree,
+    FURI_EVENT_LOOP_TREE_RANK,
+    void*, /* pointer to object we track */
+    M_PTR_OPLIST,
+    FuriEventLoopItem*, /* pointer to the FuriEventLoopItem */
+    M_PTR_OPLIST)
+
+#define M_OPL_FuriEventLoopTree_t() BPTREE_OPLIST(FuriEventLoopTree, M_POD_OPLIST)
+
+#define FURI_EVENT_LOOP_FLAG_NOTIFY_INDEX (2)
+
+typedef enum {
+    FuriEventLoopFlagEvent = (1 << 0),
+    FuriEventLoopFlagStop = (1 << 1),
+} FuriEventLoopFlag;
+
+#define FuriEventLoopFlagAll (FuriEventLoopFlagEvent | FuriEventLoopFlagStop)
+
+typedef enum {
+    FuriEventLoopProcessStatusComplete,
+    FuriEventLoopProcessStatusIncomplete,
+    FuriEventLoopProcessStatusAgain,
+} FuriEventLoopProcessStatus;
+
+typedef enum {
+    FuriEventLoopStateIdle,
+    FuriEventLoopStateProcessing,
+} FuriEventLoopState;
+
+struct FuriEventLoop {
+    // Only works if all operations are done from the same thread
+    FuriThreadId thread_id;
+
+    // Poller state
+    volatile FuriEventLoopState state;
+
+    // Tree
+    FuriEventLoopTree_t tree;
+    // Tree waiting list
+    WaitingList_t waiting_list;
+
+    // Tick event
+    uint32_t tick_interval;
+    FuriEventLoopTickCallback tick_callback;
+    void* tick_callback_context;
+};
+
+FuriEventLoop* furi_event_loop_alloc(void) {
+    FuriEventLoop* instance = malloc(sizeof(FuriEventLoop));
+
+    instance->thread_id = furi_thread_get_current_id();
+    FuriEventLoopTree_init(instance->tree);
+    WaitingList_init(instance->waiting_list);
+
+    return instance;
+}
+
+void furi_event_loop_free(FuriEventLoop* instance) {
+    furi_check(instance);
+    furi_check(instance->thread_id == furi_thread_get_current_id());
+
+    FuriEventLoopTree_clear(instance->tree);
+    free(instance);
+}
+
+static FuriEventLoopProcessStatus
+    furi_event_loop_poll_process_event(FuriEventLoop* instance, FuriEventLoopItem* item) {
+    UNUSED(instance);
+
+    if(!item->contract->get_level(item->object, item->event)) {
+        return FuriEventLoopProcessStatusComplete;
+    }
+
+    if(item->callback(item->object, item->callback_context)) {
+        return FuriEventLoopProcessStatusIncomplete;
+    } else {
+        return FuriEventLoopProcessStatusAgain;
+    }
+}
+
+void furi_event_loop_run(FuriEventLoop* instance) {
+    furi_check(instance);
+    furi_check(instance->thread_id == furi_thread_get_current_id());
+
+    uint32_t timeout = instance->tick_callback ? instance->tick_interval : FuriWaitForever;
+
+    while(true) {
+        uint32_t flags = 0;
+        BaseType_t ret = xTaskNotifyWaitIndexed(
+            FURI_EVENT_LOOP_FLAG_NOTIFY_INDEX, 0, FuriEventLoopFlagAll, &flags, timeout);
+
+        instance->state = FuriEventLoopStateProcessing;
+        if(ret == pdTRUE) {
+            if(flags & FuriEventLoopFlagStop) {
+                instance->state = FuriEventLoopStateIdle;
+                break;
+            } else if(flags & FuriEventLoopFlagEvent) {
+                FuriEventLoopItem* item = NULL;
+                FURI_CRITICAL_ENTER();
+                if(!WaitingList_empty_p(instance->waiting_list)) {
+                    item = WaitingList_pop_front(instance->waiting_list);
+                    WaitingList_init_field(item);
+                }
+                FURI_CRITICAL_EXIT();
+                if(item) {
+                    while(true) {
+                        FuriEventLoopProcessStatus ret =
+                            furi_event_loop_poll_process_event(instance, item);
+                        if(ret == FuriEventLoopProcessStatusComplete) {
+                            // Event processing complete, break from loop
+                            break;
+                        } else if(ret == FuriEventLoopProcessStatusIncomplete) {
+                            // Event processing incomplete more processing needed
+                        } else if(ret == FuriEventLoopProcessStatusAgain) { //-V547
+                            furi_event_loop_item_notify(item);
+                            break;
+                        } else {
+                            furi_crash();
+                        }
+                    }
+                }
+            }
+        } else {
+            if(instance->tick_callback) {
+                instance->tick_callback(instance->tick_callback_context);
+            }
+        }
+        instance->state = FuriEventLoopStateIdle;
+    }
+}
+
+void furi_event_loop_stop(FuriEventLoop* instance) {
+    furi_check(instance);
+    furi_check(instance->thread_id == furi_thread_get_current_id());
+
+    xTaskNotifyIndexed(
+        instance->thread_id, FURI_EVENT_LOOP_FLAG_NOTIFY_INDEX, FuriEventLoopFlagStop, eSetBits);
+}
+
+void furi_event_loop_tick_set(
+    FuriEventLoop* instance,
+    uint32_t interval,
+    FuriEventLoopTickCallback callback,
+    void* context) {
+    furi_check(instance);
+    furi_check(instance->thread_id == furi_thread_get_current_id());
+    furi_check(callback ? interval > 0 : true);
+
+    instance->tick_interval = interval;
+    instance->tick_callback = callback;
+    instance->tick_callback_context = context;
+}
+
+void furi_event_loop_message_queue_subscribe(
+    FuriEventLoop* instance,
+    FuriMessageQueue* message_queue,
+    FuriEventLoopEvent event,
+    FuriEventLoopMessageQueueCallback callback,
+    void* context) {
+    furi_check(instance);
+    furi_check(instance->thread_id == furi_thread_get_current_id());
+    furi_check(instance->state == FuriEventLoopStateIdle);
+    furi_check(message_queue);
+
+    FURI_CRITICAL_ENTER();
+
+    furi_check(FuriEventLoopTree_get(instance->tree, message_queue) == NULL);
+
+    // Allocate and setup item
+    FuriEventLoopItem* item = furi_event_loop_item_alloc(
+        instance, &furi_message_queue_event_loop_contract, message_queue, event);
+    furi_event_loop_item_set_callback(item, callback, context);
+
+    FuriEventLoopTree_set_at(instance->tree, message_queue, item);
+
+    FuriEventLoopLink* link = item->contract->get_link(message_queue);
+
+    if(item->event == FuriEventLoopEventIn) {
+        furi_check(link->item_in == NULL);
+        link->item_in = item;
+    } else if(item->event == FuriEventLoopEventOut) {
+        furi_check(link->item_out == NULL);
+        link->item_out = item;
+    } else {
+        furi_crash();
+    }
+
+    if(item->contract->get_level(item->object, item->event)) {
+        furi_event_loop_item_notify(item);
+    }
+
+    FURI_CRITICAL_EXIT();
+}
+
+void furi_event_loop_message_queue_unsubscribe(
+    FuriEventLoop* instance,
+    FuriMessageQueue* message_queue) {
+    furi_check(instance);
+    furi_check(instance->state == FuriEventLoopStateIdle);
+    furi_check(instance->thread_id == furi_thread_get_current_id());
+
+    FURI_CRITICAL_ENTER();
+
+    FuriEventLoopItem** item_ptr = FuriEventLoopTree_get(instance->tree, message_queue);
+    furi_check(item_ptr);
+
+    FuriEventLoopItem* item = *item_ptr;
+    furi_check(item);
+    furi_check(item->owner == instance);
+
+    FuriEventLoopLink* link = item->contract->get_link(message_queue);
+
+    if(item->event == FuriEventLoopEventIn) {
+        furi_check(link->item_in == item);
+        link->item_in = NULL;
+    } else if(item->event == FuriEventLoopEventOut) {
+        furi_check(link->item_out == item);
+        link->item_out = NULL;
+    } else {
+        furi_crash();
+    }
+
+    furi_event_loop_item_free(item);
+
+    FuriEventLoopTree_erase(instance->tree, message_queue);
+
+    FURI_CRITICAL_EXIT();
+}
+
+/* 
+ * Event Loop Item API, used internally
+ */
+
+static FuriEventLoopItem* furi_event_loop_item_alloc(
+    FuriEventLoop* owner,
+    const FuriEventLoopContract* contract,
+    void* object,
+    FuriEventLoopEvent event) {
+    furi_assert(owner);
+    furi_assert(object);
+
+    FuriEventLoopItem* instance = malloc(sizeof(FuriEventLoopItem));
+
+    instance->owner = owner;
+    instance->contract = contract;
+    instance->object = object;
+    instance->event = event;
+
+    WaitingList_init_field(instance);
+
+    return instance;
+}
+
+static void furi_event_loop_item_free(FuriEventLoopItem* instance) {
+    furi_assert(instance);
+    free(instance);
+}
+
+static void furi_event_loop_item_set_callback(
+    FuriEventLoopItem* instance,
+    FuriEventLoopMessageQueueCallback callback,
+    void* callback_context) {
+    furi_assert(instance);
+    furi_assert(!instance->callback);
+
+    instance->callback = callback;
+    instance->callback_context = callback_context;
+}
+
+static void furi_event_loop_item_notify(FuriEventLoopItem* instance) {
+    furi_assert(instance);
+
+    FURI_CRITICAL_ENTER();
+
+    if(!instance->WaitingList.prev && !instance->WaitingList.next) {
+        WaitingList_push_back(instance->owner->waiting_list, instance);
+    }
+
+    FURI_CRITICAL_EXIT();
+
+    xTaskNotifyIndexed(
+        instance->owner->thread_id,
+        FURI_EVENT_LOOP_FLAG_NOTIFY_INDEX,
+        FuriEventLoopFlagEvent,
+        eSetBits);
+}
+
+void furi_event_loop_link_notify(FuriEventLoopLink* instance, FuriEventLoopEvent event) {
+    furi_assert(instance);
+
+    FURI_CRITICAL_ENTER();
+
+    if(event == FuriEventLoopEventIn) {
+        if(instance->item_in) furi_event_loop_item_notify(instance->item_in);
+    } else if(event == FuriEventLoopEventOut) {
+        if(instance->item_out) furi_event_loop_item_notify(instance->item_out);
+    } else {
+        furi_crash();
+    }
+
+    FURI_CRITICAL_EXIT();
+}

furi/core/event_loop.h

@@ -0,0 +1,134 @@
+/**
+ * @file event_loop.h
+ * @brief      Furi Event Loop
+ *
+ *             This module is designed to handle application event loop in fully
+ *             asynchronous, reactive nature. On the low level this modules is
+ *             inspired by epoll/kqueue concept, on the high level by asyncio
+ *             event loop.
+ *
+ *             This module is trying to best fit into Furi OS, so we don't
+ *             provide any compatibility with other event driven APIs. But
+ *             programming concepts are the same, except some runtime
+ *             limitations from our side.
+ */
+#pragma once
+
+#include "base.h"
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+/** Event Loop events */
+typedef enum {
+    FuriEventLoopEventOut, /**< On departure: item was retrieved from container, flag reset, etc... */
+    FuriEventLoopEventIn, /**< On arrival: item was inserted into container, flag set, etc... */
+} FuriEventLoopEvent;
+
+/** Anonymous message queue type */
+typedef struct FuriEventLoop FuriEventLoop;
+
+/** Allocate Event Loop instance
+ *
+ * Couple things to keep in mind:
+ * - You can have 1 event_loop per 1 thread
+ * - You can not use event_loop instance in the other thread
+ * - Do not use blocking api to query object delegated to Event Loop
+ *
+ * @return     The Event Loop instance
+ */
+FuriEventLoop* furi_event_loop_alloc(void);
+
+/** Free Event Loop instance
+ *
+ * @param      instance  The Event Loop instance
+ */
+void furi_event_loop_free(FuriEventLoop* instance);
+
+/** Continuously poll for events
+ *
+ * Can be stopped with `furi_event_loop_stop`
+ *
+ * @param      instance  The Event Loop instance
+ */
+void furi_event_loop_run(FuriEventLoop* instance);
+
+/** Stop Event Loop instance
+ *
+ * @param      instance  The Event Loop instance
+ */
+void furi_event_loop_stop(FuriEventLoop* instance);
+
+/*
+ * Tick related API
+ */
+
+/** Tick callback type
+ *
+ * @param      context  The context for callback
+ */
+typedef void (*FuriEventLoopTickCallback)(void* context);
+
+/** Set Event Loop tick callback
+ *
+ * Tick callback called after specified inactivity time. It's not periodic. If
+ * Event Loop is busy then ticks will be skipped.
+ *
+ * @param      instance  The Event Loop instance
+ * @param[in]  interval  The tick interval
+ * @param[in]  callback  The callback to call
+ * @param      context   The context for callback
+ */
+void furi_event_loop_tick_set(
+    FuriEventLoop* instance,
+    uint32_t interval,
+    FuriEventLoopTickCallback callback,
+    void* context);
+
+/*
+ * Message queue related APIs
+ */
+
+/** Anonymous message queue type */
+typedef struct FuriMessageQueue FuriMessageQueue;
+
+/** Callback type for message queue
+ *
+ * @param      queue    The queue that triggered event
+ * @param      context  The context that was provided on
+ *                      furi_event_loop_message_queue_subscribe call
+ *
+ * @return     true if event was processed, false if we need to delay processing
+ */
+typedef bool (*FuriEventLoopMessageQueueCallback)(FuriMessageQueue* queue, void* context);
+
+/** Subscribe to message queue events
+ * 
+ * @warning you can only have one subscription for one event type.
+ *
+ * @param      instance       The Event Loop instance
+ * @param      message_queue  The message queue to add
+ * @param[in]  event          The Event Loop event to trigger on
+ * @param[in]  callback       The callback to call on event
+ * @param      context        The context for callback
+ */
+void furi_event_loop_message_queue_subscribe(
+    FuriEventLoop* instance,
+    FuriMessageQueue* message_queue,
+    FuriEventLoopEvent event,
+    FuriEventLoopMessageQueueCallback callback,
+    void* context);
+
+/** Unsubscribe from message queue
+ *
+ * @param      instance       The Event Loop instance
+ * @param      message_queue  The message queue
+ */
+void furi_event_loop_message_queue_unsubscribe(
+    FuriEventLoop* instance,
+    FuriMessageQueue* message_queue);
+
+#ifdef __cplusplus
+}
+#endif

furi/core/event_loop_i.h

@@ -0,0 +1,33 @@
+#pragma once
+
+#include "event_loop.h"
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+typedef struct FuriEventLoopItem FuriEventLoopItem;
+
+/* Link between Event Loop  */
+
+typedef struct {
+    FuriEventLoopItem* item_in;
+    FuriEventLoopItem* item_out;
+} FuriEventLoopLink;
+
+void furi_event_loop_link_notify(FuriEventLoopLink* instance, FuriEventLoopEvent event);
+
+/* Contract between event loop and an object */
+
+typedef FuriEventLoopLink* (*FuriEventLoopContractGetLink)(void* object);
+
+typedef uint32_t (*FuriEventLoopContractGetLevel)(void* object, FuriEventLoopEvent event);
+
+typedef struct {
+    const FuriEventLoopContractGetLink get_link;
+    const FuriEventLoopContractGetLevel get_level;
+} FuriEventLoopContract;
+
+#ifdef __cplusplus
+}
+#endif

furi/core/message_queue.c

@@ -1,9 +1,4 @@
-#include "kernel.h"
-#include "message_queue.h"
-#include "check.h"
-
-#include <FreeRTOS.h>
-#include <queue.h>
+#include "message_queue_i.h"
 
 // Internal FreeRTOS member names
 #define uxMessagesWaiting uxDummy4[0]
@@ -12,6 +7,10 @@
 
 struct FuriMessageQueue {
     StaticQueue_t container;
+
+    // Event Loop Link
+    FuriEventLoopLink event_loop_link;
+
     uint8_t buffer[];
 };
 
@@ -42,6 +41,10 @@ void furi_message_queue_free(FuriMessageQueue* instance) {
     furi_check(furi_kernel_is_irq_or_masked() == 0U);
     furi_check(instance);
 
+    // Event Loop must be disconnected
+    furi_check(!instance->event_loop_link.item_in);
+    furi_check(!instance->event_loop_link.item_out);
+
     vQueueDelete((QueueHandle_t)instance);
     free(instance);
 }
@@ -82,6 +85,11 @@ FuriStatus
         }
     }
 
+    if(stat == FuriStatusOk) {
+        furi_event_loop_link_notify(&instance->event_loop_link, FuriEventLoopEventIn);
+    }
+
+    /* Return execution status */
     return stat;
 }
 
@@ -120,6 +128,10 @@ FuriStatus furi_message_queue_get(FuriMessageQueue* instance, void* msg_ptr, uin
         }
     }
 
+    if(stat == FuriStatusOk) {
+        furi_event_loop_link_notify(&instance->event_loop_link, FuriEventLoopEventOut);
+    }
+
     return stat;
 }
 
@@ -182,5 +194,34 @@ FuriStatus furi_message_queue_reset(FuriMessageQueue* instance) {
         (void)xQueueReset(hQueue);
     }
 
+    if(stat == FuriStatusOk) {
+        furi_event_loop_link_notify(&instance->event_loop_link, FuriEventLoopEventOut);
+    }
+
+    /* Return execution status */
     return stat;
 }
+
+static FuriEventLoopLink* furi_message_queue_event_loop_get_link(void* object) {
+    FuriMessageQueue* instance = object;
+    furi_assert(instance);
+    return &instance->event_loop_link;
+}
+
+static uint32_t furi_message_queue_event_loop_get_level(void* object, FuriEventLoopEvent event) {
+    FuriMessageQueue* instance = object;
+    furi_assert(instance);
+
+    if(event == FuriEventLoopEventIn) {
+        return furi_message_queue_get_count(instance);
+    } else if(event == FuriEventLoopEventOut) {
+        return furi_message_queue_get_space(instance);
+    } else {
+        furi_crash();
+    }
+}
+
+const FuriEventLoopContract furi_message_queue_event_loop_contract = {
+    .get_link = furi_message_queue_event_loop_get_link,
+    .get_level = furi_message_queue_event_loop_get_level,
+};

furi/core/message_queue_i.h

@@ -0,0 +1,12 @@
+#pragma once
+
+#include "message_queue.h"
+
+#include "kernel.h"
+#include "event_loop_i.h"
+#include "check.h"
+
+#include <FreeRTOS.h>
+#include <queue.h>
+
+extern const FuriEventLoopContract furi_message_queue_event_loop_contract;

furi/core/stream_buffer.c

@@ -1,6 +1,6 @@
-#include "base.h"
-#include "check.h"
 #include "stream_buffer.h"
+
+#include "check.h"
 #include "common_defines.h"
 
 #include <FreeRTOS.h>

furi/core/stream_buffer.h

@@ -12,8 +12,8 @@
  * interrupt that will read from the buffer (the reader).
  */
 #pragma once
-#include <stdint.h>
-#include <stddef.h>
+
+#include "base.h"
 
 #ifdef __cplusplus
 extern "C" {

furi/furi.h

@@ -4,6 +4,7 @@
 
 #include "core/check.h"
 #include "core/common_defines.h"
+#include "core/event_loop.h"
 #include "core/event_flag.h"
 #include "core/kernel.h"
 #include "core/log.h"