Fine-grained task-level concurrency between launchers

core/api/daemon/src/mill/api/daemon/internal/BuildFileApi.scala

@@ -5,11 +5,18 @@ import mill.api.daemon.Watchable
 trait BuildFileApi {
   def rootModule: RootModuleApi
   def moduleWatchedValues: Seq[Watchable]
-  def evalWatchedValues: collection.mutable.Buffer[Watchable]
+
+  /**
+   * Run `body` with a fresh per-evaluation eval-watch buffer installed on the
+   * current thread; returns the body result and the accumulated watches.
+   * Concurrent evaluations get isolated buffers so they cannot lose or mix
+   * each other's watches.
+   */
+  def withEvalWatchedValues[T](body: => T): (T, Seq[Watchable])
 }
 object BuildFileApi {
   class Bootstrap(val rootModule: RootModuleApi) extends BuildFileApi {
     def moduleWatchedValues = Nil
-    def evalWatchedValues = collection.mutable.Buffer[Watchable]()
+    def withEvalWatchedValues[T](body: => T): (T, Seq[Watchable]) = (body, Nil)
   }
 }

core/api/daemon/src/mill/api/daemon/internal/EvaluatorApi.scala

@@ -40,6 +40,14 @@ trait EvaluatorApi extends AutoCloseable {
       @unused allowPositionalCommandArgs: Boolean = false
   ): Result[Boolean] = Result.Success(false)
 
