func (conR *ConsensusReactor) queryMaj23Routine(peer *p2p.Peer, ps *PeerState) {
OUTER_LOOP:
	for {
		if !peer.IsRunning() || !conR.IsRunning() {
			conR.slogger.Infof("Stopping queryMaj23Routine for %v.", peer)
			return
		}
		{
			rs := conR.conS.GetRoundState()
			prs := ps.GetRoundState()
			if rs.Height == prs.Height {
				if maj23, ok := rs.Votes.Prevotes(prs.Round).TwoThirdsMajority(); ok {
					peer.TrySendBytes(StateChannel, csspb.MarshalDataToCssMsg(&csspb.VoteSetMaj23Message{
						Height:  prs.Height,
						Round:   prs.Round,
						Type:    pbtypes.VoteType_Prevote,
						BlockID: &maj23,
					}))
					time.Sleep(peerQueryMaj23SleepDuration)
				}
			}
		}
		{
			rs := conR.conS.GetRoundState()
			prs := ps.GetRoundState()
			if rs.Height == prs.Height {
				if maj23, ok := rs.Votes.Precommits(prs.Round).TwoThirdsMajority(); ok {
					peer.TrySendBytes(StateChannel, csspb.MarshalDataToCssMsg(&csspb.VoteSetMaj23Message{
						Height:  prs.Height,
						Round:   prs.Round,
						Type:    pbtypes.VoteType_Precommit,
						BlockID: &maj23,
					}))
					time.Sleep(peerQueryMaj23SleepDuration)
				}
			}
		}
		{
			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.TrySendBytes(StateChannel, csspb.MarshalDataToCssMsg(&csspb.VoteSetMaj23Message{
						Height:  prs.Height,
						Round:   prs.ProposalPOLRound,
						Type:    pbtypes.VoteType_Prevote,
						BlockID: &maj23,
					}))
					time.Sleep(peerQueryMaj23SleepDuration)
				}
			}
		}
		{
			prs := ps.GetRoundState()
			if prs.CatchupCommitRound != -1 && 0 < prs.Height && prs.Height <= conR.conS.blockStore.Height() {
				commit := conR.conS.LoadCommit(prs.Height)
				peer.TrySendBytes(StateChannel, csspb.MarshalDataToCssMsg(&csspb.VoteSetMaj23Message{
					Height:  prs.Height,
					Round:   commit.Round(),
					Type:    pbtypes.VoteType_Precommit,
					BlockID: commit.BlockID,
				}))
				time.Sleep(peerQueryMaj23SleepDuration)
			}
		}
		time.Sleep(peerQueryMaj23SleepDuration)
		continue OUTER_LOOP
	}
}
NOTE: `queryMaj23Routine` has a simple crude design since it only comes
into play for liveness when there's a signature DDoS attack happening.
package consensus
import (
	"errors"
	"fmt"
	"reflect"
	"sync"
	"time"
	"go.uber.org/zap"
	csspb "github.com/annchain/annchain/angine/protos/consensus"
	pbtypes "github.com/annchain/annchain/angine/protos/types"
	sm "github.com/annchain/annchain/angine/state"
	agtypes "github.com/annchain/annchain/angine/types"
	. "github.com/annchain/annchain/module/lib/go-common"
	"github.com/annchain/annchain/module/lib/go-p2p"
	"github.com/annchain/annchain/module/xlib/def"
)
const (
	StateChannel       = byte(0x20)
	DataChannel        = byte(0x21)
	VoteChannel        = byte(0x22)
	VoteSetBitsChannel = byte(0x23)
	peerGossipSleepDuration     = 100 * time.Millisecond  
	peerQueryMaj23SleepDuration = 2 * time.Second         
	maxConsensusMessageSize     = 1048576                 
)
type ConsensusReactor struct {
	p2p.BaseReactor  
	conS     *ConsensusState
	fastSync bool
	evsw     agtypes.EventSwitch
	logger  *zap.Logger
	slogger *zap.SugaredLogger
}
func NewConsensusReactor(logger *zap.Logger, consensusState *ConsensusState, fastSync bool) *ConsensusReactor {
	conR := &ConsensusReactor{
		conS:     consensusState,
		fastSync: fastSync,
		logger:  logger,
		slogger: logger.Sugar(),
	}
	conR.BaseReactor = *p2p.NewBaseReactor(logger, "ConsensusReactor", conR)
	return conR
}
func (conR *ConsensusReactor) OnStart() error {
	conR.logger.Info("ConsensusReactor ", zap.Bool("fastSync", conR.fastSync))
	conR.BaseReactor.OnStart()
	conR.registerEventCallbacks()
	if !conR.fastSync {
		_, err := conR.conS.Start()
		if err != nil {
			return err
		}
	}
	return nil
}
func (conR *ConsensusReactor) OnStop() {
	conR.BaseReactor.OnStop()
	conR.conS.Stop()
}
func (conR *ConsensusReactor) SwitchToConsensus(state *sm.State) {
	cs := conR.conS
	validators := state.Validators
	height := state.LastBlockHeight + 1  
	cs.mtx.Lock()
	cs.updateHeight(height)
	cs.updateRoundStep(0, csspb.RoundStepType_NewHeight)
	cs.StartTime = cs.timeoutParams.Commit(time.Now())
	cs.Validators = validators
	cs.Proposal = nil
	cs.ProposalBlock = nil
	cs.ProposalBlockParts = nil
	cs.LockedRound = 0
	cs.LockedBlock = nil
	cs.LockedBlockParts = nil
	cs.Votes = NewHeightVoteSet(cs.config.GetString("chain_id"), height, validators)
	cs.CommitRound = -1
	cs.reconstructLastCommit(state)
	cs.LastValidators = state.LastValidators
	cs.state = state
	cs.mtx.Unlock()
	cs.newStep()
	conR.fastSync = false
	cs.Start()
}
func (conR *ConsensusReactor) GetChannels() []*p2p.ChannelDescriptor {
	return []*p2p.ChannelDescriptor{
		&p2p.ChannelDescriptor{
			ID:                StateChannel,
			Priority:          5,
			SendQueueCapacity: 100,
		},
		&p2p.ChannelDescriptor{
			ID:                 DataChannel,  
			Priority:           10,           
			SendQueueCapacity:  100,
			RecvBufferCapacity: 50 * 4096,
		},
		&p2p.ChannelDescriptor{
			ID:                 VoteChannel,
			Priority:           5,
			SendQueueCapacity:  100,
			RecvBufferCapacity: 100 * 100,
		},
		&p2p.ChannelDescriptor{
			ID:                 VoteSetBitsChannel,
			Priority:           1,
			SendQueueCapacity:  2,
			RecvBufferCapacity: 1024,
		},
	}
}
func (conR *ConsensusReactor) AddPeer(peer *p2p.Peer) {
	if !conR.IsRunning() {
		return
	}
	peerState := NewPeerState(conR.slogger, peer)
	peer.Data.Set(agtypes.PeerStateKey, peerState)
	go conR.gossipDataRoutine(peer, peerState)
	go conR.gossipVotesRoutine(peer, peerState)
	go conR.queryMaj23Routine(peer, peerState)
	if !conR.fastSync {
		conR.sendNewRoundStepMessage(peer)
	}
}
func (conR *ConsensusReactor) RemovePeer(peer *p2p.Peer, reason interface{}) {
	if !conR.IsRunning() {
		return
	}
}
func (conR *ConsensusReactor) Receive(chID byte, src *p2p.Peer, msgBytes []byte) {
	if !conR.IsRunning() {
		conR.slogger.Debugw("Receive", "src", src, "chId", chID, "bytes", msgBytes)
		return
	}
	msg, err := csspb.UnmarshalCssMsg(msgBytes)
	if err != nil {
		conR.slogger.Warnw("Error decoding message", "src", src, "chId", chID, "msg", msg, "error", err, "bytes", msgBytes)
		return
	}
	ps := src.Data.Get(agtypes.PeerStateKey).(*PeerState)
	switch chID {
	case StateChannel:
		switch msg := msg.(type) {
		case *csspb.NewRoundStepMessage:
			ps.ApplyNewRoundStepMessage(msg)
		case *csspb.CommitStepMessage:
			ps.ApplyCommitStepMessage(msg)
		case *csspb.HasVoteMessage:
			ps.ApplyHasVoteMessage(msg)
		case *csspb.VoteSetMaj23Message:
			cs := conR.conS
			cs.mtx.Lock()
			height, votes := cs.Height, cs.Votes
			cs.mtx.Unlock()
			if height != msg.Height {
				return
			}
			votes.SetPeerMaj23(msg.Round, msg.Type, ps.Peer.Key, msg.BlockID)
			var ourVotes *BitArray
			switch msg.Type {
			case pbtypes.VoteType_Prevote:
				ourVotes = votes.Prevotes(msg.Round).BitArrayByBlockID(msg.BlockID)
			case pbtypes.VoteType_Precommit:
				ourVotes = votes.Precommits(msg.Round).BitArrayByBlockID(msg.BlockID)
			default:
				conR.logger.Warn("Bad VoteSetBitsMessage field Type")
				return
			}
			src.TrySendBytes(VoteSetBitsChannel, csspb.MarshalDataToCssMsg(&csspb.VoteSetBitsMessage{
				Height:  msg.Height,
				Round:   msg.Round,
				Type:    msg.Type,
				BlockID: msg.BlockID,
				Votes:   csspb.TransferBitArray(ourVotes),
			}))
		default:
			conR.slogger.Warnf("Unknown message type %T", reflect.TypeOf(msg))
		}
	case DataChannel:
		if conR.fastSync {
			conR.slogger.Warnw("Ignoring message received during fastSync", "msg", msg)
			return
		}
		switch msg := msg.(type) {
		case *csspb.ProposalMessage:
			ps.SetHasProposal(msg.Proposal)
			conR.conS.peerMsgQueue <- genMsgInfo(msg, src.Key)
		case *csspb.ProposalPOLMessage:
			ps.ApplyProposalPOLMessage(msg)
		case *csspb.BlockPartMessage:
			ps.SetHasProposalBlockPart(msg.Height, msg.Round, int(msg.Part.Index))
			conR.conS.peerMsgQueue <- genMsgInfo(msg, src.Key)
		default:
			conR.slogger.Warnf("Unknown message type %T", msg)
		}
	case VoteChannel:
		if conR.fastSync {
			conR.slogger.Warnw("Ignoring message received during fastSync", "msg", msg)
			return
		}
		switch msg := msg.(type) {
		case *csspb.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)
			conR.conS.peerMsgQueue <- genMsgInfo(msg, src.Key)
		default:
			conR.slogger.Warnf("Unknown message type %T", msg)
		}
	case VoteSetBitsChannel:
		if conR.fastSync {
			conR.slogger.Warnw("Ignoring message received during fastSync", "msg", msg)
			return
		}
		switch msg := msg.(type) {
		case *csspb.VoteSetBitsMessage:
			cs := conR.conS
			cs.mtx.Lock()
			height, votes := cs.Height, cs.Votes
			cs.mtx.Unlock()
			if height == msg.Height {
				var ourVotes *BitArray
				switch msg.Type {
				case pbtypes.VoteType_Prevote:
					ourVotes = votes.Prevotes(msg.Round).BitArrayByBlockID(msg.BlockID)
				case pbtypes.VoteType_Precommit:
					ourVotes = votes.Precommits(msg.Round).BitArrayByBlockID(msg.BlockID)
				default:
					conR.logger.Warn("Bad VoteSetBitsMessage field Type")
					return
				}
				ps.ApplyVoteSetBitsMessage(msg, ourVotes)
			} else {
				ps.ApplyVoteSetBitsMessage(msg, nil)
			}
		default:
			conR.slogger.Warnf("Unknown message type %T", msg)
		}
	default:
		conR.slogger.Warnf("Unknown chId %X", chID)
	}
	if err != nil {
		conR.logger.Warn("Error in Receive()", zap.String("error", err.Error()))
	}
}
func (conR *ConsensusReactor) SetEventSwitch(evsw agtypes.EventSwitch) {
	conR.evsw = evsw
	conR.conS.SetEventSwitch(evsw)
}
func (conR *ConsensusReactor) registerEventCallbacks() {
	agtypes.AddListenerForEvent(conR.evsw, "conR", agtypes.EventStringNewRoundStep(), func(data agtypes.TMEventData) {
		rs := data.(agtypes.EventDataRoundState).RoundState.(*RoundState)
		conR.broadcastNewRoundStep(rs)
	})
	agtypes.AddListenerForEvent(conR.evsw, "conR", agtypes.EventStringVote(), func(data agtypes.TMEventData) {
		edv := data.(agtypes.EventDataVote)
		conR.broadcastHasVoteMessage(edv.Vote)
	})
	agtypes.AddListenerForEvent(conR.evsw, "conR", agtypes.EventStringSwitchToConsensus(), func(data agtypes.TMEventData) {
		conR.SwitchToConsensus(conR.conS.state)
	})
}
func (conR *ConsensusReactor) broadcastNewRoundStep(rs *RoundState) {
	nrsMsg, csMsg := makeRoundStepMessages(rs)
	if nrsMsg != nil {
		conR.Switch.BroadcastBytes(StateChannel, csspb.MarshalDataToCssMsg(nrsMsg))
	}
	if csMsg != nil {
		conR.Switch.BroadcastBytes(StateChannel, csspb.MarshalDataToCssMsg(csMsg))
	}
}
func (conR *ConsensusReactor) broadcastHasVoteMessage(vote *pbtypes.Vote) {
	vdata := vote.GetData()
	msg := &csspb.HasVoteMessage{
		Height: vdata.Height,
		Round:  vdata.Round,
		Type:   vdata.Type,
		Index:  vdata.ValidatorIndex,
	}
	conR.Switch.BroadcastBytes(StateChannel, csspb.MarshalDataToCssMsg(msg))
}
func makeRoundStepMessages(rs *RoundState) (nrsMsg *csspb.NewRoundStepMessage, csMsg *csspb.CommitStepMessage) {
	nrsMsg = &csspb.NewRoundStepMessage{
		Height: rs.Height,
		Round:  rs.Round,
		Step:   rs.Step,
		SecondsSinceStartTime: def.INT(time.Now().Sub(rs.StartTime).Seconds()),
		LastCommitRound:       rs.LastCommit.Round(),
	}
	if rs.Step == csspb.RoundStepType_Commit {
		header := rs.ProposalBlockParts.Header()
		csMsg = &csspb.CommitStepMessage{
			Height:           rs.Height,
			BlockPartsHeader: header,
			BlockParts:       csspb.TransferBitArray(rs.ProposalBlockParts.BitArray()),
		}
	}
	return
}
func (conR *ConsensusReactor) sendNewRoundStepMessage(peer *p2p.Peer) {
	rs := conR.conS.GetRoundState()
	nrsMsg, csMsg := makeRoundStepMessages(rs)
	if nrsMsg != nil {
		peer.SendBytes(StateChannel, csspb.MarshalDataToCssMsg(nrsMsg))
	}
	if csMsg != nil {
		peer.SendBytes(StateChannel, csspb.MarshalDataToCssMsg(csMsg))
	}
}
func (conR *ConsensusReactor) gossipDataRoutine(peer *p2p.Peer, ps *PeerState) {
OUTER_LOOP:
	for {
		if !peer.IsRunning() || !conR.IsRunning() {
			conR.slogger.Infof("Stopping gossipDataRoutine for %v.", peer)
			return
		}
		rs := conR.conS.GetRoundState()
		prs := ps.GetRoundState()
		if rs.ProposalBlockParts.HasHeader(&prs.ProposalBlockPartsHeader) {
			if index, ok := rs.ProposalBlockParts.BitArray().Sub(prs.ProposalBlockParts.Copy()).PickRandom(); ok {
				part := rs.ProposalBlockParts.GetPart(index)
				msg := &csspb.BlockPartMessage{
					Height: rs.Height,  
					Round:  rs.Round,   
					Part:   part,
				}
				peer.SendBytes(DataChannel, csspb.MarshalDataToCssMsg(msg))
				ps.SetHasProposalBlockPart(prs.Height, prs.Round, index)
				continue OUTER_LOOP
			}
		}
		if (0 < prs.Height) && (prs.Height < rs.Height) {
			if index, ok := prs.ProposalBlockParts.Not().PickRandom(); ok {
				blockMeta := conR.conS.blockStore.LoadBlockMeta(prs.Height)
				if blockMeta == nil {
					conR.slogger.Warnw("Failed to load block meta", "peer height", prs.Height, "our height", rs.Height, "blockstore height", conR.conS.blockStore.Height(), "pv", conR.conS.privValidator)
					time.Sleep(peerGossipSleepDuration)
					continue OUTER_LOOP
				} else if !blockMeta.PartsHeader.Equals(&prs.ProposalBlockPartsHeader) {
					conR.slogger.Debugw("Peer ProposalBlockPartsHeader mismatch, sleeping",
						"peerHeight", prs.Height, "blockPartsHeader", blockMeta.PartsHeader, "peerBlockPartsHeader", prs.ProposalBlockPartsHeader)
					time.Sleep(peerGossipSleepDuration)
					continue OUTER_LOOP
				}
				part := conR.conS.blockStore.LoadBlockPart(prs.Height, index)
				if part == nil {
					conR.slogger.Warnw("Could not load part", "index", index,
						"peerHeight", prs.Height, "blockPartsHeader", blockMeta.PartsHeader, "peerBlockPartsHeader", prs.ProposalBlockPartsHeader)
					time.Sleep(peerGossipSleepDuration)
					continue OUTER_LOOP
				}
				msg := &csspb.BlockPartMessage{
					Height: prs.Height,  
					Round:  prs.Round,   
					Part:   part,
				}
				peer.SendBytes(DataChannel, csspb.MarshalDataToCssMsg(msg))
				ps.SetHasProposalBlockPart(prs.Height, prs.Round, index)
				continue OUTER_LOOP
			} else {
				time.Sleep(peerGossipSleepDuration)
				continue OUTER_LOOP
			}
		}
		if (rs.Height != prs.Height) || (rs.Round != prs.Round) {
			time.Sleep(peerGossipSleepDuration)
			continue OUTER_LOOP
		}
		if rs.Proposal != nil && !prs.Proposal {
			{
				msg := &csspb.ProposalMessage{Proposal: rs.Proposal}
				peer.SendBytes(DataChannel, csspb.MarshalDataToCssMsg(msg))
				ps.SetHasProposal(rs.Proposal)
			}
			pPOLRound := rs.Proposal.GetData().GetPOLRound()
			if 0 <= pPOLRound {
				msg := &csspb.ProposalPOLMessage{
					Height:           rs.Height,
					ProposalPOLRound: pPOLRound,
					ProposalPOL:      csspb.TransferBitArray(rs.Votes.Prevotes(pPOLRound).BitArray()),
				}
				peer.SendBytes(DataChannel, csspb.MarshalDataToCssMsg(msg))
			}
			continue OUTER_LOOP
		}
		time.Sleep(peerGossipSleepDuration)
		continue OUTER_LOOP
	}
}
func (conR *ConsensusReactor) gossipVotesRoutine(peer *p2p.Peer, ps *PeerState) {
OUTER_LOOP:
	for {
		if !peer.IsRunning() || !conR.IsRunning() {
			conR.slogger.Infof("Stopping gossipVotesRoutine for %v.", peer)
			return
		}
		rs := conR.conS.GetRoundState()
		if conR.gossipVotes(rs, ps) {
			continue OUTER_LOOP
		}
		time.Sleep(peerGossipSleepDuration)
		continue OUTER_LOOP
	}
}
func (conR *ConsensusReactor) gossipVotes(rs *RoundState, ps *PeerState) bool {
	prs := ps.GetRoundState()
	if rs.Height == prs.Height {
		if prs.Step == csspb.RoundStepType_NewHeight {
			if ps.PickSendVote(rs.LastCommit) {
				return true
			}
		}
		if prs.Step <= csspb.RoundStepType_Prevote && prs.Round != -1 && prs.Round <= rs.Round {
			if ps.PickSendVote(rs.Votes.Prevotes(prs.Round)) {
				return true
			}
		}
		if prs.Step <= csspb.RoundStepType_Precommit && prs.Round != -1 && prs.Round <= rs.Round {
			if ps.PickSendVote(rs.Votes.Precommits(prs.Round)) {
				return true
			}
		}
		if prs.ProposalPOLRound != -1 {
			if polPrevotes := rs.Votes.Prevotes(prs.ProposalPOLRound); polPrevotes != nil {
				if ps.PickSendVote(polPrevotes) {
					return true
				}
			}
		}
	}
	if prs.Height != 0 && rs.Height == prs.Height+1 {
		if ps.PickSendVote(rs.LastCommit) {
			return true
		}
	}
	if prs.Height != 0 && rs.Height >= prs.Height+2 {
		commit := conR.conS.blockStore.LoadBlockCommit(prs.Height)
		conR.slogger.Debug("Loaded BlockCommit for catch-up", "height", prs.Height, "commit", commit)
		if ps.PickSendVote(commit) {
			return true
		}
	}
	return false
}