Skip to content

feat(Ristretto): Introduce SetIfPresent and refactor codebase #282

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
manishrjain merged 6 commits into from
Aug 24, 2021
Merged

feat(Ristretto): Introduce SetIfPresent and refactor codebase #282

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
722884a
Introduce ShouldUpdate and remove interfaces.
manishrjain Aug 23, 2021
0ba5d67
Introduce SetIfPresent and move metrics out to new file
manishrjain Aug 24, 2021
1af0799
Add license
manishrjain Aug 24, 2021
c10871b
fix compilation
manishrjain Aug 24, 2021
3b0c580
Rename structs for clarity
manishrjain Aug 24, 2021
de89f80
Self Review
manishrjain Aug 24, 2021
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
258 changes: 29 additions & 229 deletions cache.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,11 +20,8 @@
package ristretto

import (
"bytes"
"errors"
"fmt"
"sync"
"sync/atomic"
"time"
"unsafe"

Expand All @@ -45,9 +42,9 @@ const itemSize = int64(unsafe.Sizeof(storeItem{}))
// from as many goroutines as you want.
type Cache struct {
// store is the central concurrent hashmap where key-value items are stored.
store store
store *shardedMap
// policy determines what gets let in to the cache and what gets kicked out.
policy policy
policy *lfuPolicy
// getBuf is a custom ring buffer implementation that gets pushed to when
// keys are read.
getBuf *ringBuffer
Expand Down Expand Up @@ -126,6 +123,8 @@ type Config struct {
// Each key will be hashed using the provided function. If keyToHash value
// is not set, the default keyToHash function is used.
KeyToHash func(key interface{}) (uint64, uint64)
// shouldUpdate is called when a value already exists in cache and is being updated.
ShouldUpdate func(prev, cur interface{}) bool
// Cost evaluates a value and outputs a corresponding cost. This function
// is ran after Set is called for a new item or an item update with a cost
// param of 0.
Expand Down Expand Up @@ -168,7 +167,7 @@ func NewCache(config *Config) (*Cache, error) {
}
policy := newPolicy(config.NumCounters, config.MaxCost)
cache := &Cache{
store: newStore(),
store: newShardedMap(config.ShouldUpdate),
policy: policy,
getBuf: newRingBuffer(policy, config.BufferItems),
setBuf: make(chan *Item, setBufSize),
Expand All @@ -195,6 +194,12 @@ func NewCache(config *Config) (*Cache, error) {
}
cache.onExit(item.Value)
}
cache.store.shouldUpdate = func(prev, cur interface{}) bool {
if config.ShouldUpdate != nil {
return config.ShouldUpdate(prev, cur)
}
return true
}
if cache.keyToHash == nil {
cache.keyToHash = z.KeyToHash
}
Expand Down Expand Up @@ -254,6 +259,17 @@ func (c *Cache) Set(key, value interface{}, cost int64) bool {
// expires, which is identical to calling Set. A negative value is a no-op and the value
// is discarded.
func (c *Cache) SetWithTTL(key, value interface{}, cost int64, ttl time.Duration) bool {
return c.setInternal(key, value, cost, ttl, false)
}

// SetIfPresent is like Set, but only updates the value of an existing key. It
// does NOT add the key to cache if it's absent.
func (c *Cache) SetIfPresent(key, value interface{}, cost int64) bool {
return c.setInternal(key, value, cost, 0*time.Second, true)
}

func (c *Cache) setInternal(key, value interface{},
cost int64, ttl time.Duration, onlyUpdate bool) bool {
if c == nil || c.isClosed || key == nil {
return false
}
Expand All @@ -279,11 +295,18 @@ func (c *Cache) SetWithTTL(key, value interface{}, cost int64, ttl time.Duration
Cost: cost,
Expiration: expiration,
}
if onlyUpdate {
i.flag = itemUpdate
}
// cost is eventually updated. The expiration must also be immediately updated
// to prevent items from being prematurely removed from the map.
if prev, ok := c.store.Update(i); ok {
c.onExit(prev)
i.flag = itemUpdate
} else if onlyUpdate {
// The instruction was to update the key, but store.Update failed. So,
// this is a NOOP.
return false
}
// Attempt to send item to policy.
select {
Expand Down Expand Up @@ -494,226 +517,3 @@ func (c *Cache) processItems() {
}
}
}

// collectMetrics just creates a new *Metrics instance and adds the pointers
// to the cache and policy instances.
func (c *Cache) collectMetrics() {
c.Metrics = newMetrics()
c.policy.CollectMetrics(c.Metrics)
}

type metricType int

const (
// The following 2 keep track of hits and misses.
hit = iota
miss
// The following 3 keep track of number of keys added, updated and evicted.
keyAdd
keyUpdate
keyEvict
// The following 2 keep track of cost of keys added and evicted.
costAdd
costEvict
// The following keep track of how many sets were dropped or rejected later.
dropSets
rejectSets
// The following 2 keep track of how many gets were kept and dropped on the
// floor.
dropGets
keepGets
// This should be the final enum. Other enums should be set before this.
doNotUse
)

func stringFor(t metricType) string {
switch t {
case hit:
return "hit"
case miss:
return "miss"
case keyAdd:
return "keys-added"
case keyUpdate:
return "keys-updated"
case keyEvict:
return "keys-evicted"
case costAdd:
return "cost-added"
case costEvict:
return "cost-evicted"
case dropSets:
return "sets-dropped"
case rejectSets:
return "sets-rejected" // by policy.
case dropGets:
return "gets-dropped"
case keepGets:
return "gets-kept"
default:
return "unidentified"
}
}

// Metrics is a snapshot of performance statistics for the lifetime of a cache instance.
type Metrics struct {
all [doNotUse][]*uint64

mu sync.RWMutex
life *z.HistogramData // Tracks the life expectancy of a key.
}

func newMetrics() *Metrics {
s := &Metrics{
life: z.NewHistogramData(z.HistogramBounds(1, 16)),
}
for i := 0; i < doNotUse; i++ {
s.all[i] = make([]*uint64, 256)
slice := s.all[i]
for j := range slice {
slice[j] = new(uint64)
}
}
return s
}

func (p *Metrics) add(t metricType, hash, delta uint64) {
if p == nil {
return
}
valp := p.all[t]
// Avoid false sharing by padding at least 64 bytes of space between two
// atomic counters which would be incremented.
idx := (hash % 25) * 10
atomic.AddUint64(valp[idx], delta)
}

func (p *Metrics) get(t metricType) uint64 {
if p == nil {
return 0
}
valp := p.all[t]
var total uint64
for i := range valp {
total += atomic.LoadUint64(valp[i])
}
return total
}

// Hits is the number of Get calls where a value was found for the corresponding key.
func (p *Metrics) Hits() uint64 {
return p.get(hit)
}

// Misses is the number of Get calls where a value was not found for the corresponding key.
func (p *Metrics) Misses() uint64 {
return p.get(miss)
}

// KeysAdded is the total number of Set calls where a new key-value item was added.
func (p *Metrics) KeysAdded() uint64 {
return p.get(keyAdd)
}

// KeysUpdated is the total number of Set calls where the value was updated.
func (p *Metrics) KeysUpdated() uint64 {
return p.get(keyUpdate)
}

// KeysEvicted is the total number of keys evicted.
func (p *Metrics) KeysEvicted() uint64 {
return p.get(keyEvict)
}

// CostAdded is the sum of costs that have been added (successful Set calls).
func (p *Metrics) CostAdded() uint64 {
return p.get(costAdd)
}

// CostEvicted is the sum of all costs that have been evicted.
func (p *Metrics) CostEvicted() uint64 {
return p.get(costEvict)
}

// SetsDropped is the number of Set calls that don't make it into internal
// buffers (due to contention or some other reason).
func (p *Metrics) SetsDropped() uint64 {
return p.get(dropSets)
}

// SetsRejected is the number of Set calls rejected by the policy (TinyLFU).
func (p *Metrics) SetsRejected() uint64 {
return p.get(rejectSets)
}

// GetsDropped is the number of Get counter increments that are dropped
// internally.
func (p *Metrics) GetsDropped() uint64 {
return p.get(dropGets)
}

// GetsKept is the number of Get counter increments that are kept.
func (p *Metrics) GetsKept() uint64 {
return p.get(keepGets)
}

// Ratio is the number of Hits over all accesses (Hits + Misses). This is the
// percentage of successful Get calls.
func (p *Metrics) Ratio() float64 {
if p == nil {
return 0.0
}
hits, misses := p.get(hit), p.get(miss)
if hits == 0 && misses == 0 {
return 0.0
}
return float64(hits) / float64(hits+misses)
}

func (p *Metrics) trackEviction(numSeconds int64) {
if p == nil {
return
}
p.mu.Lock()
defer p.mu.Unlock()
p.life.Update(numSeconds)
}

func (p *Metrics) LifeExpectancySeconds() *z.HistogramData {
if p == nil {
return nil
}
p.mu.RLock()
defer p.mu.RUnlock()
return p.life.Copy()
}

// Clear resets all the metrics.
func (p *Metrics) Clear() {
if p == nil {
return
}
for i := 0; i < doNotUse; i++ {
for j := range p.all[i] {
atomic.StoreUint64(p.all[i][j], 0)
}
}
p.mu.Lock()
p.life = z.NewHistogramData(z.HistogramBounds(1, 16))
p.mu.Unlock()
}

// String returns a string representation of the metrics.
func (p *Metrics) String() string {
if p == nil {
return ""
}
var buf bytes.Buffer
for i := 0; i < doNotUse; i++ {
t := metricType(i)
fmt.Fprintf(&buf, "%s: %d ", stringFor(t), p.get(t))
}
fmt.Fprintf(&buf, "gets-total: %d ", p.get(hit)+p.get(miss))
fmt.Fprintf(&buf, "hit-ratio: %.2f", p.Ratio())
return buf.String()
}
Loading