Merge pull request #96 from getamis/feature/catchup_round_change

consensus/istanbul: catch up round before send round change

Author: markya0616

consensus/istanbul/core/core.go

@@ -193,7 +193,7 @@ func (c *core) commit() {
 		}
 
 		if err := c.backend.Commit(proposal, signatures); err != nil {
-			c.sendRoundChange()
+			c.sendNextRoundChange()
 			return
 		}
 	}
@@ -232,6 +232,7 @@ func (c *core) catchUpRound(view *istanbul.View) {
 	}
 	c.waitingForRoundChange = true
 	c.current = newRoundState(view, c.valSet)
+	c.roundChangeSet.Clear(view.Round)
 	c.newRoundChangeTimer()
 
 	logger.Trace("Catch up round", "new_round", view.Round, "new_seq", view.Sequence, "new_proposer", c.valSet)
@@ -266,7 +267,19 @@ func (c *core) newRoundChangeTimer() {
 
 	timeout := time.Duration(c.config.RequestTimeout) * time.Millisecond
 	c.roundChangeTimer = time.AfterFunc(timeout, func() {
-		c.sendRoundChange()
+		// If we're not waiting for round change yet, we can try to catch up
+		// the max round with F+1 round change message. We only need to catch up
+		// if the max round is larger than current round.
+		if !c.waitingForRoundChange {
+			maxRound := c.roundChangeSet.MaxRound(c.valSet.F() + 1)
+			if maxRound != nil && maxRound.Cmp(c.current.Round()) > 0 {
+				c.sendRoundChange(maxRound)
+			} else {
+				c.sendNextRoundChange()
+			}
+		} else {
+			c.sendNextRoundChange()
+		}
 	})
 }
 

consensus/istanbul/core/preprepare.go

