fix: disable batching for vt executor (#32785)

* fix: Disable batching for virtual thread executor dispatcher

* make tests work on older JDKs

* actually use marker and correct invert of flag

* actually working as intended, more generous timeouts

Author: johanandren

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

@@ -14,6 +14,9 @@ import com.typesafe.config.ConfigFactory
 import org.scalatest.matchers.should.Matchers
 import org.scalatest.wordspec.AnyWordSpec
 
+import scala.concurrent.ExecutionContext
+import scala.concurrent.Future
+
 object VirtualThreadDispatcherSpec {
   final case class ThreadInfo(virtual: Boolean, name: String)
 
@@ -122,6 +125,50 @@ class VirtualThreadDispatcherSpec extends AnyWordSpec with Matchers {
         TestKit.shutdownActorSystem(system)
       }
     }
+
+    "nested works as expected" 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
+                virtual-thread-executor {
+                  fallback="fork-join-executor"
+                }
+              }
+            """).withFallback(ConfigFactory.load()))
+
+        try {
+          implicit val dispatcher: ExecutionContext = system.dispatchers.lookup("my-vt-dispatcher")
+          val threadInfoProbe = TestProbe()
+          // only available on JDK 19+ so we can't use without reflection
+          def currentThreadId(): Long =
+            classOf[Thread].getMethod("threadId").invoke(Thread.currentThread()).asInstanceOf[Long]
+
+          Future {
+            threadInfoProbe.ref ! (("parent before", currentThreadId()))
+            Future {
+              threadInfoProbe.ref ! (("child before", currentThreadId()))
+              Thread.sleep(300)
+              threadInfoProbe.ref ! (("child after", currentThreadId()))
+            }
+            Thread.sleep(100)
+            threadInfoProbe.ref ! (("parent after", currentThreadId()))
+          }
+          val all = threadInfoProbe.receiveN(4)
+          val items = all.collect { case (evt: String, threadId: Long) => (evt, threadId) }
+          items.map(_._2).toSet should have size (2) // two different virtual threads
+          items.map(_._1) shouldEqual (Seq("parent before", "child before", "parent after", "child after"))
+        } finally {
+          TestKit.shutdownActorSystem(system)
+        }
+      }
+    }
   }
 
 }

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

@@ -39,6 +39,8 @@ class Dispatcher(
 
   import configurator.prerequisites._
 
+  private val batchingEnabled = !executorServiceFactoryProvider.isInstanceOf[NoBatchingExecutorFactoryProvider]
+
   private class LazyExecutorServiceDelegate(factory: ExecutorServiceFactory) extends ExecutorServiceDelegate {
     lazy val executor: ExecutorService = factory.createExecutorService
     def copy(): LazyExecutorServiceDelegate = new LazyExecutorServiceDelegate(factory)
@@ -141,6 +143,10 @@ class Dispatcher(
     } else false
   }
 
+  override def batchable(runnable: Runnable): Boolean =
+    if (batchingEnabled) super.batchable(runnable)
+    else false
+
   override val toString: String = Logging.simpleName(this) + "[" + id + "]"
 }
 

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

@@ -65,6 +65,11 @@ trait ExecutorServiceFactoryProvider {
   def createExecutorServiceFactory(id: String, threadFactory: ThreadFactory): ExecutorServiceFactory
 }
 
+/**
+ * Marker trait to disable dispatcher batching for a given executor
+ */
+trait NoBatchingExecutorFactoryProvider {}
+
 /**
  * A small configuration DSL to create ThreadPoolExecutors that can be provided as an ExecutorServiceFactoryProvider to Dispatcher
  */

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

@@ -71,7 +71,8 @@ private[akka] object VirtualThreadConfigurator {
  */
 @InternalApi
 private[akka] class VirtualThreadConfigurator(config: Config, prerequisites: DispatcherPrerequisites)
-    extends ExecutorServiceConfigurator(config, prerequisites) {
+    extends ExecutorServiceConfigurator(config, prerequisites)
+    with NoBatchingExecutorFactoryProvider {
   import VirtualThreadConfigurator._
 
   override def createExecutorServiceFactory(id: String, threadFactory: ThreadFactory): ExecutorServiceFactory = {