consensus/istanbul: add block lock

Author: markya0616

consensus/istanbul/core/backlog_test.go

@@ -37,7 +37,7 @@ func TestCheckMessage(t *testing.T) {
 		current: newRoundState(&istanbul.View{
 			Sequence: big.NewInt(1),
 			Round:    big.NewInt(0),
-		}, newTestValidatorSet(4)),
+		}, newTestValidatorSet(4), common.Hash{}, nil),
 	}
 
 	// invalid view format
@@ -209,7 +209,7 @@ func TestProcessFutureBacklog(t *testing.T) {
 		current: newRoundState(&istanbul.View{
 			Sequence: big.NewInt(1),
 			Round:    big.NewInt(0),
-		}, newTestValidatorSet(4)),
+		}, newTestValidatorSet(4), common.Hash{}, nil),
 		state: StateAcceptRequest,
 	}
 	c.subscribeEvents()
@@ -297,7 +297,7 @@ func testProcessBacklog(t *testing.T, msg *message) {
 		current: newRoundState(&istanbul.View{
 			Sequence: big.NewInt(1),
 			Round:    big.NewInt(0),
-		}, newTestValidatorSet(4)),
+		}, newTestValidatorSet(4), common.Hash{}, nil),
 	}
 	c.subscribeEvents()
 	defer c.unsubscribeEvents()

consensus/istanbul/core/commit.go

@@ -19,13 +19,26 @@ package core
 import (
 	"reflect"
 
+	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/consensus/istanbul"
 )
 
 func (c *core) sendCommit() {
+	sub := c.current.Subject()
+	c.broadcastCommit(sub)
+}
+
+func (c *core) sendCommitForOldBlock(view *istanbul.View, digest common.Hash) {
+	sub := &istanbul.Subject{
+		View:   view,
+		Digest: digest,
+	}
+	c.broadcastCommit(sub)
+}
+
+func (c *core) broadcastCommit(sub *istanbul.Subject) {
 	logger := c.logger.New("state", c.state)
 
-	sub := c.current.Subject()
 	encodedSubject, err := Encode(sub)
 	if err != nil {
 		logger.Error("Failed to encode", "subject", sub)
@@ -55,11 +68,13 @@ func (c *core) handleCommit(msg *message, src istanbul.Validator) error {
 
 	c.acceptCommit(msg, src)
 
-	// Commit the proposal once we have enough commit messages and we are not in StateCommitted.
+	// Commit the proposal once we have enough COMMIT messages and we are not in the Committed state.
 	//
 	// If we already have a proposal, we may have chance to speed up the consensus process
-	// by committing the proposal without prepare messages.
+	// by committing the proposal without PREPARE messages.
 	if c.current.Commits.Size() > 2*c.valSet.F() && c.state.Cmp(StateCommitted) < 0 {
+		// Still need to call LockHash here since state can skip Prepared state and jump directly to the Committed state.
+		c.current.LockHash()
 		c.commit()
 	}
 

consensus/istanbul/core/commit_test.go

@@ -178,6 +178,9 @@ OUTER:
 				if err != test.expectedErr {
 					t.Errorf("error mismatch: have %v, want %v", err, test.expectedErr)
 				}
+				if r0.current.IsHashLocked() {
+					t.Errorf("block should not be locked")
+				}
 				continue OUTER
 			}
 		}
@@ -191,7 +194,9 @@ OUTER:
 			if r0.current.Commits.Size() > 2*r0.valSet.F() {
 				t.Errorf("the size of commit messages should be less than %v", 2*r0.valSet.F()+1)
 			}
-
+			if r0.current.IsHashLocked() {
+				t.Errorf("block should not be locked")
+			}
 			continue
 		}
 
@@ -214,6 +219,9 @@ OUTER:
 		if signedCount <= 2*r0.valSet.F() {
 			t.Errorf("the expected signed count should be larger than %v, but got %v", 2*r0.valSet.F(), signedCount)
 		}
+		if !r0.current.IsHashLocked() {
+			t.Errorf("block should be locked")
+		}
 	}
 }
 

consensus/istanbul/core/core.go

