From 3000cd0b6c436b4ad53abc08ec4580d2f1e7b202 Mon Sep 17 00:00:00 2001 From: Umputun Date: Mon, 11 Jun 2018 14:18:24 -0500 Subject: [PATCH] separate memory cache --- app/main.go | 2 +- app/rest/cache/cache.go | 116 +----------- app/rest/cache/cache_test.go | 335 +++------------------------------- app/rest/cache/memory.go | 119 ++++++++++++ app/rest/cache/memory_test.go | 299 ++++++++++++++++++++++++++++++ app/rest/cache/options.go | 10 +- 6 files changed, 447 insertions(+), 434 deletions(-) create mode 100644 app/rest/cache/memory.go create mode 100644 app/rest/cache/memory_test.go diff --git a/app/main.go b/app/main.go index 86184ff8..b7d802dc 100644 --- a/app/main.go +++ b/app/main.go @@ -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 diff --git a/app/rest/cache/cache.go b/app/rest/cache/cache.go index 8ac138c1..f734f42f 100644 --- a/app/rest/cache/cache.go +++ b/app/rest/cache/cache.go @@ -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 { diff --git a/app/rest/cache/cache_test.go b/app/rest/cache/cache_test.go index 3f1f443f..22cabca0 100644 --- a/app/rest/cache/cache_test.go +++ b/app/rest/cache/cache_test.go @@ -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) -} diff --git a/app/rest/cache/memory.go b/app/rest/cache/memory.go new file mode 100644 index 00000000..56511ae3 --- /dev/null +++ b/app/rest/cache/memory.go @@ -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 +} diff --git a/app/rest/cache/memory_test.go b/app/rest/cache/memory_test.go new file mode 100644 index 00000000..b4b5f99b --- /dev/null +++ b/app/rest/cache/memory_test.go @@ -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()) +} diff --git a/app/rest/cache/options.go b/app/rest/cache/options.go index e077cf8f..2902b0fa 100644 --- a/app/rest/cache/options.go +++ b/app/rest/cache/options.go @@ -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 }