Commit 9b832eb5 by PLN (Algolia)

fix(foundry): a background job that takes the desktop down is not a background job

PLN, mid-cut: "you're hogging the system :(". Nine children at ~88% of a core each
put the load average at **15.6 on 16 cores**, and the machine he performs and edits
on became unusable. My fault twice over — the default was one subprocess per track
regardless of the box, and nothing told the scheduler this work was background.

Both fixed at the source rather than by remembering to pass a flag:

* `--jobs` defaults to **half the cores, capped at 6**. It is the total that matters,
  not the track count, and each child peaks near 2 GB.
* Children start at **nice 19 and SCHED_IDLE** via a preexec hook, so they only run
  when nothing else wants the CPU. The nine still saturate the sixteen cores; the
  editor now wins every scheduling decision, because nothing waits on this work but
  me. (The live run was reniced in place rather than killed — eight minutes of
  analysis is not worth a restart.)

The test for that hook is worth a note, because the first version of it was fake: it
looked the hook up through `_fan_out.__globals__`, where a nested function does not
live, got `None`, fell back to a no-op lambda, and passed. Same shape of mistake as
the finder test earlier today. So `run_at_idle_priority` is module-level now and the
test asserts on a **real child's** reported nice and scheduling policy — verified to
read `0 0` without the hook and `19 5` with it.

Also in here, from the same run: the presence gate I added to the finder was doing a
full `max(abs(window))` per candidate — at 8 bars that is ~700 k samples times
thousands of windows, which is most of why this run was slower than the last. It now
reads a per-beat peak table computed once per stem. Exact, not approximate: the
inter-beat segments partition `[int(times[i]*sr), int(times[e]*sr))` precisely, so it
visits the same samples — with a test that checks the two agree for every window
rather than trusting the argument.

And `kitgate --link` now takes each kit's staging root from the manifest instead of
`publish.samples_root()`. A pack cut with `--samples-root` lives beside its source, so
the default root would have looked in the wrong place and called every kit missing.

