p2p/enode: per-source timeout in FairMix

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 {

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()
 		}

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 {