From d4ada26c865df19a99ddbc7e697d476c041e61d1 Mon Sep 17 00:00:00 2001 From: eugene yokota Date: Tue, 25 Aug 2026 02:02:37 -0400 Subject: [PATCH] [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. --- build.sbt | 25 +++- .../src/main/scala/sbt/ForkTests.scala | 8 +- .../scala/sbt/internal/WorkerExchange.scala | 104 ++++++++++++-- .../sbt/internal/WorkerExchangeTest.scala | 18 ++- project/Utils.scala | 2 + .../java/sbt/protocol/DuplexChannels.java | 91 ++++++++++++ .../scala/sbt/protocol/ClientSocket.scala | 12 +- .../sbt/protocol/ClientSocketDuplexTest.scala | 130 ++++++++++++++++++ .../java/sbt/internal/worker1/JdkCompat.java | 23 ++++ .../sbt/internal/worker1/DuplexChannels.java | 91 ++++++++++++ .../java/sbt/internal/worker1/JdkCompat.java | 25 ++++ .../java/sbt/internal/worker1/WorkerMain.java | 15 ++ .../sbt/internal/worker1/WorkerTest.scala | 13 ++ 13 files changed, 528 insertions(+), 29 deletions(-) create mode 100644 protocol/src/main/java/sbt/protocol/DuplexChannels.java create mode 100644 protocol/src/test/scala/sbt/protocol/ClientSocketDuplexTest.scala create mode 100644 worker/src/jdk17/java/sbt/internal/worker1/JdkCompat.java create mode 100644 worker/src/main/java/sbt/internal/worker1/DuplexChannels.java create mode 100644 worker/src/main/java/sbt/internal/worker1/JdkCompat.java diff --git a/build.sbt b/build.sbt index b9da33813..fc1bde8f3 100644 --- a/build.sbt +++ b/build.sbt @@ -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) diff --git a/main-actions/src/main/scala/sbt/ForkTests.scala b/main-actions/src/main/scala/sbt/ForkTests.scala index bb996c79e..6380fe55c 100755 --- a/main-actions/src/main/scala/sbt/ForkTests.scala +++ b/main-actions/src/main/scala/sbt/ForkTests.scala @@ -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 diff --git a/main-actions/src/main/scala/sbt/internal/WorkerExchange.scala b/main-actions/src/main/scala/sbt/internal/WorkerExchange.scala index 0918faa12..af368e28f 100644 --- a/main-actions/src/main/scala/sbt/internal/WorkerExchange.scala +++ b/main-actions/src/main/scala/sbt/internal/WorkerExchange.scala @@ -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) diff --git a/main-actions/src/test/scala/sbt/internal/WorkerExchangeTest.scala b/main-actions/src/test/scala/sbt/internal/WorkerExchangeTest.scala index fd64d7fa1..52f9bca5d 100644 --- a/main-actions/src/test/scala/sbt/internal/WorkerExchangeTest.scala +++ b/main-actions/src/test/scala/sbt/internal/WorkerExchangeTest.scala @@ -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 diff --git a/project/Utils.scala b/project/Utils.scala index b3e8833af..0a183d50f 100644 --- a/project/Utils.scala +++ b/project/Utils.scala @@ -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]]("") diff --git a/protocol/src/main/java/sbt/protocol/DuplexChannels.java b/protocol/src/main/java/sbt/protocol/DuplexChannels.java new file mode 100644 index 000000000..0ea53307c --- /dev/null +++ b/protocol/src/main/java/sbt/protocol/DuplexChannels.java @@ -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(); + } + }; + } +} diff --git a/protocol/src/main/scala/sbt/protocol/ClientSocket.scala b/protocol/src/main/scala/sbt/protocol/ClientSocket.scala index b7611aee7..74cc1f41b 100644 --- a/protocol/src/main/scala/sbt/protocol/ClientSocket.scala +++ b/protocol/src/main/scala/sbt/protocol/ClientSocket.scala @@ -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) } diff --git a/protocol/src/test/scala/sbt/protocol/ClientSocketDuplexTest.scala b/protocol/src/test/scala/sbt/protocol/ClientSocketDuplexTest.scala new file mode 100644 index 000000000..604a4d954 --- /dev/null +++ b/protocol/src/test/scala/sbt/protocol/ClientSocketDuplexTest.scala @@ -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 diff --git a/worker/src/jdk17/java/sbt/internal/worker1/JdkCompat.java b/worker/src/jdk17/java/sbt/internal/worker1/JdkCompat.java new file mode 100644 index 000000000..1e29477d2 --- /dev/null +++ b/worker/src/jdk17/java/sbt/internal/worker1/JdkCompat.java @@ -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; + } +} diff --git a/worker/src/main/java/sbt/internal/worker1/DuplexChannels.java b/worker/src/main/java/sbt/internal/worker1/DuplexChannels.java new file mode 100644 index 000000000..8c2629826 --- /dev/null +++ b/worker/src/main/java/sbt/internal/worker1/DuplexChannels.java @@ -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(); + } + }; + } +} diff --git a/worker/src/main/java/sbt/internal/worker1/JdkCompat.java b/worker/src/main/java/sbt/internal/worker1/JdkCompat.java new file mode 100644 index 000000000..e3f7b50a1 --- /dev/null +++ b/worker/src/main/java/sbt/internal/worker1/JdkCompat.java @@ -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"); + } +} diff --git a/worker/src/main/java/sbt/internal/worker1/WorkerMain.java b/worker/src/main/java/sbt/internal/worker1/WorkerMain.java index 9e9657f68..c9a6eb6ed 100644 --- a/worker/src/main/java/sbt/internal/worker1/WorkerMain.java +++ b/worker/src/main/java/sbt/internal/worker1/WorkerMain.java @@ -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); diff --git a/worker/src/test/scala/sbt/internal/worker1/WorkerTest.scala b/worker/src/test/scala/sbt/internal/worker1/WorkerTest.scala index 8285f3997..51b70559b 100644 --- a/worker/src/test/scala/sbt/internal/worker1/WorkerTest.scala +++ b/worker/src/test/scala/sbt/internal/worker1/WorkerTest.scala @@ -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()