diff --git a/README.md b/README.md index e6ea3dd..98f191a 100644 --- a/README.md +++ b/README.md @@ -510,12 +510,13 @@ the moment it is wired up, and never on the read path. ## Benchmarking (throughput + memory) -There are four harnesses. Only the first needs a running server: +There are five 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_stage1.jl`](#stage-1-component-benchmark-binbench_stage1jl) | stage 1 taken apart: classify vs. rename vs. enqueue vs. logging | no | +| [`bin/bench_stage2.jl`](#stage-2-component-benchmark-binbench_stage2jl) | stage 2 taken apart: exiftool spawn vs. extraction vs. commit | no | | [`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 | @@ -530,6 +531,9 @@ julia --project=. -t auto bin/bench_model.jl # 1b. stage 1 taken apart — also no server julia --project=. -t auto bin/bench_stage1.jl +# 1c. stage 2 taken apart — needs a directory of real files, not generated ones +julia --project=. -t auto bin/bench_stage2.jl + # 2. the pipeline. Start the server in one terminal… julia --project=. -t auto bin/server.jl @@ -692,6 +696,96 @@ Reported times are the **minimum** over trials. Flags: `--files`, `--reps`, `--trials`, `--size`, `--dir`, `--model`, `--threads`, `--no-threads`, `--json PATH`. +### Stage-2 component benchmark (`bin/bench_stage2.jl`) + +Stage 2 is the one stage whose cost is dominated by something outside Julia +entirely: it forks `exiftool`, a Perl program, once per file. `bin/bench.jl` +reports the stage as a single throughput number, which can't distinguish "the +extraction is slow" from "the *spawn* is slow" — and those have opposite fixes. +`bin/bench_stage2.jl` times each piece in isolation, then times the real +`handle_known_job` end to end: + +```bash +julia --project=. -t auto bin/bench_stage2.jl +``` + +Two things make this benchmark different from the stage-1 one: + +- **The corpus must be real files.** exiftool's cost depends on what it finds; a + file of random bytes bails out early and understates the stage by ~10×. The + default corpus is `data/done` — files that already went through stage 2 on this + machine. `--corpus PATH` points it elsewhere. +- **It prices the alternatives to one-fork-per-file**, because if the fork + dominates then the only fixes are to stop paying it per file. `exiftool + (batched Nx)` runs the whole corpus through one process; `exiftool + (-stay_open)` keeps one process alive and feeds it one file at a time over a + pipe — the shape a streaming pipeline could actually adopt. Both are measured, + not assumed. + +Measured on this machine (Ryzen 7 2700X, 8 cores/16 threads, Julia 1.12, +exiftool 12.40; 150 real files / 102 MiB, minimum of 2 trials): + +| component | per file | share of the handler | +|---|---|---| +| `run_exiftool()` | 135.9 ms | 98% | +| ↳ bare fork + Perl boot (`exiftool -ver`) | 76.7 ms | 55% | +| ↳ `JSON3.read` + tag map | 10 µs | 0.0% | +| `normalize_metadata` | 1.7 µs | 0.0% | +| `commit_enriched!` (sidecar + fsyncs + rename) | 2.0 ms | 1.5% | +| per-file logging (`@info`, flush→file) | 47 µs | 0.0% | +| **`handle_known_job`** | **138.3 ms** | 100% | +| *alt:* `exiftool -stay_open` | 40.1 ms | 29% | +| *alt:* `exiftool` batched 150× | 37.3 ms | 27% | + +**Stage 2 is exiftool and nothing else.** Everything the Julia code does — +parsing, normalizing, the durable sidecar-first commit, the log line — sums to +about 1.5% of the stage. There is no point optimizing any of it. + +**More than half the stage is interpreter startup, not metadata extraction.** +The bare `exiftool -ver` (fork, Perl boot, module loads, read no file) costs +76.7 ms against a 135.9 ms full call. Both fork-free alternatives agree on what's +left: ~37–40 ms of actual work per file. So a persistent exiftool would cut the +stage by ~70%, and `-stay_open` gets there without giving up the one-file-in, +one-result-out shape the pipeline needs. That remains the single biggest +available win in this stage; it is measured here but not yet implemented. + +**This benchmark is also why `run_with_timeout` no longer polls.** The original +watchdog polled with `sleep(0.1)` and then joined the polling task, so every call +paid the remainder of an in-flight sleep *after* the child had already exited — +~25 ms per file here, and a measured 101 ms on a process that exits instantly. +Replacing it with a one-shot `Timer` took the stage from 164.6 ms to 138.3 ms per +file (6→8 files/s on one worker) and cost nothing in behavior. Stage 4 shares the +wrapper and got the same fix for free. + +Writing the missing tests for that wrapper turned up a second, worse problem: +**the timeout was never enforceable.** `wait(proc)` returns only once the +captured stdout pipe closes, and any grandchild inherits that pipe — so +signalling the child alone left the worker blocked until the whole process tree +finished on its own (a `sh -c "trap '' TERM; sleep 30"` child ran the full 30 s +against a 1 s timeout, under both the old and new watchdog). The child now runs +in its own process group and the timeout signals the group. The trade is that a +hard crash of the server orphans an in-flight child rather than taking it down; +these children are short-lived and timeout-bounded, which is the cheaper side of +it. + +**Stage 2 scales to ~8 workers, then flattens**: 8 files/s at 1 worker, 14 at 2, +28 at 4, **49 at 8**, and 49 at 16 — the machine runs out of cores to run Perl +on, which is exactly what you'd expect of a stage that is ~100% subprocess. Note +that the sweep pulls from a shared counter rather than splitting the corpus into +contiguous slices: per-file exiftool time spans two orders of magnitude on a real +corpus (one 2.1 s archive among 48 files), and a static split reports a scaling +ceiling that is really just load imbalance. + +One caveat the numbers raise but don't answer: **`fsync_dir` measures 1.75 µs**, +which is far too fast to be a real disk flush. The durability that +`commit_enriched!` is written for may not survive power loss on this filesystem, +even though the code is correct. That's a correctness question, not a speed one, +and it is not yet resolved. + +Reported times are the **minimum** over trials. Flags: `--files`, `--reps`, +`--trials`, `--corpus`, `--dir`, `--timeout`, `--threads`, `--no-threads`, +`--no-stay-open`, `--json PATH`. + ### Model microbenchmark (`bin/bench_model.jl`) `bin/bench.jl` reports stage 1 as a single number — the wall time of diff --git a/bin/bench_stage2.jl b/bin/bench_stage2.jl new file mode 100644 index 0000000..5a4022b --- /dev/null +++ b/bin/bench_stage2.jl @@ -0,0 +1,699 @@ +#!/usr/bin/env julia +# +# bench_stage2.jl — take stage 2 apart and find the slowest component. +# +# bin/bench.jl reports stage 2 as a single number (its throughput and worker +# utilization under whole-pipeline contention). It doesn't say *which part* of +# the stage costs the most, and stage 2 is the one stage whose cost is dominated +# by something outside Julia entirely: it forks `exiftool`, a Perl program, once +# per file. Per file the stage does: +# +# build_metadata +# run_exiftool fork/exec exiftool -json -G -n, capture stdout +# run_with_timeout the watchdog wrapper around the subprocess +# JSON3.read parse the dump +# normalize_metadata coalesce ~20 tag names into the sidecar schema +# finalize_known! (commit_enriched!) +# JSON3.write serialize the sidecar payload +# write + fsync durably persist the sidecar bytes to a temp name +# mv + fsync_dir commit the sidecar, then persist the rename itself +# move_to rename known/ -> done/, the commit point +# logging one @info line ("enriched") +# +# This script times each of those in isolation, then times the real +# `handle_known_job` end to end so the parts can be checked against the whole. +# +# Three things here that the stage-1 benchmark has no equivalent of: +# +# * The corpus must be real files. exiftool's cost depends on what it finds; +# random bytes exit early and would understate the stage by a lot. The +# default corpus is `data/done` — files that already went through stage 2 on +# this machine — copied back into a scratch known/ dir. +# * Two rows price the *alternatives* to one-fork-per-file, because if the +# fork dominates then the only fixes are to stop paying it per file: +# `exiftool (batched Nx)` runs the whole corpus through one process, and +# `exiftool (-stay_open)` keeps a single process alive and feeds it one file +# at a time over a pipe — the shape a streaming pipeline could actually use. +# Both are measured, not assumed. +# * `run_with_timeout` gets its own row *next to* a bare `Base.run` of the same +# command. The difference is what the watchdog costs, and its polling loop +# (`sleep(0.1)`) is a suspicious enough design to want measured rather than +# reasoned about. +# +# The `--threads` sweep runs the full handler across worker counts: subprocess +# spawning contends on things (the kernel's fork path, page cache, the logger's +# stream) that a single-threaded ranking can't reveal. +# +# Usage: +# julia --project=. -t auto bin/bench_stage2.jl [options] +# +# --files N corpus files per timed pass (default: 48). The concurrency +# sweep wants more than the component rows do — with a single +# 2 s file in the corpus, 48 files can't show more than ~4x no +# matter how many workers run, so pass --files 150 when the +# question is scaling. +# --reps N calls per timed pass for cheap, non-consuming benchmarks (default: 2000) +# --trials N timed passes; the minimum is reported (default: 3) +# --corpus PATH directory of real files to draw the corpus from (default: data/done) +# --dir PATH working directory for the corpus (default: a temp dir under data/) +# --timeout SEC exiftool timeout, as Config.exiftool_timeout (default: 30) +# --threads LIST worker counts for the concurrency sweep (default: 1,2,4,8,nthreads) +# --no-threads skip the concurrency sweep +# --no-stay-open skip the persistent-exiftool probe +# --json PATH also write the results as JSON +# +# Reported times are the *minimum* over trials: the floor is the signal and +# everything above it is scheduler, page-cache and GC noise. + +using FileServer +using JSON3 +using Logging +using Printf +using Random + +const FS = FileServer + +# ---------------------------------------------------------------- option parsing + +const DEFAULTS = Dict{String,Any}( + "files" => 48, + "reps" => 2000, + "trials" => 3, + "corpus" => "data/done", + "dir" => nothing, + "timeout" => 30, + "threads" => nothing, + "no-threads" => false, + "no-stay-open" => false, + "json" => nothing, +) + +const FLAGS = ("no-threads", "no-stay-open") +const INTS = ("files", "reps", "trials", "timeout") + +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 INTS ? 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) + +""" + best_of(pass, prepare; trials) -> ns_per_op + +Run `pass()` `trials` times and report the fastest, in nanoseconds per operation +(`pass` returns the number of operations it performed). `prepare()` runs before +each pass and is *not* timed — that is where a consuming benchmark puts the file +back where it started. `pass` comes first so callers can pass it as a `do` block. + +The first pass is thrown away: it pays Julia's JIT compilation, which on calls +this small is orders of magnitude more than the thing being measured. +""" +function best_of(pass, prepare; trials::Int) + best = Inf + for t in 0:trials + prepare() + GC.gc() + t0 = time_ns() + n = pass() + dt = Float64(time_ns() - t0) + t == 0 && continue # warm-up: compiled, not measured + best = min(best, dt / n) + end + return best +end + +noop() = nothing + +# ------------------------------------------------------------------- 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 + +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 + +rate(ns::Real) = 1e9 / max(ns, 1e-9) + +rule(n = 84) = println("-" ^ n) + +function header(title) + println() + println(title) + rule() +end + +# ------------------------------------------------------------------------- corpus + +""" + make_corpus(cfg, corpus_dir, n) -> Vector{Job} + +Copy up to `n` real files from `corpus_dir` into `known/` and build the `Job` +references a stage-2 worker would dequeue for them — the exact input +`handle_known_job` sees. + +Real files, not generated ones: exiftool's cost is a function of what it can +parse, and a file of random bytes bails out early enough to understate the stage +by an order of magnitude. `.meta.json` sidecars are skipped — they are stage-2 +*output*, and enriching them would measure the wrong population. +""" +function make_corpus(cfg::FS.Config, corpus_dir::AbstractString, n::Int) + isdir(corpus_dir) || error("corpus dir not found: $corpus_dir") + names = filter(readdir(corpus_dir)) do f + !endswith(f, ".meta.json") && isfile(joinpath(corpus_dir, f)) + end + isempty(names) && error("no usable files in corpus dir: $corpus_dir") + sort!(names) # deterministic selection across runs + length(names) > n && (names = names[1:n]) + + jobs = FS.Job[] + for (i, name) in enumerate(names) + src = joinpath(corpus_dir, name) + # Give it a fresh id/spool-style filename so nothing collides with the + # corpus the file came from. + id, spooled = FS.spool_path(cfg, @sprintf("s2-%04d-%s", i, basename(name))) + cp(src, spooled; force = true) + dest = joinpath(cfg.known_dir, basename(spooled)) + mv(spooled, dest; force = true) + push!(jobs, FS.Job(id, basename(name), dest, filesize(dest), time())) + end + return jobs +end + +""" + reknown!(cfg, jobs) + +Put every corpus file back in `known/`, wherever the last pass left it (done/ or +already home), and delete any sidecar it produced. This is the untimed `prepare` +step for benchmarks that consume their input by committing it. +""" +function reknown!(cfg::FS.Config, jobs::Vector{FS.Job}) + for job in jobs + base = basename(job.path) + for dir in (cfg.done_dir, cfg.failed_dir) + sidecar = joinpath(dir, string(base, ".meta.json")) + rm(sidecar; force = true) + rm(string(sidecar, ".tmp"); force = true) + end + isfile(job.path) && continue + for dir in (cfg.done_dir, cfg.failed_dir) + candidate = joinpath(dir, base) + if isfile(candidate) + mv(candidate, job.path; force = true) + break + end + end + end + return nothing +end + +# --------------------------------------------------------------------- loggers + +""" + with_logger_named(f, which, path) + +Run `f` under one of the loggers the cost of logging is bracketed by: + + * `:null` — `NullLogger`: the `@info` macro's own overhead, nothing else. + * `:format` — `ConsoleLogger` to `devnull`: message formatting and key/value + interpolation, but no I/O. + * `:flush` — `FlushLogger(ConsoleLogger(io))` to a real file: what + `FileServer.run` installs, under the redirect it was written + for. Stage 2's per-file line is `@info`, not `@debug`, so this + row is what the deployed server actually pays. +""" +function with_logger_named(f, which::Symbol, path::AbstractString) + if which === :null + return with_logger(f, NullLogger()) + elseif which === :format + return with_logger(f, ConsoleLogger(devnull)) + elseif which === :flush + return open(path, "w") do io + with_logger(f, FS.FlushLogger(ConsoleLogger(io, Logging.Info))) + end + end + error("unknown logger: $which") +end + +# --------------------------------------------------- exiftool spawn alternatives + +""" + capture(cmd) -> Vector{UInt8} + +Run `cmd` and return its stdout, tolerating a non-zero exit the way +`run_with_timeout` does. `read(cmd, String)` would throw instead, and a real +corpus makes that a question of when, not whether: exiftool exits 1 on a file +whose type it can't recognize, which in this pipeline is a routine outcome (it +yields a degraded sidecar, not a failure). This is `run_with_timeout` minus the +watchdog, so the gap between the two rows prices the watchdog exactly. +""" +function capture(cmd::Cmd) + out = IOBuffer() + proc = Base.run(pipeline(cmd; stdout = out, stderr = devnull); wait = false) + wait(proc) + return take!(out) +end + +""" + batched_ns(paths, trials) -> ns_per_file + +Run the whole corpus through *one* `exiftool` process and divide by the file +count. This is the floor for "what does exiftool cost if you stop paying the +interpreter startup per file" — the fork, the Perl boot and the module loads are +paid once for the batch instead of once per file. +""" +function batched_ns(paths::Vector{String}, trials::Int) + return best_of(noop; trials) do + SINK[] = capture(`exiftool -json -G -n $paths`) + length(paths) + end +end + +""" + stay_open_ns(paths, trials) -> ns_per_file (or nothing if unsupported) + +Feed files one at a time to a single long-lived `exiftool -stay_open True -@ -` +process over a pipe, reading its `{ready}` sentinel after each. Unlike the +batched row this preserves the pipeline's actual shape — one file in, one result +out, arriving whenever it arrives — so it prices the realistic fix rather than +an unrealistic one. +""" +function stay_open_ns(paths::Vector{String}, trials::Int) + inp, outp = Pipe(), Pipe() + proc = Base.run(pipeline(`exiftool -stay_open True -@ -`; + stdin = inp, stdout = outp, stderr = devnull); wait = false) + close(inp.out); close(outp.in) + + ask(path) = begin + write(inp, "-json\n-G\n-n\n", path, "\n-execute\n") + flush(inp) + readuntil(outp, "{ready}") + end + try + ask(paths[1]) # pay the one-time process startup untimed + return best_of(noop; trials) do + for p in paths + SINK[] = ask(p) + end + length(paths) + end + finally + try + write(inp, "-stay_open\nFalse\n"); flush(inp); close(inp) + wait(proc) + catch + kill(proc, Base.SIGKILL) + end + end +end + +# ------------------------------------------------------------------ components + +""" + component_rows(cfg, jobs, opts) -> Vector + +Time each piece of stage 2 on its own. The subprocess rows run once per corpus +file (they cost milliseconds and don't need repetition); the in-memory and +filesystem rows run `reps` times; the committing rows run once per corpus file +with an untimed reset between passes. +""" +function component_rows(cfg::FS.Config, jobs::Vector{FS.Job}, opts) + reps, trials = opts["reps"], opts["trials"] + timeout = opts["timeout"] + nfiles = length(jobs) + paths = String[j.path for j in jobs] + rows = [] + add!(name, part, ns) = push!(rows, (; name, part, ns)) + + # --- the bare interpreter: fork/exec + Perl boot, reading no file at all. + # Everything the real call does beyond this is actual work. + add!("exiftool -ver (spawn)", "extract", best_of(noop; trials) do + for _ in 1:nfiles + SINK[] = capture(`exiftool -ver`) + end + nfiles + end) + + # --- the real command, run directly: spawn + parse the file, no watchdog. + add!("exiftool -json (raw run)", "extract", best_of(noop; trials) do + @inbounds for p in paths + SINK[] = capture(`exiftool -json -G -n $p`) + end + nfiles + end) + + # --- the same command through the watchdog wrapper the stage actually uses. + # The gap to the row above is what the timeout costs. + add!("run_with_timeout", "extract", best_of(noop; trials) do + @inbounds for p in paths + SINK[] = FS.run_with_timeout(`exiftool -json -G -n $p`, timeout) + end + nfiles + end) + + # --- run_exiftool: the wrapper plus JSON3.read plus the group-stripped Dict. + add!("run_exiftool (total)", "extract", best_of(noop; trials) do + @inbounds for p in paths + SINK[] = FS.run_exiftool(p, timeout) + end + nfiles + end) + + # --- what a fork-free exiftool would cost, two ways (see the docstrings). + add!("exiftool (batched $(nfiles)x)", "alt", batched_ns(paths, trials)) + if !opts["no-stay-open"] + try + add!("exiftool (-stay_open)", "alt", stay_open_ns(paths, trials)) + catch e + @warn "persistent-exiftool probe failed; skipping" exception = e + end + end + + # --- parsing alone, from bytes already captured: isolates JSON3 from the fork. + raw = [FS.run_with_timeout(`exiftool -json -G -n $p`, timeout) for p in paths] + valid = [b for b in raw if b !== nothing] + if !isempty(valid) + add!("JSON3.read (parse dump)", "extract", best_of(noop; trials) do + @inbounds for i in 1:reps + SINK[] = JSON3.read(String(copy(valid[(i - 1) % length(valid) + 1]))) + end + reps + end) + end + + # --- normalize_metadata: the ~20 tag coalesces, on a tag map already in memory. + bytags = [FS.run_exiftool(p, timeout) for p in paths] + good = [(j, b) for (j, b) in zip(jobs, bytags) if b !== nothing] + isempty(good) && error("exiftool produced no parseable output for any corpus file") + add!("normalize_metadata", "extract", best_of(noop; trials) do + @inbounds for i in 1:reps + j, b = good[(i - 1) % length(good) + 1] + SINK[] = FS.normalize_metadata(j, b) + end + reps + end) + + # The sidecar payloads, built once, untimed: the commit rows below measure + # committing, not extracting. + metas = [FS.normalize_metadata(j, b) for (j, b) in good] + + # --- serializing the sidecar (the raw dump makes this bigger than it looks). + add!("JSON3.write (sidecar)", "commit", best_of(noop; trials) do + @inbounds for i in 1:reps + SINK[] = JSON3.write(metas[(i - 1) % length(metas) + 1]) + end + reps + end) + + # --- sidecar bytes: open + write + flush + fsync, to a temp name. + # Non-consuming: same path rewritten each rep, as commit_enriched! does. + tmp = joinpath(cfg.done_dir, "bench_stage2_sidecar.tmp") + blobs = [JSON3.write(m) for m in metas] + # One pass over the real sidecar population, not `reps` of them. Two reasons, + # and the first is a correctness trap: thousands of back-to-back fsyncs + # saturate the device's write cache and each one starts waiting on the + # queue, which reported this row at 16 ms/file — eight times the whole + # `commit_enriched!` that contains it. The real stage fsyncs once per file + # with ~160 ms of exiftool between, and never queues that way. Second, real + # sidecars vary hugely in size (a zip's raw dump dwarfs a jpeg's), so the + # honest per-file number is one pass over all of them, not a cycle. + nio = length(blobs) + add!("write + fsync (sidecar)", "commit", best_of(noop; trials) do + @inbounds for i in 1:nio + open(tmp, "w") do io + write(io, blobs[(i - 1) % length(blobs) + 1]) + flush(io) + FS.fsync_fd(fd(io)) + end + end + nio + end) + rm(tmp; force = true) + + # --- fsync_dir: persisting the rename itself, once per file in the real path. + add!("fsync_dir (done/)", "commit", best_of(noop; trials) do + for _ in 1:nio + SINK[] = FS.fsync_dir(cfg.done_dir) + end + nio + end) + + # --- move_to: the rename known/ -> done/. Consuming: reset each pass. + add!("move_to (rename)", "commit", best_of(() -> reknown!(cfg, jobs); trials) do + @inbounds for job in jobs + SINK[] = FS.move_to(cfg.done_dir, job) + end + nfiles + end) + + # --- commit_enriched!: the whole sidecar-first commit, extraction excluded. + committable = [j for (j, _) in good] + add!("commit_enriched! (total)", "commit", + best_of(() -> reknown!(cfg, jobs); trials) do + @inbounds for (k, job) in enumerate(committable) + SINK[] = FS.commit_enriched!(cfg.done_dir, job, metas[k]) + end + length(committable) + end) + reknown!(cfg, jobs) + + # --- the one @info line, under each logger. + job1, meta1 = good[1][1], metas[1] + logfile = joinpath(dirname(cfg.known_dir), "bench_stage2.log") + for (which, label) in ((:null, "logging (NullLogger)"), + (:format, "logging (format only)"), + (:flush, "logging (flush→file)")) + ns = with_logger_named(which, logfile) do + best_of(noop; trials) do + for _ in 1:reps + @info "enriched" worker=1 id=job1.id dest=job1.path sidecar="x.meta.json" file_type=meta1.file_type created_by=meta1.created_by degraded=(meta1.error !== nothing) + end + reps + end + end + add!(label, "log", ns) + end + rm(logfile; force = true) + + return rows +end + +""" + handler_rows(cfg, jobs, opts) -> Vector + +Time the real `handle_known_job` end to end under each logger. The difference +between the rows is the cost logging adds to a file; the `:flush` row is what the +running server actually pays. +""" +function handler_rows(cfg::FS.Config, jobs::Vector{FS.Job}, opts) + trials = opts["trials"] + nfiles = length(jobs) + logfile = joinpath(dirname(cfg.known_dir), "bench_stage2.log") + rows = [] + for (which, label) in ((:null, "handle_known_job (NullLogger)"), + (:format, "handle_known_job (format only)"), + (:flush, "handle_known_job (flush→file)")) + ns = with_logger_named(which, logfile) do + best_of(() -> reknown!(cfg, jobs); trials) do + @inbounds for job in jobs + FS.handle_known_job(job, cfg, 1) + end + nfiles + end + end + push!(rows, (; name = label, part = "total", ns)) + end + reknown!(cfg, jobs) + rm(logfile; force = true) + return rows +end + +""" + thread_rows(cfg, jobs, opts) -> Vector + +Run the full handler across worker counts, under the server's real logger. Stage +2 spends most of its time in a child process, so this is the sweep that matters +most: whether the stage scales is a question about the kernel's fork path and the +machine's cores, not about Julia. + +Workers pull from a shared atomic counter rather than taking a contiguous slice. +That matches the server (its pool pulls from one queue), and it matters here in a +way it doesn't for stage 1: per-file exiftool time spans two orders of magnitude +on a real corpus — a single 2 s archive among 48 files — so a static split leaves +whichever worker drew it running alone while the rest idle, and the sweep would +report a scaling ceiling that is really just load imbalance. +""" +function thread_rows(cfg::FS.Config, jobs::Vector{FS.Job}, opts) + trials = opts["trials"] + nfiles = length(jobs) + counts = opts["threads"] === nothing ? + unique([1; 2; 4; 8; Threads.nthreads()]) : + [parse(Int, s) for s in split(String(opts["threads"]), ",")] + counts = sort(unique(filter(k -> 1 <= k <= Threads.nthreads(), counts))) + + logfile = joinpath(dirname(cfg.known_dir), "bench_stage2.log") + rows = [] + base = 0.0 + for k in counts + ns = with_logger_named(:flush, logfile) do + next = Threads.Atomic{Int}(1) + best_of(() -> (reknown!(cfg, jobs); next[] = 1); trials) do + # Shared counter, not a contiguous slice: every worker takes the + # next unclaimed file the moment it frees up, exactly as the + # server's pool takes the next job off the known queue. + @sync for t in 1:k + Threads.@spawn begin + while true + i = Threads.atomic_add!(next, 1) + i > nfiles && break + @inbounds FS.handle_known_job(jobs[i], cfg, t) + end + end + end + nfiles + end + end + r = rate(ns) # files/sec aggregate (ns is already per file, wall-clock) + k == counts[1] && (base = r) + push!(rows, (; workers = k, ns, files_per_sec = r, speedup = r / base)) + end + reknown!(cfg, jobs) + rm(logfile; force = true) + return rows +end + +# ------------------------------------------------------------------- reporting + +function print_components(rows, total_ns) + @printf("%-34s %-9s %12s %10s %9s\n", "component", "part", "per file", "rate", "% total") + rule() + for r in rows + @printf("%-34s %-9s %12s %10s %8.1f%%\n", r.name, r.part, human_time(r.ns), + human_rate(rate(r.ns)), 100 * r.ns / total_ns) + end +end + +function print_threads(rows) + @printf("%-9s %12s %12s %9s\n", "workers", "per file", "throughput", "speedup") + rule() + for r in rows + @printf("%-9d %12s %12s %8.2fx\n", r.workers, human_time(r.ns), + human_rate(r.files_per_sec), r.speedup) + end +end + +# ------------------------------------------------------------------------- main + +function main(argv) + opts = parse_args(argv) + try + FS.assert_exiftool() + catch e + println(stderr, sprint(showerror, e)); return 1 + end + + root = opts["dir"] === nothing ? + mktempdir(pwd(); prefix = "bench_stage2_") : String(opts["dir"]) + owned = opts["dir"] === nothing + cfg = FS.Config( + spool_dir = joinpath(root, "spool"), + known_dir = joinpath(root, "known"), + done_dir = joinpath(root, "done"), + failed_dir = joinpath(root, "failed"), + exiftool_timeout = opts["timeout"], + ) + for d in (cfg.spool_dir, cfg.known_dir, cfg.done_dir, cfg.failed_dir) + mkpath(d) + end + + jobs = try + make_corpus(cfg, String(opts["corpus"]), opts["files"]) + catch e + owned && rm(root; recursive = true, force = true) + println(stderr, sprint(showerror, e)); return 1 + end + bytes = sum(j.size for j in jobs) + + println("stage-2 component benchmark") + rule() + @printf("%-22s %s\n", "julia threads", Threads.nthreads()) + @printf("%-22s %s\n", "exiftool", strip(read(`exiftool -ver`, String))) + @printf("%-22s %s\n", "corpus", "$(length(jobs)) files ($(round(bytes / 1024^2; digits=1)) MiB) from $(opts["corpus"])") + @printf("%-22s %s\n", "scratch", root) + @printf("%-22s %s\n", "reps / trials", "$(opts["reps"]) / $(opts["trials"])") + @printf("%-22s %s\n", "exiftool timeout", "$(opts["timeout"]) s") + + try + comps = component_rows(cfg, jobs, opts) + handlers = handler_rows(cfg, jobs, opts) + + # The denominator is the handler as the server actually runs it: the real + # flushing logger, one worker. Percentages are shares of that, so they are + # directly comparable and the parts can be checked against the whole. + total = only(r.ns for r in handlers if r.name == "handle_known_job (flush→file)") + + header("Components (single worker)") + print_components(comps, total) + + header("Whole handler, by logger") + print_components(handlers, total) + + pick(name) = only(r.ns for r in comps if r.name == name) + accounted = pick("run_exiftool (total)") + pick("commit_enriched! (total)") + + pick("logging (flush→file)") + println() + @printf("accounted: %s of %s (%.0f%%); unaccounted overhead %s\n", + human_time(accounted), human_time(total), 100 * accounted / total, + human_time(max(total - accounted, 0))) + + threads = nothing + if !opts["no-threads"] && Threads.nthreads() > 1 + threads = thread_rows(cfg, jobs, opts) + header("Full handler across workers (server logger)") + print_threads(threads) + end + + if opts["json"] !== nothing + open(String(opts["json"]), "w") do io + JSON3.write(io, (; + julia_threads = Threads.nthreads(), + files = length(jobs), corpus_bytes = bytes, + reps = opts["reps"], trials = opts["trials"], + exiftool_timeout = opts["timeout"], + components = comps, handlers, threads, + )) + end + println("\nwrote ", opts["json"]) + end + finally + owned && rm(root; recursive = true, force = true) + end + return 0 +end + +exit(main(ARGS)) diff --git a/src/metadata.jl b/src/metadata.jl index b188348..545501c 100644 --- a/src/metadata.jl +++ b/src/metadata.jl @@ -67,6 +67,19 @@ function coalesce_tag(bytag::Dict{String,Any}, tags) return nothing end +"How long a child gets to honor SIGTERM before `run_with_timeout` escalates to SIGKILL." +const KILL_GRACE_SECONDS = 2.0 + +""" +Send `signum` to the whole process group `pgid` (a negative pid means "the group" +to `kill(2)`). Julia's `kill(::Process, sig)` signals only the child itself, +which is not enough to enforce a timeout — see `run_with_timeout`. +""" +function signal_group(pgid::Integer, signum::Integer) + ccall(:kill, Cint, (Cint, Cint), -pgid, signum) + return nothing +end + """ run_with_timeout(cmd, timeout) -> Union{Vector{UInt8},Nothing} @@ -75,33 +88,56 @@ Run `cmd`, capturing stdout, and return the captured bytes on clean exit, or SIGKILL after a grace period) once it overruns `timeout` seconds, so one pathological input can't wedge a worker forever. Shared by the exiftool (stage 2) and github-linguist (stage 4) shells. + +The child runs in its own process group and the timeout signals the *group*, not +just the child. This is what makes the timeout enforceable: `wait` below returns +only once the captured stdout pipe closes, and any grandchild inherits that pipe, +so signalling the child alone leaves a `sh -c "...; sleep 30"`-shaped process +tree running to completion with the worker still blocked on it. The cost of the +process group is that a hard crash of the server orphans an in-flight child +rather than taking it down; these children are short-lived and timeout-bounded, +which is the cheaper side of that trade. """ function run_with_timeout(cmd::Cmd, timeout::Integer) out = IOBuffer() - proc = Base.run(pipeline(cmd; stdout=out, stderr=devnull); wait=false) + # `detach` puts the child in a fresh process group (it becomes the group + # leader, so the group id is its pid); see the docstring for why the group, + # and not the child, is what the timeout has to signal. + proc = Base.run(pipeline(detach(cmd); stdout=out, stderr=devnull); wait=false) + pgid = Base.getpid(proc) - # Kill the process if it overruns the timeout. `t` polls rather than blocking - # so we can `kill` a hung child; the poll interval bounds shutdown latency. + # A one-shot timer, cancelled the moment the child exits, rather than a + # polling loop the caller has to join. The polling version charged every + # call the remainder of its in-flight `sleep(0.1)` *after* the child had + # already exited — ~50 ms on average, and a measured 101 ms on a process + # that exits instantly. That is pure latency on the hot path of two stages + # (exiftool here, github-linguist in stage 4), and it dwarfed the work on + # anything but a slow file. Waiting on the process directly costs nothing + # when the child exits normally, which is the overwhelmingly common case. killed = Ref(false) - t = Threads.@spawn begin - waited = 0.0 - while process_running(proc) && waited < timeout - sleep(0.1); waited += 0.1 - end - if process_running(proc) - killed[] = true - kill(proc, Base.SIGTERM) - # Escalate: a process that ignores/defers SIGTERM would otherwise pin - # the worker forever on the wait(proc) below, defeating the timeout. - grace = 0.0 - while process_running(proc) && grace < 2.0 - sleep(0.1); grace += 0.1 + timer = Timer(timeout) do _ + process_running(proc) || return + killed[] = true + signal_group(pgid, Base.SIGTERM) + # Escalate: a process that ignores/defers SIGTERM would otherwise pin the + # worker forever on the wait(proc) below, defeating the timeout. This + # runs off the timer's task so the event loop isn't held during the + # grace period, and it is not joined — by the time it wakes, `wait(proc)` + # has long since returned and `process_running` settles it. + Threads.@spawn begin + deadline = time() + KILL_GRACE_SECONDS + while process_running(proc) && time() < deadline + sleep(0.05) end - process_running(proc) && kill(proc, Base.SIGKILL) + process_running(proc) && signal_group(pgid, Base.SIGKILL) end end - wait(proc) - wait(t) + + try + wait(proc) + finally + close(timer) # cancel the pending kill; a no-op if it already fired + end (killed[] || !success(proc)) && return nothing return take!(out) diff --git a/test/runtests.jl b/test/runtests.jl index 918f8d5..0f0892a 100644 --- a/test/runtests.jl +++ b/test/runtests.jl @@ -289,6 +289,42 @@ end end end + @testset "run_with_timeout: returns as soon as the child exits" begin + # Regression guard. The original implementation polled with sleep(0.1) + # and joined the polling task, so every call paid the remainder of an + # in-flight sleep after the child had already exited — ~101 ms on a + # process that exits instantly, on the hot path of stages 2 and 4. The + # bound here is deliberately loose (a loaded CI box is slow) but far + # under the 100 ms floor the polling version could not beat. + FileServer.run_with_timeout(`true`, 30) # warm up / compile + t0 = time() + out = FileServer.run_with_timeout(`echo hi`, 30) + elapsed = time() - t0 + @test out !== nothing + @test strip(String(out)) == "hi" + @test elapsed < 0.05 + end + + @testset "run_with_timeout: kills an overrunning child and reports failure" begin + t0 = time() + out = FileServer.run_with_timeout(`sleep 30`, 1) + elapsed = time() - t0 + @test out === nothing # timed out → no output, caller degrades + @test elapsed < 5 # killed near the timeout, not after 30 s + end + + @testset "run_with_timeout: escalates to SIGKILL when SIGTERM is ignored" begin + # A child that traps SIGTERM. Without the escalation the worker would + # block on wait(proc) forever and the timeout would be unenforceable. + cmd = `sh -c "trap '' TERM; sleep 30"` + t0 = time() + out = FileServer.run_with_timeout(cmd, 1) + elapsed = time() - t0 + @test out === nothing + # 1 s timeout + up to KILL_GRACE_SECONDS before SIGKILL lands. + @test elapsed < 1 + FileServer.KILL_GRACE_SECONDS + 3 + end + @testset "run_exiftool: real extraction on a PNG" begin mktempdir() do root p = joinpath(root, "pixel.png")