eth: add disc/v4 and disc/v5 sources through a fairmix

Signed-off-by: Csaba Kiraly <[email protected]>

Author: cskiraly

Diff for: eth/backend.go

@@ -70,6 +70,19 @@ const (
 	// in the next rounds, giving it overall more time but a proportionally smaller share.
 	// We expect a normal source to produce ~10 candidates per second.
 	discmixTimeout = 100 * time.Millisecond
+
+	// discoveryParallelLookups is the number of parallel lookups to perform by DHT
+	// sources (disc/v4 and disc/v5). We set this number large enough to be able to
+	// feed the dial queue with enough peers. Since the whole discovery process is triggered
+	// only when dial candidates are needed, we can keep this number high without worrying
+	// about overloading the DHT.
+	discoveryParallelLookups = 3
+
+	// discoveryPrefetchBuffer is the number of peers to pre-fetch from a discovery
+	// source. It is useful to avoid the negative effects of potential longer timeouts
+	// in the discovery, keeping dial progress while waiting for the next batch of
+	// candidates.
+	discoveryPrefetchBuffer = 32
 )
 
 // Config contains the configuration options of the ETH protocol.
@@ -502,26 +515,30 @@ func (s *Ethereum) setupDiscovery() error {
 
 	// Add DHT nodes from discv4.
 	if s.p2pServer.DiscoveryV4() != nil {
-		for i := 0; i < 1; i++ {
+		fairmix := enode.NewFairMix(0)
+		for i := 0; i < discoveryParallelLookups; i++ {
 			asyncFilter := s.p2pServer.DiscoveryV4().RequestENR
 			filter := eth.NewNodeFilter(s.blockchain)
 			iter := enode.AsyncFilter(
 				enode.NewBufferIter(
 					s.p2pServer.DiscoveryV4().RandomNodes(), 0),
-				asyncFilter, 128)
+				asyncFilter, discoveryPrefetchBuffer)
 			iter = enode.Filter(iter, filter)
-			s.discmix.AddSource(iter, fmt.Sprintf("DiscoveryV4-%d", i))
+			fairmix.AddSource(iter, fmt.Sprintf("DiscoveryV4-%d", i))
 		}
+		s.discmix.AddSource(fairmix, "DiscoveryV4")
 	}
 
 	// Add DHT nodes from discv5.
 	if s.p2pServer.DiscoveryV5() != nil {
-		for i := 0; i < 1; i++ {
-			iter := enode.NewBufferIter(s.p2pServer.DiscoveryV5().RandomNodes(), 0)
+		fairmix := enode.NewFairMix(0)
+		for i := 0; i < discoveryParallelLookups; i++ {
+			iter := enode.NewBufferIter(s.p2pServer.DiscoveryV5().RandomNodes(), discoveryPrefetchBuffer)
 			filter := eth.NewNodeFilter(s.blockchain)
 			filterIter := enode.Filter(iter, filter)
-			s.discmix.AddSource(filterIter, fmt.Sprintf("DiscoveryV5-%d", i))
+			fairmix.AddSource(filterIter, fmt.Sprintf("DiscoveryV5-%d", i))
 		}
+		s.discmix.AddSource(fairmix, "DiscoveryV5")
 	}
 
 	return nil