Merge pull request #62 from getamis/feature/refactor_seal

consensus/pbft: refactor committed, seal and viewChanged

Author: markya0616

consensus/pbft/backends/simple/backend.go

@@ -18,6 +18,7 @@ package simple
 
 import (
 	"crypto/ecdsa"
+	"sync"
 
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/consensus"
@@ -48,6 +49,7 @@ func New(config *pbft.Config, eventMux *event.TypeMux, privateKey *ecdsa.Private
 		address:      crypto.PubkeyToAddress(privateKey.PublicKey),
 		logger:       log.New("backend", "simple"),
 		db:           db,
+		commitCh:     make(chan common.Hash, 1),
 	}
 	return backend
 }
@@ -70,9 +72,9 @@ type simpleBackend struct {
 	inserter     func(block *types.Block) error
 
 	// the channels for pbft engine notifications
-	viewChange chan bool
-	commit     chan common.Hash
-	commitErr  chan error
+	commitCh          chan common.Hash
+	proposedBlockHash common.Hash
+	sealMu            sync.Mutex
 }
 
 // Address implements pbft.Backend.Address
@@ -128,33 +130,28 @@ func (sb *simpleBackend) Commit(proposal pbft.Proposal) error {
 		sb.logger.Error("Failed to commit proposal since RequestContext cannot cast to *types.Block")
 		return errCastingRequest
 	}
-	// it's a proposer
-	if sb.commit != nil {
-		sb.commitErr = make(chan error, 1)
-		closeCommitErr := func() {
-			close(sb.commitErr)
-		}
-		defer closeCommitErr()
+	// - if the proposed and committed blocks are the same, send the proposed hash
+	//   to commit channel, which is being watched inside the engine.Seal() function.
+	// - otherwise, we try to insert the block.
+	// -- if success, the ChainHeadEvent event will be broadcasted and try to build
+	//    the next block. If it's a proposer, the previous Seal() will be stopped.
+	// -- otherwise, a error will be returned and a view change event will be fired.
+	if sb.proposedBlockHash == block.Hash() {
 		// feed block hash to Seal() and wait the Seal() result
-		sb.commit <- block.Hash()
+		sb.commitCh <- block.Hash()
 		// TODO: how do we check the block is inserted correctly?
-		return <-sb.commitErr
-	} else {
-		return sb.inserter(block)
+		return nil
 	}
+	return sb.inserter(block)
 }
 
 // ViewChanged implements pbft.Backend.ViewChanged
 func (sb *simpleBackend) ViewChanged(needNewProposal bool) error {
-	// step1: update proposer
-	// step2: notify proposer and validator
-	if sb.viewChange != nil {
-		go func() {
-			sb.viewChange <- needNewProposal
-		}()
-	}
-	if sb.IsProposer() {
-		go sb.eventMux.Post(core.ChainHeadEvent{})
+	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() {
+			go sb.eventMux.Post(core.ChainHeadEvent{})
+		}
 	}
 	return nil
 }

consensus/pbft/backends/simple/engine.go

