pg_paxos: Table Replication through Distributed Consensus
CC Attribution 3.0 Unported:
You are free to use, adapt and copy, distribute and transmit the work or content in adapted or unchanged form for any legal purpose as long as the work is attributed to the author in the manner specified by the author or licensor.
Slot, Marco
This talk will discuss and demonstrate the pg_paxos extension, which adds an implementation of the Paxos distributed consensus algorithm to PostgreSQL. pgpaxos provides yet another way of doing table replication in PostgreSQL by automatically replicating a query log through Multi-Paxos. Unlike existing replication solutions, pgpaxos can provide consistent, fault-tolerant, multi-master replication across PostgreSQL servers, at the cost of high read- and write latencies and low throughput. pg_paxos is not a general replication solution, but is especially suitable for applications such as automated fail-over, distributed locking, and managing cluster membership, and can potentially replace components such as Zookeeper.
 I'm Marco from Citus Data. I'm going to talk about PG-Paxos, a word of warning if you're looking for a completely production ready solution. This is probably not the right talk to be at, but hopefully it's going to be quite interesting. So this talk will have three parts. The first part I'm going to talk about what is Paxos, how does it work, why does it work, and how can you use that to implement database replication. And then I'll go through some of the internals of PG-Paxos, an extension for Postgres that uses the Paxos algorithm to do replication. And finally I'll show you a demo of how you set up Paxos for an actual application. So the problem that Paxos addresses is that of distributed consensus. So let's say you have a number of servers, 3, 10, 100, you need exactly one of them to do something. So they all need to agree who gets to do the job, who gets to be the leader. Or you have a number of replicas of some data and there's updates to that data and they all need to agree on the order in which these updates are applied. Turns out this is an impossible problem if there are failures, which there are in distributed systems. You cannot solve consensus, you cannot write an algorithm that always reaches consensus. Let's say if you have two nodes you cannot get them to agree on who gets to be in charge if under all failure scenarios. So Paxos is a probabilistic algorithm for achieving consensus. There's basically two classes of probabilistic algorithm. One is kind of always finishes but may not succeed and the other one is always succeeds but may never finish. And Paxos is the latter category. So if there are certain failure scenarios, Paxos may take forever. But most of the time it does not. So Paxos has built up a bit of a reputation as a very hard and complicated algorithm and there's a few factors contributing to that reputation. One of which was the original paper on Paxos which was called the part time parliament. And the abstract of the paper is recent archaeological discoveries on the island of Paxos revealed that the parliament functioned despite the peripatetic propensity of its part time legislators. The legislators maintained consistent copies of the parliamentary record despite their frequent phrase from the chamber and the forgetfulness of their messengers. So the whole paper was a metaphor using some fictional kind of parliamentary system that never really existed as far as we know to try to explain this algorithm and no one really understood it. Like if you read this paper and then from that can figure out how to implement Paxos, you'd be probably a genius. But Lampert the inventor of the Paxos algorithm wrote another paper a few years later where the entire abstract was the Paxos algorithm when presented in plain English is actually very simple. And it actually is quite simple. So this is essentially the Paxos algorithm. So imagine you have a group of nodes, some well-defined group that they all know about each other. And each node has a function called Paxos with a key and a value. And for a given key, this function always returns the same value on all nodes in the group and the value is one of the inputs. So one node might call it with value one, one node might call it with value five, one node might call it with value 10, and then maybe it returns five on all of them or 10 or one. It depends on a lot of factors, but it always returns the same value on all nodes. That's useful. So it runs in two phases. If you call this function, you become a proposer. So you propose to, you basically propose a new value. So first you need to ask nodes to participate. So you ask nodes, please prepare for a new proposal. And then you need to get the majority of nodes to participate. Concept of a majority is quite important in Paxos. If you cannot get a majority to participate in your proposal for any reason whatsoever, failures, other things, you start over. If you do get a majority, you go to phase two and you request that these nodes accept a value. And if the majority of nodes at this point accept a value, you have consensus, and Paxos completes. If anything goes wrong, you start over. And there's no guarantee that this doesn't take forever. Like you could start over, but usually it doesn't take forever. So zooming in into phase one, the question that the proposer asks to the other nodes, and the proposer is kind of, it's usually one of the nodes itself, but it could be, could even be separate. Please don't accept proposals with a lower number than I. So I is some sequence number. It usually just starts at zero. And the nodes receiving this message can then respond saying, okay, I'll participate. Or well, actually I already received a competing proposal from someone else, and his sequence number was greater than yours. So I promised I would not participate in proposals with lower sequence numbers. Now they could be using the same sequence number, actually very often they are, but then you look at something else like node identifiers, something that's unique across the proposals. So in that case, the acceptor sends back, hey, I already got this proposal with higher number. And then the proposer says, darn, I'm going to start over, but this time I'm going to set my sequence number to be greater than J. So next time I will get my proposal through, hopefully. And the third response that the acceptors could give here is actually I already accepted a value before. So someone already reached phase two, and in phase two I accepted a value, and then the acceptor sends back this value to the proposer. And when the proposer gets a value from another node, from someone else's proposal, it has to start using this value instead of its own. And it fixed the value with the highest proposal number. So now it throws away its own value and it uses someone else's value. And this is kind of the trick that makes Pac-Sos work, and I'll show you why in just a sec. So the second phase of Pac-Sos, I'm going to send messages to all the nodes that are participating. Please accept this value, could be my value, could be someone else's value, for this proposal number, I. And then the acceptors could say, okay, I accept your value, or they could say, well, I already received a competing proposal with a higher sequence number, someone else ran phase one, and I promised I would not participate in proposals with lower sequence numbers, so sorry. And then the proposer starts over with a higher sequence number, and it could take a while. And then if anything goes wrong, for some reason you do not get the majority to say, okay, you also start over. And finally, there's usually a phase three, like you don't want to go through all this stuff again. So if you manage to get confirmation that tree nodes, or sorry, the majority of nodes, so that's like two out of three, or three out of four, or three out of five, accept the value, you just tell everyone else, but that they don't have to rerun Pac-Sos. But again, that could be failures here, and so it's not actually a critical step, you could even skip this step. It's more of an optimization. So why does this work? So actually, this algorithm is sufficient to provide the guarantee that it will only ever return the same value on all the nodes, and the value will be one of the inputs. So why will it never return a different value? Well, if you kind of think of it backwards, and Pac-Sos was actually kind of derived backwards from the notion of consensus. So it was more of a derived algorithm than an invented algorithm. So if a majority of nodes accept my proposal, that means that no other node has completed phase one since I did. Because if another node managed to complete phase one, I wouldn't have gotten all my OKs. One of them would have said, no, I will not participate. Because to get my, like if someone completed phase one, he would have needed a majority, and at least one node is going to be in common with my majority. That's the trick with the majorities. Two majorities always have at least one node in common. So because I completed phase one, phase two, it means that any other proposal still needs to complete phase one, and thus will see my value because they will run into this. Because they still need to complete it, so they will definitely run into this for at least one node. And it's also guaranteed that my value has the highest proposal number that got accepted since no one else has managed to complete phase one since I did. So every other proposal that got to phase two had a lower proposal number. So basically, just getting a majority of accepts in Paxos tells me this is now the value that is forever going to be returned by Paxos for a particular key. How is that useful? Like, this is a right once algorithm. You can set the value once, and then it never changes. Well, actually, I can do something which the server systems people, we call a, they'll they replicate a state machine where the idea is if you have a bunch of nodes which all have the same state machine and the same initial state, and you feed them all the same inputs, they will be in the same final state. So what I can do with Paxos is replicate a log of changes. A log is a right once thing, a right only thing. So I can say Paxos, I can use the key as kind of the sequence number in the log, or the sequence number in the log of the key, and my first item could be set x equals six, and then the second item could be set i equals seven, and then the third item could be set i equals nine. So if I know about all these three items, I know that the value of x equals six, and the value of y equals nine. So what I can do with Paxos is implement kind of this replicated log, and I can always make sure that I know what's in the log just by running Paxos for all these values. And this technique is actually called multi-Paxos, because you run Paxos multiple times. So the way I write to this log is I run Paxos for whatever I believe to be the current round number, or basically at the end of the log, and then I run it, and maybe Paxos has already completed, and I just get back someone else's value, and I say, okay, well, I'll try with the next highest round number, or what I now believe to be the highest round number. And maybe someone else comes in, completes phase one before I do, my proposal loses, and then I get back his value from Paxos. Okay, I try again with a higher round number. Eventually, hopefully, I'm going to get my value appended to the log. So my command is going to have some round number where it has reached consensus. To perform a read, what I need to do is I'm going to ask all the nodes, or at least the majority of nodes, what's the highest round number for which you accepted a value, knowing that that is the highest round number which could have consensus at the time that my read started. And if I make sure I know all the values in the log up to that round number, I'm good. Like my log is entirely up to date, up to the point when my read started, I'm going to see any preceding read. I'm going to get fresher information than any preceding read. Sorry, I'm going to see an preceding write. I'm going to get fresher information than any preceding read. But I do need to provide some value when I call Paxos. Like it takes a value as a parameter. So what I could do is just give an empty string as the value when I want to do a read. In this case, something went wrong somewhere and no value actually reached consensus in the round. And it could even be that my empty string will actually reach consensus in that round. That's fine. Like nothing bad happens, this just means nothing changes. So sometimes you could have empty entries in the log because of this. And so each note has its own copy of the log and it could be arbitrarily far behind. If a note fails and doesn't get updates for an hour like its log, it's going to be far behind. But it can almost just run the consistent read algorithm. Say note C comes back, say, okay, I want to do a consistent read on my state. Let's say my table. Now I'm going to ask note A and B what's the highest round number for which you accepted the value? So note B will say two. Let's say note B will say one. So note C runs paxos for this. Somehow it actually already accepted the value here. Maybe someone tried this insert and then crashed or whatever, something went wrong. Actually another value reached consensus. So when note C runs paxos for round one, it'll get back that value when it runs paxos for round two. It'll get back this value. I mean, it already accepted this value but didn't know whether there was consensus probably. So at that point note C is completely up to date with any proceeding writes so it can return a result. So I implemented this in Postgres as an extension. So called pgpaxos. So it's an extension you can just install in Postgres. I think it works 9.4, 9.5, possibly works on 9.6, I haven't tried. And it provides consistent, full tolerant, multi-master table replication where you can tolerate a minority of notes going down. So if you have three notes, you can tolerate one going down. The trade-off that you're making is that the right throughput is going to be incredibly low because you cannot really have concurrent writes. You can only have one write at a time for a particular log item otherwise they're going to compete and that just makes both of them take longer. So the right throughput is low, think like maybe 100 per second and the latency is generally going to be high. There's actually no guarantee on how high it could be. It could be very high. If two notes are down, let's say, then there's no way, and there are three notes in total, there's no way to get a majority so pgpaxos will take forever until one of the notes comes back. So that's the trade-off you're making, but there's no other way to get consistent, full tolerant, multi-master table replication. So it's not really an alternative to other replication solutions like streaming replication or logical replication. They have a particular purpose and it doesn't replace them in any way. It also doesn't magically make Postgres distributed. There will never really be such a thing as distributed Postgres. In distributed systems, there's not really solutions, there's just different trade-offs you can make. But it can be a very useful building block, especially for distributed systems in the way that you wouldn't do your own data storage. You wouldn't like to know, but most of the time, you don't want to write your own serialization algorithms to write things to disk and take care of truncation and concurrency and whatnot. You use a database. Similarly, there are some very hard problems that pgpaxos solves which you probably don't want to solve yourself, and you can use pgpaxos for those. I'll give you some examples later. So pgpaxos is available on GitHub under PostgresGL license. So it contains a very basic implementation of paxos and multipaxos, which is the state machine written in PLPG SQL, which sounds a bit weird, but it's a surprisingly suitable language for implementing distributed algorithms because a lot of the time, one of the things that makes paxos hard is you really need to be very careful about your local state and making sure you write that to disk in case there's a crash before you give any response, making sure you deal with truncation, making sure you serialize data. And actually, all of that comes for free in PLPG SQL. I just do an insert on a table and then let Postgres do the work. And PLPG SQL has a networking API, DB link. It's not so great, but it works. And it does make it easy to do remote procedure calls. And the other part of pgpaxos is basically C code, which uses the planner and executor hooks in Postgres to provide a replication for particular tables that you want to replicate. So you can mark certain tables as replicated, and then any insert, for example, you do, is going to be replicated to the other nodes using paxos. And I should say it is somewhat experimental. Like if someone, like the most useful thing if you want to contribute to this project is actually testing, I wouldn't, I don't feel like we're missing a lot of features, but we're definitely, like I've done a lot of testing, but like the industry standard here for these types of systems is actually Jepsen and it's actually quite a lot of work to do Jepsen test. But any type of testing you can do would be very useful in making people confident that this is a robust system. So as I said, PLPG SQL is actually surprisingly suitable for implementing distributed algorithms. You get transactional semantics. It's easy to collect a bunch of responses in a table and do some query on them. There's a simple networking API. RPC, remote procedure call, becomes pretty simple. I can just do create function, which is the one I call on the remote end, and then create another function which uses dblink to call the other function. So it's pretty simple. And because it's PLPG SQL and Amazon RDS and Heroku both have dblink installed, I could even run this on Amazon RDS. I couldn't do the sort of transparent table replication, but the paxos functions, they would work. So if you create extension pgpaxos, you will get a bunch of tables, kind of boring, but just for the concepts. There's a group table, which is the paxos groups in which this node participates. So a node can be in many groups at the same time. Host table is just the members of the group. The round table is a bit of a forename for the log. So this contains the actual replicated SQL queries. And there are sort of replicated tables, which keeps track of which tables should be replicated using pgpaxos. And the paxos extension, pgpaxos also adds some functions. One is just the paxos function that I've already described. Pretty much works exactly like that. The apply log function is pretty much the read, the consistent read function I've described. So if you give it like the max round number in the group, it just makes sure that the log is up to date. Apply and append is the write function I described. So it keeps trying to get consensus on its query for a particular round, and if it fails, it goes to the next round. But you don't really need to worry about these functions so much unless you wanted to implement your own replication solution or log. Also the time what you do is you create a table, and then on the first note in the paxos group you do a create group where you specify your own host name, such that other notes know how to connect to you. And then you do replicate table. So the data refers to the table. And after this, any insert or update on the table will get replicated, added to the paxos log, and any select will make sure that your log is up to date before the select is actually executed. So on the first note you do those three steps, and then on the second and third and whatever note, you do join group. So you specify one of the existing servers in the group. I guess they usually do original one, and then your own IP. And what this actually does is quite interesting, like if a node joins paxos, the world kind of changes. Like, before the node joins, maybe my majority was two nodes, but after the node joins, now my majority is three nodes. So I need to be aware of that. So the way that works in paxos is actually the fact that this node joins is added to the log. So it's that any subsequent query will actually see this new node. So what happens when there is an update? So as a user, you just do update the table, set to do whatever. Then in the background, paxos runs this, pgpaxos runs this, paxos apply an append function with more or less your query. It kind of deparces it from the postgres query plan, but more or less it will log back. And then it keeps trying to append. And then when it learns, okay, now I've appended it at log position 51. So I'm going to make sure that items 0 to 50 have been executed, and then I'm going to execute the update, such that my update works on a consistent copy of the table. And this makes basically all writes completely serializable. They're all executed in the same order on all the machines. Then you do a select. Basically you run the read algorithm. So in the background, it didn't transparent to the user usually. Paxos max group round gets talks to the other nodes, asks them what's the highest round number for which you accepted a value, and then it gets its log up to date up to that point. And it knows that when the select started, like any round number higher than that could not have had consensus, because at least one node in the majority would have accepted a value for it. There's some subtle to this when, like, new nodes join, then you have to repeat the max group round. But this is the general principle. So I'd like to do a little demo. But are there any questions so far? So how do you deal with transactions? So what you can do, and this is more of a manual, oh, sorry, the question was how do you deal with transactions? And essentially, like, the transparent table replication is more or less single statement. So you can't really do begin statement rollback or something. What you could do just using the faxos function is just log the whole string of your transaction, like begin, update, delete, insert, commit. You could log that as the entire string, and then it'll get executed in order along with the other writes. Yeah, that'd be the more practical way of doing it. It would also save you on network traffic. But yeah, that's entirely possible. All right, so I want to show you a demo of a distributed locking system. So I don't actually need PetitGPoxos and the nodes on which I'm doing work to be the same thing. Oh, sorry, Jeff. So if you have a single thing in the node, you'd be able to do that. So if you were able to do that, you'd be able to do that with the reportage, the state, the security, the storage, the time, and so that would be very, very safe to use there and make sure you got this in the middle of the transaction. No, I mean, it'll end up, let's say, when you do a consistent read and there's a bunch of things in the log that you need to execute, it'll end up just doing those in one transaction, basically. Is that the question? Well, any state that's recorded during the call to PetitGPoxos is actually done over the DB links. So you don't, like, PetitGPoxos doesn't even, like, the PetitGPoxos function doesn't even store state on itself. It's assumed to be completely external. So it just talks to the other nodes as if, you know, they were external. It could be talking back to itself. But so let's say in, oh, this is pretty far back. Let's say when this node responds saying, okay, I accept the value, I mean, that's a separate transaction, and so actually the PetitGPoxos function doesn't record any state at all. Everything happens in those remote procedure calls. All right. The round number needs to be persistent. You need to keep track of how much of the log you have executed. So whenever you do an execute, you also update the round number of, like, I've executed this much of the log. Yeah, and you just find that from everyone else. Yeah. Yeah, there's a separate table for keeping track of, like, which log you've applied, and then there's a table for the log itself, which contains all the items. So I'm going to show you a demo where PetitGPoxos is used to implement the distributed locking service, and the motivation here is doing distributed chron. This is actually quite a tricky problem if you have, let's say, a job that absolutely must happen every minute, but you don't want it to happen twice. So you could set up one node, and then it runs this job every minute, but if that node fails, then it doesn't happen. You could set up two nodes that both run it every minute, but that's probably not what you want. So you want exactly one of them to do it. So this is kind of a locking problem. Of course, you could set up a database where you, which has a table of locks, but then if the database goes down, you still have the same problem. But with PetitGPoxos, you can actually solve this. Like, so I've set up a little PetitGPoxos cluster using very small instances. I'm running this on EC2, so I'm using the T2 micros. Paxos doesn't have very high throughput anyway, so you might as well just use small instances. I put a load balancer in front of it, so it's that the chron node can just, will just get rooted to whichever Paxos node is up and running. And then they both try to do an insert into the same locks table with, which has a primary key on the name of the lock, and only one of them will succeed, the other one will fail and does know that the other guy has the lock. So yeah, as I said, I set up my cluster on EC2. There's three PetitGPoxos nodes and two slightly bigger chron nodes. They're not going to have to do a lot of work, but that's just how you do it. So the chron nodes, let's say, for example, you're going to have a system bank, you're going to have a system that will have a count on the node, but not quite the same. So the list of all the accounts and all the stuff using the chron one and chron two nodes. So the application I'm going to do is much simpler, I'm going to send a tweet every minute. But in that example, yeah, I mean, let's say there could be some additional database in which that is stored or they do it locally, it's kind of independent of Paxos itself. I guess technically you could actually store the accounts in Paxos as well. So what is the chron one and chron two? I mean, it's technically possible. I mean, usually you want these guys to probably have some computing power depending on what they want to do. I mean, like you can, you don't have to, let's say. I mean, you could have the chron nodes be co-located with the PgPoxos nodes or you could use them separately. It's kind of all the same. Here it's because I use these super tiny instances, the PgPoxos cluster costs almost nothing anyway. Yes, I can, I'll show you the Paxos block. Yeah, but at least if a node goes down, you don't have to actually immediately react to it. So you can automate a lot probably. So I have a bunch of nodes set up here with, this is one of my PgPoxos nodes. So it has the PgPoxos extension. I have a LOX table which has like a key and an owner and the key is the primary key. So I can only have one owner of the key at a time. And on this node, I will create a Paxos group and then replicate a table within that Paxos group. Now, that's not very interesting. So I'm going to join some additional nodes. This is one of my other Paxos nodes. And the number it's showing is actually the round in the Paxos log in which it's joined. It's kind of an internal thing. So at this point, I can do an insert, let's say, on this node. And then if I go to one of my other nodes and try to do the same insert, it's going to give me a primary key violation because they're now looking at a replicated table so only one insert will succeed. So if I, this is the third node, I can see there's now one item in my log. I can make it a bit more challenging. Let's say I do an insert on one node and then immediately a select on the other node. I should then get that insert back, right? I should see I inserted AB and now I see AB on the other node in the selected. This is guaranteed by Paxos. So I set up a load balancer in front of my Paxos node so this just connects me to any of the nodes. I don't know which one. I can't actually really distinguish them. They all have the same content. And I have a little script which will just send a tweet. And I have this same script on three nodes, two nodes, sorry, two ground nodes. And only one of them will send it. So I'll activate the script. I just have a, this and then Kron runs at, you know, when the number of seconds reaches zero so I could quickly open the Twitter account. No tweets yet. So, from what Kron is it runs once a minute. Okay, so someone grabbed the lock, a note 0.171, grabbed the lock so I should now be getting a tweet. Yes. And then both nodes are running the script but only one of them will succeed. And it's actually not like the clocks in Amazon data centers are really well synced so they run Kron almost at exactly the same time. So it kind of tends to alternate between both nodes if they're both up and running. So one thing I can do now is I can maybe stop one of my nodes, my Paxos nodes. If now I do this then I've got a bunch of warnings, I still need to kind of clean up the output of Paxos because it shows all the stack phrases of the LPG SQL. But the point is it did manage to do the select even though a note was down. It'll make one more connection attempt and then after that it will just, you know, pretend it doesn't make it. Which actually makes it even, makes it a bit faster, surprisingly enough because now it only needs to connect to two nodes. So there should be another tweet by now. Right, there's another tweet. Obviously I can also take down this node so I've stopped the Kron job over here. So at some point I should be seeing, in 26 seconds I should be seeing that the other node still manages to send the tweet, I'm not sure if Twitter shows that kind of in a live way. But so I've taken down one of the Paxos nodes, I've stopped the Kron job on one of my Kron nodes. So I could model this, let's say I put this in three different data centers, three different availability zones. I could lose an entire data center and things will just work fine. Which is quite nice about the G-Paxos. Right, so now the other node did the tweet. So not to completely spam the PGCon hashtag, I'm going to shut off the Kron job here. Okay. Any questions about the demo? All right. So what could you use this for? So this is an example of distributed locks. I didn't go into this, I kind of used a cheap trick here where the name of the lock that these Kron jobs are using is actually the number of minutes times since 1970. It doesn't actually rely heavily on synchronized clocks. If one clock is ahead, then most of the time that one will just win because it will be the first one to grab the lock, but then if that fails, then the other guy, there might be some, you know, several minutes where, let's say the other guy's several minutes behind on its clock that it might take a while before it starts tweeting again. But in Amazon data centers, it's usually synchronized up to the millisecond, so it's not really an issue. So PG-Paxos works well for any application that has low read, write volumes, but strong consistency requirements and no strong latency requirements. So distributed locks is one. If you have this type of Kron job, any type of resource management, like we can only have one node access a certain resource at a time, it works quite well for that. Also for managing cluster membership, I mean PG-Paxos manages its own cluster, but it can also manage other clusters. And this is actually one of the most common applications for Paxos. And this, for example, also includes deciding which node is the primary. And so you could imagine an application where you use Paxos to decide which out of a streaming replication group gets to be the primary. And then Paxos can tolerate one of the nodes failing. You could use it as a job scheduler. Alvaro recently did a blog post showing how you could use PG-Paxos to implement a job queue which has, like, exactly one semantics, whereas most queuing systems have at least one semantics, so the job could happen multiple times. Migrations, if you want to move over a lot of data from one place to another, you can scheme on migrations. Or as a source for metadata, like often if you have a system where, let's say, all nodes have a copy of the metadata, there's one authoritative source. But if you don't want to be subject to that source going down, you would have to use something like PG-Paxos to make sure that you have a replicated consistent copy of the metadata. So why not raft? I get this question a lot. Well, one is, Paxos can be implemented in PLPG SQL, which makes it quite suitable for implementing in Postgres, because it's actually really simple. And there's almost like a one-to-one mapping of sentences in the paper to SQL queries in the source code, because you don't have to worry about transactions and interpretation and serialization. That's a good question. The question is, what is raft? So then it's a slightly newer consensus protocol. It works by kind of electing a master, basically. It has quite a good algorithm for quickly switching, making one of the nodes master, and if the other one goes down very quickly, another one becomes the master. And well, one of the differences, like this has been a recent paper, and the paper was very instructive to engineers. It told you exactly what to do, whereas the Paxos papers are a bit more mathematical. So raft has picked up a lot of steam lately in newer distributed databases, because the paper is easier to implement in some sense. So actually Paxos, as an algorithm, is much simpler. The minimal implementation that works is a lot simpler. You can kind of play with it more to be adapted to your requirements. I really wanted it to be multi-master, so I could do writes from anywhere. And mathematically, it's very elegant. Like, in a way, Lemper didn't invent it. He just derived it from the notion of consensus, like, what would I need to do? And it's actually also optimal for achieving consensus. But actually, the more realistic answer probably is, I knew Paxos quite well. I knew PLPG SQL quite well, and I figured, I think I could implement Paxos in PLPG SQL. So that's my story. I hope you enjoyed it. Any questions? So what do you want to do with the PXO? Do you want to do a PG Paxos on the PXO? No, because PG Paxos keeps its date in tables. So on the secondaries, yeah, I suppose you could work for logical replication. Or if you create another database cluster that's co-located with these nodes. Yeah, unfortunately, it doesn't work on secondaries. Josh? I was going to ask you, too, about the user traffic cluster. But do they need to know whether or not the map is not supported by the user traffic package? Yeah. I would typically recommend running it in this type of setup where PG Paxos is actually an external component running on tiny hardware, because then it's just already up and running when you bring up your big data production cluster so you don't really have that problem. And you could reuse the PG Paxos cluster for other stuff. It's a bit similar in the way people use ZooKeeper. They just have this external ZooKeeper deployment which solves a similar problem. So the question is what is adding a node like? And it's essentially just you set up Postgres, you install the PG Paxos extension. You also need in Postgresql.conf to add it to shared preload libraries. And once you have a node, it's pretty much just you call this join group. Actually, there's one step missing here. You first need to create the same tables. It doesn't replicate the DDL yet. It seems to be a common problem with replication solution. But then you just call Paxos join group and this actually more or less clones the state of an existing node. So any data that's already in the table, it copies over. The membership table, it copies over. At this point, it might be interesting to show this. This is actually the Paxos log that I've been talking about. So these are the queries that have been executed on the table. But yeah, joining a node is mostly setting up the tables that are replicated and then calling join group. And then join group also makes sure that all the other nodes know about this node. And there's a corresponding leave group, which gives some errors because the other node is down for errors warning. But now it's left a group. So it's an elastic load balancer. So the way elastic load balancer works, so this is an Amazon service, it sets up EC2 instance more or less in every availability zone in which you have instances and then sets up users root 53 to set up DNS records pointing to each of these things. So let's say one data center goes down, like I took down one of my Paxos nodes, one of my Cron nodes, one of these guys will also go down, but I still have two left. So my clients can just talk to any of those two. And ELB is constantly doing health checks, so ELB will detect, oh, this Paxos node is no longer running, don't send any traffic there. So actually the single point of failure is kind of picking care off in ELB. But yeah, I mean, it's trickier in your own kind of architecture to do that. Probably the simple. Yeah, pretty much. It depends. So Amazon or the load balancer actually is a load balancer, so any new connection will typically go to a different one of these machines. And I mean, it performs constant health checks to figure out who's up and running. But there's no like, you know, Amazon doesn't really pick like, I'm always going to send my traffic to this machine. Every time you connect, you actually get connected to a different machine. So you have to forget about it, you have to master it, and it's just a return. And I guess that kind of thing, Amazon has to pick when you do. Who will run more than that? Who will run more than that? Who will run more than that? Because you don't know if you have to do it with a friend. Well, if you try like with an ELB and you disconnect and reconnect and disconnect and reconnect, pretty much always it connects to a different one. It actually just routes or picks a node based on your source port, kind of uses a hashing function of that source port. At least used to be the algorithm. I used to be at Amazon, so I know what it was like two years ago. I don't know what it is now. Yeah, no, it does. I think actually, so the question is, there's no conceptual technical reasons that you couldn't replicate BDL. And I'm trying to remember, I think actually if you do something like create index, it actually does replicate it because it's kind of easy. Alter table is a bit tricky because PGPUXO is kind of just before execution when the planning has already been done. And if then there's an alter table happening before that, then the execution no longer makes sense. So that's, or plan is no longer correct. So that's, there's some limitations around that, but doing something like create table actually isn't a problem, but then there's nothing to activate the PAXO's log. Once I create the locks table, that's when I start replaying the log. But if I don't have a locks table, then how do I start? I could just create a function for that or something, but it's not there at the moment. Good question. So if the PAXO's log rotated, currently it is not. It will grow infinitely. And yeah, there's also no reason it couldn't be rotated. Pretty much if you know all your notes have reached a certain point, then you can just drop everything that came before then. There's just currently no logic to do that. All right. That's my time. Thank you for listening. And if you have any other questions, I'll be around.
 I'm Marco from Citus Data.  I'm going to talk about PG-Paxos, a word of warning if you're looking for a completely  production ready solution.  This is probably not the right talk to be at, but hopefully it's going to be quite interesting.  So this talk will have three parts.  The first part I'm going to talk about what is Paxos, how does it work, why does it work,  and how can you use that to implement database replication.  And then I'll go through some of the internals of PG-Paxos, an extension for Postgres that  uses the Paxos algorithm to do replication.  And finally I'll show you a demo of how you set up Paxos for an actual application.  So the problem that Paxos addresses is that of distributed consensus.  So let's say you have a number of servers, 3, 10, 100, you need exactly one of them to  do something.  So they all need to agree who gets to do the job, who gets to be the leader.  Or you have a number of replicas of some data and there's updates to that data and they  all need to agree on the order in which these updates are applied.  Turns out this is an impossible problem if there are failures, which there are in distributed  systems.  You cannot solve consensus, you cannot write an algorithm that always reaches consensus.  Let's say if you have two nodes you cannot get them to agree on who gets to be in charge  if under all failure scenarios.  So Paxos is a probabilistic algorithm for achieving consensus.  There's basically two classes of probabilistic algorithm.  One is kind of always finishes but may not succeed and the other one is always succeeds  but may never finish.  And Paxos is the latter category.  So if there are certain failure scenarios, Paxos may take forever.  But most of the time it does not.  So Paxos has built up a bit of a reputation as a very hard and complicated algorithm and  there's a few factors contributing to that reputation.  One of which was the original paper on Paxos which was called the part time parliament.  And the abstract of the paper is recent archaeological discoveries on the island of Paxos revealed  that the parliament functioned despite the peripatetic propensity of its part time legislators.  The legislators maintained consistent copies of the parliamentary record despite their frequent  phrase from the chamber and the forgetfulness of their messengers.  So the whole paper was a metaphor using some fictional kind of parliamentary system that  never really existed as far as we know to try to explain this algorithm and no one really  understood it.  Like if you read this paper and then from that can figure out how to implement Paxos,  you'd be probably a genius.  But Lampert the inventor of the Paxos algorithm wrote another paper a few years later where  the entire abstract was the Paxos algorithm when presented in plain English is actually  very simple.  And it actually is quite simple.  So this is essentially the Paxos algorithm.  So imagine you have a group of nodes, some well-defined group that they all know about  each other.  And each node has a function called Paxos with a key and a value.  And for a given key, this function always returns the same value on all nodes in the  group and the value is one of the inputs.  So one node might call it with value one, one node might call it with value five, one  node might call it with value 10, and then maybe it returns five on all of them or 10  or one.  It depends on a lot of factors, but it always returns the same value on all nodes.  That's useful.  So it runs in two phases.  If you call this function, you become a proposer.  So you propose to, you basically propose a new value.  So first you need to ask nodes to participate.  So you ask nodes, please prepare for a new proposal.  And then you need to get the majority of nodes to participate.  Concept of a majority is quite important in Paxos.  If you cannot get a majority to participate in your proposal for any reason whatsoever,  failures, other things, you start over.  If you do get a majority, you go to phase two and you request that these nodes accept  a value.  And if the majority of nodes at this point accept a value, you have consensus, and Paxos  completes.  If anything goes wrong, you start over.  And there's no guarantee that this doesn't take forever.  Like you could start over, but usually it doesn't take forever.  So zooming in into phase one, the question that the proposer asks to the other nodes,  and the proposer is kind of, it's usually one of the nodes itself, but it could be,  could even be separate.  Please don't accept proposals with a lower number than I.  So I is some sequence number.  It usually just starts at zero.  And the nodes receiving this message can then respond saying, okay, I'll participate.  Or well, actually I already received a competing proposal from someone else, and his sequence  number was greater than yours.  So I promised I would not participate in proposals with lower sequence numbers.  Now they could be using the same sequence number, actually very often they are, but  then you look at something else like node identifiers, something that's unique across  the proposals.  So in that case, the acceptor sends back, hey, I already got this proposal with higher  number.  And then the proposer says, darn, I'm going to start over, but this time I'm going to  set my sequence number to be greater than J. So next time I will get my proposal through,  hopefully.  And the third response that the acceptors could give here is actually I already accepted  a value before.  So someone already reached phase two, and in phase two I accepted a value, and then the  acceptor sends back this value to the proposer.  And when the proposer gets a value from another node, from someone else's proposal, it has  to start using this value instead of its own.  And it fixed the value with the highest proposal number.  So now it throws away its own value and it uses someone else's value.  And this is kind of the trick that makes Pac-Sos work, and I'll show you why in just  a sec.  So the second phase of Pac-Sos, I'm going to send messages to all the nodes that are  participating.  Please accept this value, could be my value, could be someone else's value, for this proposal  number, I. And then the acceptors could say, okay, I accept your value, or they could say,  well, I already received a competing proposal with a higher sequence number, someone else  ran phase one, and I promised I would not participate in proposals with lower sequence  numbers, so sorry.  And then the proposer starts over with a higher sequence number, and it could take a while.  And then if anything goes wrong, for some reason you do not get the majority to say,  okay, you also start over.  And finally, there's usually a phase three, like you don't want to go through all this  stuff again.  So if you manage to get confirmation that tree nodes, or sorry, the majority of nodes,  so that's like two out of three, or three out of four, or three out of five, accept  the value, you just tell everyone else, but that they don't have to rerun Pac-Sos.  But again, that could be failures here, and so it's not actually a critical step, you  could even skip this step.  It's more of an optimization.  So why does this work?  So actually, this algorithm is sufficient to provide the guarantee that it will only  ever return the same value on all the nodes, and the value will be one of the inputs.  So why will it never return a different value?  Well, if you kind of think of it backwards, and Pac-Sos was actually kind of derived backwards  from the notion of consensus.  So it was more of a derived algorithm than an invented algorithm.  So if a majority of nodes accept my proposal, that means that no other node has completed  phase one since I did.  Because if another node managed to complete phase one, I wouldn't have gotten all my  OKs.  One of them would have said, no, I will not participate.  Because to get my, like if someone completed phase one, he would have needed a majority,  and at least one node is going to be in common with my majority.  That's the trick with the majorities.  Two majorities always have at least one node in common.  So because I completed phase one, phase two, it means that any other proposal still needs  to complete phase one, and thus will see my value because they will run into this.  Because they still need to complete it, so they will definitely run into this for at  least one node.  And it's also guaranteed that my value has the highest proposal number that got accepted  since no one else has managed to complete phase one since I did.  So every other proposal that got to phase two had a lower proposal number.  So basically, just getting a majority of accepts in Paxos tells me this is now the value that  is forever going to be returned by Paxos for a particular key.  How is that useful?  Like, this is a right once algorithm.  You can set the value once, and then it never changes.  Well, actually, I can do something which the server systems people, we call a, they'll  they replicate a state machine where the idea is if you have a bunch of nodes which all  have the same state machine and the same initial state, and you feed them all the same inputs,  they will be in the same final state.  So what I can do with Paxos is replicate a log of changes.  A log is a right once thing, a right only thing.  So I can say Paxos, I can use the key as kind of the sequence number in the log, or the  sequence number in the log of the key, and my first item could be set x equals six, and  then the second item could be set i equals seven, and then the third item could be set  i equals nine.  So if I know about all these three items, I know that the value of x equals six, and  the value of y equals nine.  So what I can do with Paxos is implement kind of this replicated log, and I can always make  sure that I know what's in the log just by running Paxos for all these values.  And this technique is actually called multi-Paxos, because you run Paxos multiple times.  So the way I write to this log is I run Paxos for whatever I believe to be the current  round number, or basically at the end of the log, and then I run it, and maybe Paxos has  already completed, and I just get back someone else's value, and I say, okay, well, I'll  try with the next highest round number, or what I now believe to be the highest round  number.  And maybe someone else comes in, completes phase one before I do, my proposal loses,  and then I get back his value from Paxos.  Okay, I try again with a higher round number.  Eventually, hopefully, I'm going to get my value appended to the log.  So my command is going to have some round number where it has reached consensus.  To perform a read, what I need to do is I'm going to ask all the nodes, or at least the  majority of nodes, what's the highest round number for which you accepted a value, knowing  that that is the highest round number which could have consensus at the time that my read  started.  And if I make sure I know all the values in the log up to that round number, I'm good.  Like my log is entirely up to date, up to the point when my read started, I'm going  to see any preceding read.  I'm going to get fresher information than any preceding read.  Sorry, I'm going to see an preceding write.  I'm going to get fresher information than any preceding read.  But I do need to provide some value when I call Paxos.  Like it takes a value as a parameter.  So what I could do is just give an empty string as the value when I want to do a read.  In this case, something went wrong somewhere and no value actually reached consensus in  the round.  And it could even be that my empty string will actually reach consensus in that round.  That's fine.  Like nothing bad happens, this just means nothing changes.  So sometimes you could have empty entries in the log because of this.  And so each note has its own copy of the log and it could be arbitrarily far behind.  If a note fails and doesn't get updates for an hour like its log, it's going to be far  behind.  But it can almost just run the consistent read algorithm.  Say note C comes back, say, okay, I want to do a consistent read on my state.  Let's say my table.  Now I'm going to ask note A and B what's the highest round number for which you accepted  the value?  So note B will say two.  Let's say note B will say one.  So note C runs paxos for this.  Somehow it actually already accepted the value here.  Maybe someone tried this insert and then crashed or whatever, something went wrong.  Actually another value reached consensus.  So when note C runs paxos for round one, it'll get back that value when it runs paxos for  round two.  It'll get back this value.  I mean, it already accepted this value but didn't know whether there was consensus probably.  So at that point note C is completely up to date with any proceeding writes so it can  return a result.  So I implemented this in Postgres as an extension.  So called pgpaxos.  So it's an extension you can just install in Postgres.  I think it works 9.4, 9.5, possibly works on 9.6, I haven't tried.  And it provides consistent, full tolerant, multi-master table replication where you can  tolerate a minority of notes going down.  So if you have three notes, you can tolerate one going down.  The trade-off that you're making is that the right throughput is going to be incredibly  low because you cannot really have concurrent writes.  You can only have one write at a time for a particular log item otherwise they're going  to compete and that just makes both of them take longer.  So the right throughput is low, think like maybe 100 per second and the latency is generally  going to be high.  There's actually no guarantee on how high it could be.  It could be very high.  If two notes are down, let's say, then there's no way, and there are three notes in total,  there's no way to get a majority so pgpaxos will take forever until one of the notes comes  back.  So that's the trade-off you're making, but there's no other way to get consistent, full  tolerant, multi-master table replication.  So it's not really an alternative to other replication solutions like streaming replication  or logical replication.  They have a particular purpose and it doesn't replace them in any way.  It also doesn't magically make Postgres distributed.  There will never really be such a thing as distributed Postgres.  In distributed systems, there's not really solutions, there's just different trade-offs  you can make.  But it can be a very useful building block, especially for distributed systems in the  way that you wouldn't do your own data storage.  You wouldn't like to know, but most of the time, you don't want to write your own serialization  algorithms to write things to disk and take care of truncation and concurrency and whatnot.  You use a database.  Similarly, there are some very hard problems that pgpaxos solves which you probably don't  want to solve yourself, and you can use pgpaxos for those.  I'll give you some examples later.  So pgpaxos is available on GitHub under PostgresGL license.  So it contains a very basic implementation of paxos and multipaxos, which is the state  machine written in PLPG SQL, which sounds a bit weird, but it's a surprisingly suitable  language for implementing distributed algorithms because a lot of the time, one of the things  that makes paxos hard is you really need to be very careful about your local state and  making sure you write that to disk in case there's a crash before you give any response,  making sure you deal with truncation, making sure you serialize data.  And actually, all of that comes for free in PLPG SQL.  I just do an insert on a table and then let Postgres do the work.  And PLPG SQL has a networking API, DB link.  It's not so great, but it works.  And it does make it easy to do remote procedure calls.  And the other part of pgpaxos is basically C code, which uses the planner and executor  hooks in Postgres to provide a replication for particular tables that you want to replicate.  So you can mark certain tables as replicated, and then any insert, for example, you do,  is going to be replicated to the other nodes using paxos.  And I should say it is somewhat experimental.  Like if someone, like the most useful thing if you want to contribute to this project  is actually testing, I wouldn't, I don't feel like we're missing a lot of features,  but we're definitely, like I've done a lot of testing, but like the industry standard  here for these types of systems is actually Jepsen and it's actually quite a lot of work  to do Jepsen test.  But any type of testing you can do would be very useful in making people confident that  this is a robust system.  So as I said, PLPG SQL is actually surprisingly suitable for implementing distributed algorithms.  You get transactional semantics.  It's easy to collect a bunch of responses in a table and do some query on them.  There's a simple networking API.  RPC, remote procedure call, becomes pretty simple.  I can just do create function, which is the one I call on the remote end, and then create  another function which uses dblink to call the other function.  So it's pretty simple.  And because it's PLPG SQL and Amazon RDS and Heroku both have dblink installed, I could  even run this on Amazon RDS.  I couldn't do the sort of transparent table replication, but the paxos functions, they  would work.  So if you create extension pgpaxos, you will get a bunch of tables, kind of boring, but  just for the concepts.  There's a group table, which is the paxos groups in which this node participates.  So a node can be in many groups at the same time.  Host table is just the members of the group.  The round table is a bit of a forename for the log.  So this contains the actual replicated SQL queries.  And there are sort of replicated tables, which keeps track of which tables should be replicated  using pgpaxos.  And the paxos extension, pgpaxos also adds some functions.  One is just the paxos function that I've already described.  Pretty much works exactly like that.  The apply log function is pretty much the read, the consistent read function I've described.  So if you give it like the max round number in the group, it just makes sure that the  log is up to date.  Apply and append is the write function I described.  So it keeps trying to get consensus on its query for a particular round, and if it fails,  it goes to the next round.  But you don't really need to worry about these functions so much unless you wanted to implement  your own replication solution or log.  Also the time what you do is you create a table, and then on the first note in the paxos  group you do a create group where you specify your own host name, such that other notes  know how to connect to you.  And then you do replicate table.  So the data refers to the table.  And after this, any insert or update on the table will get replicated, added to the paxos  log, and any select will make sure that your log is up to date before the select is actually  executed.  So on the first note you do those three steps, and then on the second and third and whatever  note, you do join group.  So you specify one of the existing servers in the group.  I guess they usually do original one, and then your own IP.  And what this actually does is quite interesting, like if a node joins paxos, the world kind  of changes.  Like, before the node joins, maybe my majority was two nodes, but after the node joins, now  my majority is three nodes.  So I need to be aware of that.  So the way that works in paxos is actually the fact that this node joins is added to  the log.  So it's that any subsequent query will actually see this new node.  So what happens when there is an update?  So as a user, you just do update the table, set to do whatever.  Then in the background, paxos runs this, pgpaxos runs this, paxos apply an append function  with more or less your query.  It kind of deparces it from the postgres query plan, but more or less it will log back.  And then it keeps trying to append.  And then when it learns, okay, now I've appended it at log position 51.  So I'm going to make sure that items 0 to 50 have been executed, and then I'm going  to execute the update, such that my update works on a consistent copy of the table.  And this makes basically all writes completely serializable.  They're all executed in the same order on all the machines.  Then you do a select.  Basically you run the read algorithm.  So in the background, it didn't transparent to the user usually.  Paxos max group round gets talks to the other nodes, asks them what's the highest round  number for which you accepted a value, and then it gets its log up to date up to that  point.  And it knows that when the select started, like any round number higher than that could  not have had consensus, because at least one node in the majority would have accepted a  value for it.  There's some subtle to this when, like, new nodes join, then you have to repeat the max  group round.  But this is the general principle.  So I'd like to do a little demo.  But are there any questions so far?  So how do you deal with transactions?  So what you can do, and this is more of a manual, oh, sorry, the question was how do  you deal with transactions?  And essentially, like, the transparent table replication is more or less single statement.  So you can't really do begin statement rollback or something.  What you could do just using the faxos function is just log the whole string of your transaction,  like begin, update, delete, insert, commit.  You could log that as the entire string, and then it'll get executed in order along with  the other writes.  Yeah, that'd be the more practical way of doing it.  It would also save you on network traffic.  But yeah, that's entirely possible.  All right, so I want to show you a demo of a distributed locking system.  So I don't actually need PetitGPoxos and the nodes on which I'm doing work to be the same  thing.  Oh, sorry, Jeff.  So if you have a single thing in the node, you'd be able to do that.  So if you were able to do that, you'd be able to do that with the reportage, the state,  the security, the storage, the time, and so that would be very, very safe to use there  and make sure you got this in the middle of the transaction.  No, I mean, it'll end up, let's say, when you do a consistent read and there's a bunch  of things in the log that you need to execute, it'll end up just doing those in one transaction,  basically.  Is that the question?  Well, any state that's recorded during the call to PetitGPoxos is actually done over  the DB links.  So you don't, like, PetitGPoxos doesn't even, like, the PetitGPoxos function doesn't even  store state on itself.  It's assumed to be completely external.  So it just talks to the other nodes as if, you know, they were external.  It could be talking back to itself.  But so let's say in, oh, this is pretty far back.  Let's say when this node responds saying, okay, I accept the value, I mean, that's a  separate transaction, and so actually the PetitGPoxos function doesn't record any state  at all.  Everything happens in those remote procedure calls.  All right.  The round number needs to be persistent.  You need to keep track of how much of the log you have executed.  So whenever you do an execute, you also update the round number of, like, I've executed this  much of the log.  Yeah, and you just find that from everyone else.  Yeah.  Yeah, there's a separate table for keeping track of, like, which log you've applied,  and then there's a table for the log itself, which contains all the items.  So I'm going to show you a demo where PetitGPoxos is used to implement the distributed locking  service, and the motivation here is doing distributed chron.  This is actually quite a tricky problem if you have, let's say, a job that absolutely  must happen every minute, but you don't want it to happen twice.  So you could set up one node, and then it runs this job every minute, but if that node  fails, then it doesn't happen.  You could set up two nodes that both run it every minute, but that's probably not what  you want.  So you want exactly one of them to do it.  So this is kind of a locking problem.  Of course, you could set up a database where you, which has a table of locks, but then  if the database goes down, you still have the same problem.  But with PetitGPoxos, you can actually solve this.  Like, so I've set up a little PetitGPoxos cluster using very small instances.  I'm running this on EC2, so I'm using the T2 micros.  Paxos doesn't have very high throughput anyway, so you might as well just use small instances.  I put a load balancer in front of it, so it's that the chron node can just, will just get  rooted to whichever Paxos node is up and running.  And then they both try to do an insert into the same locks table with, which has a primary  key on the name of the lock, and only one of them will succeed, the other one will fail  and does know that the other guy has the lock.  So yeah, as I said, I set up my cluster on EC2.  There's three PetitGPoxos nodes and two slightly bigger chron nodes.  They're not going to have to do a lot of work, but that's just how you do it.  So the chron nodes, let's say, for example, you're going to have a system bank, you're  going to have a system that will have a count on the node, but not quite the same.  So the list of all the accounts and all the stuff using the chron one and chron two nodes.  So the application I'm going to do is much simpler, I'm going to send a tweet every minute.  But in that example, yeah, I mean, let's say there could be some additional database in  which that is stored or they do it locally, it's kind of independent of Paxos itself.  I guess technically you could actually store the accounts in Paxos as well.  So what is the chron one and chron two?  I mean, it's technically possible.  I mean, usually you want these guys to probably have some computing power depending on what  they want to do.  I mean, like you can, you don't have to, let's say.  I mean, you could have the chron nodes be co-located with the PgPoxos nodes or you could use them  separately.  It's kind of all the same.  Here it's because I use these super tiny instances, the PgPoxos cluster costs almost nothing  anyway.  Yes, I can, I'll show you the Paxos block.  Yeah, but at least if a node goes down, you don't have to actually immediately react to  it.  So you can automate a lot probably.  So I have a bunch of nodes set up here with, this is one of my PgPoxos nodes.  So it has the PgPoxos extension.  I have a LOX table which has like a key and an owner and the key is the primary key.  So I can only have one owner of the key at a time.  And on this node, I will create a Paxos group and then replicate a table within that Paxos  group.  Now, that's not very interesting.  So I'm going to join some additional nodes.  This is one of my other Paxos nodes.  And the number it's showing is actually the round in the Paxos log in which it's joined.  It's kind of an internal thing.  So at this point, I can do an insert, let's say, on this node.  And then if I go to one of my other nodes and try to do the same insert, it's going  to give me a primary key violation because they're now looking at a replicated table so  only one insert will succeed.  So if I, this is the third node, I can see there's now one item in my log.  I can make it a bit more challenging.  Let's say I do an insert on one node and then immediately a select on the other node.  I should then get that insert back, right?  I should see I inserted AB and now I see AB on the other node in the selected.  This is guaranteed by Paxos.  So I set up a load balancer in front of my Paxos node so this just connects me to any  of the nodes.  I don't know which one.  I can't actually really distinguish them.  They all have the same content.  And I have a little script which will just send a tweet.  And I have this same script on three nodes, two nodes, sorry, two ground nodes.  And only one of them will send it.  So I'll activate the script.  I just have a, this and then Kron runs at, you know, when the number of seconds reaches  zero so I could quickly open the Twitter account.  No tweets yet.  So, from what Kron is it runs once a minute.  Okay, so someone grabbed the lock, a note 0.171, grabbed the lock so I should now be  getting a tweet.  Yes.  And then both nodes are running the script but only one of them will succeed.  And it's actually not like the clocks in Amazon data centers are really well synced  so they run Kron almost at exactly the same time.  So it kind of tends to alternate between both nodes if they're both up and running.  So one thing I can do now is I can maybe stop one of my nodes, my Paxos nodes.  If now I do this then I've got a bunch of warnings, I still need to kind of clean up  the output of Paxos because it shows all the stack phrases of the LPG SQL.  But the point is it did manage to do the select even though a note was down.  It'll make one more connection attempt and then after that it will just, you know, pretend  it doesn't make it.  Which actually makes it even, makes it a bit faster, surprisingly enough because now  it only needs to connect to two nodes.  So there should be another tweet by now.  Right, there's another tweet.  Obviously I can also take down this node so I've stopped the Kron job over here.  So at some point I should be seeing, in 26 seconds I should be seeing that the other  node still manages to send the tweet, I'm not sure if Twitter shows that kind of in  a live way.  But so I've taken down one of the Paxos nodes, I've stopped the Kron job on one of my Kron  nodes.  So I could model this, let's say I put this in three different data centers, three different  availability zones.  I could lose an entire data center and things will just work fine.  Which is quite nice about the G-Paxos.  Right, so now the other node did the tweet.  So not to completely spam the PGCon hashtag, I'm going to shut off the Kron job here.  Okay.  Any questions about the demo?  All right.  So what could you use this for?  So this is an example of distributed locks.  I didn't go into this, I kind of used a cheap trick here where the name of the lock that  these Kron jobs are using is actually the number of minutes times since 1970.  It doesn't actually rely heavily on synchronized clocks.  If one clock is ahead, then most of the time that one will just win because it will be  the first one to grab the lock, but then if that fails, then the other guy, there might  be some, you know, several minutes where, let's say the other guy's several minutes  behind on its clock that it might take a while before it starts tweeting again.  But in Amazon data centers, it's usually synchronized up to the millisecond, so it's not really  an issue.  So PG-Paxos works well for any application that has low read, write volumes, but strong  consistency requirements and no strong latency requirements.  So distributed locks is one.  If you have this type of Kron job, any type of resource management, like we can only  have one node access a certain resource at a time, it works quite well for that.  Also for managing cluster membership, I mean PG-Paxos manages its own cluster, but it can  also manage other clusters.  And this is actually one of the most common applications for Paxos.  And this, for example, also includes deciding which node is the primary.  And so you could imagine an application where you use Paxos to decide which out of a streaming  replication group gets to be the primary.  And then Paxos can tolerate one of the nodes failing.  You could use it as a job scheduler.  Alvaro recently did a blog post showing how you could use PG-Paxos to implement a job  queue which has, like, exactly one semantics, whereas most queuing systems have at least  one semantics, so the job could happen multiple times.  Migrations, if you want to move over a lot of data from one place to another, you can  scheme on migrations.  Or as a source for metadata, like often if you have a system where, let's say, all nodes  have a copy of the metadata, there's one authoritative source.  But if you don't want to be subject to that source going down, you would have to use something  like PG-Paxos to make sure that you have a replicated consistent copy of the metadata.  So why not raft?  I get this question a lot.  Well, one is, Paxos can be implemented in PLPG SQL, which makes it quite suitable for  implementing in Postgres, because it's actually really simple.  And there's almost like a one-to-one mapping of sentences in the paper to SQL queries  in the source code, because you don't have to worry about transactions and interpretation  and serialization.  That's a good question.  The question is, what is raft?  So then it's a slightly newer consensus protocol.  It works by kind of electing a master, basically.  It has quite a good algorithm for quickly switching, making one of the nodes master,  and if the other one goes down very quickly, another one becomes the master.  And well, one of the differences, like this has been a recent paper, and the paper was  very instructive to engineers.  It told you exactly what to do, whereas the Paxos papers are a bit more mathematical.  So raft has picked up a lot of steam lately in newer distributed databases, because the  paper is easier to implement in some sense.  So actually Paxos, as an algorithm, is much simpler.  The minimal implementation that works is a lot simpler.  You can kind of play with it more to be adapted to your requirements.  I really wanted it to be multi-master, so I could do writes from anywhere.  And mathematically, it's very elegant.  Like, in a way, Lemper didn't invent it.  He just derived it from the notion of consensus, like, what would I need to do?  And it's actually also optimal for achieving consensus.  But actually, the more realistic answer probably is, I knew Paxos quite well.  I knew PLPG SQL quite well, and I figured, I think I could implement Paxos in PLPG SQL.  So that's my story.  I hope you enjoyed it.  Any questions?  So what do you want to do with the PXO?  Do you want to do a PG Paxos on the PXO?  No, because PG Paxos keeps its date in tables.  So on the secondaries, yeah, I suppose you could work for logical replication.  Or if you create another database cluster that's co-located with these nodes.  Yeah, unfortunately, it doesn't work on secondaries.  Josh?  I was going to ask you, too, about the user traffic cluster.  But do they need to know whether or not the map is not supported by the user traffic  package?  Yeah.  I would typically recommend running it in this type of setup where PG Paxos is actually  an external component running on tiny hardware, because then it's just already up and running  when you bring up your big data production cluster so you don't really have that problem.  And you could reuse the PG Paxos cluster for other stuff.  It's a bit similar in the way people use ZooKeeper.  They just have this external ZooKeeper deployment which solves a similar problem.  So the question is what is adding a node like?  And it's essentially just you set up Postgres, you install the PG Paxos extension.  You also need in Postgresql.conf to add it to shared preload libraries.  And once you have a node, it's pretty much just you call this join group.  Actually, there's one step missing here.  You first need to create the same tables.  It doesn't replicate the DDL yet.  It seems to be a common problem with replication solution.  But then you just call Paxos join group and this actually more or less clones the state  of an existing node.  So any data that's already in the table, it copies over.  The membership table, it copies over.  At this point, it might be interesting to show this.  This is actually the Paxos log that I've been talking about.  So these are the queries that have been executed on the table.  But yeah, joining a node is mostly setting up the tables that are replicated and then  calling join group.  And then join group also makes sure that all the other nodes know about this node.  And there's a corresponding leave group, which gives some errors because the other node is  down for errors warning.  But now it's left a group.  So it's an elastic load balancer.  So the way elastic load balancer works, so this is an Amazon service, it sets up EC2  instance more or less in every availability zone in which you have instances and then  sets up users root 53 to set up DNS records pointing to each of these things.  So let's say one data center goes down, like I took down one of my Paxos nodes, one of  my Cron nodes, one of these guys will also go down, but I still have two left.  So my clients can just talk to any of those two.  And ELB is constantly doing health checks, so ELB will detect, oh, this Paxos node is  no longer running, don't send any traffic there.  So actually the single point of failure is kind of picking care off in ELB.  But yeah, I mean, it's trickier in your own kind of architecture to do that.  Probably the simple.  Yeah, pretty much.  It depends.  So Amazon or the load balancer actually is a load balancer, so any new connection will  typically go to a different one of these machines.  And I mean, it performs constant health checks to figure out who's up and running.  But there's no like, you know, Amazon doesn't really pick like, I'm always going to send  my traffic to this machine.  Every time you connect, you actually get connected to a different machine.  So you have to forget about it, you have to master it, and it's just a return.  And I guess that kind of thing, Amazon has to pick when you do.  Who will run more than that?  Who will run more than that?  Who will run more than that?  Because you don't know if you have to do it with a friend.  Well, if you try like with an ELB and you disconnect and reconnect and disconnect and  reconnect, pretty much always it connects to a different one.  It actually just routes or picks a node based on your source port, kind of uses a hashing  function of that source port.  At least used to be the algorithm.  I used to be at Amazon, so I know what it was like two years ago.  I don't know what it is now.  Yeah, no, it does.  I think actually, so the question is, there's no conceptual technical reasons that you couldn't  replicate BDL.  And I'm trying to remember, I think actually if you do something like create index, it actually  does replicate it because it's kind of easy.  Alter table is a bit tricky because PGPUXO is kind of just before execution when the  planning has already been done.  And if then there's an alter table happening before that, then the execution no longer makes  sense.  So that's, or plan is no longer correct.  So that's, there's some limitations around that, but doing something like create table  actually isn't a problem, but then there's nothing to activate the PAXO's log.  Once I create the locks table, that's when I start replaying the log.  But if I don't have a locks table, then how do I start?  I could just create a function for that or something, but it's not there at the moment.  Good question.  So if the PAXO's log rotated, currently it is not.  It will grow infinitely.  And yeah, there's also no reason it couldn't be rotated.  Pretty much if you know all your notes have reached a certain point, then you can just  drop everything that came before then.  There's just currently no logic to do that.  All right.  That's my time.  Thank you for listening.  And if you have any other questions, I'll be around.
