Fix race condition in ScalaCompile by selectively interrupting futures with Thread.interrupt

Prior to this change there was a bug when ScalaCompile actions were used
with dynamic execution. The bug would cause builds to fail with the
error message "compiler mirror not found". It is my understanding this
message typically indicates a problem loading the Scala Library jar
during compilation.

The error would always occur in the local multiplex worker's logs and
not the remote logs.

This commit fixes this error by not using Thread.interrupt when
cancelling the future running the ScalaCompile action.

My hypothesis on why this error occurs is as follows: imagine one of the
classloaders in the shared ScalaInstance is in use on Thread A. That
thread gets cancelled and interrupted via Thread.interrupt. This causes
some kind of persistent error inside either the classloader or another
part of Zinc or the Scala compiler. We correctly ignore the error on
Thread A because the request it was handling was cancelled. Another
thread is working a non-cancelled request, it uses that same
ScalaInstance, hit that persistent error, and fails.

By not using Thread.interrupt we don't trigger the persistent error and
thus avoid the bug.

Author: jjudd

rules/scala_proto/private/ScalaProtoWorker.scala

@@ -6,7 +6,7 @@ import higherkindness.rules_scala.common.args.implicits.*
 import higherkindness.rules_scala.common.interrupt.InterruptUtil
 import higherkindness.rules_scala.common.error.AnnexWorkerError
 import higherkindness.rules_scala.common.sandbox.SandboxUtil
-import higherkindness.rules_scala.common.worker.WorkerMain
+import higherkindness.rules_scala.common.worker.{WorkerMain, WorkTask}
 import java.io.{File, PrintStream}
 import java.nio.file.{Files, Path, Paths}
 import java.util.Collections