+  private[mill] def probeSelectiveReuse(
+      @unused scriptArgs: Seq[String],
+      @unused selectMode: SelectMode,
+      @unused previousMetadata: String,
+      @unused allowPositionalCommandArgs: Boolean = false
+  ): Result[EvaluatorApi.SelectiveReuseDecision] =
+    Result.Success(EvaluatorApi.SelectiveReuseDecision(reusable = false, previousMetadata))
+
   /**
    * Returns a copy of this evaluator with the isFinalDepth flag set to the given value.
    * Used to defer the decision of whether this is the final depth until after
@@ -48,6 +56,8 @@ trait EvaluatorApi extends AutoCloseable {
   private[mill] def withIsFinalDepth(isFinalDepth: Boolean): EvaluatorApi = this
 }
 object EvaluatorApi {
+  private[mill] case class SelectiveReuseDecision(reusable: Boolean, nextMetadata: String)
+
   trait Result[T] {
     def watchable: Seq[Watchable]
     def values: mill.api.daemon.Result[Seq[T]]

core/api/daemon/src/mill/api/daemon/internal/LauncherLocking.scala

@@ -0,0 +1,146 @@
+package mill.api.daemon.internal
+
+import java.io.PrintStream
+import java.nio.file.Path
+
+/**
+ * Locking APIs that let multiple launchers run concurrently. Acquisition order:
+ * [[metaBuildLock]] -> [[exclusiveLock]] -> [[taskLock]] (outer to inner).
+ */
+private[mill] trait LauncherLocking extends AutoCloseable {
+
+  /**
+   * Per-meta-build-depth lock. Read-then-Write upgrade during bootstrap; Read
+   * lease is retained for the rest of the launcher run after the frame is
+   * published, so concurrent launchers share the same frame via Reads.
+   */
+  def metaBuildLock(
+      depth: Int,
+      kind: LauncherLocking.LockKind,
+      waitReporter: LauncherLocking.WaitReporter
+  ): LauncherLocking.Lease
+
+  /**
+   * Non-blocking, non-queued Write attempt on the meta-build lock at
+   * `depth`. Returns [[scala.None]] if Write is not immediately
+   * available; the caller can then back off, re-probe under Read, and
+   * decide whether Write is still needed. Failed tries do NOT register
+   * as queued writers, so the caller's subsequent Read attempts are not
+   * blocked by writer-priority — this is what makes the retryable
+   * read-then-write pattern in [[mill.internal.LockUpgrade.readThenWrite]]
+   * work without poisoning the caller's own re-probe.
+   */
+  def tryMetaBuildWriteLock(depth: Int): Either[String, LauncherLocking.Lease]
+
+  /**
+   * Block on the meta-build lock at `depth` for up to `timeoutMs`,
+   * returning on the first lock state change (close/downgrade) or
+   * timeout. Used by the retryable read-then-write loop to sleep
+   * efficiently between try-Write attempts.
+   */
+  def awaitMetaBuildStateChange(depth: Int, timeoutMs: Long): Unit
+
+  /**
+   * Per-task-`dest` lock. Read-then-Write upgrade on cache miss; Read lease is
+   * retained via `Execution.LeaseTracker` until all transitive downstream
+   * terminals complete, so concurrent launchers cannot overwrite outputs that
+   * another launcher's downstream is still reading.
+   */
+  def taskLock(
+      path: Path,
+      displayLabel: String,
+      kind: LauncherLocking.LockKind,
+      waitReporter: LauncherLocking.WaitReporter
+  ): LauncherLocking.Lease
+
+  /**
+   * Non-blocking, non-queued Write counterpart of [[taskLock]].
+   * See [[tryMetaBuildWriteLock]] for semantics.
+   */
+  def tryTaskWriteLock(
+      path: Path,
+      displayLabel: String
+  ): Either[String, LauncherLocking.Lease]
+
+  /**
+   * Bounded await on per-task lock state changes; counterpart of
+   * [[awaitMetaBuildStateChange]].
+   */
+  def awaitTaskStateChange(path: Path, displayLabel: String, timeoutMs: Long): Unit
+
+  /**
+   * Daemon-wide lock taken by each task batch. Normal batches take Read so
+   * they overlap; `Task.Command(exclusive = true)` batches take Write so they
+   * run alone. Tasks that mutate in-tree sources (e.g. `build.mill`) MUST be
+   * declared `exclusive = true` — there is no per-source lock.
+   */
+  def exclusiveLock(
+      kind: LauncherLocking.LockKind,
+      waitReporter: LauncherLocking.WaitReporter
+  ): LauncherLocking.Lease
+}
+
+private[mill] object LauncherLocking {
+  enum LockKind {
+    case Read, Write
+  }
+
+  trait Lease extends AutoCloseable {
+    def downgradeToRead(): Unit = ()
+  }
+
+  /**
+   * Surface a "blocked on lock" status to the user when a lock acquisition has
+   * to wait. Implementations should display the message in a way that doesn't
+   * disturb the active console UI (e.g. via the multi-line prompt's detail
+   * line) and clear it when the returned token is closed.
+   *
+   * Implementations may also respond to holder changes by re-calling
+   * `reportWait` — each call replaces the previous wait status.
+   */
+  trait WaitReporter {
+    def reportWait(message: String): AutoCloseable
+  }
+
+  object WaitReporter {
+    private val NoopToken: AutoCloseable = () => ()
+
+    /** Discards wait events. Useful for tests and noBuildLock. */
+    val Noop: WaitReporter = (_: String) => NoopToken
+
+    /**
+     * Prints the wait message once on `reportWait` and does nothing on close.
+     * Matches the legacy behavior: a one-shot stderr line that scrolls into
+     * the user's terminal history. Used when no live prompt is available
+     * (early bootstrap, no-daemon, BSP, `--ticker false`, etc.).
+     */
+    def stderr(stream: PrintStream): WaitReporter = (msg: String) => {
+      stream.println(msg)
+      NoopToken
+    }
+  }
+
+  object Noop extends LauncherLocking {
+    private object NoopLease extends Lease {
+      override def close(): Unit = ()
+    }
+    override def metaBuildLock(
+        depth: Int,
+        kind: LockKind,
+        waitReporter: WaitReporter
+    ): Lease = NoopLease
+    override def tryMetaBuildWriteLock(depth: Int): Either[String, Lease] = Right(NoopLease)
+    override def awaitMetaBuildStateChange(depth: Int, timeoutMs: Long): Unit = ()
+    override def taskLock(
+        path: Path,
+        displayLabel: String,
+        kind: LockKind,
+        waitReporter: WaitReporter
+    ): Lease = NoopLease
+    override def tryTaskWriteLock(path: Path, displayLabel: String): Either[String, Lease] =
+      Right(NoopLease)
+    override def awaitTaskStateChange(path: Path, displayLabel: String, timeoutMs: Long): Unit = ()
+    override def exclusiveLock(kind: LockKind, waitReporter: WaitReporter): Lease = NoopLease
+    override def close(): Unit = ()
+  }
+}

