Fs2: Tagless implementation of Stream/Pull

[mpilquist]

I'm experimenting with a tagless final implementation of Stream/Pull on the topic/tagless branch. The key interpreter is Fold.

I have no idea if this will be fruitful but it's interesting nonetheless. I want to see how far we can push this without defining our own notion of bracketing/scopes. This design builds directly on Bracket[F].

[pchlupacek]

Excellent. I believe that this should be perfectly doable given last changes to interpreter. The only challenge that I am remaining unsure about is how to interpret translate

[SystemFw]

I'm officially away from my computer for the next four days :)

I had a quick look and I'm looking forward to exploring it further. We
should also introduce lighter Newtype encodings to avoid some boxing.
Anyway, excited to start working on this and see where it leads us :)

On Fri, 18 May 2018, 14:51 Michael Pilquist, [e-mail]
wrote:

I'm experimenting with a tagless final implementation of Stream/Pull on
the topic/tagless branch. I have no idea if this will be fruitful but it's
interesting nonetheless.

The key interpreter is Fold
https://github.com/functional-streams-for-scala/fs2/blob/topic/tagless/src/main/scala/fs2/Fold.scala
.

I want to see how far we can push this without defining our own notion of
bracketing/scopes. This design builds directly on Bracket[F].

—
You are receiving this because you are subscribed to this thread.
Reply to this email directly, view it on GitHub
https://github.com/functional-streams-for-scala/fs2/issues/1142, or mute
the thread
https://github.com/notifications/unsubscribe-auth/AHaN7ohgM65JQ70k5Qx47bkyNAHgwBoyks5tztHmgaJpZM4UExc0
.

[mpilquist]

I added an untested version of translate though I don't see anything obviously wrong with it... :)

Fold accomplishes a remarkable amount in very few lines of code, thanks to pushing all the heavy lifting down to the effect type. Not really sure how to implement zip or any of the combinators based on stepLeg from 0.10. Let me know if you have some ideas.

[mpilquist]

Oh btw, the bracket implementation here is different that ours -- namely, resource lifetime is not extended as a result of subsequent flatmaps.

[SystemFw]

Oh btw, the bracket implementation here is different that ours -- namely, resource lifetime is not extended as a result of subsequent flatmaps.

Just catching up, is this still the case?

[mpilquist]

Yeah, not because I want it that way but rather because I don't know how to fix it. :)

[SystemFw]

Ah, makes sense :)

[pchlupacek]

@mpilquist seems ok to me. Did you tried tests for translate? I think we have to focus at tests with stack-unsafe F.

[mpilquist]

@pchlupacek Nothing serious, just some REPL tests. So I follow the concern, we would need to test that given f: Fold[F, O, R], u: F ~> G, t: Fold[G, O, R] => Fold[F, O2, R2], t(f.translate(u)).fold(...) is stack safe for a non-stack-safe F? Note that we are guaranteed G is stack safe b/c we require a Sync[G] to fold.

[pchlupacek]

@mpilquist yeah you may be right given the fold constrain. Didn't realize that hence we did not have it on translate before, but now transitively we do. So likely you are correct, we shall be safe here.

[SystemFw]

Semi random thoughts:

• We can recover the behaviour of extending the lifetime of bracketed resources in simple cases by having F be Codensity[F] (like ManagedT), however in that case I _think_ we won't be able to modify it concurrently (i.e. in children streams forking) since the finaliser will be stored in a continuation.
• I think we will end up still needing a notion of Scope, or in any a case a Ref[DataStructure[Finalisers]]. F.bracket will probably still be needed though, to modify the Ref/ ensure the finalisers are run at the end of a Scope lifetime.
• I think this problem is essentially equivalent (or very close) to implementing monadic regions (minus the static safety some solutions try to support). Having a look at Haskell solutions for that informed my opinion that a concurrent data structure with finalisers will be needed, and that using F.bracket exclusively is not enough, and can only result in the behaviour above.

I haven't written enough code to validate any of the points above, so take them with a grain of salt.

[SystemFw]

Forgot to say: even if we end up needing Scopes again, it doesn't not mean they won't be simpler/better in this encoding :)

[pchlupacek]

@SystemFw I will be very curious to see scope-less implementation. I cannot prove it, but the fact having scope allowed us to do proper interruption. If we sacrify interruption, I am sure we may perhaps live with sort of Resources only, but then still for concurrent streams and leasing the resources in between them, that may be tricky to implement.

[mpilquist]

Ignoring zipping & merging for a moment, I think our definition of bracket works fine for the fold interpreter and is really only problematic for the step interpreter. The fold interpreter respects the stack structure of the composed Fold whereas step tunnels a value out. Not sure if this insight helps us at all... maybe stepping needs to return a scope?

This is relevant to Stream#flatMap b/c it's implemented via step.

BTW, feel free to commit to this branch with experiments.

[aloiscochard]

@SystemFw I agree with all of your points but one: concurrency.

I don't think it is strictly necessary, would you have a concrete example?

The notion of scope could be introduced by defining a monad transformer such as ResourceT:

