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

Author: He-Pin

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
 

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()
         }
       }

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