Added throughput benchmarker
Mark Qvist
parent
0e070aacf6
commit
b9278ce352
|
|
@ -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)
|
||||
|
||||
Loading…
Reference in New Issue