concurrent.join leaks memory

[pchlupacek]

concurrent.join seems to leak memory, as following program will consume memory that retains in oldGen after manual GC.

object StreamsTestApp extends App {
  import Resources._

  def innerStream(idx:Int):Stream[Task, Unit] = {
    val allocate = "*"*10000
    Stream.eval(Task.delay(println(s"XXXG START $idx" + allocate.take(1))))
  }

  concurrent.join(Int.MaxValue)(
    time.sleep_[Task](10.seconds) ++
    Stream.range(0, 7000).map { i =>
      innerStream(i)
    } ++ Stream.eval_(Task.delay(println("XXXY DONE"))) ++
      time.sleep_[Task](1.hour)
  ).run.unsafeAttemptRun()


//  (
//    time.sleep_[Task](10.seconds) ++
//    Stream.range(0, 7000).flatMap { i =>
//      innerStream(i)
//    } ++ Stream.eval_(Task.delay(println("XXXY DONE"))) ++
//      time.sleep_[Task](1.hour)
//  ).run.unsafeAttemptRun()

}

After analysing through YourKit, the JVM seems to hold exactly 7000 references to stream in heap, together with their definition.

The commented (sequential) variant is just working fine w/o leaking the memory.

Any ideas of the source ?

[mpilquist]

I think the problem is here: https://github.com/functional-streams-for-scala/fs2/blob/8fa1d5374f0ff50853eb0de4cb09fae47716a6e6/core/shared/src/main/scala/fs2/concurrent.scala#L73

When the outer stream is done, the doneQueue is never drained.

[pchlupacek]

That may be also another issue, however that wont be problem hence the leak accumulates finished inner stream when outer stream is not yet completed. I believe that the fact that you don't drain the Q will cause perhaps resource termination leakage, but not that all inner streams keeps to be strong references at their initial definition. Odesláno z iPhonu 27. 3. 2017 v 13:51, Michael Pilquist <[email protected]>:

…

I think the problem is here: https://github.com/functional-streams-for-scala/fs2/blob/8fa1d5374f0ff50853eb0de4cb09fae47716a6e6/core/shared/src/main/scala/fs2/concurrent.scala#L73 When the outer stream is done, the doneQueue is never drained. — You are receiving this because you authored the thread. Reply to this email directly, view it on GitHub, or mute the thread.

[mpilquist]

My theory is that the queue is full of the cancellation tasks, each of which closes over the inner stream.

[pchlupacek]

ok, that may be the issue. However you still don't want them to be released only after outer stream terminates. You want them to be released as soon as inner is finished?

[mpilquist]

Yeah, I'll put together a fix and then we can compare it to the new implementation.

[mpilquist]

I confirmed this theory and have most of a fix. General idea is to race the done queue with the outer stream in go. There are some edge cases that are nasty to deal with though, so if we're happy with #836, I'm not going to finish this work.

[mpilquist]

Edge cases were easy enough to fix though note that await1Async lies in its type signature -- it shouldn't be defined with Option[A] but rather A. #837.

[pchlupacek]

@mpilquist like one week in production under decent load, I am ok to promote RC1. Any opinions on your side?

[mpilquist]

@pchlupacek Sounds great. I'll backport a few things to 0.9 and release 0.9.5 tomorrow.

[mpilquist]

@pchlupacek 0.9.5 is released