Fault Tolerance with Raft: Majority Voting Explained | MIT 6.824

[00:00:00]all right well let's get started today and indeed today and tomorrow I'm gonna talk about raft both because I hope it'll be helpful you for you in implanting though the labs and also because you know it's just a case study in the details of how to get state machine replication correct so we have introduction to the problem you may have noticed a pattern in fault-tolerant systems that we've looked at so far one is that MapReduce replicates computation but the replication is controlled the whole computation is controlled by a single master another example I'd like to draw your attention to is that GFS replicates data right as this primary backup scheme for replicating the actual contents of files but it relies on a single master to choose who the primary is for every piece of data another example vmware ft replicates computational write on a primary virtual machine and a backup virtual machine but in order to figure out what to do next if one of them seems to a fail it relies on a single test and set server to help the choose to help it ensure that exactly one of the primary of the backup takes over if there's some kind of failure so in all three of these cases sure there was a replication system but sort of tucked away in a corner in the replication system there was some scheme where a single entity was required to make a critical decision about who the primary was in the cases we care about so a very nice thing about having a single entity decide who's gonna be the primary is that it can't disagree with itself right there's only one of it makes some decision that's the decision it made but the bad thing about having a single entity decide like who the primary is is that it itself as a single point of failure and so you can view these systems that we've looked at it sort of pushing the real heart of the fault tolerance Machinery into a little corner that is the single entity that decides who's going to be the primary if there's a failure now this whole thing is about how to avoid split brain the reason why we have to have have to be extremely careful about making the decision about who should be the primary if there's a failure is that otherwise we risks split brain and just make this point super clear I'm gonna just remind you what the problem is and why it's a serious problem so supposing for example where we want to build ourselves a replicated test and set server that is we're worried about the fact that vmware ft relies on this test and set server to choose who the primary is so let's build a replicated testing set server i'm gonna do this it's gonna be broken it's just an illustration for why why it's difficult to get this but brain problem correctly so you know we're gonna imagine we have a network and maybe two servers which are supposed to be replicas of our test and set service connected and you know maybe two clients they need to know who's the primary right now or actually maybe these clients in this case are the primary in the back up in vmware ft so if it's a test and set service then you know both these databases mostly servers start out with their state that is the state of this test flight back in zero and the one operation their clients can send is the test and set operation which is supposed to set the flag of the replicated service to one so i should set both copies and then return the old value so it's essentially acts as a kind of simplified lock server okay so the problem situation the lowly worried about split-brain arises when a client can talk to one of the servers but can't talk to the other so we're imagining either that when clients send a request they send it to both I'm just gonna assume that now and almost doesn't matter so let's assume that the protocol is that the clients supposed to send ordinarily any request to both servers and somehow we you know we need think through what the clients should do if one of the server's doesn't respond right or what the system should do if one of the server seems to gotten responsive so let's imagine now the client one can contact server one but not server two how should the system react one possibility is for is that we think well you know gosh we certainly don't want to just talk to client to server one because that would leave the second replica inconsistent if we set this value to one but didn't also set this value to one so maybe the rule should be that the client is always required to talk to both replicas to both servers for any operation and shouldn't be allowed to just talk to one of them so why is that the wrong answer so the rule is o in our replicated system the clients always require to talk to both replicas in order to make progress at all in fact it's worse it's worse than talking to a single server because now the system has a problem if either of these servers is crashed or or you can't talk to it at least with a non replicated service you're only depending on one server but here we am both servers have to be a lot if we require the client to talk to both servers then both servers has to be live so we can't possibly require the client to actually you know wait for both servers to respond if we don't have fault tolerance we need it to be able to proceed so another obvious answer is that if the client can't talk to both well it just talks to the one who can talk to and figures the other ones dead so what's up why is that also not the right answer the troubling scenario is if the other server is actually alive so suppose the actual problem or encountering is not that this server crashed which would be good for us but the much worse issue that something went wrong with the network cable and that this client can talk to climb one can talk to server one but not server two and there's maybe some other client out there that conduct a server two but not server one so if we make the rule that if a client can talk to both servers that it's okay in order to be fault tolerant that I just talked to one then what's just inevitably gonna happen said this cable is gonna break thus cutting the network in half client one is gonna send a test and set request to server one server one will you know set it state to one and return the previous value of zero to client one and so that mean client mom will think it has the lock and if it's a VMware ft server will think it can be takeovers primarily but this replica still of zero in it all right so now if client to who've also sends a test and set request to you know what price to send them to both sees that server one appears to be down follows the rule that says well you just send to the one server but you can talk to then it will also think that it would either quiet because client you also think that it acquired the lock and so now you know if we were imagining this test and that server was going to be used with the and where ft we have not both replicas both of these VMware machines I think they could be primary by themselves without consulting the other server so that's a complete failure so with this set up and two servers it seemed like we had this we just had to choose either you wait for both and you're not fault-tolerant or you wait for just one and you're not correct and then our correct version it's often called split brain so everybody see this well so this was basically where things stood until the late 80s and when people but people did want to build replicated systems you know like the computers that control telephone switches or the computers that ran banks you know there was placer when we spend a huge amount of money in order to have reliable service and so they would replicate they would build replicated systems and the way they would deal then way would that they would have replication but try to rule out of rule out split brain it's a couple of techniques one is they would build a network that could not fail and so usually what that means and in fact you guys use networks that essentially cannot fail all the time the wires inside your laptop you know connecting the CPU to the DRAM are effectively what you know a network that cannot fail between the between your CPU and DRAM so you know with reasonable assumptions and lots of money and you know sort of carefully controlled physical situation like you don't want to have a cable snaking across the floor that somebody can step on you know it's got to be physically designed set up with a network that cannot fail you can rule out split brain it's bit of an assumption but with enough money people get quite close to this because if the network cannot fail that basically means that the client can't talk to a server to that means server two must be down because it can't have been the network malfunctioning so that was one way that people sort of built replication systems it didn't suffer from split brain another possibility would be to have some human beings sort out the problem that is don't automatically do anything instead have the clients you know by default clients always have to wait for you know both replicas to respond or something never allowed to proceed with just one of them but you can you know call somebody's beeper to go off some human being goes to the machine room and sort of looks at the two replicas and either turns one off to make sure it's definitely dead or verifies that one of them has indeed crashed and if the other is alive and so you're essentially using the human as a as the tie breaker and the human is a you know if they were a computer it would be a single point if you themselves so for a long time people use one of the other these schemes in order to build replicated systems and it's not you know they can be made to work the humans don't respond very quickly and the network that cannot fail is expensive but it's not not doable but it turned out that you can actually build automated failover systems that can work correctly in the face of flaky networks of networks that could fail on the can partition so this split of the network in half where the two sides operate they can't talk to each other that's usually called a partition and the big insight that people came up with in order to build automated replication systems that don't suffer from split brain is the idea of a majority vote this is a concept that shows up in like every other sentence practically in the raft paper sort of fundamental way of proceeding the first step is to have an odd number of servers instead of an even number of servers like one flaw here is that it's a little bit too symmetric all right the two sides of the split here just they just look the same so they run the same software they're gonna do the same thing and that's not good but if you have an odd number of servers then it's not symmetric anymore right at least a single network split will be presumably two servers on one side and one server on the other side and they won't be symmetric at all and that's part of what majority vote majority voting schemes are appealing to so basic ideas you have an odd number of servers in order to make progress of any kind so in raft elect a leader or cause a log entry to be committed in order to make any progress at each step you have to assemble a majority of the server's more than half more than half of all the servers in order to sort of approve that step like vote for a meet or accept a new log entry and commit it so you know the most straightforward way is that two or three servers required to do anything one reason this works of course is that if there's a partition there can't be more than one partition with a majority of the server's in it that's one way to look at this a partition can have one server in it which it's not a majority or maybe you can have two but if one partition has two then the other partition has to have only one server in it and therefore will never be able to assemble a majority and won't be able to make progress and just to be totally clear when we're talking about a majority it's always a majority out of all of the server's not just a live servers this is the point that confused me for a long time but if you have a system with three servers and maybe some of them have failed or something if you need to assemble in the majority it's always two out of three even if you know that one has failed the majority is always out of the total number of servers there's a more general formulation of this because a majority voting system in which two out of three are required to make progress it can survive the failure of one server right any two servers are enough to make progress if you need to be able to if you're you know you worried about how reliable your servers are or then you can build systems that have more servers and so the more general formulation is if you have two F + 1 servers then you can withstand you know so if it's three that means F is one and the system with three servers you can tolerate F servers step one failure and still keep going all right often these are called quorum systems because the two out of three is sometimes held a quorum okay so one property I've already mentioned about these majority voting systems is that at most one partition can have a majority and therefore if the networks partitioned we can't have both halves of the network making progress another more subtle thing that's going on here is that if you always need a majority of the servers to proceed and you go through a sort of succession of operations in which reach operations somebody assembled a majority like you know votes for leaders or let's say votes for leaders arrived then at every step the majority you assemble for that step must contain at least one server that was in the previous majority that is any two majorities overlap in at least one server and it's really that property more than anything else that raft is relying on to avoid split brain it's the fact that for example when you have a leader a successful leader election and leader assembles votes from a majority its majority is guaranteed to overlap with the previous leaders majority and so for example the new leader is guaranteed to know about the term number used by the previous leader because it's a majority overlaps with the previous leaders majority and everybody in the previous leaders majority knew about the previous leaders term number similarly anything the previous leader could have committed must be present in a majority of the servers in raft and therefore any new leaders majority must overlap at at least one server with every committed entry from the previous leader this is a big part of why it is that wrapped is correct any questions about the general concept of majority voting systems these muscle ad servers it's possible intersection something maybe six in the paper explains how to add it or change the set of servers and it's possible you need to do it in a long-running system if you're running your system for five ten years you know you're gonna need to replace the servers after a while you know one of them fails permanently or you upgrade or you move machine rooms to a different machine room you really do need to be able to support changing sets of servers so that's a it certainly doesn't happen every day but it's a critical part of this or a long-term maintainability of these systems and you know the RAF authors sort of pat themselves on the back that they have a scheme that deals with this which as well they might because it's complex all right so using this idea in about 1990 or so there were two systems proposed at about the same time that realized that you could use this majority voting system to kind of get around the apparent impossibility of avoiding split brain by using basically by using three servers instead of two and taking majority votes and in one of these very early systems was called Paxos the RAF paper talks about this a lot and another of these very early systems was called view stamp replication a previa des vs r4 view stamp replication and even though Paxos pod is by far the more widely known system in this department raft is actually closer to design in design to view statment few stamp application which was invented by people at MIT and so there's a sort of a law many decade history of these systems and they only really came to the forefront and started being used a lot in deployed big distributed sisty systems about 15 years ago a good 15 years after they were originally invented okay so let me talk about Rath now raft is a takes the form of a library intended to be included in some service application so if you have a replicated service that each of the replicas in the service is gonna be some application code which you know receives rpcs or something plus a raft library and the raft libraries cooperate with each other to mean replication maintain replication so sort of software overview of a single raft replica is that at the top we can think of the replicas having the application code so it might be for lab 3 a key-value server so maybe we have some key value server and in a state the application has state that raft is helping it manage replicated state and for a key value server it's going to be a table of keys and values the next layer down is a raft layer so the key value server is gonna sort of make function calls into raft and they're gonna chitchat back and forth a little bit and raft keeps a little bit of state you can see it in Figure 2 and for our purposes really the most critical piece of state is that raft has a log of operations and a system with 3 breath will cause we're actually gonna have you know 3 servers that have exactly the same identical structure and hopefully the very same data sitting in sitting at both layers right outside of this there's gonna be clients and the game is that so we have you know client 1 and client two whole bunch of clients the clients don't really know the clients are you know just external code that needs to be able to use the service and the hope is the clients won't really need to be aware that they're talking to a replicated service that to the clients that are looking almost like it's just one server and they talked with one server and so the clients actually send client requests to the key to the application layer of the current leader the replica that's the current leader in raft and so these are gonna be you know application level requests for a database for a key value server these might be put in get requests you know put takes a key and a value and updates the table and get asked the service to get the current key current value corresponding to some key so this like has nothing about to do with raft it's just sort of client-server interaction for whatever service we're building but once one of these commands gets sent from the requests get sent from the clients of the server what actually happens is you know on a non replicated server the application code would like execute this request and say update the table and response to a book but not in a raft replicated service instead if assuming the client sends a request to leader what really happens is the application layer simply sends the request the clients request down into the raft layer to say look you know here's a request please get it committed into the replicated log and tell me when you're done and so at this point the rafts chitchat with each other until all the replicas are a majority the replicas get this new operation into their logs said it is replicated and then when its leader knows that all of the replicas of a copy of this only then as a raft sent a notification up back up to the key value they are saying aha that operation you sent me I mean it's been now committed into all the replicas and so it's safely replicated and at this point it's okay to execute that operation a raft you know the client sends a request with the key value layer Q value layer does not execute it yet so we're not sure because it hasn't been replicated only when it's in out and the logs of all the replicas then raft notifies the leader now the leader actually execute the operation which corresponds to you know for a put updating the value yet reading correct value out of the table and then finally sends the reply back to the client so that's the ordinary operation of it submitted if it's in a majority and again the reason why I can't be all is that if we want to build a fault-tolerant system it has to be able to make progress even if some of the server's have failed so yeah so ever it's committed when it's in a majority [Music] yeah and so in addition when operations finally committed each of the replicas sends the operation up each of the raft library layer sends the operation up to the local application layer in the local application layer applies that operation to its state its state and so they all so hopefully all the replicas seem the same stream of operations they show up in these up calls in the same order they get applied to the state in the same order and you know assuming the operations are deterministic which they better be the state of the replicas replicated State will evolve in identically on all the replicas so typically this this table is what the paper is talking about when it talks about state a different way of viewing this interaction and one that'll sort of notation that will come up a lot in this course is that a sort of time diagram I'll draw you a time diagram of how the messages work so let's imagine we have a client and server one is the leader that we also have server to server three and time flows downward on this diagram we imagine the client sending the original request to server one after that server ones raft layer sends an append entries RPC to each of the two replicas this is just an ordinary I'll say a put request this is append entries requests the server is now waiting for replies and the server's from other replicas as soon as replies from a majority arrive back including the leader itself so in a system with only three about because the leader only has to wait for one other replica to respond positively to an append entries as soon as it assembles positive responses from a majority the leader execute a command figures out what the answer is like forget and sends the reply back to the client I mean why of course you know if s who's actually awry alive it'll send back its response too but we're not waiting for it although it's useful to know and figure - all right everybody see this this is the sort of ordinary operation of the system no no failures oh gosh yeah I like I left out important steps so you know this point the leader knows oh I got you know I'm adora t have put it in no log I can go ahead and execute it and reply yes to the client because it's committed but server two doesn't know anything yet it just knows well you know I got this request from the leader but I don't know if it's committed yet depends on for example whether my reply got back to the leader for all server to knows it's reply was brought by the network maybe the leader never heard the reply and never decided to commit this request so there's actually another stage once the server realizes that a request is committed it then needs to tell the other replicas that fact and so there are there's there's an extra message here exactly what that message is depends a little bit on what what else is going on it's at least in raft there's not an explicit commit message instead the information is piggybacked inside the next append entries that leader sends out the next append entries RPC it sends out for whatever reason like there's a commit meter commit or something filled in that RPC and the next time the leader needs have to send a heartbeat heartbeat or needs to send out a new client request because some different client requests or something it'll send out the new hire leader commit value and at that point the replicas will execute the operation and apply it to their state yes oh yes so this is a this is a protocol that has a quite a bit of chitchat in it and it's not super fast indeed you know yeah client sends in request request has to get to the server the server talks to at least you know another instance that multiple messages has to wait for the responses send something back so there's a bunch of message round-trip times kind of embedded here yes if so this is up to you as the implementer actually exactly when the leader sends out the updated commit index if client requests a comeback only very occasionally then you know the leader may want to send out a heartbeat or send out a special append entries message if client requests come quite frequently then it doesn't matter because if they come you know there's thousand arrive per second and gee so it'll be another one along very soon and so you can piggyback so without generating an extra message which is somewhat expensive you can get the information out on the next message you were gonna send anyway in fact I I don't think the time at which the replicas execute the request is critical because nobody's waiting for it at least if there's no failures if there's no failures replicas executing the request isn't really on the critical path like the client isn't waiting for them the client saw me waiting for the leader to execute so it may not be that it may not affect client perceived latency sort of exactly how this gets staged all right one question you should ask is why does the system why is the system so focused on blogs what are the logs doing and it's sort of worth trying to come up with an explicit answers to that one answer to why the system is totally focused on logs is that the log is the kind of mechanism by which the leader orders operations it's vital for these replicated state machines that all the replicas apply not just the same client operations to their start but the same operations in the same order but they all have to apply that these operations coming from the clients in the same order and the log among many other things is part of the machinery by which the or the leader assigns an order to the incoming client operations I give you know ten clients send operations to the leader at the same time the client the leader has to pick pick an order make sure everybody all the replicas obey that order and the log is you know the fact that the log has numbered slots as part of half a meter expresses the order it's chosen another use of the log is that between this point and this point server 3 has received an operation that it is not yet sure is committed and it cannot execute it yet it has to put the this operation aside somewhere until the increment to the leader commit value comes in and so another thing that the log is doing is that on the followers the log is the place where the follower sort of sets aside operations that are still tentative that have arrived but are not yet known to be committed and they may have to be thrown away as we'll see so that's another use I'm the I sort of do love that use on the leader side is that the leader needs to remember operations in its log because it may need to retransmit them to followers if some followers offline maybe it's something briefly happened to its network action or something misses some messages the leader needs to be able to resend log messages that any followers missed and so the leader needs a place where can set aside copies of messages of client requests even ones that it's already executed in order to be able to resend them to the client I mean we send them to replicas that missed missed that operation and a final reason for all of them to keep the log is that at least in the world of figure 2 if a server crashes and restarts and wants to rejoin and you really need if it you really want a server that crashes - in fact we start and rejoin the raft cluster otherwise you're now operating with only two out of three servers and you can't survive any more failures we need to reincorporate failed and rebooted servers and the log is sort of where or what a server rebooted server uses the log persisted to its disk because one of the rules is that each raft server needs to write its log to its disk where it will still be after it crashes and restarts that log is what the server uses or replays the operations in that log from the beginning to sort of create its state as of when it crashed and then then it carries on from there so the log is also used as part of the persistence plan as a sequence of commands to rebuild the state well ultimately okay so the question is suppose the leader is capable of executing a thousand client commands a second and the followers are only incapable of executing a hundred client commands per second that's sort of sustained rate you know full speed v so one thing to note is that the the replicas the followers acknowledge commands before they execute them so they mate rate at which they acknowledge and accumulate stuff in their logs is not limited so you know maybe they can acknowledge that a thousand requests per second if they do that forever then they will build up unbounded size logs because their execution rate falls it will fall on an unbounded amount behind the rate at which the leader has given the messages sort of under the rules of our game and so what that means they will eventually run out of memory at some point so after they have a billion after they fall a billion log entries behind those just like they'll call the memory allocator for space for a new blog entry and it will fail so yeah and Raph doesn't Raph doesn't have the flow controls that's required to cope with this so I think in a real system you would actually need you know probably piggybacked and doesn't need to be real-time but you probably need some kind of additional communication here that says well here's how far I've gotten in execution so that the leader can say well you know too many thousands of requests ahead of the point in which the followers have executed yes I think there's probably you know in a production system that you're trying to push to the absolute max you would you might well need an extra message to throttle the leader if it got too far ahead okay so the question is if if one of these servers crashes it has this log that it persisted to disk because that's one of the rules of figure two so the server will be able to be just logged back from disk but of course that server doesn't know how far it got in executing the log and also it doesn't know at least when it first reboots by the rule that figure two it doesn't even know how much of the log is committed so the first answer to your question is that immediately after a restart you know after a server crashes and restarts and reads its log it is not allowed to do anything with the log because it does not know how far the system has committed in its log maybe as long as has a thousand uncommitted entries and zero committed entries for all it notes so it's a leader dye support that doesn't help either but let's suppose they've all crashed this is getting ahead of its getting a bit ahead of me but well suppose they've all crashed and so all they have is the state that was marked as non-volatile in figure 2 which includes the log and maybe the latest term and so they don't know some if there's a crash but they all crash and they always start none of them knows initially how far they had been have executed before the crash so what happens is that you leader election one of them gets picked as a leader and that leader if you sort of track through what figure 2 says about how a pendant Rees is supposed to work the leader will actually figure out as a byproduct of sending out a pendant or sending out the first heartbeat really it'll fake it'll figure out what the latest point is basically that that all of the that a majority of the replicas agree on their laws because that's the commit point another way of looking at it is that once you choose a leader through the append entries mechanism the leader forces all of the other replicas to have identical logs to the leader and at that point plus a little bit of extra the paper explains at that point since the leader knows that it's forced all the replicas to have it I didn't have logs that are identicals to it then it knows that all the replicas must also have a there must be a majority of replicas with that all those log injuries in that logs which are now are identical must also be committed because they're held on a majority of replicas and at that point a leader you know the append entries code described in Figure 2 for the leader will increment the leaders commit point and everybody can now execute the entire log from the beginning and recreate their state from scratch possibly extremely laborious Lee so that's what figure two says it's obviously this be executing from scratch is not very attractive but it's where the basic protocol does and we'll see tomorrow that the the sort of version of this is more efficient to use as checkpoints and we'll talk about tomorrow okay so this was a sequence in sort of ordinary non failure operation another thing I want to briefly mention is what this interface looks like you've probably all seen a little bit of it due to working on the labs but roughly speaking if you have let's say that this key value layer with its state and the raft layer underneath it there's on each replica there's really two main pieces of the interface between them there's this method by which the key value layer can relay if a client sends in a request the key value layer has to give it to wrap and say please you know fit this request into the log somewhere and that's the start function that you'll see in Raph go and really just takes one argument the client command the key value they're saying please I got this command to get into the log and tell me when it's committed and the other piece of the interface is that by and by the raft layer will notify the key value layer that AHA that operation that you sent to me in a start command a while ago which may well not be the most recent start right there you know a hundred client commands could come in and cause calls to start before any of them are committed so by and by this upward communication is takes the form of a message on a go channel that the raft library sends on and key value layer is supposed to read from so there's this apply called the apply channel and on it on it you send apply message this start and of course you need the the key value layer needs to be able to match up message that receives an apply channel with calls to start that it made and so the start command actually returns enough information for that matchup to happen it returns the index that start functions basically returns the index in the log where if this command is committed which it might not be it'll be committed at this index and I think it also returns the current term and some other stuff we don't care about very much and then this apply message is going to contain the index command and all the replicas will get these apply messages so they'll all know though I should apply this command figure out what this command means and apply it to my local State and they also get the index the index is really only useful I'm the leader so it can figure out what client would what client requests were talking about by the answer a slightly different question let's suppose the client sends any request in let's say it's a put or a get could be put or again it doesn't really matter I'd say it to get the point in which the it's a client sense and again and waits for a response the point at which the leader will send a response at all is after the leader knows that command is committed so this is going to be a sort of get reply so the client doesn't see anything back I mean and so that means in terms of the actual software stack that means that the key value the RPC will arrive the key value layer will call the start function the start function will return to the key value layer but the key value layer will not yet reply to the client because it does not know if it's good actually it hasn't executed the clients request now it doesn't even know if it ever will because it's not sure if the request is going to be committed right in the situation which may not be committed is if the key value layer you know guess the request calls start and immediately after starboard turn two crashes right certainly hasn't sent out its apply what append messages or whatever nothing's be committed yep so so the game is start returns time passes the relevant apply message corresponding to that client request appears to the key value server on the apply channel and only then and that causes the key value server to execute the request and send her a plot and that's like all this is very important when it doesn't really matter if all everything goes well but if there's a failure we're now at the point where we start worrying about theatres I mean extremely interested in if there was a failure what did the client see all right and so one thing that does come up is that all of you should be familiar with this that at least initially one interesting thing about the logs is that they may not be identical there are a whole bunch of situations in which at least for brief periods of time the ends of the different replicas logs may diverge like for example if a leader starts to send out a round of append messages but crashes before it's able to send all them out you know that'll mean that some of the replicas that got the append message will append you know that new log entry and the ones that didn't get that append messages RPC won't have appended them so it's easy to see that the logs are I'm gonna diverge sometimes the good news is that the the way a raft works actually ends up forcing the logs to be identical after a while there may be transient differences but in the long run all the logs will sort of be modified by the leader until the leader insurers are all identical and only then are they executed okay so I think the next there's really two big topics to talk about here for raft one is how leader election works which is lab two and the other is how the leader deals with the different replicas logs particularly after failure so first I want to talk about leader election question to ask is how come the system even has a leader why do we need a leader the part of the answer is you do not need a leader to build a system like this you it is possible to build an agreement system by which a cluster of servers agrees you know the sequence of entries in a log without having any kind of designated leader and indeed the original pack so system which the paper refers to original Paxos did not have a leader so it's possible the reason why raft has a leader is basically that there's probably a lot of reasons but one of the foremost reasons is that you can build a more efficient in the common case in which the server's don't fail it's possible to build a more efficient system if you have a leader because with a designated leader everybody knows who the leader is you can basically get agreement on requests that with one round of messages per request where as leader of this systems have more of the flavor of well you need a first round to kind of agree on a temporary leader and then a second round actually send out the requests so it's probably the case that use of a leader now speeds up the system by a factor two and it also makes it sort of easier to think about what's going on raft goes through a sequence of leaders and it uses these term numbers in order to sort of disambiguate which leader we're talking about it turns out that followers don't really need to know the identity of the leader they really just need to know what the current term number is each term has at most one leader that's a critical property you know for every term there might be no leader during that term or there might be one leader but there cannot be two leaders during the same term every term has it must most one leader how do the leaders get created in the first place every raft server keeps this election timer which is just a it's basically just out of time that it has recorded that says well if that time occurs I'm going to do something and the something that it does is that if an entire leader election period expires without the server having heard any message from the current leader then the server sort of assumes probably that the current leader is dead and starts an election so we have this election timer and if it expires we start an election and what it means to start an election is basically that you increment the term the the candidate the server that's decided it's going to be a candidate and sort of force a new election first increments this term because it wants them to be a new leader namely itself and you know leader a term can't have more than one leader so we got to start a new term in order to have a new leader and then it sends out these requests boats are pea seeds I'm going to send out a full round of request votes and you may only have to send out n minus one requests votes because one of the rules is that a new candidate always votes for itself in the election so one thing to note about this is that it's not quite the case that if the leader didn't fail we won't have an election but if the leader does fail then we will have election an election assuming any other server is up because some day the other servers election timers go will go off but as leader didn't fail we might still unfortunately get an election so if the network is slow or drops a few heartbeats or something we may end up having election timers go off and even though there was a perfectly good leader we may nevertheless have a new election so we have to sort of keep that in mind when we're thinking about the correctness and what that in turn means is that if there's a new election it could easily be the case that the old leader is still hanging around and still thinks it's the leader like if there's a network partition for example and the old leader is still alive and well in a minority partition the majority partition may run an election and indeed a successful election and choose a new leader all totally unknown to the previous leader so we also have to worry about you know what's that previous leader gonna do since it does not know there was a new election yes okay so the question is are there can there be pathological cases in which for example one-way network communication can prevent the system from making progress I believe the answer is yes certainly so for example if the current leader if its network somehow half fails in a way the current leader can send out heartbeats but can't receive any client requests then the heartbeats that it sends out which are delivered because it's outgoing network connection works its outgoing heartbeats will suppress any other server from starting an election but the fact that it's incoming Network why or apparently is broken will prevent it from hearing and executing any client commands it's absolutely the case that raft is not proof against all sort of all crazy Network problems that can come up I believe the ones I've thought about I believe are fixable in the sense that the we could solve this one by having a sort of requiring a two-way heartbeat in which if the leader sends out heartbeats but you know there were in which followers are required to reply in some way to heartbeats I guess they are already required to apply if the leader stop seeing replies to its heartbeats then after some amount of time and which is seasonals replies the leader decides to step down I feel like that specific issue can be fixed and many others can too but I but you know you're absolutely right that very strange things can happen to networks including some that the protocol is not prepared for okay so we got these meter elections we need to ensure that there is at most at most one meter per term how does Rath do that well Rath requires in order to be elected for a term Raft requires a candidate to get yes votes from a majority of the server's the servers and each server will only cast one yes vote per term so in any given term you know it basically means that in any given term Easter votes only once for only one candidate you can't have two candidates both get a majority of votes because everybody votes only once so the majorities majority rule causes there to be at most one winning candidate and so then we get at most one candidate elected per turn and in addition critically the majority rule means that you can get elected even if some servers have crashed right if a minority of servers are crashed aren't available and network problems we can still elect a leader if more than half a crash or not available or in another partition or something then actually the system will just sit there trying again and again to elect a leader and never elect one if it cannot in fact they're not a majority of live servers if an election succeeds everybody would be great if everybody learned about it I mean need to ask ourselves how do all the parties learn learn what happened the server that wins an election assuming it doesn't crash the server that wins election will actually see a majority or positive votes for its request vote from a majority of the other servers so the candidates running the election that wins it the Kennedy that wins the election will actually know directly uh I got a majority of votes but nobody else directly knows who the winner was or whether anybody one so the way that the candidate informs other servers is that heartbeat the rules and figure to say oh if you're in an election your immediately required to send out an independent trees to all the other servers now the append entries that heartbeat append entries doesn't explicitly say I won the election you know I'm a leader for term 23 it's a little more subtle than that the the way the information is communicated is that no one is allowed to send out an append entries unless they're a leader for that term so the fact that I I'm a you know I'm a server and I saw oh there's an election for term 19 and then by-and-by I sent an append entries whose term is 19 that tells me that somebody I don't know who but somebody won the election so that's how the other servers knows they were receiving append entries for that term and that append entries also has the effect of resetting everybody's election time timer so as long as the leader is up and it sends out heartbeat messages or append entries at least you know at the rate that's supposed to every time a server receives an append entries it'll reset its selection timer and sort of suppress anybody from being a new candidate so as long as everything's functioning the repeated heartbeats will prevent any further elections of course it the network fails or packets are dropped there may nevertheless be an election but if all goes well we're sort of unlikely to get an election this scheme could fail in the sense that it can't fail in the sense of electing to leaders fair term but it can fail in the sense of electing zero leaders for a term that's sort of morningway it may fail is that if too many servers are dead or unavailable or a bad network connection so if you can't assemble a majority you can't be elected nothing happens the more interesting way in which an election can fail is if everybody's up you know there's no failures no packets are dropped but two leaders become candidate close together enough in time that they split the vote between them or say three leaders so supposing we have three liters supposing we have a three replica system all their election timers go off at the same time every server both for itself and then when each of them receives a request vote from another server well it's already cast its vote for itself and so it says no so that means that it all three of the server's needs to get one vote each nobody gets a majority and nobody's elected so then their election timers will go off again because the election timers only be said if it gets an append entries but there's no leader so no append entries they'll all have their election timers go off again and if we're unlucky they'll all go off at the same time they'll all go for themselves nobody will get a majority so so clearly I'm sure you're all aware at this point there's more to this story and the way Raft makes this possibility of split votes unlikely but not impossible is by randomizing these election timers so the way to think of it and the randomization the way to think of it is that supposing you have some time line I'm gonna draw a vents on there's some point at which everybody received the last append entries right and then maybe the server died let's just assume the server send out a last heartbeat and then died well all of the followers have this we set their election timers when they received at the same time because they probably all receive this append enters at the same time they all reset their election timers for some point in the future the future but they chose different random times in the future which then we're gonna go off so it's suppose the dead leader server one so now server two and server 3 at this point set their election timers for a random point in the future let's say server to set their I like some timer to go off here and server 3 set its election timer to go off there and the crucial point about this picture is that assuming they picked different random numbers one of them is first and the other one is second right that's what's going on here and the one that's first assuming this gap is big enough the one that's first it's election time will go off first before the other ones election timer and if we're close were not unlucky it'll have time to send out a full round of vote requests and get answers from everybody who everybody's alive before the second election timer goes off from any other server so does everybody see how the randomization D synchronizes these candidates unfortunately there's a bit of art in setting the contents constants for these election timers there's some sort of competing requirements you might want to fulfill so one obvious requirement is that the election timer has to be at least as long as the expected interval between heartbeats you know this is pretty obvious that the leader sends out heartbeats every hundred milliseconds you better make sure there's no point in having the election time or anybody's election time or ever go off Borja for 100 milliseconds because then it will go off before the lower limit is certainly the lower limit is one heartbeat interval in fact because the network may drop packets you probably want to have the minimum election timer value be a couple of times the heartbeat interval so 400 millisecond heartbeats you probably want to have the very shortest possible election time or be you know say 300 milliseconds you know three times the heartbeat interval so that's the sort of minimum is the heart heartbeat so this frequent you want the minimum to be you know a couple of times that or here so what about the maximum you know you're gonna presumably randomize uniformly over some range of times you know where should we set the kind of maximum time that we're randomizing over there's a couple of considerations here in a real system you know this maximum time effect how quickly the system can recover from failure because remember from the time at which the server fails until the first election timer goes off the whole system is frozen there's no leader you know the clients requests are being thrown away because there's no leader and we're not assigning a new leader even though you know presumably these other servers are up so the beer we choose this maximum the long or delay we're imposing on clients before recovery occurs you know whether that's important depends on sort of how high performance we need this to be and how often we think there will be failures failures happen once a year then who cares we're expecting failures frequently we may care very much how long it takes to recover okay so that's one consideration the other consideration is that this gap that is the expected gap in time between the first time are going off and the second timer going off this gap really in order to be useful has to be longer than the time it takes for the candidate to assemble votes from everybody that is longer than the expected round-trip time the amount of time it takes to send an RPC and get the response and so maybe it takes 10 milliseconds to send an RPC and get a response a response from all the other servers and if that's the case we need to make maximum at least long enough that there's pretty likely to be 10 milliseconds difference between the smallest random number and the next smallest random number and for you the test code will get upset if you if you don't recover from a leader failure in a couple seconds and so just pragmatically you need to tune this maximum down so that it's highly likely that you'll be able to complete a leader election within a few seconds but that's not a very tight constraint any questions about the election time outs one tiny point is that you want to choose new random time outs every time there's every time you every time I node sets it to like me sets its election timer that is don't choose a random number when the server is first created and then we use that same number over and over again because you make an unlucky choice that is you choose this one server happens by ill chance to choose the same random number as another server that means that you're gonna have split votes over and over again forever that's why you want to almost certainly choose a different a new fresh random number for the election time out value every time you reset the timer all right so the final issue about leader election suppose we are in this situation where the old leaders partition you know the network cable is broken and the old leader is sort of out there with a couple clients and a minority of servers and there's a majority in the other half of the network and the majority of the new half of the network elects a new leader what about the old leader why why won't the old leader cause incorrect execution yes to two potential problems one is or one some non problem is that if there's a leader off in another partition and it doesn't have a majority then the next time a client sends it a request that that leader that you know in a partition with a minority yeah it'll send out append entries but because it's in the minority partition it won't be able to get responses back from a majority of the server's including itself and so it will never commit the operation it will never execute it it'll never respond to the client saying that it executed it either and so that means that yeah an old server often a different partition people many clients may send a request but they'll never get responses so no client will be fooled into thinking that that old server executed anything for it the other sort of more tricky issue which actually I'll talk about in a few minutes is the possibility that before server fails it sends out append entries to a subset of the servers and then crashes before making a commission and as a very interesting question which I'll probably spend a good 45 minutes talking about and so actually before I turn to the back topic in general any more questions about in leader election okay okay so how about the contents of the logs and how in particular how a newly elected leader possibly picking up the pieces after an awkward crash of the previous leader how does a newly elected leader sort out the possibly divergent logs on the different replicas in order to restore sort of consistent state in the system all right so the first question is what can think this is this whole topic it's really only interesting after a server crashes right if the server stays up then relatively few things can go wrong if we have a server that's up and has a majority you know during the period of time when it's up and has a majority it just tells the followers what the logs should look like and the followers are not allowed to disagree they're required to accept they just do by the rules of figure two if they've been more or less keeping up you know they just take whatever the leader sends them independent reason appended to the log and obey commit messages and execute there's hardly anything to go wrong the things that go wrong in Rapp go wrong when a the old leader crashes sort of midway through you know sending out messages or a new leader crashes you know sort of just after it's been elected but before it's done anything very useful so one thing we're very interested in is what can the logs look like after some sequence of crashes okay so here's an example supposing we have two servers and the way I'm gonna draw out these diagrams because we're gonna be looking a lot at a lot of sort of situations where the logs look like this and we're gonna be wondering is that possible and what happens if they do look like that so my notation is going to be I'm gonna write out log entries for each of the servers sort of aligned to indicate slots corresponding slots in the log and the values I'm going to write here are the term numbers rather than client operations I'm going to you know this is slot one this is thought to everybody saw a command from term three in slot 1 and server tuned server three saw command from also term three and the second slot the server one has nothing there at all and so question for this like the very first question is can this arrive could this setup arise and how yes so you know maybe server 3 was the leader for just repeating what you said maybe server 3 is the leader for term 3 he got a command that sent out to everybody everybody received a dependent at the log and then I got a server 3 got a second request from a client and maybe it sent it to all three servers but the message got lost on the way to server one or maybe server was down at the time or something and so only server to the leader always append new commands to its log before it sends out append entries and maybe the append entry RPC only got to server 2 so this situation you know it's like the simplest situation and was actually the logs are not different and we know how it could possibly arise and so if server 3 which is a leadership crash now you know the next server they're gonna need to make sure server 1 well first of all if server 3 crashes or we'll be at an election and some of the leader is chosen you know two things have to happen the new leader has got to recognize that this command could have committed it's not allowed to throw it away and it needs to make sure server one fills in this blank here with indeed this very same command that everybody else had in that slot all right so after a crash somebody you know server 3 suppose another way this can come up is server 3 might have sent out the append entries the server 2 but then crashed before sending the append entries to server 3 so if were you know electing a new leader it could because we got a crash before the message was sent here's another scenario to think about three servers again no I mean a number the slots in the law and so we can refer to them got slot 10 11 12 13 [Music] again it's same setup except now we have in slide 12 we have server 2 as a command from term for and server 3 has a term command from term 5 so you know before we analyze these to figure out what would happen and what would a server do if it saw this we need to ask could this even occur because sometimes the answer to the question oh jeez what would happen if this configuration arose sometimes the answer is it cannot arise so we do not have to worry about it the question is could this arise and how all right so any [Music] yeah in brief we know this configuration can arise and so the way we can then get the four and a five here is let's suppose in the next leader election server twos elected leader now for term for its elected leader because a request from a client it appends it to its own log and crashes so now we have this right we need a new election because the leader just crashed now in this election and then so now we have to ask whether who could be elected or we have to give him back of our heads oh gosh what could be elected so we're gonna claim server three could be elected the reason why I could be elected is because it only needs request vote responses from majority that majority is server one and server three you know there's no no problem no conflict between these two logs so server three can be elected for term five get a request from a client append it to its own log and crash and that's how you get this this configuration so you know you need to be able to to work through these things in order to get to the stage of saying yes this could happen and therefore raft must do something sensible as opposed to it cannot happen because some things can't happen all right so so what can happen now we know this can occur so hopefully we can convince ourselves that raft actually does something sensible now as for the range of things before we talk about what RAF would actually would actually do we need to have some sense of what would be an acceptable outcome right and just eyeballing this we know that the command in slot 10 since it's known by all all the replicas it could have been committed so we cannot throw it away similarly the command in slot 11 since it's in a majority of the replicas it could for all we know have been committed so we can't throw it away the command in slot 12 however neither of them could possibly have been committed so we're entitled we don't know haven't we'll actually do but raft is entitled to drop both of these even though it is not entitled to drop it and either of the commands in a 10 or 11 this is entitled dropped it's not required to drop either one of them but I mean oh it certainly must drop one at least one because you have to have identical log contents in the end this could have been committed it the we can't tell by looking at the laws exactly how far the leader got before crashing so one possibility is that for this command or even this command one possibility is that leaders send out the append messages with a new command and then immediately crashed so it never got any response back because it crashed so the old leader did not know if it was committed and if it didn't get a response back that means it didn't execute it and it didn't send out but you know it didn't send out that incremented commit index and so maybe the replicas didn't execute it either so it's actually possible that this wasn't committed so even though RAF doesn't know it could be legal for raft if raft knew more than it does know it might be legal to drop this log entry because it might not have been committed but because on the evidence there's no way to disprove it was committed based on this evidence it could have been committed and raft can't prove it wasn't so it must treat it as committed because the leader might have received it might have crashed just after receiving the append entry replies and replying to the client so just looking at this we can't rule out the possibility that either possibility that the leader responded to the client in which case we cannot throw away this entry because a client knows about it or the possibility the leader never did and yeah we could you know if we have to assume that it was committed yeah no there's no mañana server crash before getting the response it's alright well let's continue this on Thursday