[FL-3764] Expansion module service improvements (#3429)

* Separate expansion control and worker threads
* Add edge case checks
* Reduce expansion control thread stack size, add comments
* Fix crash when disabling expansion modules
* Show a different RPC icon for expansion modules
* Restore expansion interrupt on changing logging settings
* Improve responsiveness in heavy games at the expense of dropped frames
* Improve furi_hal_serial API
* Fix a typo
* Remove too optimistic furi_check, replace with condition
* Fix premature RX interrupt during serial configuration
* Disable expansion interrupt if the handle was acquired
* Do not use a timer callback

Co-authored-by: Aleksandr Kutuzov <alleteam@gmail.com>

applications/services/expansion/application.fam

@@ -8,5 +8,5 @@ App(
     ],
     requires=["rpc_start"],
     provides=["expansion_settings"],
-    order=10,
+    order=150,
 )

applications/services/expansion/expansion.c

@@ -1,19 +1,17 @@
 #include "expansion.h"
 
-#include <furi_hal_power.h>
-#include <furi_hal_serial.h>
 #include <furi_hal_serial_control.h>
 
 #include <furi.h>
+#include <toolbox/api_lock.h>
 
-#include <rpc/rpc.h>
+#include "expansion_worker.h"
-
 #include "expansion_settings.h"
-#include "expansion_protocol.h"
 
 #define TAG "ExpansionSrv"
 
-#define EXPANSION_BUFFER_SIZE (sizeof(ExpansionFrame) + sizeof(ExpansionFrameChecksum))
+#define EXPANSION_CONTROL_QUEUE_SIZE (8UL)
+#define EXPANSION_CONTROL_STACK_SIZE (768UL)
 
 typedef enum {
     ExpansionStateDisabled,
@@ -22,368 +20,196 @@ typedef enum {
 } ExpansionState;
 
 typedef enum {
-    ExpansionSessionStateHandShake,
+    ExpansionMessageTypeEnable,
-    ExpansionSessionStateConnected,
+    ExpansionMessageTypeDisable,
-    ExpansionSessionStateRpcActive,
+    ExpansionMessageTypeSetListenSerial,
-} ExpansionSessionState;
+    ExpansionMessageTypeModuleConnected,
+    ExpansionMessageTypeModuleDisconnected,
+} ExpansionMessageType;
 
-typedef enum {
+typedef union {
-    ExpansionSessionExitReasonUnknown,
+    FuriHalSerialId serial_id;
-    ExpansionSessionExitReasonUser,
+} ExpansionMessageData;
-    ExpansionSessionExitReasonError,
-    ExpansionSessionExitReasonTimeout,
-} ExpansionSessionExitReason;
 
-typedef enum {
+typedef struct {
-    ExpansionFlagStop = 1 << 0,
+    ExpansionMessageType type;
-    ExpansionFlagData = 1 << 1,
+    ExpansionMessageData data;
-    ExpansionFlagError = 1 << 2,
+    FuriApiLock api_lock;
-} ExpansionFlag;
+} ExpansionMessage;
-
-#define EXPANSION_ALL_FLAGS (ExpansionFlagData | ExpansionFlagStop)
 
 struct Expansion {
-    ExpansionState state;
+    FuriThread* thread;
-    ExpansionSessionState session_state;
+    FuriMessageQueue* queue;
-    ExpansionSessionExitReason exit_reason;
-    FuriStreamBuffer* rx_buf;
-    FuriSemaphore* tx_semaphore;
-    FuriMutex* state_mutex;
-    FuriThread* worker_thread;
     FuriHalSerialId serial_id;
-    FuriHalSerialHandle* serial_handle;
+    ExpansionWorker* worker;
-    RpcSession* rpc_session;
+    ExpansionState state;
 };
 
