Improve account block validation performance

chain/account/keys.go

@@ -6,6 +6,9 @@ var (
 	chainPlasmaKey           = []byte{5}
 	receivedBlockPrefix      = []byte{6}
 	sequencerLastReceivedKey = []byte{7}
+
+	StorageKeyPrefix = storageKeyPrefix
+	ChainPlasmaKey   = chainPlasmaKey
 )
 
 const (

chain/cache.go

@@ -0,0 +1,164 @@
+package chain
+
+import (
+	"fmt"
+	"sync"
+
+	"github.com/inconshreveable/log15"
+	"github.com/pkg/errors"
+	"github.com/zenon-network/go-zenon/chain/cache"
+	"github.com/zenon-network/go-zenon/chain/cache/storage"
+	"github.com/zenon-network/go-zenon/chain/nom"
+	"github.com/zenon-network/go-zenon/chain/store"
+	"github.com/zenon-network/go-zenon/common"
+	"github.com/zenon-network/go-zenon/common/db"
+	"github.com/zenon-network/go-zenon/common/types"
+)
+
+type chainCache struct {
+	manager storage.CacheManager
+	log     log15.Logger
+	changes sync.Mutex
+}
+
+func (c *chainCache) getFrontierStore() store.Cache {
+	if db := c.manager.DB(); db == nil {
+		return nil
+	} else {
+		return cache.NewCacheStore(storage.GetFrontierIdentifier(db), c.manager)
+	}
+}
+
+func (c *chainCache) GetFrontierCacheStore() store.Cache {
+	c.changes.Lock()
+	defer c.changes.Unlock()
+	return c.getFrontierStore()
+}
+
+func (c *chainCache) GetCacheStore(identifier types.HashHeight) store.Cache {
+	c.changes.Lock()
+	defer c.changes.Unlock()
+	return cache.NewCacheStore(identifier, c.manager)
+}
+
+func (c *chainCache) UpdateCache(insertLocker sync.Locker, detailed *nom.DetailedMomentum, changes db.Patch) error {
+	if insertLocker == nil {
+		return errors.Errorf("insertLocker can't be nil")
+	}
+	if changes == nil {
+		return errors.Errorf("changes can't be nil")
+	}
+	c.changes.Lock()
+	defer c.changes.Unlock()
+	if err := c.update(detailed, changes); err != nil {
+		return err
+	}
+	return nil
+}
+
+func (c *chainCache) update(detailed *nom.DetailedMomentum, changes db.Patch) error {
+	momentum := detailed.Momentum
+	c.log.Info("inserting new momentum to chain cache", "identifier", momentum.Identifier())
+	store := c.getFrontierStore()
+	store.ApplyMomentum(detailed, changes)
+	patch, err := store.Changes()
+	if err != nil {
+		return err
+	}
+	if err := c.manager.Add(momentum.Identifier(), patch); err != nil {
+		return err
+	}
+	return nil
+}
+
+func (c *chainCache) RollbackCacheTo(insertLocker sync.Locker, identifier types.HashHeight) error {
+	if insertLocker == nil {
+		return errors.Errorf("insertLocker can't be nil")
+	}
+	c.changes.Lock()
+	defer c.changes.Unlock()
+	if err := c.rollbackTo(identifier); err != nil {
+		return err
+	}
+	return nil
+}
+
+func (c *chainCache) rollbackTo(identifier types.HashHeight) error {
+	c.log.Info("rollbacking cache", "to-identifier", identifier)
+	frontier := c.getFrontierStore().Identifier()
+
+	if identifier.Height > frontier.Height {
+		return errors.Errorf("can't rollback cache. Expected identifier height %v is greater than frontier %v", identifier, frontier)
+	}
+
+	if frontier.Height-identifier.Height > uint64(storage.GetRollbackCacheSize()) {
+		return errors.Errorf("can't rollback cache. Target identifier %v is outside the rollback cache", identifier)
+	}
+
+	for {
+		store := c.getFrontierStore()
+		frontier := store.Identifier()
+		if frontier.Height == identifier.Height {
+			break
+		}
+		c.log.Info("rollbacking", "momentum-identifier", frontier)
+		if err := c.manager.Pop(); err != nil {
+			return err
+		}
+	}
+
+	return nil
+}
+
+func (c *chainCache) Init(chainManager db.Manager, momentumStore store.Momentum) error {
+	c.changes.Lock()
+	defer c.changes.Unlock()
+	chainFrontier := db.GetFrontierIdentifier(chainManager.Frontier())
+	cacheFrontier := storage.GetFrontierIdentifier(c.manager.DB())
+
+	if cacheFrontier.Height == chainFrontier.Height {
+		if cacheFrontier.Hash != chainFrontier.Hash {
+			return errors.Errorf("The cache's state is incorrect. " +
+				"You can fix the problem by removing the cache database manually.")
+		}
+		return nil
+	}
+
+	if cacheFrontier.Height > chainFrontier.Height {
+		if err := c.rollbackTo(chainFrontier); err != nil {
+			return err
+		}
+		return nil
+	}
+
+	if chainFrontier.Height-cacheFrontier.Height >= 100000 {
+		fmt.Println("Initializing cache: 0%")
+	}
+
+	for i := cacheFrontier.Height + 1; i <= chainFrontier.Height; i++ {
+		momentum, err := momentumStore.GetMomentumByHeight(i)
+		if err != nil {
+			return err
+		}
+		changes := chainManager.GetPatch(momentum.Identifier())
+		detailed, err := momentumStore.PrefetchMomentum(momentum)
+		if err != nil {
+			return err
+		}
+		if err := c.update(detailed, changes); err != nil {
+			return err
+		}
+		if i%100000 == 0 {
+			fmt.Printf("Initializing cache: %d%%\n", i*100/chainFrontier.Height)
+		}
+	}
+
+	return nil
+}
+
+func NewChainCache(cacheManager storage.CacheManager) *chainCache {
+	return &chainCache{
+		manager: cacheManager,
+		log:     common.ChainLogger.New("submodule", "chain-cache"),
+	}
+}

chain/cache/account.go

@@ -0,0 +1,69 @@
+package cache
+
+import (
+	"math/big"
+
+	"github.com/syndtr/goleveldb/leveldb"
+	"github.com/zenon-network/go-zenon/chain/nom"
+	"github.com/zenon-network/go-zenon/common"
+	"github.com/zenon-network/go-zenon/common/types"
+)
+
+var (
+	fusedAmountKeyPrefix = []byte{0}
+	chainPlasmaKeyPrefix = []byte{1}
+)
+
+func getFusedAmountKeyPrefix(address []byte) []byte {
+	return common.JoinBytes(accountCacheKeyPrefix, fusedAmountKeyPrefix, address)
+}
+
+func getChainPlasmaKeyPrefix(address []byte) []byte {
+	return common.JoinBytes(accountCacheKeyPrefix, chainPlasmaKeyPrefix, address)
+}
+
+func (cs *cacheStore) GetStakeBeneficialAmount(address types.Address) (*big.Int, error) {
+	value, err := cs.findValue(getFusedAmountKeyPrefix(address.Bytes()))
+	if err == leveldb.ErrNotFound {
+		return big.NewInt(0), nil
+	}
+	if err != nil {
+		return nil, err
+	}
+	return big.NewInt(0).SetBytes(value), nil
+}
+
+func (cs *cacheStore) GetChainPlasma(address types.Address) (*big.Int, error) {
+	value, err := cs.findValue(getChainPlasmaKeyPrefix(address.Bytes()))
+	if err == leveldb.ErrNotFound {
+		return big.NewInt(0), nil
+	}
+	if err != nil {
+		return nil, err
+	}
+	return big.NewInt(0).SetBytes(value), nil
+}
+
+func (cs *cacheStore) pruneAccountCache(blocks []*nom.AccountBlock) error {
+	for _, block := range blocks {
+		all := append([]*nom.AccountBlock{block}, block.DescendantBlocks...)
+		for _, b := range all {
+			prefix := getFusedAmountKeyPrefix(b.Address.Bytes())
+			fusedPlasmaKeys, err := cs.findExpiredKeys(prefix, b.MomentumAcknowledged.Height)
+			if err != nil {
+				return err
+			}
+
+			prefix = getChainPlasmaKeyPrefix(b.Address.Bytes())
+			chainPlasmaKeys, err := cs.findExpiredKeys(prefix, b.MomentumAcknowledged.Height)
+			if err != nil {
+				return err
+			}
+
+			for _, key := range append(fusedPlasmaKeys, chainPlasmaKeys...) {
+				cs.changes.Delete(key)
+			}
+		}
+	}
+	return nil
+}

chain/cache/extractor.go

@@ -0,0 +1,78 @@
+package cache
+
+import (
+	"bytes"
+
+	"github.com/zenon-network/go-zenon/chain/account"
+	"github.com/zenon-network/go-zenon/chain/momentum"
+	"github.com/zenon-network/go-zenon/chain/store"
+	"github.com/zenon-network/go-zenon/common"
+	"github.com/zenon-network/go-zenon/common/db"
+	"github.com/zenon-network/go-zenon/common/types"
+	"github.com/zenon-network/go-zenon/vm/embedded/definition"
+)
+
+type cacheExtractor struct {
+	cache  store.Cache
+	height uint64
+	patch  db.Patch
+}
+
+func (e *cacheExtractor) Put(key []byte, value []byte) {
+	if cacheKey := e.tryToGetCacheKey(key, value); cacheKey != nil {
+		e.patch.Put(cacheKey, value)
+	}
+}
+
+func (e *cacheExtractor) Delete(key []byte) {
+	if cacheKey := e.tryToGetCacheKey(key, nil); cacheKey != nil {
+		e.patch.Put(cacheKey, []byte{})
+	}
+}
+
+func (e *cacheExtractor) tryToGetCacheKey(key []byte, value []byte) []byte {
+	if bytes.HasPrefix(key, momentum.AccountStorePrefix) {
+		key = bytes.TrimPrefix(key, momentum.AccountStorePrefix)
+		keyWithoutAddress := key[types.AddressSize:]
+		address, err := types.BytesToAddress(key[:types.AddressSize])
+		common.DealWithErr(err)
+
+		// Cache fused plasma
+		if address == types.PlasmaContract {
+			prefix := common.JoinBytes(account.StorageKeyPrefix, definition.FusedAmountKeyPrefix)
+			if bytes.HasPrefix(keyWithoutAddress, prefix) {
+				beneficiary := bytes.TrimPrefix(keyWithoutAddress, prefix)
+				return e.getHeightKey(getFusedAmountKeyPrefix(beneficiary))
+			}
+		}
+
+		// Cache sporks
+		if address == types.SporkContract {
+			prefix := common.JoinBytes(account.StorageKeyPrefix, []byte{definition.SporkInfoPrefix})
+			if bytes.HasPrefix(keyWithoutAddress, prefix) {
+				sporkId := bytes.TrimPrefix(keyWithoutAddress, prefix)
+				return e.getHeightKey(getSporkInfoKeyPrefix(sporkId))
+			}
+		}
+
+		// Cache chain plasma
+		if bytes.HasPrefix(keyWithoutAddress, account.ChainPlasmaKey) {
+			if value == nil {
+				return e.getHeightKey(getChainPlasmaKeyPrefix(address.Bytes()))
+			}
+			// Verify that the state has changed
+			current, err := e.cache.GetChainPlasma(address)
+			common.DealWithErr(err)
+			if current.Cmp(common.BytesToBigInt(value)) == 0 {
+				return nil
+			}
+			return e.getHeightKey(getChainPlasmaKeyPrefix(address.Bytes()))
+		}
+	}
+
+	return nil
+}
+
+func (e *cacheExtractor) getHeightKey(prefix []byte) []byte {
+	return common.JoinBytes(prefix, common.Uint64ToBytes(e.height))
+}