@@ -45,10 +45,6 @@ var (
 	errInvalidDifficulty = errors.New("invalid difficulty")
 	// errNotProposer is returned when I'm not a proposer
 	errNotProposer = errors.New("not a proposer")
-	// errViewChanged is returned when we receive a view change event
-	errViewChanged = errors.New("view changed")
-	// errOtherBlockCommitted is returned when other block is committed.
-	errOtherBlockCommitted = errors.New("other block is committed")
 	// errInvalidPeer is returned when a message from invalid peer comes
 	errInvalidPeer = errors.New("invalid peer")
 	// errInvalidExtraDataFormat is returned when the extra data format is incorrect
@@ -270,23 +266,6 @@ func (sb *simpleBackend) Finalize(chain consensus.ChainReader, header *types.Hea
 	return types.NewBlock(header, txs, nil, receipts), nil
 }
 
-func (sb *simpleBackend) closeChannels() {
-	if sb.viewChange != nil {
-		close(sb.viewChange)
-		sb.viewChange = nil
-	}
-
-	if sb.commit != nil {
-		close(sb.commit)
-		sb.commit = nil
-	}
-}
-
-func (sb *simpleBackend) newChannels() {
-	sb.viewChange = make(chan bool, 1)
-	sb.commit = make(chan common.Hash, 1)
-}
-
 // Seal generates a new block for the given input block with the local miner's
 // seal place on top.
 func (sb *simpleBackend) Seal(chain consensus.ChainReader, block *types.Block, stop <-chan struct{}) (*types.Block, error) {
@@ -314,27 +293,28 @@ func (sb *simpleBackend) Seal(chain consensus.ChainReader, block *types.Block, s
 		return nil, errNotProposer
 	}
 
-	sb.newChannels()
-	defer sb.closeChannels()
+	// get the proposed block hash and clear it if the seal() is completed.
+	sb.sealMu.Lock()
+	sb.proposedBlockHash = block.Hash()
+	clear := func() {
+		sb.proposedBlockHash = common.Hash{}
+		sb.sealMu.Unlock()
+	}
+	defer clear()
+
 	// post block into PBFT engine
 	go sb.EventMux().Post(pbft.RequestEvent{
 		Proposal: block,
 	})
 
 	for {
 		select {
-		case needNewProposal := <-sb.viewChange:
-			if needNewProposal {
-				return nil, errViewChanged
-			}
-			// if we don't need to change block, we keep waiting events.
-		case hash := <-sb.commit:
+		case hash := <-sb.commitCh:
+			// if the block hash and the hash from channel are the same,
+			// return the block. Otherwise, keep waiting the next hash.
 			if block.Hash() == hash {
-				sb.commitErr <- nil
 				return block, nil
 			}
-			sb.commitErr <- errOtherBlockCommitted
-			return nil, errOtherBlockCommitted
 		case <-stop:
 			return nil, nil
 		}

consensus/pbft/backends/simple/engine_test.go

@@ -155,7 +155,7 @@ func TestSealStopChannel(t *testing.T) {
 	}
 }
 
-func TestSealViewChange(t *testing.T) {
+func TestSealRoundChange(t *testing.T) {
 	chain, engine := newBlockChain(4)
 	block := makeBlockWithoutSeal(chain, engine, chain.Genesis())
 	eventSub := engine.EventMux().Subscribe(pbft.RequestEvent{})
@@ -166,7 +166,7 @@ func TestSealViewChange(t *testing.T) {
 			if !ok {
 				t.Errorf("unexpected event comes, got: %v, expected: pbft.RequestEvent", reflect.TypeOf(ev.Data))
 			}
-			engine.viewChange <- false
+			engine.ViewChanged(false)
 		}
 		eventSub.Unsubscribe()
 	}
@@ -186,35 +186,10 @@ func TestSealViewChange(t *testing.T) {
 	}
 }
 
-func TestSealViewChangeNeedNewProposal(t *testing.T) {
-	chain, engine := newBlockChain(4)
-	block := makeBlockWithoutSeal(chain, engine, chain.Genesis())
-	eventSub := engine.EventMux().Subscribe(pbft.RequestEvent{})
-	eventLoop := func() {
-		select {
-		case ev := <-eventSub.Chan():
-			_, ok := ev.Data.(pbft.RequestEvent)
-			if !ok {
-				t.Errorf("unexpected event comes, got: %v, expected: pbft.RequestEvent", reflect.TypeOf(ev.Data))
-			}
-			engine.viewChange <- true
-		}
-		eventSub.Unsubscribe()
-	}
-	go eventLoop()
-
-	finalBlock, err := engine.Seal(chain, block, nil)
-	if err != errViewChanged {
-		t.Errorf("unexpected error comes, got: %v, expected: errViewChanged", err)
-	}
-	if finalBlock != nil {
-		t.Errorf("block should be nil, but got: %v", finalBlock.Hash().Hex())
-	}
-}
-
 func TestSealCommittedOtherHash(t *testing.T) {
 	chain, engine := newBlockChain(4)
 	block := makeBlockWithoutSeal(chain, engine, chain.Genesis())
+	otherBlock := makeBlockWithoutSeal(chain, engine, block)
 	eventSub := engine.EventMux().Subscribe(pbft.RequestEvent{})
 	eventLoop := func() {
 		select {
@@ -223,26 +198,22 @@ func TestSealCommittedOtherHash(t *testing.T) {
 			if !ok {
 				t.Errorf("unexpected event comes, got: %v, expected: pbft.RequestEvent", reflect.TypeOf(ev.Data))
 			}
-			engine.commitErr = make(chan error, 1)
-			closeCommitErr := func() {
-				close(engine.commitErr)
-			}
-			defer closeCommitErr()
-			engine.commit <- common.StringToHash("1234567890")
-			err := <-engine.commitErr
-			if err != errOtherBlockCommitted {
-				t.Errorf("unexpected error comes, got: %v, expected: errOtherBlockCommitted", err)
-			}
+			engine.Commit(otherBlock)
 		}
 		eventSub.Unsubscribe()
 	}
 	go eventLoop()
-	finalBlock, err := engine.Seal(chain, block, nil)
-	if err != errOtherBlockCommitted {
-		t.Errorf("unexpected error comes, got: %v, expected: errOtherBlockCommitted", err)
+	seal := func() {
+		engine.Seal(chain, block, nil)
+		t.Errorf("should not be called")
 	}
-	if finalBlock != nil {
-		t.Errorf("block should be nil, but got: %v", finalBlock)
+	go seal()
+
+	const timeoutDura = 2 * time.Second
+	timeout := time.NewTimer(timeoutDura)
+	select {
+	case <-timeout.C:
+		// wait 2 seconds to ensure we cannot get any blocks from PBFT
 	}
 }