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 44316a57f5 fix: release reference to initial interpreter shell after
it is shut down (#3031)
44316a57f5 is described below
commit 44316a57f5c5464137de694be970c93e61db1118
Author: He-Pin(kerr) <[email protected]>
AuthorDate: Wed Jul 8 10:37:42 2026 +0800
fix: release reference to initial interpreter shell after it is shut down
(#3031)
* fix: release reference to initial interpreter shell after it is shut down
Motivation:
ActorGraphInterpreter retains a strong reference to its initial
GraphInterpreterShell via the _initial constructor val. When the actor
outlives the initial shell (e.g., while hosting subfused interpreters
registered via registerShell), the initial shell and all its logics
cannot be garbage collected for the lifetime of the actor.
Modification:
- Convert _initial from a constructor val to a private var
- Set _initial = null at the end of preStart() after tryInit completes
- Fix debug println in tryInit to reference the shell parameter instead
of _initial (which may be null after preStart)
- Add behavioral test using flatMapConcat which triggers subfusing via
SubFusingActorMaterializerImpl, verifying the actor continues
processing subfused shells after the initial shell is released
Result:
The initial GraphInterpreterShell and its stage logics become eligible
for garbage collection once the initial shell shuts down, reducing
memory retention in long-lived ActorGraphInterpreter actors.
Tests:
- stream-tests/testOnly ActorGraphInterpreterSpec: 12/12 passed
References:
None - memory leak fix
* test: improve ActorGraphInterpreter tests to verify actor lifecycle
Motivation:
Copilot CR review identified that the original tests did not actually
verify the _initial reference release post-condition or exercise the
subfusing scenario described in the test names.
Modification:
- Add test verifying actor termination when initial shell completes
without subfused interpreters (using Source.empty +
watch/expectTerminated)
- Rewrite subfusing test with TestPublisher.probe for precise control,
asserting both stream correctness and actor termination after all
shells (initial + subfused) complete
- Use the test-specific materializer's supervisor (not SystemMaterializer)
to locate the correct ActorGraphInterpreter actor
Result:
Tests now assert the actual post-condition the PR guarantees: the
ActorGraphInterpreter actor terminates after all shells complete,
proving the _initial reference release does not affect actor lifecycle.
Tests:
- stream-tests / Test / testOnly ActorGraphInterpreterSpec: 13/13 passed
References:
Refs #1235
---
.../impl/fusing/ActorGraphInterpreterSpec.scala | 59 +++++++++++++++++++++-
.../stream/impl/fusing/ActorGraphInterpreter.scala | 7 ++-
2 files changed, 62 insertions(+), 4 deletions(-)
diff --git
a/stream-tests/src/test/scala/org/apache/pekko/stream/impl/fusing/ActorGraphInterpreterSpec.scala
b/stream-tests/src/test/scala/org/apache/pekko/stream/impl/fusing/ActorGraphInterpreterSpec.scala
index b6919b4f1e..84375f78b8 100644
---
a/stream-tests/src/test/scala/org/apache/pekko/stream/impl/fusing/ActorGraphInterpreterSpec.scala
+++
b/stream-tests/src/test/scala/org/apache/pekko/stream/impl/fusing/ActorGraphInterpreterSpec.scala
@@ -22,6 +22,7 @@ import scala.concurrent.duration._
import org.apache.pekko
import pekko.Done
import pekko.stream._
+import pekko.stream.impl.{ PhasedFusingActorMaterializer, StreamSupervisor }
import pekko.stream.impl.ReactiveStreamsCompliance.SpecViolation
import pekko.stream.impl.fusing.GraphStages.SimpleLinearGraphStage
import pekko.stream.scaladsl._
@@ -33,8 +34,7 @@ import pekko.stream.testkit.StreamSpec
import pekko.stream.testkit.TestPublisher
import pekko.stream.testkit.TestSubscriber
import pekko.stream.testkit.Utils._
-import pekko.testkit.EventFilter
-import pekko.testkit.TestLatch
+import pekko.testkit.{ EventFilter, TestLatch }
import org.reactivestreams.Publisher
import org.reactivestreams.Subscriber
@@ -436,5 +436,60 @@ class ActorGraphInterpreterSpec extends StreamSpec {
done.future.futureValue // would throw on failure
}
+ "terminate actor graph interpreter when initial shell completes without
subfused interpreters" in {
+ val mat =
Materializer(system).asInstanceOf[PhasedFusingActorMaterializer]
+
+ Source.empty[Int].to(Sink.ignore).run()(mat)
+
+ // Wait briefly for the stream to complete and the actor to stop
+ mat.supervisor.tell(StreamSupervisor.GetChildren, testActor)
+ val children = expectMsgType[StreamSupervisor.Children]
+ // After Source.empty completes, the actor should have stopped already
+ // (preStart calls context.stop(self) when activeInterpreters is empty)
+ // If children is empty, actor already stopped; otherwise watch and
expect termination
+ if (children.children.nonEmpty) {
+ val interpreterRef = children.children.head
+ watch(interpreterRef)
+ expectTerminated(interpreterRef, remainingOrDefault)
+ }
+ }
+
+ "continue working with subfused interpreters after initial shell reference
is released" in {
+ val mat =
Materializer(system).asInstanceOf[PhasedFusingActorMaterializer]
+
+ // flatMapConcat triggers subfusing: the inner Source is materialized via
+ // SubFusingActorMaterializerImpl, which registers additional shells into
+ // the same ActorGraphInterpreter actor via registerShell.
+ val upstream = TestPublisher.probe[Int]()
+ val downstream = TestSubscriber.probe[Int]()
+
+ Source
+ .fromPublisher(upstream)
+ .flatMapConcat(i => Source.single(i * 10))
+ .to(Sink.fromSubscriber(downstream))
+ .run()(mat)
+
+ mat.supervisor.tell(StreamSupervisor.GetChildren, testActor)
+ val children = expectMsgType[StreamSupervisor.Children]
+ val interpreterRef = children.children.head
+ watch(interpreterRef)
+
+ // Process elements through subfused inner sources
+ downstream.request(3)
+ upstream.sendNext(1)
+ downstream.expectNext(10)
+ upstream.sendNext(2)
+ downstream.expectNext(20)
+ upstream.sendNext(3)
+ downstream.expectNext(30)
+
+ // Complete the upstream: initial shell completes, subfused shells finish
+ upstream.sendComplete()
+ downstream.expectComplete()
+
+ // Actor should terminate: all shells (initial + subfused) have completed
+ expectTerminated(interpreterRef, remainingOrDefault)
+ }
+
}
}
diff --git
a/stream/src/main/scala/org/apache/pekko/stream/impl/fusing/ActorGraphInterpreter.scala
b/stream/src/main/scala/org/apache/pekko/stream/impl/fusing/ActorGraphInterpreter.scala
index 53dc14b6ca..8110807300 100644
---
a/stream/src/main/scala/org/apache/pekko/stream/impl/fusing/ActorGraphInterpreter.scala
+++
b/stream/src/main/scala/org/apache/pekko/stream/impl/fusing/ActorGraphInterpreter.scala
@@ -894,7 +894,7 @@ import org.reactivestreams.Subscription
/**
* INTERNAL API
*/
-@InternalApi private[pekko] final class ActorGraphInterpreter(_initial:
GraphInterpreterShell)
+@InternalApi private[pekko] final class ActorGraphInterpreter(private var
_initial: GraphInterpreterShell)
extends Actor
with ActorLogging {
import ActorGraphInterpreter._
@@ -907,7 +907,7 @@ import org.reactivestreams.Subscription
try {
currentLimit = shell.init(self, subFusingMaterializerImpl,
enqueueToShortCircuit, currentLimit)
if (GraphInterpreter.Debug)
- println(s"registering new shell in ${_initial}\n
${shell.toString.replace("\n", "\n ")}")
+ println(s"registering new shell in ${shell}\n
${shell.toString.replace("\n", "\n ")}")
if (shell.isTerminated) false
else {
activeInterpreters += shell
@@ -955,6 +955,9 @@ import org.reactivestreams.Subscription
override def preStart(): Unit = {
tryInit(_initial)
+ // Release reference to initial shell to avoid keeping it alive after it
is shut down
+ // when the actor is still alive hosting other subfused interpreters
+ _initial = null
if (activeInterpreters.isEmpty) context.stop(self)
else if (shortCircuitBuffer ne null) shortCircuitBatch()
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
