Fs2: Stream.broadcast behaving unintuitively

[PeterAaser]

Stream(10, 20, 30, 40, 50, 60)
      .covary[IO]
      .broadcast
      .take(2)
      .parEvalMap(2)(_.compile.toList)
      .map(println)
      .compile
      .drain

outputs
[info] List(10, 20, 30, 40, 50, 60)
[info] List(10, 20, 30, 40, 50, 60)
[info] List(10, 20, 30, 40, 50, 60)
[info] List(10, 20, 30, 40, 50, 60)
[info] List()
[info] List()
[info] List()
[info] List()
[info] List()
[info] List()

(whereas .through(Broadcast(10)) provides the expected result)

As per the docs of broadcastTo

    * Each sink is guaranteed to see all `O` pulled from the source stream, unlike `broadcast`,
    * where workers see only the elements after the start of each worker evaluation.

which is clearly what is happening here.
This is rather unintuitive, especially considering what happens if you try to take then flatten

    Stream(1,2,3)
      .covary[IO]
      .broadcast
      .take(2)
      .flatten
      .map(println)
      .compile
      .toList.unsafeRunSync()

which yields 1,2,3 instead of 1,2,3,1,2,3
(and take(2).drop(1) also yields 1,2,3)

I think either documentation or signature could benefit from being changed to avoid this issue

[SystemFw]

The behaviour is correct I think.
I'll write down a longer explanation as to why, and we can then see how to improve the docs.

[troolean]

Keep in mind, that the entire discussion started with the fact that the following app hangs:

import cats.implicits._
import cats.effect._
import fs2._

object Fs2BroadcastingDemo extends IOApp {
  override def run(args: List[String]): IO[ExitCode] = IO(ExitCode.Success) <* {
    Stream(10, 20, 30, 40, 50, 60)
      .covary[IO]
      .broadcast
      .take(2)
      .fold(List.empty[Stream[IO, Int]])(_ :+ _)
      .flatMap { case List(s1, s2) => s1.zip(s2) }
      .map(println)
      .compile
      .drain
  }
}

It seems like we're seeing at least two different things that _may_ need fixing: (1) more clarification about semantics of broadcasting and (2) usage of broadcast + fold/zip hangs an app

[SystemFw]

