You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@pekko.apache.org by "He-Pin (via GitHub)" <gi...@apache.org> on 2026/10/13 07:19:19 UTC
[PR] fix: stop dispatch failure reporting from recursing into a StackOverflowError [pekko]
He-Pin opened a new pull request, #3622:
URL: https://github.com/apache/pekko/pull/3622
### Motivation
This is defect 2 of #3613, and unlike the rest of that issue it needs no exotic workload.
Reporting a rejected execution publishes an `Error` to the EventStream. `EventStream.publish` delivers to each subscriber with `subscriber ! event` (`EventStream.scala:53`), and the default logger is an actor, so that goes straight back through `Dispatch.sendMessage` → `Dispatcher.dispatch` → `registerForExecution`. When the executor is rejecting, the logger's own dispatch is rejected too, its report publishes again, and the cycle repeats on one thread:
```
Dispatcher.registerForExecution(Dispatcher.scala:161)
-> SubchannelClassification.publish(EventBus.scala:186)
-> EventStream.publish(EventStream.scala:53)
-> RepointableActorRef.$bang -> Dispatch.sendMessage(Dispatch.scala:177)
-> Dispatcher.dispatch -> Dispatcher.registerForExecution
```
#3613 measured 1024 frames and 59 repetitions of `registerForExecution` in one trace. With the default `pekko.jvm-exit-on-fatal-error = on`, the resulting `StackOverflowError` kills the JVM. It reproduces during a plain `ActorSystem.terminate()`, where the pool is legitimately gone and `execute` rejecting is the expected path — #3613 saw exactly that in a run with zero rejections and zero stalls during its whole 19,500-iteration workload.
Note that #3621 does not help here: that PR removes one *trigger* for rejections, while this one is reachable from any rejection, including the ordinary shutdown case.
### Modification
`DispatchFailureReporting` (in `AbstractDispatcher.scala`, next to the other dispatch-internal helpers) guards failure reporting per thread:
```scala
def report(eventStream: EventStream, event: LogEvent): Unit =
if (reporting.get()) StandardOutLogger ! event
else {
reporting.set(true)
try eventStream.publish(event)
finally reporting.remove()
}
```
The outermost failure publishes to the EventStream exactly as before. A failure reached while that publish is still in flight goes to `Logging.StandardOutLogger`, which is a `MinimalActorRef` whose `!` calls `print` directly (`Logging.scala:1130-1132`) — no mailbox, no dispatcher, so it cannot re-enter the cycle. Pekko already uses it this way for initialization errors (`Logging.scala:95`, `Logging.scala:128`).
All seven dispatch failure report sites go through it, since any of them can be the one that closes the cycle:
| Site | Message |
|---|---|
| `Dispatcher.scala:112` | `executeTask was rejected twice!` |
| `Dispatcher.scala:161` | `registerForExecution was rejected twice!` |
| `Dispatch.scala:138` | `interrupted during message send` |
| `Dispatch.scala:145` | `swallowing exception during message send` |
| `AbstractDispatcher.scala:61` | `TaskInvocation.run` non-fatal |
| `AbstractDispatcher.scala:216` | `reportFailure`, `LogEventException` |
| `AbstractDispatcher.scala:218` | `reportFailure`, other |
Guarding `Dispatch.handleException` is what bounds the cycle for every *other* `eventStream.publish` call site too: a nested dispatch failure reached from `FaultHandling`, `Mailbox`'s dead-letter cleanup, or a config warning now ends in a guarded report instead of another publish.
### Result
A rejecting executor produces a bounded number of reports instead of unbounded recursion, and no `StackOverflowError`.
The EventStream still receives the outermost failure, so subscribers see what they see today. The nested failures that were previously consumed by the stack overflow are printed to stdout instead of being lost — which also addresses #3613's observation that the rejection starting the cascade was invisible, since the JVM died before a single `registerForExecution was rejected twice!` line reached the log.
Two things worth flagging for review:
- A single rejected `registerForExecution` publishes **two** Error events (one at `Dispatcher.scala:161`, one from `handleException` after the rethrow at `Dispatcher.scala:162`). That double report is pre-existing and this PR does not change it; de-duplicating it is a separate behavioural question.
- The stdout fallback is not rate limited, so a shutdown with a large backlog produces roughly two printed stack traces per rejected send. That is comparable to today's one ERROR log line per occurrence, but if it is judged too noisy, rate limiting the fallback would be a small follow-up.
Not addressed here, and left open on #3613: a double rejection still leaves a non-empty mailbox un-scheduled (`mbox.setAsIdle()` at `Dispatcher.scala:160`), and `handleException` still returns normally so `tell` reports success for a message that will not be delivered.
### Tests
`DispatchFailureReportingSpec` drives a real `ActorSystem` whose `default-dispatcher` uses a switchable executor that can be flipped into rejecting every submission. Because `pekko.loggers-dispatcher` defaults to `pekko.actor.default-dispatcher` (`reference.conf:31`), the logger actor sits on the same rejecting executor, which is what closes the cycle.
The spec needs one non-obvious setup step, worth calling out: the actor's mailbox must be drained before flipping the executor. `system.actorOf` returns while the mailbox is still `Scheduled`, and `registerForExecution` then short-circuits on `setAsScheduled()` without ever calling `execute` — no rejection, no report, and the test passes vacuously. The first version of this spec did exactly that and was green against the unfixed source; it only became a real test after a warm-up round trip.
- `sbt "actor-tests/Test/testOnly org.apache.pekko.dispatch.DispatchFailureReportingSpec"` — 3 passed
- **The spec fails against the pre-fix source**: `must not recurse into a StackOverflowError` fails with `java.lang.StackOverflowError` through `Dispatcher.registerForExecution(Dispatcher.scala:161)` → `EventStream.publish(EventStream.scala:34)`, the cycle above
- The spec also asserts the rejection was really published (`published.get() should be > 0`), so it cannot pass without exercising the path
- A third test pins the guard directly: a re-entrant `MinimalActorRef` subscriber that calls `report` again from `!` must not come back through the bus, so the observed re-entry depth stays at 1
- `sbt "actor-tests/Test/testOnly *Dispatch* *Mailbox* *ForkJoin* *ActorSystem* *Logging* *EventStream*"` — 221 passed, 31 suites
- `sbt "actor-tests/test"` — 1997 of 1998 passed. The one failure is `TcpConnectionSpec` "send incoming data to the connection handler" (3s `expectMsgClass` timeout), which passes 35/35 when re-run alone and failed identically during #3621 before this change existed
- `sbt mimaReportBinaryIssues` — no issues on 2.13.18; `sbt "++3.3.8!" actor/mimaReportBinaryIssues` — no issues. No new filters needed: `DispatchFailureReporting` is an addition
- `sbt checkCodeStyle` — passed
- `scalafmt --mode diff-ref=origin/main` — applied
### References
Refs #3613
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
Re: [PR] fix: stop dispatch failure reporting from recursing into a StackOverflowError [pekko]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #3622:
URL: https://github.com/apache/pekko/pull/3622#discussion_r4293312013
##########
actor-tests/src/test/scala/org/apache/pekko/dispatch/DispatchFailureReportingSpec.scala:
##########
@@ -0,0 +1,270 @@
+/*
+ * 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.{
+ CountDownLatch,
+ ExecutorService,
+ LinkedBlockingQueue,
+ RejectedExecutionException,
+ ThreadFactory,
+ ThreadPoolExecutor,
+ TimeUnit
+}
+import java.util.concurrent.atomic.{ AtomicBoolean, AtomicInteger }
+
+import scala.annotation.nowarn
+
+import org.apache.pekko
+import pekko.actor._
+import pekko.event.{ EventStream, Logging }
+import pekko.event.Logging.Error
+import pekko.testkit.{ PekkoSpec, TestProbe }
+
+import com.typesafe.config.{ Config, ConfigFactory }
+
+object RejectingExecutorConfigurator {
+
+ /** Flipped by the spec to make every submission fail, the way a shut down or saturated executor does. */
+ val rejecting = new AtomicBoolean(false)
+ val pendingTasks = new AtomicInteger(0)
+}
+
+/**
+ * An executor that can be switched into rejecting every task, so a spec can reproduce the
+ * `RejectedExecutionException` that a dispatcher hits during shutdown without waiting for one.
+ */
+class RejectingExecutorConfigurator(
+ @nowarn("msg=never used") config: Config,
+ @nowarn("msg=never used") prerequisites: DispatcherPrerequisites)
+ extends ExecutorServiceConfigurator(config, prerequisites) {
+
+ override def createExecutorServiceFactory(id: String, threadFactory: ThreadFactory): ExecutorServiceFactory =
+ new ExecutorServiceFactory {
+ override def createExecutorService: ExecutorService = {
+ val underlying = new ThreadPoolExecutor(
+ 4,
+ 4,
+ 60L,
+ TimeUnit.SECONDS,
+ new LinkedBlockingQueue[Runnable](),
+ threadFactory)
+ new ExecutorServiceDelegate {
+ override def executor: ExecutorService = underlying
+ override def execute(command: Runnable): Unit =
+ if (RejectingExecutorConfigurator.rejecting.get())
+ throw new RejectedExecutionException("test executor is rejecting")
+ else {
+ RejectingExecutorConfigurator.pendingTasks.incrementAndGet()
+ try executor.execute(() => {
+ try command.run()
+ finally RejectingExecutorConfigurator.pendingTasks.decrementAndGet()
+ })
+ catch {
+ case e: RejectedExecutionException =>
+ RejectingExecutorConfigurator.pendingTasks.decrementAndGet()
+ throw e
+ }
+ }
+ }
+ }
+ }
+}
+
+/**
+ * Reporting a dispatch failure publishes to the EventStream, which delivers to a logging actor with
+ * `!` and so goes back through dispatch. When the executor rejects, that re-enters the failure path
+ * and recurses until the stack overflows, which `pekko.jvm-exit-on-fatal-error` turns into a JVM
+ * kill. See https://github.com/apache/pekko/issues/3613.
+ */
+class DispatchFailureReportingSpec
+ extends PekkoSpec(
+ ConfigFactory
+ .parseString(s"""
+ pekko {
+ jvm-exit-on-fatal-error = off
+ loggers = ["org.apache.pekko.event.Logging$$DefaultLogger"]
+ loglevel = "ERROR"
+ actor.default-dispatcher.executor = "org.apache.pekko.dispatch.RejectingExecutorConfigurator"
+ }
+ """)
+ .withFallback(PekkoSpec.testConf)) {
+
+ private def echoActor(probe: TestProbe): ActorRef = {
+ val ref = system.actorOf(Props(new Actor {
+ override def receive: Receive = { case msg => probe.ref ! msg }
+ }))
+ // A reply can arrive while the mailbox is still Scheduled. Even after it becomes idle,
+ // Mailbox.run can reschedule it, so wait until the executor has finished the whole task.
+ ref ! "warmup"
+ probe.expectMsg("warmup")
+ awaitAssert(RejectingExecutorConfigurator.pendingTasks.get() shouldBe 0)
+ ref
+ }
+
+ private def errorSubscriber(onError: => Unit): MinimalActorRef = new MinimalActorRef {
+ override def path: ActorPath = RootActorPath(Address("pekko", "all-systems"), "/errorSubscriber")
+ override def provider: ActorRefProvider = throw new UnsupportedOperationException("test ref")
+ override def !(message: Any)(implicit sender: ActorRef): Unit = onError
+ }
+
+ "A message sent while the executor is rejecting" must {
+
+ "not recurse into a StackOverflowError" in {
+ val probe = TestProbe()
+ val ref = echoActor(probe)
+ val published = new AtomicInteger(0)
+ val logger = echoActor(TestProbe())
+ val counter = errorSubscriber {
+ published.incrementAndGet()
+ logger ! "log error"
+ }
+ val output = new java.io.ByteArrayOutputStream()
+ // Forward through a synchronous ref so EventStreamUnsubscriber watches that ref rather
+ // than asynchronously sending Watch to the idle actor whose dispatch must be rejected.
+ system.eventStream.subscribe(counter, classOf[Logging.Error])
+ awaitAssert(RejectingExecutorConfigurator.pendingTasks.get() shouldBe 0)
+
+ RejectingExecutorConfigurator.rejecting.set(true)
+ try {
+ // Before the guard, the rejection report published to the logger actor, whose own dispatch
+ // was rejected by the same executor, whose report published again, all on this thread.
+ Console.withOut(output) { ref ! "hello" }
Review Comment:
`Console.withOut` changes a global process-wide setting and is not thread-safe. If tests in this project run concurrently (or other tests print to stdout), this can lead to flaky assertions and cross-test interference. Consider serializing access (e.g., via a shared lock around the `withOut` blocks) or using a Pekko/testkit-provided mechanism to intercept logging output (if available) that doesn’t mutate global stdout.
##########
docs/src/main/paradox/logging.md:
##########
@@ -285,7 +285,8 @@ pekko {
```
The `stdout-loglevel` is only in effect during system startup and shutdown, and setting
-it to `OFF` as well, ensures that nothing gets logged during system startup or shutdown.
+it to `OFF` as well disables normal logging during system startup and shutdown. Recursive
+dispatcher failures can still use the emergency stdout fallback described below.
Review Comment:
Same grammar/readability issue here as in the typed logging guide (“setting it to `OFF` as well disables…”). Rephrase to a cleaner construction (or split into two sentences) to keep the docs consistent and easy to parse.
##########
docs/src/main/paradox/typed/logging.md:
##########
@@ -293,7 +293,8 @@ pekko {
```
The `stdout-loglevel` is only in effect during system startup and shutdown, and setting
-it to `OFF` as well, ensures that nothing gets logged during system startup or shutdown.
+it to `OFF` as well disables normal logging during system startup and shutdown. Recursive
+dispatcher failures can still use the emergency stdout fallback described below.
Review Comment:
The updated sentence is grammatically awkward (“setting it to `OFF` as well disables…”). Consider rephrasing to avoid the dangling “as well”, e.g. “...and setting it to `OFF` disables normal logging...”, or split into two sentences for readability.
##########
actor-tests/src/test/scala/org/apache/pekko/dispatch/DispatchFailureReportingSpec.scala:
##########
@@ -0,0 +1,270 @@
+/*
+ * 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.{
+ CountDownLatch,
+ ExecutorService,
+ LinkedBlockingQueue,
+ RejectedExecutionException,
+ ThreadFactory,
+ ThreadPoolExecutor,
+ TimeUnit
+}
+import java.util.concurrent.atomic.{ AtomicBoolean, AtomicInteger }
+
+import scala.annotation.nowarn
+
+import org.apache.pekko
+import pekko.actor._
+import pekko.event.{ EventStream, Logging }
+import pekko.event.Logging.Error
+import pekko.testkit.{ PekkoSpec, TestProbe }
+
+import com.typesafe.config.{ Config, ConfigFactory }
+
+object RejectingExecutorConfigurator {
+
+ /** Flipped by the spec to make every submission fail, the way a shut down or saturated executor does. */
+ val rejecting = new AtomicBoolean(false)
+ val pendingTasks = new AtomicInteger(0)
+}
+
+/**
+ * An executor that can be switched into rejecting every task, so a spec can reproduce the
+ * `RejectedExecutionException` that a dispatcher hits during shutdown without waiting for one.
+ */
+class RejectingExecutorConfigurator(
+ @nowarn("msg=never used") config: Config,
+ @nowarn("msg=never used") prerequisites: DispatcherPrerequisites)
+ extends ExecutorServiceConfigurator(config, prerequisites) {
+
+ override def createExecutorServiceFactory(id: String, threadFactory: ThreadFactory): ExecutorServiceFactory =
+ new ExecutorServiceFactory {
+ override def createExecutorService: ExecutorService = {
+ val underlying = new ThreadPoolExecutor(
+ 4,
+ 4,
+ 60L,
+ TimeUnit.SECONDS,
+ new LinkedBlockingQueue[Runnable](),
+ threadFactory)
+ new ExecutorServiceDelegate {
+ override def executor: ExecutorService = underlying
+ override def execute(command: Runnable): Unit =
+ if (RejectingExecutorConfigurator.rejecting.get())
+ throw new RejectedExecutionException("test executor is rejecting")
+ else {
+ RejectingExecutorConfigurator.pendingTasks.incrementAndGet()
+ try executor.execute(() => {
+ try command.run()
+ finally RejectingExecutorConfigurator.pendingTasks.decrementAndGet()
+ })
+ catch {
+ case e: RejectedExecutionException =>
+ RejectingExecutorConfigurator.pendingTasks.decrementAndGet()
+ throw e
+ }
+ }
+ }
+ }
+ }
+}
+
+/**
+ * Reporting a dispatch failure publishes to the EventStream, which delivers to a logging actor with
+ * `!` and so goes back through dispatch. When the executor rejects, that re-enters the failure path
+ * and recurses until the stack overflows, which `pekko.jvm-exit-on-fatal-error` turns into a JVM
+ * kill. See https://github.com/apache/pekko/issues/3613.
+ */
+class DispatchFailureReportingSpec
+ extends PekkoSpec(
+ ConfigFactory
+ .parseString(s"""
+ pekko {
+ jvm-exit-on-fatal-error = off
+ loggers = ["org.apache.pekko.event.Logging$$DefaultLogger"]
+ loglevel = "ERROR"
+ actor.default-dispatcher.executor = "org.apache.pekko.dispatch.RejectingExecutorConfigurator"
+ }
+ """)
+ .withFallback(PekkoSpec.testConf)) {
+
+ private def echoActor(probe: TestProbe): ActorRef = {
+ val ref = system.actorOf(Props(new Actor {
+ override def receive: Receive = { case msg => probe.ref ! msg }
+ }))
+ // A reply can arrive while the mailbox is still Scheduled. Even after it becomes idle,
+ // Mailbox.run can reschedule it, so wait until the executor has finished the whole task.
+ ref ! "warmup"
+ probe.expectMsg("warmup")
+ awaitAssert(RejectingExecutorConfigurator.pendingTasks.get() shouldBe 0)
+ ref
+ }
+
+ private def errorSubscriber(onError: => Unit): MinimalActorRef = new MinimalActorRef {
+ override def path: ActorPath = RootActorPath(Address("pekko", "all-systems"), "/errorSubscriber")
+ override def provider: ActorRefProvider = throw new UnsupportedOperationException("test ref")
+ override def !(message: Any)(implicit sender: ActorRef): Unit = onError
+ }
+
+ "A message sent while the executor is rejecting" must {
+
+ "not recurse into a StackOverflowError" in {
+ val probe = TestProbe()
+ val ref = echoActor(probe)
+ val published = new AtomicInteger(0)
+ val logger = echoActor(TestProbe())
+ val counter = errorSubscriber {
+ published.incrementAndGet()
+ logger ! "log error"
+ }
+ val output = new java.io.ByteArrayOutputStream()
+ // Forward through a synchronous ref so EventStreamUnsubscriber watches that ref rather
+ // than asynchronously sending Watch to the idle actor whose dispatch must be rejected.
+ system.eventStream.subscribe(counter, classOf[Logging.Error])
+ awaitAssert(RejectingExecutorConfigurator.pendingTasks.get() shouldBe 0)
+
+ RejectingExecutorConfigurator.rejecting.set(true)
+ try {
+ // Before the guard, the rejection report published to the logger actor, whose own dispatch
+ // was rejected by the same executor, whose report published again, all on this thread.
+ Console.withOut(output) { ref ! "hello" }
+ } catch {
+ case e: StackOverflowError =>
+ fail("reporting the rejected execution recursed until the stack overflowed", e)
+ } finally {
+ RejectingExecutorConfigurator.rejecting.set(false)
+ system.eventStream.unsubscribe(counter)
+ }
+
+ // The send above really was rejected and reported, so the assertion is not vacuous.
+ published.get() should be > 0
+ output.toString("UTF-8") should include("registerForExecution was rejected twice!")
+ }
+
+ "still dispatch normally once the executor accepts again" in {
+ val rejectedProbe = TestProbe()
+ val rejected = echoActor(rejectedProbe)
+
+ RejectingExecutorConfigurator.rejecting.set(true)
+ try rejected ! "hello"
+ finally RejectingExecutorConfigurator.rejecting.set(false)
+
+ // A fresh actor, so this does not depend on when the rejected message above is drained.
+ val probe = TestProbe()
+ val fresh = system.actorOf(Props(new Actor {
+ override def receive: Receive = { case msg => probe.ref ! msg }
+ }))
+
+ fresh ! "delivered"
+ probe.expectMsg("delivered")
+ }
+ }
+
+ "DispatchFailureReporting" must {
+
+ "send a nested failure to stdout even when it crosses event streams" in {
+ val eventStream = new EventStream(system, false)
+ val nestedEventStream = new EventStream(system, false)
+ val deliveries = new AtomicInteger(0)
+ val nestedDeliveries = new AtomicInteger(0)
+ val output = new java.io.ByteArrayOutputStream()
+ val reentrant = errorSubscriber {
+ deliveries.incrementAndGet()
+ DispatchFailureReporting.report(
+ nestedEventStream,
+ Error(Error.NoCause, "test", getClass, "nested failure"))
+ }
+ eventStream.subscribe(reentrant, classOf[Error])
+ nestedEventStream.subscribe(errorSubscriber { nestedDeliveries.incrementAndGet(); () }, classOf[Error])
+
+ Console.withOut(output) {
+ DispatchFailureReporting.report(eventStream, Error(Error.NoCause, "test", getClass, "outer failure"))
+ deliveries.get() shouldBe 1
+ nestedDeliveries.get() shouldBe 0
+
+ // A second outer report on this thread must reach the bus after the guard is released.
+ DispatchFailureReporting.report(eventStream, Error(Error.NoCause, "test", getClass, "second outer failure"))
+ deliveries.get() shouldBe 2
+ nestedDeliveries.get() shouldBe 0
+ }
Review Comment:
`Console.withOut` changes a global process-wide setting and is not thread-safe. If tests in this project run concurrently (or other tests print to stdout), this can lead to flaky assertions and cross-test interference. Consider serializing access (e.g., via a shared lock around the `withOut` blocks) or using a Pekko/testkit-provided mechanism to intercept logging output (if available) that doesn’t mutate global stdout.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
Re: [PR] fix: stop dispatch failure reporting from recursing into a StackOverflowError [pekko]
Posted by "He-Pin (via GitHub)" <gi...@apache.org>.
He-Pin merged PR #3622:
URL: https://github.com/apache/pekko/pull/3622
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
Re: [PR] fix: stop dispatch failure reporting from recursing into a StackOverflowError [pekko]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #3622:
URL: https://github.com/apache/pekko/pull/3622#discussion_r4256388924
##########
actor-tests/src/test/scala/org/apache/pekko/dispatch/DispatchFailureReportingSpec.scala:
##########
@@ -0,0 +1,238 @@
+/*
+ * 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.{
+ CountDownLatch,
+ ExecutorService,
+ LinkedBlockingQueue,
+ RejectedExecutionException,
+ ThreadFactory,
+ ThreadPoolExecutor,
+ TimeUnit
+}
+import java.util.concurrent.atomic.{ AtomicBoolean, AtomicInteger }
+
+import scala.annotation.nowarn
+
+import org.apache.pekko
+import pekko.actor._
+import pekko.event.Logging
+import pekko.event.Logging.Error
+import pekko.testkit.{ PekkoSpec, TestProbe }
+
+import com.typesafe.config.{ Config, ConfigFactory }
+
+object RejectingExecutorConfigurator {
+
+ /** Flipped by the spec to make every submission fail, the way a shut down or saturated executor does. */
+ val rejecting = new AtomicBoolean(false)
+}
+
+/**
+ * An executor that can be switched into rejecting every task, so a spec can reproduce the
+ * `RejectedExecutionException` that a dispatcher hits during shutdown without waiting for one.
+ */
+class RejectingExecutorConfigurator(
+ @nowarn("msg=never used") config: Config,
+ @nowarn("msg=never used") prerequisites: DispatcherPrerequisites)
+ extends ExecutorServiceConfigurator(config, prerequisites) {
+
+ override def createExecutorServiceFactory(id: String, threadFactory: ThreadFactory): ExecutorServiceFactory =
+ new ExecutorServiceFactory {
+ override def createExecutorService: ExecutorService = {
+ val underlying = new ThreadPoolExecutor(
+ 4,
+ 4,
+ 60L,
+ TimeUnit.SECONDS,
+ new LinkedBlockingQueue[Runnable](),
+ threadFactory)
+ new ExecutorServiceDelegate {
+ override def executor: ExecutorService = underlying
+ override def execute(command: Runnable): Unit =
+ if (RejectingExecutorConfigurator.rejecting.get())
+ throw new RejectedExecutionException("test executor is rejecting")
+ else
+ executor.execute(command)
+ }
+ }
+ }
+}
+
+/**
+ * Reporting a dispatch failure publishes to the EventStream, which delivers to a logging actor with
+ * `!` and so goes back through dispatch. When the executor rejects, that re-enters the failure path
+ * and recursed until the stack overflowed, which `pekko.jvm-exit-on-fatal-error` turned into a JVM
Review Comment:
Grammar/tense mismatch in the Scaladoc: “re-enters … and recursed” should use consistent tense (e.g., “and recurses …”).
##########
actor-tests/src/test/scala/org/apache/pekko/dispatch/DispatchFailureReportingSpec.scala:
##########
@@ -0,0 +1,238 @@
+/*
+ * 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.{
+ CountDownLatch,
+ ExecutorService,
+ LinkedBlockingQueue,
+ RejectedExecutionException,
+ ThreadFactory,
+ ThreadPoolExecutor,
+ TimeUnit
+}
+import java.util.concurrent.atomic.{ AtomicBoolean, AtomicInteger }
+
+import scala.annotation.nowarn
+
+import org.apache.pekko
+import pekko.actor._
+import pekko.event.Logging
+import pekko.event.Logging.Error
+import pekko.testkit.{ PekkoSpec, TestProbe }
+
+import com.typesafe.config.{ Config, ConfigFactory }
+
+object RejectingExecutorConfigurator {
+
+ /** Flipped by the spec to make every submission fail, the way a shut down or saturated executor does. */
+ val rejecting = new AtomicBoolean(false)
+}
+
+/**
+ * An executor that can be switched into rejecting every task, so a spec can reproduce the
+ * `RejectedExecutionException` that a dispatcher hits during shutdown without waiting for one.
+ */
+class RejectingExecutorConfigurator(
+ @nowarn("msg=never used") config: Config,
+ @nowarn("msg=never used") prerequisites: DispatcherPrerequisites)
+ extends ExecutorServiceConfigurator(config, prerequisites) {
+
+ override def createExecutorServiceFactory(id: String, threadFactory: ThreadFactory): ExecutorServiceFactory =
+ new ExecutorServiceFactory {
+ override def createExecutorService: ExecutorService = {
+ val underlying = new ThreadPoolExecutor(
+ 4,
+ 4,
+ 60L,
+ TimeUnit.SECONDS,
+ new LinkedBlockingQueue[Runnable](),
+ threadFactory)
+ new ExecutorServiceDelegate {
+ override def executor: ExecutorService = underlying
+ override def execute(command: Runnable): Unit =
+ if (RejectingExecutorConfigurator.rejecting.get())
+ throw new RejectedExecutionException("test executor is rejecting")
+ else
+ executor.execute(command)
+ }
+ }
+ }
+}
+
+/**
+ * Reporting a dispatch failure publishes to the EventStream, which delivers to a logging actor with
+ * `!` and so goes back through dispatch. When the executor rejects, that re-enters the failure path
+ * and recursed until the stack overflowed, which `pekko.jvm-exit-on-fatal-error` turned into a JVM
+ * kill. See https://github.com/apache/pekko/issues/3613.
+ */
+class DispatchFailureReportingSpec
+ extends PekkoSpec(
+ ConfigFactory
+ .parseString(s"""
+ pekko {
+ jvm-exit-on-fatal-error = off
+ loggers = ["org.apache.pekko.event.Logging$$DefaultLogger"]
+ loglevel = "ERROR"
+ actor.default-dispatcher.executor = "org.apache.pekko.dispatch.RejectingExecutorConfigurator"
+ }
+ """)
+ .withFallback(PekkoSpec.testConf)) {
+
+ private def echoActor(probe: TestProbe): ActorRef = {
+ val ref = system.actorOf(Props(new Actor {
+ override def receive: Receive = { case msg => probe.ref ! msg }
+ }))
+ // Drain the Create message first. A mailbox that is still Scheduled never reaches the executor,
+ // so without this the rejection path is not exercised at all and the test passes vacuously.
+ ref ! "warmup"
+ probe.expectMsg("warmup")
+ ref
+ }
+
+ private def errorCounter(published: AtomicInteger): MinimalActorRef = new MinimalActorRef {
+ override def path: ActorPath = RootActorPath(Address("pekko", "all-systems"), "/errorCounter")
+ override def provider: ActorRefProvider = throw new UnsupportedOperationException("test ref")
+ override def !(message: Any)(implicit sender: ActorRef): Unit = {
+ published.incrementAndGet()
+ ()
+ }
+ }
+
+ "A message sent while the executor is rejecting" must {
+
+ "not recurse into a StackOverflowError" in {
+ val probe = TestProbe()
+ val ref = echoActor(probe)
+ val published = new AtomicInteger(0)
+ val counter = errorCounter(published)
+ system.eventStream.subscribe(counter, classOf[Logging.Error])
+
+ RejectingExecutorConfigurator.rejecting.set(true)
+ try {
+ // Before the guard, the rejection report published to the logger actor, whose own dispatch
+ // was rejected by the same executor, whose report published again, all on this thread.
+ ref ! "hello"
+ } catch {
+ case e: StackOverflowError =>
+ fail("reporting the rejected execution recursed until the stack overflowed", e)
+ } finally {
+ RejectingExecutorConfigurator.rejecting.set(false)
+ system.eventStream.unsubscribe(counter)
+ }
+
+ // The send above really was rejected and reported, so the assertion is not vacuous.
+ published.get() should be > 0
+ }
+
+ "still dispatch normally once the executor accepts again" in {
+ val rejectedProbe = TestProbe()
+ val rejected = echoActor(rejectedProbe)
+
+ RejectingExecutorConfigurator.rejecting.set(true)
+ try rejected ! "hello"
+ finally RejectingExecutorConfigurator.rejecting.set(false)
+
+ // A fresh actor, so this does not depend on when the rejected message above is drained.
+ val probe = TestProbe()
+ val fresh = system.actorOf(Props(new Actor {
+ override def receive: Receive = { case msg => probe.ref ! msg }
+ }))
+
+ fresh ! "delivered"
+ probe.expectMsg("delivered")
+ }
+ }
+
+ "DispatchFailureReporting" must {
+
+ "send a nested failure to the stdout logger instead of back to the EventStream" in {
+ val depth = new AtomicInteger(0)
+ val maxDepth = new AtomicInteger(0)
+ val deliveries = new AtomicInteger(0)
+
+ val reentrant = new MinimalActorRef {
+ override def path: ActorPath = RootActorPath(Address("pekko", "all-systems"), "/reentrant")
+ override def provider: ActorRefProvider = throw new UnsupportedOperationException("test ref")
+ override def !(message: Any)(implicit sender: ActorRef): Unit = {
+ deliveries.incrementAndGet()
+ val d = depth.incrementAndGet()
+ maxDepth.updateAndGet(math.max(_, d))
+ try {
+ if (d < 100)
+ DispatchFailureReporting.report(
+ system.eventStream,
+ Error(new RuntimeException("nested"), "test", getClass, s"nested at depth $d"))
+ } finally depth.decrementAndGet()
+ }
+ }
+
+ system.eventStream.subscribe(reentrant, classOf[Logging.Error])
+ try {
+ DispatchFailureReporting.report(
+ system.eventStream,
+ Error(new RuntimeException("outer"), "test", getClass, "outer failure"))
+ // The nested report must not have come back through the bus, or this would be unbounded.
+ maxDepth.get() shouldBe 1
+ deliveries.get() shouldBe 1
+
+ // The guard is a set of threads shared by every dispatcher, so an entry that was not
+ // released would divert every later report from this thread. It must reach the bus again.
+ DispatchFailureReporting.report(
+ system.eventStream,
+ Error(new RuntimeException("second outer"), "test", getClass, "second outer failure"))
+ maxDepth.get() shouldBe 1
+ deliveries.get() shouldBe 2
+ } finally system.eventStream.unsubscribe(reentrant)
+ }
+
+ "not suppress reports that arrive concurrently from different threads" in {
+ val threads = 16
+ val deliveries = new AtomicInteger(0)
+ val counter = errorCounter(deliveries)
+ val started = new CountDownLatch(threads)
+ val proceed = new CountDownLatch(1)
+ val done = new CountDownLatch(threads)
+ system.eventStream.subscribe(counter, classOf[Logging.Error])
+ try {
+ (1 to threads).foreach { _ =>
+ new Thread(new Runnable {
+ override def run(): Unit = {
+ started.countDown()
+ try {
+ proceed.await()
+ DispatchFailureReporting.report(
+ system.eventStream,
+ Error(new RuntimeException("concurrent"), "test", getClass, "concurrent failure"))
+ } catch {
+ case _: InterruptedException => Thread.currentThread().interrupt()
+ } finally done.countDown()
+ }
+ }).start()
+ }
+ started.await(10, TimeUnit.SECONDS) shouldBe true
+ proceed.countDown()
+ done.await(10, TimeUnit.SECONDS) shouldBe true
+
+ // Every thread was the outermost reporter on its own stack, so all of them must reach the
+ // bus. A guard shared across threads rather than keyed by thread would divert some.
+ deliveries.get() shouldBe threads
Review Comment:
If `started.await(...) shouldBe true` fails, the test will throw before `proceed.countDown()`, leaving the spawned non-daemon threads blocked on `proceed.await()` and potentially hanging the test JVM. Make sure `proceed.countDown()` runs in a `finally` (or make the threads daemon and/or interrupt them on failure) so the test can always terminate cleanly even when assertions fail.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org