From 6df1e24ae3dec3f6932c4e5e46aa7252ee5f87ca Mon Sep 17 00:00:00 2001 From: Karl Seguin Date: Sun, 28 Dec 2014 11:11:32 +0700 Subject: [PATCH] 2 changes: 1 - Previously, we determined if an item should be promoted in the main getter thread. This required that we protect the item.promotions variable, as both the getter and the worker were concurrently accessing it. This change pushes the conditional promotion to the worker (from the getter's point of view, items are always promoted). Since only the worker ever accesses .promotions, we no longer must protect access to it. 2 - The total size of the cache was being maintained by both the worker thread and the calling code. This required that we protect access to cache.size. Now, only the worker ever changes the size. While this simplifies much of the code, it means that we can't easily replace an item (replacement either via Set or Replace). A replcement now involves creating a new object and deleting the old one (using the existing deletables and promotable infrastructure). The only noticeable impact frmo this change is that, despite previous documentation, Replace WILL cause the item to be promoted (but it still only does so if it exists and it still doesn't extend the original TTL). --- bucket.go | 40 ++++---------------------------- bucket_test.go | 25 ++++---------------- cache.go | 54 ++++++++++++++++++++------------------------ cache_test.go | 13 +++++++++-- item.go | 5 ++-- layeredbucket.go | 20 ++++------------ layeredcache.go | 48 ++++++++++++++++----------------------- layeredcache_test.go | 11 ++++++++- 8 files changed, 80 insertions(+), 136 deletions(-) diff --git a/bucket.go b/bucket.go index d70b4bb..e330d24 100644 --- a/bucket.go +++ b/bucket.go @@ -2,7 +2,6 @@ package ccache import ( "sync" - "sync/atomic" "time" ) @@ -17,45 +16,14 @@ func (b *bucket) get(key string) *Item { return b.lookup[key] } -func (b *bucket) set(key string, value interface{}, duration time.Duration) (*Item, bool, int64) { +func (b *bucket) set(key string, value interface{}, duration time.Duration) (*Item, *Item) { expires := time.Now().Add(duration).Unix() - b.Lock() - defer b.Unlock() - if existing, exists := b.lookup[key]; exists { - existing.value = value - existing.expires = expires - d := int64(0) - if sized, ok := value.(Sized); ok { - newSize := sized.Size() - d = newSize - existing.size - if d != 0 { - atomic.StoreInt64(&existing.size, newSize) - } - } - return existing, false, int64(d) - } item := newItem(key, value, expires) - b.lookup[key] = item - return item, true, int64(item.size) -} - -func (b *bucket) replace(key string, value interface{}) (bool, int64) { b.Lock() defer b.Unlock() - existing, exists := b.lookup[key] - if exists == false { - return false, 0 - } - d := int64(0) - if sized, ok := value.(Sized); ok { - newSize := sized.Size() - d = newSize - existing.size - if d != 0 { - atomic.StoreInt64(&existing.size, newSize) - } - } - existing.value = value - return true, d + existing := b.lookup[key] + b.lookup[key] = item + return item, existing } func (b *bucket) delete(key string) *Item { diff --git a/bucket_test.go b/bucket_test.go index 25717fb..8527400 100644 --- a/bucket_test.go +++ b/bucket_test.go @@ -32,37 +32,20 @@ func (_ *BucketTests) DeleteItemFromBucket() { func (_ *BucketTests) SetsANewBucketItem() { bucket := testBucket() - item, new, d := bucket.set("spice", TestValue("flow"), time.Minute) + item, existing := bucket.set("spice", TestValue("flow"), time.Minute) assertValue(item, "flow") item = bucket.get("spice") assertValue(item, "flow") - Expect(new).To.Equal(true) - Expect(d).To.Equal(1) + Expect(existing).To.Equal(nil) } func (_ *BucketTests) SetsAnExistingItem() { bucket := testBucket() - item, new, d := bucket.set("power", TestValue("9002"), time.Minute) + item, existing := bucket.set("power", TestValue("9001"), time.Minute) assertValue(item, "9002") item = bucket.get("power") assertValue(item, "9002") - Expect(new).To.Equal(false) - Expect(d).To.Equal(0) -} - -func (_ *BucketTests) ReplaceDoesNothingIfKeyDoesNotExist() { - bucket := testBucket() - Expect(bucket.replace("power", TestValue("9002"))).To.Equal(false) - Expect(bucket.get("power")).To.Equal(nil) -} - -func (_ *BucketTests) ReplaceReplacesThevalue() { - bucket := testBucket() - item, _, _ := bucket.set("power", TestValue("9002"), time.Minute) - Expect(bucket.replace("power", TestValue("9004"))).To.Equal(true) - Expect(item.Value().(string)).To.Equal("9004") - Expect(bucket.get("power").Value().(string)).To.Equal("9004") - //not sure how to test that the TTL hasn't changed sort of a sleep.. + assertValue(existing, "9001") } func testBucket() *bucket { diff --git a/cache.go b/cache.go index 75b069b..2ed33bb 100644 --- a/cache.go +++ b/cache.go @@ -43,13 +43,12 @@ func New(config *Configuration) *Cache { // is expired and item.TTL() to see how long until the item expires (which // will be negative for an already expired item). func (c *Cache) Get(key string) *Item { - bucket := c.bucket(key) - item := bucket.get(key) + item := c.bucket(key).get(key) if item == nil { return nil } if item.expires > time.Now().Unix() { - c.conditionalPromote(item) + c.promote(item) } return item } @@ -67,26 +66,23 @@ func (c *Cache) TrackingGet(key string) TrackedItem { // Set the value in the cache for the specified duration func (c *Cache) Set(key string, value interface{}, duration time.Duration) { - item, new, d := c.bucket(key).set(key, value, duration) - if new { - c.promote(item) - } else { - c.conditionalPromote(item) - } - if d != 0 { - atomic.AddInt64(&c.size, d) + item, existing := c.bucket(key).set(key, value, duration) + if existing != nil { + c.deletables <- existing } + c.promote(item) } // Replace the value if it exists, does not set if it doesn't. // Returns true if the item existed an was replaced, false otherwise. -// Replace does not reset item's TTL nor does it alter its position in the LRU +// Replace does not reset item's TTL func (c *Cache) Replace(key string, value interface{}) bool { - exists, d := c.bucket(key).replace(key, value) - if d != 0 { - atomic.AddInt64(&c.size, d) + item := c.bucket(key).get(key) + if item == nil { + return false } - return exists + c.Set(key, value, item.TTL()) + return true } // Attempts to get the value from the cache and calles fetch on a miss. @@ -134,13 +130,6 @@ func (c *Cache) bucket(key string) *bucket { return c.buckets[h.Sum32()&c.bucketMask] } -func (c *Cache) conditionalPromote(item *Item) { - if item.shouldPromote(c.getsPerPromote) == false { - return - } - c.promote(item) -} - func (c *Cache) promote(item *Item) { c.promotables <- item } @@ -149,14 +138,14 @@ func (c *Cache) worker() { for { select { case item := <-c.promotables: - if c.doPromote(item) && atomic.LoadInt64(&c.size) > c.maxSize { + if c.doPromote(item) && c.size > c.maxSize { c.gc() } case item := <-c.deletables: - atomic.AddInt64(&c.size, -item.size) if item.element == nil { - atomic.StoreInt32(&item.promotions, -2) + item.promotions = -2 } else { + c.size -= item.size c.list.Remove(item.element) } } @@ -165,14 +154,18 @@ func (c *Cache) worker() { func (c *Cache) doPromote(item *Item) bool { //already deleted - if atomic.LoadInt32(&item.promotions) == -2 { + if item.promotions == -2 { return false } - atomic.StoreInt32(&item.promotions, 0) if item.element != nil { //not a new item - c.list.MoveToFront(item.element) + if item.shouldPromote(c.getsPerPromote) { + c.list.MoveToFront(item.element) + item.promotions = 0 + } return false } + + c.size += item.size item.element = c.list.PushFront(item) return true } @@ -187,8 +180,9 @@ func (c *Cache) gc() { item := element.Value.(*Item) if c.tracking == false || atomic.LoadInt32(&item.refCount) == 0 { c.bucket(item.key).delete(item.key) - atomic.AddInt64(&c.size, -item.size) + c.size -= item.size c.list.Remove(element) + item.promotions = -2 } element = prev } diff --git a/cache_test.go b/cache_test.go index 1bbc5d3..3e85b8b 100644 --- a/cache_test.go +++ b/cache_test.go @@ -83,22 +83,27 @@ func (_ CacheTests) RemovesOldestItemWhenFullBySizer() { Expect(cache.Get("0")).To.Equal(nil) Expect(cache.Get("1")).To.Equal(nil) Expect(cache.Get("2")).To.Equal(nil) - Expect(cache.Get("3").Value().(*SizedItem).id).To.Equal(3) + Expect(cache.Get("3")).To.Equal(nil) + Expect(cache.Get("4").Value().(*SizedItem).id).To.Equal(4) } func (_ CacheTests) SetUpdatesSizeOnDelta() { cache := New(Configure()) cache.Set("a", &SizedItem{0, 2}, time.Minute) cache.Set("b", &SizedItem{0, 3}, time.Minute) + time.Sleep(time.Millisecond * 5) Expect(cache.size).To.Equal(int64(5)) cache.Set("b", &SizedItem{0, 3}, time.Minute) + time.Sleep(time.Millisecond * 5) Expect(cache.size).To.Equal(int64(5)) cache.Set("b", &SizedItem{0, 4}, time.Minute) + time.Sleep(time.Millisecond * 5) Expect(cache.size).To.Equal(int64(6)) cache.Set("b", &SizedItem{0, 2}, time.Minute) + time.Sleep(time.Millisecond * 5) Expect(cache.size).To.Equal(int64(4)) cache.Delete("b") - time.Sleep(time.Millisecond * 10) + time.Sleep(time.Millisecond * 100) Expect(cache.size).To.Equal(int64(2)) } @@ -108,6 +113,7 @@ func (_ CacheTests) ReplaceDoesNotchangeSizeIfNotSet() { cache.Set("2", &SizedItem{1, 2}, time.Minute) cache.Set("3", &SizedItem{1, 2}, time.Minute) cache.Replace("4", &SizedItem{1, 2}) + time.Sleep(time.Millisecond * 5) Expect(cache.size).To.Equal(int64(6)) } @@ -117,12 +123,15 @@ func (_ CacheTests) ReplaceChangesSize() { cache.Set("2", &SizedItem{1, 2}, time.Minute) cache.Replace("2", &SizedItem{1, 2}) + time.Sleep(time.Millisecond * 5) Expect(cache.size).To.Equal(int64(4)) cache.Replace("2", &SizedItem{1, 1}) + time.Sleep(time.Millisecond * 5) Expect(cache.size).To.Equal(int64(3)) cache.Replace("2", &SizedItem{1, 3}) + time.Sleep(time.Millisecond * 5) Expect(cache.size).To.Equal(int64(5)) } diff --git a/item.go b/item.go index 35e42c6..1c35ae7 100644 --- a/item.go +++ b/item.go @@ -60,14 +60,15 @@ func newItem(key string, value interface{}, expires int64) *Item { return &Item{ key: key, value: value, - promotions: -1, + promotions: 0, size: size, expires: expires, } } func (i *Item) shouldPromote(getsPerPromote int32) bool { - return atomic.AddInt32(&i.promotions, 1) == getsPerPromote + i.promotions += 1 + return i.promotions == getsPerPromote } func (i *Item) Value() interface{} { diff --git a/layeredbucket.go b/layeredbucket.go index b541c76..8cde8c2 100644 --- a/layeredbucket.go +++ b/layeredbucket.go @@ -20,7 +20,7 @@ func (b *layeredBucket) get(primary, secondary string) *Item { return bucket.get(secondary) } -func (b *layeredBucket) set(primary, secondary string, value interface{}, duration time.Duration) (*Item, bool, int64) { +func (b *layeredBucket) set(primary, secondary string, value interface{}, duration time.Duration) (*Item, *Item) { b.Lock() bkt, exists := b.buckets[primary] if exists == false { @@ -28,21 +28,9 @@ func (b *layeredBucket) set(primary, secondary string, value interface{}, durati b.buckets[primary] = bkt } b.Unlock() - item, new, d := bkt.set(secondary, value, duration) - if new { - item.group = primary - } - return item, new, d -} - -func (b *layeredBucket) replace(primary, secondary string, value interface{}) (bool, int64) { - b.Lock() - bucket, exists := b.buckets[primary] - b.Unlock() - if exists == false { - return false, 0 - } - return bucket.replace(secondary, value) + item, existing := bkt.set(secondary, value, duration) + item.group = primary + return item, existing } func (b *layeredBucket) delete(primary, secondary string) *Item { diff --git a/layeredcache.go b/layeredcache.go index 79f6d20..8645db3 100644 --- a/layeredcache.go +++ b/layeredcache.go @@ -54,13 +54,12 @@ func Layered(config *Configuration) *LayeredCache { // is expired and item.TTL() to see how long until the item expires (which // will be negative for an already expired item). func (c *LayeredCache) Get(primary, secondary string) *Item { - bucket := c.bucket(primary) - item := bucket.get(primary, secondary) + item := c.bucket(primary).get(primary, secondary) if item == nil { return nil } if item.expires > time.Now().Unix() { - c.conditionalPromote(item) + c.promote(item) } return item } @@ -78,26 +77,23 @@ func (c *LayeredCache) TrackingGet(primary, secondary string) TrackedItem { // Set the value in the cache for the specified duration func (c *LayeredCache) Set(primary, secondary string, value interface{}, duration time.Duration) { - item, new, d := c.bucket(primary).set(primary, secondary, value, duration) - if new { - c.promote(item) - } else { - c.conditionalPromote(item) - } - if d != 0 { - atomic.AddInt64(&c.size, d) + item, existing := c.bucket(primary).set(primary, secondary, value, duration) + if existing != nil { + c.deletables <- existing } + c.promote(item) } // Replace the value if it exists, does not set if it doesn't. // Returns true if the item existed an was replaced, false otherwise. // Replace does not reset item's TTL nor does it alter its position in the LRU func (c *LayeredCache) Replace(primary, secondary string, value interface{}) bool { - exists, d := c.bucket(primary).replace(primary, secondary, value) - if d != 0 { - atomic.AddInt64(&c.size, d) + item := c.bucket(primary).get(primary, secondary) + if item == nil { + return false } - return exists + c.Set(primary, secondary, value, item.TTL()) + return true } // Attempts to get the value from the cache and calles fetch on a miss. @@ -145,13 +141,6 @@ func (c *LayeredCache) bucket(key string) *layeredBucket { return c.buckets[h.Sum32()&c.bucketMask] } -func (c *LayeredCache) conditionalPromote(item *Item) { - if item.shouldPromote(c.getsPerPromote) == false { - return - } - c.promote(item) -} - func (c *LayeredCache) promote(item *Item) { c.promotables <- item } @@ -160,14 +149,14 @@ func (c *LayeredCache) worker() { for { select { case item := <-c.promotables: - if c.doPromote(item) && atomic.LoadInt64(&c.size) > c.maxSize { + if c.doPromote(item) && c.size > c.maxSize { c.gc() } case item := <-c.deletables: - atomic.AddInt64(&c.size, -item.size) if item.element == nil { item.promotions = -2 } else { + c.size -= item.size c.list.Remove(item.element) } } @@ -179,12 +168,14 @@ func (c *LayeredCache) doPromote(item *Item) bool { if item.promotions == -2 { return false } - - item.promotions = 0 if item.element != nil { //not a new item - c.list.MoveToFront(item.element) + if item.shouldPromote(c.getsPerPromote) { + c.list.MoveToFront(item.element) + item.promotions = 0 + } return false } + c.size += item.size item.element = c.list.PushFront(item) return true } @@ -198,9 +189,10 @@ func (c *LayeredCache) gc() { prev := element.Prev() item := element.Value.(*Item) if c.tracking == false || atomic.LoadInt32(&item.refCount) == 0 { - atomic.AddInt64(&c.size, -item.size) c.bucket(item.group).delete(item.group, item.key) + c.size -= item.size c.list.Remove(element) + item.promotions = -2 } element = prev } diff --git a/layeredcache_test.go b/layeredcache_test.go index 67dfa5c..e233a47 100644 --- a/layeredcache_test.go +++ b/layeredcache_test.go @@ -152,20 +152,25 @@ func (_ LayeredCacheTests) RemovesOldestItemWhenFullBySizer() { Expect(cache.Get("pri", "0")).To.Equal(nil) Expect(cache.Get("pri", "1")).To.Equal(nil) Expect(cache.Get("pri", "2")).To.Equal(nil) - Expect(cache.Get("pri", "3").Value().(*SizedItem).id).To.Equal(3) + Expect(cache.Get("pri", "3")).To.Equal(nil) + Expect(cache.Get("pri", "4").Value().(*SizedItem).id).To.Equal(4) } func (_ LayeredCacheTests) SetUpdatesSizeOnDelta() { cache := Layered(Configure()) cache.Set("pri", "a", &SizedItem{0, 2}, time.Minute) cache.Set("pri", "b", &SizedItem{0, 3}, time.Minute) + time.Sleep(time.Millisecond * 5) Expect(cache.size).To.Equal(int64(5)) cache.Set("pri", "b", &SizedItem{0, 3}, time.Minute) + time.Sleep(time.Millisecond * 5) Expect(cache.size).To.Equal(int64(5)) cache.Set("pri", "b", &SizedItem{0, 4}, time.Minute) + time.Sleep(time.Millisecond * 5) Expect(cache.size).To.Equal(int64(6)) cache.Set("pri", "b", &SizedItem{0, 2}, time.Minute) cache.Set("sec", "b", &SizedItem{0, 3}, time.Minute) + time.Sleep(time.Millisecond * 5) Expect(cache.size).To.Equal(int64(7)) cache.Delete("pri", "b") time.Sleep(time.Millisecond * 10) @@ -178,6 +183,7 @@ func (_ LayeredCacheTests) ReplaceDoesNotchangeSizeIfNotSet() { cache.Set("pri", "2", &SizedItem{1, 2}, time.Minute) cache.Set("pri", "3", &SizedItem{1, 2}, time.Minute) cache.Replace("sec", "3", &SizedItem{1, 2}) + time.Sleep(time.Millisecond * 5) Expect(cache.size).To.Equal(int64(6)) } @@ -187,11 +193,14 @@ func (_ LayeredCacheTests) ReplaceChangesSize() { cache.Set("pri", "2", &SizedItem{1, 2}, time.Minute) cache.Replace("pri", "2", &SizedItem{1, 2}) + time.Sleep(time.Millisecond * 5) Expect(cache.size).To.Equal(int64(4)) cache.Replace("pri", "2", &SizedItem{1, 1}) + time.Sleep(time.Millisecond * 5) Expect(cache.size).To.Equal(int64(3)) cache.Replace("pri", "2", &SizedItem{1, 3}) + time.Sleep(time.Millisecond * 5) Expect(cache.size).To.Equal(int64(5)) }