From f3afb6433d7543fbdcf24ba8914a02320d054345 Mon Sep 17 00:00:00 2001 From: liquidraver <504870+liquidraver@users.noreply.github.com> Date: Sun, 22 Feb 2026 18:39:49 +0100 Subject: [PATCH] rework tx cad --- zephcore/src/Dispatcher.cpp | 37 +++++++++++++++------ zephcore/src/StaticPoolPacketManager.cpp | 41 ++++++++++++++++++++++-- 2 files changed, 66 insertions(+), 12 deletions(-) diff --git a/zephcore/src/Dispatcher.cpp b/zephcore/src/Dispatcher.cpp index c0ca9a0..8e6908f 100644 --- a/zephcore/src/Dispatcher.cpp +++ b/zephcore/src/Dispatcher.cpp @@ -295,11 +295,15 @@ void Dispatcher::checkSend() int count = _mgr->getOutboundCount(now); if (count == 0) return; - if (!millisHasNowPassed(next_tx_time)) { - LOG_DBG("checkSend: waiting for tx_time (count=%d)", count); - return; - } if (_radio->isReceiving()) { + /* Channel busy — enforce retry timer so we don't hammer the check */ + if (!millisHasNowPassed(next_tx_time)) { + if (_tx_queued_cb) { + uint32_t remaining = next_tx_time - now; + _tx_queued_cb(remaining + 1, _tx_queued_user_data); + } + return; + } if (cad_busy_start == 0) { cad_busy_start = now; LOG_INF("checkSend: channel busy, starting wait"); @@ -308,7 +312,11 @@ void Dispatcher::checkSend() _err_flags |= ERR_EVENT_CAD_TIMEOUT; LOG_WRN("checkSend: CAD timeout exceeded"); } else { - next_tx_time = futureMillis((int)getCADFailRetryDelay()); + uint32_t retry = getCADFailRetryDelay(); + next_tx_time = futureMillis((int)retry); + if (_tx_queued_cb) { + _tx_queued_cb(retry + 1, _tx_queued_user_data); + } return; } } @@ -323,6 +331,9 @@ void Dispatcher::checkSend() outbound->getPayloadType()); _mgr->queueOutbound(outbound, 0, futureMillis(5000)); outbound = nullptr; + if (_tx_queued_cb) { + _tx_queued_cb(5000, _tx_queued_user_data); + } return; } LOG_INF("checkSend: got outbound type=%d payload_len=%d", outbound->getPayloadType(), outbound->payload_len); @@ -370,19 +381,27 @@ void Dispatcher::checkSend() * isReceiving() and actual TX start (serialisation + * logging can take 1-5 ms). */ if (_radio->isReceiving()) { - LOG_INF("checkSend: channel busy at TX commit, re-queuing"); - _mgr->queueOutbound(outbound, 0, futureMillis((int)getCADFailRetryDelay())); + uint32_t retry = getCADFailRetryDelay(); + LOG_INF("checkSend: channel busy at TX commit, re-queuing delay=%u", retry); + _mgr->queueOutbound(outbound, 0, futureMillis((int)retry)); outbound = nullptr; + if (_tx_queued_cb) { + _tx_queued_cb(retry, _tx_queued_user_data); + } return; } LOG_INF("checkSend: calling startSendRaw len=%d", len); bool success = _radio->startSendRaw(raw, len); if (!success) { - LOG_ERR("checkSend: startSendRaw failed!"); + uint32_t retry = getCADFailRetryDelay(); + LOG_ERR("checkSend: startSendRaw failed! re-queuing delay=%u", retry); logTxFail(outbound, outbound->getRawLength()); - releasePacket(outbound); + _mgr->queueOutbound(outbound, 0, futureMillis((int)retry)); outbound = nullptr; + if (_tx_queued_cb) { + _tx_queued_cb(retry, _tx_queued_user_data); + } } else { LOG_INF("checkSend: TX started, max_airtime=%u", max_airtime); outbound_expiry = futureMillis((int)max_airtime); diff --git a/zephcore/src/StaticPoolPacketManager.cpp b/zephcore/src/StaticPoolPacketManager.cpp index 31bf1d1..3a60703 100644 --- a/zephcore/src/StaticPoolPacketManager.cpp +++ b/zephcore/src/StaticPoolPacketManager.cpp @@ -7,10 +7,13 @@ #include #include +#include +LOG_MODULE_REGISTER(zephcore_pktpool, CONFIG_ZEPHCORE_LORA_LOG_LEVEL); + namespace mesh { -#define POOL_SIZE 16 -#define QUEUE_SIZE 16 +#define POOL_SIZE 32 +#define QUEUE_SIZE 32 struct PacketQueue { Packet *_table[QUEUE_SIZE]; @@ -74,6 +77,21 @@ struct PacketQueue { return true; } + /* Find index of the entry with the highest priority number (least + * important). Returns -1 when the queue is empty. */ + int findLowestPriority() const { + if (_num == 0) return -1; + uint8_t worst = 0; + int idx = 0; + for (int j = 0; j < _num; j++) { + if (_pri_table[j] >= worst) { + worst = _pri_table[j]; + idx = j; + } + } + return idx; + } + int count() const { return _num; } Packet *itemAt(int i) const { return (i < _num) ? _table[i] : nullptr; } }; @@ -105,7 +123,22 @@ void StaticPoolPacketManager::free(Packet *packet) void StaticPoolPacketManager::queueOutbound(Packet *packet, uint8_t priority, uint32_t scheduled_for) { - if (!_send_queue.add(packet, priority, scheduled_for)) { + if (_send_queue.add(packet, priority, scheduled_for)) return; + + /* Queue full — evict the least-important entry to make room. + * Only evict if the new packet is higher priority (lower number). */ + int worst = _send_queue.findLowestPriority(); + if (worst >= 0 && _send_queue._pri_table[worst] > priority) { + uint8_t evicted_pri = _send_queue._pri_table[worst]; + Packet *evicted = _send_queue.removeByIdx(worst); + LOG_WRN("queueOutbound: FULL — evicted type=%d pri=%d for type=%d pri=%d", + evicted->getPayloadType(), evicted_pri, + packet->getPayloadType(), priority); + free(evicted); + _send_queue.add(packet, priority, scheduled_for); + } else { + LOG_WRN("queueOutbound: FULL (%d entries) — dropping type=%d pri=%d", + _send_queue.count(), packet->getPayloadType(), priority); free(packet); } } @@ -138,6 +171,8 @@ Packet *StaticPoolPacketManager::removeOutboundByIdx(int i) void StaticPoolPacketManager::queueInbound(Packet *packet, uint32_t scheduled_for) { if (!_rx_queue.add(packet, 0, scheduled_for)) { + LOG_WRN("queueInbound: FULL (%d entries) — dropping type=%d", + _rx_queue.count(), packet->getPayloadType()); free(packet); } }