diff options
author | Janoš Guljaš <janos@users.noreply.github.com> | 2019-02-13 20:03:23 +0800 |
---|---|---|
committer | Rafael Matias <rafael@skyle.net> | 2019-02-19 20:11:52 +0800 |
commit | 8ea3d8ad90f90e7233e829ad141acfd9d911658c (patch) | |
tree | ed6e0fcb16c90c460052a62a66d3aa253ea3498d /swarm/network/stream/delivery_test.go | |
parent | a0127019c3d516e8d8cf83839583bcf71af763e0 (diff) | |
download | dexon-8ea3d8ad90f90e7233e829ad141acfd9d911658c.tar dexon-8ea3d8ad90f90e7233e829ad141acfd9d911658c.tar.gz dexon-8ea3d8ad90f90e7233e829ad141acfd9d911658c.tar.bz2 dexon-8ea3d8ad90f90e7233e829ad141acfd9d911658c.tar.lz dexon-8ea3d8ad90f90e7233e829ad141acfd9d911658c.tar.xz dexon-8ea3d8ad90f90e7233e829ad141acfd9d911658c.tar.zst dexon-8ea3d8ad90f90e7233e829ad141acfd9d911658c.zip |
swarm: fix network/stream data races (#19051)
* swarm/network/stream: newStreamerTester cleanup only if err is nil
* swarm/network/stream: raise newStreamerTester waitForPeers timeout
* swarm/network/stream: fix data races in GetPeerSubscriptions
* swarm/storage: prevent data race on LDBStore.batchesC
https://github.com/ethersphere/go-ethereum/issues/1198#issuecomment-461775049
* swarm/network/stream: fix TestGetSubscriptionsRPC data race
https://github.com/ethersphere/go-ethereum/issues/1198#issuecomment-461768477
* swarm/network/stream: correctly use Simulation.Run callback
https://github.com/ethersphere/go-ethereum/issues/1198#issuecomment-461783804
* swarm/network: protect addrCountC in Kademlia.AddrCountC function
https://github.com/ethersphere/go-ethereum/issues/1198#issuecomment-462273444
* p2p/simulations: fix a deadlock calling getRandomNode with lock
https://github.com/ethersphere/go-ethereum/issues/1198#issuecomment-462317407
* swarm/network/stream: terminate disconnect goruotines in tests
* swarm/network/stream: reduce memory consumption when testing data races
* swarm/network/stream: add watchDisconnections helper function
* swarm/network/stream: add concurrent counter for tests
* swarm/network/stream: rename race/norace test files and use const
* swarm/network/stream: remove watchSim and its panic
* swarm/network/stream: pass context in watchDisconnections
* swarm/network/stream: add concurrent safe bool for watchDisconnections
* swarm/storage: fix LDBStore.batchesC data race by not closing it
(cherry picked from commit 3fd6db2bf63ce90232de445c7f33943406a5e634)
Diffstat (limited to 'swarm/network/stream/delivery_test.go')
-rw-r--r-- | swarm/network/stream/delivery_test.go | 75 |
1 files changed, 20 insertions, 55 deletions
diff --git a/swarm/network/stream/delivery_test.go b/swarm/network/stream/delivery_test.go index cb7690f3e..e5821df4f 100644 --- a/swarm/network/stream/delivery_test.go +++ b/swarm/network/stream/delivery_test.go @@ -22,7 +22,6 @@ import ( "errors" "fmt" "sync" - "sync/atomic" "testing" "time" @@ -48,10 +47,10 @@ func TestStreamerRetrieveRequest(t *testing.T) { Syncing: SyncingDisabled, } tester, streamer, _, teardown, err := newStreamerTester(regOpts) - defer teardown() if err != nil { t.Fatal(err) } + defer teardown() node := tester.Nodes[0] @@ -100,10 +99,10 @@ func TestStreamerUpstreamRetrieveRequestMsgExchangeWithoutStore(t *testing.T) { Retrieval: RetrievalEnabled, Syncing: SyncingDisabled, //do no syncing }) - defer teardown() if err != nil { t.Fatal(err) } + defer teardown() node := tester.Nodes[0] @@ -172,10 +171,10 @@ func TestStreamerUpstreamRetrieveRequestMsgExchange(t *testing.T) { Retrieval: RetrievalEnabled, Syncing: SyncingDisabled, }) - defer teardown() if err != nil { t.Fatal(err) } + defer teardown() node := tester.Nodes[0] @@ -362,10 +361,10 @@ func TestStreamerDownstreamChunkDeliveryMsgExchange(t *testing.T) { Retrieval: RetrievalDisabled, Syncing: SyncingDisabled, }) - defer teardown() if err != nil { t.Fatal(err) } + defer teardown() streamer.RegisterClientFunc("foo", func(p *Peer, t string, live bool) (Client, error) { return &testClient{ @@ -485,7 +484,8 @@ func testDeliveryFromNodes(t *testing.T, nodes, chunkCount int, skipCheck bool) } log.Info("Starting simulation") - ctx := context.Background() + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() result := sim.Run(ctx, func(ctx context.Context, sim *simulation.Simulation) (err error) { nodeIDs := sim.UpNodeIDs() //determine the pivot node to be the first node of the simulation @@ -548,27 +548,10 @@ func testDeliveryFromNodes(t *testing.T, nodes, chunkCount int, skipCheck bool) retErrC <- err }() - log.Debug("Watching for disconnections") - disconnections := sim.PeerEvents( - context.Background(), - sim.NodeIDs(), - simulation.NewPeerEventsFilter().Drop(), - ) - - var disconnected atomic.Value - go func() { - for d := range disconnections { - if d.Error != nil { - log.Error("peer drop", "node", d.NodeID, "peer", d.PeerID) - disconnected.Store(true) - } - } - }() + disconnected := watchDisconnections(ctx, sim) defer func() { - if err != nil { - if yes, ok := disconnected.Load().(bool); ok && yes { - err = errors.New("disconnect events received") - } + if err != nil && disconnected.bool() { + err = errors.New("disconnect events received") } }() @@ -589,7 +572,7 @@ func testDeliveryFromNodes(t *testing.T, nodes, chunkCount int, skipCheck bool) return fmt.Errorf("Test failed, chunks not available on all nodes") } if err := <-retErrC; err != nil { - t.Fatalf("requesting chunks: %v", err) + return fmt.Errorf("requesting chunks: %v", err) } log.Debug("Test terminated successfully") return nil @@ -657,21 +640,22 @@ func benchmarkDeliveryFromNodes(b *testing.B, nodes, chunkCount int, skipCheck b b.Fatal(err) } - ctx := context.Background() + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() result := sim.Run(ctx, func(ctx context.Context, sim *simulation.Simulation) (err error) { nodeIDs := sim.UpNodeIDs() node := nodeIDs[len(nodeIDs)-1] item, ok := sim.NodeItem(node, bucketKeyFileStore) if !ok { - b.Fatal("No filestore") + return errors.New("No filestore") } remoteFileStore := item.(*storage.FileStore) pivotNode := nodeIDs[0] item, ok = sim.NodeItem(pivotNode, bucketKeyNetStore) if !ok { - b.Fatal("No filestore") + return errors.New("No filestore") } netStore := item.(*storage.NetStore) @@ -679,26 +663,10 @@ func benchmarkDeliveryFromNodes(b *testing.B, nodes, chunkCount int, skipCheck b return err } - disconnections := sim.PeerEvents( - context.Background(), - sim.NodeIDs(), - simulation.NewPeerEventsFilter().Drop(), - ) - - var disconnected atomic.Value - go func() { - for d := range disconnections { - if d.Error != nil { - log.Error("peer drop", "node", d.NodeID, "peer", d.PeerID) - disconnected.Store(true) - } - } - }() + disconnected := watchDisconnections(ctx, sim) defer func() { - if err != nil { - if yes, ok := disconnected.Load().(bool); ok && yes { - err = errors.New("disconnect events received") - } + if err != nil && disconnected.bool() { + err = errors.New("disconnect events received") } }() // benchmark loop @@ -713,12 +681,12 @@ func benchmarkDeliveryFromNodes(b *testing.B, nodes, chunkCount int, skipCheck b ctx := context.TODO() hash, wait, err := remoteFileStore.Store(ctx, testutil.RandomReader(i, chunkSize), int64(chunkSize), false) if err != nil { - b.Fatalf("expected no error. got %v", err) + return fmt.Errorf("store: %v", err) } // wait until all chunks stored err = wait(ctx) if err != nil { - b.Fatalf("expected no error. got %v", err) + return fmt.Errorf("wait store: %v", err) } // collect the hashes hashes[i] = hash @@ -754,10 +722,7 @@ func benchmarkDeliveryFromNodes(b *testing.B, nodes, chunkCount int, skipCheck b break Loop } } - if err != nil { - b.Fatal(err) - } - return nil + return err }) if result.Error != nil { b.Fatal(result.Error) |