Fix stream race tests

swarm/network/stream/common_test.go

@@ -151,7 +151,7 @@ func newStreamerTester(registryOptions *RegistryOptions) (*p2ptest.ProtocolTeste
 	// temp datadir
 	datadir, err := ioutil.TempDir("", "streamer")
 	if err != nil {
-		return nil, nil, nil, func() {}, err
+		return nil, nil, nil, nil, err
 	}
 	removeDataDir := func() {
 		os.RemoveAll(datadir)
@@ -163,12 +163,14 @@ func newStreamerTester(registryOptions *RegistryOptions) (*p2ptest.ProtocolTeste
 
 	localStore, err := storage.NewTestLocalStoreForAddr(params)
 	if err != nil {
-		return nil, nil, nil, removeDataDir, err
+		removeDataDir()
+		return nil, nil, nil, nil, err
 	}
 
 	netStore, err := storage.NewNetStore(localStore, nil)
 	if err != nil {
-		return nil, nil, nil, removeDataDir, err
+		removeDataDir()
+		return nil, nil, nil, nil, err
 	}
 
 	delivery := NewDelivery(to, netStore)
@@ -180,8 +182,9 @@ func newStreamerTester(registryOptions *RegistryOptions) (*p2ptest.ProtocolTeste
 	}
 	protocolTester := p2ptest.NewProtocolTester(addr.ID(), 1, streamer.runProtocol)
 
-	err = waitForPeers(streamer, 1*time.Second, 1)
+	err = waitForPeers(streamer, 10*time.Second, 1)
 	if err != nil {
+		teardown()
 		return nil, nil, nil, nil, errors.New("timeout: peer is not created")
 	}
 

swarm/network/stream/delivery_test.go

@@ -48,10 +48,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 +100,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 +172,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 +362,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{
@@ -589,7 +589,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
@@ -664,14 +664,14 @@ func benchmarkDeliveryFromNodes(b *testing.B, nodes, chunkCount int, skipCheck b
 
 		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)
 
@@ -713,12 +713,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 +754,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)

swarm/network/stream/intervals_test.go

@@ -118,13 +118,11 @@ func testIntervals(t *testing.T, live bool, history *Range, skipCheck bool) {
 
 		_, wait, err := fileStore.Store(ctx, testutil.RandomReader(1, size), int64(size), false)
 		if err != nil {
-			log.Error("Store error: %v", "err", err)
-			t.Fatal(err)
+			return fmt.Errorf("store: %v", err)
 		}
 		err = wait(ctx)
 		if err != nil {
-			log.Error("Wait error: %v", "err", err)
-			t.Fatal(err)
+			return fmt.Errorf("wait store: %v", err)
 		}
 
 		item, ok = sim.NodeItem(checker, bucketKeyRegistry)

swarm/network/stream/lightnode_test.go

@@ -29,10 +29,10 @@ func TestLigthnodeRetrieveRequestWithRetrieve(t *testing.T) {
 		Syncing:   SyncingDisabled,
 	}
 	tester, _, _, teardown, err := newStreamerTester(registryOptions)
-	defer teardown()
 	if err != nil {
 		t.Fatal(err)
 	}
+	defer teardown()
 
 	node := tester.Nodes[0]
 
@@ -68,10 +68,10 @@ func TestLigthnodeRetrieveRequestWithoutRetrieve(t *testing.T) {
 		Syncing:   SyncingDisabled,
 	}
 	tester, _, _, teardown, err := newStreamerTester(registryOptions)
-	defer teardown()
 	if err != nil {
 		t.Fatal(err)
 	}
+	defer teardown()
 
 	node := tester.Nodes[0]
 
@@ -112,10 +112,10 @@ func TestLigthnodeRequestSubscriptionWithSync(t *testing.T) {
 		Syncing:   SyncingRegisterOnly,
 	}
 	tester, _, _, teardown, err := newStreamerTester(registryOptions)
-	defer teardown()
 	if err != nil {
 		t.Fatal(err)
 	}
+	defer teardown()
 
 	node := tester.Nodes[0]
 
@@ -157,10 +157,10 @@ func TestLigthnodeRequestSubscriptionWithoutSync(t *testing.T) {
 		Syncing:   SyncingDisabled,
 	}
 	tester, _, _, teardown, err := newStreamerTester(registryOptions)
-	defer teardown()
 	if err != nil {
 		t.Fatal(err)
 	}
+	defer teardown()
 
 	node := tester.Nodes[0]
 

swarm/network/stream/stream.go

@@ -938,16 +938,22 @@ It returns a map of node IDs with an array of string representations of Stream o
 func (api *API) GetPeerSubscriptions() map[string][]string {
 	//create the empty map
 	pstreams := make(map[string][]string)
+
 	//iterate all streamer peers
+	api.streamer.peersMu.RLock()
+	defer api.streamer.peersMu.RUnlock()
+
 	for id, p := range api.streamer.peers {
 		var streams []string
 		//every peer has a map of stream servers
 		//every stream server represents a subscription
+		p.serverMu.RLock()
 		for s := range p.servers {
 			//append the string representation of the stream
 			//to the list for this peer
 			streams = append(streams, s.String())
 		}
+		p.serverMu.RUnlock()
 		//set the array of stream servers to the map
 		pstreams[id.String()] = streams
 	}