Fs2: Moving data through queue leaks memory

[avakhrenev]

The following program

Stream.eval(async.boundedQueue[IO, Either[Throwable, Option[Int]]](10)).flatMap { queue =>
  queue.dequeueAvailable.rethrow.unNoneTerminate
    .concurrently(
      Stream.constant(1, 128).covary[IO].noneTerminate.attempt.evalMap(queue.enqueue1(_)))
    .evalMap(_ => IO.unit)
}

leaks memory. Here is what Eclipse Momory Analyzer shows:

fs2 version is 0.10.0-M9

[avakhrenev]

I'm running it from sbt via runMain with this setting enabled:
connectInput in run := true
This allows me to interrupt the program by hitting Enter

Here is a full program for copy-pasting

package com.example

import cats.effect.IO
import fs2._
import fs2.StreamApp.ExitCode
import cats.syntax.apply._

object MemorySanityCheck extends StreamApp[IO] {
  import scala.concurrent.ExecutionContext.Implicits._

  override final def stream(args: List[String], requestShutdown: IO[Unit]): Stream[IO, ExitCode] =
    (io
      .readInputStream[IO](IO(System.in), 32, false)
      .takeWhile(_ != '\n') //read System.in to the end
      .drain
      .onFinalize(IO(println("Exiting...")) *> requestShutdown) mergeHaltBoth app(
      args,
      requestShutdown).evalMap(_ => IO.unit).drain).handleErrorWith(err =>
      Stream.eval(IO {
        err.printStackTrace(System.out)
        ExitCode.Error
      }))

  def app(args: List[String], requestShutdown: IO[Unit]): Stream[IO, Any] =
    Stream.eval(async.boundedQueue[IO, Either[Throwable, Option[Int]]](10)).flatMap { queue =>
      queue.dequeueAvailable.rethrow.unNoneTerminate
        .concurrently(
          Stream.constant(1, 128).covary[IO].noneTerminate.attempt.evalMap(queue.enqueue1(_)))
        .evalMap(_ => IO.unit)
    }
}

[avakhrenev]

Promise, created at this line gets filled with unset callbacks

[mpilquist]

@avakhrenev I'm surprised it's the promise in unconsAsync and not one in Queue

[avakhrenev]

That's what I've seen after instrumenting Promise creation with sourcecode and adding printlns when waiting map reaches 1000 elements. It may be not a root cause.

[mpilquist]

@SystemFw Totally guessing here, but this could occur if the cancellation action in AsyncPull#race wasn't unregistering from the promise's state. Quick spot check appears fine though.

[SystemFw]

AsyncPull#race is definitely the bit of the concurrency PR I'm less confident about. However, I'm still unsure exactly how and where unconsAsync enters the picture in the queue example...

[avakhrenev]

It may be triggered by surrounding code, which reads from input stream simultaneously with running data through queue using mergeHaltBoth combinator.

[SystemFw]

ah, mergeHaltBoth!
I didn't see that, but that's far more likely (and when you drain a stream then merge, I prefer to use concurrently or join) .

However, the minimal snippet does not have that.

[mpilquist]

@avakhrenev I just ran the sample code without the surrounding stream interruption stuff and I haven't been able to reproduce a memory leak. I suspect the real leak is in merge when one of the sides doesn't make progress.

[SystemFw]

a leak on merge makes a lot more sense to me, since that's where AsyncPull.race is involved.
Not sure if it's a problem with race, or just another case of a.drain merge b not being good enough .

[avakhrenev]

Interesting, I've just removed reading from input stream with mergeHaltBoth and got the same result: instances of Promise, allocated in ToPull#unconsAsync continue to grow. Here is the full code for reference:

package com.example

import cats.effect.IO
import fs2._
import fs2.StreamApp.ExitCode

object MemorySanityCheck extends StreamApp[IO] {
  import scala.concurrent.ExecutionContext.Implicits._

  override final def stream(args: List[String], requestShutdown: IO[Unit]): Stream[IO, ExitCode] = {
    Stream.eval(async.boundedQueue[IO, Either[Throwable, Option[Int]]](10)).flatMap { queue =>
      queue.dequeueAvailable.rethrow.unNoneTerminate
        .concurrently(
          Stream.constant(1, 128).covary[IO].noneTerminate.attempt.evalMap(queue.enqueue1(_)))
        .evalMap(_ => IO.unit)
    }.drain ++ Stream(ExitCode.Error)
  }
}

[mpilquist]

@avakhrenev Can you try without StreamApp? Here's what I tried:

object QueueTest extends App {
  import ExecutionContext.Implicits.global
  Stream.eval(async.boundedQueue[IO, Either[Throwable, Option[Int]]](10)).flatMap { queue =>
    queue.dequeueAvailable.rethrow.unNoneTerminate.concurrently(
      Stream.constant(1, 128).covary[IO].noneTerminate.attempt.evalMap(queue.enqueue1(_))
    ).evalMap(_ => IO.unit)
  }.run.unsafeRunSync()
}

[mpilquist]

This program definitely leaks:

object HungMerge extends App {
  import ExecutionContext.Implicits.global
  val hung = Stream.eval(IO.async[Int](_ => ()))
  val progress = Stream.constant(1, 128).covary[IO]
  (hung merge progress).run.unsafeRunSync()
}

[mpilquist]

Note StreamApp uses an interruptWhen on the result of def stream, resulting in a call to merge.

[avakhrenev]

Hm, your program leaks also, or maybe I'm misinterpreting results. I've added println at Promise.scala:96:

case State.Unset(waiting) =>
  val wu = waiting.updated(id, cb)
  State.Unset(wu) -> F.delay[Unit](
    if(wu.size > 100)
      println(s"Waiters size is: ${wu.size}")
    else ())

[SystemFw]

Wait, does QueueTest leak or not?

In any case, HungMerge does, so would you say this is a bug on AsyncPull.race (which I should probably start working on)? No problem with non-hung merge, I'm assuming

[mpilquist]

@avakhrenev Yeah, I see a slow leak in QueueTest, though it is much, much slower than the leak in merge -- about ~200MB over 5 minutes on my machine.

@SystemFw I'll test non-hung merge now.

[mpilquist]

Merging 2 progressing streams does not leak:

object ProgressMerge extends App {
  import ExecutionContext.Implicits.global
  val progress = Stream.constant(1, 128).covary[IO]
  (progress merge progress).run.unsafeRunSync()
}

[SystemFw]

@mpilquist Probably found the cause, cancellableGet in Promise. I'll start working on a fix/precise diagnosis and we can then see if the leaks are also fixed.

[mpilquist]

Fixed by #1027

[SystemFw]

@avakhrenev Thanks a lot for spotting this, turns out it was a pretty serious problem.
Matter of fact, I'm surprised it "only" resulted in a memory leak and didn't break any tests.
Michael fixed it, in any case :)

[avakhrenev]

@mpilquist @SystemFw Thank you for your great work!