Vihren Atlas

go.temporal.io/server/common/cache/lru.go

497 LOC · 315 covered · 182 uncovered · 91 ranges · 4913 concepts · 54 introducers · 1954 tests

File neighbourhood

The centred file is linked to every concept that introduces one of its ranges, every test that runs code from the file, and the gray connector concepts standing between those tests and the file's own introducer concepts. Undirected links join concepts to every file where they introduce source and concepts to the tests they introduce; arrows show specialization between the displayed concepts and bridge only concepts omitted from this view. Concept colors match the source ranges below; connector concepts have no source color and are shown in gray.

Focused file, its introducer and connector concepts, their introduced files, and tests that run code from the file

In the embedded map, ordinary wheel input scrolls the page; use the visible controls to zoom and drag to pan. Open the full-screen map for canvas navigation: wheel pans, Ctrl/Command plus wheel zooms, and arrow keys pan when this region is focused. On touch screens, open the full-screen map to pan or pinch. If JavaScript or WebGL is unavailable, use the related-file, concept, and source links on this page.

Graph controls are ready.

Interactive rendering requires JavaScript and WebGL. Use the related-file, concept, and source links on this page while the interactive map is unavailable.

Files sharing a concept 3

A concept that introduces code in this file also introduces code in these files.

Files with a broader (ancestor) concept 108

These files introduce code for a direct ancestor — a broader, larger-extent concept — of a concept that introduces code here.

showing 50 of 108

Files with a narrower (descendant) concept 25

These files introduce code for a direct descendant — a narrower, larger-intent concept — of a concept that introduces code here.

