diff --git a/cache/cache.go b/cache/cache.go index 04375838e..fa3f3c290 100644 --- a/cache/cache.go +++ b/cache/cache.go @@ -23,6 +23,9 @@ type Cache interface { // Returns Node for the key, if exists. nil otherwise. Get(key []byte) Node + // Peek is like Get but does not refresh the entry's LRU position. + Peek(key []byte) Node + // Has returns true if node with key exists in cache, false otherwise. Has(key []byte) bool @@ -87,6 +90,13 @@ func (c *lruCache) Get(key []byte) Node { return nil } +func (c *lruCache) Peek(key []byte) Node { + if ele, hit := c.dict[string(key)]; hit { + return ele.Value.(Node) + } + return nil +} + func (c *lruCache) Has(key []byte) bool { _, exists := c.dict[string(key)] return exists diff --git a/cache/cache_test.go b/cache/cache_test.go index 4a0c75bf4..10a7976f8 100644 --- a/cache/cache_test.go +++ b/cache/cache_test.go @@ -307,3 +307,14 @@ func randBytes(length int) []byte { rand.Read(key) //nolint:errcheck return key } + +func TestPeekDoesNotRefreshRecency(t *testing.T) { + c := cache.New(2) + c.Add(&testNode{key: []byte("a")}) + c.Add(&testNode{key: []byte("b")}) + require.NotNil(t, c.Peek([]byte("a"))) + require.Nil(t, c.Peek([]byte("zz"))) + // "a" is still the oldest, so adding "c" evicts it. + evicted := c.Add(&testNode{key: []byte("c")}) + require.Equal(t, []byte("a"), evicted.GetKey()) +} diff --git a/cache_opts_test.go b/cache_opts_test.go new file mode 100644 index 000000000..d8f676c0d --- /dev/null +++ b/cache_opts_test.go @@ -0,0 +1,130 @@ +package iavl + +import ( + "bytes" + "fmt" + "testing" + + "github.com/stretchr/testify/require" + + dbm "github.com/cosmos/iavl/db" +) + +func TestCacheValueSizeLimit(t *testing.T) { + small, big := []byte("small"), bytes.Repeat([]byte("x"), 64) + tree := NewMutableTree(dbm.NewMemDB(), 1000, false, NewNopLogger(), CacheValueSizeLimitOption(16)) + _, err := tree.Set([]byte("a"), small) + require.NoError(t, err) + _, err = tree.Set([]byte("b"), big) + require.NoError(t, err) + _, _, err = tree.SaveVersion() + require.NoError(t, err) + + ndb := tree.ndb + require.True(t, ndb.fastNodeCache.Has([]byte("a"))) + require.False(t, ndb.fastNodeCache.Has([]byte("b")), "large fast node must not be cached") + var bigLeaf int + leaves, err := ndb.leafNodes() + require.NoError(t, err) + for _, l := range leaves { + cached := ndb.nodeCache.Has(l.GetKey()) + if len(l.value) > 16 { + bigLeaf++ + require.False(t, cached, "large leaf must not be cached") + } else { + require.True(t, cached) + } + } + require.Equal(t, 1, bigLeaf) + + // reads of the large value still work and don't populate the caches. + v, err := tree.Get([]byte("b")) + require.NoError(t, err) + require.Equal(t, big, v) + require.False(t, ndb.fastNodeCache.Has([]byte("b"))) + + // growing a cached value past the limit must drop the stale cache entry. + _, err = tree.Set([]byte("a"), big) + require.NoError(t, err) + _, _, err = tree.SaveVersion() + require.NoError(t, err) + require.False(t, ndb.fastNodeCache.Has([]byte("a"))) + v, err = tree.Get([]byte("a")) + require.NoError(t, err) + require.Equal(t, big, v) +} + +func TestFastNodeCacheSizeOption(t *testing.T) { + db := dbm.NewMemDB() + require.Equal(t, defaultFastNodeCacheSize, fastCacheCap(NewMutableTree(db, 0, false, NewNopLogger()))) + + tree := NewMutableTree(db, 0, false, NewNopLogger(), FastNodeCacheSizeOption(3)) + for i := 0; i < 10; i++ { + _, err := tree.Set([]byte(fmt.Sprintf("k%d", i)), []byte("v")) + require.NoError(t, err) + } + _, _, err := tree.SaveVersion() + require.NoError(t, err) + require.Equal(t, 3, tree.ndb.fastNodeCache.Len()) +} + +func fastCacheCap(tree *MutableTree) int { + for i := 0; ; i++ { + tree.ndb.fastNodeCache.Add(&testCacheNode{key: []byte(fmt.Sprintf("%d", i))}) + if tree.ndb.fastNodeCache.Len() <= i { + return i + } + if i > 2*defaultFastNodeCacheSize { + return -1 + } + } +} + +type testCacheNode struct{ key []byte } + +func (n *testCacheNode) GetKey() []byte { return n.key } + +func TestPruningEvictsAndSkipsCache(t *testing.T) { + db := dbm.NewMemDB() + tree := NewMutableTree(db, 100000, true, NewNopLogger()) + for v := 0; v < 5; v++ { + for i := 0; i < 10; i++ { + _, err := tree.Set([]byte(fmt.Sprintf("k%d", i)), []byte(fmt.Sprintf("v%d-%d", v, i))) + require.NoError(t, err) + } + _, _, err := tree.SaveVersion() + require.NoError(t, err) + } + ndb := tree.ndb + + // nodes orphaned by the version update. + var orphans [][]byte + require.NoError(t, ndb.traverseOrphans(1, 2, func(o *Node) error { + orphans = append(orphans, o.GetKey()) + return nil + })) + require.NotEmpty(t, orphans) + for _, k := range orphans { + require.True(t, ndb.nodeCache.Has(k), "setup: orphan should be cached") + } + + require.NoError(t, ndb.DeleteVersionsTo(1)) + require.NoError(t, ndb.Commit()) + // version 1 pruned: its orphans are gone from disk and must be gone from the cache. + require.NoError(t, ndb.DeleteVersionsTo(2)) + require.NoError(t, ndb.Commit()) + for _, k := range orphans { + require.False(t, ndb.nodeCache.Has(k), "pruned node still cached: %x", k) + has, err := ndb.Has(k) + require.NoError(t, err) + require.False(t, has) + } + + // a pruning read of an uncached node must not populate the cache. + latest, err := ndb.GetRoot(5) + require.NoError(t, err) + ndb.nodeCache.Remove(latest) + _, err = ndb.getNode(latest, true) + require.NoError(t, err) + require.False(t, ndb.nodeCache.Has(latest)) +} diff --git a/iterator.go b/iterator.go index b49c35c91..ea3ab0864 100644 --- a/iterator.go +++ b/iterator.go @@ -268,26 +268,30 @@ type NodeIterator struct { nodesToVisit []*Node ndb *nodeDB err error + pruning bool // read through the node cache without refreshing or populating it } // NewNodeIterator returns a new NodeIterator to traverse the tree of the root node. func NewNodeIterator(rootKey []byte, ndb *nodeDB) (*NodeIterator, error) { + return newNodeIterator(rootKey, ndb, false) +} + +func newNodeIterator(rootKey []byte, ndb *nodeDB, pruning bool) (*NodeIterator, error) { + iter := &NodeIterator{nodesToVisit: []*Node{}, ndb: ndb, pruning: pruning} if len(rootKey) == 0 { - return &NodeIterator{ - nodesToVisit: []*Node{}, - ndb: ndb, - }, nil + return iter, nil } - node, err := ndb.GetNode(rootKey) + node, err := iter.getNode(rootKey) if err != nil { return nil, err } + iter.nodesToVisit = append(iter.nodesToVisit, node) + return iter, nil +} - return &NodeIterator{ - nodesToVisit: []*Node{node}, - ndb: ndb, - }, nil +func (iter *NodeIterator) getNode(nk []byte) (*Node, error) { + return iter.ndb.getNode(nk, iter.pruning) } // GetNode returns the current visiting node. @@ -322,14 +326,14 @@ func (iter *NodeIterator) Next(isSkipped bool) { return } - rightNode, err := iter.ndb.GetNode(node.rightNodeKey) + rightNode, err := iter.getNode(node.rightNodeKey) if err != nil { iter.err = err return } iter.nodesToVisit = append(iter.nodesToVisit, rightNode) - leftNode, err := iter.ndb.GetNode(node.leftNodeKey) + leftNode, err := iter.getNode(node.leftNodeKey) if err != nil { iter.err = err return diff --git a/nodedb.go b/nodedb.go index be1f52357..e9ceb155e 100644 --- a/nodedb.go +++ b/nodedb.go @@ -36,7 +36,7 @@ const ( // Using semantic versioning: https://semver.org/ defaultStorageVersionValue = "1.0.0" fastStorageVersionValue = "1.1.0" - fastNodeCacheSize = 100000 + defaultFastNodeCacheSize = 100000 ) var ( @@ -103,6 +103,11 @@ func newNodeDB(db corestore.KVStoreWithBatch, cacheSize int, opts Options, lg Lo } ctx, cancel := context.WithCancel(context.Background()) + fastNodeCacheSize := opts.FastNodeCacheSize + if fastNodeCacheSize <= 0 { + fastNodeCacheSize = defaultFastNodeCacheSize + } + ndb := &nodeDB{ ctx: ctx, cancel: cancel, @@ -134,6 +139,25 @@ func newNodeDB(db corestore.KVStoreWithBatch, cacheSize int, opts Options, lg Lo // It is used for both formats of nodes: legacy and new. // `legacy`: nk is the hash of the node. `new`: . func (ndb *nodeDB) GetNode(nk []byte) (*Node, error) { + return ndb.getNode(nk, false) +} + +// cacheable reports whether a value of the given size may be held in the caches. +func (ndb *nodeDB) cacheable(valueLen int) bool { + limit := ndb.opts.CacheValueSizeLimit + return limit <= 0 || valueLen <= limit +} + +// evictNode drops a node from the node cache. Used by pruning once a node is deleted. +func (ndb *nodeDB) evictNode(nk []byte) { + ndb.mtx.Lock() + defer ndb.mtx.Unlock() + ndb.nodeCache.Remove(nk) +} + +// getNode is GetNode. During pruning, it peeks without refreshing LRU +// order and skips caching misses to avoid evicting frequently accessed nodes. +func (ndb *nodeDB) getNode(nk []byte, pruning bool) (*Node, error) { ndb.mtx.Lock() defer ndb.mtx.Unlock() @@ -142,7 +166,13 @@ func (ndb *nodeDB) GetNode(nk []byte) (*Node, error) { } // Check the cache. - if cachedNode := ndb.nodeCache.Get(nk); cachedNode != nil { + var cachedNode cache.Node + if pruning { + cachedNode = ndb.nodeCache.Peek(nk) + } else { + cachedNode = ndb.nodeCache.Get(nk) + } + if cachedNode != nil { ndb.opts.Stat.IncCacheHitCnt() return cachedNode.(*Node), nil } @@ -192,7 +222,9 @@ func (ndb *nodeDB) GetNode(nk []byte) (*Node, error) { } } - ndb.nodeCache.Add(node) + if !pruning && (!node.isLeaf() || ndb.cacheable(len(node.value))) { + ndb.nodeCache.Add(node) + } return node, nil } @@ -229,7 +261,9 @@ func (ndb *nodeDB) GetFastNode(key []byte) (*fastnode.Node, error) { if err != nil { return nil, fmt.Errorf("error reading FastNode. bytes: %x, error: %w", buf, err) } - ndb.fastNodeCache.Add(fastNode) + if ndb.cacheable(len(fastNode.GetValue())) { + ndb.fastNodeCache.Add(fastNode) + } return fastNode, nil } @@ -255,7 +289,9 @@ func (ndb *nodeDB) SaveNode(node *Node) error { } ndb.logger.Debug("BATCH SAVE", "node", node) - ndb.nodeCache.Add(node) + if !node.isLeaf() || ndb.cacheable(len(node.value)) { + ndb.nodeCache.Add(node) + } return nil } @@ -381,7 +417,12 @@ func (ndb *nodeDB) saveFastNodeUnlocked(node *fastnode.Node, shouldAddToCache bo if shouldAddToCache { // defer adding the node to the cache until after commit, to ensure // that we do not have a period where the tree and the cache differ - ndb.pendingFastNodeAdditions = append(ndb.pendingFastNodeAdditions, node) + if ndb.cacheable(len(node.GetValue())) { + ndb.pendingFastNodeAdditions = append(ndb.pendingFastNodeAdditions, node) + } else { + // too large to cache so make sure no stale smaller value stays cached. + ndb.pendingFastNodeRemovals = append(ndb.pendingFastNodeRemovals, node.GetKey()) + } } return nil } @@ -475,9 +516,12 @@ func (ndb *nodeDB) deleteVersion(version int64, cache *rootkeyCache) error { if rootKey != nil { if err := ndb.traverseOrphansWithRootkeyCache(cache, version, version+1, func(orphan *Node) error { + // evict before nodeKey is rewritten below, while the cache key still matches. + ndb.evictNode(orphan.GetKey()) if orphan.nodeKey.nonce == 0 && !orphan.isLegacy { // if the orphan is a reformatted root, it can be a legacy root // so it should be removed from the pruning process. + ndb.evictNode(orphan.hash) if err := ndb.deleteFromPruning(ndb.legacyNodeKey(orphan.hash)); err != nil { return err } @@ -502,6 +546,7 @@ func (ndb *nodeDB) deleteVersion(version int64, cache *rootkeyCache) error { if rootKey == nil || !bytes.Equal(rootKey, literalRootKey) { // if the root key is not matched with the literal root key, it means the given root // is a reference root to the previous version. + ndb.evictNode(literalRootKey) if err := ndb.deleteFromPruning(ndb.nodeKey(literalRootKey)); err != nil { return err } @@ -513,10 +558,11 @@ func (ndb *nodeDB) deleteVersion(version int64, cache *rootkeyCache) error { return err } if bytes.Equal(literalRootKey, nextRootKey) { - root, err := ndb.GetNode(nextRootKey) + root, err := ndb.getNode(nextRootKey, true) if err != nil { return err } + ndb.evictNode(nextRootKey) // ensure that the given version is not included in the root search if err := ndb.deleteFromPruning(ndb.nodeKey(literalRootKey)); err != nil { return err @@ -559,6 +605,7 @@ func (ndb *nodeDB) deleteLegacyNodes(version int64, nk []byte) error { func (ndb *nodeDB) deleteLegacyVersions(legacyLatestVersion int64) error { // Delete the last version for the legacyLastVersion if err := ndb.traverseOrphans(legacyLatestVersion, legacyLatestVersion+1, func(orphan *Node) error { + ndb.evictNode(orphan.hash) return ndb.deleteFromPruning(ndb.legacyNodeKey(orphan.hash)) }); err != nil { return err @@ -572,6 +619,7 @@ func (ndb *nodeDB) deleteLegacyVersions(legacyLatestVersion int64) error { var fromVersion, toVersion int64 legacyOrphanKeyFormat.Scan(key, &toVersion, &fromVersion) if (fromVersion <= legacyLatestVersion && toVersion < legacyLatestVersion) || fromVersion > legacyLatestVersion { + ndb.evictNode(value) return ndb.deleteFromPruning(ndb.legacyNodeKey(value)) } return nil @@ -1188,7 +1236,7 @@ func (ndb *nodeDB) traverseOrphansWithRootkeyCache(cache *rootkeyCache, prevVers return err } - curIter, err := NewNodeIterator(curKey, ndb) + curIter, err := newNodeIterator(curKey, ndb, true) if err != nil { return err } @@ -1197,7 +1245,7 @@ func (ndb *nodeDB) traverseOrphansWithRootkeyCache(cache *rootkeyCache, prevVers if err != nil { return err } - prevIter, err := NewNodeIterator(prevKey, ndb) + prevIter, err := newNodeIterator(prevKey, ndb, true) if err != nil { return err } diff --git a/options.go b/options.go index ca65d0ea4..d6371c430 100644 --- a/options.go +++ b/options.go @@ -88,6 +88,14 @@ type Options struct { // AsyncPruning is a flag to enable async pruning AsyncPruning bool + // CacheValueSizeLimit is the maximum value size, in bytes, allowed in node + // and fast node caches. Zero disables the limit. + CacheValueSizeLimit int + + // FastNodeCacheSize is the maximum number of entries in the fast node cache. + // Zero uses the default size. + FastNodeCacheSize int + initialVersionSet bool } @@ -131,3 +139,19 @@ func AsyncPruningOption(asyncPruning bool) Option { opts.AsyncPruning = asyncPruning } } + +// CacheValueSizeLimitOption keeps leaves and fast nodes whose value exceeds limit bytes out of the caches. +// Zero disables the limit. +func CacheValueSizeLimitOption(limit int) Option { + return func(opts *Options) { + opts.CacheValueSizeLimit = limit + } +} + +// FastNodeCacheSizeOption sets the maximum number of entries in the fast node cache. +// Zero selects the default. +func FastNodeCacheSizeOption(size int) Option { + return func(opts *Options) { + opts.FastNodeCacheSize = size + } +}