diff --git a/.github/workflows/run-unit-tests.yml b/.github/workflows/run-unit-tests.yml index 8aa6994d2..dee98f9af 100644 --- a/.github/workflows/run-unit-tests.yml +++ b/.github/workflows/run-unit-tests.yml @@ -708,6 +708,7 @@ jobs: python3 -B test/test_t096_full_memory.py python3 -B test/test_nrf52_ble_startup.py python3 -B test/test_nrf52_ble_backpressure.py + python3 -B test/test_nrf52_ble_rx_queue.py -v python3 -B test/test_nrf52_ble_trace.py python3 -B test/test_nrf52_ble_debug_tools.py python3 -B test/test_nrf52_usb_power.py diff --git a/src/helpers/nrf52/BleRxFrameQueue.h b/src/helpers/nrf52/BleRxFrameQueue.h new file mode 100644 index 000000000..000f7dc97 --- /dev/null +++ b/src/helpers/nrf52/BleRxFrameQueue.h @@ -0,0 +1,120 @@ +#pragma once + +#include +#include +#include + +namespace mesh_nrf52 { + +// Guard serializes task access on the single-core nRF52. Keep every guarded +// operation bounded to metadata or one frame copy; UART FIFO operations must +// run outside the guard because Bluefruit takes a FreeRTOS mutex there. +template +class BleRxFrameQueue { + static_assert(Capacity > 0 && Capacity <= 255, "invalid BLE RX capacity"); + static constexpr uint16_t INVALID_HANDLE = 0xffff; + + Frame _frames[Capacity]; + uint32_t _generation = 0; + uint16_t _handle = INVALID_HANDLE; + uint16_t _resets_in_progress = 0; + uint8_t _head = 0; + uint8_t _count = 0; + bool _enabled = false; + bool _secured = false; + + void invalidate() { + ++_generation; + _head = _count = 0; + } + + bool accepts(uint16_t handle, uint32_t generation) const { + return _enabled && _secured && !_resets_in_progress + && handle == _handle && generation == _generation; + } + +public: + enum class PushResult { Accepted, Full, Stale }; + + void setEnabled(bool enabled) { + Guard guard; + _enabled = enabled; + _handle = INVALID_HANDLE; + _secured = false; + invalidate(); + } + + void connect(uint16_t handle) { + Guard guard; + _handle = _enabled ? handle : INVALID_HANDLE; + _secured = false; + invalidate(); + } + + void secure(uint16_t handle) { + Guard guard; + if (_enabled && _handle == handle && handle != INVALID_HANDLE) { + _secured = true; + } + } + + void disconnect(uint16_t handle) { + Guard guard; + if (_handle != handle) return; + _handle = INVALID_HANDLE; + _secured = false; + invalidate(); + } + + template + void reset(Flush flush) { + { + Guard guard; + ++_resets_in_progress; + invalidate(); + } + flush(); + { + Guard guard; + --_resets_in_progress; + } + } + + bool snapshot(uint16_t handle, uint32_t& generation) const { + Guard guard; + generation = _generation; + return accepts(handle, generation); + } + + bool current(uint16_t handle, uint32_t generation) const { + Guard guard; + return accepts(handle, generation); + } + + PushResult push(uint16_t handle, uint32_t generation, const Frame& frame) { + Guard guard; + if (!accepts(handle, generation)) return PushResult::Stale; + if (_count == Capacity) return PushResult::Full; + _frames[(_head + _count) % Capacity] = frame; + ++_count; + return PushResult::Accepted; + } + + size_t pop(uint8_t* destination) { + Guard guard; + if (!_count) return 0; + const Frame& frame = _frames[_head]; + const size_t length = frame.len; + memcpy(destination, frame.buf, length); + _head = (_head + 1) % Capacity; + --_count; + return length; + } + + uint8_t size() const { + Guard guard; + return _count; + } +}; + +} // namespace mesh_nrf52 diff --git a/src/helpers/nrf52/SerialBLEInterface.cpp b/src/helpers/nrf52/SerialBLEInterface.cpp index a3817d158..368cd65fb 100644 --- a/src/helpers/nrf52/SerialBLEInterface.cpp +++ b/src/helpers/nrf52/SerialBLEInterface.cpp @@ -32,9 +32,6 @@ __attribute__((used, externally_visible)) MeshBleTraceBuffer mesh_ble_trace; #define BLE_ADV_INTERVAL_MAX 244 // 152.5ms (units: 0.625ms) #define BLE_ADV_FAST_TIMEOUT 30 // seconds -// RX drain buffer size for overflow protection -#define BLE_RX_DRAIN_BUF_SIZE 32 - static SerialBLEInterface* instance = nullptr; #if defined(MESH_NRF52_BLE_TRACE) && MESH_NRF52_BLE_TRACE @@ -45,7 +42,7 @@ void SerialBLEInterface::formatDebugState(char* reply, size_t capacity) { (unsigned)_isDeviceConnected, (unsigned)Bluefruit.Advertising.isRunning(), (unsigned)Bluefruit.connected(), (unsigned)_companionDataSeen.load(std::memory_order_acquire), - (unsigned)send_queue_len, (unsigned)recv_queue_len, + (unsigned)send_queue_len, (unsigned)recv_queue.size(), (unsigned)_tx_disconnect_recovery.pending(), (unsigned)(_conn_handle != BLE_CONN_HANDLE_INVALID && bleuart.notifyEnabled(_conn_handle)), (unsigned)_debug_mtu.load()); @@ -273,6 +270,7 @@ void SerialBLEInterface::onSecured(uint16_t connection_handle) { } instance->_isDeviceConnected = true; + instance->recv_queue.secure(connection_handle); #if defined(MESH_NRF52_BLE_TRACE) && MESH_NRF52_BLE_TRACE mesh_ble_trace.secured.fetch_add(1); meshBleTrace(0xf006, connection_handle, conn->bonded(), @@ -431,6 +429,10 @@ void SerialBLEInterface::onBLEEvent(ble_evt_t* evt) { if (evt->header.evt_id == BLE_GAP_EVT_CONNECTED) { ble_gap_evt_connected_t const* connected = &evt->evt.gap_evt.params.connected; if (connected->role == BLE_GAP_ROLE_PERIPH) { + // Observe the physical session in the BLE worker before its deferred + // onConnect callback. A disable/re-enable between the event and that + // callback must not allow the old link to reopen the RX queue. + instance->recv_queue.connect(evt->evt.gap_evt.conn_handle); instance->_peer_address = connected->peer_addr; instance->_peer_address_valid = true; instance->_bond_removed_for_connection = false; @@ -480,6 +482,7 @@ void SerialBLEInterface::onBLEEvent(ble_evt_t* evt) { } else if (evt->header.evt_id == BLE_GATTS_EVT_SYS_ATTR_MISSING) { BLE_DEBUG_PRINTLN("CCCD RAM system attributes missing"); } else if (evt->header.evt_id == BLE_GAP_EVT_DISCONNECTED) { + instance->recv_queue.disconnect(evt->evt.gap_evt.conn_handle); ble_gap_evt_disconnected_t const* disconnected = &evt->evt.gap_evt.params.disconnected; if (isBondAuthenticationFailure(disconnected->reason)) { instance->removeStoredBondForPeer("authentication disconnect"); @@ -986,11 +989,10 @@ void SerialBLEInterface::cancelStealthPairingTransition() { void SerialBLEInterface::clearBuffers() { send_queue_len = 0; - recv_queue_len = 0; _last_retry_attempt = 0; _tx_stall_watchdog.reset(); _tx_disconnect_recovery.complete(); - bleuart.flush(); + recv_queue.reset([this]() { bleuart.flush(); }); } void SerialBLEInterface::shiftSendQueueLeft() { @@ -1002,15 +1004,6 @@ void SerialBLEInterface::shiftSendQueueLeft() { } } -void SerialBLEInterface::shiftRecvQueueLeft() { - if (recv_queue_len > 0) { - recv_queue_len--; - for (uint8_t i = 0; i < recv_queue_len; i++) { - recv_queue[i] = recv_queue[i + 1]; - } - } -} - size_t SerialBLEInterface::writeBleUartFrame(const Frame& frame) { BLEConnection* conn = Bluefruit.Connection(_conn_handle); if (conn == nullptr || !conn->connected() || @@ -1071,7 +1064,7 @@ void SerialBLEInterface::serviceTxRecovery(uint32_t now) { } void SerialBLEInterface::recoverStalledTx(const char* cause) { - meshBleTrace(0xf020, _conn_handle, send_queue_len, recv_queue_len); + meshBleTrace(0xf020, _conn_handle, send_queue_len, recv_queue.size()); if (_tx_disconnect_recovery.pending()) return; BLE_DEBUG_PRINTLN("SerialBLEInterface: %s; forcing reconnect", cause); @@ -1080,11 +1073,11 @@ void SerialBLEInterface::recoverStalledTx(const char* cause) { // disconnection, but make isConnected() false through the recovery state so // no more companion frames enter this damaged stream. send_queue_len = 0; - recv_queue_len = 0; + recv_queue.disconnect(_conn_handle); _last_retry_attempt = 0; _tx_stall_watchdog.reset(); _companion_start_timer.cancel(); - bleuart.flush(); + recv_queue.reset([this]() { bleuart.flush(); }); #if COMPANION_FEATURE_BLE_MOTA_SOURCE setMotaStreamActive(false); #endif @@ -1113,6 +1106,7 @@ bool SerialBLEInterface::isAdvertising() const { void SerialBLEInterface::enable() { if (_isEnabled) return; + recv_queue.setEnabled(true); _pairingRequestPending.store(false, std::memory_order_release); _isEnabled = true; clearBuffers(); @@ -1126,6 +1120,7 @@ void SerialBLEInterface::enable() { void SerialBLEInterface::disconnect() { if (_conn_handle != BLE_CONN_HANDLE_INVALID) { + recv_queue.disconnect(_conn_handle); const uint32_t result = sd_ble_gap_disconnect(_conn_handle, BLE_HCI_REMOTE_USER_TERMINATED_CONNECTION); meshBleTrace(0xf013, _conn_handle, result, 0); (void)result; @@ -1133,6 +1128,7 @@ void SerialBLEInterface::disconnect() { } void SerialBLEInterface::disable() { + recv_queue.setEnabled(false); _isEnabled = false; _pairingRequestPending.store(false, std::memory_order_release); BLE_DEBUG_PRINTLN("SerialBLEInterface: disable"); @@ -1298,13 +1294,11 @@ size_t SerialBLEInterface::checkRecvFrame(uint8_t dest[]) { // TX/recovery above must run even when full. Retain RX until a required // command reply can be admitted instead of silently losing that reply. - if (recv_queue_len > 0 && hasRequiredReplyCapacity()) { - size_t len = recv_queue[0].len; - memcpy(dest, recv_queue[0].buf, len); - - BLE_DEBUG_PRINTLN("readBytes: sz=%u, hdr=%u", (unsigned)len, (unsigned)dest[0]); - - shiftRecvQueueLeft(); + if (hasRequiredReplyCapacity()) { + const size_t len = recv_queue.pop(dest); + if (len) { + BLE_DEBUG_PRINTLN("readBytes: sz=%u, hdr=%u", (unsigned)len, (unsigned)dest[0]); + } return len; } @@ -1312,50 +1306,49 @@ size_t SerialBLEInterface::checkRecvFrame(uint8_t dest[]) { } void SerialBLEInterface::onBleUartRX(uint16_t conn_handle) { - if (!instance) { - return; - } - - meshBleTrace(0xf030, conn_handle, instance->bleuart.available(), - instance->isConnected()); - if (instance->_conn_handle != conn_handle || !instance->isConnected()) { - while (instance->bleuart.available() > 0) { - instance->bleuart.read(); - } - return; - } - - while (instance->bleuart.available() > 0) { - if (instance->recv_queue_len >= FRAME_QUEUE_SIZE) { - while (instance->bleuart.available() > 0) { - instance->bleuart.read(); - } - meshBleTrace(0xf034, conn_handle, instance->recv_queue_len); - BLE_DEBUG_PRINTLN("onBleUartRX: recv queue full, dropping data"); - break; - } - - int avail = instance->bleuart.available(); - + SerialBLEInterface* self = instance; + if (!self) return; + + uint32_t generation; + if (!self->recv_queue.snapshot(conn_handle, generation) + || !self->isConnected()) return; + meshBleTrace(0xf030, conn_handle, self->bleuart.available(), true); + + while (self->recv_queue.current(conn_handle, generation)) { + const int avail = self->bleuart.available(); + if (avail <= 0) return; if (avail > MAX_FRAME_SIZE) { BLE_DEBUG_PRINTLN("onBleUartRX: WARN: BLE RX overflow, avail=%d, draining all", avail); - uint8_t drain_buf[BLE_RX_DRAIN_BUF_SIZE]; - while (instance->bleuart.available() > 0) { - int chunk = instance->bleuart.available() > BLE_RX_DRAIN_BUF_SIZE ? BLE_RX_DRAIN_BUF_SIZE : instance->bleuart.available(); - instance->bleuart.readBytes(drain_buf, chunk); - } - continue; + self->bleuart.flush(); + return; + } + + // BLEUart::read() is an immediate FIFO read. Stream::readBytes() may wait + // for bytes flushed by a simultaneous reset. Stage outside the critical + // section, then reject the frame if its connection generation changed. + Frame frame = {}; + const int read_len = self->bleuart.read(frame.buf, avail); + if (read_len != avail) return; + frame.len = read_len; + RxQueue::PushResult result; + { + RxQueueGuard guard; + result = self->recv_queue.push(conn_handle, generation, frame); + if (result == RxQueue::PushResult::Accepted) { + self->_companionDataSeen.store(true, std::memory_order_release); + self->_companion_start_timer.cancel(); + } + } + if (result == RxQueue::PushResult::Stale) return; + if (result == RxQueue::PushResult::Full) { + self->bleuart.flush(); + meshBleTrace(0xf034, conn_handle, FRAME_QUEUE_SIZE); + BLE_DEBUG_PRINTLN("onBleUartRX: recv queue full, dropping data"); + return; } - - int read_len = avail; - instance->recv_queue[instance->recv_queue_len].len = read_len; - instance->bleuart.readBytes(instance->recv_queue[instance->recv_queue_len].buf, read_len); - instance->recv_queue_len++; #if defined(MESH_NRF52_BLE_TRACE) && MESH_NRF52_BLE_TRACE mesh_ble_trace.rx.fetch_add(1); #endif - instance->_companionDataSeen.store(true, std::memory_order_release); - instance->_companion_start_timer.cancel(); } } @@ -1393,7 +1386,7 @@ uint16_t SerialBLEInterface::motaResponseCapacity() { #endif bool SerialBLEInterface::isReadBusy() const { - return (recv_queue_len > 0); + return recv_queue.size() > 0; } bool SerialBLEInterface::isWriteBusy() const { @@ -1401,5 +1394,5 @@ bool SerialBLEInterface::isWriteBusy() const { } bool SerialBLEInterface::hasPendingIO() const { - return recv_queue_len > 0 || send_queue_len > 0; + return recv_queue.size() > 0 || send_queue_len > 0; } diff --git a/src/helpers/nrf52/SerialBLEInterface.h b/src/helpers/nrf52/SerialBLEInterface.h index e7dfbdcd3..59010f340 100644 --- a/src/helpers/nrf52/SerialBLEInterface.h +++ b/src/helpers/nrf52/SerialBLEInterface.h @@ -5,6 +5,7 @@ #include "../BleTxStallWatchdog.h" #include "../UsbLogging.h" #include "SecuritySessionTimer.h" +#include "BleRxFrameQueue.h" #include #include #if COMPANION_FEATURE_BLE_MOTA_SOURCE @@ -105,13 +106,17 @@ class SerialBLEInterface : public BaseSerialInterface { uint8_t send_queue_len; Frame send_queue[FRAME_QUEUE_SIZE]; - uint8_t recv_queue_len; - Frame recv_queue[FRAME_QUEUE_SIZE]; + struct RxQueueGuard { + RxQueueGuard() { taskENTER_CRITICAL(); } + ~RxQueueGuard() { taskEXIT_CRITICAL(); } + }; + using RxQueue = mesh_nrf52::BleRxFrameQueue; + RxQueue recv_queue; void clearBuffers(); bool hasRequiredReplyCapacity() const; void shiftSendQueueLeft(); - void shiftRecvQueueLeft(); size_t writeBleUartFrame(const Frame& frame); void recoverStalledTx(const char* cause); void serviceTxRecovery(uint32_t now); @@ -156,7 +161,6 @@ public: _bonded_only = false; _bonded_only_configure_pending = false; send_queue_len = 0; - recv_queue_len = 0; } /** diff --git a/test/fixtures/nrf52_ble_rx_queue.cpp b/test/fixtures/nrf52_ble_rx_queue.cpp new file mode 100644 index 000000000..db91a1123 --- /dev/null +++ b/test/fixtures/nrf52_ble_rx_queue.cpp @@ -0,0 +1,477 @@ +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +static unsigned critical_depth = 0; +static std::deque> deferred; +static std::function during_copy; + +// Model a single-core RTOS: a preemption requested during a critical section +// runs when the outermost guard exits. All UART/logging operations assert that +// interrupts and task switching are enabled. +static void preempt(std::function operation) { + if (critical_depth) deferred.push_back(std::move(operation)); + else operation(); +} + +struct TestGuard { + TestGuard() { ++critical_depth; } + ~TestGuard() { + assert(critical_depth); + if (--critical_depth) return; + while (!deferred.empty()) { + auto operation = std::move(deferred.front()); + deferred.pop_front(); + operation(); + } + } +}; + +static void* guarded_copy(void* dest, const void* source, size_t length) { + assert(critical_depth); + const size_t first = length / 2; + std::memcpy(dest, source, first); + if (during_copy) { + auto operation = std::move(during_copy); + during_copy = nullptr; + preempt(std::move(operation)); + } + std::memcpy(static_cast(dest) + first, + static_cast(source) + first, length - first); + return dest; +} + +#define memcpy guarded_copy +#include +#undef memcpy + +#define MAX_FRAME_SIZE 176 +#define FRAME_QUEUE_SIZE 12 +#define COMPANION_RADIO_FULL 1 +#define COMPANION_FEATURE_BLE_MOTA_SOURCE 0 +#define BLE_CONN_HANDLE_INVALID 0xffff +#define BLE_DEBUG_PRINTLN(...) do { assert(!critical_depth); } while (0) +#define meshBleTrace(...) do { assert(!critical_depth); } while (0) + +enum { + BLE_GAP_EVT_CONNECTED = 1, BLE_GAP_EVT_DISCONNECTED, + BLE_GAP_EVT_CONN_SEC_UPDATE, BLE_GATTS_EVT_WRITE, + BLE_GATTS_EVT_SYS_ATTR_MISSING, BLE_GAP_EVT_CONN_PARAM_UPDATE_REQUEST, + BLE_GAP_ROLE_PERIPH = 1, BLE_UUID_TYPE_BLE = 1, + BLE_UUID_DESCRIPTOR_CLIENT_CHAR_CONFIG = 0x2902, + BLE_HCI_AUTHENTICATION_FAILURE = 5, + BLE_HCI_STATUS_CODE_PIN_OR_KEY_MISSING = 6, + BLE_HCI_CONN_TERMINATED_DUE_TO_MIC_FAILURE = 0x3d, + BLE_HCI_REMOTE_USER_TERMINATED_CONNECTION = 0x13, + NRF_SUCCESS = 0 +}; +struct ble_gap_addr_t { uint8_t addr[6] = {}; }; +struct ble_gap_evt_connected_t { + uint8_t role = BLE_GAP_ROLE_PERIPH; + ble_gap_addr_t peer_addr; +}; +struct ble_gap_evt_disconnected_t { uint8_t reason = 0; }; +struct ble_gatts_evt_write_t { + struct { uint8_t type = 0; uint16_t uuid = 0; } uuid; + uint16_t len = 0; + uint8_t data[2] = {}; +}; +struct ble_evt_t { + struct { uint16_t evt_id = 0; } header; + struct { + struct { uint16_t conn_handle = 7; } common_evt; + struct { + uint16_t conn_handle = 7; + struct { + ble_gap_evt_connected_t connected; + ble_gap_evt_disconnected_t disconnected; + } params; + } gap_evt; + struct { + uint16_t conn_handle = 7; + struct { ble_gatts_evt_write_t write; } params; + } gatts_evt; + } evt; +}; +struct BLEConnection { + bool connected() const { return true; } + bool secured() const { return true; } + bool bonded() const { return true; } +}; +struct { + BLEConnection connection; + BLEConnection* Connection(uint16_t) { return &connection; } +} Bluefruit; +static bool cccd_written_this_connection = false; +static struct { std::atomic valid{false}; } cccd_ram_cache; +static void captureCccdDeferred(uint16_t, ble_gap_addr_t*) {} +template static bool ada_callback(Args...) { return true; } +static uint32_t sd_ble_gap_disconnect(uint16_t, uint8_t) { return NRF_SUCCESS; } +static uint32_t sd_ble_gap_conn_param_update(uint16_t, const void*) { return NRF_SUCCESS; } +static uint32_t millis() { return 100; } + +struct SerialBLEInterface { + struct Frame { + uint8_t len = 0; + uint8_t buf[MAX_FRAME_SIZE] = {}; + Frame& operator=(const Frame& source) { + assert(critical_depth); + len = source.len; + guarded_copy(buf, source.buf, sizeof(buf)); + return *this; + } + }; + using RxQueueGuard = TestGuard; + using RxQueue = mesh_nrf52::BleRxFrameQueue; + RxQueue recv_queue; + uint8_t send_queue_len = 0; + unsigned long _last_retry_attempt = 0; + mesh::BleTxStallWatchdog _tx_stall_watchdog; + mesh::BleDisconnectRecovery _tx_disconnect_recovery; + std::atomic _companionDataSeen{false}; + std::atomic _pairingRequestPending{false}; + SecuritySessionTimer _security_timer; + SecuritySessionTimer _companion_start_timer; + ble_gap_addr_t _peer_address; + bool _peer_address_valid = false, _bond_removed_for_connection = false; + bool _bonded_only = false, _isDeviceConnected = false; + uint16_t _conn_handle = BLE_CONN_HANDLE_INVALID; + unsigned removed_bonds = 0, bonded_recoveries = 0; + bool connected = true; + struct Uart { + std::vector bytes; + std::function before_read, after_read, during_flush; + unsigned reads = 0, flushes = 0; + int available() const { + assert(!critical_depth); + return bytes.size(); + } + int read(uint8_t* destination, size_t length) { + assert(!critical_depth); + ++reads; + if (before_read) { + auto operation = std::move(before_read); + before_read = nullptr; + operation(); + } + length = std::min(length, bytes.size()); + if (length) std::memcpy(destination, bytes.data(), length); + bytes.erase(bytes.begin(), bytes.begin() + length); + if (after_read) { + auto operation = std::move(after_read); + after_read = nullptr; + operation(); + } + return length; + } + void flush() { + assert(!critical_depth); + ++flushes; + bytes.clear(); + if (during_flush) { + auto operation = std::move(during_flush); + during_flush = nullptr; + operation(); + } + } + } bleuart; + void open(uint16_t handle = 7) { + recv_queue.setEnabled(true); + recv_queue.connect(handle); + recv_queue.secure(handle); + } + bool isConnected() const { assert(!critical_depth); return connected; } + bool isValidConnection(uint16_t handle) const { return handle == _conn_handle; } + bool removeStoredBondForPeer(const char*) { + assert(!critical_depth && _peer_address_valid); + ++removed_bonds; + _bond_removed_for_connection = true; + return true; + } + void requestBondedOnlyRecovery(const char*) { + assert(!critical_depth); + ++bonded_recoveries; + } + void clearBuffers(); + static void onBLEEvent(ble_evt_t*); + static void onConnect(uint16_t); + static void onDisconnect(uint16_t, uint8_t); + static void onBleUartRX(uint16_t); + bool isReadBusy() const; + bool hasPendingIO() const; +}; + +static SerialBLEInterface* instance; +@METHODS@ + +using Queue = SerialBLEInterface::RxQueue; +using Result = Queue::PushResult; + +static SerialBLEInterface::Frame frame(uint32_t id, uint8_t length = MAX_FRAME_SIZE) { + SerialBLEInterface::Frame value; + value.len = length; + std::memset(value.buf, id % 251, sizeof(value.buf)); + std::memcpy(value.buf, &id, sizeof(id)); + return value; +} + +static void expect_pop(Queue& queue, uint32_t id, uint8_t length = MAX_FRAME_SIZE) { + uint8_t destination[MAX_FRAME_SIZE] = {}; + assert(queue.pop(destination) == length); + uint32_t actual; + std::memcpy(&actual, destination, sizeof(actual)); + assert(actual == id); + for (unsigned i = sizeof(id); i < length; ++i) assert(destination[i] == id % 251); +} + +static void ring_order_capacity_and_interleaved_copy() { + SerialBLEInterface link; + link.open(); + Queue& queue = link.recv_queue; + uint32_t generation; + assert(queue.snapshot(7, generation)); + for (unsigned i = 0; i < 12; ++i) + assert(queue.push(7, generation, frame(i)) == Result::Accepted); + assert(queue.push(7, generation, frame(12)) == Result::Full); + assert(queue.size() == 12 && link.isReadBusy() && link.hasPendingIO()); + + // Request producer execution halfway through the actual pop memcpy. Its + // append must run after the copy/dequeue completes, preserving the head. + during_copy = [&]() { + assert(queue.size() == 11); + assert(queue.push(7, generation, frame(12)) == Result::Accepted); + }; + expect_pop(queue, 0); + for (unsigned i = 1; i <= 12; ++i) expect_pop(queue, i); + assert(!link.isReadBusy() && !link.hasPendingIO()); + link.send_queue_len = 1; + assert(link.hasPendingIO()); + + // Exercise both indices wrapping through many complete rings. + for (uint32_t base = 100; base < 1300; base += 12) { + for (unsigned offset = 0; offset < 12; ++offset) + assert(queue.push(7, generation, frame(base + offset)) == Result::Accepted); + for (unsigned offset = 0; offset < 12; ++offset) expect_pop(queue, base + offset); + } + + assert(queue.push(7, generation, frame(99)) == Result::Accepted); + during_copy = [&]() { link.clearBuffers(); }; + expect_pop(queue, 99); + assert(queue.size() == 0 && !queue.current(7, generation)); + + assert(queue.snapshot(7, generation)); + // Dequeue attempted during a producer copy cannot inspect an unpublished + // slot. The custom Frame assignment checks that push guards its copy too. + during_copy = [&]() { expect_pop(queue, 88); }; + assert(queue.push(7, generation, frame(88)) == Result::Accepted); + assert(queue.size() == 0); +} + +static void session_reset_and_stale_handles() { + SerialBLEInterface link; + Queue& queue = link.recv_queue; + uint32_t generation; + assert(!queue.snapshot(7, generation)); + queue.setEnabled(true); + queue.connect(7); + assert(!queue.snapshot(7, generation)); + queue.secure(8); + assert(!queue.snapshot(7, generation)); + queue.secure(7); + assert(queue.snapshot(7, generation)); + const uint32_t old = generation; + assert(queue.push(7, old, frame(1)) == Result::Accepted); + queue.reset([&]() { + assert(!critical_depth && queue.size() == 0); + assert(!queue.snapshot(7, generation)); + assert(queue.push(7, old, frame(2)) == Result::Stale); + // Overlapping resets must keep RX suspended until BOTH flushes finish. + queue.reset([&]() { assert(!queue.snapshot(7, generation)); }); + assert(!queue.snapshot(7, generation)); + }); + assert(queue.snapshot(7, generation) && generation != old); + assert(queue.push(7, old, frame(2)) == Result::Stale); + assert(queue.push(7, generation, frame(3)) == Result::Accepted); + queue.disconnect(8); // A stale disconnect cannot reset the active handle. + expect_pop(queue, 3); + queue.disconnect(7); + queue.connect(7); // The SoftDevice can reuse a connection handle. + queue.secure(7); + assert(queue.push(7, generation, frame(4)) == Result::Stale); + assert(queue.snapshot(7, generation)); + queue.setEnabled(false); + queue.setEnabled(true); + queue.secure(7); // A delayed security callback must not revive that link. + assert(!queue.snapshot(7, generation)); + queue.connect(7); + queue.secure(7); + assert(queue.snapshot(7, generation)); + assert(queue.push(7, generation, frame(5, 4)) == Result::Accepted); + expect_pop(queue, 5, 4); + + // The BLE worker records connect before enqueuing/delegating the higher + // level callbacks. A later enable cannot revive that earlier event. + queue.setEnabled(false); + queue.connect(7); + queue.setEnabled(true); + link.clearBuffers(); // Delayed onConnect clears buffers, not session ID. + queue.secure(7); + assert(!queue.snapshot(7, generation)); + queue.connect(7); + queue.setEnabled(false); + queue.setEnabled(true); + link.clearBuffers(); + queue.secure(7); + assert(!queue.snapshot(7, generation)); + queue.connect(7); + queue.secure(7); + assert(queue.snapshot(7, generation)); +} + +static void callback_read_reset_and_reconnect() { + SerialBLEInterface link; + instance = &link; + link.open(); + link._companion_start_timer.start(100); + link.bleuart.bytes = {1, 2, 3, 4}; + SerialBLEInterface::onBleUartRX(7); + uint8_t destination[MAX_FRAME_SIZE] = {}; + assert(link.recv_queue.pop(destination) == 4 && destination[3] == 4); + assert(link._companionDataSeen && !link._companion_start_timer.pending()); + + // A reset after FIFO read but before queue commit invalidates the staged + // frame. New-session bytes remain for their own callback. + link.bleuart.bytes = {10, 11, 12}; + link.bleuart.after_read = [&]() { + link.recv_queue.setEnabled(false); + link.clearBuffers(); + link.open(); + link._companionDataSeen = false; + link._companion_start_timer.start(200); + link.bleuart.bytes = {20, 21}; + }; + SerialBLEInterface::onBleUartRX(7); + assert(link.recv_queue.size() == 0 && !link._companionDataSeen); + assert(link._companion_start_timer.pending() && link.bleuart.available() == 2); + SerialBLEInterface::onBleUartRX(7); + assert(link.recv_queue.pop(destination) == 2 && destination[0] == 20); + + link.bleuart.bytes = {30, 31}; + link.bleuart.before_read = [&]() { link.clearBuffers(); }; + SerialBLEInterface::onBleUartRX(7); + assert(link.recv_queue.size() == 0); // Short FIFO reads never publish bytes. + + link.bleuart.bytes = {40, 41}; + const unsigned reads = link.bleuart.reads; + SerialBLEInterface::onBleUartRX(8); + assert(link.bleuart.reads == reads && link.bleuart.available() == 2); + SerialBLEInterface::onBleUartRX(7); + assert(link.recv_queue.pop(destination) == 2 && destination[0] == 40); + + link.bleuart.bytes.assign(MAX_FRAME_SIZE + 1, 0x55); + SerialBLEInterface::onBleUartRX(7); + assert(link.recv_queue.size() == 0 && !link.bleuart.available()); + + for (unsigned i = 0; i < FRAME_QUEUE_SIZE; ++i) { + link.bleuart.bytes = {static_cast(i)}; + SerialBLEInterface::onBleUartRX(7); + } + link.bleuart.bytes = {99}; + SerialBLEInterface::onBleUartRX(7); + assert(link.recv_queue.size() == FRAME_QUEUE_SIZE && !link.bleuart.available()); + for (unsigned i = 0; i < FRAME_QUEUE_SIZE; ++i) + assert(link.recv_queue.pop(destination) == 1 && destination[0] == i); + + // FIFO flush can block/yield; none of its work may retain a task guard. + link.bleuart.during_flush = [&]() { + assert(!critical_depth); + uint32_t token; + assert(!link.recv_queue.snapshot(7, token)); + link.clearBuffers(); + assert(!link.recv_queue.snapshot(7, token)); + }; + link.clearBuffers(); + uint32_t token; + assert(link.recv_queue.snapshot(7, token)); + assert(!critical_depth && deferred.empty()); +} + +static void production_connection_events_preserve_disconnect_cleanup() { + for (uint8_t reason : {uint8_t(BLE_HCI_AUTHENTICATION_FAILURE), + uint8_t(BLE_HCI_STATUS_CODE_PIN_OR_KEY_MISSING), + uint8_t(BLE_HCI_CONN_TERMINATED_DUE_TO_MIC_FAILURE), + uint8_t(BLE_HCI_REMOTE_USER_TERMINATED_CONNECTION)}) { + SerialBLEInterface link; + instance = &link; + link.recv_queue.setEnabled(true); + link._bonded_only = true; + link._bond_removed_for_connection = true; + cccd_written_this_connection = true; + ble_evt_t event; + event.header.evt_id = BLE_GAP_EVT_CONNECTED; + event.evt.gap_evt.params.connected.peer_addr.addr[0] = 42; + SerialBLEInterface::onBLEEvent(&event); + assert(link._peer_address_valid && link._peer_address.addr[0] == 42); + assert(!link._bond_removed_for_connection && !cccd_written_this_connection); + SerialBLEInterface::onConnect(7); + assert(link._conn_handle == 7 && link._security_timer.pending()); + link.recv_queue.secure(7); + uint32_t generation; + assert(link.recv_queue.snapshot(7, generation)); + assert(link.recv_queue.push(7, generation, frame(42)) == Result::Accepted); + link._isDeviceConnected = true; + link._companion_start_timer.start(100); + link.send_queue_len = 3; + cccd_written_this_connection = true; + + event.header.evt_id = BLE_GAP_EVT_DISCONNECTED; + event.evt.gap_evt.params.disconnected.reason = reason; + SerialBLEInterface::onBLEEvent(&event); + assert(link.recv_queue.size() == 0); + assert(!link.recv_queue.current(7, generation)); + assert(link.recv_queue.push(7, generation, frame(43)) == Result::Stale); + assert(!link._peer_address_valid && !cccd_written_this_connection); + const unsigned authentication_failure = + reason != BLE_HCI_REMOTE_USER_TERMINATED_CONNECTION; + assert(link.removed_bonds == authentication_failure); + assert(link.bonded_recoveries == authentication_failure); + SerialBLEInterface::onDisconnect(7, reason); + assert(link._conn_handle == BLE_CONN_HANDLE_INVALID && !link._isDeviceConnected); + assert(!link._security_timer.pending() && !link._companion_start_timer.pending()); + assert(link.send_queue_len == 0 && link.recv_queue.size() == 0); + } + + // Run the actual event and deferred connect methods across an enable + // transition, so the test also verifies where session identity is assigned. + SerialBLEInterface link; + instance = &link; + link.recv_queue.setEnabled(true); + ble_evt_t event; + event.header.evt_id = BLE_GAP_EVT_CONNECTED; + SerialBLEInterface::onBLEEvent(&event); + link.recv_queue.setEnabled(false); + link.recv_queue.setEnabled(true); + SerialBLEInterface::onConnect(7); + link.recv_queue.secure(7); + uint32_t generation; + assert(!link.recv_queue.snapshot(7, generation)); + SerialBLEInterface::onBLEEvent(&event); // A new physical session can open. + SerialBLEInterface::onConnect(7); + link.recv_queue.secure(7); + assert(link.recv_queue.snapshot(7, generation)); +} + +int main() { + ring_order_capacity_and_interleaved_copy(); + session_reset_and_stale_handles(); + callback_read_reset_and_reconnect(); + production_connection_events_preserve_disconnect_cleanup(); +} diff --git a/test/test_nrf52_ble_backpressure.py b/test/test_nrf52_ble_backpressure.py index f47f5c801..9e27fa37e 100644 --- a/test/test_nrf52_ble_backpressure.py +++ b/test/test_nrf52_ble_backpressure.py @@ -22,6 +22,7 @@ HARNESS = r''' #include #include #include +#include #define MAX_FRAME_SIZE 176 #define FRAME_QUEUE_SIZE 12 @@ -33,8 +34,16 @@ uint32_t millis() { return current_millis; } struct SerialBLEInterface { struct Frame { uint8_t len = 0; uint8_t buf[MAX_FRAME_SIZE] = {}; }; - uint8_t send_queue_len = 0, recv_queue_len = 0; - Frame send_queue[FRAME_QUEUE_SIZE], recv_queue[FRAME_QUEUE_SIZE]; + struct RxQueueGuard {}; + using RxQueue = mesh_nrf52::BleRxFrameQueue; + uint8_t send_queue_len = 0; + Frame send_queue[FRAME_QUEUE_SIZE]; + RxQueue recv_queue; + SerialBLEInterface() { + recv_queue.setEnabled(true); + recv_queue.connect(1); + recv_queue.secure(1); + } unsigned long _last_retry_attempt = 0; mesh::BleTxStallWatchdog _tx_stall_watchdog; mesh::BleDisconnectRecovery _tx_disconnect_recovery; @@ -53,12 +62,15 @@ struct SerialBLEInterface { return frame.len; } void receive(const uint8_t* bytes, size_t size) { - assert(recv_queue_len < FRAME_QUEUE_SIZE && size <= MAX_FRAME_SIZE); - recv_queue[recv_queue_len].len = size; - memcpy(recv_queue[recv_queue_len++].buf, bytes, size); + assert(size <= MAX_FRAME_SIZE); + uint32_t generation; + assert(recv_queue.snapshot(1, generation)); + Frame frame; + frame.len = size; + memcpy(frame.buf, bytes, size); + assert(recv_queue.push(1, generation, frame) == RxQueue::PushResult::Accepted); } void shiftSendQueueLeft(); - void shiftRecvQueueLeft(); bool hasRequiredReplyCapacity() const; bool isWriteBusy() const; size_t writeFrame(const uint8_t*, size_t); @@ -78,19 +90,19 @@ static void full_required_queue() { link.receive(command, sizeof(command)); uint8_t received[MAX_FRAME_SIZE] = {}; assert(link.checkRecvFrame(received) == 0); - assert(link.recv_queue_len == 1 && link.send_queue_len == FRAME_QUEUE_SIZE); + assert(link.recv_queue.size() == 1 && link.send_queue_len == FRAME_QUEUE_SIZE); assert(link.write_attempts == 1 && link.transmitted.empty()); ++current_millis; assert(link.checkRecvFrame(received) == 0); assert(link.write_attempts == 1); // Existing 250 ms retry throttle is retained. - assert(link.recv_queue_len == 1 && link.isWriteBusy()); + assert(link.recv_queue.size() == 1 && link.isWriteBusy()); current_millis = 351; link.allow_write = true; // A completed TX frees admission on this SAME call, before RX is examined. assert(link.checkRecvFrame(received) == sizeof(command)); assert(memcmp(received, command, sizeof(command)) == 0); - assert(link.recv_queue_len == 0 && link.send_queue_len == FRAME_QUEUE_SIZE - 1); + assert(link.recv_queue.size() == 0 && link.send_queue_len == FRAME_QUEUE_SIZE - 1); const uint8_t reply[] = {29, 'A', '7', '|', 'O', 'K'}; assert(link.writeFrame(reply, sizeof(reply)) == sizeof(reply)); for (unsigned i = 0; i < FRAME_QUEUE_SIZE + 2; ++i) { @@ -101,7 +113,7 @@ static void full_required_queue() { for (uint8_t i = 0; i < FRAME_QUEUE_SIZE; ++i) assert((link.transmitted[i] == std::vector{0x0d, i})); assert(link.transmitted.back() == std::vector(reply, reply + sizeof(reply))); - assert(link.send_queue_len == 0 && link.recv_queue_len == 0); + assert(link.send_queue_len == 0 && link.recv_queue.size() == 0); } static void best_effort_eviction() { @@ -149,7 +161,7 @@ static void producer_threshold_and_recovery_ordering() { link._tx_disconnect_recovery.begin(); uint8_t received[MAX_FRAME_SIZE] = {}; assert(link.checkRecvFrame(received) == 0); - assert(link.recovery_services == 1 && link.write_attempts == 0 && link.recv_queue_len == 1); + assert(link.recovery_services == 1 && link.write_attempts == 0 && link.recv_queue.size() == 1); SerialBLEInterface partial; partial.partial_write = true; @@ -157,7 +169,7 @@ static void producer_threshold_and_recovery_ordering() { partial.receive(command, sizeof(command)); assert(partial.checkRecvFrame(received) == 0); assert(partial.recoveries == 1 && partial.write_attempts == 1); - assert(partial.recv_queue_len == 1); // Recovery is invoked instead of dispatch. + assert(partial.recv_queue.size() == 1); // Recovery is invoked instead of dispatch. } int main() { @@ -172,7 +184,6 @@ class Nrf52BleBackpressureTests(unittest.TestCase): def test_real_queue_dispatch_throttle_and_recovery_under_sanitizers(self): source = SOURCE.read_text() signatures = ["void SerialBLEInterface::shiftSendQueueLeft()", - "void SerialBLEInterface::shiftRecvQueueLeft()", "size_t SerialBLEInterface::writeFrame(", "size_t SerialBLEInterface::checkRecvFrame(", "bool SerialBLEInterface::isWriteBusy() const"] diff --git a/test/test_nrf52_ble_rx_queue.py b/test/test_nrf52_ble_rx_queue.py new file mode 100644 index 000000000..626d116d3 --- /dev/null +++ b/test/test_nrf52_ble_rx_queue.py @@ -0,0 +1,56 @@ +#!/usr/bin/env python3 +"""Exercise the production RX ring and callback with task preemption hooks.""" + +from pathlib import Path +import os +import subprocess +import tempfile +import unittest + +from test_t096_full_memory import method + +ROOT = Path(__file__).resolve().parents[1] + + +class Nrf52BleRxQueueTests(unittest.TestCase): + def test_session_identity_is_observed_before_deferred_lifecycle_callbacks(self): + source = (ROOT / "src/helpers/nrf52/SerialBLEInterface.cpp").read_text() + events = method(source, "void SerialBLEInterface::onBLEEvent(") + self.assertIn("recv_queue.connect(evt->evt.gap_evt.conn_handle)", events) + self.assertIn("recv_queue.disconnect(evt->evt.gap_evt.conn_handle)", events) + for signature in ("void SerialBLEInterface::onConnect(", + "void SerialBLEInterface::onDisconnect("): + lifecycle = method(source, signature) + self.assertNotIn("recv_queue.connect(", lifecycle) + self.assertNotIn("recv_queue.disconnect(", lifecycle) + + def test_production_queue_and_callback_interleavings_under_sanitizers(self): + source = (ROOT / "src/helpers/nrf52/SerialBLEInterface.cpp").read_text() + signatures = ["void SerialBLEInterface::clearBuffers()", + "static bool isBondAuthenticationFailure(", + "void SerialBLEInterface::onBLEEvent(", + "void SerialBLEInterface::onConnect(", + "void SerialBLEInterface::onDisconnect(", + "void SerialBLEInterface::onBleUartRX(", + "bool SerialBLEInterface::isReadBusy() const", + "bool SerialBLEInterface::hasPendingIO() const"] + methods = "\n".join(method(source, signature) for signature in signatures) + harness = (ROOT / "test/fixtures/nrf52_ble_rx_queue.cpp").read_text() + sanitizers = [] if os.name == "nt" else [ + "-fsanitize=address,undefined", "-fno-sanitize-recover=all", + "-fno-pie", "-no-pie"] + with tempfile.TemporaryDirectory(prefix="meshcore-nrf-ble-rx-") as temp: + binary = Path(temp) / "rx_queue" + built = subprocess.run([ + "c++", "-std=c++17", "-x", "c++", "-", "-I" + str(ROOT / "src"), + *sanitizers, "-o", str(binary)], + input=harness.replace("@METHODS@", methods), + text=True, capture_output=True, timeout=30) + self.assertEqual(built.returncode, 0, built.stderr) + result = subprocess.run([str(binary)], text=True, + capture_output=True, timeout=10) + self.assertEqual(result.returncode, 0, result.stderr) + + +if __name__ == "__main__": + unittest.main()