From 1694a17a75d239e41d9a5ac5c655213d43df4fef Mon Sep 17 00:00:00 2001 From: K8 <8e4525cda44827204097dbcadf90a94c> Date: Thu, 28 May 2026 13:11:00 -0600 Subject: [PATCH 1/6] Full and configurable logfile rotation. --- RNS/__init__.py | 17 +++++++++++++---- 1 file changed, 13 insertions(+), 4 deletions(-) diff --git a/RNS/__init__.py b/RNS/__init__.py index e38684cd..4a3211b0 100755 --- a/RNS/__init__.py +++ b/RNS/__init__.py @@ -77,7 +77,8 @@ LOG_STDOUT = 0x91 LOG_FILE = 0x92 LOG_CALLBACK = 0x93 -LOG_MAXSIZE = 5*1024*1024 +LOG_MAXSIZE = 30*1024*1024 +LOG_MAXROT = 9 loglevel = LOG_NOTICE logfile = None @@ -144,9 +145,17 @@ def log(msg, level=3, _override_destination = False, pt=False): try: with open(logfile, "a") as file: file.write(logstring+"\n") if os.path.getsize(logfile) > LOG_MAXSIZE: - prevfile = logfile+".1" - if os.path.isfile(prevfile): os.unlink(prevfile) - os.rename(logfile, prevfile) + for i in range(LOG_MAXROT, 0, -1): + oldfile = f"{logfile}.{i}" + if os.path.isfile(oldfile): + if i == LOG_MAXROT: + os.unlink(oldfile) + else: + rotfile = f"{logfile}.{i+1}" + os.rename(oldfile, rotfile) + + rotfile = f"{logfile}.1" + os.rename(logfile, rotfile) except Exception as e: _always_override_destination = True From 9da66649761e375ab3ad07ce651a0064005c29c0 Mon Sep 17 00:00:00 2001 From: K8 <8e4525cda44827204097dbcadf90a94c> Date: Wed, 19 Aug 2026 17:34:35 -0600 Subject: [PATCH 2/6] Move log writing to a dedicated thread. --- RNS/__init__.py | 124 ++++++++++++++++++++++++++++++++---------------- 1 file changed, 84 insertions(+), 40 deletions(-) diff --git a/RNS/__init__.py b/RNS/__init__.py index 4a3211b0..eb26bb9d 100755 --- a/RNS/__init__.py +++ b/RNS/__init__.py @@ -36,6 +36,9 @@ import datetime import random import threading +from collections import deque +from threading import Lock, Condition + from ._version import __version__ from .Reticulum import Reticulum @@ -94,8 +97,6 @@ instance_random.seed(os.urandom(10)) _always_override_destination = False -logging_lock = threading.Lock() - def loglevelname(level): if (level == LOG_CRITICAL): return "[Critical]" if (level == LOG_ERROR): return "[Error] " @@ -123,54 +124,97 @@ def timestamp_str(time_s): def precise_timestamp_str(time_s): return datetime.datetime.now().strftime(logtimefmt_p)[:-3] +_log_thread = None +_log_thread_lock = Lock() +_log_queue = deque() +_log_cond = Condition() +def _ensure_log_thread(): + global _log_thread + if _log_thread is None or not _log_thread.is_alive(): + _log_thread = threading.Thread(target=_log_job, daemon=True) + _log_thread.start() + def sl(level=3): return loglevel >= level def log(msg, level=3, _override_destination = False, pt=False): + global compact_log_fmt + if loglevel == LOG_NONE: return - global _always_override_destination, compact_log_fmt + _ensure_log_thread() + msg = str(msg) if loglevel >= level: - if pt: logstring = "["+precise_timestamp_str(time.time())+"] "+loglevelname(level)+" "+msg - else: - if not compact_log_fmt: logstring = ("["+timestamp_str(time.time())+"] " if logtimestamps else "")+loglevelname(level)+" "+msg - else: logstring = ("["+timestamp_str(time.time())+"] " if logtimestamps else "")+msg + with _log_cond: + if pt: logstring = "["+precise_timestamp_str(time.time())+"] "+loglevelname(level)+" "+msg + else: + if not compact_log_fmt: logstring = ("["+timestamp_str(time.time())+"] " if logtimestamps else "")+loglevelname(level)+" "+msg + else: logstring = ("["+timestamp_str(time.time())+"] " if logtimestamps else "")+msg - with logging_lock: - if (logdest == LOG_STDOUT or _always_override_destination or _override_destination): - if not threading.main_thread().is_alive(): return - else: - try: print(logstring) - except: pass + _log_queue.append((logstring, level, _override_destination)) + _log_cond.notify() - elif (logdest == LOG_FILE and logfile != None): - try: - with open(logfile, "a") as file: file.write(logstring+"\n") - if os.path.getsize(logfile) > LOG_MAXSIZE: - for i in range(LOG_MAXROT, 0, -1): - oldfile = f"{logfile}.{i}" - if os.path.isfile(oldfile): - if i == LOG_MAXROT: - os.unlink(oldfile) - else: - rotfile = f"{logfile}.{i+1}" - os.rename(oldfile, rotfile) +def _log_job(): + global _always_override_destination - rotfile = f"{logfile}.1" - os.rename(logfile, rotfile) + if not _log_thread_lock.acquire(blocking=False): return + try: + file = None + if (logdest == LOG_FILE and logfile != None): + try: + file = open(logfile, "a", buffering=1) + except Exception as e: + _always_override_destination = True + log("Exception occurred while opening log file: "+str(e), LOG_CRITICAL) + log("Dumping future log events to console!", LOG_CRITICAL) - except Exception as e: - _always_override_destination = True - log("Exception occurred while writing log message to log file: "+str(e), LOG_CRITICAL) - log("Dumping future log events to console!", LOG_CRITICAL) - log(msg, level) + while True: + with _log_cond: + try: + logstring, level, _override_destination = _log_queue.popleft() + except IndexError as e: + _log_cond.wait() + continue - elif logdest == LOG_CALLBACK: - try: logcall(logstring) - except Exception as e: - _always_override_destination = True - log("Exception occurred while calling external log handler: "+str(e), LOG_CRITICAL) - log("Dumping future log events to console!", LOG_CRITICAL) - log(msg, level) - + if (logdest == LOG_STDOUT or _always_override_destination or _override_destination): + if not threading.main_thread().is_alive(): return + else: + try: print(logstring) + except: pass + + elif (logdest == LOG_FILE and logfile != None and file != None): + try: + file.write(logstring+"\n") + if os.path.getsize(logfile) > LOG_MAXSIZE: + file.close() + for i in range(LOG_MAXROT, 0, -1): + oldfile = f"{logfile}.{i}" + if os.path.isfile(oldfile): + if i == LOG_MAXROT: + os.unlink(oldfile) + else: + rotfile = f"{logfile}.{i+1}" + os.rename(oldfile, rotfile) + + rotfile = f"{logfile}.1" + os.rename(logfile, rotfile) + file = open(logfile, "a", buffering=1) + + except Exception as e: + _always_override_destination = True + log("Exception occurred while writing log message to log file: "+str(e), LOG_CRITICAL) + log("Dumping future log events to console!", LOG_CRITICAL) + log(msg, level) + + elif logdest == LOG_CALLBACK: + try: logcall(logstring) + except Exception as e: + _always_override_destination = True + log("Exception occurred while calling external log handler: "+str(e), LOG_CRITICAL) + log("Dumping future log events to console!", LOG_CRITICAL) + log(msg, level) + + finally: + if file is not None: file.close() + _log_thread_lock.release() def rand(): result = instance_random.random() From 9302415f9e61897ff07b9b7bba5083ebfb5b536f Mon Sep 17 00:00:00 2001 From: K8 <8e4525cda44827204097dbcadf90a94c> Date: Wed, 19 Aug 2026 19:57:42 -0600 Subject: [PATCH 3/6] Add live profiling results output to rnstatus. --- RNS/Reticulum.py | 14 +++++++-- RNS/Transport.py | 3 ++ RNS/Utilities/rnstatus.py | 29 +++++++++++++----- RNS/__init__.py | 64 ++++++++++++++++++++++----------------- 4 files changed, 73 insertions(+), 37 deletions(-) diff --git a/RNS/Reticulum.py b/RNS/Reticulum.py index 4618e456..94924ee9 100755 --- a/RNS/Reticulum.py +++ b/RNS/Reticulum.py @@ -190,8 +190,6 @@ class Reticulum: RNS.Transport.exit_handler() RNS.Identity.exit_handler() - if RNS.Profiler.ran(): RNS.Profiler.results() - RNS.loglevel = RNS.LOG_NONE RNS._detach_stdout() @@ -1293,6 +1291,7 @@ class Reticulum: if path == "packet_rssi": self.rpc_return(conn, self.get_packet_rssi(call["packet_hash"])) if path == "packet_snr": self.rpc_return(conn, self.get_packet_snr(call["packet_hash"])) if path == "packet_q": self.rpc_return(conn, self.get_packet_q(call["packet_hash"])) + if path == "profiling_results": self.rpc_return(conn, self.get_profiling_results()) if path == "blackholed_identities": self.rpc_return(conn, self.get_blackholed_identities()) if path == "is_blackholed": self.rpc_return(conn, self.is_blackholed(call["identity_hash"])) @@ -1848,6 +1847,17 @@ class Reticulum: return None + def get_profiling_results(self): + if self.is_connected_to_shared_instance: + rpc_connection = self.get_rpc_client() + rpc_connection.send_bytes(mp.packb({"get": "profiling_results"})) + response = mp.unpackb(rpc_connection.recv_bytes()) + return response + + else: + if RNS.Profiler.ran(): return RNS.Profiler.results() + else: return None + def halt_interface(self, interface): pass diff --git a/RNS/Transport.py b/RNS/Transport.py index 279814f7..10fc6d1e 100755 --- a/RNS/Transport.py +++ b/RNS/Transport.py @@ -3339,6 +3339,9 @@ class Transport: response.append(Transport.owner.get_interface_stats()) if data[0] == True: response.append(Transport.owner.get_link_count()) + if len(data) >= 2: + if data[1] == True: response.append(Transport.owner.get_profiling_results()) + return response except Exception as e: diff --git a/RNS/Utilities/rnstatus.py b/RNS/Utilities/rnstatus.py index 87be681b..f489d891 100644 --- a/RNS/Utilities/rnstatus.py +++ b/RNS/Utilities/rnstatus.py @@ -63,9 +63,10 @@ request_concluded = False first_remote_req = True remote_destination = None remote_link = None -def get_remote_status(destination_hash, include_lstats, identity, no_output=False, timeout=RNS.Transport.PATH_REQUEST_TIMEOUT): +def get_remote_status(destination_hash, include_lstats, include_profiling, identity, no_output=False, timeout=RNS.Transport.PATH_REQUEST_TIMEOUT): global request_result, request_concluded, first_remote_req, remote_destination, remote_link link_count = None + profiling_results = None if not RNS.Transport.has_path(destination_hash): if not no_output: @@ -114,7 +115,10 @@ def get_remote_status(destination_hash, include_lstats, identity, no_output=Fals if len(response) > 1: link_count = response[1] else: link_count = None - request_result = (status, link_count) + if len(response) > 2: profiling_results = response[2] + else: profiling_results = None + + request_result = (status, link_count, profiling_results) request_concluded = True @@ -125,7 +129,7 @@ def get_remote_status(destination_hash, include_lstats, identity, no_output=Fals print("Sending request...", end=" ") sys.stdout.flush() link.identify(identity) - link.request("/status", data = [include_lstats], response_callback = got_response, failed_callback = request_failed) + link.request("/status", data = [include_lstats, include_profiling], response_callback = got_response, failed_callback = request_failed) first_remote_req = False if not remote_link and not no_output: @@ -155,7 +159,7 @@ def get_remote_status(destination_hash, include_lstats, identity, no_output=Fals def program_setup(configdir, dispall=False, verbosity=0, name_filter=None, json=False, astats=False, pstats=False, lstats=False, sorting=None, sort_reverse=False, remote=None, management_identity=None, must_exit=True, rns_instance=None, traffic_totals=False, discovered_interfaces=False, config_entries=False, burst_filter=False, blocked_ips=False, - queue_stats=False, pps=False, remote_timeout=RNS.Transport.PATH_REQUEST_TIMEOUT): + queue_stats=False, pps=False, profiling=False, remote_timeout=RNS.Transport.PATH_REQUEST_TIMEOUT): if remote: require_shared = False else: require_shared = True @@ -175,6 +179,7 @@ def program_setup(configdir, dispall=False, verbosity=0, name_filter=None, json= link_count = None active_link_count = None stats = None + profiling_results = None details = False if config_entries: @@ -327,8 +332,8 @@ def program_setup(configdir, dispall=False, verbosity=0, name_filter=None, json= if identity == None: raise ValueError("Could not load management identity from "+str(management_identity)) try: - remote_status = get_remote_status(destination_hash, lstats, identity, no_output=json, timeout=remote_timeout) - if remote_status != None: stats, link_count = remote_status + remote_status = get_remote_status(destination_hash, lstats, profiling, identity, no_output=json, timeout=remote_timeout) + if remote_status != None: stats, link_count, profiling_results = remote_status except Exception as e: raise e except Exception as e: @@ -343,6 +348,10 @@ def program_setup(configdir, dispall=False, verbosity=0, name_filter=None, json= try: active_link_count = reticulum.get_active_link_count() except Exception as e: pass + if profiling: + try: profiling_results = reticulum.get_profiling_results() + except Exception as e: pass + try: stats = reticulum.get_interface_stats() except Exception as e: pass @@ -799,6 +808,9 @@ def program_setup(configdir, dispall=False, verbosity=0, name_filter=None, json= print(f" {pqpress}") print(f" {ilpress}") + if profiling_results: + print(f"\n Profiling :\n{RNS.Profiler.format_results(profiling_results)}") + if "transport_id" in stats and stats["transport_id"] != None: print("\n Transport Instance "+RNS.prettyhexrep(stats["transport_id"])+" running") if "network_id" in stats and stats["network_id"] != None: @@ -838,6 +850,7 @@ def main(must_exit=True, rns_instance=None): parser.add_argument("-t", "--totals", action="store_true", help="display traffic totals", default=False) parser.add_argument("-p", "--pps", action="store_true", help="display packets per second in totals", default=False) parser.add_argument("-q", "--queues", action="store_true", help="display queue stats", default=False) + parser.add_argument("-z", "--profiling", action="store_true", help="display live profiling results", default=False) parser.add_argument("-s", "--sort", action="store", help="sort interfaces by [rate, traffic, rx, tx, rxs, txs, announces, arx, atx, arxc, atxc, held, prx, ptx, prxc, ptxc, pvs, ivs, flt]", default=None, type=str) parser.add_argument("-r", "--reverse", action="store_true", help="reverse sorting", default=False) parser.add_argument("-j", "--json", action="store_true", help="output in JSON format", default=False) @@ -876,7 +889,7 @@ def main(must_exit=True, rns_instance=None): astats=args.announce_stats, pstats=args.pr_stats, lstats=args.link_stats, sorting=args.sort, sort_reverse=args.reverse, remote=args.R, management_identity=args.i, remote_timeout=args.w, must_exit=False, rns_instance=reticulum, traffic_totals=args.totals, discovered_interfaces=args.discovered, config_entries=args.D, burst_filter=args.burst, - blocked_ips=args.blocked_ips, queue_stats=args.queues, pps=args.pps) + blocked_ips=args.blocked_ips, queue_stats=args.queues, pps=args.pps, profiling=args.profiling) finally: sys.stdout = old_stdout @@ -894,7 +907,7 @@ def main(must_exit=True, rns_instance=None): astats=args.announce_stats, pstats=args.pr_stats, lstats=args.link_stats, sorting=args.sort, sort_reverse=args.reverse, remote=args.R, management_identity=args.i, remote_timeout=args.w, must_exit=must_exit, rns_instance=rns_instance, traffic_totals=args.totals, discovered_interfaces=args.discovered, config_entries=args.D, burst_filter=args.burst, - blocked_ips=args.blocked_ips, queue_stats=args.queues, pps=args.pps) + blocked_ips=args.blocked_ips, queue_stats=args.queues, pps=args.pps, profiling=args.profiling) except KeyboardInterrupt: print("") diff --git a/RNS/__init__.py b/RNS/__init__.py index eb26bb9d..e4356502 100755 --- a/RNS/__init__.py +++ b/RNS/__init__.py @@ -41,21 +41,6 @@ from threading import Lock, Condition from ._version import __version__ -from .Reticulum import Reticulum -from .Identity import Identity -from .Link import Link, RequestReceipt -from .Channel import MessageBase -from .Buffer import Buffer, RawChannelReader, RawChannelWriter -from .Transport import Transport -from .Discovery import InterfaceAnnouncer -from .Destination import Destination -from .Packet import Packet -from .Packet import PacketReceipt -from .Resolver import Resolver -from .Resource import Resource, ResourceAdvertisement -from .Cryptography import HKDF -from .Cryptography import Hashes - py_modules = glob.glob(os.path.dirname(__file__)+"/*.py") pyc_modules = glob.glob(os.path.dirname(__file__)+"/*.pyc") modules = py_modules+pyc_modules @@ -492,7 +477,7 @@ class Profiler: from statistics import mean, median, stdev results = {} - for tag in Profiler.tags: + for tag in sorted(Profiler.tags): tag_captures = [] tag_entry = Profiler.tags[tag] @@ -534,33 +519,41 @@ class Profiler: results[tag] = tag_results + return results + + @staticmethod + def format_results(results): def print_results_recursive(tag, results, level=0): - print_tag_results(tag, level+1) + results_str = print_tag_results(tag, level+1) + "\n" for tag_name in results: sub_tag = results[tag_name] if sub_tag["super"] == tag["name"]: - print_results_recursive(sub_tag, results, level=level+1) + results_str += print_results_recursive(sub_tag, results, level=level+1) + + return results_str def print_tag_results(tag, level): ind = " "*level name = tag["name"]; count = tag["count"] mean = tag["mean"]; median = tag["median"]; stdev = tag["stdev"] - print( f"{ind}{name}") - print( f"{ind} Samples : {count}") + results_str = f" {ind}{name}\n" + results_str += f" {ind} Samples : {count}\n" if stdev != None: - print(f"{ind} Mean : {prettyshorttime(mean)}") - print(f"{ind} Median : {prettyshorttime(median)}") - print(f"{ind} St.dev. : {prettyshorttime(stdev)}") - print( f"{ind} Total : {prettyshorttime(mean*count)}") - print("") + results_str += f" {ind} Mean : {prettyshorttime(mean)}\n" + results_str += f" {ind} Median : {prettyshorttime(median)}\n" + results_str += f" {ind} St.dev. : {prettyshorttime(stdev)}\n" + results_str += f" {ind} Total : {prettyshorttime(mean*count)}\n" + return results_str - print("\nProfiler results:\n") + results_str = "" for tag_name in results: tag = results[tag_name] if tag["super"] == None: - print_results_recursive(tag, results) + results_str += print_results_recursive(tag, results) + + return results_str profile = Profiler.get_profiler @@ -610,3 +603,20 @@ def bytes_to_b256(data): if not type(data) == bytes: raise TypeError("Invalid input data for base256 encode") try: return [byte_to_b256(c) for c in data] except Exception as e: raise TypeError(f"Could not encode to base256: {e}") + + +from .Reticulum import Reticulum +from .Identity import Identity +from .Link import Link, RequestReceipt +from .Channel import MessageBase +from .Buffer import Buffer, RawChannelReader, RawChannelWriter +from .Transport import Transport +from .Discovery import InterfaceAnnouncer +from .Destination import Destination +from .Packet import Packet +from .Packet import PacketReceipt +from .Resolver import Resolver +from .Resource import Resource, ResourceAdvertisement +from .Cryptography import HKDF +from .Cryptography import Hashes + From cf5d6a796ef12e40e57407e4c9c2eedacd19315e Mon Sep 17 00:00:00 2001 From: K8 <8e4525cda44827204097dbcadf90a94c> Date: Wed, 19 Aug 2026 23:16:52 -0600 Subject: [PATCH 4/6] Rework profilers for running indefinitely. --- RNS/Reticulum.py | 1 + RNS/__init__.py | 162 ++++++++++++++++++++++++++++++++--------------- 2 files changed, 113 insertions(+), 50 deletions(-) diff --git a/RNS/Reticulum.py b/RNS/Reticulum.py index 94924ee9..0804c567 100755 --- a/RNS/Reticulum.py +++ b/RNS/Reticulum.py @@ -1327,6 +1327,7 @@ class Reticulum: except Exception as e: RNS.log("An error ocurred while handling RPC call from local client: "+str(e), RNS.LOG_ERROR) + RNS.trace_exception(e) def get_rpc_client(self): return multiprocessing.connection.Client(self.rpc_addr, family=self.rpc_type, authkey=self.rpc_key) diff --git a/RNS/__init__.py b/RNS/__init__.py index e4356502..8c56abe3 100755 --- a/RNS/__init__.py +++ b/RNS/__init__.py @@ -35,6 +35,8 @@ import time import datetime import random import threading +import math +import bisect from collections import deque from threading import Lock, Condition @@ -329,7 +331,7 @@ def prettytime(time, verbose=False, compact=False): if not neg: return tstr else: return f"-{tstr}" -def prettyshorttime(time, verbose=False, compact=False): +def prettyshorttime(time, verbose=False, compact=False, tight=False): neg = False time = time*1e6 if time < 0: @@ -365,8 +367,8 @@ def prettyshorttime(time, verbose=False, compact=False): for c in components: i += 1 if i == 1: pass - elif i < len(components): tstr += ", " - elif i == len(components): tstr += " and " + elif i < len(components): tstr += ", " if not tight else " " + elif i == len(components): tstr += " and " if not tight else " " tstr += c @@ -403,20 +405,25 @@ class Profiler: profilers = {} tags = {} + # Samples per tag per thread + MAX_CAPTURES = 10000 + @staticmethod - def get_profiler(tag=None, super_tag=None): + def get_profiler(tag=None, super_tag=None, max_captures=None): if tag in Profiler.profilers: return Profiler.profilers[tag] else: - profiler = Profiler(tag, super_tag) + if max_captures is None: max_captures = Profiler.MAX_CAPTURES + profiler = Profiler(tag, super_tag, max_captures) Profiler.profilers[tag] = profiler return profiler - def __init__(self, tag=None, super_tag=None): + def __init__(self, tag=None, super_tag=None, max_captures=None): self.paused = False self.pause_time = 0 self.pause_started = None self.tag = tag self.super_tag = super_tag + self.max_captures = max_captures if max_captures is not None else Profiler.MAX_CAPTURES if self.super_tag in Profiler.profilers: self.super_profiler = Profiler.profilers[self.super_tag] @@ -436,7 +443,7 @@ class Profiler: thread_ident = threading.get_ident() if not tag in Profiler.tags: Profiler.tags[tag] = {"threads": {}, "super": super_tag} if not thread_ident in Profiler.tags[tag]["threads"]: - Profiler.tags[tag]["threads"][thread_ident] = {"current_start": None, "captures": []} + Profiler.tags[tag]["threads"][thread_ident] = {"current_start": None, "captures": deque(maxlen=self.max_captures)} Profiler.tags[tag]["threads"][thread_ident]["current_start"] = time.perf_counter() self.resume_super() @@ -452,7 +459,7 @@ class Profiler: if Profiler.tags[tag]["threads"][thread_ident]["current_start"] != None: begin = Profiler.tags[tag]["threads"][thread_ident]["current_start"] Profiler.tags[tag]["threads"][thread_ident]["current_start"] = None - Profiler.tags[tag]["threads"][thread_ident]["captures"].append(end-begin) + Profiler.tags[tag]["threads"][thread_ident]["captures"].append((begin, end-begin)) if not Profiler._ran: Profiler._ran = True self.resume_super() @@ -474,55 +481,104 @@ class Profiler: @staticmethod def results(): - from statistics import mean, median, stdev results = {} - + + def find_window_start(captures, start_time, hi=None): + hi = len(captures) if hi is None else hi + idx = bisect.bisect_left(captures, start_time, hi=hi, key=lambda c: c[0]) + return idx if len(captures) - idx > 1 else None + + # Fast one-pass calculation of summary statistics + def calc_stats(captures, start=0, end=None, key=lambda c: c): + if end is None: end = len(captures) + count = end - start + + if count <= 0: return None + elif count == 1: + return { "mean": key(captures[start]), + "median": key(captures[start]), + "min": key(captures[start]), + "max": key(captures[start]), + "stdev": None } + + med_even = count % 2 == 0 + med_idx = start + (count // 2 if med_even else (count - 1) // 2) + if med_even: c_median = key(captures[med_idx]) + else: c_median = (key(captures[med_idx]) + key(captures[med_idx+1])) / 2 + + c_mean = 0; c_min = key(captures[start]); c_max = key(captures[start]); ck = 0; ck2 = 0 + for idx in range(start, end): + c = key(captures[idx]) + c_mean += c + if c < c_min: c_min = c + if c > c_max: c_max = c + ck += c - c_median + ck2 += (c - c_median) ** 2 + c_mean /= count + c_std = math.sqrt((ck2 - (ck ** 2)/count) / (count - 1)) + + return { "mean": c_mean, "median": c_median, "min": c_min, "max": c_max, "stdev": c_std } + + now = time.perf_counter() for tag in sorted(Profiler.tags): tag_captures = [] tag_entry = Profiler.tags[tag] - + for thread_ident in tag_entry["threads"]: thread_entry = tag_entry["threads"][thread_ident] thread_captures = thread_entry["captures"] - sample_count = len(thread_captures) - - if sample_count > 1: - thread_results = { "count": sample_count, - "mean": mean(thread_captures), - "median": median(thread_captures), - "stdev": stdev(thread_captures) } - - elif sample_count == 1: - thread_results = { "count": sample_count, - "mean": mean(thread_captures), - "median": median(thread_captures), - "stdev": None } + + #sample_count = len(thread_captures) + #if sample_count > 1: + # thread_results = { "count": sample_count, + # "mean": mean(thread_captures), + # "median": median(thread_captures), + # "stdev": stdev(thread_captures) } + #elif sample_count == 1: + # thread_results = { "count": sample_count, + # "mean": mean(thread_captures), + # "median": median(thread_captures), + # "stdev": None } tag_captures.extend(thread_captures) + tag_captures.sort(key=lambda c: c[0]) - sample_count = len(tag_captures) - if sample_count > 1: - tag_results = { "name": tag, - "super": tag_entry["super"], - "count": len(tag_captures), - "mean": mean(tag_captures), - "median": median(tag_captures), - "stdev": stdev(tag_captures) } - - elif sample_count == 1: - tag_results = { "name": tag, - "super": tag_entry["super"], - "count": len(tag_captures), - "mean": mean(tag_captures), - "median": median(tag_captures), - "stdev": None } + tag_results = None + if len(tag_captures): + captures_1m = None; captures_5m = None; captures_30m = None; captures_60m = None + stats_1m = None; stats_5m = None; stats_30m = None; stats_60m = None - results[tag] = tag_results + captures_1m = find_window_start(tag_captures, now - 1*60) + if captures_1m: captures_5m = find_window_start(tag_captures, now - 5*60, hi=captures_1m) + if captures_5m: captures_30m = find_window_start(tag_captures, now - 30*60, hi=captures_5m) + if captures_30m: captures_60m = find_window_start(tag_captures, now - 60*60, hi=captures_30m) + + stats_all = calc_stats(tag_captures, 0, key=lambda c: c[1]) + if captures_1m: stats_1m = calc_stats(tag_captures, captures_1m, key=lambda c: c[1]) + if captures_5m: stats_5m = calc_stats(tag_captures, captures_5m, key=lambda c: c[1]) + if captures_30m: stats_30m = calc_stats(tag_captures, captures_30m, key=lambda c: c[1]) + if captures_60m: stats_60m = calc_stats(tag_captures, captures_60m, key=lambda c: c[1]) + + tag_results = { "name": tag, + "super": tag_entry["super"], + "count": len(tag_captures), + "threads": len(tag_entry["threads"]), + "stats_all": stats_all, + "stats_1m": stats_1m, + "stats_5m": stats_5m, + "stats_30m": stats_30m, + "stats_60m": stats_60m } + + results[tag] = tag_results return results @staticmethod def format_results(results): + def pst(time): + if time is not None: return prettyshorttime(time, tight=True) + else: return "-----" + def print_results_recursive(tag, results, level=0): results_str = print_tag_results(tag, level+1) + "\n" @@ -533,18 +589,24 @@ class Profiler: return results_str - def print_tag_results(tag, level): ind = " "*level - name = tag["name"]; count = tag["count"] - mean = tag["mean"]; median = tag["median"]; stdev = tag["stdev"] + name = tag["name"]; count = tag["count"]; threads = tag["threads"] + stats_all = tag["stats_all"]; stats_1m = tag["stats_1m"]; stats_5m = tag["stats_5m"]; stats_30m = tag["stats_30m"]; stats_60m = tag["stats_60m"] results_str = f" {ind}{name}\n" - results_str += f" {ind} Samples : {count}\n" - if stdev != None: - results_str += f" {ind} Mean : {prettyshorttime(mean)}\n" - results_str += f" {ind} Median : {prettyshorttime(median)}\n" - results_str += f" {ind} St.dev. : {prettyshorttime(stdev)}\n" - results_str += f" {ind} Total : {prettyshorttime(mean*count)}\n" + results_str += f" {ind} Samples : {count} from {threads} thread{'s' if threads > 1 else ''}\n" + if stats_all != None: + results_str += f" {ind} Total : {pst(stats_all["mean"]*count)}\n" + results_str += f" {ind} {'Mean':^15} | {'Median':^15} | {'Min':^15} | {'Max':^15} | {'St. dev':^15}\n" + results_str += f" {ind} Stats : ({pst(stats_all["mean"]):^15} | {pst(stats_all["median"]):^15} | {pst(stats_all["min"]):^15} | {pst(stats_all["max"]):^15} | {pst(stats_all["stdev"]):^15})\n" + if stats_1m != None: + results_str += f" {ind} 1m : ({pst(stats_1m["mean"]):^15} | {pst(stats_1m["median"]):^15} | {pst(stats_1m["min"]):^15} | {pst(stats_1m["max"]):^15} | {pst(stats_1m["stdev"]):^15})\n" + if stats_5m != None: + results_str += f" {ind} 5m : ({pst(stats_5m["mean"]):^15} | {pst(stats_5m["median"]):^15} | {pst(stats_5m["min"]):^15} | {pst(stats_5m["max"]):^15} | {pst(stats_5m["stdev"]):^15})\n" + if stats_30m != None: + results_str += f" {ind} 30m : ({pst(stats_30m["mean"]):^15} | {pst(stats_30m["median"]):^15} | {pst(stats_30m["min"]):^15} | {pst(stats_30m["max"]):^15} | {pst(stats_30m["stdev"]):^15})\n" + if stats_60m != None: + results_str += f" {ind} 60m : ({pst(stats_60m["mean"]):^15} | {pst(stats_60m["median"]):^15} | {pst(stats_60m["min"]):^15} | {pst(stats_60m["max"]):^15} | {pst(stats_60m["stdev"]):^15})\n" return results_str results_str = "" From 40281f91daac478d5ab15d36b9ac20dfa5eb5b04 Mon Sep 17 00:00:00 2001 From: K8 <8e4525cda44827204097dbcadf90a94c> Date: Sun, 23 Aug 2026 02:27:53 -0600 Subject: [PATCH 5/6] Decorator for profiling functions. --- RNS/__init__.py | 26 ++++++++++++++++++++++++++ 1 file changed, 26 insertions(+) diff --git a/RNS/__init__.py b/RNS/__init__.py index 8c56abe3..f71d521a 100755 --- a/RNS/__init__.py +++ b/RNS/__init__.py @@ -40,6 +40,7 @@ import bisect from collections import deque from threading import Lock, Condition +from functools import partial, wraps from ._version import __version__ @@ -417,6 +418,31 @@ class Profiler: Profiler.profilers[tag] = profiler return profiler + @staticmethod + def profile(func=None, *, tag=None, max_captures=None): + """ + Decorator to profile a function. With no arguments, it can be + used as `@RNS.Profiler.profile`, in which case it will use the + qualified name of the function as the profiler tag. It can + also be used as `@RNS.Profiler.profile(tag="prof_tag")` to use + a specific profiler tag. + + :param tag: Either a profiler tag as passed to `Profiler.get_profiler(...)` or a `Profiler` object. + :param max_captures: Maximum samples per thread to capture for the specified profiler tag. + """ + if func is None: + return partial(Profiler.profile, tag=tag, max_captures=max_captures) + + tag = func.__qualname__ if tag is None else tag + if isinstance(tag, Profiler): profiler = tag + else: profiler = Profiler.get_profiler(tag=tag, max_captures=max_captures) + + @wraps(func) + def wrapper(*args, **kwargs): + with profiler: + return func(*args, **kwargs) + return wrapper + def __init__(self, tag=None, super_tag=None, max_captures=None): self.paused = False self.pause_time = 0 From dca5b9639ea4d90b99675782eeec8ec7f797970b Mon Sep 17 00:00:00 2001 From: K8 <8e4525cda44827204097dbcadf90a94c> Date: Sun, 23 Aug 2026 21:43:49 -0600 Subject: [PATCH 6/6] Limit total profiler captures per tag, not per thread; handle reentrant profilers; make stats time windows be non-overlapping. --- RNS/__init__.py | 164 +++++++++++++++++++++++++++--------------------- 1 file changed, 92 insertions(+), 72 deletions(-) diff --git a/RNS/__init__.py b/RNS/__init__.py index f71d521a..e4b513b1 100755 --- a/RNS/__init__.py +++ b/RNS/__init__.py @@ -406,7 +406,7 @@ class Profiler: profilers = {} tags = {} - # Samples per tag per thread + # Samples per tag MAX_CAPTURES = 10000 @staticmethod @@ -428,7 +428,7 @@ class Profiler: a specific profiler tag. :param tag: Either a profiler tag as passed to `Profiler.get_profiler(...)` or a `Profiler` object. - :param max_captures: Maximum samples per thread to capture for the specified profiler tag. + :param max_captures: Maximum samples to capture for the specified profiler tag. """ if func is None: return partial(Profiler.profile, tag=tag, max_captures=max_captures) @@ -467,11 +467,18 @@ class Profiler: tag = self.tag super_tag = self.super_tag thread_ident = threading.get_ident() - if not tag in Profiler.tags: Profiler.tags[tag] = {"threads": {}, "super": super_tag} + if not tag in Profiler.tags: Profiler.tags[tag] = {"threads": {}, "super": super_tag, "captures": deque(maxlen=self.max_captures)} if not thread_ident in Profiler.tags[tag]["threads"]: - Profiler.tags[tag]["threads"][thread_ident] = {"current_start": None, "captures": deque(maxlen=self.max_captures)} + Profiler.tags[tag]["threads"][thread_ident] = {"running": deque()} - Profiler.tags[tag]["threads"][thread_ident]["current_start"] = time.perf_counter() + # The tag deque stores a shared reference to the capture, and the end + # time isn't updated until the context manager exits. This leaves the + # deque sorted by start time, except for cases where the thread is + # preemted between getting the current time and appending. We also store + # a stack of start times per thread to support reentrancy. + capture = {"start": time.perf_counter(), "end": None, "thread_ident": thread_ident} + Profiler.tags[tag]["captures"].append(capture) + Profiler.tags[tag]["threads"][thread_ident]["running"].append(capture) self.resume_super() def __exit__(self, exc_type, exc_value, traceback): @@ -481,14 +488,15 @@ class Profiler: end = time.perf_counter() - self.pause_time self.pause_time = 0 thread_ident = threading.get_ident() - if tag in Profiler.tags and thread_ident in Profiler.tags[tag]["threads"]: - if Profiler.tags[tag]["threads"][thread_ident]["current_start"] != None: - begin = Profiler.tags[tag]["threads"][thread_ident]["current_start"] - Profiler.tags[tag]["threads"][thread_ident]["current_start"] = None - Profiler.tags[tag]["threads"][thread_ident]["captures"].append((begin, end-begin)) + try: + if tag in Profiler.tags and thread_ident in Profiler.tags[tag]["threads"]: + try: capture = Profiler.tags[tag]["threads"][thread_ident]["running"].pop() + except IndexError as e: return + capture["end"] = end if not Profiler._ran: Profiler._ran = True - self.resume_super() + finally: + self.resume_super() def pause(self, pause_started=None): if not self.paused: @@ -507,69 +515,82 @@ class Profiler: @staticmethod def results(): - results = {} - def find_window_start(captures, start_time, hi=None): hi = len(captures) if hi is None else hi - idx = bisect.bisect_left(captures, start_time, hi=hi, key=lambda c: c[0]) + idx = bisect.bisect_left(captures, start_time, hi=hi, key=lambda c: c["start"]) return idx if len(captures) - idx > 1 else None # Fast one-pass calculation of summary statistics - def calc_stats(captures, start=0, end=None, key=lambda c: c): + def calc_stats(captures, start=0, end=None, key=lambda c: c, threads_key=None): if end is None: end = len(captures) + while start > 0 and start < len(captures) and start < end and key(captures[start]) is None: start += 1 + while end > 0 and end <= len(captures) and start < end and key(captures[end-1]) is None: end -= 1 count = end - start if count <= 0: return None elif count == 1: - return { "mean": key(captures[start]), - "median": key(captures[start]), - "min": key(captures[start]), - "max": key(captures[start]), - "stdev": None } + capture = key(captures[start]) + return { "count": 1, + "mean": capture, + "median": capture, + "min": capture, + "max": capture, + "stdev": None, + "sum": capture, + "threads": 1 if threads_key else None} + # Median is approximate if there are any incomplete captures in the + # middle of the range. med_even = count % 2 == 0 med_idx = start + (count // 2 if med_even else (count - 1) // 2) - if med_even: c_median = key(captures[med_idx]) - else: c_median = (key(captures[med_idx]) + key(captures[med_idx+1])) / 2 + while key(captures[med_idx]) is None and med_idx < end: med_idx += 1 + if med_idx == end: c_median = None + elif med_even: c_median = key(captures[med_idx]) + else: + med_idx2 = med_idx + 1 + while key(captures[med_idx2]) is None and med_idx2 < end: med_idx2 += 1 + if med_idx2 == end: c_median = None + else: + med_cap1 = key(captures[med_idx]) + med_cap2 = key(captures[med_idx+1]) + if med_cap1 is None or med_cap2 is None: c_median = None + else: c_median = (med_cap1 + med_cap2) / 2 - c_mean = 0; c_min = key(captures[start]); c_max = key(captures[start]); ck = 0; ck2 = 0 + c_count = 0; c_sum = 0; c_min = key(captures[start]); c_max = key(captures[start]); ck = 0; ck2 = 0 + uniq_threads = set() for idx in range(start, end): c = key(captures[idx]) - c_mean += c + if c is None: continue + else: c_count += 1 + c_sum += c if c < c_min: c_min = c if c > c_max: c_max = c ck += c - c_median ck2 += (c - c_median) ** 2 - c_mean /= count - c_std = math.sqrt((ck2 - (ck ** 2)/count) / (count - 1)) + if threads_key: uniq_threads.add(threads_key(captures[idx])) + c_mean = c_sum / c_count if c_count > 0 else None + c_std = math.sqrt((ck2 - (ck ** 2)/c_count) / (c_count - 1)) if c_count > 1 else None - return { "mean": c_mean, "median": c_median, "min": c_min, "max": c_max, "stdev": c_std } + return { "count": c_count, + "mean": c_mean, + "median": c_median, + "min": c_min, + "max": c_max, + "stdev": c_std, + "sum": c_sum, + "threads": len(uniq_threads) if threads_key else None } + def stats_key(capture): + end = capture["end"] + if end is None: return None + else: return end - capture["start"] + + results = {} now = time.perf_counter() for tag in sorted(Profiler.tags): - tag_captures = [] tag_entry = Profiler.tags[tag] + tag_captures = sorted(tag_entry["captures"], key=lambda c: c["start"]) - for thread_ident in tag_entry["threads"]: - thread_entry = tag_entry["threads"][thread_ident] - thread_captures = thread_entry["captures"] - - #sample_count = len(thread_captures) - #if sample_count > 1: - # thread_results = { "count": sample_count, - # "mean": mean(thread_captures), - # "median": median(thread_captures), - # "stdev": stdev(thread_captures) } - #elif sample_count == 1: - # thread_results = { "count": sample_count, - # "mean": mean(thread_captures), - # "median": median(thread_captures), - # "stdev": None } - - tag_captures.extend(thread_captures) - tag_captures.sort(key=lambda c: c[0]) - - tag_results = None if len(tag_captures): captures_1m = None; captures_5m = None; captures_30m = None; captures_60m = None stats_1m = None; stats_5m = None; stats_30m = None; stats_60m = None @@ -579,23 +600,23 @@ class Profiler: if captures_5m: captures_30m = find_window_start(tag_captures, now - 30*60, hi=captures_5m) if captures_30m: captures_60m = find_window_start(tag_captures, now - 60*60, hi=captures_30m) - stats_all = calc_stats(tag_captures, 0, key=lambda c: c[1]) - if captures_1m: stats_1m = calc_stats(tag_captures, captures_1m, key=lambda c: c[1]) - if captures_5m: stats_5m = calc_stats(tag_captures, captures_5m, key=lambda c: c[1]) - if captures_30m: stats_30m = calc_stats(tag_captures, captures_30m, key=lambda c: c[1]) - if captures_60m: stats_60m = calc_stats(tag_captures, captures_60m, key=lambda c: c[1]) + stats_all = calc_stats(tag_captures, 0, None, key=stats_key, threads_key=lambda c: c["thread_ident"]) + if captures_1m: stats_1m = calc_stats(tag_captures, captures_1m, None, key=stats_key) + if captures_5m: stats_5m = calc_stats(tag_captures, captures_5m, captures_1m, key=stats_key) + if captures_30m: stats_30m = calc_stats(tag_captures, captures_30m, captures_5m, key=stats_key) + if captures_60m: stats_60m = calc_stats(tag_captures, captures_60m, captures_30m, key=stats_key) - tag_results = { "name": tag, - "super": tag_entry["super"], - "count": len(tag_captures), - "threads": len(tag_entry["threads"]), - "stats_all": stats_all, - "stats_1m": stats_1m, - "stats_5m": stats_5m, - "stats_30m": stats_30m, - "stats_60m": stats_60m } + if stats_all["count"]: + results[tag] = { "name": tag, + "super": tag_entry["super"], + "stats_all": stats_all, + "stats_1m": stats_1m, + "stats_5m": stats_5m, + "stats_30m": stats_30m, + "stats_60m": stats_60m } - results[tag] = tag_results + # Yield to avoid bogging down the instance + time.sleep(0.001) return results @@ -617,22 +638,21 @@ class Profiler: def print_tag_results(tag, level): ind = " "*level - name = tag["name"]; count = tag["count"]; threads = tag["threads"] + name = tag["name"] stats_all = tag["stats_all"]; stats_1m = tag["stats_1m"]; stats_5m = tag["stats_5m"]; stats_30m = tag["stats_30m"]; stats_60m = tag["stats_60m"] results_str = f" {ind}{name}\n" - results_str += f" {ind} Samples : {count} from {threads} thread{'s' if threads > 1 else ''}\n" + results_str += f" {ind} Samples : {stats_all["count"]} from {stats_all["threads"]} thread{'s' if stats_all["threads"] > 1 else ''}\n" if stats_all != None: - results_str += f" {ind} Total : {pst(stats_all["mean"]*count)}\n" - results_str += f" {ind} {'Mean':^15} | {'Median':^15} | {'Min':^15} | {'Max':^15} | {'St. dev':^15}\n" - results_str += f" {ind} Stats : ({pst(stats_all["mean"]):^15} | {pst(stats_all["median"]):^15} | {pst(stats_all["min"]):^15} | {pst(stats_all["max"]):^15} | {pst(stats_all["stdev"]):^15})\n" + results_str += f" {ind} {'Mean':^15} | {'Median':^15} | {'Min':^15} | {'Max':^15} | {'St. dev':^15} | {'Total':^15}\n" + results_str += f" {ind} Stats : ({pst(stats_all["mean"]):^15} | {pst(stats_all["median"]):^15} | {pst(stats_all["min"]):^15} | {pst(stats_all["max"]):^15} | {pst(stats_all["stdev"]):^15} | {pst(stats_all["sum"]):^15})\n" if stats_1m != None: - results_str += f" {ind} 1m : ({pst(stats_1m["mean"]):^15} | {pst(stats_1m["median"]):^15} | {pst(stats_1m["min"]):^15} | {pst(stats_1m["max"]):^15} | {pst(stats_1m["stdev"]):^15})\n" + results_str += f" {ind} 0-1m : ({pst(stats_1m["mean"]):^15} | {pst(stats_1m["median"]):^15} | {pst(stats_1m["min"]):^15} | {pst(stats_1m["max"]):^15} | {pst(stats_1m["stdev"]):^15} | {pst(stats_1m["sum"]):^15})\n" if stats_5m != None: - results_str += f" {ind} 5m : ({pst(stats_5m["mean"]):^15} | {pst(stats_5m["median"]):^15} | {pst(stats_5m["min"]):^15} | {pst(stats_5m["max"]):^15} | {pst(stats_5m["stdev"]):^15})\n" + results_str += f" {ind} 1-5m : ({pst(stats_5m["mean"]):^15} | {pst(stats_5m["median"]):^15} | {pst(stats_5m["min"]):^15} | {pst(stats_5m["max"]):^15} | {pst(stats_5m["stdev"]):^15} | {pst(stats_5m["sum"]):^15})\n" if stats_30m != None: - results_str += f" {ind} 30m : ({pst(stats_30m["mean"]):^15} | {pst(stats_30m["median"]):^15} | {pst(stats_30m["min"]):^15} | {pst(stats_30m["max"]):^15} | {pst(stats_30m["stdev"]):^15})\n" + results_str += f" {ind} 5-30m : ({pst(stats_30m["mean"]):^15} | {pst(stats_30m["median"]):^15} | {pst(stats_30m["min"]):^15} | {pst(stats_30m["max"]):^15} | {pst(stats_30m["stdev"]):^15} | {pst(stats_30m["sum"]):^15})\n" if stats_60m != None: - results_str += f" {ind} 60m : ({pst(stats_60m["mean"]):^15} | {pst(stats_60m["median"]):^15} | {pst(stats_60m["min"]):^15} | {pst(stats_60m["max"]):^15} | {pst(stats_60m["stdev"]):^15})\n" + results_str += f" {ind} 30-60m : ({pst(stats_60m["mean"]):^15} | {pst(stats_60m["median"]):^15} | {pst(stats_60m["min"]):^15} | {pst(stats_60m["max"]):^15} | {pst(stats_60m["stdev"]):^15} | {pst(stats_60m["sum"]):^15})\n" return results_str results_str = ""