This is an automated email from the ASF dual-hosted git repository.
He-Pin pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/pekko.git
The following commit(s) were added to refs/heads/main by this push:
new d53beed463 Fix reentrant BatchingExecutor execution #1708 (#3331)
d53beed463 is described below
commit d53beed463d034adbae520d246bd7c0fc9fa2f3e
Author: He-Pin(kerr) <[email protected]>
AuthorDate: Tue Jul 14 00:58:07 2026 +0800
Fix reentrant BatchingExecutor execution #1708 (#3331)
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 ++++++++++++++++++++++
.../apache/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 36a164f9d8..4db07018c2 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 93d278a00a..ca0ac077ef 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 dc547e4ec0..33b591d7cc 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
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
