Merge pull request #17 from preminem/xiaohe/v0.32.1

P2P: Remove deleted peers after commit

consensus/state.go

@@ -43,6 +43,10 @@ var (
 	msgQueueSize = 1000
 )
 
+var (
+	Switch *p2p.Switch
+)
+
 // msgs from the reactor which may update the state
 type msgInfo struct {
 	Msg    ConsensusMessage `json:"msg"`
@@ -1267,6 +1271,9 @@ func (cs *ConsensusState) tryFinalizeCommit(height int64) {
 
 	//	go
 	cs.finalizeCommit(height)
+
+	// check if peer needs to be removed
+	Switch.CheckPeers()
 }
 
 // Increment height and goto cstypes.RoundStepNewHeight

node/node.go

@@ -676,6 +676,8 @@ func NewNode(config *cfg.Config,
 		consensusReactor, evidenceReactor, nodeInfo, nodeKey, p2pLogger,
 	)
 
+	consensus.Switch = sw
+
 	err = sw.AddPersistentPeers(splitAndTrimEmpty(config.P2P.PersistentPeers, ",", " "))
 	if err != nil {
 		return nil, errors.Wrap(err, "could not add peers from persistent_peers field")

p2p/switch.go

@@ -775,3 +775,27 @@ func (sw *Switch) addPeer(p Peer, inBound bool) error {
 
 	return nil
 }
+
+// check if peer needs to be removed
+func (sw *Switch) CheckPeers() {
+	for _, p := range sw.peers.List() {
+		errc := make(chan error, len(sw.peerFilters))
+
+		for _, f := range sw.peerFilters {
+			go func(f PeerFilterFunc, p Peer, errc chan<- error) {
+				errc <- f(sw.peers, p)
+			}(f, p, errc)
+		}
+
+		for i := 0; i < cap(errc); i++ {
+			select {
+			case err := <-errc:
+				if err != nil {
+					p.CloseConn()
+					sw.StopPeerGracefully(p)
+					sw.addrBook.RemoveAddress(p.SocketAddr())
+				}
+			}
+		}
+	}
+}