diff --git a/integration/feature/output-directory/resources/build.mill b/integration/feature/output-directory/resources/build.mill
index b5fae4e0f89..e068bbd11cb 100644
--- a/integration/feature/output-directory/resources/build.mill
+++ b/integration/feature/output-directory/resources/build.mill
@@ -5,4 +5,16 @@ import mill.scalalib._
 
 object `package` extends RootModule with ScalaModule {
   def scalaVersion = scala.util.Properties.versionNumberString
+
+  def hello = Task {
+    "Hello from hello task"
+  }
+
+  def blockWhileExists(path: os.Path) = Task.Command[String] {
+    if (!os.exists(path))
+      os.write(path, Array.emptyByteArray)
+    while (os.exists(path))
+      Thread.sleep(100L)
+    "Blocking command done"
+  }
 }
diff --git a/integration/feature/output-directory/src/OutputDirectoryLockTests.scala b/integration/feature/output-directory/src/OutputDirectoryLockTests.scala
new file mode 100644
index 00000000000..25b5a15d078
--- /dev/null
+++ b/integration/feature/output-directory/src/OutputDirectoryLockTests.scala
@@ -0,0 +1,92 @@
+package mill.integration
+
+import mill.testkit.UtestIntegrationTestSuite
+import utest._
+
+import java.io.ByteArrayOutputStream
+import java.util.concurrent.{CountDownLatch, Executors}
+
+import scala.concurrent.duration.Duration
+import scala.concurrent.{Await, ExecutionContext, Future}
+
+object OutputDirectoryLockTests extends UtestIntegrationTestSuite {
+
+  private val pool = Executors.newCachedThreadPool()
+  private val ec = ExecutionContext.fromExecutorService(pool)
+
+  override def utestAfterAll(): Unit = {
+    pool.shutdown()
+  }
+
+  def tests: Tests = Tests {
+    test("basic") - integrationTest { tester =>
+      import tester._
+      val signalFile = workspacePath / "do-wait"
+      System.err.println("Spawning blocking task")
+      val blocksFuture =
+        Future(eval(("show", "blockWhileExists", "--path", signalFile), check = true))(ec)
+      while (!os.exists(signalFile) && !blocksFuture.isCompleted)
+        Thread.sleep(100L)
+      if (os.exists(signalFile))
+        System.err.println("Blocking task is running")
+      else {
+        System.err.println("Failed to run blocking task")
+        Predef.assert(blocksFuture.isCompleted)
+        blocksFuture.value.get.get
+      }
+
+      val testCommand: os.Shellable = ("show", "hello")
+      val testMessage = "Hello from hello task"
+
+      System.err.println("Evaluating task without lock")
+      val noLockRes = eval(("--no-build-lock", testCommand), check = true)
+      assert(noLockRes.out.contains(testMessage))
+
+      System.err.println("Evaluating task without waiting for lock (should fail)")
+      val noWaitRes = eval(("--no-wait-for-build-lock", testCommand))
+      assert(noWaitRes.err.contains("Cannot proceed, another Mill process is running tasks"))
+
+      System.err.println("Evaluating task waiting for the lock")
+
+      val lock = new CountDownLatch(1)
+      val stderr = new ByteArrayOutputStream
+      var success = false
+      val futureWaitingRes = Future {
+        eval(
+          testCommand,
+          stderr = os.ProcessOutput {
+            val expectedMessage =
+              "Another Mill process is running tasks, waiting for it to be done..."
+
+            (bytes, len) =>
+              stderr.write(bytes, 0, len)
+              val output = new String(stderr.toByteArray)
+              if (output.contains(expectedMessage))
+                lock.countDown()
+          },
+          check = true
+        )
+      }(ec)
+      try {
+        lock.await()
+        success = true
+      } finally {
+        if (!success) {
+          System.err.println("Waiting task output:")
+          System.err.write(stderr.toByteArray)
+        }
+      }
+
+      System.err.println("Task is waiting for the lock, unblocking it")
+      os.remove(signalFile)
+
+      System.err.println("Blocking task should exit")
+      val blockingRes = Await.result(blocksFuture, Duration.Inf)
+      assert(blockingRes.out.contains("Blocking command done"))
+
+      System.err.println("Waiting task should be free to proceed")
+      val waitingRes = Await.result(futureWaitingRes, Duration.Inf)
+      assert(waitingRes.out.contains(testMessage))
+    }
+  }
+}
diff --git a/main/api/src/mill/api/Logger.scala b/main/api/src/mill/api/Logger.scala
index 4b93616d14f..add24defd80 100644
--- a/main/api/src/mill/api/Logger.scala
+++ b/main/api/src/mill/api/Logger.scala
@@ -2,6 +2,8 @@ package mill.api
 
 import java.io.{InputStream, PrintStream}
 
