rt/Reticulum
1
0
Fork 0
mirror of https://github.com/markqvist/Reticulum.git synced 2026-10-25 18:40:17 -04:00
Code Issues Projects Releases Packages Wiki Activity Actions

Track in-flight path requests separately

Browse source
This commit is contained in:
Mark Qvist 2026-08-21 23:16:20 +02:00
parent 4c69b376b2
commit b1aa527861
No known key found for this signature in database
1 changed files with 41 additions and 3 deletions
Download patch file Download diff file Expand all files Collapse all files

44
RNS/Transport.py
View file

@ -180,6 +180,7 @@ class Transport:
tunnels = {} # A table storing tunnels to other transport instances
announce_rate_table = {} # A table for keeping track of announce rates
path_requests = {} # A table for storing path request timestamps
inflight_path_requests = {} # A table for tracking in-flight path request
path_states = {} # A table for keeping track of path states
blackholed_identities = {} # A table for keeping track of blackholed identities
@ -208,6 +209,7 @@ class Transport:
discovery_pr_lock = Lock()
discovery_pr_tags_lock = Lock()
path_requests_lock = Lock()
inflight_path_requests_lock = Lock()
pending_local_prs_lock = Lock()
path_states_lock = Lock()
jobs_lock = Lock()
@ -958,6 +960,18 @@ class Transport:
except Exception as e:
RNS.log(f"Could not complete stale path request enumeration in this job round, retrying later: {e}", RNS.LOG_WARNING)
# Cull the pending in-flight path requests table
stale_inflight_path_requests = []
try:
inflight_path_requests_snapshot = Transport.inflight_path_requests.copy()
for destination_hash in inflight_path_requests_snapshot:
if time.time() > inflight_path_requests_snapshot[destination_hash] + Transport.PATH_REQUEST_GATE_TIMEOUT:
stale_inflight_path_requests.append(destination_hash)
RNS.log("In-flight path request entry for "+RNS.prettyhexrep(destination_hash)+" timed out and was removed", RNS.LOG_EXTREME) if RNS.sl(RNS.LOG_EXTREME) else None
except Exception as e:
RNS.log(f"Could not complete stale in-flight path request enumeration in this job round, retrying later: {e}", RNS.LOG_WARNING)
# Cull the pending discovery path requests table
stale_discovery_path_requests = []
with Transport.discovery_pr_lock:
@ -1061,6 +1075,16 @@ class Transport:
if i == 1: RNS.log("Removed "+str(i)+" path request entry", RNS.LOG_EXTREME) if RNS.sl(RNS.LOG_EXTREME) else None
else: RNS.log("Removed "+str(i)+" path request entries", RNS.LOG_EXTREME) if RNS.sl(RNS.LOG_EXTREME) else None
i = 0
with Transport.inflight_path_requests_lock:
for destination_hash in stale_inflight_path_requests:
Transport.inflight_path_requests.pop(destination_hash)
i += 1
if i > 0:
if i == 1: RNS.log("Removed "+str(i)+" in-flight path request entry", RNS.LOG_EXTREME) if RNS.sl(RNS.LOG_EXTREME) else None
else: RNS.log("Removed "+str(i)+" in-flight path request entries", RNS.LOG_EXTREME) if RNS.sl(RNS.LOG_EXTREME) else None
i = 0
with Transport.discovery_pr_lock:
for destination_hash in stale_discovery_path_requests:
@ -1738,12 +1762,12 @@ class Transport:
traffic_class = Transport.TC_INGRESS_LIMITED
path_request_inflight = False
with Transport.path_requests_lock:
if destination_hash in Transport.path_requests: path_request_inflight = True
with Transport.inflight_path_requests_lock:
if destination_hash in Transport.inflight_path_requests: path_request_inflight = True
if not path_request_inflight:
# Early registration to immediately batch duplicates
with Transport.path_requests_lock: Transport.path_requests[destination_hash] = time.time()
with Transport.inflight_path_requests_lock: Transport.inflight_path_requests[destination_hash] = time.time()
else:
if not traffic_class == Transport.TC_INGRESS_LIMITED:
RNS.log(f"Path request on {packet.receiving_interface} for {RNS.prettyhexrep(destination_hash)} already in-flight, batching to existing", RNS.LOG_PATHING) if RNS.sl(RNS.LOG_PATHING) else None
@ -2360,6 +2384,11 @@ class Transport:
tunnel_entry[IDX_TT_EXPIRES] = expires
RNS.log("Path to "+RNS.prettyhexrep(packet.destination_hash)+" associated with tunnel "+RNS.prettyhexrep(packet.receiving_interface.tunnel_id), RNS.LOG_PATHING) if RNS.sl(RNS.LOG_PATHING) else None
# Resolve potential in-flight path requests
with Transport.inflight_path_requests_lock:
if packet.destination_hash in Transport.inflight_path_requests:
Transport.inflight_path_requests.pop(packet.destination_hash)
# Call externally registered callbacks from apps
# wanting to know when an announce arrives
with Transport.announce_handler_lock:
@ -3305,6 +3334,7 @@ class Transport:
should_search_for_unknown = False
should_ingress_limit = False
search_mode_filter = None
answered = False
if attached_interface != None:
should_ingress_limit = ingress_limited or attached_interface.should_ingress_limit_pr()
@ -3335,6 +3365,7 @@ class Transport:
if destination_hash in Transport.destinations_map: local_destination = Transport.destinations_map[destination_hash]
if local_destination != None:
answered = True
local_destination.announce(path_response=True, tag=tag, attached_interface=attached_interface)
RNS.log("Answering path request for "+RNS.prettyhexrep(destination_hash)+interface_str+", destination is local to this system", RNS.LOG_PATHING) if RNS.sl(RNS.LOG_PATHING) else None
@ -3342,6 +3373,7 @@ class Transport:
packet = Transport.get_cached_packet(Transport.path_table[destination_hash][IDX_PT_PACKET], packet_type="announce")
next_hop = Transport.path_table[destination_hash][IDX_PT_NEXT_HOP]
received_from = Transport.path_table[destination_hash][IDX_PT_RVCD_IF]
answered = True
if packet == None:
RNS.log(f"Could not retrieve announce packet from cache while answering path request for {RNS.prettyhexrep(destination_hash)}, ignoring path request", RNS.LOG_DEBUG) if RNS.sl(RNS.LOG_DEBUG) else None
@ -3473,6 +3505,12 @@ class Transport:
else:
RNS.log("Ignoring path request for "+RNS.prettyhexrep(destination_hash)+interface_str+", no path known", RNS.LOG_PATHING) if RNS.sl(RNS.LOG_PATHING) else None
if answered:
# Resolve potential in-flight path requests
with Transport.inflight_path_requests_lock:
if destination_hash in Transport.inflight_path_requests:
Transport.inflight_path_requests.pop(destination_hash)
@staticmethod
def from_local_client(packet):
if hasattr(packet.receiving_interface, "parent_interface"):