From 7edbcbaf5ab08c5403766ddb24687386058afadc Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E8=99=8E=E9=B8=A3?= Date: Sat, 11 Jul 2026 23:36:52 +0800 Subject: [PATCH] Fix reentrant BatchingExecutor execution #1708 Motivation: Fork-join workers can execute another submitted batch while helping a join. The nested batch hit a single-level ThreadLocal requirement and could drop queued work. Modification: Preserve and restore the previous batch context around Batch and BlockableBatch execution. Add deterministic coverage for both modes and document nested fork-join execution. Result: Reentrant batches run without failing, and outer queued tasks continue. Normal execution retains one ThreadLocal lookup with no added allocations or locking. Tests: - sbt +actor-tests / Test / testOnly org.apache.pekko.dispatch.ExecutionContextSpec - passed on Scala 2.13 and Scala 3 - sbt headerCreateAll +headerCheckAll checkCodeStyle docs / paradox - passed - sbt +mimaReportBinaryIssues - passed - scalafmt --list --mode diff-ref=origin/main - passed - git diff --check - passed - sbt validatePullRequest - stopped at user request before completion - Qoder review - No must-fix findings References: Fixes #1708 --- .../pekko/dispatch/ExecutionContextSpec.scala | 32 +++++++++++++++++++ .../pekko/dispatch/BatchingExecutor.scala | 14 +++++--- docs/src/main/paradox/typed/dispatchers.md | 3 ++ 3 files changed, 44 insertions(+), 5 deletions(-) diff --git a/actor-tests/src/test/scala/org/apache/pekko/dispatch/ExecutionContextSpec.scala b/actor-tests/src/test/scala/org/apache/pekko/dispatch/ExecutionContextSpec.scala index 36a164f9d89..4db07018c29 100644 --- a/actor-tests/src/test/scala/org/apache/pekko/dispatch/ExecutionContextSpec.scala +++ b/actor-tests/src/test/scala/org/apache/pekko/dispatch/ExecutionContextSpec.scala @@ -96,6 +96,38 @@ class ExecutionContextSpec extends PekkoSpec(ExecutionContextSpec.config) with D Await.result(p.future, timeout.duration) should ===(()) } + "continue the outer batch after reentrant batch execution" in { + Seq(true, false).foreach { resubmit => + val submitted = new java.util.ArrayDeque[Runnable]() + val executor = new BatchingExecutor { + override protected def unbatchedExecute(runnable: Runnable): Unit = submitted.add(runnable) + override protected def resubmitOnBlock: Boolean = resubmit + override def batchable(runnable: Runnable): Boolean = true + } + val completed = new AtomicInteger(0) + var innerBatch: Runnable = null + + executor.execute(new Runnable { + override def run(): Unit = { + executor.execute(new Runnable { + override def run(): Unit = completed.incrementAndGet() + }) + // Model a pool worker executing another submitted batch while helping a join. + innerBatch.run() + completed.incrementAndGet() + } + }) + val outerBatch = submitted.remove() + executor.execute(new Runnable { + override def run(): Unit = completed.incrementAndGet() + }) + innerBatch = submitted.remove() + + outerBatch.run() + completed.get should ===(3) + } + } + "be able to avoid starvation when Batching is used and Await/blocking is called" in { implicit val dispatcher: ExecutionContextExecutor = batchingDispatcher diff --git a/actor/src/main/scala/org/apache/pekko/dispatch/BatchingExecutor.scala b/actor/src/main/scala/org/apache/pekko/dispatch/BatchingExecutor.scala index 93d278a00af..ca0ac077ef7 100644 --- a/actor/src/main/scala/org/apache/pekko/dispatch/BatchingExecutor.scala +++ b/actor/src/main/scala/org/apache/pekko/dispatch/BatchingExecutor.scala @@ -64,7 +64,7 @@ private[pekko] trait Batchable extends Runnable { @InternalApi private[pekko] trait BatchingExecutor extends Executor { - // invariant: if "_tasksLocal.get ne null" then we are inside Batch.run; if it is null, we are outside + // _tasksLocal points to the batch accepting nested tasks, or is null when batching is inactive or paused for blocking private val _tasksLocal = new ThreadLocal[AbstractBatch]() private abstract class AbstractBatch extends java.util.ArrayDeque[Runnable](4) with Runnable { @@ -86,14 +86,17 @@ private[pekko] trait BatchingExecutor extends Executor { private final class Batch extends AbstractBatch { override final def run(): Unit = { - require(_tasksLocal.get eq null) + val previousBatch = _tasksLocal.get _tasksLocal.set(this) // Install ourselves as the current batch try processBatch(this) catch { case t: Throwable => resubmitUnbatched() throw t - } finally _tasksLocal.remove() + } finally { + if (previousBatch eq null) _tasksLocal.remove() + else _tasksLocal.set(previousBatch) + } } } @@ -102,7 +105,7 @@ private[pekko] trait BatchingExecutor extends Executor { private final class BlockableBatch extends AbstractBatch with BlockContext { // this method runs in the delegate ExecutionContext's thread override final def run(): Unit = { - require(_tasksLocal.get eq null) + val previousBatch = _tasksLocal.get _tasksLocal.set(this) // Install ourselves as the current batch val firstInvocation = _blockContext.get eq null if (firstInvocation) _blockContext.set(BlockContext.current) @@ -113,7 +116,8 @@ private[pekko] trait BatchingExecutor extends Executor { resubmitUnbatched() throw t } finally { - _tasksLocal.remove() + if (previousBatch eq null) _tasksLocal.remove() + else _tasksLocal.set(previousBatch) if (firstInvocation) _blockContext.remove() } } diff --git a/docs/src/main/paradox/typed/dispatchers.md b/docs/src/main/paradox/typed/dispatchers.md index dc547e4ec09..33b591d7ccc 100644 --- a/docs/src/main/paradox/typed/dispatchers.md +++ b/docs/src/main/paradox/typed/dispatchers.md @@ -24,6 +24,9 @@ page describes how to use dispatchers with `pekko-actor-typed`, which has depend A Pekko `MessageDispatcher` is what makes Pekko Actors "tick", it is the engine of the machine so to speak. All `MessageDispatcher` implementations are also an @scala[`ExecutionContext`]@java[`Executor`], which means that they can be used to execute arbitrary code, for instance @scala[`Future`s]@java[`CompletableFuture`s]. +On fork-join dispatchers, a worker waiting for fork/join subtasks can help execute other tasks submitted to the same dispatcher. +This nested execution is supported, but work that blocks on external resources should still use a dedicated dispatcher as described +in @ref:[Blocking Needs Careful Management](#blocking-needs-careful-management). ## Default dispatcher