4861926555

Committed 02 May 2023 01:45PM UTC coverage: 99.625% (+0.01%) from 99.614%

Build # 4861926555

Build Type

github

Committed by Gökhan Özeloğlu

Commit Message

Add test for `Range` method

Pull Request Pull Request #102: Add `Range` method

Run Details

10 of 10 new or added lines in 1 file covered. (100.0%)

532 of 534 relevant lines covered (99.63%)

17.55 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

99.49

/cache.go

package ttlcache

import (
        "container/list"
        "context"
        "fmt"
        "sync"
        "time"

        "golang.org/x/sync/singleflight"
)

// Available eviction reasons.
const (
        EvictionReasonDeleted EvictionReason = iota + 1
        EvictionReasonCapacityReached
        EvictionReasonExpired
)

// EvictionReason is used to specify why a certain item was
// evicted/deleted.
type EvictionReason int

// Cache is a synchronised map of items that are automatically removed
// when they expire or the capacity is reached.
type Cache[K comparable, V any] struct {
        items struct {
                mu     sync.RWMutex
                values map[K]*list.Element

                // a generic doubly linked list would be more convenient
                // (and more performant?). It's possible that this
                // will be introduced with/in go1.19+
                lru      *list.List
                expQueue expirationQueue[K, V]

                timerCh chan time.Duration
        }

        metricsMu sync.RWMutex
        metrics   Metrics

        events struct {
                insertion struct {
                        mu     sync.RWMutex
                        nextID uint64
                        fns    map[uint64]func(*Item[K, V])
                }
                eviction struct {
                        mu     sync.RWMutex
                        nextID uint64
                        fns    map[uint64]func(EvictionReason, *Item[K, V])
                }
        }

        stopCh  chan struct{}
        options options[K, V]
}

// New creates a new instance of cache.
func New[K comparable, V any](opts ...Option[K, V]) *Cache[K, V] {
        c := &Cache[K, V]{
                stopCh: make(chan struct{}),
        }
        c.items.values = make(map[K]*list.Element)
        c.items.lru = list.New()
        c.items.expQueue = newExpirationQueue[K, V]()
        c.items.timerCh = make(chan time.Duration, 1) // buffer is important
        c.events.insertion.fns = make(map[uint64]func(*Item[K, V]))
        c.events.eviction.fns = make(map[uint64]func(EvictionReason, *Item[K, V]))

        applyOptions(&c.options, opts...)

        return c
}

// updateExpirations updates the expiration queue and notifies
// the cache auto cleaner if needed.
// Not concurrently safe.
func (c *Cache[K, V]) updateExpirations(fresh bool, elem *list.Element) {
        var oldExpiresAt time.Time

        if !c.items.expQueue.isEmpty() {
                oldExpiresAt = c.items.expQueue[0].Value.(*Item[K, V]).expiresAt
        }

        if fresh {
                c.items.expQueue.push(elem)
        } else {
                c.items.expQueue.update(elem)
        }

        newExpiresAt := c.items.expQueue[0].Value.(*Item[K, V]).expiresAt

        // check if the closest/soonest expiration timestamp changed
        if newExpiresAt.IsZero() || (!oldExpiresAt.IsZero() && !newExpiresAt.Before(oldExpiresAt)) {
                return
        }

        d := time.Until(newExpiresAt)

        // It's possible that the auto cleaner isn't active or
        // is busy, so we need to drain the channel before
        // sending a new value.
        // Also, since this method is called after locking the items' mutex,
        // we can be sure that there is no other concurrent call of this
        // method
        if len(c.items.timerCh) > 0 {
                // we need to drain this channel in a select with a default
                // case because it's possible that the auto cleaner
                // read this channel just after we entered this if
                select {
                case d1 := <-c.items.timerCh:
                        if d1 < d {
                                d = d1
                        }
                default:
                }
        }

        // since the channel has a size 1 buffer, we can be sure
        // that the line below won't block (we can't overfill the buffer
        // because we just drained it)
        c.items.timerCh <- d
}

// set creates a new item, adds it to the cache and then returns it.
// Not concurrently safe.
func (c *Cache[K, V]) set(key K, value V, ttl time.Duration) *Item[K, V] {
        if ttl == DefaultTTL {
                ttl = c.options.ttl
        }

        elem := c.get(key, false)
        if elem != nil {
                // update/overwrite an existing item
                item := elem.Value.(*Item[K, V])
                item.update(value, ttl)
                c.updateExpirations(false, elem)

                return item
        }

        if c.options.capacity != 0 && uint64(len(c.items.values)) >= c.options.capacity {
                // delete the oldest item
                c.evict(EvictionReasonCapacityReached, c.items.lru.Back())
        }

        // create a new item
        item := newItem(key, value, ttl)
        elem = c.items.lru.PushFront(item)
        c.items.values[key] = elem
        c.updateExpirations(true, elem)

        c.metricsMu.Lock()
        c.metrics.Insertions++
        c.metricsMu.Unlock()

        c.events.insertion.mu.RLock()
        for _, fn := range c.events.insertion.fns {
                fn(item)
        }
        c.events.insertion.mu.RUnlock()

        return item
}

// get retrieves an item from the cache and extends its expiration
// time if 'touch' is set to true.
// It returns nil if the item is not found or is expired.
// Not concurrently safe.
func (c *Cache[K, V]) get(key K, touch bool) *list.Element {
        elem := c.items.values[key]
        if elem == nil {
                return nil
        }

        item := elem.Value.(*Item[K, V])
        if item.isExpiredUnsafe() {
                return nil
        }

        c.items.lru.MoveToFront(elem)

        if touch && item.ttl > 0 {
                item.touch()
                c.updateExpirations(false, elem)
        }

        return elem
}

// getWithOpts wraps the get method applying the given options.
// Metrics are updated.
// It returns nil if the item is not found or is expired.
func (c *Cache[K, V]) getWithOpts(key K, opts ...Option[K, V]) *Item[K, V] {
        getOpts := options[K, V]{
                loader:            c.options.loader,
                disableTouchOnHit: c.options.disableTouchOnHit,
        }

        applyOptions(&getOpts, opts...)

        c.items.mu.Lock()
        elem := c.get(key, !getOpts.disableTouchOnHit)
        c.items.mu.Unlock()

        if elem == nil {
                c.metricsMu.Lock()
                c.metrics.Misses++
                c.metricsMu.Unlock()

                if getOpts.loader != nil {
                        return getOpts.loader.Load(c, key)
                }

                return nil
        }

        c.metricsMu.Lock()
        c.metrics.Hits++
        c.metricsMu.Unlock()

        return elem.Value.(*Item[K, V])
}

// evict deletes items from the cache.
// If no items are provided, all currently present cache items
// are evicted.
// Not concurrently safe.
func (c *Cache[K, V]) evict(reason EvictionReason, elems ...*list.Element) {
        if len(elems) > 0 {
                c.metricsMu.Lock()
                c.metrics.Evictions += uint64(len(elems))
                c.metricsMu.Unlock()

                c.events.eviction.mu.RLock()
                for i := range elems {
                        item := elems[i].Value.(*Item[K, V])
                        delete(c.items.values, item.key)
                        c.items.lru.Remove(elems[i])
                        c.items.expQueue.remove(elems[i])

                        for _, fn := range c.events.eviction.fns {
                                fn(reason, item)
                        }
                }
                c.events.eviction.mu.RUnlock()

                return
        }

        c.metricsMu.Lock()
        c.metrics.Evictions += uint64(len(c.items.values))
        c.metricsMu.Unlock()

        c.events.eviction.mu.RLock()
        for _, elem := range c.items.values {
                item := elem.Value.(*Item[K, V])

                for _, fn := range c.events.eviction.fns {
                        fn(reason, item)
                }
        }
        c.events.eviction.mu.RUnlock()

        c.items.values = make(map[K]*list.Element)
        c.items.lru.Init()
        c.items.expQueue = newExpirationQueue[K, V]()
}

// Set creates a new item from the provided key and value, adds
// it to the cache and then returns it. If an item associated with the
// provided key already exists, the new item overwrites the existing one.
func (c *Cache[K, V]) Set(key K, value V, ttl time.Duration) *Item[K, V] {
        c.items.mu.Lock()
        defer c.items.mu.Unlock()

        return c.set(key, value, ttl)
}

// Get retrieves an item from the cache by the provided key.
// Unless this is disabled, it also extends/touches an item's
// expiration timestamp on successful retrieval.
// If the item is not found, a nil value is returned.
func (c *Cache[K, V]) Get(key K, opts ...Option[K, V]) *Item[K, V] {
        return c.getWithOpts(key, opts...)
}

// Delete deletes an item from the cache. If the item associated with
// the key is not found, the method is no-op.
func (c *Cache[K, V]) Delete(key K) {
        c.items.mu.Lock()
        defer c.items.mu.Unlock()

        elem := c.items.values[key]
        if elem == nil {
                return
        }

        c.evict(EvictionReasonDeleted, elem)
}

// GetOrSet returns the existing value for the key if present.
// Otherwise, it sets and returns the given value. The retrieved
// result is true if the value was retrieved, false if set.
func (c *Cache[K, V]) GetOrSet(key K, value V, opts ...Option[K, V]) (*Item[K, V], bool) {
        elem := c.getWithOpts(key, opts...)
        if elem != nil {
                return elem, true
        }

        setOpts := options[K, V]{
                ttl: c.options.ttl,
        }

        applyOptions(&setOpts, opts...)

        return c.Set(key, value, setOpts.ttl), false
}

// GetAndDelete deletes the value for a key, returning the previous
// value if any. The retrieved result reports whether the key was present.
func (c *Cache[K, V]) GetAndDelete(key K, opts ...Option[K, V]) (*Item[K, V], bool) {
        elem := c.getWithOpts(key, opts...)
        if elem == nil {
                return nil, false
        }

        c.Delete(key)

        return elem, true
}

// DeleteAll deletes all items from the cache.
func (c *Cache[K, V]) DeleteAll() {
        c.items.mu.Lock()
        c.evict(EvictionReasonDeleted)
        c.items.mu.Unlock()
}

// DeleteExpired deletes all expired items from the cache.
func (c *Cache[K, V]) DeleteExpired() {
        c.items.mu.Lock()
        defer c.items.mu.Unlock()

        if c.items.expQueue.isEmpty() {
                return
        }

        e := c.items.expQueue[0]
        for e.Value.(*Item[K, V]).isExpiredUnsafe() {
                c.evict(EvictionReasonExpired, e)

                if c.items.expQueue.isEmpty() {
                        break
                }

                // expiration queue has a new root
                e = c.items.expQueue[0]
        }
}

