From d45fbf7027384f56469eb5a14295ab7e2bfd8896 Mon Sep 17 00:00:00 2001 From: liquidraver <504870+liquidraver@users.noreply.github.com> Date: Wed, 11 Mar 2026 10:06:52 +0100 Subject: [PATCH] reactive collision avoidance --- README.md | 22 +++ zephcore/CMakeLists.txt | 1 + zephcore/app/CompanionMesh.cpp | 20 ++- zephcore/app/RepeaterDataStore.cpp | 10 +- zephcore/app/RepeaterMesh.cpp | 41 ++--- zephcore/app/RepeaterMesh.h | 13 +- zephcore/helpers/CommonCLI.cpp | 55 +++--- zephcore/helpers/CommonCLI.h | 5 + zephcore/helpers/NodePrefs.h | 1 + zephcore/include/mesh/ContentionTracker.h | 90 ++++++++++ zephcore/include/mesh/Dispatcher.h | 5 + zephcore/include/mesh/Mesh.h | 7 + .../include/mesh/StaticPoolPacketManager.h | 2 + zephcore/src/ContentionTracker.cpp | 161 ++++++++++++++++++ zephcore/src/Dispatcher.cpp | 13 +- zephcore/src/Mesh.cpp | 41 ++++- zephcore/src/StaticPoolPacketManager.cpp | 20 +++ 17 files changed, 439 insertions(+), 68 deletions(-) create mode 100644 zephcore/include/mesh/ContentionTracker.h create mode 100644 zephcore/src/ContentionTracker.cpp diff --git a/README.md b/README.md index 9c5d611..abb4a9b 100644 --- a/README.md +++ b/README.md @@ -116,6 +116,28 @@ All code paths are event-driven. The CPU sleeps in WFI between events. | Configuration | `platformio.ini` + `variant.h` per board | Kconfig + devicetree overlays, hierarchical config inheritance | | Threading | Single `loop()` + ISRs | Explicit threads (main mesh, TX wait) + system work queue | +### Adaptive Contention Window (ZephCore-only) + +Arduino MeshCore uses three static delay knobs (`txdelay`, `rxdelay`, `direct.txdelay`) that add the same retransmit jitter regardless of local conditions. In a linear chain of repeaters where each only hears its neighbor, this adds latency for zero benefit. In dense areas with 50+ neighbors, the same value may be too low to avoid collisions. + +ZephCore replaces all three with a self-tuning system based on **observed retransmit contention**: + +1. **Dupe counting**: When a node retransmits a flood packet, it counts how many times it hears that same packet retransmitted by neighbors within a 10-second window. This is a direct measurement of local contention -- 0 dupes means a quiet linear chain, 15+ means a dense cluster. + +2. **EMA-based delay sizing**: Dupe counts feed into a rolling exponential moving average. This drives a sqrt-curve delay factor for future retransmits: near-zero delay in sparse areas, scaling up in dense ones. At ~15 dupes (moderate density), the factor matches the old Arduino default of 0.5. + +3. **Reactive per-packet backoff**: When a node is waiting to retransmit and hears a neighbor retransmit the same packet, it pushes its own TX back by a random amount (up to `backoff.multiplier` x airtime). This is real-time CSMA -- you hear the channel being used for your packet, so you defer. + +**Direct packets** (routed, single next-hop) use minimal fixed jitter (~0-45ms) instead of adaptive delay, since only the next hop retransmits them. + +The old `txdelay`, `rxdelay`, and `direct.txdelay` commands are still accepted for binary compatibility with Arduino prefs but are ignored -- the system is fully adaptive. + +**CLI commands:** +- `get txdelay` -- shows adaptive status: contention estimate and current flood delay factor +- `get/set backoff.multiplier` -- reactive backoff cap (default 0.5, range 0.0-2.0). Set to 0 to disable reactive backoff (EMA window still works). Higher values allow more per-packet deferral in dense areas. + +**Compatibility**: Purely local behavior, no wire protocol changes. Works alongside Arduino MeshCore repeaters -- their retransmits are counted as dupes just the same. + ## Power Saving - **LoRa RX duty cycle**: CAD-based receive windowing reduces LoRa RX current from ~10-15mA to ~3-5mA (configurable via `CONFIG_ZEPHCORE_LORA_RX_DUTY_CYCLE`) diff --git a/zephcore/CMakeLists.txt b/zephcore/CMakeLists.txt index 24b82dd..040e92b 100644 --- a/zephcore/CMakeLists.txt +++ b/zephcore/CMakeLists.txt @@ -358,6 +358,7 @@ target_include_directories(app PRIVATE # ========== Core Sources (both roles) ========== target_sources(app PRIVATE + src/ContentionTracker.cpp src/Dispatcher.cpp src/Identity.cpp src/Mesh.cpp diff --git a/zephcore/app/CompanionMesh.cpp b/zephcore/app/CompanionMesh.cpp index a57eb4a..662efca 100644 --- a/zephcore/app/CompanionMesh.cpp +++ b/zephcore/app/CompanionMesh.cpp @@ -1111,17 +1111,27 @@ void CompanionMesh::onRawDataRecv(mesh::Packet *packet) sendPush(buf[0], &buf[1], i - 1); } -/* Dispatcher tuning - hard-coded for companion repeat mode (matches Arduino) */ uint32_t CompanionMesh::getRetransmitDelay(const mesh::Packet *packet) { - uint32_t t = (_radio->getEstAirtimeFor(packet->getPathByteLen() + packet->payload_len + 2) * 0.5f); - return getRNG()->nextInt(0, 7 * t + 1); + float factor = getContentionTracker().getFloodDelayFactor(); + uint32_t t = (uint32_t)(_radio->getEstAirtimeFor( + packet->getPathByteLen() + packet->payload_len + 2) * factor); + uint32_t max_jitter = 5 * t; + /* Cap jitter to 2000ms to avoid excessive latency in very dense areas. + * Reactive backoff will fine-tune further if needed. */ + if (max_jitter > 2000) max_jitter = 2000; + /* Floor: give downstream nodes time to finish RX processing + * and return to RX mode before we TX (~20ms settle) */ + return 20 + getRNG()->nextInt(0, max_jitter + 1); } uint32_t CompanionMesh::getDirectRetransmitDelay(const mesh::Packet *packet) { - uint32_t t = (_radio->getEstAirtimeFor(packet->getPathByteLen() + packet->payload_len + 2) * 0.2f); - return getRNG()->nextInt(0, 7 * t + 1); + uint32_t t = _radio->getEstAirtimeFor( + packet->getPathByteLen() + packet->payload_len + 2); + /* Floor: give downstream nodes time to finish RX processing + * and return to RX mode before we TX (~20ms settle + jitter) */ + return 20 + getRNG()->nextInt(0, t / 10 + 1); } uint8_t CompanionMesh::getDutyCyclePercent() const diff --git a/zephcore/app/RepeaterDataStore.cpp b/zephcore/app/RepeaterDataStore.cpp index 7f09476..58dc6ca 100644 --- a/zephcore/app/RepeaterDataStore.cpp +++ b/zephcore/app/RepeaterDataStore.cpp @@ -147,7 +147,7 @@ bool RepeaterDataStore::loadPrefs(NodePrefs& prefs) { fs_read(&file, &prefs.tx_delay_factor, sizeof(prefs.tx_delay_factor)); fs_read(&file, &prefs.guest_password, sizeof(prefs.guest_password)); fs_read(&file, &prefs.direct_tx_delay_factor, sizeof(prefs.direct_tx_delay_factor)); - fs_read(&file, pad, 4); + fs_read(&file, &prefs.backoff_multiplier, sizeof(prefs.backoff_multiplier)); fs_read(&file, &prefs.sf, sizeof(prefs.sf)); fs_read(&file, &prefs.cr, sizeof(prefs.cr)); fs_read(&file, &prefs.allow_read_only, sizeof(prefs.allow_read_only)); @@ -169,6 +169,12 @@ bool RepeaterDataStore::loadPrefs(NodePrefs& prefs) { fs_read(&file, prefs.owner_info, sizeof(prefs.owner_info)); fs_close(&file); + + /* Migrate uninitialized backoff_multiplier (0.0 or NaN) to default */ + if (prefs.backoff_multiplier == 0.0f || prefs.backoff_multiplier != prefs.backoff_multiplier) { + prefs.backoff_multiplier = 0.5f; + } + LOG_INF("Loaded prefs from %s", path); LOG_INF(" name='%s' freq=%.3f sf=%u bw=%.1f tx_pwr=%d", prefs.node_name, (double)prefs.freq, prefs.sf, (double)prefs.bw, prefs.tx_power_dbm); @@ -224,7 +230,7 @@ bool RepeaterDataStore::savePrefs(const NodePrefs& prefs) { fs_write(&file, &prefs.tx_delay_factor, sizeof(prefs.tx_delay_factor)); fs_write(&file, &prefs.guest_password, sizeof(prefs.guest_password)); fs_write(&file, &prefs.direct_tx_delay_factor, sizeof(prefs.direct_tx_delay_factor)); - fs_write(&file, pad, 4); + fs_write(&file, &prefs.backoff_multiplier, sizeof(prefs.backoff_multiplier)); fs_write(&file, &prefs.sf, sizeof(prefs.sf)); fs_write(&file, &prefs.cr, sizeof(prefs.cr)); fs_write(&file, &prefs.allow_read_only, sizeof(prefs.allow_read_only)); diff --git a/zephcore/app/RepeaterMesh.cpp b/zephcore/app/RepeaterMesh.cpp index 5c559d0..cfb86b0 100644 --- a/zephcore/app/RepeaterMesh.cpp +++ b/zephcore/app/RepeaterMesh.cpp @@ -485,35 +485,25 @@ void RepeaterMesh::logTxFail(mesh::Packet* pkt, int len) { } } -/* Fast base^x approximation using IEEE 754 float bit tricks. - * Avoids linking pow()/powf() (~1.9KB). Accuracy ~5% which is - * more than sufficient for RX delay jitter calculation. */ -static float fast_powf(float base, float exp) -{ - /* log2(base) via IEEE 754: float bits ≈ 2^23 * (log2(x) + 127) */ - union { float f; uint32_t i; } bx = { .f = base }; - float log2_base = (float)(int32_t)(bx.i - 0x3F800000) * (1.0f / 8388608.0f); /* 1/2^23 */ - - /* exp2(exp * log2_base) via IEEE 754 */ - float y = exp * log2_base; - union { float f; uint32_t i; } ex; - ex.i = (uint32_t)((int32_t)(y * 8388608.0f) + 0x3F800000); - return ex.f; -} - -int RepeaterMesh::calcRxDelay(float score, uint32_t air_time) const { - if (_prefs.rx_delay_base <= 0.0f) return 0; - return (int)((fast_powf(_prefs.rx_delay_base, 0.85f - score) - 1.0f) * air_time); -} - uint32_t RepeaterMesh::getRetransmitDelay(const mesh::Packet* packet) { - uint32_t t = (_radio->getEstAirtimeFor(packet->getPathByteLen() + packet->payload_len + 2) * _prefs.tx_delay_factor); - return getRNG()->nextInt(0, 7 * t + 1); + float factor = getContentionTracker().getFloodDelayFactor(); + uint32_t t = (uint32_t)(_radio->getEstAirtimeFor( + packet->getPathByteLen() + packet->payload_len + 2) * factor); + uint32_t max_jitter = 5 * t; + /* Cap jitter to 2000ms to avoid excessive latency in very dense areas. + * Reactive backoff will fine-tune further if needed. */ + if (max_jitter > 2000) max_jitter = 2000; + /* Floor: give downstream nodes time to finish RX processing + * and return to RX mode before we TX (~20ms settle) */ + return 20 + getRNG()->nextInt(0, max_jitter + 1); } uint32_t RepeaterMesh::getDirectRetransmitDelay(const mesh::Packet* packet) { - uint32_t t = (_radio->getEstAirtimeFor(packet->getPathByteLen() + packet->payload_len + 2) * _prefs.direct_tx_delay_factor); - return getRNG()->nextInt(0, 7 * t + 1); + uint32_t t = _radio->getEstAirtimeFor( + packet->getPathByteLen() + packet->payload_len + 2); + /* Floor: give downstream nodes time to finish RX processing + * and return to RX mode before we TX (~20ms settle + jitter) */ + return 20 + getRNG()->nextInt(0, t / 10 + 1); } bool RepeaterMesh::filterRecvFloodPacket(mesh::Packet* pkt) { @@ -816,6 +806,7 @@ void RepeaterMesh::begin(RepeaterDataStore* store) { * NOTE: Identity is loaded in main_repeater.cpp before begin() is called, * so we skip loading it here (self_id should already be set). */ _store->loadPrefs(_prefs); + _contention.setBackoffMultiplier(_prefs.backoff_multiplier); acl.load(_store->getAclPath(), self_id); region_map.load(_store->getRegionsPath()); diff --git a/zephcore/app/RepeaterMesh.h b/zephcore/app/RepeaterMesh.h index fc602f6..0bc8f5d 100644 --- a/zephcore/app/RepeaterMesh.h +++ b/zephcore/app/RepeaterMesh.h @@ -128,8 +128,6 @@ protected: void logRx(mesh::Packet* pkt, int len, float score) override; void logTx(mesh::Packet* pkt, int len) override; void logTxFail(mesh::Packet* pkt, int len) override; - int calcRxDelay(float score, uint32_t air_time) const override; - uint32_t getRetransmitDelay(const mesh::Packet* packet) override; uint32_t getDirectRetransmitDelay(const mesh::Packet* packet) override; @@ -188,6 +186,17 @@ public: void saveIdentity(const mesh::LocalIdentity& new_id) override; void clearStats() override; + /* Adaptive contention window callbacks */ + float getContentionEstimate() const override { + return getContentionTracker().getContentionEstimate(); + } + float getFloodDelayFactor() const override { + return getContentionTracker().getFloodDelayFactor(); + } + void setBackoffMultiplier(float m) override { + getContentionTracker().setBackoffMultiplier(m); + } + void handleCommand(uint32_t sender_timestamp, char* command, char* reply); void loop(); diff --git a/zephcore/helpers/CommonCLI.cpp b/zephcore/helpers/CommonCLI.cpp index cb3de2e..937afff 100644 --- a/zephcore/helpers/CommonCLI.cpp +++ b/zephcore/helpers/CommonCLI.cpp @@ -82,7 +82,7 @@ void CommonCLI::loadPrefs(const char* path) { ok = ok && prefs_read(&file, &_prefs->tx_delay_factor, sizeof(_prefs->tx_delay_factor)); // 84 ok = ok && prefs_read(&file, &_prefs->guest_password[0], sizeof(_prefs->guest_password)); // 88 ok = ok && prefs_read(&file, &_prefs->direct_tx_delay_factor, sizeof(_prefs->direct_tx_delay_factor)); // 104 - ok = ok && prefs_read(&file, pad, 4); // 108 + ok = ok && prefs_read(&file, &_prefs->backoff_multiplier, sizeof(_prefs->backoff_multiplier)); // 108 ok = ok && prefs_read(&file, &_prefs->sf, sizeof(_prefs->sf)); // 112 ok = ok && prefs_read(&file, &_prefs->cr, sizeof(_prefs->cr)); // 113 ok = ok && prefs_read(&file, &_prefs->allow_read_only, sizeof(_prefs->allow_read_only)); // 114 @@ -122,6 +122,11 @@ void CommonCLI::loadPrefs(const char* path) { _prefs->rx_delay_base = constrain(_prefs->rx_delay_base, 0.0f, 20.0f); _prefs->tx_delay_factor = constrain(_prefs->tx_delay_factor, 0.0f, 2.0f); _prefs->direct_tx_delay_factor = constrain(_prefs->direct_tx_delay_factor, 0.0f, 2.0f); + /* Migrate uninitialized pad bytes (0.0f or NaN) to default 0.5 */ + if (_prefs->backoff_multiplier == 0.0f || _prefs->backoff_multiplier != _prefs->backoff_multiplier) { + _prefs->backoff_multiplier = 0.5f; + } + _prefs->backoff_multiplier = constrain(_prefs->backoff_multiplier, 0.0f, 2.0f); /* Migrate old AF multiplier (0-9) to duty cycle percentage (0-99) */ if (_prefs->airtime_factor > 0.0f && _prefs->airtime_factor <= 9.0f) { _prefs->airtime_factor *= 10.0f; @@ -179,7 +184,7 @@ void CommonCLI::savePrefs(const char* path) { fs_write(&file, &_prefs->tx_delay_factor, sizeof(_prefs->tx_delay_factor)); fs_write(&file, &_prefs->guest_password[0], sizeof(_prefs->guest_password)); fs_write(&file, &_prefs->direct_tx_delay_factor, sizeof(_prefs->direct_tx_delay_factor)); - fs_write(&file, pad, 4); + fs_write(&file, &_prefs->backoff_multiplier, sizeof(_prefs->backoff_multiplier)); fs_write(&file, &_prefs->sf, sizeof(_prefs->sf)); fs_write(&file, &_prefs->cr, sizeof(_prefs->cr)); fs_write(&file, &_prefs->allow_read_only, sizeof(_prefs->allow_read_only)); @@ -417,13 +422,18 @@ void CommonCLI::handleCommand(uint32_t sender_timestamp, const char* command, ch (double)_prefs->freq, (double)_prefs->bw, (uint32_t)_prefs->sf, (uint32_t)_prefs->cr); } else if (memcmp(config, "rxdelay", 7) == 0) { - snprintf(reply, CLI_REPLY_SIZE, "> %.2f", (double)_prefs->rx_delay_base); + snprintf(reply, CLI_REPLY_SIZE, "> adaptive (rxdelay deprecated)"); } else if (memcmp(config, "txdelay", 7) == 0) { - snprintf(reply, CLI_REPLY_SIZE, "> %.2f", (double)_prefs->tx_delay_factor); + float est = _callbacks->getContentionEstimate(); + float ff = _callbacks->getFloodDelayFactor(); + snprintf(reply, CLI_REPLY_SIZE, "> adaptive (est=%.1f flood=%.2f)", + (double)est, (double)ff); } else if (memcmp(config, "flood.max", 9) == 0) { snprintf(reply, CLI_REPLY_SIZE, "> %u", (uint32_t)_prefs->flood_max); } else if (memcmp(config, "direct.txdelay", 14) == 0) { - snprintf(reply, CLI_REPLY_SIZE, "> %.2f", (double)_prefs->direct_tx_delay_factor); + snprintf(reply, CLI_REPLY_SIZE, "> adaptive (direct.txdelay deprecated)"); + } else if (memcmp(config, "backoff.multiplier", 18) == 0) { + snprintf(reply, CLI_REPLY_SIZE, "> %.2f", (double)_prefs->backoff_multiplier); } else if (memcmp(config, "owner.info", 10) == 0) { *reply++ = '>'; *reply++ = ' '; @@ -578,23 +588,13 @@ void CommonCLI::handleCommand(uint32_t sender_timestamp, const char* command, ch savePrefs(); strcpy(reply, "OK"); } else if (memcmp(config, "rxdelay ", 8) == 0) { - float db = atof(&config[8]); - if (db >= 0) { - _prefs->rx_delay_base = db; - savePrefs(); - strcpy(reply, "OK"); - } else { - strcpy(reply, "Error, cannot be negative"); - } + _prefs->rx_delay_base = atof(&config[8]); + savePrefs(); + strcpy(reply, "OK (ignored: rxdelay is now adaptive)"); } else if (memcmp(config, "txdelay ", 8) == 0) { - float f = atof(&config[8]); - if (f >= 0) { - _prefs->tx_delay_factor = f; - savePrefs(); - strcpy(reply, "OK"); - } else { - strcpy(reply, "Error, cannot be negative"); - } + _prefs->tx_delay_factor = atof(&config[8]); + savePrefs(); + strcpy(reply, "OK (ignored: txdelay is now adaptive)"); } else if (memcmp(config, "flood.max ", 10) == 0) { uint8_t m = atoi(&config[10]); if (m <= 64) { @@ -605,13 +605,18 @@ void CommonCLI::handleCommand(uint32_t sender_timestamp, const char* command, ch strcpy(reply, "Error, max 64"); } } else if (memcmp(config, "direct.txdelay ", 15) == 0) { - float f = atof(&config[15]); - if (f >= 0) { - _prefs->direct_tx_delay_factor = f; + _prefs->direct_tx_delay_factor = atof(&config[15]); + savePrefs(); + strcpy(reply, "OK (ignored: direct.txdelay is now adaptive)"); + } else if (memcmp(config, "backoff.multiplier ", 19) == 0) { + float f = atof(&config[19]); + if (f >= 0.0f && f <= 2.0f) { + _prefs->backoff_multiplier = f; + _callbacks->setBackoffMultiplier(f); savePrefs(); strcpy(reply, "OK"); } else { - strcpy(reply, "Error, cannot be negative"); + strcpy(reply, "Error, range 0.0-2.0"); } } else if (memcmp(config, "owner.info ", 11) == 0) { config += 11; diff --git a/zephcore/helpers/CommonCLI.h b/zephcore/helpers/CommonCLI.h index 3676a7b..6aeee9c 100644 --- a/zephcore/helpers/CommonCLI.h +++ b/zephcore/helpers/CommonCLI.h @@ -47,6 +47,11 @@ public: virtual void clearStats() = 0; virtual void applyTempRadioParams(float freq, float bw, uint8_t sf, uint8_t cr, int timeout_mins) = 0; + // Adaptive contention window + virtual float getContentionEstimate() const { return -1.0f; } + virtual float getFloodDelayFactor() const { return 0.5f; } + virtual void setBackoffMultiplier(float m) { (void)m; } + // Sensor manager interface (for GPS) virtual double getNodeLat() const { return 0.0; } virtual double getNodeLon() const { return 0.0; } diff --git a/zephcore/helpers/NodePrefs.h b/zephcore/helpers/NodePrefs.h index fb4e49b..190c639 100644 --- a/zephcore/helpers/NodePrefs.h +++ b/zephcore/helpers/NodePrefs.h @@ -43,6 +43,7 @@ struct NodePrefs { float tx_delay_factor; char guest_password[16]; float direct_tx_delay_factor; + float backoff_multiplier; // reactive backoff cap (0.0 = sentinel → use default 0.5) uint32_t guard; uint8_t sf; uint8_t cr; diff --git a/zephcore/include/mesh/ContentionTracker.h b/zephcore/include/mesh/ContentionTracker.h new file mode 100644 index 0000000..7dcb59f --- /dev/null +++ b/zephcore/include/mesh/ContentionTracker.h @@ -0,0 +1,90 @@ +/* + * SPDX-License-Identifier: Apache-2.0 + * Adaptive Contention Window — replaces static txdelay/rxdelay + * + * Measures local retransmit contention by counting how many times + * we hear the same flood packet retransmitted by neighbors within + * a 10-second window after we decide to retransmit it ourselves. + * Feeds dupe counts into a rolling EMA to produce an adaptive + * delay factor for future retransmits. + */ + +#pragma once + +#include + +namespace mesh { + +class Packet; + +class ContentionTracker { +public: + ContentionTracker(); + + /* Cheap 32-bit hash for packet correlation (FNV-1a). + * NOT the same as the SHA256 used for dedup — this is only + * for matching packets in the 16-entry ring buffer. */ + static uint32_t computePacketHash32(const Packet *pkt); + + /* Called when we decide to retransmit a flood packet. */ + void trackRetransmit(uint32_t hash32, uint32_t now_ms); + + /* Called for every received flood packet. Returns true if + * the packet matched a tracked retransmit (dupe recorded). + * Caller should attempt reactive backoff when true. */ + bool recordDupeIfTracked(uint32_t hash32, uint32_t now_ms); + + /* Check if reactive extension is within cap for this entry. + * Returns the max additional ms allowed, 0 if cap reached. */ + uint16_t getReactiveHeadroom(uint32_t hash32, uint32_t airtime_ms) const; + + /* Record that we added reactive extension to this entry. */ + void addReactiveExtension(uint32_t hash32, uint16_t added_ms); + + /* Finalize expired entries into EMA. Call from maintenanceLoop. */ + void tick(uint32_t now_ms); + + /* Current contention estimate (EMA of dupes per retransmitted packet). */ + float getContentionEstimate() const; + + /* Adaptive delay factor for flood retransmits. + * sqrt curve: 0.05 + 0.116 * sqrt(est), cap 2.0. + * Returns 0.5 during warmup. */ + float getFloodDelayFactor() const; + + bool isWarmedUp() const { return _finalized_count >= WARMUP_PACKETS; } + + void setBackoffMultiplier(float m) { _backoff_multiplier = m; } + float getBackoffMultiplier() const { return _backoff_multiplier; } + +private: + static constexpr int RING_SIZE = 16; + static constexpr uint32_t WINDOW_MS = 10000; + static constexpr int EMA_SHIFT = 3; /* alpha = 1/8 */ + static constexpr int WARMUP_PACKETS = 4; + static constexpr float MIN_FLOOD_FACTOR = 0.05f; + static constexpr float FLOOD_SCALE = 0.116f; /* (0.5 - 0.05) / sqrt(15) */ + static constexpr float MAX_FLOOD_FACTOR = 2.0f; + static constexpr float DEFAULT_BACKOFF_MULT = 0.5f; + static constexpr uint32_t STALE_MS = 300000; /* 5 minutes */ + + struct Entry { + uint32_t hash32; + uint32_t first_seen_ms; + uint8_t dupe_count; + uint16_t reactive_added_ms; + bool active; + }; + + Entry _ring[RING_SIZE]; + int _next_idx; + uint32_t _ema_x256; + int _finalized_count; + uint32_t _last_retransmit_ms; + float _backoff_multiplier; + + void finalizeEntry(int idx); + int findEntry(uint32_t hash32) const; +}; + +} /* namespace mesh */ diff --git a/zephcore/include/mesh/Dispatcher.h b/zephcore/include/mesh/Dispatcher.h index 08ae4e4..73358a4 100644 --- a/zephcore/include/mesh/Dispatcher.h +++ b/zephcore/include/mesh/Dispatcher.h @@ -62,6 +62,8 @@ public: virtual int getFreeCount() const = 0; virtual Packet *getOutboundByIdx(int i) = 0; virtual Packet *removeOutboundByIdx(int i) = 0; + virtual uint32_t getOutboundSchedule(int i) const = 0; + virtual bool rescheduleOutbound(int i, uint32_t new_scheduled_for) = 0; virtual void queueInbound(Packet *packet, uint32_t scheduled_for) = 0; virtual Packet *getNextInbound(uint32_t now) = 0; }; @@ -104,6 +106,9 @@ protected: uint16_t _err_flags; Dispatcher(Radio &radio, MillisecondClock &ms, PacketManager &mgr); + void notifyTxQueued(uint32_t delay_ms) { + if (_tx_queued_cb) _tx_queued_cb(delay_ms, _tx_queued_user_data); + } virtual DispatcherAction onRecvPacket(Packet *pkt) = 0; virtual void logRxRaw(float snr, float rssi, const uint8_t raw[], int len) { (void)snr; (void)rssi; (void)raw; (void)len; } virtual void logRx(Packet *packet, int len, float score) { (void)packet; (void)len; (void)score; } diff --git a/zephcore/include/mesh/Mesh.h b/zephcore/include/mesh/Mesh.h index 7a2dbe2..7b9a29b 100644 --- a/zephcore/include/mesh/Mesh.h +++ b/zephcore/include/mesh/Mesh.h @@ -6,6 +6,7 @@ #pragma once #include +#include #include namespace mesh { @@ -31,6 +32,11 @@ class Mesh : public Dispatcher { DispatcherAction forwardMultipartDirect(Packet *pkt); protected: + ContentionTracker _contention; + ContentionTracker& getContentionTracker() { return _contention; } + const ContentionTracker& getContentionTracker() const { return _contention; } + void extendPendingRetransmit(uint32_t hash32); + DispatcherAction onRecvPacket(Packet *pkt) override; virtual uint32_t getCADFailRetryDelay() const override; virtual DispatcherAction routeRecvPacket(Packet *packet); @@ -57,6 +63,7 @@ public: Mesh(Radio &radio, MillisecondClock &ms, RNG &rng, RTCClock &rtc, PacketManager &mgr, MeshTables &tables); void begin(); void loop(); + void maintenanceLoop(); LocalIdentity self_id; diff --git a/zephcore/include/mesh/StaticPoolPacketManager.h b/zephcore/include/mesh/StaticPoolPacketManager.h index f0c576b..7ea35f1 100644 --- a/zephcore/include/mesh/StaticPoolPacketManager.h +++ b/zephcore/include/mesh/StaticPoolPacketManager.h @@ -20,6 +20,8 @@ public: int getFreeCount() const override; Packet *getOutboundByIdx(int i) override; Packet *removeOutboundByIdx(int i) override; + uint32_t getOutboundSchedule(int i) const override; + bool rescheduleOutbound(int i, uint32_t new_scheduled_for) override; void queueInbound(Packet *packet, uint32_t scheduled_for) override; Packet *getNextInbound(uint32_t now) override; }; diff --git a/zephcore/src/ContentionTracker.cpp b/zephcore/src/ContentionTracker.cpp new file mode 100644 index 0000000..b50ba71 --- /dev/null +++ b/zephcore/src/ContentionTracker.cpp @@ -0,0 +1,161 @@ +/* + * SPDX-License-Identifier: Apache-2.0 + * Adaptive Contention Window — dupe-counting based delay estimation + */ + +#include +#include +#include +#include + +namespace mesh { + +ContentionTracker::ContentionTracker() + : _next_idx(0), _ema_x256(0), _finalized_count(0), + _last_retransmit_ms(0), _backoff_multiplier(DEFAULT_BACKOFF_MULT) +{ + memset(_ring, 0, sizeof(_ring)); +} + +/* FNV-1a hash of payload_type + first 8 bytes of payload. + * Cheap and sufficient for 16-entry correlation. */ +uint32_t ContentionTracker::computePacketHash32(const Packet *pkt) +{ + uint32_t h = 0x811c9dc5u; /* FNV offset basis */ + uint8_t t = pkt->getPayloadType(); + h = (h ^ t) * 0x01000193u; + int n = pkt->payload_len < 8 ? pkt->payload_len : 8; + for (int i = 0; i < n; i++) { + h = (h ^ pkt->payload[i]) * 0x01000193u; + } + return h; +} + +int ContentionTracker::findEntry(uint32_t hash32) const +{ + for (int i = 0; i < RING_SIZE; i++) { + if (_ring[i].active && _ring[i].hash32 == hash32) { + return i; + } + } + return -1; +} + +void ContentionTracker::finalizeEntry(int idx) +{ + if (!_ring[idx].active) return; + + uint32_t sample_x256 = (uint32_t)_ring[idx].dupe_count << 8; + + int32_t diff = (int32_t)sample_x256 - (int32_t)_ema_x256; + + if (_finalized_count < WARMUP_PACKETS) { + /* During warmup, seed the EMA directly */ + if (_finalized_count == 0) { + _ema_x256 = sample_x256; + } else { + _ema_x256 = (uint32_t)((int32_t)_ema_x256 + (diff >> 1)); + } + } else { + /* Normal EMA update: ema += (sample - ema) >> shift */ + _ema_x256 = (uint32_t)((int32_t)_ema_x256 + (diff >> EMA_SHIFT)); + } + + _finalized_count++; + _ring[idx].active = false; +} + +void ContentionTracker::trackRetransmit(uint32_t hash32, uint32_t now_ms) +{ + _last_retransmit_ms = now_ms; + + /* If ring is full, finalize the oldest active entry */ + if (_ring[_next_idx].active) { + finalizeEntry(_next_idx); + } + + Entry &e = _ring[_next_idx]; + e.hash32 = hash32; + e.first_seen_ms = now_ms; + e.dupe_count = 0; + e.reactive_added_ms = 0; + e.active = true; + + _next_idx = (_next_idx + 1) % RING_SIZE; +} + +bool ContentionTracker::recordDupeIfTracked(uint32_t hash32, uint32_t now_ms) +{ + int idx = findEntry(hash32); + if (idx < 0) return false; + + Entry &e = _ring[idx]; + + /* Check if entry has expired */ + if (now_ms - e.first_seen_ms > WINDOW_MS) { + finalizeEntry(idx); + return false; + } + + if (e.dupe_count < 255) { + e.dupe_count++; + } + return true; +} + +uint16_t ContentionTracker::getReactiveHeadroom(uint32_t hash32, uint32_t airtime_ms) const +{ + int idx = findEntry(hash32); + if (idx < 0) return 0; + + uint32_t cap = (uint32_t)(_backoff_multiplier * (float)airtime_ms); + if (_ring[idx].reactive_added_ms >= cap) return 0; + + uint32_t remaining = cap - _ring[idx].reactive_added_ms; + return remaining > 0xFFFF ? 0xFFFF : (uint16_t)remaining; +} + +void ContentionTracker::addReactiveExtension(uint32_t hash32, uint16_t added_ms) +{ + int idx = findEntry(hash32); + if (idx < 0) return; + + uint32_t total = (uint32_t)_ring[idx].reactive_added_ms + added_ms; + _ring[idx].reactive_added_ms = total > 0xFFFF ? 0xFFFF : (uint16_t)total; +} + +void ContentionTracker::tick(uint32_t now_ms) +{ + /* Finalize expired entries */ + for (int i = 0; i < RING_SIZE; i++) { + if (_ring[i].active && now_ms - _ring[i].first_seen_ms > WINDOW_MS) { + finalizeEntry(i); + } + } + + /* Staleness decay: if no retransmit in 5 minutes, decay toward 0 */ + if (_last_retransmit_ms != 0 && now_ms - _last_retransmit_ms > STALE_MS) { + if (_ema_x256 > 0) { + _ema_x256 -= _ema_x256 >> EMA_SHIFT; + } + } +} + +float ContentionTracker::getContentionEstimate() const +{ + return (float)_ema_x256 / 256.0f; +} + +float ContentionTracker::getFloodDelayFactor() const +{ + if (!isWarmedUp()) return 0.5f; + + float est = getContentionEstimate(); + if (est <= 0.0f) return MIN_FLOOD_FACTOR; + + float factor = MIN_FLOOD_FACTOR + FLOOD_SCALE * sqrtf(est); + if (factor > MAX_FLOOD_FACTOR) factor = MAX_FLOOD_FACTOR; + return factor; +} + +} /* namespace mesh */ diff --git a/zephcore/src/Dispatcher.cpp b/zephcore/src/Dispatcher.cpp index 95dab5a..45c8641 100644 --- a/zephcore/src/Dispatcher.cpp +++ b/zephcore/src/Dispatcher.cpp @@ -256,13 +256,7 @@ void Dispatcher::checkRecv() logRx(pkt, pkt->getRawLength(), score); if (pkt->isRouteFlood()) { n_recv_flood++; - int delay = calcRxDelay(score, air_time); - if (delay < 50) { - processRecvPacket(pkt); - } else { - if (delay > (int)MAX_RX_DELAY_MILLIS) delay = MAX_RX_DELAY_MILLIS; - _mgr->queueInbound(pkt, futureMillis(delay)); - } + processRecvPacket(pkt); } else { n_recv_direct++; processRecvPacket(pkt); @@ -291,7 +285,10 @@ void Dispatcher::checkSend() { uint32_t now = (uint32_t)_ms->getMillis(); int count = _mgr->getOutboundCount(now); - if (count == 0) return; + if (count == 0) { + cad_busy_start = 0; + return; + } if (_radio->isReceiving()) { /* Channel busy — enforce retry timer so we don't hammer the check */ diff --git a/zephcore/src/Mesh.cpp b/zephcore/src/Mesh.cpp index fc494e7..7185c58 100644 --- a/zephcore/src/Mesh.cpp +++ b/zephcore/src/Mesh.cpp @@ -27,6 +27,35 @@ void Mesh::loop() Dispatcher::loop(); } +void Mesh::maintenanceLoop() +{ + Dispatcher::maintenanceLoop(); + _contention.tick((uint32_t)_ms->getMillis()); +} + +void Mesh::extendPendingRetransmit(uint32_t hash32) +{ + uint32_t now = (uint32_t)_ms->getMillis(); + int total = _mgr->getOutboundTotal(); + for (int i = 0; i < total; i++) { + Packet *pkt = _mgr->getOutboundByIdx(i); + if (pkt && pkt->isRouteFlood() + && ContentionTracker::computePacketHash32(pkt) == hash32) { + uint32_t airtime = _radio->getEstAirtimeFor(pkt->getRawLength()); + uint16_t headroom = _contention.getReactiveHeadroom(hash32, airtime); + if (headroom == 0) break; + uint32_t extra = _rng->nextInt(0, (int)headroom + 1); + /* Reschedule from NOW, not from the original EMA-based schedule. + * Hearing a dupe means the channel was just used — defer from + * this moment, don't compound on top of the base delay. */ + _mgr->rescheduleOutbound(i, now + extra); + _contention.addReactiveExtension(hash32, (uint16_t)extra); + notifyTxQueued(extra); + break; + } + } +} + bool Mesh::allowPacketForward(const Packet *packet) { (void)packet; @@ -36,7 +65,7 @@ bool Mesh::allowPacketForward(const Packet *packet) uint32_t Mesh::getRetransmitDelay(const Packet *packet) { uint32_t t = (_radio->getEstAirtimeFor(packet->getRawLength()) * 52 / 50) / 2; - return _rng->nextInt(0, 7) * t; + return _rng->nextInt(0, 5) * t; } uint32_t Mesh::getCADFailRetryDelay() const @@ -62,6 +91,8 @@ DispatcherAction Mesh::routeRecvPacket(Packet *packet) // append this node's hash to 'path' self_id.copyHashTo(&packet->path[n * packet->getPathHashSize()], packet->getPathHashSize()); packet->setPathHashCount(n + 1); + uint32_t h = ContentionTracker::computePacketHash32(packet); + _contention.trackRetransmit(h, (uint32_t)_ms->getMillis()); uint32_t d = getRetransmitDelay(packet); return ACTION_RETRANSMIT_DELAYED(packet->getPathHashCount(), d); // give priority to closer sources } @@ -170,6 +201,14 @@ DispatcherAction Mesh::onRecvPacket(Packet *pkt) if (pkt->isRouteFlood() && filterRecvFloodPacket(pkt)) return ACTION_RELEASE; + /* Record dupes for contention tracking + reactive backoff */ + if (pkt->isRouteFlood()) { + uint32_t h = ContentionTracker::computePacketHash32(pkt); + if (_contention.recordDupeIfTracked(h, (uint32_t)_ms->getMillis())) { + extendPendingRetransmit(h); + } + } + DispatcherAction action = ACTION_RELEASE; switch (pkt->getPayloadType()) { diff --git a/zephcore/src/StaticPoolPacketManager.cpp b/zephcore/src/StaticPoolPacketManager.cpp index c02fcc3..922ca00 100644 --- a/zephcore/src/StaticPoolPacketManager.cpp +++ b/zephcore/src/StaticPoolPacketManager.cpp @@ -92,6 +92,16 @@ struct PacketQueue { return idx; } + uint32_t scheduleAt(int i) const { + return (i < _num) ? _schedule_table[i] : 0; + } + + bool reschedule(int i, uint32_t new_scheduled_for) { + if (i >= _num) return false; + _schedule_table[i] = new_scheduled_for; + return true; + } + int count() const { return _num; } Packet *itemAt(int i) const { return (i < _num) ? _table[i] : nullptr; } }; @@ -173,6 +183,16 @@ Packet *StaticPoolPacketManager::removeOutboundByIdx(int i) return _send_queue.removeByIdx(i); } +uint32_t StaticPoolPacketManager::getOutboundSchedule(int i) const +{ + return _send_queue.scheduleAt(i); +} + +bool StaticPoolPacketManager::rescheduleOutbound(int i, uint32_t new_scheduled_for) +{ + return _send_queue.reschedule(i, new_scheduled_for); +} + void StaticPoolPacketManager::queueInbound(Packet *packet, uint32_t scheduled_for) { if (!_rx_queue.add(packet, 0, scheduled_for)) {