LBRYCommunity/ccache
1
0
Fork 0
Code Issues Pull requests Packages Projects Releases Wiki Activity

2 changes:

Browse source 
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).
This commit is contained in:
Karl Seguin 2014-12-28 11:11:32 +07:00
parent 557d56ec6f
commit 6df1e24ae3
8 changed files with 80 additions and 136 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

40
bucket.go
View file

@ -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 {

25
bucket_test.go
View file

@ -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 {

54
cache.go
View file

@ -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
}

13
cache_test.go
View file

@ -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))
}

5
item.go
View file

@ -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{} {

20
layeredbucket.go
View file

@ -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 {

48
layeredcache.go
View file

@ -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
}

11
layeredcache_test.go
View file

@ -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))
}