forks/go-ethereum
1
0
Fork 1
mirror of https://github.com/ethereum/go-ethereum.git synced 2026-07-24 05:36:46 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 1

eth: add multiple parallel lookup in discovery

Browse source 
Signed-off-by: Csaba Kiraly <csaba.kiraly@gmail.com>
This commit is contained in:
Csaba Kiraly 2025-04-29 11:03:02 +02:00
parent d14d4d2af0
commit 7bebcd1043
No known key found for this signature in database
GPG key ID: 0FE274EE8C95166E
1 changed files with 24 additions and 9 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

33
eth/backend.go
View file

@ -73,6 +73,13 @@ const (
// We expect a normal source to produce ~10 candidates per second. // We expect a normal source to produce ~10 candidates per second.
discmixTimeout = 100 * time.Millisecond 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 // 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 // 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 // in the discovery, keeping dial progress while waiting for the next batch of
@ -532,7 +539,6 @@ func (s *Ethereum) setupDiscovery() error {
// Add DHT nodes from discv4. // Add DHT nodes from discv4.
if s.p2pServer.DiscoveryV4() != nil { if s.p2pServer.DiscoveryV4() != nil {
iter := s.p2pServer.DiscoveryV4().RandomNodes()
resolverFunc := func(ctx context.Context, enr *enode.Node) *enode.Node { resolverFunc := func(ctx context.Context, enr *enode.Node) *enode.Node {
// RequestENR does not yet support context. It will simply time out. // RequestENR does not yet support context. It will simply time out.
// If the ENR can't be resolved, RequestENR will return nil. We don't // If the ENR can't be resolved, RequestENR will return nil. We don't
@ -540,18 +546,27 @@ func (s *Ethereum) setupDiscovery() error {
nn, _ := s.p2pServer.DiscoveryV4().RequestENR(enr) nn, _ := s.p2pServer.DiscoveryV4().RequestENR(enr)
return nn return nn
} }
iter = enode.AsyncFilter(iter, resolverFunc, maxParallelENRRequests) fairmix := enode.NewFairMix(0)
iter = enode.Filter(iter, eth.NewNodeFilter(s.blockchain)) for i := 0; i < discoveryParallelLookups; i++ {
iter = enode.NewBufferIter(iter, discoveryPrefetchBuffer) iter := s.p2pServer.DiscoveryV4().RandomNodes()
s.discmix.AddSource(iter) iter = enode.AsyncFilter(iter, resolverFunc, maxParallelENRRequests)
iter = enode.Filter(iter, eth.NewNodeFilter(s.blockchain))
iter = enode.NewBufferIter(iter, discoveryPrefetchBuffer)
fairmix.AddSource(iter)
}
s.discmix.AddSource(fairmix)
} }
// Add DHT nodes from discv5. // Add DHT nodes from discv5.
if s.p2pServer.DiscoveryV5() != nil { if s.p2pServer.DiscoveryV5() != nil {
filter := eth.NewNodeFilter(s.blockchain) fairmix := enode.NewFairMix(0)
iter := enode.Filter(s.p2pServer.DiscoveryV5().RandomNodes(), filter) for i := 0; i < discoveryParallelLookups; i++ {
iter = enode.NewBufferIter(iter, discoveryPrefetchBuffer) iter := s.p2pServer.DiscoveryV5().RandomNodes()
s.discmix.AddSource(iter) iter = enode.Filter(iter, eth.NewNodeFilter(s.blockchain))
iter = enode.NewBufferIter(iter, discoveryPrefetchBuffer)
fairmix.AddSource(iter)
}
s.discmix.AddSource(fairmix)
} }
return nil return nil