diff --git a/tests/throughput.py b/tests/throughput.py new file mode 100644 index 00000000..7112eaf0 --- /dev/null +++ b/tests/throughput.py @@ -0,0 +1,883 @@ +#!/usr/bin/env python3 +# +# Usage examples: +# +# python3 tests/transport_throughput.py +# python3 tests/transport_throughput.py --scenario transit_single_135 +# python3 tests/transport_throughput.py --mode inline --runs 5 +# python3 tests/transport_throughput.py --list-scenarios +# + +import unittest + +import os +import sys +import time +import gc +import struct +import platform +import tempfile +import statistics + +# Ensure that the Reticulum tree this suite lives in is the one being +# benchmarked, even if a different version of Reticulum is installed in +# site-packages. +_SUITE_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) +if os.path.isdir(os.path.join(_SUITE_ROOT, "RNS")): + sys.path.insert(0, _SUITE_ROOT) + +import RNS + +# --------------------------------------------------------------------------- +# Configuration +# --------------------------------------------------------------------------- + +BENCHMARK_CONFIG = { + "scenarios": None, # None = all, or a list of scenario names + "mode": "both", # "inline", "drainer" or "both" + "runs": 3, # measurement runs per scenario/mode, median is reported +} + +SCENARIO_DESCRIPTIONS = { + "transit_single_135": "SINGLE transit relay, 135 B payload, 3-hop path", + "transit_single_475": "SINGLE transit relay, 475 B payload, 3-hop path", + "transit_single_1024": "SINGLE transit relay, 1 KiB payload, 3-hop path", + "transit_single_16384": "SINGLE transit relay, 16 KiB payload, 3-hop path", + "transit_single_final": "SINGLE transit relay, final hop, header strip, 135 B", + "transit_link_135": "LINK transit relay via link table, cross-interface, 135 B", + "transit_link_475": "LINK transit relay via link table, cross-interface, 475 B", + "transit_link_1024": "LINK transit relay via link table, cross-interface, 1024 B", + "transit_link_16384": "LINK transit relay via link table, cross-interface, 16384 B", + "terminus_link": "LINK terminus delivery, token decrypt, 135 B", + "terminus_single": "SINGLE local delivery, ephemeral-key decrypt, 135 B", + "announce_ingress": "Announce ingress, fresh destinations, validation + path insert", + "outbound_path": "Outbound insertion into transport, known 3-hop path, 135 B", +} + +# Default packet counts per scenario and mode. +DEFAULT_PACKETS = { + "transit_single_135": {"inline": 20000, "drainer": 20000}, + "transit_single_475": {"inline": 20000, "drainer": 20000}, + "transit_single_1024": {"inline": 20000, "drainer": 20000}, + "transit_single_16384": {"inline": 10000, "drainer": 10000}, + "transit_single_final": {"inline": 20000, "drainer": 20000}, + "transit_link_135": {"inline": 20000, "drainer": 20000}, + "transit_link_475": {"inline": 20000, "drainer": 20000}, + "transit_link_1024": {"inline": 20000, "drainer": 20000}, + "transit_link_16384": {"inline": 10000, "drainer": 10000}, + "terminus_link": {"inline": 6000, "drainer": 6000}, + "terminus_single": {"inline": 3000, "drainer": 3000}, + "announce_ingress": {"inline": 1000, "drainer": 1000}, + "outbound_path": {"inline": 8000, "drainer": 0}, +} + +INLINE_SAMPLE_STEP = 64 # sample a timing point every N packets for p50/p95 +DRAINER_CHUNK = 256 # packets fed per drainer backpressure step + + +class BenchmarkInterface(RNS.Interfaces.Interface.Interface): + HW_MTU = 1048576 + BITRATE = 1_000_000_000 + + def __init__(self, name="bench"): + super().__init__() + self.name = name + self.IN = True + self.OUT = True + self.online = True + self.bitrate = self.BITRATE + self.HW_MTU = BenchmarkInterface.HW_MTU + self.mode = RNS.Interfaces.Interface.Interface.MODE_FULL + self.gravity = 0 + self.ifac_size = 0 + self.ifac_key = None + self.ifac_identity = None + self.parent_interface = None + self.ingress_control = False + self.announce_rate_target = RNS.Interfaces.Interface.Interface.DEFAULT_AR_TARGET + self.announce_rate_grace = RNS.Interfaces.Interface.Interface.DEFAULT_AR_GRACE + self.announce_rate_penalty = RNS.Interfaces.Interface.Interface.DEFAULT_AR_PENALTY + + def process_outgoing(self, data): + self.txb += len(data) + + def __str__(self): + return f"BenchmarkInterface[{self.name}]" + + +class Scenario: + kind = "ingress" # "ingress" or "outbound" + + def __init__(self, name, description, interface, frames, packets, runs, + inline_offset, drainer_offset, size, completion_counter=None, + fresh_per_run=False): + self.name = name + self.description = description + self.interface = interface + self.frame_size = size + self.frames = frames # pool of unique raw frames + self.packets_inline = packets.get("inline", 0) + self.packets_drainer = packets.get("drainer", 0) + self.runs = runs + self.inline_offset = inline_offset + self.drainer_offset = drainer_offset + self.completion_counter = completion_counter + # When true, each run uses a dedicated fresh slice of frames + # (required where processing is not idempotent, e.g. announces). + # When false, runs reuse the same pool with the duplicate filter + # reset in between. + self.fresh_per_run = fresh_per_run + + def can_drainer(self): + return self.packets_drainer > 0 and self.completion_counter is not None + + def n_for_mode(self, mode): + if mode == "inline": return self.packets_inline + if mode == "drainer": return self.packets_drainer + + +class CountingDestination(RNS.Destination): + """A destination that counts successful deliveries, for use as a + completion counter in drainer mode.""" + + def __init__(self, *args, **kwargs): + super().__init__(*args, **kwargs) + self.delivered = 0 + + def receive(self, packet): + result = super().receive(packet) + if result: + self.delivered += 1 + return result + + +def _register_link(link): + with RNS.Transport.active_links_lock: + RNS.Transport.active_links.append(link) + if hasattr(RNS.Transport, "active_links_map"): + RNS.Transport.active_links_map[link.link_id] = link + + +def _make_minimal_link(instance, interface): + """Constructs a minimal, active Link object suitable for benchmark delivery.""" + import importlib + _link_mod = importlib.import_module("RNS.Link") + + link = RNS.Link.__new__(RNS.Link) + link.mode = RNS.Link.MODE_DEFAULT + link.rtt = 0.1 + link.mtu = RNS.Reticulum.MTU + link.establishment_cost = 0 + link.establishment_rate = None + link.expected_rate = None + link.callbacks = _link_mod.LinkCallbacks() + link.resource_strategy = RNS.Link.ACCEPT_NONE + link.last_resource_window = None + link.last_resource_eifr = None + link.outgoing_resources = [] + link.incoming_resources = [] + link.pending_requests = [] + link.last_inbound = 0 + link.last_outbound = 0 + link.last_keepalive = 0 + link.last_proof = 0 + link.last_data = 0 + link.tx = 0 + link.rx = 0 + link.txbytes = 0 + link.rxbytes = 0 + link.rssi = None + link.snr = None + link.q = None + link.traffic_timeout_factor = RNS.Link.TRAFFIC_TIMEOUT_FACTOR + link.keepalive_timeout_factor = RNS.Link.KEEPALIVE_TIMEOUT_FACTOR + link.keepalive = RNS.Link.KEEPALIVE + link.stale_time = RNS.Link.STALE_TIME + link.watchdog_lock = False + link.status = RNS.Link.ACTIVE + link.activated_at = time.time() + link.type = RNS.Destination.LINK + link.owner = instance + link.initiator = False + link.expected_hops = 1 + link.rebalanced = None + link.attached_interface = interface + link._channel = None + link._Link__remote_identity = None + link._Link__track_phy_stats = False + link.derived_key = os.urandom(64) + link.token = RNS.Cryptography.Token(link.derived_key) + link.link_id = os.urandom(RNS.Reticulum.TRUNCATED_HASHLENGTH//8) + link.hash = link.link_id + link.hexhash = link.link_id.hex() + from RNS.Cryptography.Proxies import Ed25519PrivateKeyProxy + link.sig_prv = Ed25519PrivateKeyProxy.from_private_bytes(link.derived_key[:32]) + + class _DestinationShim: + pass + + link.destination = _DestinationShim() + link.destination.type = RNS.Destination.LINK + link.destination.proof_strategy = RNS.Destination.PROVE_NONE + _dest_mod = importlib.import_module("RNS.Destination") + link.destination.callbacks = _dest_mod.Callbacks() + link.destination.status = RNS.Link.ACTIVE + link.destination.last_outbound = 0 + link.destination.tx = 0 + link.destination.txbytes = 0 + link.destination.attached_interface = interface + link.destination.mtu = RNS.Reticulum.MTU + link.destination.rssi = None + link.destination.snr = None + link.destination.q = None + link.destination.rtt = RNS.Link.TRAFFIC_TIMEOUT_MIN_MS/1000 + link.destination.traffic_timeout_factor = 1.0 + + return link + + +def _transit_single_frame(interface, dst_h, payload): + """One HEADER_2|TRANSPORT|SINGLE|DATA frame addressed to this instance.""" + flags = (RNS.Packet.HEADER_2 << 6) | (RNS.Transport.TRANSPORT << 4) \ + | (RNS.Destination.SINGLE << 2) | RNS.Packet.DATA + return (struct.pack("!B", flags) + struct.pack("!B", 1) + + RNS.Transport.identity.hash + dst_h + + bytes([RNS.Packet.NONE]) + payload) + + +def _transit_link_frame(interface, link_id, payload): + """One HEADER_2|TRANSPORT|LINK|DATA frame addressed to this instance.""" + flags = (RNS.Packet.HEADER_2 << 6) | (RNS.Transport.TRANSPORT << 4) \ + | (RNS.Destination.LINK << 2) | RNS.Packet.DATA + return (struct.pack("!B", flags) + struct.pack("!B", 1) + + RNS.Transport.identity.hash + link_id + + bytes([RNS.Packet.NONE]) + payload) + + +def _terminus_link_frame(link, payload): + """One HEADER_1|BROADCAST|LINK|DATA frame for a local active link, + encrypted with the link token.""" + flags = (RNS.Packet.HEADER_1 << 6) | (RNS.Transport.BROADCAST << 4) \ + | (RNS.Destination.LINK << 2) | RNS.Packet.DATA + ciphertext = link.encrypt(payload) + return (struct.pack("!B", flags) + struct.pack("!B", 0) + + link.link_id + bytes([RNS.Packet.NONE]) + ciphertext) + + +def _payload(i, size, marker): + """Deterministic, unique payload of `size` bytes.""" + return i.to_bytes(8, "big") + bytes([marker]) * (size - 8) + + +def build_scenarios(instance, interface, interface_b, requested=None): + """Builds all benchmark scenarios and returns a dict name -> Scenario.""" + scenarios = {} + runs = BENCHMARK_CONFIG["runs"] + + def _slices(packets, fresh=False): + inline_n = packets.get("inline", 0) + drainer_n = packets.get("drainer", 0) + if fresh: + # Fresh frames per run for both modes + total = (inline_n + drainer_n) * runs + return total, inline_n, drainer_n, 0, inline_n * runs + else: + # One re-usable pool of the largest mode's frame count + total = max(inline_n, drainer_n) + return total, inline_n, drainer_n, 0, 0 + + # --- SINGLE transit relay, several payload sizes ----------------------- + sizes = [("135", 135), ("475", 475), ("1024", 1024), ("16384", 16384)] + for suffix, size in sizes: + name = f"transit_single_{suffix}" + total, inline_n, drainer_n, off_i, off_d = _slices(DEFAULT_PACKETS[name]) + dst_h = RNS.Cryptography.hkdf(length=16, + derive_from=f"T{suffix}dst".encode(), + salt=b"bench", context=None) + next_hop = RNS.Cryptography.hkdf(length=16, + derive_from=f"T{suffix}nh".encode(), + salt=b"bench", context=None) + now = time.time() + with RNS.Transport.path_table_lock: + RNS.Transport.path_table[dst_h] = [ + now, next_hop, 3, now + 3600, [], interface, bytes(32) + ] + frames = [_transit_single_frame(interface, dst_h, + _payload(i, size, 0x57)) + for i in range(total)] + scenarios[name] = Scenario( + name, + f"SINGLE transit relay, {size} B payload, 3-hop path", + interface, frames, DEFAULT_PACKETS[name], runs, + off_i, off_d, size, + completion_counter=lambda: RNS.Transport.tx_packets, + ) + + # --- SINGLE transit relay, final hop (header strip) -------------------- + name = "transit_single_final" + total, inline_n, drainer_n, off_i, off_d = _slices(DEFAULT_PACKETS[name]) + dst_h = RNS.Cryptography.hkdf(length=16, derive_from=b"TFdst", + salt=b"bench", context=None) + next_hop = RNS.Cryptography.hkdf(length=16, derive_from=b"TFnh", + salt=b"bench", context=None) + now = time.time() + with RNS.Transport.path_table_lock: + RNS.Transport.path_table[dst_h] = [ + now, next_hop, 1, now + 3600, [], interface, bytes(32) + ] + frames = [_transit_single_frame(interface, dst_h, + _payload(i, 135, 0x46)) + for i in range(total)] + scenarios[name] = Scenario( + name, + "SINGLE transit relay, final hop, transport header strip, 135 B", + interface, frames, DEFAULT_PACKETS[name], runs, + off_i, off_d, 135, + completion_counter=lambda: RNS.Transport.tx_packets, + ) + + # --- LINK transit relay (cross-interface) ------------------------------ + sizes = [("135", 135), ("475", 475), ("1024", 1024), ("16384", 16384)] + for suffix, size in sizes: + name = f"transit_link_{suffix}" + + total, inline_n, drainer_n, off_i, off_d = _slices(DEFAULT_PACKETS[name]) + link_id = RNS.Cryptography.hkdf(length=16, derive_from=b"Llink", + salt=b"bench", context=None) + now = time.time() + with RNS.Transport.link_table_lock: + # timestamp, next-hop transport id, outbound iface, remaining hops, + # received-on iface, taken hops, original destination hash, + # validated, proof timeout + RNS.Transport.link_table[link_id] = [ + now, RNS.Transport.identity.hash, interface_b, 2, + interface, 2, os.urandom(16), True, now + 60 + ] + frames = [_transit_link_frame(interface, link_id, + _payload(i, size, 0x4C)) + for i in range(total)] + scenarios[name] = Scenario( + name, + f"LINK transit relay via link table, cross-interface, {size} B", + interface, frames, DEFAULT_PACKETS[name], runs, + off_i, off_d, size, + completion_counter=lambda: RNS.Transport.tx_packets, + ) + + # --- LINK terminus delivery -------------------------------------------- + name = "terminus_link" + total, inline_n, drainer_n, off_i, off_d = _slices(DEFAULT_PACKETS[name]) + link = _make_minimal_link(instance, interface) + _register_link(link) + frames = [_terminus_link_frame(link, _payload(i, 135, 0x6C)) + for i in range(total)] + scenarios[name] = Scenario( + name, + "LINK terminus delivery, token decrypt, no app callback, 135 B", + interface, frames, DEFAULT_PACKETS[name], runs, + off_i, off_d, 135, + completion_counter=lambda: link.rx, + ) + + # --- SINGLE local delivery --------------------------------------------- + name = "terminus_single" + total, inline_n, drainer_n, off_i, off_d = _slices(DEFAULT_PACKETS[name]) + identity = RNS.Identity() + # The Destination initialiser automatically registers IN destinations + # with the transport core. + destination = CountingDestination(identity, RNS.Destination.IN, + RNS.Destination.SINGLE, + "bench", "terminus") + frames = [] + for i in range(total): + packet = RNS.Packet(destination, _payload(i, 135, 0x53), + RNS.Packet.DATA, create_receipt=False) + packet.pack() + frames.append(packet.raw) + scenarios[name] = Scenario( + name, + "SINGLE local delivery, ephemeral-key decrypt, 135 B", + interface, frames, DEFAULT_PACKETS[name], runs, + off_i, off_d, 135, + completion_counter=lambda: destination.delivered, + ) + + # --- Announce ingress ---------------------------------------------------- + name = "announce_ingress" + total, inline_n, drainer_n, off_i, off_d = _slices(DEFAULT_PACKETS[name], + fresh=True) + name_hash = RNS.Identity.full_hash(b"rns.throughput.bench")[ + :RNS.Identity.NAME_HASH_LENGTH//8] + frames = []; tsize = 0 + for i in range(total): + ann_identity = RNS.Identity() + dst_h = RNS.Identity.full_hash(name_hash + ann_identity.hash)[ + :RNS.Reticulum.TRUNCATED_HASHLENGTH//8] + random_hash = os.urandom(5) + int(time.time() + i).to_bytes(5, "big") + signed_data = dst_h + ann_identity.get_public_key() + name_hash \ + + random_hash + b"" + signature = ann_identity.sign(signed_data) + announce_data = ann_identity.get_public_key() + name_hash \ + + random_hash + b"" + signature + flags = (RNS.Packet.HEADER_1 << 6) \ + | (RNS.Destination.SINGLE << 2) | RNS.Packet.ANNOUNCE + fbs = struct.pack("!B", flags) + struct.pack("!B", 0) + dst_h + bytes([RNS.Packet.NONE]) + announce_data + frames.append(fbs) + tsize += len(fbs) + + scenarios[name] = Scenario( + name, + "Announce ingress, fresh destinations, validation + path insert", + interface, frames, DEFAULT_PACKETS[name], runs, + off_i, off_d, int(tsize/total), + completion_counter=lambda: len(RNS.Transport.path_table), + fresh_per_run=True, + ) + + # --- Outbound insertion into transport ---------------------------------- + name = "outbound_path" + packets = DEFAULT_PACKETS[name] + inline_n = packets.get("inline", 0) + total = inline_n # pool is re-used across runs + dst_h = RNS.Cryptography.hkdf(length=16, derive_from=b"Odst", + salt=b"bench", context=None) + next_hop = RNS.Cryptography.hkdf(length=16, derive_from=b"Onh", + salt=b"bench", context=None) + now = time.time() + with RNS.Transport.path_table_lock: + RNS.Transport.path_table[dst_h] = [ + now, next_hop, 3, now + 3600, [], interface, bytes(32) + ] + remote_id = RNS.Identity(create_keys=False) + remote_id.load_public_key(os.urandom(RNS.Identity.KEYSIZE//8)) + outbound_destination = RNS.Destination(remote_id, RNS.Destination.OUT, + RNS.Destination.SINGLE, + "bench", "outbound") + outbound_destination.hash = dst_h + outbound_destination.hexhash = dst_h.hex() + frames = [] + for i in range(total): + packet = RNS.Packet(outbound_destination, _payload(i, 135, 0x4F), + RNS.Packet.DATA, create_receipt=False) + packet.pack() + frames.append(packet) + outbound = Scenario( + name, + "Outbound insertion into transport, known 3-hop path, 135 B", + interface, frames, packets, runs, + 0, 0, 135, + completion_counter=None, + ) + outbound.kind = "outbound" + scenarios[name] = outbound + + if requested is not None: + missing = [s for s in requested if s not in scenarios] + if missing: + raise KeyError(f"Unknown scenario(s): {', '.join(missing)}. " + f"Available: {', '.join(sorted(scenarios))}") + scenarios = {name: scenarios[name] for name in requested} + + return scenarios + + +def _reset_transport_state(): + RNS.Transport.packet_hashlist = set() + RNS.Transport.packet_hashlist_prev = set() + RNS.Transport.reverse_table = {} + + +def _percentile(sorted_values, p): + if not sorted_values: + return 0.0 + index = min(len(sorted_values) - 1, int(len(sorted_values) * p)) + return sorted_values[index] + + +def _feed(frame, interface): + RNS.Transport.preprocess_inbound(frame, interface) + +def _bench_inline(interface, frames, n, runs, offset, fresh_per_run=False): + """Synchronous benchmark: process each frame in the calling thread, + bypassing the inbound queue. Returns (run_means_us, samples_us).""" + run_means = [] + all_samples = [] + + # Bypass the inbound queue so processing happens entirely in the + # calling thread; restore the previous setting afterwards. + previous_queue_state = RNS.Transport.USE_INBOUND_QUEUE + RNS.Transport.USE_INBOUND_QUEUE = False + try: + for r in range(runs): + _reset_transport_state() + gc.collect() + gc.disable() + + base = offset + (r * n if fresh_per_run else 0) + t0 = time.perf_counter() + prev = None + samples = [] + step = INLINE_SAMPLE_STEP + next_sample = step + for i in range(n): + _feed(frames[base + i], interface) + if i == next_sample: + now = time.perf_counter() + if prev is not None: + samples.append((now - prev) / step) + prev = now + next_sample += step + dt = time.perf_counter() - t0 + gc.enable() + + run_means.append(dt / n * 1e6) + all_samples.extend(samples) + finally: + RNS.Transport.USE_INBOUND_QUEUE = previous_queue_state + + return run_means, [s * 1e6 for s in all_samples] + + +def _bench_drainer(scenario, n, runs): + """Feed frames through the inbound queue from the calling thread + while the drainer thread processes them. Backpressure is applied + via the scenario completion counter, so no frames are dropped and + the pipeline stays saturated.""" + run_means = [] + + # Ensure the inbound queue and drainer are used; restore afterwards. + previous_queue_state = RNS.Transport.USE_INBOUND_QUEUE + RNS.Transport.USE_INBOUND_QUEUE = True + try: + for r in range(runs): + _reset_transport_state() + gc.collect() + gc.disable() + + base = scenario.drainer_offset \ + + (r * n if scenario.fresh_per_run else 0) + baseline = scenario.completion_counter() + fed = 0 + t0 = time.perf_counter() + chunk = DRAINER_CHUNK + while fed < n: + batch = min(chunk, n - fed) + for i in range(fed, fed + batch): + _feed(scenario.frames[base + i], scenario.interface) + fed += batch + target = baseline + fed + while scenario.completion_counter() < target: + time.sleep(0.0002) + dt = time.perf_counter() - t0 + gc.enable() + + run_means.append(dt / n * 1e6) + finally: + RNS.Transport.USE_INBOUND_QUEUE = previous_queue_state + + return run_means + + +def _bench_outbound(packets, n, runs, offset): + """Synchronous benchmark of the outbound path: insert each packet + into transport for a known path via Transport.outbound(). The packet + pool is re-used across runs.""" + run_means = [] + for r in range(runs): + gc.collect() + gc.disable() + t0 = time.perf_counter() + for i in range(n): + RNS.Transport.outbound(packets[offset + i]) + dt = time.perf_counter() - t0 + gc.enable() + run_means.append(dt / n * 1e6) + return run_means + + +# --------------------------------------------------------------------------- +# Output helpers +# --------------------------------------------------------------------------- + +def _fmt_us(value): + return f"{value:>10.2f}" + + +def _fmt_pps(value): + return f"{value:>12,.0f}" + + +def print_environment_header(): + tree_fastpath = "present" if hasattr(RNS.Transport, "_fastpath") else "absent" + print("=" * 40) + print("Reticulum Transport Throughput Benchmark") + print("=" * 40) + print(f" RNS version : {RNS.__version__}") + print(f" Mode : {'compiled' if RNS.compiled else 'interpreted'}") + print(f" crypto : {RNS.Cryptography.backend()}") + print(f" transport : enabled") + print(f" fast path : {tree_fastpath} (informational)") + print(f" python : {platform.python_version()}") + print(f" platform : {platform.platform()}") + print(f" machine : {platform.machine()}, cpus: {os.cpu_count()}") + print() + + +def print_scenario_table(scenario, results): + """results: dict mode -> dict(runs, mean_us, pps, stdev, p50, p95).""" + print(f"Scenario: {scenario.name} - {scenario.description}") + print(f" {'mode':<9}{'n':>8}{'runs':>6}{'mean µs':>13}{'pps':>16}" + f"{'p50':>11}{'p95':>11}") + print(" " + "-" * 75) + for mode in ("inline", "drainer"): + if mode not in results: + continue + r = results[mode] + n = scenario.n_for_mode(mode) + p50 = f"{r['p50']:>9.2f} µs" if r["p50"] is not None else "-" + p95 = f"{r['p95']:>9.2f} µs" if r["p95"] is not None else "-" + spread = "" + if r.get("stdev") is not None: + spread = f" ±{r['stdev']:.2f}" + if mode == "inline": mode = "direct" + print(f" {mode:<9}{n:>8}{r['runs']:>6}{_fmt_us(r['mean_us']):>13}" + f"{_fmt_pps(r['pps']):>16}{p50:>11}{p95:>11} " + f"({r['runs']} runs, median{spread})") + + print() + for mode in ("inline", "drainer"): + if mode not in results: + continue + + r = results[mode] + tp = r['pps']*scenario.frame_size*8 + print(f"{mode:<8} : {RNS.prettyspeed(tp)}") + + print() + + +def print_pps_matrix(rows): + """rows: list of (scenario_name, inline_pps, drainer_pps, fsize)""" + if not rows: + return + print("-" * 72) + print("Transport Core Throughput - PPS matrix (median of runs)") + print(f" {'scenario':<22}{'direct':>14}{'drainer':>14}") + print(" " + "-" * 70) + for name, inline_pps, drainer_pps, fsize in rows: + i = f"{inline_pps:>12,.0f}" if inline_pps else "-" + d = f"{drainer_pps:>12,.0f}" if drainer_pps else "-" + itp = RNS.prettyspeed(inline_pps*fsize*8) if inline_pps else "-" + dtp = RNS.prettyspeed(drainer_pps*fsize*8) if drainer_pps else "-" + print(f" {name:<22}{i:>14}{d:>14}{itp:>16} / {dtp}") + print() + + +# --------------------------------------------------------------------------- +# Test suite +# --------------------------------------------------------------------------- + +_instance = None +_interfaces = None +_scenarios = None +_config_dir = None + + +def _start_transport(): + global _instance, _interfaces, _scenarios, _config_dir + + _config_dir = tempfile.mkdtemp(prefix="rns-transport-bench-") + config_dir = _config_dir + with open(os.path.join(config_dir, "config"), "w") as fh: + fh.write( + "[reticulum]\n" + " enable_transport = yes\n" + " share_instance = No\n" + " panic_on_interface_error = No\n" + "\n" + "[logging]\n" + " loglevel = 0\n" + "\n" + "[interfaces]\n" + ) + + _instance = RNS.Reticulum(configdir=config_dir, + loglevel=RNS.LOG_CRITICAL, + logdest=lambda *a, **k: None) + + while not RNS.Transport.ready: + time.sleep(0.05) + + interface = BenchmarkInterface("line-a") + interface_b = BenchmarkInterface("line-b") + RNS.Transport.add_interface(interface) + RNS.Transport.add_interface(interface_b) + RNS.Transport.prioritize_interfaces() + _interfaces = (interface, interface_b) + + requested = BENCHMARK_CONFIG["scenarios"] + _scenarios = build_scenarios(_instance, interface, interface_b, + requested=requested) + + +class TestTransportThroughput(unittest.TestCase): + + @classmethod + def setUpClass(cls): + if _scenarios is None: + _start_transport() + + @classmethod + def tearDownClass(cls): + if RNS.Transport._should_run: + RNS.Transport.exit_handler() + if _config_dir is not None: + import shutil + shutil.rmtree(_config_dir, ignore_errors=True) + + def _run_scenario_matrix(self, scenario_names): + rows = [] + # Only include scenarios selected via the command line + scenario_names = [name for name in scenario_names + if name in _scenarios] + for name in scenario_names: + scenario = _scenarios[name] + results = {} + + mode_filter = BENCHMARK_CONFIG["mode"] + runs = BENCHMARK_CONFIG["runs"] + + if scenario.kind == "outbound": + # The outbound path is synchronous; it only has an + # inline-equivalent measurement, and follows the mode + # filter's "inline" selector. + if mode_filter in ("inline", "both"): + run_means = _bench_outbound(scenario.frames, + scenario.packets_inline, + runs, 0) + mean_us = statistics.median(run_means) + results["inline"] = { + "runs": runs, + "mean_us": mean_us, + "pps": 1e6 / mean_us, + "stdev": statistics.stdev(run_means) if len(run_means) > 2 else None, + "p50": None, + "p95": None, + } + else: + if mode_filter in ("inline", "both") and scenario.packets_inline: + n = scenario.packets_inline + run_means, samples = _bench_inline(scenario.interface, + scenario.frames, n, runs, + scenario.inline_offset, + scenario.fresh_per_run) + mean_us = statistics.median(run_means) + stdev = statistics.stdev(run_means) if len(run_means) > 2 else None + sorted_samples = sorted(samples) + results["inline"] = { + "runs": runs, + "mean_us": mean_us, + "pps": 1e6 / mean_us, + "stdev": stdev, + "p50": _percentile(sorted_samples, 0.50), + "p95": _percentile(sorted_samples, 0.95), + } + + if mode_filter in ("drainer", "both") and scenario.can_drainer(): + n = scenario.packets_drainer + run_means = _bench_drainer(scenario, n, runs) + mean_us = statistics.median(run_means) + stdev = statistics.stdev(run_means) if len(run_means) > 2 else None + results["drainer"] = { + "runs": runs, + "mean_us": mean_us, + "pps": 1e6 / mean_us, + "stdev": stdev, + "p50": None, + "p95": None, + } + + if results: + print_scenario_table(scenario, results) + rows.append((name, + results.get("inline", {}).get("pps"), + results.get("drainer", {}).get("pps"), + scenario.frame_size)) + + print_pps_matrix(rows) + + def test_01_transit_throughput(self): + print("") + print_environment_header() + self._run_scenario_matrix([ + "transit_single_135", + "transit_single_475", + "transit_single_1024", + "transit_single_16384", + "transit_single_final", + "transit_link_135", + "transit_link_475", + "transit_link_1024", + "transit_link_16384", + ]) + + def test_02_delivery_throughput(self): + print("") + self._run_scenario_matrix([ + "terminus_link_135", + "terminus_link_475", + "terminus_link_1024", + "terminus_link_16384", + "terminus_single", + "announce_ingress", + "outbound_path", + ]) + + +def _usage(): + return ( + "\nUsage: python3 tests/transport_throughput.py [options]\n" + "\n" + "Options:\n" + " -s, --scenario NAME Run only the named scenario (repeatable)\n" + " --mode MODE Measurement mode: inline, drainer or both\n" + " (default: both)\n" + " --runs N Measurement runs per scenario/mode, median\n" + " is reported (default: 3)\n" + " --list-scenarios List available scenarios and exit\n" + " -h, --help Show this help and exit\n" + ) + + +if __name__ == "__main__": + argv = sys.argv[1:] + rest = [] + i = 0 + while i < len(argv): + arg = argv[i] + if arg in ("-s", "--scenario"): + BENCHMARK_CONFIG["scenarios"] = BENCHMARK_CONFIG["scenarios"] or [] + BENCHMARK_CONFIG["scenarios"].append(argv[i + 1]) + i += 2 + elif arg == "--mode": + mode = argv[i + 1] + if mode not in ("inline", "drainer", "both"): + raise SystemExit(f"Invalid mode '{mode}'" + _usage()) + BENCHMARK_CONFIG["mode"] = mode + i += 2 + elif arg == "--runs": + BENCHMARK_CONFIG["runs"] = max(1, int(argv[i + 1])) + i += 2 + elif arg == "--list-scenarios": + print("\nAvailable scenarios:") + for name in sorted(SCENARIO_DESCRIPTIONS): + print(f" {name:<22} {SCENARIO_DESCRIPTIONS[name]}") + raise SystemExit(0) + elif arg in ("-h", "--help"): + raise SystemExit(_usage()) + else: + rest.append(arg) + i += 1 + + unittest.main(argv=[sys.argv[0]] + rest, verbosity=2) +