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: diff --git a/bleep-bsp-tests/src/scala/bleep/ForkCostModelTest.scala b/bleep-bsp-tests/src/scala/bleep/ForkCostModelTest.scala index 8500cc4aa..10238231d 100644 --- a/bleep-bsp-tests/src/scala/bleep/ForkCostModelTest.scala +++ b/bleep-bsp-tests/src/scala/bleep/ForkCostModelTest.scala @@ -78,12 +78,26 @@ 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 _ => + // Only macOS answers both: Linux tracks no high-water Pss, and Windows answers neither. + ProcessMemory.system should not be ProcessMemory.MacOs + } } test("the platform reader measures a real process, or honestly declines to") { 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-tests/src/scala/bleep/analysis/TimeoutAndResourceTest.scala b/bleep-bsp-tests/src/scala/bleep/analysis/TimeoutAndResourceTest.scala index c2f578f3b..d8b0e2804 100644 --- a/bleep-bsp-tests/src/scala/bleep/analysis/TimeoutAndResourceTest.scala +++ b/bleep-bsp-tests/src/scala/bleep/analysis/TimeoutAndResourceTest.scala @@ -30,6 +30,23 @@ 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". + * + * 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 = 15000L + def createTempDir(prefix: String): Path = Files.createTempDirectory(prefix) @@ -83,7 +100,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 +141,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 +180,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 +329,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) } } 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) 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-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) } 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-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) } } 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: