0187fc45-1fa7-4f9b-9c10-2aef2ebb54e0

Committed 08 May 2023 05:10PM UTC coverage: 57.153% (-0.07%) from 57.225%

Build # 0187fc45-1fa7-4f9b-9c10-2aef2ebb54e0

Build Type

push

buildkite

Committed by GitHub

Commit Message

Update persistence layer to adopt idl update for isolation (#5254)

Run Details

204 of 204 new or added lines in 15 files covered. (100.0%)

85781 of 150089 relevant lines covered (57.15%)

2419.35 hits per line

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

90.73

/common/cache/lru.go

// Copyright (c) 2017 Uber Technologies, Inc.
//
// Permission is hereby granted, free of charge, to any person obtaining a copy
// of this software and associated documentation files (the "Software"), to deal
// in the Software without restriction, including without limitation the rights
// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
// copies of the Software, and to permit persons to whom the Software is
// furnished to do so, subject to the following conditions:
//
// The above copyright notice and this permission notice shall be included in
// all copies or substantial portions of the Software.
//
// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
// THE SOFTWARE.

package cache

import (
        "container/list"
        "errors"
        "sync"
        "time"
)

var (
        // ErrCacheFull is returned if Put fails due to cache being filled with pinned elements
        ErrCacheFull = errors.New("Cache capacity is fully occupied with pinned elements")
)

// upper limit to prevent infinite growing
const cacheCountLimit = 1 << 25

// lru is a concurrent fixed size cache that evicts elements in lru order
type (
        lru struct {
                mut         sync.Mutex
                byAccess    *list.List
                byKey       map[interface{}]*list.Element
                maxCount    int
                ttl         time.Duration
                pin         bool
                rmFunc      RemovedFunc
                sizeFunc    GetCacheItemSizeFunc
                maxSize     uint64
                currSize    uint64
                sizeByKey   map[interface{}]uint64
                isSizeBased bool
        }

        iteratorImpl struct {
                lru        *lru
                createTime time.Time
                nextItem   *list.Element
        }

        entryImpl struct {
                key        interface{}
                createTime time.Time
                value      interface{}
                refCount   int
        }
)

// Close closes the iterator
func (it *iteratorImpl) Close() {
        it.lru.mut.Unlock()
}

// HasNext return true if there is more items to be returned
func (it *iteratorImpl) HasNext() bool {
        return it.nextItem != nil
}

// Next return the next item
func (it *iteratorImpl) Next() Entry {
        if it.nextItem == nil {
                panic("LRU cache iterator Next called when there is no next item")
        }

        entry := it.nextItem.Value.(*entryImpl)
        it.nextItem = it.nextItem.Next()
        // make a copy of the entry so there will be no concurrent access to this entry
        entry = &entryImpl{
                key:        entry.key,
                value:      entry.value,
                createTime: entry.createTime,
        }
        it.prepareNext()
        return entry
}

func (it *iteratorImpl) prepareNext() {
        for it.nextItem != nil {
                entry := it.nextItem.Value.(*entryImpl)
                if it.lru.isEntryExpired(entry, it.createTime) {
                        nextItem := it.nextItem.Next()
                        it.lru.deleteInternal(it.nextItem)
                        it.nextItem = nextItem
                } else {
                        return
                }
        }
}

// Iterator returns an iterator to the map. This map
// does not use re-entrant locks, so access or modification
// to the map during iteration can cause a dead lock.
func (c *lru) Iterator() Iterator {
        c.mut.Lock()
        iterator := &iteratorImpl{
                lru:        c,
                createTime: time.Now(),
                nextItem:   c.byAccess.Front(),
        }
        iterator.prepareNext()
        return iterator
}

func (entry *entryImpl) Key() interface{} {
        return entry.key
}

func (entry *entryImpl) Value() interface{} {
        return entry.value
}

func (entry *entryImpl) CreateTime() time.Time {
        return entry.createTime
}

// New creates a new cache with the given options
func New(opts *Options) Cache {
        if opts == nil || (opts.MaxCount <= 0 && (opts.MaxSize <= 0 || opts.GetCacheItemSizeFunc == nil)) {
                panic("Either MaxCount (count based) or " +
                        "MaxSize and GetCacheItemSizeFunc (size based) options must be provided for the LRU cache")
        }

        cache := &lru{
                byAccess: list.New(),
                byKey:    make(map[interface{}]*list.Element, opts.InitialCapacity),
                ttl:      opts.TTL,
                pin:      opts.Pin,
                rmFunc:   opts.RemovedFunc,
        }

        cache.isSizeBased = opts.GetCacheItemSizeFunc != nil && opts.MaxSize > 0

        if cache.isSizeBased {
                cache.sizeFunc = opts.GetCacheItemSizeFunc
                cache.maxSize = opts.MaxSize
                cache.sizeByKey = make(map[interface{}]uint64, opts.InitialCapacity)
        } else {
                // cache is count based if max size and sizeFunc are not provided
                cache.maxCount = opts.MaxCount
                cache.sizeFunc = func(interface{}) uint64 {
                        return 0
                }
        }
        return cache
}

// Get retrieves the value stored under the given key
func (c *lru) Get(key interface{}) interface{} {
        c.mut.Lock()
        defer c.mut.Unlock()

        element := c.byKey[key]
        if element == nil {
                return nil
        }

        entry := element.Value.(*entryImpl)

        if c.isEntryExpired(entry, time.Now()) {
                // Entry has expired
                c.deleteInternal(element)
                return nil
        }

        if c.pin {
                entry.refCount++
        }
        c.byAccess.MoveToFront(element)
        return entry.value
}

// Put puts a new value associated with a given key, returning the existing value (if present)
func (c *lru) Put(key interface{}, value interface{}) interface{} {
        if c.pin {
                panic("Cannot use Put API in Pin mode. Use Delete and PutIfNotExist if necessary")
        }
        val, _ := c.putInternal(key, value, true)
        return val
}

// PutIfNotExist puts a value associated with a given key if it does not exist
func (c *lru) PutIfNotExist(key interface{}, value interface{}) (interface{}, error) {
        existing, err := c.putInternal(key, value, false)
        if err != nil {
                return nil, err
        }

        if existing == nil {
                // This is a new value
                return value, err
        }

        return existing, err
}

// Delete deletes a key, value pair associated with a key
func (c *lru) Delete(key interface{}) {
        c.mut.Lock()
        defer c.mut.Unlock()

        element := c.byKey[key]
        if element != nil {
                c.deleteInternal(element)
        }
}

// Release decrements the ref count of a pinned element.
func (c *lru) Release(key interface{}) {
        c.mut.Lock()
        defer c.mut.Unlock()

        elt, ok := c.byKey[key]
        if !ok {
                return
        }
        entry := elt.Value.(*entryImpl)
        entry.refCount--
}

// Size returns the number of entries currently in the lru, useful if cache is not full
func (c *lru) Size() int {
        c.mut.Lock()
        defer c.mut.Unlock()

        return len(c.byKey)
}

// Put puts a new value associated with a given key, returning the existing value (if present)
// allowUpdate flag is used to control overwrite behavior if the value exists
func (c *lru) putInternal(key interface{}, value interface{}, allowUpdate bool) (interface{}, error) {
        valueSize := c.sizeFunc(value)
        c.mut.Lock()
        defer c.mut.Unlock()

        elt := c.byKey[key]
        if elt != nil {
                entry := elt.Value.(*entryImpl)
                if c.isEntryExpired(entry, time.Now()) {
                        // Entry has expired
                        c.deleteInternal(elt)
                } else {
                        existing := entry.value
                        if allowUpdate {
                                entry.value = value
                                if c.ttl != 0 {
                                        entry.createTime = time.Now()
                                }
                        }

                        c.byAccess.MoveToFront(elt)
                        if c.pin {
                                entry.refCount++
                        }
                        return existing, nil
                }
        }

        entry := &entryImpl{
                key:   key,
                value: value,
        }

        if c.pin {
                entry.refCount++
        }

        if c.ttl != 0 {
                entry.createTime = time.Now()
        }

        c.byKey[key] = c.byAccess.PushFront(entry)
        c.updateSizeOnAdd(key, valueSize)
        for c.isCacheFull() {
                oldest := c.byAccess.Back().Value.(*entryImpl)

                if oldest.refCount > 0 {
                        // Cache is full with pinned elements
                        // revert the insert and return
                        c.deleteInternal(c.byAccess.Front())
                        return nil, ErrCacheFull
                }

                c.deleteInternal(c.byAccess.Back())
        }
        return nil, nil
}

func (c *lru) deleteInternal(element *list.Element) {
        entry := c.byAccess.Remove(element).(*entryImpl)
        if c.rmFunc != nil {
                go c.rmFunc(entry.value)
        }
        delete(c.byKey, entry.key)
        c.updateSizeOnDelete(entry.key)
}

func (c *lru) isEntryExpired(entry *entryImpl, currentTime time.Time) bool {
        return entry.refCount == 0 && !entry.createTime.IsZero() && currentTime.After(entry.createTime.Add(c.ttl))
}

func (c *lru) isCacheFull() bool {
        count := len(c.byKey)
        // if the value size is greater than maxSize(should never happen) then the item wont be cached
        return (!c.isSizeBased && count == c.maxCount) || c.currSize > c.maxSize || count > cacheCountLimit
}

func (c *lru) updateSizeOnAdd(key interface{}, valueSize uint64) {
        if c.isSizeBased {
                c.sizeByKey[key] = valueSize
                // the int overflow should not happen here
                c.currSize += uint64(valueSize)
        }
}

func (c *lru) updateSizeOnDelete(key interface{}) {
        if c.isSizeBased {
                c.currSize -= uint64(c.sizeByKey[key])
                delete(c.sizeByKey, key)
        }
}

1	// Copyright (c) 2017 Uber Technologies, Inc.
2	//
3	// Permission is hereby granted, free of charge, to any person obtaining a copy
4	// of this software and associated documentation files (the "Software"), to deal
5	// in the Software without restriction, including without limitation the rights
6	// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
7	// copies of the Software, and to permit persons to whom the Software is
8	// furnished to do so, subject to the following conditions:
9	//
10	// The above copyright notice and this permission notice shall be included in
11	// all copies or substantial portions of the Software.
12	//
13	// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
14	// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
15	// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
16	// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
17	// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
18	// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
19	// THE SOFTWARE.
20
21	package cache
22
23	import (
24	"container/list"
25	"errors"
26	"sync"
27	"time"
28	)
29
30	var (
31	// ErrCacheFull is returned if Put fails due to cache being filled with pinned elements
32	ErrCacheFull = errors.New("Cache capacity is fully occupied with pinned elements")
33	)
34
35	// upper limit to prevent infinite growing
36	const cacheCountLimit = 1 << 25
37
38	// lru is a concurrent fixed size cache that evicts elements in lru order
39	type (
40	lru struct {
41	mut sync.Mutex
42	byAccess *list.List
43	byKey map[interface{}]*list.Element
44	maxCount int
45	ttl time.Duration
46	pin bool
47	rmFunc RemovedFunc
48	sizeFunc GetCacheItemSizeFunc
49	maxSize uint64
50	currSize uint64
51	sizeByKey map[interface{}]uint64
52	isSizeBased bool
53	}
54
55	iteratorImpl struct {
56	lru *lru
57	createTime time.Time
58	nextItem *list.Element
59	}
60
61	entryImpl struct {
62	key interface{}
63	createTime time.Time
64	value interface{}
65	refCount int
66	}
67	)
68
69	// Close closes the iterator
70	func (it *iteratorImpl) Close() {	6,488✔
71	it.lru.mut.Unlock()	6,488✔
72	}	6,488✔
73
74	// HasNext return true if there is more items to be returned
75	func (it *iteratorImpl) HasNext() bool {	16,087✔
76	return it.nextItem != nil	16,087✔
77	}	16,087✔
78
79	// Next return the next item
80	func (it *iteratorImpl) Next() Entry {	9,604✔
81	if it.nextItem == nil {	9,604✔
82	panic("LRU cache iterator Next called when there is no next item")	×
83	}
84
85	entry := it.nextItem.Value.(*entryImpl)	9,604✔
86	it.nextItem = it.nextItem.Next()	9,604✔
87	// make a copy of the entry so there will be no concurrent access to this entry	9,604✔
88	entry = &entryImpl{	9,604✔
89	key: entry.key,	9,604✔
90	value: entry.value,	9,604✔
91	createTime: entry.createTime,	9,604✔
92	}	9,604✔
93	it.prepareNext()	9,604✔
94	return entry	9,604✔
95	}
96
97	func (it *iteratorImpl) prepareNext() {	16,092✔
98	for it.nextItem != nil {	25,696✔
99	entry := it.nextItem.Value.(*entryImpl)	9,604✔
100	if it.lru.isEntryExpired(entry, it.createTime) {	9,604✔
101	nextItem := it.nextItem.Next()	×
102	it.lru.deleteInternal(it.nextItem)	×
103	it.nextItem = nextItem	×
104	} else {	9,604✔
105	return	9,604✔
106	}	9,604✔
107	}
108	}
109
110	// Iterator returns an iterator to the map. This map
111	// does not use re-entrant locks, so access or modification
112	// to the map during iteration can cause a dead lock.
113	func (c *lru) Iterator() Iterator {	6,488✔
114	c.mut.Lock()	6,488✔
115	iterator := &iteratorImpl{	6,488✔
116	lru: c,	6,488✔
117	createTime: time.Now(),	6,488✔
118	nextItem: c.byAccess.Front(),	6,488✔
119	}	6,488✔
120	iterator.prepareNext()	6,488✔
121	return iterator	6,488✔
122	}	6,488✔
123
124	func (entry *entryImpl) Key() interface{} {	5,604✔
125	return entry.key	5,604✔
126	}	5,604✔
127
128	func (entry *entryImpl) Value() interface{} {	5,604✔
129	return entry.value	5,604✔
130	}	5,604✔
131
132	func (entry *entryImpl) CreateTime() time.Time {	5,596✔
133	return entry.createTime	5,596✔
134	}	5,596✔
135
136	// New creates a new cache with the given options
137	func New(opts *Options) Cache {	1,440✔
138	if opts == nil \|\| (opts.MaxCount <= 0 && (opts.MaxSize <= 0 \|\| opts.GetCacheItemSizeFunc == nil)) {	1,443✔
139	panic("Either MaxCount (count based) or " +	3✔
140	"MaxSize and GetCacheItemSizeFunc (size based) options must be provided for the LRU cache")	3✔
141	}
142
143	cache := &lru{	1,437✔
144	byAccess: list.New(),	1,437✔
145	byKey: make(map[interface{}]*list.Element, opts.InitialCapacity),	1,437✔
146	ttl: opts.TTL,	1,437✔
147	pin: opts.Pin,	1,437✔
148	rmFunc: opts.RemovedFunc,	1,437✔
149	}	1,437✔
150		1,437✔
151	cache.isSizeBased = opts.GetCacheItemSizeFunc != nil && opts.MaxSize > 0	1,437✔
152		1,437✔
153	if cache.isSizeBased {	1,438✔
154	cache.sizeFunc = opts.GetCacheItemSizeFunc	1✔
155	cache.maxSize = opts.MaxSize	1✔
156	cache.sizeByKey = make(map[interface{}]uint64, opts.InitialCapacity)	1✔
157	} else {	1,437✔
158	// cache is count based if max size and sizeFunc are not provided	1,436✔
159	cache.maxCount = opts.MaxCount	1,436✔
160	cache.sizeFunc = func(interface{}) uint64 {	29,554✔
161	return 0	28,118✔
162	}	28,118✔
163	}
164	return cache	1,437✔
165	}
166
167	// Get retrieves the value stored under the given key
168	func (c *lru) Get(key interface{}) interface{} {	34,550✔
169	c.mut.Lock()	34,550✔
170	defer c.mut.Unlock()	34,550✔
171		34,550✔
172	element := c.byKey[key]	34,550✔
173	if element == nil {	35,656✔
174	return nil	1,106✔
175	}	1,106✔
176
177	entry := element.Value.(*entryImpl)	33,447✔
178		33,447✔
179	if c.isEntryExpired(entry, time.Now()) {	33,450✔
180	// Entry has expired	3✔
181	c.deleteInternal(element)	3✔
182	return nil	3✔
183	}	3✔
184
185	if c.pin {	43,863✔
186	entry.refCount++	10,419✔
187	}	10,419✔
188	c.byAccess.MoveToFront(element)	33,444✔
189	return entry.value	33,444✔
190	}
191
192	// Put puts a new value associated with a given key, returning the existing value (if present)
193	func (c *lru) Put(key interface{}, value interface{}) interface{} {	27,053✔
194	if c.pin {	27,053✔
195	panic("Cannot use Put API in Pin mode. Use Delete and PutIfNotExist if necessary")	×
196	}
197	val, _ := c.putInternal(key, value, true)	27,053✔
198	return val	27,053✔
199	}
200
201	// PutIfNotExist puts a value associated with a given key if it does not exist
202	func (c *lru) PutIfNotExist(key interface{}, value interface{}) (interface{}, error) {	1,075✔
203	existing, err := c.putInternal(key, value, false)	1,075✔
204	if err != nil {	1,075✔
205	return nil, err	×
206	}	×
207
208	if existing == nil {	2,150✔
209	// This is a new value	1,075✔
210	return value, err	1,075✔
211	}	1,075✔
212
213	return existing, err	×
214	}
215
216	// Delete deletes a key, value pair associated with a key
217	func (c *lru) Delete(key interface{}) {	2✔
218	c.mut.Lock()	2✔
219	defer c.mut.Unlock()	2✔
220		2✔
221	element := c.byKey[key]	2✔
222	if element != nil {	4✔
223	c.deleteInternal(element)	2✔
224	}	2✔
225	}
226
227	// Release decrements the ref count of a pinned element.
228	func (c *lru) Release(key interface{}) {	11,491✔
229	c.mut.Lock()	11,491✔
230	defer c.mut.Unlock()	11,491✔
231		11,491✔
232	elt, ok := c.byKey[key]	11,491✔
233	if !ok {	11,491✔
234	return	×
235	}	×
236	entry := elt.Value.(*entryImpl)	11,491✔
237	entry.refCount--	11,491✔
238	}
239
240	// Size returns the number of entries currently in the lru, useful if cache is not full
241	func (c *lru) Size() int {	13✔
242	c.mut.Lock()	13✔
243	defer c.mut.Unlock()	13✔
244		13✔
245	return len(c.byKey)	13✔
246	}	13✔
247
248	// Put puts a new value associated with a given key, returning the existing value (if present)
249	// allowUpdate flag is used to control overwrite behavior if the value exists
250	func (c *lru) putInternal(key interface{}, value interface{}, allowUpdate bool) (interface{}, error) {	28,125✔
251	valueSize := c.sizeFunc(value)	28,125✔
252	c.mut.Lock()	28,125✔
253	defer c.mut.Unlock()	28,125✔
254		28,125✔
255	elt := c.byKey[key]	28,125✔
256	if elt != nil {	52,601✔
257	entry := elt.Value.(*entryImpl)	24,476✔
258	if c.isEntryExpired(entry, time.Now()) {	24,476✔
259	// Entry has expired	×
260	c.deleteInternal(elt)	×
261	} else {	24,476✔
262	existing := entry.value	24,476✔
263	if allowUpdate {	48,952✔
264	entry.value = value	24,476✔
265	if c.ttl != 0 {	28,950✔
266	entry.createTime = time.Now()	4,474✔
267	}	4,474✔
268	}
269
270	c.byAccess.MoveToFront(elt)	24,476✔
271	if c.pin {	24,476✔
272	entry.refCount++	×
273	}	×
274	return existing, nil	24,476✔
275	}
276	}
277
278	entry := &entryImpl{	3,652✔
279	key: key,	3,652✔
280	value: value,	3,652✔
281	}	3,652✔
282		3,652✔
283	if c.pin {	4,727✔
284	entry.refCount++	1,075✔
285	}	1,075✔
286
287	if c.ttl != 0 {	7,275✔
288	entry.createTime = time.Now()	3,623✔
289	}	3,623✔
290
291	c.byKey[key] = c.byAccess.PushFront(entry)	3,652✔
292	c.updateSizeOnAdd(key, valueSize)	3,652✔
293	for c.isCacheFull() {	3,663✔
294	oldest := c.byAccess.Back().Value.(*entryImpl)	11✔
295		11✔
296	if oldest.refCount > 0 {	11✔
297	// Cache is full with pinned elements	×
298	// revert the insert and return	×
299	c.deleteInternal(c.byAccess.Front())	×
300	return nil, ErrCacheFull	×
301	}	×
302
303	c.deleteInternal(c.byAccess.Back())	11✔
304	}
305	return nil, nil	3,652✔
306	}
307
308	func (c lru) deleteInternal(element list.Element) {	16✔
309	entry := c.byAccess.Remove(element).(*entryImpl)	16✔
310	if c.rmFunc != nil {	19✔
311	go c.rmFunc(entry.value)	3✔
312	}	3✔
313	delete(c.byKey, entry.key)	16✔
314	c.updateSizeOnDelete(entry.key)	16✔
315	}
316
317	func (c lru) isEntryExpired(entry entryImpl, currentTime time.Time) bool {	67,524✔
318	return entry.refCount == 0 && !entry.createTime.IsZero() && currentTime.After(entry.createTime.Add(c.ttl))	67,524✔
319	}	67,524✔
320
321	func (c *lru) isCacheFull() bool {	3,663✔
322	count := len(c.byKey)	3,663✔
323	// if the value size is greater than maxSize(should never happen) then the item wont be cached	3,663✔
324	return (!c.isSizeBased && count == c.maxCount) \|\| c.currSize > c.maxSize \|\| count > cacheCountLimit	3,663✔
325	}	3,663✔
326
327	func (c *lru) updateSizeOnAdd(key interface{}, valueSize uint64) {	3,652✔
328	if c.isSizeBased {	3,659✔
329	c.sizeByKey[key] = valueSize	7✔
330	// the int overflow should not happen here	7✔
331	c.currSize += uint64(valueSize)	7✔
332	}	7✔
333	}
334
335	func (c *lru) updateSizeOnDelete(key interface{}) {	16✔
336	if c.isSizeBased {	23✔
337	c.currSize -= uint64(c.sizeByKey[key])	7✔
338	delete(c.sizeByKey, key)	7✔
339	}	7✔
340	}

uber / cadence / 0187fc45-1fa7-4f9b-9c10-2aef2ebb54e0

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