Merge PR: refactor orphan map (#1969)

* refactor orphan map (#1960)

* refactor orphan map

* func (oi *OrphanInfo) enqueueResult(res int64)

* rename

* rename

* rm 	oi.orphanItemCacheQueue

* rm 	oi.orphanItemCacheQueue

* type NodeCache struct

* simplify names

* simplify names

dev/start.sh

@@ -23,7 +23,8 @@ killbyname() {
 run() {
     LOG_LEVEL=main:debug,iavl:info,*:error,state:info,provider:info
 
-    exchaind start --pruning=nothing --rpc.unsafe \
+#    exchaind start --pruning=nothing --rpc.unsafe \
+    exchaind start --rpc.unsafe \
       --local-rpc-port 26657 \
       --log_level $LOG_LEVEL \
       --log_file json \
@@ -33,7 +34,7 @@ run() {
       --iavl-enable-async-commit \
       --enable-gid \
       --append-pid=true \
-      --iavl-commit-interval-height 10 \
+      --iavl-commit-interval-height 5 \
       --iavl-output-modules evm=0,acc=0 \
       --trace --home $HOME_SERVER --chain-id $CHAINID \
       --elapsed Round=1,CommitRound=1,Produce=1 \

libs/iavl/mutable_tree.go

@@ -121,9 +121,8 @@ func (tree *MutableTree) IsEmpty() bool {
 
 // VersionExists returns whether or not a version exists.
 func (tree *MutableTree) VersionExists(version int64) bool {
-	tree.ndb.mtx.Lock()
-	defer tree.ndb.mtx.Unlock()
-	if tree.ndb.heightOrphansMap[version] != nil {
+	_, ok := tree.ndb.findRootHash(version)
+	if ok {
 		return true
 	}
 	return tree.versions.Get(version)

libs/iavl/mutable_tree_oec.go

@@ -41,9 +41,12 @@ type commitEvent struct {
 	iavlHeight int
 }
 
+
 func (tree *MutableTree) SaveVersionAsync(version int64, useDeltas bool) ([]byte, int64, error) {
-	moduleName := tree.GetModuleName()
-	oldRoot, saved := tree.hasSaved(version)
+
+	tree.ndb.sanityCheckHandleOrphansResult(version)
+
+	oldRoot, saved := tree.ndb.findRootHash(version)
 	if saved {
 		return nil, version, fmt.Errorf("existing version: %d, root: %X", version, oldRoot)
 	}
@@ -65,34 +68,34 @@ func (tree *MutableTree) SaveVersionAsync(version int64, useDeltas bool) ([]byte
 		}
 	}
 
-	tree.ndb.SaveOrphansAsync(version, tree.orphans)
-
-	shouldPersist := (version-tree.lastPersistHeight >= CommitIntervalHeight) ||
-		(treeMap.totalPreCommitCacheSize >= MinCommitItemCount)
+	shouldPersist := (version-tree.lastPersistHeight >= CommitIntervalHeight) || (treeMap.totalPreCommitCacheSize >= MinCommitItemCount)
 
+	newOrphans := tree.orphans
 	if shouldPersist {
-		batch := tree.NewBatch()
-		if err := tree.persist(batch, version); err != nil {
-			return nil, 0, err
-		}
+		tree.ndb.saveNewOrphans(version, newOrphans, true)
+		tree.persist(version)
+		newOrphans = nil
 	}
 
+	return tree.setNewWorkingTree(version, newOrphans, shouldPersist)
+}
+
+func (tree *MutableTree) setNewWorkingTree(version int64, newOrphans []*Node, persisted bool) ([]byte, int64, error) {
 	// set new working tree
 	tree.ImmutableTree = tree.ImmutableTree.clone()
 	tree.lastSaved = tree.ImmutableTree.clone()
 	tree.orphans = make([]*Node, 0, len(tree.orphans))
 	for k := range tree.savedNodes {
 		delete(tree.savedNodes, k)
 	}
-
 	rootHash := tree.lastSaved.Hash()
-	tree.setHeightOrphansItem(version, rootHash)
 
+	tree.ndb.enqueueOrphanTask(version, rootHash, newOrphans)
 	tree.version = version
-	if shouldPersist {
+	if persisted {
 		tree.versions.Set(version, true)
 	}
-	treeMap.updateMutableTreeMap(moduleName)
+	treeMap.updateMutableTreeMap(tree.GetModuleName())
 
 	tree.removedVersions.Range(func(k, v interface{}) bool {
 		tree.log(IavlDebug, "remove version from tree version map", "Height", k.(int64))
@@ -109,31 +112,34 @@ func (tree *MutableTree) removeVersion(version int64) {
 	tree.versions.Delete(version)
 }
 
-func (tree *MutableTree) persist(batch dbm.Batch, version int64) error {
+func (tree *MutableTree) persist(version int64)  {
+	var err error
+	batch := tree.NewBatch()
 	tree.commitCh <- commitEvent{-1, nil, nil, nil, nil, 0}
 	var tpp map[string]*Node = nil
 	if EnablePruningHistoryState {
 		tree.ndb.saveCommitOrphans(batch, version, tree.commitOrphans)
 	}
 	if tree.root == nil {
 		// There can still be orphans, for example if the root is the node being removed.
-		if err := tree.ndb.SaveEmptyRoot(batch, version); err != nil {
-			return err
-		}
+		err = tree.ndb.SaveEmptyRoot(batch, version)
 	} else {
-		if err := tree.ndb.SaveRoot(batch, tree.root, version); err != nil {
-			return err
-		}
+		err = tree.ndb.SaveRoot(batch, tree.root, version)
 		tpp = tree.ndb.asyncPersistTppStart(version)
 	}
+
+	if err != nil {
+		// never going to happen in case of AC enabled
+		panic(err)
+	}
+
 	for k := range tree.commitOrphans {
 		delete(tree.commitOrphans, k)
 	}
 	versions := tree.deepCopyVersions()
 	tree.commitCh <- commitEvent{version, versions, batch,
 		tpp, nil, int(tree.Height())}
 	tree.lastPersistHeight = version
-	return nil
 }
 
 func (tree *MutableTree) commitSchedule() {
@@ -217,9 +223,6 @@ func (tree *MutableTree) log(level int, msg string, kvs ...interface{}) {
 	iavlLog(tree.GetModuleName(), level, msg, kvs...)
 }
 
-func (tree *MutableTree) setHeightOrphansItem(version int64, rootHash []byte) {
-	tree.ndb.setHeightOrphansItem(version, rootHash)
-}
 
 func (tree *MutableTree) updateCommittedStateHeightPool(batch dbm.Batch, version int64, versions map[int64]bool) {
 	queue := tree.committedHeightQueue
@@ -291,9 +294,6 @@ func (tree *MutableTree) addOrphansOptimized(orphans []*Node) {
 	}
 }
 
-func (tree *MutableTree) hasSaved(version int64) ([]byte, bool) {
-	return tree.ndb.inVersionCacheMap(version)
-}
 
 func (tree *MutableTree) deepCopyVersions() map[int64]bool {
 	if !EnablePruningHistoryState {

libs/iavl/mutable_tree_oec_test.go

@@ -110,8 +110,11 @@ func TestSaveVersionCommitIntervalHeight(t *testing.T) {
 	tree.Set([]byte(k2), []byte("k22"))
 	_, _, _, err = tree.SaveVersion(false)
 
-	require.Equal(t, 5, len(tree.ndb.prePersistNodeCache)+len(tree.ndb.nodeCache))
-	require.Equal(t, 3, len(tree.ndb.orphanNodeCache))
+	tree.ndb.sanityCheckHandleOrphansResult(tree.version+1)
+	tree.ndb.oi.enqueueResult(tree.version)
+
+	require.Equal(t, 5, len(tree.ndb.prePersistNodeCache)+tree.ndb.nc.nodeCacheLen())
+	require.Equal(t, 3, tree.ndb.oi.orphanNodeCacheLen())
 
 	_, _, _, err = tree.SaveVersion(false)
 	require.NoError(t, err)
@@ -124,7 +127,7 @@ func TestSaveVersionCommitIntervalHeight(t *testing.T) {
 	_, _, _, err = tree.SaveVersion(false)
 	require.NoError(t, err)
 	require.Equal(t, 0, len(tree.ndb.prePersistNodeCache))
-	require.Equal(t, 0, len(tree.ndb.orphanNodeCache))
+	require.Equal(t, 0, tree.ndb.oi.orphanNodeCacheLen())
 
 	//require.Equal(t, 5, len(tree.ndb.nodeCache)+len(tree.ndb.tempPrePersistNodeCache))
 	tree.Set([]byte("k5"), []byte("5555555555"))
@@ -467,7 +470,7 @@ func TestStopTree(t *testing.T) {
 	_, _, _, err := tree.SaveVersion(false)
 	require.NoError(t, err)
 	tree.StopTree()
-	require.Equal(t, 5, len(tree.ndb.nodeCache))
+	require.Equal(t, 5, tree.ndb.nc.nodeCacheLen())
 }
 
 func TestLog(t *testing.T) {

libs/iavl/nodedb.go

@@ -47,18 +47,6 @@ type nodeDB struct {
 
 	latestVersion  int64
 
-	//lruNodeCache   *lru.Cache
-
-	nodeCache      map[string]*list.Element // Node cache.
-	nodeCacheSize  int                      // Node cache size limit in elements.
-	nodeCacheQueue *syncList                // LRU queue of cache elements. Used for deletion.
-	nodeCacheMutex sync.RWMutex             // Mutex for node cache.
-
-	orphanNodeCache         map[string]*Node
-	heightOrphansCacheQueue *list.List
-	heightOrphansCacheSize  int
-	heightOrphansMap        map[int64]*heightOrphansItem
-
 	prePersistNodeCache map[string]*Node
 	tppMap              map[int64]*tppItem
 	tppVersionList      *list.List
@@ -76,6 +64,9 @@ type nodeDB struct {
 	name string
 
 	preWriteNodeCache cmap.ConcurrentMap
+
+	oi *OrphanInfo
+	nc *NodeCache
 }
 
 func makeNodeCacheMap(cacheSize int, initRatio float64) map[string]*list.Element {
@@ -97,25 +88,16 @@ func newNodeDB(db dbm.DB, cacheSize int, opts *Options) *nodeDB {
 	ndb := &nodeDB{
 		db:                      db,
 		opts:                    *opts,
-		latestVersion:           0, // initially invalid
-		nodeCache:               makeNodeCacheMap(cacheSize, IavlCacheInitRatio),
-		nodeCacheSize:           cacheSize,
-		nodeCacheQueue:          newSyncList(),
 		versionReaders:          make(map[int64]uint32, 8),
-		orphanNodeCache:         make(map[string]*Node),
-		heightOrphansCacheQueue: list.New(),
-		heightOrphansCacheSize:  HeightOrphansCacheSize,
-		heightOrphansMap:        make(map[int64]*heightOrphansItem),
 		prePersistNodeCache:     make(map[string]*Node),
 		tppMap:                  make(map[int64]*tppItem),
 		tppVersionList:          list.New(),
-		dbReadCount:             0,
-		dbReadTime:              0,
-		dbWriteCount:            0,
 		name:                    ParseDBName(db),
 		preWriteNodeCache:       cmap.New(),
 	}
 
+	ndb.oi = newOrphanInfo(ndb)
+	ndb.nc = newNodeCache(cacheSize)
 	return ndb
 }
 
@@ -138,7 +120,7 @@ func (ndb *nodeDB) getNodeFromMemory(hash []byte, promoteRecentNode bool) *Node
 		return elem
 	}
 
-	if elem, ok := ndb.orphanNodeCache[string(hash)]; ok {
+	if elem := ndb.oi.getNodeFromOrphanCache(hash); elem != nil {
 		return elem
 	}
 

libs/iavl/nodedb_cache.go

@@ -1,41 +1,32 @@
 package iavl
 
 import (
-	cmap "github.com/orcaman/concurrent-map"
-	"github.com/tendermint/go-amino"
+	"container/list"
 	"github.com/okex/exchain/libs/iavl/config"
-
+	"github.com/tendermint/go-amino"
+	"sync"
 )
 
-func (ndb *nodeDB) uncacheNodeRontine(n []*Node) {
-	for _, node := range n {
-		ndb.uncacheNode(node.hash)
-	}
+type NodeCache struct {
+	nodeCache      map[string]*list.Element // Node cache.
+	nodeCacheSize  int                      // Node cache size limit in elements.
+	nodeCacheQueue *syncList                // LRU queue of cache elements. Used for deletion.
+	nodeCacheMutex sync.RWMutex             // Mutex for node cache.
 }
 
-func (ndb *nodeDB) initPreWriteCache() {
-	if ndb.preWriteNodeCache == nil {
-		ndb.preWriteNodeCache = cmap.New()
+func newNodeCache(cacheSize int) *NodeCache {
+	return &NodeCache{
+		nodeCache:      makeNodeCacheMap(cacheSize, IavlCacheInitRatio),
+		nodeCacheSize:  cacheSize,
+		nodeCacheQueue: newSyncList(),
 	}
 }
 
-func (ndb *nodeDB) cacheNodeToPreWriteCache(n *Node) {
-	ndb.preWriteNodeCache.Set(string(n.hash), n)
-}
-
-func (ndb *nodeDB) finishPreWriteCache() {
-	ndb.preWriteNodeCache.IterCb(func(key string, v interface{}) {
-		ndb.cacheNode(v.(*Node))
-	})
-	ndb.preWriteNodeCache = nil
-}
-
-
 // ===================================================
 // ======= map[string]*list.Element implementation
 // ===================================================
 
-func (ndb *nodeDB) uncacheNode(hash []byte) {
+func (ndb *NodeCache) uncache(hash []byte) {
 	ndb.nodeCacheMutex.Lock()
 	if elem, ok := ndb.nodeCache[string(hash)]; ok {
 		ndb.nodeCacheQueue.Remove(elem)
@@ -46,7 +37,7 @@ func (ndb *nodeDB) uncacheNode(hash []byte) {
 
 // Add a node to the cache and pop the least recently used node if we've
 // reached the cache size limit.
-func (ndb *nodeDB) cacheNode(node *Node) {
+func (ndb *NodeCache) cache(node *Node) {
 	ndb.nodeCacheMutex.Lock()
 	elem := ndb.nodeCacheQueue.PushBack(node)
 	ndb.nodeCache[string(node.hash)] = elem
@@ -59,17 +50,16 @@ func (ndb *nodeDB) cacheNode(node *Node) {
 	ndb.nodeCacheMutex.Unlock()
 }
 
-func (ndb *nodeDB) cacheNodeByCheck(node *Node) {
+func (ndb *NodeCache) cacheByCheck(node *Node) {
 	ndb.nodeCacheMutex.RLock()
 	_, ok := ndb.nodeCache[string(node.hash)]
 	ndb.nodeCacheMutex.RUnlock()
 	if !ok {
-		ndb.cacheNode(node)
+		ndb.cache(node)
 	}
 }
 
-
-func (ndb *nodeDB) getNodeFromCache(hash []byte, promoteRecentNode bool) (n *Node) {
+func (ndb *NodeCache) get(hash []byte, promoteRecentNode bool) (n *Node) {
 	// Check the cache.
 	ndb.nodeCacheMutex.RLock()
 	elem, ok := ndb.nodeCache[string(hash)]
@@ -84,7 +74,7 @@ func (ndb *nodeDB) getNodeFromCache(hash []byte, promoteRecentNode bool) (n *Nod
 	return
 }
 
-func (ndb *nodeDB) nodeCacheLen() int {
+func (ndb *NodeCache) nodeCacheLen() int {
 	return len(ndb.nodeCache)
 }