swarm/network/stream: fix TestSyncerSimulation deadlock

This change fixes the synchronization between simulations network
shutdown and calling of rpc subscription Unsubscribe method.
If the unsubscribe is called after net shutdown, rpc client is
blocking.

This change also addresses the panic in benchmarkDeliveryFromNodes
function, when executing benchmarks from tests.

swarm/network/stream/delivery_test.go

@@ -22,6 +22,7 @@ import (
 	crand "crypto/rand"
 	"fmt"
 	"io"
+	"sync"
 	"testing"
 	"time"
 
@@ -318,7 +319,11 @@ func testDeliveryFromNodes(t *testing.T, nodes, conns, chunkCount int, skipCheck
 	}
 
 	sim, teardown, err := streamTesting.NewSimulation(conf)
-	defer teardown()
+	var rpcSubscriptionsWg sync.WaitGroup
+	defer func() {
+		rpcSubscriptionsWg.Wait()
+		teardown()
+	}()
 	if err != nil {
 		t.Fatal(err.Error())
 	}
@@ -347,6 +352,7 @@ func testDeliveryFromNodes(t *testing.T, nodes, conns, chunkCount int, skipCheck
 	errc := make(chan error, 1)
 	waitPeerErrC = make(chan error)
 	quitC := make(chan struct{})
+	defer close(quitC)
 
 	action := func(ctx context.Context) error {
 		// each node Subscribes to each other's swarmChunkServerStreamName
@@ -369,10 +375,15 @@ func testDeliveryFromNodes(t *testing.T, nodes, conns, chunkCount int, skipCheck
 		for j := 0; j < nodes-1; j++ {
 			id := sim.IDs[j]
 			err := sim.CallClient(id, func(client *rpc.Client) error {
-				err := streamTesting.WatchDisconnections(id, client, errc, quitC)
+				doneC, err := streamTesting.WatchDisconnections(id, client, errc, quitC)
 				if err != nil {
 					return err
 				}
+				rpcSubscriptionsWg.Add(1)
+				go func() {
+					<-doneC
+					rpcSubscriptionsWg.Done()
+				}()
 				ctx, cancel := context.WithTimeout(ctx, 1*time.Second)
 				defer cancel()
 				sid := sim.IDs[j+1]
@@ -475,6 +486,8 @@ func BenchmarkDeliveryFromNodesWithCheck(b *testing.B) {
 func benchmarkDeliveryFromNodes(b *testing.B, nodes, conns, chunkCount int, skipCheck bool) {
 	defaultSkipCheck = skipCheck
 	toAddr = network.NewAddrFromNodeID
+	createStoreFunc = createTestLocalStorageFromSim
+	registries = make(map[discover.NodeID]*TestRegistry)
 
 	timeout := 300 * time.Second
 	ctx, cancel := context.WithTimeout(context.Background(), timeout)
@@ -489,7 +502,11 @@ func benchmarkDeliveryFromNodes(b *testing.B, nodes, conns, chunkCount int, skip
 		EnableMsgEvents: false,
 	}
 	sim, teardown, err := streamTesting.NewSimulation(conf)
-	defer teardown()
+	var rpcSubscriptionsWg sync.WaitGroup
+	defer func() {
+		rpcSubscriptionsWg.Wait()
+		teardown()
+	}()
 	if err != nil {
 		b.Fatal(err.Error())
 	}
@@ -539,10 +556,15 @@ func benchmarkDeliveryFromNodes(b *testing.B, nodes, conns, chunkCount int, skip
 		for j := 0; j < nodes-1; j++ {
 			id := sim.IDs[j]
 			err = sim.CallClient(id, func(client *rpc.Client) error {
-				err := streamTesting.WatchDisconnections(id, client, disconnectC, quitC)
+				doneC, err := streamTesting.WatchDisconnections(id, client, disconnectC, quitC)
 				if err != nil {
 					return err
 				}
+				rpcSubscriptionsWg.Add(1)
+				go func() {
+					<-doneC
+					rpcSubscriptionsWg.Done()
+				}()
 				ctx, cancel := context.WithTimeout(ctx, 1*time.Second)
 				defer cancel()
 				sid := sim.IDs[j+1] // the upstream peer's id

swarm/network/stream/intervals_test.go

@@ -22,6 +22,7 @@ import (
 	"encoding/binary"
 	"fmt"
 	"io"
+	"sync"
 	"testing"
 	"time"
 
@@ -90,7 +91,11 @@ func testIntervals(t *testing.T, live bool, history *Range) {
 	}
 
 	sim, teardown, err := streamTesting.NewSimulation(conf)
-	defer teardown()
+	var rpcSubscriptionsWg sync.WaitGroup
+	defer func() {
+		rpcSubscriptionsWg.Wait()
+		teardown()
+	}()
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -136,10 +141,15 @@ func testIntervals(t *testing.T, live bool, history *Range) {
 
 			sid := sim.IDs[0]
 
-			err := streamTesting.WatchDisconnections(id, client, errc, quitC)
+			doneC, err := streamTesting.WatchDisconnections(id, client, errc, quitC)
 			if err != nil {
 				return err
 			}
+			rpcSubscriptionsWg.Add(1)
+			go func() {
+				<-doneC
+				rpcSubscriptionsWg.Done()
+			}()
 			ctx, cancel := context.WithTimeout(ctx, 100*time.Second)
 			defer cancel()
 

swarm/network/stream/snapshot_sync_test.go

@@ -199,15 +199,20 @@ func runSyncTest(chunkCount int, nodeCount int, live bool, history bool) error {
 	if err != nil {
 		return err
 	}
+	var rpcSubscriptionsWg sync.WaitGroup
 	//do cleanup after test is terminated
 	defer func() {
+		// close quitC channel to signall all goroutines to clanup
+		// before calling simulation network shutdown.
+		close(quitC)
+		//wait for all rpc subscriptions to unsubscribe
+		rpcSubscriptionsWg.Wait()
 		//shutdown the snapshot network
 		net.Shutdown()
 		//after the test, clean up local stores initialized with createLocalStoreForId
 		localStoreCleanup()
 		//finally clear all data directories
 		datadirsCleanup()
-		close(quitC)
 	}()
 	//get the nodes of the network
 	nodes := net.GetNodes()
@@ -290,7 +295,15 @@ func runSyncTest(chunkCount int, nodeCount int, live bool, history bool) error {
 				return err
 			}
 
-			watchSubscriptionEvents(ctx, id, client, errc)
+			wsDoneC := watchSubscriptionEvents(ctx, id, client, errc, quitC)
+			// doneC is nil, the error happened which is sent to errc channel, already
+			if wsDoneC == nil {
+				continue
+			}
+			go func() {
+				<-wsDoneC
+				rpcSubscriptionsWg.Done()
+			}()
 
 			if log.Lvl(*loglevel) >= log.LvlTrace {
 				//this will print the kademlia tables of all nodes
@@ -306,10 +319,15 @@ func runSyncTest(chunkCount int, nodeCount int, live bool, history bool) error {
 				log.Debug(kt)
 			}
 			//watch for peers disconnecting
-			err = streamTesting.WatchDisconnections(id, client, disconnectC, quitC)
+			wdDoneC, err := streamTesting.WatchDisconnections(id, client, disconnectC, quitC)
 			if err != nil {
 				return err
 			}
+			rpcSubscriptionsWg.Add(1)
+			go func() {
+				<-wdDoneC
+				rpcSubscriptionsWg.Done()
+			}()
 			//start syncing!
 			err = client.CallContext(ctx, nil, "stream_startSyncing")
 			if err != nil {
@@ -589,7 +607,7 @@ func initNetWithSnapshot(nodeCount int) (*simulations.Network, error) {
 
 //we want to wait for subscriptions to be established before uploading to test
 //that live syncing is working correctly
-func watchSubscriptionEvents(ctx context.Context, id discover.NodeID, client *rpc.Client, errc chan error) {
+func watchSubscriptionEvents(ctx context.Context, id discover.NodeID, client *rpc.Client, errc chan error, quitC chan struct{}) (doneC <-chan struct{}) {
 	events := make(chan *p2p.PeerEvent)
 	sub, err := client.Subscribe(context.Background(), "admin", events, "peerEvents")
 	if err != nil {
@@ -597,11 +615,18 @@ func watchSubscriptionEvents(ctx context.Context, id discover.NodeID, client *rp
 		errc <- fmt.Errorf("error getting peer events for node %v: %s", id, err)
 		return
 	}
+	c := make(chan struct{})
 	go func() {
-		defer sub.Unsubscribe()
+		defer func() {
+			log.Trace("watch subscription events: unsubscribe", "id", id)
+			sub.Unsubscribe()
+			close(c)
+		}()
 
 		for {
 			select {
+			case <-quitC:
+				return
 			case <-ctx.Done():
 				errc <- ctx.Err()
 				return
@@ -618,7 +643,7 @@ func watchSubscriptionEvents(ctx context.Context, id discover.NodeID, client *rp
 			}
 		}
 	}()
-	return
+	return c
 }
 
 //create a local store for the given node

swarm/network/stream/syncer_test.go

@@ -22,6 +22,7 @@ import (
 	"fmt"
 	"io"
 	"math"
+	"sync"
 	"testing"
 	"time"
 
@@ -69,7 +70,11 @@ func testSyncBetweenNodes(t *testing.T, nodes, conns, chunkCount int, skipCheck
 
 	// create simulation network with the config
 	sim, teardown, err := streamTesting.NewSimulation(conf)
-	defer teardown()
+	var rpcSubscriptionsWg sync.WaitGroup
+	defer func() {
+		rpcSubscriptionsWg.Wait()
+		teardown()
+	}()
 	if err != nil {
 		t.Fatal(err.Error())
 	}
@@ -154,10 +159,15 @@ func testSyncBetweenNodes(t *testing.T, nodes, conns, chunkCount int, skipCheck
 			id := sim.IDs[j]
 			err := sim.CallClient(id, func(client *rpc.Client) error {
 				// report disconnect events to the error channel cos peers should not disconnect
-				err := streamTesting.WatchDisconnections(id, client, errc, quitC)
+				doneC, err := streamTesting.WatchDisconnections(id, client, errc, quitC)
 				if err != nil {
 					return err
 				}
+				rpcSubscriptionsWg.Add(1)
+				go func() {
+					<-doneC
+					rpcSubscriptionsWg.Done()
+				}()
 				ctx, cancel := context.WithTimeout(ctx, 1*time.Second)
 				defer cancel()
 				// start syncing, i.e., subscribe to upstream peers po 1 bin

swarm/network/stream/testing/testing.go

@@ -143,7 +143,9 @@ func NewSimulation(conf *RunConfig) (*Simulation, func(), error) {
 		DefaultService: defaultService,
 	})
 	teardown := func() {
+		log.Trace("simulation: teardown adapter")
 		adapterTeardown()
+		log.Trace("simulation: teardown net")
 		net.Shutdown()
 	}
 	ids := make([]discover.NodeID, nodes)
@@ -162,8 +164,11 @@ func NewSimulation(conf *RunConfig) (*Simulation, func(), error) {
 	// set nodes number of Stores available
 	stores, storeTeardown, err := SetStores(addrs...)
 	teardown = func() {
+		log.Trace("simulation: teardown net")
 		net.Shutdown()
+		log.Trace("simulation: teardown adapter")
 		adapterTeardown()
+		log.Trace("simulation: teardown store")
 		storeTeardown()
 	}
 	if err != nil {
@@ -216,14 +221,23 @@ func (s *Simulation) Run(ctx context.Context, conf *RunConfig) (*simulations.Ste
 	return result, nil
 }
 
-func WatchDisconnections(id discover.NodeID, client *rpc.Client, errc chan error, quitC chan struct{}) error {
+// WatchDisconnections subscribes to admin peerEvents and sends peer event drop
+// errors to the errc channel. Channel quitC signals the termination of the event loop.
+// Returned doneC will be closed after the rpc subscription is unsubscribed,
+// signaling that simulations network is safe to shutdown.
+func WatchDisconnections(id discover.NodeID, client *rpc.Client, errc chan error, quitC chan struct{}) (doneC <-chan struct{}, err error) {
 	events := make(chan *p2p.PeerEvent)
 	sub, err := client.Subscribe(context.Background(), "admin", events, "peerEvents")
 	if err != nil {
-		return fmt.Errorf("error getting peer events for node %v: %s", id, err)
+		return nil, fmt.Errorf("error getting peer events for node %v: %s", id, err)
 	}
+	c := make(chan struct{})
 	go func() {
-		defer sub.Unsubscribe()
+		defer func() {
+			log.Trace("watch disconnections: unsubscribe", "id", id)
+			sub.Unsubscribe()
+			close(c)
+		}()
 		for {
 			select {
 			case <-quitC:
@@ -247,7 +261,7 @@ func WatchDisconnections(id discover.NodeID, client *rpc.Client, errc chan error
 			}
 		}
 	}()
-	return nil
+	return c, nil
 }
 
 func Trigger(d time.Duration, quitC chan struct{}, ids ...discover.NodeID) chan discover.NodeID {