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 77b5c1883a Revert unconditional fork-join task wrapping from #3621
(#3628)
77b5c1883a is described below
commit 77b5c1883a910d37d531b43c8537f9879e49a84c
Author: He-Pin(kerr) <[email protected]>
AuthorDate: Thu Oct 15 19:22:00 2026 +0800
Revert unconditional fork-join task wrapping from #3621 (#3628)
* Revert unconditional fork-join task wrapping from #3621
Motivation:
Making Mailbox a plain Runnable in #3621 already gives each mailbox
submission a fresh task. Its unrelated executor change can bypass an
existing ForkJoinTask's cancellation and leave a custom task incomplete,
causing join() to wait indefinitely.
Modification:
Restore the executor implementation exactly as it was before #3621.
Preserve the Mailbox fix, MiMa filters, and PekkoForkJoinTask's handling
of plain Runnable submissions.
Add cancellation and bounded completion regressions, move the termination
assertion out of finally, and clarify the plain-Runnable reuse control.
Result:
Mailboxes retain fresh task identities while caller-owned fork-join
tasks retain their completion and cancellation semantics. Interruption
handling for wrapped plain Runnables is unchanged.
Tests:
- MailboxTaskSubmissionSpec: 4 passed on Scala 2.13.18 and 3.3.8, JDK 17.
- The same four tests against #3621: cancellation and completion failed
as expected; both controls passed.
- Related dispatcher/mailbox/fork-join/ActorSystem suites: 189 passed,
2 existing pending tests on Scala 2.13.18, JDK 17.
- Native scalafmt, headerCreateAll, +headerCheckAll, checkCodeStyle,
and git diff --check passed.
- +mimaReportBinaryIssues passed on Scala 2.13.18 and 3.3.8.
- validatePullRequest did not complete on this ARM Mac: two TCP tests
timed out, and persistence's LevelDB JNI only contains x86_64/i386
native libraries. Stopped the remaining validation after that failure.
- Isolated TcpConnectionSpec and TcpListenerSpec rerun: 42 passed,
1 existing ignored test and 1 existing pending test.
- Auditable Qoder stdout review: no code-level must-fix findings.
The reviewed working-tree changes are included in this amended commit.
References:
Refs #3621.
* Update
actor/src/main/scala/org/apache/pekko/dispatch/ForkJoinExecutorConfigurator.scala
Co-authored-by: PJ Fanning <[email protected]>
---------
Co-authored-by: PJ Fanning <[email protected]>
---
.../pekko/dispatch/MailboxTaskSubmissionSpec.scala | 53 +++++++++++++---------
.../dispatch/ForkJoinExecutorConfigurator.scala | 11 +++--
2 files changed, 38 insertions(+), 26 deletions(-)
diff --git
a/actor-tests/src/test/scala/org/apache/pekko/dispatch/MailboxTaskSubmissionSpec.scala
b/actor-tests/src/test/scala/org/apache/pekko/dispatch/MailboxTaskSubmissionSpec.scala
index bb36d475fa..ad731104fc 100644
---
a/actor-tests/src/test/scala/org/apache/pekko/dispatch/MailboxTaskSubmissionSpec.scala
+++
b/actor-tests/src/test/scala/org/apache/pekko/dispatch/MailboxTaskSubmissionSpec.scala
@@ -33,8 +33,8 @@ import pekko.testkit.PekkoSpec
* runs again. See https://github.com/ben-manes/caffeine/issues/2021 and
* https://github.com/apache/pekko/issues/3620.
*
- * Neither race can be forced from a test, so this pins the two properties
that make them
- * unreachable: the mailbox is not a `ForkJoinTask`, and the pool never
receives one.
+ * This pins the mailbox's plain Runnable shape and resubmission after a
failing run. Existing
+ * ForkJoinTasks submitted as Runnables must retain their completion and
cancellation semantics.
*/
class MailboxTaskSubmissionSpec extends PekkoSpec {
@@ -51,18 +51,17 @@ class MailboxTaskSubmissionSpec extends PekkoSpec {
}
}
- /** The shape the pre-fix `Mailbox` had: resubmitted, `exec()` never
completes it, and rethrows. */
- private final class ReusedTask(body: Runnable) extends ForkJoinTask[Unit]
with Runnable {
+ /** Exposes a Runnable entry point so execute(Runnable) can receive an
existing ForkJoinTask. */
+ private final class RunnableForkJoinTask extends ForkJoinTask[Unit] with
Runnable {
val runs = new AtomicInteger(0)
override def getRawResult(): Unit = ()
override def setRawResult(unit: Unit): Unit = ()
override def exec(): Boolean = {
run()
- false
+ true
}
override def run(): Unit = {
runs.incrementAndGet()
- body.run()
}
}
@@ -78,37 +77,49 @@ class MailboxTaskSubmissionSpec extends PekkoSpec {
"PekkoForkJoinPool" must {
- // The mailbox is submitted through ExecutorService.execute(Runnable), so
cast to keep the
- // overload resolution identical to the dispatcher's call site.
- "run a reused ForkJoinTask again after one of its runs threw" in {
+ "not run a cancelled ForkJoinTask submitted as a Runnable" in {
val pool = newPool()
+ val task = new RunnableForkJoinTask
try {
- val firstRunFailed = new AtomicBoolean(false)
- val task = new ReusedTask(new Runnable {
- override def run(): Unit =
- if (firstRunFailed.compareAndSet(false, true)) throw new
RuntimeException("expected test failure")
- })
-
+ task.cancel(false) shouldBe true
+ // Select execute(Runnable), which must preserve the existing task's
cancellation state.
pool.execute(task: Runnable)
- awaitAssert(task.runs.get() shouldBe 1)
+ } finally pool.shutdown()
+ pool.awaitTermination(remainingOrDefault.toMillis,
java.util.concurrent.TimeUnit.MILLISECONDS) shouldBe true
+ task.isCancelled shouldBe true
+ task.runs.get() shouldBe 0
+ }
+ "complete a ForkJoinTask submitted as a Runnable" in {
+ val pool = newPool()
+ val task = new RunnableForkJoinTask
+ try {
pool.execute(task: Runnable)
- // Before the fix the task object itself was pushed, the throw
completed it exceptionally,
- // and ForkJoinTask.doExec short-circuits on a negative status, so
this stayed at 1.
- awaitAssert(task.runs.get() shouldBe 2)
+ // Bound the wait so a regression fails instead of hanging in join().
+ awaitAssert(task.isDone shouldBe true)
+ task.isCompletedNormally shouldBe true
+ task.join()
+ task.runs.get() shouldBe 1
} finally pool.shutdown()
}
- "still run a plain reused Runnable" in {
+ // Control: plain Runnables receive a fresh wrapper on every submission,
including before #3621.
+ "run a plain reused Runnable again after one of its runs threw" in {
val pool = newPool()
try {
val runs = new AtomicInteger(0)
+ val firstRunFailed = new AtomicBoolean(false)
val task = new Runnable {
- override def run(): Unit = runs.incrementAndGet()
+ override def run(): Unit = {
+ runs.incrementAndGet()
+ if (firstRunFailed.compareAndSet(false, true)) throw new
RuntimeException("expected test failure")
+ }
}
pool.execute(task)
awaitAssert(runs.get() shouldBe 1)
+ // Let the first wrapper finish failing before checking a subsequent
submission.
+ awaitAssert(pool.isQuiescent shouldBe true)
pool.execute(task)
awaitAssert(runs.get() shouldBe 2)
} finally pool.shutdown()
diff --git
a/actor/src/main/scala/org/apache/pekko/dispatch/ForkJoinExecutorConfigurator.scala
b/actor/src/main/scala/org/apache/pekko/dispatch/ForkJoinExecutorConfigurator.scala
index 870a59cf0d..5b056398fb 100644
---
a/actor/src/main/scala/org/apache/pekko/dispatch/ForkJoinExecutorConfigurator.scala
+++
b/actor/src/main/scala/org/apache/pekko/dispatch/ForkJoinExecutorConfigurator.scala
@@ -60,11 +60,12 @@ object ForkJoinExecutorConfigurator {
override def execute(r: Runnable): Unit =
if (r ne null)
- // Always hand the pool a task object it has not seen before. Pushing
a caller-owned
- // ForkJoinTask lets the same reference return to the same work-queue
slot, which a steal
- // that stalls between its base recheck and its slot CAS turns into a
permanently damaged
- // queue: https://github.com/ben-manes/caffeine/issues/2021
- super.execute(new PekkoForkJoinTask(r))
+ // An existing ForkJoinTask is pushed as-is (matching
ForkJoinPool.execute) so that its own
+ // completion and cancellation state is honoured. Callers must not
resubmit the same task
+ // instance (see https://github.com/ben-manes/caffeine/issues/2021);
this is why Mailbox
+ // is deliberately not a ForkJoinTask.
+ super.execute(
+ (if (r.isInstanceOf[ForkJoinTask[?]]) r else new
PekkoForkJoinTask(r)).asInstanceOf[ForkJoinTask[Any]])
else
throw new NullPointerException("Runnable was null")
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
