Use home-brewed futures for parallel pickling

Uses just one thread for the rest of pickling. One thread is sufficient since
there is not that much to do and we have time until the backend finishes.

We might want to partially revise that decision when we support pipelined
computation. In that case producing tasty early could be a win. But even in
that case we might want to fine-tune the number of worker threads instead
of relying on some executor. Adding more workers is easy in the new design.

Author: odersky

compiler/src/dotty/tools/dotc/transform/Pickler.scala

@@ -14,8 +14,8 @@ import Symbols._
 import Flags.Module
 import reporting.{ThrowingReporter, Profile, Message}
 import collection.mutable
-import scala.concurrent.{Future, Await, ExecutionContext}
-import scala.concurrent.duration.Duration
+import util.concurrent.{Executor, Future}
+import compiletime.uninitialized
 
 object Pickler {
   val name: String = "pickler"
@@ -70,6 +70,11 @@ class Pickler extends Phase {
         body(scratch)
       }
 
+  private val executor = Executor[Array[Byte]]()
+
+  private def useExecutor(using Context) =
+    Pickler.ParallelPickling && !ctx.settings.YtestPickler.value
+
   override def run(using Context): Unit = {
     val unit = ctx.compilationUnit
     pickling.println(i"unpickling in run ${ctx.runId}")
@@ -123,10 +128,10 @@ class Pickler extends Phase {
        *  function value.
        */
       val demandPickled: () => Array[Byte] =
-        if Pickler.ParallelPickling && !ctx.settings.YtestPickler.value then
-          val futurePickled = Future(computePickled())(using ExecutionContext.global)
+        if useExecutor then
+          val futurePickled = executor.schedule(computePickled)
           () =>
-            try Await.result(futurePickled, Duration.Inf)
+            try futurePickled.force.get
             finally reportPositionWarnings()
         else
           val pickled = computePickled()
@@ -139,7 +144,13 @@ class Pickler extends Phase {
   }
 
   override def runOn(units: List[CompilationUnit])(using Context): List[CompilationUnit] = {
-    val result = super.runOn(units)
+    val result =
+      if useExecutor then
+        executor.start()
+        try super.runOn(units)
+        finally executor.close()
+      else
+        super.runOn(units)
     if ctx.settings.YtestPickler.value then
       val ctx2 = ctx.fresh.setSetting(ctx.settings.YreadComments, true)
       testUnpickler(

compiler/src/dotty/tools/dotc/util/concurrent.scala

@@ -0,0 +1,62 @@
+package dotty.tools.dotc.util
+import scala.util.{Try, Failure, Success}
+import scala.collection.mutable.ArrayBuffer
+
+object concurrent:
+
+  class NoCompletion extends RuntimeException
+
+  class Future[T](exec: Executor[T]):
+    private var result: Option[Try[T]] = None
+    def force: Try[T] = synchronized {
+      while result.isEmpty && exec.isAlive do wait(1000 /*ms*/)
+      result.getOrElse(Failure(NoCompletion()))
+    }
+    def complete(r: Try[T]): Unit = synchronized {
+      result = Some(r)
+      notifyAll()
+    }
+  end Future
+
+  class Executor[T] extends Thread:
+    private type WorkItem = (Future[T], () => T)
+
+    private var allScheduled = false
+    private val pending = new ArrayBuffer[WorkItem]
+
+    def schedule(op: () => T): Future[T] = synchronized {
+      assert(!allScheduled)
+      val f = Future[T](this)
+      pending += ((f, op))
+      notifyAll()
+      f
+    }
+
+    def close(): Unit = synchronized {
+      allScheduled = true
+      notifyAll()
+    }
+
+    private def nextPending(): Option[WorkItem] = synchronized {
+      while pending.isEmpty && !allScheduled do wait(1000 /*ms*/)
+      if pending.isEmpty then None
+      else
+        val item = pending.head
+        pending.dropInPlace(1)
+        Some(item)
+    }
+
+    override def run(): Unit =
+      while
+        nextPending() match
+          case Some((f, op)) =>
+            f.complete(Try(op()))
+            true
+          case None =>
+            false
+      do ()
+  end Executor
+end concurrent
+
+
+