adam/pyMC_Repeater
1
0
Fork 0
mirror of https://github.com/pyMC-dev/pyMC_Repeater.git synced 2026-07-06 01:42:41 +02:00
Code Issues Packages Projects Releases Wiki Activity

perf: compute packet hash once per packet in the forwarding hot path

Browse Source 
Before this change, calculate_packet_hash() (SHA-256 + hex + upper) was called
3 times per forwarded packet and 4 times per dropped packet:
  __call__              → pkt_hash_full = packet.calculate_packet_hash()   #1
  → flood/direct_forward → is_duplicate → calculate_packet_hash()          #2
  → flood/direct_forward → mark_seen    → calculate_packet_hash()          #3
  (drop) → _get_drop_reason → is_duplicate → calculate_packet_hash()       #4

pkt_hash_full was computed in __call__ but never threaded down into
process_packet, flood_forward, direct_forward, is_duplicate, or _get_drop_reason.
Each method recomputed it independently.

Fix: add optional packet_hash: Optional[str] = None to is_duplicate,
_get_drop_reason, flood_forward, direct_forward, and process_packet.  Pass
pkt_hash_full from __call__ through the chain.  Each method uses the provided
hash or falls back to computing it — preserving backward compatibility for
external callers (TraceHelper, etc.) that have no pre-computed hash.

Result: 1 SHA-256 computation per packet in the hot path regardless of whether
the packet is forwarded or dropped.

Also adds explicit INVARIANT docstrings to flood_forward, direct_forward, and
is_duplicate documenting that these methods must remain synchronous (no await).
The is_duplicate + mark_seen pair is atomic within the asyncio event loop; adding
an await between them would allow two concurrent tasks to both pass the duplicate
check for the same packet — forwarding it twice.

