reactive collision avoidance

zephcore/src/ContentionTracker.cpp

@@ -0,0 +1,161 @@
+/*
+ * SPDX-License-Identifier: Apache-2.0
+ * Adaptive Contention Window — dupe-counting based delay estimation
+ */
+
+#include <mesh/ContentionTracker.h>
+#include <mesh/Packet.h>
+#include <math.h>
+#include <string.h>
+
+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 */

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 */

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()) {

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)) {