This is an automated email from the ASF dual-hosted git repository.
He-Pin pushed a commit to branch 1.8.x
in repository https://gitbox.apache.org/repos/asf/pekko.git
The following commit(s) were added to refs/heads/1.8.x by this push:
new 5b1b0e5b22 fix: backport mailbox task reuse fix to 1.8.x (#3621,
#3628) (#3627)
5b1b0e5b22 is described below
commit 5b1b0e5b223bd72b4d9f033a00d1a80be1babda0
Author: He-Pin(kerr) <[email protected]>
AuthorDate: Fri Oct 16 03:29:39 2026 +0800
fix: backport mailbox task reuse fix to 1.8.x (#3621, #3628) (#3627)
Motivation:
A reused mailbox must not also be a ForkJoinTask. Preserve the final main
branch fix without changing caller-owned tasks' completion or cancellation.
Modification:
Remove Mailbox's ForkJoinTask inheritance and related methods as on main.
Preserve both 1.x executor implementations and add the #3628 comments and
submission tests for Java 8 and Java 9+, including the test classpath
mapping.
Retain the existing internal MiMa filters and update dispatcher
documentation.
Result:
Mailbox submissions receive fresh wrappers; existing fork-join tasks retain
their completion and cancellation state. Java 8 support is preserved.
Tests:
- Cross-version submission suites: 24 passed on Scala 2.12, 2.13, and 3.
- JDK 8 submission suite: 4 passed after running the whole sbt process on 8.
An initial mixed JDK 17 launcher/JDK 8 fork could not read its argument
file.
- JDK 17 related suites: 192 passed, 2 existing pending tests.
- +mimaReportBinaryIssues and +headerCheckAll passed on all Scala versions.
- docs/paradox passed with java.lang opened on JDK 17; the initial run
without --add-opens failed in the existing parser's reflection code.
- Changed-file scalafmt and git diff --check passed. Full checkCodeStyle
reports only the unchanged NestedPayloadDepthSpec.scala formatting.
- Local validatePullRequest stopped and negative controls not started at
the contributor's request; remaining validation is delegated to remote CI.
- Auditable Qoder stdout review: No must-fix findings.
References:
Refs #3621, #3628, #3620, #3613.
---
.../dispatch/MailboxTaskSubmissionJdk9Spec.scala | 32 ++++++
.../pekko/dispatch/MailboxTaskSubmissionSpec.scala | 128 +++++++++++++++++++++
.../mailbox-not-a-forkjoin-task.excludes | 32 ++++++
.../pekko/dispatch/PekkoJdk9ForkJoinPool.scala | 4 +
.../dispatch/ForkJoinExecutorConfigurator.scala | 4 +
.../scala/org/apache/pekko/dispatch/Mailbox.scala | 29 ++---
build.sbt | 2 +-
docs/src/main/paradox/dispatchers.md | 6 +
8 files changed, 216 insertions(+), 21 deletions(-)
diff --git
a/actor-tests/src/test/scala-jdk9-only/org/apache/pekko/dispatch/MailboxTaskSubmissionJdk9Spec.scala
b/actor-tests/src/test/scala-jdk9-only/org/apache/pekko/dispatch/MailboxTaskSubmissionJdk9Spec.scala
new file mode 100644
index 0000000000..e31555c5ba
--- /dev/null
+++
b/actor-tests/src/test/scala-jdk9-only/org/apache/pekko/dispatch/MailboxTaskSubmissionJdk9Spec.scala
@@ -0,0 +1,32 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.pekko.dispatch
+
+import java.util.concurrent.ForkJoinPool
+
+/** Run the same submission contract against the executor selected on Java 9
and newer. */
+class MailboxTaskSubmissionJdk9Spec extends MailboxTaskSubmissionSpec {
+ override protected def newPool(): ForkJoinPool =
+ new PekkoJdk9ForkJoinPool(
+ 2,
+ new QuietThreadFactory,
+ ForkJoinPoolConstants.MaxCap,
+ quietHandler,
+ asyncMode = true,
+ minimumRunnable = 1)
+}
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
new file mode 100644
index 0000000000..ea04afd8a3
--- /dev/null
+++
b/actor-tests/src/test/scala/org/apache/pekko/dispatch/MailboxTaskSubmissionSpec.scala
@@ -0,0 +1,128 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.pekko.dispatch
+
+import java.util.concurrent.{ ForkJoinPool, ForkJoinTask, ForkJoinWorkerThread
}
+import java.util.concurrent.atomic.{ AtomicBoolean, AtomicInteger }
+
+import org.apache.pekko
+import pekko.dispatch.ForkJoinExecutorConfigurator.PekkoForkJoinPool
+import pekko.testkit.PekkoSpec
+
+/**
+ * A Mailbox is one object that its dispatcher resubmits for the actor's whole
lifetime, and
+ * `ForkJoinPool.execute` pushes a `ForkJoinTask` as-is instead of wrapping
it. Letting the mailbox
+ * itself reach the pool therefore puts the same reference back into the same
work-queue slot over
+ * and over, which a steal that stalls between its base recheck and its slot
CAS turns into a
+ * permanently damaged queue, and which one exceptional completion turns into
a mailbox that never
+ * runs again. See https://github.com/ben-manes/caffeine/issues/2021 and
+ * https://github.com/apache/pekko/issues/3620.
+ *
+ * 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 {
+
+ protected val quietHandler = new Thread.UncaughtExceptionHandler {
+ override def uncaughtException(t: Thread, e: Throwable): Unit = ()
+ }
+
+ protected final class QuietThreadFactory extends
ForkJoinPool.ForkJoinWorkerThreadFactory {
+ override def newThread(pool: ForkJoinPool): ForkJoinWorkerThread = {
+ val thread = new ForkJoinWorkerThread(pool) {}
+ thread.setName("MailboxTaskSubmissionSpec-worker")
+ thread.setUncaughtExceptionHandler(quietHandler)
+ thread
+ }
+ }
+
+ /** 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()
+ true
+ }
+ override def run(): Unit = {
+ runs.incrementAndGet()
+ }
+ }
+
+ protected def newPool(): ForkJoinPool =
+ new PekkoForkJoinPool(2, new QuietThreadFactory, quietHandler, asyncMode =
true)
+
+ "Mailbox" must {
+
+ "not be a ForkJoinTask" in {
+ classOf[ForkJoinTask[_]].isAssignableFrom(classOf[Mailbox]) shouldBe
false
+ }
+ }
+
+ "PekkoForkJoinPool" must {
+
+ "not run a cancelled ForkJoinTask submitted as a Runnable" in {
+ val pool = newPool()
+ val task = new RunnableForkJoinTask
+ try {
+ task.cancel(false) shouldBe true
+ // Select execute(Runnable), which must preserve the existing task's
cancellation state.
+ pool.execute(task: Runnable)
+ } 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)
+ // 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()
+ }
+
+ // 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()
+ 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/mima-filters/1.8.x.backwards.excludes/mailbox-not-a-forkjoin-task.excludes
b/actor/src/main/mima-filters/1.8.x.backwards.excludes/mailbox-not-a-forkjoin-task.excludes
new file mode 100644
index 0000000000..4f8d19577e
--- /dev/null
+++
b/actor/src/main/mima-filters/1.8.x.backwards.excludes/mailbox-not-a-forkjoin-task.excludes
@@ -0,0 +1,32 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+# Mailbox is no longer a ForkJoinTask. A mailbox is one object that its
dispatcher resubmits for the
+# whole lifetime of the actor, and ForkJoinPool.execute pushes a ForkJoinTask
as-is instead of
+# wrapping it, so the same reference kept returning to the same work-queue
slot. A steal that stalls
+# between its base recheck and its slot CAS can then move the queue's base
backwards and permanently
+# damage the queue (https://github.com/ben-manes/caffeine/issues/2021), and a
single exceptional
+# completion made ForkJoinTask.doExec short-circuit forever, silently
disabling the mailbox
+# (https://github.com/apache/pekko/issues/3620). Mailbox is private[pekko] and
escapes only through
+# MessageDispatcher.createMailbox.
+ProblemFilters.exclude[MissingTypesProblem]("org.apache.pekko.dispatch.Mailbox")
+ProblemFilters.exclude[DirectMissingMethodProblem]("org.apache.pekko.dispatch.Mailbox.getRawResult()Unit")
+ProblemFilters.exclude[DirectMissingMethodProblem]("org.apache.pekko.dispatch.Mailbox.setRawResult(scala.runtime.BoxedUnit)Unit")
+ProblemFilters.exclude[DirectMissingMethodProblem]("org.apache.pekko.dispatch.Mailbox.exec()Boolean")
+
+# The private SharingMailbox subclass is still present in 1.x and inherits the
same internal change.
+ProblemFilters.exclude[MissingTypesProblem]("org.apache.pekko.dispatch.BalancingDispatcher$SharingMailbox")
diff --git
a/actor/src/main/scala-jdk-9/org/apache/pekko/dispatch/PekkoJdk9ForkJoinPool.scala
b/actor/src/main/scala-jdk-9/org/apache/pekko/dispatch/PekkoJdk9ForkJoinPool.scala
index adb49072d0..e8e82595d3 100644
---
a/actor/src/main/scala-jdk-9/org/apache/pekko/dispatch/PekkoJdk9ForkJoinPool.scala
+++
b/actor/src/main/scala-jdk-9/org/apache/pekko/dispatch/PekkoJdk9ForkJoinPool.scala
@@ -43,6 +43,10 @@ private[dispatch] final class PekkoJdk9ForkJoinPool(
override def execute(r: Runnable): Unit =
if (r ne null)
+ // 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
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 903641bbb5..f82002db34 100644
---
a/actor/src/main/scala/org/apache/pekko/dispatch/ForkJoinExecutorConfigurator.scala
+++
b/actor/src/main/scala/org/apache/pekko/dispatch/ForkJoinExecutorConfigurator.scala
@@ -66,6 +66,10 @@ object ForkJoinExecutorConfigurator {
override def execute(r: Runnable): Unit =
if (r ne null)
+ // 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
diff --git a/actor/src/main/scala/org/apache/pekko/dispatch/Mailbox.scala
b/actor/src/main/scala/org/apache/pekko/dispatch/Mailbox.scala
index f8df5240a5..b0438bce10 100644
--- a/actor/src/main/scala/org/apache/pekko/dispatch/Mailbox.scala
+++ b/actor/src/main/scala/org/apache/pekko/dispatch/Mailbox.scala
@@ -60,11 +60,18 @@ private[pekko] object Mailbox {
* Mailbox and InternalMailbox is separated in two classes because ActorCell
is needed for implementation,
* but can't be exposed to user defined mailbox subclasses.
*
+ * A Mailbox must not be a `java.util.concurrent.ForkJoinTask`. It is one
object resubmitted for the
+ * whole lifetime of its actor, and `ForkJoinPool.execute` pushes a
`ForkJoinTask` as-is instead of
+ * wrapping it, so the same reference returns to the same work-queue slot
again and again. That lets a
+ * worker stalled mid-steal move the queue's `base` backwards and permanently
damage the queue, and it
+ * lets one exceptional completion disable the mailbox for good.
+ * See https://github.com/ben-manes/caffeine/issues/2021 and
+ * https://github.com/apache/pekko/issues/3620. Pinned by
MailboxTaskSubmissionSpec.
+ *
* INTERNAL API
*/
private[pekko] abstract class Mailbox(val messageQueue: MessageQueue)
- extends ForkJoinTask[Unit]
- with SystemMessageQueue
+ extends SystemMessageQueue
with Runnable {
import Mailbox._
@@ -239,24 +246,6 @@ private[pekko] abstract class Mailbox(val messageQueue:
MessageQueue)
}
}
- override final def getRawResult(): Unit = ()
- override final def setRawResult(unit: Unit): Unit = ()
- final override def exec(): Boolean =
- try {
- run(); false
- } catch {
- case _: InterruptedException =>
- Thread.currentThread().interrupt()
- false
- case anything: Throwable =>
- val t = Thread.currentThread()
- t.getUncaughtExceptionHandler match {
- case null =>
- case some => some.uncaughtException(t, anything)
- }
- throw anything
- }
-
/**
* Process the messages in the mailbox
*/
diff --git a/build.sbt b/build.sbt
index ae85048522..24430c061d 100644
--- a/build.sbt
+++ b/build.sbt
@@ -125,7 +125,7 @@ lazy val actor = pekkoModule("actor")
lazy val actorTests = pekkoModule("actor-tests")
.configs(Jdk9.TestJdk9)
- .dependsOn(testkit % "compile->compile;test->test", actor)
+ .dependsOn(testkit % "compile->compile;test->test", actor %
"compile->compile;TestJdk9->CompileJdk9")
.settings(Dependencies.actorTests)
.enablePlugins(NoPublish, Jdk9)
.disablePlugins(MimaPlugin)
diff --git a/docs/src/main/paradox/dispatchers.md
b/docs/src/main/paradox/dispatchers.md
index 3ee19c2f3e..cdf5a80272 100644
--- a/docs/src/main/paradox/dispatchers.md
+++ b/docs/src/main/paradox/dispatchers.md
@@ -44,6 +44,12 @@ You can read more about parallelism in the JDK's
[ForkJoinPool documentation](ht
When Running on Java 9+, you can use `maximum-pool-size` to set the upper
bound on the total number of threads allocated by the ForkJoinPool.
+Starting with Pekko 1.8.0, mailboxes no longer extend `ForkJoinTask`, so each
mailbox submission to a fork-join executor uses a fresh task wrapper.
+This prevents reusing a mailbox task from corrupting the work queue or leaving
the mailbox unable to run after an exception.
+The fix applies to both the Java 8 and Java 9+ executors and requires no
configuration changes.
+Existing `ForkJoinTask`s submitted as `Runnable`s are still executed directly,
preserving their own completion and cancellation state.
+Callers supplying their own fork-join tasks remain responsible for safe reuse
and reinitialization.
+
In Pekko 1.7.0, we introduced a setting for the `fork-join-executor` called
`minimum-runnable`. In Pekko 1.x, this defaults to `1` for backsward
compatibility reasons.
If you override the default and set it to `-1`, this will raise the number of
compensation threads the
pool will create when workers block, mitigating the asyncMode (FIFO)
compensation regression tracked in
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
