diff --git a/router/connection.go b/router/connection.go index 891572a5..7e380730 100644 --- a/router/connection.go +++ b/router/connection.go @@ -412,7 +412,7 @@ func (conn *LocalConnection) shutdown(err error) { } if conn.remote != nil { - conn.remote.DecrementLocalRefCount() + conn.Router.Peers.Dereference(conn.remote) conn.Router.Ourself.DeleteConnection(conn) } diff --git a/router/connection_maker.go b/router/connection_maker.go index c771bc20..d8e34e55 100644 --- a/router/connection_maker.go +++ b/router/connection_maker.go @@ -206,8 +206,13 @@ func (cm *ConnectionMaker) ourConnections() (PeerNameSet, map[string]struct{}, m func (cm *ConnectionMaker) addPeerTargets(ourConnectedPeers PeerNameSet, addTarget func(string)) { cm.peers.ForEach(func(peer *Peer) { - for conn := range peer.Connections() { - otherPeer := conn.Remote().Name + if peer == cm.ourself.Peer { + return + } + // Modifying peer.connections requires a write lock on Peers, + // and since we are holding a read lock (due to the ForEach), + // access without locking the peer is safe. + for otherPeer, conn := range peer.connections { if otherPeer == cm.ourself.Name { continue } diff --git a/router/gossip_test.go b/router/gossip_test.go index fc727d1e..9128ab39 100644 --- a/router/gossip_test.go +++ b/router/gossip_test.go @@ -55,8 +55,8 @@ func (router *Router) DeleteTestChannelConnection(r *Router) { fromPeer, _ := r.Peers.Fetch(fromName) toPeer, _ := router.Peers.Fetch(toName) - fromPeer.DecrementLocalRefCount() - toPeer.DecrementLocalRefCount() + r.Peers.Dereference(fromPeer) + router.Peers.Dereference(toPeer) conn, _ := router.Ourself.ConnectionTo(toName) router.Ourself.handleDeleteConnection(conn) @@ -79,7 +79,8 @@ func (router *Router) tp(routers ...*Router) *Peer { p := NewPeer(r.Ourself.Peer.Name, "", r.Ourself.Peer.UID, r.Ourself.Peer.version) connections[r.Ourself.Peer.Name] = newMockConnection(peer, p) } - peer.SetVersionAndConnections(router.Ourself.Peer.version, connections) + peer.version = router.Ourself.Peer.version + peer.connections = connections return peer } diff --git a/router/handshake.go b/router/handshake.go index 4a003a19..3e42633a 100644 --- a/router/handshake.go +++ b/router/handshake.go @@ -75,7 +75,7 @@ func (conn *LocalConnection) handshake(enc *gob.Encoder, dec *gob.Decoder, accep return fmt.Errorf("Found unknown remote name: %s at %s", name, conn.remoteTCPAddr) } } - if existingConn, found := conn.local.ConnectionTo(name); found && existingConn.Established() { + 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 := strconv.ParseUint(uidStr, 10, 64) diff --git a/router/json.go b/router/json.go index 2d01e8f6..9461c014 100644 --- a/router/json.go +++ b/router/json.go @@ -35,8 +35,30 @@ func (cache *MacCache) MarshalJSON() ([]byte, error) { } func (peers *Peers) MarshalJSON() ([]byte, error) { - var ps []*Peer - peers.ForEach(func(peer *Peer) { ps = append(ps, peer) }) + type p struct { + Name string + NickName string + UID uint64 + Version uint64 + Connections []Connection + } + var ps []*p + peers.ForEach(func(peer *Peer) { + var connections []Connection + if peer == peers.ourself.Peer { + for conn := range peers.ourself.Connections() { + connections = append(connections, conn) + } + } else { + // Modifying peer.connections requires a write lock on + // Peers, and since we are holding a read lock (due to the + // ForEach), access without locking the peer is safe. + for _, conn := range peer.connections { + connections = append(connections, conn) + } + } + ps = append(ps, &p{peer.Name.String(), peer.NickName, peer.UID, peer.version, connections}) + }) return json.Marshal(ps) } @@ -63,21 +85,6 @@ func (routes *Routes) MarshalJSON() ([]byte, error) { return json.Marshal(r) } -func (peer *Peer) MarshalJSON() ([]byte, error) { - conns := peer.Connections() - connections := make([]Connection, 0, len(conns)) - for conn := range conns { - connections = append(connections, conn) - } - return json.Marshal(struct { - Name string - NickName string - UID uint64 - Version uint64 - Connections []Connection - }{peer.Name.String(), peer.NickName, peer.UID, peer.version, connections}) -} - func (conn *RemoteConnection) MarshalJSON() ([]byte, error) { return json.Marshal(struct { Name string diff --git a/router/local_peer.go b/router/local_peer.go index c7218ea6..09498629 100644 --- a/router/local_peer.go +++ b/router/local_peer.go @@ -4,10 +4,12 @@ import ( "fmt" "log" "net" + "sync" "time" ) type LocalPeer struct { + sync.RWMutex *Peer router *Router actionChan chan<- LocalPeerAction @@ -75,6 +77,29 @@ func (peer *LocalPeer) RelayBroadcast(srcPeer *Peer, df bool, frame []byte, dec } } +func (peer *LocalPeer) Info() string { + peer.RLock() + defer peer.RUnlock() + return peer.Peer.Info() +} + +func (peer *LocalPeer) Connections() ConnectionSet { + connections := make(ConnectionSet) + peer.RLock() + defer peer.RUnlock() + for _, conn := range peer.connections { + connections[conn] = void + } + return connections +} + +func (peer *LocalPeer) ConnectionTo(name PeerName) (Connection, bool) { + peer.RLock() + defer peer.RUnlock() + conn, found := peer.connections[name] + return conn, found // yes, you really can't inline that. FFS. +} + func (peer *LocalPeer) ConnectionsTo(names []PeerName) []Connection { conns := make([]Connection, 0, len(names)) peer.RLock() @@ -228,6 +253,8 @@ func (peer *LocalPeer) handleDeleteConnection(conn Connection) { peer.broadcastPeerUpdate() } +// helpers + func (peer *LocalPeer) broadcastPeerUpdate(peers ...*Peer) { peer.router.Routes.Recalculate() peer.router.TopologyGossip.GossipBroadcast(NewTopologyGossipData(peer.router.Peers, append(peers, peer.Peer)...)) @@ -235,8 +262,34 @@ func (peer *LocalPeer) broadcastPeerUpdate(peers ...*Peer) { func (peer *LocalPeer) checkConnectionLimit() error { limit := peer.router.ConnLimit - if 0 != limit && peer.ConnectionCount() >= limit { + if 0 != limit && peer.connectionCount() >= limit { return fmt.Errorf("Connection limit reached (%v)", limit) } return nil } + +func (peer *LocalPeer) addConnection(conn Connection) { + peer.Lock() + defer peer.Unlock() + peer.connections[conn.Remote().Name] = conn + peer.version++ +} + +func (peer *LocalPeer) deleteConnection(conn Connection) { + peer.Lock() + defer peer.Unlock() + delete(peer.connections, conn.Remote().Name) + peer.version++ +} + +func (peer *LocalPeer) connectionEstablished(conn Connection) { + peer.Lock() + defer peer.Unlock() + peer.version++ +} + +func (peer *LocalPeer) connectionCount() int { + peer.RLock() + defer peer.RUnlock() + return len(peer.connections) +} diff --git a/router/mocks_test.go b/router/mocks_test.go index cd58a08c..674eaa3a 100644 --- a/router/mocks_test.go +++ b/router/mocks_test.go @@ -14,7 +14,7 @@ import ( func (peers *Peers) AddTestConnection(p *Peer) { toPeer := NewPeer(p.Name, "", p.UID, 0) peers.FetchWithDefault(toPeer) // Has side-effect of incrementing refcount - conn := newMockConnection(peers.ourself, toPeer) + conn := newMockConnection(peers.ourself.Peer, toPeer) peers.ourself.addConnection(conn) peers.ourself.connectionEstablished(conn) } @@ -33,7 +33,7 @@ func (peers *Peers) AddTestRemoteConnection(p1, p2 *Peer) { func (peers *Peers) DeleteTestConnection(p *Peer) { toName := p.Name toPeer, _ := peers.Fetch(toName) - toPeer.DecrementLocalRefCount() + peers.Dereference(toPeer) conn, _ := peers.ourself.ConnectionTo(toName) peers.ourself.deleteConnection(conn) } diff --git a/router/peer.go b/router/peer.go index 086e6e44..5fcf277a 100644 --- a/router/peer.go +++ b/router/peer.go @@ -3,17 +3,15 @@ package router import ( "fmt" "sort" - "sync" ) type Peer struct { - sync.RWMutex Name PeerName NameByte []byte NickName string UID uint64 version uint64 - localRefCount uint64 + localRefCount uint64 // maintained by Peers connections map[PeerName]Connection } @@ -37,85 +35,9 @@ func (peer *Peer) String() string { } func (peer *Peer) Info() string { - peer.RLock() - defer peer.RUnlock() return fmt.Sprint(peer.String(), " (v", peer.version, ") (UID ", peer.UID, ")") } -func (peer *Peer) Version() uint64 { - peer.RLock() - defer peer.RUnlock() - return peer.version -} - -func (peer *Peer) IncrementLocalRefCount() { - peer.Lock() - defer peer.Unlock() - peer.localRefCount++ -} - -func (peer *Peer) DecrementLocalRefCount() { - peer.Lock() - defer peer.Unlock() - peer.localRefCount-- -} - -func (peer *Peer) IsLocallyReferenced() bool { - peer.RLock() - defer peer.RUnlock() - return peer.localRefCount != 0 -} - -func (peer *Peer) ConnectionCount() int { - peer.RLock() - defer peer.RUnlock() - return len(peer.connections) -} - -func (peer *Peer) ConnectionTo(name PeerName) (Connection, bool) { - peer.RLock() - defer peer.RUnlock() - conn, found := peer.connections[name] - return conn, found // yes, you really can't inline that. FFS. -} - -func (peer *Peer) Connections() ConnectionSet { - connections := make(ConnectionSet) - peer.RLock() - defer peer.RUnlock() - for _, conn := range peer.connections { - connections[conn] = void - } - return connections -} - -func (peer *Peer) SetVersionAndConnections(version uint64, connections map[PeerName]Connection) { - peer.Lock() - defer peer.Unlock() - peer.version = version - peer.connections = connections -} - -func (peer *Peer) addConnection(conn Connection) { - peer.Lock() - defer peer.Unlock() - peer.connections[conn.Remote().Name] = conn - peer.version++ -} - -func (peer *Peer) deleteConnection(conn Connection) { - peer.Lock() - defer peer.Unlock() - delete(peer.connections, conn.Remote().Name) - peer.version++ -} - -func (peer *Peer) connectionEstablished(conn Connection) { - peer.Lock() - defer peer.Unlock() - peer.version++ -} - // Calculate the routing table from this peer to all peers reachable // from it, returning a "next hop" map of PeerNameX -> PeerNameY, // which says "in order to send a message to X, the peer should send @@ -137,13 +59,9 @@ func (peer *Peer) connectionEstablished(conn Connection) { // reaches that peer. The boolean return indicates whether that has // happened. // -// We acquire read locks on peers as we encounter them during the -// traversal. This prevents the connectivity graph from changing -// underneath us in ways that would invalidate the result. Thus the -// answer returned may be out of date, but never inconsistent. +// NB: This function should generally be invoked while holding a read +// lock on Peers and LocalPeer. func (peer *Peer) Routes(stopAt *Peer, establishedAndSymmetric bool) (bool, map[PeerName]PeerName) { - peer.RLock() - defer peer.RUnlock() routes := make(map[PeerName]PeerName) routes[peer.Name] = UnknownPeerName nextWorklist := []*Peer{peer} @@ -164,11 +82,7 @@ func (peer *Peer) Routes(stopAt *Peer, establishedAndSymmetric bool) (bool, map[ continue } remote := conn.Remote() - remote.RLock() - // DANGER holding rlock on remote, possibly taking - // rlock on remoteConn if remoteConn, found := remote.connections[curName]; !establishedAndSymmetric || (found && remoteConn.Established()) { - defer remote.RUnlock() nextWorklist = append(nextWorklist, remote) // We now know how to get to remoteName: the same // way we get to curPeer. Except, if curPeer is @@ -179,8 +93,6 @@ func (peer *Peer) Routes(stopAt *Peer, establishedAndSymmetric bool) (bool, map[ } else { routes[remoteName] = routes[curName] } - } else { - remote.RUnlock() } } } diff --git a/router/peers.go b/router/peers.go index ff75c0c2..ab921c17 100644 --- a/router/peers.go +++ b/router/peers.go @@ -10,7 +10,7 @@ import ( type Peers struct { sync.RWMutex - ourself *Peer + ourself *LocalPeer table map[PeerName]*Peer onGC func(*Peer) } @@ -39,7 +39,7 @@ type ConnectionSummary struct { Established bool } -func NewPeers(ourself *Peer, onGC func(*Peer)) *Peers { +func NewPeers(ourself *LocalPeer, onGC func(*Peer)) *Peers { return &Peers{ ourself: ourself, table: make(map[PeerName]*Peer), @@ -47,20 +47,17 @@ func NewPeers(ourself *Peer, onGC func(*Peer)) *Peers { } func (peers *Peers) FetchWithDefault(peer *Peer) *Peer { - peers.RLock() - res, found := peers.fetchAlias(peer) - peers.RUnlock() - if found { - return res - } peers.Lock() defer peers.Unlock() - res, found = peers.fetchAlias(peer) - if found { - return res + if existingPeer, found := peers.table[peer.Name]; found { + if existingPeer.UID != peer.UID { + return nil + } + existingPeer.localRefCount++ + return existingPeer } peers.table[peer.Name] = peer - peer.IncrementLocalRefCount() + peer.localRefCount++ return peer } @@ -71,6 +68,12 @@ func (peers *Peers) Fetch(name PeerName) (*Peer, bool) { return peer, found // GRRR, why can't I inline this!? } +func (peers *Peers) Dereference(peer *Peer) { + peers.Lock() + defer peers.Unlock() + peer.localRefCount-- +} + func (peers *Peers) ForEach(fun func(*Peer)) { peers.RLock() defer peers.RUnlock() @@ -125,19 +128,21 @@ func (peers *Peers) Names() PeerNameSet { } func (peers *Peers) EncodePeers(names PeerNameSet) []byte { + buf := new(bytes.Buffer) + enc := gob.NewEncoder(buf) peers.RLock() + defer peers.RUnlock() peerList := make([]*Peer, 0, len(names)) for name := range names { if peer, found := peers.table[name]; found { + if peer == peers.ourself.Peer { + peers.ourself.Peer.Encode(enc) + } else { + peer.Encode(enc) + } peerList = append(peerList, peer) } } - peers.RUnlock() // release lock so we don't hold it while encoding - buf := new(bytes.Buffer) - enc := gob.NewEncoder(buf) - for _, peer := range peerList { - peer.encode(enc) - } return buf.Bytes() } @@ -149,35 +154,39 @@ func (peers *Peers) GarbageCollect() []*Peer { func (peers *Peers) String() string { var buf bytes.Buffer + printConnection := func(conn Connection) { + established := "" + if !conn.Established() { + established = " (unestablished)" + } + fmt.Fprintf(&buf, " -> %s [%v%s]\n", conn.Remote(), conn.RemoteTCPAddr(), established) + } peers.ForEach(func(peer *Peer) { - fmt.Fprintln(&buf, peer.Info()) - for conn := range peer.Connections() { - established := "" - if !conn.Established() { - established = " (unestablished)" + if peer == peers.ourself.Peer { + fmt.Fprintln(&buf, peers.ourself.Info()) + for conn := range peers.ourself.Connections() { + printConnection(conn) + } + } else { + fmt.Fprintln(&buf, peer.Info()) + // Modifying peer.connections requires a write lock on + // Peers, and since we are holding a read lock (due to the + // ForEach), access without locking the peer is safe. + for _, conn := range peer.connections { + printConnection(conn) } - fmt.Fprintf(&buf, " -> %s [%v%s]\n", conn.Remote(), conn.RemoteTCPAddr(), established) } }) return buf.String() } -func (peers *Peers) fetchAlias(peer *Peer) (*Peer, bool) { - if existingPeer, found := peers.table[peer.Name]; found { - if existingPeer.UID != peer.UID { - return nil, true - } - existingPeer.IncrementLocalRefCount() - return existingPeer, true - } - return nil, false -} - func (peers *Peers) garbageCollect() []*Peer { removed := []*Peer{} + peers.ourself.RLock() _, reached := peers.ourself.Routes(nil, false) + peers.ourself.RUnlock() for name, peer := range peers.table { - if _, found := reached[peer.Name]; !found && !peer.IsLocallyReferenced() { + if _, found := reached[peer.Name]; !found && peer.localRefCount == 0 { delete(peers.table, name) peers.onGC(peer) removed = append(removed, peer) @@ -250,7 +259,7 @@ func (peers *Peers) applyUpdate(decodedUpdate []*Peer, decodedConns [][]Connecti // guaranteed to find peer in the peers.table peer := peers.table[name] if peer != newPeer && - (peer == peers.ourself || peer.Version() >= newPeer.Version()) { + (peer == peers.ourself.Peer || peer.version >= newPeer.version) { // Nobody but us updates us. And if we know more about a // peer than what's in the the update, we ignore the // latter. @@ -266,17 +275,14 @@ func (peers *Peers) applyUpdate(decodedUpdate []*Peer, decodedConns [][]Connecti // for an update would be someone else calling // router.Peers.ApplyUpdate. But ApplyUpdate takes the Lock on // the router.Peers, so there can be no race here. - conns := makeConnsMap(peer, connSummaries, peers.table) - peer.SetVersionAndConnections(newPeer.Version(), conns) + peer.version = newPeer.version + peer.connections = makeConnsMap(peer, connSummaries, peers.table) newUpdate[name] = peer } return newUpdate } -func (peer *Peer) encode(enc *gob.Encoder) { - peer.RLock() - defer peer.RUnlock() - +func (peer *Peer) Encode(enc *gob.Encoder) { checkFatal(enc.Encode(PeerSummary{ peer.NameByte, peer.NickName, @@ -297,6 +303,12 @@ func (peer *Peer) encode(enc *gob.Encoder) { checkFatal(enc.Encode(connSummaries)) } +func (peer *LocalPeer) Encode(enc *gob.Encoder) { + peer.RLock() + defer peer.RUnlock() + peer.Peer.Encode(enc) +} + func decodePeer(dec *gob.Decoder) (peerSummary PeerSummary, connSummaries []ConnectionSummary, err error) { if err = dec.Decode(&peerSummary); err != nil { return diff --git a/router/peers_test.go b/router/peers_test.go index 5fe8745c..daf1e944 100644 --- a/router/peers_test.go +++ b/router/peers_test.go @@ -17,10 +17,10 @@ import ( // - non-gc of peers that are only referenced locally func newNode(name PeerName) (*Peer, *Peers) { - peer := NewPeer(name, "", 0, 0) + peer := NewLocalPeer(name, "", nil) peers := NewPeers(peer, func(*Peer) {}) - peers.FetchWithDefault(peer) - return peer, peers + peers.FetchWithDefault(peer.Peer) + return peer.Peer, peers } // Check that ApplyUpdate copies the whole topology from peers @@ -29,7 +29,7 @@ func checkApplyUpdate(t *testing.T, peers *Peers) { // We need a new node outside of the network, with a connection // into it. _, testBedPeers := newNode(dummyName) - testBedPeers.AddTestConnection(peers.ourself) + testBedPeers.AddTestConnection(peers.ourself.Peer) testBedPeers.ApplyUpdate(peers.EncodePeers(peers.Names())) checkTopologyPeers(t, true, testBedPeers.allPeersExcept(dummyName), peers.allPeers()...) @@ -52,7 +52,7 @@ func TestPeersEncoding(t *testing.T) { case 0: from, to := rand.Intn(numNodes), rand.Intn(numNodes) if from != to { - if _, found := peer[from].ConnectionTo(peer[to].Name); !found { + if _, found := peer[from].connections[peer[to].Name]; !found { ps[from].AddTestConnection(peer[to]) conns = append(conns, struct{ from, to int }{from, to}) checkApplyUpdate(t, ps[from]) diff --git a/router/router.go b/router/router.go index 0df6a813..753853fc 100644 --- a/router/router.go +++ b/router/router.go @@ -64,9 +64,9 @@ func NewRouter(config RouterConfig, name PeerName, nickName string) *Router { } router.Ourself = NewLocalPeer(name, nickName, router) router.Macs = NewMacCache(macMaxAge, onMacExpiry) - router.Peers = NewPeers(router.Ourself.Peer, onPeerGC) + router.Peers = NewPeers(router.Ourself, onPeerGC) router.Peers.FetchWithDefault(router.Ourself.Peer) - router.Routes = NewRoutes(router.Ourself.Peer, router.Peers) + router.Routes = NewRoutes(router.Ourself, router.Peers) router.ConnectionMaker = NewConnectionMaker(router.Ourself, router.Peers, router.Port) router.TopologyGossip = router.NewGossip("topology", router) return router diff --git a/router/routes.go b/router/routes.go index 04020954..4c68e34d 100644 --- a/router/routes.go +++ b/router/routes.go @@ -9,7 +9,7 @@ import ( type Routes struct { sync.RWMutex - ourself *Peer + ourself *LocalPeer peers *Peers unicast map[PeerName]PeerName unicastAll map[PeerName]PeerName // [1] @@ -21,7 +21,7 @@ type Routes struct { // symmetric ones } -func NewRoutes(ourself *Peer, peers *Peers) *Routes { +func NewRoutes(ourself *LocalPeer, peers *Peers) *Routes { routes := &Routes{ ourself: ourself, peers: peers, @@ -166,12 +166,17 @@ func (routes *Routes) run(recalculate <-chan *struct{}, wait <-chan chan struct{ } func (routes *Routes) calculate() { + routes.peers.RLock() + routes.ourself.RLock() var ( unicast = routes.calculateUnicast(true) unicastAll = routes.calculateUnicast(false) broadcast = routes.calculateBroadcast(true) broadcastAll = routes.calculateBroadcast(false) ) + routes.ourself.RUnlock() + routes.peers.RUnlock() + routes.Lock() routes.unicast = unicast routes.unicastAll = unicastAll @@ -214,16 +219,13 @@ func (routes *Routes) calculateUnicast(establishedAndSymmetric bool) map[PeerNam // where <= is the subset relationship on keys of the returned map. func (routes *Routes) calculateBroadcast(establishedAndSymmetric bool) map[PeerName][]PeerName { broadcast := make(map[PeerName][]PeerName) - ourself := routes.ourself - ourConnections := ourself.Connections() - - routes.peers.ForEach(func(peer *Peer) { + for _, peer := range routes.peers.table { hops := []PeerName{} - if found, reached := peer.Routes(ourself, establishedAndSymmetric); found { + if found, reached := peer.Routes(routes.ourself.Peer, establishedAndSymmetric); found { // This is rather similar to the inner loop on // peer.Routes(...); the main difference is in the // locking. - for conn := range ourConnections { + for _, conn := range routes.ourself.connections { if establishedAndSymmetric && !conn.Established() { continue } @@ -231,12 +233,16 @@ func (routes *Routes) calculateBroadcast(establishedAndSymmetric bool) map[PeerN if _, found := reached[remoteName]; found { continue } - if remoteConn, found := conn.Remote().ConnectionTo(ourself.Name); !establishedAndSymmetric || (found && remoteConn.Established()) { + // conn.Remote() cannot by ourself. Modifying + // peer.connections requires a write lock on Peers, + // and since we are holding a read lock (due to the + // ForEach), access without locking the peer is safe. + if remoteConn, found := conn.Remote().connections[routes.ourself.Name]; !establishedAndSymmetric || (found && remoteConn.Established()) { hops = append(hops, remoteName) } } } broadcast[peer.Name] = hops - }) + } return broadcast }