From 4fbc190f49c61c6c08e61fb2897fc037c8155c05 Mon Sep 17 00:00:00 2001 From: paladz <453256728@qq.com> Date: Tue, 18 Jun 2019 00:23:24 +0800 Subject: [PATCH] modify the store code --- database/cache.go | 95 ++++++++++++++------------------------------------ database/cache_test.go | 8 ++--- database/store.go | 59 ++++++++++++------------------- 3 files changed, 52 insertions(+), 110 deletions(-) diff --git a/database/cache.go b/database/cache.go index ca2803f0..89ef3a45 100644 --- a/database/cache.go +++ b/database/cache.go @@ -1,7 +1,6 @@ package database import ( - "fmt" "strconv" "github.com/golang/groupcache/singleflight" @@ -13,9 +12,9 @@ import ( ) const ( - maxCachedBlockHeaders = 1024 + maxCachedBlockHeaders = 4096 maxCachedBlockTransactions = 1024 - maxCachedVoteResults = 144 // int(60 * 60 * 24 * 1000 / consensus.BlockTimeInterval / consensus.RoundVoteBlockNums) + maxCachedVoteResults = 128 ) type fillBlockHeaderFn func(hash *bc.Hash, height uint64) (*types.BlockHeader, error) @@ -43,28 +42,22 @@ type cache struct { fillBlockTransactionFn func(hash *bc.Hash) ([]*types.Tx, error) fillVoteResultFn func(seq uint64) (*state.VoteResult, error) - singleBlockHeader singleflight.Group - singleBlockTxs singleflight.Group - singleVoteResult singleflight.Group + sf singleflight.Group } func (c *cache) lookupBlockHeader(hash *bc.Hash, height uint64) (*types.BlockHeader, error) { - if bH, ok := c.getBlockHeader(hash); ok { - return bH, nil + if data, ok := c.lruBlockHeaders.Get(*hash); ok { + return data.(*types.BlockHeader), nil } - blockHeader, err := c.singleBlockHeader.Do(hash.String(), func() (interface{}, error) { - bH, err := c.fillBlockHeaderFn(hash, height) + blockHeader, err := c.sf.Do("BlockHeader:"+hash.String(), func() (interface{}, error) { + blockHeader, err := c.fillBlockHeaderFn(hash, height) if err != nil { return nil, err } - if bH == nil { - return nil, fmt.Errorf("There are no blockHeader with given hash %s", hash.String()) - } - - c.addBlockHeader(bH) - return bH, nil + c.lruBlockHeaders.Add(blockHeader.Hash(), blockHeader) + return blockHeader, nil }) if err != nil { return nil, err @@ -73,47 +66,39 @@ func (c *cache) lookupBlockHeader(hash *bc.Hash, height uint64) (*types.BlockHea } func (c *cache) lookupBlockTxs(hash *bc.Hash) ([]*types.Tx, error) { - if bTxs, ok := c.getBlockTransactions(hash); ok { - return bTxs, nil + if data, ok := c.lruBlockTxs.Get(*hash); ok { + return data.([]*types.Tx), nil } - blockTransactions, err := c.singleBlockTxs.Do(hash.String(), func() (interface{}, error) { - bTxs, err := c.fillBlockTransactionFn(hash) + blockTxs, err := c.sf.Do("BlockTxs:"+hash.String(), func() (interface{}, error) { + blockTxs, err := c.fillBlockTransactionFn(hash) if err != nil { return nil, err } - if bTxs == nil { - return nil, fmt.Errorf("There are no block transactions with given hash %s", hash.String()) - } - - c.addBlockTxs(*hash, bTxs) - return bTxs, nil + c.lruBlockTxs.Add(hash, blockTxs) + return blockTxs, nil }) if err != nil { return nil, err } - return blockTransactions.([]*types.Tx), nil + return blockTxs.([]*types.Tx), nil } func (c *cache) lookupVoteResult(seq uint64) (*state.VoteResult, error) { - if vr, ok := c.getVoteResult(seq); ok { - return vr.Fork(), nil + if data, ok := c.lruVoteResults.Get(seq); ok { + return data.(*state.VoteResult).Fork(), nil } seqStr := strconv.FormatUint(seq, 10) - voteResult, err := c.singleVoteResult.Do(seqStr, func() (interface{}, error) { - v, err := c.fillVoteResultFn(seq) + voteResult, err := c.sf.Do("VoteResult:"+seqStr, func() (interface{}, error) { + voteResult, err := c.fillVoteResultFn(seq) if err != nil { return nil, err } - if v == nil { - return nil, fmt.Errorf("There are no vote result with given seq %s", seqStr) - } - - c.addVoteResult(v) - return v, nil + c.lruVoteResults.Add(voteResult.Seq, voteResult) + return voteResult, nil }) if err != nil { return nil, err @@ -121,38 +106,10 @@ func (c *cache) lookupVoteResult(seq uint64) (*state.VoteResult, error) { return voteResult.(*state.VoteResult).Fork(), nil } -func (c *cache) getBlockHeader(hash *bc.Hash) (*types.BlockHeader, bool) { - blockHeader, ok := c.lruBlockHeaders.Get(*hash) - if blockHeader == nil { - return nil, ok - } - return blockHeader.(*types.BlockHeader), ok -} - -func (c *cache) getBlockTransactions(hash *bc.Hash) ([]*types.Tx, bool) { - txs, ok := c.lruBlockTxs.Get(*hash) - if txs == nil { - return nil, ok - } - return txs.([]*types.Tx), ok -} - -func (c *cache) getVoteResult(seq uint64) (*state.VoteResult, bool) { - voteResult, ok := c.lruVoteResults.Get(seq) - if voteResult == nil { - return nil, ok - } - return voteResult.(*state.VoteResult), ok -} - -func (c *cache) addBlockHeader(blockHeader *types.BlockHeader) { - c.lruBlockHeaders.Add(blockHeader.Hash(), blockHeader) -} - -func (c *cache) addBlockTxs(hash bc.Hash, txs []*types.Tx) { - c.lruBlockTxs.Add(hash, txs) +func (c *cache) removeBlockHeader(blockHeader *types.BlockHeader) { + c.lruBlockHeaders.Remove(blockHeader.Hash()) } -func (c *cache) addVoteResult(voteResult *state.VoteResult) { - c.lruVoteResults.Add(voteResult.Seq, voteResult) +func (c *cache) removeVoteResult(voteResult *state.VoteResult) { + c.lruVoteResults.Remove(voteResult.Seq) } diff --git a/database/cache_test.go b/database/cache_test.go index 198d4cdc..14193778 100644 --- a/database/cache_test.go +++ b/database/cache_test.go @@ -55,7 +55,7 @@ func TestBlockCache(t *testing.T) { for i := 0; i < 10; i++ { block := newBlock(uint64(i)) hash := block.Hash() - if b, _ := cache.getBlockHeader(&hash); b != nil { + if _, ok := cache.lruBlockHeaders.Get(hash); ok { t.Fatalf("find old block") } } @@ -63,7 +63,7 @@ func TestBlockCache(t *testing.T) { for i := 10; i < maxCachedBlockHeaders+10; i++ { block := newBlock(uint64(i)) hash := block.Hash() - if b, _ := cache.getBlockHeader(&hash); b == nil { + if _, ok := cache.lruBlockHeaders.Get(hash); !ok { t.Fatalf("can't find new block") } } @@ -75,14 +75,14 @@ func TestBlockCache(t *testing.T) { for i := 0; i < 10; i++ { voteResult := newVoteResult(uint64(i)) - if v, _ := cache.getVoteResult(voteResult.Seq); v != nil { + if _, ok := cache.lruVoteResults.Get(voteResult.Seq); ok { t.Fatalf("find old vote result") } } for i := 10; i < maxCachedVoteResults+10; i++ { voteResult := newVoteResult(uint64(i)) - if v, _ := cache.getVoteResult(voteResult.Seq); v == nil { + if _, ok := cache.lruVoteResults.Get(voteResult.Seq); !ok { t.Fatalf("can't find new vote result") } } diff --git a/database/store.go b/database/store.go index cddbcdf9..6de274be 100644 --- a/database/store.go +++ b/database/store.go @@ -3,11 +3,11 @@ package database import ( "encoding/binary" "encoding/json" + "fmt" "time" "github.com/golang/protobuf/proto" log "github.com/sirupsen/logrus" - "github.com/tendermint/tmlibs/common" dbm "github.com/vapor/database/leveldb" "github.com/vapor/database/storage" @@ -33,9 +33,10 @@ func loadBlockStoreStateJSON(db dbm.DB) *protocol.BlockStoreState { if bytes == nil { return nil } + bsj := &protocol.BlockStoreState{} if err := json.Unmarshal(bytes, bsj); err != nil { - common.PanicCrisis(common.Fmt("Could not unmarshal bytes: %X", bytes)) + log.WithField("err", err).Panic("fail on unmarshal BlockStoreStateJSON") } return bsj } @@ -71,26 +72,26 @@ func calcVoteResultKey(seq uint64) []byte { // GetBlockHeader return the block header by given hash and height func GetBlockHeader(db dbm.DB, hash *bc.Hash, height uint64) (*types.BlockHeader, error) { - block := &types.Block{} binaryBlockHeader := db.Get(calcBlockHeaderKey(height, hash)) if binaryBlockHeader == nil { - return nil, nil + return nil, fmt.Errorf("There are no blockHeader with given hash %s", hash.String()) } + + block := &types.Block{} if err := block.UnmarshalText(binaryBlockHeader); err != nil { return nil, err } - return &block.BlockHeader, nil } // GetBlockTransactions return the block transactions by given hash func GetBlockTransactions(db dbm.DB, hash *bc.Hash) ([]*types.Tx, error) { - block := &types.Block{} binaryBlockTxs := db.Get(calcBlockTransactionsKey(hash)) if binaryBlockTxs == nil { - return nil, errors.New("The transactions in the block is empty") + return nil, fmt.Errorf("There are no block transactions with given hash %s", hash.String()) } + block := &types.Block{} if err := block.UnmarshalText(binaryBlockTxs); err != nil { return nil, err } @@ -129,15 +130,10 @@ func NewStore(db dbm.DB) *Store { } } -// GetUtxo will search the utxo in db -func (s *Store) GetUtxo(hash *bc.Hash) (*storage.UtxoEntry, error) { - return getUtxo(s.db, hash) -} - // BlockExist check if the block is stored in disk func (s *Store) BlockExist(hash *bc.Hash, height uint64) bool { - blockHeader, err := s.cache.lookupBlockHeader(hash, height) - return err == nil && blockHeader != nil + _, err := s.cache.lookupBlockHeader(hash, height) + return err == nil } // GetBlock return the block by given hash @@ -160,20 +156,17 @@ func (s *Store) GetBlock(hash *bc.Hash, height uint64) (*types.Block, error) { // GetBlockHeader return the BlockHeader by given hash func (s *Store) GetBlockHeader(hash *bc.Hash, height uint64) (*types.BlockHeader, error) { - blockHeader, err := s.cache.lookupBlockHeader(hash, height) - if err != nil { - return nil, err - } - return blockHeader, nil + return s.cache.lookupBlockHeader(hash, height) } // GetBlockTransactions return the Block transactions by given hash func (s *Store) GetBlockTransactions(hash *bc.Hash) ([]*types.Tx, error) { - txs, err := s.cache.lookupBlockTxs(hash) - if err != nil { - return nil, err - } - return txs, nil + return s.cache.lookupBlockTxs(hash) +} + +// GetStoreStatus return the BlockStoreStateJSON +func (s *Store) GetStoreStatus() *protocol.BlockStoreState { + return loadBlockStoreStateJSON(s.db) } // GetTransactionsUtxo will return all the utxo that related to the input txs @@ -195,9 +188,9 @@ func (s *Store) GetTransactionStatus(hash *bc.Hash) (*bc.TransactionStatus, erro return ts, nil } -// GetStoreStatus return the BlockStoreStateJSON -func (s *Store) GetStoreStatus() *protocol.BlockStoreState { - return loadBlockStoreStateJSON(s.db) +// GetUtxo will search the utxo in db +func (s *Store) GetUtxo(hash *bc.Hash) (*storage.UtxoEntry, error) { + return getUtxo(s.db, hash) } // GetVoteResult retrive the voting result in specified vote sequence @@ -251,7 +244,6 @@ func (s *Store) LoadBlockIndex(stateBestHeight uint64) (*state.BlockIndex, error // SaveBlock persists a new block in the protocol. func (s *Store) SaveBlock(block *types.Block, ts *bc.TransactionStatus) error { startTime := time.Now() - binaryBlockHeader, err := block.MarshalTextForBlockHeader() if err != nil { return errors.Wrap(err, "Marshal block header") @@ -292,12 +284,7 @@ func (s *Store) SaveBlockHeader(blockHeader *types.BlockHeader) error { blockHash := blockHeader.Hash() s.db.Set(calcBlockHeaderKey(blockHeader.Height, &blockHash), binaryBlockHeader) - - // updata blockheader cache - if _, ok := s.cache.getBlockHeader(&blockHash); ok { - s.cache.addBlockHeader(blockHeader) - } - + s.cache.removeBlockHeader(blockHeader) return nil } @@ -315,9 +302,7 @@ func (s *Store) SaveChainStatus(node, irreversibleNode *state.BlockNode, view *s } batch.Set(calcVoteResultKey(vote.Seq), bytes) - if _, ok := s.cache.getVoteResult(vote.Seq); ok { - s.cache.addVoteResult(vote) - } + s.cache.removeVoteResult(vote) } bytes, err := json.Marshal(protocol.BlockStoreState{ -- 2.11.0