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