diff --git a/README.md b/README.md index ea2c3bb..90eadcf 100644 --- a/README.md +++ b/README.md @@ -447,6 +447,9 @@ curl http://127.0.0.1:8080/health # upload one or more files (multipart/form-data) curl -F "a=@report.pdf" -F "b=@data.csv" http://127.0.0.1:8080/upload # 202 {"accepted":[{"id":"","name":"report.pdf"}, ...]} + +# per-stage counters +curl http://127.0.0.1:8080/stats ``` Each file in a request becomes its own job. Responses: @@ -456,19 +459,93 @@ Each file in a request becomes its own job. Responses: - `503 Service Unavailable` — queue full, retry later - `500 Internal Server Error` — failed to write a file to disk +### `GET /stats` — per-stage counters + +The pipeline counts its own work (`src/stats.jl`), because nothing outside it +can: `known/`, `unknown/` and `text/` are *transient*, so a file can cross one +between two directory polls and an external sampler will miss exactly the stages +you most want to measure. + +```jsonc +{ + "now": 1785725300.5, "since": 1785725291.9, "uptime_seconds": 8.5, + "intake": { "requests": 400, "files": 400, "bytes": 6553600, "rejected": 0 }, + "stages": [ + { "stage": 4, "name": "language", "workers": 16, + "queue_depth": 184, "queue_capacity": 1000, + "completed": 200, "failed": 0, "bytes": 3276800, + "busy_seconds": 170.5, // summed handler time across the pool + "blocked_seconds": 0.0, // of that, time parked on a full downstream queue + "in_flight": 3 } + ] +} +``` + +Counters are monotonic since startup, Prometheus-style — rates are the reader's +job, so a scrape is stateless and two readers can't disturb each other. Take two +scrapes Δt apart and subtract: + +``` +throughput = Δcompleted / Δt +utilization = (Δbusy_seconds − Δblocked_seconds) / (Δt × workers) +``` + +**Utilization is the number that names the bottleneck.** In a pipeline every +stage completes the same files, so at steady state they all report near-identical +files/s no matter which one is the constraint; what separates them is how hard +each pool worked to keep up. The bottleneck sits near 1.0 with its queue backing +up while its neighbours idle. + +`blocked_seconds` is what keeps that true. Stages 1 and 3 apply *blocking* +backpressure — a full downstream queue means the handler parks and retries rather +than dropping the file — and that wait happens inside the handler. Counting it as +busy would pin stage 1 at 1.0 whenever stage 2 is the real jam, making every +stage upstream of a jam look like the jam. Subtracted out, utilization means +"doing its own work", and a high blocked share becomes its own signal: a stage +blocked 90% of the time is naming its successor. + +The counters are a handful of atomic adds per file, recorded in `worker_loop` — +the one place every stage's work passes through, so a new stage is instrumented +the moment it is wired up, and never on the read path. + ## Benchmarking (throughput + memory) -`bin/bench.jl` measures the pipeline against a **running** server. It must run on -the same machine (it reads the sink dirs and `/proc`), and it changes nothing in -`src/` — it only speaks HTTP and counts files. +There are three harnesses. Only the first needs a running server: + +| script | measures | server? | +|---|---|---| +| `bin/bench.jl` (below) | intake, end-to-end and per-stage throughput; server RSS | **yes** | +| [`bin/bench_model.jl`](#model-microbenchmark-binbench_modeljl) | the classifier alone: inference, feature reads, thread scaling | no | +| [`bin/cluster_calibrate.jl`](#unknown-format-discovery-stage-5-offline) | stage-5 clustering quality vs. an NCD baseline | no | + +Running all of them from a clean checkout: ```bash +julia --project=. -e 'using Pkg; Pkg.instantiate()' # once + +# 1. the model, on its own — no server involved +julia --project=. -t auto bin/bench_model.jl + +# 2. the pipeline. Start the server in one terminal… +julia --project=. -t auto bin/server.jl + +# …and drive it from another. Restart the server between memory runs: Julia's +# GC returns memory to the OS lazily, so a second run starts inflated. julia --project=. -t auto bin/bench.jl --files 2000 --size 8k --concurrency 32 julia --project=. -t auto bin/bench.jl --files 2 --size 1g --concurrency 1 # memory julia --project=. -t auto bin/bench.jl --corpus ../training_set --concurrency 16 + +# 3. stage-5 clustering quality (offline, needs a labelled corpus) +julia --project=. bin/cluster_calibrate.jl ../training_set ``` -Two properties of this design dictate how it measures: +The test suite is `julia --project=. -t auto -e 'using Pkg; Pkg.test()'`. + +`bin/bench.jl` must run on the same machine as the server (it reads the sink dirs +and `/proc`); otherwise it only speaks HTTP — `/upload`, `/health`, and `/stats` +for the per-stage numbers. + +Three properties of this design dictate how it measures: - **HTTP latency is not throughput.** `/upload` returns `202` once the bytes are spooled and a reference is enqueued — all four stages run *after* the response, @@ -476,8 +553,30 @@ Two properties of this design dictate how it measures: corpus and polls the terminal sinks (`done/`, `text_done/`, `binary/`, `failed/`) until the count stops moving, and reports both numbers separately: intake rate *and* end-to-end completion rate. It also samples the intermediate - stage dirs, so the peak depth of `spool/`/`known/`/`unknown/`/`text/` names the - bottleneck stage directly. + stage dirs, so the peak depth of `spool/`/`known/`/`unknown/`/`text/` shows + where work piles up. +- **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 in it. The bench scrapes + [`/stats`](#get-stats--per-stage-counters) before and after the run and + subtracts, giving each stage its own throughput, mean service time and + utilization: + + ``` + PER-STAGE (server counters, delta over the end-to-end window) + stage files/s MiB/s svc ms util blocked peak queue failed + 1 classify 32.52 0.51 38.7 0.08 0.0% 209/1000 0 + 2 enrich 0.24 0.0 541.0 0.01 0.0% 0/1000 0 + 3 triage 32.27 0.5 33.9 0.07 0.0% 83/1000 0 + 4 language 16.26 0.25 852.6 0.87 0.0% 184/1000 0 + bottleneck stage 4 (language) at 87.0% utilization of 16 worker(s) + ``` + + (400 mixed files, 16 KiB each, concurrency 16. Stage 4 is the constraint — + `github-linguist` is a process spawn per file at ~850 ms — and the 200 text + files queue up behind it while stages 1 and 3 idle at under 10%.) Read *down* + the util column, not the files/s column: each stage sees a different subset of + the corpus, so a low rate can just mean little work was routed there. - **Memory should be flat in file size, and the sweep is what proves it.** Both halves of the pipeline are bounded: the workers read bounded prefixes (16+16 bytes to classify, 8 KB to sniff, 64 KB to language-detect), and intake streams @@ -486,14 +585,33 @@ Two properties of this design dictate how it measures: | upload size | concurrency | RSS growth | |---|---|---| - | 256 MiB × 4 | 1 | 21.8 MiB | - | 1 GiB × 2 | 1 | 20.8 MiB | - | 2 GiB × 1 | 1 | 17.0 MiB | - | 256 MiB × 4 | 4 | 86.9 MiB (21.7 MiB per in-flight upload) | + | 256 MiB × 4 | 1 | 14.0 / 20.4 MiB (two runs) | + | 1 GiB × 2 | 1 | 22.3 MiB | + | 2 GiB × 1 | 1 | 31.3 MiB | + | 256 MiB × 4 | 4 | none measurable (peak stayed under the baseline) | - Flat across a 8× range of file sizes, and linear in concurrency — which is the - shape to expect. (The residual ~20 MiB per in-flight upload is GC churn from the - chunk reads, not retained buffers; it does not grow with the file.) Before intake + **Read the shape, not the digits.** Growth is flat across an 8× range of file + sizes — 14–31 MiB whether the upload is 256 MiB or 2 GiB — which is the claim + that matters: nothing scales with file size, so nothing is buffering. The + absolute figures are *not* precise to the megabyte. A freshly started server + settles anywhere in an ~860–985 MiB band, so run-to-run variance in the + baseline is comparable to the growth being measured; the residual is GC churn + from the chunk reads, not retained buffers. + + Two consequences worth knowing before quoting these numbers: + + - **Let the server settle ~30s after startup** before a memory run, or the + baseline is sampled mid-fall and the run reports less growth than it caused + (or none at all, as the concurrency-4 row above did). + - **Linear-in-concurrency is not currently demonstrated.** An earlier + measurement put 4 concurrent 256 MiB uploads at 86.9 MiB (21.7 per in-flight + upload), but that run predates a fix to how the baseline was sampled — it was + read *before* the peak counter was reset, against a different origin — and the + re-measurement above cannot reproduce it: the growth sits under the noise + floor. Expect concurrency to cost memory; don't trust a specific coefficient + without a quieter machine or many more runs. + + Before intake was streamed, the same 256 MiB upload grew RSS by ~700 MiB and 4 concurrent ones pushed a 950 MiB baseline past 2 GiB, so **a `--size` sweep that slopes upward is the regression signal** — it means something has started buffering bodies again. @@ -501,7 +619,9 @@ Two properties of this design dictate how it measures: Flags: `--files`, `--size` (`8k`/`64m`/`1g`), `--concurrency`, `--kind` (`binary`/`text`/`mixed` — chooses which stages get loaded), `--corpus DIR` (real files, the only way to exercise stage 2's exiftool path), `--pid`, `--no-mem`, -`--sample-ms`, `--timeout`, `--json PATH`. Full list in the script header. +`--no-stats`, `--sample-ms`, `--timeout`, `--json PATH`. Full list in the script +header. The `--json` output carries the per-stage numbers too, so a sweep can be +compared run to run. Two caveats the script reports rather than hides: it counts sink *deltas*, so it warns if the pipeline isn't idle at the start (in-flight leftovers would be @@ -511,6 +631,53 @@ resets the kernel's peak-RSS counter (`/proc//clear_refs`) per run and flag a drifted baseline, but for a clean growth figure restart the server between memory runs. +### Model microbenchmark (`bin/bench_model.jl`) + +`bin/bench.jl` reports stage 1 as a single number — the wall time of +`handle_classify_job`, which is a feature read, an inference, a rename, a log +line, and whatever contention the other three pools create. That's the right +number for capacity planning and the wrong one for "is the model slow?". +`bin/bench_model.jl` answers that separately, with no server, queue, or HTTP +involved: + +```bash +julia --project=. -t auto bin/bench_model.jl +``` + +Measured on this machine (Ryzen 7 2700X, 8 cores/16 threads, Julia 1.12): + +| what | per file | notes | +|---|---|---| +| `Lux.apply`, batch 1 | **2.3 µs** | 768 B allocated per call | +| `read_features` | **4.2–6.0 µs** | flat across a 262,144× size range (1 KiB → 256 MiB) | +| `classify()` | **8.4 µs** | 73% feature read, 27% inference | + +So the model is **not** the pipeline's problem, by three orders of magnitude: the +same run measured stage 1 at 38.7 ms per file, ~4,500× the 8.4 µs `classify()` +costs. Whatever stage 1 spends its time on, it isn't the network. + +Two findings worth acting on if stage 1 ever *does* become the constraint: + +- **Batching would buy ~13×.** A 32×1 matmul wastes most of a BLAS call: + batch 64 costs 280 ns/file and batch 512 costs 179 ns/file, against 2.34 µs + one at a time. The pipeline classifies strictly one file per job today, so it + pays the worst row in that table. +- **Inference does not scale past ~4 threads.** Concurrent `Lux.apply` on the + shared read-only `Classifier` peaks around 1.2M files/s at 4 tasks and then + *falls back* to single-thread throughput at 16. The script runs a pure-compute + control kernel through the same sweep to place the blame: the control reaches + 14.3× at 16 tasks (90% efficiency) on the same box, so the machine + parallelizes and `Lux.apply` doesn't. GC is only ~1% of it, so allocation + pressure isn't the explanation either — the cause is inside Lux/BLAS and is + not diagnosed here. The practical consequence: raising `FS_WORKERS` past ~4 + adds no classification throughput. + +Reported times are the **minimum** over trials — for a microbenchmark the floor +is the signal and everything above it is scheduler and GC noise — and every timed +loop stores its result in a sink so a pure call can't be hoisted out of the loop. +Flags: `--model`, `--reps`, `--trials`, `--batches`, `--sizes`, `--no-threads`, +`--json PATH`. + ## Layout ``` @@ -519,6 +686,7 @@ src/ config.jl Config struct + env parsing job.jl Job (the queue reference) queue.jl JobQueue seam + in-process ChannelQueue + stats.jl per-stage counters behind GET /stats (throughput, utilization) multipart.jl streaming multipart/form-data reader (intake never buffers a file) spool.jl filename sanitizing, streaming spool/move, startup recovery model.jl NN architecture + byte→feature mapping (shared with trainer) @@ -533,6 +701,7 @@ src/ bin/ server.jl entry point bench.jl throughput + memory harness against a running server + bench_model.jl classifier microbenchmark (inference, feature reads, scaling) train.jl offline training script → model/classifier.jld2 cluster_calibrate.jl offline stage-5 hyperparameter calibration + NCD baseline cluster_sweep.jl stage-5 phase-B runner: sweep binary/, update catalog, write nominations diff --git a/bin/bench.jl b/bin/bench.jl index a4709ca..4974d63 100755 --- a/bin/bench.jl +++ b/bin/bench.jl @@ -11,7 +11,16 @@ # (done/, text_done/, binary/, failed/), not the response rate. We upload a # corpus, then poll the sinks until the file count stops moving. # -# 2. Memory should be flat in file size, and that claim needs checking on two +# 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 — known/, unknown/ and text/ are transient, and a file +# can cross one between two samples. 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 @@ -24,9 +33,10 @@ # over, and a 256 MiB upload grew RSS by ~700 MiB. If a --size sweep ever # slopes upward again, something has started buffering.) # -# This is an external harness: it makes no assumptions about the server beyond -# the HTTP contract and the on-disk sink layout, and requires no changes to src/. -# It must run on the same machine as the server (it reads sink dirs and /proc). +# 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] @@ -41,6 +51,7 @@ # --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 @@ -74,13 +85,14 @@ const DEFAULTS = Dict{String,Any}( "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", "force") +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 @@ -180,14 +192,35 @@ function read_rss(pid::Int) return (rss === nothing || hwm === nothing) ? nothing : (rss, hwm) end -"Find the running server process, or `nothing`." +""" +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//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 - pids = parse.(Int, split(out)) + 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) @@ -210,6 +243,130 @@ function reset_peak_rss(pid::Int)::Bool 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 " * @@ -385,10 +542,16 @@ function main(argv) r = read_rss(pid) r === nothing && (@warn "cannot read /proc/$pid/status; skipping memory"; pid = nothing) if pid !== nothing - baseline_rss = r[1] 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 @@ -400,10 +563,18 @@ function main(argv) base_sinks = counts(sinks) interval = opts["sample-ms"] / 1000 - # --- sampler: RSS curve + stage depths, for bottleneck attribution. + # --- 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, stage dir depths, 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 @@ -414,6 +585,17 @@ function main(argv) 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 — 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 @@ -455,12 +637,25 @@ function main(argv) 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 @@ -493,19 +688,33 @@ function main(argv) "$(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 depth spool $(depth_max[:spool]) known $(depth_max[:known]) " * + println(" peak dir depth spool $(depth_max[:spool]) known $(depth_max[:known]) " * "unknown $(depth_max[:unknown]) text $(depth_max[:text])") - println(" (the stage that backs up is the bottleneck)") 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)")) - println(" growth $(human(peak_rss - baseline_rss))") - println(" per in-flight $(human((peak_rss - baseline_rss) / opts["concurrency"])) " * - "at $(human(corpusbytes / length(paths))) avg file size") - println(" (should not grow with file size — intake streams to disk)") + 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. @@ -531,13 +740,24 @@ function main(argv) 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 = percentile(lat, 0.5) * 1000, - p95_ms = percentile(lat, 0.95) * 1000, - max_ms = percentile(lat, 1.0) * 1000), + 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), diff --git a/bin/bench_model.jl b/bin/bench_model.jl new file mode 100755 index 0000000..071b958 --- /dev/null +++ b/bin/bench_model.jl @@ -0,0 +1,427 @@ +#!/usr/bin/env julia +# +# bench_model.jl — microbenchmark the classifier in isolation, with no server, +# no queue, and no disk in the way. +# +# bin/bench.jl measures the *pipeline*: it reports stage 1 as one number, the +# wall time of `handle_classify_job`, which is feature reads + inference + a +# rename + a log line, under whatever thread contention the other three pools are +# creating. That number is the right one for capacity planning and the wrong one +# for answering "is the model slow?". This script answers that question by taking +# the model apart: +# +# read_features open, read 16 bytes, seek, read 16 bytes, scale +# Lux.apply the network itself, on a feature vector already in memory +# classify both together — what stage 1 actually calls per file +# +# Three properties are worth checking beyond the raw per-file cost: +# +# * Feature reads should be flat in file size. read_features seeks to the tail +# rather than slurping, so a 1 GiB file should cost the same as a 1 KiB one. +# (This is the same claim bin/bench.jl makes about memory, on the CPU axis.) +# * Batching should be much cheaper per file. A 32x1 matmul wastes most of a +# BLAS call; if batch-64 inference is many times cheaper per file, that is +# the headroom a batching stage-1 would buy — worth knowing before building +# one, since today the pipeline classifies strictly one file at a time. +# * Inference should scale across threads. `Classifier` is shared read-only by +# the whole stage-1 pool on the claim that Lux inference is pure. If per-file +# cost degrades as tasks are added, that claim holds but BLAS threading is +# fighting the worker pool, and stage-1 workers are contending, not scaling. +# +# Usage: +# julia --project=. -t auto bin/bench_model.jl [options] +# +# --model PATH classifier artifact (default: $FS_MODEL_PATH or model/classifier.jld2) +# --reps N inference calls per timed trial (default: 20000) +# --trials N timed trials; the minimum is reported (default: 5) +# --batches LIST batch sizes to sweep, comma-separated (default: 1,8,64,512) +# --sizes LIST file sizes for the read_features sweep (default: 1k,64k,4m,256m) +# --no-threads skip the thread-scaling sweep +# --json PATH also write the results as JSON +# +# Reported times are the *minimum* over trials: for a microbenchmark the floor is +# the signal and everything above it is scheduler and GC noise. + +using JSON3 +using Random +using Statistics +using Printf + +# The script is run directly, not as part of the package, so pull in exactly the +# pieces the classifier needs. `Lux`/`JLD2` first — model.jl and classify.jl both +# assume the including scope already has them (see the note at the top of model.jl). +using Lux +using JLD2 +using LinearAlgebra + +const SRC = joinpath(dirname(@__DIR__), "src") +include(joinpath(SRC, "model.jl")) +include(joinpath(SRC, "classify.jl")) + +# ---------------------------------------------------------------- option parsing + +const DEFAULTS = Dict{String,Any}( + "model" => get(ENV, "FS_MODEL_PATH", "model/classifier.jld2"), + "reps" => 20_000, + "trials" => 5, + "batches" => "1,8,64,512", + "sizes" => "1k,64k,4m,256m", + "no-threads" => false, + "json" => nothing, +) + +const FLAGS = ("no-threads",) + +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) + opts = copy(DEFAULTS) + i = 1 + while i <= length(argv) + a = argv[i] + startswith(a, "--") || error("unexpected argument: $a") + 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") + opts[key] = key in ("reps", "trials") ? parse(Int, argv[i+1]) : argv[i+1] + i += 2 + end + return opts +end + +# ------------------------------------------------------------------- measurement + +# Every timed loop stores its result here. Without a visible side effect the +# compiler is free to hoist a pure call out of the loop and we would be timing an +# empty `for`. +const SINK = Ref{Any}(nothing) + +""" + measure(f, reps; trials) -> (ns_per_op, bytes_per_op) + +Time `f` over `reps` calls, `trials` times, and report the fastest trial. + +The first call is thrown away: it pays Julia's JIT compilation, which on a +function this small is orders of magnitude more than the thing being measured. +""" +function measure(f, reps::Int; trials::Int = 5) + SINK[] = f() # warm up (compile), and keep the result + best = Inf + for _ in 1:trials + GC.gc() + t0 = time_ns() + for _ in 1:reps + SINK[] = f() + end + best = min(best, (time_ns() - t0) / reps) + end + bytes = @allocated(SINK[] = f()) # one call, after warmup + return (Float64(best), Float64(bytes)) +end + +# ------------------------------------------------------------------- formatting + +function human_time(ns::Real) + ns < 1_000 && return @sprintf("%.0f ns", ns) + ns < 1_000_000 && return @sprintf("%.2f µs", ns / 1e3) + ns < 1e9 && return @sprintf("%.2f ms", ns / 1e6) + return @sprintf("%.2f s", ns / 1e9) +end + +human_bytes(b::Real) = b < 1024 ? @sprintf("%.0f B", b) : + b < 1024^2 ? @sprintf("%.1f KiB", b / 1024) : + @sprintf("%.1f MiB", b / 1024^2) + +rate(ns::Real) = 1e9 / max(ns, 1e-9) # calls per second + +function human_rate(r::Real) + r >= 1e6 && return @sprintf("%.2fM/s", r / 1e6) + r >= 1e3 && return @sprintf("%.1fk/s", r / 1e3) + return @sprintf("%.0f/s", r) +end + +fmt2(x::Real) = @sprintf("%.2f", x) + +# ------------------------------------------------------------- scaling control + +""" + control_kernel(x) -> Float64 + +Pure arithmetic, no allocation, no library call — deliberately dependent +(each step needs the last) so the compiler can't vectorize it away, and sized to +land in the same microsecond neighbourhood as one `Lux.apply`. +""" +function control_kernel(x::Float64) + a = x + @inbounds for i in 1:600 + a = sqrt(a + i) + end + return a +end + +""" + control_scaling(trials) -> Vector + +Run `control_kernel` over the same task counts as the model sweep. This is the +machine's own ceiling for perfectly parallel work: if the control scales and the +model doesn't, the shortfall is the model's, and no amount of `FS_WORKERS` will +recover it. +""" +function control_scaling(trials::Int) + rows = [] + base = 0.0 + per_task = 200_000 + for k in unique([1; 2; 4; 8; Threads.nthreads()]) + k > Threads.nthreads() && continue + best = Inf + for _ in 1:trials + GC.gc() + t0 = time_ns() + @sync for _ in 1:k + Threads.@spawn begin + local acc = 0.0 + for i in 1:per_task + acc += control_kernel(i % 97 + 1.0) + end + SINK[] = acc + end + end + best = min(best, Float64(time_ns() - t0)) + end + r = k * per_task / (best / 1e9) + k == 1 && (base = r) + push!(rows, (; tasks = k, ops_per_sec = r, speedup = r / base)) + end + return rows +end + +# ------------------------------------------------------------------------- corpus + +""" +Write a file of exactly `size` random bytes, in bounded chunks. + +Content is random rather than zeros so the classifier sees a realistic input — +and so the filesystem can't cheat with a sparse file, which would make the tail +`seek` unrepresentatively fast. +""" +function write_file(path::AbstractString, size::Int, rng) + chunk = 1024 * 1024 + open(path, "w") do io + remaining = size + while remaining > 0 + n = min(chunk, remaining) + write(io, rand(rng, UInt8, n)) + remaining -= n + end + end + return path +end + +# ------------------------------------------------------------------------- main + +function main(argv) + opts = parse_args(argv) + modelpath = String(opts["model"]) + isfile(modelpath) || (println(stderr, "model artifact not found: $modelpath"); return 1) + + reps, trials = opts["reps"], opts["trials"] + batches = [parse(Int, s) for s in split(String(opts["batches"]), ",")] + sizes = [parse_size(s) for s in split(String(opts["sizes"]), ",")] + + clf = load_classifier(modelpath) + println("model $modelpath") + println("architecture $(FEATURE_DIM) → 64 → 16 → 2 (Dense/relu, raw logits)") + println("julia threads $(Threads.nthreads()) BLAS threads $(BLAS.get_num_threads())") + println("timing min of $trials trials × $reps reps") + println("=" ^ 72) + + results = Dict{String,Any}() + + # --- 1. inference alone, one file at a time: the number the pipeline pays. + x1 = rand(Float32, FEATURE_DIM, 1) + infer_ns, infer_bytes = measure(reps; trials) do + Lux.apply(clf.model, x1, clf.ps, clf.st) + end + println() + println("INFERENCE (Lux.apply, batch 1 — features already in memory)") + println(" per call $(human_time(infer_ns)) $(human_rate(rate(infer_ns)))") + println(" allocations $(human_bytes(infer_bytes)) per call") + results["inference_batch1"] = (; ns = infer_ns, bytes = infer_bytes, per_sec = rate(infer_ns)) + + # --- 2. batching: how much of that is per-call overhead rather than math? + println() + println("INFERENCE BATCHED (same net, N files per apply)") + println(" batch per batch per file files/s speedup") + batch_rows = [] + for b in batches + xb = rand(Float32, FEATURE_DIM, b) + ns, _ = measure(max(1, reps ÷ b); trials) do + Lux.apply(clf.model, xb, clf.ps, clf.st) + end + per_file = ns / b + @printf(" %8d %13s %12s %13s %7.1fx\n", + b, human_time(ns), human_time(per_file), + human_rate(rate(per_file)), infer_ns / per_file) + push!(batch_rows, (; batch = b, ns_per_batch = ns, ns_per_file = per_file, + files_per_sec = rate(per_file), speedup = infer_ns / per_file)) + end + results["batched"] = batch_rows + println(" (a large speedup is headroom a batching stage 1 could claim; the pipeline") + println(" classifies one file per job today, so it pays the batch-1 row above)") + + # --- 3. feature reads: should be flat in file size (seek, not slurp). + println() + println("FEATURE READS (read_features: 16 head + 16 tail bytes, scaled)") + println(" file size per call calls/s allocations") + read_rows = [] + dir = mktempdir(; prefix = "fsmodel-") + try + rng = MersenneTwister(1234) + for sz in sizes + path = write_file(joinpath(dir, "f-$sz.bin"), sz, rng) + # Fewer reps for the big files: this touches the page cache, and the + # point is the shape of the curve, not another digit of precision. + r = max(200, reps ÷ 20) + ns, bytes = measure(() -> read_features(path), r; trials) + @printf(" %13s %14s %14s %14s\n", + human_bytes(sz), human_time(ns), human_rate(rate(ns)), human_bytes(bytes)) + push!(read_rows, (; size_bytes = sz, ns, bytes, per_sec = rate(ns))) + end + finally + rm(dir; recursive = true, force = true) + end + results["read_features"] = read_rows + flat = length(read_rows) > 1 ? + maximum(r.ns for r in read_rows) / minimum(r.ns for r in read_rows) : 1.0 + @printf(" spread across a %.0fx size range: %.1fx — %s\n", + maximum(sizes) / minimum(sizes), flat, + flat < 3 ? "flat, as designed (it seeks to the tail)" : + "NOT flat: something is reading more than 32 bytes") + + # --- 4. classify(): what stage 1 calls, I/O and inference together. + println() + println("CLASSIFY (read_features + Lux.apply — one whole stage-1 file)") + dir2 = mktempdir(; prefix = "fsmodel-") + classify_ns = 0.0 + try + path = write_file(joinpath(dir2, "sample.bin"), 64 * 1024, MersenneTwister(7)) + classify_ns, classify_bytes = measure(() -> classify(clf, path), max(200, reps ÷ 20); trials) + println(" per file $(human_time(classify_ns)) $(human_rate(rate(classify_ns)))") + println(" allocations $(human_bytes(classify_bytes)) per file") + @printf(" split %.0f%% feature read, %.0f%% inference\n", + 100 * (classify_ns - infer_ns) / classify_ns, 100 * infer_ns / classify_ns) + results["classify"] = (; ns = classify_ns, bytes = classify_bytes, per_sec = rate(classify_ns)) + finally + rm(dir2; recursive = true, force = true) + end + + # --- 5. thread scaling: does the shared read-only Classifier actually scale? + if !opts["no-threads"] && Threads.nthreads() > 1 + println() + println("THREAD SCALING (concurrent Lux.apply on the one shared Classifier)") + println(" tasks files/s per file speedup efficiency GC") + thread_rows = [] + base = 0.0 + for k in unique([1; 2; 4; 8; Threads.nthreads()]) + k > Threads.nthreads() && continue + # Work per task is held *constant* as tasks are added, so total work + # scales with `k`. Splitting a fixed total instead would shrink each + # task as the pool grows until `@spawn`/`@sync` overhead dominated, + # and the resulting curve would show a collapse that is the + # measurement's fault rather than the model's. + per_task = max(reps, 20_000) + # Each task gets its own input so we measure the model, not cache + # line ping-pong on a shared buffer. + xs = [rand(Float32, FEATURE_DIM, 1) for _ in 1:k] + best, best_gc = Inf, 0.0 + for _ in 1:trials + GC.gc() + # Julia's GC stops the world, so it is the one cost that cannot + # be parallelized away: measuring its share here is what turns a + # bad efficiency number into a diagnosis (see the note below). + gc0 = Base.gc_num().total_time + t0 = time_ns() + @sync for t in 1:k + Threads.@spawn begin + local acc = 0.0f0 + for _ in 1:per_task + y, _ = Lux.apply(clf.model, xs[t], clf.ps, clf.st) + acc += y[1] # consume the result + end + SINK[] = acc + end + end + elapsed = Float64(time_ns() - t0) + if elapsed < best + best = elapsed + best_gc = Float64(Base.gc_num().total_time - gc0) + end + end + files = k * per_task + fps = files / (best / 1e9) + k == 1 && (base = fps) + @printf(" %8d %13s %11s %7.2fx %10.0f%% %5.0f%%\n", + k, human_rate(fps), human_time(best / files), fps / base, + 100 * fps / base / k, 100 * best_gc / best) + push!(thread_rows, (; tasks = k, files_per_sec = fps, + ns_per_file = best / files, speedup = fps / base, + gc_fraction = best_gc / best)) + end + results["thread_scaling"] = thread_rows + + # A poor scaling curve has two possible authors — the model or the box — + # and the table alone can't tell them apart. So run the same sweep on a + # kernel that is pure arithmetic with no allocation and no library + # underneath: whatever *it* achieves is this machine's ceiling for + # embarrassingly parallel work, and the gap between the two curves is + # the part that belongs to Lux.apply. + ctrl = control_scaling(trials) + results["control_scaling"] = ctrl + top = last(ctrl) + println(" control pure-compute kernel, same sweep: " * + "$(fmt2(top.speedup))x at $(top.tasks) tasks " * + "($(round(Int, 100 * top.speedup / top.tasks))% efficiency)") + model_top = last(thread_rows) + if model_top.speedup < 0.6 * top.speedup + println(" → the machine parallelizes; Lux.apply does not. Stage-1") + println(" workers past ~4 buy little, whatever FS_WORKERS says.") + else + println(" → inference tracks the machine's own scaling ceiling.") + end + gc_top = maximum(r.gc_fraction for r in thread_rows) + gc_top > 0.15 && println(" ! GC is $(round(Int, 100 * gc_top))% of the " * + "worst case: apply allocates per call, and\n" * + " collection stops every thread.") + end + + println() + println("=" ^ 72) + println("Stage 1's cost per file in bin/bench.jl is this classify() figure plus a") + println("rename, a log line, and whatever contention the other three pools create.") + println("A large gap between the two is pipeline overhead, not the model.") + + if opts["json"] !== nothing + results["meta"] = (; model = modelpath, feature_dim = FEATURE_DIM, + julia_threads = Threads.nthreads(), + blas_threads = BLAS.get_num_threads(), + reps, trials) + open(String(opts["json"]), "w") do io + JSON3.write(io, results) + end + println("\nwrote $(opts["json"])") + end + return 0 +end + +if abspath(PROGRAM_FILE) == @__FILE__ + exit(main(ARGS)) +end diff --git a/src/FileServer.jl b/src/FileServer.jl index 75a4dd8..4ba6f4a 100644 --- a/src/FileServer.jl +++ b/src/FileServer.jl @@ -14,6 +14,7 @@ include("multipart.jl") # streaming multipart reader (defines UPLOAD_CHUNK_BYTES include("config.jl") include("job.jl") include("queue.jl") +include("stats.jl") # per-stage counters behind GET /stats (needs Config/Job/JobQueue) include("spool.jl") include("model.jl") # build_model() + read_features(); shared with bin/train.jl include("classify.jl") # Classifier + load_classifier/classify (needs model.jl) @@ -125,16 +126,29 @@ function run(; overrides...) recovered_text = recover_dir!(cfg.text_dir, text_queue) @info "starting file-server" host=cfg.host port=cfg.port workers=cfg.worker_count known_workers=cfg.known_worker_count unknown_workers=cfg.unknown_worker_count text_workers=cfg.text_worker_count capacity=cfg.queue_capacity known_capacity=cfg.known_queue_capacity unknown_capacity=cfg.unknown_queue_capacity text_capacity=cfg.text_queue_capacity recovered=recovered recovered_known=recovered_known recovered_unknown=recovered_unknown recovered_text=recovered_text + # Zero the counters here, not at module load: `since` should mean "serving + # since", so a scrape's totals cover the run, not the minutes spent loading + # the classifier. Nothing has been processed yet — recovery only enqueues. + reset_metrics!() + + # Each pool gets its stage's counters (src/stats.jl); `worker_loop` records + # into them, `GET /stats` reads them out. Stage 1 and 3 also hand theirs to + # their handler, which charges time parked on a full downstream queue to + # `blocked_ns` so it isn't mistaken for work. + st = METRICS.stages workers = [Threads.@spawn worker_loop(i, cfg, queue, - (job, c, wid) -> handle_classify_job(job, c, wid, known_queue, unknown_queue)) + (job, c, wid) -> handle_classify_job(job, c, wid, known_queue, unknown_queue, st.classify), + st.classify) for i in 1:cfg.worker_count] - known_workers = [Threads.@spawn worker_loop(i, cfg, known_queue, handle_known_job) + known_workers = [Threads.@spawn worker_loop(i, cfg, known_queue, handle_known_job, st.enrich) for i in 1:cfg.known_worker_count] unknown_workers = [Threads.@spawn worker_loop(i, cfg, unknown_queue, - (job, c, wid) -> handle_unknown_job(job, c, wid, text_queue)) + (job, c, wid) -> handle_unknown_job(job, c, wid, text_queue, st.triage), + st.triage) for i in 1:cfg.unknown_worker_count] text_workers = [Threads.@spawn worker_loop(i, cfg, text_queue, - (job, c, wid) -> handle_text_job(job, c, wid, DETECTOR[])) + (job, c, wid) -> handle_text_job(job, c, wid, DETECTOR[]), + st.language) for i in 1:cfg.text_worker_count] register_routes() diff --git a/src/queue.jl b/src/queue.jl index d816c88..44da94d 100644 --- a/src/queue.jl +++ b/src/queue.jl @@ -81,6 +81,16 @@ function close!(q::ChannelQueue) return nothing end +""" + capacity(q) -> Int + +How many jobs the queue can hold before `enqueue!` starts refusing. Part of the +introspection seam alongside `length`: `/stats` reports depth against capacity, +because a depth of 900 means nothing without knowing whether the limit is 1000 +or 1_000_000. +""" +capacity(q::ChannelQueue) = q.capacity + "Number of jobs currently buffered (for logging/introspection)." function Base.length(q::ChannelQueue) lock(q.cond) diff --git a/src/server.jl b/src/server.jl index 3df308b..760b006 100644 --- a/src/server.jl +++ b/src/server.jl @@ -23,6 +23,22 @@ function health_handler(_::HTTP.Request) return jsonresp(200, (; status = "ok")) end +""" +`GET /stats` — the pipeline's own counters (src/stats.jl), as JSON. + +Read-only and cheap: a few atomic loads and one `length` per queue, no pipeline +state touched. Two scrapes Δt apart give per-stage throughput and utilization — +see bin/bench.jl, which is the intended consumer. + +Unlike `/upload` this is an ordinary Oxygen route: it has no body to stream, and +being in Oxygen's middleware chain is a feature here. +""" +function stats_handler(_::HTTP.Request) + queues = (classify = QUEUE[], enrich = KNOWN_QUEUE[], + triage = UNKNOWN_QUEUE[], language = TEXT_QUEUE[]) + return jsonresp(200, stats_snapshot(CONFIG[], queues)) +end + "Write a JSON response onto a raw stream (the streaming handler's `jsonresp`)." function stream_jsonresp(stream::HTTP.Stream, status::Int, data) body = JSON3.write(data) @@ -106,6 +122,7 @@ function upload_stream_handler(stream::HTTP.Stream) end function serve_upload!(stream::HTTP.Stream) + Threads.atomic_add!(METRICS.intake.requests, 1) cfg = CONFIG[] queue = QUEUE[] chunk = cfg.upload_chunk_bytes @@ -134,6 +151,7 @@ function serve_upload!(stream::HTTP.Stream) # Already backpressured: consume the part, but don't write a file we # know we cannot enqueue. if queue_full + Threads.atomic_add!(METRICS.intake.rejected, 1) skip_part_body!(reader) continue end @@ -154,8 +172,13 @@ function serve_upload!(stream::HTTP.Stream) if enqueue!(queue, job) @info "accepted" id=job.id name=job.original_name size=job.size push!(accepted, (; id = job.id, name = job.original_name)) + # Counted at the point the job becomes stage 1's problem, so + # intake totals and stage-1 arrivals refer to the same files. + Threads.atomic_add!(METRICS.intake.files, 1) + Threads.atomic_add!(METRICS.intake.bytes, job.size) else rm(job.path; force = true) # never queued → don't leave it in spool + Threads.atomic_add!(METRICS.intake.rejected, 1) queue_full = true end end @@ -187,7 +210,8 @@ hand everything else to Oxygen unchanged. Trade-off: `/upload` bypasses Oxygen's middleware chain, so it is absent from Oxygen's built-in metrics and docs. Deliberate — flat intake memory is the point -of this service, and the pipeline's own `@info` records cover intake anyway. +of this service, and intake is covered by our own counters (`/stats`) and `@info` +records anyway. """ function root_stream_handler(middleware::Function) oxygen_handler = Oxygen.Core.stream_handler(middleware) @@ -208,5 +232,6 @@ via `root_stream_handler`, ahead of Oxygen's router. """ function register_routes() @get("/health", health_handler) + @get("/stats", stats_handler) return nothing end diff --git a/src/stats.jl b/src/stats.jl new file mode 100644 index 0000000..542a51f --- /dev/null +++ b/src/stats.jl @@ -0,0 +1,211 @@ +# Per-stage pipeline metrics. +# +# End-to-end throughput says how fast the pipeline is; it does not say which +# stage is the reason. The external harness can't answer that either: `known/`, +# `unknown/` and `text/` are *transient* — a file can pass through one between +# two directory polls — so sampling directories from outside undercounts exactly +# the stages we most want to measure. Only the pipeline itself sees every job. +# +# So each stage keeps four counters, all incremented in `worker_loop` (the one +# place every stage's work passes through) and read by `GET /stats`: +# +# completed / failed jobs finished, jobs quarantined → throughput +# bytes job bytes processed → per-stage MiB/s +# busy_ns summed handler wall time → service time +# blocked_ns of that, time parked on a full downstream queue +# in_flight handlers running right now → saturation +# +# Rates fall out of a pair of scrapes taken Δt apart: +# +# throughput = Δcompleted / Δt +# utilization = (Δbusy_ns - Δblocked_ns) / (Δt * workers) ~1.0 ⇒ bottleneck +# +# Utilization is the useful one. Throughput alone can't distinguish a stage that +# is saturated from one that is merely starved by the stage ahead of it — both +# show the same files/s — whereas utilization separates them: the saturated stage +# sits near 1.0 with its queue backing up, the starved stage sits near 0. +# +# `blocked_ns` is what keeps that true downstream. Stages 1 and 3 apply blocking +# backpressure: when the next queue is full the handler parks and retries rather +# than dropping the file (see ROUTE_ENQUEUE_RETRY_SECONDS). That parked time is +# inside the handler, so counting it as busy would show stage 1 pinned at 1.0 +# whenever stage 2 is the real bottleneck — every stage upstream of the jam would +# look like the jam. Subtracting it leaves utilization meaning "doing its own +# work", and the blocked share becomes its own signal: a stage blocked 90% of the +# time is naming its successor as the bottleneck. +# +# Everything here is monotonic since process start (or since `reset_metrics!`), +# in the Prometheus style: counters, never rates. Rates are the reader's job, so +# a scrape is stateless and two readers can't disturb each other. +# +# Cost is a handful of atomic adds per file, against handlers that spawn +# exiftool — unmeasurable in practice, and the counters are never read on the +# hot path. + +# Stage identity lives here, in declaration order, so the report, the JSON and +# the worker wiring can't drift apart: adding stage 5 to the live path means +# adding it here and passing the new `StageStats` to its `worker_loop`. +const STAGE_KEYS = (:classify, :enrich, :triage, :language) +const STAGE_TITLES = (classify = "classify", enrich = "enrich", + triage = "triage", language = "language") + +""" +Counters for one pipeline stage. All fields are atomic and monotonic: workers +only ever add, readers only ever read, so no lock is needed between them. + +`in_flight` is the exception to monotonic — it goes up and down — and is the one +counter that is a *level* rather than a total. +""" +struct StageStats + completed::Threads.Atomic{Int} + failed::Threads.Atomic{Int} + bytes::Threads.Atomic{Int} + busy_ns::Threads.Atomic{Int} + blocked_ns::Threads.Atomic{Int} + in_flight::Threads.Atomic{Int} +end + +StageStats() = StageStats(Threads.Atomic{Int}(0), Threads.Atomic{Int}(0), + Threads.Atomic{Int}(0), Threads.Atomic{Int}(0), + Threads.Atomic{Int}(0), Threads.Atomic{Int}(0)) + +""" +Counters for HTTP intake, which has no worker loop to hang them off. + +`files` counts what was actually spooled *and* queued — the same thing the 202 +body reports as `accepted` — so it lines up with stage 1's arrivals. Files +dropped because the queue was full are counted separately in `rejected`, since a +run where intake outruns the pipeline should be visible as backpressure rather +than as slow intake. +""" +struct IntakeStats + requests::Threads.Atomic{Int} + files::Threads.Atomic{Int} + bytes::Threads.Atomic{Int} + rejected::Threads.Atomic{Int} +end + +IntakeStats() = IntakeStats(Threads.Atomic{Int}(0), Threads.Atomic{Int}(0), + Threads.Atomic{Int}(0), Threads.Atomic{Int}(0)) + +""" +Every counter in the process, plus the wall-clock origin the totals are measured +from. + +`since` matters to a reader computing rates from a single scrape: without it, +"1000 files completed" has no denominator. Readers that scrape twice should use +their own Δt instead — it excludes the time before they started watching. +""" +struct Metrics + intake::IntakeStats + stages::NamedTuple{STAGE_KEYS,NTuple{4,StageStats}} + since::Base.RefValue{Float64} +end + +Metrics() = Metrics(IntakeStats(), + NamedTuple{STAGE_KEYS}(ntuple(_ -> StageStats(), 4)), + Ref(time())) + +# Process-global, like the logger: workers on every thread add to it and the +# HTTP layer reads it, with no way to thread a handle through both paths that +# wouldn't just be this by another name. +const METRICS = Metrics() + +""" + reset_metrics!() + +Zero every counter and restart the measurement window. For tests, and for a +benchmark that wants totals covering only its own run — though a harness that +scrapes before and after can subtract instead, which is safer against a server +that is also serving someone else. +""" +function reset_metrics!(m::Metrics = METRICS) + for a in (m.intake.requests, m.intake.files, m.intake.bytes, m.intake.rejected) + a[] = 0 + end + for s in m.stages + s.completed[] = 0 + s.failed[] = 0 + s.bytes[] = 0 + s.busy_ns[] = 0 + s.blocked_ns[] = 0 + s.in_flight[] = 0 + end + m.since[] = time() + return nothing +end + +"Record one finished job: `ok` distinguishes a completion from a quarantine." +function record_job!(s::StageStats, ok::Bool, bytes::Int, elapsed_ns::Int) + Threads.atomic_add!(ok ? s.completed : s.failed, 1) + Threads.atomic_add!(s.bytes, bytes) + Threads.atomic_add!(s.busy_ns, elapsed_ns) + return nothing +end + +""" + enqueue_blocking!(queue, job, stats; retry_seconds) -> nothing + +Hand `job` to a downstream queue, parking and retrying until it fits, and charge +the parked time to `stats.blocked_ns`. + +This is the routing half of stages 1 and 3: a classified file is never dropped, +so a full downstream queue means waiting, not failing. Wrapping the retry loop +here — rather than repeating `while !enqueue! sleep end` at each call site — is +what makes that wait measurable at all, and keeps the three routing paths from +drifting into three different backoff behaviours. +""" +function enqueue_blocking!(queue::JobQueue, job::Job, stats::StageStats; + retry_seconds::Real) + enqueue!(queue, job) && return nothing + t0 = time_ns() + while !enqueue!(queue, job) + sleep(retry_seconds) + end + Threads.atomic_add!(stats.blocked_ns, Int(time_ns() - t0)) + return nothing +end + +# --- snapshot --------------------------------------------------------------- + +""" + stats_snapshot(cfg, queues, m = METRICS) -> NamedTuple + +Read every counter into a plain value tree for `GET /stats`. + +`queues` is a NamedTuple of `JobQueue`s keyed by `STAGE_KEYS`, supplying each +stage's live depth and capacity — the counters above are about work *done*, and +a stage's depth is what says whether work is piling up in front of it. + +The read is not atomic across stages: counters keep moving while we walk them, +so a snapshot can show a job counted as complete by stage 1 but not yet arrived +at stage 2. Over a benchmark window that skew is a job or two and does not move +a rate; a consistent snapshot would mean stopping the pipeline to read it. +""" +function stats_snapshot(cfg::Config, queues, m::Metrics = METRICS) + workers = (classify = cfg.worker_count, enrich = cfg.known_worker_count, + triage = cfg.unknown_worker_count, language = cfg.text_worker_count) + now = time() + stages = map(STAGE_KEYS, ntuple(identity, 4)) do key, i + s, q = m.stages[key], queues[key] + (; stage = i, + name = STAGE_TITLES[key], + workers = workers[key], + queue_depth = length(q), + queue_capacity = capacity(q), + completed = s.completed[], + failed = s.failed[], + bytes = s.bytes[], + busy_seconds = s.busy_ns[] / 1e9, + blocked_seconds = s.blocked_ns[] / 1e9, + in_flight = s.in_flight[]) + end + return (; now, + since = m.since[], + uptime_seconds = now - m.since[], + intake = (; requests = m.intake.requests[], + files = m.intake.files[], + bytes = m.intake.bytes[], + rejected = m.intake.rejected[]), + stages = collect(stages)) +end diff --git a/src/worker.jl b/src/worker.jl index 563aec8..6cd9a92 100644 --- a/src/worker.jl +++ b/src/worker.jl @@ -32,23 +32,23 @@ reference is visible downstream; the moved path becomes the routed job's locatio Sub-`MIN_FILE_BYTES` files short-circuit to `:unknown` inside `classify`. """ function handle_classify_job(job::Job, cfg::Config, worker_id::Int, - known_queue::JobQueue, unknown_queue::JobQueue) + known_queue::JobQueue, unknown_queue::JobQueue, + stats::StageStats) classification = classify(CLASSIFIER[], job.path) @info "classified file" worker=worker_id id=job.id name=job.original_name size=job.size classification=classification if classification === :known dest = move_to(cfg.known_dir, job) routed = Job(job.id, job.original_name, dest, job.size, job.received_at) - while !enqueue!(known_queue, routed) - sleep(ROUTE_ENQUEUE_RETRY_SECONDS) # known queue full → back off, don't drop - end + # known queue full → park and retry, don't drop (time charged to blocked_ns) + enqueue_blocking!(known_queue, routed, stats; + retry_seconds = ROUTE_ENQUEUE_RETRY_SECONDS) @info "routed to enrichment" worker=worker_id id=job.id dest=dest else dest = move_to(cfg.unknown_dir, job) routed = Job(job.id, job.original_name, dest, job.size, job.received_at) - while !enqueue!(unknown_queue, routed) - sleep(ROUTE_ENQUEUE_RETRY_SECONDS) # unknown queue full → back off, don't drop - end + enqueue_blocking!(unknown_queue, routed, stats; + retry_seconds = ROUTE_ENQUEUE_RETRY_SECONDS) @info "routed to content triage" worker=worker_id id=job.id dest=dest end return nothing @@ -79,16 +79,17 @@ data, `text/` otherwise. A text file is then routed onward to the stage-4 language-enrichment queue, retrying on a full queue rather than dropping the file (the same blocking backpressure stage 1 uses for its downstream queues). """ -function handle_unknown_job(job::Job, cfg::Config, worker_id::Int, text_queue::JobQueue) +function handle_unknown_job(job::Job, cfg::Config, worker_id::Int, + text_queue::JobQueue, stats::StageStats) if is_binary(job.path) dest = move_to(cfg.binary_dir, job) @info "sorted unknown" worker=worker_id id=job.id name=job.original_name kind=:binary dest=dest else dest = move_to(cfg.text_dir, job) routed = Job(job.id, job.original_name, dest, job.size, job.received_at) - while !enqueue!(text_queue, routed) - sleep(ROUTE_ENQUEUE_RETRY_SECONDS) # text queue full → back off, don't drop - end + # text queue full → park and retry, don't drop (time charged to blocked_ns) + enqueue_blocking!(text_queue, routed, stats; + retry_seconds = ROUTE_ENQUEUE_RETRY_SECONDS) @info "routed to language enrichment" worker=worker_id id=job.id dest=dest end return nothing @@ -111,26 +112,46 @@ function handle_text_job(job::Job, cfg::Config, worker_id::Int, detector) end """ - worker_loop(worker_id, cfg, queue, handler) + worker_loop(worker_id, cfg, queue, handler, stats) Consume jobs from `queue` until it is closed and drained, running `handler` on each. A failure on one job is logged and the file quarantined in `failed/` — it must never kill the worker, or the pool would silently shrink. + +This loop is also where per-stage metrics are recorded (`stats`, see +src/stats.jl). Instrumenting here rather than in each handler means every stage +is measured the same way, by construction, and a new stage is measured the +moment it is wired up — there is no per-handler bookkeeping to forget. + +The timed region is the handler alone, excluding the `dequeue!` above it: time +parked waiting for work is idleness, and counting it as service time would make +an idle stage look as busy as a saturated one. A quarantined job still counts +its time — the work was done, it just ended in `failed/`. """ -function worker_loop(worker_id::Int, cfg::Config, queue::JobQueue, handler) +function worker_loop(worker_id::Int, cfg::Config, queue::JobQueue, handler, + stats::StageStats) @info "worker started" worker=worker_id while true job = dequeue!(queue) job === nothing && break # queue closed and drained → exit + Threads.atomic_add!(stats.in_flight, 1) + t0 = time_ns() + ok = true try handler(job, cfg, worker_id) catch e + ok = false @error "processing failed" worker=worker_id id=job.id name=job.original_name exception=(e, catch_backtrace()) try move_to(cfg.failed_dir, job) catch e2 @error "could not quarantine failed file" worker=worker_id id=job.id path=job.path exception=(e2, catch_backtrace()) end + finally + # In a `finally` so an InterruptException during shutdown can't leave + # in_flight permanently above zero, which would read as a stuck job. + record_job!(stats, ok, job.size, Int(time_ns() - t0)) + Threads.atomic_sub!(stats.in_flight, 1) end end @info "worker stopped" worker=worker_id diff --git a/test/runtests.jl b/test/runtests.jl index a55da9a..918f8d5 100644 --- a/test/runtests.jl +++ b/test/runtests.jl @@ -5,10 +5,12 @@ using JSON3 # Pull internals into scope. These aren't exported (only `run` is), but the # whole risk profile of this pipeline lives in these functions, so we test them # directly rather than only through the HTTP surface. -using FileServer: Job, Config, ChannelQueue, enqueue!, dequeue!, length, +using FileServer: Job, Config, ChannelQueue, enqueue!, dequeue!, close!, length, sanitize_filename, recover_dir!, normalize_metadata, build_metadata, finalize_known!, run_exiftool, - is_binary, handle_unknown_job, + is_binary, handle_unknown_job, worker_loop, + capacity, StageStats, IntakeStats, Metrics, METRICS, reset_metrics!, + record_job!, enqueue_blocking!, stats_snapshot, STAGE_KEYS, detect_natural_language, run_linguist, detect_programming_language, read_text_sample, build_text_metadata, finalize_text!, handle_text_job, linguist_available, @@ -384,12 +386,13 @@ end mktempdir() do root cfg = tmp_config(root) text_queue = ChannelQueue(10) + stats = StageStats() # A binary file (embedded NUL) lands in binary/ and is NOT enqueued. bpath = joinpath(cfg.unknown_dir, "id-b-blob.dat") write(bpath, UInt8[0x00, 0xFF, 0x10]) bjob = Job("id-b", "blob.dat", bpath, filesize(bpath), 0.0) - handle_unknown_job(bjob, cfg, 1, text_queue) + handle_unknown_job(bjob, cfg, 1, text_queue, stats) @test isfile(joinpath(cfg.binary_dir, "id-b-blob.dat")) @test !isfile(bpath) @test length(text_queue) == 0 @@ -399,7 +402,7 @@ end tpath = joinpath(cfg.unknown_dir, "id-t-notes.log") write(tpath, "just some log text\n") tjob = Job("id-t", "notes.log", tpath, filesize(tpath), 0.0) - handle_unknown_job(tjob, cfg, 1, text_queue) + handle_unknown_job(tjob, cfg, 1, text_queue, stats) moved = joinpath(cfg.text_dir, "id-t-notes.log") @test isfile(moved) @test !isfile(tpath) @@ -874,4 +877,162 @@ end end end + # ---------------------------------------------------------------- stats + # + # The counters exist to answer "which stage is the bottleneck", and every + # wrong answer they could give is a wrong *attribution*: time credited to the + # stage that was waiting rather than the stage that was slow. So these tests + # care less about exact numbers than about what is charged to whom. + + @testset "per-stage stats" begin + @testset "record_job! separates completions from quarantines" begin + s = StageStats() + record_job!(s, true, 100, 5_000_000) + record_job!(s, true, 200, 5_000_000) + record_job!(s, false, 50, 1_000_000) + @test s.completed[] == 2 + @test s.failed[] == 1 + @test s.bytes[] == 350 # a quarantined job still moved bytes + @test s.busy_ns[] == 11_000_000 + end + + @testset "reset_metrics! zeroes counters and restarts the window" begin + m = Metrics() + Threads.atomic_add!(m.intake.files, 7) + record_job!(m.stages.enrich, true, 10, 1000) + m.since[] = 0.0 + reset_metrics!(m) + @test m.intake.files[] == 0 + @test m.stages.enrich.completed[] == 0 + @test m.since[] > 0.0 + end + + @testset "worker_loop records service time, failures, and drains in_flight" begin + mktempdir() do root + cfg = tmp_config(root) + q = ChannelQueue(10) + stats = StageStats() + + # Two jobs that succeed, one that throws. The thrower is + # quarantined by worker_loop, and must still be counted. + for (i, name) in enumerate(("ok-1", "ok-2", "boom")) + p = joinpath(cfg.spool_dir, "id-$i-$name") + write(p, "x" ^ 10) + @test enqueue!(q, Job("id-$i", name, p, filesize(p), 0.0)) + end + close!(q) + + worker_loop(1, cfg, q, (job, _, _) -> begin + sleep(0.02) + job.original_name == "boom" && error("handler blew up") + nothing + end, stats) + + @test stats.completed[] == 2 + @test stats.failed[] == 1 + @test stats.bytes[] == 30 + # Each of the three handlers slept 20ms before its outcome, so + # busy time covers the failure too — the work was done either way. + @test stats.busy_ns[] > 3 * 15_000_000 + @test stats.blocked_ns[] == 0 # nothing downstream to block on + @test stats.in_flight[] == 0 # the finally in worker_loop + @test isfile(joinpath(cfg.failed_dir, "id-3-boom")) + end + end + + @testset "enqueue_blocking! charges only the parked time to blocked_ns" begin + s = StageStats() + q = ChannelQueue(1) + job = Job("id-1", "a.bin", "/tmp/a.bin", 1, 0.0) + + # Room available → no wait, and nothing charged. This is the common + # case, and it must not pay for the instrumentation. + enqueue_blocking!(q, job, s; retry_seconds = 0.01) + @test length(q) == 1 + @test s.blocked_ns[] == 0 + + # Queue full → the call parks until a consumer makes room, and that + # time lands in blocked_ns, NOT in the caller's service time (which + # worker_loop measures separately around the whole handler). + drainer = Threads.@spawn begin + sleep(0.1) + dequeue!(q) + end + enqueue_blocking!(q, Job("id-2", "b.bin", "/tmp/b.bin", 1, 0.0), s; + retry_seconds = 0.01) + wait(drainer) + @test length(q) == 1 + @test s.blocked_ns[] > 50_000_000 # parked for ~100ms + end + + @testset "a routing handler charges a full downstream queue as blocked" begin + mktempdir() do root + cfg = tmp_config(root) + stats = StageStats() + + # Stage 3 routing a text file with the stage-4 queue already + # full: it must park rather than drop, and the wait must land in + # blocked_ns instead of masquerading as slow triage work. + text_queue = ChannelQueue(1) + @test enqueue!(text_queue, Job("filler", "f", "/tmp/f", 1, 0.0)) + + p = joinpath(cfg.unknown_dir, "id-t-notes.log") + write(p, "plain text\n") + job = Job("id-t", "notes.log", p, filesize(p), 0.0) + + drainer = Threads.@spawn begin + sleep(0.1) + dequeue!(text_queue) + end + handle_unknown_job(job, cfg, 1, text_queue, stats) + wait(drainer) + @test stats.blocked_ns[] > 50_000_000 + @test length(text_queue) == 1 # the file did get through + @test isfile(joinpath(cfg.text_dir, "id-t-notes.log")) + end + end + + @testset "stats_snapshot reports depth against capacity" begin + mktempdir() do root + cfg = tmp_config(root; worker_count = 3, known_worker_count = 4, + unknown_worker_count = 5, text_worker_count = 6, + queue_capacity = 11, known_queue_capacity = 12, + unknown_queue_capacity = 13, text_queue_capacity = 14) + m = Metrics() + queues = (classify = ChannelQueue(11), + enrich = ChannelQueue(12), triage = ChannelQueue(13), + language = ChannelQueue(14)) + @test enqueue!(queues.enrich, Job("id", "n", "/tmp/n", 1, 0.0)) + record_job!(m.stages.enrich, true, 4096, 2_000_000_000) + Threads.atomic_add!(m.intake.files, 9) + + snap = stats_snapshot(cfg, queues, m) + @test length(snap.stages) == 4 + @test [s.name for s in snap.stages] == ["classify", "enrich", "triage", "language"] + @test [s.stage for s in snap.stages] == [1, 2, 3, 4] + @test [s.workers for s in snap.stages] == [3, 4, 5, 6] + @test [s.queue_capacity for s in snap.stages] == [11, 12, 13, 14] + + enrich = snap.stages[2] + @test enrich.queue_depth == 1 + @test enrich.completed == 1 + @test enrich.bytes == 4096 + @test enrich.busy_seconds ≈ 2.0 + @test snap.intake.files == 9 + @test snap.uptime_seconds >= 0 + + # It has to survive the trip through JSON — /stats is the only + # consumer, and bin/bench.jl reads these exact field names. + round_tripped = JSON3.read(JSON3.write(snap)) + @test round_tripped.stages[2].busy_seconds ≈ 2.0 + @test round_tripped.stages[2].blocked_seconds == 0.0 + @test round_tripped.stages[2].queue_depth == 1 + end + end + + @testset "capacity is part of the queue seam" begin + @test capacity(ChannelQueue(7)) == 7 + end + end + end