// Touch simulates an item's retrieval without actually returning it.
// Its main purpose is to extend an item's expiration timestamp.
// If the item is not found, the method is no-op.
func (c *Cache[K, V]) Touch(key K) {
        c.items.mu.Lock()
        c.get(key, true)
        c.items.mu.Unlock()
}

// Len returns the number of items in the cache.
func (c *Cache[K, V]) Len() int {
        c.items.mu.RLock()
        defer c.items.mu.RUnlock()

        return len(c.items.values)
}

// Keys returns all keys currently present in the cache.
func (c *Cache[K, V]) Keys() []K {
        c.items.mu.RLock()
        defer c.items.mu.RUnlock()

        res := make([]K, 0, len(c.items.values))
        for k := range c.items.values {
                res = append(res, k)
        }

        return res
}

// Items returns a copy of all items in the cache.
// It does not update any expiration timestamps.
func (c *Cache[K, V]) Items() map[K]*Item[K, V] {
        c.items.mu.RLock()
        defer c.items.mu.RUnlock()

        items := make(map[K]*Item[K, V], len(c.items.values))
        for k := range c.items.values {
                item := c.get(k, false)
                if item != nil {
                        items[k] = item.Value.(*Item[K, V])
                }
        }

        return items
}

// Metrics returns the metrics of the cache.
func (c *Cache[K, V]) Metrics() Metrics {
        c.metricsMu.RLock()
        defer c.metricsMu.RUnlock()

        return c.metrics
}

// Start starts an automatic cleanup process that
// periodically deletes expired items.
// It blocks until Stop is called.
func (c *Cache[K, V]) Start() {
        waitDur := func() time.Duration {
                c.items.mu.RLock()
                defer c.items.mu.RUnlock()

                if !c.items.expQueue.isEmpty() &&
                        !c.items.expQueue[0].Value.(*Item[K, V]).expiresAt.IsZero() {
                        d := time.Until(c.items.expQueue[0].Value.(*Item[K, V]).expiresAt)
                        if d <= 0 {
                                // execute immediately
                                return time.Microsecond
                        }

                        return d
                }

                if c.options.ttl > 0 {
                        return c.options.ttl
                }

                return time.Hour
        }

        timer := time.NewTimer(waitDur())
        stop := func() {
                if !timer.Stop() {
                        // drain the timer chan
                        select {
                        case <-timer.C:
                        default:
                        }
                }
        }

        defer stop()

        for {
                select {
                case <-c.stopCh:
                        return
                case d := <-c.items.timerCh:
                        stop()
                        timer.Reset(d)
                case <-timer.C:
                        c.DeleteExpired()
                        stop()
                        timer.Reset(waitDur())
                }
        }
}

// Stop stops the automatic cleanup process.
// It blocks until the cleanup process exits.
func (c *Cache[K, V]) Stop() {
        c.stopCh <- struct{}{}
}

// OnInsertion adds the provided function to be executed when
// a new item is inserted into the cache. The function is executed
// on a separate goroutine and does not block the flow of the cache
// manager.
// The returned function may be called to delete the subscription function
// from the list of insertion subscribers.
// When the returned function is called, it blocks until all instances of
// the same subscription function return. A context is used to notify the
// subscription function when the returned/deletion function is called.
func (c *Cache[K, V]) OnInsertion(fn func(context.Context, *Item[K, V])) func() {
        var (
                wg          sync.WaitGroup
                ctx, cancel = context.WithCancel(context.Background())
        )

        c.events.insertion.mu.Lock()
        id := c.events.insertion.nextID
        c.events.insertion.fns[id] = func(item *Item[K, V]) {
                wg.Add(1)
                go func() {
                        fn(ctx, item)
                        wg.Done()
                }()
        }
        c.events.insertion.nextID++
        c.events.insertion.mu.Unlock()

        return func() {
                cancel()

                c.events.insertion.mu.Lock()
                delete(c.events.insertion.fns, id)
                c.events.insertion.mu.Unlock()

                wg.Wait()
        }
}

// OnEviction adds the provided function to be executed when
// an item is evicted/deleted from the cache. The function is executed
// on a separate goroutine and does not block the flow of the cache
// manager.
// The returned function may be called to delete the subscription function
// from the list of eviction subscribers.
// When the returned function is called, it blocks until all instances of
// the same subscription function return. A context is used to notify the
// subscription function when the returned/deletion function is called.
func (c *Cache[K, V]) OnEviction(fn func(context.Context, EvictionReason, *Item[K, V])) func() {
        var (
                wg          sync.WaitGroup
                ctx, cancel = context.WithCancel(context.Background())
        )

        c.events.eviction.mu.Lock()
        id := c.events.eviction.nextID
        c.events.eviction.fns[id] = func(r EvictionReason, item *Item[K, V]) {
                wg.Add(1)
                go func() {
                        fn(ctx, r, item)
                        wg.Done()
                }()
        }
        c.events.eviction.nextID++
        c.events.eviction.mu.Unlock()

        return func() {
                cancel()

                c.events.eviction.mu.Lock()
                delete(c.events.eviction.fns, id)
                c.events.eviction.mu.Unlock()

                wg.Wait()
        }
}