• https://hackage.haskell.org/package/resourcet

Delta all issues about being in a strict evaluation context and limited type inference compared to the original implementation.

[SystemFw]

Hi @aloiscochard ! (nice to see you back in Scala)

So we actually do have a couple of ResourceT-like implementations in the cats ecosystem (ManagedT, which is basically codensity, and Resource).
The reason I mention concurrency is that the implementation we have now, Scope (which btw works really well, the reason we want to change it is that it doesn't exploit the fact that IO has bracket since it predates it, so we guess it can be simplified) has had most problems when used in conjunction with join, the concurrent stream combinator.

I've also looked at https://github.com/Gabriel439/Haskell-Pipes-Safe-Library/blob/master/src/Pipes/Safe.hs , which is concurrency aware (it uses atomicModifyIORef, which is kinda like our Ref.modify), but still not in enough depth to fully understand it. I'm aware that ResourceT originated in conduit, but I'm still not sure how it would all fit here, so if you have any input going forward that would be appreciated :)

I think another thing to keep in mind is that in Haskell this sort of abstraction is naturally modelled as a monad transformer, which isn't as nice in Scala.

[SystemFw]

mm, sounds like I assumed ResourceT was closer to Codensity when in fact it is not, it's closer to Scope and pipes-safe. I'll have a closer look.

[aloiscochard]

Hey @SystemFw, (thanks, it is always a pleasure to discuss Scala design with you all!)

The reason I mention ResourceT individually (not in the context of conduit), it is because I use it in Haskell independently with the machine library.

The nice property about it, is that it decouple scoping/region from the streaming library (actually there is nothing specific about stream into it). It's quite different though, for example machine have no concept of resource handling or concurrency.

I then use resourcet for resource management and scoping, while I have some combinators to use them with the async library and get concurrent processing.

The reason I don't use pipes or conduit is precisely because I don't like how they mix those concerns.

Not sure how all of that could translate in the cats ecosystem, I'm not so clear about how your IO type is defined, for example in the scalaz version they could have asynchronous exception (I highly recommend learning about this concept, it is key for some concurrency construct in haskell) if wanted.

What is the example you have in mind with join of streams? I think this would help me to motivate the trade-off I do appreciate.

A high level description of the pipeline you have in mind would be enough :-)

Thanks

[SystemFw]

The nice property about it, is that it decouple scoping/region from the streaming library (actually there is nothing specific about stream into it)

Yep. Note that Scope also has not much specific about Stream, but Stream is definitely aware of Scope.

It's quite different though, for example machine have no concept of resource handling or concurrency.

The thing is that those have a central place in fs2. fs2 takes some design aspects from pipes and conduit, some others are closer to FRP-like programming (albeit discrete), and some others for things that in Haskell would be done with async + STM on top of other things.

In particular, when using Stream for control flow (which is my favourite use case), stream level concurrency is super useful, cause it lets you use Streams as green threads on steroids (on steroids because Stream combinators lets you operate on the "thread" itself). For example if you do take(5) on a Stream you can express execute the first 5 instructions of this "thread". So Stream concurrency is the equivalent of having several of these threads running at the same time.

Not sure how all of that could translate in the cats ecosystem, I'm not so clear about how your IO type is defined, for example in the scalaz version they could have asynchronous exception (I highly recommend learning about this concept, it is key for some concurrency construct in haskell) if wanted.

Heh, I was actually fluent in Haskell before ever touching scala :P Anyway the cats IO model is a bit different, but there is cancellation, which poses similar concerns to async exceptions in terms of things like the acquire action of a bracket being masked.

What is the example you have in mind with join of streams? I think this would help me to motivate the trade-off I do appreciate.

So, there's millions because to me it's like saying "what's the example of multiple threads you have in mind", but I'll give a couple :)

First, imagine a function that takes a Socket, reads from it and writes the content to disk in a streaming fashion. This can be done with a Socket[IO] => Stream[IO, Unit]. Now, the server can be represented as a Stream[IO, Socket], which can be thought of as emitting a single Socket on every incoming connection, you can then map the function above on this stream of sockets to get Stream[IO, Stream[IO, Unit]]. Now, the inner streams obviously all have to run concurrently, you don't want to stop incoming connections while processing the old ones, so you just join the stream of streams and represent a concurrent server. This model is also quite nice because you can do things like flatMap on the _outer_ stream, for example to log the fact that an incoming connection has just happened.

A fairly different example: imagine you want to throttle a Stream. One strategy would be to keep the frequency between elements constant, and you can do it like so: the first thing would be a function that takes the Stream and enqueues each element on a bounded Queue (we have pure concurrent queues in fs2). You can then create a separate Stream that periodically (according to the rate) reads from the queue.
A throttle combinator can then just create the queue, create these two streams, and then join them. It has to join them because these two actions obviously need to happen concurrently.

Generally speaking, I think the key idea is that fs2 is not just used for streaming IO (like read from this file, transform each element and write somewhere else), but also (and in my case actually more often) for control flow, a la FRP. And in the context of control flow I think the need for concurrency and resource safety is more evident. You _could_ do the same thing in IO only, but then your logic often ends up being very imperative-looking, albeit still pure (whileM, untilM, recursion), whereas using stream gives you higher level constructs, similarly to how map is nicer than while.

