From 1332df0461b7335f0144fd34bf18041721d39dae Mon Sep 17 00:00:00 2001 From: liquidraver <504870+liquidraver@users.noreply.github.com> Date: Sun, 22 Feb 2026 15:19:03 +0100 Subject: [PATCH] future-proof lora rx queue --- zephcore/adapters/radio/LoRaRadioBase.cpp | 17 ++- zephcore/adapters/radio/radio_common.h | 8 +- zephcore/src/Dispatcher.cpp | 128 +++++++++++----------- 3 files changed, 86 insertions(+), 67 deletions(-) diff --git a/zephcore/adapters/radio/LoRaRadioBase.cpp b/zephcore/adapters/radio/LoRaRadioBase.cpp index 6969277..c43ca03 100644 --- a/zephcore/adapters/radio/LoRaRadioBase.cpp +++ b/zephcore/adapters/radio/LoRaRadioBase.cpp @@ -152,11 +152,22 @@ void LoRaRadioBase::rxCallbackStatic(const struct device *dev, uint8_t *data, LOG_DBG("RX callback: size=%u rssi=%d snr=%d", size, rssi, snr); - /* Ring buffer write */ + /* Ring buffer write — SPSC: only ISR writes _rx_head, only main + * thread writes _rx_tail. On overflow, drop the NEW packet to + * preserve this invariant (ISR must never touch _rx_tail). */ uint8_t next_head = (self->_rx_head + 1) % RX_RING_SIZE; if (next_head == self->_rx_tail) { - LOG_WRN("RX ring full, dropping oldest packet"); - self->_rx_tail = (self->_rx_tail + 1) % RX_RING_SIZE; + LOG_WRN("RX ring full, dropping new packet"); + self->_packets_recv_errors++; + /* Still restore duty cycle and notify — main loop should + * drain faster next time. */ + if (self->_rx_duty_cycle_enabled) { + self->hwSetRxDutyCycle(true); + } + if (self->_rx_cb) { + self->_rx_cb(self->_rx_cb_user_data); + } + return; } RxPacket *pkt = &self->_rx_ring[self->_rx_head]; diff --git a/zephcore/adapters/radio/radio_common.h b/zephcore/adapters/radio/radio_common.h index 2446c44..9113e13 100644 --- a/zephcore/adapters/radio/radio_common.h +++ b/zephcore/adapters/radio/radio_common.h @@ -16,8 +16,12 @@ #define NOISE_FLOOR_SAMPLING_THRESHOLD 14 /* only sample if rssi < floor + threshold */ #define DEFAULT_NOISE_FLOOR 0 /* first calibration accepts all samples */ -/* --- RX ring buffer --- */ -#define RX_RING_SIZE 4 +/* --- RX ring buffer --- + * Sized to be effectively drop-proof: even at the fastest LoRa settings + * (SF7/BW500, ~5ms per packet), 32 slots buffer 160ms+ of back-to-back + * arrivals. The main loop drain takes microseconds per packet, so + * overflow should never occur in practice. Cost: 32 × 260 = ~8.3 KB. */ +#define RX_RING_SIZE 32 /* --- TX wait thread --- */ #define TX_WAIT_THREAD_STACK_SIZE 1024 diff --git a/zephcore/src/Dispatcher.cpp b/zephcore/src/Dispatcher.cpp index d01ed79..c0ca9a0 100644 --- a/zephcore/src/Dispatcher.cpp +++ b/zephcore/src/Dispatcher.cpp @@ -152,74 +152,78 @@ void Dispatcher::loop() void Dispatcher::checkRecv() { - Packet *pkt = nullptr; - float score = 0.0f; - uint32_t air_time = 0; - - { + /* Drain ALL queued LoRa packets per wake. + * k_event is a bitfield (not a counter), so multiple ISR arrivals + * may only produce one wake. We must empty the ring each time. */ + for (;;) { uint8_t raw[MAX_TRANS_UNIT + 1]; int len = _radio->recvRaw(raw, MAX_TRANS_UNIT); - if (len > 0) { - LOG_INF("checkRecv: got raw packet len=%d", len); - logRxRaw(_radio->getLastSNR(), _radio->getLastRSSI(), raw, len); - pkt = _mgr->allocNew(); - if (pkt == nullptr) { - LOG_WRN("checkRecv: packet alloc failed"); - return; - } - LOG_INF("checkRecv: parsing packet"); - int i = 0; - pkt->header = raw[i++]; - if (pkt->hasTransportCodes()) { - memcpy(&pkt->transport_codes[0], &raw[i], 2); i += 2; - memcpy(&pkt->transport_codes[1], &raw[i], 2); i += 2; - } else { - pkt->transport_codes[0] = pkt->transport_codes[1] = 0; - } - pkt->path_len = raw[i++]; + if (len <= 0) { + break; /* ring empty — done */ + } - if (pkt->path_len > MAX_PATH_SIZE || i + pkt->path_len > len) { - LOG_WRN("checkRecv: bad path_len=%d", pkt->path_len); - _mgr->free(pkt); - pkt = nullptr; - } else { - memcpy(pkt->path, &raw[i], pkt->path_len); - i += pkt->path_len; - pkt->payload_len = len - i; - if (pkt->payload_len > (int)sizeof(pkt->payload)) { - LOG_WRN("checkRecv: payload too large %d", pkt->payload_len); - _mgr->free(pkt); - pkt = nullptr; - } else { - memcpy(pkt->payload, &raw[i], pkt->payload_len); - pkt->_snr = (int8_t)(_radio->getLastSNR() * 4.0f); - score = _radio->packetScore(_radio->getLastSNR(), len); - air_time = _radio->getEstAirtimeFor(len); - rx_air_time += air_time; - LOG_INF("checkRecv: header=0x%02x type=%d route=%s path_len=%d payload_len=%d", - pkt->header, pkt->getPayloadType(), - pkt->isRouteDirect() ? "direct" : "flood", - pkt->path_len, pkt->payload_len); - /* Log path hashes to identify forwarding nodes */ - if (pkt->path_len > 0) { - LOG_INF("checkRecv: path[0]=0x%02x%s%s", - pkt->path[0], - pkt->path_len > 1 ? " path[1]=0x" : "", - pkt->path_len > 1 ? "" : ""); - if (pkt->path_len > 1) { - LOG_INF(" path bytes: %02x %02x %02x %02x", - pkt->path[0], - pkt->path_len > 1 ? pkt->path[1] : 0, - pkt->path_len > 2 ? pkt->path[2] : 0, - pkt->path_len > 3 ? pkt->path[3] : 0); - } - } - } + LOG_INF("checkRecv: got raw packet len=%d", len); + logRxRaw(_radio->getLastSNR(), _radio->getLastRSSI(), raw, len); + + Packet *pkt = _mgr->allocNew(); + if (pkt == nullptr) { + LOG_WRN("checkRecv: packet alloc failed"); + break; + } + + float score = 0.0f; + uint32_t air_time = 0; + + LOG_INF("checkRecv: parsing packet"); + int i = 0; + pkt->header = raw[i++]; + if (pkt->hasTransportCodes()) { + memcpy(&pkt->transport_codes[0], &raw[i], 2); i += 2; + memcpy(&pkt->transport_codes[1], &raw[i], 2); i += 2; + } else { + pkt->transport_codes[0] = pkt->transport_codes[1] = 0; + } + pkt->path_len = raw[i++]; + + if (pkt->path_len > MAX_PATH_SIZE || i + pkt->path_len > len) { + LOG_WRN("checkRecv: bad path_len=%d", pkt->path_len); + _mgr->free(pkt); + continue; + } + + memcpy(pkt->path, &raw[i], pkt->path_len); + i += pkt->path_len; + pkt->payload_len = len - i; + if (pkt->payload_len > (int)sizeof(pkt->payload)) { + LOG_WRN("checkRecv: payload too large %d", pkt->payload_len); + _mgr->free(pkt); + continue; + } + + memcpy(pkt->payload, &raw[i], pkt->payload_len); + pkt->_snr = (int8_t)(_radio->getLastSNR() * 4.0f); + score = _radio->packetScore(_radio->getLastSNR(), len); + air_time = _radio->getEstAirtimeFor(len); + rx_air_time += air_time; + LOG_INF("checkRecv: header=0x%02x type=%d route=%s path_len=%d payload_len=%d", + pkt->header, pkt->getPayloadType(), + pkt->isRouteDirect() ? "direct" : "flood", + pkt->path_len, pkt->payload_len); + /* Log path hashes to identify forwarding nodes */ + if (pkt->path_len > 0) { + LOG_INF("checkRecv: path[0]=0x%02x%s%s", + pkt->path[0], + pkt->path_len > 1 ? " path[1]=0x" : "", + pkt->path_len > 1 ? "" : ""); + if (pkt->path_len > 1) { + LOG_INF(" path bytes: %02x %02x %02x %02x", + pkt->path[0], + pkt->path_len > 1 ? pkt->path[1] : 0, + pkt->path_len > 2 ? pkt->path[2] : 0, + pkt->path_len > 3 ? pkt->path[3] : 0); } } - } - if (pkt) { #if IS_ENABLED(CONFIG_ZEPHCORE_PACKET_LOGGING) /* Arduino-compatible packet logging - use printk to bypass log level filtering */ {