Unsignedio
/
Reticulum
mirror of https://github.com/markqvist/Reticulum.git
1
0
Fork 0
Code Issues Releases Wiki Activity
Reticulum/tests/throughput.py

952 lines
38 KiB
Python
Raw Permalink Blame History

#!/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",
"transit_link_32768": "LINK transit relay via link table, cross-interface, 32768 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},
"transit_link_32768": {"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), ("32768", 32768)]
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-800: # Assumes default inbound queue of 1024
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, "USE_FP_CACHE") 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}")
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
if mode == "inline": mode = "direct"
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 Throughput, 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"
)
# Raise queue lengths for to avoid drops
# at high feed rates
RNS.Transport.INBOUND_DA_QUEUE_LENGTH = 4096
RNS.Transport.INBOUND_AN_QUEUE_LENGTH = 1024
RNS.Transport.INBOUND_PR_QUEUE_LENGTH = 1024
RNS.Transport.INBOUND_IL_QUEUE_LENGTH = 1024
_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",
"transit_link_32768",
])
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\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)
# Pre:
#
# Transport Throughput - PPS matrix (median of runs)
# scenario direct drainer
# ----------------------------------------------------------------------
# transit_single_135 208,550 184,075 225.23 Mbps / 198.80 Mbps
# transit_single_475 195,652 130,808 743.48 Mbps / 497.07 Mbps
# transit_single_1024 169,828 111,446 1.39 Gbps / 912.97 Mbps
# transit_single_16384 49,618 74,101 6.50 Gbps / 9.71 Gbps
# transit_single_final 204,604 112,925 220.97 Mbps / 121.96 Mbps
# transit_link_135 272,813 234,430 294.64 Mbps / 253.18 Mbps
# transit_link_475 260,003 227,895 988.01 Mbps / 866.00 Mbps
# transit_link_1024 232,763 145,872 1.91 Gbps / 1.19 Gbps
# transit_link_16384 84,028 73,313 11.01 Gbps / 9.61 Gbps
#
# terminus_single 29,813 25,663 32.20 Mbps / 27.72 Mbps
# announce_ingress 7,561 7,242 10.10 Mbps / 9.68 Mbps
# outbound_path 696,106 - 751.79 Mbps / -
#
# No Fastpath:
#
# Transport Throughput - PPS matrix (median of runs)
# scenario direct drainer
# ----------------------------------------------------------------------
# transit_single_135 247,667 162,090 267.48 Mbps / 175.06 Mbps
# transit_single_475 233,514 159,717 887.35 Mbps / 606.92 Mbps
# transit_single_1024 211,443 175,994 1.73 Gbps / 1.44 Gbps
# transit_single_16384 78,963 72,532 10.35 Gbps / 9.51 Gbps
# transit_single_final 240,855 112,702 260.12 Mbps / 121.72 Mbps
# transit_link_135 266,614 220,571 287.94 Mbps / 238.22 Mbps
# transit_link_475 256,347 152,410 974.12 Mbps / 579.16 Mbps
# transit_link_1024 228,080 194,444 1.87 Gbps / 1.59 Gbps
# transit_link_16384 82,149 73,720 10.77 Gbps / 9.66 Gbps
#
# terminus_single 29,764 28,308 32.15 Mbps / 30.57 Mbps
# announce_ingress 7,394 7,150 9.88 Mbps / 9.55 Mbps
# outbound_path 872,747 - 942.57 Mbps / -
# ----------------------------------------------------------------------
#
#
# Fastpath:
#
# Transport Throughput - PPS matrix (median of runs)
# scenario direct drainer
# ----------------------------------------------------------------------
# transit_single_135 407,561 411,426 440.17 Mbps / 444.34 Mbps
# transit_single_475 378,867 377,964 1.44 Gbps / 1.44 Gbps
# transit_single_1024 320,878 325,910 2.63 Gbps / 2.67 Gbps
# transit_single_16384 94,265 94,357 12.36 Gbps / 12.37 Gbps
# transit_single_final 395,245 399,247 426.86 Mbps / 431.19 Mbps
# transit_link_135 453,216 451,863 489.47 Mbps / 488.01 Mbps
# transit_link_475 417,013 417,294 1.58 Gbps / 1.59 Gbps
# transit_link_1024 347,548 353,783 2.85 Gbps / 2.90 Gbps
# transit_link_16384 96,682 96,871 12.67 Gbps / 12.70 Gbps
#
# terminus_single 29,451 25,862 31.81 Mbps / 27.93 Mbps
# announce_ingress 7,495 6,996 10.01 Mbps / 9.35 Mbps
# outbound_path 840,738 - 908.00 Mbps / -
Reference in New Issue View Git Blame Copy Permalink