2019-08-13 20:35:19 +02:00
|
|
|
package workingsetcache
|
|
|
|
|
|
|
|
import (
|
|
|
|
"sync"
|
|
|
|
"sync/atomic"
|
|
|
|
"time"
|
|
|
|
|
2020-12-08 19:49:32 +01:00
|
|
|
"github.com/VictoriaMetrics/VictoriaMetrics/lib/cgroup"
|
2019-08-13 20:35:19 +02:00
|
|
|
"github.com/VictoriaMetrics/fastcache"
|
|
|
|
)
|
|
|
|
|
2019-09-08 22:21:13 +02:00
|
|
|
// Cache modes.
|
|
|
|
const (
|
|
|
|
split = 0
|
|
|
|
switching = 1
|
|
|
|
whole = 2
|
|
|
|
)
|
|
|
|
|
2019-08-13 20:35:19 +02:00
|
|
|
// Cache is a cache for working set entries.
|
|
|
|
//
|
|
|
|
// The cache evicts inactive entries after the given expireDuration.
|
|
|
|
// Recently accessed entries survive expireDuration.
|
|
|
|
//
|
|
|
|
// Comparing to fastcache, this cache minimizes the required RAM size
|
|
|
|
// to values smaller than maxBytes.
|
|
|
|
type Cache struct {
|
|
|
|
curr atomic.Value
|
|
|
|
prev atomic.Value
|
|
|
|
|
2019-09-08 22:21:13 +02:00
|
|
|
// mode indicates whether to use only curr and skip prev.
|
2019-08-15 21:57:43 +02:00
|
|
|
//
|
2019-09-08 22:21:13 +02:00
|
|
|
// This flag is set to switching if curr is filled for more than 50% space.
|
2019-08-15 21:57:43 +02:00
|
|
|
// In this case using prev would result in RAM waste,
|
|
|
|
// it is better to use only curr cache with doubled size.
|
2019-09-08 22:21:13 +02:00
|
|
|
// After the process of switching, this flag will be set to whole.
|
|
|
|
mode uint64
|
2019-08-15 21:57:43 +02:00
|
|
|
|
2019-09-08 22:21:13 +02:00
|
|
|
// mu serializes access to curr, prev and mode
|
2019-08-15 21:57:43 +02:00
|
|
|
// in expirationWorker and cacheSizeWatcher.
|
|
|
|
mu sync.Mutex
|
|
|
|
|
2019-08-13 20:35:19 +02:00
|
|
|
wg sync.WaitGroup
|
|
|
|
stopCh chan struct{}
|
|
|
|
|
2020-04-10 10:49:06 +02:00
|
|
|
// historicalStats keeps historical counters from fastcache.Stats
|
|
|
|
historicalStats fastcache.Stats
|
2019-08-13 20:35:19 +02:00
|
|
|
}
|
|
|
|
|
|
|
|
// Load loads the cache from filePath and limits its size to maxBytes
|
|
|
|
// and evicts inactive entires after expireDuration.
|
|
|
|
//
|
|
|
|
// Stop must be called on the returned cache when it is no longer needed.
|
|
|
|
func Load(filePath string, maxBytes int, expireDuration time.Duration) *Cache {
|
|
|
|
curr := fastcache.LoadFromFileOrNew(filePath, maxBytes)
|
2019-09-08 22:21:13 +02:00
|
|
|
var cs fastcache.Stats
|
|
|
|
curr.UpdateStats(&cs)
|
|
|
|
if cs.EntriesCount == 0 {
|
|
|
|
curr.Reset()
|
|
|
|
// The cache couldn't be loaded with maxBytes size.
|
|
|
|
// This may mean that the cache is split into curr and prev caches.
|
|
|
|
// Try loading it again with maxBytes / 2 size.
|
|
|
|
maxBytes /= 2
|
|
|
|
curr = fastcache.LoadFromFileOrNew(filePath, maxBytes)
|
|
|
|
return newWorkingSetCache(curr, maxBytes, expireDuration)
|
|
|
|
}
|
|
|
|
|
|
|
|
// The cache has been successfully loaded in full.
|
|
|
|
// Set its' mode to `whole`.
|
|
|
|
// There is no need in starting expirationWorker and cacheSizeWatcher.
|
|
|
|
var c Cache
|
|
|
|
c.curr.Store(curr)
|
|
|
|
c.prev.Store(fastcache.New(1024))
|
|
|
|
c.stopCh = make(chan struct{})
|
|
|
|
atomic.StoreUint64(&c.mode, whole)
|
|
|
|
return &c
|
2019-08-13 20:35:19 +02:00
|
|
|
}
|
|
|
|
|
|
|
|
// New creates new cache with the given maxBytes size and the given expireDuration
|
|
|
|
// for inactive entries.
|
|
|
|
//
|
|
|
|
// Stop must be called on the returned cache when it is no longer needed.
|
|
|
|
func New(maxBytes int, expireDuration time.Duration) *Cache {
|
2019-08-15 21:57:43 +02:00
|
|
|
// Split maxBytes between curr and prev caches.
|
|
|
|
maxBytes /= 2
|
2019-08-13 20:35:19 +02:00
|
|
|
curr := fastcache.New(maxBytes)
|
|
|
|
return newWorkingSetCache(curr, maxBytes, expireDuration)
|
|
|
|
}
|
|
|
|
|
|
|
|
func newWorkingSetCache(curr *fastcache.Cache, maxBytes int, expireDuration time.Duration) *Cache {
|
|
|
|
prev := fastcache.New(1024)
|
|
|
|
var c Cache
|
|
|
|
c.curr.Store(curr)
|
|
|
|
c.prev.Store(prev)
|
|
|
|
c.stopCh = make(chan struct{})
|
2019-09-08 22:21:13 +02:00
|
|
|
atomic.StoreUint64(&c.mode, split)
|
2019-08-15 21:57:43 +02:00
|
|
|
|
|
|
|
c.wg.Add(1)
|
|
|
|
go func() {
|
|
|
|
defer c.wg.Done()
|
|
|
|
c.expirationWorker(maxBytes, expireDuration)
|
|
|
|
}()
|
2019-08-13 20:35:19 +02:00
|
|
|
c.wg.Add(1)
|
|
|
|
go func() {
|
|
|
|
defer c.wg.Done()
|
2019-08-15 21:57:43 +02:00
|
|
|
c.cacheSizeWatcher(maxBytes)
|
|
|
|
}()
|
|
|
|
return &c
|
|
|
|
}
|
2019-08-13 20:35:19 +02:00
|
|
|
|
2019-08-15 21:57:43 +02:00
|
|
|
func (c *Cache) expirationWorker(maxBytes int, expireDuration time.Duration) {
|
|
|
|
t := time.NewTicker(expireDuration / 2)
|
|
|
|
for {
|
|
|
|
select {
|
|
|
|
case <-c.stopCh:
|
|
|
|
t.Stop()
|
|
|
|
return
|
|
|
|
case <-t.C:
|
|
|
|
}
|
|
|
|
|
|
|
|
c.mu.Lock()
|
2019-09-08 22:21:13 +02:00
|
|
|
if atomic.LoadUint64(&c.mode) == split {
|
2019-08-15 21:57:43 +02:00
|
|
|
// Expire prev cache and create fresh curr cache.
|
2019-08-13 20:35:19 +02:00
|
|
|
// Do not reuse prev cache, since it can have too big capacity.
|
|
|
|
prev := c.prev.Load().(*fastcache.Cache)
|
|
|
|
prev.Reset()
|
|
|
|
curr := c.curr.Load().(*fastcache.Cache)
|
2020-04-10 10:49:06 +02:00
|
|
|
curr.UpdateStats(&c.historicalStats)
|
2019-08-13 20:35:19 +02:00
|
|
|
c.prev.Store(curr)
|
|
|
|
curr = fastcache.New(maxBytes)
|
|
|
|
c.curr.Store(curr)
|
|
|
|
}
|
2019-08-15 21:57:43 +02:00
|
|
|
c.mu.Unlock()
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
func (c *Cache) cacheSizeWatcher(maxBytes int) {
|
|
|
|
t := time.NewTicker(time.Minute)
|
2019-09-08 22:21:13 +02:00
|
|
|
defer t.Stop()
|
|
|
|
|
2019-08-15 21:57:43 +02:00
|
|
|
for {
|
|
|
|
select {
|
|
|
|
case <-c.stopCh:
|
|
|
|
return
|
|
|
|
case <-t.C:
|
|
|
|
}
|
|
|
|
var cs fastcache.Stats
|
|
|
|
curr := c.curr.Load().(*fastcache.Cache)
|
|
|
|
curr.UpdateStats(&cs)
|
2019-09-08 22:21:13 +02:00
|
|
|
if cs.BytesSize >= uint64(maxBytes)/2 {
|
|
|
|
break
|
2019-08-15 21:57:43 +02:00
|
|
|
}
|
2019-09-08 22:21:13 +02:00
|
|
|
}
|
2019-08-15 21:57:43 +02:00
|
|
|
|
2019-09-08 22:21:13 +02:00
|
|
|
// curr cache size exceeds 50% of its capacity. It is better
|
|
|
|
// to double the size of curr cache and stop using prev cache,
|
|
|
|
// since this will result in higher summary cache capacity.
|
|
|
|
//
|
|
|
|
// Do this in the following steps:
|
|
|
|
// 1) switch to mode=switching
|
|
|
|
// 2) move curr cache to prev
|
|
|
|
// 3) create curr with the double size
|
|
|
|
// 4) wait until curr size exceeds maxBytes/2, i.e. it is populated with new data
|
|
|
|
// 5) switch to mode=whole
|
|
|
|
// 6) drop prev
|
|
|
|
|
|
|
|
c.mu.Lock()
|
|
|
|
atomic.StoreUint64(&c.mode, switching)
|
|
|
|
prev := c.prev.Load().(*fastcache.Cache)
|
|
|
|
prev.Reset()
|
|
|
|
curr := c.curr.Load().(*fastcache.Cache)
|
2020-04-10 10:49:06 +02:00
|
|
|
curr.UpdateStats(&c.historicalStats)
|
2019-09-08 22:21:13 +02:00
|
|
|
c.prev.Store(curr)
|
|
|
|
c.curr.Store(fastcache.New(maxBytes * 2))
|
|
|
|
c.mu.Unlock()
|
|
|
|
|
|
|
|
for {
|
|
|
|
select {
|
|
|
|
case <-c.stopCh:
|
|
|
|
return
|
|
|
|
case <-t.C:
|
|
|
|
}
|
|
|
|
var cs fastcache.Stats
|
|
|
|
curr := c.curr.Load().(*fastcache.Cache)
|
|
|
|
curr.UpdateStats(&cs)
|
|
|
|
if cs.BytesSize >= uint64(maxBytes)/2 {
|
|
|
|
break
|
|
|
|
}
|
2019-08-15 21:57:43 +02:00
|
|
|
}
|
2019-09-08 22:21:13 +02:00
|
|
|
|
|
|
|
c.mu.Lock()
|
|
|
|
atomic.StoreUint64(&c.mode, whole)
|
|
|
|
prev = c.prev.Load().(*fastcache.Cache)
|
|
|
|
prev.Reset()
|
|
|
|
c.prev.Store(fastcache.New(1024))
|
|
|
|
c.mu.Unlock()
|
2019-08-13 20:35:19 +02:00
|
|
|
}
|
|
|
|
|
|
|
|
// Save safes the cache to filePath.
|
|
|
|
func (c *Cache) Save(filePath string) error {
|
|
|
|
curr := c.curr.Load().(*fastcache.Cache)
|
2020-12-08 19:49:32 +01:00
|
|
|
concurrency := cgroup.AvailableCPUs()
|
2019-08-13 20:35:19 +02:00
|
|
|
return curr.SaveToFileConcurrent(filePath, concurrency)
|
|
|
|
}
|
|
|
|
|
|
|
|
// Stop stops the cache.
|
|
|
|
//
|
|
|
|
// The cache cannot be used after the Stop call.
|
|
|
|
func (c *Cache) Stop() {
|
|
|
|
close(c.stopCh)
|
|
|
|
c.wg.Wait()
|
|
|
|
|
|
|
|
c.Reset()
|
|
|
|
}
|
|
|
|
|
|
|
|
// Reset resets the cache.
|
|
|
|
func (c *Cache) Reset() {
|
|
|
|
prev := c.prev.Load().(*fastcache.Cache)
|
|
|
|
prev.Reset()
|
|
|
|
curr := c.curr.Load().(*fastcache.Cache)
|
|
|
|
curr.Reset()
|
|
|
|
}
|
|
|
|
|
|
|
|
// UpdateStats updates fcs with cache stats.
|
|
|
|
func (c *Cache) UpdateStats(fcs *fastcache.Stats) {
|
|
|
|
curr := c.curr.Load().(*fastcache.Cache)
|
|
|
|
curr.UpdateStats(fcs)
|
2020-04-10 10:49:06 +02:00
|
|
|
|
|
|
|
// Add counters from historical stats
|
|
|
|
hs := &c.historicalStats
|
|
|
|
fcs.GetCalls += atomic.LoadUint64(&hs.GetCalls)
|
|
|
|
fcs.SetCalls += atomic.LoadUint64(&hs.SetCalls)
|
|
|
|
fcs.Misses += atomic.LoadUint64(&hs.Misses)
|
|
|
|
fcs.Collisions += atomic.LoadUint64(&hs.Collisions)
|
|
|
|
fcs.Corruptions += atomic.LoadUint64(&hs.Corruptions)
|
|
|
|
|
2019-09-08 22:21:13 +02:00
|
|
|
if atomic.LoadUint64(&c.mode) == whole {
|
2019-08-13 20:35:19 +02:00
|
|
|
return
|
|
|
|
}
|
|
|
|
|
2020-04-10 10:49:06 +02:00
|
|
|
// Add stats for entries from the previous cache
|
|
|
|
// Do not add counters from the previous cache, since they are already
|
|
|
|
// taken into account via c.historicalStats.
|
2019-08-13 20:35:19 +02:00
|
|
|
prev := c.prev.Load().(*fastcache.Cache)
|
2020-04-10 10:49:06 +02:00
|
|
|
var fcsTmp fastcache.Stats
|
|
|
|
prev.UpdateStats(&fcsTmp)
|
|
|
|
fcs.EntriesCount += fcsTmp.EntriesCount
|
|
|
|
fcs.BytesSize += fcsTmp.BytesSize
|
2019-08-13 20:35:19 +02:00
|
|
|
}
|
|
|
|
|
|
|
|
// Get appends the found value for the given key to dst and returns the result.
|
|
|
|
func (c *Cache) Get(dst, key []byte) []byte {
|
|
|
|
curr := c.curr.Load().(*fastcache.Cache)
|
|
|
|
result := curr.Get(dst, key)
|
|
|
|
if len(result) > len(dst) {
|
|
|
|
// Fast path - the entry is found in the current cache.
|
|
|
|
return result
|
|
|
|
}
|
2019-09-08 22:21:13 +02:00
|
|
|
if atomic.LoadUint64(&c.mode) == whole {
|
2019-08-13 20:35:19 +02:00
|
|
|
return result
|
|
|
|
}
|
|
|
|
|
|
|
|
// Search for the entry in the previous cache.
|
|
|
|
prev := c.prev.Load().(*fastcache.Cache)
|
|
|
|
result = prev.Get(dst, key)
|
|
|
|
if len(result) <= len(dst) {
|
|
|
|
// Nothing found.
|
|
|
|
return result
|
|
|
|
}
|
|
|
|
// Cache the found entry in the current cache.
|
|
|
|
curr.Set(key, result[len(dst):])
|
|
|
|
return result
|
|
|
|
}
|
|
|
|
|
|
|
|
// Has verifies whether the cahce contains the given key.
|
|
|
|
func (c *Cache) Has(key []byte) bool {
|
|
|
|
curr := c.curr.Load().(*fastcache.Cache)
|
|
|
|
if curr.Has(key) {
|
|
|
|
return true
|
|
|
|
}
|
2019-09-08 22:21:13 +02:00
|
|
|
if atomic.LoadUint64(&c.mode) == whole {
|
2019-08-13 20:35:19 +02:00
|
|
|
return false
|
|
|
|
}
|
|
|
|
prev := c.prev.Load().(*fastcache.Cache)
|
|
|
|
return prev.Has(key)
|
|
|
|
}
|
|
|
|
|
|
|
|
// Set sets the given value for the given key.
|
|
|
|
func (c *Cache) Set(key, value []byte) {
|
|
|
|
curr := c.curr.Load().(*fastcache.Cache)
|
|
|
|
curr.Set(key, value)
|
|
|
|
}
|
|
|
|
|
|
|
|
// GetBig appends the found value for the given key to dst and returns the result.
|
|
|
|
func (c *Cache) GetBig(dst, key []byte) []byte {
|
|
|
|
curr := c.curr.Load().(*fastcache.Cache)
|
|
|
|
result := curr.GetBig(dst, key)
|
|
|
|
if len(result) > len(dst) {
|
|
|
|
// Fast path - the entry is found in the current cache.
|
|
|
|
return result
|
|
|
|
}
|
2019-09-08 22:21:13 +02:00
|
|
|
if atomic.LoadUint64(&c.mode) == whole {
|
2019-08-13 20:35:19 +02:00
|
|
|
return result
|
|
|
|
}
|
|
|
|
|
|
|
|
// Search for the entry in the previous cache.
|
|
|
|
prev := c.prev.Load().(*fastcache.Cache)
|
|
|
|
result = prev.GetBig(dst, key)
|
|
|
|
if len(result) <= len(dst) {
|
|
|
|
// Nothing found.
|
|
|
|
return result
|
|
|
|
}
|
|
|
|
// Cache the found entry in the current cache.
|
|
|
|
curr.SetBig(key, result[len(dst):])
|
|
|
|
return result
|
|
|
|
}
|
|
|
|
|
|
|
|
// SetBig sets the given value for the given key.
|
|
|
|
func (c *Cache) SetBig(key, value []byte) {
|
|
|
|
curr := c.curr.Load().(*fastcache.Cache)
|
|
|
|
curr.SetBig(key, value)
|
|
|
|
}
|