diff --git a/eth/backend.go b/eth/backend.go index ad83e6585b..a79b047365 100644 --- a/eth/backend.go +++ b/eth/backend.go @@ -491,7 +491,7 @@ func (s *Ethereum) setupDiscovery() error { if s.p2pServer.DiscoveryV4() != nil { asyncFilter := s.p2pServer.DiscoveryV4().RequestENR filter := eth.NewNodeFilter(s.blockchain) - iter := enode.AsyncFilter(s.p2pServer.DiscoveryV4().RandomNodes(), asyncFilter) + iter := enode.AsyncFilter(s.p2pServer.DiscoveryV4().RandomNodes(), asyncFilter, 16) iter = enode.Filter(iter, filter) s.discmix.AddSource(iter) } diff --git a/p2p/enode/iter.go b/p2p/enode/iter.go index 0c252a33ec..86f204c452 100644 --- a/p2p/enode/iter.go +++ b/p2p/enode/iter.go @@ -154,25 +154,52 @@ func (f *filterIter) Next() bool { // AsyncFilter wraps an iterator such that Next only returns nodes for which // the 'check' function returns a (possibly modified) node. -func AsyncFilter(it Iterator, check func(*Node) (*Node, error)) Iterator { - return &AsyncFilterIter{it, nil, check} +func AsyncFilter(it Iterator, check func(*Node) (*Node, error), workers int) Iterator { + f := &AsyncFilterIter{it, nil, check, make(chan *Node), sync.WaitGroup{}} + + taskCh := make(chan *Node) + + worker := func() { + for task := range taskCh { + if task == nil { + break + } + nn, err := f.check(task) + if err == nil { + f.passed <- nn + } + } + f.wg.Done() + } + + for range workers { + f.wg.Add(1) + go worker() + } + + go func() { + for f.it.Next() { + taskCh <- f.it.Node() + } + close(taskCh) + f.wg.Wait() + close(f.passed) + }() + + return f } type AsyncFilterIter struct { it Iterator buffer *Node check func(*Node) (*Node, error) + passed chan *Node + wg sync.WaitGroup } func (f *AsyncFilterIter) Next() bool { - for f.it.Next() { - nn, err := f.check(f.it.Node()) - if err == nil { - f.buffer = nn - return true - } - } - return false + f.buffer = <-f.passed + return f.buffer != nil } func (f *AsyncFilterIter) Node() *Node { @@ -181,6 +208,7 @@ func (f *AsyncFilterIter) Node() *Node { func (f *AsyncFilterIter) Close() { f.it.Close() + f.wg.Wait() } // FairMix aggregates multiple node iterators. The mixer itself is an iterator which ends