785 lines
34 KiB
Julia
Executable File
785 lines
34 KiB
Julia
Executable File
#!/usr/bin/env julia
|
||
#
|
||
# bench.jl: measure end-to-end throughput and server memory for a running FileServer.
|
||
#
|
||
# Two things make this pipeline awkward to benchmark with off-the-shelf tools
|
||
# (ab/hey/wrk), and both shape what this script does:
|
||
#
|
||
# 1. HTTP latency is not throughput. /upload returns 202 as soon as the bytes
|
||
# are spooled and a reference is enqueued, and all four stages run afterwards.
|
||
# So real throughput is the *arrival rate at the terminal sinks*
|
||
# (done/, text_done/, binary/, failed/), not the response rate. We upload a
|
||
# corpus, then poll the sinks until the file count stops moving.
|
||
#
|
||
# 2. End-to-end throughput doesn't name the slow stage. The four stages run
|
||
# concurrently behind their own queues, so the pipeline's rate is the
|
||
# slowest stage's rate and the others are invisible. Directory polling can't
|
||
# recover them either: every in-flight file sits in spool/ whatever stage it
|
||
# is at, since stages route by enqueueing rather than by moving bytes. So the
|
||
# server keeps per-stage counters
|
||
# (src/stats.jl) and we scrape GET /stats before and after: the deltas give
|
||
# each stage's throughput, mean service time, and worker utilization, and
|
||
# utilization is what actually names the bottleneck (see `stage_report`).
|
||
#
|
||
# 3. Memory should be flat in file size, and that claim needs checking on two
|
||
# axes. The workers read bounded prefixes (16+16 bytes to classify, 8 KB to
|
||
# sniff, 64 KB to language-detect), and intake streams each upload from the
|
||
# socket to the spool file a chunk at a time (FS_UPLOAD_CHUNK_BYTES, see
|
||
# src/multipart.jl). So peak RSS should track *concurrency*, not file size:
|
||
# sweeping --size at fixed --concurrency should be a flat line, and that is
|
||
# the regression this measures. We sample the server's RSS throughout and
|
||
# report the high-water mark.
|
||
#
|
||
# (Before intake was streamed it buffered each upload whole, several times
|
||
# over, and a 256 MiB upload grew RSS by ~700 MiB. If a --size sweep ever
|
||
# slopes upward again, something has started buffering.)
|
||
#
|
||
# The harness talks to the server only over HTTP (/upload, /health, /stats) and
|
||
# reads the on-disk sink layout; it must run on the same machine (sink dirs and
|
||
# /proc). If /stats is missing (an older build) everything else still works and
|
||
# the per-stage section is skipped.
|
||
#
|
||
# Usage:
|
||
# julia --project=. -t auto bin/bench.jl [options]
|
||
#
|
||
# --url URL server base URL (default: $FS_URL or http://127.0.0.1:8080)
|
||
# --files N number of files to upload (default: 200)
|
||
# --size S size of each generated file, e.g. 4k, 512k, 8m, 1g (default: 64k)
|
||
# --concurrency J uploads in flight at once (default: 8)
|
||
# --kind K binary | text | mixed: what to generate (default: binary)
|
||
# --corpus DIR upload an existing directory instead of generating
|
||
# (the only way to exercise stage 2: point it at real known files)
|
||
# --keep-corpus don't delete the generated corpus on exit
|
||
# --pid PID server pid for memory sampling (default: autodetect)
|
||
# --no-mem skip memory sampling entirely
|
||
# --no-stats skip the per-stage /stats scrape and its table
|
||
# --sample-ms MS sink/RSS sampling interval (default: 200)
|
||
# --timeout SEC give up after this long with no drain progress (default: 120)
|
||
# --json PATH also write the results as JSON
|
||
# --force skip the corpus-size safety check
|
||
#
|
||
# Examples:
|
||
# # throughput: many small files, high concurrency
|
||
# julia --project=. -t auto bin/bench.jl --files 2000 --size 8k --concurrency 32
|
||
#
|
||
# # memory: flat in file size? sweep --size with concurrency pinned
|
||
# julia --project=. -t auto bin/bench.jl --files 4 --size 256m --concurrency 1
|
||
# julia --project=. -t auto bin/bench.jl --files 2 --size 1g --concurrency 1
|
||
# julia --project=. -t auto bin/bench.jl --files 8 --size 1g --concurrency 8
|
||
#
|
||
# # stage 2 (exiftool) with real known files
|
||
# julia --project=. -t auto bin/bench.jl --corpus ../training_set --concurrency 16
|
||
|
||
using HTTP
|
||
using JSON3
|
||
using Random
|
||
|
||
# ---------------------------------------------------------------- option parsing
|
||
|
||
const DEFAULTS = Dict{String,Any}(
|
||
"url" => get(ENV, "FS_URL", "http://127.0.0.1:8080"),
|
||
"files" => 200,
|
||
"size" => 64 * 1024,
|
||
"concurrency" => 8,
|
||
"kind" => "binary",
|
||
"corpus" => nothing,
|
||
"keep-corpus" => false,
|
||
"pid" => nothing,
|
||
"no-mem" => false,
|
||
"no-stats" => false,
|
||
"sample-ms" => 200,
|
||
"timeout" => 120,
|
||
"json" => nothing,
|
||
"force" => false,
|
||
)
|
||
|
||
const FLAGS = ("keep-corpus", "no-mem", "no-stats", "force")
|
||
|
||
# Approximate RSS of a freshly started server (Lux + the loaded classifier + the
|
||
# language detector, measured on Julia 1.12 / -t auto). Only used to notice that
|
||
# a baseline is inflated by a previous run's un-returned GC memory, so a rough
|
||
# figure is enough.
|
||
const FRESH_RSS_HINT = 950 * 1024^2
|
||
|
||
"Parse `4k`/`8M`/`1g`/`4096` into a byte count."
|
||
function parse_size(s::AbstractString)::Int
|
||
m = match(r"^(\d+(?:\.\d+)?)\s*([kKmMgG]?)[bB]?$", strip(s))
|
||
m === nothing && error("bad --size: $s (expected e.g. 512, 64k, 8m, 1g)")
|
||
mult = Dict('k' => 1024, 'm' => 1024^2, 'g' => 1024^3)
|
||
scale = isempty(m[2]) ? 1 : mult[lowercase(m[2])[1]]
|
||
return round(Int, parse(Float64, m[1]) * scale)
|
||
end
|
||
|
||
function parse_args(argv)::Dict{String,Any}
|
||
opts = copy(DEFAULTS)
|
||
i = 1
|
||
while i <= length(argv)
|
||
a = argv[i]
|
||
startswith(a, "--") || error("unexpected argument: $a (see the header of $(PROGRAM_FILE))")
|
||
key = a[3:end]
|
||
haskey(opts, key) || error("unknown option: $a")
|
||
if key in FLAGS
|
||
opts[key] = true
|
||
i += 1
|
||
continue
|
||
end
|
||
i + 1 <= length(argv) || error("option --$key needs a value")
|
||
val = argv[i+1]
|
||
opts[key] = key == "size" ? parse_size(val) :
|
||
key in ("files", "concurrency", "sample-ms") ? parse(Int, val) :
|
||
key == "timeout" ? parse(Float64, val) :
|
||
key == "pid" ? parse(Int, val) :
|
||
val
|
||
i += 2
|
||
end
|
||
opts["kind"] in ("binary", "text", "mixed") ||
|
||
error("--kind must be binary, text or mixed (got $(opts["kind"]))")
|
||
opts["files"] >= 1 || error("--files must be >= 1")
|
||
opts["concurrency"] >= 1 || error("--concurrency must be >= 1")
|
||
return opts
|
||
end
|
||
|
||
# ------------------------------------------------------------------------- dirs
|
||
#
|
||
# Resolved from the same environment variables src/config.jl reads, so a server
|
||
# started with custom dirs is benchmarked correctly. Kept as a standalone table
|
||
# rather than `using FileServer` so the harness doesn't pay to load Lux.
|
||
|
||
sinkdirs() = (
|
||
done = get(ENV, "FS_DONE_DIR", "data/done"),
|
||
text_done = get(ENV, "FS_TEXT_DONE_DIR", "data/text_done"),
|
||
binary = get(ENV, "FS_BINARY_DIR", "data/binary"),
|
||
failed = get(ENV, "FS_FAILED_DIR", "data/failed"),
|
||
)
|
||
|
||
# One directory, not four. Files no longer move between stages: spool/ holds
|
||
# every in-flight file at every stage, and which stage it has reached lives in
|
||
# the queue holding its reference (src/worker.jl header). So this depth is
|
||
# "files in flight", full stop; per-stage depth comes from /stats, which is the
|
||
# only place that can see it at all.
|
||
stagedirs() = (spool = get(ENV, "FS_SPOOL_DIR", "data/spool"),)
|
||
|
||
"Count work items in `dir`, ignoring the .meta.json sidecars stages 2/4 write."
|
||
function count_files(dir::AbstractString)::Int
|
||
isdir(dir) || return 0
|
||
n = 0
|
||
for name in readdir(dir)
|
||
endswith(name, ".meta.json") && continue
|
||
isfile(joinpath(dir, name)) && (n += 1)
|
||
end
|
||
return n
|
||
end
|
||
|
||
counts(dirs) = NamedTuple{keys(dirs)}(map(count_files, values(dirs)))
|
||
total(c) = sum(values(c))
|
||
deltas(now_, base) = NamedTuple{keys(now_)}(map(-, values(now_), values(base)))
|
||
|
||
# ----------------------------------------------------------------------- memory
|
||
|
||
"Read (VmRSS, VmHWM) in bytes for `pid`, or `nothing` if unreadable."
|
||
function read_rss(pid::Int)
|
||
rss = hwm = nothing
|
||
try
|
||
for line in eachline("/proc/$pid/status")
|
||
if startswith(line, "VmRSS:")
|
||
rss = parse(Int, split(line)[2]) * 1024
|
||
elseif startswith(line, "VmHWM:")
|
||
hwm = parse(Int, split(line)[2]) * 1024
|
||
end
|
||
end
|
||
catch
|
||
return nothing
|
||
end
|
||
return (rss === nothing || hwm === nothing) ? nothing : (rss, hwm)
|
||
end
|
||
|
||
"""
|
||
Find the running server process, or `nothing`.
|
||
|
||
`pgrep -f` matches against the whole command line, which catches more than the
|
||
server: any shell launched with the command in its own argv (`sh -c 'julia …
|
||
bin/server.jl > log'`, a `setsid`/`nohup` wrapper, even the terminal running the
|
||
benchmark) matches the same pattern. Sampling one of those reports a few MiB of
|
||
shell as the server's memory: a wrong answer that looks plausible, which is the
|
||
worst kind.
|
||
|
||
So candidates are filtered by what each process *is* (`/proc/<pid>/comm`, the
|
||
executable name) rather than by what its arguments say. No pattern over argv can
|
||
make that distinction.
|
||
"""
|
||
function detect_pid()
|
||
out = try
|
||
readchomp(`pgrep -f "bin/server.jl"`)
|
||
catch
|
||
return nothing
|
||
end
|
||
candidates = parse.(Int, split(out))
|
||
pids = filter(candidates) do pid
|
||
comm = try
|
||
readchomp("/proc/$pid/comm")
|
||
catch
|
||
return false # exited between pgrep and here
|
||
end
|
||
startswith(comm, "julia")
|
||
end
|
||
isempty(pids) && return nothing
|
||
length(pids) > 1 && @warn "multiple server processes matched; sampling the first" pids
|
||
return first(pids)
|
||
end
|
||
|
||
"""
|
||
Reset the kernel's peak-RSS counter so VmHWM reflects only this run.
|
||
|
||
Without it, VmHWM carries the high-water mark from startup (model load) or from
|
||
an earlier benchmark, which would silently dominate a small run's result.
|
||
Requires the server to run as the same user; on failure we say so and fall back
|
||
to sampled VmRSS, which can miss a spike between samples.
|
||
"""
|
||
function reset_peak_rss(pid::Int)::Bool
|
||
try
|
||
write("/proc/$pid/clear_refs", "5")
|
||
return true
|
||
catch
|
||
return false
|
||
end
|
||
end
|
||
|
||
# ------------------------------------------------------------- per-stage counters
|
||
#
|
||
# The server exposes monotonic counters at GET /stats (src/stats.jl). Rates are
|
||
# ours to compute: scrape once before the run and once after, subtract, divide by
|
||
# the elapsed *server* clock so a slow scrape doesn't distort the window.
|
||
|
||
"Fetch and parse GET /stats, or `nothing` if the server doesn't serve it."
|
||
function scrape_stats(url::String)
|
||
try
|
||
resp = HTTP.get(string(rstrip(url, '/'), "/stats");
|
||
status_exception = false, retry = false, readtimeout = 5)
|
||
resp.status == 200 || return nothing
|
||
return JSON3.read(String(resp.body))
|
||
catch
|
||
return nothing
|
||
end
|
||
end
|
||
|
||
"One stage's activity between two scrapes."
|
||
struct StageDelta
|
||
stage::Int
|
||
name::String
|
||
workers::Int
|
||
completed::Int
|
||
failed::Int
|
||
bytes::Int
|
||
busy::Float64 # summed handler seconds across all workers in the pool
|
||
blocked::Float64 # of `busy`, seconds parked on a full downstream queue
|
||
peak_depth::Int # deepest its queue got, from the sampler
|
||
capacity::Int
|
||
end
|
||
|
||
files_per_sec(d::StageDelta, window) = d.completed / max(window, 1e-9)
|
||
mib_per_sec(d::StageDelta, window) = d.bytes / max(window, 1e-9) / 1024^2
|
||
"Mean wall time one file spends in one worker of this stage."
|
||
service_ms(d::StageDelta) = d.completed == 0 ? NaN :
|
||
(d.busy - d.blocked) / d.completed * 1000
|
||
"""
|
||
Fraction of the pool's capacity spent doing this stage's own work.
|
||
|
||
Blocked time is subtracted first: a stage parked on a full downstream queue is
|
||
waiting, not working, and leaving it in would light up every stage upstream of a
|
||
jam as though each were the jam.
|
||
"""
|
||
utilization(d::StageDelta, window) =
|
||
(d.busy - d.blocked) / max(window * d.workers, 1e-9)
|
||
blocked_share(d::StageDelta) = d.busy <= 0 ? 0.0 : d.blocked / d.busy
|
||
|
||
"Subtract two scrapes into per-stage deltas, folding in sampled peak depths."
|
||
function stage_deltas(before, after, peak_depth::Dict{Int,Int})
|
||
out = StageDelta[]
|
||
for (b, a) in zip(before.stages, after.stages)
|
||
push!(out, StageDelta(a.stage, String(a.name), a.workers,
|
||
a.completed - b.completed,
|
||
a.failed - b.failed,
|
||
a.bytes - b.bytes,
|
||
a.busy_seconds - b.busy_seconds,
|
||
a.blocked_seconds - b.blocked_seconds,
|
||
get(peak_depth, Int(a.stage), 0),
|
||
a.queue_capacity))
|
||
end
|
||
return out
|
||
end
|
||
|
||
"JSON has no NaN. A stage that completed nothing has no service time, and `null`
|
||
is the honest way to say that; writing NaN just makes JSON3 throw."
|
||
json_num(x::Real) = isfinite(x) ? x : nothing
|
||
|
||
pad(s, n) = rpad(string(s), n)
|
||
lpad_(s, n) = lpad(string(s), n)
|
||
|
||
"""
|
||
Print the per-stage table and say which stage is the bottleneck.
|
||
|
||
The verdict reads utilization, not throughput: in a pipeline every stage
|
||
completes the same files, so at steady state they all report nearly the same
|
||
files/s regardless of which one is the constraint. What separates them is how
|
||
hard each pool had to work to keep up: the bottleneck is pinned near 1.0 while
|
||
its neighbours idle.
|
||
"""
|
||
function stage_report(deltas::Vector{StageDelta}, window::Float64)
|
||
println("PER-STAGE (server counters, delta over the end-to-end window)")
|
||
println(" stage files/s MiB/s svc ms util blocked peak queue failed")
|
||
for d in deltas
|
||
svc = service_ms(d)
|
||
println(" $(d.stage) $(pad(d.name, 10)) " *
|
||
lpad_(fmt(files_per_sec(d, window)), 8) * " " *
|
||
lpad_(fmt(mib_per_sec(d, window)), 8) * " " *
|
||
lpad_(isnan(svc) ? "—" : fmt(svc, 1), 8) * " " *
|
||
lpad_(fmt(utilization(d, window)), 6) * " " *
|
||
lpad_(fmt(blocked_share(d) * 100, 0) * "%", 7) * " " *
|
||
lpad_("$(d.peak_depth)/$(d.capacity)", 11) * " " *
|
||
lpad_(d.failed, 7))
|
||
end
|
||
|
||
# Per-stage files/s are not comparable across rows and saying so costs one
|
||
# line: the stages process different subsets (stage 2 only known files,
|
||
# stage 4 only text), so a low rate can mean "little work arrived here"
|
||
# rather than "slow". Utilization is the column that compares.
|
||
println(" (files/s counts only files routed to that stage; svc is per-file wall time in")
|
||
println(" one worker; util = (busy − blocked) / (window × workers))")
|
||
|
||
worked = filter(d -> d.completed > 0, deltas)
|
||
if isempty(worked)
|
||
println(" (no stage completed a file in this window)")
|
||
return nothing
|
||
end
|
||
top = argmax(d -> utilization(d, window), worked)
|
||
println(" bottleneck stage $(top.stage) ($(top.name)) at " *
|
||
"$(fmt(utilization(top, window) * 100, 0))% utilization of " *
|
||
"$(top.workers) worker(s)")
|
||
if utilization(top, window) < 0.5
|
||
println(" ...but no stage is near saturated: the pipeline is " *
|
||
"waiting on intake,\n not on itself. Raise --concurrency " *
|
||
"or --files to load it properly.")
|
||
end
|
||
for d in worked
|
||
blocked_share(d) > 0.25 && println(" ! stage $(d.stage) ($(d.name)) spent " *
|
||
"$(fmt(blocked_share(d) * 100, 0))% of its time parked on a full downstream " *
|
||
"queue.\n It is being held up by the stage after it, not doing that work itself.")
|
||
end
|
||
return nothing
|
||
end
|
||
|
||
# ------------------------------------------------------------------------ corpus
|
||
|
||
const WORDS = split("the quick brown fox jumps over a lazy dog while parsing " *
|
||
"headers and spooling bytes onto disk for later enrichment " *
|
||
"because throughput matters more than latency here")
|
||
|
||
"Write one file of exactly `size` bytes, in bounded chunks so the generator
|
||
itself never holds a whole 1 GB file in memory."
|
||
function write_file(path::AbstractString, size::Int, kind::Symbol, rng)
|
||
chunk = 1024 * 1024
|
||
open(path, "w") do io
|
||
remaining = size
|
||
while remaining > 0
|
||
n = min(chunk, remaining)
|
||
if kind === :binary
|
||
write(io, rand(rng, UInt8, n))
|
||
else
|
||
buf = IOBuffer()
|
||
while buf.size < n
|
||
print(buf, rand(rng, WORDS), rand(rng) < 0.06 ? ".\n" : " ")
|
||
end
|
||
write(io, take!(buf)[1:n])
|
||
end
|
||
remaining -= n
|
||
end
|
||
end
|
||
return nothing
|
||
end
|
||
|
||
"Generate the corpus and return (dir, paths, total_bytes)."
|
||
function make_corpus(opts)
|
||
n, size, kind = opts["files"], opts["size"], opts["kind"]
|
||
totalbytes = n * size
|
||
if totalbytes > 16 * 1024^3 && !opts["force"]
|
||
error("corpus would be $(human(totalbytes)) on disk; pass --force if that's intended")
|
||
end
|
||
dir = mktempdir(; prefix = "fsbench-")
|
||
rng = MersenneTwister(1234)
|
||
paths = String[]
|
||
for i in 1:n
|
||
k = kind == "mixed" ? (isodd(i) ? :binary : :text) :
|
||
kind == "text" ? :text : :binary
|
||
ext = k === :text ? "txt" : "bin"
|
||
path = joinpath(dir, "bench-$(lpad(i, 6, '0')).$ext")
|
||
write_file(path, size, k, rng)
|
||
push!(paths, path)
|
||
end
|
||
return dir, paths, totalbytes
|
||
end
|
||
|
||
function existing_corpus(dir::AbstractString)
|
||
isdir(dir) || error("--corpus is not a directory: $dir")
|
||
paths = sort(filter(isfile, readdir(dir; join = true)))
|
||
isempty(paths) && error("--corpus directory is empty: $dir")
|
||
return paths, sum(filesize, paths)
|
||
end
|
||
|
||
# ----------------------------------------------------------------------- upload
|
||
|
||
struct Upload
|
||
status::Int # HTTP status, or 0 if the request threw
|
||
accepted::Int # jobs the server actually queued (from the 202/503 body)
|
||
seconds::Float64
|
||
end
|
||
|
||
"POST one file as multipart/form-data and report what the server accepted."
|
||
function upload_one(url::String, path::String)::Upload
|
||
t0 = time()
|
||
try
|
||
form = HTTP.Form(["file" => HTTP.Multipart(basename(path), open(path, "r"),
|
||
"application/octet-stream")])
|
||
resp = HTTP.post(url, [], form; status_exception = false, retry = false)
|
||
# 202 and 503 both carry an `accepted` array: a partially-accepted batch
|
||
# still queued those jobs, and they will show up in the sinks.
|
||
acc = try
|
||
length(JSON3.read(String(resp.body)).accepted)
|
||
catch
|
||
resp.status == 202 ? 1 : 0
|
||
end
|
||
return Upload(resp.status, acc, time() - t0)
|
||
catch e
|
||
e isa InterruptException && rethrow()
|
||
@warn "upload failed" file = basename(path) exception = e
|
||
return Upload(0, 0, time() - t0)
|
||
end
|
||
end
|
||
|
||
"""
|
||
Upload every path, at most `concurrency` in flight.
|
||
|
||
A bounded set of worker tasks pulling from a shared index keeps exactly
|
||
`concurrency` requests in flight for the whole run, unlike batching, where each
|
||
batch stalls on its slowest (largest) file and the real concurrency sags.
|
||
"""
|
||
function upload_all(url::String, paths::Vector{String}, concurrency::Int)
|
||
endpoint = string(rstrip(url, '/'), "/upload")
|
||
results = Vector{Upload}(undef, length(paths))
|
||
next = Threads.Atomic{Int}(1)
|
||
@sync for _ in 1:min(concurrency, length(paths))
|
||
Threads.@spawn while true
|
||
i = Threads.atomic_add!(next, 1)
|
||
i > length(paths) && break
|
||
results[i] = upload_one(endpoint, paths[i])
|
||
end
|
||
end
|
||
return results
|
||
end
|
||
|
||
# ------------------------------------------------------------------- formatting
|
||
|
||
function human(bytes::Real)
|
||
b = Float64(bytes)
|
||
for unit in ("B", "KiB", "MiB", "GiB", "TiB")
|
||
(abs(b) < 1024 || unit == "TiB") && return "$(round(b; digits = 2)) $unit"
|
||
b /= 1024
|
||
end
|
||
end
|
||
|
||
fmt(x::Real, digits::Int = 2) = string(round(Float64(x); digits = digits))
|
||
|
||
function percentile(sorted::Vector{Float64}, p::Float64)
|
||
isempty(sorted) && return NaN
|
||
idx = clamp(ceil(Int, p * length(sorted)), 1, length(sorted))
|
||
return sorted[idx]
|
||
end
|
||
|
||
# ------------------------------------------------------------------------- main
|
||
|
||
function main(argv)
|
||
opts = parse_args(argv)
|
||
url = string(opts["url"])
|
||
|
||
# Fail fast and clearly if there's no server, rather than reporting a run of zeros.
|
||
try
|
||
HTTP.get(string(rstrip(url, '/'), "/health"); retry = false, readtimeout = 5)
|
||
catch e
|
||
println(stderr, "cannot reach $url/health. Is the server running?")
|
||
println(stderr, " start it with: julia --project=. -t auto bin/server.jl")
|
||
return 1
|
||
end
|
||
|
||
sinks, stages = sinkdirs(), stagedirs()
|
||
|
||
# Leftovers mid-pipeline would land in the sinks during our window and be
|
||
# counted as our throughput, so say so up front rather than quietly skewing.
|
||
pending = total(counts(stages))
|
||
pending > 0 && @warn "pipeline is not idle: $pending file(s) still in spool/; " *
|
||
"throughput will include their completions"
|
||
|
||
# --- corpus
|
||
generated = opts["corpus"] === nothing
|
||
corpusdir, paths, corpusbytes = if generated
|
||
print("generating corpus: $(opts["files"]) × $(human(opts["size"])) ($(opts["kind"]))… ")
|
||
t = time()
|
||
d, p, b = make_corpus(opts)
|
||
println("done in $(fmt(time() - t))s → $d")
|
||
d, p, b
|
||
else
|
||
p, b = existing_corpus(String(opts["corpus"]))
|
||
println("using corpus: $(length(p)) file(s), $(human(b)) from $(opts["corpus"])")
|
||
String(opts["corpus"]), p, b
|
||
end
|
||
|
||
try
|
||
pid = opts["no-mem"] ? nothing : something(opts["pid"], detect_pid(), Some(nothing))
|
||
if pid === nothing && !opts["no-mem"]
|
||
@warn "could not find the server process; skipping memory (pass --pid PID)"
|
||
end
|
||
|
||
baseline_rss = nothing
|
||
peak_reset = false
|
||
if pid !== nothing
|
||
r = read_rss(pid)
|
||
r === nothing && (@warn "cannot read /proc/$pid/status; skipping memory"; pid = nothing)
|
||
if pid !== nothing
|
||
peak_reset = reset_peak_rss(pid)
|
||
peak_reset || @warn "could not reset the peak-RSS counter (/proc/$pid/clear_refs); " *
|
||
"reporting sampled RSS only"
|
||
# Baseline is read *after* the reset, not before. VmHWM restarts
|
||
# from whatever RSS is at the moment of the reset, so a baseline
|
||
# sampled earlier is measured against a different origin, and if
|
||
# the GC hands memory back in between, the run reports negative
|
||
# growth, which is nonsense on its face.
|
||
r2 = read_rss(pid)
|
||
baseline_rss = r2 === nothing ? r[1] : r2[1]
|
||
end
|
||
end
|
||
|
||
if Threads.nthreads() == 1 && opts["size"] > 64 * 1024^2
|
||
@warn "running with 1 thread and large files: the sampler shares a thread with " *
|
||
"blocking file reads, so the RSS curve will be coarse. Prefer -t auto."
|
||
end
|
||
|
||
base_sinks = counts(sinks)
|
||
interval = opts["sample-ms"] / 1000
|
||
|
||
# --- per-stage counters: the "before" half of the delta.
|
||
stats_before = opts["no-stats"] ? nothing : scrape_stats(url)
|
||
if stats_before === nothing && !opts["no-stats"]
|
||
@warn "no /stats endpoint on this server; skipping the per-stage table " *
|
||
"(the server predates src/stats.jl)"
|
||
end
|
||
|
||
# --- sampler: RSS curve, spool depth, and queue depths.
|
||
stop = Threads.Atomic{Bool}(false)
|
||
rss_samples = Float64[]
|
||
depth_max = Dict(k => 0 for k in keys(stages))
|
||
queue_peak = Dict{Int,Int}()
|
||
sampler = Threads.@spawn begin
|
||
while !stop[]
|
||
if pid !== nothing
|
||
r = read_rss(pid)
|
||
r !== nothing && push!(rss_samples, Float64(r[1]))
|
||
end
|
||
d = counts(stages)
|
||
for k in keys(d)
|
||
depth_max[k] = max(depth_max[k], getfield(d, k))
|
||
end
|
||
# Queue depth, unlike directory depth, can't be missed by a slow
|
||
# sample in the same way, since a file's *reference* sits in the
|
||
# queue for the whole time it waits, so this is the depth the stage
|
||
# table reports.
|
||
if stats_before !== nothing
|
||
s = scrape_stats(url)
|
||
s === nothing || for st in s.stages
|
||
k = Int(st.stage)
|
||
queue_peak[k] = max(get(queue_peak, k, 0), Int(st.queue_depth))
|
||
end
|
||
end
|
||
sleep(interval)
|
||
end
|
||
end
|
||
|
||
# --- intake
|
||
println("uploading $(length(paths)) file(s) at concurrency $(opts["concurrency"])…")
|
||
t_start = time()
|
||
ups = upload_all(url, paths, opts["concurrency"])
|
||
t_intake_end = time()
|
||
|
||
accepted = sum(u.accepted for u in ups)
|
||
n_202 = count(u -> u.status == 202, ups)
|
||
n_503 = count(u -> u.status == 503, ups)
|
||
n_err = count(u -> !(u.status in (202, 503)), ups)
|
||
intake_secs = t_intake_end - t_start
|
||
lat = sort([u.seconds for u in ups])
|
||
|
||
println(" intake: $accepted job(s) accepted in $(fmt(intake_secs))s " *
|
||
"($(fmt(accepted / max(intake_secs, 1e-9))) files/s, " *
|
||
"$(fmt(corpusbytes / max(intake_secs, 1e-9) / 1024^2)) MiB/s)")
|
||
n_503 > 0 && println(" backpressure: $n_503 request(s) got 503 (intake queue full)")
|
||
n_err > 0 && println(" errors: $n_err request(s) failed or returned an unexpected status")
|
||
|
||
# --- drain: poll the terminal sinks until they stop moving.
|
||
println("draining (polling sinks every $(opts["sample-ms"])ms)…")
|
||
completed = 0
|
||
t_last_progress = time()
|
||
t_last_completion = t_intake_end
|
||
timed_out = false
|
||
while completed < accepted
|
||
sleep(interval)
|
||
c = total(deltas(counts(sinks), base_sinks))
|
||
if c > completed
|
||
completed = c
|
||
t_last_completion = time()
|
||
t_last_progress = t_last_completion
|
||
elseif time() - t_last_progress > opts["timeout"]
|
||
timed_out = true
|
||
break
|
||
end
|
||
end
|
||
# Scrape before stopping the sampler, so the window closes as near the
|
||
# last completion as we can manage.
|
||
stats_after = stats_before === nothing ? nothing : scrape_stats(url)
|
||
stop[] = true
|
||
wait(sampler)
|
||
|
||
sink_delta = deltas(counts(sinks), base_sinks)
|
||
e2e_secs = t_last_completion - t_start
|
||
|
||
# The stage window is the server's own clock across the two scrapes, not
|
||
# e2e_secs: it starts a scrape earlier and ends a scrape later, and using
|
||
# our wall time against its counters would misattribute the difference.
|
||
stage_window, stage_delta = if stats_after === nothing
|
||
(0.0, StageDelta[])
|
||
else
|
||
(Float64(stats_after.now - stats_before.now),
|
||
stage_deltas(stats_before, stats_after, queue_peak))
|
||
end
|
||
|
||
final_rss = pid === nothing ? nothing : read_rss(pid)
|
||
peak_rss = if final_rss !== nothing && peak_reset
|
||
final_rss[2] # kernel VmHWM: catches spikes between samples
|
||
elseif !isempty(rss_samples)
|
||
maximum(rss_samples)
|
||
else
|
||
nothing
|
||
end
|
||
|
||
# --- report
|
||
println()
|
||
println("=" ^ 68)
|
||
println("corpus $(length(paths)) file(s), $(human(corpusbytes)) total, " *
|
||
"$(human(corpusbytes / length(paths))) avg")
|
||
println("concurrency $(opts["concurrency"])")
|
||
println()
|
||
println("INTAKE (HTTP 202: bytes spooled, not processed)")
|
||
println(" accepted $accepted of $(length(paths)) [202: $n_202, 503: $n_503, error: $n_err]")
|
||
println(" wall $(fmt(intake_secs))s")
|
||
println(" rate $(fmt(accepted / max(intake_secs, 1e-9))) files/s, " *
|
||
"$(fmt(corpusbytes / max(intake_secs, 1e-9) / 1024^2)) MiB/s")
|
||
println(" latency p50 $(fmt(percentile(lat, 0.5) * 1000, 1))ms " *
|
||
"p95 $(fmt(percentile(lat, 0.95) * 1000, 1))ms " *
|
||
"max $(fmt(percentile(lat, 1.0) * 1000, 1))ms")
|
||
println()
|
||
println("END-TO-END (files reaching a terminal sink)")
|
||
println(" completed $completed of $accepted accepted" * (timed_out ? " ** TIMED OUT **" : ""))
|
||
println(" wall $(fmt(e2e_secs))s (first upload → last completion)")
|
||
println(" throughput $(fmt(completed / max(e2e_secs, 1e-9))) files/s, " *
|
||
"$(fmt(corpusbytes / max(e2e_secs, 1e-9) / 1024^2)) MiB/s")
|
||
println(" sinks done $(sink_delta.done) text_done $(sink_delta.text_done) " *
|
||
"binary $(sink_delta.binary) failed $(sink_delta.failed)")
|
||
println(" peak dir depth spool $(depth_max[:spool]) known $(depth_max[:known]) " *
|
||
"unknown $(depth_max[:unknown]) text $(depth_max[:text])")
|
||
println()
|
||
if !isempty(stage_delta)
|
||
stage_report(stage_delta, stage_window)
|
||
println()
|
||
end
|
||
if peak_rss !== nothing
|
||
println("SERVER MEMORY (pid $pid)")
|
||
println(" baseline RSS $(human(baseline_rss))")
|
||
println(" peak RSS $(human(peak_rss))" *
|
||
(peak_reset ? " (kernel VmHWM, reset at start)" : " (sampled; may miss spikes)"))
|
||
growth = peak_rss - baseline_rss
|
||
if growth <= 0
|
||
# RSS never got back to where it started, so the run's own cost
|
||
# is below the noise floor of the server settling after startup.
|
||
# Printing a negative "growth" would invite reading a memory
|
||
# *saving* into what is really "too small to measure here".
|
||
println(" growth none measurable (peak never exceeded the baseline)")
|
||
println(" The baseline was still falling when we sampled it. Give the")
|
||
println(" server ~30s to settle after startup for a comparable figure.")
|
||
else
|
||
println(" growth $(human(growth))")
|
||
println(" per in-flight $(human(growth / opts["concurrency"])) " *
|
||
"at $(human(corpusbytes / length(paths))) avg file size")
|
||
println(" (should not grow with file size: intake streams to disk)")
|
||
end
|
||
# Julia's GC returns memory to the OS lazily, so a second run on the
|
||
# same process starts from an inflated baseline and under-reports
|
||
# growth. Absolute peak is the number to trust across runs.
|
||
baseline_rss > 1.3 * FRESH_RSS_HINT &&
|
||
println(" ! baseline is well above a fresh start ($(human(FRESH_RSS_HINT))): the GC " *
|
||
"has not\n returned memory from earlier work. Compare " *
|
||
"absolute peak, or restart\n the server for a clean growth figure.")
|
||
else
|
||
println("SERVER MEMORY not sampled")
|
||
end
|
||
println("=" ^ 68)
|
||
|
||
sink_delta.failed > 0 &&
|
||
println("\nnote: $(sink_delta.failed) file(s) landed in $(sinks.failed); check the server log.")
|
||
timed_out &&
|
||
println("\nnote: drain stalled with $(accepted - completed) file(s) outstanding. " *
|
||
"Check the server log and spool/; raise --timeout if the pipeline is just slow.")
|
||
|
||
if opts["json"] !== nothing
|
||
result = (
|
||
url, concurrency = opts["concurrency"], kind = opts["kind"],
|
||
files = length(paths), corpus_bytes = corpusbytes,
|
||
avg_file_bytes = corpusbytes / length(paths),
|
||
intake = (; accepted, n_202, n_503, n_err, seconds = intake_secs,
|
||
files_per_sec = accepted / max(intake_secs, 1e-9),
|
||
p50_ms = json_num(percentile(lat, 0.5) * 1000),
|
||
p95_ms = json_num(percentile(lat, 0.95) * 1000),
|
||
max_ms = json_num(percentile(lat, 1.0) * 1000)),
|
||
end_to_end = (; completed, seconds = e2e_secs, timed_out,
|
||
files_per_sec = completed / max(e2e_secs, 1e-9),
|
||
mib_per_sec = corpusbytes / max(e2e_secs, 1e-9) / 1024^2,
|
||
sinks = sink_delta, peak_stage_depth = depth_max),
|
||
stages = [(; stage = d.stage, name = d.name, workers = d.workers,
|
||
completed = d.completed, failed = d.failed, bytes = d.bytes,
|
||
busy_seconds = d.busy, blocked_seconds = d.blocked,
|
||
window_seconds = stage_window,
|
||
files_per_sec = files_per_sec(d, stage_window),
|
||
mib_per_sec = mib_per_sec(d, stage_window),
|
||
service_ms = json_num(service_ms(d)),
|
||
utilization = utilization(d, stage_window),
|
||
blocked_share = blocked_share(d),
|
||
peak_queue_depth = d.peak_depth,
|
||
queue_capacity = d.capacity) for d in stage_delta],
|
||
memory = (; pid, baseline_rss, peak_rss, peak_is_kernel_hwm = peak_reset,
|
||
growth = peak_rss === nothing ? nothing : peak_rss - baseline_rss,
|
||
samples = rss_samples),
|
||
)
|
||
open(String(opts["json"]), "w") do io
|
||
JSON3.write(io, result)
|
||
end
|
||
println("\nwrote $(opts["json"])")
|
||
end
|
||
|
||
return timed_out ? 1 : 0
|
||
finally
|
||
if generated && !opts["keep-corpus"]
|
||
rm(corpusdir; recursive = true, force = true)
|
||
elseif generated
|
||
println("\nkept corpus: $corpusdir")
|
||
end
|
||
end
|
||
end
|
||
|
||
if abspath(PROGRAM_FILE) == @__FILE__
|
||
exit(main(ARGS))
|
||
end
|