Revamp the test and run fork process infrastructure

1. Migrate to `NuProcess` and avoid the annoying Java and Scala process
APIs. Nuprocess gives us a better foundation to deal with stdout and
stderr, as well as processing standard input from the host process
(Bloop). `NuProcess` will also be more efficient than our previous
approach.

2. Migrate test server and the whole `runMain` machinery to `Task` so
that we can have a finer control over it. Note that now these actions
can be cancelled!

backend/src/main/scala/bloop/logging/ProcessLogger.scala

@@ -83,14 +83,32 @@ object ProcessLogger {
  * @param logFn  The handler that receives data from the `stream`
  * @param stream The stream that produces the data.
  */
-private class StreamLogger(logFn: String => Unit, stream: InputStream) extends Thread {
+private final class StreamLogger(logFn: String => Unit, stream: InputStream) extends Thread {
   private[this] val reader = new BufferedReader(new InputStreamReader(stream))
 
   @tailrec
-  override final def run(): Unit = {
+  override def run(): Unit = {
     Option(reader.readLine()) match {
       case Some(line) => logFn(line); run()
       case None => ()
     }
   }
 }
+
+private final class InputThread(in: InputStream, processIn: OutputStream) extends Thread {
+  override def run(): Unit = {
+    val buffer = new Array[Byte](256)
+    var read: Int = -1
+    while (true) {
+      try {
+        read = in.read(buffer)
+        if (read == -1) return
+        else {
+          processIn.write(buffer, 0, read)
+        }
+      } catch {
+        case e: java.io.IOException => return
+      }
+    }
+  }
+}

build.sbt

@@ -58,6 +58,7 @@ val jsonConfig = project
           Dependencies.metaconfigDocs,
           Dependencies.metaconfigConfig,
           Dependencies.circeDerivation,
+          Dependencies.nuprocess,
           Dependencies.scalacheck % Test,
         )
       } else {

frontend/src/main/scala/bloop/engine/tasks/Tasks.scala

@@ -4,7 +4,7 @@ import bloop.cli.ExitStatus
 import bloop.config.Config
 import bloop.engine.caches.ResultsCache
 import bloop.engine.{Dag, Leaf, Parent, State}
-import bloop.exec.ForkProcess
+import bloop.exec.Forker
 import bloop.io.AbsolutePath
 import bloop.logging.BspLogger
 import bloop.reporter.{BspReporter, LogReporter, Problem, Reporter, ReporterConfig}
@@ -271,52 +271,54 @@ object Tasks {
       isolated: Boolean,
       frameworkSpecificRawArgs: List[String],
       testFilter: String => Boolean
-  ): Task[State] = Task {
-    // TODO(jvican): This method should cache the test loader always.
+  ): Task[State] = {
     import state.logger
     import bloop.util.JavaCompat.EnrichOptional
+    def foundFrameworks (frameworks: Array[Framework])= frameworks.map(_.name).mkString(", ")
+
+    // Test arguments coming after `--` can only be used if only one mapping is found
+    def considerFrameworkArgs(frameworks: Array[Framework]) = {
+      if (frameworkSpecificRawArgs.isEmpty) Nil
+      else {
+        frameworks match {
+          case Array(oneFramework) =>
+            val rawArgs = frameworkSpecificRawArgs.toArray
+            val cls = oneFramework.getClass.getName()
+            logger.debug(s"Test options '$rawArgs' assigned to the only found framework $cls'.")
+            List(Config.TestArgument(rawArgs, Some(Config.TestFramework(List(cls)))))
+          case _ =>
+            val ignoredArgs = frameworkSpecificRawArgs.mkString(" ")
+            logger.warn(
+              s"Framework-specific test options '${ignoredArgs}' are ignored because several frameworks were found: ${foundFrameworks(frameworks)}")
+            Nil
+        }
+      }
+    }
 
     val projectsToTest = if (isolated) List(project) else Dag.dfs(state.build.getDagFor(project))
-    projectsToTest.foreach { project =>
+    val testTasks = projectsToTest.map { project =>
       val projectName = project.name
       val projectTestArgs = project.testOptions.arguments
-      val forkProcess = ForkProcess(project.javaEnv, project.classpath)
-      val testLoader = forkProcess.toExecutionClassLoader(Some(TestInternals.filteredLoader))
+      val forker = Forker(project.javaEnv, project.classpath)
+      val testLoader = forker.toExecutionClassLoader(Some(TestInternals.filteredLoader))
       val frameworks = project.testFrameworks
         .flatMap(f => TestInternals.loadFramework(testLoader, f.names, logger))
-      def foundFrameworks = frameworks.map(_.name).mkString(", ")
-      logger.debug(s"Found frameworks: $foundFrameworks")
+      logger.debug(s"Found frameworks: ${foundFrameworks(frameworks)}")
 
-      // Test arguments coming after `--` can only be used if only one mapping is found
-      val frameworkArgs = {
-        if (frameworkSpecificRawArgs.isEmpty) Nil
-        else {
-          frameworks match {
-            case Array(oneFramework) =>
-              val rawArgs = frameworkSpecificRawArgs.toArray
-              val cls = oneFramework.getClass.getName()
-              logger.debug(s"Test options '$rawArgs' assigned to the only found framework $cls'.")
-              List(Config.TestArgument(rawArgs, Some(Config.TestFramework(List(cls)))))
-            case _ =>
-              val ignoredArgs = frameworkSpecificRawArgs.mkString(" ")
-              logger.warn(
-                s"Framework-specific test options '${ignoredArgs}' are ignored because several frameworks were found: $foundFrameworks")
-              Nil
-          }
-        }
-      }
-
-      val analysis = state.results.lastSuccessfulResult(project).analysis().toOption.getOrElse {
+      val frameworkArgs = considerFrameworkArgs(frameworks)
+      val lastCompileResult = state.results.lastSuccessfulResult(project)
+      val analysis = lastCompileResult.analysis().toOption.getOrElse {
         logger.warn(s"Test execution is triggered but no compilation detected for ${projectName}.")
         Analysis.empty
       }
 
-      val discoveredTests = {
+      val discovered = {
         val tests = discoverTests(analysis, frameworks)
         val excluded = project.testOptions.excludes.toSet
         val ungroupedTests = tests.toList.flatMap {
           case (framework, tasks) => tasks.map(t => (framework, t))
         }
+
         val (includedTests, excludedTests) = ungroupedTests.partition {
           case (_, task) =>
             val fqn = task.fullyQualifiedName()
@@ -335,13 +337,17 @@ object Tasks {
         DiscoveredTests(testLoader, includedTests.groupBy(_._1).mapValues(_.map(_._2)))
       }
 
+      val opts = state.commonOptions
       val args = project.testOptions.arguments ++ frameworkArgs
-      val env = state.commonOptions.env
-      TestInternals.executeTasks(cwd, forkProcess, discoveredTests, args, handler, logger, env)
+      TestInternals.execute(cwd, forker, discovered, args, handler, logger, opts)
     }
 
-    // Return the previous state, test execution doesn't modify it.
-    state.mergeStatus(ExitStatus.Ok)
+    // For now, test execution is only sequential.
+    Task.sequence(testTasks).map { exitCodes =>
+      val isOk = exitCodes.forall(_ == 0)
+      if (isOk) state.mergeStatus(ExitStatus.Ok)
+      else state.copy(status = ExitStatus.TestExecutionError)
+    }
   }
 
   /**
@@ -357,16 +363,15 @@ object Tasks {
           project: Project,
           cwd: AbsolutePath,
           fqn: String,
-          args: Array[String]): Task[State] = Task {
+          args: Array[String]): Task[State] = {
     val classpath = project.classpath
-    val processConfig = ForkProcess(project.javaEnv, classpath)
-    val exitCode = processConfig.runMain(cwd, fqn, args, state.logger, state.commonOptions.env)
-    val exitStatus = {
-      if (exitCode == ForkProcess.EXIT_OK) ExitStatus.Ok
-      else ExitStatus.UnexpectedError
+    val processConfig = Forker(project.javaEnv, classpath)
+    processConfig.runMain(cwd, fqn, args, state.logger, state.commonOptions).map { exitCode =>
+      state.mergeStatus {
+        if (exitCode == Forker.EXIT_OK) ExitStatus.Ok
+        else ExitStatus.UnexpectedError
+      }
     }
-
-    state.mergeStatus(exitStatus)
   }
 
   /**