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/20 07:34:48 UTC
(pekko) 01/01: feat: Support setting starting number of virtual thread.
This is an automated email from the ASF dual-hosted git repository.
hepin pushed a commit to branch vtName
in repository https://gitbox.apache.org/repos/asf/pekko.git
commit c5fc6a4ff59d4218176696f24f99bfce731c1a81
Author: He-Pin <he...@gmail.com>
AuthorDate: Sat Sep 20 15:34:34 2025 +0800
feat: Support setting starting number of virtual thread.
---
.../dispatch/ForkJoinPoolVirtualThreadSpec.scala | 20 +++++++++
.../dispatch/ThreadPoolVirtualThreadSpec.scala | 21 ++++++++-
actor/src/main/resources/reference.conf | 18 +++++++-
.../apache/pekko/dispatch/AbstractDispatcher.scala | 16 +++----
.../dispatch/ForkJoinExecutorConfigurator.scala | 8 ++--
.../apache/pekko/dispatch/ThreadPoolBuilder.scala | 19 +++++---
.../pekko/dispatch/VirtualThreadSupport.scala | 52 +++++++++++-----------
7 files changed, 109 insertions(+), 45 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 550f34ef90..9b4611cc6b 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
@@ -37,6 +37,18 @@ object ForkJoinPoolVirtualThreadSpec {
| virtualize = on
| }
| }
+ | task-dispatcher-short {
+ | mailbox-type = "org.apache.pekko.dispatch.SingleConsumerOnlyUnboundedMailbox"
+ | throughput = 5
+ | executor = "fork-join-executor"
+ | fork-join-executor {
+ | parallelism-factor = 2
+ | parallelism-max = 2
+ | parallelism-min = 2
+ | virtualize = on
+ | virtual-thread-start-number = -1
+ | }
+ | }
|}
""".stripMargin)
@@ -63,6 +75,14 @@ class ForkJoinPoolVirtualThreadSpec extends PekkoSpec(ForkJoinPoolVirtualThreadS
name should include("ForkJoinPoolVirtualThreadSpec-custom.task-dispatcher-virtual-thread-")
}
}
+
+ val actor2 = system.actorOf(Props(new ThreadNameActor).withDispatcher("custom.task-dispatcher-short"))
+ for (_ <- 1 to 1000) {
+ actor2 ! "ping"
+ expectMsgPF() { case name: String =>
+ name should include("ForkJoinPoolVirtualThreadSpec-custom.task-dispatcher-short-virtual-thread")
+ }
+ }
}
}
diff --git a/actor-tests/src/test/scala-jdk21-only/org/apache/pekko/dispatch/ThreadPoolVirtualThreadSpec.scala b/actor-tests/src/test/scala-jdk21-only/org/apache/pekko/dispatch/ThreadPoolVirtualThreadSpec.scala
index 406730c233..66e3766b82 100644
--- a/actor-tests/src/test/scala-jdk21-only/org/apache/pekko/dispatch/ThreadPoolVirtualThreadSpec.scala
+++ b/actor-tests/src/test/scala-jdk21-only/org/apache/pekko/dispatch/ThreadPoolVirtualThreadSpec.scala
@@ -31,10 +31,21 @@ object ThreadPoolVirtualThreadSpec {
| throughput = 5
| executor = "thread-pool-executor"
| thread-pool-executor {
- | fixed-pool-size = 4
+ | fixed-pool-size = 1
| virtualize = on
| }
| }
+ |
+ | task-dispatcher-short {
+ | mailbox-type = "org.apache.pekko.dispatch.SingleConsumerOnlyUnboundedMailbox"
+ | throughput = 5
+ | executor = "thread-pool-executor"
+ | thread-pool-executor {
+ | fixed-pool-size = 1
+ | virtualize = on
+ | virtual-thread-start-number = -1
+ | }
+ | }
|}
""".stripMargin)
@@ -61,6 +72,14 @@ class ThreadPoolVirtualThreadSpec extends PekkoSpec(ThreadPoolVirtualThreadSpec.
name should include("ThreadPoolVirtualThreadSpec-custom.task-dispatcher-virtual-thread-")
}
}
+
+ val actor2 = system.actorOf(Props(new ThreadNameActor).withDispatcher("custom.task-dispatcher-short"))
+ for (_ <- 1 to 1000) {
+ actor2 ! "ping"
+ expectMsgPF() { case name: String =>
+ name should include("ThreadPoolVirtualThreadSpec-custom.task-dispatcher-short-virtual-thread")
+ }
+ }
}
}
diff --git a/actor/src/main/resources/reference.conf b/actor/src/main/resources/reference.conf
index 740c2975b4..e1c720fe22 100644
--- a/actor/src/main/resources/reference.conf
+++ b/actor/src/main/resources/reference.conf
@@ -492,6 +492,10 @@ pekko {
# --add-opens=java.base/jdk.internal.misc=ALL-UNNAMED
# --add-opens=java.base/java.lang=ALL-UNNAMED
virtualize = off
+
+ # When virtualize = on, you can set the starting id for virtual threads created by this dispatcher.
+ # -1 to turn off, 0 (default) to start from 0, or any positive number to start from that number.
+ virtual-thread-start-number = 0
}
# This will be used if you have set "executor = "thread-pool-executor""
@@ -529,7 +533,7 @@ pekko {
# ceil(available processors * factor)
# The maximumPoolSize will not be less than corePoolSize.
# It is only used if using a bounded task queue.
- max-pool-size-factor = 3.0
+ max-pool-size-factor = 3.0
# Max number of threads to cap factor-based maximumPoolSize number to
max-pool-size-max = 64
@@ -555,6 +559,10 @@ pekko {
# --add-opens=java.base/jdk.internal.misc=ALL-UNNAMED
# --add-opens=java.base/java.lang=ALL-UNNAMED
virtualize = off
+
+ # When virtualize = on, you can set the starting naumber for virtual threads created by this dispatcher.
+ # -1 to turn off, 0 (default) to start from 0, or any positive number to start from that number.
+ virtual-thread-start-number = 0
}
# This will be used if you have set "executor = "virtual-thread-executor"
@@ -569,6 +577,10 @@ pekko {
#jdk.virtualThreadScheduler.minRunnable
#jdk.unparker.maxPoolSize
fallback = "fork-join-executor"
+
+ # When using virtual-thread-executor, you can set the starting naumber for virtual threads created by this dispatcher.
+ # -1 to turn off, 0 (default) to start from 0, or any positive number to start from that number.
+ virtual-thread-start-number = 0
}
# How long time the dispatcher will wait for new actors until it shuts down
shutdown-timeout = 1s
@@ -623,6 +635,10 @@ pekko {
# --add-opens=java.base/jdk.internal.misc=ALL-UNNAMED
# --add-opens=java.base/java.lang=ALL-UNNAMED
virtualize = off
+
+ # When virtualize = on, you can set the starting naumber for virtual threads created by this dispatcher.
+ # -1 to turn off, 0 (default) to start from 0, or any positive number to start from that number.
+ virtual-thread-start-number = 0
}
}
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 4c4e74d448..142ac836fe 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.{ Index, JavaVersion }
+import pekko.util.Index
import com.typesafe.config.Config
@@ -412,14 +412,14 @@ abstract class MessageDispatcherConfigurator(_config: Config, val prerequisites:
final class VirtualThreadExecutorConfigurator(config: Config, prerequisites: DispatcherPrerequisites)
extends ExecutorServiceConfigurator(config, prerequisites) {
- override def isVirtualized: Boolean = true
-
+ override def isVirtualized: Boolean = VirtualThreadSupport.isSupported
+ override def virtualThreadStartNumber: Int = config.getInt("virtual-thread-start-number")
override def createExecutorServiceFactory(id: String, threadFactory: ThreadFactory): ExecutorServiceFactory = {
import VirtualThreadSupport._
val tf: ThreadFactory = threadFactory match {
case MonitorableThreadFactory(name, _, contextClassLoader, exceptionHandler, _) =>
new ThreadFactory {
- private val vtFactory = newVirtualThreadFactory(name + "-" + id)
+ private val vtFactory = newVirtualThreadFactory(name + "-" + id, virtualThreadStartNumber)
override def newThread(r: Runnable): Thread = {
val vt = vtFactory.newThread(r)
@@ -428,7 +428,7 @@ final class VirtualThreadExecutorConfigurator(config: Config, prerequisites: Dis
vt
}
}
- case _ => newVirtualThreadFactory(prerequisites.settings.name + "-" + id);
+ case _ => newVirtualThreadFactory(prerequisites.settings.name + "-" + id, virtualThreadStartNumber);
}
new ExecutorServiceFactory {
override def createExecutorService: ExecutorService with LoadMetrics = {
@@ -502,7 +502,7 @@ trait ThreadPoolExecutorServiceFactoryProvider extends ExecutorServiceFactoryPro
case m: MonitorableThreadFactory => m.name + "-" + id
case _ => id
}
- createVirtualized(tf, pool, prefixName)
+ createVirtualized(tf, pool, prefixName, virtualThreadStartNumber)
} else pool
}
}
@@ -514,8 +514,8 @@ class ThreadPoolExecutorConfigurator(config: Config, prerequisites: DispatcherPr
extends ExecutorServiceConfigurator(config, prerequisites)
with ThreadPoolExecutorServiceFactoryProvider {
override val threadPoolConfig: ThreadPoolConfig = createThreadPoolConfigBuilder(config, prerequisites).config
- override val isVirtualized: Boolean = threadPoolConfig.isVirtualized && JavaVersion.majorVersion >= 21
-
+ override val isVirtualized: Boolean = threadPoolConfig.isVirtualized && VirtualThreadSupport.isSupported
+ override def virtualThreadStartNumber: Int = config.getInt("virtual-thread-start-number")
protected def createThreadPoolConfigBuilder(
config: Config,
@nowarn("msg=never used") prerequisites: DispatcherPrerequisites): ThreadPoolConfigBuilder = {
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 7aa1aeabdd..e5d7e7f8ce 100644
--- a/actor/src/main/scala/org/apache/pekko/dispatch/ForkJoinExecutorConfigurator.scala
+++ b/actor/src/main/scala/org/apache/pekko/dispatch/ForkJoinExecutorConfigurator.scala
@@ -15,9 +15,6 @@ package org.apache.pekko.dispatch
import java.util.concurrent.{ ExecutorService, ForkJoinPool, ForkJoinTask, ThreadFactory, TimeUnit }
-import org.apache.pekko
-import pekko.util.JavaVersion
-
import com.typesafe.config.Config
object ForkJoinExecutorConfigurator {
@@ -73,7 +70,8 @@ object ForkJoinExecutorConfigurator {
class ForkJoinExecutorConfigurator(config: Config, prerequisites: DispatcherPrerequisites)
extends ExecutorServiceConfigurator(config, prerequisites) {
import ForkJoinExecutorConfigurator._
- final override val isVirtualized: Boolean = config.getBoolean("virtualize") && JavaVersion.majorVersion >= 21
+ final override val isVirtualized: Boolean = config.getBoolean("virtualize") && VirtualThreadSupport.isSupported
+ override def virtualThreadStartNumber: Int = config.getInt("virtual-thread-start-number")
def validate(t: ThreadFactory): ForkJoinPool.ForkJoinWorkerThreadFactory = t match {
case correct: ForkJoinPool.ForkJoinWorkerThreadFactory => correct
@@ -115,7 +113,7 @@ class ForkJoinExecutorConfigurator(config: Config, prerequisites: DispatcherPrer
if (isVirtualized) {
// we need to cast here,
- createVirtualized(threadFactory.asInstanceOf[ThreadFactory], pool, id)
+ createVirtualized(threadFactory.asInstanceOf[ThreadFactory], pool, id, virtualThreadStartNumber)
} 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 093c5645f1..e1321c7cab 100644
--- a/actor/src/main/scala/org/apache/pekko/dispatch/ThreadPoolBuilder.scala
+++ b/actor/src/main/scala/org/apache/pekko/dispatch/ThreadPoolBuilder.scala
@@ -77,17 +77,26 @@ trait ExecutorServiceFactory {
*/
trait ExecutorServiceFactoryProvider {
def createExecutorServiceFactory(id: String, threadFactory: ThreadFactory): ExecutorServiceFactory
+
+ /**
+ * Whether the executor service created by this factory should use virtual threads.
+ */
def isVirtualized: Boolean = false // can be overridden by implementations
- protected def createVirtualized(
- threadFactory: ThreadFactory,
+ /**
+ * The starting number of the virtual thread name, if -1, the number will not be appended.
+ */
+ def virtualThreadStartNumber: Int = 0 // can be overridden by implementations
+
+ protected def createVirtualized(threadFactory: ThreadFactory,
pool: ExecutorService with LoadMetrics,
- prefixName: String): ExecutorService = {
+ prefixName: String,
+ startNumber: Int): 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
+ private val vtFactory = newVirtualThreadFactory(name, startNumber, pool) // use the pool as the scheduler
override def newThread(r: Runnable): Thread = {
val vt = vtFactory.newThread(r)
@@ -96,7 +105,7 @@ trait ExecutorServiceFactoryProvider {
vt
}
}
- case _ => newVirtualThreadFactory(prefixName, pool); // use the pool as the scheduler
+ case _ => newVirtualThreadFactory(prefixName, startNumber, pool); // use the pool as the scheduler
}
// wrap the pool with virtualized executor service
new VirtualizedExecutorService(
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 6b3532ba99..7b80e104cf 100644
--- a/actor/src/main/scala/org/apache/pekko/dispatch/VirtualThreadSupport.scala
+++ b/actor/src/main/scala/org/apache/pekko/dispatch/VirtualThreadSupport.scala
@@ -57,51 +57,53 @@ private[dispatch] object VirtualThreadSupport {
/**
* Create a virtual thread factory with the default Virtual Thread executor.
+ * @param prefix the prefix of the virtual thread name.
+ * @param start the starting number of the virtual thread name, if -1, the number will not be appended.
*/
- def newVirtualThreadFactory(prefix: String): ThreadFactory = {
- require(isSupported, "Virtual thread is not supported.")
- try {
- val builderClass = ClassLoader.getSystemClassLoader.loadClass("java.lang.Thread$Builder")
- val ofVirtualClass = ClassLoader.getSystemClassLoader.loadClass("java.lang.Thread$Builder$OfVirtual")
- val ofVirtualMethod = lookup.findStatic(classOf[Thread], "ofVirtual", MethodType.methodType(ofVirtualClass))
- var builder = ofVirtualMethod.invoke()
- val nameMethod = lookup.findVirtual(ofVirtualClass, "name",
- MethodType.methodType(ofVirtualClass, classOf[String], classOf[Long]))
- // TODO support replace scheduler when we drop Java 8 support
- val factoryMethod = lookup.findVirtual(builderClass, "factory", MethodType.methodType(classOf[ThreadFactory]))
- builder = nameMethod.invoke(builder, prefix + "-virtual-thread-", zero)
- factoryMethod.invoke(builder).asInstanceOf[ThreadFactory]
- } catch {
- case NonFatal(e) =>
- // --add-opens java.base/java.lang=ALL-UNNAMED
- throw new UnsupportedOperationException("Failed to create virtual thread factory", e)
- }
+ def newVirtualThreadFactory(prefix: String, start: Int): ThreadFactory = {
+ newVirtualThreadFactory(prefix, start, null)
}
/**
- * Create a virtual thread factory with the specified executor as the scheduler of virtual thread.
+ * Create a virtual thread factory with the default Virtual Thread executor.
+ * @param prefix the prefix of the virtual thread name.
+ * @param start the starting number of the virtual thread name, if -1, the number will not be appended.
+ * @param executor the executor to be used as the scheduler of virtual thread. If null, the default scheduler will be used.
*/
- def newVirtualThreadFactory(prefix: String, executor: ExecutorService): ThreadFactory =
+ def newVirtualThreadFactory(prefix: String, start: Int, executor: ExecutorService): ThreadFactory = {
+ require(isSupported, "Virtual thread is not supported.")
+ require(prefix != null && prefix.nonEmpty, "prefix should not be null or empty.")
try {
val builderClass = ClassLoader.getSystemClassLoader.loadClass("java.lang.Thread$Builder")
val ofVirtualClass = ClassLoader.getSystemClassLoader.loadClass("java.lang.Thread$Builder$OfVirtual")
- val ofVirtualMethod = classOf[Thread].getDeclaredMethod("ofVirtual")
- var builder = ofVirtualMethod.invoke(null)
+ val ofVirtualMethod = lookup.findStatic(classOf[Thread], "ofVirtual", MethodType.methodType(ofVirtualClass))
+ var builder = ofVirtualMethod.invoke()
+ // set the name
+ if (start <= -1) {
+ val nameMethod = lookup.findVirtual(ofVirtualClass, "name",
+ MethodType.methodType(ofVirtualClass, classOf[String]))
+ builder = nameMethod.invoke(builder, prefix + "-virtual-thread")
+ } else {
+ val nameMethod = lookup.findVirtual(ofVirtualClass, "name",
+ MethodType.methodType(ofVirtualClass, classOf[String], classOf[Long]))
+ builder = nameMethod.invoke(builder, prefix + "-virtual-thread-", zero)
+ }
+ // set the scheduler
if (executor ne null) {
+ // Use reflection here, method handle is stricter on access control
val clazz = builder.getClass
val field = clazz.getDeclaredField("scheduler")
field.setAccessible(true)
field.set(builder, executor)
}
- val nameMethod = ofVirtualClass.getDeclaredMethod("name", classOf[String], classOf[Long])
- val factoryMethod = builderClass.getDeclaredMethod("factory")
- builder = nameMethod.invoke(builder, prefix + "-virtual-thread-", zero)
+ val factoryMethod = lookup.findVirtual(builderClass, "factory", MethodType.methodType(classOf[ThreadFactory]))
factoryMethod.invoke(builder).asInstanceOf[ThreadFactory]
} catch {
case NonFatal(e) =>
// --add-opens java.base/java.lang=ALL-UNNAMED
throw new UnsupportedOperationException("Failed to create virtual thread factory", e)
}
+ }
object CarrierThreadFactory extends ForkJoinPool.ForkJoinWorkerThreadFactory {
private val clazz = ClassLoader.getSystemClassLoader.loadClass("jdk.internal.misc.CarrierThread")
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pekko.apache.org
For additional commands, e-mail: commits-help@pekko.apache.org