Scale By The Bay 2018: Michael Pilquist, Concurrency with Cats-effect
Recording: Scale By The Bay 2018: Michael Pilquist, Concurrency with Cats-effect
so this is my fourth talk on Michael pilk list if you don't know me I maintain a bunch of open source Scala stuff and I worked at comcast rebuild video systems and we use skaaland do like a bunch of soft real-time type streaming stuff so we was talking about that come find me later this is my shortest time slot I've had at this conference normally I get like an hour this year I have like a half an hour so I thought I'd pick something really easy to talk about so they can fit into 30 minutes so I picked polymorphic defect types and concurrency because I thought you know that would definitely not be a challenge all right so let's talk about a year ago I stood here and we you know I had I had promised at the time that we were gonna release FS to zero and cats effexor dat probably nine at the time and it was a you know it turned out it didn't make it we did we didn't we didn't have those releases out in time for the conference so we had like release candidate builds instead right and and getting the those releases over the you know hurdle if you will took us until January 2018 so in January 2018 we had this type class hierarchy on the left part of the slide there the southern greens from cats core the stuff in blue is from cats effect and it was a pretty simple type class hierarchy maybe maybe too simple right we talked a little bit last year I think John had a talk last year we talked about like fiber-based concurrency and how do we do concurrency with that particular type class hierarchy and the answer was that like a downstream libraries really really had to re implement their own notions of concurrency using those particular type classes so fast forward about nine months nine months of you know debate and discussion and lots of perspectives from a lot of different parts of the overall Scala open source ecosystem and we ended up here and you know I want to I want to point out this is like 1.0 right so there's some long-term stability there's more boxes so clearly it's better right than zero ten or zero to eight but like in no way is this like necessarily where it ends right like we're committed to long-term stability on this particular type cost hierarchy but you know you'll probably see a cat's effect too that looks a lot different than this right as we continue to understand what it means to abstract over effect types alright so what are we talking about really today what I want to look at is two things one is you know principled and polymorphic resource management all right I want to be able to write code as a library author or even perhaps as an application developer in one of my systems that doesn't bind to like a specific effect type or a specific effect type plus you know some transformer stack instead I want to do that polymorphically in my effect but still have like these very crisp abstractions that I can depend on I can rely upon and I have guaranteed behaviors so anyway we have this new bracket type class it's near the top of the hierarchy right now you know it extends directly from ona tear and it gives us this you know crisp definition of resource management we'll see a bunch of this today and we'll look at all these in quite a bit of depth the other thing I want to talk about is the concurrent type class there's kind of the same idea we want to be able to do concurrency we want to be to be able to do like sort of these lightweight threads you know kind of be able to start things and and reason about them and cancel them and again do that in a way that's portable that's crisp that's that's very you know dependable regardless of like what particular you know effect type or or transformer stack we might stick into our programs okay there's a bunch of other stuff in Katz effect so we're gonna be talking a little bit about the concurrent package we're gonna spend some time on both refs and deferreds today and then there's a bunch of other stuff what what notably we're not going to be talking about is like the i/o datatype or the sink i/o datatype you know which is another one of our effect types especially useful if you're working in skellige is so you do want to talk about those come find me later but this is really more about the polymorphic capabilities provided by this library ok let's look at the concurrent type class so if you take the concurrent type classroom Kats effect and distill it down to like sort of its essence right this is what we have we have a type class concurrent it's polymorphic and some effect type F it stens this a sink type instructor I'm sorry type class you know we don't really have get into that too much the important thing he really here's this start operation so if I have a program called F a of type F of a I can start it up sort of asynchronously still do this in a completely referentially transparent way and then get back a handle to that thing that I started okay so we call these handles fibers you can sort of think of them like threads there they're not mapped to like real operating system threads bunny stretch the imagination and sort of the key operation on these fibers is this join so joining on one of these handles lets us sort of block right on the response when on on that particular program finishing with a single value of type a okay but we're not literally gonna block we're not gonna block a thread right we want to sort of register a callback sort of so you know a good way to think about this is like if you've used Scala you know you token current future I'm sure all of us have if we want to you know write a computation that's sort of dependent on a previous future right we're gonna use flatmap right and so same thing here we're gonna use that very same pattern that we're used to from like using futures we're gonna go to call join and flatmap on this f of a that comes back to represent sort of like this dependency is sort of like block for a value type a but do it in a way that's not actually like holding an operating system resource not actually holding a thread right i'm very similar by the way to like you know go routines or you know nodejs or like any of the other sort of like callback based you know sequential concurrent systems the other important operation on fibers cancel so you know cancellation you know sort of this this difficult thing to implement ad hoc after the fact and like john had to talk you know last year about that so fibers you know support cancellation so any any f of a any you know polymorphic effect that we want to be able to start up we have the ability to request cancellation you can see cancel returns this cancel token it's really just an alias for an f of you you know we'll look at a bunch of examples of both join and cancel in a moment but first I want to try to implement this particular operation so just to be clear this operation comes built-in it comes from Katz core there's a parallel type class if you have the right imports you can just call part two fold one two like iOS or two to sort of arbitrary effect types and and you get this particular operation but it's really illustrative I think to implement this in order to get a feel for what it's like working with fibres okay so what's part two fold well it's basically given two programs one that computes an F of a or one that computes an A in an F context right and one that's an F of B give me back a program that you know sort of runs those two computes about those two A's and B's and sticks them to a tuple combines them right but unlike like a normal tuple of operation we want F of a and F of B to run in parallel right we want them concurrently executing and when the last one finishes give us back that return tuple so the question is how can we do this in a you know again in this very crisp way right depending only on the abstractions that are guaranteed to us by our type classes and like here's here's a way to do it you know it feels pretty natural we import some syntax so we can call the type class methods directly on values we can say F a dot start to go start a fiber up to compute a right we call FB dot start to go start a fiber up to compute B we then wait for the you know a value to come out of that first fiber by joining on it wait on B and then yield our tuple all right very similar like you know code with Scala util concurrent future now unfortunately this is incorrect this code actually has quite a bit of issues with it but it will work leave you if you try this and you know do a couple sample programs everything appears to be working so what happens when we run this particular program so imagine our f of a argument fails with an exception in this case it's sort of you know just fails immediately with a runtime exception and there's some arbitrary F of B all right so let's think the our implementation what's going to happen at runtime we start up F of a so now concurrently something's going to go out and start computing the a right and that's going to fail sort of concurrently it's not going to fail right away so we do get back a fiber a we have we have we hold on to that fiber a we then go and kick off fiber B right we run the fight F of B program get a fiber to it now that's all fit executing so now let's use our types we see fiber a join gives us a value of type a but we know our F a program just throws an exception right so there's there's no way we can just sort of conjure up this a right out of thin air so we know from the types that that that flat lap will join cannot complete right or or rather that that a value is never made bound to a value of type a so what happens well so overall what's going to happen is our f of this a B tuple that comes back is going to fail with this propagated runtime exception right so like propagate up stack the the muriatic stack here but what happens to fiber B right so if it fails that's cool the exception propagates out but we're just often running this F of B program right and so you know in this case we're leaking we're leaking this fiber that we started up right and it's not the only leak in this case because we have fiber cancellation like someone could actually start up the overall return to program in a fiber and let it run for a little bit and then cancel that in which case we might leak both fiber a and fiber B in this case right all right so how we fix this is a pretty mechanical refactoring in this particular case where we sort of upgrade you know what we're doing and how we're binding on these fibers so this case we're using the bracket operation so brackets coming from the bracket type class it's you know sort of the top of the cats effect type clause hierarchy and concurrent certainly extends bracket so the trick here is we're gonna treat the fibers as resources sort of like as like an operating system you know you know file handle or like a socket we've opened right we've started up a fiber we want treated as a resource and no matter how we terminate we make sure those fibers get you know cleaned up so basically all we're doing if you look at like throat of the mechanical fact refactoring here is we replaced our flat map calls with calls the bracket now bracket takes two argument lists this takes you know the first argument list looks like flat map like an A to F a B the second argument list is like a clean up routine right so no matter how we terminate whether it's normal termination like we've just we've computed the value whether its termination due to an exception or termination due to cancellation we're guaranteed to have this cancel you know operation get called so anyway in this case we have like you know your place two flat maps we you know just called join and we call like the built-in tupled method to just synchronously sequence to two programs okay so this is a pretty common refactoring of like we're placing a flat map with a bracket to add cleanup logic okay let's talk about reps so a ref is a is the Katz effect version of a Java util concurrent atomic reference or another way to put this if you've never used you know atomic reference it's like the cats effect version of a bar right so like we do all this stuff to do all this cool FP stuff and then we're literally gonna use bars in some of these programs but we're still gonna do it in a referentially transparent way we're not going to give up on our functional programming principles so what does the ref look like it's polymorphic enough in effects F and a value type a it has two main methods getting set so you can imagine if we're talking about like a replacement for a VAR getting set are the two main operations right so the return type of get is not a but rather F of a right it's a program that when we evaluate we get the current value out of this ref set just takes a pure a gives us back an action like f of unit that when we evaluate that F of unit is going to set a value into that ref okay that's an important distinction there and then there's some more interesting operations maybe some more fundamental operations in in the case of this access thing the access things kinda interesting it's a it's an operation if we call we get a program that gives us a snapshot of the current value of this bar right here's just a snapshot of the a whatever is currently there along with a way to replace it with a new value so that's kind of heady but we're gonna use that in a moment and then some lower level derived operations I want to point out like construction of a raft is effectual I can't just like new RF okay so like the companion object here has this of method its return is an F of r f f/a okay so our rep itself the construction of it is suspended in our effect type and so why like why can't we just new a wrap right and the answer comes down to you know substitutability under your you know program meaning changes under substitutability if i have two expressions my program both of them have the exact same called a raft of if i replace those two expressions with one expression my program meaning is going to change right because i had two variables and i have one variable one once or a memory cell to put values all right really simple example is just a toy example we can call rafts out of we have to specify an effect type in this case I use a concrete one IO and I initialize it to the value two Refs must always be initialized to a value they are never uninitialized right just like a VAR excluding those and then we flatten up into it right we fly up into this mate you can sort of think of what comes back there is like this constructor that every time we flatten up into is going to create us a new memory cell right a new ref in this case we update the current value to five times the value we call get and you get back ten like you'd expect now if instead we had like flat mapped into make twice we would have created two different memory cells right and would have gotten back different values alright so how might we do use this stuff to do something interesting so this is a use case from an application a colleague of mine was working on the Scala jeaious application it uses react there's a bunch of UI events coming from the browser and they need to be do bounced right so if we get a bunch of the same event type back-to-back-to-back we want to take action only on the most recent event type so we sort of want to like when an event comes in kind of delay it for a little bit of you know time if we don't see any other events of that same type come in and sort of dispatch it out and go do whatever it does and so this is what we had come up with this did bounce our class so we have two operations apply and cancel cancel you can imagine it could be called like if you leave the screen right if like the overall react components sort of unmounted taken out of the you know the the browser we're going to shut down this debouncing this so the real main operation is apply applies given a program f of a and returns us back a program of f of a so we're going to change the behavior though of f of a what we're gonna do is first delay it i just kind of just slowed down the execution of it for a moment and then as long as no other new program comes in cool right we'll let it evaluate and then we'll return the overall value for the in the returned F of a now if another new T comes in though while it's waiting we want to cancel that thing and then never complete it just let it hang so if any code had flat mapped on it it's just never going to see one of these eggs come out all right here's how we might implement it remember you want to do this like pure functionally right using all the principles that we typically would use so so here's how we do it we in this case have this current Val we're going to put in this class it's going to be a ref and the ref now we need to pick like what our value type of the ref is what are we going to store in this memory so right so this is pretty pretty standard of like our concurrent state is going to be in a ref and then that ref is going to be stored as a field on a class all right so in this case we're gonna store a cancellation token a cancellation token for the last task that we queued up right or we went off and start to execute so our cancellation tokens are F of units and in this case we have an option of F of unit to represent you know the initial state where we haven't actually started up yet let's implement cancel first cuz it's pretty straightforward cancel will implement by using the modify Combinator on wraps so modify you know you pass a function to it it gives you the current value of the ref you return back a new replacement value along with sort of a side channel return okay so in this case we're gonna say modify the current cancellation token whatever see is the current cancellation tok and replace it with a nun get rid of it but then also sort of return that see on the right hand side it's like a side channel return and that side channel return is going to be like the overall result of binding into our modifying okay now since we get this f of unit now like if we flat map into this overall expression we're gonna get this F of unit representing our cancellation token or actually an option F of unit we want to go and execute it to actually cause the execution to occur to actually do the cancellation so in this case we're using sequence underscore it's just a syntax trick it's a Combinator right but basically it has the the effect of saying if this thing's a nun it does nothing and I've been this thing is a sum it's going to execute it okay in this particular case alright that's it for cancel so that's thread safe it's resource safe cancels good to go applies a little more interesting apply we're going to use this access operation so access access is kind of mind-bending we call access we flatten up into it and we get a tuple the first time of the tuple is a snapshot of the current value of the ref in this case I call it canceled and the second all in the tuple is a setter right that's it that lets us attempt to set the ref to a new value that might fail and I have to handle that if it fails right and it would fail like if there was concurrent updates occurring right so if in between the time that I called access to the time I evaluated my set if something else came in and like concurrently set my ref my my overall setter is gonna fail alright so we're gonna do the first thing we want to do is cancel the old cancellation token get rid of it right run the task to cancel that old F of a that we were running the next thing we want to do is delay execution of the new program T okay so here we're using delay and and right shark and T right to say you know wait to a certain period of time but do it in a non-blocking way like don't block a thread right and after that finishes like followed you know delay followed by go and take do the original program T all right so we're delay come from we define that as a vowel that Val we really just called the sleep operation using an implicitly available timer in the effect type F so timer is from cats effect it's one of our built-in data structures it lets us either work with you know delayed time like in this case of using sleep or working with like clock values if we wanted to read thing you know the current clock of the machine alright so first we're going to cancel the old token now we're gonna delay the new program now we don't want to just run it now though we want to you know start up or started up as a fiber and again we're gonna treat this fiber as a resource so if we're cancelled or otherwise fail we have no situation where we leak that resource so we start it we call bracket we've got to now do something with the started up delayed program but if for any reason the body where the currently of these question marks if any reason that body fails we cancel the overall returned fiber all right so what do we do inside of bracket well the general idea is we just want to store off the cancellation token for the thing we store that you started up right it'll be the new value of the of the ref so we can call try set right so try set that you know the thing that can't access gave to us as a function so we call try set and I said that might fail though right so we have to handle the case that it fails or we could ignore it but but we want to make this correct right so we handle the case where it fails we're gonna do that we flatmap into the return try set we get the flag saying whether or not we were successful or not here's what sort of interesting if we were successful we can just join our fiber now like basically just say the returned f of a that that we're gonna return just just return the overall result of that delayed program if we fail on the other hand what that means if someone else has concurrently updated our F with a new test right like there's a race condition someone replaced our current value with another task in which case we would just want to immediately cancel the thing that we tried to startup right and then hang this is what that's concurrent F dot never does hang and don't don't return value okay so anyway we we do use this exact code to debounce college AS events from from browsers okay one other main data type we have this deferred thing so deferred used to be called promise it's very similar in fact to promise we renamed it cuz we didn't we didn't want to confuse folks because there's a little bit of a difference between a deferred value and a promise deferreds have two operations getting complete so again it's you know polymorphic and F&A if we call get we get back a program that returns us an a but now unlike wrath right because ref had a get operation like this unlike ref a deferred operation is initialized without a value right and it represents an eventual completion with a single value of type a all right so if we call get and then like flat map into that you can imagine our programs going to in some way semantically block right until that new a is available so again this sort of just like flat mapping on a future now completes interesting you call completely pass at a pure value a it gives you enough of unit you evaluate that F of unit and it has the effect of completing this deferred value with that value so anyway it looks really really much like a promise why don't we call it promise it's because we don't do an error channel here there's no way to complete a deferred with an error and you know the implication of that is that get itself can't actually fail with an error right makes a little bit simpler to both implement and reason about at least arguably you know if you if you do want to like push an error channel through a deferred value you just wrap it it either okay alright so my use case for this is implementing a queue so what I want to do is implement sort of a naive queue it's going to be unbounded but it's gonna be completely pure completely reference to reference a transparent and it's going to be both resource safe and concurrent you know thread safe very completely thread safe and totally concurrent alright so like signature wise it sort of looks like deferred we're gonna have almost like in this case set and get right we're having in queue operations is given a single value of a getting back a program that sticks that a into a first in first out queue and then I'll have a DQ operation that just says get me the next value out of the queue okay and our challenge is to implement this using the existing data types we've already created alright we're gonna use the same pattern we're gonna store our concurrent state inside of a ref that ref is gonna be stored as a vowel in a class or in this case a constructor argument now the type that we're sticking the value position here is kind of mind-bending when you first look at it released its large right so the type isn't either so the state of our queue is gonna be one of two cases right either we have a bunch of elements already in the queue sitting there waiting for someone to come pick them up in that case we're gonna represent that as a right okay and so our right here it's gonna be a Scala collection immutable Q of type a they're just sitting there waiting for people to queue them right now the flip side is we might not have any elements but instead we have people waiting for elements to show up right we're gonna represent that as a left and our left we're gonna also use Scala collection immutable queue but we're gonna have a queue of these deferred values these promises for future elements all right all right so how would we then implement this well let's do in queue first the incubation is given a value of type a we're gonna use this modify operation on our ref and we're gonna do this side channel return again right so when we when when we call modify or to be given the current value we're gonna return our replacement value as well as a sort of a side channel output right and now in these cases our side channel outputs going to be another effect full value which we immediately gonna go and execute so update the ref to a new value and then execute some action as a result of that update after it's been successful this is so common there should probably be a Combinator for it it's intentional that there's not if you want to know more come talk to me I'm happy to debate that but okay so let's let's take a look at how modifies implemented if we have a new element coming in and we already have a bunch of elements sitting in memory this is easy right in that case we can just take this new element a and stick it onto the end of our current queue of elements and so we update our state with a new right just with that a you know defend it to the end now we don't want actually have to do any effect in this case we just use you know unit that pure to say don't do anything after this completes if the states are left instead well there's a couple things we have to handle just to handle like empty queues and whatnot but the general idea is take the head element out of this list of waiters and complete it with the new element that came in and now if we're left to know what nobody else waiting we're gonna update the queue to be back on the right hand side with an empty set of elements if we do it folks waiting then update the queue to still be on the left-hand side with one last waiter okay that's all this does now the side channel here we return this WS dot head complete right so we turn an action that when we flatten in a moment that's going to have the effect of having the DQ'd element you know wake up one of these listeners all right let's look at DQ real quick DQ'd we're going to create these listeners all right we're gonna go in allocate these listeners so in this case we're using the apply method on the deferred companion object we say create a promise write for this value fa we flatten up into that right because a construction is effect 'full so we flat map into that now we have some weight or W alright something that we can give out or weight on and now we want to register that W in the state of our queue we're gonna look at what that looks like we're gonna use that side channel return pattern again so after we call modify we're gonna flatten in this case right to execute some value that we return and then finally we're gonna wait for that deferred about you to show up the W dot get call is going to give us an F of a you know we've an F of a in the final position inside of a flat map so you can imagine overall we get an F of a as a result okay all right the two cases here again there's some just just nonsense to deal with the edge cases but like if the Q already has elements in it the general idea you know we check for the empty but the general idea here is highlighted we want to Amin mmediately complete the brand newly constructed waiter right away with this value okay and then on the you know on the left hand side if we have waiters in the queue already we're just going to queue up this new waiter to the end of it and we have no action to take in that case okay so this coded dub you get is interesting so we did like all this work and then we wait and you know we started one of the more complicated examples today by talking about cancellation like a part to pulled example right we had cancellation we had a problem where we leaked that fiber right upon translation well it turns out in this case we actually have a memory leak it's kind of a gnarly little memory leak and so the general idea is like what happens if an exception occurs or if cancellation occurs in this case it's really cancellation we know it's not an exception Rex I said deferred can't fail double yet can't fail but someone could call DQ give it some time and if an element doesn't show up cancel that DQ right so the common operation there is is a time to DQ like DQ an element but if I don't have anything after one second notifying me with the timeout right and the way that's implemented is is with fiber cancellation all right so why do we have a memory leak well when cancellation occurs we're gonna come off this double yet but the state of our Q class still has this W sitting around right but W is still up in that ref and so now imagine like your system has a lot of events coming in a lot of folks to queueing events but you have some of them start to timeout what its gonna look like is a lot of your events are getting eaten by the queue like some of them might be making it out right but but these outstanding w's that are never looked at again it's gonna look like the queue sort of you know dropping these events on the floor alright so how can we fix we're gonna use our same trick as earlier right we're gonna place a flat map with a bracket right and the trick in this case is we're gonna treat our resource as our deferred value the resource I mean the deferred value is the thing that we're gonna try to you know keep a handle on okay so we replaced flatten up with bracket and now we have to implement a cancellation clause to say if for what reason we come out of this come out of this block right out of this used block what do we need to do to clean up the state of our cue to handle the fact that like W is no longer going to be waited on now it may have it may have it may not be there right the cancellation might have occurred before W ever got put into the state or the cancellation would have occurred after W got you know W like get got completed so anyway our cancellation basically has to handle all those cases in this case we just do like a linear walk through the waiters and the Q and just remove the one that we're going to return or that we created in this case there are more efficient ways to do this I'm not going to go through all the more efficient ways to do this but there's a bunch of other ways this is a bit of naive implementation of a Q but hopefully it gives you like an idea of the way these Combinator's can sort of build bigger and bigger concurrent data structures off of some pretty simple primitives right we're not going to do some befores today we don't have time but if you if you are interested in semaphore is like the slides are in line and they're full implementations they're also there's a dining philosophers implementation you know it's a bit of a challenge to see if we could do it I posted today so if you want to see what dining philosophers would look like with fs2 it uses semaphore and so some before in cats effect is actually implemented you know one way but we used to implement it with deferred and wraps internally like some before was a derive data structure off of these more simple concurrent data structures all right so that's cats effect one it's pretty much I want to go through today if you use any of these libraries look at us to do be HTTP you know they're all built on this infrastructure of cats effect and all of these data types up here the cat effect io monix and zio types all of them have cats effect instances you can you know pick and choose whatever one you like the best use in your applications and all of these techniques are completely safe completely you know concurrent and and you know all all completely principled all right that's all I had I think I have time for another or for one or two questions [Applause] all right yes the question was what's next for cats effect in general I say so one like a bench before the type class hierarchy is not final in any stretch of the imagination you know one of the strengths of cats effect I think is that you have a ton of diverse opinion on how functional scholars should evolve and we continue to kind of take everybody's feedback and evolve the library in a way that you know maybe we all didn't foresee a year ago right I think you'll continue to see that occur that might entail things like you know a type class to represent effect types that can express errors like unfailing i/o types maybe we go and do like the whole by functor i/o that like John's been pioneering and zio maybe there's type classes represent that I think you'll see those types of things at the same time I think we'll continue to see cats effect concurrent data structures get built we're talking earlier about like maybe some type of pulling data structure to handle like connection polls and object polls and things like that so yeah I think you'll see both overall cat's effect has sort of like two primary purposes one is this interoperability layer right one is so like we can work on things like do B and not make a choice of an effect type for all of the users of duty right but another is to you know provide parametric City and provide you know the ability to write and tagless final style in our applications and get all the benefits that we see in our libraries but get them at the application level as well so that you'll see innovation on sort of both of those fronts any other questions all right thanks you