RELEASE Scalable Erlang
CC Attribution - NonCommercial - ShareAlike 3.0 Unported:
You are free to use, adapt and copy, distribute and transmit the work or content in adapted or unchanged form for any legal and non-commercial purpose as long as the work is attributed to the author in the manner specified by the author or licensor and the work or content is shared also in adapted form only under the conditions of this
Chechina, Natalia
In this talk I'll present Scalable Distributed (SD) Erlang - an extension of distributed Erlang functional programming language for reliable scalability. The work is a part of the RELEASE project that aims to improve the scalability of Erlang programming language. I'll start by providing an overview of the RELEASE project and discussing distributed Erlang limitations. Then I'll introduce SD Erlang, its design, motivation, and the main two components, i.e. scalable groups and semi-explicit placement. The scalable groups (s_groups) enable scaling the network of Erlang nodes by eliminating transitive connections, i.e. a node may belong to multiple s_groups where each s_group node has transitive connections with the nodes from the same s_groups and non-transitive connections with other nodes. The semi-explicit placement enables to spawn processes on nodes either in a particular s_group, or with particular attributes (e.g. available hardware or software), or with certain parameters (e.g. least load). I’ll also cover the results of the preliminary validation, and SD Erlang operational semantics and its verification. I'll conclude the talk by providing a brief overview of the ongoing work and our future plans.
 Hi, I'm Natalia and thank you for coming. Thank you for watching. And I suppose we'll start, right? So I'm Natalia, I'm from the University of Glasgow and today I will talk about the release project, the work we do there. So the project is European and it involves something like eight partners. These are five academic partners from the UK, Sweden and Greece and three industrial partners. These are EDF, Ericsson and Erlang Solutions. In particular I will talk about the work we do at Glasgow University together with Kent and Harriet Ward Universities. So I will give a brief introduction of things we do at what the release project aims in general. Then I will talk about the scalability limitations of distributed Erlang and then we'll cover what we do to scale it. So I'm not aiming to give all details but just a flavor of what we do, why we do that and how. So the purpose, the aim of the release project is to scale the radical axiomodel to build reliable software on massively problem machines. And by that we mean something like 10 to 5 cores. And for that we use Erlang. We work on three levels. So the first level is VM, then we have language level and tools. Today I will talk only about the language level and VMs and tools will be covered sometimes by different people in different talks. So typical hardware architecture, what do we mean by that? Well, we're thinking about a cloud and actually a collection of clouds, something like it's called Sky. And then clouds consist of hosts and then hosts, they consist of SMP modules, something like 6, 8 and each cell would have something like 100 of those hosts and each SMP would contain 32, 64 cores, something like that. And then Erlang, just briefly reminding you what it is. So Erlang is a functional general programming, general purpose programming language designed at Ericsson. It is dynamically typed and it was specifically designed to build distributed, scalable, software time and massively concurrent software. And then it uses these two main philosophies, which is let it crash and the second one is share nothing. The language primitives are processes and the particular very nice thing about it is that concurrency is built in in the language rather than edit on. So distributed Erlang, what's it about? Well, we have a number of nodes and those nodes, they can be right. And those nodes, they can be either on the same host, they can be on different hosts and these yellows are nodes by the way. Well, by node I mean Erlang VM. And then inside of those, we have those processes and each node, it can share a number of cores. When Erlang works in its normal mode, it means the connections are transitive. So all nodes are interconnected and they have a global common namespace. But that I mean that if we register a name on one node globally, I mean, then it will be transitively connected, transitively shared between all nodes. And if we add another node over here, it immediately as soon as we connect to one of the nodes, it will be connected to the rest of the nodes. And then it's about explicit placement. So to spawn a process to a different node, we need specifically define which node that is. And of course, Erlang and distributed Erlang is about reliability. So if one process, one node crushes, we can detect it and restart the system. So here we have a benchmark, it's a bash a bench code. So what we do is, oh my God, so we have 20 nodes, then within 18 seconds, we just kill all those process. And the system again is in a stable state. Then we start all those nodes again and system in a stable state again. But it has limitations. So we run this experiment and it's again bash a bench. Well, it's a modification of bash a bench, we call it D e bench. So what we do here is that we have a scale number of nodes, so from 10 to 100. And then we look at the throughput. And then we vary the number of the percentage of global operations that we have. So the red one is zero global operations. But then as soon as we start, you know, even this tiny, tiny number percentage of global operations, the scalability goes down. And that is when we used global module, but global operations are not only about global module. You can also use something like RPC column multiple nodes and then you will get again global operations. Another problem is with single process bottlenecks. And we can get here. So this is React benchmark. React is a DBMS system built in the Eulang. And so we look at scalability again. And here we have number of nodes. This is number of VNs, Eulang VNs. And then we scale them from 10 to 100. And then we look at throughput again. And what happens is that we have quite nice scalability up to 50, 60 nodes. And then it stops. And then it goes down with huge variation. And by the way, this is React version 11. So it was something like a year, half a year ago, one half a year ago. So they updated, they modified some things there. But the problem there is that first we have this single process bottlenecks. And another problem is that this fully connected network that we get with distributed Eulang. So design approach and principles. So we need to scale. To scale a system, to scale a program, we need to have a scalable persistent storage. And that we left to React and Xandrum. And then we need in memory data structure. And that's in Eulang, these are at stables. And this work is done by Sally University and Ericsson within the release project. And then we need a scalable computation. And that's our work. When thinking about how we want to scale it, we took as a baseline those principles. So first of all, we wanted to work at Eulang level as far as possible. Then we wanted to do those modification minimal because language is nice. Language is simple. And people love the language. And we didn't want to introduce something absolutely foreign that the Eulang community would just reject and say, no, we don't like it. It's not our philosophy. Of course, we wanted to follow the language philosophy. And then we wanted to keep the Eulang reliability model unchanged and as reliable as it is now. And then there are two things that we want to change. There's a avoid global sharing, as showed on the previous slide, and introduce an abstract notion of communication architecture. So what is SD Eulang? SD Eulang is a small, modest extension to distributed Eulang. And here we tackle two problems. The first one is network scalability. We want to use all to all connections. And the second one is semi-explicit placement. And I will talk about those two problems in details now. So network scalability. And scalable distributed Eulang, we have two types of nodes. These are free nodes. These are nodes that belong to node S groups. And another one is S groups nodes that belong to at least one S group. And so nodes in S groups, they have transitive connections. And they, so nodes within the S group, they have transitive connections within the group and non-transitive connections with the other nodes. And then if you're familiar with distributed Eulang, you can say, well, there are similarities with global groups. Yes, there are. And these are similarities that groups have their own namespace and they have transitive connection within the groups. But the difference is that the groups are overlapping. And so we don't partition them as global groups. And then the information about S groups and node is not globally shared. So let's look what happens here. So we have distributed Eulang and assume that we have two groups of nodes. So these three nodes are connected and these three nodes are also connected. Then we decide to connect nodes three and four. And what we get is a fully connected network because of the transitive connectivity. And then we look at S groups. So we group nodes. And when we connect nodes three and four, they're just connected. Connections are non-transitive and each of the groups has its own namespace. The types of connections in general, what we have. So the thing is that we don't aim to replace the existing connections and existing nodes in distributed Eulang by those groups. So if a programmer doesn't want to use them, that's fine. But if they want to use them, these are means to introduce them and communicate with other nodes. So first of all, we have these three nodes and these are free normal nodes that have transitive connections, common namespace, the connections are normal. And then we have hidden nodes. Those hidden nodes, they each node has its own namespace. So if we register something globally, it doesn't actually change much because this information is not going to be shared between other nodes. And also, they don't share connections. So they just form direct connection and don't share them with anything. Then we have S groups. So nodes are interconnected within S groups and they have transitive connections within that S group and have non-transitive connections, the rest. And here we have the same. So another question that we thought of, of course, and we often ask, so why do we use S groups? Why group nodes into those S groups? Well, before introducing them, we had some requirements. We wanted those S, we wanted nodes to join and leave us groups very easily and quickly. Then we wanted to keep the philosophy of the language. That is, each node can be connected to any other node. And then we wanted that to be simple. And we looked at three other mechanisms, so something like grouping nodes according to their hash value. But that imposed some restriction on, so if a node joins an S group, then it should change its hash value. And when it leaves an S group, it should also change its hash value. And it belongs to multiple S groups. There's again this contradiction of the hash value. So we decided to avoid that. And then another approach would be a hierarchical structure. But again, it contradicts the philosophy because in this case, node from different levels, they can't communicate directly to each other because it's just, so we decided to introduce overlapping S groups. And using these overlapping S groups, we can implement actually all other three approaches. So this, how can we group nodes? We have a set of nodes and then nodes are grouped into S groups. And one of the nodes, we can call them gateways, for example. They form another subgroups. And then gateways to those upper levels, they can form another S group. So if node A wants to communicate to node N, what happens is node A communicates to B, B communicates to D, E, F, and then it's N. So nodes actually don't form a direct connection to each other, but go through this gateways. We can also communicate, introduce something more complicated and introduce redundancy in the hierarchy. So instead of having one gateway node, we can have two in case one of them will fail. And then we can also communicate with free nodes and we can either connect them or don't connect. And they will have non-transitive direct connection to the nodes in the S groups. And then another approach. Well, we can have embedded. If we don't mind nodes to be interconnected, but we want to reduce this global space, we want to localize it somehow. Well, we can introduce embedded groups where nodes, they just share connection within their groups. So we can register names here and register names here. Then all names in this group will have that name or have all of those nodes. And how we can use those. So we can either start all that configuration at start. And in this case, we will use configuration file or we can start dynamically using this new S group function. And we can use it either with one parameter, defining the S group name or not defining it. In this case, the S group name will be generated automatically. And this is done because we don't guarantee the S group's names to be unique. So we don't collect this information anywhere. And in this case, what we do is that what we guarantee is uniqueness, high probability of name uniqueness. So not uniqueness but high probability of its uniqueness. And other function we can use. So creating new S groups, deleting S groups at nodes and removing nodes. We can also collect some sort of information that we want. So information about S groups, information about the particular nodes to which nodes it belongs to, to which, sorry, S groups it belongs to, to which, with which node it shares namespace and so on. And these are the first measurements that we did with this. So this is, sorry, what's that? Good, yes. That's good, it works. Yes. So we have this tiny percent, 0.01 percent of global operations. And we compare distributed Erlang and SD Erlang. And we have from 10 to 100 nodes. And we look at the throughput. So SD Erlang helps. That's a good news. And so to look how actually SD Erlang works, we'll look at an example of orbit. So what is orbit? Obit is a symbolic computational kernel, a generalization of transitive closure computation. So what it is? We have initially given space and then we apply a list of generators on this namespace. And from this, we generate a new list of, create a new list of numbers. And this process is repeated until no new numbers is generated. So that's the program. And we'll look how we can, how starting with non-distributed Erlang, we can come to scalable distributed Erlang, how can we build this program? Right. So the first question is, so why orbit? Well, because it is similar to React and other hash DHT, distributed hash table, no SQL database systems. So another one is, it uses peer-to-peek connections. And the third reason, well, it's very small and it's easy to understand. So we're starting to go into deep details of complicated programs. So we have this non-distributed Erlang and this is one node. We have one node and we have this master worker table and credit. So what happens is that we have a hash table and master process. It spawns the first worker process. The worker process has just one number. And then we apply some generators on it and it generates a list of numbers. So what we do with this first number? With this first number, we check whether it's already in this table. If the name is in the table, the number is in the table, then it just returns the credit to the master and it dies. If it's not, then it puts the value in the table, generates a new list and we repeat the process again. To do it, it's distributed. What we need to do is we need to a little bit modify master and worker modules. So what happens now is that master and workers, they are on different nodes. And worker nodes, they contain these distributed hash tables. So table is now not on one node, but it becomes distributed. And master, it again spawns a process, for example, to node one. But the node is defined by the value. So what we do is we have this first process. We take a hash value and this hash value defines to which node we are actually spawned the process. For example, we spawn it to node one. And then we check whether the number is already in the hash table. If it's there, then the process just dies and returns the credit to the master. And if it's not, then it generates a new process and this procedure repeats. And now we have Scalable distributed. So why we want to change it? Well, because the problem grows, right? So we can have massive, massive requirement for this orbit. And it's not possible neither for one node, neither for 60 or 100 nodes to solve this problem anymore. So we want it to scale. And for that, we use Scalable distributed error. So it's again, what happens, we have this master node, but now we have this submaster node. But the hash table is still kept on those worker nodes. So master generates the first number and then it decides, it takes the hash value and decides to which node actually it should go first, to which to spawn. For example, we want to go to group one. And then from this group one, when it rhymes there on the first submaster, it takes the second hash value. And from here, it decides to which other node to go to. And then again, a trip it. So here we need to modify master, worker again, and add to new group, to new modules. These are submaster and grouping. Well, here, as you notice probably, the transformation de-factoring is a bit more complicated than it was from non-distributed to distributed Erlang. And for that, a team at Kent University works to simplify the life a little bit. Right. So now master, master process. So what we have here is that instead of in distributed Erlang, that would be a spawning of worker processes. Now master process, it spawns to the submasters. And then on the workers, workers spawn the processes directly to other worker nodes only if it's in the same as script. Otherwise, it would just spawn to the submaster, the master would spawn to another submaster, and then we do the routing. And then submasters now, they, the submaster processes, they share the responsibility of worker processes now because they need to decide, need to root where those worker processes should go. And they also take some responsibility of their master processes because they need to collect the credit and take care of those things. And then grouping. So we need to create a few groups. These are worker groups and we need to create a master group. And this is the result. So this one is, we measured on one, up to 160 Erlang VMs and this is, this was up to almost 1300 cores. And then what we see here. So if you notice, the difference is not that large, the first thing. But orbit, first of all, we need to think about the nature of orbit. So it doesn't use global operations. So what we reduce actually here is only connections. So the only thing that shows here is what effect on orbit, on a particular program that requires all to all connection, has this reduction of all to all connections. And another thing I'd like you to notice is this flip. That Scalable distributed Erlang actually, you know, changed the scalability to the better. So what are the things we work on? Another thing is a semi-exclusive placement. So when we have this structured set of nodes and or some mechanism how to call those nodes, it can be quite simple to remember those nodes. But what if we have a hundred thousands of nodes? Well, for a programmer then explicitly naming every node, it becomes a bit of a problem. So we need something semi-exclusive. And there are a number of reasons why we want to spawn a process and why we would want to spawn a process. So probably we want to put close together processes that communicate with each other quite a lot. So we want them to keep together. Or we have a very large computation and we want to put it somewhere far away. I'm sorry, somewhere far away and just that it would be, for example, executed on a cloud and just a bit reduced load on our own machines. And when we looked at it, well, we thought, well, actually, that looks very much like a tree structure. So we have something like racks in a server, right? And then we have a cluster. And then we have a cloud. And we can think of a method how we can spawn them close and far away, depending on those distances. And here we use something defined, developed in math. Well, the math, it looks a bit scary, but it's not actually. And I'll explain it. So what we actually look here is the distance, the common distance between nodes, those paths, those paths that share, nodes share between each other. So node B and C shares two paths. And it means the distance between them is one fourth. Whereas node B and G shares only one, it means the distance is a half. And then we have one. So it means that the larger this value, the further nodes are actually located from each other. And we can use this to actually define where we want to spawn our process. We did some measurement. And these measurements made, we used our Beowulf cluster located at Heretwood University in Edinburgh. So there are actually a few clusters over here. And what we found is, actually, yes, we can observe this tree structure when looking at the communication. Just a little bit to understand. So these are names of our Beowulf nodes. And these are the communication time. So we can look something like to, but I don't remember in which time it measured. Never mind. I'm sorry. I'll check that. So it just shows how long it takes for processes between two nodes to communicate with each other. So how can we use this semics list placement? Well, we can spawn nodes to using some sort of parameters. And for that, we introduce function choose nodes. And it uses a list. And this list is introduced on purpose because we want to extend the ways how we want to spawn those processes. So, for example, we can use an sgroup name. And for example, we have a node that belong to group one. And we have nodes that belong to node group. And node three can say, well, I want to spawn processes only to the nodes that belong to both two sgroups or to one just particular group. Another thing that we can do is we can use attributes. So each node, it can have attributes. What these attributes can be? Well, that can be software. This particular node has software hardware on which this particular node runs. Or some other, I don't know, modules, something about description of this node that we want to exploit. So we have sgroup, we have attributes, and then we have use distances. So for example, nearer than 0.4, 0.4, or between 0.5 and 0.7. So something like this notion of distance. This is a bit complicated, for example, to put in a programmer to know all these distances. So what we want to introduce later is to make it more automatic. So introduce something like nearest, furthest, a bit further, very close, something like that. So more intuitive rather than forcing people to remember actually those numbers and exploit the system. And this actually, well, the semic-explicit placement is also good for portability. Why we want it? Because, well, I suppose I don't need to explain that we develop very often programs in one hardware, on one hardware architecture, and then port it to another hardware architecture. So this portability is really important. And this nearest, closest probably will be a solution, or at least a help to resolve those problems. Another thing that we work is the semantics. So why do we need semantics? We want to reason about that. We want to explain, we want to understand what our system does. We want to know what sort of properties it has. And for that, we need some formal semantics. And we introduce semantics only for SDE alone. And these are 16 functions. So something like 9, I think, 9 of these functions, they actually change the state. And other functions, they don't change the state. They return on the value. And so what do I mean by that? So for example, if we want to reduce, to create a newest group, it means that we change the state, actually. But we want, when we just want information about a nearest group, we return a value, but the state, nothing happens, we just collect it information. And this is how our semantics looks like. So we have set of groups, set of free groups, free hidden groups, and set of nodes. And this defines the state. And groups, all of them, they're associated with a set of nodes. So for hidden node, there's just one node. And for groups and free groups, there's a set of nodes. And all of these groups, they have a namespace. But as groups, they also have a name. And here, the first property, and it's quite obvious, but it's nice to see it from the semantics, that a node can belong to only one of those groups. So it can be free hidden, free normal, or belong to a nearest group. And this is how our semantics looks like. So it means that we have initial state, we apply a command on node n i, and then the state changes and it returns a volume. And this is a small example. This is the simplest one. So we have s group and we register a name. So we register a name in s group, it has name, it has speed, and it's just yes or no. So yes, registered, no, no, sorry. And we have initial state groups, free groups, free hidden groups, and nodes. We register a name in s group, name, speed, or node n i. And then we check if actually this node n i belong to this s group. And then we check if this actually node or speed are already registered. And if all these conditions satisfy, then we just register the name. And we modify the state and return true. Otherwise, return false. Well, all this is very nice and mathematics is nice. But how do we know that we actually can trust this? Well, for this, we use a quick check. And what it does is we have our s group command and then we go into two directions around it. So the first directions is we actually execute this command in the real system. So we take the command and we get a new state. And we also go through a different direction. This we take operational semantics, calculate it using our mathematics, and then get a new abstract state. And for that, here we need precondition. So precondition for this particular command is that set of nodes in the s group, it can't be empty. So s group can't be empty. And we have a post condition. And post condition is that actually abstract states, they should be the same. They should be identical. And that was really, really useful thing to do and to have. Because first of all, while writing maths, we identified quite a number of errors and misunderstandings and miscoding in the implementation. And then while running this quick check, we saw some problems with maths and again with implementation. So it was really helpful, really useful and we'll hope to move forward with that. So future work, first of all, we have quite a lot of plans with semi-explicit placement. That is, as I said already, we want to change how we think and how we apply those distances. So instead of numbers, giving actually some meaningful names to programmers. Another thing is we, oh right. So another thing is we want to look at robustness. So what happens if node fails, how the system knows about it? And of course, automation. Because just now, all this information is collected just, you know, it's hard to code it, it's just in configuration. And what we want, we want the system to provide us this information. And of course, running testing the system. So first of all, look at how run Simdiaska, our benchmark, this is an engine, run it and see how it works. And it's quite a large one. It's EDF and it's on the blue gene and it's something like 65,000 of course. Also, we want this SD Erlang to become standard. We want to improve it. We want to work on it and we want it to be included in the OTP. Of course, continue to work on SD Erlang. We found it's really useful, really just how we think, how we understand the system and the methodology. So how we actually go from distributed or non-distributed application to SD Erlang application, how we refactor it and what we need to do then. So these are some links. If you find them interesting, a bit about the release project, the OTP, the modification, I'm sorry, to the OTP. These are benchmarks that we use all here. Perceptu is a tool we work on in the release project. Then Bench. This benchmark is more if you're interested in scalability of VM because in the release project we also work on the VM level and it's already in the releases. So if you want to see how differently releases change scalability, that will be quite interesting tool to play with. And Simdiaska, if you're interested in the simulation engine that we are going to use, the information is there. And thank you. So feedback, questions, very welcome. Thanks. Any questions? Right, yeah. You mean SD Erlang, right? Yeah. Well, in the early game of the slide deck, you had some parent-bed of nodes that were obviously preferences of things. And what are they in this list? I know. I think this is just links. But if you let me know what sort of information you're interested in, I'll be happy to give links and put them into slides and just read whatever. Right. Well, great. Thank you. Thank you for coming and thank you for listening. Thank you. Thank you. Thank you. Thank you.
 Hi, I'm Natalia and thank you for coming. Thank you for watching. And I suppose we'll  start, right? So I'm Natalia, I'm from the University of Glasgow and today I will talk  about the release project, the work we do there. So the project is European and it involves  something like eight partners. These are five academic partners from the UK, Sweden and  Greece and three industrial partners. These are EDF, Ericsson and Erlang Solutions. In  particular I will talk about the work we do at Glasgow University together with Kent  and Harriet Ward Universities. So I will give a brief introduction of things we do at what  the release project aims in general. Then I will talk about the scalability limitations  of distributed Erlang and then we'll cover what we do to scale it. So I'm not aiming  to give all details but just a flavor of what we do, why we do that and how. So the  purpose, the aim of the release project is to scale the radical axiomodel to build reliable  software on massively problem machines. And by that we mean something like 10 to 5 cores.  And for that we use Erlang. We work on three levels. So the first level is VM, then we  have language level and tools. Today I will talk only about the language level and VMs  and tools will be covered sometimes by different people in different talks. So typical hardware  architecture, what do we mean by that? Well, we're thinking about a cloud and actually  a collection of clouds, something like it's called Sky. And then clouds consist of hosts  and then hosts, they consist of SMP modules, something like 6, 8 and each cell would have  something like 100 of those hosts and each SMP would contain 32, 64 cores, something  like that. And then Erlang, just briefly reminding you what it is. So Erlang is a functional  general programming, general purpose programming language designed at Ericsson. It is dynamically  typed and it was specifically designed to build distributed, scalable, software time  and massively concurrent software. And then it uses these two main philosophies, which  is let it crash and the second one is share nothing. The language primitives are processes  and the particular very nice thing about it is that concurrency is built in in the language  rather than edit on. So distributed Erlang, what's it about? Well, we have a number of  nodes and those nodes, they can be right. And those nodes, they can be either on the  same host, they can be on different hosts and these yellows are nodes by the way. Well,  by node I mean Erlang VM. And then inside of those, we have those processes and each  node, it can share a number of cores. When Erlang works in its normal mode, it means the  connections are transitive. So all nodes are interconnected and they have a global common  namespace. But that I mean that if we register a name on one node globally, I mean, then  it will be transitively connected, transitively shared between all nodes. And if we add another  node over here, it immediately as soon as we connect to one of the nodes, it will be  connected to the rest of the nodes. And then it's about explicit placement. So to spawn  a process to a different node, we need specifically define which node that is. And of course,  Erlang and distributed Erlang is about reliability. So if one process, one node crushes, we can  detect it and restart the system. So here we have a benchmark, it's a bash a bench code.  So what we do is, oh my God, so we have 20 nodes, then within 18 seconds, we just kill  all those process. And the system again is in a stable state. Then we start all those  nodes again and system in a stable state again. But it has limitations. So we run this experiment  and it's again bash a bench. Well, it's a modification of bash a bench, we call it D e  bench. So what we do here is that we have a scale number of nodes, so from 10 to 100.  And then we look at the throughput. And then we vary the number of the percentage of global  operations that we have. So the red one is zero global operations. But then as soon as  we start, you know, even this tiny, tiny number percentage of global operations, the scalability  goes down. And that is when we used global module, but global operations are not only  about global module. You can also use something like RPC column multiple nodes and then you  will get again global operations. Another problem is with single process bottlenecks.  And we can get here. So this is React benchmark. React is a DBMS system built in the Eulang.  And so we look at scalability again. And here we have number of nodes. This is number of  VNs, Eulang VNs. And then we scale them from 10 to 100. And then we look at throughput  again. And what happens is that we have quite nice scalability up to 50, 60 nodes. And then  it stops. And then it goes down with huge variation. And by the way, this is React version  11. So it was something like a year, half a year ago, one half a year ago. So they updated,  they modified some things there. But the problem there is that first we have this single process  bottlenecks. And another problem is that this fully connected network that we get with distributed  Eulang. So design approach and principles. So we need to scale. To scale a system, to  scale a program, we need to have a scalable persistent storage. And that we left to React  and Xandrum. And then we need in memory data structure. And that's in Eulang, these are  at stables. And this work is done by Sally University and Ericsson within the release  project. And then we need a scalable computation. And that's our work. When thinking about how  we want to scale it, we took as a baseline those principles. So first of all, we wanted  to work at Eulang level as far as possible. Then we wanted to do those modification minimal  because language is nice. Language is simple. And people love the language. And we didn't  want to introduce something absolutely foreign that the Eulang community would just reject  and say, no, we don't like it. It's not our philosophy. Of course, we wanted to follow  the language philosophy. And then we wanted to keep the Eulang reliability model unchanged  and as reliable as it is now. And then there are two things that we want to change. There's  a avoid global sharing, as showed on the previous slide, and introduce an abstract notion of  communication architecture. So what is SD Eulang? SD Eulang is a small, modest extension  to distributed Eulang. And here we tackle two problems. The first one is network scalability.  We want to use all to all connections. And the second one is semi-explicit placement.  And I will talk about those two problems in details now. So network scalability. And  scalable distributed Eulang, we have two types of nodes. These are free nodes. These are  nodes that belong to node S groups. And another one is S groups nodes that belong to at least  one S group. And so nodes in S groups, they have transitive connections. And they, so  nodes within the S group, they have transitive connections within the group and non-transitive  connections with the other nodes. And then if you're familiar with distributed Eulang,  you can say, well, there are similarities with global groups. Yes, there are. And these  are similarities that groups have their own namespace and they have transitive connection  within the groups. But the difference is that the groups are overlapping. And so we don't  partition them as global groups. And then the information about S groups and node is  not globally shared. So let's look what happens here. So we have distributed Eulang and assume  that we have two groups of nodes. So these three nodes are connected and these three  nodes are also connected. Then we decide to connect nodes three and four. And what we  get is a fully connected network because of the transitive connectivity. And then we look  at S groups. So we group nodes. And when we connect nodes three and four, they're just  connected. Connections are non-transitive and each of the groups has its own namespace.  The types of connections in general, what we have. So the thing is that we don't aim  to replace the existing connections and existing nodes in distributed Eulang by those groups.  So if a programmer doesn't want to use them, that's fine. But if they want to use them,  these are means to introduce them and communicate with other nodes. So first of all, we have  these three nodes and these are free normal nodes that have transitive connections, common  namespace, the connections are normal. And then we have hidden nodes. Those hidden nodes,  they each node has its own namespace. So if we register something globally, it doesn't  actually change much because this information is not going to be shared between other nodes.  And also, they don't share connections. So they just form direct connection and don't  share them with anything. Then we have S groups. So nodes are interconnected within  S groups and they have transitive connections within that S group and have non-transitive  connections, the rest. And here we have the same.  So another question that we thought of, of course, and we often ask, so why do we use  S groups? Why group nodes into those S groups? Well, before introducing them, we had some  requirements. We wanted those S, we wanted nodes to join and leave us groups very easily  and quickly. Then we wanted to keep the philosophy of the language. That is, each node can be  connected to any other node. And then we wanted that to be simple. And we looked at three  other mechanisms, so something like grouping nodes according to their hash value. But that  imposed some restriction on, so if a node joins an S group, then it should change its  hash value. And when it leaves an S group, it should also change its hash value. And  it belongs to multiple S groups. There's again this contradiction of the hash value. So we  decided to avoid that. And then another approach would be a hierarchical structure. But again,  it contradicts the philosophy because in this case, node from different levels, they can't  communicate directly to each other because it's just, so we decided to introduce overlapping  S groups. And using these overlapping S groups, we can implement actually all other three  approaches. So this, how can we group nodes? We have a set of nodes and then nodes are  grouped into S groups. And one of the nodes, we can call them gateways, for example. They  form another subgroups. And then gateways to those upper levels, they can form another  S group. So if node A wants to communicate to node N, what happens is node A communicates  to B, B communicates to D, E, F, and then it's N. So nodes actually don't form a direct  connection to each other, but go through this gateways. We can also communicate, introduce  something more complicated and introduce redundancy in the hierarchy. So instead of having one  gateway node, we can have two in case one of them will fail. And then we can also communicate  with free nodes and we can either connect them or don't connect. And they will have  non-transitive direct connection to the nodes in the S groups.  And then another approach. Well, we can have embedded. If we don't mind nodes to be interconnected,  but we want to reduce this global space, we want to localize it somehow. Well, we can  introduce embedded groups where nodes, they just share connection within their groups.  So we can register names here and register names here. Then all names in this group will  have that name or have all of those nodes. And how we can use those. So we can either  start all that configuration at start. And in this case, we will use configuration file  or we can start dynamically using this new S group function. And we can use it either  with one parameter, defining the S group name or not defining it. In this case, the S group  name will be generated automatically. And this is done because we don't guarantee the  S group's names to be unique. So we don't collect this information anywhere. And in this case,  what we do is that what we guarantee is uniqueness, high probability of name uniqueness. So not  uniqueness but high probability of its uniqueness. And other function we can use. So creating  new S groups, deleting S groups at nodes and removing nodes. We can also collect some sort  of information that we want. So information about S groups, information about the particular  nodes to which nodes it belongs to, to which, sorry, S groups it belongs to, to which, with  which node it shares namespace and so on. And these are the first measurements that  we did with this. So this is, sorry, what's that? Good, yes. That's good, it works. Yes.  So we have this tiny percent, 0.01 percent of global operations. And we compare distributed  Erlang and SD Erlang. And we have from 10 to 100 nodes. And we look at the throughput.  So SD Erlang helps. That's a good news. And so to look how actually SD Erlang works, we'll  look at an example of orbit. So what is orbit? Obit is a symbolic computational kernel,  a generalization of transitive closure computation. So what it is? We have initially given space  and then we apply a list of generators on this namespace. And from this, we generate a new  list of, create a new list of numbers. And this process is repeated until no new numbers  is generated. So that's the program. And we'll look how we can, how starting with non-distributed  Erlang, we can come to scalable distributed Erlang, how can we build this program? Right.  So the first question is, so why orbit? Well, because it is similar to React and other hash  DHT, distributed hash table, no SQL database systems. So another one is, it uses peer-to-peek  connections. And the third reason, well, it's very small and it's easy to understand. So  we're starting to go into deep details of complicated programs. So we have this non-distributed  Erlang and this is one node. We have one node and we have this master worker table and credit.  So what happens is that we have a hash table and master process. It spawns the first worker  process. The worker process has just one number. And then we apply some generators on it and  it generates a list of numbers. So what we do with this first number? With this first  number, we check whether it's already in this table. If the name is in the table, the number  is in the table, then it just returns the credit to the master and it dies. If it's not, then  it puts the value in the table, generates a new list and we repeat the process again.  To do it, it's distributed. What we need to do is we need to a little bit modify master  and worker modules. So what happens now is that master and workers, they are on different  nodes. And worker nodes, they contain these distributed hash tables. So table is now not  on one node, but it becomes distributed. And master, it again spawns a process, for example,  to node one. But the node is defined by the value. So what we do is we have this first  process. We take a hash value and this hash value defines to which node we are actually  spawned the process. For example, we spawn it to node one. And then we check whether  the number is already in the hash table. If it's there, then the process just dies and  returns the credit to the master. And if it's not, then it generates a new process and this  procedure repeats. And now we have Scalable distributed. So why we want to change it?  Well, because the problem grows, right? So we can have massive, massive requirement for  this orbit. And it's not possible neither for one node, neither for 60 or 100 nodes to  solve this problem anymore. So we want it to scale. And for that, we use Scalable distributed  error. So it's again, what happens, we have this master node, but now we have this submaster  node. But the hash table is still kept on those worker nodes. So master generates the  first number and then it decides, it takes the hash value and decides to which node actually  it should go first, to which to spawn. For example, we want to go to group one. And then  from this group one, when it rhymes there on the first submaster, it takes the second  hash value. And from here, it decides to which other node to go to. And then again, a trip  it. So here we need to modify master, worker again, and add to new group, to new modules.  These are submaster and grouping. Well, here, as you notice probably, the transformation  de-factoring is a bit more complicated than it was from non-distributed to distributed  Erlang. And for that, a team at Kent University works to simplify the life a little bit. Right.  So now master, master process. So what we have here is that instead of in distributed  Erlang, that would be a spawning of worker processes. Now master process, it spawns to  the submasters. And then on the workers, workers spawn the processes directly to other  worker nodes only if it's in the same as script. Otherwise, it would just spawn to the submaster,  the master would spawn to another submaster, and then we do the routing. And then submasters  now, they, the submaster processes, they share the responsibility of worker processes now  because they need to decide, need to root where those worker processes should go. And  they also take some responsibility of their master processes because they need to collect  the credit and take care of those things. And then grouping. So we need to create a  few groups. These are worker groups and we need to create a master group. And this is  the result. So this one is, we measured on one, up to 160 Erlang VMs and this is, this  was up to almost 1300 cores. And then what we see here. So if you notice, the difference  is not that large, the first thing. But orbit, first of all, we need to think about the nature  of orbit. So it doesn't use global operations. So what we reduce actually here is only connections.  So the only thing that shows here is what effect on orbit, on a particular program that  requires all to all connection, has this reduction of all to all connections. And another thing  I'd like you to notice is this flip. That Scalable distributed Erlang actually, you know,  changed the scalability to the better. So what are the things we work on? Another thing  is a semi-exclusive placement. So when we have this structured set of nodes and or some  mechanism how to call those nodes, it can be quite simple to remember those nodes. But  what if we have a hundred thousands of nodes? Well, for a programmer then explicitly naming  every node, it becomes a bit of a problem. So we need something semi-exclusive. And there  are a number of reasons why we want to spawn a process and why we would want to spawn a  process. So probably we want to put close together processes that communicate with each  other quite a lot. So we want them to keep together. Or we have a very large computation  and we want to put it somewhere far away. I'm sorry, somewhere far away and just that  it would be, for example, executed on a cloud and just a bit reduced load on our own machines.  And when we looked at it, well, we thought, well, actually, that looks very much like  a tree structure. So we have something like racks in a server, right? And then we have  a cluster. And then we have a cloud. And we can think of a method how we can spawn them  close and far away, depending on those distances. And here we use something defined, developed  in math. Well, the math, it looks a bit scary, but it's not actually. And I'll explain it.  So what we actually look here is the distance, the common distance between nodes, those paths,  those paths that share, nodes share between each other. So node B and C shares two paths.  And it means the distance between them is one fourth. Whereas node B and G shares only  one, it means the distance is a half. And then we have one. So it means that the larger  this value, the further nodes are actually located from each other. And we can use this  to actually define where we want to spawn our process. We did some measurement. And these  measurements made, we used our Beowulf cluster located at Heretwood University in Edinburgh.  So there are actually a few clusters over here. And what we found is, actually, yes, we can  observe this tree structure when looking at the communication. Just a little bit to understand.  So these are names of our Beowulf nodes. And these are the communication time. So we can  look something like to, but I don't remember in which time it measured. Never mind. I'm  sorry. I'll check that. So it just shows how long it takes for processes between two nodes  to communicate with each other. So how can we use this semics list placement? Well, we  can spawn nodes to using some sort of parameters. And for that, we introduce function choose  nodes. And it uses a list. And this list is introduced on purpose because we want to extend  the ways how we want to spawn those processes. So, for example, we can use an sgroup name.  And for example, we have a node that belong to group one. And we have nodes that belong  to node group. And node three can say, well, I want to spawn processes only to the nodes  that belong to both two sgroups or to one just particular group. Another thing that  we can do is we can use attributes. So each node, it can have attributes. What these attributes  can be? Well, that can be software. This particular node has software hardware on which this particular  node runs. Or some other, I don't know, modules, something about description of this node that  we want to exploit. So we have sgroup, we have attributes, and then we have use distances.  So for example, nearer than 0.4, 0.4, or between 0.5 and 0.7. So something like this notion  of distance. This is a bit complicated, for example, to put in a programmer to know all  these distances. So what we want to introduce later is to make it more automatic. So introduce  something like nearest, furthest, a bit further, very close, something like that. So more intuitive  rather than forcing people to remember actually those numbers and exploit the system. And  this actually, well, the semic-explicit placement is also good for portability. Why we want  it? Because, well, I suppose I don't need to explain that we develop very often programs  in one hardware, on one hardware architecture, and then port it to another hardware architecture.  So this portability is really important. And this nearest, closest probably will be a solution,  or at least a help to resolve those problems. Another thing that we work is the semantics.  So why do we need semantics? We want to reason about that. We want to explain, we want to  understand what our system does. We want to know what sort of properties it has. And for  that, we need some formal semantics. And we introduce semantics only for SDE alone. And  these are 16 functions. So something like 9, I think, 9 of these functions, they actually  change the state. And other functions, they don't change the state. They return on the  value. And so what do I mean by that? So for example, if we want to reduce, to create a  newest group, it means that we change the state, actually. But we want, when we just want information  about a nearest group, we return a value, but the state, nothing happens, we just collect  it information. And this is how our semantics looks like. So we have set of groups, set of  free groups, free hidden groups, and set of nodes. And this defines the state. And groups,  all of them, they're associated with a set of nodes. So for hidden node, there's just  one node. And for groups and free groups, there's a set of nodes. And all of these groups,  they have a namespace. But as groups, they also have a name. And here, the first property,  and it's quite obvious, but it's nice to see it from the semantics, that a node can belong  to only one of those groups. So it can be free hidden, free normal, or belong to a nearest  group. And this is how our semantics looks like. So it means that we have initial state,  we apply a command on node n i, and then the state changes and it returns a volume. And  this is a small example. This is the simplest one. So we have s group and we register a  name. So we register a name in s group, it has name, it has speed, and it's just yes  or no. So yes, registered, no, no, sorry. And we have initial state groups, free groups,  free hidden groups, and nodes. We register a name in s group, name, speed, or node n  i. And then we check if actually this node n i belong to this s group. And then we check  if this actually node or speed are already registered. And if all these conditions satisfy,  then we just register the name. And we modify the state and return true. Otherwise, return  false. Well, all this is very nice and mathematics is nice. But how do we know that we actually  can trust this? Well, for this, we use a quick check. And what it does is we have our s group  command and then we go into two directions around it. So the first directions is we actually  execute this command in the real system. So we take the command and we get a new state.  And we also go through a different direction. This we take operational semantics, calculate  it using our mathematics, and then get a new abstract state. And for that, here we need  precondition. So precondition for this particular command is that set of nodes in the s group,  it can't be empty. So s group can't be empty. And we have a post condition. And post condition  is that actually abstract states, they should be the same. They should be identical. And  that was really, really useful thing to do and to have. Because first of all, while writing  maths, we identified quite a number of errors and misunderstandings and miscoding in the  implementation. And then while running this quick check, we saw some problems with maths  and again with implementation. So it was really helpful, really useful and we'll hope to move  forward with that. So future work, first of all, we have quite a lot of plans with semi-explicit  placement. That is, as I said already, we want to change how we think and how we apply  those distances. So instead of numbers, giving actually some meaningful names to programmers.  Another thing is we, oh right. So another thing is we want to look at robustness. So  what happens if node fails, how the system knows about it? And of course, automation.  Because just now, all this information is collected just, you know, it's hard to code  it, it's just in configuration. And what we want, we want the system to provide us this  information. And of course, running testing the system. So first of all, look at how run  Simdiaska, our benchmark, this is an engine, run it and see how it works. And it's quite  a large one. It's EDF and it's on the blue gene and it's something like 65,000 of course.  Also, we want this SD Erlang to become standard. We want to improve it. We want to work on  it and we want it to be included in the OTP. Of course, continue to work on SD Erlang.  We found it's really useful, really just how we think, how we understand the system and  the methodology. So how we actually go from distributed or non-distributed application  to SD Erlang application, how we refactor it and what we need to do then. So these are  some links. If you find them interesting, a bit about the release project, the OTP, the  modification, I'm sorry, to the OTP. These are benchmarks that we use all here. Perceptu  is a tool we work on in the release project. Then Bench. This benchmark is more if you're  interested in scalability of VM because in the release project we also work on the VM  level and it's already in the releases. So if you want to see how differently releases  change scalability, that will be quite interesting tool to play with. And Simdiaska, if you're  interested in the simulation engine that we are going to use, the information is there.  And thank you. So feedback, questions, very welcome. Thanks. Any questions? Right, yeah.  You mean SD Erlang, right? Yeah. Well, in the early game of the slide deck, you had some  parent-bed of nodes that were obviously preferences of things. And what are they in this list?  I know. I think this is just links. But if you let me know what sort of information you're  interested in, I'll be happy to give links and put them into slides and just read whatever.  Right. Well, great. Thank you. Thank you for coming and thank you for listening.  Thank you.  Thank you.  Thank you.  Thank you.
