Repository navigation
Expand file tree
/
Copy pathlru.py
More file actions
123 lines (104 loc) · 4.24 KB
/
Copy pathlru.py
File metadata and controls
123 lines (104 loc) · 4.24 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
import threading
import typing
from collections.abc import Iterable, Iterator, MutableMapping
from .misc import SENTINEL
__all__ = ['LRU']
K = typing.TypeVar('K')
V = typing.TypeVar('V')
class LRU(MutableMapping[K, V], typing.Generic[K, V]):
"""
Implementation of a length-limited LRU map.
The mapping is thread-safe, and internally uses a lock to avoid concurrency
issues. However, access operations like ``lru[key]`` are fast and
lock-free.
"""
__slots__ = ('_count', '_lock', '_ordering', '_values')
def __init__(self, count: int, pairs: Iterable[tuple[K, V]] = ()):
assert count > 0, "LRU needs a positive count"
self._count = count
self._lock = threading.RLock()
self._values: dict[K, V] = {}
#
# The dict self._values contains the LRU items, while self._ordering
# only keeps track of their order, the most recently used ones being
# last. For performance reasons, we only use the lock when modifying
# the LRU, while reading it is lock-free (and thus faster).
#
# This strategy may result in inconsistencies between self._values and
# self._ordering. Indeed, concurrently accessed keys may be missing
# from self._ordering, but will eventually be added. This could result
# in keys being added back in self._ordering after their actual removal
# from the LRU. This results in the following invariant:
#
# self._values <= self._ordering | "keys being accessed"
#
self._ordering: dict[K, None] = {}
# Initialize
for key, value in pairs:
self[key] = value
@property
def count(self) -> int:
return self._count
def __contains__(self, key: object) -> bool:
return key in self._values
def __getitem__(self, key: K) -> V:
val = self._values[key]
# move key at the last position in self._ordering
self._ordering[key] = self._ordering.pop(key, None)
return val
def __setitem__(self, key: K, val: V):
values = self._values
ordering = self._ordering
with self._lock:
values[key] = val
ordering[key] = ordering.pop(key, None)
while True:
# if we have too many keys in ordering, filter them out
if len(ordering) > len(values):
# (copy to avoid concurrent changes on ordering)
for k in ordering.copy():
if k not in values:
ordering.pop(k, None)
# check if we have too many keys
if len(values) <= self._count:
break
# if so, pop the least recently used
try:
# have a default in case of concurrent accesses
key = next(iter(ordering), key)
except RuntimeError:
# ordering modified during iteration, retry
continue
values.pop(key, None)
ordering.pop(key, None)
def __delitem__(self, key: K):
self.pop(key)
def __len__(self) -> int:
return len(self._values)
def __iter__(self) -> Iterator[K]:
return iter(self.snapshot)
@property
def snapshot(self) -> dict[K, V]:
""" Return a copy of the LRU (ordered according to LRU first). """
with self._lock:
values = self._values
# build result in expected order (copy self._ordering to avoid concurrent changes)
result = {
key: val
for key in self._ordering.copy()
if (val := values.get(key, SENTINEL)) is not SENTINEL
}
if len(result) < len(values):
# keys in value were missing from self._ordering, add them
result.update(values)
return result
def pop(self, key: K, /, default=SENTINEL) -> V:
with self._lock:
self._ordering.pop(key, None)
if default is SENTINEL:
return self._values.pop(key)
return self._values.pop(key, default)
def clear(self):
with self._lock:
self._ordering.clear()
self._values.clear()