+import mill.main.client.lock.{Lock, Locked}
+
 /**
  * The standard logging interface of the Mill build tool.
  *
@@ -24,7 +26,7 @@ import java.io.{InputStream, PrintStream}
  * but when `show` is used both are forwarded to stderr and stdout is only
  * used to display the final `show` output for easy piping.
  */
-trait Logger {
+trait Logger extends AutoCloseable {
   def colored: Boolean
 
   def systemStreams: SystemStreams
@@ -79,4 +81,17 @@ trait Logger {
     try t
     finally removePromptLine()
   }
+
+  def waitForLock(lock: Lock, waitingAllowed: Boolean): Locked = {
+    val tryLocked = lock.tryLock()
+    if (tryLocked.isLocked())
+      tryLocked
+    else if (waitingAllowed) {
+      info("Another Mill process is running tasks, waiting for it to be done...")
+      lock.lock()
+    } else {
+      error("Cannot proceed, another Mill process is running tasks")
+      throw new Exception("Cannot acquire lock on Mill output directory")
+    }
+  }
 }
diff --git a/main/client/src/mill/main/client/OutFiles.java b/main/client/src/mill/main/client/OutFiles.java
index 04ffeecb4db..0af23e233a1 100644
--- a/main/client/src/mill/main/client/OutFiles.java
+++ b/main/client/src/mill/main/client/OutFiles.java
@@ -57,5 +57,9 @@ public class OutFiles {
      */
     final public static String millNoServer = "mill-no-server";
 
+    /**
+     * Lock file used for exclusive access to the Mill output directory
+     */
+    final public static String millLock = "mill-lock";
 
 }
diff --git a/main/client/src/mill/main/client/lock/DummyLock.java b/main/client/src/mill/main/client/lock/DummyLock.java
new file mode 100644
index 00000000000..ede8224323d
--- /dev/null
+++ b/main/client/src/mill/main/client/lock/DummyLock.java
@@ -0,0 +1,22 @@
+package mill.main.client.lock;
+
+import java.util.concurrent.locks.ReentrantLock;
+
+class DummyLock extends Lock {
+
+    public boolean probe() {
+        return true;
+    }
+
+    public Locked lock() {
+        return new DummyTryLocked();
+    }
+
+    public TryLocked tryLock() {
+        return new DummyTryLocked();
+    }
+
+    @Override
+    public void close() throws Exception {
+    }
+}
diff --git a/main/client/src/mill/main/client/lock/DummyTryLocked.java b/main/client/src/mill/main/client/lock/DummyTryLocked.java
new file mode 100644
index 00000000000..34ad7b5ea15
--- /dev/null
+++ b/main/client/src/mill/main/client/lock/DummyTryLocked.java
@@ -0,0 +1,11 @@
+package mill.main.client.lock;
+
+class DummyTryLocked implements TryLocked {
+    public DummyTryLocked() {
+    }
+
+    public boolean isLocked(){ return true; }
+
+    public void release() throws Exception {
+    }
+}
diff --git a/main/client/src/mill/main/client/lock/Lock.java b/main/client/src/mill/main/client/lock/Lock.java
index 6d729c0ebd6..3870bc07a14 100644
--- a/main/client/src/mill/main/client/lock/Lock.java
+++ b/main/client/src/mill/main/client/lock/Lock.java
@@ -15,4 +15,17 @@ public void await() throws Exception {
      */
     public abstract boolean probe() throws Exception;
     public void delete() throws Exception {}
+
+    public static Lock file(String path) throws Exception {
+        return new FileLock(path);
+    }
+
+    public static Lock memory() {
+        return new MemoryLock();
+    }
+
+    public static Lock dummy() {
+        return new DummyLock();
+    }
+
 }
diff --git a/runner/src/mill/runner/MillCliConfig.scala b/runner/src/mill/runner/MillCliConfig.scala
index 61d1ef9fac4..429a394f18a 100644
--- a/runner/src/mill/runner/MillCliConfig.scala
+++ b/runner/src/mill/runner/MillCliConfig.scala
@@ -130,7 +130,19 @@ case class MillCliConfig(
         status at the command line and falls back to the legacy ticker
       """
     )
-    disablePrompt: Flag = Flag()
+    disablePrompt: Flag = Flag(),
+    @arg(
+      hidden = true,
+      doc =
+        """Evaluate tasks / commands without acquiring an exclusive lock on the Mill output directory"""
+    )
+    noBuildLock: Flag = Flag(),
+    @arg(
+      hidden = true,
+      doc =
+        """Do not wait for an exclusive lock on the Mill output directory to evaluate tasks / commands. Fail if waiting for a lock is needed."""
+    )
+    noWaitForBuildLock: Flag = Flag()
 )
 
 import mainargs.ParserForClass
diff --git a/runner/src/mill/runner/MillMain.scala b/runner/src/mill/runner/MillMain.scala
index 3e1550ff026..10f343d1518 100644
--- a/runner/src/mill/runner/MillMain.scala
+++ b/runner/src/mill/runner/MillMain.scala
@@ -9,10 +9,12 @@ import mill.api.{MillException, SystemStreams, WorkspaceRoot, internal}
 import mill.bsp.{BspContext, BspServerResult}
 import mill.main.BuildInfo
 import mill.main.client.{OutFiles, ServerFiles}
+import mill.main.client.lock.Lock
 import mill.util.{PromptLogger, PrintLogger, Colors}
 
 import java.lang.reflect.InvocationTargetException
 import scala.util.control.NonFatal
+import scala.util.Using
 
 @internal
 object MillMain {
@@ -209,6 +211,10 @@ object MillMain {
                     .map(_ => Seq(bspCmd))
                     .getOrElse(config.leftoverArgs.value.toList)
 
+                val out = os.Path(OutFiles.out, WorkspaceRoot.workspaceRoot)
+                val outLock =
+                  if (config.noBuildLock.value || bspContext.isDefined) Lock.dummy()
+                  else Lock.file((out / OutFiles.millLock).toString)
                 var repeatForBsp = true
                 var loopRes: (Boolean, RunnerState) = (false, RunnerState.empty)
                 while (repeatForBsp) {
@@ -235,9 +241,16 @@ object MillMain {
                         colored = colored,
                         colors = colors
                       )
-                      try new MillBuildBootstrap(
+                      Using.resources(
+                        logger,
+                        logger.waitForLock(
+                          outLock,
+                          waitingAllowed = !config.noWaitForBuildLock.value
+                        )
+                      ) { (_, _) =>
+                        new MillBuildBootstrap(
                           projectRoot = WorkspaceRoot.workspaceRoot,
-                          output = os.Path(OutFiles.out, WorkspaceRoot.workspaceRoot),
+                          output = out,
                           home = config.home,
                           keepGoing = config.keepGoing.value,
                           imports = config.imports,
@@ -252,8 +265,6 @@ object MillMain {
                           config.allowPositional.value,
                           systemExit = systemExit
                         ).evaluate()
-                      finally {
-                        logger.close()
                       }
                     },
                     colors = colors