-static void expansion_detect_callback(void* context);
+static const char* const expansion_uart_names[] = {
-
+    "USART",
-// Called in UART IRQ context
+    "LPUART",
-static void expansion_serial_rx_callback(
+};
-    FuriHalSerialHandle* handle,
-    FuriHalSerialRxEvent event,
-    void* context) {
-    furi_assert(handle);
-    furi_assert(context);
-
-    Expansion* instance = context;
-
-    if(event == FuriHalSerialRxEventData) {
-        const uint8_t data = furi_hal_serial_async_rx(handle);
-        furi_stream_buffer_send(instance->rx_buf, &data, sizeof(data), 0);
-        furi_thread_flags_set(furi_thread_get_id(instance->worker_thread), ExpansionFlagData);
-    }
-}
-
-static size_t expansion_receive_callback(uint8_t* data, size_t data_size, void* context) {
-    Expansion* instance = context;
-
-    size_t received_size = 0;
-
-    while(true) {
-        received_size += furi_stream_buffer_receive(
-            instance->rx_buf, data + received_size, data_size - received_size, 0);
-
-        if(received_size == data_size) break;
-
-        const uint32_t flags = furi_thread_flags_wait(
-            EXPANSION_ALL_FLAGS, FuriFlagWaitAny, furi_ms_to_ticks(EXPANSION_PROTOCOL_TIMEOUT_MS));
-
-        if(flags & FuriFlagError) {
-            if(flags == (unsigned)FuriFlagErrorTimeout) {
-                // Exiting due to timeout
-                instance->exit_reason = ExpansionSessionExitReasonTimeout;
-            } else {
-                // Exiting due to an unspecified error
-                instance->exit_reason = ExpansionSessionExitReasonError;
-            }
-            break;
-        } else if(flags & ExpansionFlagStop) {
-            // Exiting due to explicit request
-            instance->exit_reason = ExpansionSessionExitReasonUser;
-            break;
-        } else if(flags & ExpansionFlagError) {
-            // Exiting due to RPC error
-            instance->exit_reason = ExpansionSessionExitReasonError;
-            break;
-        } else if(flags & ExpansionFlagData) {
-            // Go to buffer reading
-            continue;
-        }
-    }
-
-    return received_size;
-}
-
-static inline bool expansion_receive_frame(Expansion* instance, ExpansionFrame* frame) {
-    return expansion_protocol_decode(frame, expansion_receive_callback, instance) ==
-           ExpansionProtocolStatusOk;
-}
-
-static size_t expansion_send_callback(const uint8_t* data, size_t data_size, void* context) {
-    Expansion* instance = context;
-    furi_hal_serial_tx(instance->serial_handle, data, data_size);
-    furi_hal_serial_tx_wait_complete(instance->serial_handle);
-    return data_size;
-}
-
-static inline bool expansion_send_frame(Expansion* instance, const ExpansionFrame* frame) {
-    return expansion_protocol_encode(frame, expansion_send_callback, instance) ==
-           ExpansionProtocolStatusOk;
-}
-
-static bool expansion_send_heartbeat(Expansion* instance) {
-    const ExpansionFrame frame = {
-        .header.type = ExpansionFrameTypeHeartbeat,
-        .content.heartbeat = {},
-    };
-
-    return expansion_send_frame(instance, &frame);
-}
-
-static bool expansion_send_status_response(Expansion* instance, ExpansionFrameError error) {
-    const ExpansionFrame frame = {
-        .header.type = ExpansionFrameTypeStatus,
-        .content.status.error = error,
-    };
-
-    return expansion_send_frame(instance, &frame);
-}
-
-static bool
-    expansion_send_data_response(Expansion* instance, const uint8_t* data, size_t data_size) {
-    furi_assert(data_size <= EXPANSION_PROTOCOL_MAX_DATA_SIZE);
-
-    ExpansionFrame frame = {
-        .header.type = ExpansionFrameTypeData,
-        .content.data.size = data_size,
-    };
-
-    memcpy(frame.content.data.bytes, data, data_size);
-    return expansion_send_frame(instance, &frame);
-}
-
-// Called in Rpc session thread context
-static void expansion_rpc_send_callback(void* context, uint8_t* data, size_t data_size) {
-    Expansion* instance = context;
-
-    for(size_t sent_data_size = 0; sent_data_size < data_size;) {
-        if(furi_semaphore_acquire(
-               instance->tx_semaphore, furi_ms_to_ticks(EXPANSION_PROTOCOL_TIMEOUT_MS)) !=
-           FuriStatusOk) {
-            furi_thread_flags_set(furi_thread_get_id(instance->worker_thread), ExpansionFlagError);
-            break;
-        }
-
-        const size_t current_data_size =
-            MIN(data_size - sent_data_size, EXPANSION_PROTOCOL_MAX_DATA_SIZE);
-        if(!expansion_send_data_response(instance, data + sent_data_size, current_data_size))
-            break;
-        sent_data_size += current_data_size;
-    }
-}
-
-static bool expansion_rpc_session_open(Expansion* instance) {
-    Rpc* rpc = furi_record_open(RECORD_RPC);
-    instance->rpc_session = rpc_session_open(rpc, RpcOwnerUart);
-
-    if(instance->rpc_session) {
-        instance->tx_semaphore = furi_semaphore_alloc(1, 1);
-        rpc_session_set_context(instance->rpc_session, instance);
-        rpc_session_set_send_bytes_callback(instance->rpc_session, expansion_rpc_send_callback);
-    }
-
-    return instance->rpc_session != NULL;
-}
-
-static void expansion_rpc_session_close(Expansion* instance) {
-    if(instance->rpc_session) {
-        rpc_session_close(instance->rpc_session);
-        furi_semaphore_free(instance->tx_semaphore);
-    }
-
-    furi_record_close(RECORD_RPC);
-}
-
-static bool
-    expansion_handle_session_state_handshake(Expansion* instance, const ExpansionFrame* rx_frame) {
-    bool success = false;
-
-    do {
-        if(rx_frame->header.type != ExpansionFrameTypeBaudRate) break;
-        const uint32_t baud_rate = rx_frame->content.baud_rate.baud;
-
-        FURI_LOG_D(TAG, "Proposed baud rate: %lu", baud_rate);
-
-        if(furi_hal_serial_is_baud_rate_supported(instance->serial_handle, baud_rate)) {
-            instance->session_state = ExpansionSessionStateConnected;
-            // Send response at previous baud rate
-            if(!expansion_send_status_response(instance, ExpansionFrameErrorNone)) break;
-            furi_hal_serial_set_br(instance->serial_handle, baud_rate);
-
-        } else {
-            if(!expansion_send_status_response(instance, ExpansionFrameErrorBaudRate)) break;
-            FURI_LOG_E(TAG, "Bad baud rate");
-        }
-        success = true;
-    } while(false);
-
-    return success;
-}
-
-static bool
-    expansion_handle_session_state_connected(Expansion* instance, const ExpansionFrame* rx_frame) {
-    bool success = false;
-
-    do {
-        if(rx_frame->header.type == ExpansionFrameTypeControl) {
-            if(rx_frame->content.control.command != ExpansionFrameControlCommandStartRpc) break;
-            instance->session_state = ExpansionSessionStateRpcActive;
-            if(!expansion_rpc_session_open(instance)) break;
-            if(!expansion_send_status_response(instance, ExpansionFrameErrorNone)) break;
-
-        } else if(rx_frame->header.type == ExpansionFrameTypeHeartbeat) {
-            if(!expansion_send_heartbeat(instance)) break;
-
-        } else {
-            break;
-        }
-        success = true;
-    } while(false);
-
-    return success;
-}
-
-static bool
-    expansion_handle_session_state_rpc_active(Expansion* instance, const ExpansionFrame* rx_frame) {
-    bool success = false;
-
-    do {
-        if(rx_frame->header.type == ExpansionFrameTypeData) {
-            if(!expansion_send_status_response(instance, ExpansionFrameErrorNone)) break;
-
-            const size_t size_consumed = rpc_session_feed(
-                instance->rpc_session,
-                rx_frame->content.data.bytes,
-                rx_frame->content.data.size,
-                EXPANSION_PROTOCOL_TIMEOUT_MS);
-            if(size_consumed != rx_frame->content.data.size) break;
-
-        } else if(rx_frame->header.type == ExpansionFrameTypeControl) {
-            if(rx_frame->content.control.command != ExpansionFrameControlCommandStopRpc) break;
-            instance->session_state = ExpansionSessionStateConnected;
-            expansion_rpc_session_close(instance);
-            if(!expansion_send_status_response(instance, ExpansionFrameErrorNone)) break;
-
-        } else if(rx_frame->header.type == ExpansionFrameTypeStatus) {
-            if(rx_frame->content.status.error != ExpansionFrameErrorNone) break;
-            furi_semaphore_release(instance->tx_semaphore);
-
-        } else if(rx_frame->header.type == ExpansionFrameTypeHeartbeat) {
-            if(!expansion_send_heartbeat(instance)) break;
-
-        } else {
-            break;
-        }
-        success = true;
-    } while(false);
-
-    return success;
-}
-
-static inline void expansion_state_machine(Expansion* instance) {
-    typedef bool (*ExpansionSessionStateHandler)(Expansion*, const ExpansionFrame*);
-
-    static const ExpansionSessionStateHandler expansion_handlers[] = {
-        [ExpansionSessionStateHandShake] = expansion_handle_session_state_handshake,
-        [ExpansionSessionStateConnected] = expansion_handle_session_state_connected,
-        [ExpansionSessionStateRpcActive] = expansion_handle_session_state_rpc_active,
-    };
-
-    ExpansionFrame rx_frame;
-
-    while(true) {
-        if(!expansion_receive_frame(instance, &rx_frame)) break;
-        if(!expansion_handlers[instance->session_state](instance, &rx_frame)) break;
-    }
-}
-
-static void expansion_worker_pending_callback(void* context, uint32_t arg) {
-    furi_assert(context);
-    UNUSED(arg);
-
-    Expansion* instance = context;
-    furi_thread_join(instance->worker_thread);
-
-    // Do not re-enable detection interrupt on user-requested exit
-    if(instance->exit_reason != ExpansionSessionExitReasonUser) {
-        furi_check(furi_mutex_acquire(instance->state_mutex, FuriWaitForever) == FuriStatusOk);
-        instance->state = ExpansionStateEnabled;
-        furi_hal_serial_control_set_expansion_callback(
-            instance->serial_id, expansion_detect_callback, instance);
-        furi_mutex_release(instance->state_mutex);
-    }
-}
-
-static int32_t expansion_worker(void* context) {
-    furi_assert(context);
-    Expansion* instance = context;
-
-    furi_hal_power_insomnia_enter();
-    furi_hal_serial_control_set_expansion_callback(instance->serial_id, NULL, NULL);
-
-    instance->serial_handle = furi_hal_serial_control_acquire(instance->serial_id);
-    furi_check(instance->serial_handle);
-
-    FURI_LOG_D(TAG, "Service started");
-
-    instance->rx_buf = furi_stream_buffer_alloc(EXPANSION_BUFFER_SIZE, 1);
-    instance->session_state = ExpansionSessionStateHandShake;
-    instance->exit_reason = ExpansionSessionExitReasonUnknown;
-
-    furi_hal_serial_init(instance->serial_handle, EXPANSION_PROTOCOL_DEFAULT_BAUD_RATE);
-
-    furi_hal_serial_async_rx_start(
-        instance->serial_handle, expansion_serial_rx_callback, instance, false);
-
-    if(expansion_send_heartbeat(instance)) {
-        expansion_state_machine(instance);
-    }
-
-    if(instance->session_state == ExpansionSessionStateRpcActive) {
-        expansion_rpc_session_close(instance);
-    }
-
-    FURI_LOG_D(TAG, "Service stopped");
-
-    furi_hal_serial_control_release(instance->serial_handle);
-    furi_stream_buffer_free(instance->rx_buf);
-
-    furi_hal_power_insomnia_exit();
-    furi_timer_pending_callback(expansion_worker_pending_callback, instance, 0);
-
-    return 0;
-}
 
 // Called from the serial control thread
 static void expansion_detect_callback(void* context) {
     furi_assert(context);
     Expansion* instance = context;
 
-    furi_check(furi_mutex_acquire(instance->state_mutex, FuriWaitForever) == FuriStatusOk);
+    ExpansionMessage message = {
+        .type = ExpansionMessageTypeModuleConnected,
+        .api_lock = NULL, // Not locking the API here to avoid a deadlock
+    };
 
-    if(instance->state == ExpansionStateEnabled) {
+    // Not waiting for available queue space, discarding message if there is none
-        instance->state = ExpansionStateRunning;
+    const FuriStatus status = furi_message_queue_put(instance->queue, &message, 0);
-        furi_thread_start(instance->worker_thread);
+    UNUSED(status);
+}
+
+static void expansion_worker_callback(void* context) {
+    furi_assert(context);
+    Expansion* instance = context;
+
+    ExpansionMessage message = {
+        .type = ExpansionMessageTypeModuleDisconnected,
+        .api_lock = NULL, // Not locking the API here to avoid a deadlock
+    };
+
+    const FuriStatus status = furi_message_queue_put(instance->queue, &message, FuriWaitForever);
+    furi_check(status == FuriStatusOk);
+}
+
+static void
+    expansion_control_handler_enable(Expansion* instance, const ExpansionMessageData* data) {
+    UNUSED(data);
+
+    if(instance->state != ExpansionStateDisabled) {
+        return;
     }
 
-    furi_mutex_release(instance->state_mutex);
+    ExpansionSettings settings = {0};
+    if(!expansion_settings_load(&settings)) {
+        expansion_settings_save(&settings);
+    }
+
+    if(settings.uart_index < FuriHalSerialIdMax) {
+        instance->state = ExpansionStateEnabled;
+        instance->serial_id = settings.uart_index;
+        furi_hal_serial_control_set_expansion_callback(
+            instance->serial_id, expansion_detect_callback, instance);
+
+        FURI_LOG_D(TAG, "Detection enabled on %s", expansion_uart_names[instance->serial_id]);
+    }
+}
+
+static void
+    expansion_control_handler_disable(Expansion* instance, const ExpansionMessageData* data) {
+    UNUSED(data);
+
+    if(instance->state == ExpansionStateDisabled) {
+        return;
+    } else if(instance->state == ExpansionStateRunning) {
+        expansion_worker_stop(instance->worker);
+        expansion_worker_free(instance->worker);
+    } else {
+        furi_hal_serial_control_set_expansion_callback(instance->serial_id, NULL, NULL);
+    }
+
+    instance->state = ExpansionStateDisabled;
+
+    FURI_LOG_D(TAG, "Detection disabled");
+}
+
+static void expansion_control_handler_set_listen_serial(
+    Expansion* instance,
+    const ExpansionMessageData* data) {
+    furi_check(data->serial_id < FuriHalSerialIdMax);
+
+    if(instance->state == ExpansionStateRunning) {
+        expansion_worker_stop(instance->worker);
+        expansion_worker_free(instance->worker);
+
+    } else if(instance->state == ExpansionStateEnabled) {
+        furi_hal_serial_control_set_expansion_callback(instance->serial_id, NULL, NULL);
+    }
+
+    instance->state = ExpansionStateEnabled;
+    instance->serial_id = data->serial_id;
+
+    furi_hal_serial_control_set_expansion_callback(
+        instance->serial_id, expansion_detect_callback, instance);
+
+    FURI_LOG_D(TAG, "Listen serial changed to %s", expansion_uart_names[instance->serial_id]);
+}
+
+static void expansion_control_handler_module_connected(
+    Expansion* instance,
+    const ExpansionMessageData* data) {
+    UNUSED(data);
+    if(instance->state != ExpansionStateEnabled) {
+        return;
+    }
+
+    furi_hal_serial_control_set_expansion_callback(instance->serial_id, NULL, NULL);
+
+    instance->state = ExpansionStateRunning;
+    instance->worker = expansion_worker_alloc(instance->serial_id);
+
+    expansion_worker_set_callback(instance->worker, expansion_worker_callback, instance);
+    expansion_worker_start(instance->worker);
+}
+
+static void expansion_control_handler_module_disconnected(
+    Expansion* instance,
+    const ExpansionMessageData* data) {
+    UNUSED(data);
+    if(instance->state != ExpansionStateRunning) {
+        return;
+    }
+
+    instance->state = ExpansionStateEnabled;
+    expansion_worker_free(instance->worker);
+    furi_hal_serial_control_set_expansion_callback(
+        instance->serial_id, expansion_detect_callback, instance);
+}
+
+typedef void (*ExpansionControlHandler)(Expansion*, const ExpansionMessageData*);
+
+static const ExpansionControlHandler expansion_control_handlers[] = {
+    [ExpansionMessageTypeEnable] = expansion_control_handler_enable,
+    [ExpansionMessageTypeDisable] = expansion_control_handler_disable,
+    [ExpansionMessageTypeSetListenSerial] = expansion_control_handler_set_listen_serial,
+    [ExpansionMessageTypeModuleConnected] = expansion_control_handler_module_connected,
+    [ExpansionMessageTypeModuleDisconnected] = expansion_control_handler_module_disconnected,
+};
+
+static int32_t expansion_control(void* context) {
+    furi_assert(context);
+    Expansion* instance = context;
+
+    for(;;) {
+        ExpansionMessage message;
+
+        FuriStatus status = furi_message_queue_get(instance->queue, &message, FuriWaitForever);
+        furi_check(status == FuriStatusOk);
+
+        furi_check(message.type < COUNT_OF(expansion_control_handlers));
+        expansion_control_handlers[message.type](instance, &message.data);
+
+        if(message.api_lock != NULL) {
+            api_lock_unlock(message.api_lock);
+        }
+    }
+
+    return 0;
 }
 
 static Expansion* expansion_alloc() {
     Expansion* instance = malloc(sizeof(Expansion));
 
-    instance->state_mutex = furi_mutex_alloc(FuriMutexTypeNormal);
+    instance->queue =
-    instance->worker_thread = furi_thread_alloc_ex(TAG, 768, expansion_worker, instance);
+        furi_message_queue_alloc(EXPANSION_CONTROL_QUEUE_SIZE, sizeof(ExpansionMessage));
+    instance->thread =
+        furi_thread_alloc_ex(TAG, EXPANSION_CONTROL_STACK_SIZE, expansion_control, instance);
 
     return instance;
 }
@@ -393,6 +219,7 @@ void expansion_on_system_start(void* arg) {
 
     Expansion* instance = expansion_alloc();
     furi_record_create(RECORD_EXPANSION, instance);
+    furi_thread_start(instance->thread);
 
     expansion_enable(instance);
 }
@@ -400,42 +227,39 @@ void expansion_on_system_start(void* arg) {
 // Public API functions
 
 void expansion_enable(Expansion* instance) {
-    ExpansionSettings settings = {};
+    furi_check(instance);
-    if(!expansion_settings_load(&settings)) {
+
-        expansion_settings_save(&settings);
+    ExpansionMessage message = {
-    } else if(settings.uart_index < FuriHalSerialIdMax) {
+        .type = ExpansionMessageTypeEnable,
-        expansion_set_listen_serial(instance, settings.uart_index);
+        .api_lock = api_lock_alloc_locked(),
-    }
+    };
+
+    furi_message_queue_put(instance->queue, &message, FuriWaitForever);
+    api_lock_wait_unlock_and_free(message.api_lock);
 }
 
 void expansion_disable(Expansion* instance) {
-    furi_check(furi_mutex_acquire(instance->state_mutex, FuriWaitForever) == FuriStatusOk);
+    furi_check(instance);
 
-    if(instance->state == ExpansionStateRunning) {
+    ExpansionMessage message = {
-        furi_thread_flags_set(furi_thread_get_id(instance->worker_thread), ExpansionFlagStop);
+        .type = ExpansionMessageTypeDisable,
-        furi_thread_join(instance->worker_thread);
+        .api_lock = api_lock_alloc_locked(),
-    } else if(instance->state == ExpansionStateEnabled) {
+    };
-        FURI_LOG_D(TAG, "Detection disabled");
-        furi_hal_serial_control_set_expansion_callback(instance->serial_id, NULL, NULL);
-    }
 
-    instance->state = ExpansionStateDisabled;
+    furi_message_queue_put(instance->queue, &message, FuriWaitForever);
-
+    api_lock_wait_unlock_and_free(message.api_lock);
-    furi_mutex_release(instance->state_mutex);
 }
 
 void expansion_set_listen_serial(Expansion* instance, FuriHalSerialId serial_id) {
-    expansion_disable(instance);
+    furi_check(instance);
+    furi_check(serial_id < FuriHalSerialIdMax);
 
-    furi_check(furi_mutex_acquire(instance->state_mutex, FuriWaitForever) == FuriStatusOk);
+    ExpansionMessage message = {
+        .type = ExpansionMessageTypeSetListenSerial,
+        .data.serial_id = serial_id,
+        .api_lock = api_lock_alloc_locked(),
+    };
 
-    instance->serial_id = serial_id;
+    furi_message_queue_put(instance->queue, &message, FuriWaitForever);
-    instance->state = ExpansionStateEnabled;
+    api_lock_wait_unlock_and_free(message.api_lock);
-
-    furi_hal_serial_control_set_expansion_callback(
-        instance->serial_id, expansion_detect_callback, instance);
-
-    furi_mutex_release(instance->state_mutex);
-
-    FURI_LOG_D(TAG, "Detection enabled");
 }

applications/services/expansion/expansion_worker.c

@@ -0,0 +1,396 @@
+#include "expansion_worker.h"
+
+#include <furi_hal_power.h>
+#include <furi_hal_serial.h>
+#include <furi_hal_serial_control.h>
+
+#include <furi.h>
+#include <rpc/rpc.h>
+
+#include "expansion_protocol.h"
+
+#define TAG "ExpansionSrv"
+
+#define EXPANSION_WORKER_STACK_SZIE (768UL)
+#define EXPANSION_WORKER_BUFFER_SIZE (sizeof(ExpansionFrame) + sizeof(ExpansionFrameChecksum))
+
+typedef enum {
+    ExpansionWorkerStateHandShake,
+    ExpansionWorkerStateConnected,
+    ExpansionWorkerStateRpcActive,
+} ExpansionWorkerState;
+
+typedef enum {
+    ExpansionWorkerExitReasonUnknown,
+    ExpansionWorkerExitReasonUser,
+    ExpansionWorkerExitReasonError,
+    ExpansionWorkerExitReasonTimeout,
+} ExpansionWorkerExitReason;
+
+typedef enum {
+    ExpansionWorkerFlagStop = 1 << 0,
+    ExpansionWorkerFlagData = 1 << 1,
+    ExpansionWorkerFlagError = 1 << 2,
+} ExpansionWorkerFlag;
+
+#define EXPANSION_ALL_FLAGS (ExpansionWorkerFlagData | ExpansionWorkerFlagStop)
+
+struct ExpansionWorker {
+    FuriThread* thread;
+    FuriStreamBuffer* rx_buf;
+    FuriSemaphore* tx_semaphore;
+
+    FuriHalSerialId serial_id;
+    FuriHalSerialHandle* serial_handle;
+
+    RpcSession* rpc_session;
+
+    ExpansionWorkerState state;
+    ExpansionWorkerExitReason exit_reason;
+    ExpansionWorkerCallback callback;
+    void* cb_context;
+};
+
+// Called in UART IRQ context
+static void expansion_worker_serial_rx_callback(
+    FuriHalSerialHandle* handle,
+    FuriHalSerialRxEvent event,
+    void* context) {
+    furi_assert(handle);
+    furi_assert(context);
+
+    ExpansionWorker* instance = context;
+
+    if(event & (FuriHalSerialRxEventNoiseError | FuriHalSerialRxEventFrameError |
+                FuriHalSerialRxEventOverrunError)) {
+        furi_thread_flags_set(furi_thread_get_id(instance->thread), ExpansionWorkerFlagError);
+    } else if(event & FuriHalSerialRxEventData) {
+        while(furi_hal_serial_async_rx_available(handle)) {
+            const uint8_t data = furi_hal_serial_async_rx(handle);
+            furi_stream_buffer_send(instance->rx_buf, &data, sizeof(data), 0);
+        }
+        furi_thread_flags_set(furi_thread_get_id(instance->thread), ExpansionWorkerFlagData);
+    }
+}
+
+static size_t expansion_worker_receive_callback(uint8_t* data, size_t data_size, void* context) {
+    ExpansionWorker* instance = context;
+
+    size_t received_size = 0;
+
+    while(true) {
+        received_size += furi_stream_buffer_receive(
+            instance->rx_buf, data + received_size, data_size - received_size, 0);
+
+        if(received_size == data_size) break;
+
+        const uint32_t flags = furi_thread_flags_wait(
+            EXPANSION_ALL_FLAGS, FuriFlagWaitAny, furi_ms_to_ticks(EXPANSION_PROTOCOL_TIMEOUT_MS));
+
+        if(flags & FuriFlagError) {
+            if(flags == (unsigned)FuriFlagErrorTimeout) {
+                // Exiting due to timeout
+                instance->exit_reason = ExpansionWorkerExitReasonTimeout;
+            } else {
+                // Exiting due to an unspecified error
+                instance->exit_reason = ExpansionWorkerExitReasonError;
+            }
+            break;
+        } else if(flags & ExpansionWorkerFlagStop) {
+            // Exiting due to explicit request
+            instance->exit_reason = ExpansionWorkerExitReasonUser;
+            break;
+        } else if(flags & ExpansionWorkerFlagError) {
+            // Exiting due to RPC error
+            instance->exit_reason = ExpansionWorkerExitReasonError;
+            break;
+        } else if(flags & ExpansionWorkerFlagData) {
+            // Go to buffer reading
+            continue;
+        }
+    }
+
+    return received_size;
+}
+
+static inline bool
+    expansion_worker_receive_frame(ExpansionWorker* instance, ExpansionFrame* frame) {
+    return expansion_protocol_decode(frame, expansion_worker_receive_callback, instance) ==
+           ExpansionProtocolStatusOk;
+}
+
+static size_t
+    expansion_worker_send_callback(const uint8_t* data, size_t data_size, void* context) {
+    ExpansionWorker* instance = context;
+    furi_hal_serial_tx(instance->serial_handle, data, data_size);
+    furi_hal_serial_tx_wait_complete(instance->serial_handle);
+    return data_size;
+}
+
+static inline bool
+    expansion_worker_send_frame(ExpansionWorker* instance, const ExpansionFrame* frame) {
+    return expansion_protocol_encode(frame, expansion_worker_send_callback, instance) ==
+           ExpansionProtocolStatusOk;
+}
+
+static bool expansion_worker_send_heartbeat(ExpansionWorker* instance) {
+    const ExpansionFrame frame = {
+        .header.type = ExpansionFrameTypeHeartbeat,
+        .content.heartbeat = {},
+    };
+
+    return expansion_worker_send_frame(instance, &frame);
+}
+
+static bool
+    expansion_worker_send_status_response(ExpansionWorker* instance, ExpansionFrameError error) {
+    const ExpansionFrame frame = {
+        .header.type = ExpansionFrameTypeStatus,
+        .content.status.error = error,
+    };
+
+    return expansion_worker_send_frame(instance, &frame);
+}
+
+static bool expansion_worker_send_data_response(
+    ExpansionWorker* instance,
+    const uint8_t* data,
+    size_t data_size) {
+    furi_assert(data_size <= EXPANSION_PROTOCOL_MAX_DATA_SIZE);
+
+    ExpansionFrame frame = {
+        .header.type = ExpansionFrameTypeData,
+        .content.data.size = data_size,
+    };
+
+    memcpy(frame.content.data.bytes, data, data_size);
+    return expansion_worker_send_frame(instance, &frame);
+}
+
+// Called in Rpc session thread context
+static void expansion_worker_rpc_send_callback(void* context, uint8_t* data, size_t data_size) {
+    ExpansionWorker* instance = context;
+
+    for(size_t sent_data_size = 0; sent_data_size < data_size;) {
+        if(furi_semaphore_acquire(
+               instance->tx_semaphore, furi_ms_to_ticks(EXPANSION_PROTOCOL_TIMEOUT_MS)) !=
+           FuriStatusOk) {
+            furi_thread_flags_set(furi_thread_get_id(instance->thread), ExpansionWorkerFlagError);
+            break;
+        }
+
+        const size_t current_data_size =
+            MIN(data_size - sent_data_size, EXPANSION_PROTOCOL_MAX_DATA_SIZE);
+        if(!expansion_worker_send_data_response(instance, data + sent_data_size, current_data_size))
+            break;
+        sent_data_size += current_data_size;
+    }
+}
+
+static bool expansion_worker_rpc_session_open(ExpansionWorker* instance) {
+    Rpc* rpc = furi_record_open(RECORD_RPC);
+    instance->rpc_session = rpc_session_open(rpc, RpcOwnerUart);
+
+    if(instance->rpc_session) {
+        instance->tx_semaphore = furi_semaphore_alloc(1, 1);
+        rpc_session_set_context(instance->rpc_session, instance);
+        rpc_session_set_send_bytes_callback(
+            instance->rpc_session, expansion_worker_rpc_send_callback);
+    }
+
+    return instance->rpc_session != NULL;
+}
+
+static void expansion_worker_rpc_session_close(ExpansionWorker* instance) {
+    if(instance->rpc_session) {
+        rpc_session_close(instance->rpc_session);
+        furi_semaphore_free(instance->tx_semaphore);
+    }
+
+    furi_record_close(RECORD_RPC);
+}
+
+static bool expansion_worker_handle_state_handshake(
+    ExpansionWorker* instance,
+    const ExpansionFrame* rx_frame) {
+    bool success = false;
+
+    do {
+        if(rx_frame->header.type != ExpansionFrameTypeBaudRate) break;
+        const uint32_t baud_rate = rx_frame->content.baud_rate.baud;
+
+        FURI_LOG_D(TAG, "Proposed baud rate: %lu", baud_rate);
+
+        if(furi_hal_serial_is_baud_rate_supported(instance->serial_handle, baud_rate)) {
+            instance->state = ExpansionWorkerStateConnected;
+            // Send response at previous baud rate
+            if(!expansion_worker_send_status_response(instance, ExpansionFrameErrorNone)) break;
+            furi_hal_serial_set_br(instance->serial_handle, baud_rate);
+
+        } else {
+            if(!expansion_worker_send_status_response(instance, ExpansionFrameErrorBaudRate))
+                break;
+            FURI_LOG_E(TAG, "Bad baud rate");
+        }
+        success = true;
+    } while(false);
+
+    return success;
+}
+
+static bool expansion_worker_handle_state_connected(
+    ExpansionWorker* instance,
+    const ExpansionFrame* rx_frame) {
+    bool success = false;
+
+    do {
+        if(rx_frame->header.type == ExpansionFrameTypeControl) {
+            if(rx_frame->content.control.command != ExpansionFrameControlCommandStartRpc) break;
+            instance->state = ExpansionWorkerStateRpcActive;
+            if(!expansion_worker_rpc_session_open(instance)) break;
+            if(!expansion_worker_send_status_response(instance, ExpansionFrameErrorNone)) break;
+
+        } else if(rx_frame->header.type == ExpansionFrameTypeHeartbeat) {
+            if(!expansion_worker_send_heartbeat(instance)) break;
+
+        } else {
+            break;
+        }
+        success = true;
+    } while(false);
+
+    return success;
+}
+
+static bool expansion_worker_handle_state_rpc_active(
+    ExpansionWorker* instance,
+    const ExpansionFrame* rx_frame) {
+    bool success = false;
+
+    do {
+        if(rx_frame->header.type == ExpansionFrameTypeData) {
+            if(!expansion_worker_send_status_response(instance, ExpansionFrameErrorNone)) break;
+
+            const size_t size_consumed = rpc_session_feed(
+                instance->rpc_session,
+                rx_frame->content.data.bytes,
+                rx_frame->content.data.size,
+                EXPANSION_PROTOCOL_TIMEOUT_MS);
+            if(size_consumed != rx_frame->content.data.size) break;
+
+        } else if(rx_frame->header.type == ExpansionFrameTypeControl) {
+            if(rx_frame->content.control.command != ExpansionFrameControlCommandStopRpc) break;
+            instance->state = ExpansionWorkerStateConnected;
+            expansion_worker_rpc_session_close(instance);
+            if(!expansion_worker_send_status_response(instance, ExpansionFrameErrorNone)) break;
+
+        } else if(rx_frame->header.type == ExpansionFrameTypeStatus) {
+            if(rx_frame->content.status.error != ExpansionFrameErrorNone) break;
+            furi_semaphore_release(instance->tx_semaphore);
+
+        } else if(rx_frame->header.type == ExpansionFrameTypeHeartbeat) {
+            if(!expansion_worker_send_heartbeat(instance)) break;
+
+        } else {
+            break;
+        }
+        success = true;
+    } while(false);
+
+    return success;
+}
+
+typedef bool (*ExpansionWorkerStateHandler)(ExpansionWorker*, const ExpansionFrame*);
+
+static const ExpansionWorkerStateHandler expansion_handlers[] = {
+    [ExpansionWorkerStateHandShake] = expansion_worker_handle_state_handshake,
+    [ExpansionWorkerStateConnected] = expansion_worker_handle_state_connected,
+    [ExpansionWorkerStateRpcActive] = expansion_worker_handle_state_rpc_active,
+};
+
+static inline void expansion_worker_state_machine(ExpansionWorker* instance) {
+    ExpansionFrame rx_frame;
+
+    while(true) {
+        if(!expansion_worker_receive_frame(instance, &rx_frame)) break;
+        if(!expansion_handlers[instance->state](instance, &rx_frame)) break;
+    }
+}
+
+static int32_t expansion_worker(void* context) {
+    furi_assert(context);
+    ExpansionWorker* instance = context;
+
+    furi_hal_power_insomnia_enter();
+
+    instance->serial_handle = furi_hal_serial_control_acquire(instance->serial_id);
+    furi_check(instance->serial_handle);
+
+    FURI_LOG_D(TAG, "Worker started");
+
+    instance->state = ExpansionWorkerStateHandShake;
+    instance->exit_reason = ExpansionWorkerExitReasonUnknown;
+
+    furi_hal_serial_init(instance->serial_handle, EXPANSION_PROTOCOL_DEFAULT_BAUD_RATE);
+
+    furi_hal_serial_async_rx_start(
+        instance->serial_handle, expansion_worker_serial_rx_callback, instance, true);
+
+    if(expansion_worker_send_heartbeat(instance)) {
+        expansion_worker_state_machine(instance);
+    }
+
+    if(instance->state == ExpansionWorkerStateRpcActive) {
+        expansion_worker_rpc_session_close(instance);
+    }
+
+    FURI_LOG_D(TAG, "Worker stopped");
+
+    furi_hal_serial_control_release(instance->serial_handle);
+    furi_hal_power_insomnia_exit();
+
+    // Do not invoke worker callback on user-requested exit
+    if((instance->exit_reason != ExpansionWorkerExitReasonUser) && (instance->callback != NULL)) {
+        instance->callback(instance->cb_context);
+    }
+
+    return 0;
+}
+
+ExpansionWorker* expansion_worker_alloc(FuriHalSerialId serial_id) {
+    ExpansionWorker* instance = malloc(sizeof(ExpansionWorker));
+
+    instance->thread = furi_thread_alloc_ex(
+        TAG "Worker", EXPANSION_WORKER_STACK_SZIE, expansion_worker, instance);
+    instance->rx_buf = furi_stream_buffer_alloc(EXPANSION_WORKER_BUFFER_SIZE, 1);
+    instance->serial_id = serial_id;
+
+    // Improves responsiveness in heavy games at the expense of dropped frames
+    furi_thread_set_priority(instance->thread, FuriThreadPriorityLow);
+
+    return instance;
+}
+
+void expansion_worker_free(ExpansionWorker* instance) {
+    furi_stream_buffer_free(instance->rx_buf);
+    furi_thread_join(instance->thread);
+    furi_thread_free(instance->thread);
+    free(instance);
+}
+
+void expansion_worker_set_callback(
+    ExpansionWorker* instance,
+    ExpansionWorkerCallback callback,
+    void* context) {
+    instance->callback = callback;
+    instance->cb_context = context;
+}
+
+void expansion_worker_start(ExpansionWorker* instance) {
+    furi_thread_start(instance->thread);
+}
+
+void expansion_worker_stop(ExpansionWorker* instance) {
+    furi_thread_flags_set(furi_thread_get_id(instance->thread), ExpansionWorkerFlagStop);
+    furi_thread_join(instance->thread);
+}

applications/services/expansion/expansion_worker.h

@@ -0,0 +1,78 @@
+/**
+ * @file expansion_worker.h
+ * @brief Expansion module handling thread wrapper.
+ *
+ * The worker is started each time an expansion module is detected
+ * and handles all of the communication protocols. Likewise, it is stopped
+ * upon module disconnection or communication error.
+ *
+ * @warning This file is a private implementation detail. Please do not attempt to use it in applications.
+ */
+#pragma once
+
+#include <furi_hal_serial_types.h>
+
+/**
+ * @brief Expansion worker opaque type declaration.
+ */
+typedef struct ExpansionWorker ExpansionWorker;
+
+/**
+ * @brief Worker callback type.
+ *
+ * @see expansion_worker_set_callback()
+ *
+ * @param[in,out] context pointer to a user-defined object.
+ */
+typedef void (*ExpansionWorkerCallback)(void* context);
+
+/**
+ * @brief Create an expansion worker instance.
+ *
+ * @param[in] serial_id numerical identifier of the serial to be used by the worker.
+ * @returns pointer to the created instance.
+ */
+ExpansionWorker* expansion_worker_alloc(FuriHalSerialId serial_id);
+
+/**
+ * @brief Delete an expansion worker instance.
+ *
+ * @param[in,out] instance pointer to the instance to be deleted.
+ */
+void expansion_worker_free(ExpansionWorker* instance);
+
+/**
+ * @brief Set the module disconnect callback.
+ *
+ * The callback will be triggered upon worker stop EXCEPT
+ * when it was stopped via an expansion_worker_stop() call.
+ *
+ * In other words, the callback will ONLY be triggered if the worker was
+ * stopped due to the user disconnecting/resetting/powering down the module,
+ * or due to some communication error.
+ *
+ * @param[in,out] instance pointer to the worker instance to be modified.
+ * @param[in] callback pointer to the callback function to be called under the above conditions.
+ * @param[in] context pointer to a user-defined object, will be passed as a parameter to the callback.
+ */
+void expansion_worker_set_callback(
+    ExpansionWorker* instance,
+    ExpansionWorkerCallback callback,
+    void* context);
+
+/**
+ * @brief Start the expansion module worker.
+ *
+ * @param[in,out] instance pointer to the worker instance to be started.
+ */
+void expansion_worker_start(ExpansionWorker* instance);
+
+/**
+ * @brief Stop the expansion module worker.
+ *
+ * If the worker was stopped via this call (and not because of module disconnect/
+ * protocol error), the callback will not be triggered.
+ *
+ * @param[in,out] instance pointer to the worker instance to be stopped.
+ */
+void expansion_worker_stop(ExpansionWorker* instance);

applications/services/rpc/rpc.c

@@ -189,6 +189,12 @@ bool rpc_pb_stream_read(pb_istream_t* istream, pb_byte_t* buf, size_t count) {
     furi_assert(session);
     furi_assert(istream->bytes_left);
 
+    /* TODO FL-3768 this function may be called after
+       marking the worker for termination */
+    if(session->terminate) {
+        return false;
+    }
+
     uint32_t flags = 0;
     size_t bytes_received = 0;
 

applications/services/rpc/rpc_gui.c

@@ -54,6 +54,7 @@ typedef enum {
 typedef struct {
     RpcSession* session;
     Gui* gui;
+    const Icon* icon;
 
     // Receive part
     ViewPort* virtual_display_view_port;
@@ -380,10 +381,19 @@ static void rpc_system_gui_virtual_display_frame_process(const PB_Main* request,
     (void)session;
 }
 
+static const Icon* rpc_system_gui_get_owner_icon(RpcOwner owner) {
+    switch(owner) {
+    case RpcOwnerUart:
+        return &I_Exp_module_connected_12x8;
+    default:
+        return &I_Rpc_active_7x8;
+    }
+}
+
 static void rpc_active_session_icon_draw_callback(Canvas* canvas, void* context) {
-    UNUSED(context);
     furi_assert(canvas);
-    canvas_draw_icon(canvas, 0, 0, &I_Rpc_active_7x8);
+    RpcGuiSystem* rpc_gui = context;
+    canvas_draw_icon(canvas, 0, 0, rpc_gui->icon);
 }
 
 void* rpc_system_gui_alloc(RpcSession* session) {
@@ -394,16 +404,16 @@ void* rpc_system_gui_alloc(RpcSession* session) {
     rpc_gui->session = session;
 
     // Active session icon
+    const RpcOwner owner = rpc_session_get_owner(rpc_gui->session);
+    if(owner != RpcOwnerBle) {
+        rpc_gui->icon = rpc_system_gui_get_owner_icon(owner);
         rpc_gui->rpc_session_active_viewport = view_port_alloc();
-    view_port_set_width(rpc_gui->rpc_session_active_viewport, icon_get_width(&I_Rpc_active_7x8));
+        view_port_set_width(rpc_gui->rpc_session_active_viewport, icon_get_width(rpc_gui->icon));
         view_port_draw_callback_set(
-        rpc_gui->rpc_session_active_viewport, rpc_active_session_icon_draw_callback, session);
+            rpc_gui->rpc_session_active_viewport, rpc_active_session_icon_draw_callback, rpc_gui);
-    if(rpc_session_get_owner(rpc_gui->session) != RpcOwnerBle) {
+        gui_add_view_port(
-        view_port_enabled_set(rpc_gui->rpc_session_active_viewport, true);
+            rpc_gui->gui, rpc_gui->rpc_session_active_viewport, GuiLayerStatusBarLeft);
-    } else {
-        view_port_enabled_set(rpc_gui->rpc_session_active_viewport, false);
     }
-    gui_add_view_port(rpc_gui->gui, rpc_gui->rpc_session_active_viewport, GuiLayerStatusBarLeft);
 
     RpcHandler rpc_handler = {
         .message_handler = NULL,
@@ -445,8 +455,10 @@ void rpc_system_gui_free(void* context) {
         rpc_gui->virtual_display_not_empty = false;
     }
 
+    if(rpc_gui->rpc_session_active_viewport) {
         gui_remove_view_port(rpc_gui->gui, rpc_gui->rpc_session_active_viewport);
         view_port_free(rpc_gui->rpc_session_active_viewport);
+    }
 
     if(rpc_gui->is_streaming) {
         rpc_gui->is_streaming = false;

targets/f18/api_symbols.csv

@@ -1,5 +1,5 @@
 entry,status,name,type,params
-Version,+,55.0,,
+Version,+,55.1,,
 Header,+,applications/services/bt/bt_service/bt.h,,
 Header,+,applications/services/cli/cli.h,,
 Header,+,applications/services/cli/cli_vcp.h,,
@@ -1276,6 +1276,7 @@ Function,+,furi_hal_sd_presence_init,void,
 Function,+,furi_hal_sd_read_blocks,FuriStatus,"uint32_t*, uint32_t, uint32_t"
 Function,+,furi_hal_sd_write_blocks,FuriStatus,"const uint32_t*, uint32_t, uint32_t"
 Function,+,furi_hal_serial_async_rx,uint8_t,FuriHalSerialHandle*
+Function,+,furi_hal_serial_async_rx_available,_Bool,FuriHalSerialHandle*
 Function,+,furi_hal_serial_async_rx_start,void,"FuriHalSerialHandle*, FuriHalSerialAsyncRxCallback, void*, _Bool"
 Function,+,furi_hal_serial_async_rx_stop,void,FuriHalSerialHandle*
 Function,+,furi_hal_serial_control_acquire,FuriHalSerialHandle*,FuriHalSerialId

targets/f7/api_symbols.csv

@@ -1,5 +1,5 @@
 entry,status,name,type,params
-Version,+,55.0,,
+Version,+,55.1,,
 Header,+,applications/drivers/subghz/cc1101_ext/cc1101_ext_interconnect.h,,
 Header,+,applications/services/bt/bt_service/bt.h,,
 Header,+,applications/services/cli/cli.h,,
@@ -1442,6 +1442,7 @@ Function,+,furi_hal_sd_presence_init,void,
 Function,+,furi_hal_sd_read_blocks,FuriStatus,"uint32_t*, uint32_t, uint32_t"
 Function,+,furi_hal_sd_write_blocks,FuriStatus,"const uint32_t*, uint32_t, uint32_t"
 Function,+,furi_hal_serial_async_rx,uint8_t,FuriHalSerialHandle*
+Function,+,furi_hal_serial_async_rx_available,_Bool,FuriHalSerialHandle*
 Function,+,furi_hal_serial_async_rx_start,void,"FuriHalSerialHandle*, FuriHalSerialAsyncRxCallback, void*, _Bool"
 Function,+,furi_hal_serial_async_rx_stop,void,FuriHalSerialHandle*
 Function,+,furi_hal_serial_control_acquire,FuriHalSerialHandle*,FuriHalSerialId

targets/f7/furi_hal/furi_hal_serial.c

@@ -730,6 +730,13 @@ static void furi_hal_serial_async_rx_configure(
     FuriHalSerialHandle* handle,
     FuriHalSerialAsyncRxCallback callback,
     void* context) {
+    // Handle must be configured before enabling RX interrupt
+    // as it might be triggered right away on a misconfigured handle
+    furi_hal_serial[handle->id].rx_byte_callback = callback;
+    furi_hal_serial[handle->id].handle = handle;
+    furi_hal_serial[handle->id].rx_dma_callback = NULL;
+    furi_hal_serial[handle->id].context = context;
+
     if(handle->id == FuriHalSerialIdUsart) {
         if(callback) {
             furi_hal_serial_usart_deinit_dma_rx();
@@ -753,10 +760,6 @@ static void furi_hal_serial_async_rx_configure(
             LL_LPUART_DisableIT_RXNE_RXFNE(LPUART1);
         }
     }
-    furi_hal_serial[handle->id].rx_byte_callback = callback;
-    furi_hal_serial[handle->id].handle = handle;
-    furi_hal_serial[handle->id].rx_dma_callback = NULL;
-    furi_hal_serial[handle->id].context = context;
 }
 
 void furi_hal_serial_async_rx_start(
@@ -782,6 +785,17 @@ void furi_hal_serial_async_rx_stop(FuriHalSerialHandle* handle) {
     furi_hal_serial_async_rx_configure(handle, NULL, NULL);
 }
 
+bool furi_hal_serial_async_rx_available(FuriHalSerialHandle* handle) {
+    furi_check(FURI_IS_IRQ_MODE());
+    furi_assert(handle->id < FuriHalSerialIdMax);
+
+    if(handle->id == FuriHalSerialIdUsart) {
+        return LL_USART_IsActiveFlag_RXNE_RXFNE(USART1);
+    } else {
+        return LL_LPUART_IsActiveFlag_RXNE_RXFNE(LPUART1);
+    }
+}
+
 uint8_t furi_hal_serial_async_rx(FuriHalSerialHandle* handle) {
     furi_check(FURI_IS_IRQ_MODE());
     furi_assert(handle->id < FuriHalSerialIdMax);

targets/f7/furi_hal/furi_hal_serial.h

@@ -130,6 +130,16 @@ void furi_hal_serial_async_rx_start(
  */
 void furi_hal_serial_async_rx_stop(FuriHalSerialHandle* handle);
 
+/** Check if there is data available for reading
+ *
+ * @warning    This function must be called only from the callback
+ *             FuriHalSerialAsyncRxCallback
+ *
+ * @param      handle    Serial handle
+ * @return     true if data is available for reading, false otherwise
+ */
+bool furi_hal_serial_async_rx_available(FuriHalSerialHandle* handle);
+
 /** Get data Serial receive
  *
  * @warning    This function must be called only from the callback

targets/f7/furi_hal/furi_hal_serial_control.c

@@ -47,6 +47,7 @@ typedef struct {
     FuriHalSerialHandle* log_serial;
 
     // Expansion detection
+    FuriHalSerialHandle* expansion_serial;
     FuriHalSerialControlExpansionCallback expansion_cb;
     void* expansion_ctx;
 } FuriHalSerialControl;
@@ -58,7 +59,36 @@ static void furi_hal_serial_control_log_callback(const uint8_t* data, size_t siz
     furi_hal_serial_tx(handle, data, size);
 }
 
+static void furi_hal_serial_control_expansion_irq_callback(void* context) {
+    UNUSED(context);
+
+    FuriHalSerialControlMessage message;
+    message.type = FuriHalSerialControlMessageTypeExpansionIrq;
+    message.api_lock = NULL;
+    furi_message_queue_put(furi_hal_serial_control->queue, &message, 0);
+}
+
+static void
+    furi_hal_serial_control_enable_expansion_irq(FuriHalSerialHandle* handle, bool enable) {
+    const GpioPin* gpio = furi_hal_serial_get_gpio_pin(handle, FuriHalSerialDirectionRx);
+
+    if(enable) {
+        furi_hal_serial_disable_direction(handle, FuriHalSerialDirectionRx);
+        furi_hal_gpio_add_int_callback(gpio, furi_hal_serial_control_expansion_irq_callback, NULL);
+        furi_hal_gpio_init(gpio, GpioModeInterruptFall, GpioPullUp, GpioSpeedLow);
+    } else {
+        furi_hal_gpio_remove_int_callback(gpio);
+        furi_hal_serial_enable_direction(handle, FuriHalSerialDirectionRx);
+    }
+}
+
 static void furi_hal_serial_control_log_set_handle(FuriHalSerialHandle* handle) {
+    // Disable expansion module detection before reconfiguring UARTs
+    if(furi_hal_serial_control->expansion_serial) {
+        furi_hal_serial_control_enable_expansion_irq(
+            furi_hal_serial_control->expansion_serial, false);
+    }
+
     if(furi_hal_serial_control->log_serial) {
         furi_log_remove_handler(furi_hal_serial_control->log_handler);
         furi_hal_serial_deinit(furi_hal_serial_control->log_serial);
@@ -74,15 +104,12 @@ static void furi_hal_serial_control_log_set_handle(FuriHalSerialHandle* handle)
         furi_hal_serial_control->log_handler.context = furi_hal_serial_control->log_serial;
         furi_log_add_handler(furi_hal_serial_control->log_handler);
     }
-}
 
-static void furi_hal_serial_control_expansion_irq_callback(void* context) {
+    // Re-enable expansion module detection (if applicable)
-    UNUSED(context);
+    if(furi_hal_serial_control->expansion_serial) {
-
+        furi_hal_serial_control_enable_expansion_irq(
-    FuriHalSerialControlMessage message;
+            furi_hal_serial_control->expansion_serial, true);
-    message.type = FuriHalSerialControlMessageTypeExpansionIrq;
+    }
-    message.api_lock = NULL;
-    furi_message_queue_put(furi_hal_serial_control->queue, &message, 0);
 }
 
 static bool furi_hal_serial_control_handler_stop(void* input, void* output) {
@@ -93,16 +120,21 @@ static bool furi_hal_serial_control_handler_stop(void* input, void* output) {
 
 static bool furi_hal_serial_control_handler_acquire(void* input, void* output) {
     FuriHalSerialId serial_id = *(FuriHalSerialId*)input;
-    if(furi_hal_serial_control->handles[serial_id].in_use) {
+    FuriHalSerialHandle* handle = &furi_hal_serial_control->handles[serial_id];
+
+    if(handle->in_use) {
         *(FuriHalSerialHandle**)output = NULL;
     } else {
         // Logging
         if(furi_hal_serial_control->log_config_serial_id == serial_id) {
             furi_hal_serial_control_log_set_handle(NULL);
+            // Expansion
+        } else if(furi_hal_serial_control->expansion_serial == handle) {
+            furi_hal_serial_control_enable_expansion_irq(handle, false);
         }
         // Return handle
-        furi_hal_serial_control->handles[serial_id].in_use = true;
+        handle->in_use = true;
-        *(FuriHalSerialHandle**)output = &furi_hal_serial_control->handles[serial_id];
+        *(FuriHalSerialHandle**)output = handle;
     }
 
     return true;
@@ -116,9 +148,12 @@ static bool furi_hal_serial_control_handler_release(void* input, void* output) {
     furi_hal_serial_deinit(handle);
     handle->in_use = false;
 
-    // Return back logging
     if(furi_hal_serial_control->log_config_serial_id == handle->id) {
+        // Return back logging
         furi_hal_serial_control_log_set_handle(handle);
+    } else if(furi_hal_serial_control->expansion_serial == handle) {
+        // Re-enable expansion
+        furi_hal_serial_control_enable_expansion_irq(handle, true);
     }
 
     return true;
@@ -157,24 +192,24 @@ static bool furi_hal_serial_control_handler_expansion_set_callback(void* input,
 
     FuriHalSerialControlMessageExpCallback* message_input = input;
     FuriHalSerialHandle* handle = &furi_hal_serial_control->handles[message_input->id];
-    const GpioPin* gpio = furi_hal_serial_get_gpio_pin(handle, FuriHalSerialDirectionRx);
 
-    if(message_input->callback) {
+    const bool enable_irq = message_input->callback != NULL;
+
+    if(enable_irq) {
+        furi_check(furi_hal_serial_control->expansion_serial == NULL);
         furi_check(furi_hal_serial_control->expansion_cb == NULL);
-
+        furi_hal_serial_control->expansion_serial = handle;
-        furi_hal_serial_disable_direction(handle, FuriHalSerialDirectionRx);
-        furi_hal_gpio_add_int_callback(gpio, furi_hal_serial_control_expansion_irq_callback, NULL);
-        furi_hal_gpio_init(gpio, GpioModeInterruptFall, GpioPullUp, GpioSpeedLow);
     } else {
+        furi_check(furi_hal_serial_control->expansion_serial == handle);
         furi_check(furi_hal_serial_control->expansion_cb != NULL);
-
+        furi_hal_serial_control->expansion_serial = NULL;
-        furi_hal_gpio_remove_int_callback(gpio);
-        furi_hal_serial_enable_direction(handle, FuriHalSerialDirectionRx);
     }
 
     furi_hal_serial_control->expansion_cb = message_input->callback;
     furi_hal_serial_control->expansion_ctx = message_input->context;
 
+    furi_hal_serial_control_enable_expansion_irq(handle, enable_irq);
+
     return true;
 }