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 85bc2aa1d3 perf(stream): extract reportStageError out of
GraphInterpreter.execute hot loop (#3119)
85bc2aa1d3 is described below
commit 85bc2aa1d32a4a27f4046dd6b4ddb1c191a8e7ab
Author: He-Pin(kerr) <[email protected]>
AuthorDate: Mon Jun 22 17:38:08 2026 +0800
perf(stream): extract reportStageError out of GraphInterpreter.execute hot
loop (#3119)
* perf(stream): extract reportStageError out of GraphInterpreter.execute
hot loop
Motivation:
GraphInterpreter.execute is invoked once per event in the stream engine and
shows up as
a ~10% CPU hotspot in flamegraphs of high-throughput gRPC workloads. The
nested
`def reportStageError` closure inside the while-loop forced the Scala
compiler to emit
closure-capture bytecode (loading activeStage, chasedPush, chasedPull,
chaseCounter,
log, logics) on every iteration, even though every captured name is already
a class field
and the closure is only invoked on the error path.
Modification:
Lift `reportStageError` to a private[stream] instance method of
GraphInterpreter, placed
between `shutdownCounters` and `execute`. The three try/catch sites in
execute (normal
dispatch, chased PUSH, chased PULL) now call the method directly. Rename
the outer
`catch case NonFatal(e)` local to `ex` to remove name-shadowing with the
method parameter.
Result:
- The hot while-loop body in `execute` shrinks by ~30 lines of bytecode,
giving the JIT
a tighter compile target and a better chance of inlining
`processEvent`/`processPush`.
- Zero per-iteration closure-capture overhead (previously the nested `def`
produced a
fresh closure allocation each loop pass before JIT escape analysis could
eliminate it).
- Behavior is preserved: `GraphInterpreterSpec` 11/11 tests pass
(identity/chained/
detacher/zip/broadcast/merge/balance/cycle).
- Incidental benefit: error-path logging is now declared once and easier to
audit.
Tests:
- sbt "stream-tests/Test/testOnly
org.apache.pekko.stream.impl.fusing.GraphInterpreterSpec"
-> 11 passed, 0 failed.
References:
None - internal stream engine refactor.
* refactor(stream): address PR review on reportStageError lift
Motivation:
An independent review of #3119 identified three inaccuracies in the original
submission:
1. The claim that the nested `def reportStageError` inside `execute`
allocated
a fresh closure per loop iteration was incorrect. Scala 2.13 compiles the
nested `def` as a private synthetic method on the enclosing class when
only
class fields are referenced — verified via `javap` against the baseline
bytecode (`private final void reportStageError$1(java.lang.Throwable)`).
2. The claim that `ActorGraphInterpreter` referenced the lifted method was
false — a search of the file yields zero references.
3. The `private[stream]` visibility granted package-wide access that no
caller
actually uses.
Modification:
- Rewrite the scaladoc on `reportStageError` to describe the real benefit:
smaller `execute` bytecode for a tighter JIT compile target on the hot
`processPush`/`processPull` paths that follow each try/catch. No
allocation
reduction is claimed.
- Narrow visibility from `private[stream]` to `private` (all three call
sites
live in `GraphInterpreter.execute` itself).
- Rename the catch-bound local `e` to `ex` at the remaining two call sites
(chased PUSH and chased PULL) so all three try/catch blocks use the same
naming convention. The main-dispatch call site was already renamed in the
previous commit.
Result:
- Semantics preserved character-for-character (same log-level dispatch, same
`failStage`, same chasing-abort branch).
- `GraphInterpreterSpec` 11/11 and `GraphInterpreterFailureModesSpec` 9/9
pass locally; the existing failure spec already provides directional
coverage of the refactored method through every failure hook.
- PR description updated in #3119 to reflect the corrected justification
and to flag the JMH variance caveat.
Tests:
- sbt "stream-tests/Test/testOnly
org.apache.pekko.stream.impl.fusing.GraphInterpreterSpec
org.apache.pekko.stream.impl.fusing.GraphInterpreterFailureModesSpec"
-> 20 passed, 0 failed.
References:
Refs #3119
* refactor(stream): revert cosmetic ex rename per maintainer nit
Motivation:
A maintainer nit on #3119 flagged the catch-bound local rename `e` -> `ex`
introduced in the previous commit as unnecessary. There is no real
shadowing:
the catch-bound `e` and the method parameter `e` on `reportStageError` live
in completely different lexical scopes, so the rename adds no clarity and
diverges from the surrounding codebase convention.
Modification:
Revert `case NonFatal(ex) => reportStageError(ex)` back to
`case NonFatal(e) => reportStageError(e)` at all three call sites in
`GraphInterpreter.execute` (normal dispatch, chased PUSH, chased PULL).
No other changes.
Result:
- Code matches the surrounding convention in the stream engine.
- Semantics identical (catch-bound `e` still bound per call site, method
parameter `e` still bound per invocation).
- `GraphInterpreterSpec` 11/11 and `GraphInterpreterFailureModesSpec` 9/9
pass locally.
- PR description updated to record the revert as reviewer-driven revision
#5.
Tests:
- sbt "stream-tests/Test/testOnly
org.apache.pekko.stream.impl.fusing.GraphInterpreterSpec
org.apache.pekko.stream.impl.fusing.GraphInterpreterFailureModesSpec"
-> 20 passed, 0 failed.
References:
Refs #3119
* refactor(stream): drop redundant @InternalApi on private def
Motivation:
A maintainer comment on #3119 noted that `@InternalApi` on a class-scoped
`private def` is meaningless — the annotation documents that a *non-private*
method is part of Pekko's internal API and may change without notice. When
the method is already `private` to the enclosing class, no external caller
can see it regardless, so the annotation adds no information.
Modification:
Remove `@InternalApi` from `reportStageError`. The `InternalApi` import is
retained because it is still used on the enclosing `GraphInterpreter`
companion object, the class itself, and on `activeStage` / `nonNull` which
are `private[stream]` (package-visible) and therefore still warrant the
annotation.
Result:
- No behavioral change; the method remains class-private and its semantics
are preserved character-for-character.
- `GraphInterpreterSpec` 11/11 and `GraphInterpreterFailureModesSpec` 9/9
pass locally.
Tests:
- sbt "stream-tests/Test/testOnly
org.apache.pekko.stream.impl.fusing.GraphInterpreterSpec
org.apache.pekko.stream.impl.fusing.GraphInterpreterFailureModesSpec"
-> 20 passed, 0 failed.
References:
Refs #3119
---
.../stream/impl/fusing/GraphInterpreter.scala | 76 ++++++++++++----------
1 file changed, 43 insertions(+), 33 deletions(-)
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
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