@PeterAaser Imagine any pub sub scenario (don't even think about fs2 or code): the subscribers are typically dynamic and any number of them can arrive at any time while you have already started the broadcast to existing ones.

Now imagine that you want the guarantee that every subscribers, present and future, will always receive all the messages.
Imagine you start with subs A and B. You receive msg 1 from the source, send it to A and B, what do you do with it after? You can't throw it away, you need to store it somewhere because if an hypotethical sub C arrives, you need to send 1 to it or you'll lose the guarantee.
Now msg 2 arrives, you send it to A and B, can you throw 2 away? No, for the same reason. Can you throw 1 away now? Still no, or C will see only 2 and not both 1 and 2, which is the guarantee.
When 3 arrives, now you are storing 1,2 and 3, in case C arrives after 3.

If you continue with this pattern, you will see that it forces you to accumulate _all_ the messages you ever request from the source, even when the existing subscribers (A and B) are done processing them, i.e. you are not _really_ streaming anymore, because at some point you will have the whole source stored somewhere.

This is the same reason why things like kafka have the concept of "retention": how many messages to store before throwing older things away. An absolute guarantee that, no matter what, all possible subs will see every message basically means infinite retention (you need to save the whole history of messages). I'm not sure how useful a retention mechanism is in fs2 (not very, I think), but perhaps it can be implemented with PubSub.

However, there is another thing you could do. You can say "I'm not going to request any messages from the source until I have n subscribers ready". This ensures that the first n subscribers will see every message. Every future subscriber will only see messages that arrive after they subscribe.
And this leads us to the fundamental signature of Broadcast, which is implemented with PubSub; every other broadcast combinator is derived from this:

object Broadcast { 
  def apply[F[_]: Concurrent, O](minReady: Int): Pipe[F, O, Stream[F, O]]

or in other words

object Broadcast { 
  def apply[F[_]: Concurrent, O](minReady: Int)(source: Stream[F, O]):  Stream[F, Stream[F, O]]

The _outer_ stream is the infinite stream of all the potential subscriptions, pulling on that stream gives you a "subscription", represented as the inner Stream[F, O], which only sees messages received once it starts pulling, which is unavoidable as shown above. minReady is exactly the "other thing you could do" from the previous paragraph.

---

Now, Stream.broadcast is Broadcast(1), which obviously exposes the race you have in the very first example (even more evident with parEvalMapUnordered), where you'll get

scala> s
List(10, 20, 30, 40, 50, 60)
List(10, 20, 30, 40, 50, 60)

scala> s
List(10, 20, 30, 40, 50, 60)
List()

scala> s
List()
List(10, 20, 30, 40, 50, 60)

You say that .through(Broadcast(10)) works, but it doesn't, it hangs forever because you will never reach 10 subs, given that you take(2) on the outer stream. .through(Broadcast(2)) on the other hand work deterministically, which makes sense.

Same with

Stream(1,2,3)
      .covary[IO]
      .broadcast
      .take(2)
      .flatten // compile etc

since you use flatten instead of parJoin, you are saying that you want the first subscriber to exhaust the source before the second can even subscribe, so a result of List(1, 2, 3) also seems normal to me.

---

However, we do have different Broadcast combinators that _are_ guaranteed to see every emitted element, how is that possible? Well, let's look at the signature of Broadcast.through:

def through[F[_]: Concurrent, O, O2](pipes: Pipe[F, O, O2]*): Pipe[F, O, O2]

let's rename

object Broadcast {
 def through[F[_]: Concurrent, O, O2](subscribers: Pipe[F, O, O2]*)(source: Stream[F, O]): Stream[F, O2]

The signature has changed: we are returning a single Stream (the stream of results), and a not a Stream of Streams, we no longer have the concept of dynamic subscribers (a new subscribers every time the outer stream is pulled): we now need to provide _all_ the subscribers in advance, in the form of Pipes, and in fact the implementation is:

_.through(apply(minReady = pipes.size))
      .take(pipes.size)
      .zipWith(Stream.emits(pipes)) { case (src, pipe) => src.through(pipe) }
      .parJoinUnbounded

Note the minReady = pipes.size and parJoinUnbounded, which address your problems above.
broadcastThrough, broadcastTo etc. in Stream all follow this pattern, so

I think either documentation or signature could benefit from being changed to avoid this issue

docs can ofc be improved, especially the ones in Stream (the ones in Broadcast already explain this I think, at least to an extent), but the signatures already mark a difference: one returns a stream of streams, the others take some pipes and return another pipe. We can definitely think of ways of making this more evident through more docs though, as I said.

---

The zip case seems more complex, still looking at it.

[troolean]

@SystemFw, I have a follow-up question. What is the expected behavior of the snippet below?

Stream(1, 2, 3)
  .through(Broadcast(3))
  .take(3)
  .compile.toList
  .map { case List(s1, s2, s3) =>
    //do stuff with each stream here?
    //looks like each of the streams (s1, s2, s3) will be empty at this point
  }

Also, thank you for the thorough explanation about the behavior of broadcast in fs2.

[SystemFw]

Well, why do you want to compile.toList? In general, compile changes a lot of things (going from Stream to IO does), for example all the resources opened by the stream are closed.
Do you mean .foldMap(_.toList)?

[troolean]

i'm trying to draw a parallel to a similar example. it's confusing that some of the broadcast behavior is seemingly inconsistent with other similar behavior:

val nums = Stream(1, 2, 3).covary[IO]
val nested = Stream(nums, nums, nums)
  .covary[IO]
  .compile.toList
  .map { case List(s1, s2, s3) => Stream(s1, s2, s3).parJoinUnbounded }
Stream.eval(nested).flatten.compile.toList.map(println).unsafeRunSync()

[SystemFw]

No, those two things aren't similar at all (in general, I haven't looked at this specific example yet)
Stream operations don't "commute" over compile. There are fundamental differences in what Stream and IO can do: IO cannot have resource lifetime extend over flatMap, for example. These differences boil down to the fact that a stream can emit n values, and therefore when you compile to IO, you lose the extra powers (and that means, for example, that resources are closed).

Try coming up with an example without compile first. If you want to get all the streams in a list, do foldMap(x => List(x)).flatMap { case List(s1, s2, s3) =>

[troolean]

in the fuller example below, xs is not interchangeable with ys. even though it _seems_ like they should be since both have the same intent behind them:

val nums = Stream(1, 2, 3).covary[IO]
val xs: Stream[IO, Stream[IO, Int]] = nums.through(Broadcast(3)).take(3)
val ys: Stream[IO, Stream[IO, Int]] = Stream(nums, nums, nums).covary[IO]

val io = ys
  .compile.toList
  .map { case List(s1, s2, s3) => Stream(s1, s2, s3).parJoinUnbounded }
Stream.eval(io).flatten.compile.toList.map(println).unsafeRunSync()

[SystemFw]

Seems like you have deleted/modified your comment. Look at my answer above :)

[troolean]

Sorry, I did edit the comment. Then I updated it again to post the edited part as its own comment [[1]]. I think my follow up example is still pertinent. Before I proceed with attempting to come up with an example using foldMap/flatMap, I'd like to understand what's happening after the call to compile for both cases in [[1]]. And more generally, what exactly are the rules for handling a Stream of Streams.

[SystemFw]

Sorry, unfortunately I've exhausted my free time for today :)
There's a good chance I won't be able to help until Friday

[SystemFw]

Ok @troolean , let's unpack a few things :)

First, your example is based on an initial assumption that is incorrect regardless of compile.
The assumption is that:

// given
val nums = Stream.range(0, 100).covary[IO]
// then
val xs: Stream[IO, Stream[IO, Int]] = nums.through(Broadcast(3)).take(3)
// should be equivalent to
val ys: Stream[IO, Stream[IO, Int]] = Stream(nums, nums, nums).covary[IO]
// when run concurrently, for example in this simple case
val runner: Sink[IO, Stream[IO, Int]] = _.parJoinUnbounded.to(Sink.showLinesStdOut)

but they are actually quite different.

xs is saying "run nums once, and broadcast its elements to all the subscribers", whereas ys is saying "run nums 3 times, concurrently". They end up emitting the same elements, but you realise there's a big difference as soon as you introduce effects. For example if you change nums to

val nums = Stream.eval_(IO(println("starting"))) ++ Stream.range(0,100)

then xs.to(runner) will print "starting" once, whereas ys.to(runner) prints "starting" three times, which is a huge difference if you consider arbitrary effects. Afterall, why would we need broadcast if those two snippets were equivalent?

I'm going to explain the compile bit when I have more time :)

[SystemFw]

So, the massive difference between running effect once vs multiple times should already answer the question about the two snippets being non equivalent, but I want to explain the compile bit.

As I said above:

Stream operations don't "commute" over compile. There are fundamental differences in what Stream and IO can do: IO cannot have resource lifetime extend over flatMap, for example. These differences boil down to the fact that a stream can emit n values, and therefore when you compile to IO, you lose the extra powers (and that means, for example, that resources are closed).

The closing resources example is relevant here: Stream tracks the lifetime of resources, which are closed once the stream is compiled. You can see it in this example:

 def aa =
    Stream.emit(1)
      .onFinalize(IO(println("done")))
      .evalMap(x => IO(println(x)))
      .compile
      .lastOrError
      .unsafeRunSync

  def bb =
    Stream.emit(1)
      .onFinalize(IO(println("done")))
      .compile
      .lastOrError
      .flatMap(x => IO(println(x)))
      .unsafeRunSync

These two snippets _seem_ the same, but the first prints:

done

whereas the second prints

1

In this case it doesn't really matter because the println("1") doesn't depend on whether "done" has run or not, and this the ys case in your code.

Stream(nums, nums, nums).compile.toList

emits the list with 3 nums, and it would run finalisers on the outer stream if there were any, which there aren't since you are just doing Stream.apply.

Similarly

nums.through(Broadcast(3)).take(3).compile.toList

also emits the list with 3 nums, but in this case there are finalisers: Broadcast unsubscribes listeners when its lifetime is over, so by the time you then get a hold of the inner streams, their subscription is expired and they don't receive any elements.

Note that the assumption that compile won't make a difference is incorrect regardless of Broadcast and streams of streams, like my first aa and bb code snippet shows.

[troolean]

After playing around more with fs2 and Broadcast, I think the semantics of compile finally sank in. The observed behavior is making more sense to me now. Thank you for clarifying!

[troolean]

@SystemFw I do still have one unanswered question regarding broadcasting. Is it currently possible to achieve a broadcast which creates two _distinct_ Streams, basically something that would have the following "shape":

def dupe[F[_], A]: Pipe[F, A, (Stream[F, A], Stream[F, A])]

Note, how the output is a _pair_ thereby making each resulting Stream distinct. In my original example, that's exactly what I was trying to do, e.g.:

def dupe[F[_]: Concurrent, A]: Pipe[F, A, (Stream[F, A], Stream[F, A])] = _
  .broadcast
  .take(2)
  .fold(List.empty[Stream[F, A]])((acc, s) => s +: acc)
  .map { case List(s1, s2) => (s1, s2) }

However, that implementation definitely doesn't work.

val program = for {
  (a, b) <- Stream.emits[IO, Char]("abcdef").through(dupe)
  _ <- Stream(a.showLinesStdOut, b.showLinesStdOut).parJoin(2)
} yield ()

I'd expect that the execution of program above would result in each Char of "abcdef" to be printed twice. In other words, I'd expect the outcome of the program above to be similar to Stream.emits[IO, Char]("abcdef").broadcastTo(2)(_.showLinesStdOut). In reality, nothing at all is printed when the program above is executed.

There are two questions here:

1. What's wrong with the implementation of dupe above?
2. Does this functionality already exist in fs2?

[troolean]

Update

After, re-discovering chunkN (the existence of which I keep forgetting), I was able to get dupe to work:

def dupe[F[_]: Concurrent, A]: Pipe[F, A, (Stream[F, A], Stream[F, A])] = _
  .broadcast
  .take(2)
  .chunkN(2)
  .map(_.toList)
  .map { case List(a, b) => (a, b) }

I suspect that fold is considered to be a "terminal" operation of sorts and has a special meaning/handling behind it.

P.S.
Wouldn't grouped be more appropriate and less confusing than chunkN?