Mirror/Reticulum
1
0
Fork 0
mirror of https://github.com/markqvist/Reticulum.git synced 2026-07-20 06:58:10 -07:00
Code Issues Packages Projects Releases Wiki Activity

Improved resource handling

Browse Source
This commit is contained in:
Mark Qvist
2026-07-18 02:46:06 +02:00
parent cf6010da59
commit 3a36c367fe
2 changed files with 112 additions and 126 deletions
Download Patch File Download Diff File Expand all files Collapse all files
RNS/Link.py
+34 -31
View File
@@ -808,7 +808,7 @@ class Link:
request_data = unpacked_request[2]
if path_hash in self.destination.request_handlers:
request_handler = self.destination.request_handlers[path_hash]
request_handler = self.destination.request_handlers[path_hash]
path = request_handler[0]
response_generator = request_handler[1]
allow = request_handler[2]
@@ -1014,36 +1014,39 @@ class Link:
packet.plaintext = self.decrypt(packet.data)
if packet.plaintext != None:
self.__update_phy_stats(packet, query_shared=True)
if RNS.ResourceAdvertisement.is_request(packet):
RNS.Resource.accept(packet, callback=self.request_resource_concluded)
elif RNS.ResourceAdvertisement.is_response(packet):
request_id = RNS.ResourceAdvertisement.read_request_id(packet)
for pending_request in self.pending_requests:
if pending_request.request_id == request_id:
response_resource = RNS.Resource.accept(packet, callback=self.response_resource_concluded, progress_callback=pending_request.response_resource_progress, request_id = request_id)
if response_resource != None:
if pending_request.response_size == None:
pending_request.response_size = RNS.ResourceAdvertisement.read_size(packet)
if pending_request.response_transfer_size == None:
pending_request.response_transfer_size = 0
pending_request.response_transfer_size += RNS.ResourceAdvertisement.read_transfer_size(packet)
if pending_request.started_at == None:
pending_request.started_at = time.time()
pending_request.response_resource_progress(response_resource)
elif self.resource_strategy == Link.ACCEPT_NONE: pass
elif self.resource_strategy == Link.ACCEPT_APP:
if self.callbacks.resource != None:
try:
resource_advertisement = RNS.ResourceAdvertisement.unpack(packet.plaintext)
resource_advertisement.link = self
if self.callbacks.resource(resource_advertisement): RNS.Resource.accept(packet, self.callbacks.resource_concluded)
else: RNS.Resource.reject(packet)
except Exception as e:
RNS.log("Error while executing resource accept callback from "+str(self)+". The contained exception was: "+str(e), RNS.LOG_ERROR)
elif self.resource_strategy == Link.ACCEPT_ALL:
RNS.Resource.accept(packet, self.callbacks.resource_concluded)
try:
if RNS.ResourceAdvertisement.is_request(packet):
if self.destination.request_handlers:
RNS.Resource.accept(packet, callback=self.request_resource_concluded)
elif RNS.ResourceAdvertisement.is_response(packet):
request_id = RNS.ResourceAdvertisement.read_request_id(packet)
for pending_request in self.pending_requests:
if pending_request.request_id == request_id:
response_resource = RNS.Resource.accept(packet, callback=self.response_resource_concluded, progress_callback=pending_request.response_resource_progress, request_id = request_id)
if response_resource != None:
if pending_request.response_size == None:
pending_request.response_size = RNS.ResourceAdvertisement.read_size(packet)
if pending_request.response_transfer_size == None:
pending_request.response_transfer_size = 0
pending_request.response_transfer_size += RNS.ResourceAdvertisement.read_transfer_size(packet)
if pending_request.started_at == None:
pending_request.started_at = time.time()
pending_request.response_resource_progress(response_resource)
elif self.resource_strategy == Link.ACCEPT_NONE: pass
elif self.resource_strategy == Link.ACCEPT_APP:
if self.callbacks.resource != None:
try:
resource_advertisement = RNS.ResourceAdvertisement.unpack(packet.plaintext)
resource_advertisement.link = self
if self.callbacks.resource(resource_advertisement): RNS.Resource.accept(packet, self.callbacks.resource_concluded)
else: RNS.Resource.reject(packet)
except Exception as e:
RNS.log("Error while executing resource accept callback from "+str(self)+". The contained exception was: "+str(e), RNS.LOG_ERROR)
elif self.resource_strategy == Link.ACCEPT_ALL:
RNS.Resource.accept(packet, self.callbacks.resource_concluded)
except Exception as e:
RNS.log(f"Invalid resource advertisement on {self}: {e}", RNS.LOG_DEBUG) if RNS.sl(RNS.LOG_DEBUG) else None
self.teardown()
elif packet.context == RNS.Packet.RESOURCE_REQ:
plaintext = self.decrypt(packet.data)
RNS/Resource.py
+78 -95
View File
@@ -239,7 +239,7 @@ class Resource:
return None
except Exception as e:
RNS.log("Could not decode resource advertisement, dropping resource", RNS.LOG_DEBUG) if RNS.sl(RNS.LOG_DEBUG) else None
RNS.log(f"Could not decode resource advertisement, dropping resource: {e}", RNS.LOG_DEBUG) if RNS.sl(RNS.LOG_DEBUG) else None
return None
# Create a resource for transmission to a remote destination
@@ -314,30 +314,26 @@ class Resource:
self.input_file = data
elif isinstance(data, bytes):
data_size = len(data)
data_size = len(data)
self.total_size = data_size + self.metadata_size
resource_data = data
resource_data = data
self.total_segments = 1
self.segment_index = 1
self.split = False
elif data == None:
pass
elif data == None: pass
else:
raise TypeError("Invalid data instance type passed to resource initialisation")
else: raise TypeError("Invalid data instance type passed to resource initialisation")
if resource_data:
if self.has_metadata: data = self.metadata + resource_data
else: data = resource_data
self.status = Resource.NONE
self.link = link
if self.link.mtu:
self.sdu = self.link.mtu - RNS.Reticulum.HEADER_MAXSIZE - RNS.Reticulum.IFAC_MIN_SIZE
else:
self.sdu = link.mdu or Resource.SDU
self.link = link
if self.link.mtu: self.sdu = self.link.mtu - RNS.Reticulum.HEADER_MAXSIZE - RNS.Reticulum.IFAC_MIN_SIZE
else: self.sdu = link.mdu or Resource.SDU
self.max_retries = Resource.MAX_RETRIES
self.max_adv_retries = Resource.MAX_ADV_RETRIES
self.retries_left = self.max_retries
@@ -482,12 +478,12 @@ class Resource:
def hashmap_update_packet(self, plaintext):
if not self.status == Resource.FAILED:
self.last_activity = time.time()
self.retries_left = self.max_retries
update = umsgpack.unpackb(plaintext[RNS.Identity.HASHLENGTH//8:])
self.hashmap_update(update[0], update[1])
if self.waiting_for_hmu:
self.last_activity = time.time()
self.retries_left = self.max_retries
update = umsgpack.unpackb(plaintext[RNS.Identity.HASHLENGTH//8:])
self.hashmap_update(update[0], update[1])
def hashmap_update(self, segment, hashmap):
if not self.status == Resource.FAILED:
@@ -495,12 +491,16 @@ class Resource:
seg_len = ResourceAdvertisement.HASHMAP_MAX_LEN
hashes = len(hashmap)//Resource.MAPHASH_LEN
for i in range(0,hashes):
if self.hashmap[i+segment*seg_len] == None:
self.hashmap_height += 1
if self.hashmap[i+segment*seg_len] == None: self.hashmap_height += 1
self.hashmap[i+segment*seg_len] = hashmap[i*Resource.MAPHASH_LEN:(i+1)*Resource.MAPHASH_LEN]
self.waiting_for_hmu = False
self.request_next()
if hashes < 1:
RNS.log("Invalid HMU received, cancelling transfer", RNS.LOG_ERROR)
self.cancel()
else:
self.waiting_for_hmu = False
self.request_next()
def get_map_hash(self, data):
return RNS.Identity.full_hash(data+self.random_hash)[:Resource.MAPHASH_LEN]
@@ -517,6 +517,14 @@ class Resource:
prepare_thread = threading.Thread(target=self.__prepare_next_segment, daemon=True)
prepare_thread.start()
def ensure_link(self):
if not self.link or self.link.status != RNS.Link.ACTIVE:
RNS.log(f"Invalid link state for {self}, aborting transfer", RNS.LOG_VERBOSE)
try: self.cancel()
except Exception as e: RNS.log(f"Error while cancelling resource on link-state abort: {e}", RNS.LOG_ERROR)
return False
else: return True
def __advertise_job(self):
self.advertisement_packet = RNS.Packet(self.link, ResourceAdvertisement(self).pack(), context=RNS.Packet.RESOURCE_ADV)
while not self.link.ready_for_new_resource():
@@ -524,6 +532,7 @@ class Resource:
sleep(0.25)
try:
if not self.ensure_link(): return
self.advertisement_packet.send()
self.last_activity = time.time()
self.started_transferring = self.last_activity
@@ -541,18 +550,13 @@ class Resource:
self.watchdog_job()
def update_eifr(self):
if self.rtt == None:
rtt = self.link.rtt
else:
rtt = self.rtt
if self.rtt == None: rtt = self.link.rtt
else: rtt = self.rtt
if self.req_data_rtt_rate != 0:
expected_inflight_rate = self.req_data_rtt_rate*8
if self.req_data_rtt_rate != 0: expected_inflight_rate = self.req_data_rtt_rate*8
else:
if self.previous_eifr != None:
expected_inflight_rate = self.previous_eifr
else:
expected_inflight_rate = self.link.establishment_cost*8 / rtt
if self.previous_eifr != None: expected_inflight_rate = self.previous_eifr
else: expected_inflight_rate = self.link.establishment_cost*8 / rtt
self.eifr = expected_inflight_rate
if self.link: self.link.expected_rate = self.eifr
@@ -566,8 +570,7 @@ class Resource:
this_job_id = self.__watchdog_job_id
while self.status < Resource.ASSEMBLING and this_job_id == self.__watchdog_job_id:
while self.watchdog_lock:
sleep(0.025)
while self.watchdog_lock: sleep(0.025)
sleep_time = None
if self.status == Resource.ADVERTISED:
@@ -581,6 +584,7 @@ class Resource:
try:
RNS.log("No part requests received, retrying resource advertisement...", RNS.LOG_DEBUG) if RNS.sl(RNS.LOG_DEBUG) else None
self.retries_left -= 1
if not self.ensure_link(): return
self.advertisement_packet = RNS.Packet(self.link, ResourceAdvertisement(self).pack(), context=RNS.Packet.RESOURCE_ADV)
self.advertisement_packet.send()
self.last_activity = time.time()
@@ -657,13 +661,13 @@ class Resource:
self.last_part_sent = time.time()
sleep_time = 0.001
elif self.status == Resource.REJECTED:
elif self.status >= Resource.ASSEMBLING:
sleep_time = 0.001
if sleep_time == 0:
RNS.log("Warning! Link watchdog sleep time of 0!", RNS.LOG_DEBUG) if RNS.sl(RNS.LOG_DEBUG) else None
if sleep_time == None or sleep_time < 0:
RNS.log("Timing error, cancelling resource transfer.", RNS.LOG_ERROR)
RNS.log(f"Timing error ({sleep_time}/{self.status}), cancelling resource transfer.", RNS.LOG_ERROR)
self.cancel()
if sleep_time != None:
@@ -754,6 +758,7 @@ class Resource:
try:
proof = RNS.Identity.full_hash(self.data+self.hash)
proof_data = self.hash+proof
if not self.ensure_link(): return
proof_packet = RNS.Packet(self.link, proof_data, packet_type=RNS.Packet.PROOF, context=RNS.Packet.RESOURCE_PRF)
proof_packet.send()
RNS.Transport.cache(proof_packet, force_cache=True)
@@ -776,8 +781,8 @@ class Resource:
advertise = False,
auto_compress = self.auto_compress_option,
sent_metadata_size = self.metadata_size)
if self.__progress_callback:
self.next_segment.progress_callback(self.__progress_callback)
if self.__progress_callback: self.next_segment.progress_callback(self.__progress_callback)
def validate_proof(self, proof_data):
if not self.status == Resource.FAILED:
@@ -966,6 +971,7 @@ class Resource:
request_packet = RNS.Packet(self.link, request_data, context = RNS.Packet.RESOURCE_REQ)
try:
if not self.ensure_link(): return
request_packet.send()
self.last_activity = time.time()
self.req_sent = self.last_activity
@@ -1010,11 +1016,11 @@ class Resource:
for part in requested_parts:
try:
if not self.ensure_link(): return
if not part.sent:
part.send()
self.sent_parts += 1
else:
part.resend()
else: part.resend()
self.last_activity = time.time()
self.last_part_sent = self.last_activity
@@ -1032,8 +1038,7 @@ class Resource:
search_end = self.receiver_min_consecutive_height+ResourceAdvertisement.COLLISION_GUARD_SIZE
for part in self.parts[search_start:search_end]:
part_index += 1
if part.map_hash == last_map_hash:
break
if part.map_hash == last_map_hash: break
self.receiver_min_consecutive_height = max(part_index-1-Resource.WINDOW_MAX, 0)
@@ -1041,10 +1046,8 @@ class Resource:
RNS.log("Resource sequencing error, cancelling transfer!", RNS.LOG_ERROR)
self.cancel()
return
else:
segment = part_index // ResourceAdvertisement.HASHMAP_MAX_LEN
else: segment = part_index // ResourceAdvertisement.HASHMAP_MAX_LEN
hashmap_start = segment*ResourceAdvertisement.HASHMAP_MAX_LEN
hashmap_end = min((segment+1)*ResourceAdvertisement.HASHMAP_MAX_LEN, len(self.parts))
@@ -1052,10 +1055,16 @@ class Resource:
for i in range(hashmap_start,hashmap_end):
hashmap += self.hashmap[i*Resource.MAPHASH_LEN:(i+1)*Resource.MAPHASH_LEN]
if not hashmap:
RNS.log("Resource HMU error, cancelling transfer!", RNS.LOG_ERROR)
self.cancel()
return
hmu = self.hash+umsgpack.packb([segment, hashmap])
hmu_packet = RNS.Packet(self.link, hmu, context = RNS.Packet.RESOURCE_HMU)
try:
if not self.ensure_link(): return
hmu_packet.send()
self.last_activity = time.time()
except Exception as e:
@@ -1245,40 +1254,30 @@ class ResourceAdvertisement:
@staticmethod
def is_request(advertisement_packet):
adv = ResourceAdvertisement.unpack(advertisement_packet.plaintext)
if adv.q != None and adv.u:
return True
else:
return False
if adv.q != None and adv.u: return True
else: return False
@staticmethod
def is_response(advertisement_packet):
adv = ResourceAdvertisement.unpack(advertisement_packet.plaintext)
if adv.q != None and adv.p:
return True
else:
return False
if adv.q != None and adv.p: return True
else: return False
@staticmethod
def read_request_id(advertisement_packet):
adv = ResourceAdvertisement.unpack(advertisement_packet.plaintext)
return adv.q
@staticmethod
def read_transfer_size(advertisement_packet):
adv = ResourceAdvertisement.unpack(advertisement_packet.plaintext)
return adv.t
@staticmethod
def read_size(advertisement_packet):
adv = ResourceAdvertisement.unpack(advertisement_packet.plaintext)
return adv.d
def __init__(self, resource=None, request_id=None, is_response=False):
self.link = None
if resource != None:
@@ -1310,29 +1309,14 @@ class ResourceAdvertisement:
# Flags
self.f = 0x00 | self.x << 5 | self.p << 4 | self.u << 3 | self.s << 2 | self.c << 1 | self.e
def get_transfer_size(self):
return self.t
def get_data_size(self):
return self.d
def get_parts(self):
return self.n
def get_segments(self):
return self.l
def get_hash(self):
return self.h
def is_compressed(self):
return self.c
def has_metadata(self):
return self.x
def get_link(self):
return self.link
def get_transfer_size(self): return self.t
def get_data_size(self): return self.d
def get_parts(self): return self.n
def get_segments(self): return self.l
def get_hash(self): return self.h
def is_compressed(self): return self.c
def has_metadata(self): return self.x
def get_link(self): return self.link
def pack(self, segment=0):
hashmap_start = segment*ResourceAdvertisement.HASHMAP_MAX_LEN
@@ -1342,23 +1326,20 @@ class ResourceAdvertisement:
for i in range(hashmap_start,hashmap_end):
hashmap += self.m[i*Resource.MAPHASH_LEN:(i+1)*Resource.MAPHASH_LEN]
dictionary = {
"t": self.t, # Transfer size
"d": self.d, # Data size
"n": self.n, # Number of parts
"h": self.h, # Resource hash
"r": self.r, # Resource random hash
"o": self.o, # Original hash
"i": self.i, # Segment index
"l": self.l, # Total segments
"q": self.q, # Request ID
"f": self.f, # Resource flags
"m": hashmap
}
dictionary = { "t": self.t, # Transfer size
"d": self.d, # Data size
"n": self.n, # Number of parts
"h": self.h, # Resource hash
"r": self.r, # Resource random hash
"o": self.o, # Original hash
"i": self.i, # Segment index
"l": self.l, # Total segments
"q": self.q, # Request ID
"f": self.f, # Resource flags
"m": hashmap }
return umsgpack.packb(dictionary)
@staticmethod
def unpack(data):
dictionary = umsgpack.unpackb(data)
@@ -1382,4 +1363,6 @@ class ResourceAdvertisement:
adv.p = True if ((adv.f >> 4) & 0x01) == 0x01 else False
adv.x = True if ((adv.f >> 5) & 0x01) == 0x01 else False
if adv.t > Resource.MAX_EFFICIENT_SIZE*3: raise ValueError("Invalid transfer size")
return adv