You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
 
 
 
 
 
 

429 lines
11 KiB

package consensus
import (
"errors"
"sync"
"time"
. "github.com/tendermint/tendermint/binary"
. "github.com/tendermint/tendermint/blocks"
. "github.com/tendermint/tendermint/common"
. "github.com/tendermint/tendermint/mempool"
. "github.com/tendermint/tendermint/state"
)
const (
RoundStepStart = uint8(0x00) // Round started.
RoundStepPropose = uint8(0x01) // Did propose, broadcasting proposal.
RoundStepVote = uint8(0x02) // Did vote, broadcasting votes.
RoundStepPrecommit = uint8(0x03) // Did precommit, broadcasting precommits.
RoundStepCommit = uint8(0x04) // We committed at this round -- do not progress to the next round.
)
var (
ErrInvalidProposalSignature = errors.New("Error invalid proposal signature")
consensusStateKey = []byte("consensusState")
)
// Immutable when returned from ConsensusState.GetRoundState()
type RoundState struct {
Height uint32 // Height we are working on
Round uint16
Step uint8
StartTime time.Time
Validators *ValidatorSet
Proposer *Validator
Proposal *Proposal
ProposalBlock *Block
ProposalBlockPartSet *PartSet
ProposalPOL *POL
ProposalPOLPartSet *PartSet
LockedBlock *Block
LockedPOL *POL
Votes *VoteSet
Precommits *VoteSet
Commits *VoteSet
PrivValidator *PrivValidator
}
//-------------------------------------
// Tracks consensus state across block heights and rounds.
type ConsensusState struct {
blockStore *BlockStore
mempool *Mempool
mtx sync.Mutex
RoundState
state *State // State until height-1.
stagedBlock *Block // Cache last staged block.
stagedState *State // Cache result of staged block.
}
func NewConsensusState(state *State, blockStore *BlockStore, mempool *Mempool) *ConsensusState {
cs := &ConsensusState{
blockStore: blockStore,
mempool: mempool,
}
cs.updateToState(state)
return cs
}
func (cs *ConsensusState) GetRoundState() *RoundState {
cs.mtx.Lock()
defer cs.mtx.Unlock()
rs := cs.RoundState // copy
return &rs
}
func (cs *ConsensusState) updateToState(state *State) {
// Sanity check state.
stateHeight := state.Height()
if stateHeight > 0 && stateHeight != cs.Height+1 {
Panicf("updateToState() expected state height of %v but found %v", cs.Height+1, stateHeight)
}
// Reset fields based on state.
height := state.Height()
validators := state.Validators()
cs.Height = height
cs.Round = 0
cs.Step = RoundStepStart
cs.StartTime = state.CommitTime().Add(newBlockWaitDuration)
cs.Validators = validators
cs.Proposer = validators.GetProposer()
cs.Proposal = nil
cs.ProposalBlock = nil
cs.ProposalBlockPartSet = nil
cs.ProposalPOL = nil
cs.ProposalPOLPartSet = nil
cs.LockedBlock = nil
cs.LockedPOL = nil
cs.Votes = NewVoteSet(height, 0, VoteTypeBare, validators)
cs.Precommits = NewVoteSet(height, 0, VoteTypePrecommit, validators)
cs.Commits = NewVoteSet(height, 0, VoteTypeCommit, validators)
cs.state = state
cs.stagedBlock = nil
cs.stagedState = nil
// Update the round if we need to.
round := calcRound(cs.StartTime)
if round > 0 {
cs.setupRound(round)
}
}
func (cs *ConsensusState) SetupRound(round uint16) {
cs.mtx.Lock()
defer cs.mtx.Unlock()
if cs.Round >= round {
Panicf("ConsensusState round %v not lower than desired round %v", cs.Round, round)
}
cs.setupRound(round)
}
func (cs *ConsensusState) setupRound(round uint16) {
// Increment all the way to round.
validators := cs.Validators.Copy()
for r := cs.Round; r < round; r++ {
validators.IncrementAccum()
}
cs.Round = round
cs.Step = RoundStepStart
cs.Validators = validators
cs.Proposer = validators.GetProposer()
cs.Proposal = nil
cs.ProposalBlock = nil
cs.ProposalBlockPartSet = nil
cs.ProposalPOL = nil
cs.ProposalPOLPartSet = nil
cs.Votes = NewVoteSet(cs.Height, round, VoteTypeBare, validators)
cs.Votes.AddVotesFromCommits(cs.Commits)
cs.Precommits = NewVoteSet(cs.Height, round, VoteTypePrecommit, validators)
cs.Precommits.AddVotesFromCommits(cs.Commits)
}
func (cs *ConsensusState) SetStep(step byte) {
cs.mtx.Lock()
defer cs.mtx.Unlock()
if cs.Step < step {
cs.Step = step
} else {
panic("step regression")
}
}
func (cs *ConsensusState) SetPrivValidator(priv *PrivValidator) {
cs.mtx.Lock()
defer cs.mtx.Unlock()
cs.PrivValidator = priv
}
func (cs *ConsensusState) SetProposal(proposal *Proposal) error {
cs.mtx.Lock()
defer cs.mtx.Unlock()
// Already have one
if cs.Proposal != nil {
return nil
}
// Invalid.
if proposal.Height != cs.Height || proposal.Round != cs.Round {
return nil
}
// Verify signature
if !cs.Proposer.Verify(proposal.GenDocument(), proposal.Signature) {
return ErrInvalidProposalSignature
}
cs.Proposal = proposal
cs.ProposalBlockPartSet = NewPartSetFromMetadata(proposal.BlockPartsTotal, proposal.BlockPartsHash)
cs.ProposalPOLPartSet = NewPartSetFromMetadata(proposal.POLPartsTotal, proposal.POLPartsHash)
return nil
}
func (cs *ConsensusState) MakeProposal() {
cs.mtx.Lock()
defer cs.mtx.Unlock()
if cs.PrivValidator == nil || cs.Proposer.Id != cs.PrivValidator.Id {
return
}
var block *Block
var blockPartSet *PartSet
var pol *POL
var polPartSet *PartSet
// Decide on block and POL
if cs.LockedBlock != nil {
// If we're locked onto a block, just choose that.
block = cs.LockedBlock
pol = cs.LockedPOL
} else {
// TODO: make use of state returned from MakeProposalBlock()
block, _ = cs.mempool.MakeProposalBlock()
pol = cs.LockedPOL // If exists, is a PoUnlock.
}
blockPartSet = NewPartSetFromData(BinaryBytes(block))
if pol != nil {
polPartSet = NewPartSetFromData(BinaryBytes(pol))
}
// Make proposal
proposal := NewProposal(cs.Height, cs.Round, blockPartSet.Total(), blockPartSet.RootHash(),
polPartSet.Total(), polPartSet.RootHash())
cs.PrivValidator.SignProposal(proposal)
// Set fields
cs.Proposal = proposal
cs.ProposalBlock = block
cs.ProposalBlockPartSet = blockPartSet
cs.ProposalPOL = pol
cs.ProposalPOLPartSet = polPartSet
}
// NOTE: block is not necessarily valid.
func (cs *ConsensusState) AddProposalBlockPart(height uint32, round uint16, part *Part) (added bool, err error) {
cs.mtx.Lock()
defer cs.mtx.Unlock()
// Blocks might be reused, so round mismatch is OK
if cs.Height != height {
return false, nil
}
// We're not expecting a block part.
if cs.ProposalBlockPartSet != nil {
return false, nil // TODO: bad peer? Return error?
}
added, err = cs.ProposalBlockPartSet.AddPart(part)
if err != nil {
return added, err
}
if added && cs.ProposalBlockPartSet.IsComplete() {
var n int64
var err error
cs.ProposalBlock = ReadBlock(cs.ProposalBlockPartSet.GetReader(), &n, &err)
return true, err
}
return true, nil
}
// NOTE: POL is not necessarily valid.
func (cs *ConsensusState) AddProposalPOLPart(height uint32, round uint16, part *Part) (added bool, err error) {
cs.mtx.Lock()
defer cs.mtx.Unlock()
if cs.Height != height || cs.Round != round {
return false, nil
}
// We're not expecting a POL part.
if cs.ProposalPOLPartSet != nil {
return false, nil // TODO: bad peer? Return error?
}
added, err = cs.ProposalPOLPartSet.AddPart(part)
if err != nil {
return added, err
}
if added && cs.ProposalPOLPartSet.IsComplete() {
var n int64
var err error
cs.ProposalPOL = ReadPOL(cs.ProposalPOLPartSet.GetReader(), &n, &err)
return true, err
}
return true, nil
}
func (cs *ConsensusState) AddVote(vote *Vote) (added bool, err error) {
switch vote.Type {
case VoteTypeBare:
// Votes checks for height+round match.
return cs.Votes.AddVote(vote)
case VoteTypePrecommit:
// Precommits checks for height+round match.
return cs.Precommits.AddVote(vote)
case VoteTypeCommit:
// Commits checks for height match.
cs.Votes.AddVote(vote)
cs.Precommits.AddVote(vote)
return cs.Commits.AddVote(vote)
default:
panic("Unknown vote type")
}
}
// Lock the ProposalBlock if we have enough votes for it,
// or unlock an existing lock if +2/3 of votes were nil.
// Returns a blockhash if a block was locked.
func (cs *ConsensusState) LockOrUnlock(height uint32, round uint16) []byte {
cs.mtx.Lock()
defer cs.mtx.Unlock()
if cs.Height != height || cs.Round != round {
return nil
}
if hash, _, ok := cs.Votes.TwoThirdsMajority(); ok {
// Remember this POL. (hash may be nil)
cs.LockedPOL = cs.Votes.MakePOL()
if len(hash) == 0 {
// +2/3 voted nil. Just unlock.
cs.LockedBlock = nil
return nil
} else if cs.ProposalBlock.HashesTo(hash) {
// +2/3 voted for proposal block
// Validate the block.
// See note on ZombieValidators to see why.
if cs.stageBlock(cs.ProposalBlock) != nil {
log.Warning("+2/3 voted for an invalid block.")
return nil
}
cs.LockedBlock = cs.ProposalBlock
return hash
} else if cs.LockedBlock.HashesTo(hash) {
// +2/3 voted for already locked block
// cs.LockedBlock = cs.LockedBlock
return hash
} else {
// We don't have the block that hashes to hash.
// Unlock if we're locked.
cs.LockedBlock = nil
return nil
}
} else {
return nil
}
}
func (cs *ConsensusState) Commit(height uint32, round uint16) *Block {
cs.mtx.Lock()
defer cs.mtx.Unlock()
if cs.Height != height || cs.Round != round {
return nil
}
if hash, commitTime, ok := cs.Precommits.TwoThirdsMajority(); ok {
// There are some strange cases that shouldn't happen
// (unless voters are duplicitous).
// For example, the hash may not be the one that was
// proposed this round. These cases should be identified
// and warn the administrator. We should err on the side of
// caution and not, for example, sign a block.
// TODO: Identify these strange cases.
var block *Block
if cs.LockedBlock.HashesTo(hash) {
block = cs.LockedBlock
} else if cs.ProposalBlock.HashesTo(hash) {
block = cs.ProposalBlock
} else {
return nil
}
// The proposal must be valid.
if err := cs.stageBlock(block); err != nil {
log.Warning("Network is commiting an invalid proposal? %v", err)
return nil
}
// Save to blockStore
err := cs.blockStore.SaveBlock(block)
if err != nil {
return nil
}
// What was staged becomes committed.
state := cs.stagedState
state.Save(commitTime)
cs.updateToState(state)
// Update mempool.
cs.mempool.ResetForBlockAndState(block, state)
return block
}
return nil
}
func (cs *ConsensusState) stageBlock(block *Block) error {
// Already staged?
if cs.stagedBlock == block {
return nil
}
// Basic validation is done in state.CommitBlock().
//err := block.ValidateBasic()
//if err != nil {
// return err
//}
// Create a copy of the state for staging
stateCopy := cs.state.Copy() // Deep copy the state before staging.
// Commit block onto the copied state.
err := stateCopy.AppendBlock(block)
if err != nil {
return err
} else {
cs.stagedBlock = block
cs.stagedState = stateCopy
return nil
}
}