1 package cache
2
3 import (
4 "container/list"
5 "context"
6 "sync"
7 "time"
8
9 enumspb "go.temporal.io/api/enums/v1"
10 "go.temporal.io/api/serviceerror"
11 "go.temporal.io/server/common/clock"
12 "go.temporal.io/server/common/dynamicconfig"
13 "go.temporal.io/server/common/goro"
14 "go.temporal.io/server/common/metrics"
15 )
16
17 var (
18 // ErrCacheFull is returned if Put fails due to cache being filled with pinned elements
19 ErrCacheFull = &serviceerror.ResourceExhausted{
20 Cause: enumspb.RESOURCE_EXHAUSTED_CAUSE_SYSTEM_OVERLOADED,
21 Scope: enumspb.RESOURCE_EXHAUSTED_SCOPE_SYSTEM,
22 Message: "cache capacity is fully occupied with pinned elements",
23 }
24 // ErrCacheItemTooLarge is returned if Put fails due to item size being larger than max cache capacity
25 ErrCacheItemTooLarge = serviceerror.NewInternal("cache item size is larger than max cache capacity")
26 )
27
28 const emptyEntrySize = 0
29
30 // lru is a concurrent fixed size cache that evicts elements in lru order
31 type (
32 lru struct {
33 mut sync.Mutex
34 byAccess *list.List
35 byKey map[any]*list.Element
36 maxSize int
37 currSize int
38 pinnedSize int
39 onPut func(val any)
40 onEvict func(val any)
41 ttl time.Duration
42 pin bool
43 timeSource clock.TimeSource
44 metricsHandler metrics.Handler
45 backgroundEvict dynamicconfig.TypedPropertyFn[dynamicconfig.CacheBackgroundEvictSettings]
46 loops goro.Group
47 }
48
49 iteratorImpl struct {
50 lru *lru
51 createTime time.Time
52 nextItem *list.Element
53 }
54
55 entryImpl struct {
56 key any
57 createTime time.Time
58 value any
59 refCount int
60 size int
61 }
62 )
63
64 // Close closes the iterator
65 > func (it *iteratorImpl) Close() { lru.go ×4
66 > it.lru.mut.Unlock()
67 > }
68
69 // HasNext return true if there is more items to be returned
70 > func (it *iteratorImpl) HasNext() bool { lru.go ×4
71 > return it.nextItem != nil
72 > }
73
74 // Next return the next item
75 > func (it *iteratorImpl) Next() Entry { lru.go ×4
76 > if it.nextItem == nil {
77 panic("LRU cache iterator Next called when there is no next item")
78 }
79
80 > entry := it.nextItem.Value.(*entryImpl) lru.go ×4
81 > it.nextItem = it.nextItem.Next()
82 > // make a copy of the entry so there will be no concurrent access to this entry
83 > entry = &entryImpl{
84 > key: entry.key,
85 > value: entry.value,
86 > size: entry.size,
87 > createTime: entry.createTime,
88 > }
89 > it.prepareNext()
90 > return entry
91 }
92
93 > func (it *iteratorImpl) prepareNext() { lru.go ×4
94 > for it.nextItem != nil {
95 > entry := it.nextItem.Value.(*entryImpl) lru.go ×4
96 > if it.lru.isEntryExpired(entry, it.createTime) {
97 nextItem := it.nextItem.Next()
98 it.lru.deleteInternal(it.nextItem)
99 it.nextItem = nextItem
100 > } else { lru.go ×4
101 > return
102 > }
103 }
104 }
105
106 // Iterator returns an iterator to the map. This map
107 // does not use re-entrant locks, so access or modification
108 // to the map during iteration can cause a dead lock.
109 > func (c *lru) Iterator() Iterator { lru.go ×4
110 > c.mut.Lock()
111 > iterator := &iteratorImpl{
112 > lru: c,
113 > createTime: c.timeSource.Now().UTC(),
114 > nextItem: c.byAccess.Front(),
115 > }
116 > iterator.prepareNext()
117 > return iterator
118 > }
119
120 > func (entry *entryImpl) Key() any { lru.go ×2
121 > return entry.key
122 > }
123
124 > func (entry *entryImpl) Value() any { lru.go ×2
125 > return entry.value
126 > }
127
128 > func (entry *entryImpl) Size() int { lru.go ×1
129 > return entry.size
130 > }
131
132 > func (entry *entryImpl) CreateTime() time.Time { poller_history.go ×2
133 > return entry.createTime
134 > }
135
136 // New creates a new cache with the given options
137 > func New(maxSize int, opts *Options) StoppableCache { lru.go ×1
138 > return NewWithMetrics(maxSize, opts, metrics.NoopMetricsHandler)
139 > }
140
141 // NewWithMetrics creates a new cache that will emit capacity and ttl metrics.
142 // handler should be tagged with metrics.CacheTypeTag.
143 > func NewWithMetrics(maxSize int, opts *Options, handler metrics.Handler) StoppableCache { lru.go ×5
144 > if opts == nil {
145 > opts = &Options{} lru.go ×1
146 > }
147
148 > backgroundEvict := opts.BackgroundEvict lru.go ×5
149 > if backgroundEvict == nil {
150 > backgroundEvict = func() dynamicconfig.CacheBackgroundEvictSettings { lru.go ×1
151 > return dynamicconfig.CacheBackgroundEvictSettings{
152 > Enabled: false,
153 > }
154 > }
155 }
156
157 > timeSource := opts.TimeSource lru.go ×5
158 > if timeSource == nil {
159 > timeSource = clock.NewRealTimeSource() lru.go ×1
160 > }
161
162 > metrics.CacheSize.With(handler).Record(float64(maxSize)) lru.go ×5
163 > metrics.CacheTtl.With(handler).Record(opts.TTL)
164 > c := &lru{
165 > byAccess: list.New(),
166 > byKey: make(map[any]*list.Element),
167 > ttl: opts.TTL,
168 > maxSize: maxSize,
169 > currSize: 0,
170 > pin: opts.Pin,
171 > onPut: opts.OnPut,
172 > onEvict: opts.OnEvict,
173 > timeSource: timeSource,
174 > metricsHandler: handler,
175 > backgroundEvict: backgroundEvict,
176 > }
177 > if c.backgroundEvict().Enabled {
178 > c.loops.Go(c.bgEvictLoop) lru.go ×6
179 > }
180 > return c lru.go ×5
181 }
182
183 // NewLRU creates a new LRU cache of the given size, setting initial capacity
184 // to the max size
185 > func NewLRU(maxSize int, handler metrics.Handler) StoppableCache { lru.go ×1
186 > return New(maxSize, nil)
187 > }
188
189 // Get retrieves the value stored under the given key
190 > func (c *lru) Get(key any) any { lru.go ×1
191 > if c.maxSize == 0 { //
192 > return nil lru.go ×4
193 > }
194 > c.mut.Lock() lru.go ×1
195 > defer c.mut.Unlock()
196 >
197 > element := c.byKey[key]
198 > if element == nil {
199 > return nil lru.go ×1
200 > }
201
202 > entry := element.Value.(*entryImpl) lru.go ×1
203 >
204 > if c.isEntryExpired(entry, c.timeSource.Now().UTC()) {
205 > // Entry has expired lru.go ×1
206 > c.deleteInternal(element)
207 > return nil
208 > }
209
210 > metrics.CacheEntryAgeOnGet.With(c.metricsHandler).Record(c.timeSource.Now().UTC().Sub(entry.createTime)) lru.go ×1
211 >
212 > c.updateEntryRefCount(entry)
213 > c.byAccess.MoveToFront(element)
214 > return entry.value
215 }
216
217 // Put puts a new value associated with a given key, returning the existing value (if present)
218 > func (c *lru) Put(key any, value any) any { lru.go ×2
219 > if c.pin {
220 panic("Cannot use Put API in Pin mode. Use Delete and PutIfNotExist if necessary")
221 }
222 > val, _ := c.putInternal(key, value, true) lru.go ×2
223 > return val
224 }
225
226 // PutIfNotExist puts a value associated with a given key if it does not exist
227 > func (c *lru) PutIfNotExist(key any, value any) (any, error) { lru.go ×1
228 > existing, err := c.putInternal(key, value, false)
229 > if err != nil {
230 > return nil, err lru.go ×1
231 > }
232
233 > if existing == nil { lru.go ×1
234 > // This is a new value lru.go ×1
235 > return value, err
236 > }
237
238 > return existing, err lru.go ×1
239 }
240
241 // Delete deletes a key, value pair associated with a key
242 > func (c *lru) Delete(key any) { lru.go ×1
243 > if c.maxSize == 0 {
244 > return lru.go ×4
245 > }
246 > c.mut.Lock() lru.go ×1
247 > defer c.mut.Unlock()
248 >
249 > element := c.byKey[key]
250 > if element != nil {
251 > c.deleteInternal(element)
252 > }
253 }
254
255 // Release decrements the ref count of a pinned element.
256 > func (c *lru) Release(key any) { lru.go ×1
257 > if c.maxSize == 0 || !c.pin {
258 > return lru.go ×4
259 > }
260 > c.mut.Lock() lru.go ×1
261 > defer c.mut.Unlock()
262 >
263 > elt, ok := c.byKey[key]
264 > if !ok {
265 > return lru.go ×1
266 > }
267 > entry := elt.Value.(*entryImpl) lru.go ×3
268 > entry.refCount--
269 > if entry.refCount == 0 {
270 > c.pinnedSize -= entry.Size() lru.go ×1
271 > metrics.CachePinnedUsage.With(c.metricsHandler).Record(float64(c.pinnedSize))
272 > }
273 // Entry size might have changed. Recalculate size and evict entries if necessary.
274 > newEntrySize := getSize(entry.value) lru.go ×3
275 > c.currSize = c.calculateNewCacheSize(newEntrySize, entry.Size())
276 > entry.size = newEntrySize
277 > if c.currSize > c.maxSize {
278 > c.tryEvictUntilCacheSizeUnderLimit() lru.go ×2
279 > }
280 > metrics.CacheUsage.With(c.metricsHandler).Record(float64(c.currSize)) lru.go ×3
281 }
282
283 // Size returns the current size of the lru, useful if cache is not full. This size is calculated by summing
284 // the size of all entries in the cache. And the entry size is calculated by the size of the value.
285 // The size of the value is calculated implementing the Sizeable interface. If the value does not implement
286 // the Sizeable interface, the size is 1.
287 > func (c *lru) Size() int { lru.go ×1
288 > c.mut.Lock()
289 > defer c.mut.Unlock()
290 >
291 > return c.currSize
292 > }
293
294 // Put puts a new value associated with a given key, returning the existing value (if present)
295 // allowUpdate flag is used to control overwrite behavior if the value exists.
296 > func (c *lru) putInternal(key any, value any, allowUpdate bool) (any, error) { lru.go ×1
297 > if c.maxSize == 0 {
298 > return nil, nil lru.go ×4
299 > }
300 > newEntrySize := getSize(value) lru.go ×1
301 > if newEntrySize > c.maxSize {
302 > return nil, ErrCacheItemTooLarge lru.go ×1
303 > }
304
305 > c.mut.Lock() lru.go ×9
306 > defer c.mut.Unlock()
307 >
308 > elt := c.byKey[key]
309 > // If the entry exists, check if it has expired or update the value
310 > if elt != nil {
311 > existingEntry := elt.Value.(*entryImpl) lru.go ×2
312 > if !c.isEntryExpired(existingEntry, c.timeSource.Now().UTC()) {
313 > existingVal := existingEntry.value
314 >
315 > if allowUpdate {
316 > newCacheSize := c.calculateNewCacheSize(newEntrySize, existingEntry.Size()) lru.go ×2
317 > if newCacheSize > c.maxSize {
318 > c.tryEvictUntilEnoughSpaceWithSkipEntry(newEntrySize, existingEntry) lru.go ×3
319 > // calculate again after eviction
320 > newCacheSize = c.calculateNewCacheSize(newEntrySize, existingEntry.Size())
321 > if newCacheSize > c.maxSize {
322 // This should never happen since allowUpdate is always **true** for non-pinned cache,
323 // and if all entries are not pinned(ref==0), then the cache should never be full as long as
324 // new entry's size is less than max size.
325 // However, to prevent any unexpected behavior, it checks the cache size again.
326 return nil, ErrCacheFull
327 }
328 }
329 > existingEntry.value = value lru.go ×2
330 > existingEntry.size = newEntrySize
331 > c.currSize = newCacheSize
332 > metrics.CacheUsage.With(c.metricsHandler).Record(float64(c.currSize))
333 > c.updateEntryTTL(existingEntry)
334 >
335 > if c.onPut != nil {
336 > c.onPut(value) lru.go ×1
337 > }
338 }
339
340 > c.updateEntryRefCount(existingEntry) lru.go ×2
341 > c.byAccess.MoveToFront(elt)
342 > return existingVal, nil
343 }
344
345 // Entry has expired
346 c.deleteInternal(elt)
347 }
348
349 > c.tryEvictUntilEnoughSpaceWithSkipEntry(newEntrySize, nil) lru.go ×9
350 >
351 > // check if the new entry can fit in the cache
352 > newCacheSize := c.calculateNewCacheSize(newEntrySize, emptyEntrySize)
353 > if newCacheSize > c.maxSize {
354 > return nil, ErrCacheFull lru.go ×1
355 > }
356
357 > entry := &entryImpl{ lru.go ×9
358 > key: key,
359 > value: value,
360 > size: newEntrySize,
361 > }
362 > c.updateEntryTTL(entry)
363 > c.updateEntryRefCount(entry)
364 > element := c.byAccess.PushFront(entry)
365 > c.byKey[key] = element
366 > c.currSize = newCacheSize
367 > metrics.CacheUsage.With(c.metricsHandler).Record(float64(c.currSize))
368 >
369 > if c.onPut != nil {
370 > c.onPut(value) lru.go ×1
371 > }
372
373 > return nil, nil lru.go ×9
374 }
375
376 > func (c *lru) calculateNewCacheSize(newEntrySize int, existingEntrySize int) int { lru.go ×9
377 > return c.currSize - existingEntrySize + newEntrySize
378 > }
379
380 > func (c *lru) deleteInternal(element *list.Element) { lru.go ×1
381 > entry := c.byAccess.Remove(element).(*entryImpl)
382 > c.currSize -= entry.Size()
383 > metrics.CacheUsage.With(c.metricsHandler).Record(float64(c.currSize))
384 > metrics.CacheEntryAgeOnEviction.With(c.metricsHandler).Record(c.timeSource.Now().UTC().Sub(entry.createTime))
385 > delete(c.byKey, entry.key)
386 >
387 > if c.onEvict != nil {
388 > c.onEvict(entry.value) lru.go ×1
389 > }
390 }
391
392 // tryEvictUntilCacheSizeUnderLimit tries to evict entries until c.currSize is less than c.maxSize.
393 > func (c *lru) tryEvictUntilCacheSizeUnderLimit() { lru.go ×2
394 > c.tryEvictUntilEnoughSpaceWithSkipEntry(0, nil)
395 > }
396
397 // tryEvictUntilEnoughSpaceWithSkipEntry try to evict entries until there is enough space for the new entry without
398 // evicting the existing entry. the existing entry is skipped because it is being updated.
399 > func (c *lru) tryEvictUntilEnoughSpaceWithSkipEntry(newEntrySize int, existingEntry *entryImpl) { lru.go ×9
400 > element := c.byAccess.Back()
401 > existingEntrySize := 0
402 > if existingEntry != nil {
403 > existingEntrySize = existingEntry.Size() lru.go ×3
404 > }
405
406 > for c.calculateNewCacheSize(newEntrySize, existingEntrySize) > c.maxSize && element != nil { lru.go ×9
407 > entry := element.Value.(*entryImpl) lru.go ×3
408 > if existingEntry != nil && entry.key == existingEntry.key {
409 > element = element.Prev() lru.go ×3
410 > continue
411 }
412 > element = c.tryEvictAndGetPreviousElement(entry, element) lru.go ×3
413 }
414 }
415
416 > func (c *lru) tryEvictAndGetPreviousElement(entry *entryImpl, element *list.Element) *list.Element { lru.go ×3
417 > if entry.refCount == 0 {
418 > elementPrev := element.Prev() lru.go ×1
419 > // currSize will be updated within deleteInternal
420 > c.deleteInternal(element)
421 > return elementPrev
422 > }
423 // entry.refCount > 0
424 // skip, entry still being referenced
425 > return element.Prev() lru.go ×1
426 }
427
428 > func (c *lru) isEntryExpired(entry *entryImpl, currentTime time.Time) bool { lru.go ×1
429 > return entry.refCount == 0 && !entry.createTime.IsZero() && currentTime.After(entry.createTime.Add(c.ttl))
430 > }
431
432 > func (c *lru) updateEntryTTL(entry *entryImpl) { lru.go ×9
433 > if c.ttl != 0 {
434 > entry.createTime = c.timeSource.Now().UTC() lru.go ×1
435 > }
436 }
437
438 > func (c *lru) updateEntryRefCount(entry *entryImpl) { lru.go ×9
439 > if c.pin {
440 > entry.refCount++ lru.go ×1
441 > if entry.refCount == 1 {
442 > c.pinnedSize += entry.Size()
443 > metrics.CachePinnedUsage.With(c.metricsHandler).Record(float64(c.pinnedSize))
444 > }
445 }
446 }
447
448 > func (c *lru) Stop() { lru.go ×1
449 > c.loops.Cancel()
450 > }
451
452 > func (c *lru) bgEvictLoop(ctx context.Context) error { lru.go ×6
453 > ch, t := c.timeSource.NewTimer(c.backgroundEvict().LoopInterval)
454 > for {
455 > select {
456 > case <-ch:
457 > settings := c.backgroundEvict()
458 > if settings.Enabled {
459 > c.bgEvict(settings)
460 > }
461 > t.Reset(settings.LoopInterval)
462 > case <-ctx.Done(): lru.go ×1
463 > return ctx.Err()
464 }
465 }
466 }
467
468 > func (c *lru) bgEvict(settings dynamicconfig.CacheBackgroundEvictSettings) { lru.go ×6
469 > now := c.timeSource.Now().UTC()
470 >
471 > // Limit each iteration to scanning MaxEntryPerCall entries, to avoid holding the cache lock for too long.
472 > evictToMax := func() (again bool) {
473 > c.mut.Lock()
474 > defer c.mut.Unlock()
475 >
476 > element := c.byAccess.Back()
477 > if settings.MaxEntryPerCall <= 0 {
478 return false
479 }
480 > for n := 0; n < settings.MaxEntryPerCall; n++ { lru.go ×6
481 > if element == nil {
482 > return false
483 > }
484 > elementPrev := element.Prev()
485 > entry := element.Value.(*entryImpl) // nolint:revive
486 > if !c.isEntryExpired(entry, now) {
487 > return false
488 > }
489 > c.deleteInternal(element)
490 > element = elementPrev
491 }
492 > return element != nil lru.go ×6
493 }
494
495 > for evictToMax() { lru.go ×6
496 > }
497 }