adam/pyMC_Repeater
1
0
Fork 0
mirror of https://github.com/pyMC-dev/pyMC_Repeater.git synced 2026-08-07 17:33:16 +02:00
Code Issues Packages Projects Releases Wiki Activity

Refactor INA219 sensor integration.

Browse Source
This commit is contained in:
Lloyd
2026-05-13 12:34:05 +01:00
parent 9dbf5459c8
commit 0e7bb05208
5 changed files with 77 additions and 1115 deletions
Download Patch File Download Diff File Expand all files Collapse all files
config.yaml.example
+1
View File
@@ -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
docs/pr_hash_once.md
-346
View File
@@ -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.
docs/pr_in_flight_cap.md
-349
View File
@@ -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.
docs/pr_tx_serialization.md
-395
View File
@@ -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.
repeater/sensors/ina219.py
+76 -25
View File
@@ -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