From bd185abb29d63921301d83969ac8247bc34749b2 Mon Sep 17 00:00:00 2001 From: David Wragg Date: Wed, 10 Jun 2015 11:16:22 +0100 Subject: [PATCH] Introduce PeerShortIDs These are needed to support vxlan encapsulation. In a vxlan packet, the only place we can straightforwardly hold the source and destination peers identifiers used by the weave router is in the 24-bit VNI (vxlan network identifier) field. This gives us 12 bits to indicate a peer. The resulting limit of 4096 peers on a weave network does not seem like a problem in the near future. But it does mean that decentralized allocation makes collisions fairly likely even in weave networks of sizes that are common today. Fortunately, the problem is much more tractable than IPAM because we can detect when collisions occur and take action to resolve them. TODO: We already notify the InterHost when a short id -> peer mapping changes, but we also need to flush any entries in the router MAC cache that might be affected. --- router/handshake.go | 14 ++- router/interhost.go | 6 + router/local_peer.go | 10 +- router/mocks_test.go | 2 +- router/peer.go | 16 ++- router/peers.go | 258 +++++++++++++++++++++++++++++++++++++------ router/peers_test.go | 238 +++++++++++++++++++++++++++++++++++++++ router/routes.go | 2 +- router/sleeve.go | 4 + 9 files changed, 514 insertions(+), 36 deletions(-) diff --git a/router/handshake.go b/router/handshake.go index 850f9a75..fbfc5fe3 100644 --- a/router/handshake.go +++ b/router/handshake.go @@ -61,6 +61,7 @@ func (conn *LocalConnection) handshake(enc *gob.Encoder, dec *gob.Decoder, accep nameStr, _ := fv.Value("Name") nickNameStr, _ := fv.Value("NickName") uidStr, _ := fv.Value("UID") + shortIDStr, _ := fv.Value("ShortID") remoteConnIDStr, _ := fv.Value("ConnID") if err := fv.Err(); err != nil { return err @@ -70,18 +71,27 @@ func (conn *LocalConnection) handshake(enc *gob.Encoder, dec *gob.Decoder, accep if err != nil { return err } + if !acceptNewPeer { if _, found := conn.Router.Peers.Fetch(name); !found { return fmt.Errorf("Found unknown remote name: %s at %s", name, conn.remoteTCPAddr) } } + if existingConn, found := conn.Router.Ourself.ConnectionTo(name); found && existingConn.Established() { return fmt.Errorf("Already have connection to %s at %s", existingConn.Remote(), existingConn.RemoteTCPAddr()) } + uid, err := ParsePeerUID(uidStr) if err != nil { return err } + + shortID, err := strconv.ParseUint(shortIDStr, 10, PeerShortIDBits) + if err != nil { + return err + } + remoteConnID, err := strconv.ParseUint(remoteConnIDStr, 10, 64) if err != nil { return err @@ -110,7 +120,8 @@ func (conn *LocalConnection) handshake(enc *gob.Encoder, dec *gob.Decoder, accep conn.tcpSender = NewSimpleTCPSender(enc) } - return conn.setRemote(NewPeer(name, nickNameStr, uid, 0)) + return conn.setRemote(NewPeer(name, nickNameStr, uid, 0, + PeerShortID(shortID))) } func (conn *LocalConnection) handshakeSendRecv(localConnID uint64, usingPassword bool, enc *gob.Encoder, dec *gob.Decoder) (*FieldValidator, *[32]byte, error) { @@ -122,6 +133,7 @@ func (conn *LocalConnection) handshakeSendRecv(localConnID uint64, usingPassword "Name": conn.local.Name.String(), "NickName": conn.local.NickName, "UID": fmt.Sprint(conn.local.UID), + "ShortID": fmt.Sprint(conn.local.ShortID), "ConnID": fmt.Sprint(localConnID)} handshakeRecv := map[string]string{} diff --git a/router/interhost.go b/router/interhost.go index 2f3e9bd1..1fba3fe1 100644 --- a/router/interhost.go +++ b/router/interhost.go @@ -13,6 +13,9 @@ type InterHost interface { // be discarded. InvalidateRoutes() + // A mapping of a short id to a peer has changed + InvalidateShortIDs() + // Form a packet-forwarding connection. The remote UDPAddr // can be nil if unknown (in which case the implementation // needs to discover it). @@ -65,6 +68,9 @@ func (NullInterHost) ConsumeInterHostPackets(*Peer, *Peers, func (NullInterHost) InvalidateRoutes() { } +func (NullInterHost) InvalidateShortIDs() { +} + func (NullInterHost) MakeForwarder(*Peer, net.IP, *net.UDPAddr, uint64, InterHostCrypto, func([]byte) error) (InterHostForwarder, error) { return NullInterHost{}, nil diff --git a/router/local_peer.go b/router/local_peer.go index f41a60cb..20975313 100644 --- a/router/local_peer.go +++ b/router/local_peer.go @@ -20,7 +20,8 @@ type LocalPeerAction func() func NewLocalPeer(name PeerName, nickName string, router *Router) *LocalPeer { actionChan := make(chan LocalPeerAction, ChannelSize) peer := &LocalPeer{ - Peer: NewPeer(name, nickName, randomPeerUID(), 0), + Peer: NewPeer(name, nickName, randomPeerUID(), 0, + randomPeerShortID()), router: router, actionChan: actionChan, } @@ -293,3 +294,10 @@ func (peer *LocalPeer) connectionCount() int { defer peer.RUnlock() return len(peer.connections) } + +func (peer *LocalPeer) setShortID(shortID PeerShortID) { + peer.Lock() + defer peer.Unlock() + peer.ShortID = shortID + peer.Version++ +} diff --git a/router/mocks_test.go b/router/mocks_test.go index c0308c13..452a4dd7 100644 --- a/router/mocks_test.go +++ b/router/mocks_test.go @@ -68,7 +68,7 @@ func checkEqualConns(t *testing.T, ourName PeerName, got, wanted map[PeerName]Co // Get all the peers from a Peers in a slice func (peers *Peers) allPeers() []*Peer { var res []*Peer - for _, peer := range peers.table { + for _, peer := range peers.byName { res = append(res, peer) } return res diff --git a/router/peer.go b/router/peer.go index 7f79064e..7ee3a889 100644 --- a/router/peer.go +++ b/router/peer.go @@ -17,11 +17,24 @@ func ParsePeerUID(s string) (PeerUID, error) { return PeerUID(uid), err } +// Short IDs exist for the sake of fast datapath. They are 12 bits, +// randomly assigned, but we detect and recover from collisions. This +// does limit us to 4096 peers, but that should be sufficient for a +// while. +type PeerShortID uint16 + +const PeerShortIDBits = 12 + +func randomPeerShortID() PeerShortID { + return PeerShortID(randUint64() & (1<= newPeer.Version) { // Nobody but us updates us. And if we know more about a @@ -203,7 +392,14 @@ func (peers *Peers) ApplyUpdate(update []byte) (PeerNameSet, PeerNameSet, error) // router.Peers.ApplyUpdate. But ApplyUpdate takes the Lock on // the router.Peers, so there can be no race here. peer.Version = newPeer.Version - peer.connections = makeConnsMap(peer, connSummaries, peers.table) + peer.connections = makeConnsMap(peer, connSummaries, peers.byName) + + if newPeer.ShortID != peer.ShortID { + peers.deleteByShortID(peer) + peer.ShortID = newPeer.ShortID + peers.addByShortID(peer) + } + newUpdate[name] = void } @@ -239,7 +435,7 @@ func (peers *Peers) decodeUpdate(update []byte) (newPeers map[PeerName]*Peer, de newPeer := NewPeerFromSummary(peerSummary) decodedUpdate = append(decodedUpdate, newPeer) decodedConns = append(decodedConns, connSummaries) - existingPeer, found := peers.table[newPeer.Name] + existingPeer, found := peers.byName[newPeer.Name] if !found { newPeers[newPeer.Name] = newPeer } else if existingPeer.UID != newPeer.UID { @@ -254,7 +450,7 @@ func (peers *Peers) decodeUpdate(update []byte) (newPeers map[PeerName]*Peer, de if _, found := newPeers[remoteName]; found { continue } - if _, found := peers.table[remoteName]; found { + if _, found := peers.byName[remoteName]; found { continue } // Update refers to a peer which we have no knowledge @@ -299,11 +495,11 @@ func decodePeer(dec *gob.Decoder) (peerSummary PeerSummary, connSummaries []Conn return } -func makeConnsMap(peer *Peer, connSummaries []ConnectionSummary, table map[PeerName]*Peer) map[PeerName]Connection { +func makeConnsMap(peer *Peer, connSummaries []ConnectionSummary, byName map[PeerName]*Peer) map[PeerName]Connection { conns := make(map[PeerName]Connection) for _, connSummary := range connSummaries { name := PeerNameFromBin(connSummary.NameByte) - remotePeer := table[name] + remotePeer := byName[name] conn := NewRemoteConnection(peer, remotePeer, connSummary.RemoteTCPAddr, connSummary.Outbound, connSummary.Established) conns[name] = conn } diff --git a/router/peers_test.go b/router/peers_test.go index f0edaed0..3189c2bd 100644 --- a/router/peers_test.go +++ b/router/peers_test.go @@ -5,6 +5,7 @@ import ( wt "github.com/weaveworks/weave/testing" "math/rand" "testing" + "time" ) // TODO we should also test: @@ -110,3 +111,240 @@ func TestPeersGarbageCollection(t *testing.T) { ps1.DeleteTestConnection(p3) checkPeerArray(t, ps1.GarbageCollect(), p3) } + +func TestShortIDCollisions(t *testing.T) { + rng := rand.New(rand.NewSource(time.Now().UnixNano())) + _, peers := newNode(PeerName(1 << PeerShortIDBits)) + + // Make enough peers that short id collisions are + // overwhelmingly likely + ps := make([]*Peer, 1<