Merge pull request #385 from ethersphere/stream-test-rpc-unsubscribe-deadlock

Stream test RPC unsubscribe deadlock

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

@@ -200,15 +200,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()
@@ -292,7 +297,16 @@ 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
+			}
+			rpcSubscriptionsWg.Add(1)
+			go func() {
+				<-wsDoneC
+				rpcSubscriptionsWg.Done()
+			}()
 
 			if log.Lvl(*loglevel) >= log.LvlTrace {
 				//this will print the kademlia tables of all nodes
@@ -308,10 +322,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 {
@@ -588,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 {
@@ -596,13 +615,23 @@ 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()
+				select {
+				case errc <- ctx.Err():
+				case <-quitC:
+				}
 				return
 			case e := <-events:
 				//just catch SubscribeMsg
@@ -611,12 +640,16 @@ func watchSubscriptionEvents(ctx context.Context, id discover.NodeID, client *rp
 				}
 			case err := <-sub.Err():
 				if err != nil {
-					errc <- fmt.Errorf("error getting peer events for node %v: %v", id, err)
+					select {
+					case errc <- fmt.Errorf("error getting peer events for node %v: %v", id, err):
+					case <-quitC:
+					}
 					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

@@ -216,14 +216,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 +256,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 {