From c97e4e1a203d13852d5f7cd4a404f88b44f4064f Mon Sep 17 00:00:00 2001 From: eunhhu Date: Tue, 29 Sep 2026 06:24:43 +0100 Subject: Measure the CPU a record costs a link, against quinn (096) M5 of design 011. At first a link spent eight times quinn's CPU on a small record: scans of the backlog, a system call and allocations a record, 19 bytes of framing. After the seven changes before this one, the transport alone (transport.c; a backlog drained, every record checked) spends per record 0.30 us for 61 bytes (quinn 0.27: 1.06 to 1.12 times over three campaigns, at the design's 10 % line), 2.91 us for 1,013 bytes (quinn 3.88) and 45.2 us for 16,013 bytes (quinn 60.9): 25 % less for the larger ones, and 30 % more records a second. What is left for small records is per packet: the stream's names and the RECORDS head. quinn's numbers carry no confirmation of storing and no exactly once. Between two pure-serve nodes the links spend more per record, woken at the receiving app's pace. Filling the sender showed the outbox is written whole at every save, as the numbers heard were (094): the next thing to change. run.py gains --transport and --drain; the quinn service reads a drain a chunk at a time and its client writes 64 KiB at a time. --- research/experiments/019-idle-links/run.py | 193 ++++++++++++++++++++++++++++- 1 file changed, 192 insertions(+), 1 deletion(-) (limited to 'research/experiments/019-idle-links/run.py') diff --git a/research/experiments/019-idle-links/run.py b/research/experiments/019-idle-links/run.py index 63f88a2..96a0cbc 100644 --- a/research/experiments/019-idle-links/run.py +++ b/research/experiments/019-idle-links/run.py @@ -29,9 +29,17 @@ network namespace of its own (unshare -rn) whose loopback delays UDP: senders the node never heard from, and new quinn clients, each send one order and wait until it is stored, one at a time. +With --transport SIZE:COUNT,... (M5), transport.c drains a backlog of +COUNT records through a link and a listener in one process (no app, no +store), then quinn sends as many messages of the same size over one +stream; --rounds times each. With --drain SIZES (M5 as the design says), +two pure-serve nodes: the sender's app saves --counts orders while the +receiver is away, then the receiver starts and the link drains them; +quinn does the same over one stream. + python3 research/experiments/019-idle-links/run.py [--counts 1000,10000] [--store DIRECTORY] [--out FILE.jsonl] [--quic] [--restart [--no-memo]] - [--first-contact 5,25,50] + [--first-contact 5,25,50] [--transport 48:200000 --rounds 5] [--drain 48,1000] """ import argparse import json @@ -384,6 +392,143 @@ with tempfile.TemporaryDirectory(prefix='pure-m4-') as d: file.write(line + '\n') +def thread_cpu(pid, name): + """CPU ms of the process's threads named name (all threads when None), + from the scheduler's nanoseconds (schedstat), not clock ticks.""" + total = 0 + for task in Path(f'/proc/{pid}/task').iterdir(): + try: + if name and (task / 'comm').read_text().strip() != name: + continue + total += int((task / 'schedstat').read_text().split()[0]) + except OSError: + pass + return total / 1e6 + + +def post_all(port, n, size, threads=8): + """n orders (item of size bytes) posted to the sender app, over threads connections.""" + import http.client + import threading + body = json.dumps({'item': 'x' * size, 'count': 1}).encode() + left = [n] + lock = threading.Lock() + + def work(): + connection = http.client.HTTPConnection('127.0.0.1', port, timeout=60) + while True: + with lock: + if not left[0]: + break + left[0] -= 1 + connection.request('POST', '/inputs/buy', body, {'Content-Type': 'application/json'}) + response = connection.getresponse() + response.read() + assert response.status == 200, response.status + connection.close() + workers = [threading.Thread(target=work) for _ in range(threads)] + for w in workers: + w.start() + for w in workers: + w.join() + + +def drain_pure(work, n, size): + """M5 for Pure: n saved sends of an order whose item is size bytes wait + in the sender's outbox (the receiver is not there); then the receiver + starts, and the link drains them. Rate from the first applied to the + last; CPU of both nodes, and of their link threads alone.""" + sender_module, receiver_module = work / 'drain-sender.pmod', work / 'drain-receiver.pmod' + for module, text in ((sender_module, check.LINK_SENDER), (receiver_module, check.LINK_RECEIVER)): + source = module.with_suffix('.aup') + source.write_text(text) + check.compile_module(Path(os.environ.get('AUP', ROOT.parent / 'aup/build/aup')), source, module) + keys = {} + for name in ('a', 'b'): + node, info = start([ROOT / 'build/pure-serve', '--module', receiver_module, '--store', work / f'key-{name}', + '--port', free_port(socket.SOCK_STREAM), '--key', work / f'{name}.key']) + keys[name] = info['public_key'] + node.terminate() + node.communicate(timeout=60) + (work / 'drain-senders').write_text(f'drain-sender.ordered {keys["a"]}\n') + link_port, port_a, port_b = free_port(socket.SOCK_DGRAM), free_port(socket.SOCK_STREAM), free_port(socket.SOCK_STREAM) + a, _ = start([ROOT / 'build/pure-serve', '--module', sender_module, '--store', work / 'drain-a', '--port', port_a, + '--threads', '2', '--key', work / 'a.key', '--send', f'ordered=pure://127.0.0.1:{link_port}/order', + '--seal', f'ordered={keys["b"]}']) + post_all(port_a, n, size) + assert view(port_a, 'sent') == n + b, _ = start([ROOT / 'build/pure-serve', '--module', receiver_module, '--store', work / 'drain-b', '--port', port_b, + '--threads', '2', '--key', work / 'b.key', '--link-port', link_port, '--senders', work / 'drain-senders']) + cpu = {name: (thread_cpu(pid, None), thread_cpu(pid, thread)) for name, pid, thread in + (('sender', a.pid, 'pure-link'), ('receiver', b.pid, 'pure-listen'))} + import http.client + connection = http.client.HTTPConnection('127.0.0.1', port_b, timeout=30) + first = None + while True: # the count every 2 ms, over one connection + connection.request('GET', '/views/count') + count = json.loads(connection.getresponse().read()) + now = time.monotonic() + if count and first is None: + first = (now, count) + if count >= n: + break + time.sleep(0.002) + connection.close() + seconds = now - first[0] + row = {'messages': n, 'item_bytes': size, 'first_seen': first[1], 'seconds': round(seconds, 4), + 'per_second': round((n - first[1]) / seconds) if seconds else None} + for name, pid, thread in (('sender', a.pid, 'pure-link'), ('receiver', b.pid, 'pure-listen')): + total, link = thread_cpu(pid, None) - cpu[name][0], thread_cpu(pid, thread) - cpu[name][1] + row[f'{name}_cpu_us_per_message'] = round(total * 1000 / n, 2) + row[f'{name}_link_cpu_us_per_message'] = round(link * 1000 / n, 2) + for name, node in (('sender', a), ('receiver', b)): + node.terminate() + _, error = node.communicate(timeout=120) + stats = [json.loads(line) for line in error.decode().splitlines() if line.startswith('{"active"')][-1] + keys = ('link_packets', 'link_records', 'link_resent', 'link_lost') if name == 'sender' else \ + ('listener_received', 'listener_deliveries', 'bursts') + row[name] = {k: stats[k] for k in keys} + subprocess.run(['rm', '-rf', str(work / 'drain-a'), str(work / 'drain-b')], check=True) # tmpfs: room for the next + return row + + +def drain_quic(work, n, size): + """The same over quinn: one connection, one stream, n messages; CPU of + both sides. The service only counts them (no store).""" + binary = Path(__file__).parent / 'quic/target/release/idle-quic' + import resource + server, info = start([binary, 'serve', work / 'cert.der', work / 'key.der']) + before, children = thread_cpu(server.pid, None), resource.getrusage(resource.RUSAGE_CHILDREN) + # The same bytes as Pure's record: its prefix (count, item's length), the item and the city ("Seoul"). + client = subprocess.run([str(binary), 'drain', str(info['port']), str(n), str(size + 8 + 5)], capture_output=True, + timeout=600) + after = resource.getrusage(resource.RUSAGE_CHILDREN) + row = json.loads(client.stdout) + server_cpu = thread_cpu(server.pid, None) - before + client_cpu = (after.ru_utime + after.ru_stime - children.ru_utime - children.ru_stime) * 1000 + row['client_cpu_us_per_message'] = round(client_cpu * 1000 / n, 2) + server.kill() + server.wait() + row['service_cpu_us_per_message'] = round(server_cpu * 1000 / n, 2) + return row + + +def transport(work, size, n): + """The transport alone (transport.c): a backlog of n records of an order + with a size-byte item drained through a link and a listener in one + process, no app and no store.""" + modules = {} + for name, text in (('sender', check.LINK_SENDER), ('receiver', check.LINK_RECEIVER)): + source = work / f'{name}.aup' + source.write_text(text) + modules[name] = work / f'{name}.pmod' + check.compile_module(Path(os.environ.get('AUP', ROOT.parent / 'aup/build/aup')), source, modules[name]) + done = subprocess.run([str(ROOT / 'build/link-transport'), str(modules['sender']), str(modules['receiver']), str(n), + str(size), '0'], capture_output=True, text=True, timeout=600) + assert done.returncode == 0, done.stderr + return json.loads(done.stdout) + + def main(): parser = argparse.ArgumentParser() parser.add_argument('--counts', default='1000,10000,100000,1000000') @@ -399,8 +544,54 @@ def main(): parser.add_argument('--kill-at', type=int, default=20) parser.add_argument('--first-contact', type=str, default=None, help='M4: one-way delays in ms (for example 5,25,50), each in a namespace of its own') + parser.add_argument('--drain', type=str, default=None, + help='M5: item sizes in bytes (for example 48,1000,16000); --counts gives the messages') + parser.add_argument('--transport', type=str, default=None, + help='M5: SIZE:COUNT pairs (for example 48:200000,1000:50000): the transport alone and quinn') + parser.add_argument('--rounds', type=int, default=1) args = parser.parse_args() counts = [int(c) for c in args.counts.split(',')] + if args.transport: + subprocess.run(['make', '-s', '-C', str(ROOT), 'build/libpure.a', 'build/pure-crypto.so'], check=True) + subprocess.run(['cc', '-O2', '-g', '-std=c17', '-Wall', '-Wextra', '-Werror', f'-I{ROOT}/core', f'-I{ROOT}/serve', + '-DPURE_ASYNC_EFFECTS', str(Path(__file__).parent / 'transport.c'), str(ROOT / 'serve/link.c'), + str(ROOT / 'serve/crypto.c'), str(ROOT / 'build/libpure.a'), '-lm', '-pthread', '-ldl', '-o', + str(ROOT / 'build/link-transport')], check=True) + subprocess.run(['cargo', 'build', '--release', '--offline', '--quiet'], cwd=Path(__file__).parent / 'quic', check=True) + with tempfile.TemporaryDirectory(prefix='pure-m5-', dir=args.store) as directory: + work = Path(directory) + make_certificate(work) + for pair in args.transport.split(','): + size, n = (int(x) for x in pair.split(':')) + for _ in range(args.rounds): + for name, measure in (('pure', lambda: transport(work, size, n)), ('quinn', lambda: drain_quic(work, n, size))): + row = {'side': name, 'item_bytes': size, **measure()} + line = json.dumps(row) + print(line, flush=True) + if args.out: + with open(args.out, 'a') as out: + out.write(line + '\n') + return + if args.drain: + subprocess.run(['cargo', 'build', '--release', '--offline', '--quiet'], cwd=Path(__file__).parent / 'quic', check=True) + with tempfile.TemporaryDirectory(prefix='pure-m5-', dir=args.store) as directory: + work = Path(directory) + make_certificate(work) + for size in (int(s) for s in args.drain.split(',')): + for round_, n in enumerate(counts): + for name, measure in (('pure', drain_pure), ('quinn', drain_quic)): + case = work / f'{name}-{size}-{n}-{round_}' + case.mkdir() + if name == 'quinn': + for f in ('cert.der', 'key.der'): + (case / f).write_bytes((work / f).read_bytes()) + row = {'side': name, **measure(case, n, size)} + line = json.dumps(row) + print(line, flush=True) + if args.out: + with open(args.out, 'a') as out: + out.write(line + '\n') + return if args.first_contact: subprocess.run(['cargo', 'build', '--release', '--offline', '--quiet'], cwd=Path(__file__).parent / 'quic', check=True) for delay in args.first_contact.split(','): -- cgit v1.2.3