// Copyright (c) 2012, Suryandaru Triandana // All rights reserved. // // Use of this source code is governed by a BSD-style license that can be // found in the LICENSE file. // Package cache provides interface and implementation of a cache algorithms. package cache import ( "sort" "sync" "sync/atomic" "unsafe" "github.com/syndtr/goleveldb/leveldb/util" ) // Cacher provides interface to implements a caching functionality. // An implementation must be safe for concurrent use. type Cacher interface { // Capacity returns cache capacity. Capacity() int // SetCapacity sets cache capacity. SetCapacity(capacity int) // Promote promotes the 'cache node'. Promote(n *Node) // Ban evicts the 'cache node' and prevent subsequent 'promote'. Ban(n *Node) // Evict evicts the 'cache node'. Evict(n *Node) } // Value is a 'cache-able object'. It may implements util.Releaser, if // so the the Release method will be called once object is released. type Value interface{} // NamespaceGetter provides convenient wrapper for namespace. type NamespaceGetter struct { Cache *Cache NS uint64 } // Get simply calls Cache.Get() method. func (g *NamespaceGetter) Get(key uint64, setFunc func() (size int, value Value)) *Handle { return g.Cache.Get(g.NS, key, setFunc) } // The hash tables implementation is based on: // "Dynamic-Sized Nonblocking Hash Tables", by Yujie Liu, // Kunlong Zhang, and Michael Spear. // ACM Symposium on Principles of Distributed Computing, Jul 2014. const ( mInitialSize = 1 << 4 mOverflowThreshold = 1 << 5 mOverflowGrowThreshold = 1 << 7 ) const ( bucketUninitialized = iota bucketInitialized bucketFrozen ) type mNodes []*Node func (x mNodes) Len() int { return len(x) } func (x mNodes) Less(i, j int) bool { a, b := x[i].ns, x[j].ns if a == b { return x[i].key < x[j].key } return a < b } func (x mNodes) Swap(i, j int) { x[i], x[j] = x[j], x[i] } func (x mNodes) sort() { sort.Sort(x) } func (x mNodes) search(ns, key uint64) int { return sort.Search(len(x), func(i int) bool { a := x[i].ns if a == ns { return x[i].key >= key } return a > ns }) } type mBucket struct { mu sync.Mutex nodes mNodes state int8 } func (b *mBucket) freeze() mNodes { b.mu.Lock() defer b.mu.Unlock() if b.state == bucketInitialized { b.state = bucketFrozen } else if b.state == bucketUninitialized { panic("BUG: freeze uninitialized bucket") } return b.nodes } func (b *mBucket) frozen() bool { if b.state == bucketFrozen { return true } if b.state == bucketUninitialized { panic("BUG: accessing uninitialized bucket") } return false } func (b *mBucket) get(r *Cache, h *mHead, hash uint32, ns, key uint64, getOnly bool) (done, created bool, n *Node) { b.mu.Lock() if b.frozen() { b.mu.Unlock() return } // Find the node. i := b.nodes.search(ns, key) if i < len(b.nodes) { n = b.nodes[i] if n.ns == ns && n.key == key { atomic.AddInt32(&n.ref, 1) b.mu.Unlock() return true, false, n } } // Get only. if getOnly { b.mu.Unlock() return true, false, nil } // Create node. n = &Node{ r: r, hash: hash, ns: ns, key: key, ref: 1, } // Add node to bucket. if i == len(b.nodes) { b.nodes = append(b.nodes, n) } else { b.nodes = append(b.nodes[:i+1], b.nodes[i:]...) b.nodes[i] = n } bLen := len(b.nodes) b.mu.Unlock() // Update counter. grow := atomic.AddInt64(&r.statNodes, 1) >= h.growThreshold if bLen > mOverflowThreshold { grow = grow || atomic.AddInt32(&h.overflow, 1) >= mOverflowGrowThreshold } // Grow. if grow && atomic.CompareAndSwapInt32(&h.resizeInProgress, 0, 1) { nhLen := len(h.buckets) << 1 nh := &mHead{ buckets: make([]mBucket, nhLen), mask: uint32(nhLen) - 1, predecessor: unsafe.Pointer(h), growThreshold: int64(nhLen * mOverflowThreshold), shrinkThreshold: int64(nhLen >> 1), } ok := atomic.CompareAndSwapPointer(&r.mHead, unsafe.Pointer(h), unsafe.Pointer(nh)) if !ok { panic("BUG: failed swapping head") } atomic.AddInt32(&r.statGrow, 1) go nh.initBuckets() } return true, true, n } func (b *mBucket) delete(r *Cache, h *mHead, hash uint32, ns, key uint64) (done, deleted bool) { b.mu.Lock() if b.frozen() { b.mu.Unlock() return } // Find the node. i := b.nodes.search(ns, key) if i == len(b.nodes) { b.mu.Unlock() return true, false } n := b.nodes[i] var bLen int if n.ns == ns && n.key == key { if atomic.LoadInt32(&n.ref) == 0 { deleted = true // Save and clear value. if n.value != nil { // Call releaser. if r, ok := n.value.(util.Releaser); ok { r.Release() } n.value = nil } // Remove node from bucket. b.nodes = append(b.nodes[:i], b.nodes[i+1:]...) bLen = len(b.nodes) } } b.mu.Unlock() if deleted { // Call delete funcs. for _, f := range n.delFuncs { f() } // Update counter. atomic.AddInt64(&r.statSize, int64(n.size)*-1) shrink := atomic.AddInt64(&r.statNodes, -1) < h.shrinkThreshold if bLen >= mOverflowThreshold { atomic.AddInt32(&h.overflow, -1) } // Shrink. if shrink && len(h.buckets) > mInitialSize && atomic.CompareAndSwapInt32(&h.resizeInProgress, 0, 1) { nhLen := len(h.buckets) >> 1 nh := &mHead{ buckets: make([]mBucket, nhLen), mask: uint32(nhLen) - 1, predecessor: unsafe.Pointer(h), growThreshold: int64(nhLen * mOverflowThreshold), shrinkThreshold: int64(nhLen >> 1), } ok := atomic.CompareAndSwapPointer(&r.mHead, unsafe.Pointer(h), unsafe.Pointer(nh)) if !ok { panic("BUG: failed swapping head") } atomic.AddInt32(&r.statShrink, 1) go nh.initBuckets() } } return true, deleted } type mHead struct { buckets []mBucket mask uint32 predecessor unsafe.Pointer // *mNode resizeInProgress int32 overflow int32 growThreshold int64 shrinkThreshold int64 } func (h *mHead) initBucket(i uint32) *mBucket { b := &h.buckets[i] b.mu.Lock() if b.state >= bucketInitialized { b.mu.Unlock() return b } p := (*mHead)(atomic.LoadPointer(&h.predecessor)) if p == nil { panic("BUG: uninitialized bucket doesn't have predecessor") } var nodes mNodes if h.mask > p.mask { // Grow. m := p.initBucket(i & p.mask).freeze() // Split nodes. for _, x := range m { if x.hash&h.mask == i { nodes = append(nodes, x) } } } else { // Shrink. m0 := p.initBucket(i).freeze() m1 := p.initBucket(i + uint32(len(h.buckets))).freeze() // Merge nodes. nodes = make(mNodes, 0, len(m0)+len(m1)) nodes = append(nodes, m0...) nodes = append(nodes, m1...) nodes.sort() } b.nodes = nodes b.state = bucketInitialized b.mu.Unlock() return b } func (h *mHead) initBuckets() { for i := range h.buckets { h.initBucket(uint32(i)) } atomic.StorePointer(&h.predecessor, nil) } func (h *mHead) enumerateNodesWithCB(f func([]*Node)) { var nodes []*Node for x := range h.buckets { b := h.initBucket(uint32(x)) b.mu.Lock() nodes = append(nodes, b.nodes...) b.mu.Unlock() f(nodes) } } func (h *mHead) enumerateNodesByNS(ns uint64) []*Node { var nodes []*Node for x := range h.buckets { b := h.initBucket(uint32(x)) b.mu.Lock() i := b.nodes.search(ns, 0) for ; i < len(b.nodes); i++ { n := b.nodes[i] if n.ns != ns { break } nodes = append(nodes, n) } b.mu.Unlock() } return nodes } type Stats struct { Buckets int Nodes int64 Size int64 GrowCount int32 ShrinkCount int32 HitCount int64 MissCount int64 SetCount int64 DelCount int64 } // Cache is a 'cache map'. type Cache struct { mu sync.RWMutex mHead unsafe.Pointer // *mNode cacher Cacher closed bool statNodes int64 statSize int64 statGrow int32 statShrink int32 statHit int64 statMiss int64 statSet int64 statDel int64 } // NewCache creates a new 'cache map'. The cacher is optional and // may be nil. func NewCache(cacher Cacher) *Cache { h := &mHead{ buckets: make([]mBucket, mInitialSize), mask: mInitialSize - 1, growThreshold: int64(mInitialSize * mOverflowThreshold), shrinkThreshold: 0, } for i := range h.buckets { h.buckets[i].state = bucketInitialized } r := &Cache{ mHead: unsafe.Pointer(h), cacher: cacher, } return r } func (r *Cache) getBucket(hash uint32) (*mHead, *mBucket) { h := (*mHead)(atomic.LoadPointer(&r.mHead)) i := hash & h.mask return h, h.initBucket(i) } func (r *Cache) enumerateNodesWithCB(f func([]*Node)) { h := (*mHead)(atomic.LoadPointer(&r.mHead)) h.enumerateNodesWithCB(f) } func (r *Cache) enumerateNodesByNS(ns uint64) []*Node { h := (*mHead)(atomic.LoadPointer(&r.mHead)) return h.enumerateNodesByNS(ns) } func (r *Cache) delete(n *Node) bool { for { h, b := r.getBucket(n.hash) done, deleted := b.delete(r, h, n.hash, n.ns, n.key) if done { return deleted } } } // GetStats returns cache statistics. func (r *Cache) GetStats() Stats { return Stats{ Buckets: len((*mHead)(atomic.LoadPointer(&r.mHead)).buckets), Nodes: atomic.LoadInt64(&r.statNodes), Size: atomic.LoadInt64(&r.statSize), GrowCount: atomic.LoadInt32(&r.statGrow), ShrinkCount: atomic.LoadInt32(&r.statShrink), HitCount: atomic.LoadInt64(&r.statHit), MissCount: atomic.LoadInt64(&r.statMiss), SetCount: atomic.LoadInt64(&r.statSet), DelCount: atomic.LoadInt64(&r.statDel), } } // Nodes returns number of 'cache node' in the map. func (r *Cache) Nodes() int { return int(atomic.LoadInt64(&r.statNodes)) } // Size returns sums of 'cache node' size in the map. func (r *Cache) Size() int { return int(atomic.LoadInt64(&r.statSize)) } // Capacity returns cache capacity. func (r *Cache) Capacity() int { if r.cacher == nil { return 0 } return r.cacher.Capacity() } // SetCapacity sets cache capacity. func (r *Cache) SetCapacity(capacity int) { if r.cacher != nil { r.cacher.SetCapacity(capacity) } } // Get gets 'cache node' with the given namespace and key. // If cache node is not found and setFunc is not nil, Get will atomically creates // the 'cache node' by calling setFunc. Otherwise Get will returns nil. // // The returned 'cache handle' should be released after use by calling Release // method. func (r *Cache) Get(ns, key uint64, setFunc func() (size int, value Value)) *Handle { r.mu.RLock() defer r.mu.RUnlock() if r.closed { return nil } hash := murmur32(ns, key, 0xf00) for { h, b := r.getBucket(hash) done, created, n := b.get(r, h, hash, ns, key, setFunc == nil) if done { if created || n == nil { atomic.AddInt64(&r.statMiss, 1) } else { atomic.AddInt64(&r.statHit, 1) } if n != nil { n.mu.Lock() if n.value == nil { if setFunc == nil { n.mu.Unlock() n.unRefInternal(false) return nil } n.size, n.value = setFunc() if n.value == nil { n.size = 0 n.mu.Unlock() n.unRefInternal(false) return nil } atomic.AddInt64(&r.statSet, 1) atomic.AddInt64(&r.statSize, int64(n.size)) } n.mu.Unlock() if r.cacher != nil { r.cacher.Promote(n) } return &Handle{unsafe.Pointer(n)} } break } } return nil } // Delete removes and ban 'cache node' with the given namespace and key. // A banned 'cache node' will never inserted into the 'cache tree'. Ban // only attributed to the particular 'cache node', so when a 'cache node' // is recreated it will not be banned. // // If delFunc is not nil, then it will be executed if such 'cache node' // doesn't exist or once the 'cache node' is released. // // Delete return true is such 'cache node' exist. func (r *Cache) Delete(ns, key uint64, delFunc func()) bool { r.mu.RLock() defer r.mu.RUnlock() if r.closed { return false } hash := murmur32(ns, key, 0xf00) for { h, b := r.getBucket(hash) done, _, n := b.get(r, h, hash, ns, key, true) if done { if n != nil { if delFunc != nil { n.mu.Lock() n.delFuncs = append(n.delFuncs, delFunc) n.mu.Unlock() } if r.cacher != nil { r.cacher.Ban(n) } n.unRefInternal(true) return true } break } } if delFunc != nil { delFunc() } return false } // Evict evicts 'cache node' with the given namespace and key. This will // simply call Cacher.Evict. // // Evict return true is such 'cache node' exist. func (r *Cache) Evict(ns, key uint64) bool { r.mu.RLock() defer r.mu.RUnlock() if r.closed { return false } hash := murmur32(ns, key, 0xf00) for { h, b := r.getBucket(hash) done, _, n := b.get(r, h, hash, ns, key, true) if done { if n != nil { if r.cacher != nil { r.cacher.Evict(n) } n.unRefInternal(true) return true } break } } return false } // EvictNS evicts 'cache node' with the given namespace. This will // simply call Cacher.Evict on all nodes with the given namespace. func (r *Cache) EvictNS(ns uint64) { r.mu.RLock() defer r.mu.RUnlock() if r.closed { return } if r.cacher != nil { nodes := r.enumerateNodesByNS(ns) for _, n := range nodes { r.cacher.Evict(n) } } } func (r *Cache) evictAll() { r.enumerateNodesWithCB(func(nodes []*Node) { for _, n := range nodes { r.cacher.Evict(n) } }) } // EvictAll evicts all 'cache node'. This will simply call Cacher.EvictAll. func (r *Cache) EvictAll() { r.mu.RLock() defer r.mu.RUnlock() if r.closed { return } if r.cacher != nil { r.evictAll() } } // Close closes the 'cache map'. // All 'Cache' method is no-op after 'cache map' is closed. // All 'cache node' will be evicted from 'cacher'. // // If 'force' is true then all 'cache node' will be forcefully released // even if the 'node ref' is not zero. func (r *Cache) Close(force bool) { var head *mHead // Hold RW-lock to make sure no more in-flight operations. r.mu.Lock() if !r.closed { r.closed = true head = (*mHead)(atomic.LoadPointer(&r.mHead)) atomic.StorePointer(&r.mHead, nil) } r.mu.Unlock() if head != nil { head.enumerateNodesWithCB(func(nodes []*Node) { for _, n := range nodes { // Zeroing ref. Prevent unRefExternal to call finalizer. if force { atomic.StoreInt32(&n.ref, 0) } // Evict from cacher. if r.cacher != nil { r.cacher.Evict(n) } if force { n.callFinalizer() } } }) } } // Node is a 'cache node'. type Node struct { r *Cache hash uint32 ns, key uint64 mu sync.Mutex size int value Value ref int32 delFuncs []func() CacheData unsafe.Pointer } // NS returns this 'cache node' namespace. func (n *Node) NS() uint64 { return n.ns } // Key returns this 'cache node' key. func (n *Node) Key() uint64 { return n.key } // Size returns this 'cache node' size. func (n *Node) Size() int { return n.size } // Value returns this 'cache node' value. func (n *Node) Value() Value { return n.value } // Ref returns this 'cache node' ref counter. func (n *Node) Ref() int32 { return atomic.LoadInt32(&n.ref) } // GetHandle returns an handle for this 'cache node'. func (n *Node) GetHandle() *Handle { if atomic.AddInt32(&n.ref, 1) <= 1 { panic("BUG: Node.GetHandle on zero ref") } return &Handle{unsafe.Pointer(n)} } func (n *Node) callFinalizer() { // Call releaser. if n.value != nil { if r, ok := n.value.(util.Releaser); ok { r.Release() } n.value = nil } // Call delete funcs. for _, f := range n.delFuncs { f() } n.delFuncs = nil } func (n *Node) unRefInternal(updateStat bool) { if atomic.AddInt32(&n.ref, -1) == 0 { n.r.delete(n) if updateStat { atomic.AddInt64(&n.r.statDel, 1) } } } func (n *Node) unRefExternal() { if atomic.AddInt32(&n.ref, -1) == 0 { n.r.mu.RLock() if n.r.closed { n.callFinalizer() } else { n.r.delete(n) atomic.AddInt64(&n.r.statDel, 1) } n.r.mu.RUnlock() } } // Handle is a 'cache handle' of a 'cache node'. type Handle struct { n unsafe.Pointer // *Node } // Value returns the value of the 'cache node'. func (h *Handle) Value() Value { n := (*Node)(atomic.LoadPointer(&h.n)) if n != nil { return n.value } return nil } // Release releases this 'cache handle'. // It is safe to call release multiple times. func (h *Handle) Release() { nPtr := atomic.LoadPointer(&h.n) if nPtr != nil && atomic.CompareAndSwapPointer(&h.n, nPtr, nil) { n := (*Node)(nPtr) n.unRefExternal() } } func murmur32(ns, key uint64, seed uint32) uint32 { const ( m = uint32(0x5bd1e995) r = 24 ) k1 := uint32(ns >> 32) k2 := uint32(ns) k3 := uint32(key >> 32) k4 := uint32(key) k1 *= m k1 ^= k1 >> r k1 *= m k2 *= m k2 ^= k2 >> r k2 *= m k3 *= m k3 ^= k3 >> r k3 *= m k4 *= m k4 ^= k4 >> r k4 *= m h := seed h *= m h ^= k1 h *= m h ^= k2 h *= m h ^= k3 h *= m h ^= k4 h ^= h >> 13 h *= m h ^= h >> 15 return h }