-
Notifications
You must be signed in to change notification settings - Fork 186
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
feat: expire messages from the cache based on last seen time (#513)
* feat: expire messages from the cache based on last seen time * chore: minor renaming * fix: messages should not be found after expiration * chore: editorial * fix: use new time cache strategy consistently * fix: default to old time cache and add todo for background gc
- Loading branch information
Showing
9 changed files
with
337 additions
and
19 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,71 @@ | ||
| package timecache | ||
|
|
||
| import ( | ||
| "container/list" | ||
| "sync" | ||
| "time" | ||
| ) | ||
|
|
||
| // FirstSeenCache is a thread-safe copy of https://github.com/whyrusleeping/timecache. | ||
| type FirstSeenCache struct { | ||
| q *list.List | ||
| m map[string]time.Time | ||
| span time.Duration | ||
| guard *sync.RWMutex | ||
| } | ||
|
|
||
| func newFirstSeenCache(span time.Duration) TimeCache { | ||
| return &FirstSeenCache{ | ||
| q: list.New(), | ||
| m: make(map[string]time.Time), | ||
| span: span, | ||
| guard: new(sync.RWMutex), | ||
| } | ||
| } | ||
|
|
||
| func (tc FirstSeenCache) Add(s string) { | ||
| tc.guard.Lock() | ||
| defer tc.guard.Unlock() | ||
|
|
||
| _, ok := tc.m[s] | ||
| if ok { | ||
| panic("putting the same entry twice not supported") | ||
| } | ||
|
|
||
| // TODO(#515): Do GC in the background | ||
| tc.sweep() | ||
|
|
||
| tc.m[s] = time.Now() | ||
| tc.q.PushFront(s) | ||
| } | ||
|
|
||
| func (tc FirstSeenCache) sweep() { | ||
| for { | ||
| back := tc.q.Back() | ||
| if back == nil { | ||
| return | ||
| } | ||
|
|
||
| v := back.Value.(string) | ||
| t, ok := tc.m[v] | ||
| if !ok { | ||
| panic("inconsistent cache state") | ||
| } | ||
|
|
||
| if time.Since(t) > tc.span { | ||
| tc.q.Remove(back) | ||
| delete(tc.m, v) | ||
| } else { | ||
| return | ||
| } | ||
| } | ||
| } | ||
|
|
||
| func (tc FirstSeenCache) Has(s string) bool { | ||
| tc.guard.RLock() | ||
| defer tc.guard.RUnlock() | ||
|
|
||
| ts, ok := tc.m[s] | ||
| // Only consider the entry found if it was present in the cache AND hadn't already expired. | ||
| return ok && time.Since(ts) <= tc.span | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,39 @@ | ||
| package timecache | ||
|
|
||
| import ( | ||
| "fmt" | ||
| "testing" | ||
| "time" | ||
| ) | ||
|
|
||
| func TestFirstSeenCacheFound(t *testing.T) { | ||
| tc := newFirstSeenCache(time.Minute) | ||
|
|
||
| tc.Add("test") | ||
|
|
||
| if !tc.Has("test") { | ||
| t.Fatal("should have this key") | ||
| } | ||
| } | ||
|
|
||
| func TestFirstSeenCacheExpire(t *testing.T) { | ||
| tc := newFirstSeenCache(time.Second) | ||
| for i := 0; i < 11; i++ { | ||
| tc.Add(fmt.Sprint(i)) | ||
| time.Sleep(time.Millisecond * 100) | ||
| } | ||
|
|
||
| if tc.Has(fmt.Sprint(0)) { | ||
| t.Fatal("should have dropped this from the cache already") | ||
| } | ||
| } | ||
|
|
||
| func TestFirstSeenCacheNotFoundAfterExpire(t *testing.T) { | ||
| tc := newFirstSeenCache(time.Second) | ||
| tc.Add(fmt.Sprint(0)) | ||
| time.Sleep(1100 * time.Millisecond) | ||
|
|
||
| if tc.Has(fmt.Sprint(0)) { | ||
| t.Fatal("should have dropped this from the cache already") | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,84 @@ | ||
| package timecache | ||
|
|
||
| import ( | ||
| "sync" | ||
| "time" | ||
|
|
||
| "github.com/emirpasic/gods/maps/linkedhashmap" | ||
| ) | ||
|
|
||
| // LastSeenCache is a LRU cache that keeps entries for up to a specified time duration. After this duration has elapsed, | ||
| // "old" entries will be purged from the cache. | ||
| // | ||
| // It's also a "sliding window" cache. Every time an unexpired entry is seen again, its timestamp slides forward. This | ||
| // keeps frequently occurring entries cached and prevents them from being propagated, especially because of network | ||
| // issues that might increase the number of duplicate messages in the network. | ||
| // | ||
| // Garbage collection of expired entries is event-driven, i.e. it only happens when there is a new entry added to the | ||
| // cache. This should be ok - if existing entries are being looked up then the cache is not growing, and when a new one | ||
| // appears that would grow the cache, garbage collection will attempt to reduce the pressure on the cache. | ||
| // | ||
| // This implementation is heavily inspired by https://github.com/whyrusleeping/timecache. | ||
| type LastSeenCache struct { | ||
| m *linkedhashmap.Map | ||
| span time.Duration | ||
| guard *sync.Mutex | ||
| } | ||
|
|
||
| func newLastSeenCache(span time.Duration) TimeCache { | ||
| return &LastSeenCache{ | ||
| m: linkedhashmap.New(), | ||
| span: span, | ||
| guard: new(sync.Mutex), | ||
| } | ||
| } | ||
|
|
||
| func (tc *LastSeenCache) Add(s string) { | ||
| tc.guard.Lock() | ||
| defer tc.guard.Unlock() | ||
|
|
||
| tc.add(s) | ||
|
|
||
| // Garbage collect expired entries | ||
| // TODO(#515): Do GC in the background | ||
| tc.gc() | ||
| } | ||
|
|
||
| func (tc *LastSeenCache) add(s string) { | ||
| // We don't need a lock here because this function is always called with the lock already acquired. | ||
|
|
||
| // If an entry already exists, remove it and add a new one to the back of the list to maintain temporal ordering and | ||
| // an accurate sliding window. | ||
| tc.m.Remove(s) | ||
| now := time.Now() | ||
| tc.m.Put(s, &now) | ||
| } | ||
|
|
||
| func (tc *LastSeenCache) gc() { | ||
| // We don't need a lock here because this function is always called with the lock already acquired. | ||
| iter := tc.m.Iterator() | ||
| for iter.Next() { | ||
| key := iter.Key() | ||
| ts := iter.Value().(*time.Time) | ||
| // Exit if we've found an entry with an unexpired timestamp. Since we're iterating in order of insertion, all | ||
| // entries hereafter will be unexpired. | ||
| if time.Since(*ts) <= tc.span { | ||
| return | ||
| } | ||
| tc.m.Remove(key) | ||
| } | ||
| } | ||
|
|
||
| func (tc *LastSeenCache) Has(s string) bool { | ||
| tc.guard.Lock() | ||
| defer tc.guard.Unlock() | ||
|
|
||
| // If the entry exists and has not already expired, slide it forward. | ||
| if ts, found := tc.m.Get(s); found { | ||
| if t := ts.(*time.Time); time.Since(*t) <= tc.span { | ||
| tc.add(s) | ||
| return true | ||
| } | ||
| } | ||
| return false | ||
| } |
Oops, something went wrong.