Fs2: concurrently or merge* and toInputStream never terminates

[wedens]

This code never terminates (hangs) in 0.10.0-RC1:

import fs2._
import cats.effect.IO
import scala.concurrent.ExecutionContext.global

val in: Stream[IO, Option[Byte]] = Stream(Some(1 : Byte), None)

val s: Stream[IO, Byte] =
  Stream.eval(async.unboundedQueue[IO, Option[Byte]]).flatMap { q =>
    val write = in.to(q.enqueue)
    val read = q.dequeue.unNoneTerminate
    read concurrently write // or `read merge write`
  }

s.through(io.toInputStream).compile.drain.unsafeRunSync()

It worked fine in 0.10.0-M5.

It successfully terminates if concurrently is replaced with merge* or join. It also successfully terminates with concurrently but without toInputStream (e.g. if toInputStream is replaced by io.file.writeAll).

[wedens]

I'm not sure how it worked with merge* (perhaps I launched it with unsaved file?) but now it also consistently hangs. Still works fine with join.

[wedens]

I've found an interesting comment in prefetchN implementation. Perhaps it's related to the issue (and it doesn't use toInputStream)?

    def prefetchN(n: Int)(implicit ec: ExecutionContext, F: Effect[F]): Stream[F, O] =
      Stream.eval(async.boundedQueue[F, Option[Segment[O, Unit]]](n)).flatMap { queue =>
        // TODO Replace the join-based implementation with this one based on concurrently when it passes tests
        // queue.dequeue.unNoneTerminate
        //   .flatMap(Stream.segment(_))
        //   .concurrently(self.segments.noneTerminate.to(queue.enqueue))
        Stream(queue.dequeue.unNoneTerminate.flatMap(Stream.segment(_)),
               self.segments.noneTerminate.to(queue.enqueue).drain).joinUnbounded
      }

[SystemFw]

I'm seeing all sorts of weird behaviour on this one (weirder than your initial example seems to imply). Trying to figure out what's going on. In any case it appears to be broken with join as well (in addition to concurrently and merge). I think the bug is in toInputStream.

[SystemFw]

Although the doc say:

/**
    * Pipe that converts a stream of bytes to a stream that will emits a single `java.io.InputStream`,
    * that is closed whenever the resulting stream terminates.
    *
    * If the `close` of resulting input stream is invoked manually, then this will await until the
    * original stream completely terminates.
    *
    * Because all `InputStream` methods block (including `close`), the resulting `InputStream`
    * should be consumed on a different thread pool than the one that is backing the `ExecutionContext`.
    *
    * Note that the implementation is not thread safe -- only one thread is allowed at any time
    * to operate on the resulting `java.io.InputStream`.

I need to see if this is relevant or not, given that we aren't directly operating in the InputStream

[SystemFw]

My current hypothesis is that this is just a manifestation of InputStream not being thread safe, I'll try to substantiate why asap

[wedens]

Still hangs in 0.10.0-RC2 :(

[pchlupacek]

@wedens @SystemFw could we confirm that we are not able to minimize this, i.e. remove io. toInputStream from example?

[wedens]

Sorry, I forgot about SI-9076 and tested it in REPL.

It doesn't hang in RC2 with both concurrently and merge.

If @SystemFw haven't found other issues (https://github.com/functional-streams-for-scala/fs2/issues/1071#issuecomment-357986356), I'm ok with closing it.

[mpilquist]

Does adding -Ydelambdafy:inline also fix?

https://github.com/functional-streams-for-scala/fs2/blob/series/0.10/docs/faq.md#why-does-stream-evaluation-sometimes-hang-in-the-repl

[wedens]

No, it doesn't. I think using async.unsafeRunAsync (UPD: or just IO#unsafeRunAsync) did it (in REPL). So, it may not be related to SI-9076 now that I think of it (sorry for confusion).

[SystemFw]

you should also test with join, and not just that it doesnt' hang.
With the first code you posted, join wasn't working either, it won't execute anything (that's why it didn't hang). You'll see it by println through a Sink on enqueue and dequeue

[SystemFw]

I very stupidly lost the test case I had :(

[SystemFw]

emacs saves the day once again!

```scala

val in: Stream[IO, Option[Byte]] = Stream(Some(1: Byte), None)

val s: Stream[IO, Byte] =
Stream.eval(async.unboundedQueue[IO, Option[Byte]]).flatMap { q =>
val write = in.observe(_.evalMap(put)).to(q.enqueue)
val read = q.dequeue.observe(_.evalMap(put)).unNoneTerminate
read.concurrently(write)
}

val ss: Stream[IO, Byte] =
Stream.eval(async.unboundedQueue[IO, Option[Byte]]).flatMap { q =>
val write = in.observe(_.evalMap(put)).to(q.enqueue)
val read = q.dequeue.observe(_.evalMap(put)).unNoneTerminate

  Stream(read,write.drain).join(2)
}

def a = ss.through(io.toInputStream).compile.drain
def b = s.through(io.toInputStream).compile.drain
````

[SystemFw]

They are still both broken, but now they are in a uniform way.
They both will immediately complete without doing anything most of the time, and sometimes print something (a partial result), and I think also deadlock sometimes.
This is the behaviour I've observed with join previously, and now concurrently behaves the same

[pchlupacek]

@wedens @SystemFw not sure If I am following correctly, issue still persist yes/no?

[wedens]

@SystemFw yes, you're right. It doesn't execute anything with any of join/merge/concurrently (I haven't seen even partial results). Manually doing close on InputStream achieves nothing.

@pchlupacek Yes. Slightly differently but it's still broken. I don't think 0.10 final can be released with such issue :/

[wedens]

@SystemFw I think it works with concurrently and merge (but not join) if InputStream is consumed using io.readInputStreamAsync (but not io.readInputStream).

UPD: it does nothing if you don't call read (shifted) on InputStream. which is somewhat obvious, if not weird close behaviour.

UPD2: minimal working (with all 3 combinators) example:

val in: Stream[IO, Option[Byte]] = Stream.emit(Some(1 : Byte)).repeat.take(10) ++ Stream.emit(None)

val s: Stream[IO, Byte] =
  Stream.eval(async.unboundedQueue[IO, Option[Byte]]).flatMap { q =>
    val write = in.to(q.enqueue)
    val read = q.dequeue.unNoneTerminate
    // read concurrently write
    // read merge write.drain
    Stream(read, write.drain).join(2)
  }

s.observe(Sink.showLinesStdOut).through(io.toInputStream).evalMap { is =>
  IO.shift *> IO {
    while (is.read() > -1) {}
    is.close()
  }
}.compile.toVector.unsafeRunAsync(x => println(s"RESULT: $x"))

[SystemFw]

So, the thing is that my broken example above did not do nothing _all_ the time. One in every 50 runs it would print something, so I'd still like to know what's going on (because your working code might then still be broken in subtle ways)

[mpilquist]

@SystemFw @wedens I'm having trouble understanding what's wrong here. Is it that one in 50 runs you see something on std out but most times you don't? Isn't that explained by concurrency in the implementation of observe?

[SystemFw]

the original problem reported does not have observe in it. I can simplify my example to just use .evalMap(F.println) and we'll see what happens.

Anyway yes, most of the time nothing happens, with all combinators. (in the previous version join behaved identically, and concurrently and merged deadlocked)

[mpilquist]

OK, a simpler example that demonstrates an issue would be great. I'd like to release 0.10 tonight if possible but I'd also like to first determine if there's a bug here.

[SystemFw]

  import scala.concurrent.ExecutionContext.Implicits.global

  val in: Stream[IO, Option[Byte]] = Stream(Some(1: Byte), None)

  val s: Stream[IO, Byte] =
    Stream.eval(async.unboundedQueue[IO, Option[Byte]]).flatMap { q =>
      val write = in.evalMap(x => IO { println(x); x }).to(q.enqueue)
      val read = q.dequeue.evalMap(x => IO { println(x); x }).unNoneTerminate
      read.concurrently(write)
    }
   def b = s.through(io.toInputStream).compile.drain.unsafeRunSync
````

scala> b
Some(1)
Some(1)
None

scala> b

scala> b

scala> b

scala> b

scala> b

scala> b

scala> b

scala> b

scala> b

scala> b

scala> b
Some(1)

scala> b

scala> b

scala> b
````

join behaves similarly but it's rarer to see anything printed. It looks like a race condition to me though so take these remarks about probability with a grain of salt

[SystemFw]

Still unsure if this is a bug, or just expected behaviour since InputStream is not thread safe

[mpilquist]

That output looks correct to me and it is unrelated to thread safety of toInputStream. The overall stream emits a single input stream which is then ignored and the end of stream is reached. We then subsequently shut down the stream, which terminates the concurrent “fill” stream which makes data available to the InputStream. The output varies because of timing differences in execution — essentially there’s an expected race between shutdown and filling.

[SystemFw]

Wow, all this time and I hadn't even looked at the fact that the InputStream wasn't actually being used :(

[mpilquist]

Note The example from @wedens does read the InputStream to completion.

[SystemFw]

adding a .flatMap(s => io.readInputStream(s.pure[IO], 10)) gives expected behaviour ("fixes it"), so you're right.

[SystemFw]

But that's good, it means there's no blockers for 0.10 :smile:

[SystemFw]

working code:

  import scala.concurrent.ExecutionContext.Implicits.global

  val in: Stream[IO, Option[Byte]] = 
Stream.range(1, 100).map(_.toByte).covary[IO].noneTerminate

  val s: Stream[IO, Byte] =
    Stream.eval(async.unboundedQueue[IO, Option[Byte]]).flatMap { q =>
      val write = in.evalMap(x => IO { println(x); x }).to(q.enqueue)
      val read = q.dequeue.evalMap(x => IO { println(x); x }).unNoneTerminate
      read.concurrently(write)
    }

  def b = s.through(io.toInputStream).flatMap(s => io.readInputStream(s.pure[IO], 10))
                .compile.drain.unsafeRunSync

[mpilquist]

Okay excellent. I’m going to build tonight I think then. I’ll leave this open pending an update from @wedens. If we’re wrong and there’s some bug lurking here, we can always release 0.10.1.

[wedens]

Ok, let's close it for now.