Initial structure and parallelization of parser

src/compiler/scala/tools/nsc/Global.scala

@@ -10,10 +10,12 @@ package nsc
 import java.io.{File, FileNotFoundException, IOException}
 import java.net.URL
 import java.nio.charset.{Charset, CharsetDecoder, IllegalCharsetNameException, UnsupportedCharsetException}
+import java.util.concurrent.atomic.AtomicInteger
+
 import scala.collection.{immutable, mutable}
 import io.{AbstractFile, Path, SourceReader}
-import reporters.Reporter
-import util.{ClassPath, returning}
+import reporters.{Reporter, StoreReporter}
+import util.{ClassPath, ThreadIdentityAwareThreadLocal, returning}
 import scala.reflect.ClassTag
 import scala.reflect.internal.util.{BatchSourceFile, NoSourceFile, ScalaClassLoader, ScriptSourceFile, SourceFile, StatisticsStatics}
 import scala.reflect.internal.pickling.PickleBuffer
@@ -26,12 +28,13 @@ import typechecker._
 import transform.patmat.PatternMatching
 import transform._
 import backend.{JavaPlatform, ScalaPrimitives}
-import backend.jvm.{GenBCode, BackendStats}
-import scala.concurrent.Future
+import backend.jvm.{BackendStats, GenBCode}
+import scala.concurrent.duration.Duration
+import scala.concurrent._
 import scala.language.postfixOps
 import scala.tools.nsc.ast.{TreeGen => AstTreeGen}
 import scala.tools.nsc.classpath._
-import scala.tools.nsc.profile.Profiler
+import scala.tools.nsc.profile.{Profiler, ThreadPoolFactory}
 
 class Global(var currentSettings: Settings, reporter0: Reporter)
  extends SymbolTable
@@ -75,16 +78,18 @@ class Global(var currentSettings: Settings, reporter0: Reporter)
 
  override def settings = currentSettings
 
- private[this] var currentReporter: Reporter = { reporter = reporter0 ; currentReporter }
+ private[this] val currentReporter: ThreadIdentityAwareThreadLocal[Reporter] =
+ ThreadIdentityAwareThreadLocal(new StoreReporter, reporter0)
+
+ def reporter: Reporter = { reporter = reporter0 ; currentReporter.get }
 
- def reporter: Reporter = currentReporter
  def reporter_=(newReporter: Reporter): Unit =
- currentReporter = newReporter match {
+ currentReporter.set(newReporter match {
  case _: reporters.ConsoleReporter | _: reporters.LimitingReporter => newReporter
  case _ if settings.maxerrs.isSetByUser && settings.maxerrs.value < settings.maxerrs.default =>
  new reporters.LimitingReporter(settings, newReporter)
  case _ => newReporter
- }
+ })
 
  /** Switch to turn on detailed type logs */
  var printTypings = settings.Ytyperdebug.value
