diff --git a/ipam/paxos/paxos.go b/ipam/paxos/paxos.go new file mode 100644 index 00000000..fd67283d --- /dev/null +++ b/ipam/paxos/paxos.go @@ -0,0 +1,261 @@ +package paxos + +import ( + "fmt" + "github.com/weaveworks/weave/router" +) + +// The node identifier. The use of the UID here is important: Paxos +// acceptors must not forget their promises, so it's important that a +// node does not restart and lose its Paxos state but claim to have +// the same ID. +type NodeID struct { + Name router.PeerName + UID router.PeerUID +} + +// note all fields exported in structs so we can Gob them +type ProposalID struct { + // round numbers begin at 1. round 0 indicates an + // uninitialized ProposalID, and precedes all other ProposalIDs + Round uint + Proposer NodeID +} + +func (a ProposalID) precedes(b ProposalID) bool { + switch { + case a.Round != b.Round: + return a.Round < b.Round + case a.Proposer.Name != b.Proposer.Name: + return a.Proposer.Name < b.Proposer.Name + case a.Proposer.UID != b.Proposer.UID: + return a.Proposer.UID < b.Proposer.UID + default: + return false + } +} + +func (a ProposalID) valid() bool { + return a.Round > 0 +} + +// For seeding IPAM, the value we want consensus on is a set of peer names +type Value []router.PeerName + +// An AcceptedValue is a Value plus the proposal which originated that +// Value. The origin is not essential, but makes comparing +// AcceptedValues easy even if comparing Values is not. +type AcceptedValue struct { + Value Value + Origin ProposalID +} + +type NodeClaims struct { + // The node promises not to accept a proposal with id less + // than this. + Promise ProposalID + + // The accepted proposal, if valid + Accepted ProposalID + AcceptedVal AcceptedValue +} + +func (a NodeClaims) equals(b NodeClaims) bool { + return a.Promise == b.Promise && a.Accepted == b.Accepted && + a.AcceptedVal.Origin == b.AcceptedVal.Origin +} + +type GossipState map[NodeID]NodeClaims + +type Node struct { + id NodeID + quorum uint + knows GossipState +} + +func NewNode(name router.PeerName, uid router.PeerUID, quorum uint) *Node { + return &Node{ + id: NodeID{name, uid}, + quorum: quorum, + knows: map[NodeID]NodeClaims{}, + } +} + +func (node *Node) GossipState() GossipState { + return node.knows +} + +// Update this node's information about what other nodes know. +// Returns true if we learned something new. +func (node *Node) Update(from GossipState) bool { + changed := false + + for i, fromClaims := range from { + claims, ok := node.knows[i] + if ok { + if claims.Promise.precedes(fromClaims.Promise) { + claims.Promise = fromClaims.Promise + changed = true + } + + if claims.Accepted.precedes(fromClaims.Accepted) { + claims.Accepted = fromClaims.Accepted + claims.AcceptedVal = fromClaims.AcceptedVal + changed = true + } + } else { + claims = fromClaims + changed = true + } + + node.knows[i] = claims + } + + return changed +} + +func max(a uint, b uint) uint { + if a > b { + return a + } else { + return b + } +} + +// Initiate a new proposal, i.e. the Paxos "Prepare" step. This is +// simply a matter of gossipping a new proposal that supersedes all +// others. +func (node *Node) Propose() { + // Find the highest round number around + round := uint(0) + + for _, claims := range node.knows { + round = max(round, claims.Promise.Round) + round = max(round, claims.Accepted.Round) + } + + ourClaims := node.knows[node.id] + ourClaims.Promise = ProposalID{ + Round: round + 1, + Proposer: node.id, + } + node.knows[node.id] = ourClaims + + // With a quorum of 1, we can immediately accept our proposal + if node.quorum == 1 { + node.Think() + } +} + +// The heart of the consensus algorithm. Return true if we have +// changed our claims. +func (node *Node) Think() bool { + ourClaims := node.knows[node.id] + + // The "Promise" step of Paxos: Copy the highest known + // promise. + for _, claims := range node.knows { + if ourClaims.Promise.precedes(claims.Promise) { + ourClaims.Promise = claims.Promise + } + } + + // The "Accept Request" step of Paxos: Acting as a proposer, + // do we have a proposal that has been promised by a quorum? + // + // In Paxos, the "proposer" and "acceptor" roles are distinct, + // so in principle a node acting as a proposer could continue + // trying to get its proposal acccepted even after the same + // node as an acceptor has superseded that proposal. But + // that's pointless in a gossip context: If our promise + // supersedes our own proposal, then anyone who hears about + // that promise will not accept that proposal. So our + // proposal is only in the running if it is also our promise. + if ourClaims.Promise.Proposer == node.id { + // Determine whether a quorum has promised, and the + // best previously-accepted value if there is one. + count := uint(0) + var accepted ProposalID + var acceptedVal AcceptedValue + + for _, claims := range node.knows { + if claims.Promise == ourClaims.Promise { + count++ + + if accepted.precedes(claims.Accepted) { + accepted = claims.Accepted + acceptedVal = claims.AcceptedVal + } + } + } + + if count >= node.quorum { + if !accepted.valid() { + acceptedVal.Value = node.pickValue() + acceptedVal.Origin = ourClaims.Promise + } + + // We automatically accept our own proposal, + // and that's how we communicate the "accept + // request" to other nodes. + ourClaims.Accepted = ourClaims.Promise + ourClaims.AcceptedVal = acceptedVal + } + } + + // The "Accepted" step of Paxos: If the proposal we promised + // on got accepted by some other node, we accept it too. + for _, claims := range node.knows { + if claims.Accepted == ourClaims.Promise { + ourClaims.Accepted = claims.Accepted + ourClaims.AcceptedVal = claims.AcceptedVal + break + } + } + + if ourClaims.equals(node.knows[node.id]) { + return false + } + + node.knows[node.id] = ourClaims + return true +} + +// When we get to pick a value, we use the set of peer names we know +// about. This is not necessarily all peer names, but it is at least +// a quorum, and so good enough for seeding the ring. +func (node *Node) pickValue() Value { + val := make([]router.PeerName, len(node.knows)) + i := 0 + for id := range node.knows { + val[i] = id.Name + i++ + } + return val +} + +// Has a consensus been reached, based on the known claims of other nodes? +func (node *Node) Consensus() (bool, AcceptedValue) { + counts := map[ProposalID]uint{} + + for _, claims := range node.knows { + if claims.Accepted.valid() { + origin := claims.AcceptedVal.Origin + count := counts[origin] + 1 + counts[origin] = count + if count >= node.quorum { + return true, claims.AcceptedVal + } + } + } + + return false, AcceptedValue{} +} + +func (node *Node) String() string { + if ok, val := node.Consensus(); ok { + return fmt.Sprintf("Consensus reached with size %d", len(val.Value)) + } else { + return fmt.Sprintf("Nodes known: %d, Quorum size: %d", len(node.knows), node.quorum) + } +} diff --git a/ipam/paxos/paxos_test.go b/ipam/paxos/paxos_test.go new file mode 100644 index 00000000..cd704c5c --- /dev/null +++ b/ipam/paxos/paxos_test.go @@ -0,0 +1,372 @@ +package paxos + +import ( + "fmt" + "github.com/weaveworks/weave/router" + "math/rand" + "testing" + "time" +) + +type TestNode struct { + *Node + + // Topology + links []*Link + isolated bool + + // The first consensus the Node observed + firstConsensus AcceptedValue +} + +type Link struct { + from *TestNode + to *TestNode + converse *Link + + // A link is considered ready if it is worthwhile gossipping + // along it (i.e. unless we already gossipped along it and the + // "from" node didn't change since then). If this is <0, then + // the link is not ready. Otherwise, this gives the index in + // the Model's readyLinks slice. + ready int +} + +type Model struct { + t *testing.T + r *rand.Rand + quorum uint + nodes []TestNode + readyLinks []*Link + nextID uint +} + +func (m *Model) addLink(a, b *TestNode) { + if a == b { + // Never link a node to itself + return + } + + ab := Link{from: a, to: b, ready: len(m.readyLinks)} + ba := Link{from: b, to: a, converse: &ab, ready: len(m.readyLinks) + 1} + ab.converse = &ba + a.links = append(a.links, &ab) + b.links = append(b.links, &ba) + m.readyLinks = append(m.readyLinks, &ab, &ba) +} + +func (m *Model) linkExists(a, b *TestNode) bool { + for _, l := range a.links { + if l.to == b { + return true + } + } + + return false +} + +type TestParams struct { + // Number of nodes + nodeCount uint + + // Probability that two nodes are connected. + connectedProb float32 + + // Probability that some node will re-propose at each + // step. Setting this too high makes it likely that we'll fail + // to converge. + reproposeProb float32 + + // Probability that some node will be isolated at each step. + isolateProb float32 + + // Probability that some node willbe restarted at each step. + restartProb float32 +} + +// Make a network of nodes with random topology +func makeRandomModel(params *TestParams, r *rand.Rand, t *testing.T) *Model { + m := Model{ + t: t, + r: r, + quorum: params.nodeCount/2 + 1, + nodes: make([]TestNode, params.nodeCount), + readyLinks: []*Link{}, + nextID: params.nodeCount + 1, + } + + for i := range m.nodes { + m.nodes[i].Node = NewNode(router.PeerName(i/2+1), + router.PeerUID(r.Int63()), m.quorum) + m.nodes[i].Propose() + } + + for i := 1; i < len(m.nodes); i++ { + // was node i connected to the other nodes yet? + connected := false + + for j := 0; j < i; j++ { + if r.Float32() < params.connectedProb { + connected = true + m.addLink(&m.nodes[i], &m.nodes[j]) + } + } + + if !connected { + // node i must be connected into the graph + // somewhere. So if we didn't connect it + // already, this is a last resort. + m.addLink(&m.nodes[i], &m.nodes[r.Intn(i)]) + } + } + + return &m +} + +// Mark a link as unready +func (m *Model) unreadyLink(link *Link) { + i := link.ready + if i >= 0 { + m.readyLinks[i] = m.readyLinks[len(m.readyLinks)-1] + m.readyLinks[i].ready = i + m.readyLinks = m.readyLinks[:len(m.readyLinks)-1] + link.ready = -1 + } +} + +// Mark a link as ready +func (m *Model) readyLink(link *Link) { + if link.ready < 0 { + link.ready = len(m.readyLinks) + m.readyLinks = append(m.readyLinks, link) + } +} + +// Mark all the outgoing links from a node as ready +func (m *Model) nodeChanged(node *TestNode) { + for _, l := range node.links { + m.readyLink(l) + } +} + +// Isolate a node +func (m *Model) isolateNode(node *TestNode) { + node.isolated = true + for _, l := range node.links { + if l.ready >= 0 { + m.unreadyLink(l) + m.unreadyLink(l.converse) + } + } + + // Isolating a node could partition the network. We don't + // want to test such a case (because it could prevent + // consensus). Checking for partitions would be more code + // than its worth, so just add some links to prevent the + // possibility of partitions. + for i := 1; i < len(node.links); i++ { + m.addLink(node.links[i-1].to, node.links[i].to) + } +} + +// Restart a node +func (m *Model) restart(node *TestNode) { + node.Node = NewNode(router.PeerName(m.nextID), + router.PeerUID(m.r.Int63()), m.quorum) + m.nextID++ + node.Propose() + + // The node is now ignorant, so we need to mark the links into + // the node as ready. + for _, l := range node.links { + if !l.to.isolated { + m.readyLink(l.converse) + } + } + + // If a consensus was just accepted due to this node accepting + // it, without other nodes hearing of it, and we then restart + // this node, then a different consensus can occur later on. + // If a tree falls with no one to hear it, does it make a + // sound? + node.firstConsensus = AcceptedValue{} +} + +func (m *Model) pickNode() *TestNode { + for { + node := &m.nodes[m.r.Intn(len(m.nodes))] + if !node.isolated { + return node + } + } +} + +func (m *Model) simulate(params *TestParams) bool { + nodesLeft := uint(len(m.nodes)) + restarts := uint(0) + + for step := 0; step < 1000000; step++ { + if len(m.readyLinks) == 0 { + // Everything has converged. This can be + // because consensus was reached, or because + // consensus became impossible, e.g. everyone + // promised on a particular proposal, but then + // the proposer restarted or was isolated. So + // we detect the latter cases and force a new + // proposal + for i := range m.nodes { + ok, _ := m.nodes[i].Consensus() + if ok { + return true + } + } + + node := m.pickNode() + node.Propose() + m.nodeChanged(node) + } + + // Pick a ready link at random + i := m.r.Intn(len(m.readyLinks)) + link := m.readyLinks[i] + if link.ready != i { + m.t.Fatal("Link in readyLinks was not ready") + } + + // gossip across link + node := link.to + if node.Update(link.from.GossipState()) { + node.Think() + + if !node.firstConsensus.Origin.valid() { + ok, val := node.Consensus() + if ok { + node.firstConsensus = val + } + } + + m.nodeChanged(node) + } + + m.unreadyLink(link) + + // Re-propose? + if m.r.Float32() < params.reproposeProb { + node := m.pickNode() + node.Propose() + m.nodeChanged(node) + } + + // Isolate? + if nodesLeft > m.quorum && m.r.Float32() < params.isolateProb { + m.isolateNode(m.pickNode()) + nodesLeft-- + + // We isolated a node, so get another node to + // re-propose. In reality the lack of + // consensus would be detected via a timeout + node := m.pickNode() + node.Propose() + m.nodeChanged(node) + } + + // Restart? + if restarts < m.quorum && m.r.Float32() < params.restartProb { + restarts++ + node := m.pickNode() + m.restart(node) + m.nodeChanged(node) + } + } + + return false +} + +func (m *Model) dump() { + for i := range m.nodes { + node := &m.nodes[i] + fmt.Println(node.id) + for n, claims := range node.knows { + fmt.Printf("\t%d %v\n", n, claims) + } + } +} + +// Validate the final model state +func (m *Model) validate() { + var origin ProposalID + + for i := range m.nodes { + ok, val := m.nodes[i].Consensus() + if !ok { + //m.dump() + m.t.Fatal("Node doesn't know about consensus") + } + + firstConsensus := m.nodes[i].firstConsensus + if firstConsensus.Origin.valid() && + firstConsensus.Origin != val.Origin { + //m.dump() + m.t.Fatal("Consensus mismatch") + } + + if i == 0 { + origin = val.Origin + } else if val.Origin != origin { + m.t.Fatal("Node disagrees about consensus") + } + } +} + +func TestSingleNode(t *testing.T) { + r := rand.New(rand.NewSource(time.Now().UnixNano())) + + // Test the single node case + params := &TestParams{ + nodeCount: 1, + connectedProb: 0, + reproposeProb: 0, + isolateProb: 0, + restartProb: 0, + } + + m := makeRandomModel(params, r, t) + + if !m.simulate(params) { + m.t.Fatal("Failed to converge") + } + + m.validate() +} + +func TestPaxos(t *testing.T) { + r := rand.New(rand.NewSource(time.Now().UnixNano())) + + params := &TestParams{ + nodeCount: 10, + connectedProb: 0.5, + reproposeProb: 0.01, + isolateProb: 0.01, + + // Restarts cause failures due to the case where 1) + // the gossip ordering is such that there is an + // effective partition from the start, with a quorum + // on one side and one node less than a quorum on the + // other; 2) the quorum reaches a consensus; 3) a node + // is restarted, and at the same time jumps the + // partition, so the side that was now just less than + // a quorum becomes a quorum; 4) the new quorum + // reaches a contradictory consensus. + restartProb: 0, + } + + for i := 0; i < 1000; i++ { + m := makeRandomModel(params, r, t) + + if !m.simulate(params) { + m.t.Fatal("Failed to converge") + } + + m.validate() + } +}