separate memory cache
This commit is contained in:
Umputun
+1
-1
@@ -130,7 +130,7 @@ func New(opts Opts) (*Application, error) {
|
||||
MaxCommentSize: opts.MaxCommentSize,
|
||||
}
|
||||
|
||||
loadingCache, err := cache.NewLoadingCache(cache.MaxValSize(opts.MaxCachedValue), cache.MaxKeys(opts.MaxCachedItems),
|
||||
loadingCache, err := cache.NewMemoryCache(cache.MaxValSize(opts.MaxCachedValue), cache.MaxKeys(opts.MaxCachedItems),
|
||||
cache.PostFlushFn(postFlushFn(opts.Sites, opts.Port)))
|
||||
if err != nil {
|
||||
return nil, err
|
||||
|
||||
Vendored
+2
-114
@@ -1,12 +1,9 @@
|
||||
package cache
|
||||
|
||||
import (
|
||||
"log"
|
||||
"net/http"
|
||||
"strings"
|
||||
"sync/atomic"
|
||||
|
||||
"github.com/hashicorp/golang-lru"
|
||||
"github.com/pkg/errors"
|
||||
"github.com/umputun/remark/app/rest"
|
||||
)
|
||||
@@ -22,7 +19,8 @@ func Key(key string, scopes ...string) string {
|
||||
return strings.Join(scopes, "$$") + "@@" + key
|
||||
}
|
||||
|
||||
func parseKey(fullKey string) (key string, scopes []string, err error) {
|
||||
// ParseKey gets compound key created by Key func and split it to the actual key and scopes
|
||||
func ParseKey(fullKey string) (key string, scopes []string, err error) {
|
||||
elems := strings.Split(fullKey, "@@")
|
||||
if len(elems) != 2 {
|
||||
return "", nil, errors.Errorf("can't parse cache key %s", key)
|
||||
@@ -35,116 +33,6 @@ func parseKey(fullKey string) (key string, scopes []string, err error) {
|
||||
return key, scopes, nil
|
||||
}
|
||||
|
||||
// loadingCache implements LoadingCache interface on top of cache.Cache (go-cache)
|
||||
type loadingCache struct {
|
||||
bytesCache *lru.Cache
|
||||
postFlushFn func()
|
||||
maxKeys int
|
||||
maxValueSize int
|
||||
maxCacheSize int64
|
||||
currentSize int64
|
||||
}
|
||||
|
||||
// NewLoadingCache makes loadingCache implementation
|
||||
func NewLoadingCache(options ...Option) (LoadingCache, error) {
|
||||
res := loadingCache{
|
||||
postFlushFn: func() {},
|
||||
maxKeys: 1000,
|
||||
maxValueSize: 0,
|
||||
}
|
||||
for _, opt := range options {
|
||||
if err := opt(&res); err != nil {
|
||||
log.Printf("[WARN] failed to set cache option, %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
onEvicted := func(key interface{}, value interface{}) {
|
||||
size := len(value.([]byte))
|
||||
atomic.AddInt64(&res.currentSize, -1*int64(size))
|
||||
}
|
||||
|
||||
var err error
|
||||
// OnEvicted called automatically for expired and manually deleted
|
||||
if res.bytesCache, err = lru.NewWithEvict(res.maxKeys, onEvicted); err != nil {
|
||||
return nil, errors.Wrap(err, "failed to make cache")
|
||||
}
|
||||
|
||||
log.Printf("[DEBUG] create lru cache, maxKeys=%d, maxValueSize=%d", res.maxKeys, res.maxValueSize)
|
||||
return &res, nil
|
||||
}
|
||||
|
||||
// Get is loading cache method to get value by key or load via fn if not found
|
||||
func (lc *loadingCache) Get(key string, fn func() ([]byte, error)) (data []byte, err error) {
|
||||
if b, ok := lc.bytesCache.Get(key); ok {
|
||||
return b.([]byte), nil
|
||||
}
|
||||
|
||||
if data, err = fn(); err != nil {
|
||||
return data, err
|
||||
}
|
||||
if lc.allowed(data) {
|
||||
lc.bytesCache.Add(key, data)
|
||||
atomic.AddInt64(&lc.currentSize, int64(len(data)))
|
||||
|
||||
if lc.maxCacheSize > 0 && atomic.LoadInt64(&lc.currentSize) > lc.maxCacheSize {
|
||||
for atomic.LoadInt64(&lc.currentSize) > lc.maxCacheSize {
|
||||
lc.bytesCache.RemoveOldest()
|
||||
}
|
||||
}
|
||||
}
|
||||
return data, nil
|
||||
}
|
||||
|
||||
// Flush clears cache and calls postFlushFn async
|
||||
func (lc *loadingCache) Flush(scopes ...string) {
|
||||
|
||||
if len(scopes) == 0 {
|
||||
lc.bytesCache.Purge()
|
||||
go lc.postFlushFn()
|
||||
return
|
||||
}
|
||||
|
||||
// check if fullKey has matching scopes
|
||||
inScope := func(fullKey string) bool {
|
||||
for _, s := range scopes {
|
||||
_, keyScopes, err := parseKey(fullKey)
|
||||
if err != nil {
|
||||
return false
|
||||
}
|
||||
for _, ks := range keyScopes {
|
||||
if ks == s {
|
||||
return true
|
||||
}
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
// all matchedKeys should be collected first
|
||||
// we can't delete it from locked section, it will lock on eviction callback
|
||||
matchedKeys := []string{}
|
||||
for _, k := range lc.bytesCache.Keys() {
|
||||
key := k.(string)
|
||||
if inScope(key) {
|
||||
matchedKeys = append(matchedKeys, key)
|
||||
}
|
||||
}
|
||||
for _, mkey := range matchedKeys {
|
||||
lc.bytesCache.Remove(mkey)
|
||||
}
|
||||
|
||||
if lc.postFlushFn != nil {
|
||||
go lc.postFlushFn()
|
||||
}
|
||||
}
|
||||
|
||||
func (lc *loadingCache) allowed(data []byte) bool {
|
||||
if lc.maxValueSize > 0 && len(data) >= lc.maxValueSize {
|
||||
return false
|
||||
}
|
||||
return true
|
||||
}
|
||||
|
||||
// URLKey gets url from request to use it as cache key
|
||||
// admins will have different keys in order to prevent leak of admin-only data to regular users
|
||||
func URLKey(r *http.Request) string {
|
||||
|
||||
Vendored
+21
-314
@@ -1,187 +1,42 @@
|
||||
package cache
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"fmt"
|
||||
"math/rand"
|
||||
"net/http"
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/stretchr/testify/require"
|
||||
|
||||
"github.com/umputun/remark/app/rest"
|
||||
"github.com/umputun/remark/app/store"
|
||||
)
|
||||
|
||||
func TestLoadingCache_Get(t *testing.T) {
|
||||
var postFnCall, coldCalls int32
|
||||
lc, err := NewLoadingCache(PostFlushFn(func() { atomic.AddInt32(&postFnCall, 1) }))
|
||||
require.Nil(t, err)
|
||||
res, err := lc.Get("key", func() ([]byte, error) {
|
||||
atomic.AddInt32(&coldCalls, 1)
|
||||
return []byte("result"), nil
|
||||
})
|
||||
assert.Nil(t, err)
|
||||
assert.Equal(t, "result", string(res))
|
||||
assert.Equal(t, int32(1), atomic.LoadInt32(&coldCalls))
|
||||
assert.Equal(t, int32(0), atomic.LoadInt32(&postFnCall))
|
||||
func TestCache_Keys(t *testing.T) {
|
||||
tbl := []struct {
|
||||
key string
|
||||
scopes []string
|
||||
full string
|
||||
}{
|
||||
{"key1", []string{"s1"}, "s1@@key1"},
|
||||
{"key2", []string{"s11", "s2"}, "s11$$s2@@key2"},
|
||||
{"key3", []string{}, "@@key3"},
|
||||
}
|
||||
|
||||
res, err = lc.Get("key", func() ([]byte, error) {
|
||||
atomic.AddInt32(&coldCalls, 1)
|
||||
return []byte("result"), nil
|
||||
})
|
||||
assert.Nil(t, err)
|
||||
assert.Equal(t, "result", string(res))
|
||||
assert.Equal(t, int32(1), atomic.LoadInt32(&coldCalls))
|
||||
assert.Equal(t, int32(0), atomic.LoadInt32(&postFnCall))
|
||||
for n, tt := range tbl {
|
||||
full := Key(tt.key, tt.scopes...)
|
||||
assert.Equal(t, tt.full, full, "making key, #%d", n)
|
||||
|
||||
lc.Flush()
|
||||
time.Sleep(100 * time.Millisecond) // let postFn to do its thing
|
||||
assert.Equal(t, int32(1), atomic.LoadInt32(&postFnCall))
|
||||
|
||||
_, err = lc.Get("key", func() ([]byte, error) {
|
||||
return nil, errors.New("err")
|
||||
})
|
||||
assert.NotNil(t, err)
|
||||
}
|
||||
|
||||
func TestLoadingCache_MaxKeys(t *testing.T) {
|
||||
var postFnCall, coldCalls int32
|
||||
lc, err := NewLoadingCache(PostFlushFn(func() { atomic.AddInt32(&postFnCall, 1) }),
|
||||
MaxKeys(5), MaxValSize(10))
|
||||
require.Nil(t, err)
|
||||
|
||||
// put 5 keys to cache
|
||||
for i := 0; i < 5; i++ {
|
||||
res, e := lc.Get(fmt.Sprintf("key-%d", i), func() ([]byte, error) {
|
||||
atomic.AddInt32(&coldCalls, 1)
|
||||
return []byte(fmt.Sprintf("result-%d", i)), nil
|
||||
})
|
||||
k, s, e := ParseKey(full)
|
||||
assert.Nil(t, e)
|
||||
assert.Equal(t, fmt.Sprintf("result-%d", i), string(res))
|
||||
assert.Equal(t, int32(i+1), atomic.LoadInt32(&coldCalls))
|
||||
assert.Equal(t, int32(0), atomic.LoadInt32(&postFnCall))
|
||||
assert.Equal(t, tt.scopes, s)
|
||||
assert.Equal(t, tt.key, k)
|
||||
}
|
||||
|
||||
// check if really cached
|
||||
res, err := lc.Get("key-3", func() ([]byte, error) {
|
||||
return []byte("result-blah"), nil
|
||||
})
|
||||
assert.Nil(t, err)
|
||||
assert.Equal(t, "result-3", string(res), "should be cached")
|
||||
|
||||
// try to cache after maxKeys reached
|
||||
res, err = lc.Get("key-X", func() ([]byte, error) {
|
||||
return []byte("result-X"), nil
|
||||
})
|
||||
assert.Nil(t, err)
|
||||
assert.Equal(t, "result-X", string(res))
|
||||
|
||||
assert.Equal(t, 5, lc.(*loadingCache).bytesCache.Len())
|
||||
|
||||
// put to cache and make sure it cached
|
||||
res, err = lc.Get("key-Z", func() ([]byte, error) {
|
||||
return []byte("result-Z"), nil
|
||||
})
|
||||
assert.Nil(t, err)
|
||||
assert.Equal(t, "result-Z", string(res))
|
||||
|
||||
res, err = lc.Get("key-Z", func() ([]byte, error) {
|
||||
return []byte("result-Zzzz"), nil
|
||||
})
|
||||
assert.Nil(t, err)
|
||||
assert.Equal(t, "result-Z", string(res), "got cached value")
|
||||
assert.Equal(t, 5, lc.(*loadingCache).bytesCache.Len())
|
||||
_, _, err := ParseKey("abc")
|
||||
assert.Error(t, err)
|
||||
_, _, err = ParseKey("")
|
||||
assert.Error(t, err)
|
||||
}
|
||||
|
||||
func TestLoadingCache_MaxValueSize(t *testing.T) {
|
||||
lc, err := NewLoadingCache(MaxKeys(5), MaxValSize(10))
|
||||
require.Nil(t, err)
|
||||
// put good size value to cache and make sure it cached
|
||||
res, err := lc.Get("key-Z", func() ([]byte, error) {
|
||||
return []byte("result-Z"), nil
|
||||
})
|
||||
assert.Nil(t, err)
|
||||
assert.Equal(t, "result-Z", string(res))
|
||||
|
||||
res, err = lc.Get("key-Z", func() ([]byte, error) {
|
||||
return []byte("result-Zzzz"), nil
|
||||
})
|
||||
assert.Nil(t, err)
|
||||
assert.Equal(t, "result-Z", string(res), "got cached value")
|
||||
|
||||
// put too big value to cache and make sure it is not cached
|
||||
res, err = lc.Get("key-Big", func() ([]byte, error) {
|
||||
return []byte("1234567890"), nil
|
||||
})
|
||||
assert.Nil(t, err)
|
||||
assert.Equal(t, "1234567890", string(res))
|
||||
|
||||
res, err = lc.Get("key-Big", func() ([]byte, error) {
|
||||
return []byte("result-big"), nil
|
||||
})
|
||||
assert.Nil(t, err)
|
||||
assert.Equal(t, "result-big", string(res), "got not cached value")
|
||||
}
|
||||
|
||||
func TestLoadingCache_MaxCacheSize(t *testing.T) {
|
||||
lc, err := NewLoadingCache(MaxKeys(50), MaxCacheSize(20))
|
||||
require.Nil(t, err)
|
||||
|
||||
// put good size value to cache and make sure it cached
|
||||
res, err := lc.Get("key-Z", func() ([]byte, error) {
|
||||
return []byte("result-Z"), nil
|
||||
})
|
||||
assert.Nil(t, err)
|
||||
assert.Equal(t, "result-Z", string(res))
|
||||
assert.Equal(t, int64(8), lc.(*loadingCache).currentSize)
|
||||
|
||||
_, err = lc.Get("key-Z2", func() ([]byte, error) {
|
||||
return []byte("result-Z"), nil
|
||||
})
|
||||
assert.Nil(t, err)
|
||||
assert.Equal(t, int64(16), lc.(*loadingCache).currentSize)
|
||||
|
||||
// this will cause removal
|
||||
_, err = lc.Get("key-Z3", func() ([]byte, error) {
|
||||
return []byte("result-Z"), nil
|
||||
})
|
||||
assert.Nil(t, err)
|
||||
assert.Equal(t, int64(16), lc.(*loadingCache).currentSize)
|
||||
|
||||
assert.Equal(t, 2, lc.(*loadingCache).bytesCache.Len())
|
||||
}
|
||||
|
||||
func TestLoadingCache_MaxCacheSizeParallel(t *testing.T) {
|
||||
lc, err := NewLoadingCache(MaxCacheSize(123), MaxKeys(10000))
|
||||
require.Nil(t, err)
|
||||
|
||||
wg := sync.WaitGroup{}
|
||||
for i := 0; i < 1000; i++ {
|
||||
wg.Add(1)
|
||||
i := i
|
||||
go func() {
|
||||
time.Sleep(time.Duration(rand.Intn(100)) * time.Nanosecond)
|
||||
defer wg.Done()
|
||||
res, err := lc.Get(fmt.Sprintf("key-%d", i), func() ([]byte, error) {
|
||||
return []byte(fmt.Sprintf("result-%d", i)), nil
|
||||
})
|
||||
require.Nil(t, err)
|
||||
require.Equal(t, fmt.Sprintf("result-%d", i), string(res))
|
||||
size := atomic.LoadInt64(&lc.(*loadingCache).currentSize)
|
||||
require.True(t, size < 200 && size >= 0, "unexpected size=%d", size) // won't be exactly 123 due parallel
|
||||
}()
|
||||
}
|
||||
wg.Wait()
|
||||
assert.True(t, lc.(*loadingCache).currentSize < 123 && lc.(*loadingCache).currentSize >= 0)
|
||||
t.Log("size=", lc.(*loadingCache).currentSize)
|
||||
}
|
||||
|
||||
func TestLoadingCache_URLKey(t *testing.T) {
|
||||
func TestCache_URLKey(t *testing.T) {
|
||||
r, err := http.NewRequest("GET", "http://blah/123", nil)
|
||||
assert.Nil(t, err)
|
||||
key := URLKey(r)
|
||||
@@ -197,151 +52,3 @@ func TestLoadingCache_URLKey(t *testing.T) {
|
||||
key = URLKey(r)
|
||||
assert.Equal(t, "admin!!http://blah/123?key=v&k2=v2", key)
|
||||
}
|
||||
|
||||
func TestLoadingCache_Parallel(t *testing.T) {
|
||||
var coldCalls int32
|
||||
lc, err := NewLoadingCache()
|
||||
require.Nil(t, err)
|
||||
|
||||
res, err := lc.Get("key", func() ([]byte, error) {
|
||||
return []byte("value"), nil
|
||||
})
|
||||
assert.Nil(t, err)
|
||||
assert.Equal(t, "value", string(res))
|
||||
|
||||
wg := sync.WaitGroup{}
|
||||
for i := 0; i < 1000; i++ {
|
||||
wg.Add(1)
|
||||
i := i
|
||||
go func() {
|
||||
defer wg.Done()
|
||||
res, err := lc.Get("key", func() ([]byte, error) {
|
||||
atomic.AddInt32(&coldCalls, 1)
|
||||
return []byte(fmt.Sprintf("result-%d", i)), nil
|
||||
})
|
||||
require.Nil(t, err)
|
||||
require.Equal(t, "value", string(res))
|
||||
}()
|
||||
}
|
||||
wg.Wait()
|
||||
assert.Equal(t, int32(0), atomic.LoadInt32(&coldCalls))
|
||||
}
|
||||
|
||||
func TestLoadingCache_Scopes(t *testing.T) {
|
||||
lc, err := NewLoadingCache()
|
||||
require.Nil(t, err)
|
||||
|
||||
res, err := lc.Get(Key("key", "s1", "s2"), func() ([]byte, error) {
|
||||
return []byte("value"), nil
|
||||
})
|
||||
assert.Nil(t, err)
|
||||
assert.Equal(t, "value", string(res))
|
||||
|
||||
res, err = lc.Get(Key("key2", "s2"), func() ([]byte, error) {
|
||||
return []byte("value2"), nil
|
||||
})
|
||||
assert.Nil(t, err)
|
||||
assert.Equal(t, "value2", string(res))
|
||||
|
||||
assert.Equal(t, 2, lc.(*loadingCache).bytesCache.Len())
|
||||
lc.Flush("s1")
|
||||
assert.Equal(t, 1, lc.(*loadingCache).bytesCache.Len())
|
||||
|
||||
lc.Get(Key("key2", "s2"), func() ([]byte, error) {
|
||||
assert.Fail(t, "should stay")
|
||||
return nil, nil
|
||||
})
|
||||
|
||||
res, err = lc.Get(Key("key", "s1", "s2"), func() ([]byte, error) {
|
||||
return []byte("value-upd"), nil
|
||||
})
|
||||
assert.Nil(t, err)
|
||||
assert.Equal(t, "value-upd", string(res), "was deleted, update")
|
||||
}
|
||||
|
||||
func TestLoadingCache_Flush(t *testing.T) {
|
||||
lc, err := NewLoadingCache()
|
||||
require.Nil(t, err)
|
||||
|
||||
addToCache := func(key string, scopes ...string) {
|
||||
res, err := lc.Get(key, func() ([]byte, error) {
|
||||
return []byte("value" + key), nil
|
||||
})
|
||||
require.Nil(t, err)
|
||||
require.Equal(t, "value"+key, string(res))
|
||||
}
|
||||
|
||||
init := func() {
|
||||
lc.Flush()
|
||||
addToCache(Key("key1", "s1", "s2"))
|
||||
addToCache(Key("key2", "s1", "s2", "s3"))
|
||||
addToCache(Key("key3", "s1", "s2", "s3"))
|
||||
addToCache(Key("key4", "s2", "s3"))
|
||||
addToCache(Key("key5", "s2"))
|
||||
addToCache(Key("key6"))
|
||||
addToCache(Key("key7", "s4", "s3"))
|
||||
require.Equal(t, 7, lc.(*loadingCache).bytesCache.Len(), "cache init")
|
||||
}
|
||||
|
||||
tbl := []struct {
|
||||
scopes []string
|
||||
left int
|
||||
msg string
|
||||
}{
|
||||
{[]string{}, 0, "full flush, no scopes"},
|
||||
{[]string{"s0"}, 7, "flush wrong scope"},
|
||||
{[]string{"s1"}, 4, "flush s1 scope"},
|
||||
{[]string{"s2", "s1"}, 2, "flush s2+s1 scope"},
|
||||
{[]string{"s1", "s2"}, 2, "flush s1+s2 scope"},
|
||||
{[]string{"s1", "s2", "s4"}, 1, "flush s1+s2+s4 scope"},
|
||||
{[]string{"s1", "s2", "s3"}, 1, "flush s1+s2+s3 scope"},
|
||||
{[]string{"s1", "s2", "ss"}, 2, "flush s1+s2+wrong scope"},
|
||||
}
|
||||
|
||||
for i, tt := range tbl {
|
||||
init()
|
||||
lc.Flush(tt.scopes...)
|
||||
assert.Equal(t, tt.left, lc.(*loadingCache).bytesCache.Len(), "keys size, %s #%d", tt.msg, i)
|
||||
}
|
||||
}
|
||||
|
||||
func TestLoadingCache_FlushFailed(t *testing.T) {
|
||||
lc, err := NewLoadingCache()
|
||||
require.Nil(t, err)
|
||||
val, err := lc.Get("invalid-composite", func() ([]byte, error) {
|
||||
return []byte("value"), nil
|
||||
})
|
||||
assert.Nil(t, err)
|
||||
assert.Equal(t, "value", string(val))
|
||||
assert.Equal(t, 1, lc.(*loadingCache).bytesCache.Len())
|
||||
|
||||
lc.Flush("invalid-composite")
|
||||
assert.Equal(t, 1, lc.(*loadingCache).bytesCache.Len())
|
||||
}
|
||||
|
||||
func TestLoadingCache_Keys(t *testing.T) {
|
||||
tbl := []struct {
|
||||
key string
|
||||
scopes []string
|
||||
full string
|
||||
}{
|
||||
{"key1", []string{"s1"}, "s1@@key1"},
|
||||
{"key2", []string{"s11", "s2"}, "s11$$s2@@key2"},
|
||||
{"key3", []string{}, "@@key3"},
|
||||
}
|
||||
|
||||
for n, tt := range tbl {
|
||||
full := Key(tt.key, tt.scopes...)
|
||||
assert.Equal(t, tt.full, full, "making key, #%d", n)
|
||||
|
||||
k, s, e := parseKey(full)
|
||||
assert.Nil(t, e)
|
||||
assert.Equal(t, tt.scopes, s)
|
||||
assert.Equal(t, tt.key, k)
|
||||
}
|
||||
|
||||
_, _, err := parseKey("abc")
|
||||
assert.Error(t, err)
|
||||
_, _, err = parseKey("")
|
||||
assert.Error(t, err)
|
||||
}
|
||||
|
||||
Vendored
+119
@@ -0,0 +1,119 @@
|
||||
package cache
|
||||
|
||||
import (
|
||||
"log"
|
||||
"sync/atomic"
|
||||
|
||||
"github.com/hashicorp/golang-lru"
|
||||
"github.com/pkg/errors"
|
||||
)
|
||||
|
||||
// memoryCache implements LoadingCache interface on top of cache.Cache (go-cache)
|
||||
type memoryCache struct {
|
||||
bytesCache *lru.Cache
|
||||
postFlushFn func()
|
||||
maxKeys int
|
||||
maxValueSize int
|
||||
maxCacheSize int64
|
||||
currentSize int64
|
||||
}
|
||||
|
||||
// NewMemoryCache makes memoryCache implementation
|
||||
func NewMemoryCache(options ...Option) (LoadingCache, error) {
|
||||
res := memoryCache{
|
||||
postFlushFn: func() {},
|
||||
maxKeys: 1000,
|
||||
maxValueSize: 0,
|
||||
}
|
||||
for _, opt := range options {
|
||||
if err := opt(&res); err != nil {
|
||||
log.Printf("[WARN] failed to set cache option, %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
onEvicted := func(key interface{}, value interface{}) {
|
||||
size := len(value.([]byte))
|
||||
atomic.AddInt64(&res.currentSize, -1*int64(size))
|
||||
}
|
||||
|
||||
var err error
|
||||
// OnEvicted called automatically for expired and manually deleted
|
||||
if res.bytesCache, err = lru.NewWithEvict(res.maxKeys, onEvicted); err != nil {
|
||||
return nil, errors.Wrap(err, "failed to make cache")
|
||||
}
|
||||
|
||||
log.Printf("[DEBUG] create lru cache, maxKeys=%d, maxValueSize=%d", res.maxKeys, res.maxValueSize)
|
||||
return &res, nil
|
||||
}
|
||||
|
||||
// Get is loading cache method to get value by key or load via fn if not found
|
||||
func (m *memoryCache) Get(key string, fn func() ([]byte, error)) (data []byte, err error) {
|
||||
if b, ok := m.bytesCache.Get(key); ok {
|
||||
return b.([]byte), nil
|
||||
}
|
||||
|
||||
if data, err = fn(); err != nil {
|
||||
return data, err
|
||||
}
|
||||
if m.allowed(data) {
|
||||
m.bytesCache.Add(key, data)
|
||||
atomic.AddInt64(&m.currentSize, int64(len(data)))
|
||||
|
||||
if m.maxCacheSize > 0 && atomic.LoadInt64(&m.currentSize) > m.maxCacheSize {
|
||||
for atomic.LoadInt64(&m.currentSize) > m.maxCacheSize {
|
||||
m.bytesCache.RemoveOldest()
|
||||
}
|
||||
}
|
||||
}
|
||||
return data, nil
|
||||
}
|
||||
|
||||
// Flush clears cache and calls postFlushFn async
|
||||
func (m *memoryCache) Flush(scopes ...string) {
|
||||
|
||||
if len(scopes) == 0 {
|
||||
m.bytesCache.Purge()
|
||||
go m.postFlushFn()
|
||||
return
|
||||
}
|
||||
|
||||
// check if fullKey has matching scopes
|
||||
inScope := func(fullKey string) bool {
|
||||
for _, s := range scopes {
|
||||
_, keyScopes, err := ParseKey(fullKey)
|
||||
if err != nil {
|
||||
return false
|
||||
}
|
||||
for _, ks := range keyScopes {
|
||||
if ks == s {
|
||||
return true
|
||||
}
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
// all matchedKeys should be collected first
|
||||
// we can't delete it from locked section, it will lock on eviction callback
|
||||
matchedKeys := []string{}
|
||||
for _, k := range m.bytesCache.Keys() {
|
||||
key := k.(string)
|
||||
if inScope(key) {
|
||||
matchedKeys = append(matchedKeys, key)
|
||||
}
|
||||
}
|
||||
for _, mkey := range matchedKeys {
|
||||
m.bytesCache.Remove(mkey)
|
||||
}
|
||||
|
||||
if m.postFlushFn != nil {
|
||||
go m.postFlushFn()
|
||||
}
|
||||
}
|
||||
|
||||
func (m *memoryCache) allowed(data []byte) bool {
|
||||
if m.maxValueSize > 0 && len(data) >= m.maxValueSize {
|
||||
return false
|
||||
}
|
||||
return true
|
||||
}
|
||||
Vendored
+299
@@ -0,0 +1,299 @@
|
||||
package cache
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"math/rand"
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/pkg/errors"
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
func TestMemoryCache_Get(t *testing.T) {
|
||||
var postFnCall, coldCalls int32
|
||||
lc, err := NewMemoryCache(PostFlushFn(func() { atomic.AddInt32(&postFnCall, 1) }))
|
||||
require.Nil(t, err)
|
||||
res, err := lc.Get("key", func() ([]byte, error) {
|
||||
atomic.AddInt32(&coldCalls, 1)
|
||||
return []byte("result"), nil
|
||||
})
|
||||
assert.Nil(t, err)
|
||||
assert.Equal(t, "result", string(res))
|
||||
assert.Equal(t, int32(1), atomic.LoadInt32(&coldCalls))
|
||||
assert.Equal(t, int32(0), atomic.LoadInt32(&postFnCall))
|
||||
|
||||
res, err = lc.Get("key", func() ([]byte, error) {
|
||||
atomic.AddInt32(&coldCalls, 1)
|
||||
return []byte("result"), nil
|
||||
})
|
||||
assert.Nil(t, err)
|
||||
assert.Equal(t, "result", string(res))
|
||||
assert.Equal(t, int32(1), atomic.LoadInt32(&coldCalls))
|
||||
assert.Equal(t, int32(0), atomic.LoadInt32(&postFnCall))
|
||||
|
||||
lc.Flush()
|
||||
time.Sleep(100 * time.Millisecond) // let postFn to do its thing
|
||||
assert.Equal(t, int32(1), atomic.LoadInt32(&postFnCall))
|
||||
|
||||
_, err = lc.Get("key", func() ([]byte, error) {
|
||||
return nil, errors.New("err")
|
||||
})
|
||||
assert.NotNil(t, err)
|
||||
}
|
||||
|
||||
func TestMemoryCache_MaxKeys(t *testing.T) {
|
||||
var postFnCall, coldCalls int32
|
||||
lc, err := NewMemoryCache(PostFlushFn(func() { atomic.AddInt32(&postFnCall, 1) }),
|
||||
MaxKeys(5), MaxValSize(10))
|
||||
require.Nil(t, err)
|
||||
|
||||
// put 5 keys to cache
|
||||
for i := 0; i < 5; i++ {
|
||||
res, e := lc.Get(fmt.Sprintf("key-%d", i), func() ([]byte, error) {
|
||||
atomic.AddInt32(&coldCalls, 1)
|
||||
return []byte(fmt.Sprintf("result-%d", i)), nil
|
||||
})
|
||||
assert.Nil(t, e)
|
||||
assert.Equal(t, fmt.Sprintf("result-%d", i), string(res))
|
||||
assert.Equal(t, int32(i+1), atomic.LoadInt32(&coldCalls))
|
||||
assert.Equal(t, int32(0), atomic.LoadInt32(&postFnCall))
|
||||
}
|
||||
|
||||
// check if really cached
|
||||
res, err := lc.Get("key-3", func() ([]byte, error) {
|
||||
return []byte("result-blah"), nil
|
||||
})
|
||||
assert.Nil(t, err)
|
||||
assert.Equal(t, "result-3", string(res), "should be cached")
|
||||
|
||||
// try to cache after maxKeys reached
|
||||
res, err = lc.Get("key-X", func() ([]byte, error) {
|
||||
return []byte("result-X"), nil
|
||||
})
|
||||
assert.Nil(t, err)
|
||||
assert.Equal(t, "result-X", string(res))
|
||||
|
||||
assert.Equal(t, 5, lc.(*memoryCache).bytesCache.Len())
|
||||
|
||||
// put to cache and make sure it cached
|
||||
res, err = lc.Get("key-Z", func() ([]byte, error) {
|
||||
return []byte("result-Z"), nil
|
||||
})
|
||||
assert.Nil(t, err)
|
||||
assert.Equal(t, "result-Z", string(res))
|
||||
|
||||
res, err = lc.Get("key-Z", func() ([]byte, error) {
|
||||
return []byte("result-Zzzz"), nil
|
||||
})
|
||||
assert.Nil(t, err)
|
||||
assert.Equal(t, "result-Z", string(res), "got cached value")
|
||||
assert.Equal(t, 5, lc.(*memoryCache).bytesCache.Len())
|
||||
}
|
||||
|
||||
func TestMemoryCache_MaxValueSize(t *testing.T) {
|
||||
lc, err := NewMemoryCache(MaxKeys(5), MaxValSize(10))
|
||||
require.Nil(t, err)
|
||||
// put good size value to cache and make sure it cached
|
||||
res, err := lc.Get("key-Z", func() ([]byte, error) {
|
||||
return []byte("result-Z"), nil
|
||||
})
|
||||
assert.Nil(t, err)
|
||||
assert.Equal(t, "result-Z", string(res))
|
||||
|
||||
res, err = lc.Get("key-Z", func() ([]byte, error) {
|
||||
return []byte("result-Zzzz"), nil
|
||||
})
|
||||
assert.Nil(t, err)
|
||||
assert.Equal(t, "result-Z", string(res), "got cached value")
|
||||
|
||||
// put too big value to cache and make sure it is not cached
|
||||
res, err = lc.Get("key-Big", func() ([]byte, error) {
|
||||
return []byte("1234567890"), nil
|
||||
})
|
||||
assert.Nil(t, err)
|
||||
assert.Equal(t, "1234567890", string(res))
|
||||
|
||||
res, err = lc.Get("key-Big", func() ([]byte, error) {
|
||||
return []byte("result-big"), nil
|
||||
})
|
||||
assert.Nil(t, err)
|
||||
assert.Equal(t, "result-big", string(res), "got not cached value")
|
||||
}
|
||||
|
||||
func TestMemoryCache_MaxCacheSize(t *testing.T) {
|
||||
lc, err := NewMemoryCache(MaxKeys(50), MaxCacheSize(20))
|
||||
require.Nil(t, err)
|
||||
|
||||
// put good size value to cache and make sure it cached
|
||||
res, err := lc.Get("key-Z", func() ([]byte, error) {
|
||||
return []byte("result-Z"), nil
|
||||
})
|
||||
assert.Nil(t, err)
|
||||
assert.Equal(t, "result-Z", string(res))
|
||||
assert.Equal(t, int64(8), lc.(*memoryCache).currentSize)
|
||||
|
||||
_, err = lc.Get("key-Z2", func() ([]byte, error) {
|
||||
return []byte("result-Z"), nil
|
||||
})
|
||||
assert.Nil(t, err)
|
||||
assert.Equal(t, int64(16), lc.(*memoryCache).currentSize)
|
||||
|
||||
// this will cause removal
|
||||
_, err = lc.Get("key-Z3", func() ([]byte, error) {
|
||||
return []byte("result-Z"), nil
|
||||
})
|
||||
assert.Nil(t, err)
|
||||
assert.Equal(t, int64(16), lc.(*memoryCache).currentSize)
|
||||
|
||||
assert.Equal(t, 2, lc.(*memoryCache).bytesCache.Len())
|
||||
}
|
||||
|
||||
func TestMemoryCache_MaxCacheSizeParallel(t *testing.T) {
|
||||
lc, err := NewMemoryCache(MaxCacheSize(123), MaxKeys(10000))
|
||||
require.Nil(t, err)
|
||||
|
||||
wg := sync.WaitGroup{}
|
||||
for i := 0; i < 1000; i++ {
|
||||
wg.Add(1)
|
||||
i := i
|
||||
go func() {
|
||||
time.Sleep(time.Duration(rand.Intn(100)) * time.Nanosecond)
|
||||
defer wg.Done()
|
||||
res, err := lc.Get(fmt.Sprintf("key-%d", i), func() ([]byte, error) {
|
||||
return []byte(fmt.Sprintf("result-%d", i)), nil
|
||||
})
|
||||
require.Nil(t, err)
|
||||
require.Equal(t, fmt.Sprintf("result-%d", i), string(res))
|
||||
size := atomic.LoadInt64(&lc.(*memoryCache).currentSize)
|
||||
require.True(t, size < 200 && size >= 0, "unexpected size=%d", size) // won't be exactly 123 due parallel
|
||||
}()
|
||||
}
|
||||
wg.Wait()
|
||||
assert.True(t, lc.(*memoryCache).currentSize < 123 && lc.(*memoryCache).currentSize >= 0)
|
||||
t.Log("size=", lc.(*memoryCache).currentSize)
|
||||
}
|
||||
|
||||
func TestMemoryCache_Parallel(t *testing.T) {
|
||||
var coldCalls int32
|
||||
lc, err := NewMemoryCache()
|
||||
require.Nil(t, err)
|
||||
|
||||
res, err := lc.Get("key", func() ([]byte, error) {
|
||||
return []byte("value"), nil
|
||||
})
|
||||
assert.Nil(t, err)
|
||||
assert.Equal(t, "value", string(res))
|
||||
|
||||
wg := sync.WaitGroup{}
|
||||
for i := 0; i < 1000; i++ {
|
||||
wg.Add(1)
|
||||
i := i
|
||||
go func() {
|
||||
defer wg.Done()
|
||||
res, err := lc.Get("key", func() ([]byte, error) {
|
||||
atomic.AddInt32(&coldCalls, 1)
|
||||
return []byte(fmt.Sprintf("result-%d", i)), nil
|
||||
})
|
||||
require.Nil(t, err)
|
||||
require.Equal(t, "value", string(res))
|
||||
}()
|
||||
}
|
||||
wg.Wait()
|
||||
assert.Equal(t, int32(0), atomic.LoadInt32(&coldCalls))
|
||||
}
|
||||
|
||||
func TestMemoryCache_Scopes(t *testing.T) {
|
||||
lc, err := NewMemoryCache()
|
||||
require.Nil(t, err)
|
||||
|
||||
res, err := lc.Get(Key("key", "s1", "s2"), func() ([]byte, error) {
|
||||
return []byte("value"), nil
|
||||
})
|
||||
assert.Nil(t, err)
|
||||
assert.Equal(t, "value", string(res))
|
||||
|
||||
res, err = lc.Get(Key("key2", "s2"), func() ([]byte, error) {
|
||||
return []byte("value2"), nil
|
||||
})
|
||||
assert.Nil(t, err)
|
||||
assert.Equal(t, "value2", string(res))
|
||||
|
||||
assert.Equal(t, 2, lc.(*memoryCache).bytesCache.Len())
|
||||
lc.Flush("s1")
|
||||
assert.Equal(t, 1, lc.(*memoryCache).bytesCache.Len())
|
||||
|
||||
lc.Get(Key("key2", "s2"), func() ([]byte, error) {
|
||||
assert.Fail(t, "should stay")
|
||||
return nil, nil
|
||||
})
|
||||
|
||||
res, err = lc.Get(Key("key", "s1", "s2"), func() ([]byte, error) {
|
||||
return []byte("value-upd"), nil
|
||||
})
|
||||
assert.Nil(t, err)
|
||||
assert.Equal(t, "value-upd", string(res), "was deleted, update")
|
||||
}
|
||||
|
||||
func TestMemoryCache_Flush(t *testing.T) {
|
||||
lc, err := NewMemoryCache()
|
||||
require.Nil(t, err)
|
||||
|
||||
addToCache := func(key string, scopes ...string) {
|
||||
res, err := lc.Get(key, func() ([]byte, error) {
|
||||
return []byte("value" + key), nil
|
||||
})
|
||||
require.Nil(t, err)
|
||||
require.Equal(t, "value"+key, string(res))
|
||||
}
|
||||
|
||||
init := func() {
|
||||
lc.Flush()
|
||||
addToCache(Key("key1", "s1", "s2"))
|
||||
addToCache(Key("key2", "s1", "s2", "s3"))
|
||||
addToCache(Key("key3", "s1", "s2", "s3"))
|
||||
addToCache(Key("key4", "s2", "s3"))
|
||||
addToCache(Key("key5", "s2"))
|
||||
addToCache(Key("key6"))
|
||||
addToCache(Key("key7", "s4", "s3"))
|
||||
require.Equal(t, 7, lc.(*memoryCache).bytesCache.Len(), "cache init")
|
||||
}
|
||||
|
||||
tbl := []struct {
|
||||
scopes []string
|
||||
left int
|
||||
msg string
|
||||
}{
|
||||
{[]string{}, 0, "full flush, no scopes"},
|
||||
{[]string{"s0"}, 7, "flush wrong scope"},
|
||||
{[]string{"s1"}, 4, "flush s1 scope"},
|
||||
{[]string{"s2", "s1"}, 2, "flush s2+s1 scope"},
|
||||
{[]string{"s1", "s2"}, 2, "flush s1+s2 scope"},
|
||||
{[]string{"s1", "s2", "s4"}, 1, "flush s1+s2+s4 scope"},
|
||||
{[]string{"s1", "s2", "s3"}, 1, "flush s1+s2+s3 scope"},
|
||||
{[]string{"s1", "s2", "ss"}, 2, "flush s1+s2+wrong scope"},
|
||||
}
|
||||
|
||||
for i, tt := range tbl {
|
||||
init()
|
||||
lc.Flush(tt.scopes...)
|
||||
assert.Equal(t, tt.left, lc.(*memoryCache).bytesCache.Len(), "keys size, %s #%d", tt.msg, i)
|
||||
}
|
||||
}
|
||||
|
||||
func TestMemoryCache_FlushFailed(t *testing.T) {
|
||||
lc, err := NewMemoryCache()
|
||||
require.Nil(t, err)
|
||||
val, err := lc.Get("invalid-composite", func() ([]byte, error) {
|
||||
return []byte("value"), nil
|
||||
})
|
||||
assert.Nil(t, err)
|
||||
assert.Equal(t, "value", string(val))
|
||||
assert.Equal(t, 1, lc.(*memoryCache).bytesCache.Len())
|
||||
|
||||
lc.Flush("invalid-composite")
|
||||
assert.Equal(t, 1, lc.(*memoryCache).bytesCache.Len())
|
||||
}
|
||||
Vendored
+5
-5
@@ -1,12 +1,12 @@
|
||||
package cache
|
||||
|
||||
// Option func type
|
||||
type Option func(lc *loadingCache) error
|
||||
type Option func(lc *memoryCache) error
|
||||
|
||||
// MaxValSize functional option defines the largest value's size allowed to be cached
|
||||
// By default it is 0, which means unlimited.
|
||||
func MaxValSize(max int) Option {
|
||||
return func(lc *loadingCache) error {
|
||||
return func(lc *memoryCache) error {
|
||||
lc.maxValueSize = max
|
||||
return nil
|
||||
}
|
||||
@@ -15,7 +15,7 @@ func MaxValSize(max int) Option {
|
||||
// MaxKeys functional option defines how many keys to keep.
|
||||
// By default it is 0, which means unlimited.
|
||||
func MaxKeys(max int) Option {
|
||||
return func(lc *loadingCache) error {
|
||||
return func(lc *memoryCache) error {
|
||||
lc.maxKeys = max
|
||||
return nil
|
||||
}
|
||||
@@ -24,7 +24,7 @@ func MaxKeys(max int) Option {
|
||||
// MaxCacheSize functional option defines the total size of cached data.
|
||||
// By default it is 0, which means unlimited.
|
||||
func MaxCacheSize(max int64) Option {
|
||||
return func(lc *loadingCache) error {
|
||||
return func(lc *memoryCache) error {
|
||||
lc.maxCacheSize = max
|
||||
return nil
|
||||
}
|
||||
@@ -32,7 +32,7 @@ func MaxCacheSize(max int64) Option {
|
||||
|
||||
// PostFlushFn functional option defines how callback function called after each Flush.
|
||||
func PostFlushFn(postFlushFn func()) Option {
|
||||
return func(lc *loadingCache) error {
|
||||
return func(lc *memoryCache) error {
|
||||
lc.postFlushFn = postFlushFn
|
||||
return nil
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user