bsp: bound and unify the compile server's scheduling — zinc 2, structural sharing, one knob

.github/workflows/build.yml

@@ -12,7 +12,15 @@ concurrency:
 
 jobs:
   build:
-    timeout-minutes: 15
+    # Measured, not guessed: this job does sourcegen, a full compile, publish-local AND the whole
+    # test suite. On master it takes 12-14 minutes against a 15-minute cap — 82-92% of budget, so it
+    # was one slow runner away from red before anything changed. The zinc 2 upgrade and per-analysis
+    # interning pushed it over: two consecutive runs were cancelled at exactly 15m17s and 15m18s.
+    #
+    # Not hiding a hang. The log shows suites finishing 4 seconds apart right up to the cut, three
+    # running concurrently — steady throughput into the wall, not a stall. The native-image jobs
+    # already get 40 minutes for less work than this.
+    timeout-minutes: 30
     # need to be newer than `ubuntu-20.04` because of scalafmt native binary
     runs-on: ubuntu-latest
     if: "!contains(github.event.head_commit.message, 'ci skip')"

bleep-bsp-tests/src/scala/bleep/MachineResourcesTest.scala

@@ -19,6 +19,24 @@ class MachineResourcesTest extends AnyFunSuite with Matchers {
   private def machine(cpu: Int, memMb: Long): MachineResources =
     MachineResources.create(totalCpu = cpu, totalMemoryMb = memMb, logger = TypedLogger.DevNull, longWaitWarnMs = 200L)
 
+  test("activeCompiles is what the heap gate reads, and counts every client's compiles") {
+    // Per connection this figure was always 1 for a client running a single compile, so a dozen
+    // such clients each concluded they were alone and skipped the stagger entirely.
+    val m = machine(cpu = 8, memMb = 8192)
+    val prog = for {
+      running <- CountDownLatch[IO](3)
+      release <- CountDownLatch[IO](1)
+      held <- List("a", "b", "c").parTraverse(n => m.reserve(Compile, n, cpu = 1, memoryMb = 0L).use(_ => running.release *> release.await).start)
+      _ <- running.await
+      n <- m.activeCompiles
+      _ <- IO(n shouldBe 3)
+      _ <- release.release
+      _ <- held.traverse_(_.join)
+      afterCount <- m.activeCompiles
+    } yield afterCount shouldBe 0
+    prog.timeout(20.seconds).unsafeRunSync()
+  }
+
   test("a reservation within budget is granted immediately and released") {
     val m = machine(cpu = 4, memMb = 8192)
     val prog = for {

bleep-bsp-tests/src/scala/bleep/analysis/AnalysisCacheOwnershipTest.scala

@@ -0,0 +1,95 @@
+package bleep.analysis
+
+import bleep.model
+import org.scalatest.funsuite.AnyFunSuite
+import org.scalatest.matchers.should.Matchers
+import xsbti.compile.CompileAnalysis
+import xsbti.compile.analysis.{ReadCompilations, ReadSourceInfos, ReadStamps}
+
+import java.nio.file.{Files, Path}
+
+/** Analyses belong to a workspace, and the consequences of that: they are partitioned by it and dropped with it.
+  *
+  * The point of the ownership is not tidiness. An unowned global pool could neither say which build was holding the heap nor free it when that build was
+  * evicted — so a `BuildCache` eviction freed the resolved build (hundreds of MB) and left its analyses (gigabytes) behind.
+  */
+class AnalysisCacheOwnershipTest extends AnyFunSuite with Matchers {
+
+  /** The cache stores analyses and reports their file size; it never looks inside one. */
+  private object StubAnalysis extends CompileAnalysis {
+    def readStamps(): ReadStamps = ???
+    def readSourceInfos(): ReadSourceInfos = ???
+    def readCompilations(): ReadCompilations = ???
+  }
+
+  private def key(name: String): model.WorkspaceKey =
+    model.WorkspaceKey(Path.of(s"/tmp/$name"), model.BuildVariant.Normal)
+
+  /** A real file, because `put` sizes the entry from disk. */
+  private def analysisFile(bytes: Int): Path = {
+    val f = Files.createTempFile("analysis", ".zip")
+    Files.write(f, Array.fill(bytes)(0: Byte))
+    f.toFile.deleteOnExit()
+    f
+  }
+
+  private def cache(): AnalysisCache = new AnalysisCache
+
+  test("analyses are partitioned by workspace: one workspace cannot see another's") {
+    val c = cache()
+    val f = analysisFile(64)
+    val mtime = Files.getLastModifiedTime(f).toMillis
+    c.put(key("alpha"), f, mtime, StubAnalysis): Unit
+
+    c.get(key("alpha"), f, mtime) shouldBe defined
+    // Same analysis FILE, different workspace. Nothing is shared across workspaces on disk, and
+    // nothing is shared here either — otherwise one workspace's entries would be charged to another.
+    c.get(key("beta"), f, mtime) shouldBe empty
+  }
+
+  test("a changed file on disk invalidates the entry rather than serving a stale analysis") {
+    val c = cache()
+    val f = analysisFile(64)
+    val mtime = Files.getLastModifiedTime(f).toMillis
+    c.put(key("alpha"), f, mtime, StubAnalysis): Unit
+
+    c.get(key("alpha"), f, mtime) shouldBe defined
+    c.get(key("alpha"), f, mtime + 1) shouldBe empty // e.g. after `remote-cache pull`
+  }
+
+  test("evicting a workspace frees only its own analyses, and reports what it freed") {
+    val c = cache()
+    val a1 = analysisFile(100)
+    val a2 = analysisFile(200)
+    val b1 = analysisFile(400)
+    c.put(key("alpha"), a1, Files.getLastModifiedTime(a1).toMillis, StubAnalysis): Unit
+    c.put(key("alpha"), a2, Files.getLastModifiedTime(a2).toMillis, StubAnalysis): Unit
+    c.put(key("beta"), b1, Files.getLastModifiedTime(b1).toMillis, StubAnalysis): Unit
+
+    val freed = c.evictWorkspace(key("alpha"))
+    freed.entries shouldBe 2
+    freed.fileBytes shouldBe 300L
+
+    c.get(key("alpha"), a1, Files.getLastModifiedTime(a1).toMillis) shouldBe empty
+    c.get(key("beta"), b1, Files.getLastModifiedTime(b1).toMillis) shouldBe defined
+  }
+
+  test("evicting a workspace that holds nothing is a no-op, not an error") {
+    val c = cache()
+    c.evictWorkspace(key("never-seen")) shouldBe AnalysisCache.Freed(0, 0L)
+  }
+
+  test("stats total across workspaces and are ordered by what they hold") {
+    val c = cache()
+    val small = analysisFile(10)
+    val large = analysisFile(900)
+    c.put(key("small-ws"), small, Files.getLastModifiedTime(small).toMillis, StubAnalysis): Unit
+    c.put(key("large-ws"), large, Files.getLastModifiedTime(large).toMillis, StubAnalysis): Unit
+
+    val stats = c.stats
+    stats.entries shouldBe 2
+    stats.fileBytes shouldBe 910L
+    // Biggest first: the whole point of per-workspace stats is answering "who is holding the heap".
+    stats.perWorkspace.head.key shouldBe key("large-ws")
+  }
+}

bleep-bsp-tests/src/scala/bleep/analysis/AnalysisDedupMeasurementTest.scala

@@ -0,0 +1,134 @@
+package bleep.analysis
+
+import org.scalatest.funsuite.AnyFunSuite
+import org.scalatest.matchers.should.Matchers
+import xsbti.api.AnalyzedClass
+
+import java.nio.file.{Files, Path, Paths}
+import java.security.MessageDigest
+import scala.jdk.CollectionConverters.*
+import scala.jdk.OptionConverters.*
+
+/** How much of the compile server's heap would a leaf interner give back?
+  *
+  * Not a unit test of behaviour — a measurement, kept in the suite so the number can be reproduced rather than quoted from a session that has scrolled away. It
+  * loads real `analysis.zip` files and reports how many `AnalyzedClass` instances are duplicates of each other.
+  *
+  * ==Why AnalyzedClass is the number that matters==
+  *
+  * A class histogram of a live daemon at 7.2GB live set found ~4.5GB in `xsbti.api.*`: 31.7M `NameHash` (1.0GB), 31.1M `Id` (498MB), 7.4M `PathComponent[]`
+  * (428MB), 839K `AnalyzedClass`, 839K `NameHash[]` (267MB). `AnalyzedClass` count matches `NameHash[]` count exactly, and each one owns the lazy `Companions`
+  * tree holding all of the above — so the duplication factor measured here is the factor by which that 4.5GB shrinks.
+  *
+  * ==Reading the output==
+  *
+  * `distinct` counts by the proposed intern key — everything `AnalyzedClass` carries EXCEPT `compilationTimestamp`, which zinc 1.12 reads in one place as a
+  * fast-path gate that falls back to a full structural diff, and which zinc 2.0.0 stopped reading altogether. The run reports both keys so the cost of
+  * including the timestamp is visible rather than assumed: with it, two worktrees that compiled the same code independently do not merge.
+  *
+  * Point it at more workspaces with `BLEEP_DEDUP_ROOTS` (colon-separated workspace roots). With none set it measures this build alone, which shows only the
+  * within-workspace sharing (library stamps, shared dependencies) and not the cross-worktree case that motivates the work.
+  */
+class AnalysisDedupMeasurementTest extends AnyFunSuite with Matchers {
+
+  /** ScalaTest's `info` is swallowed by the runner's reporter, so the measurement is written where it can be read afterwards. */
+  private val report = Paths.get(sys.props.getOrElse("user.dir", ".")).resolve("dedup-report.txt").toAbsolutePath
+  private val lines = scala.collection.mutable.ListBuffer.empty[String]
+  private def say(s: String): Unit = { lines += s; println(s"[dedup] $s") }
+  private def flush(): Unit = Files.write(report, (s"report: $report\n" + lines.mkString("\n")).getBytes("UTF-8")): Unit
+
+  private def analysisFilesUnder(root: Path): List[Path] =
+    if (!Files.isDirectory(root)) Nil
+    else {
+      val stream = Files.walk(root)
+      try stream.iterator().asScala.filter(p => p.getFileName.toString == "analysis.zip" && Files.isRegularFile(p)).toList
+      finally stream.close()
+    }
+
+  /** Roots to measure, one per line in `dedup-roots.txt` beside the build, else this build's own `.bleep`.
+    *
+    * A file rather than an env var because the test runs in a forked JVM that does not inherit one.
+    */
+  private def roots: List[Path] = {
+    val cwd = Paths.get(sys.props.getOrElse("user.dir", "."))
+    val listing = cwd.resolve("dedup-roots.txt")
+    if (Files.isRegularFile(listing))
+      Files.readAllLines(listing).asScala.toList.map(_.trim).filter(_.nonEmpty).map(Paths.get(_))
+    else List(cwd.resolve(".bleep"))
+  }
+
+  /** The proposed intern key. `withTimestamp` shows what including `compilationTimestamp` would cost in hit rate. */
+  private def internKey(ac: AnalyzedClass, withTimestamp: Boolean): String = {
+    val md = MessageDigest.getInstance("SHA-256")
+    md.update(ac.name().getBytes("UTF-8"))
+    md.update(java.nio.ByteBuffer.allocate(4).putInt(ac.apiHash()).array())
+    md.update(java.nio.ByteBuffer.allocate(4).putInt(ac.extraHash()).array())
+    md.update(if (ac.hasMacro()) Array[Byte](1) else Array[Byte](0))
+    md.update(ac.provenance().getBytes("UTF-8"))
+    // nameHashes is an array of (name, scope, hash) — digest it rather than holding it in the key.
+    ac.nameHashes().sortBy(nh => (nh.name(), nh.scope().ordinal())).foreach { nh =>
+      md.update(nh.name().getBytes("UTF-8"))
+      md.update(java.nio.ByteBuffer.allocate(4).putInt(nh.scope().ordinal()).array())
+      md.update(java.nio.ByteBuffer.allocate(4).putInt(nh.hash()).array())
+    }
+    if (withTimestamp) md.update(java.nio.ByteBuffer.allocate(8).putLong(ac.compilationTimestamp()).array())
+    md.digest().map("%02x".format(_)).mkString
+  }
+
+  test("MEASUREMENT: duplicate AnalyzedClass across real analysis files") {
+    val files = roots.flatMap(analysisFilesUnder)
+    if (files.isEmpty) {
+      say(s"no analysis.zip found under ${roots.mkString(", ")} — compile something first, or set BLEEP_DEDUP_ROOTS")
+      flush()
+      cancel("nothing to measure")
+    }
+
+    val totalBytes = files.map(Files.size).sum
+    say(f"${files.size}%d analysis files, ${totalBytes / (1024.0 * 1024)}%.1f MB on disk, across ${roots.size}%d root(s)")
+
+    var total = 0L
+    val distinctNoTs = scala.collection.mutable.HashSet.empty[String]
+    val distinctWithTs = scala.collection.mutable.HashSet.empty[String]
+    var unreadable = 0
+
+    files.foreach { f =>
+      // The same store production uses, so this measures what bleep actually writes. Files left
+      // over from zinc 1.x are in the old protobuf format and simply do not read here — they count
+      // as unreadable and the run reports them.
+      val store = sbt.internal.inc.consistent.ConsistentFileAnalysisStore.binary(
+        f.toFile,
+        ZincBridge.analysisMappers(f),
+        reproducible = true,
+        parallelism = 4
+      )
+      store.get().toScala match {
+        case None           => unreadable += 1
+        case Some(contents) =>
+          contents.getAnalysis match {
+            case a: sbt.internal.inc.Analysis =>
+              val apis = a.apis
+              (apis.internal.values ++ apis.external.values).foreach { ac =>
+                total += 1
+                distinctNoTs += internKey(ac, withTimestamp = false)
+                distinctWithTs += internKey(ac, withTimestamp = true)
+              }
+            case other => say(s"unexpected analysis type ${other.getClass.getName}")
+          }
+      }
+    }
+
+    if (unreadable > 0) say(s"$unreadable file(s) could not be read")
+
+    val ratioNoTs = if (distinctNoTs.isEmpty) 0.0 else total.toDouble / distinctNoTs.size
+    val ratioWithTs = if (distinctWithTs.isEmpty) 0.0 else total.toDouble / distinctWithTs.size
+    say(f"AnalyzedClass total=$total%d  distinct(no timestamp)=${distinctNoTs.size}%d  →  ${ratioNoTs}%.2fx sharing")
+    say(f"AnalyzedClass total=$total%d  distinct(with timestamp)=${distinctWithTs.size}%d  →  ${ratioWithTs}%.2fx sharing")
+    say(f"cost of keeping timestamp in the key: ${distinctWithTs.size - distinctNoTs.size}%d extra retained instances")
+
+    flush()
+    // A measurement with no data is not a failure — a fresh checkout, or the window right after an
+    // analysis-format change, legitimately has nothing readable. Say so and skip.
+    if (total == 0L) cancel(s"no readable analyses (${files.size} file(s) found, $unreadable unreadable)")
+    distinctNoTs.size should be <= total.toInt
+  }
+}

bleep-bsp-tests/src/scala/bleep/analysis/BspTestHarness.scala

@@ -199,6 +199,7 @@ class BspTestHarness(workspaceRoot: Path, projectConfigs: Option[List[BspTestHar
     // The production server. It has no way to be handed build state directly — it compiles the
     // build its client sends — so the configs are lowered into the same payload a real bleep
     // client would send, and delivered through build/initialize below.
+    val harnessAnalysisCache = new bleep.analysis.AnalysisCache
     val server = new MultiWorkspaceBspServer(
       serverInput,
       serverToClient,
@@ -207,7 +208,8 @@ class BspTestHarness(workspaceRoot: Path, projectConfigs: Option[List[BspTestHar
       heapMonitor = HeapMonitor.system,
       // One server per harness, so fresh daemon-scoped state is the right scope here.
       kspMutexes = new KspMutexes,
-      buildCache = new BuildCache
+      buildCache = new BuildCache(bleep.model.BspServerConfig.default.effectiveMaxCachedWorkspaces, harnessAnalysisCache),
+      analysisCache = harnessAnalysisCache
     )
 
     val buildPayload: Option[BspBuildData.Payload] =

bleep-bsp-tests/src/scala/bleep/analysis/CompilerIntegrationTest.scala

@@ -1506,7 +1506,8 @@ class CompilerVersionIsolationTest extends AnyFunSuite with Matchers {
           DiagnosticListener.noop,
           CancellationToken.never,
           Map.empty,
-          ProgressListener.noop
+          ProgressListener.noop,
+          AnalysisCache.standalone(config.buildDir)
         )
         .unsafeRunSync()
 

bleep-bsp-tests/src/scala/bleep/analysis/IncrementalCompilationTest.scala

@@ -205,7 +205,9 @@ class IncrementalTrackingTest extends AnyFunSuite with Matchers {
           diagnosticListener = listener,
           cancellationToken = CancellationToken.never,
           dependencyAnalyses = Map.empty,
-          progressListener = ProgressListener.noop
+          progressListener = ProgressListener.noop,
+          ecjVersion = None,
+          analyses = AnalysisCache.standalone(config.buildDir)
         )
         .unsafeRunSync()
 
@@ -265,7 +267,9 @@ class IncrementalTrackingTest extends AnyFunSuite with Matchers {
           diagnosticListener = listener1,
           cancellationToken = CancellationToken.never,
           dependencyAnalyses = Map.empty,
-          progressListener = ProgressListener.noop
+          progressListener = ProgressListener.noop,
+          ecjVersion = None,
+          analyses = AnalysisCache.standalone(config.buildDir)
         )
         .unsafeRunSync()
 
@@ -287,7 +291,9 @@ class IncrementalTrackingTest extends AnyFunSuite with Matchers {
           diagnosticListener = listener2,
           cancellationToken = CancellationToken.never,
           dependencyAnalyses = Map.empty,
-          progressListener = ProgressListener.noop
+          progressListener = ProgressListener.noop,
+          ecjVersion = None,
+          analyses = AnalysisCache.standalone(config.buildDir)
         )
         .unsafeRunSync()
 

bleep-bsp-tests/src/scala/bleep/analysis/LinkDagIntegrationTest.scala

@@ -20,6 +20,10 @@ import org.scalatest.matchers.should.Matchers
   */
 class LinkDagIntegrationTest extends AnyFunSuite with Matchers {
 
+  /** A machine for tests: enough CPU for the parallelism a case asks for, ample fork memory so admission never turns on it. */
+  private def testMachine(cpu: Int): bleep.MachineResources =
+    bleep.MachineResources.create(totalCpu = cpu, totalMemoryMb = 64 * 1024, logger = ryddig.TypedLogger.DevNull, longWaitWarnMs = 60000L)
+
   /** Helper to create CrossProjectName from a simple string. */
   private def projectName(name: String): CrossProjectName =
     CrossProjectName(ProjectName(name), None)
@@ -264,7 +268,7 @@ class LinkDagIntegrationTest extends AnyFunSuite with Matchers {
     val result = (for {
       eventQueue <- Queue.unbounded[IO, Option[DagEvent]]
       killSignal <- Outcome.neverKillSignal
-      finalDag <- executor.execute(dag, 4, eventQueue, killSignal)
+      finalDag <- executor.execute(dag, testMachine(4), TaskDag.ForkHeaps.default, eventQueue, killSignal)
     } yield finalDag).unsafeRunSync()
 
     linkCalled shouldBe true
@@ -309,7 +313,7 @@ class LinkDagIntegrationTest extends AnyFunSuite with Matchers {
     val events = (for {
       eventQueue <- Queue.unbounded[IO, Option[DagEvent]]
       killSignal <- Outcome.neverKillSignal
-      _ <- executor.execute(dag, 4, eventQueue, killSignal)
+      _ <- executor.execute(dag, testMachine(4), TaskDag.ForkHeaps.default, eventQueue, killSignal)
       allEvents <- drainQueue(eventQueue)
     } yield allEvents).unsafeRunSync()
 
@@ -354,7 +358,7 @@ class LinkDagIntegrationTest extends AnyFunSuite with Matchers {
     val result = (for {
       eventQueue <- Queue.unbounded[IO, Option[DagEvent]]
       killSignal <- Outcome.neverKillSignal
-      finalDag <- executor.execute(dag, 4, eventQueue, killSignal)
+      finalDag <- executor.execute(dag, testMachine(4), TaskDag.ForkHeaps.default, eventQueue, killSignal)
     } yield finalDag).unsafeRunSync()
 
     result.failed should contain(TaskId.Link(project))

bleep-bsp-tests/src/scala/bleep/analysis/PlatformTestHelper.scala

@@ -144,7 +144,9 @@ trait PlatformTestHelper {
         DiagnosticListener.noop,
         CancellationToken.never,
         Map.empty,
-        ProgressListener.noop
+        ProgressListener.noop,
+        None,
+        AnalysisCache.standalone(config.buildDir)
       )
       .unsafeRunSync()
 
@@ -213,7 +215,9 @@ trait PlatformTestHelper {
         DiagnosticListener.noop,
         CancellationToken.never,
         Map.empty,
-        ProgressListener.noop
+        ProgressListener.noop,
+        None,
+        AnalysisCache.standalone(config.buildDir)
       )
       .unsafeRunSync()