swarm: fix network/stream data races (#19051)

* swarm/network/stream: newStreamerTester cleanup only if err is nil

* swarm/network/stream: raise newStreamerTester waitForPeers timeout

* swarm/network/stream: fix data races in GetPeerSubscriptions

* swarm/storage: prevent data race on LDBStore.batchesC

https://github.com/ethersphere/go-ethereum/issues/1198#issuecomment-461775049

* swarm/network/stream: fix TestGetSubscriptionsRPC data race

https://github.com/ethersphere/go-ethereum/issues/1198#issuecomment-461768477

* swarm/network/stream: correctly use Simulation.Run callback

https://github.com/ethersphere/go-ethereum/issues/1198#issuecomment-461783804

* swarm/network: protect addrCountC in Kademlia.AddrCountC function

https://github.com/ethersphere/go-ethereum/issues/1198#issuecomment-462273444

* p2p/simulations: fix a deadlock calling getRandomNode with lock

https://github.com/ethersphere/go-ethereum/issues/1198#issuecomment-462317407

* swarm/network/stream: terminate disconnect goruotines in tests

* swarm/network/stream: reduce memory consumption when testing data races

* swarm/network/stream: add watchDisconnections helper function

* swarm/network/stream: add concurrent counter for tests

* swarm/network/stream: rename race/norace test files and use const

* swarm/network/stream: remove watchSim and its panic

* swarm/network/stream: pass context in watchDisconnections

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

* swarm/storage: fix LDBStore.batchesC data race by not closing it

(cherry picked from commit 3fd6db2bf6)

swarm/network/stream/stream.go

@@ -944,16 +944,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
 	}