From d70c4faf20d5533e30eec5cbb9b5180eb837b78c Mon Sep 17 00:00:00 2001 From: Janoš Guljaš Date: Wed, 9 Jan 2019 07:05:55 +0100 Subject: swarm: Fix T.Fatal inside a goroutine in tests (#18409) * swarm/storage: fix T.Fatal inside a goroutine * swarm/network/simulation: fix T.Fatal inside a goroutine * swarm/network/stream: fix T.Fatal inside a goroutine * swarm/network/simulation: consistent failures in TestPeerEventsTimeout * swarm/network/simulation: rename sendRunSignal to triggerSimulationRun --- swarm/network/stream/delivery_test.go | 34 ++++++++++++++++++++++++------ swarm/network/stream/intervals_test.go | 14 ++++++++++-- swarm/network/stream/snapshot_sync_test.go | 23 ++++++++++++++------ swarm/network/stream/syncer_test.go | 14 ++++++++++-- 4 files changed, 68 insertions(+), 17 deletions(-) (limited to 'swarm/network/stream') diff --git a/swarm/network/stream/delivery_test.go b/swarm/network/stream/delivery_test.go index 04a4bb077..c65e1386d 100644 --- a/swarm/network/stream/delivery_test.go +++ b/swarm/network/stream/delivery_test.go @@ -19,9 +19,11 @@ package stream import ( "bytes" "context" + "errors" "fmt" "os" "sync" + "sync/atomic" "testing" "time" @@ -500,7 +502,7 @@ func testDeliveryFromNodes(t *testing.T, nodes, chunkCount int, skipCheck bool) log.Info("Starting simulation") ctx := context.Background() - result := sim.Run(ctx, func(ctx context.Context, sim *simulation.Simulation) error { + 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 pivot := nodeIDs[0] @@ -553,14 +555,13 @@ func testDeliveryFromNodes(t *testing.T, nodes, chunkCount int, skipCheck bool) } pivotFileStore := item.(*storage.FileStore) log.Debug("Starting retrieval routine") + retErrC := make(chan error) go func() { // start the retrieval on the pivot node - this will spawn retrieve requests for missing chunks // we must wait for the peer connections to have started before requesting n, err := readAll(pivotFileStore, fileHash) log.Info(fmt.Sprintf("retrieved %v", fileHash), "read", n, "err", err) - if err != nil { - t.Fatalf("requesting chunks action error: %v", err) - } + retErrC <- err }() log.Debug("Watching for disconnections") @@ -570,11 +571,19 @@ func testDeliveryFromNodes(t *testing.T, nodes, chunkCount int, skipCheck bool) 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) - t.Fatal(d.Error) + disconnected.Store(true) + } + } + }() + defer func() { + if err != nil { + if yes, ok := disconnected.Load().(bool); ok && yes { + err = errors.New("disconnect events received") } } }() @@ -595,6 +604,9 @@ func testDeliveryFromNodes(t *testing.T, nodes, chunkCount int, skipCheck bool) if !success { return fmt.Errorf("Test failed, chunks not available on all nodes") } + if err := <-retErrC; err != nil { + t.Fatalf("requesting chunks: %v", err) + } log.Debug("Test terminated successfully") return nil }) @@ -675,7 +687,7 @@ func benchmarkDeliveryFromNodes(b *testing.B, nodes, chunkCount int, skipCheck b } ctx := context.Background() - result := sim.Run(ctx, func(ctx context.Context, sim *simulation.Simulation) error { + result := sim.Run(ctx, func(ctx context.Context, sim *simulation.Simulation) (err error) { nodeIDs := sim.UpNodeIDs() node := nodeIDs[len(nodeIDs)-1] @@ -702,11 +714,19 @@ func benchmarkDeliveryFromNodes(b *testing.B, nodes, chunkCount int, skipCheck b 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) - b.Fatal(d.Error) + disconnected.Store(true) + } + } + }() + defer func() { + if err != nil { + if yes, ok := disconnected.Load().(bool); ok && yes { + err = errors.New("disconnect events received") } } }() diff --git a/swarm/network/stream/intervals_test.go b/swarm/network/stream/intervals_test.go index 7c7feeb11..121758c18 100644 --- a/swarm/network/stream/intervals_test.go +++ b/swarm/network/stream/intervals_test.go @@ -19,9 +19,11 @@ package stream import ( "context" "encoding/binary" + "errors" "fmt" "os" "sync" + "sync/atomic" "testing" "time" @@ -117,7 +119,7 @@ func testIntervals(t *testing.T, live bool, history *Range, skipCheck bool) { t.Fatal(err) } - result := sim.Run(ctx, func(ctx context.Context, sim *simulation.Simulation) error { + result := sim.Run(ctx, func(ctx context.Context, sim *simulation.Simulation) (err error) { nodeIDs := sim.UpNodeIDs() storer := nodeIDs[0] checker := nodeIDs[1] @@ -162,11 +164,19 @@ func testIntervals(t *testing.T, live bool, history *Range, skipCheck bool) { return err } + var disconnected atomic.Value go func() { for d := range disconnections { if d.Error != nil { log.Error("peer drop", "node", d.NodeID, "peer", d.PeerID) - t.Fatal(d.Error) + disconnected.Store(true) + } + } + }() + defer func() { + if err != nil { + if yes, ok := disconnected.Load().(bool); ok && yes { + err = errors.New("disconnect events received") } } }() diff --git a/swarm/network/stream/snapshot_sync_test.go b/swarm/network/stream/snapshot_sync_test.go index f86d9acca..7a883644b 100644 --- a/swarm/network/stream/snapshot_sync_test.go +++ b/swarm/network/stream/snapshot_sync_test.go @@ -21,6 +21,7 @@ import ( "os" "runtime" "sync" + "sync/atomic" "testing" "time" @@ -213,11 +214,13 @@ func testSyncingViaGlobalSync(t *testing.T, chunkCount int, nodeCount int) { simulation.NewPeerEventsFilter().Drop(), ) + var disconnected atomic.Value go func() { for d := range disconnections { - log.Error("peer drop", "node", d.NodeID, "peer", d.PeerID) - t.Fatal("unexpected disconnect") - cancelSimRun() + if d.Error != nil { + log.Error("peer drop", "node", d.NodeID, "peer", d.PeerID) + disconnected.Store(true) + } } }() @@ -226,6 +229,9 @@ func testSyncingViaGlobalSync(t *testing.T, chunkCount int, nodeCount int) { if result.Error != nil { t.Fatal(result.Error) } + if yes, ok := disconnected.Load().(bool); ok && yes { + t.Fatal("disconnect events received") + } log.Info("Simulation ended") } @@ -395,11 +401,13 @@ func testSyncingViaDirectSubscribe(t *testing.T, chunkCount int, nodeCount int) simulation.NewPeerEventsFilter().Drop(), ) + var disconnected atomic.Value go func() { for d := range disconnections { - log.Error("peer drop", "node", d.NodeID, "peer", d.PeerID) - t.Fatal("unexpected disconnect") - cancelSimRun() + if d.Error != nil { + log.Error("peer drop", "node", d.NodeID, "peer", d.PeerID) + disconnected.Store(true) + } } }() @@ -514,6 +522,9 @@ func testSyncingViaDirectSubscribe(t *testing.T, chunkCount int, nodeCount int) return result.Error } + if yes, ok := disconnected.Load().(bool); ok && yes { + t.Fatal("disconnect events received") + } log.Info("Simulation ended") return nil } diff --git a/swarm/network/stream/syncer_test.go b/swarm/network/stream/syncer_test.go index 2fdba71ff..09152ebfe 100644 --- a/swarm/network/stream/syncer_test.go +++ b/swarm/network/stream/syncer_test.go @@ -18,11 +18,13 @@ package stream import ( "context" + "errors" "fmt" "io/ioutil" "math" "os" "sync" + "sync/atomic" "testing" "time" @@ -129,7 +131,7 @@ func testSyncBetweenNodes(t *testing.T, nodes, chunkCount int, skipCheck bool, p if err != nil { t.Fatal(err) } - result := sim.Run(ctx, func(ctx context.Context, sim *simulation.Simulation) error { + result := sim.Run(ctx, func(ctx context.Context, sim *simulation.Simulation) (err error) { nodeIDs := sim.UpNodeIDs() nodeIndex := make(map[enode.ID]int) @@ -143,11 +145,19 @@ func testSyncBetweenNodes(t *testing.T, nodes, chunkCount int, skipCheck bool, p 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) - t.Fatal(d.Error) + disconnected.Store(true) + } + } + }() + defer func() { + if err != nil { + if yes, ok := disconnected.Load().(bool); ok && yes { + err = errors.New("disconnect events received") } } }() -- cgit