Refactor ZStream to use ZManaged

[iravid]

This is passing tests, so opening this up for a review.

This pull request contains a refactor to ZStream that utilizes ZManaged as the final representation. Previously, folding a stream using a function (S, A) => ZIO[R, E, S] would result in a value of ZIO[R, E, S]. The problem with this encoding is that composing streams required not only composing their computations, but also requires constructing a tree of finalizers.

On this branch, folding now returns a ZManaged[R, E, S]. Constructors that acquire resources (such as ZStream.bracket) will acquire the resources, uninterruptibly, in the reservation step of ZManaged. The folding of the stream happens, interruptibly, in the acquisition step of ZManaged. And naturally, resource release happens in the release step.

By utilizing ZManaged, we can now create a tree of finalization scopes when composing streams. For example, in the zipWith combinator, the finalizers for the streams feeding the composite stream will run strictly after the finalizer for the composite stream has run.

Over the next few weeks I'll add more tests that exercise finalization scoping and short circuiting. Also, I'll add the concurrency combinators that triggered this refactor.

All comments welcome!

[iravid]

The concurrency combinators now behave correctly with regard to finalization 🎉

[jdegoes]

Nice! 👍

[jdegoes]

Love all the new tests! ❤️

[jdegoes]

We don't need this around, do we?

[iravid]

Nope! Will remove it.

[jdegoes]

Everything is fitting together nicely! I can also see how we can do things like:

acquire ++ use ++ release

in an easy & safe fashion now (something I've needed to do for a while).

Appreciate the work towards alphabetizing the methods (it's almost there).

Let me know if you want me to take a close look at anything in particular, otherwise I'd say this is a clear win and completes the story of resource management for left-fold-style streams! 🎉

[iravid]

Thanks @jdegoes! Yes, acquire ++ use ++ release "just works" now, as the RHS of ++ gets folded before the finalizer of LHS runs.

We can also add Stream.ensuring, onComplete and so forth. I'll add them in subsequent PRs.

I'm pretty happy with the encoding too, the only thing I'm wondering about is the ergonomics - lots of ZManaged.fromEffect and ZManaged.unwrap are needed everywhere and this smells of transformer-ness. Is there something more foundational we can do here?

[jdegoes]

Thanks @jdegoes! Yes, acquire ++ use ++ release "just works" now, as the RHS of ++ gets folded before the finalizer of LHS runs.

Superb! I suppose that means with infinite ++, the resources will never be released, but then again, infinite concatenation is an anti-pattern with lazy left folds, anyway (and may leak heap, too).

[iravid]

Yup, that's correct

[iravid]

@jdegoes would love to get this in for the next RC so we can use it in zio-kafka. Will resolve the conflicts now

[jdegoes]

@iravid Sounds good, resolve the conflicts, point me to any tricky bits, and we'll get this in for testing in the next RC. 👍

[iravid]

@jdegoes would love another look at this operator for validation. I used interruptibleMask to allow the caller to run the reserve step interruptibly rather than .interruptible

[jdegoes]

This is clever and looks correct to me.

One thing to keep in mind is that this fiber will inherit the interruptibility settings of its parent, at the moment when it is forked.

Since this is done in reserve, I believe we can therefore simplify the code to just use interruptible rather than interruptibleMask, because we know what the parent interruptibility status will be (right?).

[iravid]

So we can assume that 99% of the cases would use an uninterruptible reserve, but, that's not always true, as reserve can also be run manually for advanced manipulations. Which is why I thought interruptibleMask is more compositional.

[jdegoes]

Ah, ok. Works for me. 👍

[iravid]

@jdegoes ZManaged.unwrap { fold.map(...) } repeats itself all over the file which is unfortunate

[jdegoes]

I saw that. Any way to factor out that duplication?

[iravid]

We could have this as a private helper:

  private def foldUnwrap[R1 <: R, E1 >: E, A1 >: A, S] =
    ZManaged.fromEffect(self.fold[R1, E1, A1, S])

And then the combinators look like this:

  final def ++[R1 <: R, E1 >: E, A1 >: A](other: ZStream[R1, E1, A1]): ZStream[R1, E1, A1] =
    new ZStream[R1, E1, A1] {
      def fold[R2 <: R1, E2 >: E1, A2 >: A1, S]: Fold[R2, E2, A2, S] =
        ZIO.succeedLazy { (s, cont, f) =>
          self.foldUnwrap[R2, E2, A2, S].flatMap { foldLeft =>
            foldLeft(s, cont, f).flatMap { s =>
              if (!cont(s)) ZManaged.succeed(s)
              else
                ZManaged.unwrap {
                  other.fold[R2, E2, A2, S].map(foldRight => foldRight(s, cont, f))
                }
            }
          }
        }
    }

I guess that's decent enough!

[iravid]

Another possibility is to use ZManaged directly in the Fold wrapper type. That could work too.

[iravid]

@jdegoes This step is pretty clunky; we need to prepend a step to the finalizer of driver here, which makes me feel like we're missing an operator for that on ZManaged.

[jdegoes]

Yes, I think so. We're missing finalizer, something like:

def finalizer(f: UIO[_]): ZManaged[Any, Nothing, Unit]

Then we can do something like zmanaged.flatMap(a => ZManaged.finalizer(fin).const(a))

That would do it, right?

[iravid]

Yes! Thank you. That's what I was looking for.

[iravid]

@jdegoes all done here. thanks for the review!

[iravid]

Hmm one of the new RTS tests is flaky:

[error]     x executor is hereditble
[error]  None ==> Some(...) (RTSSpec.scala:986)
[error] Actual:   None
[error] Expected: Some(scalaz.zio.internal.Executor$$anon$2@2155e0c)

[jdegoes]

@iravid Let's see if #914 gets the job done.

[iravid]

@jdegoes Looks good this time.

[jdegoes]

Nice cleanups! ❤️

[jdegoes]

@iravid Looks great! 🎉

[iravid]

🎉

[jdegoes]