@@ -25,6 +25,7 @@ import (
 
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/consensus/istanbul"
+	"github.com/ethereum/go-ethereum/core/types"
 	"github.com/ethereum/go-ethereum/event"
 	"github.com/ethereum/go-ethereum/log"
 	metrics "github.com/ethereum/go-ethereum/metrics"
@@ -171,6 +172,7 @@ func (c *core) commit() {
 		}
 
 		if err := c.backend.Commit(proposal, committedSeals); err != nil {
+			c.current.UnlockHash() //Unlock block when insertion fails
 			c.sendNextRoundChange()
 			return
 		}
@@ -203,13 +205,21 @@ func (c *core) startNewRound(newView *istanbul.View, lastProposal istanbul.Propo
 	// Clear invalid ROUND CHANGE messages
 	c.roundChangeSet = newRoundChangeSet(c.valSet)
 	// New snapshot for new round
-	c.current = newRoundState(newView, c.valSet)
+	c.updateRoundState(newView, c.valSet, roundChange)
 	// Calculate new proposer
 	c.valSet.CalcProposer(c.lastProposer, newView.Round.Uint64())
 	c.waitingForRoundChange = false
 	c.setState(StateAcceptRequest)
 	if roundChange && c.isProposer() {
-		c.backend.NextRound()
+		// If it is locked, propose the old proposal
+		if c.current != nil && c.current.IsHashLocked() {
+			r := &istanbul.Request{
+				Proposal: c.current.Proposal(), //c.current.Proposal would be the locked proposal by previous proposer, see updateRoundState
+			}
+			c.sendPreprepare(r)
+		} else {
+			c.backend.NextRound()
+		}
 	}
 	c.newRoundChangeTimer()
 
@@ -223,13 +233,25 @@ func (c *core) catchUpRound(view *istanbul.View) {
 		c.roundMeter.Mark(new(big.Int).Sub(view.Round, c.current.Round()).Int64())
 	}
 	c.waitingForRoundChange = true
-	c.current = newRoundState(view, c.valSet)
+
+	// Need to keep block locked for round catching up
+	c.updateRoundState(view, c.valSet, true)
 	c.roundChangeSet.Clear(view.Round)
 	c.newRoundChangeTimer()
 
 	logger.Trace("Catch up round", "new_round", view.Round, "new_seq", view.Sequence, "new_proposer", c.valSet)
 }
 
+// updateRoundState updates round state by checking if locking block is necessary
+func (c *core) updateRoundState(view *istanbul.View, validatorSet istanbul.ValidatorSet, roundChange bool) {
+	// Lock only if both roundChange is true and it is locked
+	if roundChange && c.current != nil && c.current.IsHashLocked() {
+		c.current = newRoundState(view, validatorSet, c.current.GetLockedHash(), c.current.Preprepare)
+	} else {
+		c.current = newRoundState(view, validatorSet, common.Hash{}, nil)
+	}
+}
+
 func (c *core) setState(state State) {
 	if c.state != state {
 		c.state = state

consensus/istanbul/core/prepare.go

@@ -49,15 +49,18 @@ func (c *core) handlePrepare(msg *message, src istanbul.Validator) error {
 		return err
 	}
 
+	// If it is locked, it can only process on the locked block.
+	// Passing verifyPrepare and checkMessage implies it is processing on the locked block since it was verified in the Preprepared state.
 	if err := c.verifyPrepare(prepare, src); err != nil {
 		return err
 	}
 
 	c.acceptPrepare(msg, src)
 
-	// Change to Prepared state if we've received enough PREPARE messages
-	// and we are in earlier state before Prepared state
-	if c.current.Prepares.Size() > 2*c.valSet.F() && c.state.Cmp(StatePrepared) < 0 {
+	// 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() || c.current.Prepares.Size() > 2*c.valSet.F()) && c.state.Cmp(StatePrepared) < 0 {
+		c.current.LockHash()
 		c.setState(StatePrepared)
 		c.sendCommit()
 	}

consensus/istanbul/core/prepare_test.go

@@ -201,6 +201,9 @@ OUTER:
 				if err != test.expectedErr {
 					t.Errorf("error mismatch: have %v, want %v", err, test.expectedErr)
 				}
+				if r0.current.IsHashLocked() {
+					t.Errorf("block should not be locked")
+				}
 				continue OUTER
 			}
 		}
@@ -214,6 +217,9 @@ OUTER:
 			if r0.current.Prepares.Size() > 2*r0.valSet.F() {
 				t.Errorf("the size of PREPARE messages should be less than %v", 2*r0.valSet.F()+1)
 			}
+			if r0.current.IsHashLocked() {
+				t.Errorf("block should not be locked")
+			}
 
 			continue
 		}
@@ -246,6 +252,9 @@ OUTER:
 		if !reflect.DeepEqual(m, expectedSubject) {
 			t.Errorf("subject mismatch: have %v, want %v", m, expectedSubject)
 		}
+		if !r0.current.IsHashLocked() {
+			t.Errorf("block should be locked")
+		}
 	}
 }
 

consensus/istanbul/core/preprepare.go

@@ -86,9 +86,17 @@ func (c *core) handlePreprepare(msg *message, src istanbul.Validator) error {
 
 	// Here is about to accept the PRE-PREPARE
 	if c.state == StateAcceptRequest {
-		c.acceptPreprepare(preprepare)
-		c.setState(StatePreprepared)
-		c.sendPrepare()
+		// Send ROUND CHANGE if the locked proposal and the received proposal are different
+		if c.current.IsHashLocked() && preprepare.Proposal.Hash() != c.current.GetLockedHash() {
+			c.sendNextRoundChange()
+		} else {
+			// Either
+			//   1. the locked proposal and the received proposal match
+			//   2. we have no locked proposal
+			c.acceptPreprepare(preprepare)
+			c.setState(StatePreprepared)
+			c.sendPrepare()
+		}
 	}
 
 	return nil