core/api/daemon/src/mill/api/daemon/internal/LauncherOutFiles.scala

@@ -0,0 +1,31 @@
+package mill.api.daemon.internal
+
+import mill.constants.DaemonFiles
+import mill.constants.OutFiles
+
+import java.nio.file.Path
+
+private[mill] trait LauncherOutFiles extends AutoCloseable {
+  def consoleTail: Path
+  def profile: Path
+  def chromeProfile: Path
+  def dependencyTree: Path
+  def invalidationTree: Path
+
+  def publishLiveArtifacts(): Unit
+
+  def publishArtifacts(): Unit
+}
+
+private[mill] object LauncherOutFiles {
+  def noop(out: Path): LauncherOutFiles = new LauncherOutFiles {
+    override val consoleTail: Path = out.resolve(DaemonFiles.millConsoleTail)
+    override val profile: Path = out.resolve(OutFiles.millProfile)
+    override val chromeProfile: Path = out.resolve(OutFiles.millChromeProfile)
+    override val dependencyTree: Path = out.resolve(OutFiles.millDependencyTree)
+    override val invalidationTree: Path = out.resolve(OutFiles.millInvalidationTree)
+    override def publishLiveArtifacts(): Unit = ()
+    override def publishArtifacts(): Unit = ()
+    override def close(): Unit = ()
+  }
+}

core/api/daemon/src/mill/api/daemon/internal/PriorityRunnable.scala

@@ -0,0 +1,30 @@
+package mill.api.daemon.internal
+
+import java.util.concurrent.atomic.AtomicLong
+
+/**
+ * Runnable wrapper used by daemon-level priority queues.
+ *
+ * This lives in `mill.api.daemon.internal` so it is loaded from the shared daemon API
+ * classloader rather than from refreshable build/evaluator classloaders.
+ */
+private[mill] class PriorityRunnable(val priority: Int, run0: () => Unit)
+    extends Runnable
+    with Comparable[PriorityRunnable] {
+  def run(): Unit = run0()
+
+  val priorityRunnableIndex: Long = PriorityRunnable.priorityRunnableCount.getAndIncrement()
+
+  override def compareTo(o: PriorityRunnable): Int = priority.compareTo(o.priority) match {
+    case 0 =>
+      // `Comparable` wants a total ordering. This index is assigned when a task
+      // is submitted, so equal-priority work runs in submission order.
+      assert(this == o || this.priorityRunnableIndex != o.priorityRunnableIndex)
+      this.priorityRunnableIndex.compareTo(o.priorityRunnableIndex)
+    case n => n
+  }
+}
+
+private object PriorityRunnable {
+  private val priorityRunnableCount = new AtomicLong()
+}

core/api/daemon/src/mill/api/daemon/internal/bsp/BspBootstrapBridge.scala

@@ -0,0 +1,25 @@
+package mill.api.daemon.internal.bsp
+
+import mill.api.daemon.Watchable
+import mill.api.daemon.internal.{CompileProblemReporter, EvaluatorApi}
+
+/**
+ * Daemon-side hook the BSP worker calls once per BSP request. The worker
+ * supplies the active command label, a meta-build compile reporter keyed by
+ * depth, and the body that consumes the freshly-bootstrapped evaluators; the
+ * daemon runs `MillBuildBootstrap` and threads the resulting evaluators (with
+ * the watches captured during the run, plus any error message) into `body`.
+ *
+ * Declared as a SAM trait rather than a Scala 3 polymorphic-function alias so
+ * the worker can locate the method via `Class.getMethod` without depending on
+ * the polymorphic-function's erased shape (the previous `[T] => (a, b, c) => T`
+ * alias erased to `scala.Function3` and caused a `NoSuchMethodException` at
+ * the reflection boundary when the lookup mistakenly used `Function2`).
+ */
+trait BspBootstrapBridge {
+  def apply[T](
+      activeCommandMessage: String,
+      metaBuildReporter: Int => Option[CompileProblemReporter],
+      body: (Seq[EvaluatorApi], Seq[Watchable], Option[String]) => T
+  ): T
+}