Docs: docs/pr_hash_once.md — problem analysis, call-chain diagram, per-method
diffs, quantification (~3-8 µs saved per packet), test plan (including hash-count
assertion), and proof that passing the original's hash to the deep-copied packet
is correct.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
TJ Downes
2026-04-21 19:28:45 -07:00
parent c82f0cfce6
commit 4e16fd040d
2 changed files with 397 additions and 21 deletions
Download Patch File Download Diff File Expand all files Collapse all files
repeater/engine.py
+51 -21
View File
@@ -189,11 +189,12 @@ class RepeaterHandler(BaseHandler):
original_path_hashes = packet.get_path_hashes_hex()
path_hash_size = packet.get_path_hash_size()
# Process for forwarding (skip if repeat disabled or if this is a local transmission)
# Process for forwarding (skip if repeat disabled or if this is a local transmission).
# Pass pkt_hash_full so flood_forward / direct_forward don't recompute SHA-256.
result = (
None
if (not allow_forward or local_transmission)
else self.process_packet(processed_packet, snr)
else self.process_packet(processed_packet, snr, packet_hash=pkt_hash_full)
)
forwarded_path_hashes = None
@@ -303,7 +304,7 @@ class RepeaterHandler(BaseHandler):
else:
# Check if packet has a specific drop reason set by handlers
drop_reason = processed_packet.drop_reason or self._get_drop_reason(
processed_packet
processed_packet, packet_hash=pkt_hash_full
)
logger.debug(f"Packet not forwarded: {drop_reason}")
@@ -612,9 +613,9 @@ class RepeaterHandler(BaseHandler):
if pkt_hash:
self._recent_hash_index[pkt_hash] = packet_record
def _get_drop_reason(self, packet: Packet) -> str:
def _get_drop_reason(self, packet: Packet, packet_hash: Optional[str] = None) -> str:
if self.is_duplicate(packet):
if self.is_duplicate(packet, packet_hash=packet_hash):
return "Duplicate"
if not packet or not packet.payload:
@@ -642,12 +643,20 @@ class RepeaterHandler(BaseHandler):
# Default reason
return "Unknown"
def is_duplicate(self, packet: Packet) -> bool:
def is_duplicate(self, packet: Packet, packet_hash: Optional[str] = None) -> bool:
"""Return True if this packet has already been seen.
pkt_hash = packet.calculate_packet_hash().hex().upper()
if pkt_hash in self.seen_packets:
return True
return False
Accepts an optional pre-computed packet_hash to avoid a redundant SHA-256
when the caller (e.g. __call__ → process_packet → flood/direct_forward)
has already calculated the hash. Falls back to computing it if not provided.
INVARIANT: this method is synchronous with no await points. The caller
(process_packet / __call__) relies on is_duplicate + mark_seen being
effectively 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
def mark_seen(self, packet: Packet, packet_hash: Optional[str] = None):
@@ -789,8 +798,13 @@ class RepeaterHandler(BaseHandler):
logger.error(f"Transport code validation error: {e}")
return False, f"Transport code validation error: {e}"
def flood_forward(self, packet: Packet) -> Optional[Packet]:
def flood_forward(self, packet: Packet, packet_hash: Optional[str] = None) -> Optional[Packet]:
"""Forward a FLOOD packet, appending our hash to the path.
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.
"""
# Validate
valid, reason = self.validate_packet(packet)
if not valid:
@@ -824,8 +838,8 @@ class RepeaterHandler(BaseHandler):
packet.drop_reason = f"FLOOD loop detected ({mode})"
return None
# Suppress duplicates
if self.is_duplicate(packet):
# Suppress duplicates — pass pre-computed hash to avoid a second SHA-256.
if self.is_duplicate(packet, packet_hash=packet_hash):
packet.drop_reason = "Duplicate"
return None
@@ -847,7 +861,7 @@ class RepeaterHandler(BaseHandler):
packet.drop_reason = "Path would exceed MAX_PATH_SIZE"
return None
self.mark_seen(packet)
self.mark_seen(packet, packet_hash=packet_hash)
# Append hash_size bytes from our public key prefix
packet.path.extend(self.local_hash_bytes[:hash_size])
@@ -855,8 +869,13 @@ class RepeaterHandler(BaseHandler):
return packet
def direct_forward(self, packet: Packet) -> Optional[Packet]:
def direct_forward(self, packet: Packet, packet_hash: Optional[str] = None) -> Optional[Packet]:
"""Forward a DIRECT packet, removing the first hop from the path.
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.
"""
# Validate packet (empty payload, oversized path, etc.)
valid, reason = self.validate_packet(packet)
if not valid:
@@ -882,12 +901,12 @@ class RepeaterHandler(BaseHandler):
packet.drop_reason = "Direct: not for us"
return None
# Suppress duplicates
if self.is_duplicate(packet):
# Suppress duplicates — pass pre-computed hash to avoid a second SHA-256.
if self.is_duplicate(packet, packet_hash=packet_hash):
packet.drop_reason = "Duplicate"
return None
self.mark_seen(packet)
self.mark_seen(packet, packet_hash=packet_hash)
# Remove first hash entry (hash_size bytes)
packet.path = bytearray(packet.path[hash_size:])
@@ -969,19 +988,30 @@ class RepeaterHandler(BaseHandler):
return delay_s
def process_packet(self, packet: Packet, snr: float = 0.0) -> Optional[Tuple[Packet, float]]:
def process_packet(
self,
packet: Packet,
snr: float = 0.0,
packet_hash: Optional[str] = None,
) -> Optional[Tuple[Packet, float]]:
"""Route a received packet to flood_forward or direct_forward.
packet_hash is the pre-computed SHA-256 hex string from __call__.
Passing it here avoids recomputing the hash in flood_forward /
direct_forward / is_duplicate / mark_seen — reducing SHA-256 calls
from 3 per forwarded packet to 1.
"""
route_type = packet.header & PH_ROUTE_MASK
if route_type == ROUTE_TYPE_FLOOD or route_type == ROUTE_TYPE_TRANSPORT_FLOOD:
fwd_pkt = self.flood_forward(packet)
fwd_pkt = self.flood_forward(packet, packet_hash=packet_hash)
if fwd_pkt is None:
return None
delay = self._calculate_tx_delay(fwd_pkt, snr)
return fwd_pkt, delay
elif route_type == ROUTE_TYPE_DIRECT or route_type == ROUTE_TYPE_TRANSPORT_DIRECT:
fwd_pkt = self.direct_forward(packet)
fwd_pkt = self.direct_forward(packet, packet_hash=packet_hash)
if fwd_pkt is None:
return None
delay = self._calculate_tx_delay(fwd_pkt, snr)