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 df023266f7 fix: stop handing a reused ForkJoinTask to the fork-join
dispatcher (#3621)
df023266f7 is described below
commit df023266f723265859880873b4a46126b8499607
Author: He-Pin(kerr) <[email protected]>
AuthorDate: Tue Oct 13 14:16:31 2026 +0800
fix: stop handing a reused ForkJoinTask to the fork-join dispatcher (#3621)
Motivation:
Mailbox extended ForkJoinTask and was resubmitted as the same object for
the whole
lifetime of its actor, while PekkoForkJoinPool.execute passed any
ForkJoinTask through
unwrapped, so the pool pushed that one reference into the same work-queue
slot over and
over. The steal path CASes on slot content and then writes an absolute
base, so a worker
that stalls between its base recheck and its CAS can match a later copy of
the same
object and move base backwards by a multiple of the array length. The queue
then reports
more tasks than its array holds while every slot is null, and nothing
repairs it: JDK 22
and later reject every submission to it with "Queue capacity exceeded",
while earlier
JDKs silently lose or overwrite tasks. Caffeine hit the same race through
its reused
PerformCleanupTask in ben-manes/caffeine#2021.
The same reuse also made one exceptional completion permanent. Mailbox.exec
rethrew
whatever escaped run(), doExec recorded it on the mailbox object, and
doExec short-circuits
on a negative status, so that mailbox never ran again and its actor
silently stopped
processing messages, system messages included.
Modification:
Mailbox no longer extends ForkJoinTask, so no pool can push it as-is, and
PekkoForkJoinPool.execute wraps every submission in a fresh
PekkoForkJoinTask, which
absorbs an exceptional completion and is then discarded. Drop the now unused
getRawResult/setRawResult/exec from Mailbox and add MiMa filters for the
removed supertype.
Result:
A mailbox object never returns to a work-queue slot, so dispatcher
submissions cannot
trigger the steal race, and a Throwable escaping a mailbox run no longer
disables the
actor. Each mailbox run allocates one 24-byte wrapper that the pool already
allocated for
every TaskInvocation, and each Mailbox instance is 8 bytes smaller.
Tests:
- sbt "actor-tests/Test/testOnly
org.apache.pekko.dispatch.MailboxTaskSubmissionSpec" - 3 passed
- the new spec fails against the pre-fix source: 2 failed, 1 passed (the
plain-Runnable control passes either way)
- sbt "actor-tests/Test/testOnly *Dispatch* *Mailbox* *ForkJoin*
*ActorSystem*" - 188 passed, 27 suites
- sbt "actor-tests/test" - one flaky failure in TcpConnectionSpec ("send
incoming data to the connection handler", 3s expectMsgClass timeout); the spec
passes 35/35 when re-run alone
- sbt "++2.13.18!" mimaReportBinaryIssues - no issues
- sbt "++3.3.8!" actor/mimaReportBinaryIssues - no issues
- sbt checkCodeStyle - passed
- scalafmt --mode diff-ref=origin/main - no changes
References:
Refs #3613, Fixes #3620
---
.../pekko/dispatch/MailboxTaskSubmissionSpec.scala | 117 +++++++++++++++++++++
.../mailbox-not-a-forkjoin-task.excludes | 29 +++++
.../dispatch/ForkJoinExecutorConfigurator.scala | 7 +-
.../scala/org/apache/pekko/dispatch/Mailbox.scala | 29 ++---
4 files changed, 160 insertions(+), 22 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
new file mode 100644
index 0000000000..bb36d475fa
--- /dev/null
+++
b/actor-tests/src/test/scala/org/apache/pekko/dispatch/MailboxTaskSubmissionSpec.scala
@@ -0,0 +1,117 @@
+/*
+ * 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.
+ *
+ * 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.
+ */
+class MailboxTaskSubmissionSpec extends PekkoSpec {
+
+ private val quietHandler = new Thread.UncaughtExceptionHandler {
+ override def uncaughtException(t: Thread, e: Throwable): Unit = ()
+ }
+
+ private 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
+ }
+ }
+
+ /** 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 {
+ val runs = new AtomicInteger(0)
+ override def getRawResult(): Unit = ()
+ override def setRawResult(unit: Unit): Unit = ()
+ override def exec(): Boolean = {
+ run()
+ false
+ }
+ override def run(): Unit = {
+ runs.incrementAndGet()
+ body.run()
+ }
+ }
+
+ private def newPool(): PekkoForkJoinPool =
+ new PekkoForkJoinPool(2, new QuietThreadFactory,
ForkJoinPoolConstants.MaxCap, quietHandler, asyncMode = true)
+
+ "Mailbox" must {
+
+ "not be a ForkJoinTask" in {
+ classOf[ForkJoinTask[?]].isAssignableFrom(classOf[Mailbox]) shouldBe
false
+ }
+ }
+
+ "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 {
+ val pool = newPool()
+ 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")
+ })
+
+ pool.execute(task: Runnable)
+ awaitAssert(task.runs.get() shouldBe 1)
+
+ 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)
+ } finally pool.shutdown()
+ }
+
+ "still run a plain reused Runnable" in {
+ val pool = newPool()
+ try {
+ val runs = new AtomicInteger(0)
+ val task = new Runnable {
+ override def run(): Unit = runs.incrementAndGet()
+ }
+
+ pool.execute(task)
+ awaitAssert(runs.get() shouldBe 1)
+ pool.execute(task)
+ awaitAssert(runs.get() shouldBe 2)
+ } finally pool.shutdown()
+ }
+ }
+}
diff --git
a/actor/src/main/mima-filters/2.0.x.backwards.excludes/mailbox-not-a-forkjoin-task.excludes
b/actor/src/main/mima-filters/2.0.x.backwards.excludes/mailbox-not-a-forkjoin-task.excludes
new file mode 100644
index 0000000000..6c419e6828
--- /dev/null
+++
b/actor/src/main/mima-filters/2.0.x.backwards.excludes/mailbox-not-a-forkjoin-task.excludes
@@ -0,0 +1,29 @@
+# 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")
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 039e33aaa1..870a59cf0d 100644
---
a/actor/src/main/scala/org/apache/pekko/dispatch/ForkJoinExecutorConfigurator.scala
+++
b/actor/src/main/scala/org/apache/pekko/dispatch/ForkJoinExecutorConfigurator.scala
@@ -60,8 +60,11 @@ object ForkJoinExecutorConfigurator {
override def execute(r: Runnable): Unit =
if (r ne null)
- super.execute(
- (if (r.isInstanceOf[ForkJoinTask[?]]) r else new
PekkoForkJoinTask(r)).asInstanceOf[ForkJoinTask[Any]])
+ // 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))
else
throw new NullPointerException("Runnable was null")
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 fe1aee48af..f5c240700e 100644
--- a/actor/src/main/scala/org/apache/pekko/dispatch/Mailbox.scala
+++ b/actor/src/main/scala/org/apache/pekko/dispatch/Mailbox.scala
@@ -63,11 +63,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._
@@ -264,24 +271,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
*/
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
