diff --git a/.github/workflows/occupancy.yml b/.github/workflows/occupancy.yml index 4fc5ec49..8e1fdf10 100644 --- a/.github/workflows/occupancy.yml +++ b/.github/workflows/occupancy.yml @@ -22,6 +22,9 @@ on: - src/pipeline/tools.jl - src/server/routes/pipeline.jl - test/occupancy_disagreement_test.jl + - test/unit/test_occupancy_grade.jl + - scripts/check-occupancy-grades.jl + - scripts/lint-occupancy-readings.sh - docs/occupancy/** - docs/occupancy-2026-10.adoc - scripts/check-occupancy.sh @@ -35,6 +38,9 @@ on: - src/pipeline/tools.jl - src/server/routes/pipeline.jl - test/occupancy_disagreement_test.jl + - test/unit/test_occupancy_grade.jl + - scripts/check-occupancy-grades.jl + - scripts/lint-occupancy-readings.sh - docs/occupancy/** - docs/occupancy-2026-10.adoc - scripts/check-occupancy.sh @@ -107,6 +113,15 @@ jobs: - name: Three orders of the vsearch stage run: julia --project=. -t 2 test/occupancy_disagreement_test.jl + # Fork issue #105: the grades the recorded run's windows imply, read back + # against its own measurement (docs/occupancy/schedules.yml). + - name: Graded reading of the recorded run + run: julia --project=. scripts/check-occupancy-grades.jl + + # And no production code reads a peak without those grades. + - name: Occupancy readings are graded + run: bash scripts/lint-occupancy-readings.sh + typed-peak: name: Occupancy typed peak runs-on: ubuntu-24.04 diff --git a/docs/occupancy-2026-10.adoc b/docs/occupancy-2026-10.adoc index b7323e45..274fb1ee 100644 --- a/docs/occupancy-2026-10.adoc +++ b/docs/occupancy-2026-10.adoc @@ -18,8 +18,9 @@ on 2026-10-14, and checks it against the typed reading. * `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`); + process it started (`peak_rss_kib`), the kernel's own VmHWM for it + (`hwm_kib`), and the sample window its children were live in + (`first_sample`, `last_sample`) — what its grade is read from; * `child_peak_rss_kib`: the largest sum of the stage's children's VmRSS at one sampling instant (the stage's occupancy). @@ -65,6 +66,36 @@ interval 10 ms. coincide. Here they did: each search's memory plateaus once its database is loaded. +== Reading the numbers + +A recorded `peak_rss_kib` is a reading, not an occupancy: what a stage occupied +depends on what the next stage inherits from it. So the probe grades a stage's +whole window (`MemoryProbe`'s `task_grades`, `stage_grade`, +`predicted_peak_rss_kib`, `cost_bound_rss_kib`, `reclaimed_kib`) and attaches the +result to a recorded stage as `grades`; a record with no sample windows is left +ungraded rather than graded on an assumed order. Tasks are read in sampled order +(first sample, ties by record order); a task's drift is its peak when the next +task still held it, and 0 when it had exited. + +For the three orders above, the algebra's predicted peak reproduces the measured +stage peak exactly — 9204, 9192 and 18380 KiB — and the slack against the cost +bound is what the cuts reclaimed: 9180, 9152 and 0 KiB. +`scripts/check-occupancy-grades.jl` checks that equality, occupancy≤cost and +non-negative reclamation against this page's raw record, rebuilding each order's +windows from `docs/occupancy/schedules.yml` (the file records no windows of its +own; the Session IR texts in this directory state the same interleavings). The +reclamation readings are theorems over the pinned siblings at these numbers: +`docs/occupancy/GradedStages.agda` — `no-reclaim` for the concurrent stage, +`reclaimed` for the serial one, where 9204 KiB are released at the frontier and +the cost reading counts them — and the two protocol-level statements +(`occupancy-at-cut`, `protocol-occupancy≤cost`) as the siblings' own two-step +over the stage's timeline. `docs/occupancy/reject` gains a fourth rejection, for +reading the concurrent stage's drift as 0. + +`scripts/lint-occupancy-readings.sh` keeps production code from reading the key +directly: outside `src/core/memory_probe.jl`, an occupancy figure must go through +those helpers. + == Status [cols="1,3,1,2",options="header"] @@ -73,7 +104,8 @@ interval 10 ms. | 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`) +| 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`; three rejections in `docs/occupancy/reject`) +| MMW-01.5 | The recorded run's grades reproduce its measured stage peaks, and the reclamation the pinned siblings' laws read off them is what the cuts freed | TESTED | `scripts/check-occupancy-grades.jl` (the recorded record), `docs/occupancy/GradedStages.agda` (`no-reclaim`, `reclaimed`, `occupancy-at-cut` at the measured numbers), `test/unit/test_occupancy_grade.jl` | 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 |=== @@ -106,6 +138,8 @@ interval 10 ms. ---- 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 +julia --project=. scripts/check-occupancy-grades.jl # the recorded run's grades +bash scripts/lint-occupancy-readings.sh # no ungraded readings in src/ ---- `scripts/fetch-siblings.sh` fetches the siblings at their pins into diff --git a/docs/occupancy/GradedStages.agda b/docs/occupancy/GradedStages.agda new file mode 100644 index 00000000..774c2edd --- /dev/null +++ b/docs/occupancy/GradedStages.agda @@ -0,0 +1,95 @@ +{-# OPTIONS --safe --without-K #-} +-- SPDX-License-Identifier: AGPL-3.0-only +-- SPDX-FileCopyrightText: 2026 Jonathan D.A. Jewell (hyperpolymath) +-- +-- The vsearch stage's grades, and occupancy-types' reclamation theorems at the +-- measured numbers. Fork issue #105 asks for exactly this: grade the stages +-- (alloc/hold/release) and compare the predicted peak with the measured one. +-- TypedPeak.agda computes the typed peak by the algebra's sequence law; this +-- module adds the grades the measurement itself gives, and the laws that say +-- what the difference between a peak and a cost reading is: occupancy≤cost, and +-- the strict reclamation at a cut where something is released. +-- +-- Consumed, never re-declared, from the siblings scripts/fetch-siblings.sh pins +-- and scripts/check-occupancy.sh checks: +-- occupancy-types af957280075699c659de0abd9a3b0defade77309 +-- resource-algebra cf9235c532712a783a404c359a1055f40134794a +-- Everything is stated in this file's own Timeline instance (`TH`, as +-- TypedPeak.agda does): the Grade-level laws are Reclaim's as they stand, and +-- the two protocol-level ones are Reclaim's own two-step — its `cut-factor` over +-- its `reclaim-law`/`occupancy≤cost` — applied to `xs` and `ys` of that instance. +-- +-- A grade is `hwm peak drift _`: what an interaction reached, and what it still +-- holds when the next one begins. The peaks below are the 2026-10-14 run's +-- (docs/occupancy-2026-10.adoc, docs/occupancy/results-2026-10-14.yml), in KiB. +module GradedStages where + +open import ResourceAlgebra.Prelude +open import ResourceAlgebra.Instances.HWM +open import OccupancyTypes.Order +open import OccupancyTypes.Reclaim + +import OccupancyTypes.Timeline +module TH = OccupancyTypes.Timeline HWM + +-- A search that has not been waited for when the next one starts: it still holds +-- everything it reached, so the cut reclaims none of it. +still : Nat → Grade +still p = hwm p p (≤-refl p) + +-- A search that has exited before the next one starts: it holds nothing, so the +-- cut reclaims all of it. +gone : Nat → Grade +gone p = hwm p 0 z≤n + +-- The full pipeline runs the two searches together: ASV is still resident when +-- OTU starts, so the stage peak is their sum and nothing is reclaimed — +-- Reclaim.no-reclaim's equality, whose `refl` argument discharges the theorem's +-- own hypothesis (drift = peak) at the measured grade. +run-together : peak (still 9176 ⊗h gone 9204) ≡ 18380 +run-together = no-reclaim (still 9176) refl (gone 9204) + +-- A single-stage re-run runs them one after the other: ASV has exited, so the +-- cut reclaims its 9204 KiB and the stage peak is the larger search's +-- (Reclaim.reclaim-law, by `refl`). +run-in-turn : peak (gone 9204 ⊗h gone 9180) ≡ 9204 +run-in-turn = refl + +-- The same claim as Reclaim.reclaimed states it: with drift < peak at the cut +-- and a suffix that does anything, occupancy is strictly below cost. 9204 KiB +-- are released at the frontier, so the cost reading (18384 KiB) overstates the +-- stage. +run-in-turn +-- +-- Must be REFUSED: reading the concurrent stage as one that hands on nothing. +-- Running the two searches together leaves the ASV search resident when the OTU +-- search starts, so the stage's drift is that search's peak — 9176 KiB, what a +-- suffix would see — and not 0. Grading the drift as 0 would read the stage's +-- COST as its occupancy, which is the very reading the measured overlap exists +-- to refute. +module ConcurrentDriftIsZero where + +open import ResourceAlgebra.Prelude +open import ResourceAlgebra.Instances.HWM +open import GradedStages + +wrong : drift (still 9176 ⊗h gone 9204) ≡ 0 +wrong = refl diff --git a/docs/occupancy/schedules.yml b/docs/occupancy/schedules.yml new file mode 100644 index 00000000..dfdedb81 --- /dev/null +++ b/docs/occupancy/schedules.yml @@ -0,0 +1,21 @@ +# SPDX-License-Identifier: AGPL-3.0-only +# SPDX-FileCopyrightText: 2026 Jonathan D.A. Jewell (hyperpolymath) +# +# The interleaving each order in results-2026-10-14.yml ran with, as the sample +# window [first, last] of each of its tasks, in the task order the record keeps. +# These are the same schedules the Session IR texts in this directory state +# (vsearch-concurrent.sir runs the two searches together; the other two run them +# one after the other), expressed in the probe's own sample indices: the windows +# of a serial order do not overlap, and the concurrent order's do. +# +# Read by test/unit/test_occupancy_grade.jl and scripts/check-occupancy-grades.jl, +# so the two cannot disagree about what the recorded run did. +asv_then_otu: + - [1, 2] + - [3, 4] +otu_then_asv: + - [1, 2] + - [3, 4] +concurrent: + - [1, 4] + - [2, 5] diff --git a/scripts/check-occupancy-grades.jl b/scripts/check-occupancy-grades.jl new file mode 100755 index 00000000..ce160d09 --- /dev/null +++ b/scripts/check-occupancy-grades.jl @@ -0,0 +1,72 @@ +# SPDX-License-Identifier: AGPL-3.0-only +# SPDX-FileCopyrightText: 2026 Jonathan D.A. Jewell (hyperpolymath) +# +# Gate for the graded reading of the recorded run (fork issue #105): rebuild each +# order's timeline from its sample window (docs/occupancy/schedules.yml), grade it +# with the module the probe itself uses (src/core/memory_probe.jl — included +# rather than the whole package, so this needs OrderedCollections and YAML only), +# and compare the algebra's predicted peak with the measured one. +# +# The recorded run is fixed evidence, so the comparison is an equality: this is +# the reading docs/occupancy-2026-10.adoc publishes, and a change that moves it +# must be a deliberate one. Two further claims hold of every order and are +# checked: the prediction is never above the cost bound (occupancy≤cost) and the +# slack between them is never negative (what the cuts reclaim). +# +# Run from the repository root: julia --project=. scripts/check-occupancy-grades.jl +include(joinpath(@__DIR__, "..", "src", "core", "memory_probe.jl")) +using .MemoryProbe +using YAML + +const ROOT = normpath(joinpath(@__DIR__, "..")) +const RESULTS = joinpath(ROOT, "docs", "occupancy", "results-2026-10-14.yml") +const SCHEDULES = joinpath(ROOT, "docs", "occupancy", "schedules.yml") + +orders = YAML.load_file(RESULTS)["orders"] +schedules = YAML.load_file(SCHEDULES) + +failures = 0 +function fail(message) + global failures += 1 + println("FAIL: ", message) +end + +println(rpad("order", 16), rpad("predicted", 11), rpad("measured", 11), + rpad("cost", 8), "reclaimed") +# An order's timeline, from the peaks the probe recorded and the window each task +# ran in. Nothing else is taken from the schedule: the grades are the +# measurement's. +function timeline(rec, windows) + tasks = Any[Dict{String,Any}("label" => t["label"], + "peak_rss_kib" => t["peak_rss_kib"], + "first_sample" => windows[i][1], + "last_sample" => windows[i][2]) + for (i, t) in enumerate(rec["tasks"])] + Dict{String,Any}("child_peak_rss_kib" => rec["child_peak_rss_kib"], "tasks" => tasks) +end + +for name in sort(collect(keys(orders))) + rec = orders[name] + occ = timeline(rec, schedules[name]) + predicted = MemoryProbe.predicted_peak_rss_kib(occ) + cost = MemoryProbe.cost_bound_rss_kib(occ) + reclaimed = MemoryProbe.reclaimed_kib(occ) + measured = rec["child_peak_rss_kib"] + println(rpad(name, 16), rpad(string(predicted), 11), rpad(string(measured), 11), + rpad(string(cost), 8), string(reclaimed)) + + predicted == measured || + fail("$name: predicted $predicted KiB != measured $measured KiB") + predicted <= cost || + fail("$name: predicted $predicted KiB above the cost bound $cost KiB (occupancy≤cost)") + reclaimed >= 0 || + fail("$name: reclaimed $reclaimed KiB is negative") + d = MemoryProbe.disagreement(occ) + d === nothing || fail("$name: $d") +end + +if failures > 0 + println("check-occupancy-grades: $failures failure(s)") + exit(1) +end +println("check-occupancy-grades: OK — the prediction reproduces the recorded measurement") diff --git a/scripts/check-occupancy.sh b/scripts/check-occupancy.sh index f9bcba36..c43f28e4 100755 --- a/scripts/check-occupancy.sh +++ b/scripts/check-occupancy.sh @@ -4,8 +4,9 @@ # # 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; +# 2. TypedPeak.agda and GradedStages.agda type-check 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 @@ -22,6 +23,29 @@ logs=_build/check-logs mkdir -p "$logs" failures=0 +# GitHub renders an `::error` line as a check annotation, and annotations -- unlike +# job logs -- can be read back through the API. A failure that only happens in CI +# must be readable from a machine that cannot download the log, so every failure +# below says what it was and repeats the prover's own last lines: % escapes to +# %25 and newlines to %0A, as the workflow-command syntax requires. +annotate() { # title file log + [ -n "${GITHUB_ACTIONS:-}" ] || return 0 + printf '::error file=%s,title=%s::%s\n' "$2" "$1" \ + "$(awk 'NR > 1 { printf "%%0A" } { gsub(/%/, "%25"); printf "%s", $0 }' "$3" | + tail -c 6000)" +} + +# One Agda module that must type-check, with its own log. +check_agda() { # file label + if ! agda "${flags[@]}" "$1" >"$logs/$2.log" 2>&1; then + cat "$logs/$2.log" + annotate "Agda refused $2" "$1" "$logs/$2.log" + echo "ERROR: $1 does not type-check" >&2 + exit 1 + fi + echo "PASS: $2" +} + # Order -> Session IR file and TypedPeak lemma. orders=(asv-then-otu otu-then-asv concurrent) @@ -36,18 +60,26 @@ for o in "${orders[@]}"; do 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" +check_agda docs/occupancy/TypedPeak.agda TypedPeak.agda + +# The graded reading (fork issue #105): occupancy-types' Reclaim theorems at the +# measured grades, from the same pinned siblings. The recorded run's grades +# against its measurement are checked by scripts/check-occupancy-grades.jl, which +# needs the project's Julia packages and so runs in the disagreement job. +check_agda docs/occupancy/GradedStages.agda GradedStages.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 + cat "$log" + annotate "Invalid claim accepted: $name" "$file" "$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 + cat "$log" + annotate "$name failed for the wrong reason" "$file" "$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 @@ -59,6 +91,8 @@ for o in "${orders[@]}"; do 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)) + echo "counted ${counted[$o]:-?} KiB != typed ${typed:-?} KiB" >"$logs/counted-$o.log" + annotate "Counted bytes != typed peak ($o)" docs/occupancy/TypedPeak.agda "$logs/counted-$o.log" fi done diff --git a/scripts/lint-occupancy-readings.sh b/scripts/lint-occupancy-readings.sh new file mode 100755 index 00000000..114779a0 --- /dev/null +++ b/scripts/lint-occupancy-readings.sh @@ -0,0 +1,40 @@ +#!/usr/bin/env bash +# SPDX-License-Identifier: AGPL-3.0-only +# SPDX-FileCopyrightText: 2026 Jonathan D.A. Jewell (hyperpolymath) +# +# Fork issue #105: a task's recorded `peak_rss_kib` is a reading, not an +# occupancy. What a stage occupied is determined only by what the next stage +# inherits from it — the drift — which is why the probe grades a whole window and +# why src/core/memory_probe.jl's helpers (`graded`, `predicted_peak_rss_kib`, +# `cost_bound_rss_kib`, `reclaimed_kib`, `task_grades`) exist. Code outside the +# probe that reaches for the key itself is comparing measured figures without +# that context; on the recorded run that reading is off by 9180 KiB for +# `asv_then_otu` and 9204 KiB for `otu_then_asv` (the reclaimed suffix). +# +# So: production code may name the key only in src/core/memory_probe.jl, where +# the record is built and graded. Tests may read it (they assert the recorded +# numbers directly, which is fine — they are evidence, not presentation). +# +# Run from the repository root: bash scripts/lint-occupancy-readings.sh +set -u -o pipefail + +if [[ ! -f src/core/memory_probe.jl ]]; then + echo "lint-occupancy-readings: run from the repository root" >&2 + exit 2 +fi + +failures=0 +while IFS= read -r hit; do + [[ -z "$hit" ]] && continue + echo "ERROR: ungraded occupancy reading: $hit" >&2 + failures=$((failures + 1)) +done < <(grep -rn --include='*.jl' '"peak_rss_kib"' src/ | + grep -v '^src/core/memory_probe\.jl:' || true) + +if (( failures > 0 )); then + echo "lint-occupancy-readings: $failures ungraded reading(s); use the grading helpers in" >&2 + echo "MemoryProbe (see docs/occupancy-2026-10.adoc, 'Reading the numbers') instead." >&2 + exit 1 +fi + +echo "lint-occupancy-readings: OK — every peak_rss_kib reading in src/ goes through the probe's grades" diff --git a/src/core/memory_probe.jl b/src/core/memory_probe.jl index 6406c37a..163f8bc9 100644 --- a/src/core/memory_probe.jl +++ b/src/core/memory_probe.jl @@ -50,8 +50,16 @@ mutable struct TaskTrace samples :: Int # samples that found at least one live process peak_kib :: Int hwm_kib :: Int + # The sample window the task's children were live in: the index of the first + # sample taken after the first of them started, and of the one taken after + # the last was waited for. 0 means never. Two tasks overlap when the later + # one's `first_sample` falls at or before the earlier one's `last_sample`, + # which is the interleaving the grade algebra sequences over. + first_sample :: Int + last_sample :: Int end -TaskTrace(label::AbstractString) = TaskTrace(String(label), Int[], 0, 0, 0, 0) +TaskTrace(label::AbstractString) = + TaskTrace(String(label), Int[], 0, 0, 0, 0, 0, 0) mutable struct Probe interval :: Float64 @@ -269,7 +277,10 @@ function watch!(proc::Base.Process) end lock(p.lock) do t.processes += 1 - pid === nothing || push!(t.live, pid) + if pid !== nothing + isempty(t.live) && (t.first_sample = p.samples) + push!(t.live, pid) + end end pid === nothing || _sample!(p) pid @@ -301,16 +312,218 @@ function unwatch!(pid::Int) p === nothing && return nothing lock(p.lock) do filter!(!=(pid), t.live) + isempty(t.live) && (t.last_sample = p.samples) + end + nothing +end + +## Grading a stage against the high-water-mark algebra +# `docs/occupancy` reads a stage as a timeline of grades and checks the peak the +# algebra predicts against the one the probe measured. The grade is the pinned +# HWM instance's (`occupancy-types`' sibling `resource-algebra`, +# `ResourceAlgebra.Instances.HWM`; both are fetched by `scripts/fetch-siblings.sh`): +# +# hwm peak drift with drift ≤ peak +# +# * `peak` — the high-water mark this interaction reached, in KiB; +# * `drift` — what it still held when the next interaction began; +# * `release` = peak − drift — what the frontier reclaims. +# +# Sequencing starts the next interaction from the previous one's drift: +# +# hwm p₁ n₁ ⊗ hwm p₂ n₂ = hwm (max p₁ (n₁ + p₂)) (n₁ + n₂) +# +# and `OccupancyTypes.Reclaim` proves of that definition +# +# peak (a ⊗ b) = max (peak a) (drift a + peak b) (reclaim-law) +# peak (a ⊗ b) ≤ peak a + peak b (occupancy≤cost) +# drift a = peak a ⇒ peak (a ⊗ c) = peak a + peak c (no-reclaim) +# drift a < peak a, 0 < peak c ⇒ peak (a ⊗ c) < peak a + peak c (reclaimed) +# +# The last two are the two shapes a stage actually has. Run together, the first +# task is still resident when the second starts (drift = peak) and the stage peak +# is the sum; run one after the other it has exited (drift = 0) and the stage peak +# is the larger. The samples already tell the two apart, so the drift is read from +# the task windows and never assumed. +# +# The prediction is an upper bound on the measurement: at a handover the task +# still live holds at most its peak, so the drift the algebra adds is at least +# what a sample could have seen. `disagreement` therefore reports only what that +# bound forbids — a measured stage peak above the prediction — which is the "one +# of them is a bug" the occupancy reading asks about. + +const GRADE_ALGEBRA = "ResourceAlgebra.Instances.HWM" + +"A grade of the pinned high-water-mark algebra: a peak and the drift it holds." +struct Grade + peak :: Int + drift :: Int + function Grade(peak::Integer, drift::Integer) + drift <= peak || throw(ArgumentError( + "a grade holds at most what it reached: drift $drift > peak $peak")) + new(Int(peak), Int(drift)) + end +end + +"`release(grade)` — what the frontier reclaims, `peak − drift`." +release(g::Grade) = g.peak - g.drift + +"`seq(a, b)` — sequence `b` after `a`: the suffix starts from `a`'s drift." +seq(a::Grade, b::Grade) = Grade(max(a.peak, a.drift + b.peak), a.drift + b.drift) + +"`alt(a, b)` — the larger grade, componentwise (what the timeline never needs, kept for the pair of laws)." +alt(a::Grade, b::Grade) = Grade(max(a.peak, b.peak), max(a.drift, b.drift)) + +"Nothing allocated and nothing held: the algebra's unit." +const NO_GRADE = Grade(0, 0) + +# The trace's sample window, or `nothing` for a record that carries none (one +# written before the windows were sampled). +function _window(rec::AbstractDict) + first_seen = get(rec, "first_sample", nothing) + last_seen = get(rec, "last_sample", nothing) + (first_seen isa Integer && last_seen isa Integer) ? (Int(first_seen), Int(last_seen)) : nothing +end + +_is_windowed(rec::AbstractDict) = _window(rec) !== nothing + +# Whether `prev` was still live when `next` started, so the handover hands its +# peak on rather than reclaiming it. A `prev` never unwatched (window end 0) was +# still live; a `next` that never started (window start 0) never overlapped. +_overlaps(prev::Tuple{Int,Int}, next::Tuple{Int,Int}) = + next[1] != 0 && (prev[2] == 0 || next[1] <= prev[2]) + +# Grades and labels in sampled order, or `nothing` when the record carries no +# windows: a comparison is then skipped rather than made on an assumed order. +function _grades_and_labels(occ::AbstractDict) + recs = get(occ, "tasks", nothing) + recs isa AbstractVector || return Grade[], String[] + all(_is_windowed, recs) || return nothing + # Sampled order, ties broken by record order so the result never depends on + # the sort's stability. + order = sort(collect(eachindex(recs)); by = i -> (_window(recs[i])[1], i)) + ordered = collect(recs)[order] + windows = [_window(r) for r in ordered] + grades = Grade[] + labels = String[] + for (i, rec) in enumerate(ordered) + peak = Int(get(rec, "peak_rss_kib", 0)) + drift = if i == length(ordered) + # The stage waits for its children: a task still resident at the end + # keeps its peak as the grade's drift, an exited one holds nothing. + (windows[i][1] != 0 && windows[i][2] == 0) ? peak : 0 + else + _overlaps(windows[i], windows[i + 1]) ? peak : 0 + end + push!(grades, Grade(peak, drift)) + push!(labels, String(get(rec, "label", "?"))) end + grades, labels +end + +""" + task_grades(occupancy) -> Union{Vector{Grade},Nothing} + +One grade per task of a stage's `occupancy` record, in sampled order: a task's +drift is its peak when it was still live at the next task's start, and 0 when it +had exited. `nothing` when the record carries no sample windows. +""" +function task_grades(occ::AbstractDict) + graded = _grades_and_labels(occ) + graded === nothing ? nothing : graded[1] +end + +""" + stage_grade(occupancy) -> Union{Grade,Nothing} + +The stage's grade: its tasks' grades folded with `seq`, the algebra's sequencing. +`nothing` when the record carries no sample windows. +""" +function stage_grade(occ::AbstractDict) + grades = task_grades(occ) + grades === nothing && return nothing + isempty(grades) ? NO_GRADE : foldl(seq, grades) +end + +"`predicted_peak_rss_kib(occupancy)` — the peak the algebra predicts, or `nothing`." +function predicted_peak_rss_kib(occ::AbstractDict) + g = stage_grade(occ) + g === nothing ? nothing : g.peak +end + +"`cost_bound_rss_kib(occupancy)` — the peaks added: the peak if no cut reclaimed anything (occupancy≤cost's right-hand side)." +function cost_bound_rss_kib(occ::AbstractDict) + grades = task_grades(occ) + grades === nothing ? nothing : sum(g -> g.peak, grades; init = 0) +end + +"`reclaimed_kib(occupancy)` — the slack between the cost bound and the predicted peak: what the cuts reclaim." +function reclaimed_kib(occ::AbstractDict) + cost = cost_bound_rss_kib(occ) + predicted = predicted_peak_rss_kib(occ) + (cost === nothing || predicted === nothing) ? nothing : cost - predicted +end + +""" + disagreement(occupancy) -> Union{String,Nothing} + +Why the measured stage peak contradicts the grade algebra, or `nothing` when it +does not. The prediction is an upper bound, so the one contradiction is a +measured peak above it: a sample then saw more resident memory than the timed +timeline of tasks can hold, and one of the two is wrong. A record with no +windows, or none the probe graded, gives `nothing`: not measured is not a +disagreement. +""" +function disagreement(occ::AbstractDict) + predicted = predicted_peak_rss_kib(occ) + measured = get(occ, "child_peak_rss_kib", nothing) + (predicted === nothing || !(measured isa Integer)) && return nothing + measured > predicted && return "measured $(measured) KiB above the predicted $(predicted) KiB" nothing end +""" + graded(occupancy) -> Union{OrderedDict,Nothing} + +The grades of a stage's `occupancy` record, ready to sit beside it in the +attestation: per task `peak_kib`/`drift_kib`/`release_kib`, the stage's +`predicted_peak_kib`, the `cost_bound_kib` (its peaks added) and the +`reclaimed_kib` slack between the two, the `measured_peak_kib` it is checked +against, and the `disagreement` between them or `nothing`. `nothing` when the +record carries no sample windows. +""" +function graded(occ::AbstractDict) + graded_pair = _grades_and_labels(occ) + graded_pair === nothing && return nothing + grades, labels = graded_pair + stage = isempty(grades) ? NO_GRADE : foldl(seq, grades) + cost = sum(g -> g.peak, grades; init = 0) + OrderedDict{String,Any}( + "algebra" => GRADE_ALGEBRA, + "tasks" => Any[OrderedDict{String,Any}( + "label" => labels[i], + "peak_kib" => grades[i].peak, + "drift_kib" => grades[i].drift, + "release_kib" => release(grades[i])) + for i in eachindex(grades)], + "predicted_peak_kib" => stage.peak, + "drift_kib" => stage.drift, + "release_kib" => release(stage), + "cost_bound_kib" => cost, + "reclaimed_kib" => cost - stage.peak, + "measured_peak_kib" => get(occ, "child_peak_rss_kib", nothing), + "disagreement" => disagreement(occ), + ) +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, + "first_sample" => t.first_sample, + "last_sample" => t.last_sample, ) """ @@ -337,6 +550,11 @@ function finish!(p::Probe) "tasks" => Any[_task_record(t) for t in traces], ) p.error === nothing || (p.record["error"] = p.error) + # The grades the sample windows imply, beside the samples they came from. A + # record whose tasks carry no windows cannot be graded, and is left as it is + # rather than graded on an assumed order. + g = graded(p.record) + g === nothing || (p.record["grades"] = g) p.record end diff --git a/test/occupancy_disagreement_test.jl b/test/occupancy_disagreement_test.jl index 00837263..15df70fd 100644 --- a/test/occupancy_disagreement_test.jl +++ b/test/occupancy_disagreement_test.jl @@ -341,6 +341,25 @@ _task(rec, label) = only(t for t in rec["tasks"] if t["label"] == label) @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" + # The grades the sample windows imply, on this same run + # (fork issue #105). The peak is recorded, not asserted — a + # runner that never has both searches resident cannot show it — + # but the algebra's own claims are asserted of every order: + # every task is timed, the prediction at least reaches what the + # samples saw (the prediction is an upper bound: at a handover + # the live task holds at most its peak), it never exceeds the + # cost bound (Reclaim.occupancy≤cost), and the slack between + # them is never negative. + for o in ORDERS + r = records[o] + @test all(t -> haskey(t, "first_sample") && haskey(t, "last_sample"), r["tasks"]) + @test haskey(r, "grades") + @test MP.predicted_peak_rss_kib(r) >= r["child_peak_rss_kib"] + @test MP.predicted_peak_rss_kib(r) <= MP.cost_bound_rss_kib(r) + @test MP.reclaimed_kib(r) >= 0 + @test MP.disagreement(r) === nothing + end + path = get(ENV, "OCCUPANCY_RESULTS", "") if !isempty(path) io = IOBuffer() diff --git a/test/runtests.jl b/test/runtests.jl index 81626927..73921614 100644 --- a/test/runtests.jl +++ b/test/runtests.jl @@ -45,6 +45,7 @@ const UNIT_FILES = [ "test_remote_stages.jl", "test_jobs.jl", "test_provenance.jl", + "test_occupancy_grade.jl", "test_install_pins.jl", "test_migrate_composition.jl", "test_sample_reads.jl", diff --git a/test/unit/test_occupancy_grade.jl b/test/unit/test_occupancy_grade.jl new file mode 100644 index 00000000..4a3fc003 --- /dev/null +++ b/test/unit/test_occupancy_grade.jl @@ -0,0 +1,179 @@ +# © 2026 Joshua Benjamin Jewell. All rights reserved. +# Licensed under the GNU Affero General Public License version 3 (AGPLv3). + +## Grading a stage against the high-water-mark algebra (MMW-01, fork issue #105) +# The algebra is proved in occupancy-types' `OccupancyTypes.Reclaim`, over the HWM +# instance in resource-algebra; `scripts/fetch-siblings.sh` pins both and +# `scripts/check-occupancy.sh` type-checks what consumes them. Nothing here +# restates a sibling's definition: what is tested is the *reading* of a probe's +# samples as grades — that a task is sequenced from the previous one's drift, +# that the siblings' laws come out of that reading, and that the prediction +# reproduces the measurement the occupancy reading recorded on 2026-10-14. + +using MetaManifold.MemoryProbe +using OrderedCollections, YAML + +const MG = MetaManifold.MemoryProbe + +const RECORDED_OCCUPANCY = + normpath(joinpath(@__DIR__, "..", "..", "docs", "occupancy", "results-2026-10-14.yml")) +const RECORDED_SCHEDULES = + normpath(joinpath(@__DIR__, "..", "..", "docs", "occupancy", "schedules.yml")) + +# One task as the probe records it, with the sample window a schedule gives it. +# The window is what the grade reads: a task still live when the next one starts +# hands its peak on as drift, an exited one hands on nothing. +function _occ_task(peak; label = "t", first_sample = 1, last_sample = 2) + OrderedDict{String,Any}("label" => label, "processes" => 1, "samples" => 3, + "peak_rss_kib" => peak, "hwm_kib" => peak, + "first_sample" => first_sample, "last_sample" => last_sample) +end + +# One stage as `finish!` writes it: its tasks and the peak the samples saw. +_occ_stage(tasks; measured) = OrderedDict{String,Any}( + "probe" => MG.PROBE, "interval_ms" => 10, "samples" => 10, + "julia_allocated_bytes" => 1000, "julia_live_bytes_delta" => 500, + "child_peak_rss_kib" => measured, "tasks" => Any[tasks...]) + +@testset "occupancy grades" begin + @testset "a grade is (peak, drift), drift ≤ peak" begin + g = MG.Grade(90, 40) + @test g.peak == 90 + @test g.drift == 40 + @test MG.release(g) == 50 # what the frontier reclaims + # The invariant the HWM instance carries as its `bounded` field. + @test_throws ArgumentError MG.Grade(40, 90) + end + + @testset "sequencing starts the suffix from the drift (Reclaim.reclaim-law)" begin + a = MG.Grade(100, 40) + b = MG.Grade(60, 10) + s = MG.seq(a, b) + # peak (a ⊗ b) ≡ max (peak a) (drift a + peak b) + @test s.peak == max(a.peak, a.drift + b.peak) + @test s.drift == a.drift + b.drift + @test MG.seq(a, MG.NO_GRADE) == a # the unit holds nothing + @test s.peak <= a.peak + b.peak # Reclaim.occupancy≤cost + # Alternatives keep the larger of each component. + @test MG.alt(a, b) == MG.Grade(100, 40) + end + + @testset "the two shapes a stage has: serial and concurrent" begin + # One after the other: the first has exited (drift = 0), so the peak is + # the larger of the two and the cut reclaims the difference. + serial = _occ_stage([_occ_task(100; label = "asv", first_sample = 1, last_sample = 2), + _occ_task(60; label = "otu", first_sample = 3, last_sample = 4)]; + measured = 100) + @test MG.predicted_peak_rss_kib(serial) == 100 + @test MG.cost_bound_rss_kib(serial) == 160 + @test MG.reclaimed_kib(serial) == 60 # Reclaim.reclaimed: strictly below cost + @test MG.disagreement(serial) === nothing + + # Together: the first is still resident when the second starts + # (drift = peak), so the peak is the sum and nothing is reclaimed. + together = _occ_stage([_occ_task(100; label = "asv", first_sample = 1, last_sample = 4), + _occ_task(60; label = "otu", first_sample = 2, last_sample = 5)]; + measured = 160) + @test MG.predicted_peak_rss_kib(together) == 160 + @test MG.cost_bound_rss_kib(together) == 160 + @test MG.reclaimed_kib(together) == 0 # Reclaim.no-reclaim: drift = peak + @test MG.disagreement(together) === nothing + end + + @testset "the graded record says what it is" begin + rec = _occ_stage([_occ_task(100; label = "vsearch:asv", first_sample = 1, last_sample = 4), + _occ_task(60; label = "vsearch:otu", first_sample = 2, last_sample = 5)]; + measured = 160) + g = MG.graded(rec) + @test g["algebra"] == "ResourceAlgebra.Instances.HWM" + @test g["predicted_peak_kib"] == 160 + @test g["cost_bound_kib"] == 160 + @test g["reclaimed_kib"] == 0 + @test g["measured_peak_kib"] == 160 + @test g["disagreement"] === nothing + @test [t["label"] for t in g["tasks"]] == ["vsearch:asv", "vsearch:otu"] + @test g["tasks"][1]["drift_kib"] == 100 # still resident at the handover + @test g["tasks"][2]["release_kib"] == 60 # the last one hands on nothing + end + + @testset "a measured peak above the prediction is a disagreement" begin + # The prediction is an upper bound, so this is the one contradiction it + # allows: a sample saw more than the timed timeline can hold. + rec = _occ_stage([_occ_task(100; first_sample = 1, last_sample = 2)]; measured = 250) + d = MG.disagreement(rec) + @test d isa String + @test occursin("250", d) && occursin("100", d) + # At or below the bound there is nothing to report. + @test MG.disagreement(_occ_stage([_occ_task(100; first_sample = 1, last_sample = 2)]; + measured = 100)) === nothing + end + + @testset "a record with no sample windows is not graded" begin + # Records written before the windows were sampled cannot be read as a + # timeline, so nothing is predicted for them rather than an order assumed. + rec = OrderedDict{String,Any}("child_peak_rss_kib" => 100, + "tasks" => Any[OrderedDict{String,Any}("label" => "t", "peak_rss_kib" => 100)]) + @test MG.predicted_peak_rss_kib(rec) === nothing + @test MG.graded(rec) === nothing + @test MG.disagreement(rec) === nothing # not measured is not a disagreement + end + + @testset "finish! grades the record it writes" begin + # A record the probe writes itself carries its tasks' sample windows, so + # `finish!` attaches the grades read from them; the run with real vsearch + # children is test/occupancy_disagreement_test.jl. + if !Sys.islinux() + @info "occupancy grades: not Linux, so there is no /proc to sample; the finish! grading check is SKIPPED" + else + probe = MG.start(Dict("occupancy_probe" => Dict("enabled" => true, + "interval_ms" => 10))) + MG.within(probe) do + MG.task("sleep") do + proc = run(`sleep 0.2`; wait = false) + pids = MG.watch!(proc) + try + wait(proc) + finally + MG.unwatch!(pids) + end + end + end + rec = MG.finish!(probe) + @test length(rec["tasks"]) == 1 # one task; no stage trace + t = rec["tasks"][1] + @test t["label"] == "sleep" && t["samples"] > 0 + @test t["peak_rss_kib"] > 0 # sampled from /proc + @test t["last_sample"] >= t["first_sample"] # the window watch!/unwatch! set + @test haskey(rec, "grades") + @test rec["grades"]["predicted_peak_kib"] == MG.predicted_peak_rss_kib(rec) + @test rec["grades"]["measured_peak_kib"] == rec["child_peak_rss_kib"] + @test MG.disagreement(rec) === nothing + end + end + + @testset "the prediction reproduces the recorded measurement" begin + # docs/occupancy/results-2026-10-14.yml is the probe's own record of the + # run docs/occupancy-2026-10.adoc publishes. The schedules are the ones the + # .sir files in docs/occupancy state — serial for two orders, together for + # the third — and the peaks are the recorded ones: nothing here is invented. + orders = YAML.load_file(RECORDED_OCCUPANCY)["orders"] + # The interleavings the run used, from the same file the gate reads + # (scripts/check-occupancy-grades.jl), so the two cannot disagree. + schedules = YAML.load_file(RECORDED_SCHEDULES) + for (name, rec) in orders + windows = schedules[name] + tasks = [_occ_task(t["peak_rss_kib"]; label = t["label"], + first_sample = windows[i][1], last_sample = windows[i][2]) + for (i, t) in enumerate(rec["tasks"])] + occ = _occ_stage(tasks; measured = rec["child_peak_rss_kib"]) + graded = MG.graded(occ) + @test MG.predicted_peak_rss_kib(occ) == rec["child_peak_rss_kib"] + @test MG.disagreement(occ) === nothing + peaks = [t["peak_rss_kib"] for t in rec["tasks"]] + # Serial: the larger peak. Together: both resident, so nothing is reclaimed. + @test graded["reclaimed_kib"] == + (name == "concurrent" ? 0 : sum(peaks) - maximum(peaks)) + @test graded["cost_bound_kib"] == sum(peaks) + end + end +end