// Range iterate over all items and calls fn function. It calls fn function
// until it returns false.
func (c *Cache[K, V]) Range(fn func(item *Item[K, V]) bool) {
        c.items.mu.Lock()
        defer c.items.mu.Unlock()

        for k := range c.items.values {
                item := c.get(k, false).Value.(*Item[K, V])
                if !fn(item) {
                        return
                }
        }
}

// Loader is an interface that handles missing data loading.
type Loader[K comparable, V any] interface {
        // Load should execute a custom item retrieval logic and
        // return the item that is associated with the key.
        // It should return nil if the item is not found/valid.
        // The method is allowed to fetch data from the cache instance
        // or update it for future use.
        Load(c *Cache[K, V], key K) *Item[K, V]
}

// LoaderFunc type is an adapter that allows the use of ordinary
// functions as data loaders.
type LoaderFunc[K comparable, V any] func(*Cache[K, V], K) *Item[K, V]

// Load executes a custom item retrieval logic and returns the item that
// is associated with the key.
// It returns nil if the item is not found/valid.
func (l LoaderFunc[K, V]) Load(c *Cache[K, V], key K) *Item[K, V] {
        return l(c, key)
}

// SuppressedLoader wraps another Loader and suppresses duplicate
// calls to its Load method.
type SuppressedLoader[K comparable, V any] struct {
        loader Loader[K, V]
        group  *singleflight.Group
}

// NewSuppressedLoader creates a new instance of suppressed loader.
// If the group parameter is nil, a newly created instance of
// *singleflight.Group is used.
func NewSuppressedLoader[K comparable, V any](loader Loader[K, V], group *singleflight.Group) *SuppressedLoader[K, V] {
        if group == nil {
                group = &singleflight.Group{}
        }

        return &SuppressedLoader[K, V]{
                loader: loader,
                group:  group,
        }
}

// Load executes a custom item retrieval logic and returns the item that
// is associated with the key.
// It returns nil if the item is not found/valid.
// It also ensures that only one execution of the wrapped Loader's Load
// method is in-flight for a given key at a time.
func (l *SuppressedLoader[K, V]) Load(c *Cache[K, V], key K) *Item[K, V] {
        // there should be a better/generic way to create a
        // singleflight Group's key. It's possible that a generic
        // singleflight.Group will be introduced with/in go1.19+
        strKey := fmt.Sprint(key)

        // the error can be discarded since the singleflight.Group
        // itself does not return any of its errors, it returns
        // the error that we return ourselves in the func below, which
        // is also nil
        res, _, _ := l.group.Do(strKey, func() (interface{}, error) {
                item := l.loader.Load(c, key)
                if item == nil {
                        return nil, nil
                }

                return item, nil
        })
        if res == nil {
                return nil
        }

        return res.(*Item[K, V])
}

