fix: release references to completed stage logics in GraphInterpreter#3030
Open
He-Pin wants to merge 1 commit into
Open
fix: release references to completed stage logics in GraphInterpreter#3030He-Pin wants to merge 1 commit into
He-Pin wants to merge 1 commit into
Conversation
Contributor
There was a problem hiding this comment.
Pull request overview
This PR attempts to reduce GraphInterpreter memory retention by clearing stage logic references after finalization and adapting snapshots/tests for released logics.
Changes:
- Clears entries from
GraphInterpreter.logicsduringfinish(). - Adds null-aware logic snapshot handling.
- Adds tests asserting the
logicsarray is nulled afterfinish().
Reviewed changes
Copilot reviewed 2 out of 2 changed files in this pull request and generated 3 comments.
| File | Description |
|---|---|
stream/src/main/scala/org/apache/pekko/stream/impl/fusing/GraphInterpreter.scala |
Updates finalization and snapshot behavior for null stage logic entries. |
stream-tests/src/test/scala/org/apache/pekko/stream/impl/fusing/GraphInterpreterSpec.scala |
Adds regression tests for releasing stage logic references. |
Comments suppressed due to low confidence (1)
stream/src/main/scala/org/apache/pekko/stream/impl/fusing/GraphInterpreter.scala:757
toSnapshotstill dereferenceslogicIndexes(connection.inOwner)/outOwner, butlogicIndexesnow excludes every released (null) logic. Afterfinish()nullslogics, any snapshot of a shell that still has itsconnectionsarray will throwNoSuchElementExceptioninstead of producing the<completed>snapshots added above. Use the owners' stablestageIdto indexlogicSnapshotsso released logics are handled.
val logicIndexes = logics.zipWithIndex.collect { case (stage, idx) if stage ne null => stage -> idx }.toMap
val connectionSnapshots = connections.filter(_ ne null).map { connection =>
ConnectionSnapshotImpl(
connection.id,
logicSnapshots(logicIndexes(connection.inOwner)),
logicSnapshots(logicIndexes(connection.outOwner)),
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
He-Pin
force-pushed
the
fix/graph-interpreter-memory-leak
branch
from
c58c59e to
d29add0
Compare
Member
|
@He-Pin there are merge conflicts and also open comments from Copilot that need actions. |
Member
Author
|
oops, will rebase and fix soon. the origin problem was reported by @jrudolph |
He-Pin
added a commit
to He-Pin/incubator-pekko
that referenced
this pull request
Jul 7, 2026
Motivation: GraphInterpreter retains strong references to completed stage logics in the logics array and Connection objects (inHandler/outHandler/inOwner/outOwner), preventing garbage collection even after stages have finished. This is especially problematic for long-running fused interpreters that host subfused shells, where completed initial stages remain referenced for the lifetime of the actor. Modification: - afterStageHasRun now scans ALL logics for completed stages instead of only checking the single activeStage. This ensures stages that complete without being activeStage (e.g., boundary stages calling complete(), cascading completions) are properly finalized and released. - finish() and afterStageHasRun null connection handlers (inHandler/ outHandler) for finalized stages, breaking handler-to-logic reference chains to enable garbage collection. - finish() additionally nulls inOwner/outOwner on all connections since the interpreter is being shut down and no more events will be processed. - logics(i) is set to null after finalizing each stage in both finish() and afterStageHasRun. - toSnapshot handles null logics gracefully with <completed> placeholder and uses connection.inOwner.stageId for positional indexing instead of identity-based map lookups that fail with NPE on null logics. - toSnapshot filters out connections with null inOwner/outOwner. - inLogicName/outLogicName/inOwnerName/outOwnerName debug methods guard against null logics and null owners. - Fix GraphStageLogicSpec "not double-terminate" test to save logic references before execute() since afterStageHasRun now releases completed logics. Result: Completed stage logics are properly finalized and released for GC during normal stream execution (not just in finish()), and all strong reference chains (logics array, connection handlers, connection owners) are broken when the interpreter is shut down, reducing memory retention in long-running fused interpreters. Tests: - stream-tests/testOnly GraphInterpreterSpec: 13/13 passed (includes 2 new regression tests) - stream-tests/testOnly GraphStageLogicSpec: 17/17 passed - stream-tests/testOnly InterpreterSpec+LifecycleInterpreterSpec+InterpreterSupervisionSpec+GraphInterpreterFailureModesSpec+ActorGraphInterpreterSpec: 65/65 passed References: Refs apache#3030, Related to akka/akka-core#23439
He-Pin
force-pushed
the
fix/graph-interpreter-memory-leak
branch
from
d29add0 to
3d0369b
Compare
He-Pin
added a commit
that referenced
this pull request
Jul 8, 2026
Motivation:\nWhen fused streams keep the GraphInterpreter alive, completed stage logics and connection owners/handlers must not remain strongly referenced.\n\nModification:\nFinalize completed stages after they run, release finalized connection-side references, and keep stable stage ids for diagnostics and snapshots after owners are cleared.\n\nResult:\nCompleted fused stages can be released while other stages keep running, and interpreter snapshots remain usable after release.\n\nTests:\n- rtk sbt "stream-tests / Test / testOnly org.apache.pekko.stream.impl.fusing.GraphInterpreterSpec"\n- rtk sbt "stream-tests / Test / testOnly org.apache.pekko.stream.impl.GraphStageLogicSpec org.apache.pekko.stream.impl.fusing.GraphInterpreterPortsSpec org.apache.pekko.stream.impl.fusing.LifecycleInterpreterSpec org.apache.pekko.stream.impl.fusing.GraphInterpreterFailureModesSpec"\n- rtk sbt "stream-tests / Test / testOnly org.apache.pekko.stream.impl.fusing.GraphInterpreterSpec org.apache.pekko.stream.impl.GraphStageLogicSpec"\n- rtk scalafmt --mode diff-ref=origin/main\n- rtk scalafmt --list --mode diff-ref=origin/main\n- rtk git diff --check\n- qodercli stdout review: No must-fix findings\n- subAgent review: No GraphInterpreter must-fix findings\n\nReferences:\nRefs #3030
Member
Author
|
@pjfanning I have address all the problems |
He-Pin
added a commit
that referenced
this pull request
Jul 8, 2026
Motivation: Fused stages keep references to all completed GraphStageLogic instances even after they finish, preventing GC while the GraphInterpreter is alive. Modification: - Null out logics(i) when a stage completes in finish() and afterStageHasRun() - Null connection handlers for finalized stages to break handler-to-logic chains - Handle null logics in inLogicName/outLogicName/toSnapshot Result: Completed stage logics can be garbage collected even if the GraphInterpreter is still alive due to other references. Tests: - sbt "stream-tests / Test / testOnly *GraphInterpreterSpec *GraphStageLogicSpec" References: Fixes #3030
He-Pin
force-pushed
the
fix/graph-interpreter-memory-leak
branch
from
a414517 to
0f76b9a
Compare
He-Pin
force-pushed
the
fix/graph-interpreter-memory-leak
branch
from
09e5103 to
be50923
Compare
Member
Author
|
@pjfanning wdyt |
pjfanning
reviewed
Jul 10, 2026
He-Pin
added a commit
that referenced
this pull request
Jul 10, 2026
Motivation: BitSet bookkeeping adds overhead for the small fused graphs handled by GraphInterpreter. Modification: Replace pending stage IDs with a boolean-gated traversal bounded by initialized stages. Add a regression test that preserves preStart/postStop ordering for zero-port stages. Result: Stage finalization no longer uses BitSet while cascading completion and zero-port lifecycle behavior remain correct. Tests: - Related GraphInterpreter and lifecycle suites: 127 tests passed - scalafmt --list --mode diff-ref=origin/main - git diff --check References: Refs #3030
He-Pin
added a commit
that referenced
this pull request
Jul 10, 2026
Motivation: Completed graph stages could leave connection payloads reachable, and snapshots could index unresolved stage ids. Modification: Reuse releaseStage from finish, clear slots after both owners are released, and safely represent unresolved snapshot owners. Add focused regression coverage. Result: Completed interpreters release in-flight payloads without breaking snapshots. Tests: - sbt "stream-tests / Test / testOnly org.apache.pekko.stream.impl.fusing.GraphInterpreterSpec org.apache.pekko.stream.impl.GraphStageLogicSpec org.apache.pekko.stream.impl.fusing.LifecycleInterpreterSpec" (44 tests passed) - sbt +mimaReportBinaryIssues (passed) - sbt +headerCheckAll (passed) - sbt checkCodeStyle (passed) - Qoder review (no must-fix findings) References: Refs #3030
He-Pin
added a commit
that referenced
this pull request
Jul 10, 2026
Motivation: Completed graph stages could remain reachable through interpreter logic, owner, handler, and connection payload references after finalization. Modification: Finalize completed stages with a bounded traversal, release their logic and connection references, retain only payloads that remain observable through grab(), and keep snapshots stable after owners are released. Add focused lifecycle and reference-release coverage. Result: Completed stages are released promptly without breaking cascading completion, initialization ordering, snapshots, or port semantics. Tests: - sbt "stream-tests / Test / testOnly org.apache.pekko.stream.impl.fusing.GraphInterpreterPortsSpec org.apache.pekko.stream.impl.fusing.GraphInterpreterSpec org.apache.pekko.stream.impl.GraphStageLogicSpec org.apache.pekko.stream.impl.fusing.LifecycleInterpreterSpec" (106 passed) - scalafmt --list --mode diff-ref=origin/main (passed) - sbt +headerCheckAll and checkCodeStyle (passed) - Qoder stdout review (No must-fix findings) - git diff --check origin/main..HEAD (passed) - +mimaReportBinaryIssues interrupted during final sbt reset after Scala 2.13/3.3 sub-runs completed; full CI requested - validatePullRequest not run - user requested remote CI References: Refs #3030
He-Pin
force-pushed
the
fix/graph-interpreter-memory-leak
branch
from
0589a50 to
7307c97
Compare
Motivation: Completed graph stages could remain reachable through interpreter logic, owner, handler, and connection payload references after finalization. Modification: Finalize completed stages with a bounded traversal, release their logic and connection references, retain only payloads that remain observable through grab(), and keep snapshots stable after owners are released. Add focused lifecycle and reference-release coverage. Result: Completed stages are released promptly without breaking cascading completion, initialization ordering, snapshots, or port semantics. Tests: - sbt "stream-tests / Test / testOnly org.apache.pekko.stream.impl.fusing.GraphInterpreterPortsSpec org.apache.pekko.stream.impl.fusing.GraphInterpreterSpec org.apache.pekko.stream.impl.GraphStageLogicSpec org.apache.pekko.stream.impl.fusing.LifecycleInterpreterSpec" (106 passed) - scalafmt --list --mode diff-ref=origin/main (passed) - sbt +headerCheckAll and checkCodeStyle (passed) - Qoder stdout review (No must-fix findings) - git diff --check origin/main..HEAD (passed) - +mimaReportBinaryIssues interrupted during final sbt reset after Scala 2.13/3.3 sub-runs completed; full CI requested - validatePullRequest not run - user requested remote CI References: Refs #3030
He-Pin
force-pushed
the
fix/graph-interpreter-memory-leak
branch
from
7307c97 to
6cbdfb9
Compare
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Motivation
When multiple stages are fused, a single alive stage keeps the GraphInterpreter alive and with it references to all already completed GraphStage logics, which may keep a significant amount of memory.
This issue was originally reported in akka/akka-core#23439.
Modification
Result
GraphInterpreter no longer keeps references to completed stage logics, allowing them to be garbage collected even if the interpreter is still alive due to other fused stages.
Tests
References