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:
1 parent
08a08be370
commit
d5eb76f32e
6 files changed
+19
-24
No files matched your search
@@ -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
@@ -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
@@ -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
@@ -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
@@ -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
@@ -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
|
||||||
})
|
})
|
||||||
|
|||||||
Reference in new issue
Block a user