Natalia_Chechina_-_RELEASE_Scalable_Erlang_0_1175.png Natalia_Chechina_-_RELEASE_Scalable_Erlang_1176_1799.png Natalia_Chechina_-_RELEASE_Scalable_Erlang_1800_2771.png Natalia_Chechina_-_RELEASE_Scalable_Erlang_2772_3587.png Natalia_Chechina_-_RELEASE_Scalable_Erlang_3588_4823.png Natalia_Chechina_-_RELEASE_Scalable_Erlang_4824_5003.png Natalia_Chechina_-_RELEASE_Scalable_Erlang_5004_6683.png Natalia_Chechina_-_RELEASE_Scalable_Erlang_6684_7187.png Natalia_Chechina_-_RELEASE_Scalable_Erlang_7188_7595.png Natalia_Chechina_-_RELEASE_Scalable_Erlang_7596_9083.png Natalia_Chechina_-_RELEASE_Scalable_Erlang_9084_10703.png Natalia_Chechina_-_RELEASE_Scalable_Erlang_10704_13139.png Natalia_Chechina_-_RELEASE_Scalable_Erlang_13140_13739.png Natalia_Chechina_-_RELEASE_Scalable_Erlang_13740_15767.png Natalia_Chechina_-_RELEASE_Scalable_Erlang_15768_16655.png Natalia_Chechina_-_RELEASE_Scalable_Erlang_16656_18647.png Natalia_Chechina_-_RELEASE_Scalable_Erlang_18648_20819.png Natalia_Chechina_-_RELEASE_Scalable_Erlang_20820_21731.png Natalia_Chechina_-_RELEASE_Scalable_Erlang_21732_22487.png Natalia_Chechina_-_RELEASE_Scalable_Erlang_22488_23327.png Natalia_Chechina_-_RELEASE_Scalable_Erlang_23328_25247.png Natalia_Chechina_-_RELEASE_Scalable_Erlang_25248_26075.png Natalia_Chechina_-_RELEASE_Scalable_Erlang_26076_28211.png Natalia_Chechina_-_RELEASE_Scalable_Erlang_28212_29423.png Natalia_Chechina_-_RELEASE_Scalable_Erlang_29424_31031.png Natalia_Chechina_-_RELEASE_Scalable_Erlang_31032_31679.png Natalia_Chechina_-_RELEASE_Scalable_Erlang_31680_32771.png Natalia_Chechina_-_RELEASE_Scalable_Erlang_32772_33131.png Natalia_Chechina_-_RELEASE_Scalable_Erlang_33132_33491.png Natalia_Chechina_-_RELEASE_Scalable_Erlang_33492_33959.png Natalia_Chechina_-_RELEASE_Scalable_Erlang_33960_34451.png Natalia_Chechina_-_RELEASE_Scalable_Erlang_34452_35183.png Natalia_Chechina_-_RELEASE_Scalable_Erlang_35184_35303.png Natalia_Chechina_-_RELEASE_Scalable_Erlang_35304_35411.png Natalia_Chechina_-_RELEASE_Scalable_Erlang_35412_39059.png Natalia_Chechina_-_RELEASE_Scalable_Erlang_39060_39395.png Natalia_Chechina_-_RELEASE_Scalable_Erlang_39396_39695.png Natalia_Chechina_-_RELEASE_Scalable_Erlang_39696_39743.png Natalia_Chechina_-_RELEASE_Scalable_Erlang_39744_40511.png Natalia_Chechina_-_RELEASE_Scalable_Erlang_40512_41759.png Natalia_Chechina_-_RELEASE_Scalable_Erlang_41760_45047.png Natalia_Chechina_-_RELEASE_Scalable_Erlang_45048_46415.png Natalia_Chechina_-_RELEASE_Scalable_Erlang_46416_47735.png Natalia_Chechina_-_RELEASE_Scalable_Erlang_47736_48071.png Natalia_Chechina_-_RELEASE_Scalable_Erlang_48072_49307.png Natalia_Chechina_-_RELEASE_Scalable_Erlang_49308_50171.png Natalia_Chechina_-_RELEASE_Scalable_Erlang_50172_50543.png Natalia_Chechina_-_RELEASE_Scalable_Erlang_50544_51551.png Natalia_Chechina_-_RELEASE_Scalable_Erlang_51552_52907.png Natalia_Chechina_-_RELEASE_Scalable_Erlang_52908_54359.png Natalia_Chechina_-_RELEASE_Scalable_Erlang_54360_55499.png Natalia_Chechina_-_RELEASE_Scalable_Erlang_55500_56267.png Natalia_Chechina_-_RELEASE_Scalable_Erlang_56268_59843.png Natalia_Chechina_-_RELEASE_Scalable_Erlang_59844_59891.png Natalia_Chechina_-_RELEASE_Scalable_Erlang_59892_59951.png Natalia_Chechina_-_RELEASE_Scalable_Erlang_59952_60023.png Natalia_Chechina_-_RELEASE_Scalable_Erlang_60024_60083.png Natalia_Chechina_-_RELEASE_Scalable_Erlang_60084_60131.png Natalia_Chechina_-_RELEASE_Scalable_Erlang_60132_60191.png Natalia_Chechina_-_RELEASE_Scalable_Erlang_60192_60275.png Natalia_Chechina_-_RELEASE_Scalable_Erlang_60276_60347.png Natalia_Chechina_-_RELEASE_Scalable_Erlang_60348_60395.png Natalia_Chechina_-_RELEASE_Scalable_Erlang_60396_60587.png Natalia_Chechina_-_RELEASE_Scalable_Erlang_60588_60647.png Natalia_Chechina_-_RELEASE_Scalable_Erlang_60648_61427.png