"""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)