From 410ed4210143fe50685b530eaad02a3e5726d701 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=C3=98yvind=20Raddum=20Berg?= Date: Wed, 29 Jul 2026 01:40:24 +0200 Subject: [PATCH 1/8] ci: keep the compile server's telemetry from every run MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The build job got ~50% slower (12-14 minutes on master before #622, 18-20 after) and the evidence needed to explain that is written by the server and then thrown away when the runner is destroyed. Two steps, both `if: always()` because a run that went wrong is the one whose telemetry is worth having: collect metrics.jsonl and the server logs into an artifact, and print a summary into the job log so the common questions need no download at all. The summary answers the question a wall clock cannot: did each compile get slower, or did fewer of them run at once? Per-project durations answer the first, machine occupancy the second, and they have different causes and different fixes. It also flags the case where cpu sits free while the queue is non-empty — admission refusing work it could take, which would be a scheduling bug rather than a slow compiler. CI is also the only environment with a realistic shape for this: 4 cores and 16GB, where the fork-memory budget actually binds. A developer machine cannot reproduce it, which is why the constrained case has gone unmeasured. Verified against a real local run, where it immediately showed 21.9% intern hit rate (sharing 1.28, well under the 2-3x that justified interning), 0% starvation, and a single compile allocating 145GB of churn. --- .github/scripts/summarise-bsp-metrics.py | 116 +++++++++++++++++++++++ .github/workflows/build.yml | 30 ++++++ 2 files changed, 146 insertions(+) create mode 100755 .github/scripts/summarise-bsp-metrics.py diff --git a/.github/scripts/summarise-bsp-metrics.py b/.github/scripts/summarise-bsp-metrics.py new file mode 100755 index 000000000..a08a09b84 --- /dev/null +++ b/.github/scripts/summarise-bsp-metrics.py @@ -0,0 +1,116 @@ +#!/usr/bin/env python3 +"""Print what the compile server did during a CI run, from its metrics.jsonl. + +The point is that the common questions should not require downloading an artifact. Two of them +recur often enough to be worth answering on every run: + + "did each compile get slower, or did fewer of them run at once?" + Those have different causes and different fixes, and the distinction is invisible in a wall + clock. Per-project compile durations answer the first; machine occupancy answers the second. + + "is bleep leaving the machine idle while work is queued?" + Idle cores with a non-empty queue means admission is refusing work it could take — a + scheduling bug rather than a slow compiler. + +Reads every metrics.jsonl under the directory given, tolerates truncated or malformed lines (a +server killed mid-write leaves one), and never fails the build: a diagnostic that breaks the run it +is diagnosing is worse than no diagnostic. +""" + +from __future__ import annotations + +import json +import pathlib +import sys +from collections import Counter, defaultdict + + +def load(path: pathlib.Path) -> list[dict]: + rows = [] + with path.open(encoding="utf-8", errors="replace") as fh: + for line in fh: + line = line.strip() + if not line: + continue + try: + rows.append(json.loads(line)) + except ValueError: + continue # truncated final line from a killed server + return rows + + +def fmt_ms(ms: float) -> str: + return f"{ms / 1000:.1f}s" if ms < 60_000 else f"{ms / 60_000:.1f}m" + + +def summarise(path: pathlib.Path, rows: list[dict]) -> None: + print(f"\n=== {path.parent.name} — {len(rows)} events ===") + by_type = Counter(r.get("type") for r in rows) + print(" events: " + ", ".join(f"{k}={v}" for k, v in by_type.most_common())) + + compiles = [r for r in rows if r.get("type") == "compile_end"] + if compiles: + total = sum(r.get("duration_ms", 0) for r in compiles) + failed = sum(1 for r in compiles if not r.get("success", True)) + print(f"\n compiles: {len(compiles)} ({failed} failed), total task time {fmt_ms(total)}") + worst = sorted(compiles, key=lambda r: -r.get("duration_ms", 0))[:8] + print(" slowest:") + for r in worst: + print(f" {fmt_ms(r.get('duration_ms', 0)):>8} {r.get('project', '?')}") + + allocs = [r for r in rows if r.get("type") == "compile_allocation"] + if allocs: + per = defaultdict(int) + for r in allocs: + per[r.get("project", "?")] += r.get("allocated_mb", 0) + print(f"\n allocation attributed for {len(allocs)} compiles; heaviest:") + for proj, mb in sorted(per.items(), key=lambda kv: -kv[1])[:5]: + print(f" {mb:>8} MB {proj}") + + machine = [r for r in rows if r.get("type") == "machine"] + if machine: + samples = len(machine) + starved = sum(1 for r in machine if r.get("waiting", 0) > 0 and r.get("used_cpu", 0) < r.get("total_cpu", 0)) + saturated = sum(1 for r in machine if r.get("used_cpu", 0) >= r.get("total_cpu", 0)) + idle = sum(1 for r in machine if r.get("used_cpu", 0) == 0 and r.get("waiting", 0) == 0) + peak_wait = max((r.get("waiting", 0) for r in machine), default=0) + print(f"\n machine: {samples} samples at 15s") + print(f" saturated (all cpu in use): {saturated:>4} ({100 * saturated / samples:.0f}%)") + print(f" idle (nothing running, nothing queued): {idle:>4} ({100 * idle / samples:.0f}%)") + print(f" STARVED (queue non-empty, cpu free): {starved:>4} ({100 * starved / samples:.0f}%) <- admission refusing work it could take") + print(f" deepest queue: {peak_wait}") + + cache = [r for r in rows if r.get("type") == "analysis_cache"] + if cache: + last = cache[-1] + hits, misses = last.get("intern_hits", 0), last.get("intern_misses", 0) + looked_up = hits + misses + rate = (100 * hits / looked_up) if looked_up else 0.0 + print(f"\n analysis cache: {last.get('entries', 0)} analyses over {last.get('workspaces', 0)} workspace(s), {last.get('file_bytes', 0) // (1024 * 1024)}MB of files") + print(f" interning: {hits} hits / {misses} misses ({rate:.1f}% hit rate), sharing factor {last.get('sharing_factor', 0)}") + if looked_up and rate < 5: + print(" NOTE: interning is paying its full cost and returning almost nothing here.") + + oom = [r for r in rows if r.get("type") in ("oom_pressure", "oom_crash")] + if oom: + print(f"\n heap events: {len(oom)}") + for r in oom[-3:]: + print(f" {r.get('type')}: used={r.get('heap_used_mb')}MB live={r.get('heap_live_mb', '?')}MB max={r.get('heap_max_mb')}MB compiles={r.get('concurrent_compiles')}") + + +def main() -> int: + root = pathlib.Path(sys.argv[1] if len(sys.argv) > 1 else ".") + found = sorted(root.rglob("metrics.jsonl")) + if not found: + print(f"no metrics.jsonl under {root}") + return 0 + for path in found: + try: + summarise(path, load(path)) + except Exception as e: # noqa: BLE001 - a broken summary must not break the build + print(f"could not summarise {path}: {e!r}") + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/.github/workflows/build.yml b/.github/workflows/build.yml index afd0bccdf..5c06263dc 100644 --- a/.github/workflows/build.yml +++ b/.github/workflows/build.yml @@ -65,6 +65,36 @@ jobs: CI: true run: ./bleep-cli.sh --dev test + # Always, including on failure and timeout — a run that went wrong is the one whose telemetry + # is worth having. CI is also the only environment with a realistic shape: 4 cores and 16GB, + # where memory bounds actually bind. A developer machine cannot reproduce that. + - name: Summarise compile-server telemetry + if: always() + shell: bash + run: | + set -uo pipefail + shopt -s nullglob + dirs=(~/.cache/bleep/socket/*/) + if [ ${#dirs[@]} -eq 0 ]; then echo "no compile-server socket dirs — nothing to summarise"; exit 0; fi + mkdir -p bsp-diagnostics + for d in "${dirs[@]}"; do + name=$(basename "$d") + mkdir -p "bsp-diagnostics/$name" + for f in "$d"metrics.jsonl "$d"output "$d"output.1 "$d"output.2; do + [ -f "$f" ] && cp "$f" "bsp-diagnostics/$name/" + done + done + python3 .github/scripts/summarise-bsp-metrics.py bsp-diagnostics + + - name: Upload compile-server telemetry + if: always() + uses: actions/upload-artifact@v7 + with: + name: bsp-diagnostics + path: bsp-diagnostics + retention-days: 14 + if-no-files-found: ignore + yaml-ls-check: runs-on: ubuntu-latest steps: From e4227d9c5d35f7b811551b49a3e5e927aea8d77a Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=C3=98yvind=20Raddum=20Berg?= Date: Wed, 29 Jul 2026 08:01:09 +0200 Subject: [PATCH 2/8] bsp: reuse the ECJ classloader across compiles ECJ keeps the JDK module image in statics -- JRTUtil.images, ctSymFiles, JRT_FILE_SYSTEMS, and classCache (a SoftClassCache of already-parsed JDK class files). Statics are per-classloader, and createEcjTools built a fresh URLClassLoader for every compile, so each Java compile started with all four caches empty: it re-opened ct.sym as a zip filesystem and re-read every JDK class it touched. Reusing one loader per ECJ version is also how ECJ is meant to be used -- the Eclipse IDE holds it for the life of the process. Found by allocation profiling a loaded daemon: 26.6% of all sampled allocation sat on ECJ's own thread in ClasspathJep247Jdk12.findClass -> ZipFileSystem.getPath, churning ZipPath and byte[]. Note that this change does NOT recover that allocation -- most of it is inherent to ECJ compiling. What it recovers is CPU time, because a fresh loader also means ECJ's 807 classes are re-loaded per compile and the JIT never stays warm on the compiler's own hot loops. Measured two ways: isolated A/B, only the classloader differs (800 files / 49.6k lines): fresh loader ~8200 ms, 6265 MB shared loader ~3300 ms, 6180 MB 2.5x faster, allocation ~unchanged end-to-end, dlab dquery/ast (2265 Java files), n=1 per condition: baseline 230.0 s fixed 209.7 s -8.8% The saving is per compile invocation, not per line, which is why one huge project only nets ~9% while the synthetic case -- ten short compiles each re-paying startup -- shows 2.5x. Builds with many Java projects pay it once per project. What this retains is bounded by the number of distinct ECJ versions the daemon serves (normally one), not by compiles or workspaces: 807 classes of metaspace plus a GC-reclaimable SoftClassCache. The previous code paid that class-loading cost per compile instead. Co-Authored-By: Claude Opus 5 (1M context) --- .../src/scala/bleep/analysis/ZincBridge.scala | 30 +++++++++++++++---- 1 file changed, 24 insertions(+), 6 deletions(-) diff --git a/bleep-bsp/src/scala/bleep/analysis/ZincBridge.scala b/bleep-bsp/src/scala/bleep/analysis/ZincBridge.scala index 5ff8637b4..3b8b6de6b 100644 --- a/bleep-bsp/src/scala/bleep/analysis/ZincBridge.scala +++ b/bleep-bsp/src/scala/bleep/analysis/ZincBridge.scala @@ -91,6 +91,23 @@ object ZincBridge { */ private val ecjJarCache = new java.util.concurrent.ConcurrentHashMap[String, Seq[Path]]() + /** Cached ECJ classloader per ECJ version, because ECJ's caches are `static` and therefore per-classloader. + * + * `org.eclipse.jdt.internal.compiler.util.JRTUtil` keeps the JDK module image in statics: `images`, `ctSymFiles`, `JRT_FILE_SYSTEMS`, and `classCache` (a + * `SoftClassCache` of already-parsed JDK class files). A fresh classloader per compile discards all four, so every Java compile re-opens `ct.sym` as a zip + * filesystem and re-reads every JDK class it touches from scratch. Allocation profiling of a loaded daemon put 26% of ALL heap allocation in + * `ClasspathJep247Jdk12.findClass` -> `ZipFileSystem.getPath`, churning `ZipPath` and `byte[]` — that is this, and it scales with the number of Java + * compiles. + * + * Sharing one loader across concurrent compiles is also the way ECJ is meant to be used (the Eclipse IDE reuses it for the life of the process); those + * statics are concurrent structures. Only the loader is shared — `EcjCompiler` still gets a per-compile cancellation token and progress listener. + * + * Never closed, deliberately: it lives as long as the daemon, and closing it would throw away the caches that are the entire point. What that retains is + * bounded by the number of distinct ECJ versions the daemon serves — normally one — not by compiles or workspaces: 807 classes of metaspace per version, + * plus a `SoftClassCache` the GC can reclaim under pressure. The previous code paid that class-loading cost per compile instead. + */ + private val ecjClassLoaderCache = new java.util.concurrent.ConcurrentHashMap[String, java.net.URLClassLoader]() + /** Threads each analysis read/write may use. * * Zinc defaults this to `availableProcessors()`, which assumes one build at a time. This daemon runs up to `parallelism` operations at once, each loading @@ -853,12 +870,7 @@ object ZincBridge { cancellationToken: CancellationToken, progressListener: ProgressListener ): xsbti.compile.JavaTools = { - // Resolve ECJ jar - val ecjJars = getEcjJars(ecjVersion) - val ecjClassLoader = new java.net.URLClassLoader( - ecjJars.map(_.toUri.toURL).toArray, - getClass.getClassLoader - ) + val ecjClassLoader = getEcjClassLoader(ecjVersion) // Create a forked Java compiler that uses ECJ val ecjCompiler = new EcjCompiler(ecjClassLoader, cancellationToken, progressListener) @@ -879,6 +891,12 @@ object ZincBridge { private def getEcjJars(version: String): Seq[Path] = ecjJarCache.computeIfAbsent(version, resolveEcj) + private def getEcjClassLoader(version: String): java.net.URLClassLoader = + ecjClassLoaderCache.computeIfAbsent( + version, + v => new java.net.URLClassLoader(getEcjJars(v).map(_.toUri.toURL).toArray, getClass.getClassLoader) + ) + /** Resolve ECJ jars from Maven */ private def resolveEcj(version: String): Seq[Path] = { val dep = bleep.model.Dep.Java("org.eclipse.jdt", "ecj", version) From 252663537f10666be0bd5bcbe6f2449875e230fd Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=C3=98yvind=20Raddum=20Berg?= Date: Wed, 29 Jul 2026 09:43:17 +0200 Subject: [PATCH 3/8] bsp: read macOS process footprint via proc_pid_rusage, not /usr/bin/footprint ourTreeFootprintMb sweeps the daemon's whole process tree every 5s, and on macOS each pid meant spawning /usr/bin/footprint: ~35ms of address-space survey per target (measured against a 6GB-heap JVM), plus parsing its output, which showed up as 2% of the daemon's total heap allocation in a profile. proc_pid_rusage is the syscall footprint itself reads. Measured on the same process: 11us/call versus 27994us/call, ~2500x, and it allocates one 296-byte buffer. RSS via a single batched `ps` was the tempting cheap fix and is wrong: on that same 6GB-heap JVM phys_footprint reported 7190MB while RSS reported 2933MB, because macOS compresses pages and RSS stops counting them. Subtracting a number 2.5x too small from the machine total is the class of error that produced a 63GB fork budget on a 48GB machine, so the metric has to stay phys_footprint. Verified proc_pid_rusage returns the same value footprint prints (7189 vs 7190MB, the drift being live allocation between the two calls), and that offsets 72 and 240 in the 296-byte rusage_info_v4 are what offsetof reports. Linker.downcallHandle is a restricted method: without --enable-native-access the JVM prints four warning lines on first call and, per JEP 472, a future release blocks the call outright. So the flag ships here rather than later: - the daemon gets it in buildServerCommand, outside javaOpts so it stays out of the daemon's identity key -- it is how bleep runs the server, not a tuning knob - bleep-bsp-tests gets it in bleep.yaml, because ForkCostModelTest measures a real process; scoped to this project rather than to every forked test JVM, since it is this suite's need and not a general one Also: say so when the daemon is killed from outside. Every shutdown the server decides on sets shutdownRequested first, so finding it false in the shutdown hook means a signal or a dead parent. This is not cosmetic -- the client reports "BSP server crashed twice (no OutOfMemoryError recorded)", indistinguishable from a crash, and a session diagnosed a thread dump taken by that very hook as an idle-watchdog bug and filed a report for what was actually a `kill`. Tested: ForkCostModelTest 8/8, including a dead pid declining rather than returning a bogus zero, and peak >= current as a guard on the struct offsets. The full bleep-bsp-tests run is NOT yet green -- it deadlocked twice against a daemon built from another branch, which is being investigated separately. Co-Authored-By: Claude Opus 5 (1M context) --- .../src/scala/bleep/ForkCostModelTest.scala | 25 +++-- .../src/scala/bleep/bsp/BspServerDaemon.scala | 18 +++- .../src/scala/bleep/ProcessMemory.scala | 91 +++++++++++-------- .../scala/bleep/bsp/BspServerOperations.scala | 8 +- bleep.yaml | 6 ++ 5 files changed, 103 insertions(+), 45 deletions(-) diff --git a/bleep-bsp-tests/src/scala/bleep/ForkCostModelTest.scala b/bleep-bsp-tests/src/scala/bleep/ForkCostModelTest.scala index 8500cc4aa..bcc3569b7 100644 --- a/bleep-bsp-tests/src/scala/bleep/ForkCostModelTest.scala +++ b/bleep-bsp-tests/src/scala/bleep/ForkCostModelTest.scala @@ -78,12 +78,25 @@ class ForkCostModelTest extends AnyFunSuite with Matchers { prog.timeout(10.seconds).unsafeRunSync() } - test("macOS footprint output parses, including units other than MB") { - ProcessMemory.MacOs.parseSize(" phys_footprint: 542 MB") shouldBe Some(542L) - ProcessMemory.MacOs.parseSize("phys_footprint_peak: 1645 MB") shouldBe Some(1645L) - ProcessMemory.MacOs.parseSize("phys_footprint: 1.6 GB") shouldBe Some(1638L) - ProcessMemory.MacOs.parseSize("phys_footprint: 8192 KB") shouldBe Some(8L) - ProcessMemory.MacOs.parseSize("phys_footprint: not-a-number") shouldBe None + test("a pid that no longer exists is declined, not guessed at") { + // The tree sweep lists pids and then measures them, so forks exit mid-sweep as a matter of course. + // proc_pid_rusage answers ESRCH for those; the contract is None rather than a bogus zero. + val goneForever = ProcessHandle.current().pid() + 100000000L + ProcessMemory.system.footprintMb(goneForever) shouldBe None + } + + test("the peak is at least the current footprint, and both are plausible") { + // Guards the struct offsets: ri_phys_footprint and ri_lifetime_max_phys_footprint are read from + // fixed positions in rusage_info_v4, so a layout that moved would show up as nonsense here. + val self = ProcessHandle.current().pid() + (ProcessMemory.system.footprintMb(self), ProcessMemory.system.peakFootprintMb(self)) match { + case (Some(current), Some(peak)) => + current should be > 0L + peak should be >= current + peak should be < 200000L + case _ => + ProcessMemory.system shouldBe ProcessMemory.Linux // Linux has no peak; only macOS answers both + } } test("the platform reader measures a real process, or honestly declines to") { diff --git a/bleep-bsp/src/scala/bleep/bsp/BspServerDaemon.scala b/bleep-bsp/src/scala/bleep/bsp/BspServerDaemon.scala index 9272efe78..5d9249002 100644 --- a/bleep-bsp/src/scala/bleep/bsp/BspServerDaemon.scala +++ b/bleep-bsp/src/scala/bleep/bsp/BspServerDaemon.scala @@ -322,7 +322,23 @@ object BspServerDaemon { // 4. Kill descendant processes Runtime.getRuntime.addShutdownHook(new Thread("bsp-shutdown") { override def run(): Unit = { - shutdownRequested.set(true) + // Every shutdown this server decides on sets the flag first: the idle watchdog, and each exit path + // out of the accept loop. So finding it still false means the JVM is going down from outside — a + // signal, someone's `kill`, or the OS reaping us — and NOT anything the server chose. + // + // Worth a line of its own because of how it presents at the other end: the client whose compile + // was in flight reports "BSP server crashed twice (no OutOfMemoryError recorded)", which reads + // exactly like a crash. Nothing in the log said otherwise, so a session diagnosed a thread dump + // taken by this very hook as an idle-watchdog bug and filed a report for a shutdown that was + // really a `kill`. The dump below is evidence of what was interrupted, not of what stopped us. + val selfInitiated = shutdownRequested.getAndSet(true) + if (!selfInitiated) + logger.warn( + s"Going down without an internal shutdown request — killed by a signal, or the parent process died " + + s"(the parent-death monitor logs its own line just above when that is the cause). " + + s"${activeClientThreads.size()} client(s) still connected; work in flight is being abandoned, " + + s"and they will report it as a crash." + ) try BspMetrics.shutdown() catch { case _: Exception => () } // Process-global lock state. Released here, at daemon shutdown — never per connection. diff --git a/bleep-core/src/scala/bleep/ProcessMemory.scala b/bleep-core/src/scala/bleep/ProcessMemory.scala index c6bd71231..e59b042e6 100644 --- a/bleep-core/src/scala/bleep/ProcessMemory.scala +++ b/bleep-core/src/scala/bleep/ProcessMemory.scala @@ -11,7 +11,7 @@ import scala.util.Properties * almost all of the difference was mapped classpath. * * Every platform has a name for the metric that excludes that double-count, and they mean the same thing: - * - macOS: `phys_footprint` (and `phys_footprint_peak`), via `/usr/bin/footprint` + * - macOS: `phys_footprint` (and its lifetime peak), via the `proc_pid_rusage` syscall * - Linux: `Pss` — proportional set size, shared pages divided by the number of sharers — via `/proc//smaps_rollup` * - Windows: private working set * @@ -43,47 +43,64 @@ object ProcessMemory { def peakFootprintMb(pid: Long): Option[Long] = None } - /** `/usr/bin/footprint -p ` prints both current and peak: - * {{{ - * phys_footprint: 542 MB - * phys_footprint_peak: 1645 MB - * }}} - * ~40ms per call, and it accepts several `-p` at once, so a whole fleet costs one process spawn. `vmmap --summary` reports the same number but walks the - * entire address space and takes ~800ms, which is too slow to do repeatedly. + /** `phys_footprint`, read from the kernel via `proc_pid_rusage` — the same source `/usr/bin/footprint` reads. + * + * This used to shell out to `/usr/bin/footprint -p `, one process per pid. [[ourTreeFootprintMb]] sweeps the daemon's whole process tree every + * [[MachineResources.BudgetRetuneInterval]], so on a busy server that was tens of spawns per sweep, each costing ~35ms of address-space survey (measured + * against a 6GB-heap JVM), plus parsing the output — which showed up as 2% of the daemon's total heap allocation in a profile. The syscall answers in + * microseconds and touches one 296-byte buffer. + * + * Resident set size is NOT a cheaper substitute, tempting as `ps` looks. Measured against that same 6GB-heap JVM, `phys_footprint` reported 7190MB while RSS + * reported 2933MB: macOS compresses pages, and RSS stops counting them once compressed. Subtracting a number 2.5x too small from the machine total is + * exactly the class of error that produced a 63GB fork budget on a 48GB machine. */ object MacOs extends ProcessMemory { - def footprintMb(pid: Long): Option[Long] = read(pid, "phys_footprint") - def peakFootprintMb(pid: Long): Option[Long] = read(pid, "phys_footprint_peak") - private def read(pid: Long, field: String): Option[Long] = - runFootprint(pid).flatMap { out => - out.linesIterator - .map(_.trim) - .collectFirst { case l if l.startsWith(s"$field:") => l } - .flatMap(parseSize) - } + /** `struct rusage_info_v4` (`sys/resource.h`): 16 bytes of uuid followed by 35 `uint64_t`, 296 bytes in total. We want two of those fields, so we index + * them directly rather than describing the whole struct. Offsets are `offsetof` on macOS 15 and derive as `16 + n*8`: `ri_phys_footprint` is the 8th + * `uint64_t` (72) and `ri_lifetime_max_phys_footprint` the 29th (240). This is a stable public ABI, but it IS an ABI assumption — the live test in + * ForkCostModelTest reads this JVM's own footprint and asserts the value is plausible, which is what would catch a layout that moved. + */ + private final val StructBytes = 296L + private final val PhysFootprintOffset = 72L + private final val LifetimeMaxPhysFootprintOffset = 240L + private final val RusageInfoV4 = 4 - private def runFootprint(pid: Long): Option[String] = - try { - val p = new ProcessBuilder("/usr/bin/footprint", "-p", pid.toString).redirectErrorStream(false).start() - val out = new String(p.getInputStream.readAllBytes(), "UTF-8") - // A dead pid exits non-zero; that is an expected race (the fork finished between listing and - // measuring), not an error worth surfacing. - if (p.waitFor() == 0) Some(out) else None - } catch { case _: java.io.IOException => None } + /** `int proc_pid_rusage(int pid, int flavor, rusage_info_t *buffer)`, from libSystem. Resolved once; a macOS without it is not a macOS we can run on, so + * failing to find it throws rather than quietly degrading to "cannot measure". + */ + private lazy val procPidRusage: java.lang.invoke.MethodHandle = { + val linker = java.lang.foreign.Linker.nativeLinker() + val symbol = linker + .defaultLookup() + .find("proc_pid_rusage") + .orElseThrow(() => new RuntimeException("proc_pid_rusage is missing from libSystem — cannot measure process memory on this macOS")) + linker.downcallHandle( + symbol, + java.lang.foreign.FunctionDescriptor.of( + java.lang.foreign.ValueLayout.JAVA_INT, + java.lang.foreign.ValueLayout.JAVA_INT, + java.lang.foreign.ValueLayout.JAVA_INT, + java.lang.foreign.ValueLayout.ADDRESS + ) + ) + } + + def footprintMb(pid: Long): Option[Long] = read(pid, PhysFootprintOffset) + def peakFootprintMb(pid: Long): Option[Long] = read(pid, LifetimeMaxPhysFootprintOffset) - /** `phys_footprint: 542 MB` / `... 1.6 GB` / `... 813 KB` */ - private[bleep] def parseSize(line: String): Option[Long] = { - val value = line.dropWhile(_ != ':').drop(1).trim - val (num, unit) = value.span(c => c.isDigit || c == '.') - num.trim.toDoubleOption.map { n => - unit.trim.toUpperCase match { - case u if u.startsWith("GB") || u == "G" => math.round(n * 1024) - case u if u.startsWith("KB") || u == "K" => math.round(n / 1024) - case u if u.startsWith("B") && u.length == 1 => math.round(n / (1024 * 1024)) - case _ => math.round(n) // MB, the usual case - } - } + private def read(pid: Long, offset: Long): Option[Long] = { + val arena = java.lang.foreign.Arena.ofConfined() + try { + val buffer = arena.allocate(StructBytes) + // invokeWithArguments rather than invokeExact: it boxes, but this runs a handful of times per + // retune and the alternative depends on Scala's handling of signature-polymorphic calls. + val rc = procPidRusage.invokeWithArguments(Integer.valueOf(pid.toInt), Integer.valueOf(RusageInfoV4), buffer).asInstanceOf[Integer] + // Non-zero is ESRCH: the process exited between being listed and being measured. That is the + // expected race in a tree sweep, not an error worth surfacing. + if (rc.intValue() != 0) None + else Some(buffer.get(java.lang.foreign.ValueLayout.JAVA_LONG, offset) / (1024L * 1024L)) + } finally arena.close() } } diff --git a/bleep-core/src/scala/bleep/bsp/BspServerOperations.scala b/bleep-core/src/scala/bleep/bsp/BspServerOperations.scala index 6abdf65c3..0d1ed5276 100644 --- a/bleep-core/src/scala/bleep/bsp/BspServerOperations.scala +++ b/bleep-core/src/scala/bleep/bsp/BspServerOperations.scala @@ -156,11 +156,17 @@ object BspServerOperations { Seq(config.javaPath.toString) ++ config.javaOpts ++ - Seq("-cp", classpath, config.serverMainClass) ++ + Seq(NativeAccessFlag, "-cp", classpath, config.serverMainClass) ++ socketArg ++ dieWithParentArg } + /** The daemon measures its own process tree through `proc_pid_rusage` (see [[bleep.ProcessMemory]]), which is a restricted method. Without this the JVM + * prints four lines of warning on the first call, and — per JEP 472 — a future release blocks the call outright rather than warning. Set here rather than in + * `javaOpts` so it does not enter the daemon's identity key: it is a constant of how bleep runs the server, not a user-visible tuning choice. + */ + private final val NativeAccessFlag = "--enable-native-access=ALL-UNNAMED" + // ========================================================================== // Connection Management // ========================================================================== diff --git a/bleep.yaml b/bleep.yaml index 17fbfa2f4..2f332d962 100644 --- a/bleep.yaml +++ b/bleep.yaml @@ -326,6 +326,12 @@ projects: - template-scala-3 - template-test isTestProject: true + platform: + # ForkCostModelTest measures a real process through ProcessMemory, which calls the restricted + # `proc_pid_rusage`. Without this the forked test JVM prints JEP 472 warnings, and a future JDK + # blocks the call outright — failing the suite. Scoped to this project rather than to every + # forked test JVM bleep runs, because it is this suite's need, not a general one. + jvmOptions: --enable-native-access=ALL-UNNAMED bleep-tests: dependencies: org.scalatest::scalatest:3.2.19 dependsOn: From e3ad7bf037f31a0debd88fd6f3dd951b03e9c3ef Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=C3=98yvind=20Raddum=20Berg?= Date: Wed, 29 Jul 2026 09:54:44 +0200 Subject: [PATCH 4/8] bsp: stop double-charging CPU for test forks, which self-deadlocked the server A TestSuiteTask is charged Cost(TestFork, cpu = 1) by the DAG interpreter at admission. JvmPool.acquire then reserved cpu = 1 from the same machine governor AGAIN, from inside the already-admitted task -- asking a finite pool for a second permit while holding the first. That is fine until admission saturates. Once every permit is held by a test task, each of those tasks queues for a permit that cannot exist, and nothing can release because releasing requires finishing: machine: cpu 18/18, mem 0/24259MB, compiles 0, running 18, waiting 18 with zero fork processes, no worker thread doing anything, and the compute pool idle. It is a self-deadlock, not contention. Because machine CPU is daemon-wide while the pool is per test run, it also starved unrelated workspaces: compiles in another worktree sat behind parked test forks for 20+ minutes, and one grant was logged only after 1466599ms, when something elsewhere happened to free a permit. The two reservations were distinguishable in the queue dump only by label -- the interpreter's `test:proj:Suite` versus this one's `test Suite` -- which is why the dump reads as 18 running and 18 waiting for what looks like the same work. Fix: the interpreter is the single authority on machine-wide capacity, so the duplicate reservation goes. What stays in the pool is the local semaphore bounding one run's parallelism, which is not machine-wide and cannot deadlock against admission, and the spawn-time memory reservation, which belongs to the process rather than the suite and outlives this scope. Reproduced on this build before the change (not only on the branch where it was first seen), and diagnosed from the daemon's own queue dump plus a thread dump showing no OS thread involved -- the parked fibers are invisible to jstack, which is what made three earlier theories (leaked reservations, memory starvation, hold-and-wait behind startLimiter) plausible and wrong. Co-Authored-By: Claude Opus 5 (1M context) --- .../src/scala/bleep/testing/JvmPool.scala | 41 +++++++++++-------- 1 file changed, 23 insertions(+), 18 deletions(-) diff --git a/bleep-core/src/scala/bleep/testing/JvmPool.scala b/bleep-core/src/scala/bleep/testing/JvmPool.scala index de9ad28cb..9ac2f115c 100644 --- a/bleep-core/src/scala/bleep/testing/JvmPool.scala +++ b/bleep-core/src/scala/bleep/testing/JvmPool.scala @@ -391,27 +391,32 @@ object JvmPool { val boundedOptions = MachineResources.withHeapBound(jvmOptions) val key = JvmKey(classpath, boundedOptions, environment, workingDirectory) - // Only CPU is reserved here. The two dimensions have genuinely different lifetimes: + // NO machine CPU reservation here. The caller already holds one. // - // cpu — the SUITE's, which is exactly this scope. A JVM idling in the pool between suites - // burns no cores, so holding one for it would throttle the machine for nothing. - // memory — the PROCESS's, which outlives this scope. An idle pooled JVM is still resident and - // still costs its whole footprint, so its reservation is taken at spawn and returned - // at destroy (see `spawnJvm` / `destroy`), not here. Tying memory to the suite is - // what let the governor believe memory was free while live JVMs still held it. + // A TestSuiteTask is charged `Cost(TestFork, cpu = 1)` by the DAG interpreter at admission (see + // TaskDag.costOf), and this method runs INSIDE that admitted task. Reserving again here asked the + // same finite pool for a second permit while holding the first, so once admission had handed out + // every permit to test tasks, all of them queued for a permit that could not exist: `cpu 18/18, + // running 18, waiting 18`, no forks spawned, no thread doing anything, forever. Compiles in other + // workspaces starved behind it, because machine CPU is daemon-wide. The two entries were + // distinguishable in the queue dump only by their labels — `test:proj:Suite` from the interpreter + // and `test Suite` from here. // - // Order also matters: the per-pool semaphore (a local counter bounding THIS run's parallelism) - // is taken FIRST and the shared machine reservation second, so we never withhold machine-wide - // capacity while merely queueing for our own run's token. + // The interpreter is the single authority on machine-wide capacity. What stays here is the local + // counter bounding THIS run's parallelism, which is not machine-wide and cannot deadlock against + // admission. + // + // Memory is different and is still taken below, not here: it belongs to the PROCESS, which outlives + // this scope. An idle pooled JVM is still resident and still costs its whole footprint, so its + // reservation is taken at spawn and returned at destroy (see `spawnJvm` / `destroy`). Tying memory + // to the suite is what let the governor believe memory was free while live JVMs still held it. Resource.make(semaphore.acquire)(_ => semaphore.release).flatMap { _ => - machine.reserve(MachineResources.ResourceKind.TestFork, s"test $label", cpu = 1, memoryMb = 0L).flatMap { _ => - Resource - .make(getOrCreate(key, classpath, boundedOptions, runnerClass, environment, workingDirectory).map(jvm => (jvm, new TestJvmImpl(jvm): TestJvm))) { - // Return JVM to pool (or destroy it); the semaphore + cpu reservation are released by their own Resources. - case (jvm, _) => release(jvm) - } - .map(_._2) - } + Resource + .make(getOrCreate(key, classpath, boundedOptions, runnerClass, environment, workingDirectory).map(jvm => (jvm, new TestJvmImpl(jvm): TestJvm))) { + // Return JVM to pool (or destroy it); the semaphore is released by its own Resource. + case (jvm, _) => release(jvm) + } + .map(_._2) } } From 5b89b2409950145245365050423df5f5295f7b99 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=C3=98yvind=20Raddum=20Berg?= Date: Wed, 29 Jul 2026 15:46:25 +0200 Subject: [PATCH 5/8] test: fix ForkCostModelTest's platform fallback, which failed on Windows The new peak-vs-current assertion falls back to identifying the platform when it gets no reading, and I wrote that fallback as `shouldBe ProcessMemory.Linux` -- thinking only of the two platforms that implement a reader. Windows implements neither and is `Unavailable`, so the test failed there while passing on macOS and Linux. CI caught it on windows-latest. Assert what the test actually means: only macOS answers both current and peak. Co-Authored-By: Claude Opus 5 (1M context) --- bleep-bsp-tests/src/scala/bleep/ForkCostModelTest.scala | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/bleep-bsp-tests/src/scala/bleep/ForkCostModelTest.scala b/bleep-bsp-tests/src/scala/bleep/ForkCostModelTest.scala index bcc3569b7..10238231d 100644 --- a/bleep-bsp-tests/src/scala/bleep/ForkCostModelTest.scala +++ b/bleep-bsp-tests/src/scala/bleep/ForkCostModelTest.scala @@ -95,7 +95,8 @@ class ForkCostModelTest extends AnyFunSuite with Matchers { peak should be >= current peak should be < 200000L case _ => - ProcessMemory.system shouldBe ProcessMemory.Linux // Linux has no peak; only macOS answers both + // Only macOS answers both: Linux tracks no high-water Pss, and Windows answers neither. + ProcessMemory.system should not be ProcessMemory.MacOs } } From 0a08417e769e504dfbcee90ffa444f97fab7ba5b Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=C3=98yvind=20Raddum=20Berg?= Date: Wed, 29 Jul 2026 15:46:50 +0200 Subject: [PATCH 6/8] bsp: move heap pressure into DAG admission, so it stops holding a CPU permit HeapPressureGate was the last scheduling decision made BELOW admission. The compile was admitted by the interpreter, took a machine-wide CPU permit, and only then slept on the gate -- up to MaxWaitMs, 60s. For that whole time it withheld capacity from every other kind of work: a test or a link that could have run waited behind a permit held by something deliberately doing nothing. That is the same shape as the test-fork double-charge fixed in e3ad7bf0, and it is the reason the resource story had three authorities instead of one. It cannot deadlock the way that one did -- the wait is capped and GC resolves the condition externally -- but under sustained pressure N compiles each sit on a permit. So the gate becomes a decision the interpreter consults at admission: HeapPressureGate.decide(usage, othersCompiling, threshold, retryMs, firstRefusedAt, now): Decision a total function of what it observes, testable without a heap, a clock or a scheduler. `TaskDag.Handlers` gains `mayAdmitCompile`, which admission calls before spending a reservation on a compile; false means "not now", the task stays ready, and its permit goes to whatever else can use it. It is reconsidered on the next wakeup -- which fires when a task completes, precisely when heap is most likely to have been freed, so the 2s polling loop is not replaced by anything. Two details the sleep loop got for free and now need saying explicitly: - MaxWaitMs is enforced via a per-run map of when each project was first deferred, because there is no longer a single sleep to measure against. - othersCompiling is `> 0`, not `> 1`: this runs BEFORE the reservation, so the compile being considered is not yet in the count. `idle` bypasses the gate exactly as it bypasses tryReserve -- with nothing running, nothing will complete to reconsider the decision, so deferring would stall the build rather than delay a start. A sole compile is never deferred. Handlers is where this belongs per its own doc: new inputs go in the bundle rather than as positional parameters cascading through callsites, and every field is required, so test callsites pass `_ => IO.pure(true)` explicitly. waitForHeapPressure is deleted rather than left beside decide(), so there is only one copy of the policy. makeCompileHandler loses heapPressureThreshold with it. Tested: bleep-bsp-tests 649 passed / 0 failed (10.2x parallelism), bleep-tests 283 passed / 0 failed. Co-Authored-By: Claude Opus 5 (1M context) --- .../analysis/LinkDagIntegrationTest.scala | 3 + .../SourcegenDagIntegrationTest.scala | 10 ++ .../SymbolProcessorDagIntegrationTest.scala | 2 + .../scala/bleep/bsp/HeapPressureGate.scala | 92 +++++++---------- .../bleep/bsp/MultiWorkspaceBspServer.scala | 99 ++++++++++++------- bleep-bsp/src/scala/bleep/bsp/TaskDag.scala | 45 ++++++++- 6 files changed, 153 insertions(+), 98 deletions(-) diff --git a/bleep-bsp-tests/src/scala/bleep/analysis/LinkDagIntegrationTest.scala b/bleep-bsp-tests/src/scala/bleep/analysis/LinkDagIntegrationTest.scala index b5893dc33..03915b305 100644 --- a/bleep-bsp-tests/src/scala/bleep/analysis/LinkDagIntegrationTest.scala +++ b/bleep-bsp-tests/src/scala/bleep/analysis/LinkDagIntegrationTest.scala @@ -246,6 +246,7 @@ class LinkDagIntegrationTest extends AnyFunSuite with Matchers { val executor = TaskDag.executor( Handlers( + mayAdmitCompile = _ => IO.pure(true), compile = (_, _) => IO.pure(TaskResult.Success), link = (lt, _) => { linkCalled = true @@ -294,6 +295,7 @@ class LinkDagIntegrationTest extends AnyFunSuite with Matchers { val executor = TaskDag.executor( Handlers( + mayAdmitCompile = _ => IO.pure(true), compile = (_, _) => IO.pure(TaskResult.Success), link = (_, _) => IO.pure( @@ -345,6 +347,7 @@ class LinkDagIntegrationTest extends AnyFunSuite with Matchers { val executor = TaskDag.executor( Handlers( + mayAdmitCompile = _ => IO.pure(true), compile = (_, _) => IO.pure(TaskResult.Success), link = (_, _) => IO.pure((TaskResult.Failure("Link error", List.empty), LinkResult.Failure("Link error", List.empty))), discover = (_, _) => IO.pure((TaskResult.Success, List.empty)), diff --git a/bleep-bsp-tests/src/scala/bleep/analysis/SourcegenDagIntegrationTest.scala b/bleep-bsp-tests/src/scala/bleep/analysis/SourcegenDagIntegrationTest.scala index 3ba20fdb0..748701eb4 100644 --- a/bleep-bsp-tests/src/scala/bleep/analysis/SourcegenDagIntegrationTest.scala +++ b/bleep-bsp-tests/src/scala/bleep/analysis/SourcegenDagIntegrationTest.scala @@ -297,6 +297,7 @@ class SourcegenDagIntegrationTest extends AnyFunSuite with Matchers { val executor = TaskDag.executor( Handlers( + mayAdmitCompile = _ => IO.pure(true), compile = (t, _) => IO(order.add(s"compile:${t.project.value}"): Unit).as(TaskResult.Success), link = (_, _) => sys.error("LinkTask should not appear here"), discover = (_, _) => sys.error("DiscoverTask should not appear here"), @@ -343,6 +344,7 @@ class SourcegenDagIntegrationTest extends AnyFunSuite with Matchers { val executor = TaskDag.executor( Handlers( + mayAdmitCompile = _ => IO.pure(true), compile = (t, _) => if (t.project == scriptsProject) IO.pure(TaskResult.Failure("compile error", Nil)) else if (t.project == target) IO(targetCompileCalled.set(true)).as(TaskResult.Success) @@ -393,6 +395,7 @@ class SourcegenDagIntegrationTest extends AnyFunSuite with Matchers { val executor = TaskDag.executor( Handlers( + mayAdmitCompile = _ => IO.pure(true), compile = (t, _) => if (t.project == target) IO(targetCompileCalled.set(true)).as(TaskResult.Success) else IO.pure(TaskResult.Success), @@ -441,6 +444,7 @@ class SourcegenDagIntegrationTest extends AnyFunSuite with Matchers { val executor = TaskDag.executor( Handlers( + mayAdmitCompile = _ => IO.pure(true), compile = (t, _) => if (t.project == target) IO(targetCompileCalled.set(true)).as(TaskResult.Success) else IO.pure(TaskResult.Success), @@ -489,6 +493,7 @@ class SourcegenDagIntegrationTest extends AnyFunSuite with Matchers { val executor = TaskDag.executor( Handlers( + mayAdmitCompile = _ => IO.pure(true), compile = (_, _) => IO.pure(TaskResult.Success), link = (_, _) => sys.error("LinkTask should not appear here"), discover = (_, _) => sys.error("DiscoverTask should not appear here"), @@ -541,6 +546,7 @@ class SourcegenDagIntegrationTest extends AnyFunSuite with Matchers { val executor = TaskDag.executor( Handlers( + mayAdmitCompile = _ => IO.pure(true), compile = (_, _) => IO.pure(TaskResult.Success), link = (_, _) => sys.error("LinkTask should not appear here"), discover = (_, _) => sys.error("DiscoverTask should not appear here"), @@ -589,6 +595,7 @@ class SourcegenDagIntegrationTest extends AnyFunSuite with Matchers { val executor = TaskDag.executor( Handlers( + mayAdmitCompile = _ => IO.pure(true), compile = (t, _) => if (t.project == target) IO(targetCompileCalled.set(true)).as(TaskResult.Success) else IO.pure(TaskResult.Success), @@ -638,6 +645,7 @@ class SourcegenDagIntegrationTest extends AnyFunSuite with Matchers { val executor = TaskDag.executor( Handlers( + mayAdmitCompile = _ => IO.pure(true), compile = (t, _) => record(s"compile:${t.project.value}").as(TaskResult.Success), link = (_, _) => sys.error("LinkTask should not appear here"), discover = (_, _) => sys.error("DiscoverTask should not appear here"), @@ -684,6 +692,7 @@ class SourcegenDagIntegrationTest extends AnyFunSuite with Matchers { val executor = TaskDag.executor( Handlers( + mayAdmitCompile = _ => IO.pure(true), compile = (_, _) => IO.raiseError(new RuntimeException("bleep-test-runner resolution returned no jars")), link = (_, _) => sys.error("LinkTask should not appear here"), discover = (_, _) => sys.error("DiscoverTask should not appear here"), @@ -736,6 +745,7 @@ class SourcegenDagIntegrationTest extends AnyFunSuite with Matchers { val executor = TaskDag.executor( Handlers( + mayAdmitCompile = _ => IO.pure(true), compile = (_, _) => IO.pure(TaskResult.Success), link = (_, _) => sys.error("LinkTask should not appear here"), discover = (_, _) => sys.error("DiscoverTask should not appear here"), diff --git a/bleep-bsp-tests/src/scala/bleep/analysis/SymbolProcessorDagIntegrationTest.scala b/bleep-bsp-tests/src/scala/bleep/analysis/SymbolProcessorDagIntegrationTest.scala index 201684144..f005f9ecb 100644 --- a/bleep-bsp-tests/src/scala/bleep/analysis/SymbolProcessorDagIntegrationTest.scala +++ b/bleep-bsp-tests/src/scala/bleep/analysis/SymbolProcessorDagIntegrationTest.scala @@ -106,6 +106,7 @@ class SymbolProcessorDagIntegrationTest extends AnyFunSuite with Matchers { eventQueue <- Queue.bounded[IO, Option[TaskDag.DagEvent]](1024) killSignal <- Deferred[IO, KillReason] handlers = Handlers( + mayAdmitCompile = _ => IO.pure(true), compile = (ct, _) => IO { timeline.add(s"compile:${ct.project.value}"); TaskResult.Success }, link = (_, _) => sys.error("LinkTask should not appear here"), discover = (_, _) => sys.error("DiscoverTask should not appear here"), @@ -138,6 +139,7 @@ class SymbolProcessorDagIntegrationTest extends AnyFunSuite with Matchers { eventQueue <- Queue.bounded[IO, Option[TaskDag.DagEvent]](1024) killSignal <- Deferred[IO, KillReason] handlers = Handlers( + mayAdmitCompile = _ => IO.pure(true), compile = (ct, _) => IO { compileInvoked.set(true); finishedTasks.add(ct.project.value -> TaskResult.Success); TaskResult.Success }, link = (_, _) => sys.error("LinkTask should not appear here"), discover = (_, _) => sys.error("DiscoverTask should not appear here"), diff --git a/bleep-bsp/src/scala/bleep/bsp/HeapPressureGate.scala b/bleep-bsp/src/scala/bleep/bsp/HeapPressureGate.scala index 8eb93cac6..d7c20be52 100644 --- a/bleep-bsp/src/scala/bleep/bsp/HeapPressureGate.scala +++ b/bleep-bsp/src/scala/bleep/bsp/HeapPressureGate.scala @@ -1,8 +1,5 @@ package bleep.bsp -import cats.effect.{Clock, IO} -import scala.concurrent.duration.* - object HeapPressureGate { val DefaultThreshold: Double = 0.80 val DefaultRetryMs: DurationMs = DurationMs(2000L) @@ -29,64 +26,43 @@ object HeapPressureGate { } } - /** Wait until it is safe to start a new compilation. + /** Whether a compile may start now, and if not, how long its stagger should be. * - * Always staggers when other compilations are running, with delay proportional to heap pressure: - * - At low heap (e.g. 2%/80%): short delay (~200ms) then proceed - * - At moderate heap (e.g. 50%/80%): longer delay (~1250ms) then proceed - * - At high heap (>= threshold): full delay, keep waiting until heap drops + * A decision rather than a wait, because the caller is [[TaskDag]]'s admission loop. Waiting here would mean a compile that had ALREADY been admitted — + * holding a machine-wide CPU permit — sleeping on it, which withholds capacity from every other kind of work while doing nothing. Deferring at admission + * leaves the permit for a test or a link that could run right now, and the compile is reconsidered on the next wakeup, which fires when a task completes: + * exactly when heap is most likely to have been freed. + */ + sealed trait Decision + object Decision { + case object Admit extends Decision + + /** Not now. `delayMs` is only the stagger this would have slept, reported to the listener so the "waiting for memory" event still carries a duration. */ + case class Defer(delayMs: Long) extends Decision + } + + /** The gate's whole policy, as a total function of what it observes. Pure so it can be tested without a heap, a clock, or a scheduler. * - * This prevents stampedes (all cores starting simultaneously) while avoiding unnecessary waits when memory is plentiful. + * `firstRefusedAt` is when this task was first deferred (None if it has never been), which is what makes [[MaxWaitMs]] enforceable across separate admission + * attempts rather than within one sleep loop. */ - def waitForHeapPressure( - heapMonitor: HeapMonitor, - activeCompiles: IO[Int], + def decide( + usage: HeapMonitor.Usage, + othersCompiling: Boolean, threshold: Double, retryMs: DurationMs, - projectName: String, - listener: Listener - )(implicit clock: Clock[IO]): IO[Unit] = { - def loop(waitStart: Option[EpochMs]): IO[Unit] = - for { - usage <- IO(heapMonitor.heapUsage()) - nowMs <- clock.realTime.map(d => EpochMs(d.toMillis)) - // Daemon-wide, not per connection. See MachineResources.activeCompiles for why that - // distinction is the difference between this gate working and this gate never firing. - compiling <- activeCompiles - othersCompiling = compiling > 1 - result <- - if (!othersCompiling) { - // We're the only active compilation — proceed immediately - waitStart match { - case Some(start) => - val waitedFor = DurationMs(nowMs.value - start.value) - IO(listener.onResume(projectName, usage.usedMb, usage.maxMb, waitedFor, nowMs)) - case None => - IO.unit - } - } else if (waitStart.isDefined && usage.fraction < threshold) { - // Already waited at least one cycle and heap is below threshold — proceed - val waitedFor = DurationMs(nowMs.value - waitStart.get.value) - IO(listener.onResume(projectName, usage.usedMb, usage.maxMb, waitedFor, nowMs)) - } else if (waitStart.exists(start => nowMs.value - start.value >= MaxWaitMs)) { - // Deadline reached while still under pressure — proceed anyway rather than loop forever. - val waitedFor = DurationMs(nowMs.value - waitStart.get.value) - IO(listener.onResume(projectName, usage.usedMb, usage.maxMb, waitedFor, nowMs)) - } else { - // Others compiling — stagger with proportional delay. - // Scale delay by how close we are to the threshold: - // fraction 0.02 / threshold 0.80 => scale 0.10 (min) => 200ms - // fraction 0.50 / threshold 0.80 => scale 0.625 => 1250ms - // fraction 0.80 / threshold 0.80 => scale 1.0 => 2000ms - val scale = math.max(MinDelayFraction, math.min(1.0, usage.fraction / threshold)) - val effectiveDelayMs = (retryMs.value * scale).toLong - val effectiveWaitStart = waitStart.getOrElse(nowMs) - IO(listener.onWait(projectName, usage.usedMb, usage.maxMb, effectiveDelayMs, nowMs)) >> - IO(BspMetrics.recordHeapPressureStall(projectName, usage.usedMb.value, usage.maxMb.value)) >> - IO.sleep(effectiveDelayMs.millis) >> - loop(Some(effectiveWaitStart)) - } - } yield result - loop(None) - } + firstRefusedAt: Option[EpochMs], + now: EpochMs + ): Decision = + if (!othersCompiling) Decision.Admit // sole compile: staggering against nobody, and deferring it would stall the build + else if (firstRefusedAt.isDefined && usage.fraction < threshold) Decision.Admit // waited at least once and the pressure is gone + else if (firstRefusedAt.exists(start => now.value - start.value >= MaxWaitMs)) Decision.Admit // deadline: proceed under pressure rather than never + else { + // Stagger proportional to how close we are to the threshold: + // fraction 0.02 / threshold 0.80 => scale 0.10 (min) => 200ms + // fraction 0.50 / threshold 0.80 => scale 0.625 => 1250ms + // fraction 0.80 / threshold 0.80 => scale 1.0 => 2000ms + val scale = math.max(MinDelayFraction, math.min(1.0, usage.fraction / threshold)) + Decision.Defer((retryMs.value * scale).toLong) + } } diff --git a/bleep-bsp/src/scala/bleep/bsp/MultiWorkspaceBspServer.scala b/bleep-bsp/src/scala/bleep/bsp/MultiWorkspaceBspServer.scala index a7099aaf8..c7ae0b0f2 100644 --- a/bleep-bsp/src/scala/bleep/bsp/MultiWorkspaceBspServer.scala +++ b/bleep-bsp/src/scala/bleep/bsp/MultiWorkspaceBspServer.scala @@ -1635,7 +1635,7 @@ class MultiWorkspaceBspServer( // KSP doesn't need an equivalent map: the runner emits files to disk that the project's source set picks up directly; no compile-time data flow. val apResults = new java.util.concurrent.ConcurrentHashMap[CrossProjectName, AnnotationProcessorResult]() - val compileHandler = makeCompileHandler(started, workspace, params.originId, serverConfig.effectiveHeapPressureThreshold, apResults, diagnosticTracker) + val compileHandler = makeCompileHandler(started, workspace, params.originId, apResults, diagnosticTracker) val sourcegenHandler = makeSourcegenHandler(started, params.originId) // Create link handler @@ -1668,7 +1668,8 @@ class MultiWorkspaceBspServer( test = testHandler, sourcegen = sourcegenHandler, annotationProcessor = apHandler, - symbolProcessor = kspHandler + symbolProcessor = kspHandler, + mayAdmitCompile = makeCompileAdmission(params.originId, serverConfig.effectiveHeapPressureThreshold) ) ) @@ -1852,23 +1853,51 @@ class MultiWorkspaceBspServer( } } - /** Wait until heap pressure is below threshold before starting compilation. + /** Heap pressure as an ADMISSION decision, for [[TaskDag.Handlers.mayAdmitCompile]]. * - * Delegates to HeapPressureGate for testable logic. Cancellation-safe: IO.sleep is cancelable. + * This replaced an `IO.sleep` loop that ran inside the compile task. The task had already been admitted by then, so it sat on a machine-wide CPU permit + * while waiting — withholding capacity from tests and links that could have run. Refusing admission instead leaves the permit available, and the compile is + * reconsidered on the next wakeup, which fires whenever a task completes: exactly when heap is most likely to have been freed. + * + * The refusal-time map is per DAG run and is what makes [[HeapPressureGate.MaxWaitMs]] enforceable at all now that there is no sleep to measure against: it + * remembers when each project was first deferred, across separate admission attempts. + * + * `othersCompiling` is `> 0`, not `> 1` as the old in-task gate used: this runs BEFORE the reservation, so this compile is not in the count yet. */ - private def waitForHeapPressure( - projectName: String, - originId: Option[String], - threshold: Double - ): IO[Unit] = - HeapPressureGate.waitForHeapPressure( - heapMonitor = heapMonitor, - activeCompiles = machine.activeCompiles, - threshold = threshold, - retryMs = HeapPressureGate.DefaultRetryMs, - projectName = projectName, - listener = makeHeapPressureListener(originId) - ) + private def makeCompileAdmission(originId: Option[String], threshold: Double): TaskDag.CompileTask => IO[Boolean] = { + val listener = makeHeapPressureListener(originId) + val firstRefusedAt = Ref.unsafe[IO, Map[String, EpochMs]](Map.empty) + + compileTask => { + val projectName = compileTask.project.value + for { + usage <- IO(heapMonitor.heapUsage()) + compiling <- machine.activeCompiles + nowMs <- IO.realTime.map(d => EpochMs(d.toMillis)) + refusedAt <- firstRefusedAt.get.map(_.get(projectName)) + admit <- HeapPressureGate.decide( + usage = usage, + othersCompiling = compiling > 0, + threshold = threshold, + retryMs = HeapPressureGate.DefaultRetryMs, + firstRefusedAt = refusedAt, + now = nowMs + ) match { + case HeapPressureGate.Decision.Admit => + refusedAt match { + case None => IO.pure(true) + case Some(start) => + firstRefusedAt.update(_ - projectName) >> + IO(listener.onResume(projectName, usage.usedMb, usage.maxMb, DurationMs(nowMs.value - start.value), nowMs)).as(true) + } + case HeapPressureGate.Decision.Defer(delayMs) => + firstRefusedAt.update(m => m.updated(projectName, m.getOrElse(projectName, nowMs))) >> + IO(listener.onWait(projectName, usage.usedMb, usage.maxMb, delayMs, nowMs)) >> + IO(BspMetrics.recordHeapPressureStall(projectName, usage.usedMb.value, usage.maxMb.value)).as(false) + } + } yield admit + } + } /** Send a structured event via BSP notification. Used for compile, link, and test events. */ private def sendEvent(originId: Option[String], taskId: String, event: BleepBspProtocol.Event): Unit = { @@ -2051,7 +2080,7 @@ class MultiWorkspaceBspServer( val apResults = new java.util.concurrent.ConcurrentHashMap[CrossProjectName, AnnotationProcessorResult]() val compileHandler = - makeCompileHandler(started, workspace, params.originId, serverConfig.effectiveHeapPressureThreshold, apResults, diagnosticTracker) + makeCompileHandler(started, workspace, params.originId, apResults, diagnosticTracker) val sourcegenHandler = makeSourcegenHandler(started, params.originId) val includeTagsSet = testOptions.includeTags.toSet @@ -2199,7 +2228,8 @@ class MultiWorkspaceBspServer( test = testHandler, sourcegen = sourcegenHandler, annotationProcessor = apHandler, - symbolProcessor = kspHandler + symbolProcessor = kspHandler, + mayAdmitCompile = makeCompileAdmission(params.originId, serverConfig.effectiveHeapPressureThreshold) ) ) @@ -2432,7 +2462,6 @@ class MultiWorkspaceBspServer( started: Started, workspace: Path, originId: Option[String], - heapPressureThreshold: Double, apResults: java.util.concurrent.ConcurrentHashMap[CrossProjectName, AnnotationProcessorResult], diagnosticTracker: BspDiagnosticTracker ): (TaskDag.CompileTask, Deferred[IO, KillReason]) => IO[TaskDag.TaskResult] = @@ -2467,7 +2496,7 @@ class MultiWorkspaceBspServer( // Reserve one core from the machine governor for this compile — the same governor test // forks reserve against, so compiles and forks can't oversubscribe the CPU. A compile // runs in the server heap (not a forked process), so it reserves no fork memory; server - // heap is staggered separately by waitForHeapPressure. + // heap pressure is handled at admission — see Handlers.mayAdmitCompile. val gatedCompile = // Admitted by the DAG before this ran — see TaskDag.admit. IO.unit.flatMap { _ => @@ -2475,21 +2504,19 @@ class MultiWorkspaceBspServer( // across every connection, and readable via `machine.activeCompiles`. The connection- // local tally that used to be maintained here counted only this client's compiles, // which is not the quantity anything wants to know. - waitForHeapPressure(projectName, originId, heapPressureThreshold) >> { - val compileStartTime = System.currentTimeMillis() - IO(BspMetrics.recordCompileStart(projectName, wsStr)) >> - compileProject(started, compileTask.project, originId, token, depAnalyses, apFlags, diagnosticTracker) - .guaranteeCase { - case cats.effect.Outcome.Succeeded(resultIO) => - resultIO.flatMap { result => - val dur = System.currentTimeMillis() - compileStartTime - val ok = result == TaskDag.TaskResult.Success - IO(BspMetrics.recordCompileEnd(projectName, wsStr, dur, ok)) - } - case _ => - IO(BspMetrics.recordCompileEnd(projectName, wsStr, System.currentTimeMillis() - compileStartTime, false)) - } - } + val compileStartTime = System.currentTimeMillis() + IO(BspMetrics.recordCompileStart(projectName, wsStr)) >> + compileProject(started, compileTask.project, originId, token, depAnalyses, apFlags, diagnosticTracker) + .guaranteeCase { + case cats.effect.Outcome.Succeeded(resultIO) => + resultIO.flatMap { result => + val dur = System.currentTimeMillis() - compileStartTime + val ok = result == TaskDag.TaskResult.Success + IO(BspMetrics.recordCompileEnd(projectName, wsStr, dur, ok)) + } + case _ => + IO(BspMetrics.recordCompileEnd(projectName, wsStr, System.currentTimeMillis() - compileStartTime, false)) + } } val waitForKill = taskKillSignal.get.map(reason => TaskDag.TaskResult.Killed(reason)) diff --git a/bleep-bsp/src/scala/bleep/bsp/TaskDag.scala b/bleep-bsp/src/scala/bleep/bsp/TaskDag.scala index d3053824d..18745832d 100644 --- a/bleep-bsp/src/scala/bleep/bsp/TaskDag.scala +++ b/bleep-bsp/src/scala/bleep/bsp/TaskDag.scala @@ -856,7 +856,17 @@ object TaskDag { test: (TestSuiteTask, Deferred[IO, KillReason]) => IO[TaskResult], sourcegen: (SourcegenTask, Deferred[IO, KillReason]) => IO[TaskResult], annotationProcessor: (ResolveAnnotationProcessorsTask, Deferred[IO, KillReason]) => IO[(TaskResult, Int)], - symbolProcessor: (RunSymbolProcessorsTask, Deferred[IO, KillReason]) => IO[(TaskResult, Int)] + symbolProcessor: (RunSymbolProcessorsTask, Deferred[IO, KillReason]) => IO[(TaskResult, Int)], + + /** Consulted at ADMISSION, before a compile is given a reservation: false means "not now, reconsider on the next wakeup". + * + * This is the one resource question the machine governor cannot answer for itself. CPU and fork memory are known before a task starts; heap pressure is + * a property of the daemon's own heap right now, and the thing that relieves it is another task finishing. It belongs here rather than inside the + * compile handler because a gate below admission holds a machine-wide CPU permit while it waits — starving tests and links that could have run. + * + * Callers with no opinion pass `_ => IO.pure(true)`. Explicitly, not by default: a no-op default is a default parameter in disguise. + */ + mayAdmitCompile: CompileTask => IO[Boolean] ) /** Create a DAG executor with the given handlers. */ @@ -880,6 +890,27 @@ object TaskDag { * the build rather than merely wait. When the machine is idle the first candidate is admitted by blocking reservation instead, which clamps its request * to the machine's totals — so a task larger than the whole machine waits for the whole machine and then runs, rather than never running. */ + /** Ask whether a compile may start, BEFORE spending a reservation on it. + * + * Heap pressure used to be handled below admission: the compile was admitted, took a machine-wide CPU permit, and only then slept on the gate. That + * withholds capacity from every other kind of work — a test or a link that could have run right now waits behind a permit held by something doing + * nothing. Deferring here instead leaves the permit available, and the compile is reconsidered on the next wakeup, which fires when a task completes: + * exactly when heap is most likely to have been freed. + * + * `idle` bypasses the gate for the same reason it bypasses `tryReserve` below — with nothing running, nothing will complete to reconsider this, so + * deferring would stall the build rather than delay a start. + */ + def mayAdmit(task: Task, idle: Boolean): IO[Boolean] = + task match { + case c: CompileTask if !idle => handlers.mayAdmitCompile(c) + case _ => IO.pure(true) + } + + def reserveFor(task: Task): IO[Option[(Task, IO[Unit])]] = { + val c = costOf(task, forkHeaps) + machine.tryReserve(c.kind, task.id.toString, c.cpu, c.memoryMb).map(_.map(release => (task, release))) + } + def admit(candidates: List[Task], idle: Boolean): IO[List[(Task, IO[Unit])]] = candidates match { case Nil => IO.pure(Nil) @@ -891,13 +922,19 @@ object TaskDag { .reserveUntilReleased(firstCost.kind, first.id.toString, firstCost.cpu, firstCost.memoryMb) .map(release => Some((first, release))) else - machine.tryReserve(firstCost.kind, first.id.toString, firstCost.cpu, firstCost.memoryMb).map(_.map(release => (first, release))) + mayAdmit(first, idle).flatMap { + case true => reserveFor(first) + case false => IO.pure(None) + } firstAdmission.flatMap { headResult => rest .traverse { task => - val c = costOf(task, forkHeaps) - machine.tryReserve(c.kind, task.id.toString, c.cpu, c.memoryMb).map(_.map(release => (task, release))) + // Never `idle` here: if the head was admitted, something is running by definition. + mayAdmit(task, idle = false).flatMap { + case true => reserveFor(task) + case false => IO.pure(None) + } } .map(tail => (headResult :: tail).flatten) } From dd423db81ec35595712cf5b1d29283b672bf8d17 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=C3=98yvind=20Raddum=20Berg?= Date: Wed, 29 Jul 2026 16:18:24 +0200 Subject: [PATCH 7/8] test: loosen cancellation wall-clock bounds that were asserting an idle machine The cancellation tests each start a job that would take 30s or 60s, cancel it, and assert both that the kill was honoured and that the call returned in under 5s (10s in one case). The first assertion is the point of the test. The second was a proxy for "it did not sit through the workload" -- but a tight wall-clock number does not measure the cancellation path, it measures how loaded the machine is, and these run alongside the rest of the suite. That proxy held only because the suite was effectively serialised: a double- charged CPU reservation capped test parallelism at ~1.4x (fixed in e3ad7bf0). At the ~11x the suite now genuinely runs at, windows-latest took 6074ms here. The behaviour was correct -- terminationReason was Killed, and only the clock assertion failed: 6074 was not less than 5000 (TimeoutAndResourceTest.scala:86) So the bound moves to a named 20s, still a small fraction of the 30s/60s work it exists to rule out. Applied to all four sites, not just the one that failed: they share the pattern and the flaw, and the other three would fail the next time a runner is busy. This is loosening a test, so to be explicit about what is and is not being given up: the invariant that cancellation short-circuits the work is kept, and the accidental assertion that nothing else is running is dropped. No production code changed. Co-Authored-By: Claude Opus 5 (1M context) --- .../analysis/TimeoutAndResourceTest.scala | 21 +++++++++++++++---- 1 file changed, 17 insertions(+), 4 deletions(-) diff --git a/bleep-bsp-tests/src/scala/bleep/analysis/TimeoutAndResourceTest.scala b/bleep-bsp-tests/src/scala/bleep/analysis/TimeoutAndResourceTest.scala index c2f578f3b..f3faab5cb 100644 --- a/bleep-bsp-tests/src/scala/bleep/analysis/TimeoutAndResourceTest.scala +++ b/bleep-bsp-tests/src/scala/bleep/analysis/TimeoutAndResourceTest.scala @@ -30,6 +30,19 @@ class TimeoutAndResourceTest extends AnyFunSuite with Matchers with TimeLimits { val mediumTimeout = Span(10, Seconds) val parallelTimeout = Span(20, Seconds) + /** Upper bound for "cancellation short-circuited the work", used by the cancellation tests below. + * + * What those tests prove is that a run which was told to stop does NOT sit through its workload — each one starts a job that would take 30s or 60s, so + * returning in any fraction of that is the evidence. The bound is deliberately generous rather than tight: a tight wall-clock number does not measure the + * cancellation path, it measures how loaded the machine is, and these run alongside the rest of the suite. + * + * They previously used 5s, which held only because the suite was effectively serialised — a double-charged CPU reservation capped test parallelism at ~1.4x + * (fixed in e3ad7bf0). At the ~11x the suite now really runs at, a Windows runner took 6074ms on a path whose behaviour was correct: the kill was honoured, + * `terminationReason` was `Killed`, and only the clock assertion failed. Raising the bound keeps the invariant that matters and drops the one that was + * silently asserting "nothing else is running". + */ + val cancellationShortCircuitMs = 20000L + def createTempDir(prefix: String): Path = Files.createTempDirectory(prefix) @@ -83,7 +96,7 @@ class TimeoutAndResourceTest extends AnyFunSuite with Matchers with TimeLimits { val duration = System.currentTimeMillis() - startTime result.terminationReason shouldBe a[TestRunnerTypes.TerminationReason.Killed] - duration should be < 5000L + duration should be < cancellationShortCircuitMs } finally deleteRecursively(tempDir) } } @@ -124,7 +137,7 @@ class TimeoutAndResourceTest extends AnyFunSuite with Matchers with TimeLimits { val duration = System.currentTimeMillis() - startTime result.terminationReason shouldBe a[TestRunnerTypes.TerminationReason.Killed] - duration should be < 10000L + duration should be < cancellationShortCircuitMs } finally deleteRecursively(tempDir) } } @@ -163,7 +176,7 @@ class TimeoutAndResourceTest extends AnyFunSuite with Matchers with TimeLimits { val duration = System.currentTimeMillis() - startTime result.terminationReason shouldBe a[TestRunnerTypes.TerminationReason.Killed] - duration should be < 5000L + duration should be < cancellationShortCircuitMs } finally deleteRecursively(tempDir) } } @@ -312,7 +325,7 @@ class TimeoutAndResourceTest extends AnyFunSuite with Matchers with TimeLimits { val duration = System.currentTimeMillis() - startTime result.terminationReason shouldBe a[TestRunnerTypes.TerminationReason.Killed] - duration should be < 5000L + duration should be < cancellationShortCircuitMs } finally deleteRecursively(tempDir) } } From 144d0227f35e75102fb50b9eda66786f4063d80c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=C3=98yvind=20Raddum=20Berg?= Date: Wed, 29 Jul 2026 16:27:09 +0200 Subject: [PATCH 8/8] test: tighten the cancellation bound from 20s to 15s MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 20s was headroom picked by feel. Against the actual numbers — sub-second healthy (all eight tests in ~1.6s together), 6074ms worst on a loaded runner, 30s for the shortest workload being ruled out — 15s clears the worst observation by ~2.5x and still fails long before a cancellation that never happened. Co-Authored-By: Claude Opus 5 (1M context) --- .../src/scala/bleep/analysis/TimeoutAndResourceTest.scala | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/bleep-bsp-tests/src/scala/bleep/analysis/TimeoutAndResourceTest.scala b/bleep-bsp-tests/src/scala/bleep/analysis/TimeoutAndResourceTest.scala index f3faab5cb..d8b0e2804 100644 --- a/bleep-bsp-tests/src/scala/bleep/analysis/TimeoutAndResourceTest.scala +++ b/bleep-bsp-tests/src/scala/bleep/analysis/TimeoutAndResourceTest.scala @@ -40,8 +40,12 @@ class TimeoutAndResourceTest extends AnyFunSuite with Matchers with TimeLimits { * (fixed in e3ad7bf0). At the ~11x the suite now really runs at, a Windows runner took 6074ms on a path whose behaviour was correct: the kill was honoured, * `terminationReason` was `Killed`, and only the clock assertion failed. Raising the bound keeps the invariant that matters and drops the one that was * silently asserting "nothing else is running". + * + * 15s is picked against the evidence rather than for round-number comfort: healthy is sub-second (all eight tests here finish in ~1.6s together), the worst + * seen on a loaded runner is 6074ms, and the shortest workload being ruled out is 30s. So it clears the worst observation by ~2.5x and still fails long + * before a cancellation that did not happen at all. */ - val cancellationShortCircuitMs = 20000L + val cancellationShortCircuitMs = 15000L def createTempDir(prefix: String): Path = Files.createTempDirectory(prefix)