core/api/daemon/src/mill/api/daemon/internal/bsp/BspServerHandle.scala

@@ -1,23 +1,14 @@
 package mill.api.daemon.internal.bsp
 
-import mill.api.daemon.internal.EvaluatorApi
-import mill.api.daemon.Watchable
-
 import scala.concurrent.Future
 
 /** With this server handle you can interact with a running Mill BSP server. */
 trait BspServerHandle {
 
   /**
-   * Starts a new session with the given evaluator. Doesn't block or wait for the session to end.
+   * Completes when the BSP server shuts down or asks the launcher to reload the workspace.
    */
-  def startSession(
-      evaluators: Seq[EvaluatorApi],
-      errored: Boolean,
-      watched: Seq[Watchable]
-  ): Future[BspServerResult]
-
-  def resetSession(): Unit
+  def shutdownFuture: Future[BspServerResult]
 
   /** Stops the BSP server. */
   def close(): Unit

core/api/java11/src/mill/api/BuildCtx.scala

@@ -37,8 +37,32 @@ object BuildCtx {
   def withFilesystemCheckerDisabled[T](block: => T): T =
     os.checker.withValue(os.Checker.Nop) { block }
 
-  protected[mill] val watchedValues: mutable.Buffer[Watchable] = mutable.Buffer.empty[Watchable]
-  protected[mill] val evalWatchedValues: mutable.Buffer[Watchable] = mutable.Buffer.empty[Watchable]
+  // Module-init watches: populated by `<clinit>` of the build classloader
+  // (serialized by the JVM), so a shared default buffer is safe.
+  private val moduleWatchedDefault: mutable.Buffer[Watchable] = mutable.Buffer.empty[Watchable]
+
+  // Eval watches: populated per evaluation via `BuildCtx.evalWatch`.
+  // The DynamicVariable + `ExecutionContexts.ThreadPool` capture/rebind
+  // give each evaluation an isolated buffer so concurrent launchers
+  // don't clobber each other's watches.
+  private[mill] val watchedValues0: DynamicVariable[mutable.Buffer[Watchable]] =
+    new DynamicVariable(moduleWatchedDefault)
+  private[mill] val evalWatchedValues0: DynamicVariable[mutable.Buffer[Watchable]] =
+    new DynamicVariable(mutable.Buffer.empty[Watchable])
+
+  protected[mill] def watchedValues: mutable.Buffer[Watchable] = watchedValues0.value
+  protected[mill] def evalWatchedValues: mutable.Buffer[Watchable] = evalWatchedValues0.value
+
+  /**
+   * Run `body` with a fresh per-evaluation `evalWatchedValues` buffer; returns
+   * the buffer alongside the body result so the caller can collect the
+   * accumulated watches without racing other evaluations.
+   */
+  private[mill] def withEvalWatchedValues[T](body: => T): (T, mutable.Buffer[Watchable]) = {
+    val buf = mutable.Buffer.empty[Watchable]
+    val result = evalWatchedValues0.withValue(buf)(body)
+    (result, buf)
+  }
 
   /**
    * Register a compute value as watched during module initialization, so Mill knows
@@ -52,23 +76,23 @@ object BuildCtx {
         v.hashCode(),
         fn.value + ":" + ln.value
       )
-      watchedValues.append(watchable)
+      watchedValues0.value.append(watchable)
       v
     }
   }
 
   def watch(p: os.Path): os.Path = withFilesystemCheckerDisabled {
     val watchable = Watchable.Path.from(PathRef(p))
-    watchedValues.append(watchable)
+    watchedValues0.value.append(watchable)
     p
   }
   def evalWatch(p: os.Path): os.Path = withFilesystemCheckerDisabled {
     val watchable = Watchable.Path.from(PathRef(p))
-    evalWatchedValues.append(watchable)
+    evalWatchedValues0.value.append(watchable)
     p
   }
 
-  def watch0(w: Watchable): Unit = watchedValues.append(w)
+  def watch0(w: Watchable): Unit = watchedValues0.value.append(w)
 
-  def evalWatch0(w: Watchable): Unit = evalWatchedValues.append(w)
+  def evalWatch0(w: Watchable): Unit = evalWatchedValues0.value.append(w)
 }