|
|
|
@ -25,7 +25,6 @@ import ( |
|
|
|
|
proto_discovery "github.com/harmony-one/harmony/api/proto/discovery" |
|
|
|
|
"github.com/harmony-one/harmony/api/proto/message" |
|
|
|
|
proto_node "github.com/harmony-one/harmony/api/proto/node" |
|
|
|
|
"github.com/harmony-one/harmony/api/service" |
|
|
|
|
"github.com/harmony-one/harmony/contracts/structs" |
|
|
|
|
"github.com/harmony-one/harmony/core" |
|
|
|
|
"github.com/harmony-one/harmony/core/types" |
|
|
|
@ -201,8 +200,6 @@ func (node *Node) messageHandler(content []byte, sender libp2p_peer.ID) { |
|
|
|
|
} |
|
|
|
|
case proto_node.PING: |
|
|
|
|
node.pingMessageHandler(msgPayload, sender) |
|
|
|
|
case proto_node.PONG: |
|
|
|
|
node.pongMessageHandler(msgPayload) |
|
|
|
|
case proto_node.ShardState: |
|
|
|
|
if err := node.epochShardStateMessageHandler(msgPayload); err != nil { |
|
|
|
|
ctxerror.Log15(utils.GetLogger().Warn, err) |
|
|
|
@ -713,15 +710,6 @@ func getGenesisNodeByConsensusKey(key types.BlsPublicKey) *genesisNode { |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
func (node *Node) pingMessageHandler(msgPayload []byte, sender libp2p_peer.ID) int { |
|
|
|
|
senderStr := string(sender) |
|
|
|
|
if senderStr != "" { |
|
|
|
|
_, ok := node.duplicatedPing.LoadOrStore(senderStr, true) |
|
|
|
|
if ok { |
|
|
|
|
// duplicated ping message return
|
|
|
|
|
return 0 |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
ping, err := proto_discovery.GetPingMessage(msgPayload) |
|
|
|
|
if err != nil { |
|
|
|
|
utils.Logger().Error(). |
|
|
|
@ -746,15 +734,22 @@ func (node *Node) pingMessageHandler(msgPayload []byte, sender libp2p_peer.ID) i |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
var k types.BlsPublicKey |
|
|
|
|
if err := k.FromLibBLSPublicKey(peer.ConsensusPubKey); err != nil { |
|
|
|
|
err = ctxerror.New("cannot convert BLS public key").WithCause(err) |
|
|
|
|
ctxerror.Log15(utils.GetLogger().Warn, err) |
|
|
|
|
} |
|
|
|
|
utils.Logger().Info(). |
|
|
|
|
Str("Peer Version", ping.NodeVer). |
|
|
|
|
Interface("PeerID", peer). |
|
|
|
|
Msg("received ping message") |
|
|
|
|
Str("Version", ping.NodeVer). |
|
|
|
|
Str("BlsKey", peer.ConsensusPubKey.SerializeToHexStr()). |
|
|
|
|
Str("IP", peer.IP). |
|
|
|
|
Str("Port", peer.Port). |
|
|
|
|
Interface("PeerID", peer.PeerID). |
|
|
|
|
Msg("[PING] PeerInfo") |
|
|
|
|
|
|
|
|
|
senderStr := string(sender) |
|
|
|
|
if senderStr != "" { |
|
|
|
|
_, ok := node.duplicatedPing.LoadOrStore(senderStr, true) |
|
|
|
|
if ok { |
|
|
|
|
// duplicated ping message return
|
|
|
|
|
return 0 |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
// add to incoming peer list
|
|
|
|
|
//node.host.AddIncomingPeer(*peer)
|
|
|
|
@ -776,140 +771,27 @@ func (node *Node) pingMessageHandler(msgPayload []byte, sender libp2p_peer.ID) i |
|
|
|
|
return 1 |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
// SendPongMessage is the a goroutine to periodcally send pong message to all peers
|
|
|
|
|
func (node *Node) SendPongMessage() { |
|
|
|
|
tick := time.NewTicker(2 * time.Second) |
|
|
|
|
tick2 := time.NewTicker(120 * time.Second) |
|
|
|
|
|
|
|
|
|
numPeers := node.numPeers |
|
|
|
|
sentMessage := false |
|
|
|
|
firstTime := true |
|
|
|
|
|
|
|
|
|
// Send Pong Message only when there is change on the number of peers
|
|
|
|
|
// bootstrapConsensus is the a goroutine to check number of peers and start the consensus
|
|
|
|
|
func (node *Node) bootstrapConsensus() { |
|
|
|
|
tick := time.NewTicker(5 * time.Second) |
|
|
|
|
for { |
|
|
|
|
select { |
|
|
|
|
case <-tick.C: |
|
|
|
|
peers := node.Consensus.GetValidatorPeers() |
|
|
|
|
numPeersNow := node.numPeers |
|
|
|
|
|
|
|
|
|
// no peers, wait for another tick
|
|
|
|
|
if numPeersNow == 0 { |
|
|
|
|
utils.Logger().Info(). |
|
|
|
|
Int("numPeers", numPeers). |
|
|
|
|
Int("numPeersNow", numPeersNow). |
|
|
|
|
Msg("[PONG] No peers, continue") |
|
|
|
|
continue |
|
|
|
|
} |
|
|
|
|
// new peers added
|
|
|
|
|
if numPeersNow != numPeers { |
|
|
|
|
utils.Logger().Info(). |
|
|
|
|
Int("numPeers", numPeers). |
|
|
|
|
Int("numPeersNow", numPeersNow). |
|
|
|
|
Msg("[PONG] Different number of peers") |
|
|
|
|
sentMessage = false |
|
|
|
|
} else { |
|
|
|
|
// stable number of peers, sent the pong message
|
|
|
|
|
// also make sure number of peers is greater than the minimal required number
|
|
|
|
|
if !sentMessage && numPeersNow >= node.Consensus.MinPeers { |
|
|
|
|
pong := proto_discovery.NewPongMessage(peers, node.Consensus.PublicKeys, node.Consensus.GetLeaderPubKey(), node.Consensus.ShardID) |
|
|
|
|
buffer := pong.ConstructPongMessage() |
|
|
|
|
err := node.host.SendMessageToGroups([]p2p.GroupID{node.NodeConfig.GetShardGroupID()}, host.ConstructP2pMessage(byte(0), buffer)) |
|
|
|
|
if err != nil { |
|
|
|
|
utils.Logger().Error(). |
|
|
|
|
Str("group", string(node.NodeConfig.GetShardGroupID())). |
|
|
|
|
Msg("[PONG] Failed to send pong message") |
|
|
|
|
continue |
|
|
|
|
} else { |
|
|
|
|
utils.Logger().Info(). |
|
|
|
|
Str("group", string(node.NodeConfig.GetShardGroupID())). |
|
|
|
|
Int("# nodes", numPeersNow). |
|
|
|
|
Msg("[PONG] Sent pong message to") |
|
|
|
|
} |
|
|
|
|
sentMessage = true |
|
|
|
|
|
|
|
|
|
// only need to notify consensus leader once to start the consensus
|
|
|
|
|
if firstTime { |
|
|
|
|
// Leader stops sending ping message
|
|
|
|
|
node.serviceManager.TakeAction(&service.Action{Action: service.Stop, ServiceType: service.PeerDiscovery}) |
|
|
|
|
utils.Logger().Info().Msg("[PONG] StartConsensus") |
|
|
|
|
node.startConsensus <- struct{}{} |
|
|
|
|
firstTime = false |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
numPeers = numPeersNow |
|
|
|
|
case <-tick2.C: |
|
|
|
|
// send pong message regularly to make sure new node received all the public keys
|
|
|
|
|
// also nodes offline/online will receive the public keys
|
|
|
|
|
peers := node.Consensus.GetValidatorPeers() |
|
|
|
|
pong := proto_discovery.NewPongMessage(peers, node.Consensus.PublicKeys, node.Consensus.GetLeaderPubKey(), node.Consensus.ShardID) |
|
|
|
|
buffer := pong.ConstructPongMessage() |
|
|
|
|
err := node.host.SendMessageToGroups([]p2p.GroupID{node.NodeConfig.GetShardGroupID()}, host.ConstructP2pMessage(byte(0), buffer)) |
|
|
|
|
if err != nil { |
|
|
|
|
utils.Logger().Error(). |
|
|
|
|
Str("group", string(node.NodeConfig.GetShardGroupID())). |
|
|
|
|
Msg("[PONG] Failed to send regular pong message") |
|
|
|
|
Msg("No peers, continue") |
|
|
|
|
continue |
|
|
|
|
} else { |
|
|
|
|
utils.Logger().Info(). |
|
|
|
|
Str("group", string(node.NodeConfig.GetShardGroupID())). |
|
|
|
|
Int("# nodes", len(peers)). |
|
|
|
|
Msg("[PONG] Sent regular pong message to") |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
func (node *Node) pongMessageHandler(msgPayload []byte) int { |
|
|
|
|
pong, err := proto_discovery.GetPongMessage(msgPayload) |
|
|
|
|
if err != nil { |
|
|
|
|
utils.Logger().Error(). |
|
|
|
|
Err(err). |
|
|
|
|
Msg("Can't get Pong Message") |
|
|
|
|
return -1 |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
if pong.ShardID != node.Consensus.ShardID { |
|
|
|
|
utils.Logger().Error(). |
|
|
|
|
Uint32("receivedShardID", pong.ShardID). |
|
|
|
|
Uint32("expectedShardID", node.Consensus.ShardID). |
|
|
|
|
Msg("Received Pong message for the wrong shard") |
|
|
|
|
return 0 |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
peers := make([]*p2p.Peer, 0) |
|
|
|
|
|
|
|
|
|
for _, p := range pong.Peers { |
|
|
|
|
peer := new(p2p.Peer) |
|
|
|
|
peer.IP = p.IP |
|
|
|
|
peer.Port = p.Port |
|
|
|
|
peer.PeerID = p.PeerID |
|
|
|
|
|
|
|
|
|
peer.ConsensusPubKey = &bls.PublicKey{} |
|
|
|
|
if len(p.PubKey) != 0 { // TODO: add the check in bls library
|
|
|
|
|
err = peer.ConsensusPubKey.Deserialize(p.PubKey[:]) |
|
|
|
|
if err != nil { |
|
|
|
|
utils.Logger().Error(). |
|
|
|
|
Err(err). |
|
|
|
|
Msg("Deserialize ConsensusPubKey Failed") |
|
|
|
|
continue |
|
|
|
|
if numPeersNow >= node.Consensus.MinPeers { |
|
|
|
|
utils.Logger().Info().Msg("[bootstrap] StartConsensus") |
|
|
|
|
node.startConsensus <- struct{}{} |
|
|
|
|
return |
|
|
|
|
} |
|
|
|
|
} |
|
|
|
|
peers = append(peers, peer) |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
if len(peers) > 0 { |
|
|
|
|
node.AddPeers(peers) |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
// Stop discovery service after received pong message
|
|
|
|
|
data := make(map[string]interface{}) |
|
|
|
|
data["peer"] = p2p.GroupAction{Name: node.NodeConfig.GetShardGroupID(), Action: p2p.ActionPause} |
|
|
|
|
|
|
|
|
|
node.serviceManager.TakeAction(&service.Action{Action: service.Notify, ServiceType: service.PeerDiscovery, Params: data}) |
|
|
|
|
|
|
|
|
|
// TODO: remove this after fully migrating to beacon chain-based committee membership
|
|
|
|
|
return 0 |
|
|
|
|
} |
|
|
|
|
|
|
|
|
|
func (node *Node) epochShardStateMessageHandler(msgPayload []byte) error { |
|
|
|
|