Fix race condition in PosixPluginFrontend

Author: nadavsamet-db

.github/workflows/ci.yml

@@ -25,7 +25,7 @@ jobs:
         key: ${{ runner.os }}-sbt-${{ hashFiles('**/*.sbt') }}
     - name: Compile and test
       run: |
-       sbt test
+       sbt bridgeStressTest/compile test
       shell: bash
     - name: Format check
       if: ${{ runner.os == 'Linux' }}

bridge-stress-test/src/test/scala/protocbridge/frontend/PosixPluginFrontendSpec.scala

@@ -0,0 +1,52 @@
+package protocbridge.frontend
+
+import org.apache.commons.io.IOUtils
+import org.scalatest.flatspec.AnyFlatSpec
+import org.scalatest.matchers.must.Matchers
+import protocbridge.{ExtraEnv, ProtocCodeGenerator}
+
+import java.io.ByteArrayOutputStream
+import scala.sys.process.ProcessIO
+import scala.util.Random
+
+class PosixPluginFrontendSpec extends AnyFlatSpec with Matchers {
+  if (!PluginFrontend.isWindows) {
+    it must "execute a program that forwards input and output to given stream" in {
+      val random = new Random()
+      val toSend = Array.fill(123)(random.nextInt(256).toByte)
+      val toReceive = Array.fill(456)(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 10,000 times since named pipes on macOS are flaky.
+      val repeatCount = 10000
+      for (i <- 1 to repeatCount) {
+        if (i % 100 == 1) println(s"Running iteration $i of $repeatCount at ${java.time.LocalDateTime.now}")
+
+        val (path, state) = PosixPluginFrontend.prepare(
+          fakeGenerator,
+          env
+        )
+        val actualOutput = new ByteArrayOutputStream()
+        val process = sys.process
+          .Process(path.toAbsolutePath.toString)
+          .run(new ProcessIO(writeInput => {
+            writeInput.write(toSend)
+            writeInput.close()
+          }, processOutput => {
+            IOUtils.copy(processOutput, actualOutput)
+            processOutput.close()
+          }, _.close()))
+        process.exitValue()
+        actualOutput.toByteArray mustBe toReceive
+        PosixPluginFrontend.cleanup(state)
+      }
+    }
+  }
+}

bridge/src/main/scala/protocbridge/frontend/MacPluginFrontend.scala

@@ -137,7 +137,6 @@ object PluginFrontend {
 
   def newInstance: PluginFrontend = {
     if (isWindows) WindowsPluginFrontend
-    else if (isMac) MacPluginFrontend
     else PosixPluginFrontend
   }
 }

bridge/src/main/scala/protocbridge/frontend/PluginFrontend.scala

@@ -39,16 +39,13 @@ object PosixPluginFrontend extends PluginFrontend {
 
     Future {
       blocking {
+        // Each of the following two calls below block until the shell script opens the
+        // corresponding pipe. This ensures that we are not writing to any of the pipes
         val fsin = Files.newInputStream(inputPipe)
+        val fsout = Files.newOutputStream(outputPipe)
+
         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()
       }
@@ -77,8 +74,12 @@ object PosixPluginFrontend extends PluginFrontend {
       "",
       s"""|#!$shell
           |set -e
-          |cat /dev/stdin > "$inputPipe"
-          |cat "$outputPipe"
+          |exec 4> "$inputPipe"
+          |exec 5< "$outputPipe"
+          |cat /dev/stdin >&4
+          |exec 4>&-
+          |cat <&5
+          |exec 5<&-
       """.stripMargin
     )
     val perms = new ju.HashSet[PosixFilePermission]

bridge/src/main/scala/protocbridge/frontend/PosixPluginFrontend.scala

@@ -156,7 +156,7 @@ class ProtocIntegrationSpec extends AnyFlatSpec with Matchers {
 
     Await.result(
       Future.sequence(invocations),
-      Duration(60, SECONDS)
+      Duration(120, SECONDS)
     ) must be(List.fill(parallelProtocInvocations)(0))
   }
 }

bridge/src/test/scala/protocbridge/ProtocIntegrationSpec.scala

@@ -46,6 +46,17 @@ lazy val bridge: Project = project
     )
   )
 
+lazy val bridgeStressTest = project
+  .in(file("bridge-stress-test"))
+  .dependsOn(bridge)
+  .settings(
+    name := "bridge-stress-test",
+    libraryDependencies ++= Seq(
+      "org.scalatest" %% "scalatest" % "3.2.19" % "test",
+      "commons-io" % "commons-io" % "2.11.0" % "test"
+    )
+  )
+
 lazy val protocCacheCoursier = project
   .in(file("protoc-cache-coursier"))
   .dependsOn(bridge)