fix: release references to completed stage logics in GraphInterpreter#3030
Open
He-Pin wants to merge 2 commits into
Open
fix: release references to completed stage logics in GraphInterpreter#3030He-Pin wants to merge 2 commits into
He-Pin wants to merge 2 commits 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 |
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 #3030, Related to akka/akka-core#23439
He-Pin
force-pushed
the
fix/graph-interpreter-memory-leak
branch
from
d29add0 to
3d0369b
Compare
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 |
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