From 3e6eef5ebd0702038711e65d90f35cdddd32c663 Mon Sep 17 00:00:00 2001 From: Matthias Radestock Date: Mon, 11 May 2015 07:59:10 +0100 Subject: [PATCH 1/9] refactor(ish): move a bunch of methods from Peer to LocalPeer in order to reduce the former's API surface --- router/local_peer.go | 30 +++++++++++++++++++++++++++++- router/mocks_test.go | 2 +- router/peer.go | 26 -------------------------- router/peers.go | 6 +++--- router/peers_test.go | 8 ++++---- router/router.go | 2 +- 6 files changed, 38 insertions(+), 36 deletions(-) diff --git a/router/local_peer.go b/router/local_peer.go index c7218ea6..2e4398fe 100644 --- a/router/local_peer.go +++ b/router/local_peer.go @@ -228,6 +228,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 +237,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..7168034a 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) } diff --git a/router/peer.go b/router/peer.go index 086e6e44..a497a9a6 100644 --- a/router/peer.go +++ b/router/peer.go @@ -66,12 +66,6 @@ func (peer *Peer) IsLocallyReferenced() bool { 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() @@ -96,26 +90,6 @@ func (peer *Peer) SetVersionAndConnections(version uint64, connections map[PeerN 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 diff --git a/router/peers.go b/router/peers.go index ff75c0c2..fff0cd84 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), @@ -250,7 +250,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. diff --git a/router/peers_test.go b/router/peers_test.go index 5fe8745c..c38e6588 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()...) diff --git a/router/router.go b/router/router.go index 0df6a813..dcd33f44 100644 --- a/router/router.go +++ b/router/router.go @@ -64,7 +64,7 @@ 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.ConnectionMaker = NewConnectionMaker(router.Ourself, router.Peers, router.Port) From 505ced5cc825b910d9f52bb29e6f2d3acaee5a5d Mon Sep 17 00:00:00 2001 From: Matthias Radestock Date: Mon, 11 May 2015 09:36:19 +0100 Subject: [PATCH 2/9] move localRefCount accounting responsibility from Peer to Peers This makes for a more symmetric API - Peers.FetchWithDefault does the increment and Peers.Dereference does the decrement, whereas previously the latter was done by Peer.DecrementLocalRefCount. This does increase lock granularity, i.e. modifying any ref count now requires a write lock on Peers. --- router/connection.go | 2 +- router/gossip_test.go | 4 ++-- router/mocks_test.go | 2 +- router/peer.go | 20 +------------------- router/peers.go | 36 ++++++++++++++---------------------- 5 files changed, 19 insertions(+), 45 deletions(-) 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/gossip_test.go b/router/gossip_test.go index fc727d1e..09240735 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) diff --git a/router/mocks_test.go b/router/mocks_test.go index 7168034a..674eaa3a 100644 --- a/router/mocks_test.go +++ b/router/mocks_test.go @@ -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 a497a9a6..50b55e9a 100644 --- a/router/peer.go +++ b/router/peer.go @@ -13,7 +13,7 @@ type Peer struct { NickName string UID uint64 version uint64 - localRefCount uint64 + localRefCount uint64 // maintained by Peers connections map[PeerName]Connection } @@ -48,24 +48,6 @@ func (peer *Peer) Version() uint64 { 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) ConnectionTo(name PeerName) (Connection, bool) { peer.RLock() defer peer.RUnlock() diff --git a/router/peers.go b/router/peers.go index fff0cd84..db7577c5 100644 --- a/router/peers.go +++ b/router/peers.go @@ -47,20 +47,17 @@ func NewPeers(ourself *LocalPeer, 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() @@ -162,22 +165,11 @@ func (peers *Peers) String() string { 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{} _, reached := peers.ourself.Routes(nil, false) 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) From 0229185ffe035d2a2ff347674a866fbd55f7a6d7 Mon Sep 17 00:00:00 2001 From: Matthias Radestock Date: Mon, 11 May 2015 10:20:51 +0100 Subject: [PATCH 3/9] move ConnectionTo from Peer to LocalPeer further shrinking the former's API --- router/handshake.go | 2 +- router/local_peer.go | 7 +++++++ router/peer.go | 7 ------- router/peers_test.go | 2 +- router/routes.go | 6 +++++- 5 files changed, 14 insertions(+), 10 deletions(-) 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/local_peer.go b/router/local_peer.go index 2e4398fe..bfc21823 100644 --- a/router/local_peer.go +++ b/router/local_peer.go @@ -75,6 +75,13 @@ func (peer *LocalPeer) RelayBroadcast(srcPeer *Peer, df bool, frame []byte, dec } } +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() diff --git a/router/peer.go b/router/peer.go index 50b55e9a..1ba834b8 100644 --- a/router/peer.go +++ b/router/peer.go @@ -48,13 +48,6 @@ func (peer *Peer) Version() uint64 { return peer.version } -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() diff --git a/router/peers_test.go b/router/peers_test.go index c38e6588..daf1e944 100644 --- a/router/peers_test.go +++ b/router/peers_test.go @@ -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/routes.go b/router/routes.go index 04020954..1f50f213 100644 --- a/router/routes.go +++ b/router/routes.go @@ -231,7 +231,11 @@ 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[ourself.Name]; !establishedAndSymmetric || (found && remoteConn.Established()) { hops = append(hops, remoteName) } } From 5bf688925038f8527cf1366db4321d47401c8041 Mon Sep 17 00:00:00 2001 From: Matthias Radestock Date: Mon, 11 May 2015 11:04:53 +0100 Subject: [PATCH 4/9] remove Peer.Version further shrinking the Peer API --- router/peer.go | 6 ------ router/peers.go | 4 ++-- 2 files changed, 2 insertions(+), 8 deletions(-) diff --git a/router/peer.go b/router/peer.go index 1ba834b8..ffe56580 100644 --- a/router/peer.go +++ b/router/peer.go @@ -42,12 +42,6 @@ func (peer *Peer) Info() string { 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) Connections() ConnectionSet { connections := make(ConnectionSet) peer.RLock() diff --git a/router/peers.go b/router/peers.go index db7577c5..8362e4bd 100644 --- a/router/peers.go +++ b/router/peers.go @@ -242,7 +242,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 || 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. @@ -259,7 +259,7 @@ func (peers *Peers) applyUpdate(decodedUpdate []*Peer, decodedConns [][]Connecti // 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.SetVersionAndConnections(newPeer.version, conns) newUpdate[name] = peer } return newUpdate From 86ba3be4f704fd53034549009ad428ad709d73a4 Mon Sep 17 00:00:00 2001 From: Matthias Radestock Date: Mon, 11 May 2015 12:12:46 +0100 Subject: [PATCH 5/9] move Connections from Peer to LocalPeer further shrinking the former's API --- router/connection_maker.go | 9 +++++++-- router/json.go | 41 ++++++++++++++++++++++---------------- router/local_peer.go | 10 ++++++++++ router/peer.go | 10 ---------- router/peers.go | 22 +++++++++++++++----- router/router.go | 2 +- router/routes.go | 6 +++--- 7 files changed, 62 insertions(+), 38 deletions(-) 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/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 bfc21823..239fce4c 100644 --- a/router/local_peer.go +++ b/router/local_peer.go @@ -75,6 +75,16 @@ func (peer *LocalPeer) RelayBroadcast(srcPeer *Peer, df bool, frame []byte, dec } } +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() diff --git a/router/peer.go b/router/peer.go index ffe56580..0ec4172a 100644 --- a/router/peer.go +++ b/router/peer.go @@ -42,16 +42,6 @@ func (peer *Peer) Info() string { return fmt.Sprint(peer.String(), " (v", peer.version, ") (UID ", peer.UID, ")") } -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() diff --git a/router/peers.go b/router/peers.go index 8362e4bd..45e5a75c 100644 --- a/router/peers.go +++ b/router/peers.go @@ -152,14 +152,26 @@ 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 { + for conn := range peers.ourself.Connections() { + printConnection(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 { + printConnection(conn) } - fmt.Fprintf(&buf, " -> %s [%v%s]\n", conn.Remote(), conn.RemoteTCPAddr(), established) } }) return buf.String() diff --git a/router/router.go b/router/router.go index dcd33f44..753853fc 100644 --- a/router/router.go +++ b/router/router.go @@ -66,7 +66,7 @@ func NewRouter(config RouterConfig, name PeerName, nickName string) *Router { router.Macs = NewMacCache(macMaxAge, onMacExpiry) 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 1f50f213..92197fb5 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, @@ -219,7 +219,7 @@ func (routes *Routes) calculateBroadcast(establishedAndSymmetric bool) map[PeerN routes.peers.ForEach(func(peer *Peer) { hops := []PeerName{} - if found, reached := peer.Routes(ourself, establishedAndSymmetric); found { + if found, reached := peer.Routes(ourself.Peer, establishedAndSymmetric); found { // This is rather similar to the inner loop on // peer.Routes(...); the main difference is in the // locking. From 426d8edb314ff49ec22c0b8abc2acbc97b0e07ff Mon Sep 17 00:00:00 2001 From: Matthias Radestock Date: Mon, 11 May 2015 12:35:11 +0100 Subject: [PATCH 6/9] remove Peer.SetVersionAndConnections further shrinking Peer API --- router/gossip_test.go | 3 ++- router/peer.go | 7 ------- router/peers.go | 4 ++-- 3 files changed, 4 insertions(+), 10 deletions(-) diff --git a/router/gossip_test.go b/router/gossip_test.go index 09240735..9128ab39 100644 --- a/router/gossip_test.go +++ b/router/gossip_test.go @@ -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/peer.go b/router/peer.go index 0ec4172a..46d024b0 100644 --- a/router/peer.go +++ b/router/peer.go @@ -42,13 +42,6 @@ func (peer *Peer) Info() string { return fmt.Sprint(peer.String(), " (v", peer.version, ") (UID ", peer.UID, ")") } -func (peer *Peer) SetVersionAndConnections(version uint64, connections map[PeerName]Connection) { - peer.Lock() - defer peer.Unlock() - peer.version = version - peer.connections = connections -} - // 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 diff --git a/router/peers.go b/router/peers.go index 45e5a75c..6a5606d9 100644 --- a/router/peers.go +++ b/router/peers.go @@ -270,8 +270,8 @@ 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 From c210f0e3c13af25f9e74d63d52ad6d3b41b18f3e Mon Sep 17 00:00:00 2001 From: Matthias Radestock Date: Mon, 11 May 2015 12:35:54 +0100 Subject: [PATCH 7/9] make Peer.Info lock-free --- router/local_peer.go | 6 ++++++ router/peer.go | 2 -- router/peers.go | 3 ++- 3 files changed, 8 insertions(+), 3 deletions(-) diff --git a/router/local_peer.go b/router/local_peer.go index 239fce4c..f20b0676 100644 --- a/router/local_peer.go +++ b/router/local_peer.go @@ -75,6 +75,12 @@ 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() diff --git a/router/peer.go b/router/peer.go index 46d024b0..29ffe6d7 100644 --- a/router/peer.go +++ b/router/peer.go @@ -37,8 +37,6 @@ 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, ")") } diff --git a/router/peers.go b/router/peers.go index 6a5606d9..1074b267 100644 --- a/router/peers.go +++ b/router/peers.go @@ -160,12 +160,13 @@ func (peers *Peers) String() string { fmt.Fprintf(&buf, " -> %s [%v%s]\n", conn.Remote(), conn.RemoteTCPAddr(), established) } peers.ForEach(func(peer *Peer) { - fmt.Fprintln(&buf, peer.Info()) 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. From 5bd8730353a71f3588f2adac13d4d3f097999402 Mon Sep 17 00:00:00 2001 From: Matthias Radestock Date: Mon, 11 May 2015 13:05:14 +0100 Subject: [PATCH 8/9] make Peer.Encode lock-free --- router/peers.go | 25 +++++++++++++++---------- 1 file changed, 15 insertions(+), 10 deletions(-) diff --git a/router/peers.go b/router/peers.go index 1074b267..1aac62ef 100644 --- a/router/peers.go +++ b/router/peers.go @@ -128,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() } @@ -278,10 +280,7 @@ func (peers *Peers) applyUpdate(decodedUpdate []*Peer, decodedConns [][]Connecti 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, @@ -302,6 +301,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 From e246e6acc604b034e3e1c8ada6f574100f0058af Mon Sep 17 00:00:00 2001 From: Matthias Radestock Date: Mon, 11 May 2015 14:36:39 +0100 Subject: [PATCH 9/9] coarser lock granularity - get rid of individual Peer locks - hold Peers read (or write) lock and LocalPeer read lock while performing route calculations This eliminates the deadlock resulting from the inconsistent lock acquisition order in route calculcation. Fixes #643. --- router/local_peer.go | 2 ++ router/peer.go | 16 ++-------------- router/peers.go | 2 ++ router/routes.go | 18 ++++++++++-------- 4 files changed, 16 insertions(+), 22 deletions(-) diff --git a/router/local_peer.go b/router/local_peer.go index f20b0676..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 diff --git a/router/peer.go b/router/peer.go index 29ffe6d7..5fcf277a 100644 --- a/router/peer.go +++ b/router/peer.go @@ -3,11 +3,9 @@ package router import ( "fmt" "sort" - "sync" ) type Peer struct { - sync.RWMutex Name PeerName NameByte []byte NickName string @@ -61,13 +59,9 @@ func (peer *Peer) Info() string { // 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} @@ -88,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 @@ -103,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 1aac62ef..ab921c17 100644 --- a/router/peers.go +++ b/router/peers.go @@ -182,7 +182,9 @@ func (peers *Peers) String() string { 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.localRefCount == 0 { delete(peers.table, name) diff --git a/router/routes.go b/router/routes.go index 92197fb5..4c68e34d 100644 --- a/router/routes.go +++ b/router/routes.go @@ -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.Peer, 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 } @@ -235,12 +237,12 @@ func (routes *Routes) calculateBroadcast(establishedAndSymmetric bool) map[PeerN // 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[ourself.Name]; !establishedAndSymmetric || (found && remoteConn.Established()) { + if remoteConn, found := conn.Remote().connections[routes.ourself.Name]; !establishedAndSymmetric || (found && remoteConn.Established()) { hops = append(hops, remoteName) } } } broadcast[peer.Name] = hops - }) + } return broadcast }