sfscala.org: Tristan Penman, Distributed Hash Table in Scala and Akka
Recording: sfscala.org: Tristan Penman, Distributed Hash Table in Scala and Akka
good evening everybody I start off by introducing myself my name is Tristan I'm actually a relatively new arrival to the San Francisco area thank you it's been great so far I've actually been to maybe five meetups and each one has been amazingly cated I haven't paid for a dinner and maybe a week so no complaints there keep up the good work I also feel obligated to let you know that I started programming in Scala probably I'd say seven or eight months ago so I wouldn't be surprised if there's a few of you out there who might spot errors or mistakes and things that I've done tonight I'd appreciate it if you didn't just call it out and say hey you got that wrong because that'll embarrass me a little bit but I'm going to be hanging around after the talk for a little bit so if you've got any tips at the end of this feel free to share them now I'll start by telling you a bit about what we're going to talk about tonight so first and foremost I am going to start talking about a concept in computer science called a distributed hash table a distributed hash table is really just a way of taking an ordinary hash table and spreading it across multiple nodes so we'll dig into that a little bit i'm also going to talk about something called the cord protocol so the cord protocol is just one way to implement a distributed hash table it's one that I learned about years and years back and thought would be interesting for this exercise and we'll dig in a little bit into how the protocols so how that protocol works just in terms of its its algorithms then we'll jump across to a demo app so you know what's motivated this is that I have actually been writing code figuring out how all this works and trying to take best practices patterns and lessons from that that I could share with the scholar community once we've seen the demo how it works may have a bit of an intuition for cord and how it works to implement a distributed hash table will actually look at how you can take a protocol like cord and actually model it as an actor system then we'll look at some arca patterns so just ask and pipe a lot of you are probably familiar with them but I thought they were worth sharing because for a long time I wasn't and then we'll finish up with some closing remarks few useful resources so let's jump into it the first thing I wanted to do here was to do a quick archive refresher last time I delivered this talk not everyone in the room was really familiar with our car so I really just wanted to introduce two core concepts first is that you know our crew is a framework for building concurrent distributed highly resilient message driven apps so that's a bit of a mouthful basically what that means is that you have a system that's composed of multiple components that talk to one another by sending messages back and forth what's really nice about this is that it allows each component in that system to be developed and tested independently and as long as you've got a clean concept of what the protocol is or what kinds of messages are flowing backwards and forwards you can then start the reason about the system as a whole that's basically the reason I chose ARCA for this project that's about all I'm going to do for the intro stuff so hopefully that's enough for you to catch everything else as it comes along so let's jump into the computer science ii part of the talk so we're going to talk about distributed hash tables and the cord protocol but we'll start off with something that's even more fundamentals of computer science which is hash tables so this is my one slide reintroduction to hash tables for anyone who hasn't thought about them recently although I'd be surprised if that was the case in San Francisco since pretty much every tech interview seems bring them up at some point so a hash table is basically a data structure that we use to implement an associative array can also be used to build basic sets things like that but what we're really interested in is its performance so when we use a hash table we get look up in insert operations that run in constant time which is really nice it's worth pointing out this constant in average not as a worst-case and basically the way they work is that you have a an array in memory and that array contains a number of buckets and the keys in your hash table map to one of these buckets a bucket can contain more than one key in which case you might have to use some sort of a collision resolution strategy the simplest one would be something like a linked list so this should all be pretty familiar it's going back to your basic computer science but what really gets quite interesting is when you start thinking about what you do when you've got too much data to fit into memory on one node and this is becoming even more relevant now with some of the more modern database architectures that are looking at at the performance problem by storing things entirely in memory instead of on disk instead of going back to data structures like be trees which were the bread and butter of relational data stores we're now starting to look at what happens when we just put it all in RAM so that's where distributed hash tables come in now what makes a distributed hash table particularly interesting is that sure you have to figure out how to distribute this data across the network but more importantly you have to figure out what the network actually looks like because once you add another node you're increasing the likelihood that a node in your system is going to fail and especially in highly dynamic networks where nodes are likely to join in leave the network at any time you need an efficient way to maintain that network overlay all that network topology luckily way way back in the early 2000s a bunch of smart people at MIT came up with the cord protocol so the cord protocol is basically a set of algorithms and a protocol specification for how to maintain that overlay Network it provides one core operation which is to figure out who the owner or which node is the owner of a given key in the network and that's all it really does for us and how it does that is really quite fascinating key features are that that look up operation is logarithmic in the number of nodes so that's a nice quality and network stabilization which is how chord figures out what the network actually looks like is an entirely asynchronous process so it's constantly happening in the background and while that's happening the application layer can still use your overlay Network as though it's completely stable that's worth drawing attention to the fact that core doesn't actually specify how your application layer should work so as I said it gives you a lookup operation but that's not a lookup operation to say get me this key that look up operation is to say tell me who might have this key tell me which node is responsible for this key and that's that allows cord to be completely app application agnostic so you can quite easily use it to implement a distributed hash table but in the same way you could also use it in sensor networks and the whole Internet of Things concept that's become so popular recently it also means that there are some problems in implementing something like a distributed hash table that core doesn't solve for us one of them is redistributing Keys as the network changes so it does actually say what should happen when a node drops out of the network or how we should handle resilience where the replicas are used or we just allow data to to die with a node and that really is stuff that's application dependence so it seems like the right decision so the way the white cord maintains its its network or its mapping from your key space to the nodes in the system relies on this concept called consistent hashing so anyone who's played around with memcache may be familiar with the concept because it's one way of configuring a memcache cluster so that it can handle failure of other node in the cluster what it does is it says given a key space that is consistent so the size of the key space can't change you know so typically we'd say is four byte integer key space so you know roughly four billion possible keys and what we do is we we create this concept of a ring and each key in the key space corresponds to a place on that ring and then the nodes in the network are randomly placed around this ring based on IDs that they have and take ownership for a segment based on where they are so in the case of cord and if you look at this diagram we've got four nodes on this ring these are highlighted in blue and each of these nodes is responsible for keys going the all the way back from its own ID so taking three for example all the way back to fifteen back here and in consistent hashing this mapping wraps around so you can see here that although three wraps back past zero it's still responsible for fifteen over here and then if we wanted to go further we have a node at 14 here just responsible all the way down to 11 so in cord these these nodes that are responsible for all of those nodes they're often called the successor of all of those keys look at that I've already talked a lot about consistent hatching so you can just skip right past that one what gets really interesting is when we look at how the network behaves when we add a node so here we've added a node down here at position 6 on the ring and what this does is it changes the ownership of this segment from four to eight by splitting it in two so now we have two segments one is from four to six which belonged to node 6 and then from seven to eight which belonged to no date so with that that with that ring concept in mind let's revisit the lookup and insert operations in a distributed hash table and think about how they would work so like an ordinary hash table we have a hash function and that's actually what we're using to figure out where on the ring a key should go we actually use the same hash function for both the keys we're storing in our hash table and for coming up with the IDS of the nodes on that ring so going back here see we had a string say it was hello and it mapped to node 4 we're actually using the same hash function perhaps on the IP address of the node that corresponds to node 3 here and that ensures that everything is mapped consistently to the same key space as I mentioned before the lookup operation doesn't tell us whether or not the key is there it just tells us we're in the ring it is or might be and then the application is responsible for further logic the other side of that is that when we have an insert operation we're relying on the application to actually store a data there now this is a really interesting part of cord which is the network stabilization protocol so most of cord if you actually grab the cord MIT paper and have a flip through it is all about how this works and basically what what it does is it defines a set of data that has to be stored with these node first thing is the successor of a given node in the system so going back to our ring again node 14 here has to know about node 3 in order to maintain the network topology and so we would say that node 3 is the successor of 14 similarly we have a predecessor and that just goes the other way one of cords innovations is that it introduces the concept of a finger table and that's kind of like a shortcut across the ring so and this is I guess where the name cord came from is that this finger table allows you to to jump you know at most half way across that entire ring when you're trying to locate a given key in the network and it does this by storing the address of not just the successor but of increasingly far away nodes in the network and the stabilisation process is responsible for maintaining those links and figuring out whether or not they should be updated so what does it actually look like when stabilization happens so what I've got here is an example where we've got a network a node joins and then we've got to think about what what's actually changing so if you imagine we've got cord ring node 3 has just during the network and when we ask that node to join the network we told it that its successor or its seed node was no 10 so it's got a successor pointer pointing to 10 what we need to do is find a way for three to learn its true successor which is eight and it does this by kicking off a stabilization process every couple of milliseconds and what it does is it goes to its current success in which is 10 and says a 10 who's your predecessor now in a stable network the predecessor of 10 will be so when three ask its successor for its predecessor it'll just be three that would mean that's everything stable but in this case it's going to ask ten what's your predecessor and 10 is going to respond well my predecessors eight so 3 is going to look at that and say well that 8 is somewhere between ten and three so i'll use that as my successor instead and that's what's happening here is three is now updated its successor pointer and it sends a notification to eight and says hey 8 i'm now your predecessor and eight updates its pointer accordingly and so you end up with these nice clean links around the ring now the cord paper itself introduces a handful of algorithms that all work together to achieve stabilization but the focus of this talk is just on the core stabilized operation so with that I'm actually going to jump across to a quick demo because a lot of this stuff it's really hard to get your head around so you've actually seen something a little bit more interactive and that's part of the reason why I started coding it is it's all well and good to read the paper but it was actually getting it up and running and seeing how it behaves with a real network or at least a small demo network that gave me a better intuition for how it works just bear with me while I figure out how this works awesome show thank you so just leave that one I click you should save just a second thank you this is important part of the video thank you well what's a demo without a technical failure of some kind screen thank you all right we've got it now hopefully everyone will be able to read that does anyone need me to increase the font size at all oh that's as far as I'm going to go so as I mentioned I've I've written a small demo app to actually explore some of the concepts here it's what motivated the talk and this is all code that I've put up on github project is called cordial it's funny I guess and I encourage you to go and have a look at the code I also encourage you to star it because I really like getting stars on my github repos you can do that now if you like what you'll find if you go to the cordial repo is small projects and it's basically two parts one is a core which is the actual chord part of my implementation so if we dig it down into this so it's basically just a handful of actor classes I've got an actor class for each of the algorithms that I've implemented in the cord protocol I've got one for a node then also got one called pointers which I'll talk a bit about later but i'm not going to jump into code for the demo here because just it'll just be too difficult to follow the other half of this code base is the actual demo sub-project the demo sub-project is just a little web app so if i bring this up i can send kull requests to it that will allow us to see how a small core network would behave I can tell it to create a new node to create a new one using a particular node as a seed and then I can use a web UI to actually see what's happening with that so what I'll do is I'll just get that up and running so this is actually a fun little distraction in this project because it gave me an excuse to see how the spray library worked just want to hadn't played with before and I also got to play with d3 which is nice because I really wanted to have a more visual representation of what was happening in the network so it gives us actually a kind of restful api which is nice so if I do a get request on the nodes collection right now it's empty because this network has no nodes in it but I can also post to it and this will just create a new node and so you'll see what's happened here is a node has jumped into the network and it's all on its own that little line that's jutting out from it is actually a cord from this node to itself because right now it is its own successor and its own predecessor so we can't actually see what what's going on with that link you'll notice here that when we posted to that collection we got back a little chunk of JSON that tells us what it's ideas and its successor and that it's active just good wouldn't want it to be inactive so we can now post another one and we'll tell it to use 12 as its seed so now we end up with another node in the network now what if you were watching really keenly you might have seen that there was that little moment where that node existed before it actually stabilized and right now although it seems like there's one line there's actually two because these nodes see each other as their successes so if we did this get request again let's pipe that through JQ we actually have two nodes so you can see one node has ID 12 the other one has the ID 56 and then they're pointing to one another as their successes I can throw another one in there and will we use 56 as the see this time so starting to see some more interesting things happen but what really gives you a sense for how this works is to insert lots of them really quickly so it's getting a little bit busy but this is actually a chord network stabilizing the whole bunch of fresh nodes they're all talking to one another and figuring out where they fit into the network topology based on their IDs and on this ring so we're actually missing one to make that just perfect so let's get that in there and we'll just wait for this guy to catch up okay that that satisfies the perfectionist in me so hopefully that gives you a bit of an intuition for what's actually happening when when a chord network stabilizes these nodes jump in there they have their successor it might be wrong but then they talk or gossip amongst themselves to figure out where they actually fit in so that brings us to the second part of the talk which is modeling chord using actors hopefully this will be the part that's quite interesting to everybody here because really what we're going to talk about is the actor model and how you can decompose a protocol or a problem light cord into a set of actors that really do justice to the problem and also give you a nice maintainable solution to work with so start by thinking about the actor model and what it means to have something in the actor model what we're doing is for taking a computation in this case stabilization and we're trying to define it in terms of individual components that respond to well defined messages and send well defined messages back to other actors in the system the group of actors is called an actor system all of the message passing happens asynchronously so we don't necessarily have to have one part of the system blocking on another and individual nodes process their messages sequentially so we can start the reason about things in terms of finite state machines and once we do that we can actually start thinking about the whole system in a way that's more rigorous and what would get with a lot of traditional networking solutions so the obvious question to ask when we are trying to design an actor system is what are our actors or who are our actors depending on how you want to phrase it and the case of chord nodes are really the obvious starting point because we've got a cord node it has a set of operations or algorithms that it has to support we we know it has to store network pointers because nodes have to be able to talk to one another so they need to know who to talk to and we also need some timing logic in there because we've got stabilization that happens periodically I've got to make sure it actually happens but we're starting to have a bit of a monolithic solution here we've got all of these conflicting concerns testing this thing would be a nightmare and because this is actually how i started i can tell you the code is just awful when you've got all these competing concerns so what we want to do is we want to decompose the node actor into several different actors each with its own responsibilities ideally each one can be thought about it as a finite state machine can reason about it and really we're looking for a set of properties that should be familiar to us as functional programmers first is determinism we want to know that given a set of inputs over a period of time that the the state of our node is going to be the same if we replay those inputs we want immutability so we don't want to be using them we don't to be using mutable data structures internally we don't want to rely on having to change state in other nodes and we want to be able to use well defined functions to describe what's happening in our node rather than imperatively saying go do this and then in a case like when we're using something like ARCA we're also interested in thread safety because we've got all these different nodes talking to one another and potentially all these actors that are supporting a node we need to make sure that what they're doing is thread-safe and although a lot of us will probably think oh well it's a message passing system it's always going to be thread-safe there are actually exceptions to that so that this is really why I wanted to look at stabilization before is because it gives us such a great case study for how to structure an actor system so we need to perform stabilization periodically and what we want to do is break the problem down in such a way that we achieve certain constraints around how stabilisation should work so let's have a quick look at what stabilization might look like if we were doing it the wrong way so in this case we have we have our node I'm sorry 21 way up the back it's when I prepared this talk the first time and wrote this code was for a much smaller room but what we have is we have a node actor and it starts off with its own successor we have a do stabilization method which just gives us back a future it's basically saying yeah sure I'll stabilize when I feel like it and then we have our received method that actually does all the message processing our sieve method will respond to a stabilized request by calling do stabilization we'll kick all that off and then when it's done we'll send a message back saying everything's good you'll notice down the bottom here that we have a scheduled message that we're sending to self so this is basically the note saying yeah I'll take responsibility for that I'll make sure your stabilization happens and it's hard coded for three seconds it's quite a long time be nice to how to change that but there's something that's really not nice about this and that's that active state should only really ever evolve in response to messages received from the outside so why it's important it's important because when we're testing something we don't want to be relying on intricate timing logic or implicit timing of of an actor or a part of our system to verify that it's correct that makes our test brittle and even worse it makes them hard code so maybe the tests aren't even correct so the solution to a problem like this is to to say okay we better lift that timing logic outside of the actor so one potential solution here is to lift it into the application so what we're doing is the same okay application you've got your node actor and you know how often you want it to stabilize said it a message to do so okay every so often this means that we can now take our our actor that does the stabilization we can test it and we don't have to worry about any of that time in logic we all notice here couple parts that I've highlighted in orange that's because this is still not really that good we still got that mutable variable the counter and we want to be able to increment that counter without using mutable variables so what can we do there well if we follow the next rule or best practice which is that actor state should only ever be mutated with calls to contacts become that basically tells us exactly what to do actually you can see that this is a more concrete example of how that can be that scenario can be misused because although we have a vowel which feels immutable we're actually using a mutable set so doesn't actually solve the problem so what we want to do instead is say that when we're processing a message that needs to be Tate state we should be handling that through the receive closure that we're using as an argument so what we've got here is we've got the is inset argument or parameter that we're passing to active set and every time we receive an ad message we're saying okay well we're going to advance the state of our act using its receive closure by adding a key to that set and then calling that function again sorry so there was one final one that I wanted to cover as well so this will give us three which is that active state should not be allowed to link to leak into a synchronous closures so this is the thread safety issue that I was talking about before this is how it can come up so often when we're working with futures it's really tempting to do things like this where we process a message we kick off something that we want to have have happen asynchronously and then we say well when it's done call me back in this case for calling context I've become with our new config but this is a really dangerous thing to do because although we can call context I've become this future may actually be executing in another thread may not be on the main thread and the main thread is the only place where we should be changing the context of the actor so what we should be doing here is is defining another message type or another case class that we can use to advance that state so when the future finishes it can use its reference to self which is thread-safe and send a message to it saying in this case the content the config has been loaded and then that will be put into the message queue along with everything else and the actor can deal with it when it's ready to so if we take our initial stabilization example you can look at that with all these nice best practices and that looks a little bit nicer because what we're doing now is we are making sure that we only advanced state in response to messages everything's thread safe because we're using a message to encapsulate any state changes or context changes and overall we've made sure that the entire system is testable because we're not actually scheduling stabilization from within this class so to summarize the best practices and it's very small but I should highlight that I didn't come up with these I borrowed them from somebody else's get repo and the link is right down the bottom of the slide here the three are active state should only ever evolve in response to messages received from the outside it should only ever be mutated with the calls the context become so no mutable variables and we should make sure that actor state doesn't leak into asynchronous closures futures so it's digging a bit more into how we decompose the node actor based on all this so first thing we do is we lift stabilization into its own actor because that's a big chunk of logic that we don't need in the node this make sure that everything is deterministic immutable reverentially transparent and thread safe we can also lift the timing logic into its own actor so reason we want to do this is because we don't want the node scheduling its own stabilization and then what that gives us is this rough design for our actor system where we have our node which was our obvious starting place we have our timing logic and we have our stabilization algorithm and then we've also got these two ghost dish nodes so ghost ish classes one is interface and the other is application and that's a reminder that although we've implemented a node it doesn't actually specify how the application should work we still need to think about those things as a separate issue already covered all of that so let's look at what this actually looks like in code and this is where the two are kapat urns that i mentioned come up so first one's asking the second one is pipe and the reason these are useful is because there's a couple of constraints on how stabilization should work in a cord node first one is the only one stabilization request should be in progress at any given time and we need to make sure that stabilization fails if the algorithm exceeds given time out and when it finishes we want to send a notification back to the timing logic actor because that actors should be able to slow speed up its stabilization requests depending on how many failures there are or how long they're taking to complete so a bit of back pressure logic if you like so the answer pattern is pretty nice and pretty simple too so the idea is that instead of telling an actor to do something by sending a message to it you're asking it for a response you still have to send it a message for what you're saying is ok send this message to this actor and now I'm just going to sit here and wait until I get something back but this all happens asynchronously so your application logic and fire and forget and then all of this can be handled as a future because that's what we're actually getting back from our last call what I'd like to call out is this is a lot nicer than a set received timeout and I have to admit I actually use set received time out and wrote a lot of the code involved in doing this manually and probably had it there for something like three months before I realized that there actually better ways of doing this so nobody told me about the ask Fadden and I was delighted when I found it so this is actually what it looks like in code so here's an example of what happens when we want to ask a note for its ID and we want to do that asynchronously we call ask on our load reference we send at the get ID message and we pass in a timeout but this can be implicit we we take the response we get and we map it to this type and then we can actually break that out into other types that inherit from that if something goes wrong we've got this recover clause here which tells us what to do with the exception that may have been thrown as a result of this ask for a complement to that is the pipe pattern because one of the things we often want to do when we've asked for something is then forward the message somewhere else so this fits really well into the request response model that I've used with this cord implementation because when I kick off one of these algorithms I eventually want to send a response back to the client that asks for that algorithm to be executed so we can combine these quite nicely and in the interest of time I'll just jump straight to the code that we get from that this is what it actually looks like when we want to run stabilization asynchronously we handle the stabilized message here and we actually have a reference to our stabilization algorithm which is this guy there we ask that to start so we say start stabilization now and then we only ever expect one response from that algorithm actor it's either going to say it's complete already running or its failed in some way and we map all of the responses into a response that our client can understand and that's where the the pipe to send a part helps us because we've done all the mapping work here and the last thing we want to do is say well when that's done when this feature completes pass that along to our original sender one of the things that's really useful to do here is to use sealed traits so I mentioned that we're mapping to a particular response type and then breaking that down even further but it's quite easy for us to make silly mistakes like leaving out this line here then what happens when the our stabilization algorithm says already running well it's going to be kind of ugly we're going to get something like a classcastexception and then we've got to handle that using the sealed trait ensures that the compiler can actually tell us whether or not our our pattern match is exhaustive and it it's actually quite useful for ide autocomplete functionality as well you don't have to read this whole slide but I wanted to finish up with a really interesting edge case that came up after I'd implemented the stabilization algorithm and started tinkering with how the network behaves under weird scenarios and one of the scenarios that came up with was okay stabilization has started it can take a little while depending on how many steps it has to go through what if while it's stabilizing or while a note is stabilizing I tell it to join a new network reason this is a problem is because although you know the obvious thing to do thing to do is I gave kill the current stabilization algorithm get that out of the way creating new one and reset all of our network pointers doing this in a way that's nice means using the ordinary shutdown process in ARCA which sends messages so what we're doing is we're actually cueing a message telling the stabilization worker to shut shut down and before it processes that message it may still send updates to us plain you know update this pointer to X but if that occurs in a particular sequence we might actually end up with a node that has information that refers to two networks now although the cord protocol automatically resolve this issue and stabilize it does so inefficiently and we might end up with a period of time where the network is kind of flapping between two states as a result of this and it may even fail to join completely it'll still be stable in one of the networks we just don't know which one yeah so I just wanted to throw out a other a potential solution I'm not sure if this is the best approach yet so definitely welcoming feedback here and what I did with this is I actually broke the node actor down step further so what I ended up doing is having a node that kind of acts like a controller and then another actor that is really just the data model for the pointers in the network so what happens is when we create a stabilization algorithm we say here's your model this is the one that you update anytime you need to make changes just don't update in anything in the node itself what we get with that is that when we do a join not only do we kill the old stabilization worker but we also kill the old pointers model and create a new one and this solve I think this solves a problem because it means that even if that old stabilization algorithm still has a couple of updates queued then those updates are only ever going to be sent to the old pointers model the one that we're not actually linked to any more and then those two actors will eventually die when they process air shutdown messages it also has a nice feel to it which is that it gels really well with the MVC idea of the MVC architecture so we're starting to think about a node that encapsulate slaw jek and another one that encapsulates data and some of the benefits of testability that come with that so definitely keen for feedback on that approach after this because still feels overcomplicated feels like maybe somebody else has already thought about this problem and solved a better but I'm just not sure yet so I'll just wrap this up a couple of links to the resources that I've found useful over the last six months or so learning scholar I can really recommend the principles of reactive programming course it's a lot of content in a short period of time and probably not enough time to go into any depth on any of it but it's enough to get you started the functional programming in schuyler book was great especially the exercises the book is not really going to help you unless you actually sit down you go through the exercises and then the Skyler best practices that I mentioned is much more complete than what i just showed here you know I only talked about three tonight the creator of this repo has got these three but dozens more that relate to skyler in general if you're interested in court itself these are the papers consistent hashing it's difficult paper to get through but in taraba Blee worth it if you're interested in that kind of thing and then a couple of the libraries that I found really useful in implementing the demo in particular I like style checking so highly recommend that you Skylar style or something like that on all of your projects because it will pick up on things that that give you a better outcome so with that thanks for listening and thanks for having me at the meet up and I've been told that we have time for two questions yeah it seems like most your actors are finite state machines or brought there's probably like a finance state machine specification for different components in the original papers there's a fsm trait that you can use napa a few up there accusing that yeah so the coolest yeah so the question crossed with an observation is that there is already a finite state machine concept or implement yeah trait in in ARCA why didn't I use that or have I considered using that i have actually considered using it but haven't had the time to really experiment with it as this architecture is design is gradually moved towards being more robust and more finite state machine like it feels more and more urgent that i actually start to experiment with that all right looks like we can wrap it up thank you