mirrors/remark42
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

basic scopes for cache invalidation #44

Browse Source
This commit is contained in:
Umputun
2018-05-25 01:11:48 -05:00
parent 38095ec3b5
commit 83afa4d57c
4 changed files with 132 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/rest/api/rest.go
+12 -11
View File
@@ -212,7 +212,8 @@ func (s *Rest) createCommentCtrl(w http.ResponseWriter, r *http.Request) {
rest.SendErrorJSON(w, r, http.StatusInternalServerError, err, "can't load created comment")
return
}
s.Cache.Flush() // reset all caches
s.Cache.Flush(comment.Locator.URL, "last", comment.User.ID)
render.Status(r, http.StatusCreated)
render.JSON(w, r, &finalComment)
}
@@ -293,7 +294,7 @@ func (s *Rest) updateCommentCtrl(w http.ResponseWriter, r *http.Request) {
return
}
s.Cache.Flush() // reset all caches
s.Cache.Flush(locator.URL, "last", user.ID)
render.JSON(w, r, res)
}
@@ -307,7 +308,7 @@ func (s *Rest) findCommentsCtrl(w http.ResponseWriter, r *http.Request) {
}
log.Printf("[DEBUG] get comments for %+v, sort %s, format %s", locator, sort, r.URL.Query().Get("format"))
data, err := s.Cache.Get(rest.URLKey(r), 4*time.Hour, func() ([]byte, error) {
data, err := s.Cache.Get(rest.CacheKey(rest.URLKey(r), locator.SiteID, locator.URL), 4*time.Hour, func() ([]byte, error) {
comments, e := s.DataService.Find(locator, sort)
if e != nil {
return nil, e
@@ -332,16 +333,16 @@ func (s *Rest) findCommentsCtrl(w http.ResponseWriter, r *http.Request) {
// GET /last/{limit}?site=siteID - last comments for the siteID, across all posts, sorted by time
func (s *Rest) lastCommentsCtrl(w http.ResponseWriter, r *http.Request) {
log.Printf("[DEBUG] get last comments for %s", r.URL.Query().Get("site"))
siteID := r.URL.Query().Get("site")
log.Printf("[DEBUG] get last comments for %s", siteID)
limit, err := strconv.Atoi(chi.URLParam(r, "limit"))
if err != nil {
limit = 0
}
data, err := s.Cache.Get(rest.URLKey(r), 4*time.Hour, func() ([]byte, error) {
comments, e := s.DataService.Last(r.URL.Query().Get("site"), limit)
data, err := s.Cache.Get(rest.CacheKey(rest.URLKey(r), "last", siteID), 4*time.Hour, func() ([]byte, error) {
comments, e := s.DataService.Last(siteID, limit)
if e != nil {
return nil, e
}
@@ -403,7 +404,7 @@ func (s *Rest) findUserCommentsCtrl(w http.ResponseWriter, r *http.Request) {
log.Printf("[DEBUG] get comments for userID %s, %s", userID, siteID)
data, err := s.Cache.Get(rest.URLKey(r), 4*time.Hour, func() ([]byte, error) {
data, err := s.Cache.Get(rest.CacheKey(rest.URLKey(r), userID, siteID), 4*time.Hour, func() ([]byte, error) {
comments, count, e := s.DataService.User(siteID, userID, limit)
if e != nil {
return nil, e
@@ -492,7 +493,7 @@ func (s *Rest) countMultiCtrl(w http.ResponseWriter, r *http.Request) {
}
sha := base64.URLEncoding.EncodeToString(hasher.Sum(nil))
data, err := s.Cache.Get(sha, 8*time.Hour, func() ([]byte, error) {
data, err := s.Cache.Get(rest.CacheKey(sha, siteID), 8*time.Hour, func() ([]byte, error) {
counts, e := s.DataService.Counts(siteID, posts)
if e != nil {
return nil, e
@@ -520,7 +521,7 @@ func (s *Rest) listCtrl(w http.ResponseWriter, r *http.Request) {
skip = v
}
data, err := s.Cache.Get(rest.URLKey(r), 8*time.Hour, func() ([]byte, error) {
data, err := s.Cache.Get(rest.CacheKey(rest.URLKey(r), siteID), 8*time.Hour, func() ([]byte, error) {
posts, e := s.DataService.List(siteID, limit, skip)
if e != nil {
return nil, e
@@ -554,7 +555,7 @@ func (s *Rest) voteCtrl(w http.ResponseWriter, r *http.Request) {
rest.SendErrorJSON(w, r, http.StatusBadRequest, err, "can't vote for comment")
return
}
s.Cache.Flush()
s.Cache.Flush(locator.URL)
render.JSON(w, r, JSON{"id": comment.ID, "score": comment.Score})
}
app/rest/api/rest_test.go
+1 -1
View File
@@ -567,4 +567,4 @@ func (mc *mockCache) Get(key string, ttl time.Duration, fn func() ([]byte, error
return fn()
}
func (mc *mockCache) Flush() {}
func (mc *mockCache) Flush(scopes ...string) {}
app/rest/cache.go
+70 -3
View File
@@ -4,15 +4,33 @@ import (
"log"
"net/http"
"strings"
"sync"
"time"
"github.com/patrickmn/go-cache"
"github.com/pkg/errors"
)
// LoadingCache defines interface for caching
type LoadingCache interface {
Get(key string, ttl time.Duration, fn func() ([]byte, error)) (data []byte, err error)
Flush()
Flush(scopes ...string)
}
func CacheKey(key string, scopes ...string) string {
return strings.Join(scopes, "$$") + "@@" + key
}
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)
}
scopes = strings.Split(elems[0], "$$")
if len(scopes) == 1 && scopes[0] == "" {
scopes = []string{}
}
key = elems[1]
return key, scopes, nil
}
// loadingCache implements LoadingCache interface on top of cache.Cache (go-cache)
@@ -23,6 +41,9 @@ type loadingCache struct {
cleanupInterval time.Duration
maxKeys int
maxValueSize int
activeKeys map[string]struct{}
lock sync.Mutex
}
// NewLoadingCache makes loadingCache implementation
@@ -33,6 +54,7 @@ func NewLoadingCache(options ...CacheOption) LoadingCache {
postFlushFn: func() {},
maxKeys: 0,
maxValueSize: 0,
activeKeys: map[string]struct{}{},
}
for _, opt := range options {
if err := opt(&res); err != nil {
@@ -40,6 +62,11 @@ func NewLoadingCache(options ...CacheOption) LoadingCache {
}
}
res.bytesCache = cache.New(res.defaultExpiration, res.cleanupInterval)
res.bytesCache.OnEvicted(func(key string, _ interface{}) {
res.withLock(func() { delete(res.activeKeys, key) })
})
log.Printf("[DEBUG] create cache with cleanupInterval=%s, maxKeys=%d, maxValueSize=%d",
res.cleanupInterval, res.maxKeys, res.maxValueSize)
@@ -57,13 +84,53 @@ func (lc *loadingCache) Get(key string, ttl time.Duration, fn func() ([]byte, er
}
if lc.allowed(data) {
lc.bytesCache.Set(key, data, ttl)
lc.withLock(func() { lc.activeKeys[key] = struct{}{} })
}
return data, nil
}
func (lc *loadingCache) withLock(fn func()) {
lc.lock.Lock()
fn()
lc.lock.Unlock()
}
// Flush clears cache and calls postFlushFn async
func (lc *loadingCache) Flush() {
lc.bytesCache.Flush()
func (lc *loadingCache) Flush(scopes ...string) {
if len(scopes) == 0 {
lc.bytesCache.Flush()
}
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
}
if len(scopes) > 0 {
matchedKeys := []string{}
lc.withLock(func() {
for k := range lc.activeKeys {
if inScope(k) {
matchedKeys = append(matchedKeys, k)
}
}
})
for _, mkey := range matchedKeys {
lc.bytesCache.Delete(mkey)
}
}
if lc.postFlushFn != nil {
go lc.postFlushFn()
}
app/rest/cache_test.go
+49
View File
@@ -171,3 +171,52 @@ func TestLoadingCache_Parallel(t *testing.T) {
wg.Wait()
assert.Equal(t, int32(0), atomic.LoadInt32(&coldCalls))
}
func TestLoadingCache_Scopes(t *testing.T) {
lc := NewLoadingCache(CleanupInterval(time.Second))
res, err := lc.Get(CacheKey("key", "s1", "s2"), time.Minute, func() ([]byte, error) {
return []byte("value"), nil
})
assert.Nil(t, err)
assert.Equal(t, "value", string(res))
res, err = lc.Get(CacheKey("key2", "s2"), time.Minute, func() ([]byte, error) {
return []byte("value2"), nil
})
assert.Nil(t, err)
assert.Equal(t, "value2", string(res))
lc.Flush("s1")
lc.Get(CacheKey("key2", "s2"), time.Minute, func() ([]byte, error) {
assert.Fail(t, "should stay")
return nil, nil
})
res, err = lc.Get(CacheKey("key", "s1", "s2"), time.Minute, func() ([]byte, error) {
return []byte("value-upd"), nil
})
assert.Equal(t, "value-upd", string(res), "was deleted, update")
}
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 := CacheKey(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)
}
}