bughz/3proxy
1
0
Fork 0
mirror of https://github.com/3proxy/3proxy.git synced 2026-10-15 16:15:49 +08:00
Code Issues Actions 12 Packages Projects Releases Wiki Activity
a146570533
3proxy/bench/bench.py
Vladimir Dubrovin a146570533 Benchmarks: splice by direction, grace delay, tunnels, more proxies 
bench.py:
- 2000 clients posting 2 KB and getting 50 KB back, next to the GET one
- haproxy with splice-response and splice-request, like -s1 and -s2
- CONNECT and SOCKSv5 scenarios: wrk and iperf3 run under proxychains-ng,
  so every connection asks the proxy for a tunnel; and the same loads
  through a TCP relay to a fixed destination
- tinyproxy and squid (HTTP proxy, CONNECT), dante (SOCKS); each proxy
  takes part only in the scenarios it supports
- 3proxy also gets its socks service and a tcppm to the web server

report.py: one throughput table for several machines (--machine), each
cell their medians and their average; bold for the best of a group, * for
the best of the row; --split puts the splice proxies in a table of their
own, --prefix picks scenarios.

RESULTS.md: the comparison as one table for both machines, splice by
direction (devel -s0/-s1/-s3 against haproxy's settings) and grace delay
with splice, with their data.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-12 00:40:19 +03:00

877 lines
33 KiB
Python
Raw Blame History

#!/usr/bin/env python3
"""Compares the speed and the cost of proxies on Linux: 3proxy versions, and
other servers doing the same job.
The client, the proxy and the origin each run in a network namespace of their
own, linked by veth pairs, and in a cgroup of their own pinned to separate
CPUs, so they neither share a loopback nor take CPU from each other, and the
proxy's CPU time and memory can be read on their own. See README.md.
Needs root, iproute2, iperf3, wrk and nginx (the origin); haproxy only when
it is one of the proxies compared.
sudo bench/bench.py --proxy 3proxy:bench/bin/3proxy-1.0.1 \\
--proxy 3proxy:bench/bin/3proxy-devel --proxy nginx --proxy none
"""
import argparse
import csv
import datetime
import json
import os
import re
import resource
import shutil
import signal
import statistics
import subprocess
import sys
import tempfile
import threading
import time
CG = "/sys/fs/cgroup/3proxy-bench"
# client 10.0.1.2 -- 10.0.1.1 proxy 10.0.2.1 -- 10.0.2.2 origin
CLI_ADDR, PRX_IN, PRX_OUT, SRV_ADDR = "10.0.1.2", "10.0.1.1", "10.0.2.1", "10.0.2.2"
HTTP_PORT, TCP_PORT, ORIGIN_HTTP, ORIGIN_TCP = 3128, 5201, 8080, 5201
# a TCP relay to the origin's web server, standing in for CONNECT where the
# proxy has none (nginx, haproxy)
RELAY_HTTP = 5280
SOCKS_PORT = 1080
FILES = {"1k": 1024, "50k": 50 * 1024, "100k": 100 * 1024, "1m": 1024 * 1024}
# name: (kind, parameters)
SCENARIOS = {
"stream1": ("stream", {"streams": 1}),
"stream16": ("stream", {"streams": 16}),
"stream1-down": ("stream", {"streams": 1, "reverse": True}),
"http-ka-1k": ("http", {"file": "1k", "conns": 64}),
"http-ka-1m": ("http", {"file": "1m", "conns": 16}),
"http-ka-50k-2000": ("http", {"file": "50k", "conns": 2000}),
"http-ka-2k-50k-2000": ("http", {"file": "50k", "conns": 2000, "body": 2048}),
"http-close-1k": ("http", {"file": "1k", "conns": 64, "close": True}),
# Through tunnels: the load generators run under proxychains, so each of
# their connections asks the proxy for one to the origin, with CONNECT
# (HTTP proxies) or SOCKSv5. And the same load through a TCP relay to a
# fixed destination (3proxy tcppm, nginx stream, haproxy mode tcp).
"connect-stream1-down": ("stream", {"streams": 1, "reverse": True, "via": "connect"}),
"connect-stream16": ("stream", {"streams": 16, "via": "connect"}),
"connect-ka-1k": ("http", {"file": "1k", "conns": 64, "via": "connect"}),
"connect-ka-50k-2000": ("http", {"file": "50k", "conns": 2000, "via": "connect"}),
"connect-close-1k": ("http", {"file": "1k", "conns": 64, "close": True, "via": "connect"}),
"socks-stream1-down": ("stream", {"streams": 1, "reverse": True, "via": "socks"}),
"socks-stream16": ("stream", {"streams": 16, "via": "socks"}),
"socks-ka-1k": ("http", {"file": "1k", "conns": 64, "via": "socks"}),
"socks-ka-50k-2000": ("http", {"file": "50k", "conns": 2000, "via": "socks"}),
"socks-close-1k": ("http", {"file": "1k", "conns": 64, "close": True, "via": "socks"}),
"relay-stream1-down": ("stream", {"streams": 1, "reverse": True, "via": "relay"}),
"relay-stream16": ("stream", {"streams": 16, "via": "relay"}),
"relay-ka-1k": ("http", {"file": "1k", "conns": 64, "via": "relay"}),
"relay-ka-50k-2000": ("http", {"file": "50k", "conns": 2000, "via": "relay"}),
"relay-close-1k": ("http", {"file": "1k", "conns": 64, "close": True, "via": "relay"}),
}
TOOLS = {}
def tool(*names):
"""A tool by its absolute path: sudo's secure_path may leave out where
it was installed, /usr/local/bin for a wrk built by hand. Several names
are alternatives."""
name = names[0]
if name not in TOOLS:
for n in names:
path = shutil.which(n) or shutil.which(
n, path="/usr/local/bin:/usr/local/sbin:/usr/sbin:/usr/bin")
if path:
break
if not path:
sys.exit(f"{' or '.join(names)} not found")
TOOLS[name] = path
return TOOLS[name]
def sh(cmd, check=True, **kw):
return subprocess.run(cmd, shell=True, check=check, text=True,
stdout=subprocess.PIPE, stderr=subprocess.STDOUT, **kw).stdout
def cpulist(spec):
cpus = []
for part in spec.split(","):
a, _, b = part.partition("-")
cpus.extend(range(int(a), int(b or a) + 1))
return cpus
def cpumask(spec):
return "%x" % sum(1 << c for c in cpulist(spec))
# --- namespaces --------------------------------------------------------------
def net_up(args):
net_down()
for ns in ("bcli", "bprx", "bsrv"):
sh(f"ip netns add {ns}")
sh(f"ip -n {ns} link set lo up")
sh("ip link add bc0 netns bcli type veth peer name bp0 netns bprx")
sh("ip link add bp1 netns bprx type veth peer name bs0 netns bsrv")
for ns, dev, addr in (("bcli", "bc0", CLI_ADDR), ("bprx", "bp0", PRX_IN),
("bprx", "bp1", PRX_OUT), ("bsrv", "bs0", SRV_ADDR)):
sh(f"ip -n {ns} addr add {addr}/24 dev {dev}")
sh(f"ip -n {ns} link set {dev} mtu {args.mtu} up")
# the direct baseline goes through the proxy namespace, routed by the
# kernel
sh(f"ip -n bcli route add 10.0.2.0/24 via {PRX_IN}")
sh(f"ip -n bsrv route add 10.0.1.0/24 via {PRX_OUT}")
sh("ip netns exec bprx sysctl -qw net.ipv4.ip_forward=1")
for ns in ("bcli", "bprx", "bsrv"):
sh(f"ip netns exec {ns} sysctl -qw net.core.somaxconn=65535 "
"net.ipv4.tcp_max_syn_backlog=65535 net.ipv4.tcp_tw_reuse=1 "
"net.ipv4.ip_local_port_range='1024 65535'")
# A veth hands what it receives to the CPU that sent it; steer it to the
# receiver's own CPUs instead.
for ns, dev, role in (("bcli", "bc0", "cli"), ("bprx", "bp0", "prx"),
("bprx", "bp1", "prx"), ("bsrv", "bs0", "srv")):
sh(f"ip netns exec {ns} sh -c 'echo {cpumask(args.cpus[role])} "
f"> /sys/class/net/{dev}/queues/rx-0/rps_cpus'", check=False)
if args.netem:
for ns, dev in (("bcli", "bc0"), ("bprx", "bp0")):
sh(f"ip netns exec {ns} tc qdisc add dev {dev} root netem {args.netem}")
def net_down():
for ns in ("bcli", "bprx", "bsrv"):
sh(f"ip netns del {ns}", check=False)
# --- cgroups -----------------------------------------------------------------
def cg_write(path, value):
with open(path, "w") as f:
f.write(value)
def cg_read(path):
with open(path) as f:
return f.read()
def cg_up(args):
os.makedirs(CG, exist_ok=True)
cg_write("/sys/fs/cgroup/cgroup.subtree_control", "+cpuset +cpu +memory")
cg_write(f"{CG}/cgroup.subtree_control", "+cpuset +cpu +memory")
for role in ("cli", "srv"):
cg_make(role, args.cpus[role])
def cg_make(role, cpus):
path = f"{CG}/{role}"
os.makedirs(path, exist_ok=True)
cg_write(f"{path}/cpuset.cpus", cpus)
return path
def cg_remove(role):
path = f"{CG}/{role}"
if not os.path.isdir(path):
return
for _ in range(50):
try:
os.rmdir(path)
return
except OSError:
time.sleep(0.1)
def cg_kill(role):
"""Whatever is left in a role's cgroup, e.g. after an interrupted run."""
try:
pids = cg_read(f"{CG}/{role}/cgroup.procs").split()
except OSError:
return
for pid in pids:
try:
os.kill(int(pid), signal.SIGKILL)
except ProcessLookupError:
pass
def cg_down():
for role in ("cli", "prx", "srv"):
cg_kill(role)
cg_remove(role)
try:
os.rmdir(CG)
except OSError:
pass
def cg_usage(role):
"""CPU seconds used so far by everything in the cgroup."""
for line in cg_read(f"{CG}/{role}/cpu.stat").splitlines():
name, value = line.split()
if name == "usage_usec":
return int(value) / 1e6
return 0.0
TICK = os.sysconf("SC_CLK_TCK")
def cpu_busy():
"""Busy and total ticks per CPU. Busy counts interrupts and softirqs too:
a kernel with IRQ time accounting leaves them out of the cgroup's CPU
time, and that is where much of the network work is done."""
busy = {}
with open("/proc/stat") as f:
for line in f:
if not line.startswith("cpu") or line.startswith("cpu "):
continue
name, *v = line.split()
v = [int(x) for x in v]
# user nice system idle iowait irq softirq steal
busy[int(name[3:])] = (v[0] + v[1] + v[2] + v[5] + v[6] + v[7], sum(v[:8]))
return busy
def role_busy(before, after, cpus):
"""Busy seconds of a role's CPUs, and the share of them that was busy."""
b = t = 0
for c in cpus:
b += after[c][0] - before[c][0]
t += after[c][1] - before[c][1]
return b / TICK, (b / t if t else 0.0)
def thread_times(role):
"""CPU ticks per thread of everything in a role's cgroup."""
times = {}
try:
tids = cg_read(f"{CG}/{role}/cgroup.threads").split()
except OSError:
return times
for tid in tids:
try:
with open(f"/proc/{tid}/stat") as f:
v = f.read().rsplit(")", 1)[1].split()
times[tid] = int(v[11]) + int(v[12])
except (OSError, IndexError, ValueError):
pass
return times
def busiest_thread(before, after, wall):
"""The busiest thread's share of one CPU: one thread at 100% is a limit
however many CPUs its role has (iperf3 runs a stream in a thread)."""
top = 0
for tid, t in after.items():
top = max(top, t - before.get(tid, 0))
return top / TICK / wall if wall else 0.0
def cg_peak(role):
try:
return int(cg_read(f"{CG}/{role}/memory.peak")) / 2**20
except OSError:
return float("nan")
# --- processes ---------------------------------------------------------------
def spawn(ns, role, argv, log):
"""Start a process in a namespace and a cgroup; threads it starts stay
there with it."""
procs = f"{CG}/{role}/cgroup.procs"
def enter():
cg_write(procs, str(os.getpid()))
resource.setrlimit(resource.RLIMIT_NOFILE, (1048576, 1048576))
os.setsid()
return subprocess.Popen(["ip", "netns", "exec", ns] + argv, stdout=log,
stderr=subprocess.STDOUT, preexec_fn=enter)
LAST_CLIENT_THREAD = [0.0]
def run_in(ns, role, argv, timeout):
"""Run a load generator to its end. Its busiest thread's share of a CPU
is kept in LAST_CLIENT_THREAD, read just before it exits."""
p = spawn(ns, role, argv, subprocess.PIPE)
t0 = time.time()
top = {}
buf = []
reader = threading.Thread(target=lambda: buf.append(p.stdout.read()))
reader.start()
while p.poll() is None:
for tid, t in thread_times(role).items():
top[tid] = t
time.sleep(0.5)
reader.join(timeout)
wall = time.time() - t0
LAST_CLIENT_THREAD[0] = max(top.values()) / TICK / wall if top and wall else 0.0
return (buf[0] if buf else b"").decode("utf-8", "replace")
def stop(p):
if p and p.poll() is None:
try:
os.killpg(p.pid, signal.SIGTERM)
p.wait(timeout=5)
except (ProcessLookupError, subprocess.TimeoutExpired):
try:
os.killpg(p.pid, signal.SIGKILL)
except ProcessLookupError:
pass
p.wait()
def wait_port(ns, addr, port, timeout=10):
deadline = time.time() + timeout
while time.time() < deadline:
# /dev/tcp is bash's
if subprocess.run(["ip", "netns", "exec", ns, "bash", "-c",
f"exec 3<>/dev/tcp/{addr}/{port}"],
stderr=subprocess.DEVNULL).returncode == 0:
return True
time.sleep(0.1)
return False
# --- the origin --------------------------------------------------------------
def origin_up(args, work):
www = os.path.join(work, "www")
os.makedirs(www, exist_ok=True)
for name, size in FILES.items():
with open(os.path.join(www, name), "wb") as f:
f.write(b"x" * size)
n = len(cpulist(args.cpus["srv"]))
conf = os.path.join(work, "origin.conf")
with open(conf, "w") as f:
f.write(f"""daemon off;
worker_processes {n};
pid {work}/origin.pid;
error_log {work}/origin.err;
worker_rlimit_nofile 1048576;
events {{ worker_connections 65536; }}
http {{
access_log off;
sendfile on;
tcp_nopush on;
keepalive_requests 1000000;
keepalive_timeout 120s;
server {{
listen {SRV_ADDR}:{ORIGIN_HTTP} backlog=65535 reuseport;
root {www};
default_type application/octet-stream;
# the files answer a POST too, its body read and thrown away
error_page 405 =200 $uri;
}}
}}
""")
log = open(os.path.join(work, "origin.out"), "w")
procs = [spawn("bsrv", "srv", [tool("nginx"), "-c", conf, "-p", work], log),
spawn("bsrv", "srv", [tool("iperf3"), "-s", "-B", SRV_ADDR, "-p", str(ORIGIN_TCP)], log)]
if not (wait_port("bsrv", SRV_ADDR, ORIGIN_HTTP) and wait_port("bsrv", SRV_ADDR, ORIGIN_TCP)):
for p in procs:
stop(p)
sys.exit("the origin did not start, see " + work)
return procs
# --- proxies -----------------------------------------------------------------
def proxy_conf(kind, work, ncpu, label, opts=""):
"""Configuration for one proxy: an HTTP proxy on HTTP_PORT and a TCP relay
on TCP_PORT, both towards the origin. Returns the argv. opts are added to
both 3proxy services, e.g. -s0 to keep a version from using splice()."""
if kind == "3proxy":
conf = os.path.join(work, f"{label}.cfg")
with open(conf, "w") as f:
f.write(f"""maxconn 100000
auth none
proxy {opts} -i{PRX_IN} -e{PRX_OUT} -p{HTTP_PORT}
tcppm {opts} -i{PRX_IN} -e{PRX_OUT} {TCP_PORT} {SRV_ADDR} {ORIGIN_TCP}
tcppm {opts} -i{PRX_IN} -e{PRX_OUT} {RELAY_HTTP} {SRV_ADDR} {ORIGIN_HTTP}
socks {opts} -i{PRX_IN} -e{PRX_OUT} -p{SOCKS_PORT}
""")
return [conf]
if kind == "tinyproxy":
conf = os.path.join(work, f"{label}.conf")
with open(conf, "w") as f:
f.write(f"""Port {HTTP_PORT}
Listen {PRX_IN}
Bind {PRX_OUT}
Timeout 600
MaxClients 100000
LogLevel Critical
LogFile "{work}/{label}.log"
PidFile "{work}/{label}.pid"
DisableViaHeader Yes
""")
return ["-d", "-c", conf]
if kind == "squid":
conf = os.path.join(work, f"{label}.conf")
with open(conf, "w") as f:
f.write(f"""http_port {PRX_IN}:{HTTP_PORT}
tcp_outgoing_address {PRX_OUT}
http_access allow all
cache deny all
cache_mem 8 MB
access_log none
cache_log {work}/{label}.log
pid_filename {work}/{label}.pid
coredump_dir {work}
cache_effective_user nobody
max_filedescriptors 65536
shutdown_lifetime 1 seconds
via off
forwarded_for delete
""")
return ["-N", "-f", conf]
if kind == "dante":
conf = os.path.join(work, f"{label}.conf")
with open(conf, "w") as f:
f.write(f"""logoutput: {work}/{label}.log
internal: {PRX_IN} port = {SOCKS_PORT}
external: {PRX_OUT}
socksmethod: none
clientmethod: none
user.privileged: root
user.unprivileged: nobody
client pass {{ from: 0.0.0.0/0 to: 0.0.0.0/0 }}
socks pass {{ from: 0.0.0.0/0 to: 0.0.0.0/0 }}
""")
return ["-f", conf, "-p", f"{work}/{label}.pid", "-N", "1"]
if kind == "nginx":
conf = os.path.join(work, f"{label}.conf")
with open(conf, "w") as f:
# distributions build the stream module as a dynamic one
load = ""
for d in ("/usr/lib/nginx/modules", "/usr/lib64/nginx/modules"):
stream = os.path.join(d, "ngx_stream_module.so")
if os.path.exists(stream):
load = f"load_module {stream};\n"
break
f.write(f"""{load}daemon off;
worker_processes {ncpu};
pid {work}/{label}.pid;
error_log {work}/{label}.err;
worker_rlimit_nofile 1048576;
events {{ worker_connections 65536; }}
http {{
access_log off;
upstream origin {{ server {SRV_ADDR}:{ORIGIN_HTTP}; keepalive 1024; }}
server {{
listen {PRX_IN}:{HTTP_PORT} backlog=65535 reuseport;
keepalive_requests 1000000;
location / {{
proxy_pass http://origin;
proxy_http_version 1.1;
proxy_set_header Connection "";
proxy_bind {PRX_OUT};
}}
}}
}}
stream {{
server {{
listen {PRX_IN}:{TCP_PORT} backlog=65535 reuseport;
proxy_pass {SRV_ADDR}:{ORIGIN_TCP};
proxy_bind {PRX_OUT};
}}
server {{
listen {PRX_IN}:{RELAY_HTTP} backlog=65535 reuseport;
proxy_pass {SRV_ADDR}:{ORIGIN_HTTP};
proxy_bind {PRX_OUT};
}}
}}
""")
return ["-c", conf, "-p", work]
if kind == "haproxy":
# splice-auto: haproxy's heuristic, splicing a direction once reads
# fill its buffer three times in a row; splice-response, splice-request
# and splice: the server's data, the client's, or both, always
# (3proxy's -s1, -s2 and -s3)
splice = {"": "", "splice-auto": " option splice-auto\n",
"splice-response": " option splice-response\n",
"splice-request": " option splice-request\n",
"splice": " option splice-request\n option splice-response\n"}
if opts not in splice:
sys.exit("haproxy options: " + ", ".join(k for k in splice if k))
splice = splice[opts]
conf = os.path.join(work, f"{label}.cfg")
with open(conf, "w") as f:
f.write(f"""global
nbthread {ncpu}
maxconn 200000
defaults
{splice} timeout connect 5s
timeout client 60s
timeout server 60s
frontend http
mode http
bind {PRX_IN}:{HTTP_PORT}
option http-keep-alive
default_backend origin
backend origin
mode http
http-reuse always
source {PRX_OUT}
server origin {SRV_ADDR}:{ORIGIN_HTTP}
listen tcp
mode tcp
bind {PRX_IN}:{TCP_PORT}
source {PRX_OUT}
server origin {SRV_ADDR}:{ORIGIN_TCP}
listen tcphttp
mode tcp
bind {PRX_IN}:{RELAY_HTTP}
source {PRX_OUT}
server origin {SRV_ADDR}:{ORIGIN_HTTP}
""")
return ["-f", conf]
raise ValueError(kind)
class Proxy:
def __init__(self, spec):
kind, _, path = spec.partition(":")
self.kind = kind
self.opts = ""
if kind == "3proxy":
path, _, self.opts = path.partition(":")
if not path:
sys.exit("--proxy 3proxy:<binary>[:<service options>]")
self.binary = os.path.abspath(path)
self.label = os.path.basename(path) + (self.opts.replace(" ", "") if self.opts else "")
elif kind in ("nginx", "haproxy"):
path, _, self.opts = path.partition(":")
if kind == "nginx" and self.opts:
sys.exit("nginx takes no options (it does not use splice() at all)")
self.binary = path or tool(kind)
self.label = kind + ("-" + self.opts if self.opts else "")
elif kind in ("tinyproxy", "squid", "dante"):
# dante's server is sockd, danted on Debian
self.binary = path or (tool("sockd", "danted") if kind == "dante" else tool(kind))
self.label = kind
elif kind == "none":
self.binary = None
self.label = "direct"
else:
sys.exit("unknown proxy " + spec)
# what each proxy listens on, and which scenarios it can take part in:
# tinyproxy and squid are HTTP proxies, dante a SOCKS server only
PORTS = {"3proxy": (HTTP_PORT, TCP_PORT, SOCKS_PORT, RELAY_HTTP),
"nginx": (HTTP_PORT, TCP_PORT, RELAY_HTTP),
"haproxy": (HTTP_PORT, TCP_PORT, RELAY_HTTP),
"tinyproxy": (HTTP_PORT,), "squid": (HTTP_PORT,), "dante": (SOCKS_PORT,)}
def serves(self, scenario):
kind, params = SCENARIOS[scenario]
via = params.get("via")
if self.kind == "dante":
return via == "socks"
if self.kind in ("tinyproxy", "squid"):
return via == "connect" or (kind == "http" and not via)
if self.kind in ("nginx", "haproxy"):
# no CONNECT, no SOCKS
return via in (None, "relay")
return True
def start(self, args, work):
cg_remove("prx")
cg_make("prx", args.cpus["prx"])
if self.kind == "none":
return None
argv = [self.binary] + proxy_conf(self.kind, work, len(cpulist(args.cpus["prx"])),
self.label, self.opts)
log = open(os.path.join(work, self.label + ".out"), "a")
p = spawn("bprx", "prx", argv, log)
if not all(wait_port("bcli", PRX_IN, port, 30) for port in self.PORTS[self.kind]):
stop(p)
sys.exit(f"{self.label} did not start, see {work}")
return p
# --- load --------------------------------------------------------------------
WRK_SCRIPT = """
local path, close, size = "%s", %s, %d
local headers = {["Host"] = "%s:%d"}
if close then headers["Connection"] = "close" end
local req
if size > 0 then
req = wrk.format("POST", path, headers, string.rep("u", size))
else
req = wrk.format("GET", path, headers)
end
request = function() return req end
"""
UNITS = {"": 1, "k": 1e3, "K": 1e3, "M": 1e6, "G": 1e9, "KB": 2**10, "MB": 2**20,
"GB": 2**30, "B": 1}
def to_ms(text):
m = re.match(r"([\d.]+)(us|ms|s|m)", text)
v, u = float(m.group(1)), m.group(2)
return v * {"us": 1e-3, "ms": 1, "s": 1e3, "m": 6e4}[u]
def tunnel(via, work):
"""argv prefix running a load generator so that its connections go
through tunnels of the proxy: CONNECT, or SOCKSv5."""
conf = os.path.join(work, f"proxychains-{via}.conf")
line = f"http {PRX_IN} {HTTP_PORT}" if via == "connect" else f"socks5 {PRX_IN} {SOCKS_PORT}"
with open(conf, "w") as f:
f.write(f"strict_chain\nquiet_mode\ntcp_read_time_out 15000\n"
f"tcp_connect_time_out 8000\n[ProxyList]\n{line}\n")
return [tool("proxychains4"), "-q", "-f", conf]
def target(proxy, params, work, direct_port, proxy_port, relay_port):
"""Where the load generator connects, and what it runs under: the origin
itself with no proxy or through a tunnel, the proxy's relay otherwise."""
if proxy.kind == "none":
return SRV_ADDR, direct_port, []
if params.get("via") == "relay":
return PRX_IN, relay_port, []
if params.get("via"):
return SRV_ADDR, direct_port, tunnel(params["via"], work)
return PRX_IN, proxy_port, []
def run_http(proxy, params, args, work):
host, port, prefix = target(proxy, params, work, ORIGIN_HTTP, HTTP_PORT, RELAY_HTTP)
path = "/" + params["file"]
if host == PRX_IN and port == HTTP_PORT:
path = f"http://{SRV_ADDR}:{ORIGIN_HTTP}{path}"
script = os.path.join(work, "wrk.lua")
with open(script, "w") as f:
f.write(WRK_SCRIPT % (path, "true" if params.get("close") else "false",
params.get("body", 0), SRV_ADDR, ORIGIN_HTTP))
threads = len(cpulist(args.cpus["cli"]))
if prefix:
# proxychains asks for each tunnel inside connect(), blocking the
# thread: more threads keep the client's CPUs busy meanwhile
threads = min(params["conns"], threads * 4)
conns = max(params["conns"], threads)
out = run_in("bcli", "cli", prefix + [tool("wrk"), "-t", str(threads), "-c", str(conns),
"-d", f"{args.duration}s", "--latency", "-s", script,
f"http://{host}:{port}/"], args.duration + 30)
res = {"raw": out}
m = re.search(r"(\d+) requests in", out)
res["requests"] = int(m.group(1)) if m else 0
m = re.search(r"Requests/sec:\s+([\d.]+)", out)
res["rate"] = float(m.group(1)) if m else 0.0
m = re.search(r"Transfer/sec:\s+([\d.]+)(\w+)", out)
res["gbit"] = float(m.group(1)) * UNITS[m.group(2)] * 8 / 1e9 if m else 0.0
for pct in ("50", "99"):
m = re.search(rf"^\s+{pct}%\s+(\S+)", out, re.M)
res[f"p{pct}_ms"] = to_ms(m.group(1)) if m else float("nan")
errors = 0
res["timeouts"] = 0
m = re.search(r"Socket errors: connect (\d+), read (\d+), write (\d+), timeout (\d+)", out)
if m:
errors += sum(int(x) for x in m.groups()[:3])
# requests slower than wrk's 2 seconds: a latency tail, not a failure
res["timeouts"] = int(m.group(4))
m = re.search(r"Non-2xx or 3xx responses: (\d+)", out)
if m:
errors += int(m.group(1))
if not res["requests"]:
errors += 1 # wrk did not run, or reported nothing
res["errors"] = errors
res["retrans"] = 0
res["bytes"] = res["gbit"] * 1e9 / 8 * args.duration
return res
def run_stream(proxy, params, args, work):
host, port, prefix = target(proxy, params, work, ORIGIN_TCP, TCP_PORT, TCP_PORT)
argv = prefix + [tool("iperf3"), "-c", host, "-p", str(port), "-t", str(args.duration),
"-P", str(params["streams"]), "-J"]
if params.get("reverse"):
argv.append("-R")
out = run_in("bcli", "cli", argv, args.duration + 30)
res = {"raw": out, "rate": float("nan"), "p50_ms": float("nan"),
"p99_ms": float("nan"), "requests": 0, "errors": 0, "retrans": 0, "timeouts": 0}
try:
j = json.loads(out)
s = j["end"]["sum_received"]
res["gbit"] = s["bits_per_second"] / 1e9
res["bytes"] = s["bytes"]
res["retrans"] = j["end"].get("sum_sent", {}).get("retransmits", 0)
if "error" in j:
res["errors"] = 1
except (ValueError, KeyError):
res["gbit"], res["bytes"], res["errors"] = 0.0, 0, 1
return res
BUSY = 0.9
def measure(proxy, scenario, args, work):
kind, params = SCENARIOS[scenario]
p = proxy.start(args, work)
try:
time.sleep(0.5)
task0 = cg_usage("prx")
cpu0 = cpu_busy()
thr0 = {r: thread_times(r) for r in ("cli", "srv")}
t0 = time.time()
res = (run_http if kind == "http" else run_stream)(proxy, params, args, work)
res["cli_thread"] = LAST_CLIENT_THREAD[0]
wall = time.time() - t0
cpu1 = cpu_busy()
# threads of the load generator have ended by now: take what the
# client's cgroup reports for the run instead
thr1 = {r: thread_times(r) for r in ("cli", "srv")}
res["prx_task_s"] = cg_usage("prx") - task0
res["peak_mb"] = cg_peak("prx")
finally:
stop(p)
for r in ("cli", "prx", "srv"):
# the busy time of the role's CPUs, its own processes and the network
# work done there for it
res[f"{r}_cpu_s"], res[f"{r}_util"] = role_busy(cpu0, cpu1, cpulist(args.cpus[r]))
res["srv_thread"] = busiest_thread(thr0["srv"], thr1["srv"], wall)
# Which side set the pace: a role whose CPUs, or a single thread of it,
# were busy all the time. A run limited by the client or the origin says
# only that the proxy can do at least this much.
if res["srv_util"] >= BUSY or res["srv_thread"] >= BUSY:
res["bound"] = "origin"
elif res["cli_util"] >= BUSY or res["cli_thread"] >= BUSY:
res["bound"] = "client"
elif res["prx_util"] >= BUSY:
res["bound"] = "proxy"
else:
res["bound"] = "none"
gb = res["bytes"] / 1e9
res["cpu_s_per_gb"] = res["prx_cpu_s"] / gb if gb else float("nan")
res["cpu_us_per_req"] = (res["prx_cpu_s"] * 1e6 / res["requests"]
if res["requests"] else float("nan"))
return res
# --- main --------------------------------------------------------------------
FIELDS = ["proxy", "scenario", "rep", "gbit", "rate", "p50_ms", "p99_ms", "errors", "timeouts", "retrans",
"prx_cpu_s", "prx_task_s", "cpu_s_per_gb", "cpu_us_per_req", "peak_mb",
"cli_util", "prx_util", "srv_util", "cli_thread", "srv_thread", "bound"]
def summary(rows):
keys = []
for r in rows:
if (r["proxy"], r["scenario"]) not in keys:
keys.append((r["proxy"], r["scenario"]))
cols = ["gbit", "rate", "p99_ms", "cpu_s_per_gb", "cpu_us_per_req", "peak_mb",
"prx_util", "cli_util", "srv_util", "errors"]
print("\nmedians (" + ", ".join(cols) + ")")
print("%-24s %-14s " % ("proxy", "scenario") + " ".join("%10s" % c[:10] for c in cols))
for proxy, scenario in sorted(keys, key=lambda k: (k[1], k[0])):
sel = [r for r in rows if r["proxy"] == proxy and r["scenario"] == scenario]
vals = []
for c in cols:
xs = [r[c] for r in sel if r[c] == r[c]]
vals.append("%10.3g" % statistics.median(xs) if xs else "%10s" % "-")
print("%-24s %-14s " % (proxy, scenario) + " ".join(vals))
def main():
ap = argparse.ArgumentParser(description=__doc__.split("\n\n")[0])
ap.add_argument("--proxy", action="append", required=True,
help="3proxy:<binary>[:<service options>], nginx[:<binary>], "
"tinyproxy, squid, dante (sockd), "
"haproxy[:<binary>[:splice-auto|splice-response|splice-request|splice]] "
"or none (the kernel routing, an upper bound); repeat")
ap.add_argument("--scenario", action="append", choices=sorted(SCENARIOS),
help="default: all")
ap.add_argument("--reps", type=int, default=3)
ap.add_argument("--duration", type=int, default=10, help="seconds per run")
ap.add_argument("--cpus", default="cli=0-3,prx=4,srv=5-7",
help="CPUs per role (default %(default)s)")
ap.add_argument("--mtu", type=int, default=1500)
ap.add_argument("--netem", help='netem on the client link, e.g. "delay 1ms rate 10gbit"')
ap.add_argument("--out", default="bench/results", help="directory for the CSV and logs")
args = ap.parse_args()
if os.geteuid():
sys.exit("needs root")
args.cpus = dict(x.split("=") for x in args.cpus.split(","))
proxies = [Proxy(s) for s in args.proxy]
scenarios = args.scenario or list(SCENARIOS)
# fail now rather than after the first scenarios
for name in ("nginx", "iperf3", "wrk"):
tool(name)
if any(SCENARIOS[s][1].get("via") in ("connect", "socks") for s in scenarios):
tool("proxychains4")
stamp = datetime.datetime.now().strftime("%Y%m%d-%H%M%S")
os.makedirs(args.out, exist_ok=True)
work = tempfile.mkdtemp(prefix="3proxy-bench-")
# squid writes its log and pid file there after giving up root
os.chmod(work, 0o1777)
csvpath = os.path.join(args.out, f"{stamp}.csv")
rawpath = os.path.join(args.out, f"{stamp}.raw.txt")
with open(os.path.join(args.out, f"{stamp}.env.txt"), "w") as f:
f.write(" ".join(sys.argv) + "\n")
f.write(sh("uname -a; lscpu | head -20; nproc; "
"cat /sys/devices/system/cpu/cpu0/cpufreq/scaling_governor 2>/dev/null; "
f"{tool('iperf3')} --version | head -1; {tool('wrk')} -v 2>&1 | head -1; "
f"{tool('nginx')} -v 2>&1; "
"haproxy -v 2>/dev/null | head -1", check=False))
rows = []
origin = []
try:
net_up(args)
cg_up(args)
origin = origin_up(args, work)
with open(csvpath, "w", newline="") as f, open(rawpath, "w") as raw:
w = csv.DictWriter(f, FIELDS, extrasaction="ignore")
w.writeheader()
# Every repetition runs each proxy once, rotating the order, so a
# drift over time is shared by all of them.
for rep in range(args.reps):
for scenario in scenarios:
order = proxies[rep % len(proxies):] + proxies[:rep % len(proxies)]
for proxy in order:
if not proxy.serves(scenario):
continue
res = measure(proxy, scenario, args, work)
res.update(proxy=proxy.label, scenario=scenario, rep=rep)
rows.append(res)
w.writerow(res)
f.flush()
raw.write(f"=== {proxy.label} {scenario} {rep}\n{res['raw']}\n")
print("%-24s %-14s rep %d: %7.2f Gbit/s %10.0f req/s proxy cpu %5.2fs "
"busy cli %3.0f%% prx %3.0f%% srv %3.0f%% (threads %3.0f%%/%3.0f%%) "
"bound %-6s errors %d" % (
proxy.label, scenario, rep, res["gbit"],
res["rate"] if res["rate"] == res["rate"] else 0,
res["prx_cpu_s"], res["cli_util"] * 100,
res["prx_util"] * 100, res["srv_util"] * 100,
res["cli_thread"] * 100, res["srv_thread"] * 100,
res["bound"], res["errors"]), flush=True)
finally:
for p in origin:
stop(p)
cg_down()
net_down()
summary(rows)
print(f"\n{csvpath}\nlogs and configurations: {work}")
if __name__ == "__main__":
main()
Reference in New Issue View Git Blame Copy Permalink