@@ -385,45 +390,83 @@ class Global(var currentSettings: Settings, reporter0: Reporter)
  abstract class GlobalPhase(prev: Phase) extends Phase(prev) {
  phaseWithId(id) = this
 
- def run(): Unit = {
+ def apply(unit: CompilationUnit): Unit
+
+ def run() {
+ assertOnMainThread()
  echoPhaseSummary(this)
- currentRun.units foreach applyPhase
+ Await.result(Future.sequence(currentRun.units map processUnit), Duration.Inf)
  }
 
- def apply(unit: CompilationUnit): Unit
+ final def applyPhase(unit: CompilationUnit): Unit = Await.result(processUnit(unit), Duration.Inf)
 
- /** Is current phase cancelled on this unit? */
- def cancelled(unit: CompilationUnit) = {
- // run the typer only if in `createJavadoc` mode
- val maxJavaPhase = if (createJavadoc) currentRun.typerPhase.id else currentRun.namerPhase.id
- reporter.cancelled || unit.isJava && this.id > maxJavaPhase
+ implicit val ec: ExecutionContext = {
+ val threadPoolFactory = ThreadPoolFactory(Global.this, this)
+ val javaExecutor = threadPoolFactory.newUnboundedQueueFixedThreadPool(parallelThreads, "worker")
+ scala.concurrent.ExecutionContext.fromExecutorService(javaExecutor, (_) => ())
  }
 
- final def withCurrentUnit(unit: CompilationUnit)(task: => Unit): Unit = {
- if ((unit ne null) && unit.exists)
- lastSeenSourceFile = unit.source
-
+ private def processUnit(unit: CompilationUnit)(implicit ec: ExecutionContext): Future[Unit] = {
  if (settings.debug && (settings.verbose || currentRun.size < 5))
  inform("[running phase " + name + " on " + unit + "]")
- if (!cancelled(unit)) {
- currentRun.informUnitStarting(this, unit)
- try withCurrentUnitNoLog(unit)(task)
- finally currentRun.advanceUnit()
+
+ def runWithCurrentUnit(): Unit = {
+ val threadName = Thread.currentThread().getName
+ if (!threadName.contains("worker")) Thread.currentThread().setName(s"$threadName-worker")
+ val unit0 = currentUnit
+
+ try {
+ if ((unit ne null) && unit.exists) lastSeenSourceFile = unit.source
+ currentRun.currentUnit = unit
+ apply(unit)
+ } finally {
+ currentRun.currentUnit = unit0
+ currentRun.advanceUnit()
+ Thread.currentThread().setName(threadName)
+
+ // If we are on main thread it means there are no worker threads at all.
+ // That in turn means we were already using main reporter all the time, so there is nothing more to do.
+ // Otherwise we have to forward messages from worker thread reporter to main one.
+ reporter match {
+ case rep: StoreReporter =>
+ val mainReporter = currentReporter.main
+ mainReporter.synchronized {
+ rep.infos.foreach { info =>
+ info.severity.toString match {
+ case "INFO" => mainReporter.info(info.pos, info.msg, force = false)
+ case "WARNING" => mainReporter.warning(info.pos, info.msg)
+ case "ERROR" => mainReporter.error(info.pos, info.msg)
+ }
+ }
+ }
+ case _ =>
+ }
+ }
  }
+
+ if (cancelled(unit)) Future.successful(())
+ else if (isParallel) Future(runWithCurrentUnit())
+ else Future.fromTry(scala.util.Try(runWithCurrentUnit()))
  }
 
- final def withCurrentUnitNoLog(unit: CompilationUnit)(task: => Unit): Unit = {
- val unit0 = currentUnit
- try {
- currentRun.currentUnit = unit
- task
- } finally {
- //assert(currentUnit == unit)
- currentRun.currentUnit = unit0
- }
+ private def adjustWorkerThreadName(): Unit = {
+ val currentThreadName = Thread.currentThread().getName
  }
 
- final def applyPhase(unit: CompilationUnit) = withCurrentUnit(unit)(apply(unit))
+ private def parallelThreads = settings.YparallelThreads.value
+
+ private def isParallel = settings.YparallelPhases.containsPhase(this)
+
+ /** Is current phase cancelled on this unit? */
+ private def cancelled(unit: CompilationUnit) = {
+ // run the typer only if in `createJavadoc` mode
+ val maxJavaPhase = if (createJavadoc) currentRun.typerPhase.id else currentRun.namerPhase.id
+ reporter.cancelled || unit.isJava && this.id > maxJavaPhase
+ }
+
+ private def assertOnMainThread(): Unit = {
+ assert("main".equals(Thread.currentThread().getName), "")
+ }
  }
 
  // phaseName = "parser"
@@ -953,7 +996,9 @@ class Global(var currentSettings: Settings, reporter0: Reporter)
  * of what file was being compiled when it broke. Since I really
  * really want to know, this hack.
  */
- protected var lastSeenSourceFile: SourceFile = NoSourceFile
+ protected var _lastSeenSourceFile: ThreadIdentityAwareThreadLocal[SourceFile] = ThreadIdentityAwareThreadLocal(NoSourceFile)
+ @inline protected def lastSeenSourceFile: SourceFile = _lastSeenSourceFile.get
+ @inline protected def lastSeenSourceFile_=(source: SourceFile): Unit = _lastSeenSourceFile.set(source)
 
  /** Let's share a lot more about why we crash all over the place.
  * People will be very grateful.
@@ -1090,7 +1135,9 @@ class Global(var currentSettings: Settings, reporter0: Reporter)
  */
  var isDefined = false
  /** The currently compiled unit; set from GlobalPhase */
- var currentUnit: CompilationUnit = NoCompilationUnit
+ private final val _currentUnit: ThreadIdentityAwareThreadLocal[CompilationUnit] = ThreadIdentityAwareThreadLocal(NoCompilationUnit, NoCompilationUnit)
+ def currentUnit: CompilationUnit = _currentUnit.get
+ def currentUnit_=(unit: CompilationUnit): Unit = _currentUnit.set(unit)
 
  val profiler: Profiler = Profiler(settings)
  keepPhaseStack = settings.log.isSetByUser
@@ -1128,8 +1175,8 @@ class Global(var currentSettings: Settings, reporter0: Reporter)
  /** A map from compiled top-level symbols to their picklers */
  val symData = new mutable.AnyRefMap[Symbol, PickleBuffer]
 
- private var phasec: Int  = 0 // phases completed
- private var unitc: Int = 0 // units completed this phase
+ private var phasec: Int = 0  // phases completed
+ private final val unitc: AtomicInteger = new AtomicInteger(0) // units completed this phase
 
  def size = unitbuf.size
  override def toString = "scalac Run for:\n " + compiledFiles.toList.sorted.mkString("\n ")
@@ -1250,22 +1297,22 @@ class Global(var currentSettings: Settings, reporter0: Reporter)
  * (for progress reporting)
  */
  def advancePhase(): Unit = {
- unitc = 0
+ unitc.set(0)
  phasec += 1
  refreshProgress()
  }
  /** take note that a phase on a unit is completed
  * (for progress reporting)
  */
  def advanceUnit(): Unit = {
- unitc += 1
+ unitc.incrementAndGet()
  refreshProgress()
  }
 
  // for sbt
  def cancel(): Unit = { reporter.cancelled = true }
 
- private def currentProgress = (phasec * size) + unitc
+ private def currentProgress = (phasec * size) + unitc.get()
  private def totalProgress = (phaseDescriptors.size - 1) * size // -1: drops terminal phase
  private def refreshProgress() = if (size > 0) progress(currentProgress, totalProgress)
 

src/compiler/scala/tools/nsc/ast/Positions.scala

@@ -5,7 +5,6 @@ trait Positions extends scala.reflect.internal.Positions {
  self: Global =>
 
  class ValidatingPosAssigner extends PosAssigner {
- var pos: Position = _
  override def traverse(t: Tree): Unit = {
  if (t eq EmptyTree) ()
  else if (t.pos == NoPosition) super.traverse(t setPos pos)

src/compiler/scala/tools/nsc/settings/ScalaSettings.scala

@@ -143,7 +143,10 @@ trait ScalaSettings extends AbsScalaSettings
  val reporter = StringSetting ("-Xreporter", "classname", "Specify a custom reporter for compiler messages.", "scala.tools.nsc.reporters.ConsoleReporter")
  val source = ScalaVersionSetting ("-Xsource", "version", "Treat compiler input as Scala source for the specified version, see scala/bug#8126.", initial = ScalaVersion("2.13"))
 
- val XnoPatmatAnalysis = BooleanSetting ("-Xno-patmat-analysis", "Don't perform exhaustivity/unreachability analysis. Also, ignore @switch annotation.")
+ val XnoPatmatAnalysis = BooleanSetting ("-Xno-patmat-analysis", "Don't perform exhaustivity/unreachability analysis. Also, ignore @switch annotation.")
+
+ val YparallelPhases = PhasesSetting ("-Yparallel-phases", "Which phases to run in parallel")
+ val YparallelThreads = IntSetting ("-Yparallel-threads", "worker threads for parallel compilation", 4, Some((0,64)), _ => None )
 
  val XmixinForceForwarders = ChoiceSetting(
  name = "-Xmixin-force-forwarders",

src/compiler/scala/tools/nsc/util/ThreadIdentityAwareThreadLocal.scala

@@ -0,0 +1,39 @@
+package scala.tools.nsc.util
+
+object ThreadIdentityAwareThreadLocal {
+ def apply[T](valueOnWorker: => T) = new ThreadIdentityAwareThreadLocal[T](valueOnWorker, null.asInstanceOf[T])
+ def apply[T](valueOnWorker: => T, valueOnMain: => T) = new ThreadIdentityAwareThreadLocal[T](valueOnWorker, valueOnMain)
+}
+
+// `ThreadIdentityAwareThreadLocal` allows us to have different (sub)type of values on main and worker threads.
+// It's useful in cases like reporter, when on workers we want to just store messages and on main we want to print them,
+// but also in the cases when we do not expect some value to be read/write on the main thread,
+// and we want to discover violations of that rule.
+class ThreadIdentityAwareThreadLocal[T](valueOnWorker: => T, valueOnMain: => T) {
+ var main: T = valueOnMain
+
+ private val worker: ThreadLocal[T] = new ThreadLocal[T] {
+ override def initialValue(): T = valueOnWorker
+ }
+
+ // That logic may look a little bit funky because we need to consider cases
+ // where there is only one thread which is both main and worker at a given time.
+ private def isOnMainThread: Boolean = {
+ val currentThreadName = Thread.currentThread().getName
+ def isMainDefined = currentThreadName.startsWith("main") && valueOnMain != null
+ def isWorkerDefined = currentThreadName.contains("-worker")
+
+ assert(isMainDefined || isWorkerDefined, "Variable cannot be accessed on the main thread")
+
+ isMainDefined
+ }
+
+ def get: T = if (isOnMainThread) main else worker.get()
+
+ def set(value: T): Unit = if (isOnMainThread) main = value else worker.set(value)
+
+ def reset(): Unit = {
+ main = valueOnMain
+ worker.set(valueOnWorker)
+ }
+}

src/reflect/scala/reflect/api/Trees.scala

@@ -6,6 +6,8 @@ package scala
 package reflect
 package api
 
+import scala.reflect.runtime.ThreadIdentityAwareThreadLocal
+
 /**
  * <span class="badge badge-red" style="float: right;">EXPERIMENTAL</span>
  *
@@ -49,8 +51,7 @@ package api
  * @groupprio Factories 1
  * @groupname Copying Tree Copying
  * @groupprio Copying 1
- *
- * @contentDiagram hideNodes "*Api"
+ * @contentDiagram hideNodes "*Api"
  * @group ReflectionAPI
  */
 trait Trees { self: Universe =>
@@ -2463,7 +2464,9 @@ trait Trees { self: Universe =>
  * @group Traversal
  */
  class Traverser {
- protected[scala] var currentOwner: Symbol = rootMirror.RootClass
+ protected[scala] def currentOwner: Symbol = _currentOwner.get
+ protected[scala] def currentOwner_=(sym: Symbol): Unit = _currentOwner.set(sym)
+ private val _currentOwner: ThreadIdentityAwareThreadLocal[Symbol] = ThreadIdentityAwareThreadLocal[Symbol](rootMirror.RootClass)
 
  /** Traverse something which Trees contain, but which isn't a Tree itself. */
  def traverseName(name: Name): Unit = ()
@@ -2535,7 +2538,9 @@ trait Trees { self: Universe =>
  val treeCopy: TreeCopier = newLazyTreeCopier
 
  /** The current owner symbol. */
- protected[scala] var currentOwner: Symbol = rootMirror.RootClass
+ protected[scala] def currentOwner: Symbol = _currentOwner.get
+ protected[scala] def currentOwner_=(sym: Symbol): Unit = _currentOwner.set(sym)
+ private val _currentOwner: ThreadIdentityAwareThreadLocal[Symbol] = ThreadIdentityAwareThreadLocal[Symbol](rootMirror.RootClass)
 
  /** The enclosing method of the currently transformed tree. */
  protected def currentMethod = {

src/reflect/scala/reflect/internal/Names.scala

@@ -27,7 +27,7 @@ trait Names extends api.Names {
  // detect performance regressions.
  //
  // Discussion: https://groups.google.com/forum/#!search/biased$20scala-internals/scala-internals/0cYB7SkJ-nM/47MLhsgw8jwJ
- protected def synchronizeNames: Boolean = false
+ protected def synchronizeNames: Boolean = true
  private val nameLock: Object = new Object
 
  /** Memory to store all names sequentially. */

src/reflect/scala/reflect/internal/Positions.scala

@@ -5,6 +5,7 @@ package internal
 import scala.collection.mutable
 import util._
 import scala.collection.mutable.ListBuffer
+import scala.reflect.runtime.ThreadIdentityAwareThreadLocal
 
 /** Handling range positions
  * atPos, the main method in this trait, will add positions to a tree,
@@ -278,12 +279,13 @@ trait Positions extends api.Positions { self: SymbolTable =>
  }
 
  trait PosAssigner extends InternalTraverser {
- var pos: Position
+ private val _pos = ThreadIdentityAwareThreadLocal[Position](NoPosition)
+ @inline def pos: Position = _pos.get
+ @inline def pos_=(position: Position): Unit = _pos.set(position)
  }
  protected[this] lazy val posAssigner: PosAssigner = new DefaultPosAssigner
 
  protected class DefaultPosAssigner extends PosAssigner {
- var pos: Position = _
  override def traverse(t: Tree) {
  if (!t.canHaveAttrs) ()
  else if (t.pos == NoPosition) {

src/reflect/scala/reflect/internal/Reporting.scala

@@ -79,6 +79,8 @@ import util.Position
 abstract class Reporter {
  protected def info0(pos: Position, msg: String, severity: Severity, force: Boolean): Unit
 
+ def infoRaw(pos: Position, msg: String, severity: Severity, force: Boolean): Unit = info0(pos, msg, severity, force)
+
  def echo(pos: Position, msg: String): Unit = info0(pos, msg, INFO, force = true)
  def warning(pos: Position, msg: String): Unit = info0(pos, msg, WARNING, force = false)
  def error(pos: Position, msg: String): Unit = info0(pos, msg, ERROR, force = false)