Declarative distributed concurrency in S...

Scale By The Bay 2018: Sergei Winitzki, Declarative distributed concurrency in Scala

Recording: Scale By The Bay 2018: Sergei Winitzki, Declarative distributed concurrency in Scala

thank you very much so I'm going to be talking about my open source project which is very much experimental and challenging for everybody to understand and use including myself but if it promises should have been full screen but it's not doing full screen okay so this paradigm is called the chemical machine okay you've never heard of it but this paradigm promises to implement everything in 15 lines of code or less it's so declarative that your programs are correct by construction they're obviously correct and no need to debug anything and the only problem is you need to start thinking in a new way okay so that's my goal today is to explain how to think in this new way in different different metaphors and show code that uses this I have presented this before in this presentation today there is a significant new element which is distributed programming so the previous implementations that I had where concurrent programming only which means you can implement any concurrent tasks in 15 lines of code or less but now you can implement any distributed concurrent tasks in 15 lines of code or less so that's a new element in this presentation not in my talk will be any kind of academic theory and I'm formally a theoretical physicist so I can bash academics if I want to this chemical machine paradigm is based on something called join calculus so can I ask how many people have ever heard of that want to in the back right not a lot of people and I can't blame you for not knowing about during calculus because if you even get this idea of looking it up in the Wikipedia and he will read something that's that's like this communication occurs only on defined names and replicated reception is synthetic li merged into the definition okay so this is completely unhelpful okay this I'm I don't really understand what they wanted to say in this web page although I have implemented this paradigm several times I know very well what this is supposed to do how it's working but I don't understand this jargon and they're and they don't care to explain it to you so let me just close this tab it's so useless so that's basically what my talk is going to be about we all know that imperative concurrency and parallelism is very hard and we don't want to do that because there are all kinds of problems with reasoning about mutable state reasoning about processes that run at the same time it's very hard to test you have non-deterministic behavior at runtime there's never enough testing basically so you you have all kinds of race conditions deadlocks live looks people don't want to program in threads and callbacks and semaphores and people have other approaches people have these more declarative approaches to concurrency and parallelism which i stratify into for complexity classes my classification is from top to bottom increasing complexity so the easiest way is synchronous parallelism and this is MapReduce as a spark this is color parallel collections where you split your task into pieces each piece is separate and independent you run it and you merge the results from the type theory point of view this is applicative functor applicative functors are well known to be easily runnable in parallel because computations are independent of each other the next complexity class is a synchronous streaming graph or without cycles and that is realized for instance with future async/await and various reactive framework streaming frameworks there many many libraries now that do this from a type theory point of view this is a magnetic functor like future is a monadic functor and you can see that is strictly more powerful than a colligative and so this is really a different complexity class from the point of view of type theory already so that's not just arbitrary distinction the next complexity class is unrestricted streaming where your graph can be cyclic it can be paralyzing forking and joining parallel things and so on but it's still a graph there are different implementations of this we will here in this conference I'm sure about FS to z io and other such things flink is another one extremes can probably do this to and from the type theory point of view this is a little more complicated than just just a monad that it's a recursive monad so the graph can feed itself asynchronously and the final level of complexity is unrestricted concurrency which means any kind of process can start at any time wait for anything and start other processes or terminate them or whatever it's not a graph it's not a stream it's more complicated than that I have no idea what type class that means I don't want to try to say that and yet more complicated it's distributed computing so on top of all that you need to coordinate between different partitions of the network and so on consensus clustering discovery all that is another layer of complexity on top of this fourth layer which I call unrestricted concurrency and I would like just mention that Apaches keeper and etcd and other such may be other things like kwang-soo they provide distributed coordination as a service and that's what I'm going to be using as implementation for my library but this is just a just an aside so let me first start with ordinary concurrency so the example of concurrency is the dining philosophers problem can I ask who does not know about dining philosophers problem who has not heard about it one two three that's excellent I'm not going to have a lot of trouble then so basically these philosophers whom you can easily recognize by their faces they have resource contention so they eat and think concurrently and in parallel but they have resource contention they're contending on the forks and so this is a problem in I'm in in my opinion it is not a graph it's not a stream there's more complicated than that I don't know how to prove mathematically that this cannot be implemented as say an expression of type future of X where X is some type in Scala I don't know maybe you can maybe you cannot in my intuition this is more complicated in the stream because you have resource contention stream without resource contention is less complicated than the stream with resource contention so what I call unrestricted concurrency you can think it's more than stream because it has resource contention at the same time as it has concurrency now in different programming languages I wanted to just show very quickly the solution right so there is this great website called rosetta code and it has solutions for different programming problems and very different programming languages so you see a lot of people contribute their favorite programming languages so for example let's take I don't know f-sharp strategy work faith waiting for it was a fair way to do a sink sleep kind of complicated anyway C++ is much worse so that's that's you know that's not easy to simulate this problem this dining philosophers and let me show you this solution if it's on one screen it's certainly less than 15 lines of code is it kind of they are almost a 5 5 4 up there and one yeah 15 lines of code yeah so that is the chemical machine parallel if you can recognize various less than 15 lines of code and this solution is in the language called Joe camel which is a fork of Oh camel to implement the chemical machine and you see this solution is completely declarative it just says that one philosopher and two forks leads to the eating process and another and two forks are released and philosopher is not hungry and then and non hungry philosopher leads to a thinking process and a hungry philosopher that's your that's your declaration that's your specification of what you want to do and this is actually executable this is right this runs and prints the philosophers thinking and eating and so on so the chemical machine code is purely declarative in this sense it's a domain-specific language for concurrency unfortunately because of insufficient documentation and it's a fairly new paradigm nobody knows about it so I'm trying to spread the good word and I'm trying to show that you can program or at least you can think about concurrency maybe this is you know you don't want to reuse this in production maybe because it's my side project which is not supported but it will make you think about concurrency in a much clearer way once you learn how to think in this way you can always do other things so in order to explain the chemical machine paradigm I have three different metaphors the first the easiest one perhaps is the comparison compared with Amazon lambda and Amazon lambda is almost the chemical machine except for one feature so what does Amazon lambda do it waits for some kind of signal that arrives Ignis that the date data is here data has arrived then it runs a computation on that data the output data is pushed back into the system and it might be another trigger when that happens add two requirements one that lambda should be able to wait for several triggers at once not just one trigger but several triggers at once to when the one or more triggers arrive they're consumed atomically so if two more lambdas can wait on the same trigger and contend on it only one gets the data and the other doesn't now this could be quite difficult to implement actually in Amazon's distributed setting I don't know maybe it is easy made I know I don't think it's it easy but what do I know if you do that then instead of Amazon lambda you get Amazon PI I'm making a pun here on PI calculus you get basically unrestricted concurrency with current Amazon as far as I know they do not allow you to have multiple triggers that is streaming with this extra feature it becomes first level of computer unrestricted ii so ii metaphor is compared with act normal i assume everybody knows actor model here again actor models are actor model is very powerful but we need to modify it in a few ways so that we can get the chemical machine out of it and that helps you think about the chemical machine first modification actors are created automatically by the system you don't create actors you create props you just create the body the function body of the author of the actor but not instance not not actual actor ref you don't manipulate actor refs they're all automatic and they're created automatically in parallel whenever messages arrive so you actually send messages to mailboxes and when messages arrive automatically the system will make actors for you and actors may wait atomically for messages in several different mailboxes consume several messages at once start and then disappear so from these requirements i can logically derive that for instance instances of actors must be stateless and all you do is you define mailboxes and computations so here's a comparison so akka code roughly looks like this you declare an actor reference and the actor receives a message and does something receives a message and does some and you send messages in my library called chemists this is the actual scala syntax of the DSL very similar except you can have more than one receive statement so to speak at once so you receive B and C so B and C are mailbox references a is another mailbox reference XYZ our values on messages so this this function will consume X and do thing with it and this function will consume y&z and do something with it and then you send messages to a B and C so you send messages to mailboxes and not to specific axis yes so the question is the question is how do we understand the notion of receiving two messages at the same time where whereas in the actor model messages are linearly received one at a time and processed one at a time so you think of different mailboxes so each mailbox gets one message at a time but you are now allowed to wait on to mail built to two mailboxes at the same time and so when one mailbox has a message and the other does not you don't do anything you wait until the other has a message too and then you grab two messages atomically at once yes so the question is how do we handle contention if one a chemist's actor so to speak wait on B and C in the other ways and B and D and we get a B message in what happens nothing happens until at least one of these gets all the data it needs and then it's non-deterministic now all actor programs can be straight fully translated into this so these are called reactions it's a I'll show you the chemical metaphor in a second yeah so the question is do you insure atomicity when when you consume messages so you can specify what you're waiting for one or more messages all of them are consumed atomic weight ones and it is a schedulers job not your code but the scheduler the runtimes job to ensure that this works contention not determinism it needs to be consistent this finally let me look at the chemical metaphor so in the chemistry in real life molecules meets to like in this chemical reaction I hope everybody knows what these symbols mean in real life molecules meet randomly so imagine some kind of chemical reactor and all we do is stir and the molecules just randomly meet each other and start reacting nobody is telling them to wait or anything they're just completely naturally distributed concurrent that's how we want our programs to be naturally distributed concurrent parallel we don't want to tell each molecule to or wait we need this process this thread no I don't want to do this I just want to say what reactions exist in nature that's not what happens nature in those reactions and it just runs that's what we want to do we want to have our computations run as naturally concurrently in parallel and distributed as they do in real life so to begin doing this first we go from real chemistry to abstract chemistry which means we just declare molecules of any kind and reactions of any kind we don't care about real physics or real chemistry so program code just defines our molecules ABC and these laws for example a plus B goes to a whatever I it doesn't matter in the real life of course this cannot happen but it doesn't matter the runtime system will grab these laws look at them then it will simulate this I have an a I have an a I have a B or I can run the first one let's run it and then just keep doing that and simulator can be implemented quite easily in any language and I've done this in Objective C I've done this in Java it runs it's pretty easy now in order to put this to use you need to actually compute something and the key idea is that each molecule needs to carry a value on it so they are not just molecule they're actually containers for values and these values are now concurrent data they can float in this concurrent world weight to be consumed each reaction will consume for example a and B but they now will consume the value on a and the value on be when it emits a at the end it needs to put a value on a say Z where would this Z come from well it obviously it needs to be computed from X&Y so there you have a function from x and y which corresponds to each reaction naturally therefore your program becomes a specification of molecules and their data types and the number of functions that take inputs and produce outputs and put these outputs on some molecules and take the inputs from some molecules here's the actual syntax in my chemist DSL for this kind of program reactions are partial functions in the Scala syntax this this plus is the chemical plus is not really an operation it's just a separator I'm taking a with value X and I'm taking also be with value Y I'm making a computation and I'm omitting a with values very concise yes [Music] not the term so question is what happens when the requirements are meant for two different functions which means data is available for two different computations to start which one will start non-deterministic it is up to you to like in nature what happens when molecules can react this way or that way they just do whatever they feel like it's up to you like it is up to the chemical engineer to design the process so that the correct molecules come out and it's always possible but otherwise is non-deterministic sometimes it's good to have non determinism in your code because you don't care and what order certain things are computed but it is necessary to design everything such that the results are deterministic of course so basically we need to learn how to think in molecules or concurrent data and reactions to compare this with the actor model very quickly actors are similar to this but you still have to reason about instances of actors I create instances of actors and this is like saying I compute five times six plus seven I need to have an instance of plus how many did I already allocate and an instance of times because I have five times second is an instance of time do I have already as instance of time can I send it to you don't think like this you just say five times six plus seven you don't think about how many instances of plus you have similarly in the chemical machine you don't think about instances of actors you specify your computation there is no state there's no supervision there is no lifecycle there's no dead letter there's no router there's not none of this and the code is typically several times shorter than a code as a result I will be posting this throughout this code here's one first example ok throttling oh sorry there's a question the question is how do I manage complexity of I have a lot of reactions I have a lot of these computations each of the computation is a local value I can put them in local scopes in different modules I can think about one of them at a time so it is certainly it's it's just like modularizing your program you split it into independent parts so you see there's a concept of a reaction site which is this site function and each site is independent it runs independently on its own thread pool and you think about that site as something that contains on its own input molecules independently of all others so that's how you manage the complexity each little reaction site has maybe a couple of reactions you think about them and you think only about them you don't have this and the spaghetti code of all kinds of things connected to everything else that's and let me just show you this example you will see how this is managed so the the task is I'm trying to throttle emission of a signal so in the chemical machine there's only one way to send a signal it is to emit a molecule emit data on on the molecule suppose I have some molecule s I want to throttle it so I don't want to be emitted s too many times I need some minimum delay how do I do that I make another molecule called R which can be emitted at will but when you emit AR s will be emitted with throttling every time so R has a value on X on ethical X with type X I define a new molecule called R I define a new molecule called enable because the only way to prevent a reaction from happening is to withhold input molecules once you have input molecules in your chemical reactor nothing can tell them not to react to each other the only way to stop or delay something is to withhold input so therefore I create auxiliary input which I call enable which doesn't have any value it's just a unit value name but I withhold it until I'm ready to run the reaction then I define a reaction which takes our takes enable it emits the s and then it sleeps for this interval and then it emits enable so enable is emitted only after asleep while it is sleeping somebody could have emitted a million hours but no reaction will start because enable isn't there that's how you think what molecules are present when are they present everything will work your computation will start but what are present in what do you have enough of those how do you prevent them from being emitted second thing you see this is a function six color function deff so all of these things are defined in its local scope and a completely invisible outside nobody can screw up with this it's it's it's sealed you return R which means R is of type M of X which is molecule emitter with values of type X so whoever calls this throttle gets them are out but they don't get enable out they cannot it screw up with this logic so I guarantee that enable is emitted once initially in here and nobody else can ever emit enable because it's in a local scope so in this way I have local reactions local molecules and I carefully separate the scopes note there are no threads now semaphores no locks no mutable state and everything works in far less than 15 lines of code whereas all these other implementations in these other great libraries are much longer yes they are not this is asynchronous emission is asynchronous correct I don't understand catches on fire so the question is what happens if the process crashes and this enable molecules never emitted and we pile up the R and it's now yeah it's a deadlock it's a deadlock so yeah if things if things crash yeah you get a deadlock in this case all right let me just jump through a few examples this is MapReduce akka is a lot longer this is parallel merge sort akka is longer this is dining philosophers and I have a lot more examples and and akka is longer so if this were implemented really well and really efficiently which I'm trying to do what I'm just one person and it's an open source project and that I do in my spare time if this were really efficient I would never touch akka again okay because I don't want to write two hundred lines of code if I can write 15 or less and I said 15 or less anything you want to do you do 15 or less so reasoning is very easy you just think about what molecules exist and what data is available for computations you don't think about processes run run concurrently or not and okay let me just before I go to before I go to my distributed so this I already already explained previous in previous presentations now you see if you look at when certain concepts were invented and when they were adopted by in three there's a about a 40-year gap 38.2 year gap right so point two years are like eight 11 weeks so 38 years and 11 weeks you have to wait and at the end of the 11th week the industry says great we're all going to use it like streaming closures you know everybody every programming language has a lambda now everyone even even like like C++ right anyway I've got a lot of stuff I'm skipping here has anyone heard of Petri Nets more people heard of them the chemical machine implements them a very rich version of Petri net it's one to one and it's it's got more features molecules are values reactions or values you can put in our channels and channels and channels whatever you want it's very very flexible and the Petri net can be translated into a chemical machine very easily ok new stuff I've got five minutes before questions but I've got already a lot of questions answered hopefully new stuff distributed chemical machine I figured out how to do distributed programming in the same model 15 lines of code or less here's how it works you just say this molecule is distributed instead of M so M is this constructor for a molecule I do DM which is a distributed molecule constructor that's it now this data is distributed that's it oh yeah and you have to put an implicit Val with a cluster config which is the data structure that shows how to connect to the cluster and it's a zookeeper instance and you just connect to it you're done that's it the language is modified by adding the concept of a distributed molecule which means data so previously we had a concept of molecule which is data that is automatically concurrent it is ready to be consumed by whoever is there and whatever you define now we have a concept of distributed molecule data that is ready to be consumed by any peer on the cluster and that's it let's see the only concept now you know everything you can just go on and program I don't have to explain anything more more to you this is the entire language here's an example of how to do distributed MapReduce 15 lines of coreless and ik of course is 400 lines of code who would ever write that now there are there details that I'm not going to be able to explain in the limited time but one important thing yeah distributed cache ten lines of code Street a peer-to-peer chat fifteen lines of code acha code is longer the main interesting things about this concept is that every peer operates in the same way it's as if you have this chemical reactor that has just distributed between different plants and they all exchange the soup the solution with the Chatterley they exchange this the solution in a random way so whoever is free to start processing it will just grab some molecules do the processing put the molecules back into the cluster and then grab some other molecule do the processing put it back there is no master worker driver nothing it's just everything is completely symmetric if you disconnect from the cluster and connect again I just keep going there's no nobody knows about you my code does not know how many nodes are in the cluster how to connect to them doesn't you know this is the entire code and you reason about it in the same way you don't think about how to connect you know URLs for is this is this actor remote actor is it up has it been supervised none of that the semantics of your program does not depend on the number of nodes even if there are no nodes in the cluster your program still runs correctly just slowly and until some but you never know when some nodes will connect or disconnect and the semantics does not does not guarantee that okay I will skip to the conclusion because I want to give you time from our questions okay so basically declarative and almost purely functional I don't I shouldn't have said really functional because emitting a molecule is an effect defining a reaction is an effect it's not it's not the value that you can preferentially transparently put into an array or anything emitting a molecule is an effect other than that things are purely functional it's more purely functional than actors for sure significantly easier to reason about I have an open source implementation I'm planning to have some very good industry strength features I have a tutorial book that explains how to work with it and there are some interesting promising applications especially in workflow management distributed peer-to-peer systems and I'm working on getting the chemical machine to be distributed so that you can actually run it right now I'm still in progress on that the non distributed version works in Syria version is in progress and I have some slides with technical implementation details I'll just skim them so thank you [Applause] so the question is can I make a purely functional can I instead of doing an effect immediately can I return some kind of effect type from these things from these functions I can I have a function here that returns an effect type instead of actually executing the effect very good question I've no idea how to do that maybe you can try to do that in some way but right now I'm not able to give you an answer so and another part of your question was what if it crashes I would like to make it more resilient so that you can have error correction so a concept of a reaction that crashes and then something should happen if that crashes so the first thing that I already want to do and I know that it'll work is that if it crashes then the cluster should know that this distributed data has not been processed it's like a commit in the Kafka world so if it did not crash I commit but if it crashed I do not commit and so that's kind of what I am thinking of doing but or some kind of recovery some kind of retry I have a retry feature I don't know if it's very useful but that's what I'm trying to do so I'm not I'm not going into the direction of purely functional because I don't really know how to do it well in in this context but this isn't this is a good question yeah so the question is are the cues for each mailbox unbounded they are however the runtime can be made more clever and so you can maintain counts of how many threads are busy and so on and you can you can try to schedule fewer reactions or more reactions depending on how full mailboxes are right now they're unbounded and you can pile up a lot of a lot of unprocessed data however molecules are just little pieces of data on the heap with some wrapper they're not threads they're not heavyweight they don't have their stack molecules are not like processes they don't have stack they don't have date thread information so you can pile them up really pretty pretty big and I was I was so there is a lot of research on queueing theory which tells you what is the capacity of a queue and when you can expect things to be done and so on maybe you can just and since in this in this kind of setting your entire computation structure is known in advance you can reason about it before you run it and you can decide to monitor this reaction because it's got this input-output problem or something like that so I I think that's a good way of solving this problem rather than making a mailbox bounded by some random Ameri of our network okay so the question is is there any mechanism of favoring one kind of reaction of another in case of determinism for instance the favorite cheaper reaction or to favor a local reaction over a distributed one right now there is no such mechanism for various reasons it's it's probably not productive to have priorities on reactions however local reactions can be very easily preferred to distributed ones which is probably what I'm going to do the problem is you don't know when reactions start and when you don't know that and you think that a reaction will start because it has a higher priority you're going to be mistaken you're going to have race conditions okay so the question is can we optimize that so my goal is right now they're optimizing CPU usage and that's what I I'm going to do with thread pools and so on I'm not going to prefer one reaction over another so right now I don't want to put weights on reactions and have some but in principle you could just say this reaction has a higher higher value to us and we would prefer to run it and maybe you can schedule it first and prefer it too you can try to do that in it will it will probably it will certainly be dangerous to use that mechanism in order to make your business logic work because that's a race condition right there you know if yeah sorry yes the question is can it can't have conditions on values so yes it can it's a scholar partial function can arbitrary things you can have guard conditions match you know part pattern match and whatever you want yeah we are out of time thank you very much you [Applause]