[FL-3957] EventLoop unsubscribe fix (#4109)

* Fix memory leak during event loop unsubscription
* Event better memory leak fix
* unit test for the fix

Co-authored-by: Georgii Surkov <georgii.surkov@outlook.com>

applications/debug/unit_tests/tests/furi/furi_event_loop_test.c

@@ -446,6 +446,55 @@ static int32_t test_furi_event_loop_consumer(void* p) {
     return 0;
 }
 
+typedef struct {
+    FuriEventLoop* event_loop;
+    FuriSemaphore* semaphore;
+    size_t counter;
+} SelfUnsubTestTimerContext;
+
+static void test_self_unsub_semaphore_callback(FuriEventLoopObject* object, void* context) {
+    furi_event_loop_unsubscribe(context, object); // shouldn't crash here
+}
+
+static void test_self_unsub_timer_callback(void* arg) {
+    SelfUnsubTestTimerContext* context = arg;
+
+    if(context->counter == 0) {
+        furi_semaphore_release(context->semaphore);
+    } else if(context->counter == 1) {
+        furi_event_loop_stop(context->event_loop);
+    }
+
+    context->counter++;
+}
+
+void test_furi_event_loop_self_unsubscribe(void) {
+    FuriEventLoop* event_loop = furi_event_loop_alloc();
+
+    FuriSemaphore* semaphore = furi_semaphore_alloc(1, 0);
+    furi_event_loop_subscribe_semaphore(
+        event_loop,
+        semaphore,
+        FuriEventLoopEventIn,
+        test_self_unsub_semaphore_callback,
+        event_loop);
+
+    SelfUnsubTestTimerContext timer_context = {
+        .event_loop = event_loop,
+        .semaphore = semaphore,
+        .counter = 0,
+    };
+    FuriEventLoopTimer* timer = furi_event_loop_timer_alloc(
+        event_loop, test_self_unsub_timer_callback, FuriEventLoopTimerTypePeriodic, &timer_context);
+    furi_event_loop_timer_start(timer, furi_ms_to_ticks(20));
+
+    furi_event_loop_run(event_loop);
+
+    furi_event_loop_timer_free(timer);
+    furi_semaphore_free(semaphore);
+    furi_event_loop_free(event_loop);
+}
+
 void test_furi_event_loop(void) {
     TestFuriEventLoopData data = {};
 

applications/debug/unit_tests/tests/furi/furi_test.c

@@ -8,6 +8,7 @@ void test_furi_concurrent_access(void);
 void test_furi_pubsub(void);
 void test_furi_memmgr(void);
 void test_furi_event_loop(void);
+void test_furi_event_loop_self_unsubscribe(void);
 void test_errno_saving(void);
 void test_furi_primitives(void);
 void test_stdin(void);
@@ -46,6 +47,10 @@ MU_TEST(mu_test_furi_event_loop) {
     test_furi_event_loop();
 }
 
+MU_TEST(mu_test_furi_event_loop_self_unsubscribe) {
+    test_furi_event_loop_self_unsubscribe();
+}
+
 MU_TEST(mu_test_errno_saving) {
     test_errno_saving();
 }
@@ -68,6 +73,7 @@ MU_TEST_SUITE(test_suite) {
     MU_RUN_TEST(mu_test_furi_pubsub);
     MU_RUN_TEST(mu_test_furi_memmgr);
     MU_RUN_TEST(mu_test_furi_event_loop);
+    MU_RUN_TEST(mu_test_furi_event_loop_self_unsubscribe);
     MU_RUN_TEST(mu_test_stdio);
     MU_RUN_TEST(mu_test_errno_saving);
     MU_RUN_TEST(mu_test_furi_primitives);

furi/core/event_loop.c

@@ -32,14 +32,6 @@ static void furi_event_loop_item_notify(FuriEventLoopItem* instance);
 
 static bool furi_event_loop_item_is_waiting(FuriEventLoopItem* instance);
 
-static void furi_event_loop_process_pending_callbacks(FuriEventLoop* instance) {
-    for(; !PendingQueue_empty_p(instance->pending_queue);
-        PendingQueue_pop_back(NULL, instance->pending_queue)) {
-        const FuriEventLoopPendingQueueItem* item = PendingQueue_back(instance->pending_queue);
-        item->callback(item->context);
-    }
-}
-
 static bool furi_event_loop_signal_callback(uint32_t signal, void* arg, void* context) {
     furi_assert(context);
     FuriEventLoop* instance = context;
@@ -130,12 +122,16 @@ static inline FuriEventLoopProcessStatus
         furi_event_loop_unsubscribe(instance, item->object);
     }
 
+    instance->current_item = item;
+
     if(item->event & FuriEventLoopEventFlagEdge) {
         status = furi_event_loop_process_edge_event(item);
     } else {
         status = furi_event_loop_process_level_event(item);
     }
 
+    instance->current_item = NULL;
+
     if(item->owner == NULL) {
         status = FuriEventLoopProcessStatusFreeLater;
     }
@@ -193,6 +189,14 @@ static void furi_event_loop_process_waiting_list(FuriEventLoop* instance) {
     furi_event_loop_sync_flags(instance);
 }
 
+static void furi_event_loop_process_pending_callbacks(FuriEventLoop* instance) {
+    for(; !PendingQueue_empty_p(instance->pending_queue);
+        PendingQueue_pop_back(NULL, instance->pending_queue)) {
+        const FuriEventLoopPendingQueueItem* item = PendingQueue_back(instance->pending_queue);
+        item->callback(item->context);
+    }
+}
+
 static void furi_event_loop_restore_flags(FuriEventLoop* instance, uint32_t flags) {
     if(flags) {
         xTaskNotifyIndexed(
@@ -203,7 +207,6 @@ static void furi_event_loop_restore_flags(FuriEventLoop* instance, uint32_t flag
 void furi_event_loop_run(FuriEventLoop* instance) {
     furi_check(instance);
     furi_check(instance->thread_id == furi_thread_get_current_id());
-
     FuriThread* thread = furi_thread_get_current();
 
     // Set the default signal callback if none was previously set
@@ -213,9 +216,9 @@ void furi_event_loop_run(FuriEventLoop* instance) {
 
     furi_event_loop_init_tick(instance);
 
-    while(true) {
-        instance->state = FuriEventLoopStateIdle;
+    instance->state = FuriEventLoopStateRunning;
 
+    while(true) {
         const TickType_t ticks_to_sleep =
             MIN(furi_event_loop_get_timer_wait_time(instance),
                 furi_event_loop_get_tick_wait_time(instance));
@@ -224,8 +227,6 @@ void furi_event_loop_run(FuriEventLoop* instance) {
         BaseType_t ret = xTaskNotifyWaitIndexed(
             FURI_EVENT_LOOP_FLAG_NOTIFY_INDEX, 0, FuriEventLoopFlagAll, &flags, ticks_to_sleep);
 
-        instance->state = FuriEventLoopStateProcessing;
-
         if(ret == pdTRUE) {
             if(flags & FuriEventLoopFlagStop) {
                 instance->state = FuriEventLoopStateStopped;
@@ -448,7 +449,7 @@ void furi_event_loop_unsubscribe(FuriEventLoop* instance, FuriEventLoopObject* o
         WaitingList_unlink(item);
     }
 
-    if(instance->state == FuriEventLoopStateProcessing) {
+    if(instance->current_item == item) {
         furi_event_loop_item_free_later(item);
     } else {
         furi_event_loop_item_free(item);

furi/core/event_loop_i.h

@@ -64,8 +64,7 @@ typedef enum {
 
 typedef enum {
     FuriEventLoopStateStopped,
-    FuriEventLoopStateIdle,
-    FuriEventLoopStateProcessing,
+    FuriEventLoopStateRunning,
 } FuriEventLoopState;
 
 typedef struct {
@@ -81,6 +80,7 @@ struct FuriEventLoop {
 
     // Poller state
     volatile FuriEventLoopState state;
+    volatile FuriEventLoopItem* current_item;
 
     // Event handling
     FuriEventLoopTree_t tree;