Merge pull request #60 from getamis/feature/implement-round-change-protocol

consensus/pbft: implement round change protocol

Author: alanchchen

consensus/pbft/backend.go

@@ -41,8 +41,8 @@ type Backend interface {
 	// Commit is to deliver a final result to write into blockchain
 	Commit(Proposal) error
 
-	// ViewChanged is called when view change occurred
-	ViewChanged(needNewProposal bool) error
+	// RoundChanged is called when round change occurred
+	RoundChanged(needNewProposal bool) error
 
 	// Verify is to verify the proposal request
 	Verify(Proposal) error

consensus/pbft/backends/simple/backend.go

@@ -121,7 +121,7 @@ func (sb *simpleBackend) Broadcast(payload []byte) error {
 
 // Commit implements pbft.Backend.Commit
 func (sb *simpleBackend) Commit(proposal pbft.Proposal) error {
-	sb.logger.Info("Committed", "address", sb.Address().Hex(), "proposal", proposal)
+	sb.logger.Info("Committed", "address", sb.Address().Hex(), "hash", proposal.Hash(), "number", proposal.Number().Uint64())
 	// step1: update validator set from extra data of block
 	// step2: insert chain
 	block := &types.Block{}
@@ -145,8 +145,8 @@ func (sb *simpleBackend) Commit(proposal pbft.Proposal) error {
 	return sb.inserter(block)
 }
 
-// ViewChanged implements pbft.Backend.ViewChanged
-func (sb *simpleBackend) ViewChanged(needNewProposal bool) error {
+// RoundChanged implements pbft.Backend.RoundChanged
+func (sb *simpleBackend) RoundChanged(needNewProposal bool) error {
 	if needNewProposal {
 		// if I was or I'm a proposer now, fire a ChainHeadEvent to trigger next Seal()
 		if !common.EmptyHash(sb.proposedBlockHash) || sb.IsProposer() {

consensus/pbft/backends/simple/engine_test.go

@@ -166,7 +166,7 @@ func TestSealRoundChange(t *testing.T) {
 			if !ok {
 				t.Errorf("unexpected event comes, got: %v, expected: pbft.RequestEvent", reflect.TypeOf(ev.Data))
 			}
-			engine.ViewChanged(false)
+			engine.RoundChanged(true)
 		}
 		eventSub.Unsubscribe()
 	}

consensus/pbft/config.go

@@ -31,7 +31,7 @@ type Config struct {
 }
 
 var DefaultConfig = &Config{
-	RequestTimeoutMsec: 3000,
+	RequestTimeoutMsec: 10000,
 	BlockPeriod:        1,
 	BlockPauseTime:     2,
 	ProposerPolicy:     RoundRobin,

consensus/pbft/core/backlog.go

@@ -26,7 +26,7 @@ var (
 		StateAcceptRequest: msgPreprepare,
 		StatePreprepared:   msgPrepare,
 		StatePrepared:      msgCommit,
-		StateCommitted:     msgNewView,
+		StateCommitted:     msgAll,
 	}
 )
 
@@ -56,7 +56,7 @@ func (c *core) storeBacklog(msg *message, src pbft.Validator) {
 		return
 	}
 
-	logger.Debug("Store future message")
+	logger.Trace("Store future message")
 
 	c.backlogsMu.Lock()
 	defer c.backlogsMu.Unlock()
@@ -123,15 +123,15 @@ func (c *core) processBacklog() {
 			}
 			// Push back if it's a future message
 			if c.isFutureMessage(msg.Code, view) {
-				logger.Debug("Stop processing backlog", "msg", msg)
+				logger.Trace("Stop processing backlog", "msg", msg)
 				backlog.Push(msg, prio)
 				isFuture = true
 				break
 			}
 
-			logger.Debug("Post backlog event", "msg", msg)
+			logger.Trace("Post backlog event", "msg", msg)
 
-			go c.sendInternalEvent(backlogEvent{
+			go c.sendEvent(backlogEvent{
 				src: src,
 				msg: msg,
 			})

consensus/pbft/core/backlog_test.go

@@ -33,9 +33,11 @@ import (
 
 func TestIsFutureMessage(t *testing.T) {
 	c := &core{
-		state:    StateAcceptRequest,
-		sequence: big.NewInt(1),
-		round:    big.NewInt(0),
+		state: StateAcceptRequest,
+		current: newSnapshot(&pbft.View{
+			Sequence: big.NewInt(1),
+			Round:    big.NewInt(0),
+		}, newTestValidatorSet(4)),
 	}
 
 	// invalid view format
@@ -147,7 +149,7 @@ func TestStoreBacklog(t *testing.T) {
 	p := validator.New(common.StringToAddress("12345667890"))
 	// push preprepare msg
 	preprepare := &pbft.Preprepare{
-		View: v,
+		View:     v,
 		Proposal: makeBlock(1),
 	}
 	prepreparePayload, _ := Encode(preprepare)
@@ -192,14 +194,15 @@ func TestProcessFutureBacklog(t *testing.T) {
 		events: new(event.TypeMux),
 	}
 	c := &core{
-		logger:      log.New("backend", "test", "id", 0),
-		backlogs:    make(map[pbft.Validator]*prque.Prque),
-		backlogsMu:  new(sync.Mutex),
-		backend:     backend,
-		sequence:    big.NewInt(1),
-		round:       big.NewInt(0),
-		state:       StateAcceptRequest,
-		internalMux: new(event.TypeMux),
+		logger:     log.New("backend", "test", "id", 0),
+		backlogs:   make(map[pbft.Validator]*prque.Prque),
+		backlogsMu: new(sync.Mutex),
+		backend:    backend,
+		current: newSnapshot(&pbft.View{
+			Sequence: big.NewInt(1),
+			Round:    big.NewInt(0),
+		}, newTestValidatorSet(4)),
+		state: StateAcceptRequest,
 	}
 	c.subscribeEvents()
 	defer c.unsubscribeEvents()
@@ -225,7 +228,7 @@ func TestProcessFutureBacklog(t *testing.T) {
 	const timeoutDura = 2 * time.Second
 	timeout := time.NewTimer(timeoutDura)
 	select {
-	case <-c.internalEvents.Chan():
+	case <-c.events.Chan():
 		t.Errorf("Should not receive any events")
 
 	case <-timeout.C:
@@ -239,7 +242,7 @@ func TestProcessBacklog(t *testing.T) {
 		Sequence: big.NewInt(1),
 	}
 	preprepare := &pbft.Preprepare{
-		View: v,
+		View:     v,
 		Proposal: makeBlock(1),
 	}
 	prepreparePayload, _ := Encode(preprepare)
@@ -276,14 +279,15 @@ func testProcessBacklog(t *testing.T, msg *message) {
 		peers:  vset,
 	}
 	c := &core{
-		logger:      log.New("backend", "test", "id", 0),
-		backlogs:    make(map[pbft.Validator]*prque.Prque),
-		backlogsMu:  new(sync.Mutex),
-		backend:     backend,
-		internalMux: new(event.TypeMux),
-		state:       State(msg.Code),
-		sequence:    big.NewInt(1),
-		round:       big.NewInt(0),
+		logger:     log.New("backend", "test", "id", 0),
+		backlogs:   make(map[pbft.Validator]*prque.Prque),
+		backlogsMu: new(sync.Mutex),
+		backend:    backend,
+		state:      State(msg.Code),
+		current: newSnapshot(&pbft.View{
+			Sequence: big.NewInt(1),
+			Round:    big.NewInt(0),
+		}, newTestValidatorSet(4)),
 		subject: &pbft.Subject{
 			View: &pbft.View{
 				Sequence: big.NewInt(1),
@@ -300,7 +304,7 @@ func testProcessBacklog(t *testing.T, msg *message) {
 	const timeoutDura = 2 * time.Second
 	timeout := time.NewTimer(timeoutDura)
 	select {
-	case ev := <-c.internalEvents.Chan():
+	case ev := <-c.events.Chan():
 		e, ok := ev.Data.(backlogEvent)
 		if !ok {
 			t.Fatalf("Unexpected event comes")

consensus/pbft/core/checkpoint.go

@@ -24,7 +24,7 @@ import (
 
 func (c *core) sendCheckpoint(cp *pbft.Subject) {
 	logger := c.logger.New("state", c.state)
-	logger.Debug("sendCheckpoint")
+	logger.Trace("sendCheckpoint")
 
 	newCp, err := Encode(cp)
 	if err != nil {
@@ -54,23 +54,27 @@ func (c *core) handleCheckpoint(msg *message, src pbft.Validator) error {
 
 	var snapshot *snapshot
 
-	logger.Debug("handleCheckpoint")
+	logger.Trace("handleCheckpoint")
 
 	c.snapshotsMu.Lock()
 	defer c.snapshotsMu.Unlock()
 
 	// Look for matching snapshot
-	if cp.View.Sequence.Cmp(c.current.Sequence) == 0 { // current
+	if cp.View.Sequence.Cmp(c.current.Sequence()) == 0 { // current
+		// If we're waiting for RoundChange, ignore this
+		if c.waitingForRoundChange {
+			return pbft.ErrIgnored
+		}
 		snapshot = c.current
-	} else if cp.View.Sequence.Cmp(c.current.Sequence) < 0 { // old checkpoint
+	} else if cp.View.Sequence.Cmp(c.current.Sequence()) < 0 { // old checkpoint
 		snapshotIndex := sort.Search(len(c.snapshots),
 			func(i int) bool {
-				return c.snapshots[i].Sequence.Cmp(cp.View.Sequence) >= 0
+				return c.snapshots[i].Sequence().Cmp(cp.View.Sequence) >= 0
 			},
 		)
 
 		// If there is no such index, Search returns len(c.snapshots).
-		if snapshotIndex < len(c.snapshots) && c.snapshots[snapshotIndex].Sequence.Cmp(cp.View.Sequence) == 0 {
+		if snapshotIndex < len(c.snapshots) && c.snapshots[snapshotIndex].Sequence().Cmp(cp.View.Sequence) == 0 {
 			snapshot = c.snapshots[snapshotIndex]
 		} else {
 			logger.Warn("Failed to find snapshot entry", "seq", cp.View.Sequence, "current", c.current.Sequence)
@@ -93,7 +97,7 @@ func (c *core) handleCheckpoint(msg *message, src pbft.Validator) error {
 func (c *core) buildStableCheckpoint() {
 	var stableCheckpoint *snapshot
 	stableCheckpointIndex := -1
-	logger := c.logger.New("seq", c.sequence)
+	logger := c.logger.New("seq", c.current.Sequence())
 
 	c.snapshotsMu.Lock()
 	for i := len(c.snapshots) - 1; i >= 0; i-- {

consensus/pbft/core/checkpoint_test.go

@@ -34,17 +34,16 @@ func TestHandleCheckpoint(t *testing.T) {
 	}
 	system := NewTestSystemWithBackend(N, F)
 	c := system.backends[0].engine.(*core)
-	c.current = newSnapshot(preprepare, system.backends[0].Validators())
-	c.snapshots = append(c.snapshots, newSnapshot(&pbft.Preprepare{
-		View: &pbft.View{
-			Round:    big.NewInt(0),
-			Sequence: big.NewInt(1),
-		},
-	}, system.backends[0].Validators()), newSnapshot(&pbft.Preprepare{
-		View: &pbft.View{
-			Round:    big.NewInt(0),
-			Sequence: big.NewInt(2),
-		},
+	c.current = newSnapshot(&pbft.View{
+		Sequence: big.NewInt(3),
+		Round:    big.NewInt(0),
+	}, system.backends[0].Validators())
+	c.snapshots = append(c.snapshots, newSnapshot(&pbft.View{
+		Round:    big.NewInt(0),
+		Sequence: big.NewInt(1),
+	}, system.backends[0].Validators()), newSnapshot(&pbft.View{
+		Round:    big.NewInt(0),
+		Sequence: big.NewInt(2),
 	}, system.backends[0].Validators()))
 
 	testCases := []struct {
@@ -55,7 +54,7 @@ func TestHandleCheckpoint(t *testing.T) {
 		expectedErr error
 	}{
 		// empty subject
-		{system, &pbft.Subject{}, system.backends[0].Validators().List()[0], nil, pbft.ErrInvalidMessage},
+		{system, &pbft.Subject{View: &pbft.View{Sequence: big.NewInt(0), Round: big.NewInt(0)}}, system.backends[0].Validators().List()[0], nil, pbft.ErrInvalidMessage},
 		// current sequence
 		{system, &pbft.Subject{View: &pbft.View{Sequence: preprepare.View.Sequence}}, system.backends[0].Validators().List()[0], c.current, nil},
 		// old sequence
@@ -102,20 +101,25 @@ func TestBuildStableCheckpoint(t *testing.T) {
 	c := system.backends[0].engine.(*core)
 	v := system.backends[0].Validators().List()[0]
 	proposal := makeBlock(1)
-	expectedStableSnapshot := newSnapshot(&pbft.Preprepare{
-		View: &pbft.View{
-			Round:    big.NewInt(0),
-			Sequence: big.NewInt(1),
-		},
+	view := &pbft.View{
+		Round:    big.NewInt(0),
+		Sequence: big.NewInt(1),
+	}
+	expectedStableSnapshot := newSnapshot(view, system.backends[0].Validators())
+	expectedStableSnapshot.Preprepare = &pbft.Preprepare{
+		View:     view,
 		Proposal: proposal,
-	}, system.backends[0].Validators())
-	c.snapshots = append(c.snapshots, expectedStableSnapshot, newSnapshot(&pbft.Preprepare{
-		View: &pbft.View{
-			Round:    big.NewInt(0),
-			Sequence: big.NewInt(2),
-		},
+	}
+	view = &pbft.View{
+		Round:    big.NewInt(0),
+		Sequence: big.NewInt(2),
+	}
+	s := newSnapshot(view, system.backends[0].Validators())
+	s.Preprepare = &pbft.Preprepare{
+		View:     view,
 		Proposal: proposal,
-	}, system.backends[0].Validators()))
+	}
+	c.snapshots = append(c.snapshots, expectedStableSnapshot, s)
 
 	sub := &pbft.Subject{View: expectedStableSnapshot.Preprepare.View}
 	b, _ := Encode(sub)
@@ -133,7 +137,7 @@ func TestBuildStableCheckpoint(t *testing.T) {
 	}
 	if stableCheckpoint == nil {
 		t.Errorf("cannot get the stable checkpoint")
-	} else if stableCheckpoint.Sequence.Cmp(expectedStableSnapshot.Sequence) != 0 {
-		t.Errorf("unexpected stable check point, got: %v, expected: %v", stableCheckpoint.Sequence, expectedStableSnapshot.Sequence)
+	} else if stableCheckpoint.Sequence().Cmp(expectedStableSnapshot.Sequence()) != 0 {
+		t.Errorf("unexpected stable check point, got: %v, expected: %v", stableCheckpoint.Sequence(), expectedStableSnapshot.Sequence())
 	}
 }

consensus/pbft/core/commit.go

@@ -24,7 +24,7 @@ import (
 
 func (c *core) sendCommit() {
 	logger := c.logger.New("state", c.state)
-	logger.Debug("sendCommit")
+	logger.Trace("sendCommit")
 
 	subject, err := Encode(c.subject)
 	if err != nil {
@@ -39,7 +39,12 @@ func (c *core) sendCommit() {
 
 func (c *core) handleCommit(msg *message, src pbft.Validator) error {
 	logger := c.logger.New("from", src.Address().Hex(), "state", c.state)
-	logger.Debug("handleCommit")
+	logger.Trace("handleCommit")
+
+	if c.waitingForRoundChange {
+		logger.Warn("Waiting for a RoundChange, ignore", "msg", msg)
+		return pbft.ErrIgnored
+	}
 
 	var commit *pbft.Subject
 	err := msg.Decode(&commit)