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 2fe223701f fix(stream): replace O(n) finalization scan with O(1)
counter in afterStageHasRun (#3373)
2fe223701f is described below
commit 2fe223701f9118117fa2224c3065e8e1293c198e
Author: He-Pin(kerr) <[email protected]>
AuthorDate: Wed Jul 22 19:37:32 2026 +0800
fix(stream): replace O(n) finalization scan with O(1) counter in
afterStageHasRun (#3373)
Motivation:
PR #3030 (commit 3fbb1848f3) introduced releaseStage() which nulls out
connection references during stage finalization. The afterStageHasRun
method was changed to scan all initializedStages on every event dispatch
to find completed stages. With 1M+ stages (InterpreterStressSpec), this
causes O(n^2) behavior — the test thread hangs indefinitely at
GraphInterpreter.afterStageHasRun (issue #3372).
Modification:
Replace the pendingFinalization Boolean with a pendingFinalizations Int
counter. The O(1) fast path finalizes the active stage directly and
decrements the counter by 1. The O(n) scan loop only runs when the
counter is still > 0 (cascading completion where multiple stages
completed during one event dispatch). A Boolean cannot distinguish
"one stage pending" from "multiple stages pending" — the fast path
would erase signals for other stages, breaking cascading finalization.
Result:
afterStageHasRun is O(1) in the common case (long chains where each
event completes exactly one stage). The O(n) scan only runs for
cascading completions (rare in practice). InterpreterStressSpec with
1M stages completes in bounded time instead of hanging.
Tests:
- sbt "stream-tests / Test / testOnly
org.apache.pekko.stream.impl.fusing.GraphInterpreterSpec" — 14/14 passed
- sbt "stream-tests / Test / testOnly
org.apache.pekko.stream.impl.fusing.GraphInterpreterPortsSpec
org.apache.pekko.stream.impl.fusing.LifecycleInterpreterSpec" — 75/75 passed
- sbt "stream / scalafmt" — no formatting changes needed
References:
Fixes #3372
---
.../stream/impl/fusing/GraphInterpreter.scala | 53 +++++++++++++++-------
1 file changed, 37 insertions(+), 16 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 10f06936c9..6d79526528 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
@@ -281,10 +281,14 @@ import pekko.stream.stage._
private var chaseCounter = 0 // the first events in preStart blocks should
be not chased
private var chasedPush: Connection = NoEvent
private var chasedPull: Connection = NoEvent
- // Set whenever a stage's shutdownCounter transitions to 0 (i.e. the stage
just became completed and
- // needs finalization). Lets the chase / dispatch loops skip the
per-iteration shutdownCounter array
- // load in afterStageHasRun when no stage has completed since the last
finalization pass.
- private var pendingFinalization: Boolean = false
+ // Counts stages whose shutdownCounter just transitioned to 0 (completed,
awaiting finalization).
+ // Incremented by completeConnection/setKeepGoing; decremented by the O(1)
fast path in
+ // afterStageHasRun. When > 0 after the fast path, the scan loop finalizes
remaining stages.
+ // A counter (not Boolean) is required: multiple stages may complete during
one event dispatch
+ // (e.g. cascading completion), and the fast path must not erase signals for
other pending stages.
+ // PERFORMANCE: the scan loop is O(initializedStages) — with 1M+ stages this
causes O(n^2) if it
+ // runs on every dispatch. The counter ensures it only runs when stages are
genuinely pending.
+ private var pendingFinalizations: Int = 0
private def queueStatus: String = {
val contents = (queueHead until queueTail).map(idx => {
@@ -375,7 +379,7 @@ import pekko.stream.stage._
i += 1
}
activeStage = null
- pendingFinalization = false
+ pendingFinalizations = 0
// Release connection references and any in-flight payloads retained by
the stopped interpreter
var j = 0
@@ -493,7 +497,7 @@ import pekko.stream.stage._
catch {
case NonFatal(e) => reportStageError(e)
}
- if (pendingFinalization) {
+ if (pendingFinalizations > 0) {
afterStageHasRun(activeStage)
}
@@ -528,7 +532,7 @@ import pekko.stream.stage._
catch {
case NonFatal(e) => reportStageError(e)
}
- if (pendingFinalization) {
+ if (pendingFinalizations > 0) {
afterStageHasRun(activeStage)
}
}
@@ -541,7 +545,7 @@ import pekko.stream.stage._
catch {
case NonFatal(e) => reportStageError(e)
}
- if (pendingFinalization) {
+ if (pendingFinalizations > 0) {
afterStageHasRun(activeStage)
}
}
@@ -687,11 +691,22 @@ import pekko.stream.stage._
}
def afterStageHasRun(logic: GraphStageLogic): Unit = {
- if ((logic ne null) && isStageCompleted(logic) && !isStageFinalized(logic))
- pendingFinalization = true
+ // O(1) fast path: finalize the stage that just ran if it completed.
+ // Guard the decrement: zero-port stages are completed from construction
(shutdownCounter == 0)
+ // without a corresponding completeConnection increment — must not
underflow the counter.
+ if ((logic ne null) && isStageCompleted(logic) &&
!isStageFinalized(logic)) {
+ markStageFinalized(logic)
+ runningStages -= 1
+ if (pendingFinalizations > 0) pendingFinalizations -= 1
+ finalizeStage(logic)
+ releaseStage(logic)
+ }
- while (pendingFinalization) {
- pendingFinalization = false
+ // O(n) scan: only entered when other stages also completed (cascading
completion).
+ // Uses while (not if) because finalizeStage → postStop → cancel can
complete further stages
+ // mid-scan, incrementing the counter again.
+ while (pendingFinalizations > 0) {
+ pendingFinalizations = 0
var stageId = 0
while (stageId < initializedStages) {
val completedLogic = logics(stageId)
@@ -764,16 +779,22 @@ import pekko.stream.stage._
if (activeConnections > 0) {
val next = activeConnections - 1
shutdownCounter(stageId) = next
- if (next == 0) pendingFinalization = true
+ if (next == 0) pendingFinalizations += 1
}
}
private[stream] def setKeepGoing(logic: GraphStageLogic, enabled: Boolean):
Unit =
if (enabled) shutdownCounter(logic.stageId) |= KeepGoingFlag
else {
- val next = shutdownCounter(logic.stageId) & KeepGoingMask
- shutdownCounter(logic.stageId) = next
- if (next == 0) pendingFinalization = true
+ val current = shutdownCounter(logic.stageId)
+ // Only signal finalization if the KeepGoing flag was actually set.
Without this guard,
+ // internalCompleteStage's trailing setKeepGoing(false) would spuriously
increment the
+ // counter when shutdownCounter is already 0 (no KeepGoing was ever
enabled).
+ if ((current & KeepGoingFlag) != 0) {
+ val next = current & KeepGoingMask
+ shutdownCounter(logic.stageId) = next
+ if (next == 0) pendingFinalizations += 1
+ }
}
@InternalStableApi
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