P.S.

Nice historical fact, the very first inspiration for fs2, years before my time, was actually machines, rather than pipes/conduit.

[SystemFw]

Ah, I guess I should mention: join in fs2 is not monadic join, that's called flatten in cats and it's completely sequential, just like flatMap is. fs2.join has the same type of flatten but runs all inner streams concurrently rather than sequentially, and has nothing to do with the Monad instance.

[mpilquist]

@diesalbla You might be interested in the branch linked to from this issue if you're looking at FreeC and Algebra.

[mpilquist]

I created a new branch for the tagless encoding -- topic/tagless2: https://github.com/functional-streams-for-scala/fs2/blob/topic/tagless2/core/shared/src/main/scala/fs2/Tagless.scala

[SystemFw]

ah nice, was thinking about this recently andI thought that one other crucial case, similar to bracket, is something like:
streamA.interruptAfter(2.seconds) ++ streamB

So basically interruption with resumption, that isn't easily representable by the F notion of interruption, and that currently is spread between core interpreter, FreeC and Scope

[mpilquist]

I just pushed a simpler encoding that has a working Stream.bracket. Take a look when you get a chance.

[SystemFw]

wow, nice :)

[mpilquist]

I made a bunch of progress, including scopes and most of the implementation of uncons. I ran in to an issue implementing translate for uncons and I'm not sure how to fix. Would appreciate a second pair of eyes or two.

[SystemFw]

I started having a look, albeit very cursory. I have an initial very stupid question, what's the reason for named classes for each op? Merely code organisation or there's something else?

[mpilquist]

Not tied to it -- did that to avoid type parameter name confusion. E.g., defining them inline means making weird choices for names like F2, F3, etc.

[mpilquist]

Update: I cleaned up the prototype a bit and moved it to the topic/fs3 branch. At this point, the bulk of Stream and Pull are working with the exceptions of interruption and leasing. Hence, concurrent operations like merge / concurrently / parJoin fail.

I'm currently investigating how best to implement interruption. Porting the old version might not be the best path forward, as it relied on FreeC#transformWith and the associated Result.Interrupted constructor.

[pchlupacek]

@mpilquist one of the ideas for nextgen interruption with tagless I had in mind was to capture next state at each eval node. And actually when interruption occur continue from that node. It is rough idea and not sure if its clearly explainable, but essentially instead of racing the eval today, I had in mind that before eval you will store the next node (r => F[....]) and in case of interruption, that will get strait evaluated. Currently running eval will be then just interrupted and discarded.
its sort of inverse what we do today, My guess will be also that it will have positive impact on performance

[mpilquist]

@pchlupacek I was able to port the old interruption logic to the topic/fs3 branch. I'm still working on porting the interruption tests but the first 10 are all passing. Working on the remaining ~10 now.

[pchlupacek]

@mpilquist thats excellent.

[mpilquist]

Spoke too soon -- first 11 tests pass but next 3 have failed.

Streams of the shape s.interruptWhen(...).flatMap(_ => Stream.eval_(hang)) are not working -- for some reason, the eval occurs in the parent scope -- not the interruption scope introduced by interruptWhen.

I also need to figure out how to make "resume on append" work for interruption, as there's definitely no support for that now.

[pchlupacek]

@mpilquist yeah, don't be frustrated. It was always like this with interruption implementation. Always when you think you have it, there somethings pops out and sometimes this implies full rework it :-(

[mpilquist]

Made some progress -- 19 out of the 24 interruption tests are passing now. Everything is pushed.

[nigredo-tori]

I've played a bit with a slightly different Pull encoding in https://github.com/nigredo-tori/fs2/blob/fs3-split-monads/core/shared/src/main/scala/fs2/Pull.scala. I've changed step to this:

protected def step[F2[x] >: F[x], H[_]: Sync, O2 >: O, R2 >: R](
    ctx: PullContext[F2, H]
): H[StepResult[F2, H, O2, R2]]

where H is the target monad, and PullContext[F, H] is basically a Scope[H] with F ~> H (these are grouped together temporarily to simplify experimentation).

This means that:

1. translate doesn't need to be implemented for each Pull separately - we just change the natural transformation in the PullContext, and translate any continuations.
2. uncons works, translated or not.
3. This is closer to the tagless final pattern (if you squint hard enough).

PullContext is more complicated than it could be; this is to preserve the Scope/Concurrent logic that is present in the fs3 branch. The main reason why this is complicated is because Scope handling in the fs3 branch is broken - there are invalid asInstanceOf calls, weird Concurrent captures that break after translate, etc. Fixing all this is not in scope (ha!) of this exercise. However, the new encoding does highlight why things like Pull.getScope can't be sound with the current way Scope is implemented: we only ever have Scope[H] for some H, and no way to build Scope[F]. The closest we can get is probably something like Scope[Pull[F, Nothing, ?], where any operations will get compiled to H eventually.