diff options
Diffstat (limited to 'research/experiments/020-lossy-paths')
| -rw-r--r-- | research/experiments/020-lossy-paths/run.py | 247 |
1 files changed, 247 insertions, 0 deletions
diff --git a/research/experiments/020-lossy-paths/run.py b/research/experiments/020-lossy-paths/run.py new file mode 100644 index 0000000..4b98d6b --- /dev/null +++ b/research/experiments/020-lossy-paths/run.py @@ -0,0 +1,247 @@ +#!/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 beside TCP CUBIC. + +Each case runs in `unshare -rn` (an unprivileged network namespace) whose +loopback is the path, both ways: MTU 1500, and TCP's segments no larger +(gso_max_size), so each is a packet; a 50 Mbit/s bottleneck (tbf) whose +queue holds about a round trip of it; under it netem's delay (half the +RTT), jitter (packets reorder) and random loss. Pure and quinn 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: 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 or a second TCP flow beside a TCP CUBIC flow, both + for --seconds; each one's share of the goodput from 5 s on. + + python3 research/experiments/020-lossy-paths/run.py [--rounds 3] [--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' +TC = shutil.which('tc') or '/usr/sbin/tc' +IP = shutil.which('ip') or '/usr/sbin/ip' +RATE_MBIT, SIZE = 50, 1000 +sys.path.insert(0, str(ROOT / 'tests')) + + +def build(work): + """The transport harness, quinn'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) + 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.""" + 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 + subprocess.run([TC, 'qdisc', 'add', 'dev', 'lo', 'root', 'handle', '1:', 'tbf', 'rate', f'{RATE_MBIT}mbit', + 'burst', '15000', 'limit', str(queue)], check=True) + netem = ['delay', f'{rtt_ms / 2}ms'] + ([f'{jitter_ms}ms'] if jitter_ms else []) + \ + (['loss', f'{loss}%'] if loss else []) + ['limit', '100000'] + subprocess.run([TC, 'qdisc', 'add', 'dev', 'lo', 'parent', '1:1', 'handle', '10:', 'netem'] + netem, 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 pure(work, seconds, rate=0, count=2000000): + done = subprocess.run([str(ROOT / 'build/link-transport'), str(work / 'sender.pmod'), str(work / 'receiver.pmod'), + str(count), str(SIZE), '20000', str(rate), str(seconds)], capture_output=True, text=True, + env={**os.environ, 'TRANSPORT_GSO': '0'}, timeout=seconds + 120) + assert done.returncode == 0, done.stderr[-2000:] + return json.loads(done.stdout) + + +def quinn_server(work): + server = subprocess.Popen([str(QUINN), 'serve', str(work / 'cert.der'), str(work / 'key.der')], + stdout=subprocess.PIPE, text=True, env={**os.environ, 'QUIC_GSO': '0'}) + port = json.loads(server.stdout.readline())['port'] + return server, port + + +def quinn(work, seconds, rate=0, count=2000000): + server, port = quinn_server(work) + try: + # The bytes of Pure's record: its prefix (count, item's length), the item and the city ("Seoul"). + client = subprocess.run([str(QUINN), 'drain', str(port), str(count), str(SIZE + 8 + 5), str(rate), str(seconds)], + capture_output=True, text=True, env={**os.environ, 'QUIC_GSO': '0'}, + timeout=seconds + 300) + assert client.returncode == 0, client.stderr[-2000:] + line = server.stdout.readline() + finally: + server.kill() + server.wait() + return json.loads(line) + + +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): + self.seconds = seconds + self.listener = socket.socket() + self.listener.bind(('127.0.0.1', 0)) + 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 quinn(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 + tcp = Tcp(seconds) + tcp.start() + if who == 'tcp': + other = Tcp(seconds) + other.start() + theirs = [b / (SIZE + 13) for b in other.join()] + else: + theirs = (pure(work, seconds) if who == 'pure' else quinn(work, seconds))['timeline_100ms'] + mine = tcp.join() + row['mbit'] = goodput(theirs, SIZE + 13, 5.0, seconds - 1) + row['tcp_mbit'] = goodput(mine, 1, 5.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): + for rtt in (20, 100): + for loss in (0, 1, 5): + for jitter in (0, 3): + for who in ('pure', 'quinn'): + 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'): + yield {'kind': 'paced', 'who': who, 'rtt_ms': rtt, 'loss': loss, 'jitter_ms': 0, 'seconds': seconds} + for who in ('pure', 'quinn', 'tcp'): + yield {'kind': 'share', 'who': who, 'rtt_ms': rtt, 'loss': 0, 'jitter_ms': 0, '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=30) + parser.add_argument('--only', default=None, help='kinds to run, for example bulk,share') + parser.add_argument('--out', type=Path, default=None) + parser.add_argument('--inner', default=None, help=argparse.SUPPRESS) + args = parser.parse_args() + 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): + 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() |