@@ -72,9 +72,12 @@ object ScalaProtoWorker extends WorkerMain[Unit] {
 
   override def init(args: Option[Array[String]]): Unit = ()
 
-  protected def work(ctx: Unit, args: Array[String], out: PrintStream, workDir: Path, verbosity: Int): Unit = {
-    val workRequest = ScalaProtoRequest(workDir, ArgsUtil.parseArgsOrFailSafe(args, argParser, out))
-    InterruptUtil.throwIfInterrupted()
+  protected def work(task: WorkTask[Unit]): Unit = {
+    val workRequest = ScalaProtoRequest(
+      task.workDir,
+      ArgsUtil.parseArgsOrFailSafe(task.args, argParser, task.output),
+    )
+    InterruptUtil.throwIfInterrupted(task.isCancelled)
 
     val scalaOut = workRequest.outputDir
     Files.createDirectories(scalaOut)
@@ -93,7 +96,7 @@ object ScalaProtoWorker extends WorkerMain[Unit] {
       }
     }
 
-    InterruptUtil.throwIfInterrupted()
+    InterruptUtil.throwIfInterrupted(task.isCancelled)
     val exitCode = ProtocBridge.runWithGenerators(
       new MyProtocRunner,
       namedGenerators = List("scala" -> ScalaPbCodeGenerator),
@@ -102,7 +105,7 @@ object ScalaProtoWorker extends WorkerMain[Unit] {
     if (exitCode != 0) {
       throw new AnnexWorkerError(exitCode)
     }
-    InterruptUtil.throwIfInterrupted()
+    InterruptUtil.throwIfInterrupted(task.isCancelled)
   }
 
 }

rules/scalafmt/scalafmt/ScalafmtRunner.scala

@@ -4,7 +4,7 @@ import higherkindness.rules_scala.common.args.ArgsUtil
 import higherkindness.rules_scala.common.args.ArgsUtil.PathArgumentType
 import higherkindness.rules_scala.common.interrupt.InterruptUtil
 import higherkindness.rules_scala.common.sandbox.SandboxUtil
-import higherkindness.rules_scala.common.worker.WorkerMain
+import higherkindness.rules_scala.common.worker.{WorkTask, WorkerMain}
 import higherkindness.rules_scala.workers.common.Color
 import java.io.{File, PrintStream}
 import java.nio.file.{Files, Path}
@@ -45,16 +45,19 @@ object ScalafmtRunner extends WorkerMain[Unit] {
 
   protected def init(args: Option[Array[String]]): Unit = {}
 
-  protected def work(worker: Unit, args: Array[String], out: PrintStream, workDir: Path, verbosity: Int): Unit = {
-    val workRequest = ScalafmtRequest(workDir, ArgsUtil.parseArgsOrFailSafe(args, argParser, out))
-    InterruptUtil.throwIfInterrupted()
+  protected def work(task: WorkTask[Unit]): Unit = {
+    val workRequest = ScalafmtRequest(
+      task.workDir,
+      ArgsUtil.parseArgsOrFailSafe(task.args, argParser, task.output),
+    )
+    InterruptUtil.throwIfInterrupted(task.isCancelled)
 
     val source = FileOps.readFile(workRequest.inputFile)(Codec.UTF8)
 
     val config = ScalafmtConfig.fromHoconFile(workRequest.configFile).get
     @tailrec
     def format(code: String): String = {
-      InterruptUtil.throwIfInterrupted()
+      InterruptUtil.throwIfInterrupted(task.isCancelled)
       val formatted = Scalafmt.format(code, config).get
       if (code == formatted) code else format(formatted)
     }
@@ -76,7 +79,7 @@ object ScalafmtRunner extends WorkerMain[Unit] {
       }
 
     Files.write(workRequest.outputFile, output.getBytes)
-    InterruptUtil.throwIfInterrupted()
+    InterruptUtil.throwIfInterrupted(task.isCancelled)
   }
 
 }

src/main/scala/higherkindness/rules_scala/common/interrupt/InterruptUtil.scala

@@ -1,10 +1,14 @@
 package higherkindness.rules_scala
 package common.interrupt
 
+import java.util.concurrent.CancellationException
+
 object InterruptUtil {
-  def throwIfInterrupted(): Unit = {
+  def throwIfInterrupted(isCancelled: Function0[Boolean]): Unit = {
     if (Thread.interrupted()) {
-      throw new InterruptedException("WorkRequest was cancelled.")
+      throw new InterruptedException("WorkRequest was cancelled via Thread interruption.")
+    } else if (isCancelled()) {
+      throw new CancellationException("WorkRequest was cancelled via FutureTask cancellation.")
     }
   }
 }

src/main/scala/higherkindness/rules_scala/common/worker/CancellableTask.scala

@@ -16,12 +16,12 @@ import scala.util.Try
  * Heavily inspired by the following: https://github.com/NthPortal/cancellable-task/tree/master
  * https://stackoverflow.com/a/39986418/6442597
  */
-class CancellableTask[S] private (fn: => S) {
+class CancellableTask[S] private (fn: Function1[Function0[Boolean], S]) {
   private val promise = Promise[S]()
   val future: Future[S] = promise.future
 
   private val fnCallable = new Callable[S]() {
-    def call(): S = fn
+    def call(): S = fn(isCancelled)
   }
 
   private val task = new FutureTaskWaitOnCancel[S](fnCallable) {
@@ -39,10 +39,16 @@ class CancellableTask[S] private (fn: => S) {
   def cancel(mayInterruptIfRunning: Boolean): Boolean = task.cancel(mayInterruptIfRunning)
 
   def execute(executionContext: ExecutionContext): Unit = executionContext.execute(task)
+
+  def isCancelled(): Boolean = task.isCancelled()
 }
 
 object CancellableTask {
   def apply[S](fn: => S): CancellableTask[S] = {
+    new CancellableTask((_: Function0[Boolean]) => fn)
+  }
+
+  def apply[S](fn: Function1[Function0[Boolean], S]): CancellableTask[S] = {
     new CancellableTask(fn)
   }
 }

src/main/scala/higherkindness/rules_scala/common/worker/WorkTask.scala

@@ -0,0 +1,14 @@
+package higherkindness.rules_scala
+package common.worker
+
+import java.io.PrintStream
+import java.nio.file.Path
+
+case class WorkTask[S](
+  context: S,
+  args: Array[String],
+  output: PrintStream,
+  workDir: Path,
+  verbosity: Int,
+  isCancelled: Function0[Boolean],
+)

src/main/scala/higherkindness/rules_scala/common/worker/WorkerMain.scala

@@ -4,6 +4,7 @@ package common.worker
 import common.error.{AnnexDuplicateActiveRequestException, AnnexWorkerError}
 import com.google.devtools.build.lib.worker.WorkerProtocol
 import java.io.{ByteArrayInputStream, ByteArrayOutputStream, InputStream, OutputStream, PrintStream}
+import java.nio.channels.ClosedByInterruptException
 import java.nio.file.{Path, Paths}
 import java.util.concurrent.{Callable, CancellationException, ConcurrentHashMap, ForkJoinPool, FutureTask}
 import scala.annotation.tailrec
@@ -15,10 +16,27 @@ abstract class WorkerMain[S](stdin: InputStream = System.in, stdout: PrintStream
 
   protected def init(args: Option[Array[String]]): S
 
-  protected def work(ctx: S, args: Array[String], out: PrintStream, workDir: Path, verbosity: Int): Unit
-
+  /**
+   * isCancelled is used to determine whether the FutureTask that is executing this work request has been cancelled. If
+   * it is safe to Thread.interrupt a process, then that is done and can be checked. Not all workers can be
+   * Thread.interrupted safely.
+   *
+   * TODO(James): document the rest of this function
+   */
+  protected def work(workRequest: WorkTask[S]): Unit
+
+  /**
+   * Indicates whether this program is being executed as a worker or as a regular process. It is a var because we won't
+   * know until runtime which one it is.
+   */
   protected var isWorker = false
 
+  /**
+   * Used to determine whether to interrupt the FutureTasks being executed by this worker using Thread.interrupt or not.
+   * It's safe to interrupt many things with Thread.interrupt, but not all things.
+   */
+  protected val mayInterruptWorkerTasks = true
+
   final def main(args: Array[String]): Unit = {
     args.toList match {
       case "--persistent_worker" :: args =>
@@ -113,7 +131,9 @@ abstract class WorkerMain[S](stdin: InputStream = System.in, stdout: PrintStream
             Option(activeRequests.get(requestId)).foreach { activeRequest =>
               // Cancel will wait for the thread to complete or be interrupted, so we do it in a future
               // to prevent blocking the worker from processing more requests
-              Future(activeRequest.cancel(mayInterruptIfRunning = true))(scala.concurrent.ExecutionContext.global)
+              Future(activeRequest.cancel(mayInterruptIfRunning = mayInterruptWorkerTasks))(
+                scala.concurrent.ExecutionContext.global,
+              )
             }
           } else {
             val args = request.getArgumentsList.toArray(Array.empty[String])
@@ -131,13 +151,13 @@ abstract class WorkerMain[S](stdin: InputStream = System.in, stdout: PrintStream
               maybeOut.map(_.flush())
             }
 
-            val workTask = CancellableTask {
+            def doWork(isCancelled: Function0[Boolean]) = {
               val outStream = new ByteArrayOutputStream()
               val out = new PrintStream(outStream)
               maybeOutStream = Some(outStream)
               maybeOut = Some(out)
               try {
-                work(ctx, args, out, sandboxDir, verbosity)
+                work(WorkTask(ctx, args, out, sandboxDir, verbosity, isCancelled))
                 0
               } catch {
                 case e @ AnnexWorkerError(code, _, _) =>
@@ -146,6 +166,8 @@ abstract class WorkerMain[S](stdin: InputStream = System.in, stdout: PrintStream
               }
             }
 
+            val workTask = CancellableTask(doWork)
+
             workTask.future
               .andThen {
                 // Work task succeeded or failed in an expected way
@@ -178,7 +200,7 @@ abstract class WorkerMain[S](stdin: InputStream = System.in, stdout: PrintStream
                   }
 
                 // Task successfully cancelled
-                case Failure(e: CancellationException) =>
+                case Failure(e @ (_: CancellationException | _: ClosedByInterruptException)) =>
                   flushOut()
                   writeResponse(requestId, None, None, wasCancelled = true)
                   logVerbose(
@@ -235,11 +257,14 @@ abstract class WorkerMain[S](stdin: InputStream = System.in, stdout: PrintStream
           val returnCode =
             try {
               work(
-                init(args = None),
-                args.toArray,
-                out,
-                workDir = Path.of(""),
-                verbosity = 0,
+                WorkTask(
+                  init(args = None),
+                  args.toArray,
+                  out,
+                  workDir = Path.of(""),
+                  verbosity = 0,
+                  isCancelled = () => false,
+                ),
               )
 
               0

src/main/scala/higherkindness/rules_scala/workers/bloop/compile/BloopRunner.scala

@@ -1,13 +1,13 @@
 package higherkindness.rules_scala
 package workers.bloop.compile
 
-import common.worker.WorkerMain
+import common.worker.{WorkTask, WorkerMain}
 
 import bloop.Bloop
 import java.io.PrintStream
 import java.nio.file.Path
 
 object BloopRunner extends WorkerMain[Unit] {
   override def init(args: Option[Array[String]]): Unit = ()
-  override def work(ctx: Unit, args: Array[String], out: PrintStream, workDir: Path, verbosity: Int): Unit = Bloop
+  override def work(workRequest: WorkTask[Unit]): Unit = Bloop
 }

src/main/scala/higherkindness/rules_scala/workers/deps/DepsRunner.scala

@@ -7,7 +7,7 @@ import common.args.implicits.*
 import common.error.AnnexWorkerError
 import common.interrupt.InterruptUtil
 import common.sandbox.SandboxUtil
-import common.worker.WorkerMain
+import common.worker.{WorkTask, WorkerMain}
 import workers.common.AnnexMapper
 import workers.common.FileUtil
 import java.io.{File, PrintStream}
@@ -111,9 +111,12 @@ object DepsRunner extends WorkerMain[Unit] {
 
   override def init(args: Option[Array[String]]): Unit = ()
 
-  override def work(ctx: Unit, args: Array[String], out: PrintStream, workDir: Path, verbosity: Int): Unit = {
-    val workRequest = DepsRunnerRequest(workDir, ArgsUtil.parseArgsOrFailSafe(args, argParser, out))
-    InterruptUtil.throwIfInterrupted()
+  override def work(task: WorkTask[Unit]): Unit = {
+    val workRequest = DepsRunnerRequest(
+      task.workDir,
+      ArgsUtil.parseArgsOrFailSafe(task.args, argParser, task.output),
+    )
+    InterruptUtil.throwIfInterrupted(task.isCancelled)
 
     val groupLabelToJarPaths = workRequest.groups.map { group =>
       group.label -> group.jars
@@ -133,10 +136,10 @@ object DepsRunner extends WorkerMain[Unit] {
 
       potentialLabels.collect(groupLabelToJarPaths).flatten
     }
-    val readWriteMappers = AnnexMapper.mappers(workDir, isIncremental = false)
+    val readWriteMappers = AnnexMapper.mappers(task.workDir, isIncremental = false)
     val readMapper = readWriteMappers.getReadMapper()
 
-    InterruptUtil.throwIfInterrupted()
+    InterruptUtil.throwIfInterrupted(task.isCancelled)
     val usedPaths = Files
       .readAllLines(workRequest.usedDepsFile)
       .asScala
@@ -165,7 +168,7 @@ object DepsRunner extends WorkerMain[Unit] {
       Nil
     }
 
-    InterruptUtil.throwIfInterrupted()
+    InterruptUtil.throwIfInterrupted(task.isCancelled)
     val labelsToAdd = if (workRequest.checkDirect) {
       (usedPaths -- (workRequest.directDepLabels :++ workRequest.unusedDepWhitelist).flatMap(pathsForLabel))
         .flatMap { path =>
@@ -198,6 +201,6 @@ object DepsRunner extends WorkerMain[Unit] {
       throw new AnnexWorkerError(1, errorMessage.result())
     }
 
-    InterruptUtil.throwIfInterrupted()
+    InterruptUtil.throwIfInterrupted(task.isCancelled)
   }
 }

src/main/scala/higherkindness/rules_scala/workers/jacoco/instrumenter/JacocoInstrumenter.scala

@@ -5,7 +5,7 @@ import common.args.ArgsUtil
 import common.error.AnnexWorkerError
 import common.interrupt.InterruptUtil
 import common.sandbox.SandboxUtil
-import common.worker.WorkerMain
+import common.worker.{WorkTask, WorkerMain}
 import java.io.{BufferedInputStream, BufferedOutputStream, PrintStream}
 import java.net.URI
 import java.nio.file.Files
@@ -68,14 +68,17 @@ object JacocoInstrumenter extends WorkerMain[Unit] {
 
   override def init(args: Option[Array[String]]): Unit = ()
 
-  override def work(ctx: Unit, args: Array[String], out: PrintStream, workDir: Path, verbosity: Int): Unit = {
-    val workRequest = JacocoRequest(workDir, ArgsUtil.parseArgsOrFailSafe(args, argParser, out))
+  override def work(task: WorkTask[Unit]): Unit = {
+    val workRequest = JacocoRequest(
+      task.workDir,
+      ArgsUtil.parseArgsOrFailSafe(task.args, argParser, task.output),
+    )
 
     val jacoco = new Instrumenter(new OfflineInstrumentationAccessGenerator)
 
     workRequest.jars.foreach { case (inPath, outPath) =>
       Using.Manager { use =>
-        InterruptUtil.throwIfInterrupted()
+        InterruptUtil.throwIfInterrupted(task.isCancelled)
 
         val inFS = use(FileSystems.newFileSystem(inPath, null: ClassLoader))
         val outFS =
@@ -107,6 +110,6 @@ object JacocoInstrumenter extends WorkerMain[Unit] {
       }.get
     }
 
-    InterruptUtil.throwIfInterrupted()
+    InterruptUtil.throwIfInterrupted(task.isCancelled)
   }
 }