From 3873a7314d6c82f3128a627e55b38a7437a30f9f Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jano=C5=A1=20Gulja=C5=A1?= Date: Thu, 25 Apr 2019 21:33:18 +0200 Subject: [PATCH] swarm/network: fix data races in TestInitialPeersMsg test (#19490) * swarm/network: fix data races in TestInitialPeersMsg test * swarm/network: add Kademlia.Saturation method with lock * swarm/network: add Hive.Peer method to safely retrieve a bzz peer * swarm/network: remove duplicate comment * p2p/testing: prevent goroutine leak in ProtocolTester * swarm/network: fix data race in newBzzBaseTesterWithAddrs * swarm/network: fix goroutone leaks in testInitialPeersMsg * swarm/network: raise number of peer check attempts in testInitialPeersMsg * swarm/network: use Hive.Peer in Hive.PeerInfo function * swarm/network: reduce the scope of mutex lock in newBzzBaseTesterWithAddrs * swarm/storage: disable TestCleanIndex with race detector --- p2p/testing/protocoltester.go | 1 + swarm/network/discovery_test.go | 8 +++++--- swarm/network/hive.go | 13 ++++++++++--- swarm/network/kademlia.go | 9 ++++++++- swarm/network/protocol_test.go | 9 ++++++++- swarm/storage/ldbstore_test.go | 7 +++++-- 6 files changed, 37 insertions(+), 10 deletions(-) diff --git a/p2p/testing/protocoltester.go b/p2p/testing/protocoltester.go index 1e1752af8..e798240a5 100644 --- a/p2p/testing/protocoltester.go +++ b/p2p/testing/protocoltester.go @@ -110,6 +110,7 @@ func NewProtocolTester(prvkey *ecdsa.PrivateKey, nodeCount int, run func(*p2p.Pe // Stop stops the p2p server func (t *ProtocolTester) Stop() { t.Server.Stop() + t.network.Shutdown() } // Connect brings up the remote peer node and connects it using the diff --git a/swarm/network/discovery_test.go b/swarm/network/discovery_test.go index 04e1b36fe..cf7ddc4fb 100644 --- a/swarm/network/discovery_test.go +++ b/swarm/network/discovery_test.go @@ -154,13 +154,15 @@ func testInitialPeersMsg(t *testing.T, peerPO, peerDepth int) { if err != nil { t.Fatal(err) } + defer s.Stop() // peerID to use in the protocol tester testExchange expect/trigger peerID := s.Nodes[0].ID() // block until control peer is found among hive peers found := false - for attempts := 0; attempts < 20; attempts++ { - if _, found = hive.peers[peerID]; found { + for attempts := 0; attempts < 2000; attempts++ { + found = hive.Peer(peerID) != nil + if found { break } time.Sleep(1 * time.Millisecond) @@ -171,7 +173,7 @@ func testInitialPeersMsg(t *testing.T, peerPO, peerDepth int) { } // pivotDepth is the advertised depth of the pivot node we expect in the outgoing subPeersMsg - pivotDepth := hive.saturation() + pivotDepth := hive.Saturation() // the test exchange is as follows: // 1. pivot sends to the control peer a `subPeersMsg` advertising its depth (ignored) // 2. peer sends to pivot a `subPeersMsg` advertising its own depth (arbitrarily chosen) diff --git a/swarm/network/hive.go b/swarm/network/hive.go index a0b6b988a..2eb521f1d 100644 --- a/swarm/network/hive.go +++ b/swarm/network/hive.go @@ -192,9 +192,7 @@ func (h *Hive) NodeInfo() interface{} { // PeerInfo function is used by the p2p.server RPC interface to display // protocol specific information any connected peer referred to by their NodeID func (h *Hive) PeerInfo(id enode.ID) interface{} { - h.lock.Lock() - p := h.peers[id] - h.lock.Unlock() + p := h.Peer(id) if p == nil { return nil @@ -209,6 +207,15 @@ func (h *Hive) PeerInfo(id enode.ID) interface{} { } } +// Peer returns a bzz peer from the Hive. If there is no peer +// with the provided enode id, a nil value is returned. +func (h *Hive) Peer(id enode.ID) *BzzPeer { + h.lock.Lock() + defer h.lock.Unlock() + + return h.peers[id] +} + // loadPeers, savePeer implement persistence callback/ func (h *Hive) loadPeers() error { var as []*BzzAddr diff --git a/swarm/network/kademlia.go b/swarm/network/kademlia.go index 304f9cd77..dd6de44fd 100644 --- a/swarm/network/kademlia.go +++ b/swarm/network/kademlia.go @@ -735,8 +735,15 @@ func NewPeerPotMap(neighbourhoodSize int, addrs [][]byte) map[string]*PeerPot { return ppmap } -// saturation returns the smallest po value in which the node has less than MinBinSize peers +// Saturation returns the smallest po value in which the node has less than MinBinSize peers // if the iterator reaches neighbourhood radius, then the last bin + 1 is returned +func (k *Kademlia) Saturation() int { + k.lock.RLock() + defer k.lock.RUnlock() + + return k.saturation() +} + func (k *Kademlia) saturation() int { prev := -1 radius := neighbourhoodRadiusForPot(k.conns, k.NeighbourhoodSize, k.base) diff --git a/swarm/network/protocol_test.go b/swarm/network/protocol_test.go index b562a4253..2207ba308 100644 --- a/swarm/network/protocol_test.go +++ b/swarm/network/protocol_test.go @@ -86,9 +86,12 @@ func newBzzBaseTester(n int, prvkey *ecdsa.PrivateKey, spec *protocols.Spec, run func newBzzBaseTesterWithAddrs(prvkey *ecdsa.PrivateKey, addrs [][]byte, spec *protocols.Spec, run func(*BzzPeer) error) (*bzzTester, [][]byte, error) { n := len(addrs) cs := make(map[enode.ID]chan bool) + var csMu sync.Mutex srv := func(p *BzzPeer) error { defer func() { + csMu.Lock() + defer csMu.Unlock() if cs[p.ID()] != nil { close(cs[p.ID()]) } @@ -99,8 +102,8 @@ func newBzzBaseTesterWithAddrs(prvkey *ecdsa.PrivateKey, addrs [][]byte, spec *p nodeToAddr := make(map[enode.ID][]byte) protocol := func(p *p2p.Peer, rw p2p.MsgReadWriter) error { mu.Lock() - defer mu.Unlock() nodeToAddr[p.ID()] = addrs[0] + mu.Unlock() bzzAddr := &BzzAddr{addrs[0], []byte(p.Node().String())} addrs = addrs[1:] return srv(&BzzPeer{Peer: protocols.NewPeer(p, rw, spec), BzzAddr: bzzAddr}) @@ -120,10 +123,12 @@ func newBzzBaseTesterWithAddrs(prvkey *ecdsa.PrivateKey, addrs [][]byte, spec *p } addr := getENRBzzAddr(nod) + csMu.Lock() for _, node := range s.Nodes { log.Warn("node", "node", node) cs[node.ID()] = make(chan bool) } + csMu.Unlock() var nodeAddrs [][]byte pt := &bzzTester{ @@ -131,9 +136,11 @@ func newBzzBaseTesterWithAddrs(prvkey *ecdsa.PrivateKey, addrs [][]byte, spec *p ProtocolTester: s, cs: cs, } + mu.Lock() for _, n := range pt.Nodes { nodeAddrs = append(nodeAddrs, nodeToAddr[n.ID()]) } + mu.Unlock() return pt, nodeAddrs, nil } diff --git a/swarm/storage/ldbstore_test.go b/swarm/storage/ldbstore_test.go index 70b0d6bb4..1cd4947be 100644 --- a/swarm/storage/ldbstore_test.go +++ b/swarm/storage/ldbstore_test.go @@ -27,12 +27,11 @@ import ( "strings" "testing" - "github.com/ethereum/go-ethereum/swarm/testutil" - "github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/swarm/chunk" "github.com/ethereum/go-ethereum/swarm/log" "github.com/ethereum/go-ethereum/swarm/storage/mock/mem" + "github.com/ethereum/go-ethereum/swarm/testutil" ldberrors "github.com/syndtr/goleveldb/leveldb/errors" ) @@ -606,6 +605,10 @@ func TestLDBStoreCollectGarbageAccessUnlikeIndex(t *testing.T) { } func TestCleanIndex(t *testing.T) { + if testutil.RaceEnabled { + t.Skip("disabled because it times out with race detector") + } + capacity := 5000 n := 3