Suite 122 → 125.
parent e2b701ea
...@@ -201,6 +201,23 @@ def _true_seam(y: np.ndarray, s_smp: int, e_smp: int, sr: int, n_beats: int = 0) ...@@ -201,6 +201,23 @@ def _true_seam(y: np.ndarray, s_smp: int, e_smp: int, sr: int, n_beats: int = 0)
_EMPTY_AMP = 10.0 ** (G.THRESH["empty_dbfs"] / 20.0) _EMPTY_AMP = 10.0 ** (G.THRESH["empty_dbfs"] / 20.0)
def _beat_peaks(y: np.ndarray, sr: int, times: np.ndarray) -> np.ndarray:
"""Peak amplitude in each inter-beat segment, so the presence gate is free.
A window's peak is the max over the beats it spans, and the segments partition
`[int(times[i]*sr), int(times[e]*sr))` exactly — the same samples the naive
`max(abs(slice))` would visit — so this is not an approximation. It just does the
O(samples) work once for the whole stem instead of once per candidate: at 8 bars a
window is ~700 k samples and there are thousands of windows, which turned a free
check into the most expensive line in the loop.
"""
mono = y if y.ndim == 1 else np.abs(y).max(axis=1)
a = np.abs(mono)
edges = [int(t * sr) for t in times]
return np.array([float(a[s:e].max()) if e > s else 0.0
for s, e in zip(edges, edges[1:])] + [0.0], dtype=np.float64)
def analyze_stem(y: np.ndarray, sr: int, *, bars=(2, 4, 8), top_n=8, def analyze_stem(y: np.ndarray, sr: int, *, bars=(2, 4, 8), top_n=8,
grid=None, weights=None, verify=True) -> list[LoopCandidate]: grid=None, weights=None, verify=True) -> list[LoopCandidate]:
"""Rank loop candidates within one stem. """Rank loop candidates within one stem.
...@@ -228,6 +245,7 @@ def analyze_stem(y: np.ndarray, sr: int, *, bars=(2, 4, 8), top_n=8, ...@@ -228,6 +245,7 @@ def analyze_stem(y: np.ndarray, sr: int, *, bars=(2, 4, 8), top_n=8,
ssm = beat_sync_ssm(y, sr, peaks) # peaks are onset-envelope frames ssm = beat_sync_ssm(y, sr, peaks) # peaks are onset-envelope frames
n = ssm.shape[0] n = ssm.shape[0]
nov = np.maximum(foote_novelty(ssm, 2), foote_novelty(ssm, 8)) if n else np.zeros(nb) nov = np.maximum(foote_novelty(ssm, 2), foote_novelty(ssm, 8)) if n else np.zeros(nb)
beat_peak = _beat_peaks(y, sr, times) # for the presence gate, computed once
cands: list[LoopCandidate] = [] cands: list[LoopCandidate] = []
for B in bars: for B in bars:
...@@ -249,7 +267,7 @@ def analyze_stem(y: np.ndarray, sr: int, *, bars=(2, 4, 8), top_n=8, ...@@ -249,7 +267,7 @@ def analyze_stem(y: np.ndarray, sr: int, *, bars=(2, 4, 8), top_n=8,
# Rejecting at export (via `grade`) is not enough: BIGHEN's strings stem had # Rejecting at export (via `grade`) is not enough: BIGHEN's strings stem had
# all six of its top windows inside a rest, so the kit went from four silent # all six of its top windows inside a rest, so the kit went from four silent
# loops to one real one instead of to four real ones. # loops to one real one instead of to four real ones.
if float(np.max(np.abs(sl))) < _EMPTY_AMP: if float(np.max(beat_peak[i:e])) < _EMPTY_AMP:
continue continue
seam, _ = G.seam_score(sl) seam, _ = G.seam_score(sl)
zc, _, _ = G.zc_score(sl, sr) zc, _, _ = G.zc_score(sl, sr)
......
...@@ -659,7 +659,7 @@ def main(argv=None) -> int: ...@@ -659,7 +659,7 @@ def main(argv=None) -> int:
ap.add_argument("--dry-run", action="store_true", help="analyze and report, write nothing") ap.add_argument("--dry-run", action="store_true", help="analyze and report, write nothing")
ap.add_argument("--report-only", action="store_true", ap.add_argument("--report-only", action="store_true",
help="rebuild the report/cheatsheet from an existing --out, no cutting") help="rebuild the report/cheatsheet from an existing --out, no cutting")
ap.add_argument("-j", "--jobs", type=int, default=3, ap.add_argument("-j", "--jobs", type=int, default=_default_jobs(),
help="tracks analyzed in parallel (each pins one core)") help="tracks analyzed in parallel (each pins one core)")
a = ap.parse_args(argv) a = ap.parse_args(argv)
...@@ -737,11 +737,45 @@ def _report_only(out: Path) -> int: ...@@ -737,11 +737,45 @@ def _report_only(out: Path) -> int:
return 0 return 0
def run_at_idle_priority() -> None:
"""Drop this process to the lowest priority the scheduler has.
Passed to `subprocess` as a preexec hook, so it runs in the child between fork and
exec. SCHED_IDLE means "only when nothing else wants the CPU"; nice 19 is the
fallback when the policy call is not permitted.
"""
import os
os.nice(19)
try:
os.sched_setscheduler(0, os.SCHED_IDLE, os.sched_param(0))
except (AttributeError, OSError, PermissionError):
pass
def _default_jobs() -> int:
"""Half the cores, and never more than 6.
One child per track is tempting and wrong: it is the total that matters, and the
machine this runs on is the machine PLN performs and edits on. Half leaves the
other half for the desktop; the cap keeps a bigger box from turning a background
job into a 24-way memory hog (each child peaks around 2 GB).
"""
import os
n = len(os.sched_getaffinity(0)) if hasattr(os, "sched_getaffinity") else (os.cpu_count() or 4)
return max(1, min(6, n // 2))
def _fan_out(todo: list[Path], a, out: Path) -> tuple[list, list]: def _fan_out(todo: list[Path], a, out: Path) -> tuple[list, list]:
"""Run one `--jobs 1` subprocess per track, then merge their results. """Run one `--jobs 1` subprocess per track, then merge their results.
Children are told `--no-clap`: tagging loads a ~500 MB model, so it happens once Children are told `--no-clap`: tagging loads a ~500 MB model, so it happens once
in the parent over the merged set rather than once per track. in the parent over the merged set rather than once per track.
They also run at **nice 19 and idle I/O**. This is not politeness — PLN works on
this machine while a cut is running, and nine children at 88% of a core each took
the load average to 15.6 on 16 cores and made the desktop unusable. Niced, the same
nine saturate the same sixteen cores and the editor still wins every scheduling
decision, because the work is genuinely background: nothing waits on it but me.
""" """
import subprocess import subprocess
from concurrent.futures import ThreadPoolExecutor from concurrent.futures import ThreadPoolExecutor
...@@ -766,7 +800,8 @@ def _fan_out(todo: list[Path], a, out: Path) -> tuple[list, list]: ...@@ -766,7 +800,8 @@ def _fan_out(todo: list[Path], a, out: Path) -> tuple[list, list]:
sub.mkdir(parents=True, exist_ok=True) sub.mkdir(parents=True, exist_ok=True)
with open(sub / "run.log", "w") as lf: with open(sub / "run.log", "w") as lf:
r = subprocess.run(cmd, cwd=Path(__file__).resolve().parent.parent, r = subprocess.run(cmd, cwd=Path(__file__).resolve().parent.parent,
stdout=lf, stderr=subprocess.STDOUT, text=True) stdout=lf, stderr=subprocess.STDOUT, text=True,
preexec_fn=run_at_idle_priority)
return d, sub, r return d, sub, r
results, cuts = [], [] results, cuts = [], []
......
...@@ -208,9 +208,15 @@ def main(argv=None) -> int: ...@@ -208,9 +208,15 @@ def main(argv=None) -> int:
if a.link: if a.link:
from engine import publish as P from engine import publish as P
acted = Counter() acted = Counter()
for kit in sorted({c["kit"] for c in cuts}): # Where the audio is comes from the MANIFEST, not from `publish.samples_root()`.
# A pack cut with `--samples-root` is staged beside its source, so the default
# root would look in the wrong place and report every kit as missing — and the
# manifest is the one thing that already knows, per kit, exactly where its files
# were written.
roots = {c["kit"]: Path(c["path"]).parent.parent for c in cuts}
for kit in sorted(roots):
try: try:
acted[P.link_kit(kit).action] += 1 acted[P.link_kit(kit, samples=roots[kit]).action] += 1
except Exception as e: except Exception as e:
print(f" ! {kit}: {e}") print(f" ! {kit}: {e}")
acted["failed"] += 1 acted["failed"] += 1
......
...@@ -444,3 +444,21 @@ def test_the_finder_will_not_spend_its_candidate_slots_on_silence(): ...@@ -444,3 +444,21 @@ def test_the_finder_will_not_spend_its_candidate_slots_on_silence():
played_s = 8 * bar_s played_s = 8 * bar_s
assert all(c.start_s < played_s for c in cands), \ assert all(c.start_s < played_s for c in cands), \
[(c.start_s, c.score) for c in cands] [(c.start_s, c.score) for c in cands]
def test_the_beat_peak_table_is_exact_not_an_approximation():
"""The presence gate reads a per-beat peak table instead of scanning each candidate
window, which is only legitimate if it returns the same number. The inter-beat
segments partition the window exactly, so it must — and if that ever stops being
true the gate would start rejecting real material with no visible symptom."""
import numpy as np
from engine import loops as L
sr = 8000
rng = np.random.RandomState(3)
y = (rng.randn(sr * 6) * 0.2).astype(np.float32)
times = np.arange(0, 6.0, 0.5)
bp = L._beat_peaks(y, sr, times)
for i in range(len(times) - 4):
e = i + 4
naive = float(np.max(np.abs(y[int(times[i] * sr):int(times[e] * sr)])))
assert float(np.max(bp[i:e])) == naive, (i, e)
...@@ -364,3 +364,33 @@ def test_the_two_presence_lenses_use_one_number(): ...@@ -364,3 +364,33 @@ def test_the_two_presence_lenses_use_one_number():
failed.""" failed."""
from engine import grade as G from engine import grade as G
assert K.EMPTY_DBFS == G.THRESH["empty_dbfs"] assert K.EMPTY_DBFS == G.THRESH["empty_dbfs"]
# ── a background job has to behave like one ──────────────────────────────────
def test_the_default_job_count_leaves_the_machine_usable():
"""PLN performs and edits on this machine. Nine children at ~88% of a core each took
the load average to 15.6 on 16 cores and made the desktop unusable — so the default
is half the cores, capped, rather than one per track."""
import os
import engine.stempack as SP
n = len(os.sched_getaffinity(0))
j = SP._default_jobs()
assert 1 <= j <= 6
assert j <= max(1, n // 2)
def test_children_are_started_at_idle_priority():
"""Assert on a real child, not on the hook's existence. `_fan_out` passes this as
subprocess's preexec hook, so if it silently did nothing the run would still work —
it would just take the desktop down with it, which is how this was noticed."""
import os
import subprocess
import sys
import engine.stempack as SP
src = "import os;print(os.nice(0), os.sched_getscheduler(0))"
out = subprocess.run([sys.executable, "-c", src], capture_output=True, text=True,
preexec_fn=SP.run_at_idle_priority)
assert out.returncode == 0, out.stderr
nice, policy = out.stdout.split()
assert int(nice) == 19, f"child ran at nice {nice}"
assert int(policy) == os.SCHED_IDLE, f"child policy {policy}, wanted SCHED_IDLE"
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment