Merge pull request #148 from getamis/feature/round_change_backlog

consensus/istanbul: add backlog for roundChange messages

Author: markya0616

consensus/istanbul/core/backlog.go

@@ -40,6 +40,15 @@ func (c *core) checkMessage(msgCode uint64, view *istanbul.View) error {
 		return errInvalidMessage
 	}
 
+	if msgCode == msgRoundChange {
+		if view.Sequence.Cmp(c.currentView().Sequence) > 0 {
+			return errFutureMessage
+		} else if view.Cmp(c.currentView()) < 0 {
+			return errOldMessage
+		}
+		return nil
+	}
+
 	if view.Cmp(c.currentView()) > 0 {
 		return errFutureMessage
 	}
@@ -90,7 +99,7 @@ func (c *core) storeBacklog(msg *message, src istanbul.Validator) {
 		if err == nil {
 			backlog.Push(msg, toPriority(msg.Code, p.View))
 		}
-		// for istanbul.MsgPrepare and istanbul.MsgCommit cases
+		// for msgRoundChange, msgPrepare and msgCommit cases
 	default:
 		var p *istanbul.Subject
 		err := msg.Decode(&p)
@@ -127,7 +136,7 @@ func (c *core) processBacklog() {
 				if err == nil {
 					view = m.View
 				}
-				// for istanbul.MsgPrepare and istanbul.MsgCommit cases
+				// for msgRoundChange, msgPrepare and msgCommit cases
 			default:
 				var sub *istanbul.Subject
 				err := msg.Decode(&sub)
@@ -162,6 +171,10 @@ func (c *core) processBacklog() {
 }
 
 func toPriority(msgCode uint64, view *istanbul.View) float32 {
+	if msgCode == msgRoundChange {
+		// For msgRoundChange, set the message priority based on its sequence
+		return -float32(view.Sequence.Uint64() * 1000)
+	}
 	// FIXME: round will be reset as 0 while new sequence
 	// 10 * Round limits the range of message code is from 0 to 9
 	// 1000 * Sequence limits the range of round is from 0 to 99

consensus/istanbul/core/backlog_test.go

@@ -47,7 +47,7 @@ func TestCheckMessage(t *testing.T) {
 	}
 
 	testStates := []State{StateAcceptRequest, StatePreprepared, StatePrepared, StateCommitted}
-	testCode := []uint64{msgPreprepare, msgPrepare, msgCommit}
+	testCode := []uint64{msgPreprepare, msgPrepare, msgCommit, msgRoundChange}
 
 	// future sequence
 	v := &istanbul.View{
@@ -73,7 +73,11 @@ func TestCheckMessage(t *testing.T) {
 		c.state = testStates[i]
 		for j := 0; j < len(testCode); j++ {
 			err := c.checkMessage(testCode[j], v)
-			if err != errFutureMessage {
+			if testCode[j] == msgRoundChange {
+				if err != nil {
+					t.Errorf("error mismatch: have %v, want nil", err)
+				}
+			} else if err != errFutureMessage {
 				t.Errorf("error mismatch: have %v, want %v", err, errFutureMessage)
 			}
 		}
@@ -89,7 +93,11 @@ func TestCheckMessage(t *testing.T) {
 		c.state = testStates[i]
 		for j := 0; j < len(testCode); j++ {
 			err := c.checkMessage(testCode[j], v)
-			if err != errFutureMessage {
+			if testCode[j] == msgRoundChange {
+				if err != nil {
+					t.Errorf("error mismatch: have %v, want nil", err)
+				}
+			} else if err != errFutureMessage {
 				t.Errorf("error mismatch: have %v, want %v", err, errFutureMessage)
 			}
 		}
@@ -101,7 +109,11 @@ func TestCheckMessage(t *testing.T) {
 	c.state = StateAcceptRequest
 	for i := 0; i < len(testCode); i++ {
 		err = c.checkMessage(testCode[i], v)
-		if testCode[i] == msgPreprepare {
+		if testCode[i] == msgRoundChange {
+			if err != nil {
+				t.Errorf("error mismatch: have %v, want nil", err)
+			}
+		} else if testCode[i] == msgPreprepare {
 			if err != nil {
 				t.Errorf("error mismatch: have %v, want nil", err)
 			}
@@ -116,7 +128,11 @@ func TestCheckMessage(t *testing.T) {
 	c.state = StatePreprepared
 	for i := 0; i < len(testCode); i++ {
 		err = c.checkMessage(testCode[i], v)
-		if err != nil {
+		if testCode[i] == msgRoundChange {
+			if err != nil {
+				t.Errorf("error mismatch: have %v, want nil", err)
+			}
+		} else if err != nil {
 			t.Errorf("error mismatch: have %v, want nil", err)
 		}
 	}
@@ -125,7 +141,11 @@ func TestCheckMessage(t *testing.T) {
 	c.state = StatePrepared
 	for i := 0; i < len(testCode); i++ {
 		err = c.checkMessage(testCode[i], v)
-		if err != nil {
+		if testCode[i] == msgRoundChange {
+			if err != nil {
+				t.Errorf("error mismatch: have %v, want nil", err)
+			}
+		} else if err != nil {
 			t.Errorf("error mismatch: have %v, want nil", err)
 		}
 	}
@@ -134,7 +154,11 @@ func TestCheckMessage(t *testing.T) {
 	c.state = StateCommitted
 	for i := 0; i < len(testCode); i++ {
 		err = c.checkMessage(testCode[i], v)
-		if err != nil {
+		if testCode[i] == msgRoundChange {
+			if err != nil {
+				t.Errorf("error mismatch: have %v, want nil", err)
+			}
+		} else if err != nil {
 			t.Errorf("error mismatch: have %v, want nil", err)
 		}
 	}
@@ -195,6 +219,17 @@ func TestStoreBacklog(t *testing.T) {
 	if !reflect.DeepEqual(msg, m) {
 		t.Errorf("message mismatch: have %v, want %v", msg, m)
 	}
+
+	// push roundChange msg
+	m = &message{
+		Code: msgRoundChange,
+		Msg:  subjectPayload,
+	}
+	c.storeBacklog(m, p)
+	msg = c.backlogs[p].PopItem()
+	if !reflect.DeepEqual(msg, m) {
+		t.Errorf("message mismatch: have %v, want %v", msg, m)
+	}
 }
 
 func TestProcessFutureBacklog(t *testing.T) {
@@ -276,6 +311,10 @@ func TestProcessBacklog(t *testing.T) {
 			Code: msgCommit,
 			Msg:  subjectPayload,
 		},
+		&message{
+			Code: msgRoundChange,
+			Msg:  subjectPayload,
+		},
 	}
 	for i := 0; i < len(msgs); i++ {
 		testProcessBacklog(t, msgs[i])

consensus/istanbul/core/handler.go

@@ -172,7 +172,7 @@ func (c *core) handleCheckedMsg(msg *message, src istanbul.Validator) error {
 	case msgCommit:
 		return testBacklog(c.handleCommit(msg, src))
 	case msgRoundChange:
-		return c.handleRoundChange(msg, src)
+		return testBacklog(c.handleRoundChange(msg, src))
 	default:
 		logger.Error("Invalid message", "msg", msg)
 	}

consensus/istanbul/core/prepare.go

@@ -59,7 +59,7 @@ func (c *core) handlePrepare(msg *message, src istanbul.Validator) error {
 
 	// Change to Prepared state if we've received enough PREPARE messages or it is locked
 	// and we are in earlier state before Prepared state.
-	if ((c.current.IsHashLocked() && prepare.Digest == c.current.GetLockedHash()) || c.current.Prepares.Size() > 2*c.valSet.F()) &&
+	if ((c.current.IsHashLocked() && prepare.Digest == c.current.GetLockedHash()) || c.current.GetPrepareOrCommitSize() > 2*c.valSet.F()) &&
 		c.state.Cmp(StatePrepared) < 0 {
 		c.current.LockHash()
 		c.setState(StatePrepared)

consensus/istanbul/core/roundchange.go

@@ -48,10 +48,9 @@ func (c *core) sendRoundChange(round *big.Int) {
 
 	// Now we have the new round number and sequence number
 	cv = c.currentView()
-	rc := &roundChange{
-		Round:    new(big.Int).Set(cv.Round),
-		Sequence: new(big.Int).Set(cv.Sequence),
-		Digest:   common.Hash{},
+	rc := &istanbul.Subject{
+		View:   cv,
+		Digest: common.Hash{},
 	}
 
 	payload, err := Encode(rc)
@@ -70,29 +69,22 @@ func (c *core) handleRoundChange(msg *message, src istanbul.Validator) error {
 	logger := c.logger.New("state", c.state, "from", src.Address().Hex())
 
 	// Decode ROUND CHANGE message
-	var rc *roundChange
+	var rc *istanbul.Subject
 	if err := msg.Decode(&rc); err != nil {
 		logger.Error("Failed to decode ROUND CHANGE", "err", err)
 		return errInvalidMessage
 	}
 
-	cv := c.currentView()
-
-	// We never accept ROUND CHANGE message with different sequence number
-	if rc.Sequence.Cmp(cv.Sequence) != 0 {
-		logger.Warn("Inconsistent sequence number", "expected", cv.Sequence, "got", rc.Sequence)
-		return errInvalidMessage
+	if err := c.checkMessage(msgRoundChange, rc.View); err != nil {
+		return err
 	}
 
-	// We never accept ROUND CHANGE message with smaller round number
-	if rc.Round.Cmp(cv.Round) < 0 {
-		logger.Warn("Old round change", "from", src, "expected", cv.Round, "got", rc.Round)
-		return errOldMessage
-	}
+	cv := c.currentView()
+	roundView := rc.View
 
 	// 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(rc.Round, msg)
+	num, err := c.roundChangeSet.Add(roundView.Round, msg)
 	if err != nil {
 		logger.Warn("Failed to add round change message", "from", src, "msg", msg, "err", err)
 		return err
@@ -102,18 +94,17 @@ func (c *core) handleRoundChange(msg *message, src istanbul.Validator) error {
 	// If our round number is smaller than the certificate's round number, we would
 	// try to catch up the round number.
 	if c.waitingForRoundChange && num == int(c.valSet.F()+1) {
-		if cv.Round.Cmp(rc.Round) < 0 {
-			c.sendRoundChange(rc.Round)
+		if cv.Round.Cmp(roundView.Round) < 0 {
+			c.sendRoundChange(roundView.Round)
 		}
 		return nil
-	} else if num == int(2*c.valSet.F()+1) && (c.waitingForRoundChange || cv.Round.Cmp(rc.Round) < 0) {
+	} else if num == int(2*c.valSet.F()+1) && (c.waitingForRoundChange || cv.Round.Cmp(roundView.Round) < 0) {
 		// We've received 2f+1 ROUND CHANGE messages, start a new round immediately.
-		c.startNewRound(rc.Round)
+		c.startNewRound(roundView.Round)
 		return nil
-	} else if cv.Round.Cmp(rc.Round) < 0 {
-		// We consider the message with larger round as future messages and not
-		// gossip it to other validators.
-		return errFutureMessage
+	} else if cv.Round.Cmp(roundView.Round) < 0 {
+		// Only gossip the message with current round to other validators.
+		return errIgnored
 	}
 	return nil
 }

consensus/istanbul/core/roundchange_test.go

@@ -33,10 +33,9 @@ func TestRoundChangeSet(t *testing.T) {
 		Sequence: big.NewInt(1),
 		Round:    big.NewInt(1),
 	}
-	r := &roundChange{
-		Round:    view.Round,
-		Sequence: view.Sequence,
-		Digest:   common.Hash{},
+	r := &istanbul.Subject{
+		View:   view,
+		Digest: common.Hash{},
 	}
 	m, _ := Encode(r)
 

consensus/istanbul/core/types.go

@@ -19,7 +19,6 @@ package core
 import (
 	"fmt"
 	"io"
-	"math/big"
 
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/rlp"
@@ -163,34 +162,3 @@ func (m *message) String() string {
 func Encode(val interface{}) ([]byte, error) {
 	return rlp.EncodeToBytes(val)
 }
-
-// ----------------------------------------------------------------------------
-
-type roundChange struct {
-	Round    *big.Int
-	Sequence *big.Int
-	Digest   common.Hash
-}
-
-// EncodeRLP serializes rc into the Ethereum RLP format.
-func (rc *roundChange) EncodeRLP(w io.Writer) error {
-	return rlp.Encode(w, []interface{}{
-		rc.Round,
-		rc.Sequence,
-		rc.Digest,
-	})
-}
-
-// DecodeRLP implements rlp.Decoder, and load the consensus fields from a RLP stream.
-func (rc *roundChange) DecodeRLP(s *rlp.Stream) error {
-	var rawRoundChange struct {
-		Round    *big.Int
-		Sequence *big.Int
-		Digest   common.Hash
-	}
-	if err := s.Decode(&rawRoundChange); err != nil {
-		return err
-	}
-	rc.Round, rc.Sequence, rc.Digest = rawRoundChange.Round, rawRoundChange.Sequence, rawRoundChange.Digest
-	return nil
-}

consensus/istanbul/core/types_test.go

@@ -172,47 +172,8 @@ func testSubjectWithSignature(t *testing.T) {
 	}
 }
 
-func testRoundChange(t *testing.T) {
-	rc := &roundChange{
-		Round:    big.NewInt(1),
-		Sequence: big.NewInt(2),
-		Digest:   common.StringToHash("1234567890"),
-	}
-	RoundChangePayload, _ := Encode(rc)
-
-	m := &message{
-		Code:    msgRoundChange,
-		Msg:     RoundChangePayload,
-		Address: common.HexToAddress("0x1234567890"),
-	}
-
-	msgPayload, err := m.Payload()
-	if err != nil {
-		t.Errorf("error mismatch: have %v, want nil", err)
-	}
-
-	decodedMsg := new(message)
-	err = decodedMsg.FromPayload(msgPayload, nil)
-	if err != nil {
-		t.Errorf("error mismatch: have %v, want nil", err)
-	}
-
-	var decodedRC *roundChange
-	err = decodedMsg.Decode(&decodedRC)
-	if err != nil {
-		t.Errorf("error mismatch: have %v, want nil", err)
-	}
-
-	// if block is encoded/decoded by rlp, we cannot to compare interface data type using reflect.DeepEqual. (like istanbul.Proposal)
-	// so individual comparison here.
-	if !reflect.DeepEqual(rc, decodedRC) {
-		t.Errorf("round change mismatch: have %v, want %v", decodedRC, rc)
-	}
-}
-
 func TestMessageEncodeDecode(t *testing.T) {
 	testPreprepare(t)
 	testSubject(t)
 	testSubjectWithSignature(t)
-	testRoundChange(t)
 }