feat: virtual thread executor support (#32689)

Author: johanandren

akka-actor-tests/src/test/scala/akka/dispatch/ForkJoinPoolStarvationSpec.scala

@@ -5,9 +5,9 @@
 package akka.dispatch
 
 import com.typesafe.config.ConfigFactory
-
 import akka.actor.{ Actor, Props }
 import akka.testkit.{ AkkaSpec, ImplicitSender }
+import akka.util.JavaVersion
 
 object ForkJoinPoolStarvationSpec {
   val config = ConfigFactory.parseString("""
@@ -52,8 +52,7 @@ class ForkJoinPoolStarvationSpec extends AkkaSpec(ForkJoinPoolStarvationSpec.con
 
     "not starve tasks arriving from external dispatchers under high internal traffic" in {
       // TODO issue #31117: starvation with JDK 17 FJP
-      val javaSpecVersion = System.getProperty("java.specification.version").toInt
-      if (javaSpecVersion >= 17)
+      if (JavaVersion.majorVersion >= 17)
         pending
 
       // Two busy actors that will occupy the threads of the dispatcher

akka-actor-tests/src/test/scala/akka/dispatch/VirtualThreadDispatcherSpec.scala

@@ -0,0 +1,127 @@
+/*
+ * Copyright (C) 2009-2025 Lightbend Inc. <https://www.lightbend.com>
+ */
+
+package akka.dispatch
+
+import akka.actor.Actor
+import akka.actor.ActorSystem
+import akka.actor.Props
+import akka.testkit.TestKit
+import akka.testkit.TestProbe
+import akka.util.JavaVersion
+import com.typesafe.config.ConfigFactory
+import org.scalatest.matchers.should.Matchers
+import org.scalatest.wordspec.AnyWordSpec
+
+object VirtualThreadDispatcherSpec {
+  final case class ThreadInfo(virtual: Boolean, name: String)
+
+  object ThreadInfoActor {
+    def props() = Props(new ThreadInfoActor)
+  }
+  private class ThreadInfoActor extends Actor {
+    override def receive: Receive = {
+      case "give-me-info" =>
+        sender() ! reflectiveVirtualThreadInfo()
+    }
+  }
+
+  private def reflectiveVirtualThreadInfo(): ThreadInfo = {
+    val thread = Thread.currentThread()
+    // can't use methods directly or test won't compile on jdk < 21
+    val isVirtualMethod = thread.getClass.getMethod("isVirtual")
+    val isVirtual = isVirtualMethod.invoke(thread).asInstanceOf[Boolean]
+    ThreadInfo(isVirtual, thread.getName)
+  }
+}
+
+class VirtualThreadDispatcherSpec extends AnyWordSpec with Matchers {
+  import VirtualThreadDispatcherSpec._
+
+  "The virtual thread support" should {
+
+    "run tasks on virtual threads" in {
+      if (JavaVersion.majorVersion < 21) {
+        // loom not available yet here
+        pending
+      } else {
+        implicit val system: ActorSystem = ActorSystem(
+          classOf[VirtualThreadDispatcherSpec].getSimpleName,
+          ConfigFactory.parseString("""
+              my-vt-dispatcher {
+                type = "Dispatcher"
+                executor = virtual-thread-executor
+              }
+            """).withFallback(ConfigFactory.load()))
+
+        try {
+          val vtDispatcher = system.dispatchers.lookup("my-vt-dispatcher")
+          vtDispatcher shouldBe a[BatchingExecutor]
+
+          val threadIsVirtualProbe = TestProbe()
+          vtDispatcher.execute(() => {
+            threadIsVirtualProbe.ref ! reflectiveVirtualThreadInfo()
+          })
+          val info = threadIsVirtualProbe.expectMsgType[ThreadInfo]
+          info.virtual shouldBe true
+          info.name should endWith("my-vt-dispatcher")
+        } finally {
+          TestKit.shutdownActorSystem(system)
+        }
+
+      }
+    }
+
+    "can be used as default dispatcher" in {
+      if (JavaVersion.majorVersion < 21) {
+        // loom not available yet here
+        pending
+      } else {
+        // not necessarily a good idea because of the virtual thread per task overhead, but to know it works
+        // and to cover running actors on it
+        implicit val system: ActorSystem = ActorSystem(
+          classOf[VirtualThreadDispatcherSpec].getSimpleName,
+          ConfigFactory.parseString("""
+              akka.actor.default-dispatcher.executor="virtual-thread-executor"
+            """).withFallback(ConfigFactory.load()))
+        try {
+          val echo = system.actorOf(ThreadInfoActor.props())
+          val responseProbe = TestProbe()
+          echo.tell("give-me-info", responseProbe.ref)
+          val info = responseProbe.expectMsgType[ThreadInfo]
+          info.virtual shouldBe true
+          info.name should endWith("akka.actor.default-dispatcher")
+        } finally {
+          TestKit.shutdownActorSystem(system)
+        }
+      }
+    }
+
+    "can be configured with a fallback for work on all JVMs" in {
+      implicit val system: ActorSystem = ActorSystem(
+        classOf[VirtualThreadDispatcherSpec].getSimpleName,
+        ConfigFactory.parseString("""
+              my-vt-dispatcher {
+                type = "Dispatcher"
+                executor = virtual-thread-executor
+                virtual-thread-executor {
+                  fallback="fork-join-executor"
+                }
+              }
+            """).withFallback(ConfigFactory.load()))
+
+      try {
+        val dispatcher = system.dispatchers.lookup("my-vt-dispatcher")
+        val threadInfoProbe = TestProbe()
+        dispatcher.execute(() => {
+          threadInfoProbe.ref ! "ok"
+        })
+        threadInfoProbe.expectMsg("ok")
+      } finally {
+        TestKit.shutdownActorSystem(system)
+      }
+    }
+  }
+
+}

akka-actor/src/main/resources/reference.conf

@@ -370,6 +370,7 @@ akka {
       #  - "fork-join-executor" requires a "fork-join-executor" section
       #  - "thread-pool-executor" requires a "thread-pool-executor" section
       #  - "affinity-pool-executor" requires an "affinity-pool-executor" section
+      #  - "virtual-thread-executor" requires a "virtual-thread-executor" section
       #  - A FQCN of a class extending ExecutorServiceConfigurator
       executor = "default-executor"
 
@@ -527,6 +528,31 @@ akka {
         allow-core-timeout = on
       }
 
+      # This will be used if you have set "executor = "virtual-thread-executor"
+      # The virtual thread executor can only be used on JDK 21 an newer and runs each
+      # task in a virtual thread.
+      #
+      # Note that while running actors on this will make sure they do not interfer with each other by starving
+      # the dispatcher, an individual actor should still not be blocked since that means it cannot act on messages
+      # while blocked, for example system messages such as termination of the actor. Using the virtual-thread-executor
+      # comes at the price of some overhead, since each scheduled task is wrapped in a virtual thread instance, in
+      # benchmarks this has shown up as around 30% lower throughput, a higher allocation rate and 10% more time spent on GC
+      #
+      # For other use cases, such as future calls invoking blocking work, or using as blocking-io-dispatcher the executor
+      # may be a more sound choice.
+      #
+      # It is not possible to tune the carrier thread pool per executor however it is possible to globally
+      # control the JVM carrier thread pool size through system properties. See JVM
+      # Thread documentation for details: https://docs.oracle.com/en/java/javase/21/docs/api/java.base/java/lang/Thread.html
+      virtual-thread-executor {
+        # If run on a mixture of JDK versions, some not supporting virtual threads (JDK 17 and earlier),
+        # this can be used to specify an alternative executor to make sure an application always works
+        # by default this is empty so that expecting to be able to run virtual threads when it is not possible
+        # will not go unnoticed.
+        fallback = ""
+      }
+
+
       # How long time the dispatcher will wait for new actors until it shuts down
       shutdown-timeout = 1s
 

akka-actor/src/main/scala/akka/dispatch/AbstractDispatcher.scala

@@ -6,21 +6,19 @@ package akka.dispatch
 
 import java.{ util => ju }
 import java.util.concurrent._
-
 import scala.annotation.nowarn
 import scala.annotation.tailrec
 import scala.concurrent.{ ExecutionContext, ExecutionContextExecutor }
 import scala.concurrent.duration.{ Duration, FiniteDuration }
 import scala.util.control.NonFatal
-
 import com.typesafe.config.Config
-
 import akka.actor._
 import akka.annotation.InternalStableApi
 import akka.dispatch.affinity.AffinityPoolConfigurator
 import akka.dispatch.sysmsg._
 import akka.event.EventStream
 import akka.event.Logging.{ Debug, Error, LogEventException }
+import akka.util.JavaVersion
 import akka.util.{ Index, Unsafe }
 
 final case class Envelope private (message: Any, sender: ActorRef) {
@@ -365,6 +363,19 @@ abstract class MessageDispatcherConfigurator(_config: Config, val prerequisites:
         new ThreadPoolExecutorConfigurator(config.getConfig("thread-pool-executor"), prerequisites)
       case "affinity-pool-executor" =>
         new AffinityPoolConfigurator(config.getConfig("affinity-pool-executor"), prerequisites)
+      case "virtual-thread-executor" =>
+        val executorConfig = config.getConfig("virtual-thread-executor")
+        if (VirtualThreadConfigurator.virtualThreadsSupported()) {
+          new VirtualThreadConfigurator(executorConfig, prerequisites)
+        } else {
+          val fallbackExecutorName = executorConfig.getString("fallback")
+          if (fallbackExecutorName.isEmpty)
+            throw new RuntimeException(
+              s"Dispatcher configured to use virtual threads, but JVM version ${JavaVersion.majorVersion} does not support that. " +
+              "Use a newer Java version (21 or later), or configure 'virtual-thread-executor.fallback' for an alternative executor on older Java versions.")
+          else
+            configurator(fallbackExecutorName)
+        }
 
       case fqcn =>
         val args = List(classOf[Config] -> config, classOf[DispatcherPrerequisites] -> prerequisites)

akka-actor/src/main/scala/akka/dispatch/VirtualThreadConfigurator.scala

@@ -0,0 +1,98 @@
+/*
+ * Copyright (C) 2009-2025 Lightbend Inc. <https://www.lightbend.com>
+ */
+
+package akka.dispatch
+
+import akka.actor.DynamicAccess
+import akka.annotation.InternalApi
+import akka.util.JavaVersion
+import com.typesafe.config.Config
+
+import java.util.concurrent.ExecutorService
+import java.util.concurrent.Executors
+import java.util.concurrent.ThreadFactory
+
+/**
+ * INTERNAL API
+ */
+@InternalApi
+private[akka] object VirtualThreadConfigurator {
+
+  def virtualThreadsSupported(): Boolean = JavaVersion.majorVersion >= 21
+
+  // Note: since we still support JDK 11 and 17, we need to access the factory method reflectively
+  private def createVirtualThreadFactory(
+      dynamicAccess: DynamicAccess,
+      virtualThreadName: String,
+      uncaughtExceptionHandler: Option[Thread.UncaughtExceptionHandler]): ThreadFactory = {
+    val ofVirtualMethod =
+      try {
+        classOf[Thread].getMethod("ofVirtual")
+      } catch {
+        case _: NoSuchMethodError =>
+          throw new IllegalStateException("Virtual thread executors only supported on JDK 21 and newer")
+      }
+    val ofVirtual = ofVirtualMethod.invoke(null)
+    val ofVirtualInterface = dynamicAccess.getClassFor[AnyRef]("java.lang.Thread$Builder$OfVirtual").get
+
+    // thread names
+    val ofVirtualWithName =
+      if (virtualThreadName.nonEmpty) {
+        val nameMethod = ofVirtualInterface.getMethod("name", classOf[String])
+        nameMethod.invoke(ofVirtual, virtualThreadName)
+      } else ofVirtual
+
+    // uncaught exception handler
+    val ofVirtualWithUEH = uncaughtExceptionHandler match {
+      case Some(ueh) =>
+        val uncaughtExceptionHandlerMethod =
+          ofVirtualInterface.getMethod("uncaughtExceptionHandler", classOf[Thread.UncaughtExceptionHandler])
+        uncaughtExceptionHandlerMethod.invoke(ofVirtualWithName, ueh)
+      case None => ofVirtualWithName
+    }
+    val factoryMethod = ofVirtualInterface.getMethod("factory")
+    factoryMethod.invoke(ofVirtualWithUEH).asInstanceOf[ThreadFactory]
+  }
+
+  private def threadPerTaskExecutor(threadFactory: ThreadFactory): ExecutorService = {
+    val newThreadPerTaskMethod = classOf[Executors].getMethod("newThreadPerTaskExecutor", classOf[ThreadFactory])
+    newThreadPerTaskMethod.invoke(null, threadFactory).asInstanceOf[ExecutorService]
+  }
+
+  private class VirtualThreadExecutorServiceFactory(tf: ThreadFactory) extends ExecutorServiceFactory {
+    override def createExecutorService: ExecutorService = threadPerTaskExecutor(tf)
+  }
+
+}
+
+/**
+ * INTERNAL API
+ */
+@InternalApi
+private[akka] class VirtualThreadConfigurator(config: Config, prerequisites: DispatcherPrerequisites)
+    extends ExecutorServiceConfigurator(config, prerequisites) {
+  import VirtualThreadConfigurator._
+
+  override def createExecutorServiceFactory(id: String, threadFactory: ThreadFactory): ExecutorServiceFactory = {
+
+    val tf = threadFactory match {
+      case m: MonitorableThreadFactory =>
+        // add the dispatcher id to the thread names
+        val virtualThreadFactory =
+          createVirtualThreadFactory(prerequisites.dynamicAccess, m.name + "-" + id, Some(m.exceptionHandler))
+
+        // Note: daemonic false not allowed for virtual threads, so we ignore that
+        m.contextClassLoader.fold[ThreadFactory](virtualThreadFactory)(classLoader =>
+          (r: Runnable) => {
+            val virtualThread = virtualThreadFactory.newThread(r)
+            virtualThread.setContextClassLoader(classLoader)
+            virtualThread
+          })
+
+      case _ => createVirtualThreadFactory(prerequisites.dynamicAccess, "-" + id, None)
+    }
+    new VirtualThreadExecutorServiceFactory(tf)
+  }
+
+}

akka-bench-jmh/src/main/scala/akka/actor/ActorBenchmark.scala

@@ -45,7 +45,7 @@ class ActorBenchmark {
       "akka.actor.JCToolsMailbox"))
   var mailbox = ""
 
-  @Param(Array("fjp-dispatcher")) //  @Param(Array("fjp-dispatcher", "affinity-dispatcher"))
+  @Param(Array("fjp-dispatcher", "vt-dispatcher")) //  @Param(Array("fjp-dispatcher", "affinity-dispatcher"))
   var dispatcher = ""
 
   implicit var system: ActorSystem = _
@@ -85,6 +85,11 @@ class ActorBenchmark {
             throughput = $tpt
             mailbox-type = "$mailbox"
          }
+         vt-dispatcher {
+           executor = "virtual-thread-executor"
+           throughput = $tpt
+           mailbox-type = "$mailbox"
+         }
        }
       """))
   }

