diff --git a/bridge/src/main/scala/protocbridge/frontend/MacPluginFrontend.scala b/bridge/src/main/scala/protocbridge/frontend/MacPluginFrontend.scala new file mode 100644 index 0000000..fd2a686 --- /dev/null +++ b/bridge/src/main/scala/protocbridge/frontend/MacPluginFrontend.scala @@ -0,0 +1,72 @@ +package protocbridge.frontend + +import org.newsclub.net.unix.AFUNIXServerSocket +import protocbridge.{ExtraEnv, ProtocCodeGenerator} + +import java.net.ServerSocket +import java.nio.file.attribute.PosixFilePermission +import java.nio.file.{Files, Path} +import java.{util => ju} + +/** PluginFrontend for macOS. + * + * Creates a server socket and uses `nc` to communicate with the socket. We use + * a server socket instead of named pipes because named pipes are unreliable on + * macOS: https://github.com/scalapb/protoc-bridge/issues/366 + * + * Since `nc` is widely available on macOS, this is the simplest alternative + * for macOS. However, raw `nc` is also not very reliable on macOS: + * https://github.com/scalapb/protoc-bridge/issues/379 + * + * The most reliable way to communicate is found to be with a domain socket and + * a server-side read timeout, which are implemented here. + */ +object MacPluginFrontend extends SocketBasedPluginFrontend { + case class InternalState( + shellScript: Path, + tempDirPath: Path, + socketPath: Path, + serverSocket: ServerSocket + ) + + override def prepare( + plugin: ProtocCodeGenerator, + env: ExtraEnv + ): (Path, InternalState) = { + val tempDirPath = Files.createTempDirectory("protocbridge") + val socketPath = tempDirPath.resolve("socket") + val serverSocket = AFUNIXServerSocket.bindOn(socketPath, true) + val sh = createShellScript(socketPath) + + runWithSocket(plugin, env, serverSocket) + + (sh, InternalState(sh, tempDirPath, socketPath, serverSocket)) + } + + override def cleanup(state: InternalState): Unit = { + state.serverSocket.close() + if (sys.props.get("protocbridge.debug") != Some("1")) { + Files.delete(state.tempDirPath) + Files.delete(state.shellScript) + } + } + + private def createShellScript(socketPath: Path): Path = { + val shell = sys.env.getOrElse("PROTOCBRIDGE_SHELL", "/bin/sh") + val scriptName = PluginFrontend.createTempFile( + "", + s"""|#!$shell + |set -e + |nc -U "$socketPath" + """.stripMargin + ) + val perms = new ju.HashSet[PosixFilePermission] + perms.add(PosixFilePermission.OWNER_EXECUTE) + perms.add(PosixFilePermission.OWNER_READ) + Files.setPosixFilePermissions( + scriptName, + perms + ) + scriptName + } +} diff --git a/bridge/src/main/scala/protocbridge/frontend/PluginFrontend.scala b/bridge/src/main/scala/protocbridge/frontend/PluginFrontend.scala index 7415f06..3b83cfa 100644 --- a/bridge/src/main/scala/protocbridge/frontend/PluginFrontend.scala +++ b/bridge/src/main/scala/protocbridge/frontend/PluginFrontend.scala @@ -5,8 +5,6 @@ import java.nio.file.{Files, Path} import protocbridge.{ProtocCodeGenerator, ExtraEnv} -import scala.util.Try - /** A PluginFrontend instance provides a platform-dependent way for protoc to * communicate with a JVM based ProtocCodeGenerator. * @@ -47,13 +45,7 @@ object PluginFrontend { gen: ProtocCodeGenerator, request: Array[Byte] ): Array[Byte] = { - Try { - gen.run(request) - }.recover { case throwable => - createCodeGeneratorResponseWithError( - throwable.toString + "\n" + getStackTrace(throwable) - ) - }.get + gen.run(request) } def createCodeGeneratorResponseWithError(error: String): Array[Byte] = { @@ -116,9 +108,17 @@ object PluginFrontend { gen: ProtocCodeGenerator, fsin: InputStream, env: ExtraEnv - ): Array[Byte] = { + ): Array[Byte] = try { val bytes = readInputStreamToByteArrayWithEnv(fsin, env) runWithBytes(gen, bytes) + } catch { + // This covers all Throwable including OutOfMemoryError, StackOverflowError, etc. + // We need to make a best effort to return a response to protoc, + // otherwise protoc can hang indefinitely. + case throwable: Throwable => + createCodeGeneratorResponseWithError( + throwable.toString + "\n" + getStackTrace(throwable) + ) } def createTempFile(extension: String, content: String): Path = { @@ -131,8 +131,13 @@ object PluginFrontend { def isWindows: Boolean = sys.props("os.name").startsWith("Windows") + def isMac: Boolean = sys.props("os.name").startsWith("Mac") || sys + .props("os.name") + .startsWith("Darwin") + def newInstance: PluginFrontend = { if (isWindows) WindowsPluginFrontend + else if (isMac) MacPluginFrontend else PosixPluginFrontend } } diff --git a/bridge/src/main/scala/protocbridge/frontend/PosixPluginFrontend.scala b/bridge/src/main/scala/protocbridge/frontend/PosixPluginFrontend.scala index 5f70120..65935e0 100644 --- a/bridge/src/main/scala/protocbridge/frontend/PosixPluginFrontend.scala +++ b/bridge/src/main/scala/protocbridge/frontend/PosixPluginFrontend.scala @@ -12,10 +12,13 @@ import scala.concurrent.ExecutionContext.Implicits.global import scala.sys.process._ import java.{util => ju} -/** PluginFrontend for Unix-like systems (Linux, Mac, etc) +/** PluginFrontend for Unix-like systems except macOS (Linux, FreeBSD, + * etc) * * Creates a pair of named pipes for input/output and a shell script that - * communicates with them. + * communicates with them. Compared with `SocketBasedPluginFrontend`, this + * frontend doesn't rely on `nc` that might not be available in some + * distributions. */ object PosixPluginFrontend extends PluginFrontend { case class InternalState( @@ -40,6 +43,11 @@ object PosixPluginFrontend extends PluginFrontend { val response = PluginFrontend.runWithInputStream(plugin, fsin, env) fsin.close() + // Note that the output pipe must be opened after the input pipe is consumed. + // Otherwise, there might be a deadlock that + // - The shell script is stuck writing to the input pipe (which has a full buffer), + // and doesn't open the write end of the output pipe. + // - This thread is stuck waiting for the write end of the output pipe to be opened. val fsout = Files.newOutputStream(outputPipe) fsout.write(response) fsout.close() diff --git a/bridge/src/main/scala/protocbridge/frontend/SocketBasedPluginFrontend.scala b/bridge/src/main/scala/protocbridge/frontend/SocketBasedPluginFrontend.scala new file mode 100644 index 0000000..9f65cb4 --- /dev/null +++ b/bridge/src/main/scala/protocbridge/frontend/SocketBasedPluginFrontend.scala @@ -0,0 +1,44 @@ +package protocbridge.frontend + +import protocbridge.{ExtraEnv, ProtocCodeGenerator} + +import java.net.ServerSocket +import scala.concurrent.ExecutionContext.Implicits.global +import scala.concurrent.{Future, blocking} + +/** PluginFrontend for Windows and macOS where a server socket is used. + */ +abstract class SocketBasedPluginFrontend extends PluginFrontend { + + protected def runWithSocket( + plugin: ProtocCodeGenerator, + env: ExtraEnv, + serverSocket: ServerSocket + ): Unit = { + Future { + blocking { + // Accept a single client connection from the shell script. + val client = serverSocket.accept() + // It's found on macOS that a `junixsocket` domain socket server + // might not receive the EOF sent by the other end, leading to a hang: + // https://github.com/scalapb/protoc-bridge/issues/379 + // However, confusingly, adding an arbitrary read timeout resolves the issue. + // We thus add a read timeout of 1 minute here, which should be more than enough. + // It also helps to prevent an infinite hang on both Windows and macOS due to + // unexpected issues. + // client.setSoTimeout(60000) + try { + val response = + PluginFrontend.runWithInputStream( + plugin, + client.getInputStream, + env + ) + client.getOutputStream.write(response) + } finally { + client.close() + } + } + } + } +} diff --git a/bridge/src/main/scala/protocbridge/frontend/WindowsPluginFrontend.scala b/bridge/src/main/scala/protocbridge/frontend/WindowsPluginFrontend.scala index 490211d..0f82cca 100644 --- a/bridge/src/main/scala/protocbridge/frontend/WindowsPluginFrontend.scala +++ b/bridge/src/main/scala/protocbridge/frontend/WindowsPluginFrontend.scala @@ -1,53 +1,38 @@ package protocbridge.frontend +import protocbridge.{ExtraEnv, ProtocCodeGenerator} + import java.net.ServerSocket import java.nio.file.{Files, Path, Paths} -import protocbridge.ExtraEnv -import protocbridge.ProtocCodeGenerator - -import scala.concurrent.blocking - -import scala.concurrent.ExecutionContext.Implicits.global -import scala.concurrent.Future - /** A PluginFrontend that binds a server socket to a local interface. The plugin * is a batch script that invokes BridgeApp.main() method, in a new JVM with * the same parameters as the currently running JVM. The plugin will * communicate its stdin and stdout to this socket. */ -object WindowsPluginFrontend extends PluginFrontend { - - case class InternalState(batFile: Path) +object WindowsPluginFrontend extends SocketBasedPluginFrontend { + case class InternalState(shellScript: Path, serverSocket: ServerSocket) override def prepare( plugin: ProtocCodeGenerator, env: ExtraEnv ): (Path, InternalState) = { - val ss = new ServerSocket(0) - val state = createWindowsScript(ss.getLocalPort) + val ss = new ServerSocket(0) // Bind to any available port. + val sh = createShellScript(ss.getLocalPort) - Future { - blocking { - val client = ss.accept() - val response = - PluginFrontend.runWithInputStream(plugin, client.getInputStream, env) - client.getOutputStream.write(response) - client.close() - ss.close() - } - } + runWithSocket(plugin, env, ss) - (state.batFile, state) + (sh, InternalState(sh, ss)) } override def cleanup(state: InternalState): Unit = { + state.serverSocket.close() if (sys.props.get("protocbridge.debug") != Some("1")) { - Files.delete(state.batFile) + Files.delete(state.shellScript) } } - private def createWindowsScript(port: Int): InternalState = { + private def createShellScript(port: Int): Path = { val classPath = Paths.get(getClass.getProtectionDomain.getCodeSource.getLocation.toURI) val classPathBatchString = classPath.toString.replace("%", "%%") @@ -62,6 +47,6 @@ object WindowsPluginFrontend extends PluginFrontend { ].getName} $port """.stripMargin ) - InternalState(batchFile) + batchFile } } diff --git a/bridge/src/test/scala/protocbridge/frontend/MacPluginFrontendSpec.scala b/bridge/src/test/scala/protocbridge/frontend/MacPluginFrontendSpec.scala new file mode 100644 index 0000000..6e8b972 --- /dev/null +++ b/bridge/src/test/scala/protocbridge/frontend/MacPluginFrontendSpec.scala @@ -0,0 +1,15 @@ +package protocbridge.frontend + +class MacPluginFrontendSpec extends OsSpecificFrontendSpec { + if (PluginFrontend.isMac) { + it must "execute a program that forwards input and output to given stream" in { + val state = testSuccess(MacPluginFrontend) + state.serverSocket.isClosed mustBe true + } + + it must "not hang if there is an error in generator" in { + val state = testFailure(MacPluginFrontend) + state.serverSocket.isClosed mustBe true + } + } +} diff --git a/bridge/src/test/scala/protocbridge/frontend/OsSpecificFrontendSpec.scala b/bridge/src/test/scala/protocbridge/frontend/OsSpecificFrontendSpec.scala new file mode 100644 index 0000000..de79098 --- /dev/null +++ b/bridge/src/test/scala/protocbridge/frontend/OsSpecificFrontendSpec.scala @@ -0,0 +1,111 @@ +package protocbridge.frontend + +import org.apache.commons.io.IOUtils +import org.scalatest.exceptions.TestFailedException +import org.scalatest.flatspec.AnyFlatSpec +import org.scalatest.matchers.must.Matchers +import protocbridge.{ExtraEnv, ProtocCodeGenerator} + +import java.io.ByteArrayOutputStream +import scala.concurrent.ExecutionContext.Implicits.global +import scala.concurrent.duration.DurationInt +import scala.concurrent.{Await, Future, TimeoutException} +import scala.sys.process.ProcessIO +import scala.util.Random + +class OsSpecificFrontendSpec extends AnyFlatSpec with Matchers { + + protected def testPluginFrontend( + frontend: PluginFrontend, + generator: ProtocCodeGenerator, + env: ExtraEnv, + request: Array[Byte] + ): (frontend.InternalState, Array[Byte]) = { + val (path, state) = frontend.prepare( + generator, + env + ) + val actualOutput = new ByteArrayOutputStream() + val process = sys.process + .Process(path.toAbsolutePath.toString) + .run( + new ProcessIO( + writeInput => { + writeInput.write(request) + writeInput.close() + }, + processOutput => { + IOUtils.copy(processOutput, actualOutput) + processOutput.close() + }, + processError => { + IOUtils.copy(processError, System.err) + processError.close() + } + ) + ) + try { + Await.result(Future { process.exitValue() }, 5.seconds) + } catch { + case _: TimeoutException => + System.err.println(s"Timeout") + process.destroy() + } + frontend.cleanup(state) + (state, actualOutput.toByteArray) + } + + protected def testSuccess( + frontend: PluginFrontend + ): frontend.InternalState = { + val random = new Random() + val toSend = Array.fill(100000)(random.nextInt(256).toByte) + val toReceive = Array.fill(100000)(random.nextInt(256).toByte) + val env = new ExtraEnv(secondaryOutputDir = "tmp") + + val fakeGenerator = new ProtocCodeGenerator { + override def run(request: Array[Byte]): Array[Byte] = { + request mustBe (toSend ++ env.toByteArrayAsField) + toReceive + } + } + // Repeat 100,000 times since named pipes on macOS are flaky. + val repeatCount = 100000 + for (i <- 1 until repeatCount) { + if (i % 100 == 1) println(s"Running iteration $i of $repeatCount") + val (state, response) = + testPluginFrontend(frontend, fakeGenerator, env, toSend) + if (!(response sameElements toReceive)) { + System.err.println( + s"Failed on iteration $i of $repeatCount ($state): ${response.length} != ${toReceive.length}" + ) + } + } + val (state, response) = + testPluginFrontend(frontend, fakeGenerator, env, toSend) + if (!(response sameElements toReceive)) { + System.err.println( + s"Failed on iteration $repeatCount of $repeatCount ($state): ${response.length} != ${toReceive.length}" + ) + } + state + } + + protected def testFailure( + frontend: PluginFrontend + ): frontend.InternalState = { + val random = new Random() + val toSend = Array.fill(123)(random.nextInt(256).toByte) + val env = new ExtraEnv(secondaryOutputDir = "tmp") + + val fakeGenerator = new ProtocCodeGenerator { + override def run(request: Array[Byte]): Array[Byte] = { + throw new OutOfMemoryError("test error") + } + } + val (state, response) = + testPluginFrontend(frontend, fakeGenerator, env, toSend) + response.length must be > 0 + state + } +} diff --git a/bridge/src/test/scala/protocbridge/frontend/PosixPluginFrontendSpec.scala b/bridge/src/test/scala/protocbridge/frontend/PosixPluginFrontendSpec.scala new file mode 100644 index 0000000..1c615d2 --- /dev/null +++ b/bridge/src/test/scala/protocbridge/frontend/PosixPluginFrontendSpec.scala @@ -0,0 +1,13 @@ +package protocbridge.frontend + +class PosixPluginFrontendSpec extends OsSpecificFrontendSpec { + if (!PluginFrontend.isWindows && !PluginFrontend.isMac) { + it must "execute a program that forwards input and output to given stream" in { + testSuccess(MacPluginFrontend) + } + + it must "not hang if there is an OOM in generator" in { + testFailure(MacPluginFrontend) + } + } +} diff --git a/bridge/src/test/scala/protocbridge/frontend/WindowsPluginFrontendSpec.scala b/bridge/src/test/scala/protocbridge/frontend/WindowsPluginFrontendSpec.scala index 6385ad7..db0bc65 100644 --- a/bridge/src/test/scala/protocbridge/frontend/WindowsPluginFrontendSpec.scala +++ b/bridge/src/test/scala/protocbridge/frontend/WindowsPluginFrontendSpec.scala @@ -1,38 +1,15 @@ package protocbridge.frontend -import java.io.ByteArrayInputStream - -import protocbridge.{ProtocCodeGenerator, ExtraEnv} - -import scala.sys.process.ProcessLogger -import org.scalatest.flatspec.AnyFlatSpec -import org.scalatest.matchers.must.Matchers - -class WindowsPluginFrontendSpec extends AnyFlatSpec with Matchers { +class WindowsPluginFrontendSpec extends OsSpecificFrontendSpec { if (PluginFrontend.isWindows) { it must "execute a program that forwards input and output to given stream" in { - val toSend = "ping" - val toReceive = "pong" - val env = new ExtraEnv(secondaryOutputDir = "tmp") + val state = testSuccess(WindowsPluginFrontend) + state.serverSocket.isClosed mustBe true + } - val fakeGenerator = new ProtocCodeGenerator { - override def run(request: Array[Byte]): Array[Byte] = { - request mustBe (toSend.getBytes ++ env.toByteArrayAsField) - toReceive.getBytes - } - } - val (path, state) = WindowsPluginFrontend.prepare( - fakeGenerator, - env - ) - val actualOutput = scala.collection.mutable.Buffer.empty[String] - val process = sys.process - .Process(path.toAbsolutePath.toString) - .#<(new ByteArrayInputStream(toSend.getBytes)) - .run(ProcessLogger(o => actualOutput.append(o))) - process.exitValue() - actualOutput.mkString mustBe toReceive - WindowsPluginFrontend.cleanup(state) + it must "not hang if there is an OOM in generator" in { + val state = testFailure(WindowsPluginFrontend) + state.serverSocket.isClosed mustBe true } } } diff --git a/build.sbt b/build.sbt index 568684d..6498eb2 100644 --- a/build.sbt +++ b/build.sbt @@ -29,7 +29,8 @@ lazy val bridge: Project = project "org.scalatest" %% "scalatest" % "3.2.17" % "test", "org.scalacheck" %% "scalacheck" % "1.17.0" % "test", "org.scala-lang.modules" %% "scala-collection-compat" % "2.11.0" % "test", - "io.get-coursier" %% "coursier" % coursierVersion % "test" + "io.get-coursier" %% "coursier" % coursierVersion % "test", + "com.kohlschutter.junixsocket" % "junixsocket-core" % "2.10.0" ), scalacOptions ++= (if (scalaVersion.value.startsWith("2.13.")) Seq("-Wconf:origin=.*JavaConverters.*:s") diff --git a/domain_socket_stress_test.sh b/domain_socket_stress_test.sh new file mode 100755 index 0000000..0747233 --- /dev/null +++ b/domain_socket_stress_test.sh @@ -0,0 +1,88 @@ +#!/usr/bin/env bash + +BYTE_LENGTH="$1" +SERVER_MODE="${2:-nc-save}" +CLIENT_MODE="${3:-nc}" + +TEST_FILE_PATH="/tmp/domain_socket_test_file" +SOCKET_PATH="/tmp/domain_socket_test.sck" +SERVER_RESULT_PATH="$TEST_FILE_PATH.server" +CLIENT_RESULT_PATH="$TEST_FILE_PATH.client" +dd if=/dev/urandom of="$TEST_FILE_PATH" bs=1 count="$BYTE_LENGTH" 2>/dev/null + +if [[ "$SERVER_MODE" == *"-save" ]]; then + TEST_RESULT_PATH="$SERVER_RESULT_PATH" +else + TEST_RESULT_PATH="$CLIENT_RESULT_PATH" +fi + +test_socket() { + # Start a process to consume the data from the socket + if [[ "$SERVER_MODE" == "nc-save" ]]; then + (nc -l -U "$SOCKET_PATH" > "$SERVER_RESULT_PATH" && echo "Completed saving random bytes from the socket") & + elif [[ "$SERVER_MODE" == "ncat-save" ]]; then + (ncat -l -U "$SOCKET_PATH" > "$SERVER_RESULT_PATH" && echo "Completed saving random bytes from the socket") & + elif [[ "$SERVER_MODE" == "socat-save" ]]; then + (socat UNIX-LISTEN:"$SOCKET_PATH" - > "$SERVER_RESULT_PATH" && echo "Completed saving random bytes from the socket") & + elif [[ "$SERVER_MODE" == "socat-echo" ]]; then + (socat UNIX-LISTEN:"$SOCKET_PATH" EXEC:"/bin/cat" && echo "Completed echoing random bytes from the socket") & + else + echo "Invalid server mode: $SERVER_MODE" + exit 1 + fi + SERVER_PID=$! + echo "Starting the server (PID: $SERVER_PID)" + + # Wait for the socket file to be created so that the server has started + while [ ! -e "$SOCKET_PATH" ]; do + sleep 0.001 + done + echo "The server has started and is listening to the socket (PID: $SERVER_PID)" + + # `nc` can fail even if we wait for another second to ensure the server has started + # sleep 1 + + # Start dumping random bytes to the socket in the background + if [[ "$CLIENT_MODE" == "nc" ]]; then + (nc -U "$SOCKET_PATH" < "$TEST_FILE_PATH" > "$CLIENT_RESULT_PATH" && echo "Completed dumping random bytes to the socket") & + elif [[ "$CLIENT_MODE" == "ncat" ]]; then + (ncat -U "$SOCKET_PATH" < "$TEST_FILE_PATH" > "$CLIENT_RESULT_PATH" && echo "Completed dumping random bytes to the socket") & + elif [[ "$CLIENT_MODE" == "socat" ]]; then + (socat - UNIX-CONNECT:"$SOCKET_PATH" < "$TEST_FILE_PATH" > "$CLIENT_RESULT_PATH" && echo "Completed dumping random bytes to the socket") & + else + echo "Invalid client mode: $CLIENT_MODE" + exit 1 + fi + CLIENT_PID=$! + echo "Started dumping random bytes to the socket (PID: $CLIENT_PID)" + + # Ensure the client process is killed + wait $CLIENT_PID 2>/dev/null + echo "The client process has stopped (PID: $CLIENT_PID)" + + # Ensure the server process is killed + wait $SERVER_PID 2>/dev/null + echo "The server process has stopped (PID: $SERVER_PID)" + + # Check the size of the data read from the socket + DATA_SIZE=$(wc -c < "$TEST_RESULT_PATH") + if [ "$DATA_SIZE" -ne "$BYTE_LENGTH" ]; then + echo "Error: Expected $BYTE_LENGTH bytes, but read $DATA_SIZE bytes" + exit 1 + else + echo "Successfully read $BYTE_LENGTH bytes from the socket" + fi + + rm -f "$SOCKET_PATH" + rm -f "$SERVER_RESULT_PATH" + rm -f "$CLIENT_RESULT_PATH" +} + +rm -f "$SOCKET_PATH" + +# Repeat the process +counter=0; +while test_socket; do + ((counter++)); echo "Iterations completed: $counter"; +done +echo "Command failed after $counter successful iterations."