1461 lines
43 KiB
Go
1461 lines
43 KiB
Go
package consensus
|
|
|
|
import (
|
|
"bytes"
|
|
"context"
|
|
"fmt"
|
|
"reflect"
|
|
"sync"
|
|
"time"
|
|
|
|
"github.com/pkg/errors"
|
|
|
|
wire "github.com/tendermint/go-wire"
|
|
cmn "github.com/tendermint/tmlibs/common"
|
|
"github.com/tendermint/tmlibs/log"
|
|
|
|
cstypes "github.com/tendermint/tendermint/consensus/types"
|
|
"github.com/tendermint/tendermint/p2p"
|
|
sm "github.com/tendermint/tendermint/state"
|
|
"github.com/tendermint/tendermint/types"
|
|
)
|
|
|
|
const (
|
|
StateChannel = byte(0x20)
|
|
DataChannel = byte(0x21)
|
|
VoteChannel = byte(0x22)
|
|
VoteSetBitsChannel = byte(0x23)
|
|
|
|
maxConsensusMessageSize = 1048576 // 1MB; NOTE/TODO: keep in sync with types.PartSet sizes.
|
|
|
|
blocksToContributeToBecomeGoodPeer = 10000
|
|
)
|
|
|
|
//-----------------------------------------------------------------------------
|
|
|
|
// ConsensusReactor defines a reactor for the consensus service.
|
|
type ConsensusReactor struct {
|
|
p2p.BaseReactor // BaseService + p2p.Switch
|
|
|
|
conS *ConsensusState
|
|
|
|
mtx sync.RWMutex
|
|
fastSync bool
|
|
eventBus *types.EventBus
|
|
}
|
|
|
|
// NewConsensusReactor returns a new ConsensusReactor with the given consensusState.
|
|
func NewConsensusReactor(consensusState *ConsensusState, fastSync bool) *ConsensusReactor {
|
|
conR := &ConsensusReactor{
|
|
conS: consensusState,
|
|
fastSync: fastSync,
|
|
}
|
|
conR.BaseReactor = *p2p.NewBaseReactor("ConsensusReactor", conR)
|
|
return conR
|
|
}
|
|
|
|
// OnStart implements BaseService.
|
|
func (conR *ConsensusReactor) OnStart() error {
|
|
conR.Logger.Info("ConsensusReactor ", "fastSync", conR.FastSync())
|
|
if err := conR.BaseReactor.OnStart(); err != nil {
|
|
return err
|
|
}
|
|
|
|
err := conR.startBroadcastRoutine()
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
if !conR.FastSync() {
|
|
err := conR.conS.Start()
|
|
if err != nil {
|
|
return err
|
|
}
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
// OnStop implements BaseService
|
|
func (conR *ConsensusReactor) OnStop() {
|
|
conR.BaseReactor.OnStop()
|
|
conR.conS.Stop()
|
|
}
|
|
|
|
// SwitchToConsensus switches from fast_sync mode to consensus mode.
|
|
// It resets the state, turns off fast_sync, and starts the consensus state-machine
|
|
func (conR *ConsensusReactor) SwitchToConsensus(state sm.State, blocksSynced int) {
|
|
conR.Logger.Info("SwitchToConsensus")
|
|
conR.conS.reconstructLastCommit(state)
|
|
// NOTE: The line below causes broadcastNewRoundStepRoutine() to
|
|
// broadcast a NewRoundStepMessage.
|
|
conR.conS.updateToState(state)
|
|
|
|
conR.mtx.Lock()
|
|
conR.fastSync = false
|
|
conR.mtx.Unlock()
|
|
|
|
if blocksSynced > 0 {
|
|
// dont bother with the WAL if we fast synced
|
|
conR.conS.doWALCatchup = false
|
|
}
|
|
err := conR.conS.Start()
|
|
if err != nil {
|
|
conR.Logger.Error("Error starting conS", "err", err)
|
|
}
|
|
}
|
|
|
|
// GetChannels implements Reactor
|
|
func (conR *ConsensusReactor) GetChannels() []*p2p.ChannelDescriptor {
|
|
// TODO optimize
|
|
return []*p2p.ChannelDescriptor{
|
|
{
|
|
ID: StateChannel,
|
|
Priority: 5,
|
|
SendQueueCapacity: 100,
|
|
},
|
|
{
|
|
ID: DataChannel, // maybe split between gossiping current block and catchup stuff
|
|
Priority: 10, // once we gossip the whole block there's nothing left to send until next height or round
|
|
SendQueueCapacity: 100,
|
|
RecvBufferCapacity: 50 * 4096,
|
|
},
|
|
{
|
|
ID: VoteChannel,
|
|
Priority: 5,
|
|
SendQueueCapacity: 100,
|
|
RecvBufferCapacity: 100 * 100,
|
|
},
|
|
{
|
|
ID: VoteSetBitsChannel,
|
|
Priority: 1,
|
|
SendQueueCapacity: 2,
|
|
RecvBufferCapacity: 1024,
|
|
},
|
|
}
|
|
}
|
|
|
|
// AddPeer implements Reactor
|
|
func (conR *ConsensusReactor) AddPeer(peer p2p.Peer) {
|
|
if !conR.IsRunning() {
|
|
return
|
|
}
|
|
|
|
// Create peerState for peer
|
|
peerState := NewPeerState(peer).SetLogger(conR.Logger)
|
|
peer.Set(types.PeerStateKey, peerState)
|
|
|
|
// Begin routines for this peer.
|
|
go conR.gossipDataRoutine(peer, peerState)
|
|
go conR.gossipVotesRoutine(peer, peerState)
|
|
go conR.queryMaj23Routine(peer, peerState)
|
|
|
|
// Send our state to peer.
|
|
// If we're fast_syncing, broadcast a RoundStepMessage later upon SwitchToConsensus().
|
|
if !conR.FastSync() {
|
|
conR.sendNewRoundStepMessages(peer)
|
|
}
|
|
}
|
|
|
|
// RemovePeer implements Reactor
|
|
func (conR *ConsensusReactor) RemovePeer(peer p2p.Peer, reason interface{}) {
|
|
if !conR.IsRunning() {
|
|
return
|
|
}
|
|
// TODO
|
|
//peer.Get(PeerStateKey).(*PeerState).Disconnect()
|
|
}
|
|
|
|
// Receive implements Reactor
|
|
// NOTE: We process these messages even when we're fast_syncing.
|
|
// Messages affect either a peer state or the consensus state.
|
|
// Peer state updates can happen in parallel, but processing of
|
|
// proposals, block parts, and votes are ordered by the receiveRoutine
|
|
// NOTE: blocks on consensus state for proposals, block parts, and votes
|
|
func (conR *ConsensusReactor) Receive(chID byte, src p2p.Peer, msgBytes []byte) {
|
|
if !conR.IsRunning() {
|
|
conR.Logger.Debug("Receive", "src", src, "chId", chID, "bytes", msgBytes)
|
|
return
|
|
}
|
|
|
|
_, msg, err := DecodeMessage(msgBytes)
|
|
if err != nil {
|
|
conR.Logger.Error("Error decoding message", "src", src, "chId", chID, "msg", msg, "err", err, "bytes", msgBytes)
|
|
conR.Switch.StopPeerForError(src, err)
|
|
return
|
|
}
|
|
conR.Logger.Debug("Receive", "src", src, "chId", chID, "msg", msg)
|
|
|
|
// Get peer states
|
|
ps := src.Get(types.PeerStateKey).(*PeerState)
|
|
|
|
switch chID {
|
|
case StateChannel:
|
|
switch msg := msg.(type) {
|
|
case *NewRoundStepMessage:
|
|
ps.ApplyNewRoundStepMessage(msg)
|
|
case *CommitStepMessage:
|
|
ps.ApplyCommitStepMessage(msg)
|
|
case *HasVoteMessage:
|
|
ps.ApplyHasVoteMessage(msg)
|
|
case *VoteSetMaj23Message:
|
|
cs := conR.conS
|
|
cs.mtx.Lock()
|
|
height, votes := cs.Height, cs.Votes
|
|
cs.mtx.Unlock()
|
|
if height != msg.Height {
|
|
return
|
|
}
|
|
// Peer claims to have a maj23 for some BlockID at H,R,S,
|
|
err := votes.SetPeerMaj23(msg.Round, msg.Type, ps.Peer.ID(), msg.BlockID)
|
|
if err != nil {
|
|
conR.Switch.StopPeerForError(src, err)
|
|
return
|
|
}
|
|
// Respond with a VoteSetBitsMessage showing which votes we have.
|
|
// (and consequently shows which we don't have)
|
|
var ourVotes *cmn.BitArray
|
|
switch msg.Type {
|
|
case types.VoteTypePrevote:
|
|
ourVotes = votes.Prevotes(msg.Round).BitArrayByBlockID(msg.BlockID)
|
|
case types.VoteTypePrecommit:
|
|
ourVotes = votes.Precommits(msg.Round).BitArrayByBlockID(msg.BlockID)
|
|
default:
|
|
conR.Logger.Error("Bad VoteSetBitsMessage field Type")
|
|
return
|
|
}
|
|
src.TrySend(VoteSetBitsChannel, struct{ ConsensusMessage }{&VoteSetBitsMessage{
|
|
Height: msg.Height,
|
|
Round: msg.Round,
|
|
Type: msg.Type,
|
|
BlockID: msg.BlockID,
|
|
Votes: ourVotes,
|
|
}})
|
|
case *ProposalHeartbeatMessage:
|
|
hb := msg.Heartbeat
|
|
conR.Logger.Debug("Received proposal heartbeat message",
|
|
"height", hb.Height, "round", hb.Round, "sequence", hb.Sequence,
|
|
"valIdx", hb.ValidatorIndex, "valAddr", hb.ValidatorAddress)
|
|
default:
|
|
conR.Logger.Error(cmn.Fmt("Unknown message type %v", reflect.TypeOf(msg)))
|
|
}
|
|
|
|
case DataChannel:
|
|
if conR.FastSync() {
|
|
conR.Logger.Info("Ignoring message received during fastSync", "msg", msg)
|
|
return
|
|
}
|
|
switch msg := msg.(type) {
|
|
case *ProposalMessage:
|
|
ps.SetHasProposal(msg.Proposal)
|
|
conR.conS.peerMsgQueue <- msgInfo{msg, src.ID()}
|
|
case *ProposalPOLMessage:
|
|
ps.ApplyProposalPOLMessage(msg)
|
|
case *BlockPartMessage:
|
|
ps.SetHasProposalBlockPart(msg.Height, msg.Round, msg.Part.Index)
|
|
if numBlocks := ps.RecordBlockPart(msg); numBlocks%blocksToContributeToBecomeGoodPeer == 0 {
|
|
conR.Switch.MarkPeerAsGood(src)
|
|
}
|
|
conR.conS.peerMsgQueue <- msgInfo{msg, src.ID()}
|
|
default:
|
|
conR.Logger.Error(cmn.Fmt("Unknown message type %v", reflect.TypeOf(msg)))
|
|
}
|
|
|
|
case VoteChannel:
|
|
if conR.FastSync() {
|
|
conR.Logger.Info("Ignoring message received during fastSync", "msg", msg)
|
|
return
|
|
}
|
|
switch msg := msg.(type) {
|
|
case *VoteMessage:
|
|
cs := conR.conS
|
|
cs.mtx.Lock()
|
|
height, valSize, lastCommitSize := cs.Height, cs.Validators.Size(), cs.LastCommit.Size()
|
|
cs.mtx.Unlock()
|
|
ps.EnsureVoteBitArrays(height, valSize)
|
|
ps.EnsureVoteBitArrays(height-1, lastCommitSize)
|
|
ps.SetHasVote(msg.Vote)
|
|
if blocks := ps.RecordVote(msg.Vote); blocks%blocksToContributeToBecomeGoodPeer == 0 {
|
|
conR.Switch.MarkPeerAsGood(src)
|
|
}
|
|
|
|
cs.peerMsgQueue <- msgInfo{msg, src.ID()}
|
|
|
|
default:
|
|
// don't punish (leave room for soft upgrades)
|
|
conR.Logger.Error(cmn.Fmt("Unknown message type %v", reflect.TypeOf(msg)))
|
|
}
|
|
|
|
case VoteSetBitsChannel:
|
|
if conR.FastSync() {
|
|
conR.Logger.Info("Ignoring message received during fastSync", "msg", msg)
|
|
return
|
|
}
|
|
switch msg := msg.(type) {
|
|
case *VoteSetBitsMessage:
|
|
cs := conR.conS
|
|
cs.mtx.Lock()
|
|
height, votes := cs.Height, cs.Votes
|
|
cs.mtx.Unlock()
|
|
|
|
if height == msg.Height {
|
|
var ourVotes *cmn.BitArray
|
|
switch msg.Type {
|
|
case types.VoteTypePrevote:
|
|
ourVotes = votes.Prevotes(msg.Round).BitArrayByBlockID(msg.BlockID)
|
|
case types.VoteTypePrecommit:
|
|
ourVotes = votes.Precommits(msg.Round).BitArrayByBlockID(msg.BlockID)
|
|
default:
|
|
conR.Logger.Error("Bad VoteSetBitsMessage field Type")
|
|
return
|
|
}
|
|
ps.ApplyVoteSetBitsMessage(msg, ourVotes)
|
|
} else {
|
|
ps.ApplyVoteSetBitsMessage(msg, nil)
|
|
}
|
|
default:
|
|
// don't punish (leave room for soft upgrades)
|
|
conR.Logger.Error(cmn.Fmt("Unknown message type %v", reflect.TypeOf(msg)))
|
|
}
|
|
|
|
default:
|
|
conR.Logger.Error(cmn.Fmt("Unknown chId %X", chID))
|
|
}
|
|
|
|
if err != nil {
|
|
conR.Logger.Error("Error in Receive()", "err", err)
|
|
}
|
|
}
|
|
|
|
// SetEventBus sets event bus.
|
|
func (conR *ConsensusReactor) SetEventBus(b *types.EventBus) {
|
|
conR.eventBus = b
|
|
conR.conS.SetEventBus(b)
|
|
}
|
|
|
|
// FastSync returns whether the consensus reactor is in fast-sync mode.
|
|
func (conR *ConsensusReactor) FastSync() bool {
|
|
conR.mtx.RLock()
|
|
defer conR.mtx.RUnlock()
|
|
return conR.fastSync
|
|
}
|
|
|
|
//--------------------------------------
|
|
|
|
// startBroadcastRoutine subscribes for new round steps, votes and proposal
|
|
// heartbeats using the event bus and starts a go routine to broadcasts events
|
|
// to peers upon receiving them.
|
|
func (conR *ConsensusReactor) startBroadcastRoutine() error {
|
|
const subscriber = "consensus-reactor"
|
|
ctx := context.Background()
|
|
|
|
// new round steps
|
|
stepsCh := make(chan interface{})
|
|
err := conR.eventBus.Subscribe(ctx, subscriber, types.EventQueryNewRoundStep, stepsCh)
|
|
if err != nil {
|
|
return errors.Wrapf(err, "failed to subscribe %s to %s", subscriber, types.EventQueryNewRoundStep)
|
|
}
|
|
|
|
// votes
|
|
votesCh := make(chan interface{})
|
|
err = conR.eventBus.Subscribe(ctx, subscriber, types.EventQueryVote, votesCh)
|
|
if err != nil {
|
|
return errors.Wrapf(err, "failed to subscribe %s to %s", subscriber, types.EventQueryVote)
|
|
}
|
|
|
|
// proposal heartbeats
|
|
heartbeatsCh := make(chan interface{})
|
|
err = conR.eventBus.Subscribe(ctx, subscriber, types.EventQueryProposalHeartbeat, heartbeatsCh)
|
|
if err != nil {
|
|
return errors.Wrapf(err, "failed to subscribe %s to %s", subscriber, types.EventQueryProposalHeartbeat)
|
|
}
|
|
|
|
go func() {
|
|
for {
|
|
select {
|
|
case data, ok := <-stepsCh:
|
|
if ok { // a receive from a closed channel returns the zero value immediately
|
|
edrs := data.(types.TMEventData).Unwrap().(types.EventDataRoundState)
|
|
conR.broadcastNewRoundStep(edrs.RoundState.(*cstypes.RoundState))
|
|
}
|
|
case data, ok := <-votesCh:
|
|
if ok {
|
|
edv := data.(types.TMEventData).Unwrap().(types.EventDataVote)
|
|
conR.broadcastHasVoteMessage(edv.Vote)
|
|
}
|
|
case data, ok := <-heartbeatsCh:
|
|
if ok {
|
|
edph := data.(types.TMEventData).Unwrap().(types.EventDataProposalHeartbeat)
|
|
conR.broadcastProposalHeartbeatMessage(edph)
|
|
}
|
|
case <-conR.Quit():
|
|
conR.eventBus.UnsubscribeAll(ctx, subscriber)
|
|
return
|
|
}
|
|
}
|
|
}()
|
|
|
|
return nil
|
|
}
|
|
|
|
func (conR *ConsensusReactor) broadcastProposalHeartbeatMessage(heartbeat types.EventDataProposalHeartbeat) {
|
|
hb := heartbeat.Heartbeat
|
|
conR.Logger.Debug("Broadcasting proposal heartbeat message",
|
|
"height", hb.Height, "round", hb.Round, "sequence", hb.Sequence)
|
|
msg := &ProposalHeartbeatMessage{hb}
|
|
conR.Switch.Broadcast(StateChannel, struct{ ConsensusMessage }{msg})
|
|
}
|
|
|
|
func (conR *ConsensusReactor) broadcastNewRoundStep(rs *cstypes.RoundState) {
|
|
nrsMsg, csMsg := makeRoundStepMessages(rs)
|
|
if nrsMsg != nil {
|
|
conR.Switch.Broadcast(StateChannel, struct{ ConsensusMessage }{nrsMsg})
|
|
}
|
|
if csMsg != nil {
|
|
conR.Switch.Broadcast(StateChannel, struct{ ConsensusMessage }{csMsg})
|
|
}
|
|
}
|
|
|
|
// Broadcasts HasVoteMessage to peers that care.
|
|
func (conR *ConsensusReactor) broadcastHasVoteMessage(vote *types.Vote) {
|
|
msg := &HasVoteMessage{
|
|
Height: vote.Height,
|
|
Round: vote.Round,
|
|
Type: vote.Type,
|
|
Index: vote.ValidatorIndex,
|
|
}
|
|
conR.Switch.Broadcast(StateChannel, struct{ ConsensusMessage }{msg})
|
|
/*
|
|
// TODO: Make this broadcast more selective.
|
|
for _, peer := range conR.Switch.Peers().List() {
|
|
ps := peer.Get(PeerStateKey).(*PeerState)
|
|
prs := ps.GetRoundState()
|
|
if prs.Height == vote.Height {
|
|
// TODO: Also filter on round?
|
|
peer.TrySend(StateChannel, struct{ ConsensusMessage }{msg})
|
|
} else {
|
|
// Height doesn't match
|
|
// TODO: check a field, maybe CatchupCommitRound?
|
|
// TODO: But that requires changing the struct field comment.
|
|
}
|
|
}
|
|
*/
|
|
}
|
|
|
|
func makeRoundStepMessages(rs *cstypes.RoundState) (nrsMsg *NewRoundStepMessage, csMsg *CommitStepMessage) {
|
|
nrsMsg = &NewRoundStepMessage{
|
|
Height: rs.Height,
|
|
Round: rs.Round,
|
|
Step: rs.Step,
|
|
SecondsSinceStartTime: int(time.Since(rs.StartTime).Seconds()),
|
|
LastCommitRound: rs.LastCommit.Round(),
|
|
}
|
|
if rs.Step == cstypes.RoundStepCommit {
|
|
csMsg = &CommitStepMessage{
|
|
Height: rs.Height,
|
|
BlockPartsHeader: rs.ProposalBlockParts.Header(),
|
|
BlockParts: rs.ProposalBlockParts.BitArray(),
|
|
}
|
|
}
|
|
return
|
|
}
|
|
|
|
func (conR *ConsensusReactor) sendNewRoundStepMessages(peer p2p.Peer) {
|
|
rs := conR.conS.GetRoundState()
|
|
nrsMsg, csMsg := makeRoundStepMessages(rs)
|
|
if nrsMsg != nil {
|
|
peer.Send(StateChannel, struct{ ConsensusMessage }{nrsMsg})
|
|
}
|
|
if csMsg != nil {
|
|
peer.Send(StateChannel, struct{ ConsensusMessage }{csMsg})
|
|
}
|
|
}
|
|
|
|
func (conR *ConsensusReactor) gossipDataRoutine(peer p2p.Peer, ps *PeerState) {
|
|
logger := conR.Logger.With("peer", peer)
|
|
|
|
OUTER_LOOP:
|
|
for {
|
|
// Manage disconnects from self or peer.
|
|
if !peer.IsRunning() || !conR.IsRunning() {
|
|
logger.Info("Stopping gossipDataRoutine for peer")
|
|
return
|
|
}
|
|
rs := conR.conS.GetRoundState()
|
|
prs := ps.GetRoundState()
|
|
|
|
// Send proposal Block parts?
|
|
if rs.ProposalBlockParts.HasHeader(prs.ProposalBlockPartsHeader) {
|
|
if index, ok := rs.ProposalBlockParts.BitArray().Sub(prs.ProposalBlockParts.Copy()).PickRandom(); ok {
|
|
part := rs.ProposalBlockParts.GetPart(index)
|
|
msg := &BlockPartMessage{
|
|
Height: rs.Height, // This tells peer that this part applies to us.
|
|
Round: rs.Round, // This tells peer that this part applies to us.
|
|
Part: part,
|
|
}
|
|
logger.Debug("Sending block part", "height", prs.Height, "round", prs.Round)
|
|
if peer.Send(DataChannel, struct{ ConsensusMessage }{msg}) {
|
|
ps.SetHasProposalBlockPart(prs.Height, prs.Round, index)
|
|
}
|
|
continue OUTER_LOOP
|
|
}
|
|
}
|
|
|
|
// If the peer is on a previous height, help catch up.
|
|
if (0 < prs.Height) && (prs.Height < rs.Height) {
|
|
heightLogger := logger.With("height", prs.Height)
|
|
|
|
// if we never received the commit message from the peer, the block parts wont be initialized
|
|
if prs.ProposalBlockParts == nil {
|
|
blockMeta := conR.conS.blockStore.LoadBlockMeta(prs.Height)
|
|
if blockMeta == nil {
|
|
cmn.PanicCrisis(cmn.Fmt("Failed to load block %d when blockStore is at %d",
|
|
prs.Height, conR.conS.blockStore.Height()))
|
|
}
|
|
ps.InitProposalBlockParts(blockMeta.BlockID.PartsHeader)
|
|
// continue the loop since prs is a copy and not effected by this initialization
|
|
continue OUTER_LOOP
|
|
}
|
|
conR.gossipDataForCatchup(heightLogger, rs, prs, ps, peer)
|
|
continue OUTER_LOOP
|
|
}
|
|
|
|
// If height and round don't match, sleep.
|
|
if (rs.Height != prs.Height) || (rs.Round != prs.Round) {
|
|
//logger.Info("Peer Height|Round mismatch, sleeping", "peerHeight", prs.Height, "peerRound", prs.Round, "peer", peer)
|
|
time.Sleep(conR.conS.config.PeerGossipSleep())
|
|
continue OUTER_LOOP
|
|
}
|
|
|
|
// By here, height and round match.
|
|
// Proposal block parts were already matched and sent if any were wanted.
|
|
// (These can match on hash so the round doesn't matter)
|
|
// Now consider sending other things, like the Proposal itself.
|
|
|
|
// Send Proposal && ProposalPOL BitArray?
|
|
if rs.Proposal != nil && !prs.Proposal {
|
|
// Proposal: share the proposal metadata with peer.
|
|
{
|
|
msg := &ProposalMessage{Proposal: rs.Proposal}
|
|
logger.Debug("Sending proposal", "height", prs.Height, "round", prs.Round)
|
|
if peer.Send(DataChannel, struct{ ConsensusMessage }{msg}) {
|
|
ps.SetHasProposal(rs.Proposal)
|
|
}
|
|
}
|
|
// ProposalPOL: lets peer know which POL votes we have so far.
|
|
// Peer must receive ProposalMessage first.
|
|
// rs.Proposal was validated, so rs.Proposal.POLRound <= rs.Round,
|
|
// so we definitely have rs.Votes.Prevotes(rs.Proposal.POLRound).
|
|
if 0 <= rs.Proposal.POLRound {
|
|
msg := &ProposalPOLMessage{
|
|
Height: rs.Height,
|
|
ProposalPOLRound: rs.Proposal.POLRound,
|
|
ProposalPOL: rs.Votes.Prevotes(rs.Proposal.POLRound).BitArray(),
|
|
}
|
|
logger.Debug("Sending POL", "height", prs.Height, "round", prs.Round)
|
|
peer.Send(DataChannel, struct{ ConsensusMessage }{msg})
|
|
}
|
|
continue OUTER_LOOP
|
|
}
|
|
|
|
// Nothing to do. Sleep.
|
|
time.Sleep(conR.conS.config.PeerGossipSleep())
|
|
continue OUTER_LOOP
|
|
}
|
|
}
|
|
|
|
func (conR *ConsensusReactor) gossipDataForCatchup(logger log.Logger, rs *cstypes.RoundState,
|
|
prs *cstypes.PeerRoundState, ps *PeerState, peer p2p.Peer) {
|
|
|
|
if index, ok := prs.ProposalBlockParts.Not().PickRandom(); ok {
|
|
// Ensure that the peer's PartSetHeader is correct
|
|
blockMeta := conR.conS.blockStore.LoadBlockMeta(prs.Height)
|
|
if blockMeta == nil {
|
|
logger.Error("Failed to load block meta",
|
|
"ourHeight", rs.Height, "blockstoreHeight", conR.conS.blockStore.Height())
|
|
time.Sleep(conR.conS.config.PeerGossipSleep())
|
|
return
|
|
} else if !blockMeta.BlockID.PartsHeader.Equals(prs.ProposalBlockPartsHeader) {
|
|
logger.Info("Peer ProposalBlockPartsHeader mismatch, sleeping",
|
|
"blockPartsHeader", blockMeta.BlockID.PartsHeader, "peerBlockPartsHeader", prs.ProposalBlockPartsHeader)
|
|
time.Sleep(conR.conS.config.PeerGossipSleep())
|
|
return
|
|
}
|
|
// Load the part
|
|
part := conR.conS.blockStore.LoadBlockPart(prs.Height, index)
|
|
if part == nil {
|
|
logger.Error("Could not load part", "index", index,
|
|
"blockPartsHeader", blockMeta.BlockID.PartsHeader, "peerBlockPartsHeader", prs.ProposalBlockPartsHeader)
|
|
time.Sleep(conR.conS.config.PeerGossipSleep())
|
|
return
|
|
}
|
|
// Send the part
|
|
msg := &BlockPartMessage{
|
|
Height: prs.Height, // Not our height, so it doesn't matter.
|
|
Round: prs.Round, // Not our height, so it doesn't matter.
|
|
Part: part,
|
|
}
|
|
logger.Debug("Sending block part for catchup", "round", prs.Round, "index", index)
|
|
if peer.Send(DataChannel, struct{ ConsensusMessage }{msg}) {
|
|
ps.SetHasProposalBlockPart(prs.Height, prs.Round, index)
|
|
} else {
|
|
logger.Debug("Sending block part for catchup failed")
|
|
}
|
|
return
|
|
} else {
|
|
//logger.Info("No parts to send in catch-up, sleeping")
|
|
time.Sleep(conR.conS.config.PeerGossipSleep())
|
|
return
|
|
}
|
|
}
|
|
|
|
func (conR *ConsensusReactor) gossipVotesRoutine(peer p2p.Peer, ps *PeerState) {
|
|
logger := conR.Logger.With("peer", peer)
|
|
|
|
// Simple hack to throttle logs upon sleep.
|
|
var sleeping = 0
|
|
|
|
OUTER_LOOP:
|
|
for {
|
|
// Manage disconnects from self or peer.
|
|
if !peer.IsRunning() || !conR.IsRunning() {
|
|
logger.Info("Stopping gossipVotesRoutine for peer")
|
|
return
|
|
}
|
|
rs := conR.conS.GetRoundState()
|
|
prs := ps.GetRoundState()
|
|
|
|
switch sleeping {
|
|
case 1: // First sleep
|
|
sleeping = 2
|
|
case 2: // No more sleep
|
|
sleeping = 0
|
|
}
|
|
|
|
//logger.Debug("gossipVotesRoutine", "rsHeight", rs.Height, "rsRound", rs.Round,
|
|
// "prsHeight", prs.Height, "prsRound", prs.Round, "prsStep", prs.Step)
|
|
|
|
// If height matches, then send LastCommit, Prevotes, Precommits.
|
|
if rs.Height == prs.Height {
|
|
heightLogger := logger.With("height", prs.Height)
|
|
if conR.gossipVotesForHeight(heightLogger, rs, prs, ps) {
|
|
continue OUTER_LOOP
|
|
}
|
|
}
|
|
|
|
// Special catchup logic.
|
|
// If peer is lagging by height 1, send LastCommit.
|
|
if prs.Height != 0 && rs.Height == prs.Height+1 {
|
|
if ps.PickSendVote(rs.LastCommit) {
|
|
logger.Debug("Picked rs.LastCommit to send", "height", prs.Height)
|
|
continue OUTER_LOOP
|
|
}
|
|
}
|
|
|
|
// Catchup logic
|
|
// If peer is lagging by more than 1, send Commit.
|
|
if prs.Height != 0 && rs.Height >= prs.Height+2 {
|
|
// Load the block commit for prs.Height,
|
|
// which contains precommit signatures for prs.Height.
|
|
commit := conR.conS.blockStore.LoadBlockCommit(prs.Height)
|
|
if ps.PickSendVote(commit) {
|
|
logger.Debug("Picked Catchup commit to send", "height", prs.Height)
|
|
continue OUTER_LOOP
|
|
}
|
|
}
|
|
|
|
if sleeping == 0 {
|
|
// We sent nothing. Sleep...
|
|
sleeping = 1
|
|
logger.Debug("No votes to send, sleeping", "rs.Height", rs.Height, "prs.Height", prs.Height,
|
|
"localPV", rs.Votes.Prevotes(rs.Round).BitArray(), "peerPV", prs.Prevotes,
|
|
"localPC", rs.Votes.Precommits(rs.Round).BitArray(), "peerPC", prs.Precommits)
|
|
} else if sleeping == 2 {
|
|
// Continued sleep...
|
|
sleeping = 1
|
|
}
|
|
|
|
time.Sleep(conR.conS.config.PeerGossipSleep())
|
|
continue OUTER_LOOP
|
|
}
|
|
}
|
|
|
|
func (conR *ConsensusReactor) gossipVotesForHeight(logger log.Logger, rs *cstypes.RoundState, prs *cstypes.PeerRoundState, ps *PeerState) bool {
|
|
|
|
// If there are lastCommits to send...
|
|
if prs.Step == cstypes.RoundStepNewHeight {
|
|
if ps.PickSendVote(rs.LastCommit) {
|
|
logger.Debug("Picked rs.LastCommit to send")
|
|
return true
|
|
}
|
|
}
|
|
// If there are prevotes to send...
|
|
if prs.Step <= cstypes.RoundStepPrevote && prs.Round != -1 && prs.Round <= rs.Round {
|
|
if ps.PickSendVote(rs.Votes.Prevotes(prs.Round)) {
|
|
logger.Debug("Picked rs.Prevotes(prs.Round) to send", "round", prs.Round)
|
|
return true
|
|
}
|
|
}
|
|
// If there are precommits to send...
|
|
if prs.Step <= cstypes.RoundStepPrecommit && prs.Round != -1 && prs.Round <= rs.Round {
|
|
if ps.PickSendVote(rs.Votes.Precommits(prs.Round)) {
|
|
logger.Debug("Picked rs.Precommits(prs.Round) to send", "round", prs.Round)
|
|
return true
|
|
}
|
|
}
|
|
// If there are POLPrevotes to send...
|
|
if prs.ProposalPOLRound != -1 {
|
|
if polPrevotes := rs.Votes.Prevotes(prs.ProposalPOLRound); polPrevotes != nil {
|
|
if ps.PickSendVote(polPrevotes) {
|
|
logger.Debug("Picked rs.Prevotes(prs.ProposalPOLRound) to send",
|
|
"round", prs.ProposalPOLRound)
|
|
return true
|
|
}
|
|
}
|
|
}
|
|
return false
|
|
}
|
|
|
|
// NOTE: `queryMaj23Routine` has a simple crude design since it only comes
|
|
// into play for liveness when there's a signature DDoS attack happening.
|
|
func (conR *ConsensusReactor) queryMaj23Routine(peer p2p.Peer, ps *PeerState) {
|
|
logger := conR.Logger.With("peer", peer)
|
|
|
|
OUTER_LOOP:
|
|
for {
|
|
// Manage disconnects from self or peer.
|
|
if !peer.IsRunning() || !conR.IsRunning() {
|
|
logger.Info("Stopping queryMaj23Routine for peer")
|
|
return
|
|
}
|
|
|
|
// Maybe send Height/Round/Prevotes
|
|
{
|
|
rs := conR.conS.GetRoundState()
|
|
prs := ps.GetRoundState()
|
|
if rs.Height == prs.Height {
|
|
if maj23, ok := rs.Votes.Prevotes(prs.Round).TwoThirdsMajority(); ok {
|
|
peer.TrySend(StateChannel, struct{ ConsensusMessage }{&VoteSetMaj23Message{
|
|
Height: prs.Height,
|
|
Round: prs.Round,
|
|
Type: types.VoteTypePrevote,
|
|
BlockID: maj23,
|
|
}})
|
|
time.Sleep(conR.conS.config.PeerQueryMaj23Sleep())
|
|
}
|
|
}
|
|
}
|
|
|
|
// Maybe send Height/Round/Precommits
|
|
{
|
|
rs := conR.conS.GetRoundState()
|
|
prs := ps.GetRoundState()
|
|
if rs.Height == prs.Height {
|
|
if maj23, ok := rs.Votes.Precommits(prs.Round).TwoThirdsMajority(); ok {
|
|
peer.TrySend(StateChannel, struct{ ConsensusMessage }{&VoteSetMaj23Message{
|
|
Height: prs.Height,
|
|
Round: prs.Round,
|
|
Type: types.VoteTypePrecommit,
|
|
BlockID: maj23,
|
|
}})
|
|
time.Sleep(conR.conS.config.PeerQueryMaj23Sleep())
|
|
}
|
|
}
|
|
}
|
|
|
|
// Maybe send Height/Round/ProposalPOL
|
|
{
|
|
rs := conR.conS.GetRoundState()
|
|
prs := ps.GetRoundState()
|
|
if rs.Height == prs.Height && prs.ProposalPOLRound >= 0 {
|
|
if maj23, ok := rs.Votes.Prevotes(prs.ProposalPOLRound).TwoThirdsMajority(); ok {
|
|
peer.TrySend(StateChannel, struct{ ConsensusMessage }{&VoteSetMaj23Message{
|
|
Height: prs.Height,
|
|
Round: prs.ProposalPOLRound,
|
|
Type: types.VoteTypePrevote,
|
|
BlockID: maj23,
|
|
}})
|
|
time.Sleep(conR.conS.config.PeerQueryMaj23Sleep())
|
|
}
|
|
}
|
|
}
|
|
|
|
// Little point sending LastCommitRound/LastCommit,
|
|
// These are fleeting and non-blocking.
|
|
|
|
// Maybe send Height/CatchupCommitRound/CatchupCommit.
|
|
{
|
|
prs := ps.GetRoundState()
|
|
if prs.CatchupCommitRound != -1 && 0 < prs.Height && prs.Height <= conR.conS.blockStore.Height() {
|
|
commit := conR.conS.LoadCommit(prs.Height)
|
|
peer.TrySend(StateChannel, struct{ ConsensusMessage }{&VoteSetMaj23Message{
|
|
Height: prs.Height,
|
|
Round: commit.Round(),
|
|
Type: types.VoteTypePrecommit,
|
|
BlockID: commit.BlockID,
|
|
}})
|
|
time.Sleep(conR.conS.config.PeerQueryMaj23Sleep())
|
|
}
|
|
}
|
|
|
|
time.Sleep(conR.conS.config.PeerQueryMaj23Sleep())
|
|
|
|
continue OUTER_LOOP
|
|
}
|
|
}
|
|
|
|
// String returns a string representation of the ConsensusReactor.
|
|
// NOTE: For now, it is just a hard-coded string to avoid accessing unprotected shared variables.
|
|
// TODO: improve!
|
|
func (conR *ConsensusReactor) String() string {
|
|
// better not to access shared variables
|
|
return "ConsensusReactor" // conR.StringIndented("")
|
|
}
|
|
|
|
// StringIndented returns an indented string representation of the ConsensusReactor
|
|
func (conR *ConsensusReactor) StringIndented(indent string) string {
|
|
s := "ConsensusReactor{\n"
|
|
s += indent + " " + conR.conS.StringIndented(indent+" ") + "\n"
|
|
for _, peer := range conR.Switch.Peers().List() {
|
|
ps := peer.Get(types.PeerStateKey).(*PeerState)
|
|
s += indent + " " + ps.StringIndented(indent+" ") + "\n"
|
|
}
|
|
s += indent + "}"
|
|
return s
|
|
}
|
|
|
|
//-----------------------------------------------------------------------------
|
|
|
|
var (
|
|
ErrPeerStateHeightRegression = errors.New("Error peer state height regression")
|
|
ErrPeerStateInvalidStartTime = errors.New("Error peer state invalid startTime")
|
|
)
|
|
|
|
// PeerState contains the known state of a peer, including its connection
|
|
// and threadsafe access to its PeerRoundState.
|
|
type PeerState struct {
|
|
Peer p2p.Peer
|
|
logger log.Logger
|
|
|
|
mtx sync.Mutex
|
|
cstypes.PeerRoundState
|
|
|
|
stats *peerStateStats
|
|
}
|
|
|
|
// peerStateStats holds internal statistics for a peer.
|
|
type peerStateStats struct {
|
|
lastVoteHeight int64
|
|
votes int
|
|
|
|
lastBlockPartHeight int64
|
|
blockParts int
|
|
}
|
|
|
|
func (pss peerStateStats) String() string {
|
|
return fmt.Sprintf("peerStateStats{votes: %d, blockParts: %d}", pss.votes, pss.blockParts)
|
|
}
|
|
|
|
// NewPeerState returns a new PeerState for the given Peer
|
|
func NewPeerState(peer p2p.Peer) *PeerState {
|
|
return &PeerState{
|
|
Peer: peer,
|
|
logger: log.NewNopLogger(),
|
|
PeerRoundState: cstypes.PeerRoundState{
|
|
Round: -1,
|
|
ProposalPOLRound: -1,
|
|
LastCommitRound: -1,
|
|
CatchupCommitRound: -1,
|
|
},
|
|
stats: &peerStateStats{},
|
|
}
|
|
}
|
|
|
|
func (ps *PeerState) SetLogger(logger log.Logger) *PeerState {
|
|
ps.logger = logger
|
|
return ps
|
|
}
|
|
|
|
// GetRoundState returns an atomic snapshot of the PeerRoundState.
|
|
// There's no point in mutating it since it won't change PeerState.
|
|
func (ps *PeerState) GetRoundState() *cstypes.PeerRoundState {
|
|
ps.mtx.Lock()
|
|
defer ps.mtx.Unlock()
|
|
|
|
prs := ps.PeerRoundState // copy
|
|
return &prs
|
|
}
|
|
|
|
// GetHeight returns an atomic snapshot of the PeerRoundState's height
|
|
// used by the mempool to ensure peers are caught up before broadcasting new txs
|
|
func (ps *PeerState) GetHeight() int64 {
|
|
ps.mtx.Lock()
|
|
defer ps.mtx.Unlock()
|
|
return ps.PeerRoundState.Height
|
|
}
|
|
|
|
// SetHasProposal sets the given proposal as known for the peer.
|
|
func (ps *PeerState) SetHasProposal(proposal *types.Proposal) {
|
|
ps.mtx.Lock()
|
|
defer ps.mtx.Unlock()
|
|
|
|
if ps.Height != proposal.Height || ps.Round != proposal.Round {
|
|
return
|
|
}
|
|
if ps.Proposal {
|
|
return
|
|
}
|
|
|
|
ps.Proposal = true
|
|
ps.ProposalBlockPartsHeader = proposal.BlockPartsHeader
|
|
ps.ProposalBlockParts = cmn.NewBitArray(proposal.BlockPartsHeader.Total)
|
|
ps.ProposalPOLRound = proposal.POLRound
|
|
ps.ProposalPOL = nil // Nil until ProposalPOLMessage received.
|
|
}
|
|
|
|
// InitProposalBlockParts initializes the peer's proposal block parts header and bit array.
|
|
func (ps *PeerState) InitProposalBlockParts(partsHeader types.PartSetHeader) {
|
|
ps.mtx.Lock()
|
|
defer ps.mtx.Unlock()
|
|
|
|
if ps.ProposalBlockParts != nil {
|
|
return
|
|
}
|
|
|
|
ps.ProposalBlockPartsHeader = partsHeader
|
|
ps.ProposalBlockParts = cmn.NewBitArray(partsHeader.Total)
|
|
}
|
|
|
|
// SetHasProposalBlockPart sets the given block part index as known for the peer.
|
|
func (ps *PeerState) SetHasProposalBlockPart(height int64, round int, index int) {
|
|
ps.mtx.Lock()
|
|
defer ps.mtx.Unlock()
|
|
|
|
if ps.Height != height || ps.Round != round {
|
|
return
|
|
}
|
|
|
|
ps.ProposalBlockParts.SetIndex(index, true)
|
|
}
|
|
|
|
// PickSendVote picks a vote and sends it to the peer.
|
|
// Returns true if vote was sent.
|
|
func (ps *PeerState) PickSendVote(votes types.VoteSetReader) bool {
|
|
if vote, ok := ps.PickVoteToSend(votes); ok {
|
|
msg := &VoteMessage{vote}
|
|
ps.logger.Debug("Sending vote message", "ps", ps, "vote", vote)
|
|
return ps.Peer.Send(VoteChannel, struct{ ConsensusMessage }{msg})
|
|
}
|
|
return false
|
|
}
|
|
|
|
// PickVoteToSend picks a vote to send to the peer.
|
|
// Returns true if a vote was picked.
|
|
// NOTE: `votes` must be the correct Size() for the Height().
|
|
func (ps *PeerState) PickVoteToSend(votes types.VoteSetReader) (vote *types.Vote, ok bool) {
|
|
ps.mtx.Lock()
|
|
defer ps.mtx.Unlock()
|
|
|
|
if votes.Size() == 0 {
|
|
return nil, false
|
|
}
|
|
|
|
height, round, type_, size := votes.Height(), votes.Round(), votes.Type(), votes.Size()
|
|
|
|
// Lazily set data using 'votes'.
|
|
if votes.IsCommit() {
|
|
ps.ensureCatchupCommitRound(height, round, size)
|
|
}
|
|
ps.ensureVoteBitArrays(height, size)
|
|
|
|
psVotes := ps.getVoteBitArray(height, round, type_)
|
|
if psVotes == nil {
|
|
return nil, false // Not something worth sending
|
|
}
|
|
if index, ok := votes.BitArray().Sub(psVotes).PickRandom(); ok {
|
|
ps.setHasVote(height, round, type_, index)
|
|
return votes.GetByIndex(index), true
|
|
}
|
|
return nil, false
|
|
}
|
|
|
|
func (ps *PeerState) getVoteBitArray(height int64, round int, type_ byte) *cmn.BitArray {
|
|
if !types.IsVoteTypeValid(type_) {
|
|
return nil
|
|
}
|
|
|
|
if ps.Height == height {
|
|
if ps.Round == round {
|
|
switch type_ {
|
|
case types.VoteTypePrevote:
|
|
return ps.Prevotes
|
|
case types.VoteTypePrecommit:
|
|
return ps.Precommits
|
|
}
|
|
}
|
|
if ps.CatchupCommitRound == round {
|
|
switch type_ {
|
|
case types.VoteTypePrevote:
|
|
return nil
|
|
case types.VoteTypePrecommit:
|
|
return ps.CatchupCommit
|
|
}
|
|
}
|
|
if ps.ProposalPOLRound == round {
|
|
switch type_ {
|
|
case types.VoteTypePrevote:
|
|
return ps.ProposalPOL
|
|
case types.VoteTypePrecommit:
|
|
return nil
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
if ps.Height == height+1 {
|
|
if ps.LastCommitRound == round {
|
|
switch type_ {
|
|
case types.VoteTypePrevote:
|
|
return nil
|
|
case types.VoteTypePrecommit:
|
|
return ps.LastCommit
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// 'round': A round for which we have a +2/3 commit.
|
|
func (ps *PeerState) ensureCatchupCommitRound(height int64, round int, numValidators int) {
|
|
if ps.Height != height {
|
|
return
|
|
}
|
|
/*
|
|
NOTE: This is wrong, 'round' could change.
|
|
e.g. if orig round is not the same as block LastCommit round.
|
|
if ps.CatchupCommitRound != -1 && ps.CatchupCommitRound != round {
|
|
cmn.PanicSanity(cmn.Fmt("Conflicting CatchupCommitRound. Height: %v, Orig: %v, New: %v", height, ps.CatchupCommitRound, round))
|
|
}
|
|
*/
|
|
if ps.CatchupCommitRound == round {
|
|
return // Nothing to do!
|
|
}
|
|
ps.CatchupCommitRound = round
|
|
if round == ps.Round {
|
|
ps.CatchupCommit = ps.Precommits
|
|
} else {
|
|
ps.CatchupCommit = cmn.NewBitArray(numValidators)
|
|
}
|
|
}
|
|
|
|
// EnsureVoteVitArrays ensures the bit-arrays have been allocated for tracking
|
|
// what votes this peer has received.
|
|
// NOTE: It's important to make sure that numValidators actually matches
|
|
// what the node sees as the number of validators for height.
|
|
func (ps *PeerState) EnsureVoteBitArrays(height int64, numValidators int) {
|
|
ps.mtx.Lock()
|
|
defer ps.mtx.Unlock()
|
|
ps.ensureVoteBitArrays(height, numValidators)
|
|
}
|
|
|
|
func (ps *PeerState) ensureVoteBitArrays(height int64, numValidators int) {
|
|
if ps.Height == height {
|
|
if ps.Prevotes == nil {
|
|
ps.Prevotes = cmn.NewBitArray(numValidators)
|
|
}
|
|
if ps.Precommits == nil {
|
|
ps.Precommits = cmn.NewBitArray(numValidators)
|
|
}
|
|
if ps.CatchupCommit == nil {
|
|
ps.CatchupCommit = cmn.NewBitArray(numValidators)
|
|
}
|
|
if ps.ProposalPOL == nil {
|
|
ps.ProposalPOL = cmn.NewBitArray(numValidators)
|
|
}
|
|
} else if ps.Height == height+1 {
|
|
if ps.LastCommit == nil {
|
|
ps.LastCommit = cmn.NewBitArray(numValidators)
|
|
}
|
|
}
|
|
}
|
|
|
|
// RecordVote updates internal statistics for this peer by recording the vote.
|
|
// It returns the total number of votes (1 per block). This essentially means
|
|
// the number of blocks for which peer has been sending us votes.
|
|
func (ps *PeerState) RecordVote(vote *types.Vote) int {
|
|
if ps.stats.lastVoteHeight >= vote.Height {
|
|
return ps.stats.votes
|
|
}
|
|
ps.stats.lastVoteHeight = vote.Height
|
|
ps.stats.votes += 1
|
|
return ps.stats.votes
|
|
}
|
|
|
|
// VotesSent returns the number of blocks for which peer has been sending us
|
|
// votes.
|
|
func (ps *PeerState) VotesSent() int {
|
|
return ps.stats.votes
|
|
}
|
|
|
|
// RecordVote updates internal statistics for this peer by recording the block part.
|
|
// It returns the total number of block parts (1 per block). This essentially means
|
|
// the number of blocks for which peer has been sending us block parts.
|
|
func (ps *PeerState) RecordBlockPart(bp *BlockPartMessage) int {
|
|
if ps.stats.lastBlockPartHeight >= bp.Height {
|
|
return ps.stats.blockParts
|
|
}
|
|
|
|
ps.stats.lastBlockPartHeight = bp.Height
|
|
ps.stats.blockParts += 1
|
|
return ps.stats.blockParts
|
|
}
|
|
|
|
// BlockPartsSent returns the number of blocks for which peer has been sending
|
|
// us block parts.
|
|
func (ps *PeerState) BlockPartsSent() int {
|
|
return ps.stats.blockParts
|
|
}
|
|
|
|
// SetHasVote sets the given vote as known by the peer
|
|
func (ps *PeerState) SetHasVote(vote *types.Vote) {
|
|
ps.mtx.Lock()
|
|
defer ps.mtx.Unlock()
|
|
|
|
ps.setHasVote(vote.Height, vote.Round, vote.Type, vote.ValidatorIndex)
|
|
}
|
|
|
|
func (ps *PeerState) setHasVote(height int64, round int, type_ byte, index int) {
|
|
logger := ps.logger.With("peerH/R", cmn.Fmt("%d/%d", ps.Height, ps.Round), "H/R", cmn.Fmt("%d/%d", height, round))
|
|
logger.Debug("setHasVote", "type", type_, "index", index)
|
|
|
|
// NOTE: some may be nil BitArrays -> no side effects.
|
|
psVotes := ps.getVoteBitArray(height, round, type_)
|
|
if psVotes != nil {
|
|
psVotes.SetIndex(index, true)
|
|
}
|
|
}
|
|
|
|
// ApplyNewRoundStepMessage updates the peer state for the new round.
|
|
func (ps *PeerState) ApplyNewRoundStepMessage(msg *NewRoundStepMessage) {
|
|
ps.mtx.Lock()
|
|
defer ps.mtx.Unlock()
|
|
|
|
// Ignore duplicates or decreases
|
|
if CompareHRS(msg.Height, msg.Round, msg.Step, ps.Height, ps.Round, ps.Step) <= 0 {
|
|
return
|
|
}
|
|
|
|
// Just remember these values.
|
|
psHeight := ps.Height
|
|
psRound := ps.Round
|
|
//psStep := ps.Step
|
|
psCatchupCommitRound := ps.CatchupCommitRound
|
|
psCatchupCommit := ps.CatchupCommit
|
|
|
|
startTime := time.Now().Add(-1 * time.Duration(msg.SecondsSinceStartTime) * time.Second)
|
|
ps.Height = msg.Height
|
|
ps.Round = msg.Round
|
|
ps.Step = msg.Step
|
|
ps.StartTime = startTime
|
|
if psHeight != msg.Height || psRound != msg.Round {
|
|
ps.Proposal = false
|
|
ps.ProposalBlockPartsHeader = types.PartSetHeader{}
|
|
ps.ProposalBlockParts = nil
|
|
ps.ProposalPOLRound = -1
|
|
ps.ProposalPOL = nil
|
|
// We'll update the BitArray capacity later.
|
|
ps.Prevotes = nil
|
|
ps.Precommits = nil
|
|
}
|
|
if psHeight == msg.Height && psRound != msg.Round && msg.Round == psCatchupCommitRound {
|
|
// Peer caught up to CatchupCommitRound.
|
|
// Preserve psCatchupCommit!
|
|
// NOTE: We prefer to use prs.Precommits if
|
|
// pr.Round matches pr.CatchupCommitRound.
|
|
ps.Precommits = psCatchupCommit
|
|
}
|
|
if psHeight != msg.Height {
|
|
// Shift Precommits to LastCommit.
|
|
if psHeight+1 == msg.Height && psRound == msg.LastCommitRound {
|
|
ps.LastCommitRound = msg.LastCommitRound
|
|
ps.LastCommit = ps.Precommits
|
|
} else {
|
|
ps.LastCommitRound = msg.LastCommitRound
|
|
ps.LastCommit = nil
|
|
}
|
|
// We'll update the BitArray capacity later.
|
|
ps.CatchupCommitRound = -1
|
|
ps.CatchupCommit = nil
|
|
}
|
|
}
|
|
|
|
// ApplyCommitStepMessage updates the peer state for the new commit.
|
|
func (ps *PeerState) ApplyCommitStepMessage(msg *CommitStepMessage) {
|
|
ps.mtx.Lock()
|
|
defer ps.mtx.Unlock()
|
|
|
|
if ps.Height != msg.Height {
|
|
return
|
|
}
|
|
|
|
ps.ProposalBlockPartsHeader = msg.BlockPartsHeader
|
|
ps.ProposalBlockParts = msg.BlockParts
|
|
}
|
|
|
|
// ApplyProposalPOLMessage updates the peer state for the new proposal POL.
|
|
func (ps *PeerState) ApplyProposalPOLMessage(msg *ProposalPOLMessage) {
|
|
ps.mtx.Lock()
|
|
defer ps.mtx.Unlock()
|
|
|
|
if ps.Height != msg.Height {
|
|
return
|
|
}
|
|
if ps.ProposalPOLRound != msg.ProposalPOLRound {
|
|
return
|
|
}
|
|
|
|
// TODO: Merge onto existing ps.ProposalPOL?
|
|
// We might have sent some prevotes in the meantime.
|
|
ps.ProposalPOL = msg.ProposalPOL
|
|
}
|
|
|
|
// ApplyHasVoteMessage updates the peer state for the new vote.
|
|
func (ps *PeerState) ApplyHasVoteMessage(msg *HasVoteMessage) {
|
|
ps.mtx.Lock()
|
|
defer ps.mtx.Unlock()
|
|
|
|
if ps.Height != msg.Height {
|
|
return
|
|
}
|
|
|
|
ps.setHasVote(msg.Height, msg.Round, msg.Type, msg.Index)
|
|
}
|
|
|
|
// ApplyVoteSetBitsMessage updates the peer state for the bit-array of votes
|
|
// it claims to have for the corresponding BlockID.
|
|
// `ourVotes` is a BitArray of votes we have for msg.BlockID
|
|
// NOTE: if ourVotes is nil (e.g. msg.Height < rs.Height),
|
|
// we conservatively overwrite ps's votes w/ msg.Votes.
|
|
func (ps *PeerState) ApplyVoteSetBitsMessage(msg *VoteSetBitsMessage, ourVotes *cmn.BitArray) {
|
|
ps.mtx.Lock()
|
|
defer ps.mtx.Unlock()
|
|
|
|
votes := ps.getVoteBitArray(msg.Height, msg.Round, msg.Type)
|
|
if votes != nil {
|
|
if ourVotes == nil {
|
|
votes.Update(msg.Votes)
|
|
} else {
|
|
otherVotes := votes.Sub(ourVotes)
|
|
hasVotes := otherVotes.Or(msg.Votes)
|
|
votes.Update(hasVotes)
|
|
}
|
|
}
|
|
}
|
|
|
|
// String returns a string representation of the PeerState
|
|
func (ps *PeerState) String() string {
|
|
return ps.StringIndented("")
|
|
}
|
|
|
|
// StringIndented returns a string representation of the PeerState
|
|
func (ps *PeerState) StringIndented(indent string) string {
|
|
ps.mtx.Lock()
|
|
defer ps.mtx.Unlock()
|
|
return fmt.Sprintf(`PeerState{
|
|
%s Key %v
|
|
%s PRS %v
|
|
%s Stats %v
|
|
%s}`,
|
|
indent, ps.Peer.ID(),
|
|
indent, ps.PeerRoundState.StringIndented(indent+" "),
|
|
indent, ps.stats,
|
|
indent)
|
|
}
|
|
|
|
//-----------------------------------------------------------------------------
|
|
// Messages
|
|
|
|
const (
|
|
msgTypeNewRoundStep = byte(0x01)
|
|
msgTypeCommitStep = byte(0x02)
|
|
msgTypeProposal = byte(0x11)
|
|
msgTypeProposalPOL = byte(0x12)
|
|
msgTypeBlockPart = byte(0x13) // both block & POL
|
|
msgTypeVote = byte(0x14)
|
|
msgTypeHasVote = byte(0x15)
|
|
msgTypeVoteSetMaj23 = byte(0x16)
|
|
msgTypeVoteSetBits = byte(0x17)
|
|
|
|
msgTypeProposalHeartbeat = byte(0x20)
|
|
)
|
|
|
|
// ConsensusMessage is a message that can be sent and received on the ConsensusReactor
|
|
type ConsensusMessage interface{}
|
|
|
|
var _ = wire.RegisterInterface(
|
|
struct{ ConsensusMessage }{},
|
|
wire.ConcreteType{&NewRoundStepMessage{}, msgTypeNewRoundStep},
|
|
wire.ConcreteType{&CommitStepMessage{}, msgTypeCommitStep},
|
|
wire.ConcreteType{&ProposalMessage{}, msgTypeProposal},
|
|
wire.ConcreteType{&ProposalPOLMessage{}, msgTypeProposalPOL},
|
|
wire.ConcreteType{&BlockPartMessage{}, msgTypeBlockPart},
|
|
wire.ConcreteType{&VoteMessage{}, msgTypeVote},
|
|
wire.ConcreteType{&HasVoteMessage{}, msgTypeHasVote},
|
|
wire.ConcreteType{&VoteSetMaj23Message{}, msgTypeVoteSetMaj23},
|
|
wire.ConcreteType{&VoteSetBitsMessage{}, msgTypeVoteSetBits},
|
|
wire.ConcreteType{&ProposalHeartbeatMessage{}, msgTypeProposalHeartbeat},
|
|
)
|
|
|
|
// DecodeMessage decodes the given bytes into a ConsensusMessage.
|
|
// TODO: check for unnecessary extra bytes at the end.
|
|
func DecodeMessage(bz []byte) (msgType byte, msg ConsensusMessage, err error) {
|
|
msgType = bz[0]
|
|
n := new(int)
|
|
r := bytes.NewReader(bz)
|
|
msgI := wire.ReadBinary(struct{ ConsensusMessage }{}, r, maxConsensusMessageSize, n, &err)
|
|
msg = msgI.(struct{ ConsensusMessage }).ConsensusMessage
|
|
return
|
|
}
|
|
|
|
//-------------------------------------
|
|
|
|
// NewRoundStepMessage is sent for every step taken in the ConsensusState.
|
|
// For every height/round/step transition
|
|
type NewRoundStepMessage struct {
|
|
Height int64
|
|
Round int
|
|
Step cstypes.RoundStepType
|
|
SecondsSinceStartTime int
|
|
LastCommitRound int
|
|
}
|
|
|
|
// String returns a string representation.
|
|
func (m *NewRoundStepMessage) String() string {
|
|
return fmt.Sprintf("[NewRoundStep H:%v R:%v S:%v LCR:%v]",
|
|
m.Height, m.Round, m.Step, m.LastCommitRound)
|
|
}
|
|
|
|
//-------------------------------------
|
|
|
|
// CommitStepMessage is sent when a block is committed.
|
|
type CommitStepMessage struct {
|
|
Height int64
|
|
BlockPartsHeader types.PartSetHeader
|
|
BlockParts *cmn.BitArray
|
|
}
|
|
|
|
// String returns a string representation.
|
|
func (m *CommitStepMessage) String() string {
|
|
return fmt.Sprintf("[CommitStep H:%v BP:%v BA:%v]", m.Height, m.BlockPartsHeader, m.BlockParts)
|
|
}
|
|
|
|
//-------------------------------------
|
|
|
|
// ProposalMessage is sent when a new block is proposed.
|
|
type ProposalMessage struct {
|
|
Proposal *types.Proposal
|
|
}
|
|
|
|
// String returns a string representation.
|
|
func (m *ProposalMessage) String() string {
|
|
return fmt.Sprintf("[Proposal %v]", m.Proposal)
|
|
}
|
|
|
|
//-------------------------------------
|
|
|
|
// ProposalPOLMessage is sent when a previous proposal is re-proposed.
|
|
type ProposalPOLMessage struct {
|
|
Height int64
|
|
ProposalPOLRound int
|
|
ProposalPOL *cmn.BitArray
|
|
}
|
|
|
|
// String returns a string representation.
|
|
func (m *ProposalPOLMessage) String() string {
|
|
return fmt.Sprintf("[ProposalPOL H:%v POLR:%v POL:%v]", m.Height, m.ProposalPOLRound, m.ProposalPOL)
|
|
}
|
|
|
|
//-------------------------------------
|
|
|
|
// BlockPartMessage is sent when gossipping a piece of the proposed block.
|
|
type BlockPartMessage struct {
|
|
Height int64
|
|
Round int
|
|
Part *types.Part
|
|
}
|
|
|
|
// String returns a string representation.
|
|
func (m *BlockPartMessage) String() string {
|
|
return fmt.Sprintf("[BlockPart H:%v R:%v P:%v]", m.Height, m.Round, m.Part)
|
|
}
|
|
|
|
//-------------------------------------
|
|
|
|
// VoteMessage is sent when voting for a proposal (or lack thereof).
|
|
type VoteMessage struct {
|
|
Vote *types.Vote
|
|
}
|
|
|
|
// String returns a string representation.
|
|
func (m *VoteMessage) String() string {
|
|
return fmt.Sprintf("[Vote %v]", m.Vote)
|
|
}
|
|
|
|
//-------------------------------------
|
|
|
|
// HasVoteMessage is sent to indicate that a particular vote has been received.
|
|
type HasVoteMessage struct {
|
|
Height int64
|
|
Round int
|
|
Type byte
|
|
Index int
|
|
}
|
|
|
|
// String returns a string representation.
|
|
func (m *HasVoteMessage) String() string {
|
|
return fmt.Sprintf("[HasVote VI:%v V:{%v/%02d/%v}]", m.Index, m.Height, m.Round, m.Type)
|
|
}
|
|
|
|
//-------------------------------------
|
|
|
|
// VoteSetMaj23Message is sent to indicate that a given BlockID has seen +2/3 votes.
|
|
type VoteSetMaj23Message struct {
|
|
Height int64
|
|
Round int
|
|
Type byte
|
|
BlockID types.BlockID
|
|
}
|
|
|
|
// String returns a string representation.
|
|
func (m *VoteSetMaj23Message) String() string {
|
|
return fmt.Sprintf("[VSM23 %v/%02d/%v %v]", m.Height, m.Round, m.Type, m.BlockID)
|
|
}
|
|
|
|
//-------------------------------------
|
|
|
|
// VoteSetBitsMessage is sent to communicate the bit-array of votes seen for the BlockID.
|
|
type VoteSetBitsMessage struct {
|
|
Height int64
|
|
Round int
|
|
Type byte
|
|
BlockID types.BlockID
|
|
Votes *cmn.BitArray
|
|
}
|
|
|
|
// String returns a string representation.
|
|
func (m *VoteSetBitsMessage) String() string {
|
|
return fmt.Sprintf("[VSB %v/%02d/%v %v %v]", m.Height, m.Round, m.Type, m.BlockID, m.Votes)
|
|
}
|
|
|
|
//-------------------------------------
|
|
|
|
// ProposalHeartbeatMessage is sent to signal that a node is alive and waiting for transactions for a proposal.
|
|
type ProposalHeartbeatMessage struct {
|
|
Heartbeat *types.Heartbeat
|
|
}
|
|
|
|
// String returns a string representation.
|
|
func (m *ProposalHeartbeatMessage) String() string {
|
|
return fmt.Sprintf("[HEARTBEAT %v]", m.Heartbeat)
|
|
}
|