[pong] actively sending pong message

Signed-off-by: Leo Chen <leo@harmony.one>

consensus/consensus.go

@@ -404,13 +404,11 @@ func (consensus *Consensus) AddPeers(peers []*p2p.Peer) int {
 	count := 0
 
 	for _, peer := range peers {
-		_, ok := consensus.validators.Load(peer.GetAddressHex())
+		_, ok := consensus.validators.LoadOrStore(peer.GetAddressHex(), *peer)
 		if !ok {
-			consensus.validators.Store(peer.GetAddressHex(), *peer)
 			consensus.pubKeyLock.Lock()
 			consensus.PublicKeys = append(consensus.PublicKeys, peer.ConsensusPubKey)
 			consensus.pubKeyLock.Unlock()
-			//			utils.GetLogInstance().Debug("[SYNC]", "new peer added", peer)
 		}
 		count++
 	}

node/node.go

@@ -321,7 +321,9 @@ func (node *Node) AddPeers(peers []*p2p.Peer) int {
 		}
 	}
 
-	if count > 0 {
+	// Only leader needs to add the peer info into consensus
+	// Validators will receive the updated peer info from Leader via pong message
+	if count > 0 && node.NodeConfig.IsLeader() {
 		node.Consensus.AddPeers(peers)
 		// TODO: make peers into a context object shared by consensus and drand
 		node.DRand.AddPeers(peers)

node/node_handler.go

@@ -385,10 +385,7 @@ func (node *Node) pingMessageHandler(msgPayload []byte, sender string) int {
 	if ping.Node.Role == proto_node.ClientRole {
 		utils.GetLogInstance().Info("Add Client Peer to Node", "Address", node.Consensus.GetSelfAddress(), "Client", peer)
 		node.ClientPeer = peer
-		return 0
+	} else {
-	}
-
-	if node.NodeConfig.IsLeader() {
 		utils.GetLogInstance().Info("Add Peer to Node", "Address", node.Consensus.GetSelfAddress(), "Pear", peer)
 		node.AddPeers([]*p2p.Peer{peer})
 	}
@@ -398,10 +395,11 @@ func (node *Node) pingMessageHandler(msgPayload []byte, sender string) int {
 
 // SendPongMessage is the a goroutine to periodcally send pong message to all peers
 func (node *Node) SendPongMessage() {
-	tick := time.NewTicker(3 * time.Second)
+	tick := time.NewTicker(2 * time.Second)
 	numPeers := len(node.Consensus.GetValidatorPeers())
 	numPubKeys := len(node.Consensus.PublicKeys)
 	sentMessage := false
+	firstTime := true
 
 	// Send Pong Message only when there is change on the number of peers
 	for {
@@ -417,6 +415,7 @@ func (node *Node) SendPongMessage() {
 			}
 			// new peers added
 			if numPubKeysNow != numPubKeys || numPeersNow != numPeers {
+				utils.GetLogInstance().Info("[PONG] different number of peers", "numPeers", numPeers, "numPeersNow", numPeersNow)
 				sentMessage = false
 			} else {
 				// stable number of peers/pubkeys, sent the pong message
@@ -436,7 +435,11 @@ func (node *Node) SendPongMessage() {
 					node.serviceManager.TakeAction(&service.Action{Action: service.Stop, ServiceType: service.PeerDiscovery})
 					// wait a bit until all validators received pong message
 					time.Sleep(200 * time.Millisecond)
+					// only need to notify consensus leader once to start the consensus
+					if firstTime {
 						node.startConsensus <- struct{}{}
+						firstTime = false
+					}
 				}
 			}
 			numPeers = numPeersNow