pg_paxos_Paxos_based_table_replication_d6IZuLt8_Ao_0_131.png pg_paxos_Paxos_based_table_replication_d6IZuLt8_Ao_132_1127.png pg_paxos_Paxos_based_table_replication_d6IZuLt8_Ao_1128_4715.png pg_paxos_Paxos_based_table_replication_d6IZuLt8_Ao_4716_7247.png pg_paxos_Paxos_based_table_replication_d6IZuLt8_Ao_7248_12587.png pg_paxos_Paxos_based_table_replication_d6IZuLt8_Ao_12588_14243.png pg_paxos_Paxos_based_table_replication_d6IZuLt8_Ao_14244_14447.png pg_paxos_Paxos_based_table_replication_d6IZuLt8_Ao_14448_15275.png pg_paxos_Paxos_based_table_replication_d6IZuLt8_Ao_15276_17459.png pg_paxos_Paxos_based_table_replication_d6IZuLt8_Ao_17460_22619.png pg_paxos_Paxos_based_table_replication_d6IZuLt8_Ao_22620_26183.png pg_paxos_Paxos_based_table_replication_d6IZuLt8_Ao_26184_31091.png pg_paxos_Paxos_based_table_replication_d6IZuLt8_Ao_31092_32339.png pg_paxos_Paxos_based_table_replication_d6IZuLt8_Ao_32340_33131.png pg_paxos_Paxos_based_table_replication_d6IZuLt8_Ao_33132_34343.png pg_paxos_Paxos_based_table_replication_d6IZuLt8_Ao_34344_35543.png pg_paxos_Paxos_based_table_replication_d6IZuLt8_Ao_35544_41207.png pg_paxos_Paxos_based_table_replication_d6IZuLt8_Ao_41208_41723.png pg_paxos_Paxos_based_table_replication_d6IZuLt8_Ao_41724_45563.png pg_paxos_Paxos_based_table_replication_d6IZuLt8_Ao_45564_45839.png pg_paxos_Paxos_based_table_replication_d6IZuLt8_Ao_45840_45887.png pg_paxos_Paxos_based_table_replication_d6IZuLt8_Ao_45888_47303.png pg_paxos_Paxos_based_table_replication_d6IZuLt8_Ao_47304_48479.png pg_paxos_Paxos_based_table_replication_d6IZuLt8_Ao_48480_49163.png pg_paxos_Paxos_based_table_replication_d6IZuLt8_Ao_49164_49379.png pg_paxos_Paxos_based_table_replication_d6IZuLt8_Ao_49380_49475.png pg_paxos_Paxos_based_table_replication_d6IZuLt8_Ao_49476_49955.png pg_paxos_Paxos_based_table_replication_d6IZuLt8_Ao_49956_50099.png pg_paxos_Paxos_based_table_replication_d6IZuLt8_Ao_50100_50291.png pg_paxos_Paxos_based_table_replication_d6IZuLt8_Ao_50292_50519.png pg_paxos_Paxos_based_table_replication_d6IZuLt8_Ao_50520_51959.png pg_paxos_Paxos_based_table_replication_d6IZuLt8_Ao_51960_52007.png pg_paxos_Paxos_based_table_replication_d6IZuLt8_Ao_52008_52055.png pg_paxos_Paxos_based_table_replication_d6IZuLt8_Ao_52056_52103.png pg_paxos_Paxos_based_table_replication_d6IZuLt8_Ao_52104_52151.png pg_paxos_Paxos_based_table_replication_d6IZuLt8_Ao_52152_52235.png pg_paxos_Paxos_based_table_replication_d6IZuLt8_Ao_52236_52283.png pg_paxos_Paxos_based_table_replication_d6IZuLt8_Ao_52284_52523.png pg_paxos_Paxos_based_table_replication_d6IZuLt8_Ao_52524_52571.png pg_paxos_Paxos_based_table_replication_d6IZuLt8_Ao_52572_52727.png pg_paxos_Paxos_based_table_replication_d6IZuLt8_Ao_52728_52787.png pg_paxos_Paxos_based_table_replication_d6IZuLt8_Ao_52788_52835.png pg_paxos_Paxos_based_table_replication_d6IZuLt8_Ao_52836_53327.png pg_paxos_Paxos_based_table_replication_d6IZuLt8_Ao_53328_53483.png pg_paxos_Paxos_based_table_replication_d6IZuLt8_Ao_53484_53603.png pg_paxos_Paxos_based_table_replication_d6IZuLt8_Ao_53604_54095.png pg_paxos_Paxos_based_table_replication_d6IZuLt8_Ao_54096_54539.png pg_paxos_Paxos_based_table_replication_d6IZuLt8_Ao_54540_54815.png pg_paxos_Paxos_based_table_replication_d6IZuLt8_Ao_54816_54899.png pg_paxos_Paxos_based_table_replication_d6IZuLt8_Ao_54900_55811.png pg_paxos_Paxos_based_table_replication_d6IZuLt8_Ao_55812_55931.png pg_paxos_Paxos_based_table_replication_d6IZuLt8_Ao_55932_56207.png pg_paxos_Paxos_based_table_replication_d6IZuLt8_Ao_56208_65519.png pg_paxos_Paxos_based_table_replication_d6IZuLt8_Ao_65520_66215.png pg_paxos_Paxos_based_table_replication_d6IZuLt8_Ao_66216_66659.png pg_paxos_Paxos_based_table_replication_d6IZuLt8_Ao_66660_67379.png pg_paxos_Paxos_based_table_replication_d6IZuLt8_Ao_67380_67835.png pg_paxos_Paxos_based_table_replication_d6IZuLt8_Ao_67836_67907.png pg_paxos_Paxos_based_table_replication_d6IZuLt8_Ao_67908_74713.png