diff --git a/compiler/src/dotty/tools/backend/jvm/BTypes.scala b/compiler/src/dotty/tools/backend/jvm/BTypes.scala
index 5539bf44aa17..ba5e3e360e88 100644
--- a/compiler/src/dotty/tools/backend/jvm/BTypes.scala
+++ b/compiler/src/dotty/tools/backend/jvm/BTypes.scala
@@ -31,8 +31,7 @@ abstract class BTypes { self =>
    * Concurrent because stack map frames are computed when in the class writer, which might run
    * on multiple classes concurrently.
    */
-  protected def classBTypeFromInternalNameMap: collection.concurrent.Map[String, ClassBType]
-    // NOTE: Should be a lazy val but scalac does not allow abstract lazy vals (dotty does)
+  protected lazy val classBTypeFromInternalNameMap: collection.concurrent.Map[String, ClassBType]
 
   /**
    * Obtain a previously constructed ClassBType for a given internal name.
diff --git a/compiler/src/dotty/tools/backend/jvm/BTypesFromSymbols.scala b/compiler/src/dotty/tools/backend/jvm/BTypesFromSymbols.scala
index 884dd19ee64f..dc5cf48de6fe 100644
--- a/compiler/src/dotty/tools/backend/jvm/BTypesFromSymbols.scala
+++ b/compiler/src/dotty/tools/backend/jvm/BTypesFromSymbols.scala
@@ -55,14 +55,15 @@ class BTypesFromSymbols[I <: DottyBackendInterface](val int: I, val frontendAcce
       (classSym != defn.NothingClass && classSym != defn.NullClass),
       s"Cannot create ClassBType for special class symbol ${classSym.showFullName}")
 
-    convertedClasses.getOrElse(classSym, {
-      val internalName = classSym.javaBinaryName
-      // We first create and add the ClassBType to the hash map before computing its info. This
-      // allows initializing cylic dependencies, see the comment on variable ClassBType._info.
-      val classBType = new ClassBType(internalName)
-      convertedClasses(classSym) = classBType
-      setClassInfo(classSym, classBType)
-    })
+    convertedClasses.synchronized:
+      convertedClasses.getOrElse(classSym, {
+        val internalName = classSym.javaBinaryName
+        // We first create and add the ClassBType to the hash map before computing its info. This
+        // allows initializing cylic dependencies, see the comment on variable ClassBType._info.
+        val classBType = new ClassBType(internalName)
+        convertedClasses(classSym) = classBType
+        setClassInfo(classSym, classBType)
+      })
   }
 
   final def mirrorClassBTypeFromSymbol(moduleClassSym: Symbol): ClassBType = {
diff --git a/compiler/src/dotty/tools/backend/jvm/BackendUtils.scala b/compiler/src/dotty/tools/backend/jvm/BackendUtils.scala
index 2eaaccdd441d..d781479fdf1e 100644
--- a/compiler/src/dotty/tools/backend/jvm/BackendUtils.scala
+++ b/compiler/src/dotty/tools/backend/jvm/BackendUtils.scala
@@ -103,12 +103,10 @@ class BackendUtils(val postProcessor: PostProcessor) {
     // stack map frames and invokes the `getCommonSuperClass` method. This method expects all
     // ClassBTypes mentioned in the source code to exist in the map.
 
-    val serlamObjDesc = MethodBType(jliSerializedLambdaRef :: Nil, ObjectRef).descriptor
-
-    val mv = cw.visitMethod(ACC_PRIVATE + ACC_STATIC + ACC_SYNTHETIC, "$deserializeLambda$", serlamObjDesc, null, null)
+    val mv = cw.visitMethod(ACC_PRIVATE + ACC_STATIC + ACC_SYNTHETIC, "$deserializeLambda$", serializedLamdaObjDesc, null, null)
     def emitLambdaDeserializeIndy(targetMethods: Seq[Handle]): Unit = {
       mv.visitVarInsn(ALOAD, 0)
-      mv.visitInvokeDynamicInsn("lambdaDeserialize", serlamObjDesc, jliLambdaDeserializeBootstrapHandle, targetMethods: _*)
+      mv.visitInvokeDynamicInsn("lambdaDeserialize", serializedLamdaObjDesc, jliLambdaDeserializeBootstrapHandle, targetMethods: _*)
     }
 
     val targetMethodGroupLimit = 255 - 1 - 3 // JVM limit. See See MAX_MH_ARITY in CallSite.java
@@ -133,6 +131,11 @@ class BackendUtils(val postProcessor: PostProcessor) {
     mv.visitInsn(ARETURN)
   }
 
+  private lazy val serializedLamdaObjDesc = {
+    import coreBTypes.{ObjectRef, jliSerializedLambdaRef}
+    MethodBType(jliSerializedLambdaRef :: Nil, ObjectRef).descriptor
+  }
+
   /**
    * Visit the class node and collect all referenced nested classes.
    */
diff --git a/compiler/src/dotty/tools/backend/jvm/ClassfileWriter.scala b/compiler/src/dotty/tools/backend/jvm/ClassfileWriter.scala
deleted file mode 100644
index 08e84de92dca..000000000000
--- a/compiler/src/dotty/tools/backend/jvm/ClassfileWriter.scala
+++ /dev/null
@@ -1,142 +0,0 @@
-package dotty.tools.backend.jvm
-
-import java.io.{DataOutputStream, IOException, PrintWriter, StringWriter}
-import java.nio.file.Files
-import java.util.jar.Attributes.Name
-
-import scala.tools.asm.ClassReader
-import scala.tools.asm.tree.ClassNode
-import dotty.tools.io.*
-import dotty.tools.dotc.core.Decorators.*
-import dotty.tools.dotc.util.NoSourcePosition
-import java.nio.charset.StandardCharsets
-import java.nio.channels.ClosedByInterruptException
-import BTypes.InternalName
-import scala.language.unsafeNulls
-
-class ClassfileWriter(frontendAccess: PostProcessorFrontendAccess) {
-  import frontendAccess.{backendReporting, compilerSettings}
-
-  // if non-null, classfiles are additionally written to this directory
-  private val dumpOutputDir: AbstractFile = getDirectoryOrNull(compilerSettings.dumpClassesDirectory)
-
-  // if non-null, classfiles are written to a jar instead of the output directory
-  private val jarWriter: JarWriter | Null = compilerSettings.outputDirectory match {
-    case jar: JarArchive =>
-      val mainClass = compilerSettings.mainClass.orElse {
-        // If no main class was specified, see if there's only one
-        // entry point among the classes going into the jar.
-        frontendAccess.getEntryPoints match {
-          case name :: Nil =>
-            backendReporting.log(i"Unique entry point: setting Main-Class to $name")
-            Some(name)
-          case names =>
-            if names.isEmpty then backendReporting.warning(em"No Main-Class designated or discovered.")
-            else backendReporting.warning(em"No Main-Class due to multiple entry points:\n  ${names.mkString("\n  ")}")
-            None
-        }
-      }
-      jar.underlyingSource.map{ source =>
-        if jar.isEmpty then
-          val jarMainAttrs = mainClass.map(Name.MAIN_CLASS -> _).toList
-          new Jar(source.file).jarWriter(jarMainAttrs: _*)
-        else
-          // Writing to non-empty JAR might be an undefined behaviour, e.g. in case if other files where
-          // created using `AbstractFile.bufferedOutputStream`instead of JarWritter
-          backendReporting.warning(em"Tried to write to non-empty JAR: $source")
-          null
-      }.orNull
-
-    case _ => null
-  }
-
-  private def getDirectoryOrNull(dir: Option[String]): AbstractFile =
-    dir.map(d => new PlainDirectory(Directory(d))).orNull
-
-  private def getFile(base: AbstractFile, clsName: String, suffix: String): AbstractFile = {
-    if (base.file != null) {
-      fastGetFile(base, clsName, suffix)
-    } else {
-      def ensureDirectory(dir: AbstractFile): AbstractFile =
-        if (dir.isDirectory) dir
-        else throw new FileConflictException(s"${base.path}/$clsName$suffix: ${dir.path} is not a directory", dir)
-      var dir = base
-      val pathParts = clsName.split("[./]").toList
-      for (part <- pathParts.init) dir = ensureDirectory(dir) subdirectoryNamed part
-      ensureDirectory(dir) fileNamed pathParts.last + suffix
-    }
-  }
-
-  private def fastGetFile(base: AbstractFile, clsName: String, suffix: String) = {
-    val index = clsName.lastIndexOf('/')
-    val (packageName, simpleName) = if (index > 0) {
-      (clsName.substring(0, index), clsName.substring(index + 1))
-    } else ("", clsName)
-    val directory = base.file.toPath.resolve(packageName)
-    new PlainFile(Path(directory.resolve(simpleName + suffix)))
-  }
-
-  private def writeBytes(outFile: AbstractFile, bytes: Array[Byte]): Unit = {
-    if (outFile.file != null) {
-      val outPath = outFile.file.toPath
-      try Files.write(outPath, bytes)
-      catch {
-        case _: java.nio.file.NoSuchFileException =>
-          Files.createDirectories(outPath.getParent)
-          Files.write(outPath, bytes)
-      }
-    } else {
-      val out = new DataOutputStream(outFile.bufferedOutput)
-      try out.write(bytes, 0, bytes.length)
-      finally out.close()
-    }
-  }
-
-  def writeClass(className: InternalName, bytes: Array[Byte], sourceFile: AbstractFile): AbstractFile | Null = try {
-    // val writeStart = Statistics.startTimer(BackendStats.bcodeWriteTimer)
-    val outFile = writeToJarOrFile(className, bytes, ".class")
-    // Statistics.stopTimer(BackendStats.bcodeWriteTimer, writeStart)
-
-    if (dumpOutputDir != null) {
-      val dumpFile = getFile(dumpOutputDir, className, ".class")
-      writeBytes(dumpFile, bytes)
-    }
-    outFile
-  } catch {
-    case e: FileConflictException =>
-      backendReporting.error(em"error writing $className: ${e.getMessage}")
-      null
-    case e: java.nio.file.FileSystemException =>
-      if compilerSettings.debug then e.printStackTrace()
-      backendReporting.error(em"error writing $className: ${e.getClass.getName} ${e.getMessage}")
-      null
-  }
-
-  def writeTasty(className: InternalName, bytes: Array[Byte]): Unit =
-    writeToJarOrFile(className, bytes, ".tasty")
-
-  private def writeToJarOrFile(className: InternalName, bytes: Array[Byte], suffix: String): AbstractFile | Null = {
-    if jarWriter == null then
-      val outFolder = compilerSettings.outputDirectory
-      val outFile = getFile(outFolder, className, suffix)
-      try writeBytes(outFile, bytes)
-      catch case ex: ClosedByInterruptException =>
-        try outFile.delete() // don't leave an empty or half-written files around after an interrupt
-        catch case _: Throwable => ()
-        finally throw ex
-      outFile
-    else
-      val path = className + suffix
-      val out = jarWriter.newOutputStream(path)
-      try out.write(bytes, 0, bytes.length)
-      finally out.flush()
-      null
-  }
-
-  def close(): Unit = {
-    if (jarWriter != null) jarWriter.close()
-  }
-}
-
-/** Can't output a file due to the state of the file system. */
-class FileConflictException(msg: String, val file: AbstractFile) extends IOException(msg)
diff --git a/compiler/src/dotty/tools/backend/jvm/ClassfileWriters.scala b/compiler/src/dotty/tools/backend/jvm/ClassfileWriters.scala
new file mode 100644
index 000000000000..928d19c92041
--- /dev/null
+++ b/compiler/src/dotty/tools/backend/jvm/ClassfileWriters.scala
@@ -0,0 +1,288 @@
+package dotty.tools.backend.jvm
+
+import java.io.{DataOutputStream, IOException, BufferedOutputStream, FileOutputStream}
+import java.nio.ByteBuffer
+import java.nio.channels.{ClosedByInterruptException, FileChannel}
+import java.nio.charset.StandardCharsets.UTF_8
+import java.nio.file._
+import java.nio.file.attribute.FileAttribute
+import java.util
+import java.util.concurrent.ConcurrentHashMap
+import java.util.zip.{CRC32, Deflater, ZipEntry, ZipOutputStream}
+
+import dotty.tools.dotc.core.Contexts.*
+import dotty.tools.dotc.core.Decorators.em
+import dotty.tools.io.{AbstractFile, PlainFile}
+import dotty.tools.io.PlainFile.toPlainFile
+import BTypes.InternalName
+import scala.util.chaining._
+import dotty.tools.io.JarArchive
+
+import scala.language.unsafeNulls
+
+
+class ClassfileWriters(frontendAccess: PostProcessorFrontendAccess) {
+  type NullableFile =  AbstractFile | Null
+  import frontendAccess.{compilerSettings, backendReporting}
+
+  sealed trait TastyWriter {
+    def writeTasty(name: InternalName, bytes: Array[Byte], sourceFile: AbstractFile): Unit
+  }
+
+  /**
+   * The interface to writing classfiles. GeneratedClassHandler calls these methods to generate the
+   * directory and files that are created, and eventually calls `close` when the writing is complete.
+   *
+   * The companion object is responsible for constructing a appropriate and optimal implementation for
+   * the supplied settings.
+   *
+   * Operations are threadsafe.
+   */
+  sealed trait ClassfileWriter extends TastyWriter {
+    /**
+     * Write a classfile
+     */
+    def writeClass(name: InternalName, bytes: Array[Byte], sourceFile: AbstractFile): NullableFile
+
+
+    /**
+     * Close the writer. Behavior is undefined after a call to `close`.
+     */
+    def close(): Unit
+
+    protected def classRelativePath(className: InternalName, suffix: String = ".class"): String =
+      className.replace('.', '/').nn + suffix
+  }
+
+  object ClassfileWriter {
+    private def getDirectory(dir: String): Path = Paths.get(dir).nn
+
+    def apply(): ClassfileWriter = {
+      val jarManifestMainClass: Option[String] = compilerSettings.mainClass.orElse {
+        frontendAccess.getEntryPoints match {
+          case List(name) => Some(name)
+          case es =>
+            if es.isEmpty then backendReporting.log("No Main-Class designated or discovered.")
+            else backendReporting.log(s"No Main-Class due to multiple entry points:\n  ${es.mkString("\n  ")}")
+            None
+        }
+      }
+
+      // In Scala 2 depenening on cardinality of distinct output dirs MultiClassWriter could have been used
+      // In Dotty we always use single output directory
+      val basicClassWriter = new SingleClassWriter(
+        FileWriter(compilerSettings.outputDirectory, jarManifestMainClass)
+      )
+
+      val withAdditionalFormats =
+        compilerSettings.dumpClassesDirectory
+          .map(getDirectory)
+          .filter{path => Files.exists(path).tap{ok => if !ok then backendReporting.error(em"Output dir does not exist: ${path.toString}")}}
+          .map(out => FileWriter(out.toPlainFile, None))
+          .fold[ClassfileWriter](basicClassWriter)(new DebugClassWriter(basicClassWriter, _))
+
+      // val enableStats = settings.areStatisticsEnabled && settings.YaddBackendThreads.value == 1
+      // if (enableStats) new WithStatsWriter(withAdditionalFormats) else
+      withAdditionalFormats
+    }
+
+    private final class SingleClassWriter(underlying: FileWriter) extends ClassfileWriter {
+      override def writeClass(className: InternalName, bytes: Array[Byte], sourceFile: AbstractFile): NullableFile = {
+        underlying.writeFile(classRelativePath(className), bytes)
+      }
+      override def writeTasty(className: InternalName, bytes: Array[Byte], sourceFile: AbstractFile): Unit = {
+        underlying.writeFile(classRelativePath(className, ".tasty"), bytes)
+      }
+
+
+      override def close(): Unit = underlying.close()
+    }
+
+    private final class DebugClassWriter(basic: ClassfileWriter, dump: FileWriter) extends ClassfileWriter {
+      override def writeClass(className: InternalName, bytes: Array[Byte], sourceFile: AbstractFile): NullableFile = {
+        val outFile = basic.writeClass(className, bytes, sourceFile)
+        dump.writeFile(classRelativePath(className), bytes)
+        outFile
+      }
+
+      override def writeTasty(className: InternalName, bytes: Array[Byte], sourceFile: AbstractFile): Unit = {
+        basic.writeTasty(className, bytes, sourceFile)
+      }
+
+      override def close(): Unit = {
+        basic.close()
+        dump.close()
+      }
+    }
+  }
+
+  sealed trait FileWriter {
+    def writeFile(relativePath: String, bytes: Array[Byte]): NullableFile
+    def close(): Unit
+  }
+
+  object FileWriter {
+    def apply(file: AbstractFile, jarManifestMainClass: Option[String]): FileWriter =
+      if (file.isInstanceOf[JarArchive]) {
+        val jarCompressionLevel = compilerSettings.jarCompressionLevel
+        // Writing to non-empty JAR might be an undefined behaviour, e.g. in case if other files where
+        // created using `AbstractFile.bufferedOutputStream`instead of JarWritter
+        val jarFile = file.underlyingSource.getOrElse{
+          throw new IllegalStateException("No underlying source for jar")
+        }
+        assert(file.isEmpty, s"Unsafe writing to non-empty JAR: $jarFile")
+        new JarEntryWriter(jarFile, jarManifestMainClass, jarCompressionLevel)
+      }
+      else if (file.isVirtual) new VirtualFileWriter(file)
+      else if (file.isDirectory) new DirEntryWriter(file.file.toPath.nn)
+      else throw new IllegalStateException(s"don't know how to handle an output of $file [${file.getClass}]")
+  }
+
+  private final class JarEntryWriter(file: AbstractFile, mainClass: Option[String], compressionLevel: Int) extends FileWriter {
+    //keep these imports local - avoid confusion with scala naming
+    import java.util.jar.Attributes.Name.{MANIFEST_VERSION, MAIN_CLASS}
+    import java.util.jar.{JarOutputStream, Manifest}
+
+    val storeOnly = compressionLevel == Deflater.NO_COMPRESSION
+
+    val jarWriter: JarOutputStream = {
+      import scala.util.Properties._
+      val manifest = new Manifest
+      val attrs = manifest.getMainAttributes.nn
+      attrs.put(MANIFEST_VERSION, "1.0")
+      attrs.put(ScalaCompilerVersion, versionNumberString)
+      mainClass.foreach(c => attrs.put(MAIN_CLASS, c))
+
+      val jar = new JarOutputStream(new BufferedOutputStream(new FileOutputStream(file.file), 64000), manifest)
+      jar.setLevel(compressionLevel)
+      if (storeOnly) jar.setMethod(ZipOutputStream.STORED)
+      jar
+    }
+
+    lazy val crc = new CRC32
+
+    override def writeFile(relativePath: String, bytes: Array[Byte]): NullableFile = this.synchronized {
+      val entry = new ZipEntry(relativePath)
+      if (storeOnly) {
+        // When using compression method `STORED`, the ZIP spec requires the CRC and compressed/
+        // uncompressed sizes to be written before the data. The JarOutputStream could compute the
+        // values while writing the data, but not patch them into the stream after the fact. So we
+        // need to pre-compute them here. The compressed size is taken from size.
+        // https://stackoverflow.com/questions/1206970/how-to-create-uncompressed-zip-archive-in-java/5868403
+        // With compression method `DEFLATED` JarOutputStream computes and sets the values.
+        entry.setSize(bytes.length)
+        crc.reset()
+        crc.update(bytes)
+        entry.setCrc(crc.getValue)
+      }
+      jarWriter.putNextEntry(entry)
+      try jarWriter.write(bytes, 0, bytes.length)
+      finally jarWriter.flush()
+      null
+    }
+
+    override def close(): Unit = this.synchronized(jarWriter.close())
+  }
+
+  private final class DirEntryWriter(base: Path) extends FileWriter {
+    val builtPaths = new ConcurrentHashMap[Path, java.lang.Boolean]()
+    val noAttributes = Array.empty[FileAttribute[_]]
+    private val isWindows = scala.util.Properties.isWin
+
+    private def checkName(component: Path): Unit = if (isWindows) {
+      val specials = raw"(?i)CON|PRN|AUX|NUL|COM[1-9]|LPT[1-9]".r
+      val name = component.toString
+      def warnSpecial(): Unit = backendReporting.warning(em"path component is special Windows device: ${name}")
+      specials.findPrefixOf(name).foreach(prefix => if (prefix.length == name.length || name(prefix.length) == '.') warnSpecial())
+    }
+
+    def ensureDirForPath(baseDir: Path, filePath: Path): Unit = {
+      import java.lang.Boolean.TRUE
+      val parent = filePath.getParent
+      if (!builtPaths.containsKey(parent)) {
+        parent.iterator.forEachRemaining(checkName)
+        try Files.createDirectories(parent, noAttributes: _*)
+        catch {
+          case e: FileAlreadyExistsException =>
+            // `createDirectories` reports this exception if `parent` is an existing symlink to a directory
+            // but that's fine for us (and common enough, `scalac -d /tmp` on mac targets symlink).
+            if (!Files.isDirectory(parent))
+              throw new FileConflictException(s"Can't create directory $parent; there is an existing (non-directory) file in its path", e)
+        }
+        builtPaths.put(baseDir, TRUE)
+        var current = parent
+        while ((current ne null) && (null ne builtPaths.put(current, TRUE))) {
+          current = current.getParent
+        }
+      }
+      checkName(filePath.getFileName())
+    }
+
+    // the common case is that we are are creating a new file, and on MS Windows the create and truncate is expensive
+    // because there is not an options in the windows API that corresponds to this so the truncate is applied as a separate call
+    // even if the file is new.
+    // as this is rare, its best to always try to create a new file, and it that fails, then open with truncate if that fails
+
+    private val fastOpenOptions = util.EnumSet.of(StandardOpenOption.CREATE_NEW, StandardOpenOption.WRITE)
+    private val fallbackOpenOptions = util.EnumSet.of(StandardOpenOption.CREATE, StandardOpenOption.WRITE, StandardOpenOption.TRUNCATE_EXISTING)
+
+    override def writeFile(relativePath: String, bytes: Array[Byte]): NullableFile = {
+      val path = base.resolve(relativePath)
+      try {
+        ensureDirForPath(base, path)
+        val os = if (isWindows) {
+          try FileChannel.open(path, fastOpenOptions)
+          catch {
+            case _: FileAlreadyExistsException => FileChannel.open(path, fallbackOpenOptions)
+          }
+        } else FileChannel.open(path, fallbackOpenOptions)
+
+        try os.write(ByteBuffer.wrap(bytes), 0L)
+        catch {
+          case ex: ClosedByInterruptException =>
+            try Files.deleteIfExists(path) // don't leave a empty of half-written classfile around after an interrupt
+            catch { case _: Throwable => () }
+            throw ex
+        }
+        os.close()
+      } catch {
+        case e: FileConflictException =>
+          backendReporting.error(em"error writing ${path.toString}: ${e.getMessage}")
+        case e: java.nio.file.FileSystemException =>
+          if (compilerSettings.debug) e.printStackTrace()
+          backendReporting.error(em"error writing ${path.toString}: ${e.getClass.getName} ${e.getMessage}")
+      }
+      AbstractFile.getFile(path)
+    }
+
+    override def close(): Unit = ()
+  }
+
+  private final class VirtualFileWriter(base: AbstractFile) extends FileWriter {
+    private def getFile(base: AbstractFile, path: String): AbstractFile = {
+      def ensureDirectory(dir: AbstractFile): AbstractFile =
+        if (dir.isDirectory) dir
+        else throw new FileConflictException(s"${base.path}/${path}: ${dir.path} is not a directory")
+      val components = path.split('/')
+      var dir = base
+      for (i <- 0 until components.length - 1) dir = ensureDirectory(dir) subdirectoryNamed components(i).toString
+      ensureDirectory(dir) fileNamed components.last.toString
+    }
+
+    private def writeBytes(outFile: AbstractFile, bytes: Array[Byte]): Unit = {
+      val out = new DataOutputStream(outFile.bufferedOutput)
+      try out.write(bytes, 0, bytes.length)
+      finally out.close()
+    }
+
+    override def writeFile(relativePath: String, bytes: Array[Byte]):NullableFile = {
+      val outFile = getFile(base, relativePath)
+      writeBytes(outFile, bytes)
+      outFile
+    }
+    override def close(): Unit = ()
+  }
+
+  /** Can't output a file due to the state of the file system. */
+  class FileConflictException(msg: String, cause: Throwable = null) extends IOException(msg, cause)
+}
diff --git a/compiler/src/dotty/tools/backend/jvm/CodeGen.scala b/compiler/src/dotty/tools/backend/jvm/CodeGen.scala
index 4e00597d5b96..9572777095e0 100644
--- a/compiler/src/dotty/tools/backend/jvm/CodeGen.scala
+++ b/compiler/src/dotty/tools/backend/jvm/CodeGen.scala
@@ -41,7 +41,15 @@ class CodeGen(val int: DottyBackendInterface, val primitives: DottyPrimitives)(
 
   private lazy val mirrorCodeGen = Impl.JMirrorBuilder()
 
-  def genUnit(unit: CompilationUnit): GeneratedDefs = {
+  private def genBCode(using Context) = Phases.genBCodePhase.asInstanceOf[GenBCode]
+  private def postProcessor(using Context) = genBCode.postProcessor
+  private def generatedClassHandler(using Context) = genBCode.generatedClassHandler
+
+  /**
+   * Generate ASM ClassNodes for classes found in a compilation unit. The resulting classes are
+   * passed to the `GenBCode.generatedClassHandler`.
+   */
+  def genUnit(unit: CompilationUnit)(using ctx: Context): Unit = {
     val generatedClasses = mutable.ListBuffer.empty[GeneratedClass]
     val generatedTasty = mutable.ListBuffer.empty[GeneratedTasty]
 
@@ -50,25 +58,32 @@ class CodeGen(val int: DottyBackendInterface, val primitives: DottyPrimitives)(
         val sym = cd.symbol
         val sourceFile = unit.source.file
 
-        def registerGeneratedClass(classNode: ClassNode, isArtifact: Boolean): Unit =
-          generatedClasses += GeneratedClass(classNode, sourceFile, isArtifact, onFileCreated(classNode, sym, unit.source))
-
-        val plainC = genClass(cd, unit)
-        registerGeneratedClass(plainC, isArtifact = false)
 
-        val attrNode =
-          if !sym.isTopLevelModuleClass then plainC
-          else if sym.companionClass == NoSymbol then
-            val mirrorC = genMirrorClass(sym, unit)
-            registerGeneratedClass(mirrorC, isArtifact = true)
-            mirrorC
+        val mainClassNode = genClass(cd, unit)
+        val mirrorClassNode =
+          if !sym.isTopLevelModuleClass then null
+          else if sym.companionClass == NoSymbol then genMirrorClass(sym, unit)
           else
             report.log(s"No mirror class for module with linked class: ${sym.fullName}", NoSourcePosition)
-            plainC
+            null
 
         if sym.isClass then
-          genTastyAndSetAttributes(sym, attrNode)
+          val tastyAttrNode = if (mirrorClassNode ne null) mirrorClassNode else mainClassNode
+          genTastyAndSetAttributes(sym, tastyAttrNode)
+
+        def registerGeneratedClass(classNode: ClassNode, isArtifact: Boolean): Unit =
+          if classNode ne null then
+            generatedClasses += GeneratedClass(classNode,
+              sourceClassName = sym.javaClassName,
+              position = sym.srcPos.sourcePos,
+              isArtifact = isArtifact,
+              onFileCreated = onFileCreated(classNode, sym, unit.source)
+            )
+
+        registerGeneratedClass(mainClassNode, isArtifact = false)
+        registerGeneratedClass(mirrorClassNode, isArtifact = true)
       catch
+        case ex: InterruptedException => throw ex
         case ex: Throwable =>
           ex.printStackTrace()
           report.error(s"Error while emitting ${unit.source}\n${ex.getMessage}", NoSourcePosition)
@@ -99,26 +114,29 @@ class CodeGen(val int: DottyBackendInterface, val primitives: DottyPrimitives)(
         case EmptyTree => ()
         case PackageDef(_, stats) => stats foreach genClassDefs
         case ValDef(_, _, _) => () // module val not emitted
-        case td: TypeDef =>  genClassDef(td)
+        case td: TypeDef => frontendAccess.frontendSynch(genClassDef(td))
       }
 
     genClassDefs(unit.tpdTree)
-    GeneratedDefs(generatedClasses.toList, generatedTasty.toList)
+    generatedClassHandler.process(
+      GeneratedCompilationUnit(unit.source.file, generatedClasses.toList, generatedTasty.toList)
+    )
   }
 
   // Creates a callback that will be evaluated in PostProcessor after creating a file
-  private def onFileCreated(cls: ClassNode, claszSymbol: Symbol, sourceFile: util.SourceFile): AbstractFile => Unit = clsFile => {
+  private def onFileCreated(cls: ClassNode, claszSymbol: Symbol, sourceFile: util.SourceFile): AbstractFile => Unit = {
     val (fullClassName, isLocal) = atPhase(sbtExtractDependenciesPhase) {
       (ExtractDependencies.classNameAsString(claszSymbol), claszSymbol.isLocal)
     }
-
-    val className = cls.name.replace('/', '.')
-    if (ctx.compilerCallback != null)
-      ctx.compilerCallback.onClassGenerated(sourceFile, convertAbstractFile(clsFile), className)
-
-    ctx.withIncCallback: cb =>
-      if (isLocal) cb.generatedLocalClass(sourceFile, clsFile.jpath)
-      else cb.generatedNonLocalClass(sourceFile, clsFile.jpath, className, fullClassName)
+    clsFile => {
+      val className = cls.name.replace('/', '.')
+      if (ctx.compilerCallback != null)
+        ctx.compilerCallback.onClassGenerated(sourceFile, convertAbstractFile(clsFile), className)
+
+      ctx.withIncCallback: cb =>
+        if (isLocal) cb.generatedLocalClass(sourceFile, clsFile.jpath)
+        else cb.generatedNonLocalClass(sourceFile, clsFile.jpath, className, fullClassName)
+    }
   }
 
   /** Convert a `dotty.tools.io.AbstractFile` into a
@@ -132,48 +150,20 @@ class CodeGen(val int: DottyBackendInterface, val primitives: DottyPrimitives)(
     }
 
   private def genClass(cd: TypeDef, unit: CompilationUnit): ClassNode = {
-    val b = new Impl.PlainClassBuilder(unit)
+    val b = new Impl.SyncAndTryBuilder(unit) {}
     b.genPlainClass(cd)
-    val cls = b.cnode
-    checkForCaseConflict(cls.name, cd.symbol)
-    cls
+    b.cnode
   }
 
   private def genMirrorClass(classSym: Symbol, unit: CompilationUnit): ClassNode = {
-    val cls = mirrorCodeGen.genMirrorClass(classSym, unit)
-    checkForCaseConflict(cls.name, classSym)
-    cls
+    mirrorCodeGen.genMirrorClass(classSym, unit)
   }
 
-  private val lowerCaseNames = mutable.HashMap.empty[String, Symbol]
-  private def checkForCaseConflict(javaClassName: String, classSymbol: Symbol) = {
-    val lowerCaseName = javaClassName.toLowerCase
-    lowerCaseNames.get(lowerCaseName) match {
-      case None =>
-        lowerCaseNames.put(lowerCaseName, classSymbol)
-      case Some(dupClassSym) =>
-        // Order is not deterministic so we enforce lexicographic order between the duplicates for error-reporting
-        val (cl1, cl2) =
-          if (classSymbol.effectiveName.toString < dupClassSym.effectiveName.toString) (classSymbol, dupClassSym)
-          else (dupClassSym, classSymbol)
-        val same = classSymbol.effectiveName.toString == dupClassSym.effectiveName.toString
-        atPhase(typerPhase) {
-          if same then
-             // FIXME: This should really be an error, but then FromTasty tests fail
-            report.warning(s"${cl1.show} and ${cl2.showLocated} produce classes that overwrite one another", cl1.sourcePos)
-          else
-            report.warning(s"${cl1.show} differs only in case from ${cl2.showLocated}. " +
-              "Such classes will overwrite one another on case-insensitive filesystems.", cl1.sourcePos)
-        }
-    }
-  }
 
   sealed transparent trait ImplEarlyInit{
     val int: self.int.type = self.int
     val bTypes: self.bTypes.type = self.bTypes
     protected val primitives: DottyPrimitives = self.primitives
   }
-  object Impl extends ImplEarlyInit with BCodeSyncAndTry {
-    class PlainClassBuilder(unit: CompilationUnit) extends SyncAndTryBuilder(unit)
-  }
+  object Impl extends ImplEarlyInit with BCodeSyncAndTry
 }
diff --git a/compiler/src/dotty/tools/backend/jvm/CoreBTypes.scala b/compiler/src/dotty/tools/backend/jvm/CoreBTypes.scala
index 30ad6b29b9f0..0e65850d4798 100644
--- a/compiler/src/dotty/tools/backend/jvm/CoreBTypes.scala
+++ b/compiler/src/dotty/tools/backend/jvm/CoreBTypes.scala
@@ -58,6 +58,7 @@ abstract class CoreBTypesFromSymbols[I <: DottyBackendInterface] extends CoreBTy
   import bTypes._
   import int.given
   import DottyBackendInterface._
+  import frontendAccess.frontendSynch
   import dotty.tools.dotc.core.Contexts.Context
 
   /**
@@ -80,7 +81,7 @@ abstract class CoreBTypesFromSymbols[I <: DottyBackendInterface] extends CoreBTy
    * Map from primitive types to their boxed class type. Useful when pushing class literals onto the
    * operand stack (ldc instruction taking a class literal), see genConstant.
    */
-  lazy val boxedClassOfPrimitive: Map[PrimitiveBType, ClassBType] = Map(
+  lazy val boxedClassOfPrimitive: Map[PrimitiveBType, ClassBType] = frontendSynch(Map(
       UNIT   -> classBTypeFromSymbol(requiredClass[java.lang.Void]),
       BOOL   -> classBTypeFromSymbol(requiredClass[java.lang.Boolean]),
       BYTE   -> classBTypeFromSymbol(requiredClass[java.lang.Byte]),
@@ -90,7 +91,7 @@ abstract class CoreBTypesFromSymbols[I <: DottyBackendInterface] extends CoreBTy
       LONG   -> classBTypeFromSymbol(requiredClass[java.lang.Long]),
       FLOAT  -> classBTypeFromSymbol(requiredClass[java.lang.Float]),
       DOUBLE -> classBTypeFromSymbol(requiredClass[java.lang.Double])
-    )
+    ))
 
   lazy val boxedClasses: Set[ClassBType] = boxedClassOfPrimitive.values.toSet
 
@@ -116,6 +117,10 @@ abstract class CoreBTypesFromSymbols[I <: DottyBackendInterface] extends CoreBTy
     yield unboxMethodSym -> primitiveTypeMap(valueClassSym)
   }
 
+  // Used to synchronize initialization of Context dependent ClassBTypes which can be accessed from multiple-threads
+  // Unsychronized initialization might lead errors in either CodeGen or PostProcessor
+  inline private def synchClassBTypeFromSymbol(inline sym: Symbol) = frontendSynch(classBTypeFromSymbol(sym))
+
   /*
    * srNothingRef and srNullRef exist at run-time only. They are the bytecode-level manifestation (in
    * method signatures only) of what shows up as NothingClass (scala.Nothing) resp. NullClass (scala.Null) in Scala ASTs.
@@ -124,35 +129,35 @@ abstract class CoreBTypesFromSymbols[I <: DottyBackendInterface] extends CoreBTy
    * names of NothingClass and NullClass can't be emitted as-is.
    * TODO @lry Once there's a 2.11.3 starr, use the commented argument list. The current starr crashes on the type literal `scala.runtime.Nothing$`
    */
-  lazy val srNothingRef : ClassBType = classBTypeFromSymbol(requiredClass("scala.runtime.Nothing$"))
-  lazy val srNullRef    : ClassBType = classBTypeFromSymbol(requiredClass("scala.runtime.Null$"))
-
-  lazy val ObjectRef : ClassBType = classBTypeFromSymbol(defn.ObjectClass)
-  lazy val StringRef : ClassBType = classBTypeFromSymbol(defn.StringClass)
-
-  lazy val jlStringBuilderRef       : ClassBType = classBTypeFromSymbol(requiredClass[java.lang.StringBuilder])
-  lazy val jlStringBufferRef        : ClassBType = classBTypeFromSymbol(requiredClass[java.lang.StringBuffer])
-  lazy val jlCharSequenceRef        : ClassBType = classBTypeFromSymbol(requiredClass[java.lang.CharSequence])
-  lazy val jlClassRef               : ClassBType = classBTypeFromSymbol(requiredClass[java.lang.Class[_]])
-  lazy val jlThrowableRef           : ClassBType = classBTypeFromSymbol(defn.ThrowableClass)
-  lazy val jlCloneableRef           : ClassBType = classBTypeFromSymbol(defn.JavaCloneableClass)
-  lazy val jiSerializableRef        : ClassBType = classBTypeFromSymbol(requiredClass[java.io.Serializable])
-  lazy val jlClassCastExceptionRef  : ClassBType = classBTypeFromSymbol(requiredClass[java.lang.ClassCastException])
-  lazy val jlIllegalArgExceptionRef : ClassBType = classBTypeFromSymbol(requiredClass[java.lang.IllegalArgumentException])
-  lazy val jliSerializedLambdaRef   : ClassBType = classBTypeFromSymbol(requiredClass[java.lang.invoke.SerializedLambda])
-
-  lazy val srBoxesRuntimeRef: ClassBType = classBTypeFromSymbol(requiredClass[scala.runtime.BoxesRunTime])
-
-  private lazy val jliCallSiteRef            : ClassBType = classBTypeFromSymbol(requiredClass[java.lang.invoke.CallSite])
-  private lazy val jliLambdaMetafactoryRef   : ClassBType = classBTypeFromSymbol(requiredClass[java.lang.invoke.LambdaMetafactory])
-  private lazy val jliMethodHandleRef        : ClassBType = classBTypeFromSymbol(defn.MethodHandleClass)
-  private lazy val jliMethodHandlesLookupRef : ClassBType = classBTypeFromSymbol(defn.MethodHandlesLookupClass)
-  private lazy val jliMethodTypeRef          : ClassBType = classBTypeFromSymbol(requiredClass[java.lang.invoke.MethodType])
-  private lazy val jliStringConcatFactoryRef : ClassBType = classBTypeFromSymbol(requiredClass("java.lang.invoke.StringConcatFactory")) // since JDK 9
-
-  lazy val srLambdaDeserialize               : ClassBType = classBTypeFromSymbol(requiredClass[scala.runtime.LambdaDeserialize])
-
-  lazy val jliLambdaMetaFactoryMetafactoryHandle = new Handle(
+  lazy val srNothingRef : ClassBType = synchClassBTypeFromSymbol(requiredClass("scala.runtime.Nothing$"))
+  lazy val srNullRef    : ClassBType = synchClassBTypeFromSymbol(requiredClass("scala.runtime.Null$"))
+
+  lazy val ObjectRef : ClassBType = synchClassBTypeFromSymbol(defn.ObjectClass)
+  lazy val StringRef : ClassBType = synchClassBTypeFromSymbol(defn.StringClass)
+
+  lazy val jlStringBuilderRef       : ClassBType = synchClassBTypeFromSymbol(requiredClass[java.lang.StringBuilder])
+  lazy val jlStringBufferRef        : ClassBType = synchClassBTypeFromSymbol(requiredClass[java.lang.StringBuffer])
+  lazy val jlCharSequenceRef        : ClassBType = synchClassBTypeFromSymbol(requiredClass[java.lang.CharSequence])
+  lazy val jlClassRef               : ClassBType = synchClassBTypeFromSymbol(requiredClass[java.lang.Class[_]])
+  lazy val jlThrowableRef           : ClassBType = synchClassBTypeFromSymbol(defn.ThrowableClass)
+  lazy val jlCloneableRef           : ClassBType = synchClassBTypeFromSymbol(defn.JavaCloneableClass)
+  lazy val jiSerializableRef        : ClassBType = synchClassBTypeFromSymbol(requiredClass[java.io.Serializable])
+  lazy val jlClassCastExceptionRef  : ClassBType = synchClassBTypeFromSymbol(requiredClass[java.lang.ClassCastException])
+  lazy val jlIllegalArgExceptionRef : ClassBType = synchClassBTypeFromSymbol(requiredClass[java.lang.IllegalArgumentException])
+  lazy val jliSerializedLambdaRef   : ClassBType = synchClassBTypeFromSymbol(requiredClass[java.lang.invoke.SerializedLambda])
+
+  lazy val srBoxesRuntimeRef: ClassBType = synchClassBTypeFromSymbol(requiredClass[scala.runtime.BoxesRunTime])
+
+  private lazy val jliCallSiteRef            : ClassBType = synchClassBTypeFromSymbol(requiredClass[java.lang.invoke.CallSite])
+  private lazy val jliLambdaMetafactoryRef   : ClassBType = synchClassBTypeFromSymbol(requiredClass[java.lang.invoke.LambdaMetafactory])
+  private lazy val jliMethodHandleRef        : ClassBType = synchClassBTypeFromSymbol(defn.MethodHandleClass)
+  private lazy val jliMethodHandlesLookupRef : ClassBType = synchClassBTypeFromSymbol(defn.MethodHandlesLookupClass)
+  private lazy val jliMethodTypeRef          : ClassBType = synchClassBTypeFromSymbol(requiredClass[java.lang.invoke.MethodType])
+  private lazy val jliStringConcatFactoryRef : ClassBType = synchClassBTypeFromSymbol(requiredClass("java.lang.invoke.StringConcatFactory")) // since JDK 9
+
+  lazy val srLambdaDeserialize               : ClassBType = synchClassBTypeFromSymbol(requiredClass[scala.runtime.LambdaDeserialize])
+
+  lazy val jliLambdaMetaFactoryMetafactoryHandle = frontendSynch{ new Handle(
     Opcodes.H_INVOKESTATIC,
     jliLambdaMetafactoryRef.internalName,
     "metafactory",
@@ -160,9 +165,9 @@ abstract class CoreBTypesFromSymbols[I <: DottyBackendInterface] extends CoreBTy
       List(jliMethodHandlesLookupRef, StringRef, jliMethodTypeRef, jliMethodTypeRef, jliMethodHandleRef, jliMethodTypeRef),
       jliCallSiteRef
     ).descriptor,
-    /* itf = */ false)
+    /* itf = */ false)}
 
-  lazy val jliLambdaMetaFactoryAltMetafactoryHandle = new Handle(
+  lazy val jliLambdaMetaFactoryAltMetafactoryHandle = frontendSynch{ new Handle(
     Opcodes.H_INVOKESTATIC,
     jliLambdaMetafactoryRef.internalName,
     "altMetafactory",
@@ -170,9 +175,9 @@ abstract class CoreBTypesFromSymbols[I <: DottyBackendInterface] extends CoreBTy
       List(jliMethodHandlesLookupRef, StringRef, jliMethodTypeRef, ArrayBType(ObjectRef)),
       jliCallSiteRef
     ).descriptor,
-    /* itf = */ false)
+    /* itf = */ false)}
 
-  lazy val jliLambdaDeserializeBootstrapHandle: Handle = new Handle(
+  lazy val jliLambdaDeserializeBootstrapHandle: Handle = frontendSynch{ new Handle(
     Opcodes.H_INVOKESTATIC,
     srLambdaDeserialize.internalName,
     "bootstrap",
@@ -180,9 +185,9 @@ abstract class CoreBTypesFromSymbols[I <: DottyBackendInterface] extends CoreBTy
       List(jliMethodHandlesLookupRef, StringRef, jliMethodTypeRef, ArrayBType(jliMethodHandleRef)),
       jliCallSiteRef
     ).descriptor,
-    /* itf = */ false)
+    /* itf = */ false)}
 
-  lazy val jliStringConcatFactoryMakeConcatWithConstantsHandle = new Handle(
+  lazy val jliStringConcatFactoryMakeConcatWithConstantsHandle = frontendSynch{ new Handle(
     Opcodes.H_INVOKESTATIC,
     jliStringConcatFactoryRef.internalName,
     "makeConcatWithConstants",
@@ -190,7 +195,7 @@ abstract class CoreBTypesFromSymbols[I <: DottyBackendInterface] extends CoreBTy
       List(jliMethodHandlesLookupRef, StringRef, jliMethodTypeRef, StringRef, ArrayBType(ObjectRef)),
       jliCallSiteRef
     ).descriptor,
-    /* itf = */ false)
+    /* itf = */ false)}
 
   /**
    * Methods in scala.runtime.BoxesRuntime
diff --git a/compiler/src/dotty/tools/backend/jvm/GenBCode.scala b/compiler/src/dotty/tools/backend/jvm/GenBCode.scala
index 469a6ea57679..8c1691e88b80 100644
--- a/compiler/src/dotty/tools/backend/jvm/GenBCode.scala
+++ b/compiler/src/dotty/tools/backend/jvm/GenBCode.scala
@@ -66,25 +66,30 @@ class GenBCode extends Phase { self =>
     _postProcessor.nn
   }
 
-  override def run(using ctx: Context): Unit =
-    // CompilationUnit is the only component that will differ between each run invocation
-    // We need to update it to have correct source positions.
-    // FreshContext is always enforced when creating backend interface
-    backendInterface.ctx
+  private var _generatedClassHandler: GeneratedClassHandler | Null = _
+  def generatedClassHandler(using Context): GeneratedClassHandler = {
+    if _generatedClassHandler eq null then
+      _generatedClassHandler = GeneratedClassHandler(postProcessor)
+    _generatedClassHandler.nn
+  }
+
+  override def run(using Context): Unit =
+    frontendAccess.frontendSynch {
+      backendInterface.ctx
       .asInstanceOf[FreshContext]
       .setCompilationUnit(ctx.compilationUnit)
-    val generated = codeGen.genUnit(ctx.compilationUnit)
-    // In Scala 2, the backend might use global optimizations which might delay post-processing to build the call graph.
-    // In Scala 3, we don't perform backend optimizations and always perform post-processing immediately.
-    // https://github.com/scala/scala/pull/6057
-    postProcessor.postProcessAndSendToDisk(generated)
+    }
+    codeGen.genUnit(ctx.compilationUnit)
     (ctx.compilerCallback: CompilerCallback | Null) match {
       case cb: CompilerCallback => cb.onSourceCompiled(ctx.source)
       case null => ()
     }
 
   override def runOn(units: List[CompilationUnit])(using ctx:Context): List[CompilationUnit] = {
-    try super.runOn(units)
+    try
+      val result = super.runOn(units)
+      generatedClassHandler.complete()
+      result
     finally
       // frontendAccess and postProcessor are created lazilly, clean them up only if they were initialized
       if _frontendAccess ne null then
@@ -100,6 +105,7 @@ class GenBCode extends Phase { self =>
         }
       if _postProcessor ne null then
         postProcessor.classfileWriter.close()
+      generatedClassHandler.close()
   }
 }
 
diff --git a/compiler/src/dotty/tools/backend/jvm/GeneratedClassHandler.scala b/compiler/src/dotty/tools/backend/jvm/GeneratedClassHandler.scala
new file mode 100644
index 000000000000..fc02d9597efe
--- /dev/null
+++ b/compiler/src/dotty/tools/backend/jvm/GeneratedClassHandler.scala
@@ -0,0 +1,191 @@
+package dotty.tools.backend.jvm
+
+import java.nio.channels.ClosedByInterruptException
+import java.util.concurrent.ThreadPoolExecutor.CallerRunsPolicy
+import java.util.concurrent._
+
+import scala.collection.mutable.ListBuffer
+import scala.concurrent.duration.Duration
+import scala.concurrent.{Await, ExecutionContext, ExecutionContextExecutor, Future}
+import dotty.tools.dotc.core.Contexts.*
+import dotty.tools.io.AbstractFile
+import dotty.tools.dotc.profile.ThreadPoolFactory
+import scala.util.control.NonFatal
+import dotty.tools.dotc.core.Phases
+import dotty.tools.dotc.core.Decorators.em
+
+import scala.language.unsafeNulls
+
+/**
+ * Interface to handle post-processing and classfile writing (see [[PostProcessor]]) of generated
+ * classes, potentially in parallel.
+ */
+private[jvm] sealed trait GeneratedClassHandler {
+  val postProcessor: PostProcessor
+
+  /**
+    * Pass the result of code generation for a compilation unit to this handler for post-processing
+    */
+  def process(unit: GeneratedCompilationUnit): Unit
+
+  /**
+   * If running in parallel, block until all generated classes are handled
+   */
+  def complete(): Unit
+
+  /**
+    * Invoked at the end of the jvm phase
+    */
+  def close(): Unit = ()
+}
+
+private[jvm] object GeneratedClassHandler {
+  def apply(postProcessor: PostProcessor)(using ictx: Context): GeneratedClassHandler = {
+    val compilerSettings = postProcessor.frontendAccess.compilerSettings
+    val handler = compilerSettings.backendParallelism match {
+      case 1 => new SyncWritingClassHandler(postProcessor)
+
+      case maxThreads =>
+        // if (settings.areStatisticsEnabled)
+        //   runReporting.warning(
+        //     NoPosition,
+        //     "JVM statistics are not reliable with multi-threaded JVM class writing.\n" +
+        //     "To collect compiler statistics remove the " + settings.YaddBackendThreads.name + " setting.",
+        //     WarningCategory.Other,
+        //     site = ""
+        //   )
+        val additionalThreads = maxThreads - 1
+        // The thread pool queue is limited in size. When it's full, the `CallerRunsPolicy` causes
+        // a new task to be executed on the main thread, which provides back-pressure.
+        // The queue size is large enough to ensure that running a task on the main thread does
+        // not take longer than to exhaust the queue for the backend workers.
+        val queueSize = compilerSettings.backendMaxWorkerQueue.getOrElse(maxThreads * 2)
+        val threadPoolFactory = ThreadPoolFactory(Phases.genBCodePhase)
+        val javaExecutor = threadPoolFactory.newBoundedQueueFixedThreadPool(additionalThreads, queueSize, new CallerRunsPolicy, "non-ast")
+        new AsyncWritingClassHandler(postProcessor, javaExecutor)
+    }
+
+    // if (settings.optInlinerEnabled || settings.optClosureInvocations) new GlobalOptimisingGeneratedClassHandler(postProcessor, handler)
+    // else
+    handler
+  }
+
+  sealed abstract class WritingClassHandler(val javaExecutor: Executor) extends GeneratedClassHandler {
+    import postProcessor.bTypes.frontendAccess
+
+    def tryStealing: Option[Runnable]
+
+    private val processingUnits = ListBuffer.empty[CompilationUnitInPostProcess]
+
+    def process(unit: GeneratedCompilationUnit): Unit = {
+      val unitInPostProcess = new CompilationUnitInPostProcess(unit.classes, unit.tasty, unit.sourceFile)(using unit.ctx)
+      postProcessUnit(unitInPostProcess)
+      processingUnits += unitInPostProcess
+    }
+
+    protected implicit val executionContext: ExecutionContextExecutor = ExecutionContext.fromExecutor(javaExecutor)
+
+    final def postProcessUnit(unitInPostProcess: CompilationUnitInPostProcess): Unit = {
+      unitInPostProcess.task = Future:
+        frontendAccess.withThreadLocalReporter(unitInPostProcess.bufferedReporting):
+          // we 'take' classes to reduce the memory pressure
+          // as soon as the class is consumed and written, we release its data
+          unitInPostProcess.takeClasses().foreach:
+            postProcessor.sendToDisk(_, unitInPostProcess.sourceFile)
+          unitInPostProcess.takeTasty().foreach:
+            postProcessor.sendToDisk(_, unitInPostProcess.sourceFile)
+    }
+
+    protected def takeProcessingUnits(): List[CompilationUnitInPostProcess] = {
+      val result = processingUnits.result()
+      processingUnits.clear()
+      result
+    }
+
+    final def complete(): Unit = {
+      import frontendAccess.directBackendReporting
+
+      def stealWhileWaiting(unitInPostProcess: CompilationUnitInPostProcess): Unit = {
+        val task = unitInPostProcess.task
+        while (!task.isCompleted)
+          tryStealing match {
+            case Some(r) => r.run()
+            case None => Await.ready(task, Duration.Inf)
+          }
+      }
+
+      /*
+       * Go through each task in submission order, wait for it to finish and report its messages.
+       * When finding task that has not completed, steal work from the executor's queue and run
+       * it on the main thread (which we are on here), until the task is done.
+       *
+       * We could consume the results when they are ready, via use of a [[java.util.concurrent.CompletionService]]
+       * or something similar, but that would lead to non deterministic reports from backend threads, as the
+       * compilation unit could complete in a different order than when they were submitted, and thus the relayed
+       * reports would be in a different order.
+       * To avoid that non-determinism we read the result in order of submission, with a potential minimal performance
+       * loss, due to the memory being retained longer for tasks than it might otherwise.
+       * Most of the memory in the CompilationUnitInPostProcess is reclaimable anyway as the classes are dereferenced after use.
+       */
+      takeProcessingUnits().foreach { unitInPostProcess =>
+        try
+          stealWhileWaiting(unitInPostProcess)
+          unitInPostProcess.bufferedReporting.relayReports(directBackendReporting)
+          // We know the future is complete, throw the exception if it completed with a failure
+          unitInPostProcess.task.value.get.get
+        catch
+          case _: ClosedByInterruptException => throw new InterruptedException()
+          case NonFatal(t) =>
+            t.printStackTrace()
+            frontendAccess.backendReporting.error(em"unable to write ${unitInPostProcess.sourceFile} $t")
+      }
+    }
+  }
+
+  private final class SyncWritingClassHandler(val postProcessor: PostProcessor)
+    extends WritingClassHandler(_.nn.run()) {
+
+    override def toString: String = "SyncWriting"
+
+    def tryStealing: Option[Runnable] = None
+  }
+
+  private final class AsyncWritingClassHandler(val postProcessor: PostProcessor, override val javaExecutor: ThreadPoolExecutor)
+    extends WritingClassHandler(javaExecutor) {
+
+    override def toString: String = s"AsyncWriting[additional threads:${javaExecutor.getMaximumPoolSize}]"
+
+    override def close(): Unit = {
+      super.close()
+      javaExecutor.shutdownNow()
+    }
+
+    def tryStealing: Option[Runnable] = Option(javaExecutor.getQueue.poll())
+  }
+
+}
+
+/**
+ * State for a compilation unit being post-processed.
+ *   - Holds the classes to post-process (released for GC when no longer used)
+ *   - Keeps a reference to the future that runs the post-processor
+ *   - Buffers messages reported during post-processing
+ */
+final private class CompilationUnitInPostProcess(private var classes: List[GeneratedClass], private var tasty: List[GeneratedTasty], val sourceFile: AbstractFile)(using Context) {
+  def takeClasses(): List[GeneratedClass] = {
+    val c = classes
+    classes = Nil
+    c
+  }
+
+  def takeTasty(): List[GeneratedTasty] = {
+    val v = tasty
+    tasty = Nil
+    v
+  }
+
+  /** the main async task submitted onto the scheduler */
+  var task: Future[Unit] = _
+
+  val bufferedReporting = new PostProcessorFrontendAccess.BufferingBackendReporting()
+}
\ No newline at end of file
diff --git a/compiler/src/dotty/tools/backend/jvm/PostProcessor.scala b/compiler/src/dotty/tools/backend/jvm/PostProcessor.scala
index d1f9c6216a22..680e587844c7 100644
--- a/compiler/src/dotty/tools/backend/jvm/PostProcessor.scala
+++ b/compiler/src/dotty/tools/backend/jvm/PostProcessor.scala
@@ -1,5 +1,7 @@
 package dotty.tools.backend.jvm
 
