Fs2: Deprecate StepLeg?

[mpilquist]

I was looking in to #1624 and was considering improving the error message thrown here: https://github.com/functional-streams-for-scala/fs2/blob/829279df39c6cb779e512061ed6b4cdc8f622bb1/core/shared/src/main/scala/fs2/internal/Algebra.scala#L233

In doing so, I tried to reproduce the original error that led to the creation of StepLeg but I couldn't get things to fail. I wrote this test class:

package fs2

class Proof extends Fs2Spec {
  "proof" - {
    "zip proof" in {
      def brokenZip[F[_], A, B](s1: Stream[F, A], s2: Stream[F, B]): Stream[F, (A, B)] = {
        def go(s1: Stream[F, A], s2: Stream[F, B]): Pull[F, (A, B), Unit] =
          s1.pull.uncons1.flatMap {
            case Some((hd1, tl1)) =>
              s2.pull.uncons1.flatMap {
                case Some((hd2, tl2)) =>
                  Pull.output1((hd1, hd2)) >> go(tl1, tl2)
                case None => Pull.done
              }
            case None => Pull.done
          }
        go(s1, s2).stream
      }
      val s = Stream(0).scope
      brokenZip(s ++ s, s).toList shouldBe List((0, 0))
    }
  }
}

The brokenZip method implements zip by calling uncons1 on each source stream. The assertion comes from our test suite, where we'd run in to the scope id issue due to interleaving of streams.

This test passes on the current head and fails on v0.10.0. I ran a git bisect and determined that this test has been passing since a9f8a1d437603b6a08f29b019e9891521e886727.

On HEAD, I reimplemented zipWith_ using uncons instead of stepLeg and all tests pass.

So, should we deprecate StepLeg? Or is there a better example program that demonstrates the need for it?

/cc @AdamChlupacek @pchlupacek

[mpilquist]

I haven't succeeded in reproducing an exception with that original test case, but I have succeeded in showing why we need StepLeg. Here's two definitions of deterministically merging sorted streams, one based on uncons1 and the other on stepLeg:

def sortedMerge[F[_], O: Order](s1: Stream[F, O], s2: Stream[F, O]): Stream[F, O] = {
      def go(h1h: O, s1: Stream[F, O], h2h: O, s2: Stream[F, O]): Pull[F, O, Unit] = {
        if (Order[O].compare(h1h, h2h) <= 0) {
          Pull.output1(h1h) >> s1.pull.uncons1.flatMap {
            case Some((h1h, s1)) => go(h1h, s1, h2h, s2)
            case None => Pull.output1(h2h) >> s2.pull.echo
          }
        } else {
          Pull.output1(h2h) >> s2.pull.uncons1.flatMap {
            case Some((h2h, s2)) => go(h1h, s1, h2h, s2)
            case None => Pull.output1(h1h) >> s1.pull.echo
          }
        }
      }
      s1.pull.uncons1.flatMap {
        case Some((h1h, s1)) => s2.pull.uncons1.flatMap {
          case Some((h2h, s2)) => go(h1h, s1, h2h, s2)
          case None => s1.pull.echo
        }
        case None => s2.pull.echo
      }.stream
    }

def sortedMergeLeg[F[_], O: Order](s1: Stream[F, O], s2: Stream[F, O]): Stream[F, O] = {
      def go(s1: Stream.StepLeg[F, O], s2: Stream.StepLeg[F, O]): Pull[F, O, Unit] = {
        // These chunks are guaranteed to have size == 1 due to the `flatMap(emit)` normalization
        val h1h = s1.head(0)
        val h2h = s2.head(0)
        if (Order[O].compare(h1h, h2h) <= 0) {
          Pull.output1(h1h) >> s1.stepLeg.flatMap {
            case Some(s1) => go(s1, s2)
            case None => Pull.output(s2.head) >> s2.stream.pull.echo
          }
        } else {
          Pull.output1(h2h) >> s2.stepLeg.flatMap {
            case Some(s2) => go(s1, s2)
            case None => Pull.output(s1.head) >> s1.stream.pull.echo
          }
        }
      }
      s1.flatMap(Stream.emit).pull.stepLeg.flatMap {
        case Some(sl1) => s2.flatMap(Stream.emit).pull.stepLeg.flatMap {
          case Some(sl2) => go(sl1, sl2)
          case None => sl1.stream.pull.echo
        }
        case None => s2.pull.echo
      }.stream
    }

Now consider the following:

@ def out(msg: String) = IO(println(msg))
defined function out

@ def res[A](value: A): Stream[IO, A] = Stream.bracket(out(s"acquired $value").as(value))(_ => out(s"released $value"))
defined function res

@ def foo(tag: String) = res(tag).flatMap { _ => res(1) ++ res(2) }
defined function foo

@ sortedMergeLeg(foo("left"), foo("right")).compile.toList.unsafeRunSync
acquired left
acquired 1
acquired right
acquired 1
released 1
acquired 2
released 1
acquired 2
released 2
released left
released 2
released right
res58: List[Int] = List(1, 1, 2, 2)

@ sortedMerge(foo("left"), foo("right")).compile.toList.unsafeRunSync
acquired left
acquired 1
acquired right
acquired 1
released 1
released right
released 1
acquired 2
acquired 2
released 2
released 2
released left
res59: List[Int] = List(1, 1, 2, 2)

Okay, wow, the scopes are all sorts of messed up when using the non-stepLeg version. This is why stepLeg exists.

[mpilquist]

Here's a new reproducer:

val s = Stream(0).scope

(s ++ s).zip(s.zip(s)).compile.toList // Works fine

brokenZip(s ++ s, s.zip(s)).compile.toList
//java.lang.Throwable: Fail to find scope for next step: current: Token(2b07d8da), step: Step(CloseScope(Token(2ad48485),None,Completed),Some(Token(2ad48485)))

brokenZip(s ++ s, brokenZip(s, s)).compile.toList // Works but scopes are all messed up

Using:

def brokenZip[F[_], A, B](s1: Stream[F, A], s2: Stream[F, B]): Stream[F, (A, B)] = {
    def go(s1: Stream[F, A], s2: Stream[F, B]): Pull[F, (A, B), Unit] =
      s1.pull.uncons1.flatMap {
        case Some((hd1, tl1)) =>
          s2.pull.uncons1.flatMap {
            case Some((hd2, tl2)) =>
              Pull.output1((hd1, hd2)) >> go(tl1, tl2)
            case None => Pull.done
          }
        case None => Pull.done
      }
    go(s1, s2).stream
  }

[johnynek]

This was helpful to me, if anyone else comes here looking for how to merge several, I made an implementation that uses a heap interally:

https://gist.github.com/johnynek/689199b4ac49364e7c94abef996ae59f