fix dead lock

[bytom/vapor.git] / protocol / protocol.go

diff --git a/protocol/protocol.go b/protocol/protocol.go
index d7ea944..59880a7 100644 (file)
--- a/protocol/protocol.go
+++ b/protocol/protocol.go
@@ -7,6 +7,7 @@ import (
 
        "github.com/vapor/common"
        "github.com/vapor/config"
+       "github.com/vapor/errors"
        "github.com/vapor/event"
        "github.com/vapor/protocol/bc"
        "github.com/vapor/protocol/bc/types"
@@ -18,12 +19,23 @@ const (
        maxKnownTxs           = 32768 // Maximum transactions hashes to keep in the known list (prevent DOS)
 )
 
+type Protocoler interface {
+       Name() string
+       BeforeProposalBlock(capacity int, nodeProgram []byte) ([]*types.Tx, error)
+       ChainStatus() (uint64, *bc.Hash, error)
+       ValidateBlock(block *types.Block, verifyResults []*bc.TxVerifyResult) error
+       ValidateTxs(txs []*types.Tx, verifyResults []*bc.TxVerifyResult) error
+       ApplyBlock(block *types.Block) error
+       DetachBlock(block *types.Block) error
+}
+
 // Chain provides functions for working with the Bytom block chain.
 type Chain struct {
        orphanManage   *OrphanManage
        txPool         *TxPool
        store          Store
        processBlockCh chan *processBlockMsg
+       subProtocols   []Protocoler
 
        signatureCache  *common.Cache
        eventDispatcher *event.Dispatcher
@@ -36,12 +48,13 @@ type Chain struct {
 }
 
 // NewChain returns a new Chain using store as the underlying storage.
-func NewChain(store Store, txPool *TxPool, eventDispatcher *event.Dispatcher) (*Chain, error) {
+func NewChain(store Store, txPool *TxPool, subProtocols []Protocoler, eventDispatcher *event.Dispatcher) (*Chain, error) {
        knownTxs, _ := common.NewOrderedSet(maxKnownTxs)
        c := &Chain{
                orphanManage:    NewOrphanManage(),
                txPool:          txPool,
                store:           store,
+               subProtocols:    subProtocols,
                signatureCache:  common.NewCache(maxSignatureCacheSize),
                eventDispatcher: eventDispatcher,
                processBlockCh:  make(chan *processBlockMsg, maxProcessBlockChSize),
@@ -67,6 +80,13 @@ func NewChain(store Store, txPool *TxPool, eventDispatcher *event.Dispatcher) (*
        if err != nil {
                return nil, err
        }
+
+       for _, p := range c.subProtocols {
+               if err := c.syncProtocolStatus(p); err != nil {
+                       return nil, errors.Wrap(err, p.Name(), "sync sub protocol status")
+               }
+       }
+
        go c.blockProcesser()
        return c, nil
 }
@@ -146,6 +166,10 @@ func (c *Chain) InMainChain(hash bc.Hash) bool {
        return *blockHash == hash
 }
 
+func (c *Chain) SubProtocols() []Protocoler {
+       return c.subProtocols
+}
+
 // trace back to the tail of the chain from the given block header
 func (c *Chain) traceLongestChainTail(blockHeader *types.BlockHeader) (*types.BlockHeader, error) {
        longestTail, workQueue := blockHeader, []*types.BlockHeader{blockHeader}
@@ -182,6 +206,55 @@ func (c *Chain) markTransactions(txs ...*types.Tx) {
        }
 }
 
+func (c *Chain) syncProtocolStatus(subProtocol Protocoler) error {
+       protocolHeight, protocolHash, err := subProtocol.ChainStatus()
+       if err != nil {
+               return errors.Wrap(err, "failed on get sub protocol status")
+       }
+
+       if protocolHeight == c.bestBlockHeader.Height && *protocolHash == c.bestBlockHeader.Hash() {
+               return nil
+       }
+
+       for !c.InMainChain(*protocolHash) {
+               block, err := c.GetBlockByHash(protocolHash)
+               if err != nil {
+                       return errors.Wrap(err, subProtocol.Name(), "can't get block by hash in chain")
+               }
+
+               if err := subProtocol.DetachBlock(block); err != nil {
+                       return errors.Wrap(err, subProtocol.Name(), "sub protocol detach block err")
+               }
+
+               protocolHeight, protocolHash, err = subProtocol.ChainStatus()
+               if err != nil {
+                       return errors.Wrap(err, "failed on get sub protocol status")
+               }
+       }
+
+       for height := protocolHeight + 1; height <= c.BestBlockHeight(); height++ {
+               block, err := c.GetBlockByHeight(height)
+               if err != nil {
+                       return errors.Wrap(err, subProtocol.Name(), "can't get block by height in chain")
+               }
+
+               if err := subProtocol.ApplyBlock(block); err != nil {
+                       return errors.Wrap(err, subProtocol.Name(), "sub protocol apply block err")
+               }
+
+               protocolHeight, protocolHash, err = subProtocol.ChainStatus()
+               if err != nil {
+                       return errors.Wrap(err, "failed on get sub protocol status")
+               }
+
+               if *protocolHash != block.Hash() {
+                       return errors.Wrap(errors.New("sub protocol status sync err"), subProtocol.Name())
+               }
+       }
+
+       return nil
+}
+
 // This function must be called with mu lock in above level
 func (c *Chain) setState(blockHeader, irrBlockHeader *types.BlockHeader, mainBlockHeaders []*types.BlockHeader, view *state.UtxoViewpoint, consensusResults []*state.ConsensusResult) error {
        if err := c.store.SaveChainStatus(blockHeader, irrBlockHeader, mainBlockHeaders, view, consensusResults); err != nil {