Coverage for adhoc-cicd-odoo-odoo / odoo / tools / lru.py: 72%

1import threading 

2import typing 

3from collections.abc import Iterable, Iterator, MutableMapping 

4 

5from .misc import SENTINEL 

6 

7__all__ = ['LRU'] 

8 

9K = typing.TypeVar('K') 

10V = typing.TypeVar('V') 

11 

12 

13class LRU(MutableMapping[K, V], typing.Generic[K, V]): 

14 """ 

15 Implementation of a length-limited LRU map. 

16 

17 The mapping is thread-safe, and internally uses a lock to avoid concurrency 

18 issues. However, access operations like ``lru[key]`` are fast and 

19 lock-free. 

20 """ 

21 

22 __slots__ = ('_count', '_lock', '_ordering', '_values') 

23 

24 def __init__(self, count: int, pairs: Iterable[tuple[K, V]] = ()): 

25 assert count > 0, "LRU needs a positive count" 

26 self._count = count 

27 self._lock = threading.RLock() 

28 self._values: dict[K, V] = {} 

29 # 

30 # The dict self._values contains the LRU items, while self._ordering 

31 # only keeps track of their order, the most recently used ones being 

32 # last. For performance reasons, we only use the lock when modifying 

33 # the LRU, while reading it is lock-free (and thus faster). 

34 # 

35 # This strategy may result in inconsistencies between self._values and 

36 # self._ordering. Indeed, concurrently accessed keys may be missing 

37 # from self._ordering, but will eventually be added. This could result 

38 # in keys being added back in self._ordering after their actual removal 

39 # from the LRU. This results in the following invariant: 

40 # 

41 # self._values <= self._ordering | "keys being accessed" 

42 # 

43 self._ordering: dict[K, None] = {} 

44 

45 # Initialize 

46 for key, value in pairs: 46 ↛ 47line 46 didn't jump to line 47 because the loop on line 46 never started

47 self[key] = value 

48 

49 @property 

50 def count(self) -> int: 

51 return self._count 

52 

53 def __contains__(self, key: object) -> bool: 

54 return key in self._values 

55 

56 def __getitem__(self, key: K) -> V: 

57 val = self._values[key] 

58 # move key at the last position in self._ordering 

59 self._ordering[key] = self._ordering.pop(key, None) 

60 return val 

61 

62 def __setitem__(self, key: K, val: V): 

63 values = self._values 

64 ordering = self._ordering 

65 with self._lock: 

66 values[key] = val 

67 ordering[key] = ordering.pop(key, None) 

68 while True: 

69 # if we have too many keys in ordering, filter them out 

70 if len(ordering) > len(values): 70 ↛ 72line 70 didn't jump to line 72 because the condition on line 70 was never true

71 # (copy to avoid concurrent changes on ordering) 

72 for k in ordering.copy(): 

73 if k not in values: 

74 ordering.pop(k, None) 

75 # check if we have too many keys 

76 if len(values) <= self._count: 

77 break 

78 # if so, pop the least recently used 

79 try: 

80 # have a default in case of concurrent accesses 

81 key = next(iter(ordering), key) 

82 except RuntimeError: 

83 # ordering modified during iteration, retry 

84 continue 

85 values.pop(key, None) 

86 ordering.pop(key, None) 

87 

88 def __delitem__(self, key: K): 

89 self.pop(key) 

90 

91 def __len__(self) -> int: 

92 return len(self._values) 

93 

94 def __iter__(self) -> Iterator[K]: 

95 return iter(self.snapshot) 

96 

97 @property 

98 def snapshot(self) -> dict[K, V]: 

99 """ Return a copy of the LRU (ordered according to LRU first). """ 

100 with self._lock: 

101 values = self._values 

102 # build result in expected order (copy self._ordering to avoid concurrent changes) 

103 result = { 

104 key: val 

105 for key in self._ordering.copy() 

106 if (val := values.get(key, SENTINEL)) is not SENTINEL 

107 } 

108 if len(result) < len(values): 108 ↛ 110line 108 didn't jump to line 110 because the condition on line 108 was never true

109 # keys in value were missing from self._ordering, add them 

110 result.update(values) 

111 return result 

112 

113 def pop(self, key: K, /, default=SENTINEL) -> V: 

114 with self._lock: 

115 self._ordering.pop(key, None) 

116 if default is SENTINEL: 

117 return self._values.pop(key) 

118 return self._values.pop(key, default) 

119 

120 def clear(self): 

121 with self._lock: 

122 self._ordering.clear() 

123 self._values.clear()