Merge pull request #2263 from cheng762/fixp2p

fix p2p close

eth/backend.go

@@ -648,6 +648,7 @@ func (s *Ethereum) Start() error {
 func (s *Ethereum) Stop() error {
 	s.ethDialCandidates.Close()
 	s.snapDialCandidates.Close()
+	s.p2pServer.CloseConsensusDial()
 	s.handler.Stop()
 
 	// Then stop everything else.

p2p/consensus_dialed.go

@@ -111,7 +111,7 @@ func (tasks *dialedTasks) size() int {
 }
 
 // clear queue
-/*func (tasks *dialedTasks) clear() bool {
+func (tasks *dialedTasks) clear() bool {
 	if tasks.isEmpty() {
 		log.Info("queue is empty!")
 		return false
@@ -121,7 +121,7 @@ func (tasks *dialedTasks) size() int {
 	}
 	tasks.queue = nil
 	return true
-}*/
+}
 
 // whether the queue is empty
 func (tasks *dialedTasks) isEmpty() bool {

p2p/dial.go

@@ -108,6 +108,7 @@ type dialScheduler struct {
 
 	addconsensus    chan *enode.Node
 	removeconsensus chan *enode.Node
+	clearConsensus  chan struct{}
 
 	// Everything below here belongs to loop and
 	// should only be accessed by code on the loop goroutine.
@@ -185,6 +186,7 @@ func newDialScheduler(config dialConfig, it enode.Iterator, setupFunc dialSetupF
 		remPeerCh:              make(chan *conn),
 		addconsensus:           make(chan *enode.Node),
 		removeconsensus:        make(chan *enode.Node),
+		clearConsensus:         make(chan struct{}),
 		updateConsensusPeersCh: make(chan int),
 		consensusPool:          NewDialedTasks(config.MaxConsensusPeers*2, nil),
 	}
@@ -239,6 +241,13 @@ func (d *dialScheduler) removeConsensus(n *enode.Node) {
 	}
 }
 
+func (d *dialScheduler) closeConsensusDial() {
+	select {
+	case d.clearConsensus <- struct{}{}:
+	case <-d.ctx.Done():
+	}
+}
+
 func (d *dialScheduler) removeConsensusFromQueue(n *enode.Node) {
 	d.history.remove(string(n.ID().Bytes()))
 }
@@ -346,6 +355,8 @@ loop:
 			d.consensusPool.AddTask(newDialTask(node, dynDialedConn|consensusDialedConn))
 		case node := <-d.removeconsensus:
 			d.consensusPool.RemoveTask(node.ID())
+		case <-d.clearConsensus:
+			d.consensusPool.clear()
 		case num := <-d.updateConsensusPeersCh:
 			d.log.Debug("update added consensus peers num", "num", num)
 			d.consensusPeers = num

p2p/server.go

@@ -399,6 +399,11 @@ func (srv *Server) RemoveConsensusPeer(node *enode.Node) {
 	}
 }
 
+func (srv *Server) CloseConsensusDial() {
+	srv.dialsched.closeConsensusDial()
+	srv.log.Info("Close consensus Dial")
+}
+
 // AddTrustedPeer adds the given node to a reserved whitelist which allows the
 // node to always connect, even if the slot are full.
 func (srv *Server) AddTrustedPeer(node *enode.Node) {