@@ -70,7 +70,7 @@ func (c *core) handlePreprepare(msg *message, src istanbul.Validator) error {
 	// Verify the proposal we received
 	if err := c.backend.Verify(preprepare.Proposal); err != nil {
 		logger.Warn("Failed to verify proposal", "err", err)
-		c.sendRoundChange()
+		c.sendNextRoundChange()
 		return err
 	}
 

consensus/istanbul/core/roundchange.go

@@ -24,14 +24,26 @@ import (
 	"github.com/ethereum/go-ethereum/consensus/istanbul"
 )
 
-func (c *core) sendRoundChange() {
+// sendNextRoundChange sends the round change message with current round + 1
+func (c *core) sendNextRoundChange() {
+	cv := c.currentView()
+	c.sendRoundChange(new(big.Int).Add(cv.Round, common.Big1))
+}
+
+// sendRoundChange sends the round change message with the given round
+func (c *core) sendRoundChange(round *big.Int) {
 	logger := c.logger.New("state", c.state)
 	logger.Trace("sendRoundChange")
 
 	cv := c.currentView()
+	if cv.Round.Cmp(round) >= 0 {
+		logger.Error("Cannot send out the round change", "current round", cv.Round, "target round", round)
+		return
+	}
+
 	c.catchUpRound(&istanbul.View{
 		// The round number we'd like to transfer to.
-		Round:    new(big.Int).Add(cv.Round, common.Big1),
+		Round:    new(big.Int).Set(round),
 		Sequence: new(big.Int).Set(cv.Sequence),
 	})
 
@@ -82,10 +94,7 @@ func (c *core) handleRoundChange(msg *message, src istanbul.Validator) error {
 
 	// Add the round change message to its message set and return how many
 	// messages we've got with the same round number and sequence number.
-	num, err := c.roundChangeSet.Add(&istanbul.View{
-		Round:    new(big.Int).Set(rc.Round),
-		Sequence: new(big.Int).Set(rc.Sequence),
-	}, msg)
+	num, err := c.roundChangeSet.Add(rc.Round, msg)
 	if err != nil {
 		logger.Warn("Failed to add round change message", "from", src, "msg", msg, "err", err)
 		return err
@@ -94,13 +103,9 @@ func (c *core) handleRoundChange(msg *message, src istanbul.Validator) error {
 	// Once we received f+1 round change messages, those messages form a weak certificate.
 	// If our round number is smaller than the certificate's round number, we would
 	// try to catch up the round number.
-	if num == int(c.valSet.F()+1) {
+	if c.waitingForRoundChange && num == int(c.valSet.F()+1) {
 		if cv.Round.Cmp(rc.Round) < 0 {
-			c.catchUpRound(&istanbul.View{
-				Round:    new(big.Int).Sub(rc.Round, common.Big1),
-				Sequence: new(big.Int).Set(rc.Sequence),
-			})
-			c.sendRoundChange()
+			c.sendRoundChange(rc.Round)
 		}
 	}
 
@@ -120,49 +125,59 @@ func (c *core) handleRoundChange(msg *message, src istanbul.Validator) error {
 func newRoundChangeSet(valSet istanbul.ValidatorSet) *roundChangeSet {
 	return &roundChangeSet{
 		validatorSet: valSet,
-		roundChanges: make(map[common.Hash]*messageSet),
+		roundChanges: make(map[uint64]*messageSet),
 		mu:           new(sync.Mutex),
 	}
 }
 
 type roundChangeSet struct {
 	validatorSet istanbul.ValidatorSet
-	roundChanges map[common.Hash]*messageSet
+	roundChanges map[uint64]*messageSet
 	mu           *sync.Mutex
 }
 
-func (rcs *roundChangeSet) Add(v *istanbul.View, msg *message) (int, error) {
+// Add adds the round and message into round change set
+func (rcs *roundChangeSet) Add(r *big.Int, msg *message) (int, error) {
 	rcs.mu.Lock()
 	defer rcs.mu.Unlock()
 
-	h := istanbul.RLPHash(v)
-	if rcs.roundChanges[h] == nil {
-		rcs.roundChanges[h] = newMessageSet(rcs.validatorSet)
+	round := r.Uint64()
+	if rcs.roundChanges[round] == nil {
+		rcs.roundChanges[round] = newMessageSet(rcs.validatorSet)
 	}
-	err := rcs.roundChanges[h].Add(msg)
+	err := rcs.roundChanges[round].Add(msg)
 	if err != nil {
 		return 0, err
 	}
-	return rcs.roundChanges[h].Size(), nil
+	return rcs.roundChanges[round].Size(), nil
 }
 
-func (rcs *roundChangeSet) Clear(v *istanbul.View) {
+// Clear deletes the messages with smaller round
+func (rcs *roundChangeSet) Clear(round *big.Int) {
 	rcs.mu.Lock()
 	defer rcs.mu.Unlock()
 
 	for k, rms := range rcs.roundChanges {
-		if len(rms.Values()) == 0 {
+		if len(rms.Values()) == 0 || k < round.Uint64() {
 			delete(rcs.roundChanges, k)
 		}
+	}
+}
+
+// MaxRound returns the max round which the number of messages is equal or larger than num
+func (rcs *roundChangeSet) MaxRound(num int) *big.Int {
+	rcs.mu.Lock()
+	defer rcs.mu.Unlock()
 
-		var rc *roundChange
-		if err := rms.Values()[0].Decode(&rc); err != nil {
+	var maxRound *big.Int
+	for k, rms := range rcs.roundChanges {
+		if rms.Size() < num {
 			continue
 		}
-
-		if rc.Sequence.Cmp(v.Sequence) < 0 ||
-			(rc.Sequence.Cmp(v.Sequence) == 0 && rc.Round.Cmp(v.Round) < 0) {
-			delete(rcs.roundChanges, k)
+		r := big.NewInt(int64(k))
+		if maxRound == nil || maxRound.Cmp(r) < 0 {
+			maxRound = r
 		}
 	}
+	return maxRound
 }

consensus/istanbul/core/roundchange_test.go

@@ -0,0 +1,98 @@
+// Copyright 2017 AMIS Technologies
+// This file is part of the go-ethereum library.
+//
+// The go-ethereum library is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Lesser General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// The go-ethereum library is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU Lesser General Public License for more details.
+//
+// You should have received a copy of the GNU Lesser General Public License
+// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
+
+package core
+
+import (
+	"math/big"
+	"testing"
+
+	"github.com/ethereum/go-ethereum/common"
+	"github.com/ethereum/go-ethereum/consensus/istanbul"
+	"github.com/ethereum/go-ethereum/consensus/istanbul/validator"
+)
+
+func TestRoundChangeSet(t *testing.T) {
+	vset := validator.NewSet(generateValidators(4), istanbul.RoundRobin)
+	rc := newRoundChangeSet(vset)
+
+	view := &istanbul.View{
+		Sequence: big.NewInt(1),
+		Round:    big.NewInt(1),
+	}
+	r := &roundChange{
+		Round:    view.Round,
+		Sequence: view.Sequence,
+		Digest:   common.Hash{},
+	}
+	m, _ := Encode(r)
+
+	// Test Add()
+	// Add message from all validators
+	for i, v := range vset.List() {
+		msg := &message{
+			Code:    msgRoundChange,
+			Msg:     m,
+			Address: v.Address(),
+		}
+		rc.Add(view.Round, msg)
+		if rc.roundChanges[view.Round.Uint64()].Size() != i+1 {
+			t.Errorf("unexpected round change message size, got: %v, expected: %v",
+				rc.roundChanges[view.Round.Uint64()].Size(), i+1)
+		}
+	}
+
+	// Add message again from all validators, but the size should be the same
+	for _, v := range vset.List() {
+		msg := &message{
+			Code:    msgRoundChange,
+			Msg:     m,
+			Address: v.Address(),
+		}
+		rc.Add(view.Round, msg)
+		if rc.roundChanges[view.Round.Uint64()].Size() != vset.Size() {
+			t.Errorf("unexpected round change message size, got: %v, expected: %v",
+				rc.roundChanges[view.Round.Uint64()].Size(), vset.Size())
+		}
+	}
+
+	// Test MaxRound()
+	for i := 0; i < 10; i++ {
+		maxRound := rc.MaxRound(i)
+		if i <= vset.Size() {
+			if maxRound == nil || maxRound.Cmp(view.Round) != 0 {
+				t.Errorf("unexpected max round, got: %v, expected: %v", maxRound, view.Round)
+			}
+		} else if maxRound != nil {
+			t.Errorf("max round should be nil, but got: %v", maxRound)
+
+		}
+	}
+
+	// Test Clear()
+	for i := int64(0); i < 2; i++ {
+		rc.Clear(big.NewInt(i))
+		if rc.roundChanges[view.Round.Uint64()].Size() != vset.Size() {
+			t.Errorf("unexpected round change message size, got: %v, expected: %v",
+				rc.roundChanges[view.Round.Uint64()].Size(), vset.Size())
+		}
+	}
+	rc.Clear(big.NewInt(2))
+	if rc.roundChanges[view.Round.Uint64()] != nil {
+		t.Errorf("round change set should be nil, but got: %v",
+			rc.roundChanges[view.Round.Uint64()])
+	}
+}

consensus/istanbul/validator/default.go

@@ -17,12 +17,11 @@
 package validator
 
 import (
+	"math"
 	"reflect"
 	"sort"
 	"sync"
 
-	"math"
-
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/consensus/istanbul"
 )