LoraApi callbacks refactored to events (#632)

Documentation/ideas.md

@@ -11,14 +11,13 @@
 
 ## Higher Priority
 
+- Add tests for app stdin/stdout
+- AppHubApp: Prevent download callbacks from accessing a destroyed view. Create a "download task" concept that emits events.
 - CrashDiagnostics shouldn't show a QR when there's no callstack
-- Apps should be able to specify stack size in their manifest, per architecture.
-  Use thread_get_stack_space() to find out the unused bytes
 - Apps currently have a `Context` object with an `appInstanceId` in it, purely for being able to close the app.
   Change it so that the app has its own termination signal that it waits for in the loop, it should subscribe to the event group.
 - stopAppFromToolbar() in Tactility.cpp stops the top-most app. Change it so the toolbar knows for which app id it is created, so it can rely on that.
 - Warn if file operations are done from prohibited tasks (e.g. lvgl task)
-- AppHubApp: Prevent download callbacks from accessing a destroyed view.
 - Move USB host task stacks to SPIRAM when available: esp32_usbhost*.cpp
 - Get rid of WiFi service (Wifi.cpp/h) in Tactility.cpp
 - Make it more clear to end-users that an SD card is required to run Tactility
@@ -51,12 +50,12 @@
 
 ## Medium Priority
 
+- esp_lvgl_port settings has a large stack size (~9kB) to fix stackoverflow when LVGL events (e.g. button click) do actions like file operations do actions like file operations. Can we reduce the callstack?
 - `struct Driver` has an `.owner`, but it's not always set. Either validate on Module construct that it matches, or otherwise set it during module start. The problem: NULL parent currently means that driver is not removable. This clashes with setting it dynamically. Consider some kind of flag to determine removability.
 - Consider moving certain drivers into separate modules: audio, bt, wifi, etc
 - Consider using https://github.com/Graphify-Labs/graphify
 - Consider implementing LVGL gridnav in apps https://lvgl.io/docs/open/9.3/details/auxiliary-modules/gridnav.html
 - Make USB host driver disabled by default, so it doesn't consume memory
-- Diceware app has large "+" and "-' buttons on Cardputer. It should be smaller.
 - TactilityTool: Make API compatibility table (and check for compatibility in the tool itself)
 - Improve EspLcdDisplay to contain all the standard configuration options, and implement a default init function. Add a configuration class.
 - Unify the way displays are dimmed. Some implementations turn off the display when it's fully dimmed. Make this a separate functionality.
@@ -64,6 +63,7 @@
 
 ## Lower Priority
 
+- Diceware app has large "+" and "-' buttons on Cardputer. It should be smaller.
 - lvgl-module has a keyboard.cpp that creates a `keyboard_group`. This group is set as the default group, so it can also work with trackball(= LVGL "encoder").
   Make a separate group that is the default group. The keyboard can then use it (or use its own).
   The basic idea is to invert the ownership: now the keyboard group is made the default group, but it's probably more logical to have the default group used by the keyboard.
@@ -74,7 +74,6 @@
 - CrashHandler: use "corrupted" flag
 - CrashHandler: process other types of crashes (WDT?)
 - Use GPS time to set/update the current time
-- Fix bug in T-Deck/etc: esp_lvgl_port settings has a large stack size (~9kB) to fix an issue where the T-Deck would get a stackoverflow. This sometimes happens when WiFi is auto-enabled and you open the app while it is still connecting.
 - Consider using non_null (either via MS GSL, or custom)
 - Fix system time to not be 1980 (use build year as a minimum). Consider keeping track of the last known time.
 - Use std::span or string_view in StringUtils https://youtu.be/FRkJCvHWdwQ?t=2754 
@@ -96,8 +95,6 @@
 - Audio recording app
 - OTA updates
 - If present, use LED to show boot/wifi status
-- On crash, try to save the current log to flash or SD card? (this is risky, though, so ask in Discord first)
-- Support more than 1 hardware keyboard (see lvgl::hardware_keyboard_set_indev()). LVGL init currently calls keyboard init, but that part should probably be done from the KeyboardDevice base class.
 
 # App Ideas
 

Drivers/sx126x-module/private/drivers/sx1262_radio.cpp

@@ -8,7 +8,6 @@
 #include <tactility/drivers/gpio_controller.h>
 #include <tactility/log.h>
 
-#include <algorithm>
 #include <initializer_list>
 
 #include <RadioLib.h>
@@ -345,12 +344,9 @@ void Sx1262Radio::setState(enum LoraRadioState newState) {
     }
     LOG_I(TAG, "State: %s -> %s", toString(state), toString(newState));
     state = newState;
-    auto callbacks = stateCallbacks;
     unlock();
 
-    for (const auto& entry : callbacks) {
-        entry.callback(settings.device, entry.context, newState);
-    }
+    lora_state_event_emit(settings.device, newState);
 }
 
 error_t Sx1262Radio::setModulation(enum LoraModulation newModulation) {
@@ -377,76 +373,12 @@ enum LoraModulation Sx1262Radio::getModulation() const {
     return result;
 }
 
-// Callbacks are invoked on a snapshot of the list, with the radio mutex released:
-// consumers take their own locks in callbacks and also call into this API while
-// holding those locks, so invoking under the radio mutex would set up an AB-BA
-// deadlock between the radio thread and any consumer thread.
-void Sx1262Radio::publishRx(const struct LoraRxPacket& packet) {
-    lock();
-    auto callbacks = rxCallbacks;
-    unlock();
-
-    for (const auto& entry : callbacks) {
-        entry.callback(settings.device, entry.context, &packet);
-    }
+error_t Sx1262Radio::publishRx(const uint8_t* data, size_t length, float rssi, float snr) {
+    return lora_rx_event_emit(settings.device, data, length, rssi, snr);
 }
 
-void Sx1262Radio::publishTx(LoraTxId id, enum LoraTransmissionState txState) {
-    lock();
-    auto callbacks = txCallbacks;
-    unlock();
-
-    for (const auto& entry : callbacks) {
-        entry.callback(settings.device, entry.context, id, txState);
-    }
-}
-
-error_t Sx1262Radio::addRxCallback(void* context, LoraRxCallback callback) {
-    lock();
-    rxCallbacks.push_back({context, callback});
-    unlock();
-    return ERROR_NONE;
-}
-
-error_t Sx1262Radio::removeRxCallback(LoraRxCallback callback) {
-    lock();
-    const auto old_size = rxCallbacks.size();
-    std::erase_if(rxCallbacks, [callback](const auto& entry) { return entry.callback == callback; });
-    const auto result = (rxCallbacks.size() == old_size) ? ERROR_NOT_FOUND : ERROR_NONE;
-    unlock();
-    return result;
-}
-
-error_t Sx1262Radio::addStateCallback(void* context, LoraStateCallback callback) {
-    lock();
-    stateCallbacks.push_back({context, callback});
-    unlock();
-    return ERROR_NONE;
-}
-
-error_t Sx1262Radio::removeStateCallback(LoraStateCallback callback) {
-    lock();
-    const auto old_size = stateCallbacks.size();
-    std::erase_if(stateCallbacks, [callback](const auto& entry) { return entry.callback == callback; });
-    const auto result = (stateCallbacks.size() == old_size) ? ERROR_NOT_FOUND : ERROR_NONE;
-    unlock();
-    return result;
-}
-
-error_t Sx1262Radio::addTxCallback(void* context, LoraTxCallback callback) {
-    lock();
-    txCallbacks.push_back({context, callback});
-    unlock();
-    return ERROR_NONE;
-}
-
-error_t Sx1262Radio::removeTxCallback(LoraTxCallback callback) {
-    lock();
-    const auto old_size = txCallbacks.size();
-    std::erase_if(txCallbacks, [callback](const auto& entry) { return entry.callback == callback; });
-    const auto result = (txCallbacks.size() == old_size) ? ERROR_NOT_FOUND : ERROR_NONE;
-    unlock();
-    return result;
+error_t Sx1262Radio::publishTx(LoraTxId id, enum LoraTransmissionState txState) {
+    return lora_tx_event_emit(settings.device, id, txState);
 }
 
 // endregion
@@ -952,15 +884,10 @@ void Sx1262Radio::doReceive() {
     } else if (rxSize == 0) {
         // Empty read: skip silently to avoid log flooding on spurious IRQs.
     } else {
-        const struct LoraRxPacket packet = {
-            .data = data.data(),
-            .length = data.size(),
-            .rssi = radio.getRSSI(),
-            .snr = radio.getSNR(),
-        };
-
-        LOG_I(TAG, "RX: %u bytes, RSSI %.1f dBm, SNR %.1f dB", (unsigned)packet.length, packet.rssi, packet.snr);
-        publishRx(packet);
+        const float rssi = radio.getRSSI();
+        const float snr = radio.getSNR();
+        LOG_I(TAG, "RX: %u bytes, RSSI %.1f dBm, SNR %.1f dB", (unsigned)data.size(), rssi, snr);
+        publishRx(data.data(), data.size(), rssi, snr);
         radio.finishReceive();
     }
 }

Drivers/sx126x-module/private/drivers/sx1262_radio.h

@@ -19,10 +19,9 @@ struct Device;
 struct GpioDescriptor;
 
 /**
- * SX1262 radio engine: owns the radio thread, the TX queue and the callback lists.
- * The public methods are thread-safe. Callbacks are invoked on a snapshot of the list with
- * the internal mutex released, either from the radio thread (RX, TX progress, state) or from
- * the caller of transmit() (QUEUED).
+ * SX1262 radio engine: owns the radio thread and the TX queue. The public methods are
+ * thread-safe. State/RX/TX events are published via lora_event_emit(), either from the radio
+ * thread (RX, TX progress, state) or from the caller of transmit() (QUEUED).
  *
  * The RadioLib types live behind the RadioParts indirection: RadioLib declares a global
  * `class Module` that collides with the kernel's `struct Module` when both are visible
@@ -59,12 +58,6 @@ private:
         std::vector<uint8_t> data;
     };
 
-    template<typename Callback>
-    struct CallbackEntry {
-        void* context;
-        Callback callback;
-    };
-
     const Settings settings;
     RadioParts* parts;
     mutable RecursiveMutex mutex = {};
@@ -80,10 +73,6 @@ private:
     TxItem currentTx;
     LoraTxId lastTxId = 0;
 
-    std::vector<CallbackEntry<LoraStateCallback>> stateCallbacks;
-    std::vector<CallbackEntry<LoraRxCallback>> rxCallbacks;
-    std::vector<CallbackEntry<LoraTxCallback>> txCallbacks;
-
     // Parameter store, applied on the next doBegin(). Frequencies/rates are held in base SI
     // units (Hz, bit/s) and converted to RadioLib's MHz/kHz/kbps floats in doBegin().
     int8_t power = -9;
@@ -111,8 +100,8 @@ private:
     int32_t threadMain();
 
     void setState(enum LoraRadioState newState);
-    void publishRx(const struct LoraRxPacket& packet);
-    void publishTx(LoraTxId id, enum LoraTransmissionState txState);
+    error_t publishRx(const uint8_t* data, size_t length, float rssi, float snr);
+    error_t publishTx(LoraTxId id, enum LoraTransmissionState txState);
 
     size_t getTxQueueSize() const;
     TxItem popNextQueuedTx();
@@ -165,11 +154,4 @@ public:
     error_t getParameter(enum LoraParameter parameter, int32_t* value) const;
 
     error_t transmit(const uint8_t* data, size_t length, LoraTxId* id);
-
-    error_t addRxCallback(void* context, LoraRxCallback callback);
-    error_t removeRxCallback(LoraRxCallback callback);
-    error_t addStateCallback(void* context, LoraStateCallback callback);
-    error_t removeStateCallback(LoraStateCallback callback);
-    error_t addTxCallback(void* context, LoraTxCallback callback);
-    error_t removeTxCallback(LoraTxCallback callback);
 };

Drivers/sx126x-module/source/sx1262.cpp

@@ -312,42 +312,6 @@ static error_t api_transmit(Device* device, const uint8_t* data, size_t length,
     return radio->transmit(data, length, id);
 }
 
-static error_t api_add_rx_callback(Device* device, void* callback_context, LoraRxCallback callback) {
-    auto* radio = get_radio(device);
-    if (radio == nullptr) return ERROR_INVALID_STATE;
-    return radio->addRxCallback(callback_context, callback);
-}
-
-static error_t api_remove_rx_callback(Device* device, LoraRxCallback callback) {
-    auto* radio = get_radio(device);
-    if (radio == nullptr) return ERROR_INVALID_STATE;
-    return radio->removeRxCallback(callback);
-}
-
-static error_t api_add_state_callback(Device* device, void* callback_context, LoraStateCallback callback) {
-    auto* radio = get_radio(device);
-    if (radio == nullptr) return ERROR_INVALID_STATE;
-    return radio->addStateCallback(callback_context, callback);
-}
-
-static error_t api_remove_state_callback(Device* device, LoraStateCallback callback) {
-    auto* radio = get_radio(device);
-    if (radio == nullptr) return ERROR_INVALID_STATE;
-    return radio->removeStateCallback(callback);
-}
-
-static error_t api_add_tx_callback(Device* device, void* callback_context, LoraTxCallback callback) {
-    auto* radio = get_radio(device);
-    if (radio == nullptr) return ERROR_INVALID_STATE;
-    return radio->addTxCallback(callback_context, callback);
-}
-
-static error_t api_remove_tx_callback(Device* device, LoraTxCallback callback) {
-    auto* radio = get_radio(device);
-    if (radio == nullptr) return ERROR_INVALID_STATE;
-    return radio->removeTxCallback(callback);
-}
-
 static const struct LoraApi sx1262_lora_api = {
     .get_radio_state = api_get_radio_state,
     .set_enabled = api_set_enabled,
@@ -358,12 +322,6 @@ static const struct LoraApi sx1262_lora_api = {
     .set_parameter = api_set_parameter,
     .get_parameter = api_get_parameter,
     .transmit = api_transmit,
-    .add_rx_callback = api_add_rx_callback,
-    .remove_rx_callback = api_remove_rx_callback,
-    .add_state_callback = api_add_state_callback,
-    .remove_state_callback = api_remove_state_callback,
-    .add_tx_callback = api_add_tx_callback,
-    .remove_tx_callback = api_remove_tx_callback,
 };
 
 // endregion

TactilityKernel/include/tactility/drivers/lora.h

@@ -5,6 +5,7 @@
 #include <stddef.h>
 #include <stdint.h>
 
+#include <tactility/concurrent/task_event_group.h>
 #include <tactility/error.h>
 
 #ifdef __cplusplus
@@ -93,27 +94,8 @@ enum LoraTransmissionState {
     LORA_TRANSMISSION_STATE_ERROR,
 };
 
-struct LoraRxPacket {
-    /** Packet payload. Only valid for the duration of the RX callback. */
-    const uint8_t* data;
-    size_t length;
-    /** Received signal strength in dBm */
-    float rssi;
-    /** Signal-to-noise ratio in dB */
-    float snr;
-};
-
-/**
- * Callbacks are invoked without any driver lock held, either from the driver's radio
- * thread (RX, TX progress, state) or from the thread calling transmit()/set_enabled()
- * (the QUEUED TX event and enable/disable state changes). Taking consumer locks in a
- * callback is therefore safe, but keep callbacks short: RX processing stalls while
- * they run. After remove_*_callback returns, a callback that was already in flight
- * may still complete once — disable the radio before destroying callback context.
- */
-typedef void (*LoraStateCallback)(struct Device* device, void* context, enum LoraRadioState state);
-typedef void (*LoraRxCallback)(struct Device* device, void* context, const struct LoraRxPacket* packet);
-typedef void (*LoraTxCallback)(struct Device* device, void* context, LoraTxId id, enum LoraTransmissionState state);
+/** Largest RX payload any supported modem can hand back in one packet. */
+#define LORA_RX_MAX_PACKET_LENGTH 255
 
 struct LoraApi {
     /**
@@ -198,63 +180,279 @@ struct LoraApi {
      * @return ERROR_NONE on success
      */
     error_t (*transmit)(struct Device* device, const uint8_t* data, size_t length, LoraTxId* id);
-
-    /**
-     * Add a callback for received packets.
-     * @param[in] device the lora device
-     * @param[in] callback_context the context to pass to the callback
-     * @param[in] callback the callback function
-     * @return ERROR_NONE on success
-     */
-    error_t (*add_rx_callback)(struct Device* device, void* callback_context, LoraRxCallback callback);
-
-    /**
-     * Remove a callback for received packets.
-     * @param[in] device the lora device
-     * @param[in] callback the callback function
-     * @return ERROR_NONE on success
-     */
-    error_t (*remove_rx_callback)(struct Device* device, LoraRxCallback callback);
-
-    /**
-     * Add a callback for radio state changes.
-     * @param[in] device the lora device
-     * @param[in] callback_context the context to pass to the callback
-     * @param[in] callback the callback function
-     * @return ERROR_NONE on success
-     */
-    error_t (*add_state_callback)(struct Device* device, void* callback_context, LoraStateCallback callback);
-
-    /**
-     * Remove a callback for radio state changes.
-     * @param[in] device the lora device
-     * @param[in] callback the callback function
-     * @return ERROR_NONE on success
-     */
-    error_t (*remove_state_callback)(struct Device* device, LoraStateCallback callback);
-
-    /**
-     * Add a callback for transmission progress.
-     * @param[in] device the lora device
-     * @param[in] callback_context the context to pass to the callback
-     * @param[in] callback the callback function
-     * @return ERROR_NONE on success
-     */
-    error_t (*add_tx_callback)(struct Device* device, void* callback_context, LoraTxCallback callback);
-
-    /**
-     * Remove a callback for transmission progress.
-     * @param[in] device the lora device
-     * @param[in] callback the callback function
-     * @return ERROR_NONE on success
-     */
-    error_t (*remove_tx_callback)(struct Device* device, LoraTxCallback callback);
 };
 
-extern const struct DeviceType LORA_TYPE;
+// region State events
 
-/** @return the first registered lora device, regardless of started state, or NULL if none exists */
-struct Device* lora_find_first_registered_device(void);
+struct LoraStateEvent {
+    /** Stamped by lora_state_event_emit(); any value passed in by the caller is ignored. */
+    uint64_t timestamp;
+    enum LoraRadioState state;
+};
+
+/**
+ * Number of events that can be queued per subscription before lora_state_event_emit() starts
+ * returning ERROR_RESOURCE (dropping the newest event, preserving FIFO order of what's
+ * already queued).
+ */
+#define LORA_STATE_EVENT_QUEUE_CAPACITY 4
+
+/**
+ * Caller-owned subscription node, registered with lora_state_event_subscribe() and drained with
+ * lora_state_event_poll(). Events queue by value (FIFO).
+ * @warning Fields other than `bit` are for internal use only; do not read or write them directly.
+ */
+struct LoraStateEventSubscription {
+    /** Set by lora_state_event_subscribe(). Read-only for the caller: OR it into a
+     * task_event_group_wait() mask (alongside other subscriptions sharing the same
+     * `event_group`) to block on this subscription and other event sources with one call. */
+    uint32_t bit;
+
+    struct {
+        /** The lora device this subscription receives events for; set by lora_state_event_subscribe(). */
+        struct Device* device;
+
+        /** Caller-owned, borrowed; set by lora_state_event_subscribe(). */
+        struct TaskEventGroup* event_group;
+
+        struct LoraStateEvent queue[LORA_STATE_EVENT_QUEUE_CAPACITY];
+        uint8_t head;
+        uint8_t count;
+
+        struct LoraStateEventSubscription* next;
+    } internal;
+};
+
+/**
+ * Register a subscription for radio state changes emitted by @a device.
+ * @warning Does not work in ISR context.
+ * @param[in,out] sub subscription to register; owns the storage, must stay alive (and
+ * stationary) until unsubscribed
+ * @param[in] event_group caller-owned group to wait on; must outlive @a sub (i.e. be
+ * destructed only after lora_state_event_unsubscribe()). To block for an event, call
+ * task_event_group_wait()/task_event_group_wait_any() on this group (OR sub->bit into the mask,
+ * or use _wait_any() to include every subscription sharing it), then drain with
+ * lora_state_event_poll().
+ * @param[in] device the lora device this subscription receives events for
+ * @retval ERROR_NONE on success
+ * @retval ERROR_RESOURCE @a event_group has no free bits left to claim; @a sub was not registered
+ * @retval ERROR_INVALID_STATE @a sub is already registered
+ */
+error_t lora_state_event_subscribe(struct LoraStateEventSubscription* sub, struct TaskEventGroup* event_group, struct Device* device);
+
+/**
+ * Remove a previously registered subscription.
+ * @warning Does not work in ISR context.
+ * @return ERROR_NONE on success, ERROR_NOT_FOUND if no matching subscription exists
+ */
+error_t lora_state_event_unsubscribe(struct LoraStateEventSubscription* sub);
+
+/**
+ * Non-blocking: pop the next event for @a sub if one is already queued.
+ * @warning Never blocks. To wait for an event, block in task_event_group_wait()/
+ * task_event_group_wait_any() on @a sub's event group first (see lora_state_event_subscribe()),
+ * then drain with this in a loop.
+ * @retval ERROR_NONE @a out_event was filled
+ * @retval ERROR_TIMEOUT nothing queued right now
+ */
+error_t lora_state_event_poll(struct LoraStateEventSubscription* sub, struct LoraStateEvent* out_event);
+
+/**
+ * Emit a radio state event to every subscriber of @a device. Called by LoraApi driver
+ * implementations; not intended for consumers of the API.
+ * @param[in] device the lora device the event originates from
+ * @param[in] state the new radio state
+ * @return ERROR_NONE on success, ERROR_NOT_FOUND if @a device has no subscribers
+ */
+error_t lora_state_event_emit(struct Device* device, enum LoraRadioState state);
+
+// endregion
+
+// region RX events
+
+struct LoraRxEvent {
+    /** Stamped by lora_rx_event_emit(); any value passed in by the caller is ignored. */
+    uint64_t timestamp;
+    uint8_t data[LORA_RX_MAX_PACKET_LENGTH];
+    size_t length;
+    /** Received signal strength in dBm */
+    float rssi;
+    /** Signal-to-noise ratio in dB */
+    float snr;
+};
+
+/**
+ * Number of events that can be queued per subscription before lora_rx_event_emit() starts
+ * returning ERROR_RESOURCE (dropping the newest event, preserving FIFO order of what's
+ * already queued).
+ */
+#define LORA_RX_EVENT_QUEUE_CAPACITY 8
+
+/**
+ * Caller-owned subscription node, registered with lora_rx_event_subscribe() and drained with
+ * lora_rx_event_poll(). Events queue by value (FIFO): a subscription that falls behind keeps
+ * every queued packet; once the queue is full, newly received packets are dropped instead.
+
+ * @warning Fields other than `bit` are for internal use only; do not read or write them directly.
+ */
+struct LoraRxEventSubscription {
+    /** Set by lora_rx_event_subscribe(). Read-only for the caller: OR it into a
+     * task_event_group_wait() mask (alongside other subscriptions sharing the same
+     * `event_group`) to block on this subscription and other event sources with one call. */
+    uint32_t bit;
+
+    struct {
+        /** The lora device this subscription receives events for; set by lora_rx_event_subscribe(). */
+        struct Device* device;
+
+        /** Caller-owned, borrowed; set by lora_rx_event_subscribe(). */
+        struct TaskEventGroup* event_group;
+
+        struct LoraRxEvent queue[LORA_RX_EVENT_QUEUE_CAPACITY];
+        uint8_t head;
+        uint8_t count;
+
+        struct LoraRxEventSubscription* next;
+    } internal;
+};
+
+/**
+ * Register a subscription for received packets emitted by @a device.
+ * @warning Does not work in ISR context.
+ * @param[in,out] sub subscription to register; owns the storage, must stay alive (and
+ * stationary) until unsubscribed
+ * @param[in] event_group caller-owned group to wait on; must outlive @a sub (i.e. be
+ * destructed only after lora_rx_event_unsubscribe()). To block for an event, call
+ * task_event_group_wait()/task_event_group_wait_any() on this group (OR sub->bit into the mask,
+ * or use _wait_any() to include every subscription sharing it), then drain with
+ * lora_rx_event_poll().
+ * @param[in] device the lora device this subscription receives events for
+ * @retval ERROR_NONE on success
+ * @retval ERROR_RESOURCE @a event_group has no free bits left to claim; @a sub was not registered
+ * @retval ERROR_INVALID_STATE @a sub is already registered
+ */
+error_t lora_rx_event_subscribe(struct LoraRxEventSubscription* sub, struct TaskEventGroup* event_group, struct Device* device);
+
+/**
+ * Remove a previously registered subscription.
+ * @warning Does not work in ISR context.
+ * @return ERROR_NONE on success, ERROR_NOT_FOUND if no matching subscription exists
+ */
+error_t lora_rx_event_unsubscribe(struct LoraRxEventSubscription* sub);
+
+/**
+ * Non-blocking: pop the next event for @a sub if one is already queued.
+ * @warning Never blocks. To wait for an event, block in task_event_group_wait()/
+ * task_event_group_wait_any() on @a sub's event group first (see lora_rx_event_subscribe()),
+ * then drain with this in a loop.
+ * @retval ERROR_NONE @a out_event was filled
+ * @retval ERROR_TIMEOUT nothing queued right now
+ */
+error_t lora_rx_event_poll(struct LoraRxEventSubscription* sub, struct LoraRxEvent* out_event);
+
+/**
+ * Emit a received-packet event to every subscriber of @a device. Called by LoraApi driver
+ * implementations; not intended for consumers of the API.
+ * @param[in] device the lora device the event originates from
+ * @param[in] data the packet payload; copied, truncated to LORA_RX_MAX_PACKET_LENGTH
+ * @param[in] length the payload length in bytes
+ * @param[in] rssi received signal strength in dBm
+ * @param[in] snr signal-to-noise ratio in dB
+ * @return ERROR_NONE on success, ERROR_NOT_FOUND if @a device has no subscribers
+ */
+error_t lora_rx_event_emit(struct Device* device, const uint8_t* data, size_t length, float rssi, float snr);
+
+// endregion
+
+// region TX events
+
+struct LoraTxEvent {
+    /** Stamped by lora_tx_event_emit(); any value passed in by the caller is ignored. */
+    uint64_t timestamp;
+    LoraTxId id;
+    enum LoraTransmissionState state;
+};
+
+/**
+ * Number of events that can be queued per subscription before lora_tx_event_emit() starts
+ * returning ERROR_RESOURCE (dropping the newest event, preserving FIFO order of what's
+ * already queued).
+ */
+#define LORA_TX_EVENT_QUEUE_CAPACITY 4
+
+/**
+ * Caller-owned subscription node, registered with lora_tx_event_subscribe() and drained with
+ * lora_tx_event_poll(). Events queue by value (FIFO).
+ * @warning Fields other than `bit` are for internal use only; do not read or write them directly.
+ */
+struct LoraTxEventSubscription {
+    /** Set by lora_tx_event_subscribe(). Read-only for the caller: OR it into a
+     * task_event_group_wait() mask (alongside other subscriptions sharing the same
+     * `event_group`) to block on this subscription and other event sources with one call. */
+    uint32_t bit;
+
+    struct {
+        /** The lora device this subscription receives events for; set by lora_tx_event_subscribe(). */
+        struct Device* device;
+
+        /** Caller-owned, borrowed; set by lora_tx_event_subscribe(). */
+        struct TaskEventGroup* event_group;
+
+        struct LoraTxEvent queue[LORA_TX_EVENT_QUEUE_CAPACITY];
+        uint8_t head;
+        uint8_t count;
+
+        struct LoraTxEventSubscription* next;
+    } internal;
+};
+
+/**
+ * Register a subscription for transmission progress emitted by @a device.
+ * @warning Does not work in ISR context.
+ * @param[in,out] sub subscription to register; owns the storage, must stay alive (and
+ * stationary) until unsubscribed
+ * @param[in] event_group caller-owned group to wait on; must outlive @a sub (i.e. be
+ * destructed only after lora_tx_event_unsubscribe()). To block for an event, call
+ * task_event_group_wait()/task_event_group_wait_any() on this group (OR sub->bit into the mask,
+ * or use _wait_any() to include every subscription sharing it), then drain with
+ * lora_tx_event_poll().
+ * @param[in] device the lora device this subscription receives events for
+ * @retval ERROR_NONE on success
+ * @retval ERROR_RESOURCE @a event_group has no free bits left to claim; @a sub was not registered
+ * @retval ERROR_INVALID_STATE @a sub is already registered
+ */
+error_t lora_tx_event_subscribe(struct LoraTxEventSubscription* sub, struct TaskEventGroup* event_group, struct Device* device);
+
+/**
+ * Remove a previously registered subscription.
+ * @warning Does not work in ISR context.
+ * @return ERROR_NONE on success, ERROR_NOT_FOUND if no matching subscription exists
+ */
+error_t lora_tx_event_unsubscribe(struct LoraTxEventSubscription* sub);
+
+/**
+ * Non-blocking: pop the next event for @a sub if one is already queued.
+ * @warning Never blocks. To wait for an event, block in task_event_group_wait()/
+ * task_event_group_wait_any() on @a sub's event group first (see lora_tx_event_subscribe()),
+ * then drain with this in a loop.
+ * @retval ERROR_NONE @a out_event was filled
+ * @retval ERROR_TIMEOUT nothing queued right now
+ */
+error_t lora_tx_event_poll(struct LoraTxEventSubscription* sub, struct LoraTxEvent* out_event);
+
+/**
+ * Emit a transmission-progress event to every subscriber of @a device. Called by LoraApi driver
+ * implementations; not intended for consumers of the API.
+ * @param[in] device the lora device the event originates from
+ * @param[in] id the transmission this event reports progress for
+ * @param[in] state the transmission's new state
+ * @return ERROR_NONE on success, ERROR_NOT_FOUND if @a device has no subscribers
+ */
+error_t lora_tx_event_emit(struct Device* device, LoraTxId id, enum LoraTransmissionState state);
+
+// endregion
+
+extern const struct DeviceType LORA_TYPE;
 
 error_t lora_get_radio_state(struct Device* device, enum LoraRadioState* state);
 error_t lora_set_enabled(struct Device* device, bool enabled);
@@ -265,12 +463,6 @@ bool lora_can_receive(struct Device* device, enum LoraModulation modulation);
 error_t lora_set_parameter(struct Device* device, enum LoraParameter parameter, int32_t value);
 error_t lora_get_parameter(struct Device* device, enum LoraParameter parameter, int32_t* value);
 error_t lora_transmit(struct Device* device, const uint8_t* data, size_t length, LoraTxId* id);
-error_t lora_add_rx_callback(struct Device* device, void* callback_context, LoraRxCallback callback);
-error_t lora_remove_rx_callback(struct Device* device, LoraRxCallback callback);
-error_t lora_add_state_callback(struct Device* device, void* callback_context, LoraStateCallback callback);
-error_t lora_remove_state_callback(struct Device* device, LoraStateCallback callback);
-error_t lora_add_tx_callback(struct Device* device, void* callback_context, LoraTxCallback callback);
-error_t lora_remove_tx_callback(struct Device* device, LoraTxCallback callback);
 
 #ifdef __cplusplus
 }

