"""Native/Python sampling under HTTP load; diagnostic throughput is not a benchmark.""" import argparse import hashlib import http.client import json import os from pathlib import Path import signal import socket import subprocess import sys import time root = Path(__file__).resolve().parents[1] p = argparse.ArgumentParser(description=__doc__) p.add_argument('--profiler', required=True) p.add_argument('--kind', choices=['py-spy', 'perf'], default='py-spy') p.add_argument('--python-only', action='store_true') p.add_argument('--baseline', type=Path) p.add_argument('--server', choices=['skirr', 'peregrine', 'weft'], default='skirr') p.add_argument('--app', choices=['raw', 'fastapi'], default='raw') p.add_argument('--path', default='/tiny') p.add_argument('--python-perf', action='store_true', help='CPython DWARF trampolines; diagnostic overhead') p.add_argument('--connections', type=int, default=64) p.add_argument('--duration', type=int, default=11) p.add_argument('--wrk', default='wrk') p.add_argument('--rate', type=int) p.add_argument('--output', required=True, type=Path) a = p.parse_args() a.output.mkdir(parents=True, exist_ok=True) with socket.socket() as s: s.bind(('127.0.0.1', 0)) port = s.getsockname()[1] source = a.baseline/'python' if a.baseline else root/'python' env = dict(os.environ, PYTHONPATH=str(source) + os.pathsep + str(root/'bench')) if a.python_perf: env['PYTHON_PERF_JIT_SUPPORT'] = '1' server = [sys.executable, str(root/'bench/server.py'), a.server, '--app', a.app, '--port', str(port), '--workers', '1', '--cpus', '0'] command = [a.profiler, 'record', '--subprocesses', '--native', '--rate', '500', '--duration', str(a.duration + 3), '--format', 'raw', '--output', str(a.output/'stacks.txt'), '--', *server] if a.python_only: command.remove('--native') if a.kind == 'perf': data = Path('/tmp') / (a.output.name + '.perf.data') data.unlink(missing_ok=True) command = [a.profiler, 'record', '-e', 'cycles:u', '-F', '499', '--call-graph', 'dwarf,8192', '--delay', '3000', '-o', str(data), '--', *server] if a.kind == 'perf' and a.python_perf: command[2:2] = ['-k', '1'] os.sched_setaffinity(0, {7}) with (a.output/'profiler.log').open('w') as log: proc = subprocess.Popen(command, env=env, stdout=log, stderr=subprocess.STDOUT, start_new_session=True) try: deadline = time.monotonic() + 10 while True: if proc.poll() is not None: raise RuntimeError((a.output/'profiler.log').read_text()) client = http.client.HTTPConnection('127.0.0.1', port, timeout=.2) try: client.request('GET', '/_benchmark_pid') r = client.getresponse() if r.status == 200 and (pid := int(r.read())): worker_affinity = sorted(os.sched_getaffinity(pid)) if worker_affinity != [0]: raise RuntimeError(f'Worker affinity differs from CPU 0: {worker_affinity}') (a.output/'maps.txt').write_text(Path(f'/proc/{pid}/maps').read_text()) break except (OSError, http.client.HTTPException): if time.monotonic() > deadline: raise time.sleep(.03) finally: client.close() client = http.client.HTTPConnection('127.0.0.1', port, timeout=2) try: payload = b'x' * 4096 if a.path == '/echo' else None client.request('POST' if payload is not None else 'GET', a.path, body=payload) response = client.getresponse() body = response.read() expected = {'/json': b'{"hello":"world"}', '/tiny': b'Hello world!', '/echo': b'x' * 4096, '/bytes': b'x' * 65536}[a.path] if response.status != 200 or body != expected: raise RuntimeError(f'Invalid profiling workload: status {response.status}, body {body[:80]!r}') (a.output/'validation.json').write_text(json.dumps({'status': 200, 'body_bytes': len(body), 'app': a.app, 'path': a.path})) finally: client.close() load_command = ['taskset', '-c', '4,5,6,7', a.wrk, '-t4', f'-c{a.connections}', f'-d{a.duration}s', '--latency'] if a.rate is not None: load_command += ['-R', str(a.rate)] if a.path == '/echo': load_command += ['-s', str(root/'bench/post_4k.lua')] load_command += [f'http://127.0.0.1:{port}{a.path}'] load = subprocess.run(load_command, capture_output=True, text=True, timeout=a.duration + 10, check=True) (a.output/'diagnostic-load.txt').write_text(load.stdout) if 'Non-2xx or 3xx' in load.stdout or 'Socket errors:' in load.stdout: raise RuntimeError('Invalid diagnostic load: ' + load.stdout) if a.kind == 'perf': os.killpg(proc.pid, signal.SIGINT) proc.wait(timeout=20) if proc.returncode and not (a.kind == 'perf' and proc.returncode in (130, -signal.SIGINT) and data.exists()): raise RuntimeError((a.output/'profiler.log').read_text()) finally: try: os.killpg(proc.pid, signal.SIGTERM) except ProcessLookupError: pass proc.wait(timeout=5) (a.output/'command.json').write_text(json.dumps(command, indent=2)) if a.server == 'skirr': library = source/'skirr/_native.so' else: mappings = (a.output/'maps.txt').read_text().splitlines() module = '_native.' if a.server == 'peregrine' else '_weft.' library = Path(next(line.split()[-1] for line in mappings if f'/{a.server}/{module}' in line and '.so' in line)) try: power_profile = subprocess.check_output(['powerprofilesctl', 'get'], text=True).strip() except (OSError, subprocess.CalledProcessError): power_profile = None (a.output/'metadata.json').write_text(json.dumps({ 'server': a.server, 'power_profile': power_profile, 'workload': dict(app=a.app, path=a.path, connections=a.connections, threads=4, rate=a.rate, duration=a.duration, load_generator=a.wrk, worker_pid=pid), 'verified_worker_affinity': worker_affinity, 'python_only': a.python_only, 'python_perf_trampolines': a.python_perf, 'sample_subprocesses': a.kind == 'py-spy', 'library': str(library), 'binary_sha256': hashlib.sha256(library.read_bytes()).hexdigest(), 'python_source_sha256': hashlib.sha256((source/'skirr/__init__.py').read_bytes()).hexdigest() if a.server == 'skirr' else None, 'source_sha256': {p.name: hashlib.sha256(p.read_bytes()).hexdigest() for p in (a.baseline or root).joinpath('src').glob('*.rs')} if a.server == 'skirr' else {}, 'startup_sampling_delay_ms': 3000 if a.kind == 'perf' else 0, 'diagnostic_only': True, }, indent=2)) if a.kind == 'perf': if a.python_perf: injected = data.with_suffix('.jit.data') subprocess.run([a.profiler, 'inject', '--jit', '-i', str(data), '-o', str(injected)], check=True) data = injected for label, extra in [('exclusive', ['--no-children']), ('inclusive', ['--children']), ('flat', ['--no-children', '-g', 'none'])]: report = subprocess.run([a.profiler, 'report', '--stdio', '--percent-limit', '0.5', '-i', str(data), *extra, '--sort', 'symbol,dso'], capture_output=True, text=True, check=True) (a.output/(label+'.txt')).write_text(report.stdout) if a.python_perf: from collections import Counter import re stacks = subprocess.run([a.profiler, 'script', '-i', str(data), '-F', 'comm,pid,tid,time,period,event,ip,sym,dso'], capture_output=True, text=True, check=True) counts = Counter() total = 0 for stack in stacks.stdout.split('\n\n'): lines = stack.splitlines() if not lines: continue sample = re.search(r':\s+(\d+)\s+\S+:', lines[0]) if not sample: continue weight = int(sample[1]) total += weight frames = [line.strip().split('py::', 1)[1].rsplit(' (', 1)[0] for line in lines[1:] if 'py::' in line] counts[frames[0] if frames else '[no Python trampoline frame]'] += weight (a.output/'python-attribution.json').write_text(json.dumps({ 'total_sample_period': total, 'nearest_python_frame_percent': {key: round(value/total*100, 2) for key, value in counts.most_common(40)}, 'limitations': 'Nearest Python caller includes its native/C-API callees; normalized diagnostic samples.', }, indent=2)) (a.output/'script-diagnostics.txt').write_text(stacks.stderr) print(a.output)