+import java.util.concurrent.ConcurrentHashMap
+
 import scala.collection.mutable.ListBuffer
 import dotty.tools.dotc.util.{SourcePosition, NoSourcePosition}
 import dotty.tools.io.AbstractFile
@@ -14,49 +16,70 @@ import scala.tools.asm.tree.ClassNode
  */
 class PostProcessor(val frontendAccess: PostProcessorFrontendAccess, val bTypes: BTypes) {
   self =>
-  import bTypes.*
+  import bTypes.{classBTypeFromInternalName, int}
   import frontendAccess.{backendReporting, compilerSettings}
   import int.given
 
   val backendUtils = new BackendUtils(this)
-  val classfileWriter  = ClassfileWriter(frontendAccess)
-
-  def postProcessAndSendToDisk(generatedDefs: GeneratedDefs): Unit = {
-    val GeneratedDefs(classes, tasty) = generatedDefs
-    if !ctx.settings.YoutputOnlyTasty.value then
-      postProcessClassesAndSendToDisk(classes)
-    postProcessTastyAndSendToDisk(tasty)
-  }
-
-  private def postProcessClassesAndSendToDisk(classes: List[GeneratedClass]): Unit = {
-    for (GeneratedClass(classNode, sourceFile, isArtifact, onFileCreated) <- classes) {
-      val bytes =
-        try
-          if !isArtifact then setSerializableLambdas(classNode)
-          setInnerClasses(classNode)
-          serializeClass(classNode)
-        catch
-          case e: java.lang.RuntimeException if e.getMessage != null && e.getMessage.nn.contains("too large!") =>
-            backendReporting.error(em"Could not write class ${classNode.name} because it exceeds JVM code size limits. ${e.getMessage}")
-            null
-          case ex: Throwable =>
-            ex.printStackTrace()
-            backendReporting.error(em"Error while emitting ${classNode.name}\n${ex.getMessage}")
-            null
-
-      if (bytes != null) {
-        if (AsmUtils.traceSerializedClassEnabled && classNode.name.nn.contains(AsmUtils.traceSerializedClassPattern))
+  val classfileWriters = new ClassfileWriters(frontendAccess)
+  val classfileWriter  = classfileWriters.ClassfileWriter()
+
+  type ClassnamePosition = (String, SourcePosition)
+  private val caseInsensitively = new ConcurrentHashMap[String, ClassnamePosition]
+
+  def sendToDisk(clazz: GeneratedClass, sourceFile: AbstractFile): Unit = if !ctx.settings.YoutputOnlyTasty.value then {
+    val classNode = clazz.classNode
+    val internalName = classNode.name.nn
+    val bytes =
+      try
+        if !clazz.isArtifact then setSerializableLambdas(classNode)
+        warnCaseInsensitiveOverwrite(clazz)
+        setInnerClasses(classNode)
+        serializeClass(classNode)
+      catch
+        case e: java.lang.RuntimeException if e.getMessage != null && e.getMessage.nn.contains("too large!") =>
+          backendReporting.error(em"Could not write class $internalName because it exceeds JVM code size limits. ${e.getMessage}")
+          null
+        case ex: Throwable =>
+          if compilerSettings.debug then ex.printStackTrace()
+          backendReporting.error(em"Error while emitting $internalName\n${ex.getMessage}")
+          null
+
+     if bytes != null then
+        if (AsmUtils.traceSerializedClassEnabled && internalName.contains(AsmUtils.traceSerializedClassPattern))
           AsmUtils.traceClass(bytes)
+        val clsFile = classfileWriter.writeClass(internalName, bytes, sourceFile)
+        if clsFile != null then clazz.onFileCreated(clsFile)
+  }
 
-        val clsFile = classfileWriter.writeClass(classNode.name.nn, bytes, sourceFile)
-        if clsFile != null then onFileCreated(clsFile)
-      }
-    }
+  def sendToDisk(tasty: GeneratedTasty, sourceFile: AbstractFile): Unit = {
+    val GeneratedTasty(classNode, tastyGenerator) = tasty
+    val internalName = classNode.name.nn
+    classfileWriter.writeTasty(classNode.name.nn, tastyGenerator(), sourceFile)
   }
 
-  private def postProcessTastyAndSendToDisk(tasty: List[GeneratedTasty]): Unit = {
-    for (GeneratedTasty(classNode, binaryGen) <- tasty) {
-      classfileWriter.writeTasty(classNode.name.nn, binaryGen())
+  private def warnCaseInsensitiveOverwrite(clazz: GeneratedClass) = {
+    val name = clazz.classNode.name.nn
+    val lowerCaseJavaName = name.nn.toLowerCase
+    val clsPos = clazz.position
+    caseInsensitively.putIfAbsent(lowerCaseJavaName, (name, clsPos)) match {
+      case null => ()
+      case (dupName, dupPos) =>
+        // Order is not deterministic so we enforce lexicographic order between the duplicates for error-reporting
+        val ((pos1, pos2), (name1, name2)) =
+          if (name < dupName) ((clsPos, dupPos), (name, dupName))
+          else ((dupPos, clsPos), (dupName, name))
+        val locationAddendum =
+          if pos1.source.path == pos2.source.path then ""
+          else s" (defined in ${pos2.source.file.name})"
+        def nicify(name: String): String = name.replace('/', '.').nn
+        if name1 == name2 then
+          backendReporting.warning(
+            em"${nicify(name1)} and ${nicify(name2)} produce classes that overwrite one another", pos1)
+        else
+          backendReporting.warning(
+            em"""Generated class ${nicify(name1)} differs only in case from ${nicify(name2)}$locationAddendum.
+                |  Such classes will overwrite one another on case-insensitive filesystems.""", pos1)
     }
   }
 
@@ -114,12 +137,12 @@ class PostProcessor(val frontendAccess: PostProcessorFrontendAccess, val bTypes:
 /**
  * The result of code generation. [[isArtifact]] is `true` for mirror.
  */
-case class GeneratedClass(classNode: ClassNode, sourceFile: AbstractFile, isArtifact: Boolean, onFileCreated: AbstractFile => Unit)
+case class GeneratedClass(
+  classNode: ClassNode,
+  sourceClassName: String,
+  position: SourcePosition,
+  isArtifact: Boolean,
+  onFileCreated: AbstractFile => Unit)
 case class GeneratedTasty(classNode: ClassNode, tastyGen: () => Array[Byte])
-case class GeneratedDefs(classes: List[GeneratedClass], tasty: List[GeneratedTasty])
+case class GeneratedCompilationUnit(sourceFile: AbstractFile, classes: List[GeneratedClass], tasty: List[GeneratedTasty])(using val ctx: Context)
 
-// Temporary class, will be refactored in a future commit
-trait ClassWriterForPostProcessor {
-  type InternalName = String
-  def write(bytes: Array[Byte], className: InternalName, sourceFile: AbstractFile): Unit
-}
diff --git a/compiler/src/dotty/tools/backend/jvm/PostProcessorFrontendAccess.scala b/compiler/src/dotty/tools/backend/jvm/PostProcessorFrontendAccess.scala
index 80ee68bc94c3..e6911c766655 100644
--- a/compiler/src/dotty/tools/backend/jvm/PostProcessorFrontendAccess.scala
+++ b/compiler/src/dotty/tools/backend/jvm/PostProcessorFrontendAccess.scala
@@ -17,11 +17,15 @@ sealed abstract class PostProcessorFrontendAccess {
   import PostProcessorFrontendAccess._
 
   def compilerSettings: CompilerSettings
+
+  def withThreadLocalReporter[T](reporter: BackendReporting)(fn: => T): T
   def backendReporting: BackendReporting
+  def directBackendReporting: BackendReporting
+
   def getEntryPoints: List[String]
 
   private val frontendLock: AnyRef = new Object()
-  inline final def frontendSynch[T](inline x: => T): T = frontendLock.synchronized(x)
+  inline final def frontendSynch[T](inline x: T): T = frontendLock.synchronized(x)
 }
 
 object PostProcessorFrontendAccess {
@@ -33,16 +37,49 @@ object PostProcessorFrontendAccess {
     def outputDirectory: AbstractFile
 
     def mainClass: Option[String]
+
+    def jarCompressionLevel: Int
+    def backendParallelism: Int
+    def backendMaxWorkerQueue: Option[Int]
   }
 
   sealed trait BackendReporting {
-    def error(message: Context ?=> Message): Unit
-    def warning(message: Context ?=> Message): Unit
-    def log(message: Context ?=> String): Unit
+    def error(message: Context ?=> Message, position: SourcePosition): Unit
+    def warning(message: Context ?=> Message, position: SourcePosition): Unit
+    def log(message: String): Unit
+
+    def error(message: Context ?=> Message): Unit = error(message, NoSourcePosition)
+    def warning(message: Context ?=> Message): Unit = warning(message, NoSourcePosition)
   }
 
-  class Impl[I <: DottyBackendInterface](val int: I, entryPoints: HashSet[String]) extends PostProcessorFrontendAccess {
-    import int.given
+  final class BufferingBackendReporting(using Context) extends BackendReporting {
+    // We optimise access to the buffered reports for the common case - that there are no warning/errors to report
+    // We could use a listBuffer etc - but that would be extra allocation in the common case
+    // Note - all access is externally synchronized, as this allow the reports to be generated in on thread and
+    // consumed in another
+    private var bufferedReports = List.empty[Report]
+    enum Report(val relay: BackendReporting => Unit):
+      case Error(message: Message, position: SourcePosition) extends Report(_.error(message, position))
+      case Warning(message: Message, position: SourcePosition) extends Report(_.warning(message, position))
+      case Log(message: String) extends Report(_.log(message))
+
+    def error(message: Context ?=> Message, position: SourcePosition): Unit = synchronized:
+      bufferedReports ::= Report.Error(message, position)
+
+    def warning(message: Context ?=> Message, position: SourcePosition): Unit = synchronized:
+      bufferedReports ::= Report.Warning(message, position)
+
+    def log(message: String): Unit = synchronized:
+      bufferedReports ::= Report.Log(message)
+
+    def relayReports(toReporting: BackendReporting): Unit = synchronized:
+      if bufferedReports.nonEmpty then
+        bufferedReports.reverse.foreach(_.relay(toReporting))
+        bufferedReports = Nil
+  }
+
+
+  class Impl[I <: DottyBackendInterface](val int: I, entryPoints: HashSet[String])(using ctx: Context) extends PostProcessorFrontendAccess {
     lazy val compilerSettings: CompilerSettings = buildCompilerSettings()
 
     private def buildCompilerSettings(): CompilerSettings = new CompilerSettings {
@@ -66,14 +103,34 @@ object PostProcessorFrontendAccess {
       lazy val dumpClassesDirectory: Option[String] = s.Ydumpclasses.valueSetByUser
       lazy val outputDirectory: AbstractFile = s.outputDir.value
       lazy val mainClass: Option[String] = s.XmainClass.valueSetByUser
+      lazy val jarCompressionLevel: Int = s.YjarCompressionLevel.value
+      lazy val backendParallelism: Int = s.YbackendParallelism.value
+      lazy val backendMaxWorkerQueue: Option[Int] = s.YbackendWorkerQueue.valueSetByUser
+     }
+
+     private lazy val localReporter = new ThreadLocal[BackendReporting]
+
+     override def withThreadLocalReporter[T](reporter: BackendReporting)(fn: => T): T = {
+       val old = localReporter.get()
+       localReporter.set(reporter)
+       try fn
+       finally
+         if old eq null then localReporter.remove()
+         else localReporter.set(old)
+     }
+
+     override def backendReporting: BackendReporting = {
+       val local = localReporter.get()
+       if local eq null then directBackendReporting
+       else local.nn
      }
 
-    object backendReporting extends BackendReporting {
-      def error(message: Context ?=> Message): Unit = frontendSynch(report.error(message))
-      def warning(message: Context ?=> Message): Unit = frontendSynch(report.warning(message))
-      def log(message: Context ?=> String): Unit = frontendSynch(report.log(message))
+    object directBackendReporting extends BackendReporting {
+      def error(message: Context ?=> Message, position: SourcePosition): Unit = frontendSynch(report.error(message, position))
+      def warning(message: Context ?=> Message, position: SourcePosition): Unit = frontendSynch(report.warning(message, position))
+      def log(message: String): Unit = frontendSynch(report.log(message))
     }
 
     def getEntryPoints: List[String] = frontendSynch(entryPoints.toList)
   }
-}
\ No newline at end of file
+}
diff --git a/compiler/src/dotty/tools/dotc/config/ScalaSettings.scala b/compiler/src/dotty/tools/dotc/config/ScalaSettings.scala
index 43c2cd5a2ff9..925be029746e 100644
--- a/compiler/src/dotty/tools/dotc/config/ScalaSettings.scala
+++ b/compiler/src/dotty/tools/dotc/config/ScalaSettings.scala
@@ -13,6 +13,8 @@ import Setting.ChoiceWithHelp
 
 import scala.util.chaining._
 
+import java.util.zip.Deflater
+
 class ScalaSettings extends SettingGroup with AllScalaSettings
 
 object ScalaSettings:
@@ -360,6 +362,9 @@ private sealed trait YSettings:
   val YnoPredef: Setting[Boolean] = BooleanSetting("-Yno-predef", "Compile without importing Predef.")
   val Yskip: Setting[List[String]] = PhasesSetting("-Yskip", "Skip")
   val Ydumpclasses: Setting[String] = StringSetting("-Ydump-classes", "dir", "Dump the generated bytecode to .class files (useful for reflective compilation that utilizes in-memory classloaders).", "")
+  val YjarCompressionLevel: Setting[Int] = IntChoiceSetting("-Yjar-compression-level", "compression level to use when writing jar files", Deflater.DEFAULT_COMPRESSION to Deflater.BEST_COMPRESSION, Deflater.DEFAULT_COMPRESSION)
+  val YbackendParallelism: Setting[Int] = IntChoiceSetting("-Ybackend-parallelism", "maximum worker threads for backend", 1 to 16, 1)
+  val YbackendWorkerQueue: Setting[Int] = IntChoiceSetting("-Ybackend-worker-queue", "backend threads worker queue size", 0 to 1000, 0)
   val YstopAfter: Setting[List[String]] = PhasesSetting("-Ystop-after", "Stop after", aliases = List("-stop")) // backward compat
   val YstopBefore: Setting[List[String]] = PhasesSetting("-Ystop-before", "Stop before") // stop before erasure as long as we have not debugged it fully
   val YshowSuppressedErrors: Setting[Boolean] = BooleanSetting("-Yshow-suppressed-errors", "Also show follow-on errors and warnings that are normally suppressed.")
diff --git a/compiler/src/dotty/tools/dotc/profile/AsyncHelper.scala b/compiler/src/dotty/tools/dotc/profile/ThreadPoolFactory.scala
similarity index 71%
rename from compiler/src/dotty/tools/dotc/profile/AsyncHelper.scala
rename to compiler/src/dotty/tools/dotc/profile/ThreadPoolFactory.scala
index 61bee4d9f32a..1e9969e0bc65 100644
--- a/compiler/src/dotty/tools/dotc/profile/AsyncHelper.scala
+++ b/compiler/src/dotty/tools/dotc/profile/ThreadPoolFactory.scala
@@ -9,38 +9,49 @@ import java.util.concurrent.atomic.AtomicInteger
 import dotty.tools.dotc.core.Phases.Phase
 import dotty.tools.dotc.core.Contexts._
 
-sealed trait AsyncHelper {
-
-  def newUnboundedQueueFixedThreadPool
-  (nThreads: Int,
-   shortId: String, priority : Int = Thread.NORM_PRIORITY) : ThreadPoolExecutor
-  def newBoundedQueueFixedThreadPool
-  (nThreads: Int, maxQueueSize: Int, rejectHandler: RejectedExecutionHandler,
-   shortId: String, priority : Int = Thread.NORM_PRIORITY) : ThreadPoolExecutor
+sealed trait ThreadPoolFactory {
+
+  def newUnboundedQueueFixedThreadPool(
+    nThreads: Int,
+    shortId: String,
+    priority : Int = Thread.NORM_PRIORITY) : ThreadPoolExecutor
+
+  def newBoundedQueueFixedThreadPool(
+    nThreads: Int,
+    maxQueueSize: Int,
+    rejectHandler: RejectedExecutionHandler,
+    shortId: String,
+    priority : Int = Thread.NORM_PRIORITY) : ThreadPoolExecutor
 }
 
 
-object AsyncHelper {
-  def apply(phase: Phase)(using Context): AsyncHelper = ctx.profiler match {
-    case NoOpProfiler => new BasicAsyncHelper(phase)
-    case r: RealProfiler => new ProfilingAsyncHelper(phase, r)
+object ThreadPoolFactory {
+  def apply(phase: Phase)(using Context): ThreadPoolFactory = ctx.profiler match {
+    case NoOpProfiler => new BasicThreadPoolFactory(phase)
+    case r: RealProfiler => new ProfilingThreadPoolFactory(phase, r)
   }
 
-  private abstract class BaseAsyncHelper(phase: Phase)(using Context) extends AsyncHelper {
+  private abstract class BaseThreadPoolFactory(phase: Phase) extends ThreadPoolFactory {
     val baseGroup = new ThreadGroup(s"dotc-${phase.phaseName}")
+
     private def childGroup(name: String) = new ThreadGroup(baseGroup, name)
 
-    protected def wrapRunnable(r: Runnable, shortId:String): Runnable
+    // Invoked when a new `Worker` is created, see `CommonThreadFactory.newThread`
+    protected def wrapWorker(worker: Runnable, shortId:String): Runnable = worker
 
-    protected class CommonThreadFactory(shortId: String,
-                                        daemon: Boolean = true,
-                                        priority: Int) extends ThreadFactory {
+    protected final class CommonThreadFactory(
+        shortId: String,
+        daemon: Boolean = true,
+        priority: Int) extends ThreadFactory {
       private val group: ThreadGroup = childGroup(shortId)
       private val threadNumber: AtomicInteger = new AtomicInteger(1)
       private val namePrefix = s"${baseGroup.getName}-$shortId-"
 
-      override def newThread(r: Runnable): Thread = {
-        val wrapped = wrapRunnable(r, shortId)
+      // Invoked by the `ThreadPoolExecutor` when creating a new worker thread. The argument
+      // runnable is the `Worker` (which extends `Runnable`). Its `run` method gets tasks from
+      // the thread pool and executes them (on the thread created here).
+      override def newThread(worker: Runnable): Thread = {
+        val wrapped = wrapWorker(worker, shortId)
         val t: Thread = new Thread(group, wrapped, namePrefix + threadNumber.getAndIncrement, 0)
         if (t.isDaemon != daemon) t.setDaemon(daemon)
         if (t.getPriority != priority) t.setPriority(priority)
@@ -49,7 +60,7 @@ object AsyncHelper {
     }
   }
 
-  private final class BasicAsyncHelper(phase: Phase)(using Context) extends BaseAsyncHelper(phase) {
+  private final class BasicThreadPoolFactory(phase: Phase) extends BaseThreadPoolFactory(phase) {
 
     override def newUnboundedQueueFixedThreadPool(nThreads: Int, shortId: String, priority: Int): ThreadPoolExecutor = {
       val threadFactory = new CommonThreadFactory(shortId, priority = priority)
@@ -62,11 +73,9 @@ object AsyncHelper {
       //like Executors.newFixedThreadPool
       new ThreadPoolExecutor(nThreads, nThreads, 0L, TimeUnit.MILLISECONDS, new ArrayBlockingQueue[Runnable](maxQueueSize), threadFactory, rejectHandler)
     }
-
-    override protected def wrapRunnable(r: Runnable, shortId:String): Runnable = r
   }
 
-  private class ProfilingAsyncHelper(phase: Phase, private val profiler: RealProfiler)(using Context) extends BaseAsyncHelper(phase) {
+  private class ProfilingThreadPoolFactory(phase: Phase, private val profiler: RealProfiler) extends BaseThreadPoolFactory(phase) {
 
     override def newUnboundedQueueFixedThreadPool(nThreads: Int, shortId: String, priority: Int): ThreadPoolExecutor = {
       val threadFactory = new CommonThreadFactory(shortId, priority = priority)
@@ -80,13 +89,13 @@ object AsyncHelper {
       new SinglePhaseInstrumentedThreadPoolExecutor(nThreads, nThreads, 0L, TimeUnit.MILLISECONDS, new ArrayBlockingQueue[Runnable](maxQueueSize), threadFactory, rejectHandler)
     }
 
-    override protected def wrapRunnable(r: Runnable, shortId:String): Runnable = {
+    override protected def wrapWorker(worker: Runnable, shortId: String): Runnable = {
       () =>
       val data = new ThreadProfileData
       localData.set(data)
 
       val profileStart = profiler.snapThread(0)
-      try r.run finally {
+      try worker.run finally {
         val snap = profiler.snapThread(data.idleNs)
         val threadRange = ProfileRange(profileStart, snap, phase, shortId, data.taskCount, Thread.currentThread())
         profiler.completeBackground(threadRange)
diff --git a/compiler/test/dotty/tools/dotc/CompilationTests.scala b/compiler/test/dotty/tools/dotc/CompilationTests.scala
index dfa6c370c7ab..4f1998cd50fd 100644
--- a/compiler/test/dotty/tools/dotc/CompilationTests.scala
+++ b/compiler/test/dotty/tools/dotc/CompilationTests.scala
@@ -18,6 +18,7 @@ import scala.concurrent.duration._
 import TestSources.sources
 import reporting.TestReporter
 import vulpix._
+import dotty.tools.dotc.config.ScalaSettings
 
 class CompilationTests {
   import ParallelTesting._
@@ -219,7 +220,6 @@ class CompilationTests {
     compileFilesInDir("tests/init/neg", options).checkExpectedErrors()
     compileFilesInDir("tests/init/pos", options).checkCompile()
     compileFilesInDir("tests/init/crash", options.without("-Xfatal-warnings")).checkCompile()
-
     // The regression test for i12128 has some atypical classpath requirements.
     // The test consists of three files: (a) Reflect_1  (b) Macro_2  (c) Test_3
     // which must be compiled separately. In addition:
@@ -241,6 +241,38 @@ class CompilationTests {
       tests.foreach(_.delete())
     }
   }
+
+  // parallel backend tests
+  @Test def parallelBackend: Unit = {
+    given TestGroup = TestGroup("parallelBackend")
+    val parallelism = Runtime.getRuntime().availableProcessors().min(16)
+    assumeTrue("Not enough available processors to run parallel tests", parallelism > 1)
+
+    val options = defaultOptions.and(s"-Ybackend-parallelism:${parallelism}")
+    def parCompileDir(directory: String) = compileDir(directory, options)
+
+    // Compilation units containing more than 1 source file
+    aggregateTests(
+      parCompileDir("tests/pos/i10477"),
+      parCompileDir("tests/pos/i4758"),
+      parCompileDir("tests/pos/scala2traits"),
+      parCompileDir("tests/pos/class-gadt"),
+      parCompileDir("tests/pos/tailcall"),
+      parCompileDir("tests/pos/reference"),
+      parCompileDir("tests/pos/pos_valueclasses")
+    ).checkCompile()
+
+    aggregateTests(
+      parCompileDir("tests/neg/package-implicit"),
+      parCompileDir("tests/neg/package-export")
+    ).checkExpectedErrors()
+
+    aggregateTests(
+      parCompileDir("tests/run/decorators"),
+      parCompileDir("tests/run/generic")
+    ).checkRuns()
+
+  }
 }
 
 object CompilationTests extends ParallelTesting {