1	package ttlcache
2
3	import (
4	"container/list"
5	"context"
6	"fmt"
7	"sync"
8	"time"
9
10	"golang.org/x/sync/singleflight"
11	)
12
13	// Available eviction reasons.
14	const (
15	EvictionReasonDeleted EvictionReason = iota + 1
16	EvictionReasonCapacityReached
17	EvictionReasonExpired
18	)
19
20	// EvictionReason is used to specify why a certain item was
21	// evicted/deleted.
22	type EvictionReason int
23
24	// Cache is a synchronised map of items that are automatically removed
25	// when they expire or the capacity is reached.
26	type Cache[K comparable, V any] struct {
27	items struct {
28	mu sync.RWMutex
29	values map[K]*list.Element
30
31	// a generic doubly linked list would be more convenient
32	// (and more performant?). It's possible that this
33	// will be introduced with/in go1.19+
34	lru *list.List
35	expQueue expirationQueue[K, V]
36
37	timerCh chan time.Duration
38	}
39
40	metricsMu sync.RWMutex
41	metrics Metrics
42
43	events struct {
44	insertion struct {
45	mu sync.RWMutex
46	nextID uint64
47	fns map[uint64]func(*Item[K, V])
48	}
49	eviction struct {
50	mu sync.RWMutex
51	nextID uint64
52	fns map[uint64]func(EvictionReason, *Item[K, V])
53	}
54	}
55
56	stopCh chan struct{}
57	options options[K, V]
58	}
59
60	// New creates a new instance of cache.
61	func New[K comparable, V any](opts ...Option[K, V]) *Cache[K, V] {	1✔
62	c := &Cache[K, V]{	1✔
63	stopCh: make(chan struct{}),	1✔
64	}	1✔
65	c.items.values = make(map[K]*list.Element)	1✔
66	c.items.lru = list.New()	1✔
67	c.items.expQueue = newExpirationQueue[K, V]()	1✔
68	c.items.timerCh = make(chan time.Duration, 1) // buffer is important	1✔
69	c.events.insertion.fns = make(map[uint64]func(*Item[K, V]))	1✔
70	c.events.eviction.fns = make(map[uint64]func(EvictionReason, *Item[K, V]))	1✔
71		1✔
72	applyOptions(&c.options, opts...)	1✔
73		1✔
74	return c	1✔
75	}	1✔
76
77	// updateExpirations updates the expiration queue and notifies
78	// the cache auto cleaner if needed.
79	// Not concurrently safe.
80	func (c Cache[K, V]) updateExpirations(fresh bool, elem list.Element) {	31✔
81	var oldExpiresAt time.Time	31✔
82		31✔
83	if !c.items.expQueue.isEmpty() {	60✔
84	oldExpiresAt = c.items.expQueue[0].Value.(*Item[K, V]).expiresAt	29✔
85	}	29✔
86
87	if fresh {	47✔
88	c.items.expQueue.push(elem)	16✔
89	} else {	31✔
90	c.items.expQueue.update(elem)	15✔
91	}	15✔
92
93	newExpiresAt := c.items.expQueue[0].Value.(*Item[K, V]).expiresAt	31✔
94		31✔
95	// check if the closest/soonest expiration timestamp changed	31✔
96	if newExpiresAt.IsZero() \|\| (!oldExpiresAt.IsZero() && !newExpiresAt.Before(oldExpiresAt)) {	44✔
97	return	13✔
98	}	13✔
99
100	d := time.Until(newExpiresAt)	18✔
101		18✔
102	// It's possible that the auto cleaner isn't active or	18✔
103	// is busy, so we need to drain the channel before	18✔
104	// sending a new value.	18✔
105	// Also, since this method is called after locking the items' mutex,	18✔
106	// we can be sure that there is no other concurrent call of this	18✔
107	// method	18✔
108	if len(c.items.timerCh) > 0 {	23✔
109	// we need to drain this channel in a select with a default	5✔
110	// case because it's possible that the auto cleaner	5✔
111	// read this channel just after we entered this if	5✔
112	select {	5✔
113	case d1 := <-c.items.timerCh:	5✔
114	if d1 < d {	7✔
115	d = d1	2✔
116	}	2✔
117	default:	×
118	}
119	}
120
121	// since the channel has a size 1 buffer, we can be sure
122	// that the line below won't block (we can't overfill the buffer
123	// because we just drained it)
124	c.items.timerCh <- d	18✔
125	}
126
127	// set creates a new item, adds it to the cache and then returns it.
128	// Not concurrently safe.
129	func (c Cache[K, V]) set(key K, value V, ttl time.Duration) Item[K, V] {	13✔
130	if ttl == DefaultTTL {	17✔
131	ttl = c.options.ttl	4✔
132	}	4✔
133
134	elem := c.get(key, false)	13✔
135	if elem != nil {	17✔
136	// update/overwrite an existing item	4✔
137	item := elem.Value.(*Item[K, V])	4✔
138	item.update(value, ttl)	4✔
139	c.updateExpirations(false, elem)	4✔
140		4✔
141	return item	4✔
142	}	4✔
143
144	if c.options.capacity != 0 && uint64(len(c.items.values)) >= c.options.capacity {	10✔
145	// delete the oldest item	1✔
146	c.evict(EvictionReasonCapacityReached, c.items.lru.Back())	1✔
147	}	1✔
148
149	// create a new item
150	item := newItem(key, value, ttl)	9✔
151	elem = c.items.lru.PushFront(item)	9✔
152	c.items.values[key] = elem	9✔
153	c.updateExpirations(true, elem)	9✔
154		9✔
155	c.metricsMu.Lock()	9✔
156	c.metrics.Insertions++	9✔
157	c.metricsMu.Unlock()	9✔
158		9✔
159	c.events.insertion.mu.RLock()	9✔
160	for _, fn := range c.events.insertion.fns {	19✔
161	fn(item)	10✔
162	}	10✔
163	c.events.insertion.mu.RUnlock()	9✔
164		9✔
165	return item	9✔
166	}
167
168	// get retrieves an item from the cache and extends its expiration
169	// time if 'touch' is set to true.
170	// It returns nil if the item is not found or is expired.
171	// Not concurrently safe.
172	func (c Cache[K, V]) get(key K, touch bool) list.Element {	39✔
173	elem := c.items.values[key]	39✔
174	if elem == nil {	56✔
175	return nil	17✔
176	}	17✔
177
178	item := elem.Value.(*Item[K, V])	22✔
179	if item.isExpiredUnsafe() {	25✔
180	return nil	3✔
181	}	3✔
182
183	c.items.lru.MoveToFront(elem)	19✔
184		19✔
185	if touch && item.ttl > 0 {	24✔
186	item.touch()	5✔
187	c.updateExpirations(false, elem)	5✔
188	}	5✔
189
190	return elem	19✔
191	}
192
193	// getWithOpts wraps the get method applying the given options.
194	// Metrics are updated.
195	// It returns nil if the item is not found or is expired.
196	func (c Cache[K, V]) getWithOpts(key K, opts ...Option[K, V]) Item[K, V] {	14✔
197	getOpts := options[K, V]{	14✔
198	loader: c.options.loader,	14✔
199	disableTouchOnHit: c.options.disableTouchOnHit,	14✔
200	}	14✔
201		14✔
202	applyOptions(&getOpts, opts...)	14✔
203		14✔
204	c.items.mu.Lock()	14✔
205	elem := c.get(key, !getOpts.disableTouchOnHit)	14✔
206	c.items.mu.Unlock()	14✔
207		14✔
208	if elem == nil {	23✔
209	c.metricsMu.Lock()	9✔
210	c.metrics.Misses++	9✔
211	c.metricsMu.Unlock()	9✔
212		9✔
213	if getOpts.loader != nil {	13✔
214	return getOpts.loader.Load(c, key)	4✔
215	}	4✔
216
217	return nil	5✔
218	}
219
220	c.metricsMu.Lock()	5✔
221	c.metrics.Hits++	5✔
222	c.metricsMu.Unlock()	5✔
223		5✔
224	return elem.Value.(*Item[K, V])	5✔
225	}
226
227	// evict deletes items from the cache.
228	// If no items are provided, all currently present cache items
229	// are evicted.
230	// Not concurrently safe.
231	func (c Cache[K, V]) evict(reason EvictionReason, elems ...list.Element) {	12✔
232	if len(elems) > 0 {	22✔
233	c.metricsMu.Lock()	10✔
234	c.metrics.Evictions += uint64(len(elems))	10✔
235	c.metricsMu.Unlock()	10✔
236		10✔
237	c.events.eviction.mu.RLock()	10✔
238	for i := range elems {	21✔
239	item := elems[i].Value.(*Item[K, V])	11✔
240	delete(c.items.values, item.key)	11✔
241	c.items.lru.Remove(elems[i])	11✔
242	c.items.expQueue.remove(elems[i])	11✔
243		11✔
244	for _, fn := range c.events.eviction.fns {	28✔
245	fn(reason, item)	17✔
246	}	17✔
247	}
248	c.events.eviction.mu.RUnlock()	10✔
249		10✔
250	return	10✔
251	}
252
253	c.metricsMu.Lock()	2✔
254	c.metrics.Evictions += uint64(len(c.items.values))	2✔
255	c.metricsMu.Unlock()	2✔
256		2✔
257	c.events.eviction.mu.RLock()	2✔
258	for _, elem := range c.items.values {	8✔
259	item := elem.Value.(*Item[K, V])	6✔
260		6✔
261	for _, fn := range c.events.eviction.fns {	18✔
262	fn(reason, item)	12✔
263	}	12✔
264	}
265	c.events.eviction.mu.RUnlock()	2✔
266		2✔
267	c.items.values = make(map[K]*list.Element)	2✔
268	c.items.lru.Init()	2✔
269	c.items.expQueue = newExpirationQueue[K, V]()	2✔
270	}
271
272	// Set creates a new item from the provided key and value, adds
273	// it to the cache and then returns it. If an item associated with the
274	// provided key already exists, the new item overwrites the existing one.
275	func (c Cache[K, V]) Set(key K, value V, ttl time.Duration) Item[K, V] {	5✔
276	c.items.mu.Lock()	5✔
277	defer c.items.mu.Unlock()	5✔
278		5✔
279	return c.set(key, value, ttl)	5✔
280	}	5✔
281
282	// Get retrieves an item from the cache by the provided key.
283	// Unless this is disabled, it also extends/touches an item's
284	// expiration timestamp on successful retrieval.
285	// If the item is not found, a nil value is returned.
286	func (c Cache[K, V]) Get(key K, opts ...Option[K, V]) Item[K, V] {	8✔
287	return c.getWithOpts(key, opts...)	8✔
288	}	8✔
289
290	// Delete deletes an item from the cache. If the item associated with
291	// the key is not found, the method is no-op.
292	func (c *Cache[K, V]) Delete(key K) {	3✔
293	c.items.mu.Lock()	3✔
294	defer c.items.mu.Unlock()	3✔
295		3✔
296	elem := c.items.values[key]	3✔
297	if elem == nil {	4✔
298	return	1✔
299	}	1✔
300
301	c.evict(EvictionReasonDeleted, elem)	2✔
302	}
303
304	// GetOrSet returns the existing value for the key if present.
305	// Otherwise, it sets and returns the given value. The retrieved	3✔
306	// result is true if the value was retrieved, false if set.	3✔
307	func (c Cache[K, V]) GetOrSet(key K, value V, opts ...Option[K, V]) (Item[K, V], bool) {	3✔
308	elem := c.getWithOpts(key, opts...)	3✔
309	if elem != nil {	3✔
310	return elem, true	3✔
311	}	3✔
312
313	setOpts := options[K, V]{
314	ttl: c.options.ttl,
315	}
316		4✔
317	applyOptions(&setOpts, opts...)	4✔
318		5✔
319	return c.Set(key, value, setOpts.ttl), false	1✔
320	}	1✔
321
322	// GetAndDelete deletes the value for a key, returning the previous	3✔
323	// value if any. The retrieved result reports whether the key was present.	3✔
324	func (c Cache[K, V]) GetAndDelete(key K, opts ...Option[K, V]) (Item[K, V], bool) {	3✔
325	elem := c.getWithOpts(key, opts...)	3✔
326	if elem == nil {	3✔
327	return nil, false	3✔
328	}	3✔
329
330	c.Delete(key)
331
332	return elem, true
333	}	2✔
334		2✔
335	// DeleteAll deletes all items from the cache.	3✔
336	func (c *Cache[K, V]) DeleteAll() {	1✔
337	c.items.mu.Lock()	1✔
338	c.evict(EvictionReasonDeleted)
339	c.items.mu.Unlock()	1✔
340	}	1✔
341		1✔
342	// DeleteExpired deletes all expired items from the cache.
343	func (c *Cache[K, V]) DeleteExpired() {
344	c.items.mu.Lock()
345	defer c.items.mu.Unlock()	1✔
346		1✔
347	if c.items.expQueue.isEmpty() {	1✔
348	return	1✔
349	}	1✔
350
351	e := c.items.expQueue[0]
352	for e.Value.(*Item[K, V]).isExpiredUnsafe() {	7✔
353	c.evict(EvictionReasonExpired, e)	7✔
354		7✔
355	if c.items.expQueue.isEmpty() {	7✔
356	break	8✔
357	}	1✔
358		1✔
359	// expiration queue has a new root
360	e = c.items.expQueue[0]	6✔
361	}	12✔
362	}	6✔
363		6✔
364	// Touch simulates an item's retrieval without actually returning it.	9✔
365	// Its main purpose is to extend an item's expiration timestamp.	3✔
366	// If the item is not found, the method is no-op.
367	func (c *Cache[K, V]) Touch(key K) {
368	c.items.mu.Lock()
369	c.get(key, true)	3✔
370	c.items.mu.Unlock()
371	}
372
373	// Len returns the number of items in the cache.
374	func (c *Cache[K, V]) Len() int {
375	c.items.mu.RLock()
376	defer c.items.mu.RUnlock()	1✔
377		1✔
378	return len(c.items.values)	1✔
379	}	1✔
380		1✔
381	// Keys returns all keys currently present in the cache.
382	func (c *Cache[K, V]) Keys() []K {
383	c.items.mu.RLock()	1✔
384	defer c.items.mu.RUnlock()	1✔
385		1✔
386	res := make([]K, 0, len(c.items.values))	1✔
387	for k := range c.items.values {	1✔
388	res = append(res, k)	1✔
389	}
390
391	return res	1✔
392	}	1✔
393		1✔
394	// Items returns a copy of all items in the cache.	1✔
395	// It does not update any expiration timestamps.	1✔
396	func (c Cache[K, V]) Items() map[K]Item[K, V] {	4✔
397	c.items.mu.RLock()	3✔
398	defer c.items.mu.RUnlock()	3✔
399
400	items := make(map[K]*Item[K, V], len(c.items.values))	1✔
401	for k := range c.items.values {
402	item := c.get(k, false)
403	if item != nil {
404	items[k] = item.Value.(*Item[K, V])
405	}	1✔
406	}	1✔
407		1✔
408	return items	1✔
409	}	1✔
410		4✔
411	// Metrics returns the metrics of the cache.	3✔
412	func (c *Cache[K, V]) Metrics() Metrics {	6✔
413	c.metricsMu.RLock()	3✔
414	defer c.metricsMu.RUnlock()	3✔
415
416	return c.metrics
417	}	1✔
418
419	// Start starts an automatic cleanup process that
420	// periodically deletes expired items.
421	// It blocks until Stop is called.	1✔
422	func (c *Cache[K, V]) Start() {	1✔
423	waitDur := func() time.Duration {	1✔
424	c.items.mu.RLock()	1✔
425	defer c.items.mu.RUnlock()	1✔
426		1✔
427	if !c.items.expQueue.isEmpty() &&
428	!c.items.expQueue[0].Value.(*Item[K, V]).expiresAt.IsZero() {
429	d := time.Until(c.items.expQueue[0].Value.(*Item[K, V]).expiresAt)
430	if d <= 0 {
431	// execute immediately	1✔
432	return time.Microsecond	6✔
433	}	5✔
434		5✔
435	return d	5✔
436	}	5✔
437		7✔
438	if c.options.ttl > 0 {	2✔
439	return c.options.ttl	3✔
440	}	1✔
441		1✔
442	return time.Hour	1✔
443	}
444		1✔
445	timer := time.NewTimer(waitDur())
446	stop := func() {
447	if !timer.Stop() {	5✔
448	// drain the timer chan	2✔
449	select {	2✔
450	case <-timer.C:
451	default:	1✔
452	}
453	}
454	}	1✔
455		8✔
456	defer stop()	11✔
457		4✔
458	for {	4✔
459	select {	×
460	case <-c.stopCh:	4✔
461	return
462	case d := <-c.items.timerCh:
463	stop()
464	timer.Reset(d)
465	case <-timer.C:	1✔
466	c.DeleteExpired()	1✔
467	stop()	8✔
468	timer.Reset(waitDur())	7✔
469	}	1✔
470	}	1✔
471	}	2✔
472		2✔
473	// Stop stops the automatic cleanup process.	2✔
474	// It blocks until the cleanup process exits.	4✔
475	func (c *Cache[K, V]) Stop() {	4✔
476	c.stopCh <- struct{}{}	4✔
477	}	4✔
478
479	// OnInsertion adds the provided function to be executed when
480	// a new item is inserted into the cache. The function is executed
481	// on a separate goroutine and does not block the flow of the cache
482	// manager.
483	// The returned function may be called to delete the subscription function
484	// from the list of insertion subscribers.	1✔
485	// When the returned function is called, it blocks until all instances of	1✔
486	// the same subscription function return. A context is used to notify the	1✔
487	// subscription function when the returned/deletion function is called.
488	func (c Cache[K, V]) OnInsertion(fn func(context.Context, Item[K, V])) func() {
489	var (
490	wg sync.WaitGroup
491	ctx, cancel = context.WithCancel(context.Background())
492	)
493
494	c.events.insertion.mu.Lock()
495	id := c.events.insertion.nextID
496	c.events.insertion.fns[id] = func(item *Item[K, V]) {
497	wg.Add(1)	2✔
498	go func() {	2✔
499	fn(ctx, item)	2✔
500	wg.Done()	2✔
501	}()	2✔
502	}	2✔
503	c.events.insertion.nextID++	2✔
504	c.events.insertion.mu.Unlock()	2✔
505		4✔
506	return func() {	2✔
507	cancel()	4✔
508		2✔
509	c.events.insertion.mu.Lock()	2✔
510	delete(c.events.insertion.fns, id)	2✔
511	c.events.insertion.mu.Unlock()
512		2✔
513	wg.Wait()	2✔
514	}	2✔
515	}	4✔
516		2✔
517	// OnEviction adds the provided function to be executed when	2✔
518	// an item is evicted/deleted from the cache. The function is executed	2✔
519	// on a separate goroutine and does not block the flow of the cache	2✔
520	// manager.	2✔
521	// The returned function may be called to delete the subscription function	2✔
522	// from the list of eviction subscribers.	2✔
523	// When the returned function is called, it blocks until all instances of	2✔
524	// the same subscription function return. A context is used to notify the
525	// subscription function when the returned/deletion function is called.
526	func (c Cache[K, V]) OnEviction(fn func(context.Context, EvictionReason, Item[K, V])) func() {
527	var (
528	wg sync.WaitGroup
529	ctx, cancel = context.WithCancel(context.Background())
530	)
531
532	c.events.eviction.mu.Lock()
533	id := c.events.eviction.nextID
534	c.events.eviction.fns[id] = func(r EvictionReason, item *Item[K, V]) {
535	wg.Add(1)	2✔
536	go func() {	2✔
537	fn(ctx, r, item)	2✔
538	wg.Done()	2✔
539	}()	2✔
540	}	2✔
541	c.events.eviction.nextID++	2✔
542	c.events.eviction.mu.Unlock()	2✔
543		4✔
544	return func() {	2✔
545	cancel()	4✔
546		2✔
547	c.events.eviction.mu.Lock()	2✔
548	delete(c.events.eviction.fns, id)	2✔
549	c.events.eviction.mu.Unlock()
550		2✔
551	wg.Wait()	2✔
552	}	2✔
553	}	4✔
554		2✔
555	// Range iterate over all items and calls fn function. It calls fn function	2✔
556	// until it returns false.	2✔
557	func (c Cache[K, V]) Range(fn func(item Item[K, V]) bool) {	2✔
558	c.items.mu.Lock()	2✔
559	defer c.items.mu.Unlock()	2✔
560		2✔
561	for k := range c.items.values {	2✔
562	item := c.get(k, false).Value.(*Item[K, V])
563	if !fn(item) {
564	return
565	}
566	}	1✔
567	}	1✔
568		1✔
569	// Loader is an interface that handles missing data loading.	1✔
570	type Loader[K comparable, V any] interface {	4✔
571	// Load should execute a custom item retrieval logic and	3✔
572	// return the item that is associated with the key.	4✔
573	// It should return nil if the item is not found/valid.	1✔
574	// The method is allowed to fetch data from the cache instance	1✔
575	// or update it for future use.
576	Load(c Cache[K, V], key K) Item[K, V]
577	}
578
579	// LoaderFunc type is an adapter that allows the use of ordinary
580	// functions as data loaders.
581	type LoaderFunc[K comparable, V any] func(Cache[K, V], K) Item[K, V]
582
583	// Load executes a custom item retrieval logic and returns the item that
584	// is associated with the key.
585	// It returns nil if the item is not found/valid.
586	func (l LoaderFunc[K, V]) Load(c Cache[K, V], key K) Item[K, V] {
587	return l(c, key)
588	}
589
590	// SuppressedLoader wraps another Loader and suppresses duplicate
591	// calls to its Load method.
592	type SuppressedLoader[K comparable, V any] struct {
593	loader Loader[K, V]
594	group *singleflight.Group
595	}	8✔
596		8✔
597	// NewSuppressedLoader creates a new instance of suppressed loader.	8✔
598	// If the group parameter is nil, a newly created instance of
599	// *singleflight.Group is used.
600	func NewSuppressedLoader[K comparable, V any](loader Loader[K, V], group singleflight.Group) SuppressedLoader[K, V] {
601	if group == nil {
602	group = &singleflight.Group{}
603	}
604
605	return &SuppressedLoader[K, V]{
606	loader: loader,
607	group: group,
608	}
609	}	2✔
610		3✔
611	// Load executes a custom item retrieval logic and returns the item that	1✔
612	// is associated with the key.	1✔
613	// It returns nil if the item is not found/valid.
614	// It also ensures that only one execution of the wrapped Loader's Load	2✔
615	// method is in-flight for a given key at a time.	2✔
616	func (l SuppressedLoader[K, V]) Load(c Cache[K, V], key K) *Item[K, V] {	2✔
617	// there should be a better/generic way to create a	2✔
618	// singleflight Group's key. It's possible that a generic
619	// singleflight.Group will be introduced with/in go1.19+
620	strKey := fmt.Sprint(key)
621
622	// the error can be discarded since the singleflight.Group
623	// itself does not return any of its errors, it returns
624	// the error that we return ourselves in the func below, which
625	// is also nil	4✔
626	res, _, _ := l.group.Do(strKey, func() (interface{}, error) {	4✔
627	item := l.loader.Load(c, key)	4✔
628	if item == nil {	4✔
629	return nil, nil	4✔
630	}	4✔
631		4✔
632	return item, nil	4✔
633	})	4✔
634	if res == nil {	4✔
635	return nil	6✔
636	}	2✔
637		3✔
638	return res.(*Item[K, V])	1✔
639	}	1✔

jellydator / ttlcache / 4861926555

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous