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.

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,71 @@
+package dotty.tools.dotc.util
+import scala.util.{Try, Failure, Success}
+import scala.annotation.tailrec
+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 {
+      @tailrec def recur(): Try[T] = result match
+        case Some(r) => r
+        case None =>
+          if exec.isRunning then
+            wait(1000 /*ms*/)
+            recur()
+          else
+            Failure(NoCompletion())
+      recur()
+    }
+    def complete(r: Try[T]): Unit = synchronized {
+      result = Some(r)
+      notifyAll()
+    }
+  end Future
+
+  class Executor[T] extends Thread:
+    private type WorkItem = (Future[T], () => T)
+
+    @volatile private var terminated = false
+    private var allScheduled = false
+    private val pending = new ArrayBuffer[WorkItem]
+
+    def isRunning: Boolean = !terminated
+
+    def schedule(op: () => T): Future[T] = synchronized {
+      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 =
+      @tailrec def recur(): Unit = nextPending() match
+        case Some((f, op)) =>
+          f.complete(Try(op()))
+          recur()
+        case None =>
+      try recur()
+      finally terminated = true
+  end Executor
+end concurrent
+
+
+