fix: Fix leak in FlatMapPrefix operator.

[He-Pin]

extracted from #1621

Motivation:
In #1566 , @queimadus points out a leak

Modification:

1. change the builder from a buffer to a vector builder.
2. free that builder to null after the prefix is generated.

Result:
leak fixed.

import org.apache.pekko.actor.ActorSystem
import org.apache.pekko.stream.scaladsl.{Flow, Sink, Source}
import org.apache.pekko.util.ByteString

import scala.concurrent.Await


object PekkoQuickstart extends App {
  private implicit val system: ActorSystem = ActorSystem()

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

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

  Await.result(r, scala.concurrent.duration.Duration.Inf)
  println(r.value)

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

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

[He-Pin]

@queimadus would you like to give this patch a try?

[queimadus]

I'll give it a try but I'm not sure this will work because prefixAndTail uses a similar implementation and has the same problem described in #1566.

[He-Pin]

@queimadus There is a fix for prefixAndTail too #1623

[He-Pin]

There do have leaks in the interpreter, eg the logic is still being refs.
I did a try in #1621, which needs changing more code to clean up the logics.

[queimadus]

This change does seem to help when running the provided example. This crash by lack of memory. Previous run would crash out at around the 2 minute mark.

[He-Pin]

[queimadus]

My original code before I simplified it with what looked to be the MRE, was more of a:

  def myLogic(prefix: Seq[ByteString]): Flow[ByteString, ByteString, NotUsed] =
    Flow[ByteString]

  val s = Source
    .repeat(())
    .map(_ => ByteString('a' * 400000))
    .take(1000000)
    .prefixAndTail(50000)
    .flatMapConcat { case (prefix, tail) =>
      Source(prefix).concatLazy(tail).via(myLogic(prefix))
    }

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

Where I would need to use the prefix to generate the remaining flow logic and then emit the prefix back again. It still runs out of heap in this case, but it may be a different problem?

Nonetheless, I think this change makes sense.

[He-Pin]

prefixAndTail fix is in another PR, could you verify that too?
@queimadus #1623 this one

[queimadus]

For reference, this would be the flatMapPrefix equivalent that wouldn't work with Concats but work with flatMapConcat (the commented code):

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

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

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

[He-Pin]

I think the concat should be reimplemented to reduce memory usage somehow.