akka-docs/src/main/paradox/typed/dispatchers.md

@@ -330,6 +330,28 @@ applications.
 
 For a similar discussion specifically about Akka HTTP, refer to @extref[Handling blocking operations in Akka HTTP](akka.http:handling-blocking-operations-in-akka-http-routes.html).
 
+### Solution: Virtual threads dispatcher for blocking operations
+
+If running on Java 21 or later, it is possible to use virtual threads for a blocking dispatcher, configure
+the executor of the dispatcher to be `virtual-thread-executor`.
+
+The virtual thread executor will run every task in a virtual thread, which can detach from of the OS-level thread
+when it is waiting for a blocking operation, much like how an async task allows threads to be handed back to
+a thread pool, until some task completes.
+
+Re-configuring the built-in blocking dispatcher to use virtual threads can be done like this:
+
+```ruby
+akka.actor.default-blocking-io-dispatcher {
+  executor = "virtual-thread-executor"
+}
+```
+
+Note that there is a difference in behavior compared to using a thread pool dispatcher in that there is no limit
+to how many virtual threads can block, for example hitting a service and waiting for a response, 
+while the threadpool executor puts an upper limit (16 by default) on how many threads are actually in flight, 
+once that limit has been reached, additional tasks are queued until a thread becomes available.
+
 ### Available solutions to blocking operations
 
 The non-exhaustive list of adequate solutions to the “blocking problem”