perf(stream): extract reportStageError out of GraphInterpreter.execute hot loop

[He-Pin]

Motivation

GraphInterpreter.execute is on the hottest path of the stream engine — it runs once per queued event. Inside the main event loop it declares a local def reportStageError(e: Throwable), which the Scala 2.13 compiler lifts into a private final synthetic method on the enclosing class (not a per-iteration closure allocation, since the body only references class-level fields and methods). Lifting it explicitly out of the loop makes the intent obvious, shrinks the body of execute, and gives the JIT a tighter compile target for inlining the genuinely hot processPush/processPull paths that follow each catch block.

Modification

Move the nested reportStageError definition out of execute into a private instance method on GraphInterpreter, placed between shutdownCounters and execute. The three call sites in execute (normal dispatch, chased PUSH, chased PULL) now invoke the instance method directly. The method body and the catch-bound locals at the call sites are unchanged.

Visibility is private (not private[stream]) because the only call sites live in GraphInterpreter.execute itself — ActorGraphInterpreter does not reference this method, and there is no planned external caller. If one materialises, widening later is a one-line change.

Because activeStage, chasedPush, chasedPull, chaseCounter, log, logics, defaultErrorReportingLogLevel, enqueue, and afterStageHasRun are all fields/methods of the enclosing GraphInterpreter, the lifted method observes exactly the same state as the previous nested def, and the chasing abort semantics (reset chaseCounter, re-enqueue pending chased push/pull) are preserved character-for-character.

Result

• Smaller execute bytecode → friendlier to JIT inlining of the hot processPush/processPull calls that follow each try/catch.
• Identical error-path semantics: log via the configured LogLevels attribute, call activeStage.failStage, reset chaseCounter, and re-enqueue any pending chased push/pull.
• No allocation reduction — a previous revision of this PR claimed per-iteration closure allocation; that was incorrect. Scala 2.13 already compiles the nested def as a synthetic method when only class fields are referenced.

Benchmarks

Initial JMH data with InterpreterBenchmark (graph_interpreter_100k_elements, 100k elements per invocation) showed improvement in the shallow-pipeline case:

numberOfIds	Baseline (main)	With this PR	Δ
1	45,474.58 ± 7,621.99	50,295.10 ± 1,221.59	+10.6%
5	11,731.50 ± 1,012.70	12,224.65 ± 866.05	+4.2%
10	5,879.32 ± 148.51	5,795.71 ± 233.77	~0%

Caveats, flagged by the reviewer:

• The baseline's error band on numberOfIds=1 (±7,622, ~17% relative) is far wider than the PR run (±1,222, ~2.4%), and the two confidence intervals overlap. This suggests the baseline run was noisy (GC, thermal, background processes) and the +10.6% should be treated as indicative rather than statistically decisive.
• The effect shrinks with pipeline depth, as expected — per-event overhead matters less when stage work dominates.

The honest takeaway is: this change produces smaller, cleaner bytecode for execute, which is genuinely beneficial for JIT compilation regardless of the exact throughput delta. Reviewers should treat any performance claim here as "likely small, not regressive, and hard to measure at this level."

Tests

• sbt "stream-tests/Test/testOnly org.apache.pekko.stream.impl.fusing.GraphInterpreterSpec" — 11/11 passed.
• sbt "stream-tests/Test/testOnly org.apache.pekko.stream.impl.fusing.GraphInterpreterFailureModesSpec" — 9/9 passed. All 9 failure scenarios (failure in onPull, onPush, onUpstreamFinish, onUpstreamFailure, onDownstreamFinish, preStart, postStop, plus pending cancel/complete) exercise reportStageError through execute's catch blocks.
• sbt "bench-jmh/Jmh/run ... .*InterpreterBenchmark.*" — results above.

No new test was added. The existing GraphInterpreterFailureModesSpec already provides directional coverage of the refactored method through every failure hook; a separate test targeting reportStageError in isolation would duplicate that coverage without improving regression detection (the method is private and reachable only through execute, which is precisely what the existing specs invoke).

Reviewer-Driven Revisions

Following an independent review, the following were corrected from the initial submission:

1. Removed the incorrect "per-iteration closure allocation" claim. Scala 2.13 compiles the nested def as a synthetic method on the enclosing class when only class fields are referenced, so no closure is allocated per iteration.
2. Narrowed private[stream] to private (the only callers are in this class; ActorGraphInterpreter does not reference the method).
3. Removed the incorrect ActorGraphInterpreter usage claim from the description.
4. Rewrote the scaladoc to describe the real benefit (smaller execute bytecode for JIT) rather than allocation reduction.
5. Reverted a cosmetic rename of the catch-bound local from e to ex — per maintainer nit, there is no real shadowing since the catch-bound e and the method parameter e live in completely different scopes.

References

None.

[pjfanning]

lgtm

A good candidate for backporting to 1.7.x branch

[He-Pin]

I think we can have better numbers in Pekko GRPC 2.0.0. https://github.com/LesnyRumcajs/grpc_bench

[He-Pin]

Addressed both maintainer nits in two follow-up commits on this branch:

• 1913c46 — Reverted the e → ex rename at all three call sites. Confirmed no real shadowing: the catch-bound e and the method parameter e live in different lexical scopes.
• 3963499 — Removed the redundant @InternalApi from reportStageError. The InternalApi import is retained because it is still used on the enclosing GraphInterpreter companion object, the class itself, and the private[stream] activeStage/nonNull members (where the annotation remains meaningful since those members are package-visible).

Both commits verified locally: GraphInterpreterSpec 11/11 and GraphInterpreterFailureModesSpec 9/9 pass.

PR description updated to record these as reviewer-driven revisions. Thanks for the careful review.

[pjfanning]

@He-Pin is this worth backporting to 1.7.x?

[He-Pin]

I don't think so, this is just a micro optimization, wdyt, otherwise we have much to backport.