You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pekko.apache.org by he...@apache.org on 2025/09/07 07:50:21 UTC
(pekko) 01/01: feat: Add virtualize support for thread-pool-executor
This is an automated email from the ASF dual-hosted git repository.
hepin pushed a commit to branch threadPool
in repository https://gitbox.apache.org/repos/asf/pekko.git
commit a4cbfc52c17ac92b0df660c9c6e7f3c074c132bc
Author: He-Pin <he...@gmail.com>
AuthorDate: Sun Sep 7 15:48:03 2025 +0800
feat: Add virtualize support for thread-pool-executor
---
.../dispatch/ForkJoinPoolVirtualThreadSpec.scala | 2 +-
...pec.scala => ThreadPoolVirtualThreadSpec.scala} | 19 ++---
actor/src/main/resources/reference.conf | 23 ++++++
.../apache/pekko/dispatch/AbstractDispatcher.scala | 8 +-
.../dispatch/ForkJoinExecutorConfigurator.scala | 27 +------
.../apache/pekko/dispatch/ThreadPoolBuilder.scala | 90 ++++++++++++++++++----
.../pekko/dispatch/VirtualThreadSupport.scala | 4 +-
7 files changed, 115 insertions(+), 58 deletions(-)
diff --git a/actor-tests/src/test/scala-jdk21-only/org/apache/pekko/dispatch/ForkJoinPoolVirtualThreadSpec.scala b/actor-tests/src/test/scala-jdk21-only/org/apache/pekko/dispatch/ForkJoinPoolVirtualThreadSpec.scala
index 455a4e2f1f..1d4ca6db94 100644
--- a/actor-tests/src/test/scala-jdk21-only/org/apache/pekko/dispatch/ForkJoinPoolVirtualThreadSpec.scala
+++ b/actor-tests/src/test/scala-jdk21-only/org/apache/pekko/dispatch/ForkJoinPoolVirtualThreadSpec.scala
@@ -22,7 +22,6 @@ import com.typesafe.config.ConfigFactory
import org.apache.pekko
import pekko.actor.{ Actor, Props }
import pekko.testkit.{ ImplicitSender, PekkoSpec }
-import pekko.util.JavaVersion
object ForkJoinPoolVirtualThreadSpec {
val config = ConfigFactory.parseString("""
@@ -30,6 +29,7 @@ object ForkJoinPoolVirtualThreadSpec {
| task-dispatcher {
| mailbox-type = "org.apache.pekko.dispatch.SingleConsumerOnlyUnboundedMailbox"
| throughput = 5
+ | executor = "fork-join-executor"
| fork-join-executor {
| parallelism-factor = 2
| parallelism-max = 2
diff --git a/actor-tests/src/test/scala-jdk21-only/org/apache/pekko/dispatch/ForkJoinPoolVirtualThreadSpec.scala b/actor-tests/src/test/scala-jdk21-only/org/apache/pekko/dispatch/ThreadPoolVirtualThreadSpec.scala
similarity index 77%
copy from actor-tests/src/test/scala-jdk21-only/org/apache/pekko/dispatch/ForkJoinPoolVirtualThreadSpec.scala
copy to actor-tests/src/test/scala-jdk21-only/org/apache/pekko/dispatch/ThreadPoolVirtualThreadSpec.scala
index 455a4e2f1f..583f911877 100644
--- a/actor-tests/src/test/scala-jdk21-only/org/apache/pekko/dispatch/ForkJoinPoolVirtualThreadSpec.scala
+++ b/actor-tests/src/test/scala-jdk21-only/org/apache/pekko/dispatch/ThreadPoolVirtualThreadSpec.scala
@@ -18,22 +18,19 @@
package org.apache.pekko.dispatch
import com.typesafe.config.ConfigFactory
-
import org.apache.pekko
import pekko.actor.{ Actor, Props }
import pekko.testkit.{ ImplicitSender, PekkoSpec }
-import pekko.util.JavaVersion
-object ForkJoinPoolVirtualThreadSpec {
+object ThreadPoolVirtualThreadSpec {
val config = ConfigFactory.parseString("""
|custom {
| task-dispatcher {
| mailbox-type = "org.apache.pekko.dispatch.SingleConsumerOnlyUnboundedMailbox"
| throughput = 5
- | fork-join-executor {
- | parallelism-factor = 2
- | parallelism-max = 2
- | parallelism-min = 2
+ | executor = "thread-pool-executor"
+ | thread-pool-executor {
+ | fixed-pool-size = 4
| virtualize = on
| }
| }
@@ -50,17 +47,17 @@ object ForkJoinPoolVirtualThreadSpec {
}
-class ForkJoinPoolVirtualThreadSpec extends PekkoSpec(ForkJoinPoolVirtualThreadSpec.config) with ImplicitSender {
- import ForkJoinPoolVirtualThreadSpec._
+class ThreadPoolVirtualThreadSpec extends PekkoSpec(ThreadPoolVirtualThreadSpec.config) with ImplicitSender {
+ import ThreadPoolVirtualThreadSpec._
- "PekkoForkJoinPool" must {
+ "ThreadPool" must {
"support virtualization with Virtual Thread" in {
val actor = system.actorOf(Props(new ThreadNameActor).withDispatcher("custom.task-dispatcher"))
for (_ <- 1 to 1000) {
actor ! "ping"
expectMsgPF() { case name: String =>
- name should include("ForkJoinPoolVirtualThreadSpec-custom.task-dispatcher-virtual-thread-")
+ name should include("ThreadPoolVirtualThreadSpec-custom.task-dispatcher-virtual-thread-")
}
}
}
diff --git a/actor/src/main/resources/reference.conf b/actor/src/main/resources/reference.conf
index 0cb04fabcc..a0500258ba 100644
--- a/actor/src/main/resources/reference.conf
+++ b/actor/src/main/resources/reference.conf
@@ -543,6 +543,18 @@ pekko {
# Allow core threads to time out
allow-core-timeout = on
+
+ # This config is new in Pekko v1.2.0 and only has an effect if you are running with JDK 21 and above,
+ # When set to `on` but underlying runtime does not support virtual threads, an Exception will throw.
+ # Virtualize this dispatcher as a virtual-thread-executor
+ # Valid values are: `on`, `off`
+ #
+ # Requirements:
+ # 1. JDK 21+
+ # 2. add options to the JVM:
+ # --add-opens=java.base/jdk.internal.misc=ALL-UNNAMED
+ # --add-opens=java.base/java.lang=ALL-UNNAMED
+ virtualize = off
}
# This will be used if you have set "executor = "virtual-thread-executor"
@@ -600,6 +612,17 @@ pekko {
thread-pool-executor {
fixed-pool-size = 16
+ # This config is new in Pekko v1.2.0 and only has an effect if you are running with JDK 21 and above,
+ # When set to `on` but underlying runtime does not support virtual threads, an Exception will throw.
+ # Virtualize this dispatcher as a virtual-thread-executor
+ # Valid values are: `on`, `off`
+ #
+ # Requirements:
+ # 1. JDK 21+
+ # 2. add options to the JVM:
+ # --add-opens=java.base/jdk.internal.misc=ALL-UNNAMED
+ # --add-opens=java.base/java.lang=ALL-UNNAMED
+ virtualize = off
}
}
diff --git a/actor/src/main/scala/org/apache/pekko/dispatch/AbstractDispatcher.scala b/actor/src/main/scala/org/apache/pekko/dispatch/AbstractDispatcher.scala
index 274c58c53f..b89131d3f9 100644
--- a/actor/src/main/scala/org/apache/pekko/dispatch/AbstractDispatcher.scala
+++ b/actor/src/main/scala/org/apache/pekko/dispatch/AbstractDispatcher.scala
@@ -28,7 +28,7 @@ import pekko.dispatch.affinity.AffinityPoolConfigurator
import pekko.dispatch.sysmsg._
import pekko.event.EventStream
import pekko.event.Logging.{ emptyMDC, Debug, Error, LogEventException, Warning }
-import pekko.util.{ unused, Index }
+import pekko.util.{ unused, Index, JavaVersion }
import com.typesafe.config.Config
@@ -466,7 +466,7 @@ final class VirtualThreadExecutorConfigurator(config: Config, prerequisites: Dis
class ThreadPoolExecutorConfigurator(config: Config, prerequisites: DispatcherPrerequisites)
extends ExecutorServiceConfigurator(config, prerequisites) {
-
+ override val isVirtualized: Boolean = config.getBoolean("virtualize") && JavaVersion.majorVersion >= 21
val threadPoolConfig: ThreadPoolConfig = createThreadPoolConfigBuilder(config, prerequisites).config
protected def createThreadPoolConfigBuilder(
@@ -477,6 +477,7 @@ class ThreadPoolExecutorConfigurator(config: Config, prerequisites: DispatcherPr
ThreadPoolConfigBuilder(ThreadPoolConfig())
.setKeepAliveTime(config.getMillisDuration("keep-alive-time"))
.setAllowCoreThreadTimeout(config.getBoolean("allow-core-timeout"))
+ .isVirtualized(isVirtualized)
.configure(Some(config.getInt("task-queue-size")).flatMap {
case size if size > 0 =>
Some(config.getString("task-queue-type"))
@@ -506,8 +507,9 @@ class ThreadPoolExecutorConfigurator(config: Config, prerequisites: DispatcherPr
builder.setFixedPoolSize(config.getInt("fixed-pool-size"))
}
- def createExecutorServiceFactory(id: String, threadFactory: ThreadFactory): ExecutorServiceFactory =
+ def createExecutorServiceFactory(id: String, threadFactory: ThreadFactory): ExecutorServiceFactory = {
threadPoolConfig.createExecutorServiceFactory(id, threadFactory)
+ }
}
class DefaultExecutorServiceConfigurator(
diff --git a/actor/src/main/scala/org/apache/pekko/dispatch/ForkJoinExecutorConfigurator.scala b/actor/src/main/scala/org/apache/pekko/dispatch/ForkJoinExecutorConfigurator.scala
index 66d1e36918..49dd5da43f 100644
--- a/actor/src/main/scala/org/apache/pekko/dispatch/ForkJoinExecutorConfigurator.scala
+++ b/actor/src/main/scala/org/apache/pekko/dispatch/ForkJoinExecutorConfigurator.scala
@@ -15,10 +15,9 @@ package org.apache.pekko.dispatch
import com.typesafe.config.Config
import org.apache.pekko
-import pekko.dispatch.VirtualThreadSupport.newVirtualThreadFactory
import pekko.util.JavaVersion
-import java.util.concurrent.{ Executor, ExecutorService, ForkJoinPool, ForkJoinTask, ThreadFactory, TimeUnit }
+import java.util.concurrent.{ ExecutorService, ForkJoinPool, ForkJoinTask, ThreadFactory, TimeUnit }
object ForkJoinExecutorConfigurator {
@@ -114,28 +113,8 @@ class ForkJoinExecutorConfigurator(config: Config, prerequisites: DispatcherPrer
val pool = new PekkoForkJoinPool(parallelism, tf, maxPoolSize, MonitorableThreadFactory.doNothing, asyncMode)
if (isVirtualized) {
- // when virtualized, we need enhanced thread factory
- val factory: ThreadFactory = threadFactory match {
- case MonitorableThreadFactory(name, _, contextClassLoader, exceptionHandler, _) =>
- new ThreadFactory {
- private val vtFactory = newVirtualThreadFactory(name, pool) // use the pool as the scheduler
-
- override def newThread(r: Runnable): Thread = {
- val vt = vtFactory.newThread(r)
- vt.setUncaughtExceptionHandler(exceptionHandler)
- contextClassLoader.foreach(vt.setContextClassLoader)
- vt
- }
- }
- case _ => newVirtualThreadFactory(prerequisites.settings.name, pool); // use the pool as the scheduler
- }
- // wrap the pool with virtualized executor service
- new VirtualizedExecutorService(
- factory, // the virtual thread factory
- pool, // the underlying pool
- (_: Executor) => pool.atFullThrottle(), // the load metrics provider, we use the pool itself
- cascadeShutdown = true // cascade shutdown
- )
+ // we need to cast here,
+ createVirtualized(threadFactory.asInstanceOf[ThreadFactory], pool, prerequisites.settings.name)
} else {
pool
}
diff --git a/actor/src/main/scala/org/apache/pekko/dispatch/ThreadPoolBuilder.scala b/actor/src/main/scala/org/apache/pekko/dispatch/ThreadPoolBuilder.scala
index b10cc46724..73319a064e 100644
--- a/actor/src/main/scala/org/apache/pekko/dispatch/ThreadPoolBuilder.scala
+++ b/actor/src/main/scala/org/apache/pekko/dispatch/ThreadPoolBuilder.scala
@@ -13,11 +13,14 @@
package org.apache.pekko.dispatch
+import org.apache.pekko.dispatch.VirtualThreadSupport.newVirtualThreadFactory
+
import java.util.Collection
import java.util.concurrent.{
ArrayBlockingQueue,
BlockingQueue,
Callable,
+ Executor,
ExecutorService,
ForkJoinPool,
ForkJoinWorkerThread,
@@ -30,7 +33,6 @@ import java.util.concurrent.{
TimeUnit
}
import java.util.concurrent.atomic.{ AtomicLong, AtomicReference }
-
import scala.concurrent.{ BlockContext, CanAwait }
import scala.concurrent.duration.Duration
@@ -73,6 +75,33 @@ trait ExecutorServiceFactory {
trait ExecutorServiceFactoryProvider {
def createExecutorServiceFactory(id: String, threadFactory: ThreadFactory): ExecutorServiceFactory
def isVirtualized: Boolean = false // can be overridden by implementations
+
+ protected def createVirtualized(threadFactory: ThreadFactory,
+ pool: ExecutorService with LoadMetrics,
+ prefixName: String): ExecutorService = {
+ // when virtualized, we need enhanced thread factory
+ val factory: ThreadFactory = threadFactory match {
+ case MonitorableThreadFactory(name, _, contextClassLoader, exceptionHandler, _) =>
+ new ThreadFactory {
+ private val vtFactory = newVirtualThreadFactory(name, pool) // use the pool as the scheduler
+
+ override def newThread(r: Runnable): Thread = {
+ val vt = vtFactory.newThread(r)
+ vt.setUncaughtExceptionHandler(exceptionHandler)
+ contextClassLoader.foreach(vt.setContextClassLoader)
+ vt
+ }
+ }
+ case _ => newVirtualThreadFactory(prefixName, pool); // use the pool as the scheduler
+ }
+ // wrap the pool with virtualized executor service
+ new VirtualizedExecutorService(
+ factory, // the virtual thread factory
+ pool, // the underlying pool
+ (_: Executor) => pool.atFullThrottle(), // the load metrics provider, we use the pool itself
+ cascadeShutdown = true // cascade shutdown
+ )
+ }
}
/**
@@ -84,7 +113,9 @@ final case class ThreadPoolConfig(
maxPoolSize: Int = ThreadPoolConfig.defaultMaxPoolSize,
threadTimeout: Duration = ThreadPoolConfig.defaultTimeout,
queueFactory: ThreadPoolConfig.QueueFactory = ThreadPoolConfig.linkedBlockingQueue(),
- rejectionPolicy: RejectedExecutionHandler = ThreadPoolConfig.defaultRejectionPolicy)
+ rejectionPolicy: RejectedExecutionHandler = ThreadPoolConfig.defaultRejectionPolicy,
+ override val isVirtualized: Boolean = false) // whether to use virtual threads if supported
+//TODO ThreadPoolConfig should not extend ExecutorServiceFactoryProvider ?
extends ExecutorServiceFactoryProvider {
// Written explicitly to permit non-inlined defn; this is necessary for downstream instrumentation that stores extra
// context information on the config
@@ -95,26 +126,48 @@ final case class ThreadPoolConfig(
maxPoolSize: Int = maxPoolSize,
threadTimeout: Duration = threadTimeout,
queueFactory: ThreadPoolConfig.QueueFactory = queueFactory,
- rejectionPolicy: RejectedExecutionHandler = rejectionPolicy
+ rejectionPolicy: RejectedExecutionHandler = rejectionPolicy,
+ isVirtualized: Boolean = isVirtualized
): ThreadPoolConfig =
- ThreadPoolConfig(allowCorePoolTimeout, corePoolSize, maxPoolSize, threadTimeout, queueFactory, rejectionPolicy)
+ ThreadPoolConfig(allowCorePoolTimeout, corePoolSize, maxPoolSize, threadTimeout, queueFactory, rejectionPolicy,
+ isVirtualized)
- class ThreadPoolExecutorServiceFactory(val threadFactory: ThreadFactory) extends ExecutorServiceFactory {
+ private class ThreadPoolExecutorServiceFactory(id: String, val threadFactory: ThreadFactory)
+ extends ExecutorServiceFactory {
def createExecutorService: ExecutorService = {
- val service: ThreadPoolExecutor = new ThreadPoolExecutor(
- corePoolSize,
- maxPoolSize,
- threadTimeout.length,
- threadTimeout.unit,
- queueFactory(),
- threadFactory,
- rejectionPolicy) with LoadMetrics {
- def atFullThrottle(): Boolean = this.getActiveCount >= this.getPoolSize
+ val tf = if (isVirtualized) {
+ threadFactory match {
+ // we need to use the thread factory to create carrier thread
+ case m: MonitorableThreadFactory => m.withName(m.name + "-CarrierThread")
+ case _ => threadFactory
+ }
+ } else threadFactory
+
+ val pool: ThreadPoolExecutor with LoadMetrics = {
+ val threadPool = new ThreadPoolExecutor(
+ corePoolSize,
+ maxPoolSize,
+ threadTimeout.length,
+ threadTimeout.unit,
+ queueFactory(),
+ tf,
+ rejectionPolicy) with LoadMetrics {
+ def atFullThrottle(): Boolean = this.getActiveCount >= this.getPoolSize
+ }
+ threadPool.allowCoreThreadTimeOut(allowCorePoolTimeout)
+ threadPool
}
- service.allowCoreThreadTimeOut(allowCorePoolTimeout)
- service
+
+ if (isVirtualized) {
+ val prefixName = threadFactory match {
+ case m: MonitorableThreadFactory => m.name
+ case _ => id
+ }
+ createVirtualized(threadFactory, pool, prefixName)
+ } else pool
}
}
+
def createExecutorServiceFactory(id: String, threadFactory: ThreadFactory): ExecutorServiceFactory = {
val tf = threadFactory match {
case m: MonitorableThreadFactory =>
@@ -122,7 +175,7 @@ final case class ThreadPoolConfig(
m.withName(m.name + "-" + id)
case other => other
}
- new ThreadPoolExecutorServiceFactory(tf)
+ new ThreadPoolExecutorServiceFactory(id, tf)
}
}
@@ -179,6 +232,9 @@ final case class ThreadPoolConfigBuilder(config: ThreadPoolConfig) {
def setQueueFactory(newQueueFactory: QueueFactory): ThreadPoolConfigBuilder =
this.copy(config = config.copy(queueFactory = newQueueFactory))
+ def isVirtualized(isVirtualized: Boolean): ThreadPoolConfigBuilder =
+ this.copy(config = config.copy(isVirtualized = isVirtualized))
+
def configure(fs: Option[Function[ThreadPoolConfigBuilder, ThreadPoolConfigBuilder]]*): ThreadPoolConfigBuilder =
fs.foldLeft(this)((c, f) => f.map(_(c)).getOrElse(c))
}
diff --git a/actor/src/main/scala/org/apache/pekko/dispatch/VirtualThreadSupport.scala b/actor/src/main/scala/org/apache/pekko/dispatch/VirtualThreadSupport.scala
index 86dea7d581..a51f6161a4 100644
--- a/actor/src/main/scala/org/apache/pekko/dispatch/VirtualThreadSupport.scala
+++ b/actor/src/main/scala/org/apache/pekko/dispatch/VirtualThreadSupport.scala
@@ -93,8 +93,8 @@ private[dispatch] object VirtualThreadSupport {
}
val nameMethod = ofVirtualClass.getDeclaredMethod("name", classOf[String], classOf[Long])
val factoryMethod = builderClass.getDeclaredMethod("factory")
- val zero = java.lang.Long.valueOf(0L)
- builder = nameMethod.invoke(builder, prefix + "-virtual-thread-", zero)
+ // TODO drop the counter?
+ builder = nameMethod.invoke(builder, prefix + "-virtual-thread-", 0L)
factoryMethod.invoke(builder).asInstanceOf[ThreadFactory]
} catch {
case NonFatal(e) =>
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pekko.apache.org
For additional commands, e-mail: commits-help@pekko.apache.org