Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/dulwich/lru_cache.py: 21%
Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1# lru_cache.py -- Simple LRU cache for dulwich
2# Copyright (C) 2006, 2008 Canonical Ltd
3# Copyright (C) 2022 Jelmer Vernooij <jelmer@jelmer.uk>
4#
5# SPDX-License-Identifier: Apache-2.0 OR GPL-2.0-or-later
6# Dulwich is dual-licensed under the Apache License, Version 2.0 and the GNU
7# General Public License as published by the Free Software Foundation; version 2.0
8# or (at your option) any later version. You can redistribute it and/or
9# modify it under the terms of either of these two licenses.
10#
11# Unless required by applicable law or agreed to in writing, software
12# distributed under the License is distributed on an "AS IS" BASIS,
13# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14# See the License for the specific language governing permissions and
15# limitations under the License.
16#
17# You should have received a copy of the licenses; if not, see
18# <http://www.gnu.org/licenses/> for a copy of the GNU General Public License
19# and <http://www.apache.org/licenses/LICENSE-2.0> for a copy of the Apache
20# License, Version 2.0.
21#
23"""A simple least-recently-used (LRU) cache."""
25from collections.abc import Iterable, Iterator
26from typing import Callable, Generic, Optional, TypeVar, Union, cast
28_null_key = object()
31K = TypeVar("K")
32V = TypeVar("V")
35class _LRUNode(Generic[K, V]):
36 """This maintains the linked-list which is the lru internals."""
38 __slots__ = ("cleanup", "key", "next_key", "prev", "size", "value")
40 prev: Optional["_LRUNode[K, V]"]
41 next_key: Union[K, object]
42 size: Optional[int]
44 def __init__(
45 self, key: K, value: V, cleanup: Optional[Callable[[K, V], None]] = None
46 ) -> None:
47 self.prev = None
48 self.next_key = _null_key
49 self.key = key
50 self.value = value
51 self.cleanup = cleanup
52 # TODO: We could compute this 'on-the-fly' like we used to, and remove
53 # one pointer from this object, we just need to decide if it
54 # actually costs us much of anything in normal usage
55 self.size = None
57 def __repr__(self) -> str:
58 if self.prev is None:
59 prev_key = None
60 else:
61 prev_key = self.prev.key
62 return f"{self.__class__.__name__}({self.key!r} n:{self.next_key!r} p:{prev_key!r})"
64 def run_cleanup(self) -> None:
65 if self.cleanup is not None:
66 self.cleanup(self.key, self.value)
67 self.cleanup = None
68 # Just make sure to break any refcycles, etc
69 del self.value
72class LRUCache(Generic[K, V]):
73 """A class which manages a cache of entries, removing unused ones."""
75 _least_recently_used: Optional[_LRUNode[K, V]]
76 _most_recently_used: Optional[_LRUNode[K, V]]
78 def __init__(
79 self, max_cache: int = 100, after_cleanup_count: Optional[int] = None
80 ) -> None:
81 """Initialize LRUCache.
83 Args:
84 max_cache: Maximum number of entries to cache
85 after_cleanup_count: Number of entries to keep after cleanup
86 """
87 self._cache: dict[K, _LRUNode[K, V]] = {}
88 # The "HEAD" of the lru linked list
89 self._most_recently_used = None
90 # The "TAIL" of the lru linked list
91 self._least_recently_used = None
92 self._update_max_cache(max_cache, after_cleanup_count)
94 def __contains__(self, key: K) -> bool:
95 """Check if key is in cache."""
96 return key in self._cache
98 def __getitem__(self, key: K) -> V:
99 """Get item from cache and mark as recently used."""
100 cache = self._cache
101 node = cache[key]
102 # Inlined from _record_access to decrease the overhead of __getitem__
103 # We also have more knowledge about structure if __getitem__ is
104 # succeeding, then we know that self._most_recently_used must not be
105 # None, etc.
106 mru = self._most_recently_used
107 if node is mru:
108 # Nothing to do, this node is already at the head of the queue
109 return node.value
110 # Remove this node from the old location
111 node_prev = node.prev
112 next_key = node.next_key
113 # benchmarking shows that the lookup of _null_key in globals is faster
114 # than the attribute lookup for (node is self._least_recently_used)
115 if next_key is _null_key:
116 # 'node' is the _least_recently_used, because it doesn't have a
117 # 'next' item. So move the current lru to the previous node.
118 self._least_recently_used = node_prev
119 else:
120 node_next = cache[cast(K, next_key)]
121 node_next.prev = node_prev
122 assert node_prev
123 assert mru
124 node_prev.next_key = next_key
125 # Insert this node at the front of the list
126 node.next_key = mru.key
127 mru.prev = node
128 self._most_recently_used = node
129 node.prev = None
130 return node.value
132 def __len__(self) -> int:
133 """Return number of items in cache."""
134 return len(self._cache)
136 def _walk_lru(self) -> Iterator[_LRUNode[K, V]]:
137 """Walk the LRU list, only meant to be used in tests."""
138 node = self._most_recently_used
139 if node is not None:
140 if node.prev is not None:
141 raise AssertionError(
142 "the _most_recently_used entry is not"
143 " supposed to have a previous entry"
144 f" {node}"
145 )
146 while node is not None:
147 if node.next_key is _null_key:
148 if node is not self._least_recently_used:
149 raise AssertionError(
150 f"only the last node should have no next value: {node}"
151 )
152 node_next = None
153 else:
154 node_next = self._cache[cast(K, node.next_key)]
155 if node_next.prev is not node:
156 raise AssertionError(
157 f"inconsistency found, node.next.prev != node: {node}"
158 )
159 if node.prev is None:
160 if node is not self._most_recently_used:
161 raise AssertionError(
162 "only the _most_recently_used should"
163 f" not have a previous node: {node}"
164 )
165 else:
166 if node.prev.next_key != node.key:
167 raise AssertionError(
168 f"inconsistency found, node.prev.next != node: {node}"
169 )
170 yield node
171 node = node_next
173 def add(
174 self, key: K, value: V, cleanup: Optional[Callable[[K, V], None]] = None
175 ) -> None:
176 """Add a new value to the cache.
178 Also, if the entry is ever removed from the cache, call
179 cleanup(key, value).
181 Args:
182 key: The key to store it under
183 value: The object to store
184 cleanup: None or a function taking (key, value) to indicate
185 'value' should be cleaned up.
186 """
187 if key is _null_key:
188 raise ValueError("cannot use _null_key as a key")
189 if key in self._cache:
190 node = self._cache[key]
191 node.run_cleanup()
192 node.value = value
193 node.cleanup = cleanup
194 else:
195 node = _LRUNode(key, value, cleanup=cleanup)
196 self._cache[key] = node
197 self._record_access(node)
199 if len(self._cache) > self._max_cache:
200 # Trigger the cleanup
201 self.cleanup()
203 def cache_size(self) -> int:
204 """Get the number of entries we will cache."""
205 return self._max_cache
207 def get(self, key: K, default: Optional[V] = None) -> Optional[V]:
208 """Get value from cache with default if not found.
210 Args:
211 key: Key to look up
212 default: Default value if key not found
214 Returns:
215 Value from cache or default
216 """
217 node = self._cache.get(key, None)
218 if node is None:
219 return default
220 self._record_access(node)
221 return node.value
223 def keys(self) -> Iterable[K]:
224 """Get the list of keys currently cached.
226 Note that values returned here may not be available by the time you
227 request them later. This is simply meant as a peak into the current
228 state.
230 Returns: An unordered list of keys that are currently cached.
231 """
232 return self._cache.keys()
234 def items(self) -> dict[K, V]:
235 """Get the key:value pairs as a dict."""
236 return {k: n.value for k, n in self._cache.items()}
238 def cleanup(self) -> None:
239 """Clear the cache until it shrinks to the requested size.
241 This does not completely wipe the cache, just makes sure it is under
242 the after_cleanup_count.
243 """
244 # Make sure the cache is shrunk to the correct size
245 while len(self._cache) > self._after_cleanup_count:
246 self._remove_lru()
248 def __setitem__(self, key: K, value: V) -> None:
249 """Add a value to the cache, there will be no cleanup function."""
250 self.add(key, value, cleanup=None)
252 def _record_access(self, node: _LRUNode[K, V]) -> None:
253 """Record that key was accessed."""
254 # Move 'node' to the front of the queue
255 if self._most_recently_used is None:
256 self._most_recently_used = node
257 self._least_recently_used = node
258 return
259 elif node is self._most_recently_used:
260 # Nothing to do, this node is already at the head of the queue
261 return
262 # We've taken care of the tail pointer, remove the node, and insert it
263 # at the front
264 # REMOVE
265 if node is self._least_recently_used:
266 self._least_recently_used = node.prev
267 if node.prev is not None:
268 node.prev.next_key = node.next_key
269 if node.next_key is not _null_key:
270 node_next = self._cache[cast(K, node.next_key)]
271 node_next.prev = node.prev
272 # INSERT
273 node.next_key = self._most_recently_used.key
274 self._most_recently_used.prev = node
275 self._most_recently_used = node
276 node.prev = None
278 def _remove_node(self, node: _LRUNode[K, V]) -> None:
279 if node is self._least_recently_used:
280 self._least_recently_used = node.prev
281 self._cache.pop(node.key)
282 # If we have removed all entries, remove the head pointer as well
283 if self._least_recently_used is None:
284 self._most_recently_used = None
285 node.run_cleanup()
286 # Now remove this node from the linked list
287 if node.prev is not None:
288 node.prev.next_key = node.next_key
289 if node.next_key is not _null_key:
290 node_next = self._cache[cast(K, node.next_key)]
291 node_next.prev = node.prev
292 # And remove this node's pointers
293 node.prev = None
294 node.next_key = _null_key
296 def _remove_lru(self) -> None:
297 """Remove one entry from the lru, and handle consequences.
299 If there are no more references to the lru, then this entry should be
300 removed from the cache.
301 """
302 assert self._least_recently_used
303 self._remove_node(self._least_recently_used)
305 def clear(self) -> None:
306 """Clear out all of the cache."""
307 # Clean up in LRU order
308 while self._cache:
309 self._remove_lru()
311 def resize(self, max_cache: int, after_cleanup_count: Optional[int] = None) -> None:
312 """Change the number of entries that will be cached."""
313 self._update_max_cache(max_cache, after_cleanup_count=after_cleanup_count)
315 def _update_max_cache(
316 self, max_cache: int, after_cleanup_count: Optional[int] = None
317 ) -> None:
318 self._max_cache = max_cache
319 if after_cleanup_count is None:
320 self._after_cleanup_count = self._max_cache * 8 / 10
321 else:
322 self._after_cleanup_count = min(after_cleanup_count, self._max_cache)
323 self.cleanup()
326class LRUSizeCache(LRUCache[K, V]):
327 """An LRUCache that removes things based on the size of the values.
329 This differs in that it doesn't care how many actual items there are,
330 it just restricts the cache to be cleaned up after so much data is stored.
332 The size of items added will be computed using compute_size(value), which
333 defaults to len() if not supplied.
334 """
336 _compute_size: Callable[[V], int]
338 def __init__(
339 self,
340 max_size: int = 1024 * 1024,
341 after_cleanup_size: Optional[int] = None,
342 compute_size: Optional[Callable[[V], int]] = None,
343 ) -> None:
344 """Create a new LRUSizeCache.
346 Args:
347 max_size: The max number of bytes to store before we start
348 clearing out entries.
349 after_cleanup_size: After cleaning up, shrink everything to this
350 size.
351 compute_size: A function to compute the size of the values. We
352 use a function here, so that you can pass 'len' if you are just
353 using simple strings, or a more complex function if you are using
354 something like a list of strings, or even a custom object.
355 The function should take the form "compute_size(value) => integer".
356 If not supplied, it defaults to 'len()'
357 """
358 self._value_size = 0
359 if compute_size is None:
360 self._compute_size = cast(Callable[[V], int], len)
361 else:
362 self._compute_size = compute_size
363 self._update_max_size(max_size, after_cleanup_size=after_cleanup_size)
364 LRUCache.__init__(self, max_cache=max(int(max_size / 512), 1))
366 def add(
367 self, key: K, value: V, cleanup: Optional[Callable[[K, V], None]] = None
368 ) -> None:
369 """Add a new value to the cache.
371 Also, if the entry is ever removed from the cache, call
372 cleanup(key, value).
374 Args:
375 key: The key to store it under
376 value: The object to store
377 cleanup: None or a function taking (key, value) to indicate
378 'value' should be cleaned up.
379 """
380 if key is _null_key:
381 raise ValueError("cannot use _null_key as a key")
382 node = self._cache.get(key, None)
383 value_len = self._compute_size(value)
384 if value_len >= self._after_cleanup_size:
385 # The new value is 'too big to fit', as it would fill up/overflow
386 # the cache all by itself
387 if node is not None:
388 # We won't be replacing the old node, so just remove it
389 self._remove_node(node)
390 if cleanup is not None:
391 cleanup(key, value)
392 return
393 if node is None:
394 node = _LRUNode(key, value, cleanup=cleanup)
395 self._cache[key] = node
396 else:
397 assert node.size is not None
398 self._value_size -= node.size
399 node.size = value_len
400 self._value_size += value_len
401 self._record_access(node)
403 if self._value_size > self._max_size:
404 # Time to cleanup
405 self.cleanup()
407 def cleanup(self) -> None:
408 """Clear the cache until it shrinks to the requested size.
410 This does not completely wipe the cache, just makes sure it is under
411 the after_cleanup_size.
412 """
413 # Make sure the cache is shrunk to the correct size
414 while self._value_size > self._after_cleanup_size:
415 self._remove_lru()
417 def _remove_node(self, node: _LRUNode[K, V]) -> None:
418 assert node.size is not None
419 self._value_size -= node.size
420 LRUCache._remove_node(self, node)
422 def resize(self, max_size: int, after_cleanup_size: Optional[int] = None) -> None:
423 """Change the number of bytes that will be cached."""
424 self._update_max_size(max_size, after_cleanup_size=after_cleanup_size)
425 max_cache = max(int(max_size / 512), 1)
426 self._update_max_cache(max_cache)
428 def _update_max_size(
429 self, max_size: int, after_cleanup_size: Optional[int] = None
430 ) -> None:
431 self._max_size = max_size
432 if after_cleanup_size is None:
433 self._after_cleanup_size = self._max_size * 8 // 10
434 else:
435 self._after_cleanup_size = min(after_cleanup_size, self._max_size)