"""Repeated same-app HTTP/1 benchmarks, pinned to separate server/client cores.
Raw results are retained; failures are reported, never discarded.
"""
import argparse
from concurrent.futures import ThreadPoolExecutor
import csv
import datetime
import http.client
import hashlib
from importlib import metadata
import json
import os
from pathlib import Path
import random
import re
import resource
import signal
import shutil
import socket
import statistics
import subprocess
import sys
import time
from zoneinfo import ZoneInfo
import psutil
BENCH = Path(__file__).resolve().parent
REPO = BENCH.parent
WORKLOADS = {
'raw_get': ('raw', 'GET', '/tiny', None, b'Hello world!'),
'fastapi_json': ('fastapi', 'GET', '/json', None, b'{"hello":"world"}'),
'post_4k': ('raw', 'POST', '/echo', b'x' * 4096, b'x' * 4096),
'get_64k': ('raw', 'GET', '/bytes', None, b'x' * 65536),
}
SERVERS = ('skirr', 'weft', 'peregrine', 'skirr-before')
def power_profile():
try:
return subprocess.check_output(['powerprofilesctl', 'get'], text=True).strip()
except (FileNotFoundError, subprocess.CalledProcessError):
return None
def package_version(package):
try:
return metadata.version(package)
except metadata.PackageNotFoundError:
return None
def seconds(value):
match = re.fullmatch(r'([\d.]+)(us|ms|s|m)', value)
if not match:
raise ValueError('Unrecognized latency: ' + value)
return float(match[1]) * {'us': 0.001, 'ms': 1, 's': 1000, 'm': 60000}[match[2]]
def parse_wrk(text):
rps = re.search(r'Requests/sec:\s+([\d.]+)', text)
latency = re.search(r'^\s+Latency\s+(\S+)\s+(\S+)\s+(\S+)', text, re.M)
p99 = re.search(r'^\s+99(?:\.0+)?%\s+(\S+)', text, re.M)
count = re.search(r'(\d+) requests in ([\d.]+)s', text)
sockets = re.search(r'Socket errors: connect (\d+), read (\d+), write (\d+), timeout (\d+)', text)
non2xx = re.search(r'Non-2xx or 3xx responses:\s+(\d+)', text)
if not all((rps, latency, p99, count)):
raise RuntimeError('Incomplete wrk output:\n' + text)
return {
'requests_per_second': float(rps[1]), 'requests': int(count[1]),
'wrk_seconds': float(count[2]), 'mean_latency_ms': seconds(latency[1]),
'p99_latency_ms': seconds(p99[1]),
'socket_errors': sum(map(int, sockets.groups())) if sockets else 0,
'socket_error_detail': dict(zip(('connect', 'read', 'write', 'timeout'), map(int, sockets.groups()))) if sockets else {},
'non_2xx_3xx': int(non2xx[1]) if non2xx else 0,
}
def process_tree(root):
try:
return [root] + root.children(recursive=True)
except psutil.NoSuchProcess:
return []
def sample_tree(root):
cpu, rss, pss, pids = {}, 0, 0, []
for process in process_tree(root):
try:
times = process.cpu_times()
cpu[process.pid] = times.user + times.system
memory = process.memory_full_info()
rss += memory.rss
pss += getattr(memory, 'pss', memory.rss)
pids.append(process.pid)
except (psutil.NoSuchProcess, psutil.AccessDenied, ProcessLookupError):
pass
return cpu, rss, pss, pids
class Server:
def __init__(self, name, app, workers, args, label):
self.args, self.name, self.workers = args, name, workers
self.ready_pids = set()
with socket.socket() as sock:
sock.bind(('127.0.0.1', 0))
self.port = sock.getsockname()[1]
source = args.baseline / 'python' if name == 'skirr-before' else REPO / 'python'
env = dict(os.environ)
paths = [BENCH, source, *args.dependency_paths]
if env.get('PYTHONPATH'):
paths.append(env['PYTHONPATH'])
env['PYTHONPATH'] = os.pathsep.join(map(str, paths))
cpus = args.server_cpus[:workers]
self.command = [sys.executable, str(BENCH / 'server.py'), name,
'--app', app, '--lifespan', args.lifespan, '--workers', str(workers), '--port', str(self.port),
'--cpus', ','.join(map(str, cpus))]
self.log_path = args.output / 'logs' / (label + '.server.log')
self.log = self.log_path.open('w')
self.process = subprocess.Popen(self.command, cwd=BENCH, env=env,
stdout=self.log, stderr=subprocess.STDOUT,
start_new_session=True)
self.root = psutil.Process(self.process.pid)
def ready(self, workload):
deadline = time.monotonic() + 20
def get_pid(_):
client = http.client.HTTPConnection('127.0.0.1', self.port, timeout=0.5)
try:
client.request('GET', '/_benchmark_pid')
response = client.getresponse()
body = response.read()
return int(body) if response.status == 200 else None
except (OSError, http.client.HTTPException, ValueError):
return None
finally:
client.close()
with ThreadPoolExecutor(max_workers=self.workers * 4) as pool:
while time.monotonic() < deadline:
if self.process.poll() is not None:
raise RuntimeError('Server exited: ' + self.log_path.read_text())
self.ready_pids.update(pid for pid in pool.map(get_pid, range(self.workers * 8)) if pid)
if len(self.ready_pids) == self.workers:
for _ in range(5):
self.validate(workload)
self.affinities()
return
time.sleep(0.05)
raise TimeoutError('Not all workers became ready: {} {}'.format(self.ready_pids, self.log_path))
def affinities(self):
actual = {str(pid): psutil.Process(pid).cpu_affinity() for pid in sorted(self.ready_pids)}
allowed = set(self.args.server_cpus[:self.workers])
if any(not cpus or not set(cpus) <= allowed for cpus in actual.values()):
raise RuntimeError(f"Worker CPU affinity differs from configured server CPUs: {actual}")
return actual
def validate(self, workload):
_, method, path, body, expected = WORKLOADS[workload]
client = http.client.HTTPConnection('127.0.0.1', self.port, timeout=1)
try:
client.request(method, path, body=body)
response = client.getresponse()
actual = response.read()
if response.status != 200 or actual != expected:
raise AssertionError((self.name, workload, response.status, len(actual), actual[:100]))
finally:
client.close()
def stop(self):
if self.process.poll() is None:
self.process.send_signal(signal.SIGTERM)
try:
self.process.wait(timeout=15)
except subprocess.TimeoutExpired:
os.killpg(self.process.pid, signal.SIGKILL)
self.process.wait()
self.log.close()
# Includes process-manager helper children that might otherwise linger.
try:
os.killpg(self.process.pid, signal.SIGTERM)
except ProcessLookupError:
pass
def wrk_command(args, server, workload, duration):
command = ['taskset', '-c', ','.join(map(str, args.client_cpus)), args.wrk,
'-t', str(args.threads), '-c', str(args.connections),
'-d', str(duration) + 's', '--latency', '--timeout', '3s']
if args.rate is not None:
command += ['-R', str(args.rate)]
if workload == 'post_4k':
command += ['-s', str(BENCH / 'post_4k.lua')]
return command + ['http://127.0.0.1:{}{}'.format(server.port, WORKLOADS[workload][2])]
def run_load(args, server, workload, repeat, ordinal):
server.validate(workload)
warmup = subprocess.run(wrk_command(args, server, workload, args.warmup),
stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, timeout=args.warmup + 10)
if warmup.returncode:
raise RuntimeError(warmup.stdout)
before_cpu, _, _, _ = sample_tree(server.root)
before_times = {}
for member in process_tree(server.root):
try:
before_times[member.pid] = member.cpu_times()
except (psutil.NoSuchProcess, psutil.AccessDenied):
pass
child_cpu = resource.getrusage(resource.RUSAGE_CHILDREN)
before_child = child_cpu.ru_utime + child_cpu.ru_stime
started = time.monotonic()
process = subprocess.Popen(wrk_command(args, server, workload, args.duration),
stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True)
samples = []
frequencies = []
psutil.cpu_percent(percpu=True)
while process.poll() is None:
_, rss, pss, pids = sample_tree(server.root)
samples.append((rss, pss, psutil.cpu_percent(percpu=True)))
frequencies.append({str(cpu): int(Path(f'/sys/devices/system/cpu/cpu{cpu}/cpufreq/scaling_cur_freq').read_text())
for cpu in args.server_cpus[:server.workers]})
if server.process.poll() is not None:
process.terminate()
raise RuntimeError('Server crashed: ' + str(server.log_path))
if time.monotonic() - started > args.duration + 12:
process.kill()
raise TimeoutError('Load generator did not complete')
time.sleep(0.2)
output, _ = process.communicate(timeout=2)
elapsed = time.monotonic() - started
if process.returncode:
raise RuntimeError(output)
after_cpu, _, _, pids = sample_tree(server.root)
child_cpu = resource.getrusage(resource.RUSAGE_CHILDREN)
client_cpu = child_cpu.ru_utime + child_cpu.ru_stime - before_child
cpu_used = sum(max(0, value - before_cpu.get(pid, 0)) for pid, value in after_cpu.items())
cpu_parts = {'user': 0.0, 'system': 0.0}
for member in process_tree(server.root):
try:
current = member.cpu_times()
previous = before_times.get(member.pid)
for part in cpu_parts:
cpu_parts[part] += max(0, getattr(current, part) - (getattr(previous, part) if previous else 0))
except (psutil.NoSuchProcess, psutil.AccessDenied):
pass
row = dict(server=server.name, workers=server.workers,
workload=workload, repeat=repeat, ordinal=ordinal,
timestamp=datetime.datetime.now(ZoneInfo('America/New_York')).isoformat(),
**parse_wrk(output), server_cpu_percent=100 * cpu_used / elapsed,
server_cpu_seconds=cpu_parts,
client_cpu_percent=100 * client_cpu / elapsed,
peak_tree_rss_mib=max(s[0] for s in samples) / 2**20,
peak_tree_pss_mib=max(s[1] for s in samples) / 2**20,
mean_system_cpu_percent=[statistics.mean(s[2][i] for s in samples) for i in range(psutil.cpu_count())],
server_frequency_khz=frequencies,
server_pids=pids, verified_worker_pids=sorted(server.ready_pids),
verified_worker_affinities=server.affinities())
path = args.output / 'logs' / ('{:03d}-{}-w{}-{}-r{}.wrk.txt'.format(ordinal, row['server'], row['workers'], workload, repeat))
path.write_text(output)
row['raw_log'] = str(path.relative_to(args.output))
server.validate(workload)
print('{ordinal:03d} {server:8s} w{workers} {workload:13s} r{repeat}: {requests_per_second:9,.0f} req/s; p99 {p99_latency_ms:.2f} ms; errors {socket_errors}/{non_2xx_3xx}; CPU srv/client {server_cpu_percent:.0f}/{client_cpu_percent:.0f}%'.format(**row), flush=True)
return row
def measure_stream(server, count=20):
first, total = [], []
for _ in range(count):
client = http.client.HTTPConnection('127.0.0.1', server.port, timeout=2)
try:
started = time.perf_counter()
client.request('GET', '/stream')
response = client.getresponse()
prefix = response.read(1)
first.append((time.perf_counter() - started) * 1000)
body = prefix + response.read()
total.append((time.perf_counter() - started) * 1000)
assert response.status == 200 and body == b'first\nlast\n', (server.name, body)
finally:
client.close()
return dict(server=server.name, first_body_byte_ms=first, total_ms=total,
median_first_body_byte_ms=statistics.median(first), median_total_ms=statistics.median(total))
def main():
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument('--servers', nargs='+', choices=SERVERS, default=list(SERVERS[:3]))
parser.add_argument('--workloads', nargs='+', choices=list(WORKLOADS), default=list(WORKLOADS))
parser.add_argument('--baseline', type=Path, help='Preserved source/binary directory for skirr-before')
parser.add_argument('--output', type=Path, required=True)
parser.add_argument('--dependency-paths', type=Path, nargs='*', default=[])
parser.add_argument('--server-cpus', type=lambda s: list(map(int, s.split(','))), default=[0, 1, 2, 3])
parser.add_argument('--client-cpus', type=lambda s: list(map(int, s.split(','))), default=[4, 5, 6, 7])
parser.add_argument('--workers', type=lambda s: list(map(int, s.split(','))), default=[1])
parser.add_argument('--wrk', default='wrk', help='wrk-compatible load-generator binary')
parser.add_argument('--rate', type=int, help='Total fixed rate; requires a wrk2-compatible generator')
parser.add_argument('--duration', type=int, default=10)
parser.add_argument('--warmup', type=int, default=1)
parser.add_argument('--repeats', type=int, default=3)
parser.add_argument('--threads', type=int, default=4)
parser.add_argument('--lifespan', choices=['off', 'on'], default='off')
parser.add_argument('--connections', type=int, default=256)
parser.add_argument('--seed', type=int, default=20261108)
parser.add_argument('--smoke', action='store_true')
args = parser.parse_args()
if 'skirr-before' in args.servers and args.baseline is None:
parser.error('--baseline is required for skirr-before')
args.output = args.output.resolve()
args.output.mkdir(parents=True, exist_ok=True)
(args.output / 'logs').mkdir(exist_ok=True)
if set(args.server_cpus) & set(args.client_cpus):
parser.error('Client and server CPU sets must not overlap')
if max(args.workers) > len(args.server_cpus):
parser.error('Each worker needs a dedicated allowed server CPU')
if args.smoke:
args.duration, args.warmup, args.repeats = 2, 1, 1
os.sched_setaffinity(0, set(args.client_cpus))
if shutil.which(args.wrk) is None:
parser.error('Load generator not found: ' + args.wrk)
versions = {p: package_version(p) for p in ('weft-server', 'peregrine-server', 'uvloop', 'fastapi')}
metadata_out = dict(python=sys.version, versions=versions,
load_generator=dict(path=shutil.which(args.wrk), sha256=hashlib.sha256(Path(shutil.which(args.wrk)).read_bytes()).hexdigest()),
build='cargo build --release --features extension-module; native Linux epoll',
source_sha256={str(p.relative_to(REPO)): hashlib.sha256(p.read_bytes()).hexdigest() for p in [REPO/'Cargo.toml', REPO/'Cargo.lock', REPO/'pyproject.toml', *sorted((REPO/'src').glob('*.rs')), REPO/'python/skirr/__init__.py', REPO/'python/skirr/_native.so']},
runtime_policy='Server allocator/GC defaults retained; Skirr does not change application GC policy.',
baseline_sha256={str(p.relative_to(args.baseline)): hashlib.sha256(p.read_bytes()).hexdigest() for p in [args.baseline/'Cargo.toml', args.baseline/'Cargo.lock', *sorted((args.baseline/'src').glob('*.rs')), args.baseline/'python/skirr/__init__.py', args.baseline/'python/skirr/_native.so']} if args.baseline else None,
configuration={k: str(v) if isinstance(v, Path) else [str(x) for x in v] if k == 'dependency_paths' else v for k, v in vars(args).items()},
cpu_info=subprocess.check_output(['lscpu'], text=True),
governor=Path('/sys/devices/system/cpu/cpu0/cpufreq/scaling_governor').read_text().strip(),
energy_preference=Path('/sys/devices/system/cpu/cpu0/cpufreq/energy_performance_preference').read_text().strip(),
power_profile=power_profile(),
initial_load_average=os.getloadavg(), mode='smoke' if args.smoke else 'measured')
result = dict(metadata=metadata_out, runs=[], streaming=[])
def save():
(args.output / 'results.json').write_text(json.dumps(result, indent=2))
save()
ordinal = 0
rng = random.Random(args.seed)
for repeat in range(1, args.repeats + 1):
groups = [(name, workers, app) for name in args.servers for workers in args.workers for app in sorted({WORKLOADS[w][0] for w in args.workloads})]
rng.shuffle(groups)
for name, workers, app in groups:
workloads = [name for name, definition in WORKLOADS.items() if definition[0] == app and name in args.workloads]
rng.shuffle(workloads)
label = '{}-w{}-{}-r{}'.format(name, workers, app, repeat)
server = Server(name, app, workers, args, label)
try:
server.ready(workloads[0])
if app == 'raw' and repeat == 1:
timing = measure_stream(server)
timing['workers'] = workers
result['streaming'].append(timing)
save()
for workload in workloads:
ordinal += 1
result['runs'].append(run_load(args, server, workload, repeat, ordinal))
save()
finally:
server.stop()
fields = ['server', 'workers', 'workload', 'repeat', 'requests_per_second', 'mean_latency_ms', 'p99_latency_ms',
'socket_errors', 'non_2xx_3xx', 'server_cpu_percent', 'client_cpu_percent', 'peak_tree_rss_mib', 'peak_tree_pss_mib']
with (args.output / 'runs.csv').open('w', newline='') as output:
writer = csv.DictWriter(output, fields, extrasaction='ignore')
writer.writeheader()
writer.writerows(result['runs'])
print('Saved {} runs to {}'.format(len(result['runs']), args.output), flush=True)
if __name__ == '__main__':
main()