dandri/ZephCore
1
0
Fork 0
mirror of https://github.com/liquidraver/ZephCore.git synced 2026-09-01 21:08:19 +00:00
Code Issues Packages Projects Releases Wiki Activity

future-proof lora rx queue

Browse Source
This commit is contained in:
liquidraver
2026-02-22 15:19:03 +01:00
parent 27e31f7274
commit 1332df0461
3 changed files with 86 additions and 67 deletions
Download Patch File Download Diff File Expand all files Collapse all files
zephcore/adapters/radio/LoRaRadioBase.cpp
+14 -3
View File
@@ -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];
zephcore/adapters/radio/radio_common.h
+6 -2
View File
@@ -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
zephcore/src/Dispatcher.cpp
+66 -62
View File
@@ -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 */
{