Parallel analyzer

gzm0 · gzm0 · commit 41f178df06c7 · 2023-08-20T09:42:40.000+02:00
op                            variant  `median(t_ns)`
  &lt;fct&gt;                         &lt;fct&gt;             &lt;dbl&gt;
1 Linker: Compute reachability  main          323617749
2 Linker: Compute reachability  parallel      163542533
3 Refiner: Compute reachability main          255269676
4 Refiner: Compute reachability parallel      170701844
diff --git a/linker/shared/src/main/scala/org/scalajs/linker/analyzer/Analyzer.scala b/linker/shared/src/main/scala/org/scalajs/linker/analyzer/Analyzer.scala
@@ -65,7 +65,7 @@ final class Analyzer(config: CommonPhaseConfig, initial: Boolean,
 
   private[this] val _errors = new AtomicReference[List[Error]](Nil)
 
-  private var workQueue: WorkQueue = _
+  private var workTracker: WorkTracker = _
 
   private val fromAnalyzer = FromCore("analyzer")
 
@@ -78,13 +78,14 @@ final class Analyzer(config: CommonPhaseConfig, initial: Boolean,
 
     infoLoader.update(logger)
 
-    workQueue = new WorkQueue(ec)
+    workTracker = new WorkTracker
     classLoader = new ClassLoader
 
     loadObjectClass(() => loadEverything(moduleInitializers, symbolRequirements))
 
-    workQueue.join()
-      .map(_ => postLoad(moduleInitializers, logger))(ec)
+    workTracker
+      .future
+      .map(_ => postLoad(moduleInitializers, logger))
       .andThen { case _ => infoLoader.cleanAfterRun() }
   }
 
@@ -93,7 +94,7 @@ final class Analyzer(config: CommonPhaseConfig, initial: Boolean,
 
   private def resetState(): Unit = {
     objectClassInfo = null
-    workQueue = null
+    workTracker = null
     _errors.set(Nil)
     classLoader = null
     _topLevelExportInfos.clear()
@@ -331,7 +332,7 @@ final class Analyzer(config: CommonPhaseConfig, initial: Boolean,
 
   private def lookupClass(className: ClassName)(
       onSuccess: ClassInfo => Unit)(implicit from: From): Unit = {
-    workQueue.enqueue(classLoader.lookupClass(className)) {
+    workTracker.track(classLoader.lookupClass(className)) {
       case info: ClassInfo =>
         info.link()
         onSuccess(info)
@@ -862,13 +863,13 @@ final class Analyzer(config: CommonPhaseConfig, initial: Boolean,
           ()
 
         case onlyCandidate :: Nil =>
-          // Fast path that does not require workQueue.enqueue
+          // Fast path that does not require workTracker.track
           val proxy = createReflProxy(proxyName, onlyCandidate.methodName)
           onSuccess(proxy)
 
         case _ =>
           val targetFuture = computeMostSpecificProxyMatch(candidates)
-          workQueue.enqueue(targetFuture) { reflectiveTarget =>
+          workTracker.track(targetFuture) { reflectiveTarget =>
             val proxy = createReflProxy(proxyName, reflectiveTarget.methodName)
             onSuccess(proxy)
           }
@@ -934,7 +935,7 @@ final class Analyzer(config: CommonPhaseConfig, initial: Boolean,
 
       // Starting here, we just do data juggling, so it can run on any thread.
       locally {
-        implicit val iec = workQueue.ec
+        implicit val iec = workTracker.ec
 
         val hasMoreSpecific = Future.traverse(specificityChecks)(
             checks => Future.sequence(checks).map(_.contains(true)))
@@ -1555,58 +1556,27 @@ object Analyzer {
   private val getSuperclassMethodName =
     MethodName("getSuperclass", Nil, ClassRef(ClassClass))
 
-  private class WorkQueue(val ec: ExecutionContext) {
-    private val queue = new ConcurrentLinkedQueue[() => Unit]()
-    private val working = new AtomicBoolean(false)
+  private class WorkTracker(implicit val ec: ExecutionContext) {
     private val pending = new AtomicInteger(0)
     private val promise = Promise[Unit]()
 
-    def enqueue[T](fut: Future[T])(onSuccess: T => Unit): Unit = {
+    def track[T](fut: Future[T])(onSuccess: T => Unit): Unit = {
       val got = pending.incrementAndGet()
       assert(got > 0)
 
-      fut.onComplete {
-        case Success(r) =>
-          queue.add(() => onSuccess(r))
-          tryDoWork()
-
-        case Failure(t) =>
-          promise.tryFailure(t)
-      } (ec)
-    }
-
-    def join(): Future[Unit] = {
-      tryDoWork()
-      promise.future
+      fut.map(onSuccess).onComplete {
+        case Success(_) => taskDone()
+        case Failure(t) => promise.tryFailure(t)
+      }
     }
 
-    @tailrec
-    private def tryDoWork(): Unit = {
-      if (!working.getAndSet(true)) {
-        while (!queue.isEmpty) {
-          try {
-            val work = queue.poll()
-            work()
-          } catch {
-            case t: Throwable => promise.tryFailure(t)
-          }
-
-          pending.decrementAndGet()
-        }
-
-        if (pending.compareAndSet(0, -1)) {
-          assert(queue.isEmpty)
+    private def taskDone(): Unit = {
+      if (pending.decrementAndGet() == 0) {
+        if (pending.compareAndSet(0, -1))
           promise.trySuccess(())
-        }
-
-        working.set(false)
-
-        /* Another thread might have inserted work in the meantime but not yet
-         * seen that we released the lock. Try and work steal again if this
-         * happens.
-         */
-        if (!queue.isEmpty) tryDoWork()
       }
     }
+
+    def future: Future[Unit] = promise.future
   }
 }
