You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@pekko.apache.org by "He-Pin (via GitHub)" <gi...@apache.org> on 2026/04/27 12:36:08 UTC
[PR] fix: Fix virtualized dispatcher shutdown ordering [pekko]
He-Pin opened a new pull request, #2925:
URL: https://github.com/apache/pekko/pull/2925
Motivation:
JDK 21 and JDK 25 can reject virtual-thread continuations when a custom scheduler is shut down before already-started virtual threads have finished. The nightly failure shows the continuation being resubmitted to the custom ForkJoin scheduler after it has already been shut down.
Modification:
- Shut down the virtual-thread thread-per-task executor first.
- Cascade shutdown to the owned underlying scheduler only after the virtual-thread executor has terminated.
- Keep ExecutorService.isShutdown tied to the virtual-thread executor contract.
- Add a regression test that verifies a blocked virtual thread can resume and finish after dispatcher shutdown.
Result:
Verified locally:
- JDK 25: actor-tests / TestJdk9 / testOnly org.apache.pekko.dispatch.ForkJoinPoolVirtualThreadSpec
- JDK 21: actor-tests / TestJdk9 / testOnly org.apache.pekko.dispatch.ForkJoinPoolVirtualThreadSpec
- JDK 25 + Scala 2.13 nightly opts: stream-tests-tck / Test / testOnly org.apache.pekko.stream.tck.FanoutPublisherTest
- JDK 25 + Scala 3 nightly opts: cluster / Test / testOnly org.apache.pekko.cluster.ClusterDeathWatchNotificationSpec
- JDK 21 + Scala 2.13 and Scala 3 nightly opts: cluster-typed / Test / testOnly org.apache.pekko.cluster.typed.RemoteDeployNotAllowedSpec
- JDK 17: actor / Compile / compile, actor-tests / Test / compile, checkCodeStyle
References:
Nightly run 24984219012
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
Re: [PR] fix: Fix virtualized dispatcher shutdown ordering [pekko]
Posted by "He-Pin (via GitHub)" <gi...@apache.org>.
He-Pin merged PR #2925:
URL: https://github.com/apache/pekko/pull/2925
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
Re: [PR] fix: Fix virtualized dispatcher shutdown ordering [pekko]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #2925:
URL: https://github.com/apache/pekko/pull/2925#discussion_r3152276930
##########
actor/src/main/scala/org/apache/pekko/dispatch/VirtualizedExecutorService.scala:
##########
@@ -40,34 +41,29 @@ final class VirtualizedExecutorService(
require(loadMetricsProvider != null, "Load metrics provider must not be null")
private val executor = VirtualThreadSupport.newThreadPerTaskExecutor(vtFactory)
+ private val underlyingShutdownScheduled = new AtomicBoolean(false)
+ private val underlyingShutdownStarted = new AtomicBoolean(false)
override def atFullThrottle(): Boolean = loadMetricsProvider(this)
override def shutdown(): Unit = {
executor.shutdown()
- if (cascadeShutdown && (underlying ne null)) {
- underlying.shutdown()
- }
+ shutdownUnderlyingWhenVirtualThreadsHaveTerminated()
}
override def shutdownNow(): util.List[Runnable] = {
- val r = executor.shutdownNow()
- if (cascadeShutdown && (underlying ne null)) {
- underlying.shutdownNow()
- }
- r
+ val result = executor.shutdownNow()
+ shutdownUnderlyingWhenVirtualThreadsHaveTerminated()
+ result
Review Comment:
`shutdownNow()` no longer cascades to `underlying.shutdownNow()` (it eventually calls `underlying.shutdown()` via `shutdownUnderlying()`), which changes the expected “attempt to stop/interrupt immediately” semantics when `cascadeShutdown=true`. Consider tracking whether `shutdownNow` was requested and, once the virtual-thread executor has terminated, invoking `underlying.shutdownNow()` (vs `shutdown()`) to preserve the stronger shutdownNow contract for the owned scheduler.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
Re: [PR] fix: Fix virtualized dispatcher shutdown ordering [pekko]
Posted by "pjfanning (via GitHub)" <gi...@apache.org>.
pjfanning commented on code in PR #2925:
URL: https://github.com/apache/pekko/pull/2925#discussion_r3152293747
##########
actor/src/main/scala/org/apache/pekko/dispatch/VirtualizedExecutorService.scala:
##########
@@ -76,42 +72,78 @@ final class VirtualizedExecutorService(
override def awaitTermination(timeout: Long, unit: TimeUnit): Boolean = {
if (cascadeShutdown) {
- executor.awaitTermination(timeout, unit) && ((underlying eq null) || underlying.awaitTermination(timeout, unit))
+ val timeoutNanos = unit.toNanos(timeout)
+ val deadline = System.nanoTime() + timeoutNanos
+ executor.awaitTermination(timeout, unit) && {
+ shutdownUnderlying()
+ (underlying eq null) || {
+ val remainingNanos = deadline - System.nanoTime()
+ if (remainingNanos <= 0L) underlying.isTerminated
+ else underlying.awaitTermination(remainingNanos, TimeUnit.NANOSECONDS)
+ }
+ }
} else {
executor.awaitTermination(timeout, unit)
}
}
- override def submit[T](task: Callable[T]): Future[T] = {
- executor.submit(task)
+ private def shutdownUnderlyingWhenVirtualThreadsHaveTerminated(): Unit = {
+ if (cascadeShutdown && (underlying ne null) && underlyingShutdownScheduled.compareAndSet(false, true)) {
+ if (executor.isTerminated) {
+ shutdownUnderlying()
+ } else {
+ VirtualThreadSupport.startVirtualThread(
+ "pekko-virtualized-executor-shutdown",
+ new Runnable {
+ override def run(): Unit = {
+ var interrupted = false
+ try {
+ while (!executor.isTerminated) {
+ try executor.awaitTermination(1L, TimeUnit.DAYS)
+ catch {
+ case _: InterruptedException => interrupted = true
+ }
+ }
+ } finally {
+ shutdownUnderlying()
+ if (interrupted) {
+ Thread.currentThread().interrupt()
+ }
+ }
+ }
+ })
+ }
+ }
}
- override def submit[T](task: Runnable, result: T): Future[T] = {
- executor.submit(task, result)
+ private def shutdownUnderlying(): Unit = {
+ if (cascadeShutdown && (underlying ne null) && underlyingShutdownStarted.compareAndSet(false, true)) {
+ underlying.shutdown()
+ }
}
- override def submit(task: Runnable): Future[_] = {
+ override def submit[T](task: Callable[T]): Future[T] =
+ executor.submit(task)
+
+ override def submit[T](task: Runnable, result: T): Future[T] =
+ executor.submit(task, result)
+
+ override def submit(task: Runnable): Future[_] =
executor.submit(task)
- }
- override def invokeAll[T](tasks: util.Collection[_ <: Callable[T]]): util.List[Future[T]] = {
+ override def invokeAll[T](tasks: util.Collection[_ <: Callable[T]]): util.List[Future[T]] =
executor.invokeAll(tasks)
- }
override def invokeAll[T](
- tasks: util.Collection[_ <: Callable[T]], timeout: Long, unit: TimeUnit): util.List[Future[T]] = {
+ tasks: util.Collection[_ <: Callable[T]], timeout: Long, unit: TimeUnit): util.List[Future[T]] =
executor.invokeAll(tasks, timeout, unit)
- }
- override def invokeAny[T](tasks: util.Collection[_ <: Callable[T]]): T = {
+ override def invokeAny[T](tasks: util.Collection[_ <: Callable[T]]): T =
executor.invokeAny(tasks)
- }
- override def invokeAny[T](tasks: util.Collection[_ <: Callable[T]], timeout: Long, unit: TimeUnit): T = {
+ override def invokeAny[T](tasks: util.Collection[_ <: Callable[T]], timeout: Long, unit: TimeUnit): T =
executor.invokeAny(tasks, timeout, unit)
- }
- override def execute(command: Runnable): Unit = {
+ override def execute(command: Runnable): Unit =
Review Comment:
don't reformat code unrelated to the PR
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org