diff --git a/.github/workflows/occupancy.yml b/.github/workflows/occupancy.yml new file mode 100644 index 00000000..92433886 --- /dev/null +++ b/.github/workflows/occupancy.yml @@ -0,0 +1,145 @@ +# SPDX-License-Identifier: AGPL-3.0-only +# SPDX-FileCopyrightText: 2026 Jonathan D.A. Jewell (hyperpolymath) +# +# The occupancy reading of the vsearch stage (docs/occupancy-2026-10.adoc): +# * the two-thread disagreement test (test/occupancy_disagreement_test.jl) on +# the pinned vsearch, and +# * Session IR's counted bytes against the typed peak computed in Agda from +# the pinned proof siblings (scripts/check-occupancy.sh). +# +# This account allows only GitHub's own actions and its own, pinned by SHA, so +# ci.yml (julia-actions, setup-bun, codecov) cannot start on this fork. That is +# why everything here installs from pinned URLs checked against pinned SHA256s, +# and why the disagreement test runs here as well as in ci.yml's full suite. +name: Occupancy reading + +on: + push: + branches: [main] + paths: + - src/core/memory_probe.jl + - src/core/provenance.jl + - src/pipeline/tools.jl + - src/server/routes/pipeline.jl + - test/occupancy_disagreement_test.jl + - docs/occupancy/** + - docs/occupancy-2026-10.adoc + - scripts/check-occupancy.sh + - scripts/fetch-siblings.sh + - .github/workflows/occupancy.yml + # Any base branch, so a PR stacked on another one is checked too. + pull_request: + paths: + - src/core/memory_probe.jl + - src/core/provenance.jl + - src/pipeline/tools.jl + - src/server/routes/pipeline.jl + - test/occupancy_disagreement_test.jl + - docs/occupancy/** + - docs/occupancy-2026-10.adoc + - scripts/check-occupancy.sh + - scripts/fetch-siblings.sh + - .github/workflows/occupancy.yml + +permissions: + contents: read + +concurrency: + group: ${{ github.workflow }}-${{ github.ref }} + cancel-in-progress: ${{ github.event_name != 'push' }} + +env: + # Julia 1.12.5, the version Manifest.toml was resolved with. The checksum is + # julialang's own (bin/checksums/julia-1.12.5.sha256), as the official Docker + # image (docker-library/julia a39df01) records it. + JULIA_URL: https://julialang-s3.julialang.org/bin/linux/x64/1.12/julia-1.12.5-linux-x86_64.tar.gz + JULIA_SHA256: 41b84d727e4e96fbf3ed9e92fa195d773d247b9097f73fad688f8b699758bae7 + +jobs: + disagreement: + name: Occupancy disagreement test + runs-on: ubuntu-24.04 + timeout-minutes: 45 + env: + # As in ci.yml: the root .Rprofile would otherwise bootstrap renv. + RENV_CONFIG_AUTOLOADER_ENABLED: "FALSE" + steps: + - uses: actions/checkout@11d5960a326750d5838078e36cf38b85af677262 # v4.4.0 + with: + persist-credentials: false + + - name: Install Julia + run: | + source scripts/ci/fetch_pinned.sh + fetch_pinned "$JULIA_URL" julia.tar.gz + echo "$JULIA_SHA256 julia.tar.gz" | sha256sum -c - + sudo mkdir -p /opt/julia + sudo tar xzf julia.tar.gz -C /opt/julia --strip-components=1 + rm julia.tar.gz + echo /opt/julia/bin >> "$GITHUB_PATH" + + # RCall builds against R. The occupancy test uses no R package. + - name: Install R + run: | + sudo apt-get update -qq + sudo apt-get install -y --no-install-recommends r-base-core + + - name: Instantiate the project + run: julia --project=. -e 'using Pkg; Pkg.instantiate()' + + # The pinned vsearch (config/defaults/tool_versions.yml), refused unless it + # hashes to the pinned SHA256, as ci.yml installs it. + - name: Install vsearch + run: | + eval "$(julia --project=. -e ' + using YAML + v = YAML.load_file("config/defaults/tool_versions.yml")["tools"]["vsearch"] + a = v["archives"]["linux-x86_64"] + println("VSEARCH_VERSION=", v["version"]); println("VSEARCH_URL=", a["url"]); println("VSEARCH_SHA256=", a["sha256"])')" + source scripts/ci/fetch_pinned.sh + fetch_pinned "$VSEARCH_URL" vsearch.tar.gz + echo "$VSEARCH_SHA256 vsearch.tar.gz" | sha256sum -c - + tar xzf vsearch.tar.gz --warning=no-unknown-keyword + sudo mv "vsearch-$VSEARCH_VERSION-linux-x86_64/bin/vsearch" /usr/local/bin/vsearch + rm -rf vsearch.tar.gz "vsearch-$VSEARCH_VERSION-linux-x86_64" + vsearch --version + + - name: Three orders of the vsearch stage + run: julia --project=. -t 2 test/occupancy_disagreement_test.jl + + typed-peak: + name: Occupancy typed peak + runs-on: ubuntu-24.04 + # The Agda 2.6.4.3 Debian trixie packages, in a digest-pinned image, as the + # upstream proof job uses. The siblings need no standard library. + container: debian:trixie-20250929-slim@sha256:1caf1c703c8f7e15dcf2e7769b35000c764e6f50e4d7401c355fb0248f3ddfdb + timeout-minutes: 20 + env: + DEBIAN_FRONTEND: noninteractive + steps: + # git before checkout, so the checkout is a repository and the siblings can be fetched. + - name: Install git, curl and Agda + run: | + apt-get update -qq + apt-get install -y --no-install-recommends ca-certificates curl git agda-bin=2.6.4.3-1+b2 + agda --version | grep -qx 'Agda version 2.6.4.3' + + - uses: actions/checkout@11d5960a326750d5838078e36cf38b85af677262 # v4.4.0 + with: + persist-credentials: false # the siblings are public; nothing here pushes + + # The Session IR stepper is Julia (standard library only). + - name: Install Julia + shell: bash + run: | + source scripts/ci/fetch_pinned.sh + fetch_pinned "$JULIA_URL" julia.tar.gz + echo "$JULIA_SHA256 julia.tar.gz" | sha256sum -c - + mkdir -p /opt/julia + tar xzf julia.tar.gz -C /opt/julia --strip-components=1 + rm julia.tar.gz + echo /opt/julia/bin >> "$GITHUB_PATH" + + - name: Counted bytes against typed peak + shell: bash + run: bash scripts/check-occupancy.sh diff --git a/.gitignore b/.gitignore index a6ed2b7d..1436ebab 100644 --- a/.gitignore +++ b/.gitignore @@ -250,6 +250,9 @@ vignettes/*.pdf ## Documentation kept local except published release notes docs/* !docs/release-notes/ +# and the dated occupancy reading (MMW-01), whose table the gate checks +!docs/occupancy-2026-10.adoc +!docs/occupancy/ # translation temp files @@ -332,3 +335,7 @@ LocalPreferences.toml # Development server log (dev/dev.sh) dev/server.log + +# Proof siblings fetched by scripts/fetch-siblings.sh, and Agda interface files +/_build/ +*.agdai diff --git a/config/defaults/pipeline.yml b/config/defaults/pipeline.yml index 11198e7a..6f50ed68 100644 --- a/config/defaults/pipeline.yml +++ b/config/defaults/pipeline.yml @@ -354,3 +354,18 @@ analysis: # Fraction of samples, 0 to 1, in which a taxon must have reads to be # tested. Taxa below it are listed as filtered, without a p-value. min_prevalence: 0.0 + +## Occupancy probe (MMW-01) +# Measures each attested stage's memory: Julia's allocation total, and the +# sampled resident memory of the tools it runs as child processes (Linux /proc). +# Recorded under the stage's `occupancy` key in attestation.yml. See the +# header of src/core/memory_probe.jl. +# +# Off unless set, and deliberately left commented out here: the run config is +# embedded in attestation.yml, so a default key would change every run's record +# even with the probe off. Set it by editing a pipeline.yml (the config editor +# only offers keys that have a factory default). +# occupancy_probe: +# enabled: true +# # Sampling interval in milliseconds (at least 1). +# interval_ms: 10 diff --git a/docs/occupancy-2026-10.adoc b/docs/occupancy-2026-10.adoc new file mode 100644 index 00000000..b7323e45 --- /dev/null +++ b/docs/occupancy-2026-10.adoc @@ -0,0 +1,125 @@ +// SPDX-License-Identifier: AGPL-3.0-only +// SPDX-FileCopyrightText: 2026 Jonathan D.A. Jewell (hyperpolymath) += Occupancy of the vsearch stage +:revdate: 2026-10-14 + +The vsearch stage runs two independent searches, ASV and OTU. The full +pipeline spawns them together; a single-stage re-run runs them one after the +other. The high-water-mark (HWM) algebra of occupancy-types predicts that the +stage's cost is the same in every order but its memory peak is not: serially +the peak is the larger search's, concurrently it can be their sum +(choreographic-types A-2 and A-3). This page records one measurement of that, +on 2026-10-14, and checks it against the typed reading. + +== What was measured + +`MemoryProbe` (`src/core/memory_probe.jl`), off unless a pipeline.yml sets +`occupancy_probe.enabled: true`, records for each attested stage: + +* `julia_allocated_bytes`: the `Base.gc_num()` delta over the stage (cost); +* per task (one spawned search): the largest sampled VmRSS of the child + process it started (`peak_rss_kib`), and the kernel's own VmHWM for it + (`hwm_kib`); +* `child_peak_rss_kib`: the largest sum of the stage's children's VmRSS at + one sampling instant (the stage's occupancy). + +The run: `test/occupancy_disagreement_test.jl`, which runs the same two searches +in three orders under the probe, on a fixture it generates: 1000 reference +sequences of 400 bp, 300 ASV and 150 OTU queries. Linux 6.18 with 4 CPUs, +Julia 1.12.5 with 2 threads, vsearch 2.30.5 with `--threads 1`. Sampling +interval 10 ms. + +== Results + +[cols="2,1,1,1,1,1,1",options="header"] +|=== +| Order | ASV peak (KiB) | OTU peak (KiB) | Stage peak, measured (KiB) | Counted bytes, Session IR | Typed peak, HWM | Julia allocation (bytes) +| ASV then OTU | 9204 | 9180 | 9204 | 9204 | 9204 | 978432 +| OTU then ASV | 9152 | 9192 | 9192 | 9192 | 9192 | 978720 +| Concurrent | 9176 | 9204 | 18380 | 18380 | 18380 | 978816 +|=== + +* *Cost commutes.* The allocation totals agree to 0.04%. The test asserts + agreement within 1%; a planted extra search in one order breaks it (spread + 25%). +* *Each search's peak is the same in every order*, to within 0.6% + (9152-9204 KiB). Each search's VmHWM equals its sampled peak, so the + sampling caught the peaks. +* *The stage peak is not the same.* Serially it is the larger search's + (9192-9204 KiB); concurrently it is 18380 KiB, twice that. This is the A-3 + order dependence, and nothing in the stage's outputs shows it: the hits are + byte-identical in all three orders. The prediction would have been falsified + had the concurrent peak matched the serial one; it did not. +* *Counted bytes equal the typed peak*, in all three orders. The counted bytes + come from occupancy-types' Session IR stepper running + `docs/occupancy/vsearch-*.sir`, one buffer per search. The typed peak is + computed, by `refl`, from occupancy-types' `Timeline` over resource-algebra's + HWM grade (`docs/occupancy/TypedPeak.agda`). A search alone exits holding + nothing (drift 0); in the concurrent order the first is still resident + when the second starts (drift = peak). The sequence law + `max (peak a) (drift a + peak b)` then gives the larger peak serially and the + sum concurrently. +* *The measured stage peak also equals the typed peak* here. For the concurrent + order the typed peak is the interleaving in which both are resident at once; + a measurement could come in lower if the two searches' peaks did not + coincide. Here they did: each search's memory plateaus once its database is + loaded. + +== Status + +[cols="1,3,1,2",options="header"] +|=== +| ID | Claim | Status | Artefact +| MMW-01.1 | With the flag off, a stage record has exactly its earlier keys; with it on, it gains `occupancy`; old records still parse and merge | TESTED | `test/occupancy_disagreement_test.jl` ("flag off", "round trip", "through the pipeline's _attest") +| MMW-01.2 | For this workload, the vsearch stage's cost (Julia allocation) is the same in all three orders | TESTED | same file, "two-thread disagreement" (asserted, within 1%) +| MMW-01.3 | For this workload, the stage peak depends on the order (A-3), and each search's own peak does not | TESTED | same file, recorded in the test log; table above +| MMW-01.4 | For the three Session IR texts, the stepper's counted bytes equal the HWM typed peak | TESTED | `scripts/check-occupancy.sh` (stepper output against `TypedPeak.agda`; two rejections in `docs/occupancy/reject`) +| bridge | Counted bytes are bounded by the typed peak in general (occupancy-types' v0 → typed bridge) | OPEN | unchanged; this page gives three instances, not a proof +|=== + +== Limits + +* *Sampled.* A per-task peak is the largest VmRSS seen at 10 ms intervals. VmHWM + is exact up to the last sample, but growth after it is not seen. A child that + lives for less than one interval may not be sampled at all (`samples: 0`). +* *Child memory only.* The children are counted; the Julia process is not. Its + `julia_live_bytes_delta` is recorded, but without a forced collection it + reflects when the GC last ran, not what the stage kept. +* *Allocation is process-wide.* While the study loop runs several runs' + pipelines at once, a stage's `julia_allocated_bytes` includes the others' + allocations too. The child figures are per stage, since each stage samples + only the children it started. +* *Linux only*, since it reads /proc. For a stage offloaded with `remote:`, + the local child is an ssh client, so its memory says nothing about the work. +* *One thread for vsearch.* With several threads, vsearch writes its hits in + whatever order the threads finish. `_sort_hits_by_query!` then does a + different amount of work, because its sort key splits each line on every + comparison. That moved the stage's allocation total by up to 12% between + repeats of the same order. So in production the cost is order-independent + only up to that noise; precomputing the sort key would remove it. This PR + does not change it. +* *One host, one day.* CI runs the same test on every push and logs its own + numbers; they are not compared with these. + +== Reproducing + +---- +julia --project=. -t 2 test/occupancy_disagreement_test.jl # set OCCUPANCY_RESULTS= for YAML +bash scripts/check-occupancy.sh # needs julia and agda 2.6.4.x +---- + +`scripts/fetch-siblings.sh` fetches the siblings at their pins into +`_build/dependencies`: occupancy-types `af95728` and resource-algebra +`cf9235c`. + +== As the proofs progress + +* *Per-buffer (n-ary) occupancy* in occupancy-types: each search is already one + buffer in the Session IR texts, so the typed peak could then be read per + search. +* *Signed drift*: frees could be graded as negative drift, so the Julia side's + collections (the pipeline's `GC.gc(false)` hints) become modelled + reclamation rather than unmodelled noise. +* *choreographic-types' operational `step` as a function*: the three orders + become three traces of one typed protocol. The frontier certificate its + two-thread application note describes could then be displayed in the WebUI. diff --git a/docs/occupancy/TypedPeak.agda b/docs/occupancy/TypedPeak.agda new file mode 100644 index 00000000..6899efb2 --- /dev/null +++ b/docs/occupancy/TypedPeak.agda @@ -0,0 +1,45 @@ +{-# OPTIONS --safe --without-K #-} +-- SPDX-License-Identifier: AGPL-3.0-only +-- SPDX-FileCopyrightText: 2026 Jonathan D.A. Jewell (hyperpolymath) +-- +-- The vsearch stage's typed peak, computed (TESTED, by refl) with +-- occupancy-types' Timeline over resource-algebra's HWM grade, both consumed +-- by pin through scripts/fetch-siblings.sh: +-- occupancy-types af957280075699c659de0abd9a3b0defade77309 +-- resource-algebra cf9235c532712a783a404c359a1055f40134794a +-- Nothing here re-declares their definitions. +-- +-- Each search is one interaction graded by its measured peak, in KiB, from the +-- 2026-10-14 run in docs/occupancy-2026-10.adoc. A search's process exits +-- holding nothing, so run alone its drift is 0. Run concurrently, the first is +-- still resident when the second starts, so that is graded with drift = peak. +-- That is the interleaving the measurement observed (both resident at once). +-- The algebra's sequence law, max (peak a) (drift a + peak b), then gives: +-- serial = the larger peak; concurrent = the sum. +module TypedPeak where + +open import ResourceAlgebra.Prelude +open import ResourceAlgebra.Instances.HWM + +import OccupancyTypes.Timeline +module TH = OccupancyTypes.Timeline HWM + +-- A search whose process has exited before the next one starts. +exits : Nat → Grade +exits p = hwm p 0 z≤n + +-- A search still resident when the next one starts. +resident : Nat → Grade +resident p = hwm p p (≤-refl p) + +-- ASV 9204 then OTU 9180. +asv-then-otu : peak (TH.grade (exits 9204 TH.∷ exits 9180 TH.∷ TH.[])) ≡ 9204 +asv-then-otu = refl + +-- OTU 9192 then ASV 9152. +otu-then-asv : peak (TH.grade (exits 9192 TH.∷ exits 9152 TH.∷ TH.[])) ≡ 9192 +otu-then-asv = refl + +-- ASV 9176 and OTU 9204 together. +concurrent : peak (TH.grade (resident 9176 TH.∷ exits 9204 TH.∷ TH.[])) ≡ 18380 +concurrent = refl diff --git a/docs/occupancy/reject/ConcurrentIsSerial.agda b/docs/occupancy/reject/ConcurrentIsSerial.agda new file mode 100644 index 00000000..592de6ee --- /dev/null +++ b/docs/occupancy/reject/ConcurrentIsSerial.agda @@ -0,0 +1,14 @@ +{-# OPTIONS --safe --without-K #-} +-- SPDX-License-Identifier: AGPL-3.0-only +-- SPDX-FileCopyrightText: 2026 Jonathan D.A. Jewell (hyperpolymath) +-- +-- Must be REFUSED: the claim that running the two searches concurrently leaves +-- the peak where a serial run puts it (the larger of the two). +module ConcurrentIsSerial where + +open import ResourceAlgebra.Prelude +open import ResourceAlgebra.Instances.HWM +open import TypedPeak + +wrong : peak (TH.grade (resident 9176 TH.∷ exits 9204 TH.∷ TH.[])) ≡ 9204 +wrong = refl diff --git a/docs/occupancy/reject/SerialIsCost.agda b/docs/occupancy/reject/SerialIsCost.agda new file mode 100644 index 00000000..dbb6f4ba --- /dev/null +++ b/docs/occupancy/reject/SerialIsCost.agda @@ -0,0 +1,14 @@ +{-# OPTIONS --safe --without-K #-} +-- SPDX-License-Identifier: AGPL-3.0-only +-- SPDX-FileCopyrightText: 2026 Jonathan D.A. Jewell (hyperpolymath) +-- +-- Must be REFUSED: reading a serial run's peak as the sum of the two peaks, +-- i.e. occupancy as cost (occupancy-types R-6). +module SerialIsCost where + +open import ResourceAlgebra.Prelude +open import ResourceAlgebra.Instances.HWM +open import TypedPeak + +wrong : peak (TH.grade (exits 9204 TH.∷ exits 9180 TH.∷ TH.[])) ≡ 18384 +wrong = refl diff --git a/docs/occupancy/results-2026-10-14.yml b/docs/occupancy/results-2026-10-14.yml new file mode 100644 index 00000000..394236d6 --- /dev/null +++ b/docs/occupancy/results-2026-10-14.yml @@ -0,0 +1,63 @@ +# SPDX-License-Identifier: AGPL-3.0-only +# The raw MemoryProbe records behind docs/occupancy-2026-10.adoc, written by +# test/occupancy_disagreement_test.jl with OCCUPANCY_RESULTS set (2026-10-14). +vsearch: "vsearch v2.30.5_linux_x86_64, 15.7GB RAM, 4 cores" +julia: "1.12.5" +threads: 2 +cpus: 4 +cost_spread: 0.00039231071008238524 +orders: + asv_then_otu: + probe: "MemoryProbe 1" + interval_ms: 10 + samples: 56 + julia_allocated_bytes: 978432 + julia_live_bytes_delta: 978544 + child_peak_rss_kib: 9204 + tasks: + - label: "vsearch:asv" + processes: 1 + samples: 33 + peak_rss_kib: 9204 + hwm_kib: 9204 + - label: "vsearch:otu" + processes: 1 + samples: 20 + peak_rss_kib: 9180 + hwm_kib: 9180 + otu_then_asv: + probe: "MemoryProbe 1" + interval_ms: 10 + samples: 57 + julia_allocated_bytes: 978720 + julia_live_bytes_delta: 978832 + child_peak_rss_kib: 9192 + tasks: + - label: "vsearch:otu" + processes: 1 + samples: 22 + peak_rss_kib: 9192 + hwm_kib: 9192 + - label: "vsearch:asv" + processes: 1 + samples: 33 + peak_rss_kib: 9152 + hwm_kib: 9152 + concurrent: + probe: "MemoryProbe 1" + interval_ms: 10 + samples: 33 + julia_allocated_bytes: 978816 + julia_live_bytes_delta: 978928 + child_peak_rss_kib: 18380 + tasks: + - label: "vsearch:asv" + processes: 1 + samples: 30 + peak_rss_kib: 9176 + hwm_kib: 9176 + - label: "vsearch:otu" + processes: 1 + samples: 24 + peak_rss_kib: 9204 + hwm_kib: 9204 diff --git a/docs/occupancy/vsearch-asv-then-otu.sir b/docs/occupancy/vsearch-asv-then-otu.sir new file mode 100644 index 00000000..e462fb5e --- /dev/null +++ b/docs/occupancy/vsearch-asv-then-otu.sir @@ -0,0 +1,13 @@ +(* SPDX-License-Identifier: AGPL-3.0-only *) +(* vsearch-asv-then-otu.sir - the vsearch stage with the two searches run one + after the other, ASV first: the single-stage re-run's order. Each search is + one buffer, sized by its measured peak in KiB (docs/occupancy-2026-10.adoc, + 2026-10-14 run); each process exits holding nothing, so its buffer is dropped + before the next search starts. + + Expected: steps = 0 (no communication), peak_live = max 9204 9180 = 9204. *) +let asv = alloc 9204 in +let _ = drop asv in +let otu = alloc 9180 in +let _ = drop otu in +unit diff --git a/docs/occupancy/vsearch-concurrent.sir b/docs/occupancy/vsearch-concurrent.sir new file mode 100644 index 00000000..93be40bf --- /dev/null +++ b/docs/occupancy/vsearch-concurrent.sir @@ -0,0 +1,12 @@ +(* SPDX-License-Identifier: AGPL-3.0-only *) +(* vsearch-concurrent.sir - the two searches spawned together, as the full + pipeline runs them. Session IR has no parallel composition, so this is the + interleaving the measurement observed: both buffers live at once, then both + dropped. + + Expected: steps = 0, peak_live = 9176 + 9204 = 18380. *) +let asv = alloc 9176 in +let otu = alloc 9204 in +let _ = drop asv in +let _ = drop otu in +unit diff --git a/docs/occupancy/vsearch-otu-then-asv.sir b/docs/occupancy/vsearch-otu-then-asv.sir new file mode 100644 index 00000000..ba575591 --- /dev/null +++ b/docs/occupancy/vsearch-otu-then-asv.sir @@ -0,0 +1,9 @@ +(* SPDX-License-Identifier: AGPL-3.0-only *) +(* vsearch-otu-then-asv.sir - the same two searches, OTU first. + + Expected: steps = 0, peak_live = max 9192 9152 = 9192. *) +let otu = alloc 9192 in +let _ = drop otu in +let asv = alloc 9152 in +let _ = drop asv in +unit diff --git a/scripts/check-occupancy.sh b/scripts/check-occupancy.sh new file mode 100755 index 00000000..f9bcba36 --- /dev/null +++ b/scripts/check-occupancy.sh @@ -0,0 +1,69 @@ +#!/usr/bin/env bash +# SPDX-License-Identifier: AGPL-3.0-only +# SPDX-FileCopyrightText: 2026 Jonathan D.A. Jewell (hyperpolymath) +# +# Gate for the occupancy reading of the vsearch stage (docs/occupancy-2026-10.adoc): +# 1. each Session IR text checks, and occupancy-types' stepper counts its bytes; +# 2. TypedPeak.agda type-checks against the pinned siblings, and each module in +# docs/occupancy/reject is refused at its own declaration; +# 3. the counted bytes equal the typed peak, order by order. +# Needs julia and agda (2.6.4.x) on PATH. Exit 0 only if all three hold. +set -euo pipefail +cd -- "$(dirname -- "${BASH_SOURCE[0]}")/.." +export LC_ALL="${LC_ALL:-C.UTF-8}" + +bash scripts/fetch-siblings.sh +deps=_build/dependencies +sir="$deps/occupancy-types/src/julia" +agda --version | head -1 +flags=(--safe --without-K --no-libraries --ignore-interfaces --double-check + -i docs/occupancy -i "$deps/occupancy-types/src" -i "$deps/resource-algebra/src") +logs=_build/check-logs +mkdir -p "$logs" +failures=0 + +# Order -> Session IR file and TypedPeak lemma. +orders=(asv-then-otu otu-then-asv concurrent) + +echo "== Session IR (occupancy-types v0 stepper)" +declare -A counted +for o in "${orders[@]}"; do + file="docs/occupancy/vsearch-$o.sir" + julia --startup-file=no --project="$sir" "$sir/bin/sessionir.jl" check "$file" + out="$(julia --startup-file=no --project="$sir" "$sir/bin/sessionir.jl" run "$file" --trace)" + echo "$out" + counted[$o]="$(sed -n 's/.*peak_live=\([0-9]*\).*/\1/p' <<<"$out" | head -1)" +done + +echo "== Typed peak (Agda, HWM Timeline)" +agda "${flags[@]}" docs/occupancy/TypedPeak.agda >"$logs/TypedPeak.log" 2>&1 || + { cat "$logs/TypedPeak.log"; echo "ERROR: TypedPeak.agda does not type-check" >&2; exit 1; } +echo "PASS: TypedPeak.agda" +for file in docs/occupancy/reject/*.agda; do + name="$(basename -- "$file" .agda)" + log="$logs/$name.log" + if agda "${flags[@]}" -i docs/occupancy/reject "$file" >"$log" 2>&1; then + cat "$log"; echo "ERROR: invalid claim accepted: $name" >&2; failures=$((failures + 1)); continue + fi + if ! grep -Eq "/$name\.agda:[0-9]+,[0-9]+" "$log" || ! grep -q 'when checking that' "$log" || + grep -Eq 'Not in scope|Failed to find|Unsolved|Parse error' "$log"; then + cat "$log"; echo "ERROR: $name failed for the wrong reason" >&2; failures=$((failures + 1)); continue + fi + echo "PASS: rejected $name ($(grep -m1 -E '[0-9]+ != [0-9]+' "$log"))" +done + +echo "== Counted bytes against typed peak (KiB)" +printf '%-14s %10s %10s\n' order counted typed +for o in "${orders[@]}"; do + typed="$(grep -A1 -E "^$o :" docs/occupancy/TypedPeak.agda | sed -n 's/.*≡ \([0-9]*\)$/\1/p' | head -1)" + printf '%-14s %10s %10s\n' "$o" "${counted[$o]}" "$typed" + if [ -z "${counted[$o]}" ] || [ -z "$typed" ] || [ "${counted[$o]}" != "$typed" ]; then + echo "ERROR: $o: counted ${counted[$o]:-?} != typed ${typed:-?}" >&2; failures=$((failures + 1)) + fi +done + +if (( failures > 0 )); then + echo "check-occupancy: $failures failure(s)" >&2 + exit 1 +fi +echo "check-occupancy: OK" diff --git a/scripts/fetch-siblings.sh b/scripts/fetch-siblings.sh new file mode 100755 index 00000000..31a98bc1 --- /dev/null +++ b/scripts/fetch-siblings.sh @@ -0,0 +1,29 @@ +#!/usr/bin/env bash +# SPDX-License-Identifier: AGPL-3.0-only +# SPDX-FileCopyrightText: 2026 Jonathan D.A. Jewell (hyperpolymath) +# +# Fetch the proof siblings the occupancy reading (docs/occupancy-2026-10.adoc) +# consumes, each at its pinned commit, into _build/dependencies. Their +# definitions are used, never copied. Re-pin one sibling per commit. +set -euo pipefail +cd -- "$(dirname -- "${BASH_SOURCE[0]}")/.." + +# occupancy-types: the HWM Timeline (Agda) and the SessionIR stepper (Julia). +OCCUPANCY_TYPES_COMMIT="${OCCUPANCY_TYPES_COMMIT:-af957280075699c659de0abd9a3b0defade77309}" +# resource-algebra: the HWM grade algebra occupancy-types builds on. +RESOURCE_ALGEBRA_COMMIT="${RESOURCE_ALGEBRA_COMMIT:-cf9235c532712a783a404c359a1055f40134794a}" + +fetch() { + local name="$1" commit="$2" dir="_build/dependencies/$1" + if [ ! -d "$dir/.git" ]; then + git init --quiet "$dir" + git -C "$dir" remote add origin "https://github.com/hyperpolymath/$name.git" + fi + git -C "$dir" fetch --quiet --depth 1 origin "$commit" + git -C "$dir" checkout --quiet --detach "$commit" + echo "$name at $(git -C "$dir" rev-parse --short HEAD)" +} + +mkdir -p _build/dependencies +fetch occupancy-types "$OCCUPANCY_TYPES_COMMIT" +fetch resource-algebra "$RESOURCE_ALGEBRA_COMMIT" diff --git a/src/MetaManifold.jl b/src/MetaManifold.jl index e3c70714..31db2a5c 100644 --- a/src/MetaManifold.jl +++ b/src/MetaManifold.jl @@ -5,6 +5,8 @@ include("core/types.jl") include("core/r_runtime.jl") # Provenance probes R through the shared runtime lock, so it follows r_runtime.jl. include("core/provenance.jl") +# Tools registers the children it starts with the probe, so it precedes pipeline/. +include("core/memory_probe.jl") include("core/log.jl") include("core/config.jl") include("core/databases.jl") diff --git a/src/core/memory_probe.jl b/src/core/memory_probe.jl new file mode 100644 index 00000000..6406c37a --- /dev/null +++ b/src/core/memory_probe.jl @@ -0,0 +1,343 @@ +# SPDX-License-Identifier: AGPL-3.0-only +# SPDX-FileCopyrightText: 2026 Jonathan D.A. Jewell (hyperpolymath) + +module MemoryProbe + +## Occupancy probe +# Measures what an attested stage occupied, for the stage's Attestation record: +# +# * Julia-side allocation, as the `Base.gc_num()` delta over the stage. This is +# a cost: it only ever adds up, whatever order the stage's work ran in. +# * Each child process's resident memory, sampled from /proc. The tools run as +# child processes (vsearch, swarm, cd-hit, ...), so `Sys.maxrss()` of this +# process never sees them. +# +# Two figures come from the samples. A task's `peak_rss_kib` is the largest VmRSS +# sampled for the processes that task started. The stage's `child_peak_rss_kib` is +# the largest SUM of VmRSS over every process of the stage at one sampling +# instant. For two tasks run one after the other it is the larger of their peaks; +# for two that run concurrently it can approach their sum. That order dependence, +# invisible in every output file, is what the measurement exists to show. +# +# `hwm_kib` is the kernel's own high-water mark (VmHWM) for a task's processes, as +# last read before they exited. It is exact up to the last sample, so it is at +# least `peak_rss_kib`; growth after the last sample is not seen. A process that +# lives for less than one interval may not be sampled at all, which `samples: 0` +# reports rather than a misleading zero peak. +# +# The probe is off unless the run config sets `occupancy_probe.enabled: true`. +# Off, `start` returns `nothing` and every other entry point is a pass-through, +# so the stage records and the outputs are exactly what they were without it. +# +# Linux only (/proc). Elsewhere, and for stages offloaded with `remote:` (where +# the local child is an ssh client), the child figures do not describe the work. +# The allocation figure is process-wide: while the study loop runs several runs' +# pipelines at once, a stage's figure includes the others' allocations too. The +# child figures are not affected, since each stage samples only its own children. + +using Base.ScopedValues: ScopedValue, with +using OrderedCollections: OrderedDict + +const PROBE = "MemoryProbe 1" +const DEFAULT_INTERVAL_MS = 10 + +# One labelled unit of a stage's work (one spawned search, say) and the child +# processes it started. +mutable struct TaskTrace + label :: String + live :: Vector{Int} # pids started and not yet waited for + processes :: Int # every child started under this label + samples :: Int # samples that found at least one live process + peak_kib :: Int + hwm_kib :: Int +end +TaskTrace(label::AbstractString) = TaskTrace(String(label), Int[], 0, 0, 0, 0) + +mutable struct Probe + interval :: Float64 + lock :: ReentrantLock + tasks :: Vector{TaskTrace} + stage :: TaskTrace # children started outside any `task` + samples :: Int + peak_kib :: Int + gc_before :: Base.GC_Num + live_before :: Int + running :: Bool + sampler :: Union{Task,Nothing} + timer :: Union{Timer,Nothing} + record :: Union{OrderedDict{String,Any},Nothing} + error :: Union{String,Nothing} # why sampling stopped early, if it did + buf :: Vector{UInt8} # /proc reads; preallocated so a sample + path :: Vector{UInt8} # allocates nothing that would count as work +end + +const STAGE = ScopedValue{Union{Probe,Nothing}}(nothing) +const TASK = ScopedValue{Union{TaskTrace,Nothing}}(nothing) + +## Configuration +_section(cfg) = cfg isa AbstractDict ? get(cfg, "occupancy_probe", nothing) : nothing + +""" + enabled(run_cfg) -> Bool + +Whether the run config turns the probe on: `occupancy_probe.enabled: true`. Any +other value, or no `occupancy_probe` section at all, leaves it off. +""" +function enabled(cfg) + s = _section(cfg) + s isa AbstractDict && get(s, "enabled", false) === true +end + +""" + interval_seconds(run_cfg) -> Float64 + +The sampling interval, from `occupancy_probe.interval_ms` (default 10 ms, at +least 1 ms). +""" +function interval_seconds(cfg) + s = _section(cfg) + ms = s isa AbstractDict ? get(s, "interval_ms", DEFAULT_INTERVAL_MS) : DEFAULT_INTERVAL_MS + (ms isa Real && !(ms isa Bool) && ms >= 1) || (ms = DEFAULT_INTERVAL_MS) + Float64(ms) / 1000 +end + +## Reading /proc without allocating +# "/proc//status" into `path`, NUL-terminated. +function _status_path!(path::Vector{UInt8}, pid::Int) + k = 0 + for c in codeunits("/proc/"); k += 1; path[k] = c; end + nd = ndigits(pid); d = pid + for i in nd:-1:1 + path[k + i] = UInt8('0') + UInt8(d % 10); d ÷= 10 + end + k += nd + for c in codeunits("/status"); k += 1; path[k] = c; end + path[k + 1] = 0x00 + nothing +end + +# Bytes of /proc//status read into `buf`, or -1 if the process is gone. +function _read_status!(buf::Vector{UInt8}, path::Vector{UInt8}, pid::Int) + _status_path!(path, pid) + fd = GC.@preserve path ccall(:open, Cint, (Ptr{UInt8}, Cint), path, 0) + fd < 0 && return -1 + total = 0 + GC.@preserve buf while total < length(buf) + r = ccall(:read, Cssize_t, (Cint, Ptr{UInt8}, Csize_t), + fd, pointer(buf, total + 1), length(buf) - total) + r <= 0 && break + total += r + end + ccall(:close, Cint, (Cint,), fd) + total +end + +# The kB value of the line starting `key` (6 bytes, e.g. "VmRSS:"), or -1. +function _field_kib(buf::Vector{UInt8}, n::Int, key::NTuple{6,UInt8}) + i = 1 + while i <= n - 6 + if (i == 1 || buf[i - 1] == UInt8('\n')) && + buf[i] == key[1] && buf[i + 1] == key[2] && buf[i + 2] == key[3] && + buf[i + 3] == key[4] && buf[i + 4] == key[5] && buf[i + 5] == key[6] + j = i + 6 + while j <= n && (buf[j] == UInt8(' ') || buf[j] == UInt8('\t')); j += 1; end + v = 0 + while j <= n && UInt8('0') <= buf[j] <= UInt8('9') + v = 10v + Int(buf[j] - UInt8('0')); j += 1 + end + return v + end + i += 1 + end + -1 +end + +const VMRSS = Tuple(codeunits("VmRSS:")) +const VMHWM = Tuple(codeunits("VmHWM:")) + +## Sampling +# One sample: each live child's VmRSS and VmHWM, summed per task and over the stage. +function _sample!(p::Probe) + lock(p.lock) do + total = 0 + for t in p.tasks + total += _sample_task!(p, t) + end + total += _sample_task!(p, p.stage) + p.samples += 1 + p.peak_kib = max(p.peak_kib, total) + end + nothing +end + +function _sample_task!(p::Probe, t::TaskTrace) + isempty(t.live) && return 0 + rss = 0; seen = false + for pid in t.live + n = _read_status!(p.buf, p.path, pid) + n > 0 || continue + r = _field_kib(p.buf, n, VMRSS) + r < 0 && continue # a zombie has no memory lines + seen = true + rss += r + t.hwm_kib = max(t.hwm_kib, _field_kib(p.buf, n, VMHWM)) + end + seen || return 0 + t.samples += 1 + t.peak_kib = max(t.peak_kib, rss) + rss +end + +""" + start(run_cfg) -> Union{Probe,Nothing} + +A running probe for one stage, or `nothing` when the run config does not enable +it. Pass the result to `within` and then `finish!`. +""" +function start(cfg) + enabled(cfg) || return nothing + p = Probe(interval_seconds(cfg), ReentrantLock(), TaskTrace[], TaskTrace("(stage)"), + 0, 0, Base.gc_num(), Base.gc_live_bytes(), true, nothing, nothing, nothing, + nothing, Vector{UInt8}(undef, 8192), Vector{UInt8}(undef, 64)) + p.timer = Timer(0; interval = p.interval) + timer = p.timer + # A probe must never fail the stage it measures: a sampling error ends the + # sampling and is reported in the record instead. + p.sampler = Threads.@spawn begin + try + while p.running + _sample!(p) + isopen(timer) || break + wait(timer) + end + catch err + (err isa EOFError && !p.running) || (p.error = sprint(showerror, err)) + end + end + p +end + +""" + within(f, probe) + +Run `f()` with `probe` as the current stage's probe, so the children it starts +(directly or from tasks it spawns) are sampled. With `probe === nothing` this is +just `f()`. +""" +within(f, ::Nothing) = f() +within(f, p::Probe) = with(f, STAGE => p) + +""" + task(f, label) + +Run `f()` as one labelled unit of the current stage. Children it starts are +reported under `label`. Outside a probed stage this is just `f()`. +""" +function task(f, label::AbstractString) + p = STAGE[] + p === nothing && return f() + t = TaskTrace(label) + lock(() -> push!(p.tasks, t), p.lock) + with(f, TASK => t) +end + +# The trace a child started now belongs to, or nothing when nothing is probed. +function _current() + p = STAGE[] + p === nothing && return nothing, nothing + t = TASK[] + p, something(t, p.stage) +end + +""" + watch!(process) -> Union{Int,Nothing} + +Register a just-started child process with the current stage's probe, sample it +at once, and return its pid for `unwatch!`. Called by `Tools._run_killable`. +Returns `nothing`, having done nothing, when the stage is not probed. A child that +has already exited is counted but cannot be sampled. Never throws: a probe must +not fail the stage it measures. +""" +function watch!(proc::Base.Process) + p, t = _current() + p === nothing && return nothing + # `getpid` throws once the child has exited, so the pid is read now or never. + pid = try + Int(getpid(proc)) + catch + nothing + end + lock(p.lock) do + t.processes += 1 + pid === nothing || push!(t.live, pid) + end + pid === nothing || _sample!(p) + pid +end + +# A pipeline of commands: each of its processes, as one task's children. +function watch!(chain::Base.ProcessChain) + pids = Int[] + for proc in chain.processes + pid = watch!(proc) + pid === nothing || push!(pids, pid) + end + isempty(pids) ? nothing : pids +end + +# Anything else `run` can return is not sampled. +watch!(_) = nothing + +""" + unwatch!(pid) + +Stop sampling a child that has been waited for (the pid `watch!` returned), so +the pid cannot be mistaken for a later, unrelated process. A no-op for `nothing`. +""" +unwatch!(::Nothing) = nothing +unwatch!(pids::Vector{Int}) = foreach(unwatch!, pids) +function unwatch!(pid::Int) + p, t = _current() + p === nothing && return nothing + lock(p.lock) do + filter!(!=(pid), t.live) + end + nothing +end + +_task_record(t::TaskTrace) = OrderedDict{String,Any}( + "label" => t.label, + "processes" => t.processes, + "samples" => t.samples, + "peak_rss_kib" => t.peak_kib, + "hwm_kib" => t.hwm_kib, +) + +""" + finish!(probe) -> Union{OrderedDict,Nothing} + +Stop sampling and return the stage's `occupancy` record (the same record on every +later call). `nothing` for `nothing`. +""" +finish!(::Nothing) = nothing +function finish!(p::Probe) + p.record === nothing || return p.record + p.running = false + close(p.timer) + wait(p.sampler) + allocated = Base.GC_Diff(Base.gc_num(), p.gc_before).allocd + traces = [p.tasks; p.stage.processes > 0 ? [p.stage] : TaskTrace[]] + p.record = OrderedDict{String,Any}( + "probe" => PROBE, + "interval_ms" => round(Int, p.interval * 1000), + "samples" => p.samples, + "julia_allocated_bytes" => allocated, + "julia_live_bytes_delta" => Base.gc_live_bytes() - p.live_before, + "child_peak_rss_kib" => p.peak_kib, + "tasks" => Any[_task_record(t) for t in traces], + ) + p.error === nothing || (p.record["error"] = p.error) + p.record +end + +end # module MemoryProbe diff --git a/src/core/provenance.jl b/src/core/provenance.jl index aa232c06..b28912d4 100644 --- a/src/core/provenance.jl +++ b/src/core/provenance.jl @@ -818,11 +818,15 @@ _stage_index(att::Attestation, name::AbstractString) = findfirst(s -> get(s, "name", nothing) == String(name), stages(att)) """ - record_stage!(att, name, env; started, finished, status, commands, outputs) -> Attestation + record_stage!(att, name, env; started, finished, status, commands, outputs, occupancy) -> Attestation Record one stage's own captured environment, replacing any earlier record of that stage. +`occupancy` is the stage's `MemoryProbe` record. It is stored under the stage's +`occupancy` key only when given, so a stage recorded without the probe has +exactly the keys it always had. + Re-running a single stage of an existing run must update that stage alone, because a run directory is an accretion and the other stages' records remain true. The run-level summary is re-derived on every call. @@ -832,7 +836,8 @@ function record_stage!(att::Attestation, name::AbstractString, env::CapturedEnvi finished::AbstractString = timestamp(), status::AbstractString = "ok", commands::AbstractVector = String[], - outputs::AbstractVector = OrderedDict{String,Any}[]) + outputs::AbstractVector = OrderedDict{String,Any}[], + occupancy::Union{AbstractDict,Nothing} = nothing) stage = OrderedDict{String,Any}( "name" => String(name), "started" => String(started), @@ -847,6 +852,7 @@ function record_stage!(att::Attestation, name::AbstractString, env::CapturedEnvi "outputs" => Any[OrderedDict{String,Any}(string(k) => v for (k, v) in o) for o in outputs], ) + isnothing(occupancy) || (stage["occupancy"] = occupancy) index = _stage_index(att, name) isnothing(index) ? push!(stages(att), stage) : (stages(att)[index] = stage) return summarise!(att) diff --git a/src/pipeline/tools.jl b/src/pipeline/tools.jl index d8e12062..ac212228 100644 --- a/src/pipeline/tools.jl +++ b/src/pipeline/tools.jl @@ -14,6 +14,7 @@ export cutadapt, vsearch, multiqc, cdhit, tool_bin, _sq, _run_logged, _run_killa using ..PipelineTypes using ..PipelineLog using ..Config + import ..MemoryProbe _sq(s::AbstractString) = "'" * replace(s, "'" => "'\\''") * "'" @@ -24,13 +25,18 @@ export cutadapt, vsearch, multiqc, cdhit, tool_bin, _sq, _run_logged, _run_killa # Truncation is the stage's own affair, done once at entry via reset_tool_logs. # Run a command and wait for it. If the waiting task is interrupted (a job # cancel), the child process is killed so it cannot keep writing outputs. + # When the stage is probed (occupancy_probe.enabled), the child's memory is + # sampled while it runs; otherwise watch!/unwatch! do nothing. function _run_killable(cmd::Base.AbstractCmd) p = run(cmd; wait=false) + watched = MemoryProbe.watch!(p) try wait(p) catch process_running(p) && kill(p) rethrow() + finally + MemoryProbe.unwatch!(watched) end success(p) || throw(ProcessFailedException(p)) nothing diff --git a/src/server/routes/pipeline.jl b/src/server/routes/pipeline.jl index 306d8c1a..ee1bdedd 100644 --- a/src/server/routes/pipeline.jl +++ b/src/server/routes/pipeline.jl @@ -9,6 +9,7 @@ using JSON3, RCall using ..RRuntime: with_r_lock const Provenance = MetaManifold.Provenance +const MemoryProbe = MetaManifold.MemoryProbe ## Tagging helpers @@ -180,17 +181,26 @@ end # # A stage that skipped leaves its earlier record alone: writing the current # environment over it would attribute old outputs to new tool versions. +# +# With `occupancy_probe.enabled: true` in the run config, the stage's memory is +# measured (MemoryProbe) and recorded under the stage's `occupancy` key. Off, the +# probe is `nothing` and the record is exactly what it was without it. function _attest(f::Function, att, env, name::AbstractString; config_path::String) started = Provenance.timestamp() ran_before = stage_run_count(config_path) + probe = MemoryProbe.start(get(get(att.doc, "config", Dict()), "run_config", nothing)) try - result = f() + result = MemoryProbe.within(f, probe) stage_run_count(config_path) > ran_before && - Provenance.record_stage!(att, name, env; started, outputs=_stage_outputs(result)) + Provenance.record_stage!(att, name, env; started, outputs=_stage_outputs(result), + occupancy=MemoryProbe.finish!(probe)) return result catch - Provenance.record_stage!(att, name, env; started, status="failed") + Provenance.record_stage!(att, name, env; started, status="failed", + occupancy=MemoryProbe.finish!(probe)) rethrow() + finally + MemoryProbe.finish!(probe) end end @@ -261,9 +271,13 @@ function _run_full_pipeline(study::String, run::String, db_config::String, # OTU search it claims to cover. asv_tax, otu_tax = if vsearch_enabled _attest(att, env, "vsearch"; config_path) do - asv_task = Threads.@spawn(vsearch(project, asvs, dbs["$(db_name)_vsearch"])) + asv_task = Threads.@spawn MemoryProbe.task("vsearch:asv") do + vsearch(project, asvs, dbs["$(db_name)_vsearch"]) + end otu_task = (swarm_enabled && !isnothing(otus)) ? - Threads.@spawn(vsearch(project, otus, dbs["$(db_name)_vsearch"])) : nothing + Threads.@spawn(MemoryProbe.task("vsearch:otu") do + vsearch(project, otus, dbs["$(db_name)_vsearch"]) + end) : nothing # Wait for both before rethrowing, so a failed ASV search cannot leave the OTU search running. tasks = filter(!isnothing, Any[asv_task, otu_task]) foreach(t -> (try wait(t) catch end), tasks) @@ -438,12 +452,18 @@ function _run_stage(study::String, run::String, stage::String, end elseif step == "vsearch" if vsearch_enabled + # Serial here, concurrent in the full pipeline: the same two + # searches in the two shapes the occupancy probe distinguishes. _record("vsearch") do asvs = _asvresult_from_disk(run_dir) - vsearch(project, asvs, dbs["$(db_name)_vsearch"]) + MemoryProbe.task("vsearch:asv") do + vsearch(project, asvs, dbs["$(db_name)_vsearch"]) + end otus = _oturesult_from_disk(run_dir) if !isnothing(otus) && swarm_enabled - vsearch(project, otus, dbs["$(db_name)_vsearch"]) + MemoryProbe.task("vsearch:otu") do + vsearch(project, otus, dbs["$(db_name)_vsearch"]) + end end end end diff --git a/test/occupancy_disagreement_test.jl b/test/occupancy_disagreement_test.jl new file mode 100644 index 00000000..b609ed38 --- /dev/null +++ b/test/occupancy_disagreement_test.jl @@ -0,0 +1,320 @@ +# SPDX-License-Identifier: AGPL-3.0-only +# SPDX-FileCopyrightText: 2026 Jonathan D.A. Jewell (hyperpolymath) +# +# MMW-01: the occupancy probe, and the two-thread disagreement test. +# +# The vsearch stage runs two independent searches, ASV and OTU, concurrently in +# the full pipeline and one after the other in a single-stage re-run. The +# high-water-mark algebra (occupancy-types, resource-algebra `Instances/HWM`) +# predicts: +# +# * cost commutes: the work done, measured as Julia's allocation total for the +# stage, is the same in every order (choreographic-types A-2). This is +# ASSERTED, and it can fail. +# * each search's own peak is the same in every order; +# * the stage's peak does not commute: serially it is the larger of the two +# searches' peaks (each child exits holding nothing, so its drift is 0), while +# concurrently it can approach their sum (A-3). These are RECORDED in the test +# log, not asserted. +# +# If the stage peak came out the same in every order, the prediction would be +# falsified for this workload: the searches' memory would not overlap. +# +# The fixture is the smallest of a calibrated ladder (250, 500, 1000, 2000 +# reference sequences of 400 bp; 300 ASV and 150 OTU queries) for which, in five +# repeats each on a 4-core host, both searches were resident together for at +# least 15 samples at 10 ms and the concurrent stage peak was at least 1.9 times +# the larger search's: 1000 sequences (17-20 samples, 2.0 times). 250 and 500 +# also overlap, by 12-18 samples, too close to the edge for a slower runner. It is +# generated here, deterministically, so no data is checked in. +# +# vsearch runs with --threads 1 here. With several threads it writes its hits in +# whatever order its threads finish, and `_sort_hits_by_query!` then does a +# different amount of work (its sort key splits each line on every comparison), +# which moved the stage's allocation total by up to 12% between repeats of the +# SAME order. On one thread the output order is fixed, and the three orders +# agreed to within 0.08% over five repeats each. Run directly with +# julia --project=. -t 2 test/occupancy_disagreement_test.jl +# and set OCCUPANCY_RESULTS= to also write the numbers as YAML. + +using Test, YAML, OrderedCollections +using MetaManifold + +const MP = MetaManifold.MemoryProbe +const PV = MetaManifold.Provenance +const TL = MetaManifold.Tools + +const PROBE_ON = Dict("occupancy_probe" => Dict("enabled" => true, "interval_ms" => 10)) + +# Relative spread of the stage's Julia allocation total across the three orders +# that still counts as "the same cost". Measured: under 0.08% (see above). +const COST_TOLERANCE = 0.01 + +## Fixture +# A 64-bit LCG, so the fixture is the same bytes on every Julia version (the +# streams of Random's generators are not guaranteed to be). +mutable struct _LCG + s :: UInt64 +end +_next!(g::_LCG) = (g.s = g.s * 0x5851f42d4c957f2d + 0x14057b7ef767814f; g.s >> 33) + +const _NT = (UInt8('A'), UInt8('C'), UInt8('G'), UInt8('T')) +_randseq(g, n) = String([_NT[_next!(g) % 4 + 1] for _ in 1:n]) + +# A copy of `s` with about `per_1000` substitutions per thousand bases: close +# enough to its reference (~96% identity) to pass vsearch's --id 0.75. +function _mutate(g, s, per_1000) + b = Vector{UInt8}(s) + for i in eachindex(b) + _next!(g) % 1000 < per_1000 && (b[i] = _NT[_next!(g) % 4 + 1]) + end + String(b) +end + +function occupancy_fixture(dir; refs = 1000, asv = 300, otu = 150, len = 400) + g = _LCG(0x4d4d57) + mkpath(dir) + seqs = [_randseq(g, len) for _ in 1:refs] + open(joinpath(dir, "ref.fasta"), "w") do io + for (i, s) in enumerate(seqs) + println(io, ">ref$i;tax=k:K$(i % 7)") + println(io, s) + end + end + for (name, n) in (("asv", asv), ("otu", otu)) + open(joinpath(dir, "$name.fasta"), "w") do io + for i in 1:n + println(io, ">$name$i") + println(io, _mutate(g, seqs[_next!(g) % refs + 1], 40)) + end + end + end + dir +end + +## Running the stage in one order +function _search(dir, out, name) + TL.vsearch(joinpath(dir, "$name.fasta"), joinpath(dir, "ref.fasta"), joinpath(out, name); + optional_args = "--id 0.75 --query_cov 0.8 --threads 1", + vsearch_bin = something(Sys.which("vsearch"), "vsearch")) +end + +const ORDERS = ("asv_then_otu", "otu_then_asv", "concurrent") + +# The stage's work in `order`, under a fresh probe. Returns the probe's record. +function run_order(dir, out, order) + mkpath(out) + probe = MP.start(PROBE_ON) + MP.within(probe) do + if order == "asv_then_otu" + MP.task(() -> _search(dir, out, "asv"), "vsearch:asv") + MP.task(() -> _search(dir, out, "otu"), "vsearch:otu") + elseif order == "otu_then_asv" + MP.task(() -> _search(dir, out, "otu"), "vsearch:otu") + MP.task(() -> _search(dir, out, "asv"), "vsearch:asv") + else + # The shape of the full pipeline's vsearch stage. + a = Threads.@spawn MP.task(() -> _search(dir, out, "asv"), "vsearch:asv") + o = Threads.@spawn MP.task(() -> _search(dir, out, "otu"), "vsearch:otu") + fetch(a); fetch(o) + end + end + MP.finish!(probe) +end + +_task(rec, label) = only(t for t in rec["tasks"] if t["label"] == label) + +@testset "Occupancy probe (MMW-01)" begin + + @testset "off unless the run config enables it" begin + @test !MP.enabled(Dict()) + @test !MP.enabled(Dict("occupancy_probe" => Dict())) + @test !MP.enabled(Dict("occupancy_probe" => Dict("enabled" => "true"))) + @test !MP.enabled(Dict("occupancy_probe" => true)) + @test MP.enabled(PROBE_ON) + @test MP.start(Dict()) === nothing + @test MP.finish!(nothing) === nothing + @test MP.within(() -> 42, nothing) == 42 + @test MP.task(() -> 7, "outside a probe") == 7 + @test MP.interval_seconds(Dict()) == 0.010 + @test MP.interval_seconds(Dict("occupancy_probe" => Dict("interval_ms" => 25))) == 0.025 + @test MP.interval_seconds(Dict("occupancy_probe" => Dict("interval_ms" => 0))) == 0.010 + end + + @testset "flag off: the stage record is unchanged" begin + # The keys a stage record had before the probe existed. With the probe + # off, `finish!` gives `nothing` and the record must be exactly these. + before = ["name", "started", "finished", "status", "degraded", + "degraded_components", "runtimes", "tools", "databases", + "commands", "outputs"] + att = PV.Attestation() + PV.record_stage!(att, "vsearch", PV.CapturedEnvironment(); + occupancy = MP.finish!(MP.start(Dict()))) + @test collect(keys(only(att.doc["stages"]))) == before + + # Negative control: with the probe on, the key is there. + on = PV.Attestation() + PV.record_stage!(on, "vsearch", PV.CapturedEnvironment(); + occupancy = MP.finish!(MP.start(PROBE_ON))) + @test collect(keys(only(on.doc["stages"]))) == [before; "occupancy"] + end + + @testset "attestation round trip; older records still parse" begin + mktempdir() do tmp + att = PV.Attestation() + rec = MP.finish!(MP.start(PROBE_ON)) + PV.record_stage!(att, "vsearch", PV.CapturedEnvironment(); occupancy = rec) + PV.record_stage!(att, "swarm", PV.CapturedEnvironment()) # an unprobed stage + path = joinpath(tmp, "attestation.yml") + PV.write_attestation(att, path) + back = PV.read_attestation(path) + stages = Dict(s["name"] => s for s in back.doc["stages"]) + @test stages["vsearch"]["occupancy"]["probe"] == "MemoryProbe 1" + @test stages["vsearch"]["occupancy"]["interval_ms"] == 10 + @test !haskey(stages["swarm"], "occupancy") + + # A record written before the probe existed merges with a new one. + old = PV.Attestation() + PV.record_stage!(old, "cutadapt", PV.CapturedEnvironment()) + PV.write_attestation(old, path; merge_existing = false) + PV.write_attestation(att, path) + merged = Dict(s["name"] => s for s in PV.read_attestation(path).doc["stages"]) + @test !haskey(merged["cutadapt"], "occupancy") + @test haskey(merged["vsearch"], "occupancy") + end + end + + if !Sys.islinux() + @info "Occupancy probe: not Linux, so there is no /proc to sample; the child-memory tests are SKIPPED" + else + @testset "samples a child's resident memory" begin + probe = MP.start(PROBE_ON) + MP.within(probe) do + # bash holds a 64 MiB string for half a second. + MP.task("64MiB") do + TL._run_killable(`bash -c 'x=$(head -c 67108864 /dev/zero | tr "\0" a); sleep 0.5; : "${#x}"'`) + end + # Negative control: a child that allocates nothing. + MP.task("sleep") do + TL._run_killable(`sleep 0.5`) + end + end + rec = MP.finish!(probe) + big, small = _task(rec, "64MiB"), _task(rec, "sleep") + @test big["processes"] == 1 && big["samples"] > 0 + @test big["peak_rss_kib"] >= 64 * 1024 + @test big["hwm_kib"] >= big["peak_rss_kib"] + @test small["samples"] > 0 + @test 0 < small["peak_rss_kib"] < 16 * 1024 + # Serial: the stage peak is the larger task's, not the sum. + @test rec["child_peak_rss_kib"] < big["peak_rss_kib"] + small["peak_rss_kib"] + @test !haskey(rec, "error") + end + + @testset "through the pipeline's _attest, flag off and on" begin + SV = MetaManifold.Server + mktempdir() do tmp + cfg_path = joinpath(tmp, "run_config.yml") + for probed in (false, true) + cfg = Dict{String,Any}("vsearch" => Dict("enabled" => true)) + probed && (cfg["occupancy_probe"] = Dict("enabled" => true)) + YAML.write_file(cfg_path, cfg) + att = PV.Attestation(; config = cfg, config_sha256 = PV.file_sha256(cfg_path)) + SV._attest(att, PV.CapturedEnvironment(), "vsearch"; config_path = cfg_path) do + MetaManifold.Config._begin_section(cfg_path, "vsearch", joinpath(tmp, "vsearch.hash")) + MP.task(() -> TL._run_killable(`sleep 0.2`), "vsearch:asv") + nothing + end + stage = only(att.doc["stages"]) + @test stage["status"] == "ok" + @test haskey(stage, "occupancy") == probed + probed && @test only(stage["occupancy"]["tasks"])["label"] == "vsearch:asv" + + # A stage that throws is recorded as failed, with its + # occupancy when probed, and the error still propagates. + failed = PV.Attestation(; config = cfg, config_sha256 = PV.file_sha256(cfg_path)) + @test_throws ProcessFailedException SV._attest(failed, PV.CapturedEnvironment(), "swarm"; + config_path = cfg_path) do + TL._run_killable(`false`) + end + stage = only(failed.doc["stages"]) + @test stage["status"] == "failed" + @test haskey(stage, "occupancy") == probed + end + end + end + + @testset "a pipeline of commands counts each process" begin + probe = MP.start(PROBE_ON) + MP.within(() -> MP.task(() -> TL._run_killable(pipeline(`sleep 0.3`, `cat`)), "chain"), probe) + t = _task(MP.finish!(probe), "chain") + @test t["processes"] == 2 + @test t["samples"] > 0 + end + + @testset "a child that exits at once is counted, not sampled" begin + probe = MP.start(PROBE_ON) + MP.within(() -> MP.task(() -> TL._run_killable(`true`), "instant"), probe) + t = _task(MP.finish!(probe), "instant") + @test t["processes"] == 1 + @test t["samples"] == 0 || t["peak_rss_kib"] > 0 + end + + vsearch_bin = Sys.which("vsearch") + if isnothing(vsearch_bin) + @info "vsearch not on PATH - the two-thread disagreement test is SKIPPED" + else + @testset "two-thread disagreement: vsearch ASV and OTU in three orders" begin + mktempdir() do tmp + dir = occupancy_fixture(joinpath(tmp, "fixture")) + # Warm-up: compile every path once, so no order pays for it. + for order in ORDERS + run_order(dir, joinpath(tmp, "warmup", order), order) + end + GC.gc() + records = OrderedDict{String,Any}() + for order in ORDERS + GC.gc() + records[order] = run_order(dir, joinpath(tmp, order), order) + end + + # The same work: identical hits in every order. + for name in ("asv", "otu") + hits = [read(joinpath(tmp, o, name, "taxonomy.tsv"), String) for o in ORDERS] + @test !isempty(first(hits)) + @test allequal(hits) + end + + # Cost commutes (asserted). + cost = [records[o]["julia_allocated_bytes"] for o in ORDERS] + spread = (maximum(cost) - minimum(cost)) / maximum(cost) + @test spread <= COST_TOLERANCE + + # Peaks (recorded, not asserted). + for o in ORDERS + r = records[o] + a, u = _task(r, "vsearch:asv"), _task(r, "vsearch:otu") + @info "occupancy" order = o julia_allocated_bytes = r["julia_allocated_bytes"] julia_live_bytes_delta = r["julia_live_bytes_delta"] child_peak_rss_kib = r["child_peak_rss_kib"] asv_peak_rss_kib = a["peak_rss_kib"] asv_hwm_kib = a["hwm_kib"] otu_peak_rss_kib = u["peak_rss_kib"] otu_hwm_kib = u["hwm_kib"] samples = r["samples"] + end + conc = records["concurrent"] + serial_max = maximum(records[o]["child_peak_rss_kib"] for o in ORDERS[1:2]) + overlap = conc["child_peak_rss_kib"] > serial_max + @info "occupancy: concurrent stage peak vs the larger serial stage peak" concurrent = conc["child_peak_rss_kib"] serial = serial_max overlapped = overlap cost_spread = round(spread; sigdigits = 3) + overlap || @warn "occupancy: the concurrent searches did not overlap in memory on this runner, so this run cannot show the order dependence" + + path = get(ENV, "OCCUPANCY_RESULTS", "") + if !isempty(path) + io = IOBuffer() + run(pipeline(`$vsearch_bin --version`; stdout = io, stderr = io)) + version = first(split(String(take!(io)), '\n')) + YAML.write_file(path, OrderedDict{String,Any}( + "vsearch" => version, + "julia" => string(VERSION), "threads" => Threads.nthreads(), + "cpus" => Sys.CPU_THREADS, "cost_spread" => spread, + "orders" => records)) + end + end + end + end + end +end diff --git a/test/runtests.jl b/test/runtests.jl index 4a8f47fa..a406baa1 100644 --- a/test/runtests.jl +++ b/test/runtests.jl @@ -83,6 +83,12 @@ using MetaManifold.Categories, MetaManifold.CompositionLibrary include(joinpath(@__DIR__, "unit", f)) end + ## Occupancy probe and the two-thread disagreement test (MMW-01), with the + ## full suite. Its vsearch part skips, saying so, when vsearch is not on PATH. + if isempty(filter(a -> !startswith(a, "--"), ARGS)) + include("occupancy_disagreement_test.jl") + end + ## Integration tests (opt-in) if RUN_INTEGRATION using MetaManifold.DADA2, MetaManifold.OTUPipeline