diff --git a/stream/src/main/scala/org/apache/pekko/stream/impl/fusing/GraphInterpreter.scala b/stream/src/main/scala/org/apache/pekko/stream/impl/fusing/GraphInterpreter.scala index edaa45f31f..7c7db1bca0 100644 --- a/stream/src/main/scala/org/apache/pekko/stream/impl/fusing/GraphInterpreter.scala +++ b/stream/src/main/scala/org/apache/pekko/stream/impl/fusing/GraphInterpreter.scala @@ -365,6 +365,49 @@ import pekko.stream.stage._ private def shutdownCounters: String = shutdownCounter.map(x => if (x >= KeepGoingFlag) s"${x & KeepGoingMask}(KeepGoing)" else x.toString).mkString(",") + /** + * INTERNAL API + * + * Reports a stage error: logs it according to the configured level, fails the active stage, and aborts any + * in-flight event chasing by re-enqueuing chased events. Lifted out of [[execute]] to shrink the hot loop's + * bytecode and give the JIT a tighter compile target for inlining `processPush`/`processPull` — the nested + * `def` that previously lived inside the loop was already compiled as a synthetic method on this class (Scala + * 2.13 does not allocate a fresh closure per iteration when the body only references class fields), so the + * gain here is purely in method-body size, not in allocation reduction. + */ + private def reportStageError(e: Throwable): Unit = { + if (activeStage eq null) throw e + else { + val logAt: Logging.LogLevel = activeStage.attributes.get[LogLevels] match { + case Some(levels) => levels.onFailure + case None => defaultErrorReportingLogLevel + } + logAt match { + case Logging.ErrorLevel => + log.error(e, "Error in stage [{}]: {}", activeStage.toString, e.getMessage) + case Logging.WarningLevel => + log.warning(e, "Error in stage [{}]: {}", activeStage.toString, e.getMessage) + case Logging.InfoLevel => + log.info("Error in stage [{}]: {}", activeStage.toString, e.getMessage) + case Logging.DebugLevel => + log.debug("Error in stage [{}]: {}", activeStage.toString, e.getMessage) + case _ => // Off, nop + } + activeStage.failStage(e) + + // Abort chasing + chaseCounter = 0 + if (chasedPush ne NoEvent) { + enqueue(chasedPush) + chasedPush = NoEvent + } + if (chasedPull ne NoEvent) { + enqueue(chasedPull) + chasedPull = NoEvent + } + } + } + /** * Executes pending events until the given limit is met. If there were remaining events, isSuspended will return * true. @@ -382,39 +425,6 @@ import pekko.stream.stage._ eventsRemaining -= 1 chaseCounter = math.min(ChaseLimit, eventsRemaining) - def reportStageError(e: Throwable): Unit = { - if (activeStage eq null) throw e - else { - val logAt: Logging.LogLevel = activeStage.attributes.get[LogLevels] match { - case Some(levels) => levels.onFailure - case None => defaultErrorReportingLogLevel - } - logAt match { - case Logging.ErrorLevel => - log.error(e, "Error in stage [{}]: {}", activeStage.toString, e.getMessage) - case Logging.WarningLevel => - log.warning(e, "Error in stage [{}]: {}", activeStage.toString, e.getMessage) - case Logging.InfoLevel => - log.info("Error in stage [{}]: {}", activeStage.toString, e.getMessage) - case Logging.DebugLevel => - log.debug("Error in stage [{}]: {}", activeStage.toString, e.getMessage) - case _ => // Off, nop - } - activeStage.failStage(e) - - // Abort chasing - chaseCounter = 0 - if (chasedPush ne NoEvent) { - enqueue(chasedPush) - chasedPush = NoEvent - } - if (chasedPull ne NoEvent) { - enqueue(chasedPull) - chasedPull = NoEvent - } - } - } - /* * This is the "normal" event processing code which dequeues directly from the internal event queue. Since * most execution paths tend to produce either a Push that will be propagated along a longer chain we take