Memory leak when using flatMapPrefix/prefixAndTail with Concat

[queimadus]

When using flatMapPrefix or prefixAndTail (or any other flow that closes over data) followed by concat the accumulated portion doesn't seem to be GC'd.

I reduced a problem I found into this snippet:

val s = Source
  .repeat(())
  .map(_ => ByteString('a' * 40000))
  .take(1000000)
  .flatMapPrefix(50000) { prefix => Flow[ByteString] } 

Source.empty
  .concatAllLazy(List.tabulate(30000)(_ => s): _*)
  .runWith(Sink.ignore).onComplete(println(_))  

If we don't use concat, it runs fine:

  Source
    .repeat(s)
    .take(30000)
    .flatMapConcat(x => x)
    .runWith(Sink.ignore)
    .onComplete(println(_))

[He-Pin]

@queimadus What about wrapping it into a Source. lazySource ? the concatAll will materialize all sources

[He-Pin]

I tooked a while of the code in the graph stage , where there is an array of input handers and connections, I think we should clean that array slot after a stage is completed, then we need more changes eg when the slot is null, we should treat the connection is closed where it's now protected by flags.

I did not tried that today, would you take a try? I mean clone the project and made the changes to see if it works.

[queimadus]

I can take a look at it somewhere in the next few days!

[He-Pin]

Me too, I will take your repo as a starting point tomorrow.

[queimadus]

Tested it with a very naive change, just to check if it would make any difference in the test example, and got no difference in heap behavior. Maybe too much of a naive change 😅

diff --git a/stream/src/main/scala/org/apache/pekko/stream/scaladsl/Graph.scala b/stream/src/main/scala/org/apache/pekko/stream/scaladsl/Graph.scala
index bf28108..aef5833 100755
--- a/stream/src/main/scala/org/apache/pekko/stream/scaladsl/Graph.scala
+++ b/stream/src/main/scala/org/apache/pekko/stream/scaladsl/Graph.scala
@@ -1349,6 +1349,7 @@ final class Concat[T](val inputPorts: Int) extends GraphStage[UniformFanInShape[
             }
 
             override def onUpstreamFinish() = {
+              connDelete(i)
               if (idx == activeStream) {
                 activeStream += 1
                 // Skip closed inputs
diff --git a/stream/src/main/scala/org/apache/pekko/stream/stage/GraphStage.scala b/stream/src/main/scala/org/apache/pekko/stream/stage/GraphStage.scala
index b186308..e09f123 100644
--- a/stream/src/main/scala/org/apache/pekko/stream/stage/GraphStage.scala
+++ b/stream/src/main/scala/org/apache/pekko/stream/stage/GraphStage.scala
@@ -483,6 +483,10 @@ abstract class GraphStageLogic private[stream] (val inCount: Int, val outCount:
 
   private def conn(in: Inlet[_]): Connection = portToConn(in.id)
   private def conn(out: Outlet[_]): Connection = portToConn(out.id + inCount)
+  def connDelete(in: Inlet[_]): Unit = {
+    portToConn(in.id) = null
+    println((portToConn.count(_ eq null), portToConn.size))
+  }
 
   /**
    * Retrieves the current callback for the events on the given [[Outlet]]

[He-Pin]

Thanks, Another point of leak is the AsyncCallback, where is actually a queue behind of it, i mean the mailbox of the interpreter's backing actor.