From 0e7bb0520879d870ff8be166684700f406dc02c2 Mon Sep 17 00:00:00 2001 From: Lloyd Date: Wed, 13 May 2026 12:34:05 +0100 Subject: [PATCH] Refactor INA219 sensor integration. --- config.yaml.example | 1 + docs/pr_hash_once.md | 346 ------------------------------- docs/pr_in_flight_cap.md | 349 ------------------------------- docs/pr_tx_serialization.md | 395 ------------------------------------ repeater/sensors/ina219.py | 101 ++++++--- 5 files changed, 77 insertions(+), 1115 deletions(-) delete mode 100644 docs/pr_hash_once.md delete mode 100644 docs/pr_in_flight_cap.md delete mode 100644 docs/pr_tx_serialization.md diff --git a/config.yaml.example b/config.yaml.example index 2776f59..cc54790 100644 --- a/config.yaml.example +++ b/config.yaml.example @@ -233,6 +233,7 @@ sensors: # auto_install_packages: true # settings: # i2c_address: 64 # 0x40 in decimal + # bus_number: 1 # I2C bus number (1 for Raspberry Pi default) # max_expected_amps: 2.0 # shunt_ohms: 0.1 diff --git a/docs/pr_hash_once.md b/docs/pr_hash_once.md deleted file mode 100644 index d8e7cef..0000000 --- a/docs/pr_hash_once.md +++ /dev/null @@ -1,346 +0,0 @@ -# PR: Compute Packet Hash Once Per Forwarded Packet - -**Branch:** `perf/hash-once` -**Base:** `rightup/fix-perfom-speed` -**Files changed:** `repeater/engine.py` (1 file, ~51 lines net) - ---- - -## Problem - -`packet.calculate_packet_hash()` runs a SHA-256 digest over the full serialised -packet bytes, converts the result to a hex string, and uppercases it. Before -this change the hot forwarding path triggered this computation **three times per -packet**: - -| Call site | Where | When | -|-----------|-------|------| -| `__call__` line 162 | `pkt_hash_full = packet.calculate_packet_hash()...` | Every received packet | -| `flood_forward` / `direct_forward` via `is_duplicate` | `pkt_hash = packet.calculate_packet_hash()...` | Every packet that reaches the forward check | -| `flood_forward` / `direct_forward` via `mark_seen` | `pkt_hash = packet_hash or packet.calculate_packet_hash()...` | Every packet that passes the duplicate check | - -And on the drop path, a fourth computation: - -| Call site | Where | When | -|-----------|-------|------| -| `_get_drop_reason` → `is_duplicate` | `pkt_hash = packet.calculate_packet_hash()...` | Every dropped packet | - -The hash computed in `__call__` was already available as `pkt_hash_full` but was -never passed into `process_packet`, `flood_forward`, `direct_forward`, -`is_duplicate`, `mark_seen`, or `_get_drop_reason`. Each of those methods -recomputed it independently. - ---- - -## Root Cause - -The `packet_hash` optional parameter existed on `mark_seen` but not on -`is_duplicate`, `flood_forward`, `direct_forward`, `process_packet`, or -`_get_drop_reason`. The call chain therefore had no way to propagate the -already-computed hash. - ---- - -## Solution - -Thread the pre-computed `pkt_hash_full` from `__call__` down through the call -chain as an optional `packet_hash: Optional[str] = None` parameter. Each method -uses the provided hash if present, or falls back to computing it — preserving -backward compatibility for any caller that doesn't have a pre-computed hash. - -``` -Before: - __call__ → calculate_packet_hash() #1 - → process_packet - → flood_forward - → is_duplicate → calculate_packet_hash() #2 - → mark_seen → calculate_packet_hash() #3 - (drop path) - → _get_drop_reason - → is_duplicate → calculate_packet_hash() #4 - -After: - __call__ → calculate_packet_hash() #1 (only computation) - → process_packet(packet_hash=pkt_hash_full) - → flood_forward(packet_hash=pkt_hash_full) - → is_duplicate(packet_hash=pkt_hash_full) uses provided hash ✓ - → mark_seen(packet_hash=pkt_hash_full) uses provided hash ✓ - (drop path) - → _get_drop_reason(packet_hash=pkt_hash_full) - → is_duplicate(packet_hash=pkt_hash_full) uses provided hash ✓ -``` - ---- - -## Methods Changed - -### `is_duplicate(packet, packet_hash=None)` - -```python -# Before -def is_duplicate(self, packet: Packet) -> bool: - pkt_hash = packet.calculate_packet_hash().hex().upper() # always recomputed - if pkt_hash in self.seen_packets: - return True - return False - -# After -def is_duplicate(self, packet: Packet, packet_hash: Optional[str] = None) -> bool: - """... - INVARIANT: purely synchronous — no await points. The caller relies on - is_duplicate + mark_seen being atomic within the asyncio event loop. - Do NOT add any await here without revisiting that invariant. - """ - pkt_hash = packet_hash or packet.calculate_packet_hash().hex().upper() - return pkt_hash in self.seen_packets -``` - -### `_get_drop_reason(packet, packet_hash=None)` - -```python -# Before -def _get_drop_reason(self, packet: Packet) -> str: - if self.is_duplicate(packet): ... # recomputes hash - -# After -def _get_drop_reason(self, packet: Packet, packet_hash: Optional[str] = None) -> str: - if self.is_duplicate(packet, packet_hash=packet_hash): ... # propagates hash -``` - -### `flood_forward(packet, packet_hash=None)` - -```python -# Before -def flood_forward(self, packet: Packet) -> Optional[Packet]: - ... - if self.is_duplicate(packet): ... # recomputes - self.mark_seen(packet) # recomputes - -# After -def flood_forward(self, packet: Packet, packet_hash: Optional[str] = None) -> Optional[Packet]: - """... - INVARIANT: purely synchronous — no await points. - """ - ... - if self.is_duplicate(packet, packet_hash=packet_hash): ... # propagates - self.mark_seen(packet, packet_hash=packet_hash) # propagates -``` - -### `direct_forward(packet, packet_hash=None)` — same pattern as `flood_forward` - -### `process_packet(packet, snr=0.0, packet_hash=None)` - -```python -# Before -def process_packet(self, packet, snr=0.0): - fwd_pkt = self.flood_forward(packet) # no hash - -# After -def process_packet(self, packet, snr=0.0, packet_hash=None): - """... - packet_hash: pre-computed SHA-256 hex from __call__; eliminates 2 SHA-256 - calls per forwarded packet by propagating the hash through the call chain. - """ - fwd_pkt = self.flood_forward(packet, packet_hash=packet_hash) -``` - -### `__call__` — two call-site changes - -```python -# Before -result = (None if ... else self.process_packet(processed_packet, snr)) -... -drop_reason = processed_packet.drop_reason or self._get_drop_reason(processed_packet) - -# After -result = (None if ... else self.process_packet(processed_packet, snr, packet_hash=pkt_hash_full)) -... -drop_reason = processed_packet.drop_reason or self._get_drop_reason( - processed_packet, packet_hash=pkt_hash_full -) -``` - ---- - -## What Was Not Changed - -`record_packet_only` (line 446) and `record_duplicate` (line 486) each compute -the hash independently. These are separate recording paths (called from the -inject path and from the raw-packet subscriber, respectively) that have no -`pkt_hash_full` from `__call__` in scope. Changing them would require a larger -refactor with no benefit to the forwarding hot path, so they are left unchanged. - -The fallback `packet_hash or packet.calculate_packet_hash()...` pattern in -`is_duplicate`, `mark_seen`, and `_build_packet_record` ensures external callers -(e.g. `TraceHelper.is_duplicate(packet)` from trace processing) continue to work -without any change. - ---- - -## Invariant Comments Added - -`flood_forward`, `direct_forward`, and `is_duplicate` now carry explicit docstring -invariants: - -> **INVARIANT:** purely synchronous — no await points. The is_duplicate + -> mark_seen pair is atomic within the asyncio event loop. Do NOT add any await -> here without revisiting that invariant in `__call__` / `process_packet`. - -These invariants were implicit before. Making them explicit means a future -contributor adding an `await` inside these methods will see the warning and -understand the consequence: the duplicate-check and mark-seen can no longer be -guaranteed atomic, allowing the same packet to be forwarded twice under concurrent -task dispatch. - ---- - -## Quantification - -On a Raspberry Pi running CPython 3.13, `hashlib.sha256` on a 50–200 byte -LoRa payload takes approximately 1–3 µs. The `.hex().upper()` string conversion -adds another ~0.5 µs. Savings per forwarded packet: ~3–8 µs. - -At 3 packets/second sustained forwarding rate this saves ~10–25 µs/second, which -is negligible in absolute terms. The more significant benefit is correctness and -clarity: - -- One canonical hash value per packet in the forwarding path. -- No possibility of the hash changing between the `is_duplicate` check and the - `mark_seen` call if `calculate_packet_hash` had any mutable state (it doesn't, - but the pattern is now provably correct). -- Explicit invariant documentation closes a latent trap for future contributors. - ---- - -## Test Plan - -### Unit tests (no hardware) - -**T1 — Hash computed exactly once per forwarded packet** - -```python -async def test_hash_computed_once_for_flood(): - call_count = 0 - original = Packet.calculate_packet_hash - - def counting_hash(self): - nonlocal call_count - call_count += 1 - return original(self) - - with patch.object(Packet, "calculate_packet_hash", counting_hash): - await engine(flood_packet, metadata={}) - - assert call_count == 1, f"Expected 1 hash computation, got {call_count}" -``` - -**T2 — Hash computed exactly once per dropped (duplicate) packet** - -```python -async def test_hash_computed_once_for_duplicate(): - # Mark packet seen first - engine.seen_packets[packet.calculate_packet_hash().hex().upper()] = time.time() - - call_count = 0 - original = Packet.calculate_packet_hash - def counting_hash(self): - nonlocal call_count; call_count += 1; return original(self) - - with patch.object(Packet, "calculate_packet_hash", counting_hash): - await engine(packet, metadata={}) - - # One computation in __call__ for pkt_hash_full; should not trigger again - # in process_packet → flood_forward → is_duplicate (drop path via _get_drop_reason) - assert call_count == 1 -``` - -**T3 — External callers of `is_duplicate` without hash still work** - -```python -def test_is_duplicate_without_hash(): - """TraceHelper and other external callers pass no hash — must still work.""" - pkt = make_test_packet() - engine.seen_packets[pkt.calculate_packet_hash().hex().upper()] = time.time() - - assert engine.is_duplicate(pkt) is True # no packet_hash arg - assert engine.is_duplicate(pkt, packet_hash="WRONGHASH") is False -``` - -**T4 — mark_seen / is_duplicate agree on the same hash** - -```python -def test_mark_then_is_duplicate_consistent(): - pkt = make_test_packet() - pkt_hash = pkt.calculate_packet_hash().hex().upper() - - assert engine.is_duplicate(pkt, packet_hash=pkt_hash) is False - engine.mark_seen(pkt, packet_hash=pkt_hash) - assert engine.is_duplicate(pkt, packet_hash=pkt_hash) is True - # Same result without the pre-computed hash (fallback path) - assert engine.is_duplicate(pkt) is True -``` - -**T5 — flood_forward / direct_forward signatures are backward compatible** - -```python -def test_flood_forward_no_hash_arg(): - """Callers that don't pass packet_hash must still work (fallback compute).""" - pkt = make_flood_packet() - result = engine.flood_forward(pkt) # no packet_hash — must not raise - assert result is not None or pkt.drop_reason is not None -``` - -### Integration / field tests (with hardware) - -**T6 — Forwarding throughput unchanged** - -1. Forward 100 packets at maximum duty-cycle budget. -2. Verify all eligible packets are forwarded (same count as before change). -3. Verify no `Duplicate` drops that were not present before. - -**T7 — Duplicate detection unchanged** - -1. Send the same packet twice within 1 second. -2. Verify the first is forwarded and the second is logged as `"Duplicate"`. - -**T8 — CPU profile shows reduced `calculate_packet_hash` calls** - -1. Enable Python profiling (`cProfile`) on the repeater for 60 seconds. -2. Compare `calculate_packet_hash` call count before and after. - -**Expected:** call count approximately halved for workloads where most packets -are forwarded (≤ 1 call per forwarded packet vs ≥ 3 before). - ---- - -## Proof of Correctness - -### Why the fallback `packet_hash or packet.calculate_packet_hash()` is safe - -`packet_hash` is either the correct hash (passed from `__call__`) or `None`. -If it is `None`, the fallback computes the hash fresh — identical to the old -behaviour. There is no case where a wrong hash is used: the only source of a -non-None `packet_hash` is `pkt_hash_full = packet.calculate_packet_hash()...` -in `__call__`, computed over the same `processed_packet` (a deep copy of the -received packet, unchanged between hash computation and the call to -`process_packet`). - -### Why passing the hash through a deep-copied packet is correct - -`processed_packet = copy.deepcopy(packet)` (line 178) happens before -`pkt_hash_full` is passed to `process_packet`. The deep copy does not change -the packet's wire representation — `calculate_packet_hash()` calls -`packet.write_to()` which serialises the packet's fields. The copy has the -same fields, so `deepcopy(packet).calculate_packet_hash() == packet.calculate_packet_hash()`. -Passing the hash computed from the original to the copy is correct. - -### Why the invariant is critical - -asyncio only yields execution at `await` points. `flood_forward` and -`direct_forward` have no `await`, so they run atomically from the event loop's -perspective. The `is_duplicate` check and the `mark_seen` call inside them -cannot be interleaved with another coroutine. If a future change added an -`await` between them, two concurrent `_route_packet` tasks could both pass the -duplicate check for the same packet before either marked it seen — sending the -same packet twice. The invariant comment documents this so the risk is visible -at the point where it could be broken. diff --git a/docs/pr_in_flight_cap.md b/docs/pr_in_flight_cap.md deleted file mode 100644 index df6cbc9..0000000 --- a/docs/pr_in_flight_cap.md +++ /dev/null @@ -1,349 +0,0 @@ -# PR: Bounded In-Flight Task Counter + Simplified Route Task Management - -**Branch:** `perf/in-flight-cap` -**Base:** `rightup/fix-perfom-speed` -**Files changed:** `repeater/packet_router.py` (1 file, ~33 lines net) - ---- - -## Background - -The queue loop dispatches each incoming packet as an `asyncio.create_task` so TX -delay timers run concurrently — this is correct behaviour. The previous -implementation tracked these tasks in a `set[asyncio.Task]` (`_route_tasks`) for -two reasons: - -1. **Error surfacing** — the done-callback read `task.result()` to log exceptions. -2. **Shutdown cancellation** — `stop()` cancelled and awaited all tasks in the set. - -This PR replaces the set with a simple integer counter and tightens the companion -deduplication prune threshold. - ---- - -## Problems - -### Problem 1 — Unbounded task accumulation - -LoRa airtime naturally limits steady-state throughput to a handful of in-flight -tasks at any time. But burst arrivals can spike the count temporarily: - -- **Multi-hop flood amplification**: a single source packet is forwarded by every - repeater in range, each of which re-broadcasts it. A node at a mesh junction - may receive 5–10 copies within 100 ms, each scheduling a separate `delayed_send` - task. -- **Collision retries**: hardware-level collisions produce duplicate RF bursts that - all arrive within the same RX window. -- **Bridge nodes**: high-traffic gateway nodes connect multiple mesh segments and - forward both directions simultaneously. - -Under these conditions `_route_tasks` can accumulate dozens of sleeping tasks. -Each holds a reference to the packet, the forwarded packet copy, a closure over -`delayed_send`, and associated asyncio task overhead. There is no cap; the set -grows until the duty-cycle gate finally fires for each task. - -### Problem 2 — `_route_tasks` set adds O(1) cost on every packet but O(n) cost on shutdown - -Every packet adds one entry to `_route_tasks` and removes it in the done-callback. -This is O(1) per operation, but the `stop()` shutdown path iterates the entire set -to cancel and gather all tasks — O(n) where n is however many tasks happen to be -in-flight at shutdown time. On a busy node this could delay clean shutdown. - -### Problem 3 — `_COMPANION_DEDUPE_PRUNE_THRESHOLD = 1000` is too high - -The companion delivery deduplication dict prunes itself only when it exceeds 1000 -entries. With a 60-second TTL, each PATH/protocol-response packet adds one entry. -On a busy mesh with 50+ nodes sending adverts and PATH packets, the dict can grow -to hundreds of entries before a prune is triggered — keeping stale entries in -memory for up to 60 seconds × 1000/rate entries worth of time. - ---- - -## Solution - -### Replace `_route_tasks` set with `_in_flight` counter - -An integer counter provides the same protection (tasks complete; done-callback -fires) without holding strong references to each task object: - -```python -# __init__ -self._in_flight: int = 0 -self._max_in_flight: int = 30 - -# _process_queue — drop early if cap reached -if self._in_flight >= self._max_in_flight: - logger.warning("In-flight task cap reached (%d/%d), dropping packet", ...) - continue -self._in_flight += 1 -task = asyncio.create_task(self._route_packet(packet)) -task.add_done_callback(self._on_route_done) - -# done-callback -def _on_route_done(self, task): - self._in_flight -= 1 - if not task.cancelled() and task.exception(): - logger.error("_route_packet raised: %s", task.exception(), ...) -``` - -### Cap at 30 concurrent in-flight tasks - -30 is chosen as a ceiling that is: -- **Never reached in normal operation**: LoRa airtime at SF8/125 kHz limits - throughput to ~2–3 packets per second; with delays of 0.5–5 s each, the - steady-state in-flight count is at most 5–15 tasks. -- **High enough not to drop legitimate traffic**: a burst of 30 nearly-simultaneous - packets would require every node in a large mesh to transmit within 1 second. -- **Low enough to protect against pathological scenarios**: a misconfigured node - flooding the channel or a software bug causing infinite re-queuing. - -### Tighten companion dedup prune threshold to 200 - -200 entries at 60 s TTL means a sweep is triggered after ~200 unique PATH/response -packets arrive without any expiry. This is far more than a typical companion -session (which sees a handful of active connections) but prevents multi-hour -accumulation on a busy mesh. - ---- - -## Trade-off: Shutdown Cancellation - -The previous `_route_tasks` set allowed `stop()` to explicitly cancel and await -all in-flight tasks on shutdown. The counter approach does not. - -**Why this is acceptable:** - -1. In-flight `_route_packet` tasks are sleeping inside `delayed_send` (waiting for - their TX delay timer). When the event loop is shut down — whether via - `asyncio.run()` completing, `loop.stop()`, or `SIGTERM` handling — Python - cancels all pending tasks automatically. - -2. Even under the old approach, cancelling a sleeping `delayed_send` means the - packet is not transmitted. The result is the same whether cancellation happens - explicitly in `stop()` or implicitly when the event loop closes. - -3. For a graceful shutdown where we want to *wait* for in-flight packets to - complete transmission, the right mechanism is `stop()` awaiting the queue to - drain *before* cancelling the router task — not cancelling sleeping tasks. - Neither the old code nor this PR implements that, so no regression. - ---- - -## Why This Is the Right Approach - -### Alternative A — Keep `_route_tasks` set, add a size cap - -```python -if len(self._route_tasks) >= 30: - logger.warning(...) - continue -``` - -Works, but the set still holds a strong reference to every Task object for the -duration of its sleep. The counter holds an integer. Task objects in Python 3.12+ -are already strongly referenced by the event loop scheduler; the set reference is -redundant for preventing GC cancellation. - -### Alternative B — `asyncio.Semaphore` - -```python -self._sem = asyncio.Semaphore(30) -async with self._sem: - await self._route_packet(packet) -``` - -Correct but changes the queue loop from fire-and-forget to blocking: the loop -would wait at `async with self._sem` for a slot to open, stalling packet reads -while a slot is occupied. That reintroduces the queue freeze the concurrent -dispatch was designed to prevent. A semaphore is the right tool for *rate- -limiting* producers; a counter cap at the dispatch site is the right tool for -bounding *background* tasks. - -### Alternative C — Integer counter (this PR) - -- O(1) increment and decrement. -- No strong reference to task objects beyond the event loop's own reference. -- Drop decision is synchronous and immediate — no sleeping on semaphore. -- Error logging preserved in `_on_route_done`. -- Simpler code, easier to reason about. - ---- - -## Changes — `repeater/packet_router.py` only - -| Location | Change | Reason | -|----------|--------|--------| -| Module level | Remove `_COMPANION_DEDUPE_PRUNE_THRESHOLD = 1000` | Replaced with inline literal `200`; no need for a named constant for a single usage site | -| `__init__` | Remove `self._route_tasks = set()`; add `self._in_flight = 0`, `self._max_in_flight = 30` | Replace set-based tracking with counter | -| `stop()` | Remove `_route_tasks` cancellation block | Tasks complete or are cancelled by event loop shutdown; explicit cancellation not needed | -| `_on_route_task_done` → `_on_route_done` | Simpler done-callback: decrement counter + log exceptions | Error logging preserved; set management removed | -| `_should_deliver_path_to_companions` | `> _COMPANION_DEDUPE_PRUNE_THRESHOLD` → `> 200` with explanatory comment | Lower threshold; comment explains the sizing rationale | -| `_process_queue` | Check `_in_flight >= _max_in_flight` before `create_task`; increment `_in_flight`; use `_on_route_done` | Cap accumulation; counter tracks live task count | - ---- - -## Test Plan - -### Unit tests (no hardware) - -**T1 — Counter increments and decrements correctly** - -```python -async def test_in_flight_counter(): - router = PacketRouter(mock_daemon) - await router.start() - - assert router._in_flight == 0 - - # Enqueue a packet that takes time to process - async def slow_route(pkt): - await asyncio.sleep(0.1) - - router._route_packet = slow_route - await router.enqueue(make_test_packet()) - await asyncio.sleep(0.01) # let queue loop run - - assert router._in_flight == 1 # task is sleeping - - await asyncio.sleep(0.15) # task finishes - assert router._in_flight == 0 # counter decremented by done-callback -``` - -**T2 — Cap enforced: packet dropped when at limit** - -```python -async def test_cap_drops_packet_at_limit(): - router = PacketRouter(mock_daemon) - router._max_in_flight = 2 - router._in_flight = 2 # simulate cap reached - - dropped = [] - original_create_task = asyncio.create_task - asyncio.create_task = lambda coro: dropped.append(coro) - - await router._process_queue_once(make_test_packet()) - - assert dropped == [], "create_task must not be called when cap is reached" - asyncio.create_task = original_create_task -``` - -**T3 — Exceptions in `_route_packet` are logged, not swallowed** - -```python -async def test_exception_logged(): - router = PacketRouter(mock_daemon) - - async def failing_route(pkt): - raise ValueError("simulated error") - - router._route_packet = failing_route - with patch("repeater.packet_router.logger") as mock_log: - task = asyncio.create_task(failing_route(make_test_packet())) - router._in_flight = 1 - task.add_done_callback(router._on_route_done) - await asyncio.gather(task, return_exceptions=True) - mock_log.error.assert_called_once() - - assert router._in_flight == 0 -``` - -**T4 — Companion dedup dict pruned at 200, not 1000** - -```python -def test_companion_dedup_prune_threshold(): - router = PacketRouter(mock_daemon) - future_time = time.time() + 999 - - # Fill with 199 entries (all unexpired) — no prune - router._companion_delivered = {f"key{i}": future_time for i in range(199)} - pkt = make_path_packet() - router._should_deliver_path_to_companions(pkt) - assert len(router._companion_delivered) == 200 # added one, no prune yet - - # 201st entry triggers prune — all unexpired so count stays at 201 - router._companion_delivered[f"key_extra"] = future_time - assert len(router._companion_delivered) == 201 - - # Force prune by making all existing entries expired - past_time = time.time() - 1 - router._companion_delivered = {f"key{i}": past_time for i in range(201)} - router._should_deliver_path_to_companions(pkt) - # All expired entries pruned; only the new entry remains - assert len(router._companion_delivered) == 1 -``` - -### Integration / field tests (with hardware) - -**T5 — Burst flood: verify cap fires under pathological load** - -1. Configure a test mesh with 4+ nodes all in range of the repeater. -2. Have all nodes send a flood packet simultaneously. -3. Observe repeater logs. - -**Expected:** `_in_flight` peaks in low single digits (LoRa airtime prevents -large bursts); no `"In-flight task cap reached"` warning fires under normal -conditions, confirming the cap is never a bottleneck in practice. - -**T6 — Counter reaches zero after all packets processed** - -1. Send a burst of 10 packets. -2. Wait 10 seconds (longer than max TX delay of 5 s). -3. Query `router._in_flight` from a debug endpoint or log. - -**Expected:** `_in_flight == 0` after all delays expire and packets transmit. - -**T7 — Error in `_route_packet` is logged and counter is decremented** - -1. Temporarily introduce a deliberate exception in `_route_packet`. -2. Send a packet. -3. Check logs for the error message and verify the repeater continues operating - (counter decremented, queue still draining). - -**T8 — Normal forwarding throughput unchanged** - -1. Send packets at a steady rate of 1 every 10 seconds for 5 minutes. -2. Verify all packets are forwarded with no warnings or errors. -3. Confirm `_in_flight` never exceeds 3–4 during normal operation. - ---- - -## Proof of Correctness - -### Counter vs set: why the counter is sufficient - -The `_route_tasks` set solved two problems: - -1. **GC protection**: In Python < 3.12, a task with no strong references other - than the event loop's internal weakref could be garbage collected before - completing. Python 3.12+ strengthened task references in the event loop. - However, even in earlier versions, the set was unnecessary once `create_task` - returns — the caller holds the reference, and the done-callback fires reliably - because the event loop holds the task alive until completion. - -2. **Explicit shutdown cancellation**: The counter loses this. As argued above, - the outcome is identical — sleeping tasks are cancelled either explicitly by - `stop()` or implicitly by the event loop at shutdown — and no packet that - hasn't been transmitted yet can complete its send after the radio is shut down - anyway. - -### Why `_on_route_done` is a done-callback and not a `try/finally` inside `_route_packet` - -A `try/finally` block inside `_route_packet` would also decrement the counter. -Done-callbacks are preferable because: - -- They fire even if the task is externally cancelled (e.g. by event loop shutdown), - whereas `finally` may not run if `CancelledError` is not caught. -- They decouple counter management from `_route_packet` logic — `_route_packet` - has no knowledge of or dependency on the cap mechanism. -- They keep the pattern consistent with the rest of the codebase's use of - `add_done_callback` for task lifecycle management. - -### Why 30 and not a smaller number like 10 - -At SF8, 125 kHz bandwidth, a 30-byte payload takes ~111 ms airtime and produces -a TX delay of roughly 0.5–3 s. With a 60-second duty-cycle window and 3.6 s -max airtime, the node can forward at most ~32 packets per minute at full budget. -If all 32 arrive within one second (they cannot physically, but as an upper -bound), 32 tasks would be in-flight simultaneously. A cap of 30 is aggressive -enough to protect against unbounded growth but not so low that it would drop -legitimate traffic under any realistic burst scenario. diff --git a/docs/pr_tx_serialization.md b/docs/pr_tx_serialization.md deleted file mode 100644 index b9e84d0..0000000 --- a/docs/pr_tx_serialization.md +++ /dev/null @@ -1,395 +0,0 @@ -# PR: Serialise Radio TX and Close Duty-Cycle TOCTOU Race - -**Branch:** `fix/tx-serialization` -**Base:** `rightup/fix-perfom-speed` -**Files changed:** `repeater/engine.py` (1 file, ~30 lines net) - ---- - -## Problem - -Two separate bugs share the same root cause: concurrent `delayed_send` coroutines -racing each other at transmission time. - -### Bug 1 — Interleaved SPI/serial commands to the radio - -The queue loop (added in an earlier commit) dispatches each incoming packet as an -`asyncio.create_task`, so multiple `delayed_send` coroutines can have their sleep -timers running concurrently. That is correct and intentional — it mirrors how -firmware nodes use a hardware timer so the radio keeps listening during a TX delay. - -However the LoRa radio is **half-duplex**: it can only transmit one packet at a -time. When two delay timers expire at nearly the same moment both coroutines call -`dispatcher.send_packet` simultaneously. `send_packet` issues a sequence of -SPI/serial register writes to the radio; two tasks interleaving these writes -produces undefined radio state and the transmission of neither packet is reliable. - -### Bug 2 — TOCTOU gap in duty-cycle enforcement - -`__call__` calls `can_transmit()` before scheduling a task: - -```python -# __call__ (before this fix) -can_tx, wait_time = self.airtime_mgr.can_transmit(airtime_ms) -if not can_tx: - ... # drop or defer -tx_task = await self.schedule_retransmit(fwd_pkt, delay, airtime_ms, ...) -``` - -`record_tx()` is only called later, inside `delayed_send`, after the sleep -completes. Between the check and the debit there is a window that spans the -entire TX delay (up to several seconds). Two packets that both pass the check -before either has slept and recorded its airtime will **both** be transmitted even -if transmitting both would exceed the duty-cycle budget. - -Under normal single-packet conditions this window is harmless. Under burst -conditions — multi-hop amplification, collision retries, or a busy mesh segment -where several packets arrive within the same delay window — multiple tasks pass -the advisory check simultaneously, and the duty-cycle limit is exceeded. - ---- - -## Root Cause - -There is no mutual exclusion around the radio send path. Each `delayed_send` -coroutine independently checks duty-cycle, sleeps, and transmits without -coordinating with any other concurrent coroutine doing the same thing. - ---- - -## Solution - -Add `self._tx_lock = asyncio.Lock()` (initialised in `__init__`) and acquire it -inside `delayed_send` **after** the sleep completes: - -``` -Delay timers run concurrently (unchanged): - Task A: sleep(1.2s) ──────────────────► acquire _tx_lock → check → TX A → release - Task B: sleep(0.9s) ──────────────────► acquire _tx_lock (waits) ──────────► check → TX B → release - Task C: sleep(2.1s) ────────────────────────────────────────────────────────────────► ... - -Radio: one packet at a time, duty-cycle state always stable inside the lock. -``` - -Inside the lock, a **second** `can_transmit()` call is made immediately before -sending. Because only one task holds the lock at a time, airtime state is stable -at this point and `record_tx()` follows on success — check and debit are -effectively atomic. This closes the TOCTOU window completely. - -The upfront `can_transmit()` in `__call__` is retained as an **advisory** fast -path: it still drops or defers packets that are obviously over budget before a -delay task is even scheduled, avoiding unnecessary sleep timers. It is no longer -the enforcement point. - ---- - -## Why This Is the Right Approach - -### Alternative A — Move `record_tx()` before the sleep - -```python -# hypothetical -self.airtime_mgr.record_tx(airtime_ms) # reserve before sleeping -await asyncio.sleep(delay) -await self.dispatcher.send_packet(...) # actual TX -``` - -Records airtime even if the send fails (exception, LBT busy, radio error) — -the budget is debited for a packet that was never transmitted. Over time this -inflates the apparent airtime, causing the node to throttle legitimate traffic -it actually has budget for. Requires a compensating `release_airtime()` on -every failure path, creating new complexity and failure modes. - -### Alternative B — A single global advisory check (status quo before this PR) - -Already demonstrated to fail under burst conditions (two tasks both pass before -either records its airtime). - -### Alternative C — asyncio.Lock (this PR) - -- Delay timers remain concurrent — no regression on the primary non-blocking TX - improvement. -- The check-and-debit pair is atomic within the lock — no TOCTOU window. -- No phantom airtime on send failure — `record_tx()` is only called on success. -- One `asyncio.Lock` object, no new state machines or compensating paths. -- The lock is `async`, so it only blocks other TX tasks, not the event loop or - the packet RX queue. - -### Why `asyncio.Lock` rather than `threading.Lock` - -The entire repeater runs on a single asyncio event loop. `asyncio.Lock` only -yields at `await` points; it does not involve OS threads or context switches. -A `threading.Lock` would work but is semantically wrong here (this is not a -thread-safety problem) and would block the event loop thread if held across an -`await`. - ---- - -## Changes - -### `repeater/engine.py` - -**1. Move `import random` to module level** - -```python -# before (inside _calculate_tx_delay): -def _calculate_tx_delay(self, packet, snr=0.0): - import random - ... - -# after (top of file, with other stdlib imports): -import random -``` - -This is a housekeeping fix bundled with this PR because `random` is a stdlib -module that should never be imported inside a hot-path function — Python caches -the import after the first call, but the attribute lookup and cache check still -run on every call. Moving it to module level is the standard pattern. - -**2. Add `self._tx_lock` to `__init__`** - -```python -# Serialise all radio TX calls. -# -# Background: since the queue loop dispatches each packet as an -# asyncio.create_task, multiple _route_packet coroutines can have their -# TX delay timers running concurrently — which is the intended behaviour -# (firmware nodes do the same with a hardware timer). However, the -# LoRa radio is half-duplex: it can only transmit one packet at a time. -# Without serialisation, two tasks whose delay timers expire near- -# simultaneously both call dispatcher.send_packet, interleaving SPI/serial -# commands to the radio and both passing the LBT check before either has -# actually transmitted. -# -# _tx_lock is acquired after each delay sleep and held for the entire -# send_packet call. Delays still run concurrently; only the radio -# access is serialised. This also eliminates the TOCTOU gap in duty-cycle -# enforcement — see schedule_retransmit / delayed_send for details. -self._tx_lock = asyncio.Lock() -``` - -**3. Acquire lock inside `delayed_send`, add authoritative duty-cycle gate** - -```python -async def delayed_send(): - await asyncio.sleep(delay) - - # Acquire the TX lock *after* the delay so that delay timers for - # multiple packets still run concurrently (matching firmware). Only - # one coroutine enters the radio send path at a time. - async with self._tx_lock: - # ── Authoritative duty-cycle gate ───────────────────────────── - # The upfront can_transmit() call in __call__ is advisory: it - # avoids scheduling packets that are obviously over budget, but - # it cannot prevent a race between two tasks whose delay timers - # expire at almost the same moment. Both tasks pass the advisory - # check before either has recorded its airtime, then both try to - # transmit. - # - # Inside _tx_lock only one task runs at a time, so airtime state - # is stable here. The check and the subsequent record_tx() are - # effectively atomic — no TOCTOU window. - if airtime_ms > 0: - can_tx_now, _ = self.airtime_mgr.can_transmit(airtime_ms) - if not can_tx_now: - logger.warning( - "Packet dropped at TX time: duty-cycle exceeded " - "(airtime=%.1fms)", airtime_ms, - ) - return - - last_error = None - for attempt in range(2 if local_transmission else 1): - try: - await self.dispatcher.send_packet(fwd_pkt, wait_for_ack=False) - self._record_packet_sent(fwd_pkt) - if airtime_ms > 0: - self.airtime_mgr.record_tx(airtime_ms) - ... -``` - ---- - -## Invariants Maintained - -| Property | Before | After | -|----------|--------|-------| -| Delay timers run concurrently | ✅ | ✅ | -| Radio accessed by one task at a time | ❌ | ✅ | -| Duty-cycle check and debit atomic | ❌ | ✅ | -| Airtime recorded only on TX success | ✅ | ✅ | -| Event loop not blocked by lock | ✅ | ✅ (asyncio.Lock) | - ---- - -## Test Plan - -### Unit tests (can run without hardware) - -**T1 — Serial TX ordering** - -```python -import asyncio -from unittest.mock import AsyncMock, MagicMock, patch - -async def test_tx_serialized(): - """Two tasks whose delays expire simultaneously must not interleave.""" - send_order = [] - send_lock = asyncio.Lock() - - async def mock_send(pkt, **kw): - # Confirm the _tx_lock is already held when we enter send_packet - assert send_lock.locked(), "send_packet called without _tx_lock held" - send_order.append(pkt) - await asyncio.sleep(0) # yield; a second task must not enter here - - engine._tx_lock = send_lock # replace with the mock lock reference - engine.dispatcher.send_packet = mock_send - - t1 = asyncio.create_task(engine.schedule_retransmit(pkt_a, delay=0.01, airtime_ms=100)) - t2 = asyncio.create_task(engine.schedule_retransmit(pkt_b, delay=0.01, airtime_ms=100)) - await asyncio.gather(t1, t2) - - assert len(send_order) == 2 # both transmitted - assert send_order[0] is not send_order[1] # different packets -``` - -**T2 — Authoritative duty-cycle gate blocks over-budget second packet** - -```python -async def test_second_packet_dropped_when_over_budget(): - """When first TX fills the budget, second task must be dropped inside the lock.""" - # Set a tiny budget: 50ms per minute - engine.airtime_mgr.max_airtime_per_minute = 50 - - sent = [] - async def mock_send(pkt, **kw): - sent.append(pkt) - - engine.dispatcher.send_packet = mock_send - - # Each packet costs ~111ms (SF8, BW125, 30-byte payload) — first passes, second must not - t1 = asyncio.create_task(engine.schedule_retransmit(pkt_a, delay=0.01, airtime_ms=111)) - t2 = asyncio.create_task(engine.schedule_retransmit(pkt_b, delay=0.01, airtime_ms=111)) - await asyncio.gather(t1, t2) - - assert len(sent) == 1, f"Expected 1 TX, got {len(sent)}" -``` - -**T3 — Airtime not debited on TX failure** - -```python -async def test_airtime_not_recorded_on_send_failure(): - before = engine.airtime_mgr.total_airtime_ms - - async def failing_send(pkt, **kw): - raise RuntimeError("radio error") - - engine.dispatcher.send_packet = failing_send - - with pytest.raises(RuntimeError): - await engine.schedule_retransmit(pkt, delay=0, airtime_ms=100) - - assert engine.airtime_mgr.total_airtime_ms == before, \ - "Airtime must not be recorded when send raises" -``` - -**T4 — Advisory check still drops before scheduling (fast path not regressed)** - -```python -async def test_advisory_check_still_drops_obvious_overage(): - """__call__ should not even schedule a task when clearly over budget.""" - engine.airtime_mgr.max_airtime_per_minute = 0 # budget exhausted - - tasks_created = [] - original = asyncio.create_task - asyncio.create_task = lambda coro: tasks_created.append(coro) or original(coro) - - await engine(over_budget_packet, metadata={}) - - assert not tasks_created, "No task should be created when advisory check fails" -``` - -### Integration / field tests (with hardware) - -**T5 — Burst scenario: 5 packets arrive within the same delay window** - -1. Connect the repeater to a radio. -2. Using a second node, send 5 FLOOD packets in quick succession (< 100 ms apart) - with a low RSSI score so the repeater's delay is ~1–2 s for all of them. -3. Monitor the radio with a spectrum analyser or a third node running in monitor - mode. - -**Expected (after this fix):** -- Transmissions are sequential — no overlapping on-air signals. -- `Retransmitted packet` log lines appear one after another, each with a non-zero - airtime value. -- No `Retransmit failed` errors in the log. -- Duty-cycle log shows airtime accumulating correctly. - -**Expected (before this fix, to confirm the bug existed):** -- Occasional `Retransmit failed` errors under burst load. -- Airtime tracking diverging from actual on-air time (double-counted or missed). - -**T6 — Duty-cycle enforcement under burst** - -1. Set `max_airtime_per_minute` to a low value (e.g. 500 ms) in config. -2. Send 10 packets rapidly so the repeater tries to forward all 10. -3. Observe logs. - -**Expected:** -- First N packets transmitted (total airtime ≤ 500 ms). -- Subsequent packets log `"Packet dropped at TX time: duty-cycle exceeded"` from - inside `delayed_send` (not just the advisory drop). -- `airtime_mgr.get_stats()["utilization_percent"]` reads ≤ 100%. - -**T7 — Normal single-packet forwarding not regressed** - -1. Send one packet every 5 seconds (well within duty-cycle budget). -2. Verify each packet is forwarded with correct airtime logged. -3. Verify no lock contention warnings in the log. - -**T8 — Local TX retry path (local_transmission=True) still works** - -1. Send a command that triggers a local transmission (e.g. a ping reply). -2. Briefly block the radio (simulate with a mock) so the first attempt fails. -3. Verify the retry fires after 1 s and the packet is eventually transmitted. - ---- - -## Proof of Correctness - -### Why `asyncio.Lock` is sufficient (no OS-level synchronisation needed) - -Python's asyncio event loop is **single-threaded**. All coroutines share one -thread and only yield execution at `await` points. Between two consecutive -`await` calls in a coroutine, the event loop does not switch to another coroutine. - -`asyncio.Lock.acquire()` suspends the current coroutine if the lock is held, -returning control to the event loop. `asyncio.Lock.release()` wakes the next -waiter. Because `send_packet` is awaited inside the lock, no other TX task can -run until the current one releases the lock and the event loop gets a chance to -schedule the next waiter. - -There is no possibility of the race seen with `threading.Lock` where an OS thread -can be preempted mid-instruction. - -### Why the advisory check in `__call__` cannot be removed - -The advisory check is still necessary as a fast path. If it were removed, every -incoming packet — even when the node is clearly at 100% duty-cycle — would -schedule a `delayed_send` task that would sleep for the full TX delay (up to -several seconds) before the lock drops it. Under a sustained flood of incoming -packets this wastes memory and CPU. The advisory check prunes the queue early at -negligible cost. - -### Why `record_tx()` must be inside the lock (not before or after) - -- **Before the send:** records airtime for a packet that may never be transmitted - (send could fail, LBT could reject it). Budget is overcounted. -- **After releasing the lock:** a second task could pass the authoritative - `can_transmit()` check between `send_packet` returning and `record_tx()` being - called — the TOCTOU window reopens at a smaller scale. -- **Inside the lock, after a successful send:** the budget is debited exactly once - for exactly the packets that were actually transmitted. The lock ensures no - other task reads airtime state between the check and the debit. diff --git a/repeater/sensors/ina219.py b/repeater/sensors/ina219.py index 9a0b9cf..a0f063b 100644 --- a/repeater/sensors/ina219.py +++ b/repeater/sensors/ina219.py @@ -1,7 +1,7 @@ """ INA219 current/voltage/power monitor sensor plug-in. -Requires: pip install adafruit-circuitpython-ina219 +Requires: pip install smbus2 Config example: - type: ina219 @@ -10,6 +10,7 @@ Config example: auto_install_packages: false settings: i2c_address: 0x40 # Default INA219 I2C address + bus_number: 1 # I2C bus number (1 for Raspberry Pi default) max_expected_amps: 2.0 shunt_ohms: 0.1 # 0.1 Ohm shunt resistor """ @@ -21,6 +22,17 @@ from typing import Any, Dict, Optional from .base import SensorBase from .registry import SensorRegistry +# INA219 register map +_REG_CONFIG = 0x00 +_REG_SHUNT_VOLTAGE = 0x01 +_REG_BUS_VOLTAGE = 0x02 +_REG_POWER = 0x03 +_REG_CURRENT = 0x04 +_REG_CALIBRATION = 0x05 + +# 32V range, 320mV shunt range, 12-bit ADC, continuous shunt+bus conversion +_CONFIG_32V_320MV_CONTINUOUS = 0x399F + @SensorRegistry.register("ina219") class INA219Sensor(SensorBase): @@ -29,63 +41,102 @@ class INA219Sensor(SensorBase): def __init__(self, name: str, config: Optional[Dict[str, Any]] = None, log=None): super().__init__(name=name, config=config, log=log) - self.i2c_address = self.settings.get("i2c_address", 0x40) + self.i2c_address = int(self.settings.get("i2c_address", 0x40)) + self.bus_number = int(self.settings.get("bus_number", 1)) self.max_expected_amps = float(self.settings.get("max_expected_amps", 2.0)) self.shunt_ohms = float(self.settings.get("shunt_ohms", 0.1)) + # INA219 calibration math from datasheet + if self.max_expected_amps <= 0: + self.max_expected_amps = 2.0 + if self.shunt_ohms <= 0: + self.shunt_ohms = 0.1 + + self.current_lsb = self.max_expected_amps / 32768.0 + cal = int(0.04096 / (self.current_lsb * self.shunt_ohms)) + self.calibration = max(1, min(cal, 0xFFFF)) + self.power_lsb = self.current_lsb * 20.0 + self.available = False if not self.ensure_python_modules( [ - ("board", "board"), - ("busio", "adafruit-blinka"), - ("adafruit_ina219", "adafruit-circuitpython-ina219"), + ("smbus2", "smbus2"), ] ): return try: - import board # type: ignore[import-not-found] - import busio # type: ignore[import-not-found] - from adafruit_ina219 import Adafruit_INA219 # type: ignore[import-not-found] + import smbus2 # type: ignore[import-not-found] - # Create I2C interface - i2c = busio.I2C(board.SCL, board.SDA) + self._smbus2 = smbus2 - # Create sensor object - self.ina219 = Adafruit_INA219( - i2c_addr=self.i2c_address, - i2c=i2c, - ) - - # Configure for expected current range - self.ina219.set_calibration_32V_2A() if self.max_expected_amps <= 2.0 else None + # Verify bus is accessible and program sensor once + bus = smbus2.SMBus(self.bus_number) + try: + self._write_register(bus, _REG_CONFIG, _CONFIG_32V_320MV_CONTINUOUS) + self._write_register(bus, _REG_CALIBRATION, self.calibration) + finally: + bus.close() self.available = True self.log.info( - "INA219 initialized (addr=0x%02X, shunt=%.3fΩ, max_A=%.1f)", + "INA219 initialized (addr=0x%02X, bus=%d, shunt=%.3fΩ, max_A=%.1f)", self.i2c_address, + self.bus_number, self.shunt_ohms, self.max_expected_amps, ) except Exception as exc: self.log.warning( - "INA219 init failed (addr=0x%02X): %s", + "INA219 init failed (addr=0x%02X, bus=%d): %s", self.i2c_address, + self.bus_number, exc, ) self.available = False + @staticmethod + def _swap_word(value: int) -> int: + return ((value & 0xFF) << 8) | ((value >> 8) & 0xFF) + + def _write_register(self, bus, register: int, value: int) -> None: + bus.write_word_data(self.i2c_address, register, self._swap_word(value & 0xFFFF)) + + def _read_register(self, bus, register: int) -> int: + return self._swap_word(bus.read_word_data(self.i2c_address, register)) + + @staticmethod + def _to_signed_16(value: int) -> int: + return value - 0x10000 if value & 0x8000 else value + def _read(self) -> Dict[str, Any]: """Read voltage, current, and power from INA219.""" if not self.available: raise RuntimeError("INA219 device not available") - + try: + bus = self._smbus2.SMBus(self.bus_number) + try: + # Reapply calibration in case the chip was reset externally. + self._write_register(bus, _REG_CALIBRATION, self.calibration) + + raw_bus = self._read_register(bus, _REG_BUS_VOLTAGE) + raw_shunt = self._to_signed_16(self._read_register(bus, _REG_SHUNT_VOLTAGE)) + raw_current = self._to_signed_16(self._read_register(bus, _REG_CURRENT)) + raw_power = self._read_register(bus, _REG_POWER) + finally: + bus.close() + + bus_voltage_v = ((raw_bus >> 3) & 0x1FFF) * 0.004 + shunt_voltage_v = raw_shunt * 0.00001 + current_ma = raw_current * self.current_lsb * 1000.0 + power_mw = raw_power * self.power_lsb * 1000.0 + return { - "bus_voltage_v": round(self.ina219.bus_voltage, 3), - "shunt_voltage_v": round(self.ina219.shunt_voltage, 4), - "current_ma": round(self.ina219.current, 2), - "power_mw": round(self.ina219.power, 2), + "bus_voltage_v": round(bus_voltage_v, 3), + "shunt_voltage_v": round(shunt_voltage_v, 5), + "current_ma": round(current_ma, 2), + "power_mw": round(power_mw, 2), } except Exception as exc: raise RuntimeError(f"INA219 read failed: {exc}") from exc