#!/usr/bin/env python3 """M6 of docs/design/011-pure-transport.md: a link on a path of delay, loss, reordering and a bottleneck, against quinn and ngtcp2, and beside TCP CUBIC. Each case runs in `unshare -rn` (an unprivileged network namespace) whose loopback is the path: MTU 1500, and TCP's segments no larger (gso_max_size), so each is a packet. Each way netem gives delay (half the RTT), jitter (packets reorder) and random loss; the way to the receivers (ports 40000-40007) then goes through a 50 Mbit/s bottleneck (tbf) whose queue holds a round trip of it, and the way back (answers, TCP's acknowledgments) does not queue behind it. Pure and the QUICs send without UDP GSO: a qdisc takes a GSO burst as one packet. bulk a backlog of 1 KiB records for --seconds (Pure: a link and a listener, transport.c; quinn and ngtcp2: one stream); the goodput from the receiver's records in each 100 ms, the first 2 s left out. paced 50 records a second for --seconds (well within what each carries); the time each took from queued (written) to handed over (read): p50, p99, max. share Pure, quinn, ngtcp2 or a second TCP flow beside a TCP CUBIC flow, each a process of its own, started together, for --share-seconds; each one's share of the goodput from 20 s on (CUBIC's flows converge slowly at an RTT of 100 ms). The path's delay wavers by 1 ms: with a fixed delay a drop-tail queue drops one flow's packets far more than the other's by their phases alone (Floyd and Jacobson, 1992), and which flow that is says nothing of either. python3 research/experiments/020-lossy-paths/run.py [--rounds 3] [--who pure,ngtcp2] [--out FILE] """ import argparse import json import os import shutil import socket import statistics import subprocess import sys import tempfile import threading import time from pathlib import Path ROOT = Path(__file__).resolve().parents[3] HERE = Path(__file__).resolve().parent QUINN = ROOT / 'research/experiments/019-idle-links/quic/target/release/idle-quic' NGTCP2 = ROOT / 'build/third/ngtcp2-drain' TC = shutil.which('tc') or '/usr/sbin/tc' IP = shutil.which('ip') or '/usr/sbin/ip' RATE_MBIT, SIZE = 50, 1000 PURE_PORT, QUINN_PORT, TCP_PORT, OTHER_TCP_PORT, NGTCP2_PORT = 40001, 40002, 40003, 40004, 40005 # the receivers, beyond the bottleneck sys.path.insert(0, str(ROOT / 'tests')) def build(work): """The transport harness, quinn's, ngtcp2's, the two modules and a certificate.""" import check import importlib.util spec = importlib.util.spec_from_file_location('idle_links', ROOT / 'research/experiments/019-idle-links/run.py') idle_links = importlib.util.module_from_spec(spec) spec.loader.exec_module(idle_links) subprocess.run(['make', '-s', '-C', str(ROOT), 'build/libpure.a', 'build/libpure-link.a', 'build/pure-crypto.so'], check=True) subprocess.run(['cc', '-O2', '-g', '-std=c17', '-Wall', '-Wextra', '-Werror', f'-I{ROOT}/core', f'-I{ROOT}/link', str(ROOT / 'research/experiments/019-idle-links/transport.c'), str(ROOT / 'build/libpure-link.a'), str(ROOT / 'build/libpure.a'), '-lm', '-pthread', '-ldl', '-o', str(ROOT / 'build/link-transport')], check=True) subprocess.run(['cargo', 'build', '--release', '--offline', '--quiet'], cwd=QUINN.parents[2], check=True) subprocess.run(['sh', str(ROOT / 'research/experiments/019-idle-links/ngtcp2/build.sh')], check=True) for name, text in (('sender', check.LINK_SENDER), ('receiver', check.LINK_RECEIVER)): (work / f'{name}.aup').write_text(text) check.compile_module(Path(os.environ.get('AUP', ROOT.parent / 'aup/build/aup')), work / f'{name}.aup', work / f'{name}.pmod') idle_links.make_certificate(work) # ---- Inside the namespace def path(rtt_ms, loss, jitter_ms): """The loopback as the path: back (band 1:1) and forth (1:2, to the receivers' ports, with the bottleneck after its netem).""" subprocess.run([IP, 'link', 'set', 'lo', 'up', 'mtu', '1500', 'gso_max_size', '1500', 'gso_max_segs', '1'], check=True) queue = max(30000, RATE_MBIT * 1000000 // 8 * rtt_ms // 1000) # a round trip of the bottleneck, in bytes netem = ['delay', f'{rtt_ms / 2}ms'] + ([f'{jitter_ms}ms'] if jitter_ms else []) + \ (['loss', f'{loss}%'] if loss else []) + ['limit', '100000'] for rule in (['qdisc', 'add', 'dev', 'lo', 'root', 'handle', '1:', 'prio', 'bands', '2', 'priomap'] + ['0'] * 16, ['qdisc', 'add', 'dev', 'lo', 'parent', '1:1', 'handle', '10:', 'netem'] + netem, ['qdisc', 'add', 'dev', 'lo', 'parent', '1:2', 'handle', '20:', 'netem'] + netem, ['qdisc', 'add', 'dev', 'lo', 'parent', '20:1', 'handle', '21:', 'tbf', 'rate', f'{RATE_MBIT}mbit', 'burst', '15000', 'limit', str(queue)], ['filter', 'add', 'dev', 'lo', 'protocol', 'ip', 'parent', '1:0', 'prio', '1', 'u32', 'match', 'ip', 'dport', '40000', '0xfff8', 'flowid', '1:2']): subprocess.run([TC] + rule, check=True) def goodput(timeline, record_bytes, skip_s=2.0, until_s=None): """Mbit/s of records from skip_s on (to until_s).""" first = int(skip_s * 10) last = len(timeline) if until_s is None else min(len(timeline), int(until_s * 10)) counted = timeline[first:last] return round(sum(counted) * record_bytes * 8 / (len(counted) / 10) / 1e6, 2) if counted else 0.0 def start_pure(work, seconds, rate=0, count=2000000): return subprocess.Popen([str(ROOT / 'build/link-transport'), str(work / 'sender.pmod'), str(work / 'receiver.pmod'), str(count), str(SIZE), '20000', str(rate), str(seconds)], stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, env={**os.environ, 'TRANSPORT_GSO': '0', 'TRANSPORT_PORT': str(PURE_PORT)}) def finish_pure(process, seconds): out, error = process.communicate(timeout=seconds + 120) assert process.returncode == 0, error[-2000:] return json.loads(out) def pure(work, seconds, rate=0): return finish_pure(start_pure(work, seconds, rate), seconds) def quinn_server(work): server = subprocess.Popen([str(QUINN), 'serve', str(work / 'cert.der'), str(work / 'key.der'), str(QUINN_PORT)], stdout=subprocess.PIPE, text=True, env={**os.environ, 'QUIC_GSO': '0'}) port = json.loads(server.stdout.readline())['port'] return server, port def start_quinn(work, seconds, rate=0, count=2000000): server, port = quinn_server(work) # The bytes of Pure's record: its prefix (count, item's length), the item and the city ("Seoul"). client = subprocess.Popen([str(QUINN), 'drain', str(port), str(count), str(SIZE + 8 + 5), str(rate), str(seconds)], stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, env={**os.environ, 'QUIC_GSO': '0'}) return server, client def finish_quinn(started, seconds): server, client = started try: _, error = client.communicate(timeout=seconds + 300) assert client.returncode == 0, error[-2000:] line = server.stdout.readline() finally: server.kill() server.wait() return json.loads(line) def start_ngtcp2(work, seconds, rate=0, count=2000000): """As quinn's: the same messages over ngtcp2 (019's ngtcp2/drain.c).""" env = {**os.environ, 'DRAIN_GSO': '0'} server = subprocess.Popen([str(NGTCP2), 'serve', str(work / 'cert.der'), str(work / 'key.der'), str(NGTCP2_PORT)], stdout=subprocess.PIPE, text=True, env=env) port = json.loads(server.stdout.readline())['port'] client = subprocess.Popen([str(NGTCP2), 'drain', str(port), str(count), str(SIZE + 8 + 5), str(rate), str(seconds)], stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, env=env) return server, client def quic(who, work, seconds, rate=0): started = (start_quinn if who == 'quinn' else start_ngtcp2)(work, seconds, rate) return finish_quinn(started, seconds) def start_tcp(port, seconds): """A TCP CUBIC flow in a process of its own (--tcp).""" return subprocess.Popen([sys.executable, __file__, '--tcp', f'{port}:{seconds}'], stdout=subprocess.PIPE, text=True) def finish_tcp(process, seconds): out, _ = process.communicate(timeout=seconds + 60) assert process.returncode == 0 return json.loads(out)['timeline_100ms'] class Tcp: """A TCP CUBIC flow: a sender writing as fast as it may for seconds, a receiver counting bytes in each 100 ms from when it starts.""" def __init__(self, seconds, port): self.seconds = seconds self.listener = socket.socket() self.listener.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) self.listener.bind(('127.0.0.1', port)) self.listener.listen(1) self.timeline = [] self.began = None def start(self): self.began = time.monotonic() self.receiver = threading.Thread(target=self.receive) self.sender = threading.Thread(target=self.send) self.receiver.start() self.sender.start() def receive(self): connection, _ = self.listener.accept() with connection: while True: data = connection.recv(1 << 20) if not data: break bucket = int((time.monotonic() - self.began) * 10) while len(self.timeline) <= bucket: self.timeline.append(0) self.timeline[bucket] += len(data) def send(self): s = socket.socket() s.setsockopt(socket.IPPROTO_TCP, socket.TCP_CONGESTION, b'cubic') s.connect(self.listener.getsockname()) chunk = b'x' * 65536 while time.monotonic() - self.began < self.seconds: s.sendall(chunk) s.close() def join(self): self.sender.join() self.receiver.join() self.listener.close() return self.timeline def inner(case): """One case, in the namespace: a line of JSON.""" work = Path(case['work']) path(case['rtt_ms'], case['loss'], case['jitter_ms']) seconds, kind, who = case['seconds'], case['kind'], case['who'] row = dict(case) del row['work'] if kind in ('bulk', 'paced'): rate = 50 if kind == 'paced' else 0 result = pure(work, seconds, rate) if who == 'pure' else quic(who, work, seconds, rate) timeline = result['timeline_100ms'] row['mbit'] = goodput(timeline, SIZE + 13, 2.0, seconds) for key in ('waited_ms_p50', 'waited_ms_p99', 'waited_ms_max', 'resent', 'lost', 'late', 'undone', 'packets'): if key in result: row[key] = result[key] else: # share: who beside a TCP CUBIC flow, started together tcp = start_tcp(TCP_PORT, seconds) if who == 'tcp': theirs = [b / (SIZE + 13) for b in finish_tcp(start_tcp(OTHER_TCP_PORT, seconds), seconds)] elif who == 'pure': theirs = finish_pure(start_pure(work, seconds), seconds)['timeline_100ms'] else: theirs = finish_quinn((start_quinn if who == 'quinn' else start_ngtcp2)(work, seconds), seconds)['timeline_100ms'] mine = finish_tcp(tcp, seconds) row['mbit'] = goodput(theirs, SIZE + 13, 20.0, seconds - 1) row['tcp_mbit'] = goodput(mine, 1, 20.0, seconds - 1) total = row['mbit'] + row['tcp_mbit'] row['share'] = round(row['mbit'] / total, 3) if total else 0.0 return row # ---- Outside def cases(seconds, share_seconds, whom=('pure', 'quinn', 'ngtcp2', 'tcp')): for rtt in (20, 100): for loss in (0, 1, 5): for jitter in (0, 3): for who in ('pure', 'quinn', 'ngtcp2'): if who in whom: yield {'kind': 'bulk', 'who': who, 'rtt_ms': rtt, 'loss': loss, 'jitter_ms': jitter, 'seconds': seconds} for loss in (0, 1, 5): for who in ('pure', 'quinn', 'ngtcp2'): if who in whom: yield {'kind': 'paced', 'who': who, 'rtt_ms': rtt, 'loss': loss, 'jitter_ms': 0, 'seconds': seconds} for who in ('pure', 'quinn', 'ngtcp2', 'tcp'): if who in whom: yield {'kind': 'share', 'who': who, 'rtt_ms': rtt, 'loss': 0, 'jitter_ms': 1, 'seconds': share_seconds} def main(): parser = argparse.ArgumentParser() parser.add_argument('--rounds', type=int, default=1) parser.add_argument('--seconds', type=int, default=20) parser.add_argument('--share-seconds', type=int, default=60) parser.add_argument('--only', default=None, help='kinds to run, for example bulk,share') parser.add_argument('--who', default='pure,quinn,ngtcp2,tcp', help='whom to run, for example pure,ngtcp2') parser.add_argument('--out', type=Path, default=None) parser.add_argument('--inner', default=None, help=argparse.SUPPRESS) parser.add_argument('--tcp', default=None, help=argparse.SUPPRESS) args = parser.parse_args() if args.tcp: port, seconds = (int(v) for v in args.tcp.split(':')) flow = Tcp(seconds, port) flow.start() print(json.dumps({'timeline_100ms': flow.join()}), flush=True) return if args.inner: print(json.dumps(inner(json.loads(args.inner))), flush=True) return with tempfile.TemporaryDirectory(prefix='pure-m6-') as directory: work = Path(directory) build(work) only = set(args.only.split(',')) if args.only else None for round_ in range(args.rounds): for case in cases(args.seconds, args.share_seconds, tuple(args.who.split(','))): if only and case['kind'] not in only: continue case = {**case, 'round': round_, 'work': str(work)} done = subprocess.run(['unshare', '-rn', sys.executable, __file__, '--inner', json.dumps(case)], capture_output=True, text=True, timeout=case['seconds'] + 600) if done.returncode: print(json.dumps({**case, 'error': done.stderr[-1500:]}), flush=True) continue line = done.stdout.strip().splitlines()[-1] print(line, flush=True) if args.out: with open(args.out, 'a') as out: out.write(line + '\n') if __name__ == '__main__': main()