ScalafmtRunner: allow dedicated format thread pool

scalameta · Feb 21, 2025 · 4b8c16c · 4b8c16c
1 parent b03f619
commit 4b8c16c
Show file tree

Hide file tree

Showing 5 changed files with 36 additions and 9 deletions.
diff --git a/scalafmt-cli/shared/src/main/scala/org/scalafmt/cli/CliArgParser.scala b/scalafmt-cli/shared/src/main/scala/org/scalafmt/cli/CliArgParser.scala
@@ -132,6 +132,12 @@ object CliArgParser {
         val offset = if (from == to) 0 else -1
         c.copy(range = c.range + Range(from - 1, to + offset))
       }.text("(experimental) only format line range from=to")
+      opt[Unit]("async-format").action((_, c) => c.copy(asyncFormat = true))
+        .text(
+          """|Use a dedicated thread pool for formatting, rather than
+             |continuing after read completes, thus separating I/O-bound
+             |input and CPU-bound formatting thread pools.""".stripMargin,
+        )
 
       note(
         s"""|Examples:

diff --git a/scalafmt-cli/shared/src/main/scala/org/scalafmt/cli/CliOptions.scala b/scalafmt-cli/shared/src/main/scala/org/scalafmt/cli/CliOptions.scala
@@ -98,6 +98,7 @@ case class CliOptions(
     noStdErr: Boolean = false,
     private val error: Boolean = false,
     check: Boolean = false,
+    asyncFormat: Boolean = false,
 ) {
   val writeMode: WriteMode = writeModeOpt.getOrElse(WriteMode.Override)
 

diff --git a/scalafmt-cli/shared/src/main/scala/org/scalafmt/cli/ScalafmtRunner.scala b/scalafmt-cli/shared/src/main/scala/org/scalafmt/cli/ScalafmtRunner.scala
@@ -79,7 +79,14 @@ trait ScalafmtRunner {
   )(f: (String, Path) => Either[ExitCode, String]): Future[ExitCode] = {
     val termDisplay = newTermDisplay(options, inputMethods, termDisplayMessage)
 
-    import PlatformRunOps.parasiticExecutionContext
+    implicit val ec = PlatformRunOps.parasiticExecutionContext
+    val (formatContext, writeContext) =
+      if (options.asyncFormat) (
+        PlatformRunOps.formatExecutionContext,
+        PlatformRunOps.outputExecutionContext,
+      )
+      else (ec, ec)
+
     implicit class FutureExt[A](private val obj: Future[A]) extends AnyRef {
       def td(f: (TermDisplay, Try[A]) => Unit): Future[A] = termDisplay
         .fold(obj)(td => obj.transform { r => f(td, r); r })
@@ -98,13 +105,13 @@ trait ScalafmtRunner {
       if (!completed.isCompleted) {
         val read = inputMethod.readInput(options)
         val format = read.td { case (td, r) => td.doneRead(ok = r.isSuccess) }
-          .map(code => f(code, inputMethod.path).map((code, _)))
+          .map(code => f(code, inputMethod.path).map((code, _)))(formatContext)
           .td { case (td, r) => td.doneFormat(ok = r.toOption.exists(_.isRight)) }
         val future = format.flatMap {
           case Left(exitCode) => exitCode.future
           case Right((code, formattedCode)) => inputMethod
               .write(formattedCode, code, options)
-        }.transform { r =>
+        }(writeContext).transform { r =>
           val ok = r == Success(ExitCode.Ok)
           termDisplay.foreach(_.doneWrite(ok = ok))
           if (!ok && options.check) completed.trySuccess(asExit(r))

diff --git a/scalafmt-sysops/js/src/main/scala/org/scalafmt/sysops/PlatformRunOps.scala b/scalafmt-sysops/js/src/main/scala/org/scalafmt/sysops/PlatformRunOps.scala
@@ -15,7 +15,9 @@ private[scalafmt] object PlatformRunOps {
   implicit def executionContext: ExecutionContext =
     scala.scalajs.concurrent.JSExecutionContext.Implicits.queue
 
-  val inputExecutionContext: ExecutionContext = executionContext
+  def inputExecutionContext: ExecutionContext = executionContext
+  def formatExecutionContext: ExecutionContext = executionContext
+  def outputExecutionContext: ExecutionContext = executionContext
 
   implicit def parasiticExecutionContext: ExecutionContext =
     GranularDialectAsyncOps.parasiticExecutionContext

diff --git a/scalafmt-sysops/jvm-native/src/main/scala/org/scalafmt/sysops/PlatformRunOps.scala b/scalafmt-sysops/jvm-native/src/main/scala/org/scalafmt/sysops/PlatformRunOps.scala
@@ -2,6 +2,9 @@ package org.scalafmt.sysops
 
 import java.nio.file.Path
 import java.util.concurrent.Executors
+import java.util.concurrent.SynchronousQueue
+import java.util.concurrent.ThreadPoolExecutor
+import java.util.concurrent.TimeUnit
 
 import scala.concurrent.ExecutionContext
 import scala.concurrent.ExecutionContextExecutorService
@@ -14,15 +17,23 @@ private[scalafmt] object PlatformRunOps {
 
   implicit def executionContext: ExecutionContext = ExecutionContext.global
 
+  private val ncores = Runtime.getRuntime.availableProcessors()
+
   // creates non-daemon threads
   val inputExecutionContext: ExecutionContextExecutorService = ExecutionContext
-    .fromExecutorService(
-      Executors.newFixedThreadPool(Runtime.getRuntime.availableProcessors()),
+    .fromExecutorService(Executors.newFixedThreadPool(ncores))
+
+  lazy val formatExecutionContext: ExecutionContext = {
+    val queue = new SynchronousQueue[Runnable]() {
+      override def offer(e: Runnable): Boolean = { put(e); true } // blocks
+    }
+    ExecutionContext.fromExecutorService(
+      new ThreadPoolExecutor(ncores, ncores, 0L, TimeUnit.MILLISECONDS, queue),
     )
+  }
+
   val outputExecutionContext: ExecutionContextExecutorService = ExecutionContext
-    .fromExecutorService(
-      Executors.newFixedThreadPool(Runtime.getRuntime.availableProcessors()),
-    )
+    .fromExecutorService(Executors.newFixedThreadPool(ncores))
 
   implicit def parasiticExecutionContext: ExecutionContext =
     GranularDialectAsyncOps.parasiticExecutionContext