TactilityKernel/source/drivers/lora.cpp

@@ -1,21 +1,326 @@
 // SPDX-License-Identifier: Apache-2.0
 #include <tactility/drivers/lora.h>
+#include <tactility/concurrent/mutex.h>
 #include <tactility/device.h>
 #include <tactility/driver.h>
+#include <tactility/time.h>
+
+#include <algorithm>
+#include <cstring>
 
 #define LORA_API(device) ((const struct LoraApi*)device_get_driver(device)->api)
 
+struct LoraEventMutex {
+    Mutex handle {};
+    LoraEventMutex() { mutex_construct(&handle); }
+    ~LoraEventMutex() { mutex_destruct(&handle); }
+};
+
+// Each event kind gets its own intrusive singly-linked subscription list, keyed by device, and
+// its own coarse-grained mutex. Notifying a subscriber never invokes caller code (just a struct
+// copy and a task_event_group_signal), so there is no reentrancy concern requiring a
+// snapshot-then-unlock dance.
+static LoraStateEventSubscription* lora_state_event_subscriptions = nullptr;
+static LoraEventMutex lora_state_event_subscriptions_mutex;
+
+static LoraRxEventSubscription* lora_rx_event_subscriptions = nullptr;
+static LoraEventMutex lora_rx_event_subscriptions_mutex;
+
+static LoraTxEventSubscription* lora_tx_event_subscriptions = nullptr;
+static LoraEventMutex lora_tx_event_subscriptions_mutex;
+
 extern "C" {
 
-struct Device* lora_find_first_registered_device() {
-    struct Device* found = nullptr;
-    device_for_each_of_type(&LORA_TYPE, &found, [](struct Device* dev, void* ctx) -> bool {
-        *static_cast<struct Device**>(ctx) = dev;
-        return false;
-    });
-    return found;
+// region State events
+
+error_t lora_state_event_subscribe(LoraStateEventSubscription* sub, TaskEventGroup* event_group, Device* device) {
+    uint32_t bit;
+    error_t claim_result = task_event_group_claim_bit(event_group, &bit);
+    if (claim_result != ERROR_NONE) {
+        return claim_result;
     }
 
+    mutex_lock(&lora_state_event_subscriptions_mutex.handle);
+
+    // Avoid cyclic subscription list that would loop forever
+    for (LoraStateEventSubscription* existing = lora_state_event_subscriptions; existing != nullptr; existing = existing->internal.next) {
+        if (existing == sub) {
+            mutex_unlock(&lora_state_event_subscriptions_mutex.handle);
+            task_event_group_release_bit(event_group, bit);
+            return ERROR_INVALID_STATE;
+        }
+    }
+
+    sub->bit = bit;
+    sub->internal.device = device;
+    sub->internal.event_group = event_group;
+    sub->internal.head = 0;
+    sub->internal.count = 0;
+    sub->internal.next = lora_state_event_subscriptions;
+    lora_state_event_subscriptions = sub;
+    mutex_unlock(&lora_state_event_subscriptions_mutex.handle);
+
+    return ERROR_NONE;
+}
+
+error_t lora_state_event_unsubscribe(LoraStateEventSubscription* sub) {
+    error_t result = ERROR_NOT_FOUND;
+
+    mutex_lock(&lora_state_event_subscriptions_mutex.handle);
+    for (LoraStateEventSubscription** link = &lora_state_event_subscriptions; *link != nullptr; link = &(*link)->internal.next) {
+        if (*link == sub) {
+            *link = sub->internal.next;
+            result = ERROR_NONE;
+            break;
+        }
+    }
+    mutex_unlock(&lora_state_event_subscriptions_mutex.handle);
+
+    if (result == ERROR_NONE) {
+        task_event_group_release_bit(sub->internal.event_group, sub->bit);
+    }
+
+    return result;
+}
+
+error_t lora_state_event_emit(Device* device, enum LoraRadioState state) {
+    LoraStateEvent stamped_event { .timestamp = get_micros_since_boot(), .state = state };
+
+    error_t result = ERROR_NOT_FOUND;
+
+    mutex_lock(&lora_state_event_subscriptions_mutex.handle);
+    for (LoraStateEventSubscription* sub = lora_state_event_subscriptions; sub != nullptr; sub = sub->internal.next) {
+        if (sub->internal.device != device) {
+            continue;
+        }
+
+        if (sub->internal.count >= LORA_STATE_EVENT_QUEUE_CAPACITY) {
+            result = ERROR_RESOURCE;
+            continue;
+        }
+
+        uint8_t tail = (sub->internal.head + sub->internal.count) % LORA_STATE_EVENT_QUEUE_CAPACITY;
+        sub->internal.queue[tail] = stamped_event;
+        sub->internal.count++;
+        if (result != ERROR_RESOURCE) {
+            result = ERROR_NONE;
+        }
+        task_event_group_signal(sub->internal.event_group, sub->bit);
+    }
+    mutex_unlock(&lora_state_event_subscriptions_mutex.handle);
+
+    return result;
+}
+
+error_t lora_state_event_poll(LoraStateEventSubscription* sub, LoraStateEvent* out_event) {
+    mutex_lock(&lora_state_event_subscriptions_mutex.handle);
+    bool has_event = sub->internal.count > 0;
+    if (has_event) {
+        *out_event = sub->internal.queue[sub->internal.head];
+        sub->internal.head = (sub->internal.head + 1) % LORA_STATE_EVENT_QUEUE_CAPACITY;
+        sub->internal.count--;
+    }
+    mutex_unlock(&lora_state_event_subscriptions_mutex.handle);
+    return has_event ? ERROR_NONE : ERROR_TIMEOUT;
+}
+
+// endregion
+
+// region RX events
+
+error_t lora_rx_event_subscribe(LoraRxEventSubscription* sub, TaskEventGroup* event_group, Device* device) {
+    uint32_t bit;
+    error_t claim_result = task_event_group_claim_bit(event_group, &bit);
+    if (claim_result != ERROR_NONE) {
+        return claim_result;
+    }
+
+    mutex_lock(&lora_rx_event_subscriptions_mutex.handle);
+
+    // Avoid cyclic subscription list that would loop forever
+    for (LoraRxEventSubscription* existing = lora_rx_event_subscriptions; existing != nullptr; existing = existing->internal.next) {
+        if (existing == sub) {
+            mutex_unlock(&lora_rx_event_subscriptions_mutex.handle);
+            task_event_group_release_bit(event_group, bit);
+            return ERROR_INVALID_STATE;
+        }
+    }
+
+    sub->bit = bit;
+    sub->internal.device = device;
+    sub->internal.event_group = event_group;
+    sub->internal.head = 0;
+    sub->internal.count = 0;
+    sub->internal.next = lora_rx_event_subscriptions;
+    lora_rx_event_subscriptions = sub;
+    mutex_unlock(&lora_rx_event_subscriptions_mutex.handle);
+
+    return ERROR_NONE;
+}
+
+error_t lora_rx_event_unsubscribe(LoraRxEventSubscription* sub) {
+    error_t result = ERROR_NOT_FOUND;
+
+    mutex_lock(&lora_rx_event_subscriptions_mutex.handle);
+    for (LoraRxEventSubscription** link = &lora_rx_event_subscriptions; *link != nullptr; link = &(*link)->internal.next) {
+        if (*link == sub) {
+            *link = sub->internal.next;
+            result = ERROR_NONE;
+            break;
+        }
+    }
+    mutex_unlock(&lora_rx_event_subscriptions_mutex.handle);
+
+    if (result == ERROR_NONE) {
+        task_event_group_release_bit(sub->internal.event_group, sub->bit);
+    }
+
+    return result;
+}
+
+error_t lora_rx_event_emit(Device* device, const uint8_t* data, size_t length, float rssi, float snr) {
+    LoraRxEvent stamped_event {};
+    stamped_event.timestamp = get_micros_since_boot();
+    stamped_event.length = std::min(length, sizeof(stamped_event.data));
+    std::memcpy(stamped_event.data, data, stamped_event.length);
+    stamped_event.rssi = rssi;
+    stamped_event.snr = snr;
+
+    error_t result = ERROR_NOT_FOUND;
+
+    mutex_lock(&lora_rx_event_subscriptions_mutex.handle);
+    for (LoraRxEventSubscription* sub = lora_rx_event_subscriptions; sub != nullptr; sub = sub->internal.next) {
+        if (sub->internal.device != device) {
+            continue;
+        }
+
+        if (sub->internal.count >= LORA_RX_EVENT_QUEUE_CAPACITY) {
+            result = ERROR_RESOURCE;
+            continue;
+        }
+
+        uint8_t tail = (sub->internal.head + sub->internal.count) % LORA_RX_EVENT_QUEUE_CAPACITY;
+        sub->internal.queue[tail] = stamped_event;
+        sub->internal.count++;
+        if (result != ERROR_RESOURCE) {
+            result = ERROR_NONE;
+        }
+        task_event_group_signal(sub->internal.event_group, sub->bit);
+    }
+    mutex_unlock(&lora_rx_event_subscriptions_mutex.handle);
+
+    return result;
+}
+
+error_t lora_rx_event_poll(LoraRxEventSubscription* sub, LoraRxEvent* out_event) {
+    mutex_lock(&lora_rx_event_subscriptions_mutex.handle);
+    bool has_event = sub->internal.count > 0;
+    if (has_event) {
+        *out_event = sub->internal.queue[sub->internal.head];
+        sub->internal.head = (sub->internal.head + 1) % LORA_RX_EVENT_QUEUE_CAPACITY;
+        sub->internal.count--;
+    }
+    mutex_unlock(&lora_rx_event_subscriptions_mutex.handle);
+    return has_event ? ERROR_NONE : ERROR_TIMEOUT;
+}
+
+// endregion
+
+// region TX events
+
+error_t lora_tx_event_subscribe(LoraTxEventSubscription* sub, TaskEventGroup* event_group, Device* device) {
+    uint32_t bit;
+    error_t claim_result = task_event_group_claim_bit(event_group, &bit);
+    if (claim_result != ERROR_NONE) {
+        return claim_result;
+    }
+
+    mutex_lock(&lora_tx_event_subscriptions_mutex.handle);
+
+    // Avoid cyclic subscription list that would loop forever
+    for (LoraTxEventSubscription* existing = lora_tx_event_subscriptions; existing != nullptr; existing = existing->internal.next) {
+        if (existing == sub) {
+            mutex_unlock(&lora_tx_event_subscriptions_mutex.handle);
+            task_event_group_release_bit(event_group, bit);
+            return ERROR_INVALID_STATE;
+        }
+    }
+
+    sub->bit = bit;
+    sub->internal.device = device;
+    sub->internal.event_group = event_group;
+    sub->internal.head = 0;
+    sub->internal.count = 0;
+    sub->internal.next = lora_tx_event_subscriptions;
+    lora_tx_event_subscriptions = sub;
+    mutex_unlock(&lora_tx_event_subscriptions_mutex.handle);
+
+    return ERROR_NONE;
+}
+
+error_t lora_tx_event_unsubscribe(LoraTxEventSubscription* sub) {
+    error_t result = ERROR_NOT_FOUND;
+
+    mutex_lock(&lora_tx_event_subscriptions_mutex.handle);
+    for (LoraTxEventSubscription** link = &lora_tx_event_subscriptions; *link != nullptr; link = &(*link)->internal.next) {
+        if (*link == sub) {
+            *link = sub->internal.next;
+            result = ERROR_NONE;
+            break;
+        }
+    }
+    mutex_unlock(&lora_tx_event_subscriptions_mutex.handle);
+
+    if (result == ERROR_NONE) {
+        task_event_group_release_bit(sub->internal.event_group, sub->bit);
+    }
+
+    return result;
+}
+
+error_t lora_tx_event_emit(Device* device, LoraTxId id, enum LoraTransmissionState state) {
+    LoraTxEvent stamped_event { .timestamp = get_micros_since_boot(), .id = id, .state = state };
+
+    error_t result = ERROR_NOT_FOUND;
+
+    mutex_lock(&lora_tx_event_subscriptions_mutex.handle);
+    for (LoraTxEventSubscription* sub = lora_tx_event_subscriptions; sub != nullptr; sub = sub->internal.next) {
+        if (sub->internal.device != device) {
+            continue;
+        }
+
+        if (sub->internal.count >= LORA_TX_EVENT_QUEUE_CAPACITY) {
+            result = ERROR_RESOURCE;
+            continue;
+        }
+
+        uint8_t tail = (sub->internal.head + sub->internal.count) % LORA_TX_EVENT_QUEUE_CAPACITY;
+        sub->internal.queue[tail] = stamped_event;
+        sub->internal.count++;
+        if (result != ERROR_RESOURCE) {
+            result = ERROR_NONE;
+        }
+        task_event_group_signal(sub->internal.event_group, sub->bit);
+    }
+    mutex_unlock(&lora_tx_event_subscriptions_mutex.handle);
+
+    return result;
+}
+
+error_t lora_tx_event_poll(LoraTxEventSubscription* sub, LoraTxEvent* out_event) {
+    mutex_lock(&lora_tx_event_subscriptions_mutex.handle);
+    bool has_event = sub->internal.count > 0;
+    if (has_event) {
+        *out_event = sub->internal.queue[sub->internal.head];
+        sub->internal.head = (sub->internal.head + 1) % LORA_TX_EVENT_QUEUE_CAPACITY;
+        sub->internal.count--;
+    }
+    mutex_unlock(&lora_tx_event_subscriptions_mutex.handle);
+    return has_event ? ERROR_NONE : ERROR_TIMEOUT;
+}
+
+// endregion
+
 error_t lora_get_radio_state(struct Device* device, enum LoraRadioState* state) {
     return LORA_API(device)->get_radio_state(device, state);
 }
@@ -52,30 +357,6 @@ error_t lora_transmit(struct Device* device, const uint8_t* data, size_t length,
     return LORA_API(device)->transmit(device, data, length, id);
 }
 
-error_t lora_add_rx_callback(struct Device* device, void* callback_context, LoraRxCallback callback) {
-    return LORA_API(device)->add_rx_callback(device, callback_context, callback);
-}
-
-error_t lora_remove_rx_callback(struct Device* device, LoraRxCallback callback) {
-    return LORA_API(device)->remove_rx_callback(device, callback);
-}
-
-error_t lora_add_state_callback(struct Device* device, void* callback_context, LoraStateCallback callback) {
-    return LORA_API(device)->add_state_callback(device, callback_context, callback);
-}
-
-error_t lora_remove_state_callback(struct Device* device, LoraStateCallback callback) {
-    return LORA_API(device)->remove_state_callback(device, callback);
-}
-
-error_t lora_add_tx_callback(struct Device* device, void* callback_context, LoraTxCallback callback) {
-    return LORA_API(device)->add_tx_callback(device, callback_context, callback);
-}
-
-error_t lora_remove_tx_callback(struct Device* device, LoraTxCallback callback) {
-    return LORA_API(device)->remove_tx_callback(device, callback);
-}
-
 const struct DeviceType LORA_TYPE = {
     .name = "lora"
 };

TactilityKernel/source/symbols.c

@@ -397,7 +397,6 @@ const struct ModuleSymbol KERNEL_SYMBOLS[] = {
     // wifi_auto_scan
     DEFINE_MODULE_SYMBOL(wifi_auto_scan_set_paused),
     // drivers/lora
-    DEFINE_MODULE_SYMBOL(lora_find_first_registered_device),
     DEFINE_MODULE_SYMBOL(lora_get_radio_state),
     DEFINE_MODULE_SYMBOL(lora_set_enabled),
     DEFINE_MODULE_SYMBOL(lora_set_modulation),
@@ -407,12 +406,15 @@ const struct ModuleSymbol KERNEL_SYMBOLS[] = {
     DEFINE_MODULE_SYMBOL(lora_set_parameter),
     DEFINE_MODULE_SYMBOL(lora_get_parameter),
     DEFINE_MODULE_SYMBOL(lora_transmit),
-    DEFINE_MODULE_SYMBOL(lora_add_rx_callback),
-    DEFINE_MODULE_SYMBOL(lora_remove_rx_callback),
-    DEFINE_MODULE_SYMBOL(lora_add_state_callback),
-    DEFINE_MODULE_SYMBOL(lora_remove_state_callback),
-    DEFINE_MODULE_SYMBOL(lora_add_tx_callback),
-    DEFINE_MODULE_SYMBOL(lora_remove_tx_callback),
+    DEFINE_MODULE_SYMBOL(lora_state_event_subscribe),
+    DEFINE_MODULE_SYMBOL(lora_state_event_unsubscribe),
+    DEFINE_MODULE_SYMBOL(lora_state_event_poll),
+    DEFINE_MODULE_SYMBOL(lora_rx_event_subscribe),
+    DEFINE_MODULE_SYMBOL(lora_rx_event_unsubscribe),
+    DEFINE_MODULE_SYMBOL(lora_rx_event_poll),
+    DEFINE_MODULE_SYMBOL(lora_tx_event_subscribe),
+    DEFINE_MODULE_SYMBOL(lora_tx_event_unsubscribe),
+    DEFINE_MODULE_SYMBOL(lora_tx_event_poll),
     DEFINE_MODULE_SYMBOL(LORA_TYPE),
     // drivers/usb_host_hid
     DEFINE_MODULE_SYMBOL(usb_host_hid_is_connected),