replace Peer.ForEachConnection with Peer.Connections

The former is prone to triggering deadlocks since the lock is held on
the peer during the iteration.
This commit is contained in:
Matthias Radestock committed 2015-03-17 16:24:19 +00:00
1 parent 08a08be370
commit d5eb76f32e
6 files changed
+19 -24

No files matched your search

+6 -5
View File
@@ -119,10 +119,10 @@ func (cm *ConnectionMaker) checkStateAndAttemptConnections() time.Duration {
// copy the set of things we are connected to, so we can access them without locking // copy the set of things we are connected to, so we can access them without locking
ourConnectedPeers := make(map[PeerName]bool) ourConnectedPeers := make(map[PeerName]bool)
ourConnectedTargets := make(map[string]bool) ourConnectedTargets := make(map[string]bool)
cm.ourself.ForEachConnection(func(peer PeerName, conn Connection) { for _, conn := range cm.ourself.Connections() {
ourConnectedPeers[peer] = true ourConnectedPeers[conn.Remote().Name] = true
ourConnectedTargets[conn.RemoteTCPAddr()] = true ourConnectedTargets[conn.RemoteTCPAddr()] = true
}) }
addTarget := func(address string) { addTarget := func(address string) {
if !ourConnectedTargets[address] { if !ourConnectedTargets[address] {
@@ -139,7 +139,8 @@ func (cm *ConnectionMaker) checkStateAndAttemptConnections() time.Duration {
// Add targets for peers that someone else is connected to, but we // Add targets for peers that someone else is connected to, but we
// aren't // aren't
cm.peers.ForEach(func(name PeerName, peer *Peer) { cm.peers.ForEach(func(name PeerName, peer *Peer) {
peer.ForEachConnection(func(otherPeer PeerName, conn Connection) { for _, conn := range peer.Connections() {
otherPeer := conn.Remote().Name
if otherPeer == cm.ourself.Name || ourConnectedPeers[otherPeer] { if otherPeer == cm.ourself.Name || ourConnectedPeers[otherPeer] {
return return
} }
@@ -151,7 +152,7 @@ func (cm *ConnectionMaker) checkStateAndAttemptConnections() time.Duration {
if host, _, err := net.SplitHostPort(address); err == nil { if host, _, err := net.SplitHostPort(address); err == nil {
addTarget(NormalisePeerAddr(host)) addTarget(NormalisePeerAddr(host))
} }
}) }
}) })
now := time.Now() // make sure we catch items just added now := time.Now() // make sure we catch items just added
+1 -6
View File
@@ -117,12 +117,7 @@ func deliverGossip(channel *GossipChannel, srcName PeerName, _ []byte, dec *gob.
func (c *GossipChannel) SendGossipMsg(buf []byte) { func (c *GossipChannel) SendGossipMsg(buf []byte) {
protocolMsg := c.gossipMsg(buf) protocolMsg := c.gossipMsg(buf)
// Copy the connections so we don't hold a lock for _, conn := range c.ourself.Connections() {
connections := make([]Connection, 0)
c.ourself.ForEachConnection(func(_ PeerName, conn Connection) {
connections = append(connections, conn)
})
for _, conn := range connections {
conn.(ProtocolSender).SendProtocolMsg(protocolMsg) conn.(ProtocolSender).SendProtocolMsg(protocolMsg)
} }
} }
+1 -5
View File
@@ -61,16 +61,12 @@ func (routes *Routes) MarshalJSON() ([]byte, error) {
} }
func (peer *Peer) MarshalJSON() ([]byte, error) { func (peer *Peer) MarshalJSON() ([]byte, error) {
conns := make([]Connection, 0)
peer.ForEachConnection(func(_ PeerName, conn Connection) {
conns = append(conns, conn)
})
return json.Marshal(struct { return json.Marshal(struct {
Name string Name string
UID uint64 UID uint64
Version uint64 Version uint64
Connections []Connection Connections []Connection
}{peer.Name.String(), peer.UID, peer.version, conns}) }{peer.Name.String(), peer.UID, peer.version, peer.Connections()})
} }
func (conn *RemoteConnection) MarshalJSON() ([]byte, error) { func (conn *RemoteConnection) MarshalJSON() ([]byte, error) {
+5 -3
View File
@@ -73,12 +73,14 @@ func (peer *Peer) ConnectionTo(name PeerName) (Connection, bool) {
return conn, found // yes, you really can't inline that. FFS. return conn, found // yes, you really can't inline that. FFS.
} }
func (peer *Peer) ForEachConnection(fun func(PeerName, Connection)) { func (peer *Peer) Connections() []Connection {
peer.RLock() peer.RLock()
defer peer.RUnlock() defer peer.RUnlock()
for name, conn := range peer.connections { connections := make([]Connection, 0, len(peer.connections))
fun(name, conn) for _, conn := range peer.connections {
connections = append(connections, conn)
} }
return connections
} }
func (peer *Peer) SetVersionAndConnections(version uint64, connections map[PeerName]Connection) { func (peer *Peer) SetVersionAndConnections(version uint64, connections map[PeerName]Connection) {
+3 -3
View File
@@ -123,13 +123,13 @@ func (peers *Peers) String() string {
var buf bytes.Buffer var buf bytes.Buffer
peers.ForEach(func(name PeerName, peer *Peer) { peers.ForEach(func(name PeerName, peer *Peer) {
buf.WriteString(fmt.Sprint(peer, "\n")) buf.WriteString(fmt.Sprint(peer, "\n"))
peer.ForEachConnection(func(remoteName PeerName, conn Connection) { for _, conn := range peer.Connections() {
established := "" established := ""
if !conn.Established() { if !conn.Established() {
established = " (unestablished)" established = " (unestablished)"
} }
buf.WriteString(fmt.Sprintf(" -> %v (%v) [%v%s]\n", remoteName, conn.Remote().NickName, conn.RemoteTCPAddr(), established)) buf.WriteString(fmt.Sprintf(" -> %v (%v) [%v%s]\n", conn.Remote().Name, conn.Remote().NickName, conn.RemoteTCPAddr(), established))
}) }
}) })
return buf.String() return buf.String()
} }
+3 -2
View File
@@ -140,17 +140,18 @@ func (routes *Routes) calculateBroadcast() map[PeerName][]PeerName {
// This is rather similar to the inner loop on // This is rather similar to the inner loop on
// peer.Routes(...); the main difference is in the // peer.Routes(...); the main difference is in the
// locking. // locking.
ourself.ForEachConnection(func(remoteName PeerName, conn Connection) { for _, conn := range ourself.Connections() {
if !conn.Established() { if !conn.Established() {
return return
} }
remoteName := conn.Remote().Name
if _, found := reached[remoteName]; found { if _, found := reached[remoteName]; found {
return return
} }
if remoteConn, found := conn.Remote().ConnectionTo(ourself.Name); found && remoteConn.Established() { if remoteConn, found := conn.Remote().ConnectionTo(ourself.Name); found && remoteConn.Established() {
hops = append(hops, remoteName) hops = append(hops, remoteName)
} }
}) }
} }
broadcast[name] = hops broadcast[name] = hops
}) })