swarm/network/stream: add concurrent safe bool for watchDisconnections

swarm/network/stream/common_test.go

@@ -323,13 +323,14 @@ func createTestLocalStorageForID(id enode.ID, addr *network.BzzAddr) (storage.Ch
 
 // watchDisconnections receives simulation peer events in a new goroutine and sets atomic value
 // disconnected to true in case of a disconnect event.
-func watchDisconnections(ctx context.Context, sim *simulation.Simulation) (disconnected atomic.Value) {
+func watchDisconnections(ctx context.Context, sim *simulation.Simulation) (disconnected *boolean) {
 	log.Debug("Watching for disconnections")
 	disconnections := sim.PeerEvents(
 		ctx,
 		sim.NodeIDs(),
 		simulation.NewPeerEventsFilter().Drop(),
 	)
+	disconnected = new(boolean)
 	go func() {
 		for {
 			select {
@@ -341,9 +342,32 @@ func watchDisconnections(ctx context.Context, sim *simulation.Simulation) (disco
 				} else {
 					log.Error("peer drop", "node", d.NodeID, "peer", d.PeerID)
 				}
-				disconnected.Store(true)
+				disconnected.set(true)
 			}
 		}
 	}()
 	return disconnected
 }
+
+// boolean is used to concurrently set
+// and read a boolean value.
+type boolean struct {
+	v  bool
+	mu sync.RWMutex
+}
+
+// set sets the value.
+func (b *boolean) set(v bool) {
+	b.mu.Lock()
+	defer b.mu.Unlock()
+
+	b.v = v
+}
+
+// bool reads the value.
+func (b *boolean) bool() bool {
+	b.mu.RLock()
+	defer b.mu.RUnlock()
+
+	return b.v
+}

swarm/network/stream/delivery_test.go

@@ -550,10 +550,8 @@ func testDeliveryFromNodes(t *testing.T, nodes, chunkCount int, skipCheck bool)
 
 		disconnected := watchDisconnections(ctx, sim)
 		defer func() {
-			if err != nil {
+			if err != nil && disconnected.bool() {
-				if yes, ok := disconnected.Load().(bool); ok && yes {
+				err = errors.New("disconnect events received")
-					err = errors.New("disconnect events received")
-				}
 			}
 		}()
 
@@ -667,10 +665,8 @@ func benchmarkDeliveryFromNodes(b *testing.B, nodes, chunkCount int, skipCheck b
 
 		disconnected := watchDisconnections(ctx, sim)
 		defer func() {
-			if err != nil {
+			if err != nil && disconnected.bool() {
-				if yes, ok := disconnected.Load().(bool); ok && yes {
+				err = errors.New("disconnect events received")
-					err = errors.New("disconnect events received")
-				}
 			}
 		}()
 		// benchmark loop

swarm/network/stream/intervals_test.go

@@ -140,10 +140,8 @@ func testIntervals(t *testing.T, live bool, history *Range, skipCheck bool) {
 
 		disconnected := watchDisconnections(ctx, sim)
 		defer func() {
-			if err != nil {
+			if err != nil && disconnected.bool() {
-				if yes, ok := disconnected.Load().(bool); ok && yes {
+				err = errors.New("disconnect events received")
-					err = errors.New("disconnect events received")
-				}
 			}
 		}()
 

swarm/network/stream/snapshot_sync_test.go

@@ -195,10 +195,8 @@ func runSim(conf *synctestConfig, ctx context.Context, sim *simulation.Simulatio
 	return sim.Run(ctx, func(ctx context.Context, sim *simulation.Simulation) (err error) {
 		disconnected := watchDisconnections(ctx, sim)
 		defer func() {
-			if err != nil {
+			if err != nil && disconnected.bool() {
-				if yes, ok := disconnected.Load().(bool); ok && yes {
+				err = errors.New("disconnect events received")
-					err = errors.New("disconnect events received")
-				}
 			}
 		}()
 

swarm/network/stream/syncer_test.go

@@ -141,10 +141,8 @@ func testSyncBetweenNodes(t *testing.T, nodes, chunkCount int, skipCheck bool, p
 
 		disconnected := watchDisconnections(ctx, sim)
 		defer func() {
-			if err != nil {
+			if err != nil && disconnected.bool() {
-				if yes, ok := disconnected.Load().(bool); ok && yes {
+				err = errors.New("disconnect events received")
-					err = errors.New("disconnect events received")
-				}
 			}
 		}()