[2.x] fix: Fixes JDK 17 domain socket deadlocks (#9671)

**Problem** Channels.newInputStream/newOutputStream share a channel-wide lock, which would deadlock for a duplex communication. **Solution** This implements an alternative DuplexChannels functions that's capable of duplex communication expected of a "socket". This also duplicates the Java implementation to the worker app, so we can use JDK domain socket for forked test communication.
2026-09-04 16:54:29 +02:00 · 2026-08-25 02:02:37 -04:00
parent 108623f848
commit d4ada26c86
13 changed files with 528 additions and 29 deletions
@@ -7,6 +7,7 @@ import java.util.Locale
 import sbt.internal.inc.Analysis
 import sbt.Tags
 import com.eed3si9n.jarjarabrams.ModuleCoordinate
+import Utils.JDK17

 // ThisBuild settings take lower precedence,
 // but can be shared across the multi projects.
@@ -465,8 +466,10 @@ lazy val testingProj = (project in file("testing"))

 lazy val workerProj = (project in file("worker"))
  .dependsOn(exampleWorkProj % Test)
+  .configs(JDK17)
  .settings(
    name := "worker",
+    inConfig(JDK17)(Defaults.compileSettings),
    Test / classLoaderLayeringStrategy := ClassLoaderLayeringStrategy.Raw,
    testedBaseSettings,
    Compile / doc / javacOptions := Nil,
@@ -474,8 +477,23 @@ lazy val workerProj = (project in file("worker"))
    autoScalaLibrary := false,
    libraryDependencies ++= Seq(gson, testInterface),
    libraryDependencies += "org.scala-lang" %% "scala3-library" % scalaVersion.value % Test,
-    // run / fork := false,
    Test / fork := true,
+    Compile / javacOptions := Seq("--release", "8"),
+    JDK17 / javacOptions := Seq("--release", "17"),
+    Compile / packageBin / packageOptions += Pkg.ManifestAttributes("Multi-Release" -> "true"),
+    Compile / packageBin / mappings ++= {
+      val _ = (JDK17 / compile).value
+      val conv = fileConverter.value
+      val dir = (Utils.JDK17 / classDirectory).value
+      fileTreeView.value
+        .list(Glob(dir) / **)
+        .map(_._1)
+        .map(_.toFile())
+        .pair(Path.rebase(dir, "META-INF/versions/17"))
+        .map: (file, rel) =>
+          val vf: xsbti.HashedVirtualFileRef = conv.toVirtualFile(file.toPath())
+          vf -> rel
+    },
    mimaSettings,
    mimaBinaryIssueFilters ++= Vector(
    ),
@@ -590,6 +608,11 @@ lazy val actionsProj = (project in file("main-actions"))
    Test / classLoaderLayeringStrategy := ClassLoaderLayeringStrategy.Flat,
    mimaSettings,
    mimaBinaryIssueFilters ++= Vector(
+      // WorkerConnection gained an Ipc(path) case; mixing a parameterized case into
+      // the enum drops the synthetic values()/valueOf() Java-enum forwarders. This is
+      // an internal (sbt.internal) type not meant for external consumption.
+      exclude[DirectMissingMethodProblem]("sbt.internal.WorkerConnection.valueOf"),
+      exclude[DirectMissingMethodProblem]("sbt.internal.WorkerConnection.values"),
    ),
  )
  .dependsOn(lmCore)
@@ -28,7 +28,6 @@ import scala.util.Random
 import scala.util.control.NonFatal
 import scala.jdk.CollectionConverters.*
 import scala.sys.process.Process
-import sbt.internal.WorkerConnection

 /**
 * This implements forked testing, in cooperation with the worker CLI,
@@ -141,8 +140,7 @@ private[sbt] object ForkTests:
      )
      testListeners.foreach(_.doInit())
      val result =
-        val ct = WorkerConnection.Tcp
-        val w = WorkerExchange.startWorker(fork, if virtualClasspath then Nil else cpFiles, ct)
+        val w = WorkerExchange.startWorker(fork, if virtualClasspath then Nil else cpFiles)
        val wl = React(randomId, log, opts.testListeners, resultsAcc, w.process)
        try
          WorkerExchange.registerListener(wl)
@@ -152,7 +150,9 @@ private[sbt] object ForkTests:
          if wl.blockForResponse() != 0 then
            throw MessageOnlyException("Forked test harness failed")
          testOutputResult
-        finally WorkerExchange.unregisterListener(wl)
+        finally
+          w.close()
+          WorkerExchange.unregisterListener(wl)
      testListeners.foreach(_.doComplete(result.overall))
      result
    } // end task
@@ -11,21 +11,62 @@ package internal

 import org.scalasbt.shadedgson.com.google.gson.Gson
 import java.io.*
-import java.net.{ InetAddress, ServerSocket }
+import java.net.{ InetAddress, ServerSocket, StandardProtocolFamily, UnixDomainSocketAddress }
+import java.nio.channels.{ ServerSocketChannel, SocketChannel }
+import java.nio.file.{ Files, Path as NioPath }
 import java.util.Scanner
 import sbt.io.IO
 import sbt.internal.io.Retry
 import sbt.internal.worker1.*
+import sbt.protocol.DuplexChannels
 import sbt.testing.Framework
 import scala.sys.process.{ BasicIO, Process, ProcessIO }
 import scala.collection.mutable
+import scala.collection.concurrent.TrieMap
 import scala.collection.mutable.ListBuffer
 import scala.concurrent.{ Await, Promise }
 import scala.concurrent.duration.*
+import scala.util.control.NonFatal

 object WorkerExchange:
  val listeners: mutable.ListBuffer[WorkerResponseListener] = ListBuffer.empty
  private val loopback = InetAddress.getByName(null)
+  private val jdkIpcSupportCache = TrieMap.empty[Option[File], Boolean]
+
+  /**
+   * Start a worker process.
+   */
+  def startWorker(fo: ForkOptions, extraCp: Seq[File]): WorkerProxy =
+    val ct =
+      if supportsUnixDomainSockets(fo.javaHome) then WorkerConnection.Ipc(newIpcSocketPath())
+      else WorkerConnection.Stdio
+    startWorker(fo, extraCp, ct)
+
+  /**
+   * True if `javaHome` (None meaning the JDK currently running sbt) is JDK 16+.
+   */
+  private def supportsUnixDomainSockets(javaHome: Option[File]): Boolean =
+    def doDetect: Boolean =
+      javaHome match
+        case None       => true // the JDK running sbt itself, which requires 17+
+        case Some(home) =>
+          try
+            val releaseFile = File(home, "release")
+            val props = java.util.Properties()
+            val in = FileInputStream(releaseFile)
+            try props.load(in)
+            finally in.close()
+            val raw = Option(props.getProperty("JAVA_VERSION")).getOrElse("")
+            val version = raw.stripPrefix("\"").stripSuffix("\"")
+            val digits =
+              version.takeWhile(c => c.isDigit || c == '.').split('.').flatMap(_.toIntOption)
+            val major = digits match
+              case Array(1, minor, _*) => minor // legacy 1.8-style versioning
+              case Array(m, _*)        => m
+              case _                   => 0
+            major >= 16
+          catch case NonFatal(_) => false
+    jdkIpcSupportCache.getOrElseUpdate(javaHome, doDetect)

  /**
   * Start a worker process.
@@ -42,28 +83,52 @@ object WorkerExchange:
      IO.classLocationPath(classOf[Gson]).toFile,
    )
    val inputRef = Promise[OutputStream]()
-    val socketOpt = connectionType match
+    def runAccepter(out: OutputStream, in: InputStream): Unit =
+      inputRef.success(out)
+      val scanner = Scanner(in, "UTF-8")
+      while scanner.hasNextLine() do notifyListeners(scanner.nextLine())
+    val (connArgs, closer): (Seq[String], Option[AutoCloseable]) = connectionType match
      case WorkerConnection.Tcp =>
        val serverSocket = Retry(ServerSocket(0, 1, loopback))
        val accepter = Thread(() => {
          val socket = serverSocket.accept()
-          inputRef.success(socket.getOutputStream())
-          val scanner = Scanner(socket.getInputStream(), "UTF-8")
-          while scanner.hasNextLine() do notifyListeners(scanner.nextLine())
+          runAccepter(socket.getOutputStream(), socket.getInputStream())
        })
        accepter.setName("sbt-fork-test-response-reader")
        accepter.setPriority(Thread.NORM_PRIORITY + 1)
        accepter.start()
-        Some(serverSocket)
-      case _ => None
+        (Seq("--tcp", serverSocket.getLocalPort().toString()), Some(serverSocket))
+      case WorkerConnection.Ipc(path) =>
+        val serverChannel = Retry {
+          Files.deleteIfExists(path)
+          val ch = ServerSocketChannel.open(StandardProtocolFamily.UNIX)
+          ch.bind(UnixDomainSocketAddress.of(path))
+          ch
+        }
+        @volatile var acceptedChannel: SocketChannel = null
+        val accepter = Thread(() => {
+          val channel = serverChannel.accept()
+          acceptedChannel = channel
+          runAccepter(
+            DuplexChannels.newOutputStream(channel),
+            DuplexChannels.newInputStream(channel)
+          )
+        })
+        accepter.setName("sbt-fork-test-response-reader")
+        accepter.setPriority(Thread.NORM_PRIORITY + 1)
+        accepter.start()
+        val closer: AutoCloseable = () => {
+          if acceptedChannel != null then acceptedChannel.close()
+          serverChannel.close()
+          Files.deleteIfExists(path)
+        }
+        (Seq("--ipc", path.toString()), Some(closer))
+      case WorkerConnection.Stdio => (Nil, None)
    val options = Seq(
      "-classpath",
      fullCp.mkString(File.pathSeparator),
      classOf[WorkerMain].getCanonicalName,
-    ) ++
-      (socketOpt match
-        case Some(s) => Seq("--tcp", s.getLocalPort().toString())
-        case _       => Nil)
+    ) ++ connArgs
    val onStdoutLine: String => Unit = connectionType match
      case WorkerConnection.Stdio => notifyListeners
      case _                      => (line) => scala.Console.out.println(line)
@@ -80,7 +145,17 @@ object WorkerExchange:
    val p = Fork.java.fork(forkWithIo, options)
    val forkTimeout = fo.connectionTimeout.getOrElse(30.seconds)
    val input = Await.result(inputRef.future, forkTimeout)
-    WorkerProxy(input, p, options, socketOpt)
+    WorkerProxy(input, p, options, closer)
+
+  /** Generates a fresh path suitable for binding a `WorkerConnection.Ipc` socket. */
+  def newIpcSocketPath(): NioPath =
+    val dir = NioPath
+      .of(sys.env.getOrElse("XDG_RUNTIME_DIR", sys.props("java.io.tmpdir")))
+      .resolve(".sbt-fork-ipc")
+    Files.createDirectories(dir)
+    val path = Files.createTempFile(dir, "fork-", ".sock")
+    Files.deleteIfExists(path)
+    path

  def registerListener(listener: WorkerResponseListener): Unit =
    synchronized:
@@ -104,12 +179,12 @@ class WorkerProxy(
    input: OutputStream,
    val process: Process,
    val options: Seq[String],
-    serverSocket: Option[ServerSocket],
+    closer: Option[AutoCloseable],
 ) extends AutoCloseable:
  lazy val inputStream = PrintStream(input)
  def close(): Unit =
    input.close()
-    serverSocket.foreach(_.close())
+    closer.foreach(_.close())
  def blockForExitCode(): Int =
    if !process.isAlive() then process.exitValue()
    else Fork.blockForExitCode(process)
@@ -132,3 +207,4 @@ abstract class WorkerResponseListener extends Function1[String, Unit]:
 enum WorkerConnection:
  case Stdio
  case Tcp
+  case Ipc(path: NioPath)
@@ -9,13 +9,18 @@ import scala.sys.process.Process

 object WorkerExchangeTest extends Properties:
  given Gen[WorkerConnection] =
-    Gen.choice1(Gen.constant(WorkerConnection.Stdio), Gen.constant(WorkerConnection.Tcp))
+    Gen.choice1(
+      Gen.constant(WorkerConnection.Stdio),
+      Gen.constant(WorkerConnection.Tcp),
+      Gen.constant(WorkerConnection.Ipc(WorkerExchange.newIpcSocketPath())),
+    )

  def gen[A1: Gen]: Gen[A1] = summon[Gen[A1]]

  override lazy val tests: List[Test] = List(
    propertyN("non-jsonrpc should return exit code 1", propBadInput, 10),
    propertyN("bye should return response json with a result", propBye, 10),
+    example("startWorker(fo, extraCp) auto-detects a working connection type", exampleAutoDetect),
  )

  def propertyN(name: String, result: => Property, n: Int): Test =
@@ -47,6 +52,17 @@ object WorkerExchangeTest extends Properties:
        .and(Result.assert(l.sb.toString() == s"""{ "jsonrpc": "2.0", "result": 0, "id": $i }"""))
        .log(s"\"${l.sb.toString()}\"")

+  def exampleAutoDetect: Result =
+    val w = WorkerExchange.startWorker(ForkOptions(), Nil)
+    withListener: l =>
+      w.println("""{"jsonrpc": "2.0", "method": "bye", "params": {}, "id": 1}""")
+      val exitCode = w.blockForExitCode()
+      l.awaitResponse()
+      Result
+        .assert(exitCode == 0)
+        .and(Result.assert(l.sb.toString() == """{ "jsonrpc": "2.0", "result": 0, "id": 1 }"""))
+        .log(s"\"${l.sb.toString()}\"")
+
  def withListener[A1](f: ConcreteListener => A1) =
    val l = ConcreteListener()
    try
@@ -8,6 +8,8 @@ import PublishBinPlugin.autoImport.publishLocalBin
 import sbt.internal.inc.Analysis

 object Utils {
+  val JDK17 = config("jdk17")
+
  val ExclusiveTest: Tags.Tag = Tags.Tag("exclusive-test")

  val componentID: SettingKey[Option[String]] = settingKey[Option[String]]("")
@@ -0,0 +1,91 @@
+/*
+ * sbt
+ * Copyright 2023, Scala center
+ * Copyright 2011 - 2022, Lightbend, Inc.
+ * Copyright 2008 - 2010, Mark Harrah
+ * Licensed under Apache License 2.0 (see LICENSE)
+ */
+
+package sbt.protocol;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.net.Socket;
+import java.nio.ByteBuffer;
+import java.nio.channels.SocketChannel;
+
+/**
+ * java.nio.channels.Channels.newInputStream/newOutputStream both synchronize on the channel's
+ * blockingLock() for the duration of each blocking call, so a thread parked in a blocking read
+ * holds that lock for as long as the read blocks, and a concurrent writer on the same channel can
+ * never acquire it. These factories talk to the channel directly instead, so a SocketChannel can
+ * safely be read and written from different threads at the same time.
+ */
+public final class DuplexChannels {
+  private DuplexChannels() {}
+
+  public static OutputStream newOutputStream(SocketChannel ch) {
+    return new OutputStream() {
+      @Override
+      public void write(int b) throws IOException {
+        ByteBuffer bb = ByteBuffer.wrap(new byte[] {(byte) b});
+        while (bb.hasRemaining()) ch.write(bb);
+      }
+
+      @Override
+      public void write(byte[] b, int off, int len) throws IOException {
+        ByteBuffer bb = ByteBuffer.wrap(b, off, len);
+        while (bb.hasRemaining()) ch.write(bb);
+      }
+    };
+  }
+
+  public static InputStream newInputStream(SocketChannel ch) {
+    return new InputStream() {
+      @Override
+      public int read() throws IOException {
+        ByteBuffer bb = ByteBuffer.allocate(1);
+        int n = ch.read(bb);
+        return n <= 0 ? -1 : (bb.get(0) & 0xff);
+      }
+
+      @Override
+      public int read(byte[] b, int off, int len) throws IOException {
+        if (len == 0) return 0;
+        return ch.read(ByteBuffer.wrap(b, off, len));
+      }
+    };
+  }
+
+  /**
+   * Wraps a connected SocketChannel as a Socket backed by {@link #newInputStream}/{@link
+   * #newOutputStream}.
+   */
+  public static Socket newSocket(SocketChannel ch) {
+    return new Socket() {
+      private final InputStream in = newInputStream(ch);
+      private final OutputStream out = newOutputStream(ch);
+
+      @Override
+      public InputStream getInputStream() {
+        return in;
+      }
+
+      @Override
+      public OutputStream getOutputStream() {
+        return out;
+      }
+
+      @Override
+      public void close() throws IOException {
+        ch.close();
+      }
+
+      @Override
+      public boolean isClosed() {
+        return !ch.isOpen();
+      }
+    };
+  }
+}
@@ -9,9 +9,9 @@
 package sbt
 package protocol

-import java.io.{ File, InputStream, OutputStream }
+import java.io.File
 import java.net.{ InetAddress, Socket, StandardProtocolFamily, URI, UnixDomainSocketAddress }
-import java.nio.channels.{ Channels, SocketChannel }
+import java.nio.channels.SocketChannel
 import scala.util.control.NonFatal
 import sjsonnew.BasicJsonProtocol
 import sjsonnew.support.scalajson.unsafe.{ Parser, Converter }
@@ -59,11 +59,5 @@ object ClientSocket {
  def bootSocket(path: String): Socket =
    val ch = SocketChannel.open(StandardProtocolFamily.UNIX)
    ch.connect(UnixDomainSocketAddress.of(path))
-    new Socket:
-      private val in = Channels.newInputStream(ch)
-      private val out = Channels.newOutputStream(ch)
-      override def getInputStream: InputStream = in
-      override def getOutputStream: OutputStream = out
-      override def close(): Unit = ch.close()
-      override def isClosed: Boolean = !ch.isOpen
+    DuplexChannels.newSocket(ch)
 }
@@ -0,0 +1,130 @@
+/*
+ * sbt
+ * Copyright 2023, Scala center
+ * Copyright 2011 - 2022, Lightbend, Inc.
+ * Copyright 2008 - 2010, Mark Harrah
+ * Licensed under Apache License 2.0 (see LICENSE)
+ */
+
+package sbt.protocol
+
+import hedgehog.{ Gen, Property, Result }
+import hedgehog.core.{ ShrinkLimit, SuccessCount }
+import hedgehog.runner.*
+import java.io.{ EOFException, InputStream }
+import java.net.{ StandardProtocolFamily, UnixDomainSocketAddress }
+import java.nio.ByteBuffer
+import java.nio.channels.{ ServerSocketChannel, SocketChannel }
+import java.util.concurrent.{ CountDownLatch, LinkedBlockingQueue, TimeUnit }
+import scala.util.Using
+import scala.util.control.NonFatal
+import sbt.io.IO
+
+/**
+ * Regression test: bootSocket used to wrap its channel with
+ * Channels.newInputStream/newOutputStream, which share the channel's blockingLock() and deadlock
+ * a blocking read against a concurrent write on the same channel. It now uses DuplexChannels,
+ * which talks to the channel directly and has no such shared lock. What matters for reproducing
+ * the deadlock is timing, not payload content, so the reader and writer threads are each given an
+ * independent startup delay drawn from {0, 100, 300}ms to exercise the read starting well before,
+ * around the same time as, and well after the write.
+ */
+object ClientSocketDuplexTest extends Properties:
+  override def tests: List[Test] = List(
+    propertyN(
+      "bootSocket: a concurrent read and write on the same channel do not deadlock",
+      propDuplex,
+      20,
+    ),
+  )
+
+  def propertyN(name: String, result: => Property, n: Int): Test =
+    Test(name, result)
+      .config(_.copy(testLimit = SuccessCount(n), shrinkLimit = ShrinkLimit(n * 10)))
+
+  private val toServer: Array[Byte] = Array[Byte](1)
+  private val toClient: Array[Byte] = Array[Byte](2)
+
+  val sleepMsGen: Gen[Int] = Gen.element1(0, 100, 300)
+
+  def propDuplex: Property =
+    for
+      readerSleepMs <- sleepMsGen.log("reader startup delay (ms)")
+      writerSleepMs <- sleepMsGen.log("writer startup delay (ms)")
+    yield runDuplexRound(readerSleepMs, writerSleepMs)
+
+  private def runDuplexRound(readerSleepMs: Int, writerSleepMs: Int): Result =
+    IO.withTemporaryDirectory: dir =>
+      val path = dir.toPath.resolve("boot.sock")
+      Using.resource(ServerSocketChannel.open(StandardProtocolFamily.UNIX)): serverChannel =>
+        serverChannel.bind(UnixDomainSocketAddress.of(path))
+        Using.resource(ClientSocket.bootSocket(path.toString)): client =>
+          Using.resource(serverChannel.accept()): serverSide =>
+            // The client's reader is parked waiting for toClient before the server has sent
+            // anything, so it's mid-read (and would be holding blockingLock() under the old
+            // Channels-based implementation) while we race the write below against it.
+            val readOutcome = new LinkedBlockingQueue[Either[Throwable, Array[Byte]]]()
+            val reader = new Thread(() =>
+              readOutcome.put(
+                try
+                  Thread.sleep(readerSleepMs.toLong)
+                  Right(readNBytes(client.getInputStream(), toClient.length))
+                catch case NonFatal(e) => Left(e)
+              )
+            )
+            reader.setDaemon(true)
+            reader.start()
+
+            val writeDone = new CountDownLatch(1)
+            val writer = new Thread(() =>
+              try
+                Thread.sleep(writerSleepMs.toLong)
+                client.getOutputStream().write(toServer)
+              catch case NonFatal(_) => ()
+              finally writeDone.countDown()
+            )
+            writer.setDaemon(true)
+            writer.start()
+
+            if !writeDone.await(3, TimeUnit.SECONDS) then
+              Result.failure.log(
+                "write blocked behind the concurrent read: possible regression of the " +
+                  "Channels.newInputStream/newOutputStream blockingLock() deadlock"
+              )
+            else
+              val fromClient = readNBytes(serverSide, toServer.length)
+              serverSide.write(ByteBuffer.wrap(toClient))
+              readOutcome.poll(3, TimeUnit.SECONDS) match
+                case null =>
+                  Result.failure.log(
+                    "client's blocked read never completed after the server replied"
+                  )
+                case Left(e)           => Result.failure.log(s"client read failed: $e")
+                case Right(fromServer) =>
+                  Result.all(
+                    List(
+                      Result
+                        .assert(fromClient.sameElements(toServer))
+                        .log("server received a different payload than the client sent"),
+                      Result
+                        .assert(fromServer.sameElements(toClient))
+                        .log("client received a different payload than the server sent"),
+                    )
+                  )
+
+  private def readNBytes(in: InputStream, n: Int): Array[Byte] =
+    val buf = new Array[Byte](n)
+    var total = 0
+    while total < n do
+      val r = in.read(buf, total, n - total)
+      if r < 0 then throw new EOFException(s"expected $n bytes, got $total")
+      total += r
+    buf
+
+  private def readNBytes(ch: SocketChannel, n: Int): Array[Byte] =
+    val bb = ByteBuffer.allocate(n)
+    while bb.hasRemaining() do
+      val r = ch.read(bb)
+      if r < 0 then throw new EOFException(s"expected $n bytes, got ${bb.position()}")
+    bb.array()
+end ClientSocketDuplexTest
@@ -0,0 +1,23 @@
+/*
+ * sbt
+ * Copyright 2023, Scala center
+ * Copyright 2011 - 2022, Lightbend, Inc.
+ * Copyright 2008 - 2010, Mark Harrah
+ * Licensed under Apache License 2.0 (see LICENSE)
+ */
+
+package sbt.internal.worker1;
+
+import java.io.IOException;
+import java.net.StandardProtocolFamily;
+import java.net.UnixDomainSocketAddress;
+import java.nio.channels.SocketChannel;
+import java.nio.file.Path;
+
+public class JdkCompat {
+  public static SocketChannel connectUnixSocket(Path socketPath) throws IOException {
+    SocketChannel client = SocketChannel.open(StandardProtocolFamily.UNIX);
+    client.connect(UnixDomainSocketAddress.of(socketPath));
+    return client;
+  }
+}
@@ -0,0 +1,91 @@
+/*
+ * sbt
+ * Copyright 2023, Scala center
+ * Copyright 2011 - 2022, Lightbend, Inc.
+ * Copyright 2008 - 2010, Mark Harrah
+ * Licensed under Apache License 2.0 (see LICENSE)
+ */
+
+package sbt.internal.worker1;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.net.Socket;
+import java.nio.ByteBuffer;
+import java.nio.channels.SocketChannel;
+
+/**
+ * java.nio.channels.Channels.newInputStream/newOutputStream both synchronize on the channel's
+ * blockingLock() for the duration of each blocking call, so a thread parked in a blocking read
+ * holds that lock for as long as the read blocks, and a concurrent writer on the same channel can
+ * never acquire it. These factories talk to the channel directly instead, so a SocketChannel can
+ * safely be read and written from different threads at the same time.
+ */
+public final class DuplexChannels {
+  private DuplexChannels() {}
+
+  public static OutputStream newOutputStream(SocketChannel ch) {
+    return new OutputStream() {
+      @Override
+      public void write(int b) throws IOException {
+        ByteBuffer bb = ByteBuffer.wrap(new byte[] {(byte) b});
+        while (bb.hasRemaining()) ch.write(bb);
+      }
+
+      @Override
+      public void write(byte[] b, int off, int len) throws IOException {
+        ByteBuffer bb = ByteBuffer.wrap(b, off, len);
+        while (bb.hasRemaining()) ch.write(bb);
+      }
+    };
+  }
+
+  public static InputStream newInputStream(SocketChannel ch) {
+    return new InputStream() {
+      @Override
+      public int read() throws IOException {
+        ByteBuffer bb = ByteBuffer.allocate(1);
+        int n = ch.read(bb);
+        return n <= 0 ? -1 : (bb.get(0) & 0xff);
+      }
+
+      @Override
+      public int read(byte[] b, int off, int len) throws IOException {
+        if (len == 0) return 0;
+        return ch.read(ByteBuffer.wrap(b, off, len));
+      }
+    };
+  }
+
+  /**
+   * Wraps a connected SocketChannel as a Socket backed by {@link #newInputStream}/{@link
+   * #newOutputStream}.
+   */
+  public static Socket newSocket(SocketChannel ch) {
+    return new Socket() {
+      private final InputStream in = newInputStream(ch);
+      private final OutputStream out = newOutputStream(ch);
+
+      @Override
+      public InputStream getInputStream() {
+        return in;
+      }
+
+      @Override
+      public OutputStream getOutputStream() {
+        return out;
+      }
+
+      @Override
+      public void close() throws IOException {
+        ch.close();
+      }
+
+      @Override
+      public boolean isClosed() {
+        return !ch.isOpen();
+      }
+    };
+  }
+}
@@ -0,0 +1,25 @@
+/*
+ * sbt
+ * Copyright 2023, Scala center
+ * Copyright 2011 - 2022, Lightbend, Inc.
+ * Copyright 2008 - 2010, Mark Harrah
+ * Licensed under Apache License 2.0 (see LICENSE)
+ */
+
+package sbt.internal.worker1;
+
+import java.io.IOException;
+import java.nio.channels.SocketChannel;
+import java.nio.file.Path;
+
+/**
+ * Base (Java 8) fallback. The Multi-Release variant under src/jdk17/java implements this using JDK
+ * 16+ Unix domain socket APIs (StandardProtocolFamily.UNIX, UnixDomainSocketAddress); it's the one
+ * actually loaded when the worker runs on Java 17+.
+ */
+public class JdkCompat {
+  public static SocketChannel connectUnixSocket(Path socketPath) throws IOException {
+    throw new UnsupportedOperationException(
+        "Unix domain sockets require Java 16+; this worker JVM is running on an older version");
+  }
+}
@@ -17,6 +17,7 @@ import java.net.MalformedURLException;
 import java.net.Socket;
 import java.net.URL;
 import java.net.URLClassLoader;
+import java.nio.channels.SocketChannel;
 import java.nio.file.Files;
 import java.nio.file.Path;
 import java.nio.file.Paths;
@@ -76,6 +77,10 @@ public final class WorkerMain {
        int serverPort = Integer.parseInt(args[1]);
        app.socketWork(serverPort);
        System.exit(0);
+      } else if (args.length == 2 && args[0].equals("--ipc")) {
+        WorkerMain app = new WorkerMain();
+        app.ipcWork(Paths.get(args[1]));
+        System.exit(0);
      } else {
        System.err.println("missing args");
        System.exit(1);
@@ -121,6 +126,16 @@ public final class WorkerMain {
    }
  }

+  void ipcWork(Path socketPath) throws Exception {
+    SocketChannel client = JdkCompat.connectUnixSocket(socketPath);
+    this.jsonOut = new PrintStream(DuplexChannels.newOutputStream(client), true, "UTF-8");
+    this.inScanner = new Scanner(DuplexChannels.newInputStream(client), "UTF-8");
+    if (this.inScanner.hasNextLine()) {
+      String line = this.inScanner.nextLine();
+      process(line);
+    }
+  }
+
  /** This processes single request of supposed JSON line. */
  void process(String json) throws Exception {
    JsonElement elem = JsonParser.parseString(json);
@@ -1,10 +1,23 @@
 package sbt.internal.worker1

+import java.net.{ StandardProtocolFamily, UnixDomainSocketAddress }
+import java.nio.channels.ServerSocketChannel
+import scala.util.Using
 import sbt.io.IO

 object WorkerTest extends verify.BasicTestSuite:
  val main = WorkerMain()

+  test("JDK UNIX domain socket connects via Multi-Release JAR"):
+    IO.withTemporaryDirectory: dir =>
+      val path = dir.toPath.resolve("test.sock")
+      Using.resource(ServerSocketChannel.open(StandardProtocolFamily.UNIX)): server =>
+        server.bind(UnixDomainSocketAddress.of(path))
+        Using.resource(JdkCompat.connectUnixSocket(path)): client =>
+          Using.resource(server.accept()): accepted =>
+            assert(client.isConnected)
+            assert(accepted != null)
+
  test("process") {
    val u0 = IO.classLocationPath(classOf[example.Hello]).toUri()
    val u1 = IO.classLocationPath(classOf[scala.quoted.Quotes]).toUri()