fluencelabs/tendermint
1
0
Fork 0
mirror of https://github.com/fluencelabs/tendermint synced 2025-06-20 00:21:21 +00:00
Code Issues Projects Releases Wiki Activity

fix tests

Browse Source
This commit is contained in:
Jae Kwon
2014-10-30 03:32:09 -07:00
parent f6093caac9
commit e42771e36f
20 changed files with 1045 additions and 794 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
README.md
View File

@ -11,7 +11,8 @@ TenderMint - proof of concept
### Development Status ### Development Status
* Testnet *pending* * Testnet *pending*
* Bootstrapping *now* * Blockchain catchup *now*
* Bootstrapping *complete*
* Mempool *complete* * Mempool *complete*
* Consensus *complete* * Consensus *complete*
* Block propagation *sidelined* * Block propagation *sidelined*

34
blocks/block.go
View File

@ -55,6 +55,7 @@ func (b *Block) ValidateBasic(lastBlockHeight uint32, lastBlockHash []byte) erro
if !bytes.Equal(b.Header.LastBlockHash, lastBlockHash) { if !bytes.Equal(b.Header.LastBlockHash, lastBlockHash) {
return ErrBlockInvalidLastBlockHash return ErrBlockInvalidLastBlockHash
} }
// XXX We need to validate LastBlockParts too.
// XXX more validation // XXX more validation
return nil return nil
} }
@ -117,6 +118,7 @@ type Header struct {
Time time.Time Time time.Time
Fees uint64 Fees uint64
LastBlockHash []byte LastBlockHash []byte
LastBlockParts PartSetHeader
StateHash []byte StateHash []byte
// Volatile // Volatile
@ -133,6 +135,7 @@ func ReadHeader(r io.Reader, n *int64, err *error) (h Header) {
Time: ReadTime(r, n, err), Time: ReadTime(r, n, err),
Fees: ReadUInt64(r, n, err), Fees: ReadUInt64(r, n, err),
LastBlockHash: ReadByteSlice(r, n, err), LastBlockHash: ReadByteSlice(r, n, err),
LastBlockParts: ReadPartSetHeader(r, n, err),
StateHash: ReadByteSlice(r, n, err), StateHash: ReadByteSlice(r, n, err),
} }
} }
@ -143,6 +146,7 @@ func (h *Header) WriteTo(w io.Writer) (n int64, err error) {
WriteTime(w, h.Time, &n, &err) WriteTime(w, h.Time, &n, &err)
WriteUInt64(w, h.Fees, &n, &err) WriteUInt64(w, h.Fees, &n, &err)
WriteByteSlice(w, h.LastBlockHash, &n, &err) WriteByteSlice(w, h.LastBlockHash, &n, &err)
WriteBinary(w, h.LastBlockParts, &n, &err)
WriteByteSlice(w, h.StateHash, &n, &err) WriteByteSlice(w, h.StateHash, &n, &err)
return return
} }
@ -166,6 +170,7 @@ func (h *Header) StringWithIndent(indent string) string {
%s Time: %v %s Time: %v
%s Fees: %v %s Fees: %v
%s LastBlockHash: %X %s LastBlockHash: %X
%s LastBlockParts: %v
%s StateHash: %X %s StateHash: %X
%s}#%X`, %s}#%X`,
indent, h.Network, indent, h.Network,
@ -173,6 +178,7 @@ func (h *Header) StringWithIndent(indent string) string {
indent, h.Time, indent, h.Time,
indent, h.Fees, indent, h.Fees,
indent, h.LastBlockHash, indent, h.LastBlockHash,
indent, h.LastBlockParts,
indent, h.StateHash, indent, h.StateHash,
indent, h.hash) indent, h.hash)
} }
@ -180,36 +186,28 @@ func (h *Header) StringWithIndent(indent string) string {
//----------------------------------------------------------------------------- //-----------------------------------------------------------------------------
type Validation struct { type Validation struct {
Signatures []Signature Commits []RoundSignature
// Volatile // Volatile
hash []byte hash []byte
} }
func ReadValidation(r io.Reader, n *int64, err *error) Validation { func ReadValidation(r io.Reader, n *int64, err *error) Validation {
numSigs := ReadUInt32(r, n, err)
sigs := make([]Signature, 0, numSigs)
for i := uint32(0); i < numSigs; i++ {
sigs = append(sigs, ReadSignature(r, n, err))
}
return Validation{ return Validation{
Signatures: sigs, Commits: ReadRoundSignatures(r, n, err),
} }
} }
func (v *Validation) WriteTo(w io.Writer) (n int64, err error) { func (v *Validation) WriteTo(w io.Writer) (n int64, err error) {
WriteUInt32(w, uint32(len(v.Signatures)), &n, &err) WriteRoundSignatures(w, v.Commits, &n, &err)
for _, sig := range v.Signatures {
WriteBinary(w, sig, &n, &err)
}
return return
} }
func (v *Validation) Hash() []byte { func (v *Validation) Hash() []byte {
if v.hash == nil { if v.hash == nil {
bs := make([]Binary, len(v.Signatures)) bs := make([]Binary, len(v.Commits))
for i, sig := range v.Signatures { for i, commit := range v.Commits {
bs[i] = Binary(sig) bs[i] = Binary(commit)
} }
v.hash = merkle.HashFromBinaries(bs) v.hash = merkle.HashFromBinaries(bs)
} }
@ -217,14 +215,14 @@ func (v *Validation) Hash() []byte {
} }
func (v *Validation) StringWithIndent(indent string) string { func (v *Validation) StringWithIndent(indent string) string {
sigStrings := make([]string, len(v.Signatures)) commitStrings := make([]string, len(v.Commits))
for i, sig := range v.Signatures { for i, commit := range v.Commits {
sigStrings[i] = sig.String() commitStrings[i] = commit.String()
} }
return fmt.Sprintf(`Validation{ return fmt.Sprintf(`Validation{
%s %v %s %v
%s}#%X`, %s}#%X`,
indent, strings.Join(sigStrings, "\n"+indent+" "), indent, strings.Join(commitStrings, "\n"+indent+" "),
indent, v.hash) indent, v.hash)
} }

11
blocks/block_test.go
View File

@ -11,6 +11,10 @@ func randSig() Signature {
return Signature{RandUInt64Exp(), RandBytes(32)} return Signature{RandUInt64Exp(), RandBytes(32)}
} }
func randRoundSig() RoundSignature {
return RoundSignature{RandUInt16(), randSig()}
}
func randBaseTx() BaseTx { func randBaseTx() BaseTx {
return BaseTx{0, RandUInt64Exp(), randSig()} return BaseTx{0, RandUInt64Exp(), randSig()}
} }
@ -64,7 +68,7 @@ func randBlock() *Block {
StateHash: RandBytes(32), StateHash: RandBytes(32),
}, },
Validation: Validation{ Validation: Validation{
Signatures: []Signature{randSig(), randSig()}, Commits: []RoundSignature{randRoundSig(), randRoundSig()},
}, },
Data: Data{ Data: Data{
Txs: []Tx{sendTx, nameTx, bondTx, unbondTx, dupeoutTx}, Txs: []Tx{sendTx, nameTx, bondTx, unbondTx, dupeoutTx},
@ -99,8 +103,9 @@ func TestBlock(t *testing.T) {
expectChange(func(b *Block) { b.Header.Time = RandTime() }, "Expected hash to depend on Time") expectChange(func(b *Block) { b.Header.Time = RandTime() }, "Expected hash to depend on Time")
expectChange(func(b *Block) { b.Header.LastBlockHash = RandBytes(32) }, "Expected hash to depend on LastBlockHash") expectChange(func(b *Block) { b.Header.LastBlockHash = RandBytes(32) }, "Expected hash to depend on LastBlockHash")
expectChange(func(b *Block) { b.Header.StateHash = RandBytes(32) }, "Expected hash to depend on StateHash") expectChange(func(b *Block) { b.Header.StateHash = RandBytes(32) }, "Expected hash to depend on StateHash")
expectChange(func(b *Block) { b.Validation.Signatures[0].SignerId += 1 }, "Expected hash to depend on Validation Signature") expectChange(func(b *Block) { b.Validation.Commits[0].Round += 1 }, "Expected hash to depend on Validation Commit")
expectChange(func(b *Block) { b.Validation.Signatures[0].Bytes = RandBytes(32) }, "Expected hash to depend on Validation Signature") expectChange(func(b *Block) { b.Validation.Commits[0].SignerId += 1 }, "Expected hash to depend on Validation Commit")
expectChange(func(b *Block) { b.Validation.Commits[0].Bytes = RandBytes(32) }, "Expected hash to depend on Validation Commit")
expectChange(func(b *Block) { b.Data.Txs[0].(*SendTx).Signature.SignerId += 1 }, "Expected hash to depend on tx Signature") expectChange(func(b *Block) { b.Data.Txs[0].(*SendTx).Signature.SignerId += 1 }, "Expected hash to depend on tx Signature")
expectChange(func(b *Block) { b.Data.Txs[0].(*SendTx).Amount += 1 }, "Expected hash to depend on send tx Amount") expectChange(func(b *Block) { b.Data.Txs[0].(*SendTx).Amount += 1 }, "Expected hash to depend on send tx Amount")

104
consensus/part_set.go → blocks/part_set.go
View File

@ -1,4 +1,4 @@
package consensus package blocks
import ( import (
"bytes" "bytes"
@ -40,34 +40,34 @@ func ReadPart(r io.Reader, n *int64, err *error) *Part {
} }
} }
func (b *Part) WriteTo(w io.Writer) (n int64, err error) { func (part *Part) WriteTo(w io.Writer) (n int64, err error) {
WriteUInt16(w, b.Index, &n, &err) WriteUInt16(w, part.Index, &n, &err)
WriteByteSlices(w, b.Trail, &n, &err) WriteByteSlices(w, part.Trail, &n, &err)
WriteByteSlice(w, b.Bytes, &n, &err) WriteByteSlice(w, part.Bytes, &n, &err)
return return
} }
func (pt *Part) Hash() []byte { func (part *Part) Hash() []byte {
if pt.hash != nil { if part.hash != nil {
return pt.hash return part.hash
} else { } else {
hasher := sha256.New() hasher := sha256.New()
_, err := hasher.Write(pt.Bytes) _, err := hasher.Write(part.Bytes)
if err != nil { if err != nil {
panic(err) panic(err)
} }
pt.hash = hasher.Sum(nil) part.hash = hasher.Sum(nil)
return pt.hash return part.hash
} }
} }
func (pt *Part) String() string { func (part *Part) String() string {
return pt.StringWithIndent("") return part.StringWithIndent("")
} }
func (pt *Part) StringWithIndent(indent string) string { func (part *Part) StringWithIndent(indent string) string {
trailStrings := make([]string, len(pt.Trail)) trailStrings := make([]string, len(part.Trail))
for i, hash := range pt.Trail { for i, hash := range part.Trail {
trailStrings[i] = fmt.Sprintf("%X", hash) trailStrings[i] = fmt.Sprintf("%X", hash)
} }
return fmt.Sprintf(`Part{ return fmt.Sprintf(`Part{
@ -75,7 +75,7 @@ func (pt *Part) StringWithIndent(indent string) string {
%s Trail: %s Trail:
%s %v %s %v
%s}`, %s}`,
indent, pt.Index, indent, part.Index,
indent, indent,
indent, strings.Join(trailStrings, "\n"+indent+" "), indent, strings.Join(trailStrings, "\n"+indent+" "),
indent) indent)
@ -83,8 +83,41 @@ func (pt *Part) StringWithIndent(indent string) string {
//------------------------------------- //-------------------------------------
type PartSetHeader struct {
Hash []byte
Total uint16
}
func ReadPartSetHeader(r io.Reader, n *int64, err *error) PartSetHeader {
return PartSetHeader{
Hash: ReadByteSlice(r, n, err),
Total: ReadUInt16(r, n, err),
}
}
func (psh PartSetHeader) WriteTo(w io.Writer) (n int64, err error) {
WriteByteSlice(w, psh.Hash, &n, &err)
WriteUInt16(w, psh.Total, &n, &err)
return
}
func (psh PartSetHeader) IsZero() bool {
return psh.Total == 0
}
func (psh PartSetHeader) String() string {
return fmt.Sprintf("PartSet{%X/%v}", psh.Hash, psh.Total)
}
func (psh PartSetHeader) Equals(other PartSetHeader) bool {
return bytes.Equal(psh.Hash, other.Hash) &&
psh.Total == other.Total
}
//-------------------------------------
type PartSet struct { type PartSet struct {
rootHash []byte hash []byte
total uint16 total uint16
mtx sync.Mutex mtx sync.Mutex
@ -115,43 +148,54 @@ func NewPartSetFromData(data []byte) *PartSet {
parts[i].Trail = merkle.HashTrailForIndex(hashTree, i) parts[i].Trail = merkle.HashTrailForIndex(hashTree, i)
} }
return &PartSet{ return &PartSet{
hash: hashTree[len(hashTree)/2],
total: uint16(total),
parts: parts, parts: parts,
partsBitArray: partsBitArray, partsBitArray: partsBitArray,
rootHash: hashTree[len(hashTree)/2],
total: uint16(total),
count: uint16(total), count: uint16(total),
} }
} }
// Returns an empty PartSet ready to be populated. // Returns an empty PartSet ready to be populated.
func NewPartSetFromMetadata(total uint16, rootHash []byte) *PartSet { func NewPartSetFromHeader(header PartSetHeader) *PartSet {
return &PartSet{ return &PartSet{
parts: make([]*Part, total), hash: header.Hash,
partsBitArray: NewBitArray(uint(total)), total: header.Total,
rootHash: rootHash, parts: make([]*Part, header.Total),
total: total, partsBitArray: NewBitArray(uint(header.Total)),
count: 0, count: 0,
} }
} }
func (ps *PartSet) Header() PartSetHeader {
if ps == nil {
return PartSetHeader{}
} else {
return PartSetHeader{
Hash: ps.hash,
Total: ps.total,
}
}
}
func (ps *PartSet) BitArray() BitArray { func (ps *PartSet) BitArray() BitArray {
ps.mtx.Lock() ps.mtx.Lock()
defer ps.mtx.Unlock() defer ps.mtx.Unlock()
return ps.partsBitArray.Copy() return ps.partsBitArray.Copy()
} }
func (ps *PartSet) RootHash() []byte { func (ps *PartSet) Hash() []byte {
if ps == nil { if ps == nil {
return nil return nil
} }
return ps.rootHash return ps.hash
} }
func (ps *PartSet) HashesTo(rootHash []byte) bool { func (ps *PartSet) HashesTo(hash []byte) bool {
if ps == nil { if ps == nil {
return false return false
} }
return bytes.Equal(ps.rootHash, rootHash) return bytes.Equal(ps.hash, hash)
} }
func (ps *PartSet) Count() uint16 { func (ps *PartSet) Count() uint16 {
@ -183,7 +227,7 @@ func (ps *PartSet) AddPart(part *Part) (bool, error) {
} }
// Check hash trail // Check hash trail
if !merkle.VerifyHashTrailForIndex(int(part.Index), part.Hash(), part.Trail, ps.rootHash) { if !merkle.VerifyHashTrailForIndex(int(part.Index), part.Hash(), part.Trail, ps.hash) {
return false, ErrPartSetInvalidTrail return false, ErrPartSetInvalidTrail
} }

14
consensus/part_set_test.go → blocks/part_set_test.go
View File

@ -1,4 +1,4 @@
package consensus package blocks
import ( import (
"bytes" "bytes"
@ -14,8 +14,8 @@ func TestBasicPartSet(t *testing.T) {
data := RandBytes(partSize * 100) data := RandBytes(partSize * 100)
partSet := NewPartSetFromData(data) partSet := NewPartSetFromData(data)
if len(partSet.RootHash()) == 0 { if len(partSet.Hash()) == 0 {
t.Error("Expected to get rootHash") t.Error("Expected to get hash")
} }
if partSet.Total() != 100 { if partSet.Total() != 100 {
t.Errorf("Expected to get 100 parts, but got %v", partSet.Total()) t.Errorf("Expected to get 100 parts, but got %v", partSet.Total())
@ -25,7 +25,7 @@ func TestBasicPartSet(t *testing.T) {
} }
// Test adding parts to a new partSet. // Test adding parts to a new partSet.
partSet2 := NewPartSetFromMetadata(partSet.Total(), partSet.RootHash()) partSet2 := NewPartSetFromHeader(PartSetHeader{partSet.Hash(), partSet.Total()})
for i := uint16(0); i < partSet.Total(); i++ { for i := uint16(0); i < partSet.Total(); i++ {
part := partSet.GetPart(i) part := partSet.GetPart(i)
@ -36,8 +36,8 @@ func TestBasicPartSet(t *testing.T) {
} }
} }
if !bytes.Equal(partSet.RootHash(), partSet2.RootHash()) { if !bytes.Equal(partSet.Hash(), partSet2.Hash()) {
t.Error("Expected to get same rootHash") t.Error("Expected to get same hash")
} }
if partSet2.Total() != 100 { if partSet2.Total() != 100 {
t.Errorf("Expected to get 100 parts, but got %v", partSet2.Total()) t.Errorf("Expected to get 100 parts, but got %v", partSet2.Total())
@ -65,7 +65,7 @@ func TestWrongTrail(t *testing.T) {
partSet := NewPartSetFromData(data) partSet := NewPartSetFromData(data)
// Test adding a part with wrong data. // Test adding a part with wrong data.
partSet2 := NewPartSetFromMetadata(partSet.Total(), partSet.RootHash()) partSet2 := NewPartSetFromHeader(PartSetHeader{partSet.Hash(), partSet.Total()})
// Test adding a part with wrong trail. // Test adding a part with wrong trail.
part := partSet.GetPart(0) part := partSet.GetPart(0)

44
blocks/signature.go
View File

@ -60,3 +60,47 @@ func WriteSignatures(w io.Writer, sigs []Signature, n *int64, err *error) {
} }
} }
} }
//-----------------------------------------------------------------------------
type RoundSignature struct {
Round uint16
Signature
}
func ReadRoundSignature(r io.Reader, n *int64, err *error) RoundSignature {
return RoundSignature{
ReadUInt16(r, n, err),
ReadSignature(r, n, err),
}
}
func (rsig RoundSignature) WriteTo(w io.Writer) (n int64, err error) {
WriteUInt16(w, rsig.Round, &n, &err)
WriteBinary(w, rsig.Signature, &n, &err)
return
}
func (rsig RoundSignature) IsZero() bool {
return rsig.Round == 0 && rsig.SignerId == 0 && len(rsig.Bytes) == 0
}
//-------------------------------------
func ReadRoundSignatures(r io.Reader, n *int64, err *error) (rsigs []RoundSignature) {
length := ReadUInt32(r, n, err)
for i := uint32(0); i < length; i++ {
rsigs = append(rsigs, ReadRoundSignature(r, n, err))
}
return
}
func WriteRoundSignatures(w io.Writer, rsigs []RoundSignature, n *int64, err *error) {
WriteUInt32(w, uint32(len(rsigs)), n, err)
for _, rsig := range rsigs {
WriteBinary(w, rsig, n, err)
if *err != nil {
return
}
}
}

17
blocks/vote.go
View File

@ -6,6 +6,7 @@ import (
"io" "io"
. "github.com/tendermint/tendermint/binary" . "github.com/tendermint/tendermint/binary"
. "github.com/tendermint/tendermint/common"
) )
const ( const (
@ -28,6 +29,7 @@ type Vote struct {
Round uint16 Round uint16
Type byte Type byte
BlockHash []byte // empty if vote is nil. BlockHash []byte // empty if vote is nil.
BlockParts PartSetHeader // zero if vote is nil.
Signature Signature
} }
@ -37,6 +39,7 @@ func ReadVote(r io.Reader, n *int64, err *error) *Vote {
Round: ReadUInt16(r, n, err), Round: ReadUInt16(r, n, err),
Type: ReadByte(r, n, err), Type: ReadByte(r, n, err),
BlockHash: ReadByteSlice(r, n, err), BlockHash: ReadByteSlice(r, n, err),
BlockParts: ReadPartSetHeader(r, n, err),
Signature: ReadSignature(r, n, err), Signature: ReadSignature(r, n, err),
} }
} }
@ -46,6 +49,7 @@ func (v *Vote) WriteTo(w io.Writer) (n int64, err error) {
WriteUInt16(w, v.Round, &n, &err) WriteUInt16(w, v.Round, &n, &err)
WriteByte(w, v.Type, &n, &err) WriteByte(w, v.Type, &n, &err)
WriteByteSlice(w, v.BlockHash, &n, &err) WriteByteSlice(w, v.BlockHash, &n, &err)
WriteBinary(w, v.BlockParts, &n, &err)
WriteBinary(w, v.Signature, &n, &err) WriteBinary(w, v.Signature, &n, &err)
return return
} }
@ -59,18 +63,17 @@ func (v *Vote) SetSignature(sig Signature) {
} }
func (v *Vote) String() string { func (v *Vote) String() string {
blockHash := v.BlockHash var typeString string
if len(v.BlockHash) == 0 {
blockHash = make([]byte, 6) // for printing
}
switch v.Type { switch v.Type {
case VoteTypePrevote: case VoteTypePrevote:
return fmt.Sprintf("Prevote{%v/%v:%X:%v}", v.Height, v.Round, blockHash, v.SignerId) typeString = "Prevote"
case VoteTypePrecommit: case VoteTypePrecommit:
return fmt.Sprintf("Precommit{%v/%v:%X:%v}", v.Height, v.Round, blockHash, v.SignerId) typeString = "Precommit"
case VoteTypeCommit: case VoteTypeCommit:
return fmt.Sprintf("Commit{%v/%v:%X:%v}", v.Height, v.Round, blockHash, v.SignerId) typeString = "Commit"
default: default:
panic("Unknown vote type") panic("Unknown vote type")
} }
return fmt.Sprintf("%v{%v/%v:%X:%v:%v}", typeString, v.Height, v.Round, Fingerprint(v.BlockHash), v.BlockParts, v.SignerId)
} }

8
common/bit_array.go
View File

@ -21,6 +21,9 @@ func NewBitArray(bits uint) BitArray {
func ReadBitArray(r io.Reader, n *int64, err *error) BitArray { func ReadBitArray(r io.Reader, n *int64, err *error) BitArray {
bits := ReadUVarInt(r, n, err) bits := ReadUVarInt(r, n, err)
if bits == 0 {
return BitArray{}
}
elemsWritten := ReadUVarInt(r, n, err) elemsWritten := ReadUVarInt(r, n, err)
if *err != nil { if *err != nil {
return BitArray{} return BitArray{}
@ -36,6 +39,10 @@ func ReadBitArray(r io.Reader, n *int64, err *error) BitArray {
} }
func (bA BitArray) WriteTo(w io.Writer) (n int64, err error) { func (bA BitArray) WriteTo(w io.Writer) (n int64, err error) {
WriteUVarInt(w, bA.bits, &n, &err)
if bA.bits == 0 {
return
}
// Count the last element > 0. // Count the last element > 0.
elemsToWrite := 0 elemsToWrite := 0
for i, elem := range bA.elems { for i, elem := range bA.elems {
@ -43,7 +50,6 @@ func (bA BitArray) WriteTo(w io.Writer) (n int64, err error) {
elemsToWrite = i + 1 elemsToWrite = i + 1
} }
} }
WriteUVarInt(w, bA.bits, &n, &err)
WriteUVarInt(w, uint(elemsToWrite), &n, &err) WriteUVarInt(w, uint(elemsToWrite), &n, &err)
for i, elem := range bA.elems { for i, elem := range bA.elems {
if i >= elemsToWrite { if i >= elemsToWrite {

7
common/byteslice.go Normal file
View File

@ -0,0 +1,7 @@
package common
func Fingerprint(bytez []byte) []byte {
fingerprint := make([]byte, 6)
copy(fingerprint, bytez)
return fingerprint
}

38
consensus/pol.go
View File

@ -16,9 +16,9 @@ type POL struct {
Height uint32 Height uint32
Round uint16 Round uint16
BlockHash []byte // Could be nil, which makes this a proof of unlock. BlockHash []byte // Could be nil, which makes this a proof of unlock.
BlockParts PartSetHeader // When BlockHash is nil, this is zero.
Votes []Signature // Vote signatures for height/round/hash Votes []Signature // Vote signatures for height/round/hash
Commits []Signature // Commit signatures for height/hash Commits []RoundSignature // Commit signatures for height/hash
CommitRounds []uint16 // Rounds of the commits, less than POL.Round.
} }
func ReadPOL(r io.Reader, n *int64, err *error) *POL { func ReadPOL(r io.Reader, n *int64, err *error) *POL {
@ -26,9 +26,9 @@ func ReadPOL(r io.Reader, n *int64, err *error) *POL {
Height: ReadUInt32(r, n, err), Height: ReadUInt32(r, n, err),
Round: ReadUInt16(r, n, err), Round: ReadUInt16(r, n, err),
BlockHash: ReadByteSlice(r, n, err), BlockHash: ReadByteSlice(r, n, err),
BlockParts: ReadPartSetHeader(r, n, err),
Votes: ReadSignatures(r, n, err), Votes: ReadSignatures(r, n, err),
Commits: ReadSignatures(r, n, err), Commits: ReadRoundSignatures(r, n, err),
CommitRounds: ReadUInt16s(r, n, err),
} }
} }
@ -36,9 +36,9 @@ func (pol *POL) WriteTo(w io.Writer) (n int64, err error) {
WriteUInt32(w, pol.Height, &n, &err) WriteUInt32(w, pol.Height, &n, &err)
WriteUInt16(w, pol.Round, &n, &err) WriteUInt16(w, pol.Round, &n, &err)
WriteByteSlice(w, pol.BlockHash, &n, &err) WriteByteSlice(w, pol.BlockHash, &n, &err)
WriteBinary(w, pol.BlockParts, &n, &err)
WriteSignatures(w, pol.Votes, &n, &err) WriteSignatures(w, pol.Votes, &n, &err)
WriteSignatures(w, pol.Commits, &n, &err) WriteRoundSignatures(w, pol.Commits, &n, &err)
WriteUInt16s(w, pol.CommitRounds, &n, &err)
return return
} }
@ -46,8 +46,11 @@ func (pol *POL) WriteTo(w io.Writer) (n int64, err error) {
func (pol *POL) Verify(vset *state.ValidatorSet) error { func (pol *POL) Verify(vset *state.ValidatorSet) error {
talliedVotingPower := uint64(0) talliedVotingPower := uint64(0)
voteDoc := BinaryBytes(&Vote{Height: pol.Height, Round: pol.Round, voteDoc := BinaryBytes(&Vote{
Type: VoteTypePrevote, BlockHash: pol.BlockHash}) Height: pol.Height, Round: pol.Round, Type: VoteTypePrevote,
BlockHash: pol.BlockHash,
BlockParts: pol.BlockParts,
})
seenValidators := map[uint64]struct{}{} seenValidators := map[uint64]struct{}{}
for _, sig := range pol.Votes { for _, sig := range pol.Votes {
@ -69,8 +72,9 @@ func (pol *POL) Verify(vset *state.ValidatorSet) error {
talliedVotingPower += val.VotingPower talliedVotingPower += val.VotingPower
} }
for i, sig := range pol.Commits { for _, rsig := range pol.Commits {
round := pol.CommitRounds[i] round := rsig.Round
sig := rsig.Signature
// Validate // Validate
if _, seen := seenValidators[sig.SignerId]; seen { if _, seen := seenValidators[sig.SignerId]; seen {
@ -84,8 +88,11 @@ func (pol *POL) Verify(vset *state.ValidatorSet) error {
return Errorf("Invalid commit round %v for POL %v", round, pol) return Errorf("Invalid commit round %v for POL %v", round, pol)
} }
commitDoc := BinaryBytes(&Vote{Height: pol.Height, Round: round, commitDoc := BinaryBytes(&Vote{
Type: VoteTypeCommit, BlockHash: pol.BlockHash}) // TODO cache Height: pol.Height, Round: round, Type: VoteTypeCommit,
BlockHash: pol.BlockHash,
BlockParts: pol.BlockParts,
})
if !val.VerifyBytes(commitDoc, sig) { if !val.VerifyBytes(commitDoc, sig) {
return Errorf("Invalid signature for commit %v for POL %v", sig, pol) return Errorf("Invalid signature for commit %v for POL %v", sig, pol)
} }
@ -108,10 +115,7 @@ func (pol *POL) Description() string {
if pol == nil { if pol == nil {
return "nil-POL" return "nil-POL"
} else { } else {
blockHash := pol.BlockHash return fmt.Sprintf("POL{H:%v R:%v BH:%X}", pol.Height, pol.Round,
if blockHash != nil { Fingerprint(pol.BlockHash), pol.BlockParts)
blockHash = blockHash[:6]
}
return fmt.Sprintf("POL{H:%v R:%v BH:%X}", pol.Height, pol.Round, blockHash)
} }
} }

12
consensus/pol_test.go
View File

@ -78,8 +78,7 @@ func TestVerifyCommits(t *testing.T) {
} }
for i := 0; i < 7; i++ { for i := 0; i < 7; i++ {
privAccounts[i].Sign(vote) privAccounts[i].Sign(vote)
pol.Commits = append(pol.Commits, vote.Signature) pol.Commits = append(pol.Commits, RoundSignature{1, vote.Signature})
pol.CommitRounds = append(pol.CommitRounds, 1)
} }
// Check that validation succeeds. // Check that validation succeeds.
@ -103,8 +102,7 @@ func TestVerifyInvalidCommits(t *testing.T) {
privAccounts[i].Sign(vote) privAccounts[i].Sign(vote)
// Mutate the signature. // Mutate the signature.
vote.Signature.Bytes[0] += byte(0x01) vote.Signature.Bytes[0] += byte(0x01)
pol.Commits = append(pol.Commits, vote.Signature) pol.Commits = append(pol.Commits, RoundSignature{1, vote.Signature})
pol.CommitRounds = append(pol.CommitRounds, 1)
} }
// Check that validation fails. // Check that validation fails.
@ -126,8 +124,7 @@ func TestVerifyInvalidCommitRounds(t *testing.T) {
} }
for i := 0; i < 7; i++ { for i := 0; i < 7; i++ {
privAccounts[i].Sign(vote) privAccounts[i].Sign(vote)
pol.Commits = append(pol.Commits, vote.Signature) pol.Commits = append(pol.Commits, RoundSignature{2, vote.Signature})
pol.CommitRounds = append(pol.CommitRounds, 2)
} }
// Check that validation fails. // Check that validation fails.
@ -149,8 +146,7 @@ func TestVerifyInvalidCommitRounds2(t *testing.T) {
} }
for i := 0; i < 7; i++ { for i := 0; i < 7; i++ {
privAccounts[i].Sign(vote) privAccounts[i].Sign(vote)
pol.Commits = append(pol.Commits, vote.Signature) pol.Commits = append(pol.Commits, RoundSignature{3, vote.Signature})
pol.CommitRounds = append(pol.CommitRounds, 3)
} }
// Check that validation fails. // Check that validation fails.

34
consensus/proposal.go
View File

@ -17,24 +17,18 @@ var (
type Proposal struct { type Proposal struct {
Height uint32 Height uint32
Round uint16 Round uint16
BlockPartsTotal uint16 BlockParts PartSetHeader
BlockPartsHash []byte POLParts PartSetHeader
POLPartsTotal uint16
POLPartsHash []byte
Signature Signature Signature Signature
} }
func NewProposal(height uint32, round uint16, func NewProposal(height uint32, round uint16, blockParts, polParts PartSetHeader) *Proposal {
blockPartsTotal uint16, blockPartsHash []byte,
polPartsTotal uint16, polPartsHash []byte) *Proposal {
return &Proposal{ return &Proposal{
Height: height, Height: height,
Round: round, Round: round,
BlockPartsTotal: blockPartsTotal, BlockParts: blockParts,
BlockPartsHash: blockPartsHash, POLParts: polParts,
POLPartsTotal: polPartsTotal,
POLPartsHash: polPartsHash,
} }
} }
@ -42,10 +36,8 @@ func ReadProposal(r io.Reader, n *int64, err *error) *Proposal {
return &Proposal{ return &Proposal{
Height: ReadUInt32(r, n, err), Height: ReadUInt32(r, n, err),
Round: ReadUInt16(r, n, err), Round: ReadUInt16(r, n, err),
BlockPartsTotal: ReadUInt16(r, n, err), BlockParts: ReadPartSetHeader(r, n, err),
BlockPartsHash: ReadByteSlice(r, n, err), POLParts: ReadPartSetHeader(r, n, err),
POLPartsTotal: ReadUInt16(r, n, err),
POLPartsHash: ReadByteSlice(r, n, err),
Signature: ReadSignature(r, n, err), Signature: ReadSignature(r, n, err),
} }
} }
@ -53,10 +45,8 @@ func ReadProposal(r io.Reader, n *int64, err *error) *Proposal {
func (p *Proposal) WriteTo(w io.Writer) (n int64, err error) { func (p *Proposal) WriteTo(w io.Writer) (n int64, err error) {
WriteUInt32(w, p.Height, &n, &err) WriteUInt32(w, p.Height, &n, &err)
WriteUInt16(w, p.Round, &n, &err) WriteUInt16(w, p.Round, &n, &err)
WriteUInt16(w, p.BlockPartsTotal, &n, &err) WriteBinary(w, p.BlockParts, &n, &err)
WriteByteSlice(w, p.BlockPartsHash, &n, &err) WriteBinary(w, p.POLParts, &n, &err)
WriteUInt16(w, p.POLPartsTotal, &n, &err)
WriteByteSlice(w, p.POLPartsHash, &n, &err)
WriteBinary(w, p.Signature, &n, &err) WriteBinary(w, p.Signature, &n, &err)
return return
} }
@ -70,8 +60,6 @@ func (p *Proposal) SetSignature(sig Signature) {
} }
func (p *Proposal) String() string { func (p *Proposal) String() string {
return fmt.Sprintf("Proposal{%v/%v %X/%v %X/%v %v}", p.Height, p.Round, return fmt.Sprintf("Proposal{%v/%v %v %v %v}", p.Height, p.Round,
p.BlockPartsHash, p.BlockPartsTotal, p.BlockParts, p.POLParts, p.Signature)
p.POLPartsHash, p.POLPartsTotal,
p.Signature)
} }

384
consensus/reactor.go
View File

@ -5,7 +5,6 @@ import (
"errors" "errors"
"fmt" "fmt"
"io" "io"
"math"
"sync" "sync"
"sync/atomic" "sync/atomic"
"time" "time"
@ -25,98 +24,26 @@ const (
peerStateKey = "ConsensusReactor.peerState" peerStateKey = "ConsensusReactor.peerState"
voteTypeNil = byte(0x00)
voteTypeBlock = byte(0x01)
roundDuration0 = 60 * time.Second // The first round is 60 seconds long.
roundDurationDelta = 15 * time.Second // Each successive round lasts 15 seconds longer.
roundDeadlinePrevote = float64(1.0 / 3.0) // When the prevote is due.
roundDeadlinePrecommit = float64(2.0 / 3.0) // When the precommit vote is due.
finalizeDuration = roundDuration0 / 3 // The time to wait between commitTime and startTime of next consensus rounds.
peerGossipSleepDuration = 50 * time.Millisecond // Time to sleep if there's nothing to send. peerGossipSleepDuration = 50 * time.Millisecond // Time to sleep if there's nothing to send.
hasVotesThreshold = 50 // After this many new votes we'll send a HasVotesMessage. hasVotesThreshold = 50 // After this many new votes we'll send a HasVotesMessage.
) )
//----------------------------------------------------------------------------- //-----------------------------------------------------------------------------
// total duration of given round
func calcRoundDuration(round uint16) time.Duration {
return roundDuration0 + roundDurationDelta*time.Duration(round)
}
// startTime is when round zero started.
func calcRoundStartTime(round uint16, startTime time.Time) time.Time {
return startTime.Add(roundDuration0*time.Duration(round) +
roundDurationDelta*(time.Duration((int64(round)*int64(round)-int64(round))/2)))
}
// calculates the current round given startTime of round zero.
// NOTE: round is zero if startTime is in the future.
func calcRound(startTime time.Time) uint16 {
now := time.Now()
if now.Before(startTime) {
return 0
}
// Start + D_0 * R + D_delta * (R^2 - R)/2 <= Now; find largest integer R.
// D_delta * R^2 + (2D_0 - D_delta) * R + 2(Start - Now) <= 0.
// AR^2 + BR + C <= 0; A = D_delta, B = (2_D0 - D_delta), C = 2(Start - Now).
// R = Floor((-B + Sqrt(B^2 - 4AC))/2A)
A := float64(roundDurationDelta)
B := 2.0*float64(roundDuration0) - float64(roundDurationDelta)
C := 2.0 * float64(startTime.Sub(now))
R := math.Floor((-B + math.Sqrt(B*B-4.0*A*C)) / (2 * A))
if math.IsNaN(R) {
panic("Could not calc round, should not happen")
}
if R > math.MaxInt16 {
Panicf("Could not calc round, round overflow: %v", R)
}
if R < 0 {
return 0
}
return uint16(R)
}
// convenience
// NOTE: elapsedRatio can be negative if startTime is in the future.
func calcRoundInfo(startTime time.Time) (round uint16, roundStartTime time.Time, roundDuration time.Duration,
roundElapsed time.Duration, elapsedRatio float64) {
round = calcRound(startTime)
roundStartTime = calcRoundStartTime(round, startTime)
roundDuration = calcRoundDuration(round)
roundElapsed = time.Now().Sub(roundStartTime)
elapsedRatio = float64(roundElapsed) / float64(roundDuration)
return
}
//-----------------------------------------------------------------------------
type RoundAction struct {
Height uint32 // The block height for which consensus is reaching for.
Round uint16 // The round number at given height.
Action RoundActionType // Action to perform.
}
//-----------------------------------------------------------------------------
type ConsensusReactor struct { type ConsensusReactor struct {
sw *p2p.Switch sw *p2p.Switch
quit chan struct{}
started uint32 started uint32
stopped uint32 stopped uint32
quit chan struct{}
conS *ConsensusState conS *ConsensusState
doActionCh chan RoundAction
} }
func NewConsensusReactor(blockStore *BlockStore, mempool *mempool.Mempool, state *state.State) *ConsensusReactor { func NewConsensusReactor(blockStore *BlockStore, mempool *mempool.Mempool, state *state.State) *ConsensusReactor {
conS := NewConsensusState(state, blockStore, mempool) conS := NewConsensusState(state, blockStore, mempool)
conR := &ConsensusReactor{ conR := &ConsensusReactor{
quit: make(chan struct{}), quit: make(chan struct{}),
conS: conS, conS: conS,
doActionCh: make(chan RoundAction, 1),
} }
return conR return conR
} }
@ -126,7 +53,8 @@ func (conR *ConsensusReactor) Start(sw *p2p.Switch) {
if atomic.CompareAndSwapUint32(&conR.started, 0, 1) { if atomic.CompareAndSwapUint32(&conR.started, 0, 1) {
log.Info("Starting ConsensusReactor") log.Info("Starting ConsensusReactor")
conR.sw = sw conR.sw = sw
go conR.stepTransitionRoutine() conR.conS.Start()
go conR.broadcastNewRoundStepRoutine()
} }
} }
@ -134,10 +62,15 @@ func (conR *ConsensusReactor) Start(sw *p2p.Switch) {
func (conR *ConsensusReactor) Stop() { func (conR *ConsensusReactor) Stop() {
if atomic.CompareAndSwapUint32(&conR.stopped, 0, 1) { if atomic.CompareAndSwapUint32(&conR.stopped, 0, 1) {
log.Info("Stopping ConsensusReactor") log.Info("Stopping ConsensusReactor")
conR.conS.Stop()
close(conR.quit) close(conR.quit)
} }
} }
func (conR *ConsensusReactor) IsStopped() bool {
return atomic.LoadUint32(&conR.stopped) == 1
}
// Implements Reactor // Implements Reactor
func (conR *ConsensusReactor) GetChannels() []*p2p.ChannelDescriptor { func (conR *ConsensusReactor) GetChannels() []*p2p.ChannelDescriptor {
// TODO optimize // TODO optimize
@ -238,7 +171,7 @@ func (conR *ConsensusReactor) Receive(chId byte, peer *p2p.Peer, msgBytes []byte
return return
} }
ps.EnsureVoteBitArrays(rs.Height, rs.Round, rs.Validators.Size()) ps.EnsureVoteBitArrays(rs.Height, rs.Round, rs.Validators.Size())
ps.SetHasVote(rs.Height, rs.Round, vote.Type, index) ps.SetHasVote(rs.Height, rs.Round, index, vote)
added, err := conR.conS.AddVote(vote) added, err := conR.conS.AddVote(vote)
if err != nil { if err != nil {
log.Warning("Error attempting to add vote: %v", err) log.Warning("Error attempting to add vote: %v", err)
@ -257,13 +190,6 @@ func (conR *ConsensusReactor) Receive(chId byte, peer *p2p.Peer, msgBytes []byte
} }
conR.sw.Broadcast(StateCh, msg) conR.sw.Broadcast(StateCh, msg)
} }
// Maybe run RoundActionCommitWait.
if vote.Type == VoteTypeCommit &&
rs.Commits.HasTwoThirdsMajority() &&
rs.Step < RoundStepCommitWait {
// NOTE: Do not call RunAction*() methods here directly.
conR.doActionCh <- RoundAction{rs.Height, rs.Round, RoundActionCommitWait}
}
} }
default: default:
@ -283,211 +209,36 @@ func (conR *ConsensusReactor) SetPrivValidator(priv *PrivValidator) {
conR.conS.SetPrivValidator(priv) conR.conS.SetPrivValidator(priv)
} }
func (conR *ConsensusReactor) IsStopped() bool {
return atomic.LoadUint32(&conR.stopped) == 1
}
//-------------------------------------- //--------------------------------------
// Source of all round state transitions (and votes). // XXX We need to ensure that Proposal* etc are also set appropriately.
func (conR *ConsensusReactor) stepTransitionRoutine() { // Listens for changes to the ConsensusState.Step by pulling
// on conR.conS.NewStepCh().
// Schedule the next action by pushing a RoundAction{} to conR.doActionCh func (conR *ConsensusReactor) broadcastNewRoundStepRoutine() {
// when it is due.
scheduleNextAction := func() {
rs := conR.conS.GetRoundState()
round, roundStartTime, roundDuration, _, elapsedRatio := calcRoundInfo(rs.StartTime)
log.Debug("Called scheduleNextAction. round:%v roundStartTime:%v elapsedRatio:%v",
round, roundStartTime, elapsedRatio)
go func() {
switch rs.Step {
case RoundStepStart:
// It's a new RoundState.
if elapsedRatio < 0 {
// startTime is in the future.
time.Sleep(time.Duration((-1.0 * elapsedRatio) * float64(roundDuration)))
}
conR.doActionCh <- RoundAction{rs.Height, rs.Round, RoundActionPropose}
case RoundStepPropose:
// Wake up when it's time to vote.
time.Sleep(time.Duration((roundDeadlinePrevote - elapsedRatio) * float64(roundDuration)))
conR.doActionCh <- RoundAction{rs.Height, rs.Round, RoundActionPrevote}
case RoundStepPrevote:
// Wake up when it's time to precommit.
time.Sleep(time.Duration((roundDeadlinePrecommit - elapsedRatio) * float64(roundDuration)))
conR.doActionCh <- RoundAction{rs.Height, rs.Round, RoundActionPrecommit}
case RoundStepPrecommit:
// Wake up when the round is over.
time.Sleep(time.Duration((1.0 - elapsedRatio) * float64(roundDuration)))
conR.doActionCh <- RoundAction{rs.Height, rs.Round, RoundActionTryCommit}
case RoundStepCommit:
panic("Should not happen: RoundStepCommit waits until +2/3 commits.")
case RoundStepCommitWait:
// Wake up when it's time to finalize commit.
if rs.CommitTime.IsZero() {
panic("RoundStepCommitWait requires rs.CommitTime")
}
time.Sleep(rs.CommitTime.Sub(time.Now()) + finalizeDuration)
conR.doActionCh <- RoundAction{rs.Height, rs.Round, RoundActionFinalize}
default:
panic("Should not happen")
}
}()
}
scheduleNextAction()
// NOTE: All ConsensusState.RunAction*() calls must come from here.
// Since only one routine calls them, it is safe to assume that
// the RoundState Height/Round/Step won't change concurrently.
// However, other fields like Proposal could change, due to gossip.
ACTION_LOOP:
for { for {
roundAction := <-conR.doActionCh // Get RoundState with new Step or quit.
var rs *RoundState
select {
case rs = <-conR.conS.NewStepCh():
case <-conR.quit:
return
}
height := roundAction.Height
round := roundAction.Round
action := roundAction.Action
rs := conR.conS.GetRoundState()
log.Info("Running round action A:%X %v", action, rs.Description())
// NOTE: This function should only be called
// when the cs.Height is still rs.Height.
broadcastNewRoundStep := func(step RoundStep) {
// Get seconds since beginning of height. // Get seconds since beginning of height.
// Due to the condition documented, this is safe. // Due to the condition documented, this is safe.
timeElapsed := rs.StartTime.Sub(time.Now()) timeElapsed := rs.StartTime.Sub(time.Now())
// Broadcast NewRoundStepMessage // Broadcast NewRoundStepMessage
msg := &NewRoundStepMessage{ msg := &NewRoundStepMessage{
Height: height, Height: rs.Height,
Round: round, Round: rs.Round,
Step: step, Step: rs.Step,
SecondsSinceStartTime: uint32(timeElapsed.Seconds()), SecondsSinceStartTime: uint32(timeElapsed.Seconds()),
} }
conR.sw.Broadcast(StateCh, msg) conR.sw.Broadcast(StateCh, msg)
} }
// Continue if action is not relevant
if height != rs.Height {
continue
}
// If action >= RoundActionCommitWait, the round doesn't matter.
if action < RoundActionCommitWait && round != rs.Round {
continue
}
// Run action
switch action {
case RoundActionPropose:
if rs.Step != RoundStepStart {
continue ACTION_LOOP
}
conR.conS.RunActionPropose(rs.Height, rs.Round)
broadcastNewRoundStep(RoundStepPropose)
scheduleNextAction()
continue ACTION_LOOP
case RoundActionPrevote:
if rs.Step >= RoundStepPrevote {
continue ACTION_LOOP
}
vote := conR.conS.RunActionPrevote(rs.Height, rs.Round)
broadcastNewRoundStep(RoundStepPrevote)
if vote != nil {
conR.broadcastVote(rs, vote)
}
scheduleNextAction()
continue ACTION_LOOP
case RoundActionPrecommit:
if rs.Step >= RoundStepPrecommit {
continue ACTION_LOOP
}
vote := conR.conS.RunActionPrecommit(rs.Height, rs.Round)
broadcastNewRoundStep(RoundStepPrecommit)
if vote != nil {
conR.broadcastVote(rs, vote)
}
scheduleNextAction()
continue ACTION_LOOP
case RoundActionTryCommit:
if rs.Step >= RoundStepCommit {
continue ACTION_LOOP
}
if rs.Precommits.HasTwoThirdsMajority() {
// NOTE: Duplicated in RoundActionCommitWait.
vote := conR.conS.RunActionCommit(rs.Height, rs.Round)
broadcastNewRoundStep(RoundStepCommit)
if vote != nil {
conR.broadcastVote(rs, vote)
// If we have +2/3 commits, queue an action to RoundActionCommitWait.
// Likely this is a duplicate action being pushed.
// See also Receive() where RoundActionCommitWait can be pushed in
// response to a vote from the network.
if rs.Commits.HasTwoThirdsMajority() {
conR.doActionCh <- RoundAction{rs.Height, rs.Round, RoundActionCommitWait}
}
}
// do not schedule next action.
continue ACTION_LOOP
} else {
// Could not commit, move onto next round.
conR.conS.SetupRound(rs.Round + 1)
scheduleNextAction()
continue ACTION_LOOP
}
case RoundActionCommitWait:
if rs.Step >= RoundStepCommitWait {
continue ACTION_LOOP
}
// Commit first we haven't already.
if rs.Step < RoundStepCommit {
// NOTE: Duplicated in RoundActionCommit.
vote := conR.conS.RunActionCommit(rs.Height, rs.Round)
broadcastNewRoundStep(RoundStepCommit)
if vote != nil {
conR.broadcastVote(rs, vote)
}
}
// Wait for more commit votes.
conR.conS.RunActionCommitWait(rs.Height, rs.Round)
scheduleNextAction()
continue ACTION_LOOP
case RoundActionFinalize:
if rs.Step != RoundStepCommitWait {
panic("This shouldn't happen")
}
conR.conS.RunActionFinalize(rs.Height, rs.Round)
// Height has been incremented, step is now RoundStepStart.
scheduleNextAction()
continue ACTION_LOOP
default:
panic("Unknown action")
}
// For clarity, ensure that all switch cases call "continue"
panic("Should not happen.")
}
} }
func (conR *ConsensusReactor) broadcastVote(rs *RoundState, vote *Vote) {
// Get our validator index
index, _ := rs.Validators.GetById(vote.SignerId)
msg := p2p.TypedMessage{msgTypeVote, vote}
for _, peer := range conR.sw.Peers().List() {
peer.Send(VoteCh, msg)
ps := peer.Data.Get(peerStateKey).(*PeerState)
ps.SetHasVote(rs.Height, rs.Round, vote.Type, index)
}
}
//--------------------------------------
func (conR *ConsensusReactor) gossipDataRoutine(peer *p2p.Peer, ps *PeerState) { func (conR *ConsensusReactor) gossipDataRoutine(peer *p2p.Peer, ps *PeerState) {
OUTER_LOOP: OUTER_LOOP:
@ -503,7 +254,7 @@ OUTER_LOOP:
// Send proposal Block parts? // Send proposal Block parts?
// NOTE: if we or peer is at RoundStepCommit*, the round // NOTE: if we or peer is at RoundStepCommit*, the round
// won't necessarily match, but that's OK. // won't necessarily match, but that's OK.
if rs.ProposalBlockParts.HashesTo(prs.ProposalBlockPartsHash) { if rs.ProposalBlockParts.Header().Equals(prs.ProposalBlockParts) {
if index, ok := rs.ProposalBlockParts.BitArray().Sub( if index, ok := rs.ProposalBlockParts.BitArray().Sub(
prs.ProposalBlockBitArray).PickRandom(); ok { prs.ProposalBlockBitArray).PickRandom(); ok {
msg := &PartMessage{ msg := &PartMessage{
@ -533,7 +284,7 @@ OUTER_LOOP:
} }
// Send proposal POL parts? // Send proposal POL parts?
if rs.ProposalPOLParts.HashesTo(prs.ProposalPOLPartsHash) { if rs.ProposalPOLParts.Header().Equals(prs.ProposalPOLParts) {
if index, ok := rs.ProposalPOLParts.BitArray().Sub( if index, ok := rs.ProposalPOLParts.BitArray().Sub(
prs.ProposalPOLBitArray).PickRandom(); ok { prs.ProposalPOLBitArray).PickRandom(); ok {
msg := &PartMessage{ msg := &PartMessage{
@ -574,62 +325,37 @@ OUTER_LOOP:
// Ensure that peer's prevote/precommit/commit bitarrays of of sufficient capacity // Ensure that peer's prevote/precommit/commit bitarrays of of sufficient capacity
ps.EnsureVoteBitArrays(rs.Height, rs.Round, rs.Validators.Size()) ps.EnsureVoteBitArrays(rs.Height, rs.Round, rs.Validators.Size())
// If there are prevotes to send... trySendVote := func(voteSet *VoteSet, peerVoteSet BitArray) (sent bool) {
if rs.Round == prs.Round && prs.Step <= RoundStepPrevote { // TODO: give priority to our vote.
index, ok := rs.Prevotes.BitArray().Sub(prs.Prevotes).PickRandom() index, ok := voteSet.BitArray().Sub(peerVoteSet).PickRandom()
if ok { if ok {
valId, val := rs.Validators.GetByIndex(index) vote := voteSet.GetByIndex(index)
if val != nil { // NOTE: vote may be a commit.
vote := rs.Prevotes.Get(valId)
// NOTE: vote may be a commit
msg := p2p.TypedMessage{msgTypeVote, vote} msg := p2p.TypedMessage{msgTypeVote, vote}
peer.Send(VoteCh, msg) peer.Send(VoteCh, msg)
ps.SetHasVote(rs.Height, rs.Round, VoteTypePrevote, index) ps.SetHasVote(rs.Height, rs.Round, index, vote)
if vote.Type == VoteTypeCommit { return true
ps.SetHasVote(rs.Height, rs.Round, VoteTypePrecommit, index)
ps.SetHasVote(rs.Height, rs.Round, VoteTypeCommit, index)
} }
return false
}
// If there are prevotes to send...
if rs.Round == prs.Round && prs.Step <= RoundStepPrevote {
if trySendVote(rs.Prevotes, prs.Prevotes) {
continue OUTER_LOOP continue OUTER_LOOP
} else {
log.Error("index is not a valid validator index")
}
} }
} }
// If there are precommits to send... // If there are precommits to send...
if rs.Round == prs.Round && prs.Step <= RoundStepPrecommit { if rs.Round == prs.Round && prs.Step <= RoundStepPrecommit {
index, ok := rs.Precommits.BitArray().Sub(prs.Precommits).PickRandom() if trySendVote(rs.Precommits, prs.Precommits) {
if ok {
valId, val := rs.Validators.GetByIndex(index)
if val != nil {
vote := rs.Precommits.Get(valId)
// NOTE: vote may be a commit
msg := p2p.TypedMessage{msgTypeVote, vote}
peer.Send(VoteCh, msg)
ps.SetHasVote(rs.Height, rs.Round, VoteTypePrecommit, index)
if vote.Type == VoteTypeCommit {
ps.SetHasVote(rs.Height, rs.Round, VoteTypeCommit, index)
}
continue OUTER_LOOP continue OUTER_LOOP
} else {
log.Error("index is not a valid validator index")
}
} }
} }
// If there are any commits to send... // If there are any commits to send...
index, ok := rs.Commits.BitArray().Sub(prs.Commits).PickRandom() if trySendVote(rs.Commits, prs.Commits) {
if ok {
valId, val := rs.Validators.GetByIndex(index)
if val != nil {
vote := rs.Commits.Get(valId)
msg := p2p.TypedMessage{msgTypeVote, vote}
peer.Send(VoteCh, msg)
ps.SetHasVote(rs.Height, rs.Round, VoteTypeCommit, index)
continue OUTER_LOOP continue OUTER_LOOP
} else {
log.Error("index is not a valid validator index")
}
} }
// We sent nothing. Sleep... // We sent nothing. Sleep...
@ -647,10 +373,10 @@ type PeerRoundState struct {
Step RoundStep // Step peer is at Step RoundStep // Step peer is at
StartTime time.Time // Estimated start of round 0 at this height StartTime time.Time // Estimated start of round 0 at this height
Proposal bool // True if peer has proposal for this round Proposal bool // True if peer has proposal for this round
ProposalBlockPartsHash []byte // Block parts merkle root ProposalBlockParts PartSetHeader //
ProposalBlockBitArray BitArray // Block parts bitarray ProposalBlockBitArray BitArray // True bit -> has part
ProposalPOLPartsHash []byte // POL parts merkle root ProposalPOLParts PartSetHeader //
ProposalPOLBitArray BitArray // POL parts bitarray ProposalPOLBitArray BitArray // True bit -> has part
Prevotes BitArray // All votes peer has for this round Prevotes BitArray // All votes peer has for this round
Precommits BitArray // All precommits peer has for this round Precommits BitArray // All precommits peer has for this round
Commits BitArray // All commits peer has for this height Commits BitArray // All commits peer has for this height
@ -693,10 +419,10 @@ func (ps *PeerState) SetHasProposal(proposal *Proposal) {
} }
ps.Proposal = true ps.Proposal = true
ps.ProposalBlockPartsHash = proposal.BlockPartsHash ps.ProposalBlockParts = proposal.BlockParts
ps.ProposalBlockBitArray = NewBitArray(uint(proposal.BlockPartsTotal)) ps.ProposalBlockBitArray = NewBitArray(uint(proposal.BlockParts.Total))
ps.ProposalPOLPartsHash = proposal.POLPartsHash ps.ProposalPOLParts = proposal.POLParts
ps.ProposalPOLBitArray = NewBitArray(uint(proposal.POLPartsTotal)) ps.ProposalPOLBitArray = NewBitArray(uint(proposal.POLParts.Total))
} }
func (ps *PeerState) SetHasProposalBlockPart(height uint32, round uint16, index uint16) { func (ps *PeerState) SetHasProposalBlockPart(height uint32, round uint16, index uint16) {
@ -740,20 +466,24 @@ func (ps *PeerState) EnsureVoteBitArrays(height uint32, round uint16, numValidat
} }
} }
func (ps *PeerState) SetHasVote(height uint32, round uint16, type_ uint8, index uint) { func (ps *PeerState) SetHasVote(height uint32, round uint16, index uint, vote *Vote) {
ps.mtx.Lock() ps.mtx.Lock()
defer ps.mtx.Unlock() defer ps.mtx.Unlock()
if ps.Height != height || (ps.Round != round && type_ != VoteTypeCommit) { if ps.Height != height {
return return
} }
switch type_ { switch vote.Type {
case VoteTypePrevote: case VoteTypePrevote:
ps.Prevotes.SetIndex(index, true) ps.Prevotes.SetIndex(index, true)
case VoteTypePrecommit: case VoteTypePrecommit:
ps.Precommits.SetIndex(index, true) ps.Precommits.SetIndex(index, true)
case VoteTypeCommit: case VoteTypeCommit:
if vote.Round < round {
ps.Prevotes.SetIndex(index, true)
ps.Precommits.SetIndex(index, true)
}
ps.Commits.SetIndex(index, true) ps.Commits.SetIndex(index, true)
default: default:
panic("Invalid vote type") panic("Invalid vote type")
@ -776,9 +506,9 @@ func (ps *PeerState) ApplyNewRoundStepMessage(msg *NewRoundStepMessage, rs *Roun
ps.StartTime = startTime ps.StartTime = startTime
if psHeight != msg.Height || psRound != msg.Round { if psHeight != msg.Height || psRound != msg.Round {
ps.Proposal = false ps.Proposal = false
ps.ProposalBlockPartsHash = nil ps.ProposalBlockParts = PartSetHeader{}
ps.ProposalBlockBitArray = BitArray{} ps.ProposalBlockBitArray = BitArray{}
ps.ProposalPOLPartsHash = nil ps.ProposalPOLParts = PartSetHeader{}
ps.ProposalPOLBitArray = BitArray{} ps.ProposalPOLBitArray = BitArray{}
// We'll update the BitArray capacity later. // We'll update the BitArray capacity later.
ps.Prevotes = BitArray{} ps.Prevotes = BitArray{}

635
consensus/state.go
View File

@ -1,9 +1,62 @@
/*
Consensus State Machine Overview:
* Propose, Prevote, Precommit represent state machine stages. (aka RoundStep, or step).
Each take a predetermined amount of time depending on the round number.
* The Commit step can be entered by two means:
1. After the Precommit step, +2/3 Precommits were found
2. At any time, +2/3 Commits were found
* Once in the Commit stage, two conditions must both be satisfied
before proceeding to the next height NewHeight.
* The Propose step of the next height does not begin until
at least Delta duration *after* +2/3 Commits were found.
The step stays at NewHeight until this timeout occurs before
proceeding to Propose.
+-------------------------------------+
| |
v |(Wait til CommitTime + Delta)
+-----------+ +-----+-----+
+----------> | Propose +--------------+ | NewHeight |
| +-----------+ | +-----------+
| | ^
| | |
| | |
|(Else) v |
+-----+-----+ +-----------+ |
| Precommit | <------------------------+ Prevote | |
+-----+-----+ +-----------+ |
|(If +2/3 Precommits found) |
| |
| + (When +2/3 Commits found) |
| | |
v v |
+------------------------------------------------------------------------------+
| Commit | |
| | |
| +----------------+ * Save Block | |
| |Get Block Parts |---> * Stage Block +--+ + |
| +----------------+ * Broadcast Commit | * Setup New Height |
| | * Move Commits set to |
| +--> LastCommits to continue |
| | collecting commits |
| +-----------------+ | * Broadcast New State |
| |Get +2/3 Commits |--> * Set CommitTime +--+ |
| +-----------------+ |
| |
+------------------------------------------------------------------------------+
*/
package consensus package consensus
import ( import (
"errors" "errors"
"fmt" "fmt"
"math"
"sync" "sync"
"sync/atomic"
"time" "time"
. "github.com/tendermint/tendermint/binary" . "github.com/tendermint/tendermint/binary"
@ -18,35 +71,38 @@ type RoundStep uint8
type RoundActionType uint8 type RoundActionType uint8
const ( const (
RoundStepStart = RoundStep(0x00) // Round started. RoundStepNewHeight = RoundStep(0x00) // Round0 for new height started, wait til CommitTime + Delta
RoundStepPropose = RoundStep(0x01) // Did propose, gossip proposal. RoundStepNewRound = RoundStep(0x01) // Pseudostep, immediately goes to RoundStepPropose
RoundStepPrevote = RoundStep(0x02) // Did prevote, gossip prevotes. RoundStepPropose = RoundStep(0x10) // Did propose, gossip proposal
RoundStepPrecommit = RoundStep(0x03) // Did precommit, gossip precommits. RoundStepPrevote = RoundStep(0x11) // Did prevote, gossip prevotes
RoundStepCommit = RoundStep(0x10) // Did commit, gossip commits. RoundStepPrecommit = RoundStep(0x12) // Did precommit, gossip precommits
RoundStepCommitWait = RoundStep(0x11) // Found +2/3 commits, wait more. RoundStepCommit = RoundStep(0x20) // Entered commit state machine
// If a block could not be committed at a given round, RoundActionPropose = RoundActionType(0xA0) // Propose and goto RoundStepPropose
// we progress to the next round, skipping RoundStepCommit. RoundActionPrevote = RoundActionType(0xA1) // Prevote and goto RoundStepPrevote
// RoundActionPrecommit = RoundActionType(0xA2) // Precommit and goto RoundStepPrecommit
// If a block was committed, we goto RoundStepCommit, RoundActionTryCommit = RoundActionType(0xC0) // Goto RoundStepCommit, or RoundStepPropose for next round.
// then wait "finalizeDuration" to gather more commits, RoundActionTryFinalize = RoundActionType(0xC1) // Maybe goto RoundStepPropose for next round.
// then we progress to the next height at round 0.
// TODO: document how RoundStepCommit transcends all rounds.
RoundActionPropose = RoundActionType(0x00) // Goto RoundStepPropose roundDuration0 = 60 * time.Second // The first round is 60 seconds long.
RoundActionPrevote = RoundActionType(0x01) // Goto RoundStepPrevote roundDurationDelta = 15 * time.Second // Each successive round lasts 15 seconds longer.
RoundActionPrecommit = RoundActionType(0x02) // Goto RoundStepPrecommit roundDeadlinePrevote = float64(1.0 / 3.0) // When the prevote is due.
RoundActionTryCommit = RoundActionType(0x10) // Goto RoundStepCommit or RoundStepStart next round roundDeadlinePrecommit = float64(2.0 / 3.0) // When the precommit vote is due.
RoundActionCommitWait = RoundActionType(0x11) // Goto RoundStepCommitWait newHeightDelta = roundDuration0 / 3 // The time to wait between commitTime and startTime of next consensus rounds.
RoundActionFinalize = RoundActionType(0x12) // Goto RoundStepStart next height
) )
var ( var (
ErrInvalidProposalSignature = errors.New("Error invalid proposal signature") ErrInvalidProposalSignature = errors.New("Error invalid proposal signature")
consensusStateKey = []byte("consensusState")
) )
type RoundAction struct {
Height uint32 // The block height for which consensus is reaching for.
Round uint16 // The round number at given height.
Action RoundActionType // Action to perform.
}
//-----------------------------------------------------------------------------
// Immutable when returned from ConsensusState.GetRoundState() // Immutable when returned from ConsensusState.GetRoundState()
type RoundState struct { type RoundState struct {
Height uint32 // Height we are working on Height uint32 // Height we are working on
@ -111,12 +167,18 @@ func (rs *RoundState) Description() string {
rs.Height, rs.Round, rs.Step, rs.StartTime) rs.Height, rs.Round, rs.Step, rs.StartTime)
} }
//------------------------------------- //-----------------------------------------------------------------------------
// Tracks consensus state across block heights and rounds. // Tracks consensus state across block heights and rounds.
type ConsensusState struct { type ConsensusState struct {
started uint32
stopped uint32
quit chan struct{}
blockStore *BlockStore blockStore *BlockStore
mempool *mempool.Mempool mempool *mempool.Mempool
runActionCh chan RoundAction
newStepCh chan *RoundState
mtx sync.Mutex mtx sync.Mutex
RoundState RoundState
@ -127,8 +189,11 @@ type ConsensusState struct {
func NewConsensusState(state *state.State, blockStore *BlockStore, mempool *mempool.Mempool) *ConsensusState { func NewConsensusState(state *state.State, blockStore *BlockStore, mempool *mempool.Mempool) *ConsensusState {
cs := &ConsensusState{ cs := &ConsensusState{
quit: make(chan struct{}),
blockStore: blockStore, blockStore: blockStore,
mempool: mempool, mempool: mempool,
runActionCh: make(chan RoundAction, 1),
newStepCh: make(chan *RoundState, 1),
} }
cs.updateToState(state) cs.updateToState(state)
return cs return cs
@ -137,10 +202,176 @@ func NewConsensusState(state *state.State, blockStore *BlockStore, mempool *memp
func (cs *ConsensusState) GetRoundState() *RoundState { func (cs *ConsensusState) GetRoundState() *RoundState {
cs.mtx.Lock() cs.mtx.Lock()
defer cs.mtx.Unlock() defer cs.mtx.Unlock()
return cs.getRoundState()
}
func (cs *ConsensusState) getRoundState() *RoundState {
rs := cs.RoundState // copy rs := cs.RoundState // copy
return &rs return &rs
} }
func (cs *ConsensusState) NewStepCh() chan *RoundState {
return cs.newStepCh
}
func (cs *ConsensusState) Start() {
if atomic.CompareAndSwapUint32(&cs.started, 0, 1) {
log.Info("Starting ConsensusState")
go cs.stepTransitionRoutine()
}
}
func (cs *ConsensusState) Stop() {
if atomic.CompareAndSwapUint32(&cs.stopped, 0, 1) {
log.Info("Stopping ConsensusState")
close(cs.quit)
}
}
func (cs *ConsensusState) IsStopped() bool {
return atomic.LoadUint32(&cs.stopped) == 1
}
// Source of all round state transitions (and votes).
func (cs *ConsensusState) stepTransitionRoutine() {
// For clarity, all state transitions that happen after some timeout are here.
// Schedule the next action by pushing a RoundAction{} to cs.runActionCh.
scheduleNextAction := func() {
go func() {
rs := cs.getRoundState()
round, roundStartTime, roundDuration, _, elapsedRatio := calcRoundInfo(rs.StartTime)
log.Debug("Called scheduleNextAction. round:%v roundStartTime:%v elapsedRatio:%v", round, roundStartTime, elapsedRatio)
switch rs.Step {
case RoundStepNewHeight:
// We should run RoundActionPropose when rs.StartTime passes.
if elapsedRatio < 0 {
// startTime is in the future.
time.Sleep(time.Duration((-1.0 * elapsedRatio) * float64(roundDuration)))
}
cs.runActionCh <- RoundAction{rs.Height, rs.Round, RoundActionPropose}
case RoundStepNewRound:
// Pseudostep: Immediately goto propose.
cs.runActionCh <- RoundAction{rs.Height, rs.Round, RoundActionPropose}
case RoundStepPropose:
// Wake up when it's time to vote.
time.Sleep(time.Duration((roundDeadlinePrevote - elapsedRatio) * float64(roundDuration)))
cs.runActionCh <- RoundAction{rs.Height, rs.Round, RoundActionPrevote}
case RoundStepPrevote:
// Wake up when it's time to precommit.
time.Sleep(time.Duration((roundDeadlinePrecommit - elapsedRatio) * float64(roundDuration)))
cs.runActionCh <- RoundAction{rs.Height, rs.Round, RoundActionPrecommit}
case RoundStepPrecommit:
// Wake up when the round is over.
time.Sleep(time.Duration((1.0 - elapsedRatio) * float64(roundDuration)))
cs.runActionCh <- RoundAction{rs.Height, rs.Round, RoundActionTryCommit}
case RoundStepCommit:
// There's nothing to scheudle, we're waiting for
// ProposalBlockParts.IsComplete() &&
// Commits.HasTwoThirdsMajority()
panic("The next action from RoundStepCommit is not scheduled by time")
default:
panic("Should not happen")
}
}()
}
scheduleNextAction()
// NOTE: All ConsensusState.RunAction*() calls come from here.
// Since only one routine calls them, it is safe to assume that
// the RoundState Height/Round/Step won't change concurrently.
// However, other fields like Proposal could change, due to gossip.
ACTION_LOOP:
for {
var roundAction RoundAction
select {
case roundAction = <-cs.runActionCh:
case <-cs.quit:
return
}
height, round, action := roundAction.Height, roundAction.Round, roundAction.Action
rs := cs.GetRoundState()
log.Info("Running round action A:%X %v", action, rs.Description())
// Continue if action is not relevant
if height != rs.Height {
continue
}
// If action <= RoundActionPrecommit, the round must match too.
if action <= RoundActionPrecommit && round != rs.Round {
continue
}
// Run action
switch action {
case RoundActionPropose:
if rs.Step != RoundStepNewHeight && rs.Step != RoundStepNewRound {
continue ACTION_LOOP
}
cs.RunActionPropose(rs.Height, rs.Round)
scheduleNextAction()
continue ACTION_LOOP
case RoundActionPrevote:
if rs.Step >= RoundStepPrevote {
continue ACTION_LOOP
}
cs.RunActionPrevote(rs.Height, rs.Round)
scheduleNextAction()
continue ACTION_LOOP
case RoundActionPrecommit:
if rs.Step >= RoundStepPrecommit {
continue ACTION_LOOP
}
cs.RunActionPrecommit(rs.Height, rs.Round)
scheduleNextAction()
continue ACTION_LOOP
case RoundActionTryCommit:
if rs.Step >= RoundStepCommit {
continue ACTION_LOOP
}
if rs.Precommits.HasTwoThirdsMajority() {
// Enter RoundStepCommit and commit.
cs.RunActionCommit(rs.Height)
// Maybe finalize already
cs.runActionCh <- RoundAction{rs.Height, rs.Round, RoundActionTryFinalize}
continue ACTION_LOOP
} else {
// Could not commit, move onto next round.
cs.SetupNewRound(rs.Height, rs.Round+1)
// cs.Step is now at RoundStepNewRound
scheduleNextAction()
continue ACTION_LOOP
}
case RoundActionTryFinalize:
if cs.TryFinalizeCommit(rs.Height) {
// Now at new height
// cs.Step is at RoundStepNewHeight or RoundStepNewRound.
scheduleNextAction()
continue ACTION_LOOP
} else {
// do not schedule next action.
continue ACTION_LOOP
}
default:
panic("Unknown action")
}
// For clarity, ensure that all switch cases call "continue"
panic("Should not happen.")
}
}
// Updates ConsensusState and increments height to match that of state.
// If calculated round is greater than 0 (based on BlockTime or calculated StartTime)
// then also sets up the appropriate round, and cs.Step becomes RoundStepNewRound.
// Otherwise the round is 0 and cs.Step becomes RoundStepNewHeight.
func (cs *ConsensusState) updateToState(state *state.State) { func (cs *ConsensusState) updateToState(state *state.State) {
// Sanity check state. // Sanity check state.
if cs.Height > 0 && cs.Height != state.Height { if cs.Height > 0 && cs.Height != state.Height {
@ -153,11 +384,11 @@ func (cs *ConsensusState) updateToState(state *state.State) {
height := state.Height + 1 // next desired block height height := state.Height + 1 // next desired block height
cs.Height = height cs.Height = height
cs.Round = 0 cs.Round = 0
cs.Step = RoundStepStart cs.Step = RoundStepNewHeight
if cs.CommitTime.IsZero() { if cs.CommitTime.IsZero() {
cs.StartTime = state.BlockTime.Add(finalizeDuration) cs.StartTime = state.BlockTime.Add(newHeightDelta)
} else { } else {
cs.StartTime = cs.CommitTime.Add(finalizeDuration) cs.StartTime = cs.CommitTime.Add(newHeightDelta)
} }
cs.CommitTime = time.Time{} cs.CommitTime = time.Time{}
cs.Validators = validators cs.Validators = validators
@ -181,20 +412,15 @@ func (cs *ConsensusState) updateToState(state *state.State) {
// Update the round if we need to. // Update the round if we need to.
round := calcRound(cs.StartTime) round := calcRound(cs.StartTime)
if round > 0 { if round > 0 {
cs.setupRound(round) cs.setupNewRound(round)
} }
} }
func (cs *ConsensusState) SetupRound(round uint16) { func (cs *ConsensusState) setupNewRound(round uint16) {
cs.mtx.Lock() // Sanity check
defer cs.mtx.Unlock() if round == 0 {
if cs.Round >= round { panic("setupNewRound() should never be called for round 0")
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. // Increment all the way to round.
validators := cs.Validators.Copy() validators := cs.Validators.Copy()
@ -203,7 +429,7 @@ func (cs *ConsensusState) setupRound(round uint16) {
} }
cs.Round = round cs.Round = round
cs.Step = RoundStepStart cs.Step = RoundStepNewRound
cs.Validators = validators cs.Validators = validators
cs.Proposal = nil cs.Proposal = nil
cs.ProposalBlock = nil cs.ProposalBlock = nil
@ -224,6 +450,21 @@ func (cs *ConsensusState) SetPrivValidator(priv *PrivValidator) {
//----------------------------------------------------------------------------- //-----------------------------------------------------------------------------
// Set up the round to desired round and set step to RoundStepNewRound
func (cs *ConsensusState) SetupNewRound(height uint32, desiredRound uint16) bool {
cs.mtx.Lock()
defer cs.mtx.Unlock()
if cs.Height != height {
return false
}
if desiredRound <= cs.Round {
return false
}
cs.setupNewRound(desiredRound)
cs.newStepCh <- cs.getRoundState()
return true
}
func (cs *ConsensusState) RunActionPropose(height uint32, round uint16) { func (cs *ConsensusState) RunActionPropose(height uint32, round uint16) {
cs.mtx.Lock() cs.mtx.Lock()
defer cs.mtx.Unlock() defer cs.mtx.Unlock()
@ -231,7 +472,9 @@ func (cs *ConsensusState) RunActionPropose(height uint32, round uint16) {
return return
} }
cs.Step = RoundStepPropose cs.Step = RoundStepPropose
cs.newStepCh <- cs.getRoundState()
// Nothing to do if it's not our turn.
if cs.PrivValidator == nil || cs.Validators.Proposer().Id != cs.PrivValidator.Id { if cs.PrivValidator == nil || cs.Validators.Proposer().Id != cs.PrivValidator.Id {
return return
} }
@ -248,6 +491,7 @@ func (cs *ConsensusState) RunActionPropose(height uint32, round uint16) {
blockParts = cs.LockedBlockParts blockParts = cs.LockedBlockParts
pol = cs.LockedPOL pol = cs.LockedPOL
} else { } else {
// Otherwise we should create a new proposal.
var validation Validation var validation Validation
if cs.Height == 1 { if cs.Height == 1 {
// We're creating a proposal for the first block. // We're creating a proposal for the first block.
@ -285,9 +529,7 @@ func (cs *ConsensusState) RunActionPropose(height uint32, round uint16) {
} }
// Make proposal // Make proposal
proposal := NewProposal(cs.Height, cs.Round, proposal := NewProposal(cs.Height, cs.Round, blockParts.Header(), polParts.Header())
blockParts.Total(), blockParts.RootHash(),
polParts.Total(), polParts.RootHash())
cs.PrivValidator.Sign(proposal) cs.PrivValidator.Sign(proposal)
// Set fields // Set fields
@ -298,167 +540,186 @@ func (cs *ConsensusState) RunActionPropose(height uint32, round uint16) {
cs.ProposalPOLParts = polParts cs.ProposalPOLParts = polParts
} }
func (cs *ConsensusState) RunActionPrevote(height uint32, round uint16) *Vote { // Prevote for LockedBlock if we're locked, or ProposealBlock if valid.
// Otherwise vote nil.
func (cs *ConsensusState) RunActionPrevote(height uint32, round uint16) {
cs.mtx.Lock() cs.mtx.Lock()
defer cs.mtx.Unlock() defer cs.mtx.Unlock()
if cs.Height != height || cs.Round != round { if cs.Height != height || cs.Round != round {
Panicf("RunActionPrevote(%v/%v), expected %v/%v", height, round, cs.Height, cs.Round) Panicf("RunActionPrevote(%v/%v), expected %v/%v", height, round, cs.Height, cs.Round)
} }
cs.Step = RoundStepPrevote cs.Step = RoundStepPrevote
cs.newStepCh <- cs.getRoundState()
// If a block is locked, prevote that. // If a block is locked, prevote that.
if cs.LockedBlock != nil { if cs.LockedBlock != nil {
return cs.signAddVote(VoteTypePrevote, cs.LockedBlock.Hash()) cs.signAddVote(VoteTypePrevote, cs.LockedBlock.Hash(), cs.LockedBlockParts.Header())
return
} }
// If ProposalBlock is nil, prevote nil. // If ProposalBlock is nil, prevote nil.
if cs.ProposalBlock == nil { if cs.ProposalBlock == nil {
return nil cs.signAddVote(VoteTypePrevote, nil, PartSetHeader{})
return
} }
// Try staging proposed block.
// Try staging cs.ProposalBlock
err := cs.stageBlock(cs.ProposalBlock) err := cs.stageBlock(cs.ProposalBlock)
if err != nil { if err != nil {
// Prevote nil. // ProposalBlock is invalid, prevote nil.
return nil cs.signAddVote(VoteTypePrevote, nil, PartSetHeader{})
} else { return
// Prevote block.
return cs.signAddVote(VoteTypePrevote, cs.ProposalBlock.Hash())
} }
// Prevote cs.ProposalBlock
cs.signAddVote(VoteTypePrevote, cs.ProposalBlock.Hash(), cs.ProposalBlockParts.Header())
return
} }
// Lock the ProposalBlock if we have enough prevotes for it, // Lock & Precommit the ProposalBlock if we have enough prevotes for it,
// or unlock an existing lock if +2/3 of prevotes were nil. // or unlock an existing lock if +2/3 of prevotes were nil.
// Returns a blockhash if a block was locked. func (cs *ConsensusState) RunActionPrecommit(height uint32, round uint16) {
func (cs *ConsensusState) RunActionPrecommit(height uint32, round uint16) *Vote {
cs.mtx.Lock() cs.mtx.Lock()
defer cs.mtx.Unlock() defer cs.mtx.Unlock()
if cs.Height != height || cs.Round != round { if cs.Height != height || cs.Round != round {
Panicf("RunActionPrecommit(%v/%v), expected %v/%v", height, round, cs.Height, cs.Round) Panicf("RunActionPrecommit(%v/%v), expected %v/%v", height, round, cs.Height, cs.Round)
} }
cs.Step = RoundStepPrecommit cs.Step = RoundStepPrecommit
cs.newStepCh <- cs.getRoundState()
if hash, ok := cs.Prevotes.TwoThirdsMajority(); ok { hash, partsHeader, ok := cs.Prevotes.TwoThirdsMajority()
if !ok {
// If we don't have two thirds of prevotes,
// don't do anything at all.
return
}
// Remember this POL. (hash may be nil) // Remember this POL. (hash may be nil)
cs.LockedPOL = cs.Prevotes.MakePOL() cs.LockedPOL = cs.Prevotes.MakePOL()
// If +2/3 prevoted nil. Just unlock.
if len(hash) == 0 { if len(hash) == 0 {
// +2/3 prevoted nil. Just unlock.
cs.LockedBlock = nil cs.LockedBlock = nil
cs.LockedBlockParts = nil cs.LockedBlockParts = nil
return nil return
} else if cs.ProposalBlock.HashesTo(hash) { }
// +2/3 prevoted for proposal block
// If +2/3 prevoted for already locked block, precommit it.
if cs.LockedBlock.HashesTo(hash) {
cs.signAddVote(VoteTypePrecommit, hash, partsHeader)
return
}
// If +2/3 prevoted for cs.ProposalBlock, lock it and precommit it.
if cs.ProposalBlock.HashesTo(hash) {
// Validate the block. // Validate the block.
// See note on ZombieValidators to see why.
if err := cs.stageBlock(cs.ProposalBlock); err != nil { if err := cs.stageBlock(cs.ProposalBlock); err != nil {
// Prevent zombies.
log.Warning("+2/3 prevoted for an invalid block: %v", err) log.Warning("+2/3 prevoted for an invalid block: %v", err)
return nil return
} }
cs.LockedBlock = cs.ProposalBlock cs.LockedBlock = cs.ProposalBlock
cs.LockedBlockParts = cs.ProposalBlockParts cs.LockedBlockParts = cs.ProposalBlockParts
return cs.signAddVote(VoteTypePrecommit, hash) cs.signAddVote(VoteTypePrecommit, hash, partsHeader)
} else if cs.LockedBlock.HashesTo(hash) { return
// +2/3 prevoted for already locked block }
return cs.signAddVote(VoteTypePrecommit, hash)
} else { // We don't have the block that validators prevoted.
// We don't have the block that hashes to hash.
// Unlock if we're locked. // Unlock if we're locked.
cs.LockedBlock = nil cs.LockedBlock = nil
cs.LockedBlockParts = nil cs.LockedBlockParts = nil
return nil return
}
} else {
return nil
}
} }
// Commits a block if we have enough precommits (and we have the block). // Enter commit step. See the diagram for details.
// If successful, saves the block and state and resets mempool, func (cs *ConsensusState) RunActionCommit(height uint32) {
// and returns the committed block.
// Commit is not finalized until FinalizeCommit() is called.
// This allows us to stay at this height and gather more commits.
func (cs *ConsensusState) RunActionCommit(height uint32, round uint16) *Vote {
cs.mtx.Lock() cs.mtx.Lock()
defer cs.mtx.Unlock() defer cs.mtx.Unlock()
if cs.Height != height || cs.Round != round { if cs.Height != height {
Panicf("RunActionCommit(%v/%v), expected %v/%v", height, round, cs.Height, cs.Round) Panicf("RunActionCommit(%v), expected %v", height, cs.Height)
}
// There are two ways to enter:
// 1. +2/3 precommits at the end of RoundStepPrecommit
// 2. +2/3 commits at any time
hash, partsHeader, ok := cs.Precommits.TwoThirdsMajority()
if !ok {
hash, partsHeader, ok = cs.Commits.TwoThirdsMajority()
if !ok {
panic("RunActionCommit() expects +2/3 precommits or commits")
}
} }
cs.Step = RoundStepCommit cs.Step = RoundStepCommit
cs.newStepCh <- cs.getRoundState()
if hash, ok := cs.Precommits.TwoThirdsMajority(); ok { // Clear the Locked* fields and use cs.Proposed*
// 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
var blockParts *PartSet
if cs.LockedBlock.HashesTo(hash) { if cs.LockedBlock.HashesTo(hash) {
block = cs.LockedBlock cs.ProposalBlock = cs.LockedBlock
blockParts = cs.LockedBlockParts cs.ProposalBlockParts = cs.LockedBlockParts
} else if cs.ProposalBlock.HashesTo(hash) { cs.LockedBlock = nil
block = cs.ProposalBlock cs.LockedBlockParts = nil
blockParts = cs.ProposalBlockParts cs.LockedPOL = nil
} else {
return nil
} }
// The proposal must be valid. // If we don't have the block being committed, set up to get it.
if err := cs.stageBlock(block); err != nil {
log.Warning("Network is commiting an invalid proposal? %v", err)
return nil
}
// Keep block in cs.Proposal*
if !cs.ProposalBlock.HashesTo(hash) { if !cs.ProposalBlock.HashesTo(hash) {
cs.ProposalBlock = block if !cs.ProposalBlockParts.Header().Equals(partsHeader) {
cs.ProposalBlockParts = blockParts // We're getting the wrong block.
// Set up ProposalBlockParts and keep waiting.
cs.ProposalBlock = nil
cs.ProposalBlockParts = NewPartSetFromHeader(partsHeader)
} else {
// We just need to keep waiting.
}
} else {
// We have the block, so save/stage/sign-commit-vote.
cs.processBlockForCommit(cs.ProposalBlock, cs.ProposalBlockParts)
} }
// Save to blockStore // If we have +2/3 commits, set the CommitTime
cs.blockStore.SaveBlock(block)
// Save the state
cs.stagedState.Save()
// Update mempool.
cs.mempool.ResetForBlockAndState(block, cs.stagedState)
return cs.signAddVote(VoteTypeCommit, block.Hash())
}
return nil
}
func (cs *ConsensusState) RunActionCommitWait(height uint32, round uint16) {
cs.mtx.Lock()
defer cs.mtx.Unlock()
if cs.Height != height || cs.Round != round {
Panicf("RunActionCommitWait(%v/%v), expected %v/%v", height, round, cs.Height, cs.Round)
}
cs.Step = RoundStepCommitWait
if cs.Commits.HasTwoThirdsMajority() { if cs.Commits.HasTwoThirdsMajority() {
cs.CommitTime = time.Now() cs.CommitTime = time.Now()
} else {
panic("RunActionCommitWait() expects +2/3 commits")
} }
} }
func (cs *ConsensusState) RunActionFinalize(height uint32, round uint16) { // Returns true if Finalize happened, which increments height && sets
// the step to RoundStepNewHeight (or RoundStepNewRound, but probably not).
func (cs *ConsensusState) TryFinalizeCommit(height uint32) bool {
cs.mtx.Lock() cs.mtx.Lock()
defer cs.mtx.Unlock() defer cs.mtx.Unlock()
if cs.Height != height || cs.Round != round {
Panicf("RunActionFinalize(%v/%v), expected %v/%v", height, round, cs.Height, cs.Round) if cs.Height != height {
Panicf("TryFinalizeCommit(%v), expected %v", height, cs.Height)
} }
// What was staged becomes committed. if cs.Step == RoundStepCommit &&
// XXX it's possible that this node never received the block to stage!!! cs.Commits.HasTwoThirdsMajority() &&
cs.ProposalBlockParts.IsComplete() {
// Sanity check
if cs.ProposalBlock == nil {
Panicf("Expected ProposalBlock to exist")
}
hash, header, _ := cs.Commits.TwoThirdsMajority()
if !cs.ProposalBlock.HashesTo(hash) {
Panicf("Expected ProposalBlock to hash to commit hash")
}
if !cs.ProposalBlockParts.Header().Equals(header) {
Panicf("Expected ProposalBlockParts header to be commit header")
}
err := cs.stageBlock(cs.ProposalBlock)
if err == nil {
// Increment height.
cs.updateToState(cs.stagedState) cs.updateToState(cs.stagedState)
// cs.Step is now RoundStepNewHeight or RoundStepNewRound
cs.newStepCh <- cs.getRoundState()
return true
} else {
// Prevent zombies.
Panicf("+2/3 committed an invalid block: %v", err)
}
}
return false
} }
//----------------------------------------------------------------------------- //-----------------------------------------------------------------------------
@ -472,23 +733,29 @@ func (cs *ConsensusState) SetProposal(proposal *Proposal) error {
return nil return nil
} }
// Invalid. // Does not apply
if proposal.Height != cs.Height || proposal.Round != cs.Round { if proposal.Height != cs.Height || proposal.Round != cs.Round {
return nil return nil
} }
// We don't care about the proposal if we're already in RoundStepCommit.
if cs.Step == RoundStepCommit {
return nil
}
// Verify signature // Verify signature
if !cs.Validators.Proposer().Verify(proposal) { if !cs.Validators.Proposer().Verify(proposal) {
return ErrInvalidProposalSignature return ErrInvalidProposalSignature
} }
cs.Proposal = proposal cs.Proposal = proposal
cs.ProposalBlockParts = NewPartSetFromMetadata(proposal.BlockPartsTotal, proposal.BlockPartsHash) cs.ProposalBlockParts = NewPartSetFromHeader(proposal.BlockParts)
cs.ProposalPOLParts = NewPartSetFromMetadata(proposal.POLPartsTotal, proposal.POLPartsHash) cs.ProposalPOLParts = NewPartSetFromHeader(proposal.POLParts)
return nil return nil
} }
// NOTE: block is not necessarily valid. // NOTE: block is not necessarily valid.
// NOTE: This function may increment the height.
func (cs *ConsensusState) AddProposalBlockPart(height uint32, round uint16, part *Part) (added bool, err error) { func (cs *ConsensusState) AddProposalBlockPart(height uint32, round uint16, part *Part) (added bool, err error) {
cs.mtx.Lock() cs.mtx.Lock()
defer cs.mtx.Unlock() defer cs.mtx.Unlock()
@ -499,7 +766,7 @@ func (cs *ConsensusState) AddProposalBlockPart(height uint32, round uint16, part
} }
// We're not expecting a block part. // We're not expecting a block part.
if cs.ProposalBlockParts != nil { if cs.ProposalBlockParts == nil {
return false, nil // TODO: bad peer? Return error? return false, nil // TODO: bad peer? Return error?
} }
@ -511,6 +778,7 @@ func (cs *ConsensusState) AddProposalBlockPart(height uint32, round uint16, part
var n int64 var n int64
var err error var err error
cs.ProposalBlock = ReadBlock(cs.ProposalBlockParts.GetReader(), &n, &err) cs.ProposalBlock = ReadBlock(cs.ProposalBlockParts.GetReader(), &n, &err)
cs.runActionCh <- RoundAction{cs.Height, cs.Round, RoundActionTryFinalize}
return true, err return true, err
} }
return true, nil return true, nil
@ -526,7 +794,7 @@ func (cs *ConsensusState) AddProposalPOLPart(height uint32, round uint16, part *
} }
// We're not expecting a POL part. // We're not expecting a POL part.
if cs.ProposalPOLParts != nil { if cs.ProposalPOLParts == nil {
return false, nil // TODO: bad peer? Return error? return false, nil // TODO: bad peer? Return error?
} }
@ -543,6 +811,7 @@ func (cs *ConsensusState) AddProposalPOLPart(height uint32, round uint16, part *
return true, nil return true, nil
} }
// NOTE: This function may increment the height.
func (cs *ConsensusState) AddVote(vote *Vote) (added bool, err error) { func (cs *ConsensusState) AddVote(vote *Vote) (added bool, err error) {
switch vote.Type { switch vote.Type {
case VoteTypePrevote: case VoteTypePrevote:
@ -553,14 +822,22 @@ func (cs *ConsensusState) AddVote(vote *Vote) (added bool, err error) {
return cs.Precommits.Add(vote) return cs.Precommits.Add(vote)
case VoteTypeCommit: case VoteTypeCommit:
// Commits checks for height match. // Commits checks for height match.
// No need to check if vote.Round < cs.Round ...
// Prevotes && Precommits already checks that.
cs.Prevotes.Add(vote) cs.Prevotes.Add(vote)
cs.Precommits.Add(vote) cs.Precommits.Add(vote)
return cs.Commits.Add(vote) added, err = cs.Commits.Add(vote)
if added && cs.Commits.HasTwoThirdsMajority() {
cs.runActionCh <- RoundAction{cs.Height, cs.Round, RoundActionTryFinalize}
}
return added, err
default: default:
panic("Unknown vote type") panic("Unknown vote type")
} }
} }
//-----------------------------------------------------------------------------
func (cs *ConsensusState) stageBlock(block *Block) error { func (cs *ConsensusState) stageBlock(block *Block) error {
if block == nil { if block == nil {
panic("Cannot stage nil block") panic("Cannot stage nil block")
@ -586,7 +863,7 @@ func (cs *ConsensusState) stageBlock(block *Block) error {
} }
} }
func (cs *ConsensusState) signAddVote(type_ byte, hash []byte) *Vote { func (cs *ConsensusState) signAddVote(type_ byte, hash []byte, header PartSetHeader) *Vote {
if cs.PrivValidator == nil || !cs.Validators.HasId(cs.PrivValidator.Id) { if cs.PrivValidator == nil || !cs.Validators.HasId(cs.PrivValidator.Id) {
return nil return nil
} }
@ -595,8 +872,82 @@ func (cs *ConsensusState) signAddVote(type_ byte, hash []byte) *Vote {
Round: cs.Round, Round: cs.Round,
Type: type_, Type: type_,
BlockHash: hash, BlockHash: hash,
BlockParts: header,
} }
cs.PrivValidator.Sign(vote) cs.PrivValidator.Sign(vote)
cs.AddVote(vote) cs.AddVote(vote)
return vote return vote
} }
func (cs *ConsensusState) processBlockForCommit(block *Block, blockParts *PartSet) {
// The proposal must be valid.
if err := cs.stageBlock(block); err != nil {
// Prevent zombies.
log.Warning("+2/3 precommitted an invalid block: %v", err)
return
}
// Save to blockStore
cs.blockStore.SaveBlock(block)
// Save the state
cs.stagedState.Save()
// Update mempool.
cs.mempool.ResetForBlockAndState(block, cs.stagedState)
cs.signAddVote(VoteTypeCommit, block.Hash(), blockParts.Header())
}
//-----------------------------------------------------------------------------
// total duration of given round
func calcRoundDuration(round uint16) time.Duration {
return roundDuration0 + roundDurationDelta*time.Duration(round)
}
// startTime is when round zero started.
func calcRoundStartTime(round uint16, startTime time.Time) time.Time {
return startTime.Add(roundDuration0*time.Duration(round) +
roundDurationDelta*(time.Duration((int64(round)*int64(round)-int64(round))/2)))
}
// calculates the current round given startTime of round zero.
// NOTE: round is zero if startTime is in the future.
func calcRound(startTime time.Time) uint16 {
now := time.Now()
if now.Before(startTime) {
return 0
}
// Start + D_0 * R + D_delta * (R^2 - R)/2 <= Now; find largest integer R.
// D_delta * R^2 + (2D_0 - D_delta) * R + 2(Start - Now) <= 0.
// AR^2 + BR + C <= 0; A = D_delta, B = (2_D0 - D_delta), C = 2(Start - Now).
// R = Floor((-B + Sqrt(B^2 - 4AC))/2A)
A := float64(roundDurationDelta)
B := 2.0*float64(roundDuration0) - float64(roundDurationDelta)
C := 2.0 * float64(startTime.Sub(now))
R := math.Floor((-B + math.Sqrt(B*B-4.0*A*C)) / (2 * A))
if math.IsNaN(R) {
panic("Could not calc round, should not happen")
}
if R > math.MaxInt16 {
Panicf("Could not calc round, round overflow: %v", R)
}
if R < 0 {
return 0
}
return uint16(R)
}
// convenience
// NOTE: elapsedRatio can be negative if startTime is in the future.
func calcRoundInfo(startTime time.Time) (round uint16, roundStartTime time.Time, roundDuration time.Duration,
roundElapsed time.Duration, elapsedRatio float64) {
round = calcRound(startTime)
roundStartTime = calcRoundStartTime(round, startTime)
roundDuration = calcRoundDuration(round)
roundElapsed = time.Now().Sub(roundStartTime)
elapsedRatio = float64(roundElapsed) / float64(roundDuration)
return
}

64
consensus/state_test.go
View File

@ -53,7 +53,7 @@ func makeConsensusState() (*ConsensusState, []*state.PrivAccount) {
func assertPanics(t *testing.T, msg string, f func()) { func assertPanics(t *testing.T, msg string, f func()) {
defer func() { defer func() {
if err := recover(); err == nil { if err := recover(); err == nil {
t.Error("Should have panic'd, but didn't. %v", msg) t.Errorf("Should have panic'd, but didn't: %v", msg)
} }
}() }()
f() f()
@ -74,36 +74,32 @@ func TestSetupRound(t *testing.T) {
// Ensure that vote appears in RoundState. // Ensure that vote appears in RoundState.
rs0 := cs.GetRoundState() rs0 := cs.GetRoundState()
if vote := rs0.Prevotes.Get(0); vote == nil || vote.Type != VoteTypePrevote { if vote := rs0.Prevotes.GetById(0); vote == nil || vote.Type != VoteTypePrevote {
t.Errorf("Expected to find prevote but got %v", vote) t.Errorf("Expected to find prevote but got %v", vote)
} }
if vote := rs0.Precommits.Get(0); vote == nil || vote.Type != VoteTypePrecommit { if vote := rs0.Precommits.GetById(0); vote == nil || vote.Type != VoteTypePrecommit {
t.Errorf("Expected to find precommit but got %v", vote) t.Errorf("Expected to find precommit but got %v", vote)
} }
if vote := rs0.Commits.Get(0); vote == nil || vote.Type != VoteTypeCommit { if vote := rs0.Commits.GetById(0); vote == nil || vote.Type != VoteTypeCommit {
t.Errorf("Expected to find commit but got %v", vote) t.Errorf("Expected to find commit but got %v", vote)
} }
// Setup round 1 (next round) // Setup round 1 (next round)
cs.SetupRound(1) cs.SetupNewRound(1, 1)
<-cs.NewStepCh() // TODO: test this value too.
// Now the commit should be copied over to prevotes and precommits. // Now the commit should be copied over to prevotes and precommits.
rs1 := cs.GetRoundState() rs1 := cs.GetRoundState()
if vote := rs1.Prevotes.Get(0); vote == nil || vote.Type != VoteTypeCommit { if vote := rs1.Prevotes.GetById(0); vote == nil || vote.Type != VoteTypeCommit {
t.Errorf("Expected to find commit but got %v", vote) t.Errorf("Expected to find commit but got %v", vote)
} }
if vote := rs1.Precommits.Get(0); vote == nil || vote.Type != VoteTypeCommit { if vote := rs1.Precommits.GetById(0); vote == nil || vote.Type != VoteTypeCommit {
t.Errorf("Expected to find commit but got %v", vote) t.Errorf("Expected to find commit but got %v", vote)
} }
if vote := rs1.Commits.Get(0); vote == nil || vote.Type != VoteTypeCommit { if vote := rs1.Commits.GetById(0); vote == nil || vote.Type != VoteTypeCommit {
t.Errorf("Expected to find commit but got %v", vote) t.Errorf("Expected to find commit but got %v", vote)
} }
// Setup round 1 (should fail)
assertPanics(t, "Round did not increment", func() {
cs.SetupRound(1)
})
} }
func TestRunActionProposeNoPrivValidator(t *testing.T) { func TestRunActionProposeNoPrivValidator(t *testing.T) {
@ -154,18 +150,21 @@ func TestRunActionPrecommitCommitFinalize(t *testing.T) {
priv := NewPrivValidator(db_.NewMemDB(), privAccounts[0]) priv := NewPrivValidator(db_.NewMemDB(), privAccounts[0])
cs.SetPrivValidator(priv) cs.SetPrivValidator(priv)
vote := cs.RunActionPrecommit(1, 0) cs.RunActionPrecommit(1, 0)
if vote != nil { <-cs.NewStepCh() // TODO: test this value too.
if cs.Precommits.GetById(0) != nil {
t.Errorf("RunActionPrecommit should return nil without a proposal") t.Errorf("RunActionPrecommit should return nil without a proposal")
} }
cs.RunActionPropose(1, 0) cs.RunActionPropose(1, 0)
<-cs.NewStepCh() // TODO: test this value too.
// Test RunActionPrecommit failures: // Test RunActionPrecommit failures:
assertPanics(t, "Wrong height ", func() { cs.RunActionPrecommit(2, 0) }) assertPanics(t, "Wrong height ", func() { cs.RunActionPrecommit(2, 0) })
assertPanics(t, "Wrong round", func() { cs.RunActionPrecommit(1, 1) }) assertPanics(t, "Wrong round", func() { cs.RunActionPrecommit(1, 1) })
vote = cs.RunActionPrecommit(1, 0) cs.RunActionPrecommit(1, 0)
if vote != nil { <-cs.NewStepCh() // TODO: test this value too.
if cs.Precommits.GetById(0) != nil {
t.Errorf("RunActionPrecommit should return nil, not enough prevotes") t.Errorf("RunActionPrecommit should return nil, not enough prevotes")
} }
@ -176,21 +175,23 @@ func TestRunActionPrecommitCommitFinalize(t *testing.T) {
Round: 0, Round: 0,
Type: VoteTypePrevote, Type: VoteTypePrevote,
BlockHash: cs.ProposalBlock.Hash(), BlockHash: cs.ProposalBlock.Hash(),
BlockParts: cs.ProposalBlockParts.Header(),
} }
privAccounts[i].Sign(vote) privAccounts[i].Sign(vote)
cs.AddVote(vote) cs.AddVote(vote)
} }
// Test RunActionPrecommit success: // Test RunActionPrecommit success:
vote = cs.RunActionPrecommit(1, 0) cs.RunActionPrecommit(1, 0)
if vote == nil { <-cs.NewStepCh() // TODO: test this value too.
if cs.Precommits.GetById(0) == nil {
t.Errorf("RunActionPrecommit should have succeeded") t.Errorf("RunActionPrecommit should have succeeded")
} }
checkRoundState(t, cs, 1, 0, RoundStepPrecommit) checkRoundState(t, cs, 1, 0, RoundStepPrecommit)
// Test RunActionCommit failures: // Test RunActionCommit failures:
assertPanics(t, "Wrong height ", func() { cs.RunActionCommit(2, 0) }) assertPanics(t, "Wrong height ", func() { cs.RunActionCommit(2) })
assertPanics(t, "Wrong round", func() { cs.RunActionCommit(1, 1) }) assertPanics(t, "Wrong round", func() { cs.RunActionCommit(1) })
// Add at least +2/3 precommits. // Add at least +2/3 precommits.
for i := 0; i < 7; i++ { for i := 0; i < 7; i++ {
@ -199,14 +200,16 @@ func TestRunActionPrecommitCommitFinalize(t *testing.T) {
Round: 0, Round: 0,
Type: VoteTypePrecommit, Type: VoteTypePrecommit,
BlockHash: cs.ProposalBlock.Hash(), BlockHash: cs.ProposalBlock.Hash(),
BlockParts: cs.ProposalBlockParts.Header(),
} }
privAccounts[i].Sign(vote) privAccounts[i].Sign(vote)
cs.AddVote(vote) cs.AddVote(vote)
} }
// Test RunActionCommit success: // Test RunActionCommit success:
vote = cs.RunActionCommit(1, 0) cs.RunActionCommit(1)
if vote == nil { <-cs.NewStepCh() // TODO: test this value too.
if cs.Commits.GetById(0) == nil {
t.Errorf("RunActionCommit should have succeeded") t.Errorf("RunActionCommit should have succeeded")
} }
checkRoundState(t, cs, 1, 0, RoundStepCommit) checkRoundState(t, cs, 1, 0, RoundStepCommit)
@ -223,19 +226,14 @@ func TestRunActionPrecommitCommitFinalize(t *testing.T) {
Round: uint16(i), // Doesn't matter what round Round: uint16(i), // Doesn't matter what round
Type: VoteTypeCommit, Type: VoteTypeCommit,
BlockHash: cs.ProposalBlock.Hash(), BlockHash: cs.ProposalBlock.Hash(),
BlockParts: cs.ProposalBlockParts.Header(),
} }
privAccounts[i].Sign(vote) privAccounts[i].Sign(vote)
cs.AddVote(vote) cs.AddVote(vote)
} }
// Test RunActionCommitWait: // Test TryFinalizeCommit:
cs.RunActionCommitWait(1, 0) cs.TryFinalizeCommit(1)
if cs.CommitTime.IsZero() { <-cs.NewStepCh() // TODO: test this value too.
t.Errorf("Expected CommitTime to have been set") checkRoundState(t, cs, 2, 0, RoundStepNewHeight)
}
checkRoundState(t, cs, 1, 0, RoundStepCommitWait)
// Test RunActionFinalize:
cs.RunActionFinalize(1, 0)
checkRoundState(t, cs, 2, 0, RoundStepStart)
} }

83
consensus/vote_set.go
View File

@ -6,6 +6,7 @@ import (
"strings" "strings"
"sync" "sync"
. "github.com/tendermint/tendermint/binary"
. "github.com/tendermint/tendermint/blocks" . "github.com/tendermint/tendermint/blocks"
. "github.com/tendermint/tendermint/common" . "github.com/tendermint/tendermint/common"
"github.com/tendermint/tendermint/state" "github.com/tendermint/tendermint/state"
@ -26,10 +27,11 @@ type VoteSet struct {
vset *state.ValidatorSet vset *state.ValidatorSet
votes map[uint64]*Vote votes map[uint64]*Vote
votesBitArray BitArray votesBitArray BitArray
votesByBlockHash map[string]uint64 votesByBlock map[string]uint64 // string(blockHash)+string(blockParts) -> vote sum.
totalVotes uint64 totalVotes uint64
twoThirdsMajority []byte maj23Hash []byte
twoThirdsExists bool maj23Parts PartSetHeader
maj23Exists bool
} }
// Constructs a new VoteSet struct used to accumulate votes for each round. // Constructs a new VoteSet struct used to accumulate votes for each round.
@ -44,7 +46,7 @@ func NewVoteSet(height uint32, round uint16, type_ byte, vset *state.ValidatorSe
vset: vset, vset: vset,
votes: make(map[uint64]*Vote, vset.Size()), votes: make(map[uint64]*Vote, vset.Size()),
votesBitArray: NewBitArray(vset.Size()), votesBitArray: NewBitArray(vset.Size()),
votesByBlockHash: make(map[string]uint64), votesByBlock: make(map[string]uint64),
totalVotes: 0, totalVotes: 0,
} }
} }
@ -96,15 +98,17 @@ func (vs *VoteSet) addVote(vote *Vote) (bool, error) {
return false, ErrVoteInvalidAccount return false, ErrVoteInvalidAccount
} }
vs.votesBitArray.SetIndex(uint(voterIndex), true) vs.votesBitArray.SetIndex(uint(voterIndex), true)
totalBlockHashVotes := vs.votesByBlockHash[string(vote.BlockHash)] + val.VotingPower blockKey := string(vote.BlockHash) + string(BinaryBytes(vote.BlockParts))
vs.votesByBlockHash[string(vote.BlockHash)] = totalBlockHashVotes totalBlockHashVotes := vs.votesByBlock[blockKey] + val.VotingPower
vs.votesByBlock[blockKey] = totalBlockHashVotes
vs.totalVotes += val.VotingPower vs.totalVotes += val.VotingPower
// If we just nudged it up to two thirds majority, add it. // If we just nudged it up to two thirds majority, add it.
if totalBlockHashVotes > vs.vset.TotalVotingPower()*2/3 && if totalBlockHashVotes > vs.vset.TotalVotingPower()*2/3 &&
(totalBlockHashVotes-val.VotingPower) <= vs.vset.TotalVotingPower()*2/3 { (totalBlockHashVotes-val.VotingPower) <= vs.vset.TotalVotingPower()*2/3 {
vs.twoThirdsMajority = vote.BlockHash vs.maj23Hash = vote.BlockHash
vs.twoThirdsExists = true vs.maj23Parts = vote.BlockParts
vs.maj23Exists = true
} }
return true, nil return true, nil
@ -126,7 +130,19 @@ func (vs *VoteSet) BitArray() BitArray {
return vs.votesBitArray.Copy() return vs.votesBitArray.Copy()
} }
func (vs *VoteSet) Get(id uint64) *Vote { func (vs *VoteSet) GetByIndex(index uint) *Vote {
vs.mtx.Lock()
defer vs.mtx.Unlock()
id, val := vs.vset.GetByIndex(index)
if val == nil {
panic("GetByIndex(index) returned nil")
}
return vs.votes[id]
}
func (vs *VoteSet) GetById(id uint64) *Vote {
vs.mtx.Lock() vs.mtx.Lock()
defer vs.mtx.Unlock() defer vs.mtx.Unlock()
return vs.votes[id] return vs.votes[id]
@ -148,18 +164,18 @@ func (vs *VoteSet) HasTwoThirdsMajority() bool {
} }
vs.mtx.Lock() vs.mtx.Lock()
defer vs.mtx.Unlock() defer vs.mtx.Unlock()
return vs.twoThirdsExists return vs.maj23Exists
} }
// Returns either a blockhash (or nil) that received +2/3 majority. // Returns either a blockhash (or nil) that received +2/3 majority.
// If there exists no such majority, returns (nil, false). // If there exists no such majority, returns (nil, false).
func (vs *VoteSet) TwoThirdsMajority() (hash []byte, ok bool) { func (vs *VoteSet) TwoThirdsMajority() (hash []byte, parts PartSetHeader, ok bool) {
vs.mtx.Lock() vs.mtx.Lock()
defer vs.mtx.Unlock() defer vs.mtx.Unlock()
if vs.twoThirdsExists { if vs.maj23Exists {
return vs.twoThirdsMajority, true return vs.maj23Hash, vs.maj23Parts, true
} else { } else {
return nil, false return nil, PartSetHeader{}, false
} }
} }
@ -169,27 +185,30 @@ func (vs *VoteSet) MakePOL() *POL {
} }
vs.mtx.Lock() vs.mtx.Lock()
defer vs.mtx.Unlock() defer vs.mtx.Unlock()
if !vs.twoThirdsExists { if !vs.maj23Exists {
return nil return nil
} }
majHash := vs.twoThirdsMajority // hash may be nil.
pol := &POL{ pol := &POL{
Height: vs.height, Height: vs.height,
Round: vs.round, Round: vs.round,
BlockHash: majHash, BlockHash: vs.maj23Hash,
BlockParts: vs.maj23Parts,
} }
for _, vote := range vs.votes { for _, vote := range vs.votes {
if bytes.Equal(vote.BlockHash, majHash) { if !bytes.Equal(vote.BlockHash, vs.maj23Hash) {
continue
}
if !vote.BlockParts.Equals(vs.maj23Parts) {
continue
}
if vote.Type == VoteTypePrevote { if vote.Type == VoteTypePrevote {
pol.Votes = append(pol.Votes, vote.Signature) pol.Votes = append(pol.Votes, vote.Signature)
} else if vote.Type == VoteTypeCommit { } else if vote.Type == VoteTypeCommit {
pol.Commits = append(pol.Votes, vote.Signature) pol.Commits = append(pol.Commits, RoundSignature{vote.Round, vote.Signature})
pol.CommitRounds = append(pol.CommitRounds, vote.Round)
} else { } else {
Panicf("Unexpected vote type %X", vote.Type) Panicf("Unexpected vote type %X", vote.Type)
} }
} }
}
return pol return pol
} }
@ -199,18 +218,26 @@ func (vs *VoteSet) MakeValidation() Validation {
} }
vs.mtx.Lock() vs.mtx.Lock()
defer vs.mtx.Unlock() defer vs.mtx.Unlock()
if len(vs.twoThirdsMajority) == 0 { if len(vs.maj23Hash) == 0 {
panic("Cannot MakeValidation() unless a blockhash has +2/3") panic("Cannot MakeValidation() unless a blockhash has +2/3")
} }
sigs := []Signature{} rsigs := make([]RoundSignature, vs.vset.Size())
for _, vote := range vs.votes { vs.vset.Iterate(func(index uint, val *state.Validator) bool {
if !bytes.Equal(vote.BlockHash, vs.twoThirdsMajority) { vote := vs.votes[val.Id]
continue if vote == nil {
return false
} }
sigs = append(sigs, vote.Signature) if !bytes.Equal(vote.BlockHash, vs.maj23Hash) {
return false
} }
if !vote.BlockParts.Equals(vs.maj23Parts) {
return false
}
rsigs[index] = RoundSignature{vote.Round, vote.Signature}
return false
})
return Validation{ return Validation{
Signatures: sigs, Commits: rsigs,
} }
} }

36
consensus/vote_set_test.go
View File

@ -14,14 +14,14 @@ func TestAddVote(t *testing.T) {
// t.Logf(">> %v", voteSet) // t.Logf(">> %v", voteSet)
if voteSet.Get(0) != nil { if voteSet.GetById(0) != nil {
t.Errorf("Expected Get(0) to be nil") t.Errorf("Expected GetById(0) to be nil")
} }
if voteSet.BitArray().GetIndex(0) { if voteSet.BitArray().GetIndex(0) {
t.Errorf("Expected BitArray.GetIndex(0) to be false") t.Errorf("Expected BitArray.GetIndex(0) to be false")
} }
hash, ok := voteSet.TwoThirdsMajority() hash, header, ok := voteSet.TwoThirdsMajority()
if hash != nil || ok { if hash != nil || !header.IsZero() || ok {
t.Errorf("There should be no 2/3 majority") t.Errorf("There should be no 2/3 majority")
} }
@ -29,14 +29,14 @@ func TestAddVote(t *testing.T) {
privAccounts[0].Sign(vote) privAccounts[0].Sign(vote)
voteSet.Add(vote) voteSet.Add(vote)
if voteSet.Get(0) == nil { if voteSet.GetById(0) == nil {
t.Errorf("Expected Get(0) to be present") t.Errorf("Expected GetById(0) to be present")
} }
if !voteSet.BitArray().GetIndex(0) { if !voteSet.BitArray().GetIndex(0) {
t.Errorf("Expected BitArray.GetIndex(0) to be true") t.Errorf("Expected BitArray.GetIndex(0) to be true")
} }
hash, ok = voteSet.TwoThirdsMajority() hash, header, ok = voteSet.TwoThirdsMajority()
if hash != nil || ok { if hash != nil || !header.IsZero() || ok {
t.Errorf("There should be no 2/3 majority") t.Errorf("There should be no 2/3 majority")
} }
} }
@ -50,8 +50,8 @@ func Test2_3Majority(t *testing.T) {
privAccounts[i].Sign(vote) privAccounts[i].Sign(vote)
voteSet.Add(vote) voteSet.Add(vote)
} }
hash, ok := voteSet.TwoThirdsMajority() hash, header, ok := voteSet.TwoThirdsMajority()
if hash != nil || ok { if hash != nil || !header.IsZero() || ok {
t.Errorf("There should be no 2/3 majority") t.Errorf("There should be no 2/3 majority")
} }
@ -59,8 +59,8 @@ func Test2_3Majority(t *testing.T) {
vote.BlockHash = CRandBytes(32) vote.BlockHash = CRandBytes(32)
privAccounts[6].Sign(vote) privAccounts[6].Sign(vote)
voteSet.Add(vote) voteSet.Add(vote)
hash, ok = voteSet.TwoThirdsMajority() hash, header, ok = voteSet.TwoThirdsMajority()
if hash != nil || ok { if hash != nil || !header.IsZero() || ok {
t.Errorf("There should be no 2/3 majority") t.Errorf("There should be no 2/3 majority")
} }
@ -68,8 +68,8 @@ func Test2_3Majority(t *testing.T) {
vote.BlockHash = nil vote.BlockHash = nil
privAccounts[7].Sign(vote) privAccounts[7].Sign(vote)
voteSet.Add(vote) voteSet.Add(vote)
hash, ok = voteSet.TwoThirdsMajority() hash, header, ok = voteSet.TwoThirdsMajority()
if hash != nil || !ok { if hash != nil || !header.IsZero() || !ok {
t.Errorf("There should be 2/3 majority for nil") t.Errorf("There should be 2/3 majority for nil")
} }
@ -128,8 +128,8 @@ func TestAddCommitsToPrevoteVotes(t *testing.T) {
privAccounts[i].Sign(vote) privAccounts[i].Sign(vote)
voteSet.Add(vote) voteSet.Add(vote)
} }
hash, ok := voteSet.TwoThirdsMajority() hash, header, ok := voteSet.TwoThirdsMajority()
if hash != nil || ok { if hash != nil || !header.IsZero() || ok {
t.Errorf("There should be no 2/3 majority") t.Errorf("There should be no 2/3 majority")
} }
@ -174,8 +174,8 @@ func TestAddCommitsToPrevoteVotes(t *testing.T) {
} }
// We should have 2/3 majority // We should have 2/3 majority
hash, ok = voteSet.TwoThirdsMajority() hash, header, ok = voteSet.TwoThirdsMajority()
if hash != nil || !ok { if hash != nil || !header.IsZero() || !ok {
t.Errorf("There should be 2/3 majority for nil") t.Errorf("There should be 2/3 majority for nil")
} }

55
state/state.go
View File

@ -288,11 +288,56 @@ func (s *State) releaseValidator(accountId uint64) {
// at an invalid state. Copy the state before calling AppendBlock! // at an invalid state. Copy the state before calling AppendBlock!
func (s *State) AppendBlock(b *Block, checkStateHash bool) error { func (s *State) AppendBlock(b *Block, checkStateHash bool) error {
// Basic block validation. // Basic block validation.
// XXX We need to validate LastBlockParts too.
err := b.ValidateBasic(s.Height, s.BlockHash) err := b.ValidateBasic(s.Height, s.BlockHash)
if err != nil { if err != nil {
return err return err
} }
// Validate block Validation.
if b.Height == 1 {
if len(b.Validation.Commits) != 0 {
return errors.New("Block at height 1 (first block) should have no Validation commits")
}
} else {
if uint(len(b.Validation.Commits)) != s.BondedValidators.Size() {
return errors.New("Invalid block validation size")
}
var sumVotingPower uint64
s.BondedValidators.Iterate(func(index uint, val *Validator) bool {
rsig := b.Validation.Commits[index]
if rsig.IsZero() {
return false
} else {
if rsig.SignerId != val.Id {
err = errors.New("Invalid validation order")
return true
}
vote := &Vote{
Height: b.Height,
Round: rsig.Round,
Type: VoteTypeCommit,
BlockHash: b.LastBlockHash,
BlockParts: b.LastBlockParts,
Signature: rsig.Signature,
}
if val.Verify(vote) {
sumVotingPower += val.VotingPower
return false
} else {
err = errors.New("Invalid validation signature")
return true
}
}
})
if err != nil {
return err
}
if sumVotingPower <= s.BondedValidators.TotalVotingPower()*2/3 {
return errors.New("Insufficient validation voting power")
}
}
// Commit each tx // Commit each tx
for _, tx := range b.Data.Txs { for _, tx := range b.Data.Txs {
err := s.ExecTx(tx) err := s.ExecTx(tx)
@ -301,9 +346,9 @@ func (s *State) AppendBlock(b *Block, checkStateHash bool) error {
} }
} }
// Update LastCommitHeight as necessary. // Update Validator.LastCommitHeight as necessary.
for _, sig := range b.Validation.Signatures { for _, rsig := range b.Validation.Commits {
_, val := s.BondedValidators.GetById(sig.SignerId) _, val := s.BondedValidators.GetById(rsig.SignerId)
if val == nil { if val == nil {
return ErrStateInvalidSignature return ErrStateInvalidSignature
} }
@ -317,7 +362,7 @@ func (s *State) AppendBlock(b *Block, checkStateHash bool) error {
// If any unbonding periods are over, // If any unbonding periods are over,
// reward account with bonded coins. // reward account with bonded coins.
toRelease := []*Validator{} toRelease := []*Validator{}
s.UnbondingValidators.Iterate(func(val *Validator) bool { s.UnbondingValidators.Iterate(func(index uint, val *Validator) bool {
if val.UnbondHeight+unbondingPeriodBlocks < b.Height { if val.UnbondHeight+unbondingPeriodBlocks < b.Height {
toRelease = append(toRelease, val) toRelease = append(toRelease, val)
} }
@ -330,7 +375,7 @@ func (s *State) AppendBlock(b *Block, checkStateHash bool) error {
// If any validators haven't signed in a while, // If any validators haven't signed in a while,
// unbond them, they have timed out. // unbond them, they have timed out.
toTimeout := []*Validator{} toTimeout := []*Validator{}
s.BondedValidators.Iterate(func(val *Validator) bool { s.BondedValidators.Iterate(func(index uint, val *Validator) bool {
if val.LastCommitHeight+validatorTimeoutBlocks < b.Height { if val.LastCommitHeight+validatorTimeoutBlocks < b.Height {
toTimeout = append(toTimeout, val) toTimeout = append(toTimeout, val)
} }

1
state/state_test.go
View File

@ -89,6 +89,7 @@ func TestGenesisSaveLoad(t *testing.T) {
Height: 1, Height: 1,
StateHash: nil, StateHash: nil,
}, },
Validation: Validation{},
Data: Data{ Data: Data{
Txs: []Tx{}, Txs: []Tx{},
}, },

9
state/validator_set.go
View File

@ -127,9 +127,12 @@ func (vset *ValidatorSet) Remove(validatorId uint64) (val *Validator, removed bo
return val_.(*Validator), removed return val_.(*Validator), removed
} }
func (vset *ValidatorSet) Iterate(fn func(val *Validator) bool) { func (vset *ValidatorSet) Iterate(fn func(index uint, val *Validator) bool) {
index := uint(0)
vset.validators.Iterate(func(key_ interface{}, val_ interface{}) bool { vset.validators.Iterate(func(key_ interface{}, val_ interface{}) bool {
return fn(val_.(*Validator)) stop := fn(index, val_.(*Validator))
index++
return stop
}) })
} }
@ -139,7 +142,7 @@ func (vset *ValidatorSet) String() string {
func (vset *ValidatorSet) StringWithIndent(indent string) string { func (vset *ValidatorSet) StringWithIndent(indent string) string {
valStrings := []string{} valStrings := []string{}
vset.Iterate(func(val *Validator) bool { vset.Iterate(func(index uint, val *Validator) bool {
valStrings = append(valStrings, val.String()) valStrings = append(valStrings, val.String())
return false return false
}) })