Use a single Mill server process (#5066)

Fixes #5031

* We now only have a single server process running at a time, and no
longer keep up to 5 separate `serverDir-1` to `serverDir-5` folders.
Essentially we are migrating from the style of concurrency used by
`gradle` and `mvnd` to the style of concurrency used by `bazel`

* We now only hold the `clientLock` until the server process is
initialized (and has taken it's `processLock`) to avoid redundant
initialization, and after that we release it to allow multiple clients
to connect

* Since there is only one server and one `serverDir`, we have to remove
several per-server files since there is no way to ensure they are unique
in the presence of multiple clients
* `runArgs` can no longer be a file, and instead we send its contents
over the socket before starting the `InputPumper`

* We can no longer rely on `clientLock` probing to check if the client
has disconnected, and so we instead have the server send
`ProxyStream.HEARTBEAT` bytes (127) and look for errors

* `Server#run` can no longer accept socket connections and process them
in a single-threaded loop, and instead needs to spawn off threads to
handle each connected socket

* We cannot perform the mutex at the server/client/connection level
because there are scenarios we want multiple clients connected at once,
e.g. when one client is in `--watch` mode and idle we want to allow
another client to proceed with execution. So locking needs to be done at
a finer-grained granularity within the server

* The original `millLock` file (renamed `millOutLock`) has been combined
with a `MemoryLock` via `DoubleLock`, which lets us check the
memory-lock first to avoid taking the file-lock twice on the same JVM
and causing a `OverlappingFileLockException`. We still need to keep the
file-lock to provide mutex between server and no-server processes, or
between multiple no-server processes

Now that things are in one process, that opens up the possibility of
loosening the concurrency constraints in future, e.g. allowing multiple
concurrent evaluations that do not interfere with each other.

Author: lihaoyi

core/constants/src/mill/constants/OutFiles.java

@@ -60,7 +60,7 @@ public class OutFiles {
   /**
    * Lock file used for exclusive access to the Mill output directory
    */
-  public static final String millLock = "mill-lock";
+  public static final String millOutLock = "mill-out-lock";
 
   /**
    * Any active Mill command that is currently run, for debugging purposes

core/constants/src/mill/constants/ProxyStream.java

@@ -35,6 +35,7 @@ public class ProxyStream {
   public static final int OUT = 1;
   public static final int ERR = -1;
   public static final int END = 0;
+  public static final int HEARTBEAT = 127;
 
   public static void sendEnd(OutputStream out) throws IOException {
     synchronized (out) {
@@ -43,6 +44,13 @@ public static void sendEnd(OutputStream out) throws IOException {
     }
   }
 
+  public static void sendHeartbeat(OutputStream out) throws IOException {
+    synchronized (out) {
+      out.write(ProxyStream.HEARTBEAT);
+      out.flush();
+    }
+  }
+
   public static class Output extends java.io.OutputStream {
     private final java.io.OutputStream destination;
     private final int key;
@@ -75,7 +83,7 @@ public void write(byte[] b, int off, int len) throws IOException {
       synchronized (destination) {
         int i = 0;
         while (i < len && i + off < b.length) {
-          int chunkLength = Math.min(len - i, 127);
+          int chunkLength = Math.min(len - i, 126);
           if (chunkLength > 0) {
             destination.write(chunkLength * key);
             destination.write(b, off + i, Math.min(b.length - off - i, chunkLength));
@@ -136,7 +144,8 @@ public void run() {
           // that only header values > 0 represent actual data to read:
           // - sign((byte)header) represents which stream the data should be sent to
           // - abs((byte)header) represents the length of the data to read and send
-          if (header == -1 || header == 0) break;
+          if (header == -1 || header == END) break;
+          else if (header == HEARTBEAT) continue;
           else {
             int stream = (byte) header > 0 ? 1 : -1;
             int quantity0 = (byte) header;

core/constants/src/mill/constants/ServerFiles.java

@@ -20,7 +20,7 @@ public class ServerFiles {
    * folder. If multiple servers are spawned in the same folder, only one takes
    * the lock and the others fail to do so and terminate immediately.
    */
-  public static final String processLock = "processLock";
+  public static final String serverLock = "serverLock";
 
   /**
    * The port used to connect between server and client
@@ -48,12 +48,6 @@ public static String pipe(String base) {
    */
   public static final String serverLog = "server.log";
 
-  /**
-   * File that the client writes to pass the arguments, environment variables,
-   * and other necessary metadata to the Mill server to kick off a run
-   */
-  public static final String runArgs = "runArgs";
-
   /**
    * File the server writes to pass the exit code of a completed run back to the
    * client

example/fundamentals/out-dir/1-out-files/build.mill

@@ -29,7 +29,7 @@ out/mill-build/...
 out/mill-profile.json
 out/mill-runner-state.json
 out/mill-dependency-tree.json
-out/mill-lock
+out/mill-out-lock
 out/mill-invalidation-tree.json
 out/mill-chrome-profile.json
 out/mill-server/...

runner/bsp/worker/src/mill/bsp/worker/BspWorkerImpl.scala

@@ -5,6 +5,7 @@ import mill.bsp.BuildInfo
 import mill.api.internal.{BspServerHandle, BspServerResult, EvaluatorApi}
 import mill.bsp.Constants
 import mill.api.{Result, SystemStreams}
+import mill.client.lock.Lock
 import org.eclipse.lsp4j.jsonrpc.Launcher
 
 import java.io.PrintWriter
@@ -17,7 +18,8 @@ object BspWorkerImpl {
       topLevelBuildRoot: os.Path,
       streams: SystemStreams,
       logDir: os.Path,
-      canReload: Boolean
+      canReload: Boolean,
+      outLock: Lock
   ): mill.api.Result[BspServerHandle] = {
 
     try {
@@ -33,7 +35,8 @@ object BspWorkerImpl {
           logStream = streams.err,
           canReload = canReload,
           debugMessages = Option(System.getenv("MILL_BSP_DEBUG")).contains("true"),
-          onShutdown = () => listening.cancel(true)
+          onShutdown = () => listening.cancel(true),
+          outLock = outLock
         ) with MillJvmBuildServer with MillJavaBuildServer with MillScalaBuildServer
 
       lazy val launcher = new Launcher.Builder[BuildClient]()

runner/bsp/worker/src/mill/bsp/worker/MillBuildServer.scala

@@ -6,12 +6,14 @@ import com.google.gson.JsonObject
 import mill.api.*
 import mill.api.internal.{JvmBuildTarget, ScalaBuildTarget, *}
 import mill.api.Segment.Label
-import mill.bsp.{Constants}
+import mill.bsp.Constants
 import mill.bsp.worker.Utils.{makeBuildTarget, outputPaths, sanitizeUri}
+import mill.client.lock.Lock
 import mill.server.Server
 
 import java.io.PrintStream
 import java.util.concurrent.CompletableFuture
+import java.util.concurrent.locks.ReentrantLock
 import scala.collection.mutable
 import scala.concurrent.Promise
 import scala.jdk.CollectionConverters.*
@@ -28,7 +30,8 @@ private class MillBuildServer(
     logStream: PrintStream,
     canReload: Boolean,
     debugMessages: Boolean,
-    onShutdown: () => Unit
+    onShutdown: () => Unit,
+    outLock: Lock
 )(implicit ec: scala.concurrent.ExecutionContext) extends BuildServer {
 
   import MillBuildServer._
@@ -724,7 +727,8 @@ private class MillBuildServer(
         case n: NamedTaskApi[_] => n.label
         case t => t.toString
       },
-      streams = logger0.streams
+      streams = logger0.streams,
+      outLock = outLock
     ) {
       evaluator.executeApi(
         goals,

runner/client/src/mill/client/ServerLauncher.java

@@ -9,7 +9,6 @@
 import java.nio.file.Path;
 import java.util.Map;
 import mill.client.lock.Locks;
-import mill.client.lock.TryLocked;
 import mill.constants.InputPumper;
 import mill.constants.ProxyStream;
 import mill.constants.ServerFiles;
@@ -23,10 +22,10 @@
  *
  * - Client:
  *   - Take clientLock
- *   - If processLock is not yet taken, it means server is not running, so spawn a server
+ *   - If serverLock is not yet taken, it means server is not running, so spawn a server
  *   - Wait for server socket to be available for connection
  * - Server:
- *   - Take processLock.
+ *   - Take serverLock.
  *     - If already taken, it means another server was running
  *       (e.g. spawned by a different client) so exit immediately
  * - Server: loop:
@@ -46,7 +45,6 @@ public static class Result {
     public Path serverDir;
   }
 
-  final int serverProcessesLimit = 5;
   final int serverInitWaitMillis = 10000;
 
   public abstract void initServer(Path serverDir, boolean b, Locks locks) throws Exception;
@@ -58,7 +56,7 @@ public static class Result {
   PrintStream stderr;
   Map<String, String> env;
   String[] args;
-  Locks[] memoryLocks;
+  Locks memoryLock;
   int forceFailureForTestingMillisDelay;
 
   public ServerLauncher(
@@ -67,7 +65,7 @@ public ServerLauncher(
       PrintStream stderr,
       Map<String, String> env,
       String[] args,
-      Locks[] memoryLocks,
+      Locks memoryLock,
       int forceFailureForTestingMillisDelay) {
     this.stdin = stdin;
     this.stdout = stdout;
@@ -78,56 +76,32 @@ public ServerLauncher(
     // For testing in memory, we need to pass in the locks separately, so that the
     // locks can be shared between the different instances of `ServerLauncher` the
     // same way file locks are shared between different Mill client/server processes
-    this.memoryLocks = memoryLocks;
+    this.memoryLock = memoryLock;
 
     this.forceFailureForTestingMillisDelay = forceFailureForTestingMillisDelay;
   }
 
-  public Result acquireLocksAndRun(Path serverDir0) throws Exception {
-
+  public Result acquireLocksAndRun(Path serverDir) throws Exception {
     final boolean setJnaNoSys = System.getProperty("jna.nosys") == null;
-    if (setJnaNoSys) {
-      System.setProperty("jna.nosys", "true");
-    }
+    if (setJnaNoSys) System.setProperty("jna.nosys", "true");
 
-    int serverIndex = 0;
-    while (serverIndex < serverProcessesLimit) { // Try each possible server process (-1 to -5)
-      serverIndex++;
-      final Path serverDir =
-          serverDir0.getParent().resolve(serverDir0.getFileName() + "-" + serverIndex);
-
-      Files.createDirectories(serverDir);
-
-      try (Locks locks = memoryLocks != null
-              ? memoryLocks[serverIndex - 1]
-              : Locks.files(serverDir.toString());
-          TryLocked clientLocked = locks.clientLock.tryLock()) {
-        if (clientLocked.isLocked()) {
-          Result result = new Result();
-          preRun(serverDir);
-          result.exitCode = run(serverDir, setJnaNoSys, locks);
-          result.serverDir = serverDir;
-          return result;
-        }
-      }
-    }
-    throw new ServerCouldNotBeStarted(
-        "Reached max server processes limit: " + serverProcessesLimit);
-  }
-
-  int run(Path serverDir, boolean setJnaNoSys, Locks locks) throws Exception {
+    Files.createDirectories(serverDir);
 
-    try (OutputStream f = Files.newOutputStream(serverDir.resolve(ServerFiles.runArgs))) {
-      f.write(Util.hasConsole() ? 1 : 0);
-      ClientUtil.writeString(f, BuildInfo.millVersion);
-      ClientUtil.writeArgs(args, f);
-      ClientUtil.writeMap(env, f);
-    }
+    Result result = new Result();
+    preRun(serverDir);
+    result.exitCode = run(serverDir, setJnaNoSys);
+    result.serverDir = serverDir;
+    return result;
+  }
 
-    if (locks.processLock.probe()) initServer(serverDir, setJnaNoSys, locks);
+  int run(Path serverDir, boolean setJnaNoSys) throws Exception {
 
-    while (locks.processLock.probe()) Thread.sleep(1);
+    try (Locks locks = memoryLock != null ? memoryLock : Locks.files(serverDir.toString());
+        mill.client.lock.Locked locked = locks.clientLock.lock()) {
 
+      if (locks.serverLock.probe()) initServer(serverDir, setJnaNoSys, locks);
+      while (locks.serverLock.probe()) Thread.sleep(1);
+    }
     long retryStart = System.currentTimeMillis();
     Socket ioSocket = null;
     Throwable socketThrowable = null;
@@ -147,6 +121,10 @@ int run(Path serverDir, boolean setJnaNoSys, Locks locks) throws Exception {
 
     InputStream outErr = ioSocket.getInputStream();
     OutputStream in = ioSocket.getOutputStream();
+    in.write(Util.hasConsole() ? 1 : 0);
+    ClientUtil.writeString(in, BuildInfo.millVersion);
+    ClientUtil.writeArgs(args, in);
+    ClientUtil.writeMap(env, in);
     ProxyStream.Pumper outPumper = new ProxyStream.Pumper(outErr, stdout, stderr);
     InputPumper inPump = new InputPumper(() -> stdin, () -> in, true);
     Thread outPumperThread = new Thread(outPumper, "outPump");
@@ -156,13 +134,13 @@ int run(Path serverDir, boolean setJnaNoSys, Locks locks) throws Exception {
     outPumperThread.start();
     inThread.start();
 
-    if (forceFailureForTestingMillisDelay > 0) {
-      Thread.sleep(forceFailureForTestingMillisDelay);
-      throw new Exception("Force failure for testing: " + serverDir);
-    }
-    outPumperThread.join();
-
     try {
+      if (forceFailureForTestingMillisDelay > 0) {
+        Thread.sleep(forceFailureForTestingMillisDelay);
+        throw new Exception("Force failure for testing: " + serverDir);
+      }
+      outPumperThread.join();
+
       Path exitCodeFile = serverDir.resolve(ServerFiles.exitCode);
       if (Files.exists(exitCodeFile)) {
         return Integer.parseInt(Files.readAllLines(exitCodeFile).get(0));

runner/client/src/mill/client/lock/DoubleLock.java

@@ -0,0 +1,51 @@
+package mill.client.lock;
+
+public class DoubleLock extends Lock {
+
+  private final Lock lock1;
+  private final Lock lock2;
+
+  public DoubleLock(Lock lock1, Lock lock2) throws Exception {
+    this.lock1 = lock1;
+    this.lock2 = lock2;
+  }
+
+  @Override
+  public Locked lock() throws Exception {
+    return new DoubleLocked(lock1.lock(), lock2.lock());
+  }
+
+  @Override
+  public TryLocked tryLock() throws Exception {
+    TryLocked l1 = lock1.tryLock();
+    TryLocked l2 = lock2.tryLock();
+    if (l1.isLocked() && l2.isLocked()) {
+      return new DoubleTryLocked(l1, l2);
+    } else {
+      l1.release();
+      l2.release();
+      return new DoubleTryLocked(null, null);
+    }
+  }
+
+  @Override
+  public boolean probe() throws Exception {
+    TryLocked tl = tryLock();
+    if (!tl.isLocked()) return true;
+    else {
+      tl.release();
+      return false;
+    }
+  }
+
+  @Override
+  public void close() throws Exception {
+    lock1.close();
+    lock2.close();
+  }
+
+  @Override
+  public void delete() throws Exception {
+    close();
+  }
+}

runner/client/src/mill/client/lock/DoubleLocked.java

@@ -0,0 +1,18 @@
+package mill.client.lock;
+
+class DoubleLocked implements Locked {
+
+  protected final Locked lock1;
+  protected final Locked lock2;
+
+  public DoubleLocked(Locked lock1, Locked lock2) {
+    this.lock1 = lock1;
+    this.lock2 = lock2;
+  }
+
+  @Override
+  public void release() throws Exception {
+    this.lock1.release();
+    this.lock2.release();
+  }
+}

runner/client/src/mill/client/lock/DoubleTryLocked.java

@@ -0,0 +1,18 @@
+package mill.client.lock;
+
+class DoubleTryLocked extends DoubleLocked implements TryLocked {
+
+  public DoubleTryLocked(TryLocked lock1, TryLocked lock2) {
+    super(lock1, lock2);
+  }
+
+  @Override
+  public boolean isLocked() {
+    return lock1 != null && lock2 != null;
+  }
+
+  @Override
+  public void release() throws Exception {
+    if (isLocked()) super.release();
+  }
+}