diff --git a/p2p/dial.go b/p2p/dial.go index a33fc5dc5..336ad35e0 100644 --- a/p2p/dial.go +++ b/p2p/dial.go @@ -106,7 +106,7 @@ type dialScheduler struct { // Everything below here belongs to loop and // should only be accessed by code on the loop goroutine. dialing map[enode.ID]*dialTask // active tasks - peers map[enode.ID]connFlag // all connected peers + peers map[enode.ID]struct{} // all connected peers dialPeers int // current number of dialed peers // The static map tracks all static dial tasks. The subset of usable static dial tasks @@ -165,7 +165,7 @@ func newDialScheduler(config dialConfig, it enode.Iterator, setupFunc dialSetupF setupFunc: setupFunc, dialing: make(map[enode.ID]*dialTask), static: make(map[enode.ID]*dialTask), - peers: make(map[enode.ID]connFlag), + peers: make(map[enode.ID]struct{}), doneCh: make(chan *dialTask), nodesIn: make(chan *enode.Node), addStaticCh: make(chan *enode.Node), @@ -258,7 +258,7 @@ loop: d.dialPeers++ } id := c.node.ID() - d.peers[id] = c.flags + d.peers[id] = struct{}{} // Remove from static pool because the node is now connected. task := d.static[id] if task != nil && task.staticPoolIndex >= 0 { diff --git a/p2p/enode/iter.go b/p2p/enode/iter.go index 664964f53..b8ab4a758 100644 --- a/p2p/enode/iter.go +++ b/p2p/enode/iter.go @@ -203,27 +203,34 @@ func (m *FairMix) Close() { func (m *FairMix) Next() bool { m.cur = nil - var timeout <-chan time.Time - if m.timeout >= 0 { - timer := time.NewTimer(m.timeout) - timeout = timer.C - defer timer.Stop() - } for { source := m.pickSource() if source == nil { return m.nextFromAny() } + + var timeout <-chan time.Time + if source.timeout >= 0 { + timer := time.NewTimer(source.timeout) + timeout = timer.C + defer timer.Stop() + } + select { case n, ok := <-source.next: if ok { - m.cur = n + // Here, the timeout is reset to the configured value + // because the source delivered a node. source.timeout = m.timeout + m.cur = n return true } // This source has ended. m.deleteSource(source) case <-timeout: + // The selected source did not deliver a node within the timeout, so the + // timeout duration is halved for next time. This is supposed to improve + // latency with stuck sources. source.timeout /= 2 return m.nextFromAny() } diff --git a/p2p/enode/iter_test.go b/p2p/enode/iter_test.go index 6009661f3..5014346af 100644 --- a/p2p/enode/iter_test.go +++ b/p2p/enode/iter_test.go @@ -268,7 +268,7 @@ func (s *genIter) Node() *Node { } func (s *genIter) Close() { - s.index = ^uint32(0) + atomic.StoreUint32(&s.index, ^uint32(0)) } func testNode(id, seq uint64) *Node {