From 710c272ac5100c536138f9203eab2fe6e9f5221f Mon Sep 17 00:00:00 2001 From: Thomas Waldmann Date: Wed, 5 Aug 2026 13:08:01 +0200 Subject: [PATCH 1/2] lrucache: assigning to a key that is present replaces its value LRUCache refused that with an assertion, so callers had to write if key in cache: cache.replace(key, value) else: cache[key] = value and a caller that just assigned crashed - which is a nasty failure mode for a cache: it is a correctness-neutral operation that suddenly is not. It bites especially when several threads use one cache (see the FUSE data cache), where two threads can miss on the same key and both insert the value they fetched. Assignment now replaces the value, disposing the old one (it left the cache, just like a deleted one would) unless it is the identical object, and counts as a use, so the entry moves to the most-recently-used end. replace() stays for the one case that really needs it: updating an entry while keeping its old value alive - the legacy repository's fd cache re-timestamps its entries that way and must not have the open file closed. Co-Authored-By: Claude Opus 5 --- src/borg/helpers/lrucache.py | 29 +++++++++++++++------ src/borg/testsuite/helpers/lrucache_test.py | 27 +++++++++++++++++++ 2 files changed, 48 insertions(+), 8 deletions(-) diff --git a/src/borg/helpers/lrucache.py b/src/borg/helpers/lrucache.py index a3bf1edb62..d06522bd84 100644 --- a/src/borg/helpers/lrucache.py +++ b/src/borg/helpers/lrucache.py @@ -9,8 +9,10 @@ class LRUCache(MutableMapping[K, V]): """ Mapping which maintains a maximum size by removing the least recently used value. - Items are passed to dispose before being removed and setting an item which is - already in the cache has to be done using the replace method. + + A value that leaves the cache - evicted, deleted, or replaced by another value under + the same key - is passed to *dispose* first, so that a cache of e.g. open files can + close them. Use replace() to update an entry while keeping its old value alive. """ _cache: OrderedDict[K, V] @@ -25,12 +27,17 @@ def __init__(self, capacity: int, dispose: Callable[[V], None] = lambda _: None) self._dispose = dispose def __setitem__(self, key: K, value: V) -> None: - assert ( - key not in self._cache - ), "Unexpected attempt to replace a cached item without first deleting the old item." + try: + previous = self._cache.pop(key) + except KeyError: + pass + else: + # the old value is no longer in the cache, so it is disposed like a deleted one. + if previous is not value: + self._dispose(previous) while len(self._cache) >= self._capacity: self._dispose(self._cache.popitem(last=False)[1]) - self._cache[key] = value # add new entry at the end + self._cache[key] = value # add the new (or refreshed) entry at the end def __getitem__(self, key: K) -> V: self._cache.move_to_end(key) # raise KeyError if not found @@ -46,8 +53,14 @@ def __len__(self) -> int: return len(self._cache) def replace(self, key: K, value: V) -> None: - """Replace an item that is already present, not disposing it in the process.""" - # this method complements __setitem__ which should be used for the normal use case. + """Replace the value of an entry that is present, without disposing the old value. + + This is for the rare case where the old value must stay alive, e.g. because the new + value still refers to it (the legacy repository's fd cache re-times-stamps its + entries this way, keeping the open file object). It also keeps the entry where it + is in the LRU order, i.e. it does not count as a use. Everything else should just + assign to the cache. + """ assert key in self._cache, "Unexpected attempt to update a non-existing item." self._cache[key] = value diff --git a/src/borg/testsuite/helpers/lrucache_test.py b/src/borg/testsuite/helpers/lrucache_test.py index 8897702f26..e152a3189f 100644 --- a/src/borg/testsuite/helpers/lrucache_test.py +++ b/src/borg/testsuite/helpers/lrucache_test.py @@ -35,6 +35,33 @@ def test_lrucache(self): c.clear() assert c.items() == set() + def test_assign_to_existing_key(self): + # assigning to a key that is present replaces the value and counts as a use, + # so the entry becomes the most recently used one. + c = LRUCache(2) + c["a"] = 1 + c["b"] = 2 + c["a"] = 3 # replaces, and refreshes "a" + assert len(c) == 2 + assert c["a"] == 3 + c["c"] = 4 # evicts the least recently used entry, which is "b" now + assert set(c.keys()) == {"a", "c"} + + def test_dispose_on_replacement(self): + disposed = [] + c = LRUCache(2, dispose=disposed.append) + c["a"] = "first" + c["a"] = "second" # the replaced value leaves the cache, so it is disposed + assert disposed == ["first"] + assert c["a"] == "second" + value = "same object" + c["b"] = value + c["b"] = value # assigning the identical object does not dispose it + assert disposed == ["first"] + c.replace("b", "replacement") # replace() never disposes, see there + assert disposed == ["first"] + assert c["b"] == "replacement" + def test_dispose(self): c = LRUCache(2, dispose=lambda f: f.close()) f1 = TemporaryFile() From fce88eeff4283aaf8265329ce5052cd92b6fbe93 Mon Sep 17 00:00:00 2001 From: Thomas Waldmann Date: Wed, 5 Aug 2026 13:26:03 +0200 Subject: [PATCH 2/2] lrucache: make it thread-safe Several threads share one LRUCache in borg: mfusepy serves the FUSE requests of one mount from a thread pool, and "borg webdav" from one thread per connection. Until now every caller had to serialize the access itself - a rule that is easy to miss, and missing it does not fail loudly at the cache, it fails somewhere far away (mfusepy turns any exception from a FUSE operation into EINVAL, so a mount just reports "Invalid argument" for a healthy file). Getting, setting, deleting, popping, replacing and clearing now each hold a lock, so they are atomic. A *sequence* of them still is not - two threads can miss on the same key and both store a value, which for a cache is wasteful, not wrong. pop() is implemented here now instead of inheriting it: MutableMapping builds it from __getitem__ + __delitem__, which is not atomic (two threads popping the same key raise KeyError, see the new test), and it would dispose the value it hands to the caller - for e.g. a cache of open files it returned a closed one. Cost, measured with timeit: about 75 ns per operation (get 70 -> 148 ns, set 148 -> 222 ns). That is nothing next to what these caches guard - a chunk decrypt is in the milliseconds, an item unpack in the microseconds - and a borg mount read benchmark (1 GiB sequential, and stat+read of 200 small files, mfusepy) shows no difference at all. Co-Authored-By: Claude Opus 5 --- src/borg/helpers/lrucache.py | 71 +++++++++++++++------ src/borg/testsuite/helpers/lrucache_test.py | 49 ++++++++++++++ 2 files changed, 100 insertions(+), 20 deletions(-) diff --git a/src/borg/helpers/lrucache.py b/src/borg/helpers/lrucache.py index d06522bd84..7f280cd5f1 100644 --- a/src/borg/helpers/lrucache.py +++ b/src/borg/helpers/lrucache.py @@ -1,10 +1,13 @@ +import threading from collections import OrderedDict from collections.abc import Callable, ItemsView, Iterator, KeysView, MutableMapping, ValuesView -from typing import TypeVar +from typing import Any, TypeVar K = TypeVar("K") V = TypeVar("V") +_MISSING = object() + class LRUCache(MutableMapping[K, V]): """ @@ -13,6 +16,16 @@ class LRUCache(MutableMapping[K, V]): A value that leaves the cache - evicted, deleted, or replaced by another value under the same key - is passed to *dispose* first, so that a cache of e.g. open files can close them. Use replace() to update an entry while keeping its old value alive. + + Thread safety: getting, setting, deleting, popping, replacing and clearing are each + atomic, so several threads can share one cache without corrupting it. A *sequence* of + them is not: two threads can miss on the same key and both compute and store a value, + which for a cache is wasteful, not wrong - and the mixin methods built on top of the + above (setdefault, popitem, update) inherit that. Iterating (also via keys(), values(), + items()) yields a live view of the underlying dict, so it needs external + synchronization if another thread may write meanwhile. + + *dispose* is called while the lock is held, so it must not use the cache. """ _cache: OrderedDict[K, V] @@ -25,26 +38,30 @@ def __init__(self, capacity: int, dispose: Callable[[V], None] = lambda _: None) self._cache = OrderedDict() self._capacity = capacity self._dispose = dispose + self._lock = threading.Lock() def __setitem__(self, key: K, value: V) -> None: - try: - previous = self._cache.pop(key) - except KeyError: - pass - else: - # the old value is no longer in the cache, so it is disposed like a deleted one. - if previous is not value: - self._dispose(previous) - while len(self._cache) >= self._capacity: - self._dispose(self._cache.popitem(last=False)[1]) - self._cache[key] = value # add the new (or refreshed) entry at the end + with self._lock: + try: + previous = self._cache.pop(key) + except KeyError: + pass + else: + # the old value is no longer in the cache, so it is disposed like a deleted one. + if previous is not value: + self._dispose(previous) + while len(self._cache) >= self._capacity: + self._dispose(self._cache.popitem(last=False)[1]) + self._cache[key] = value # add the new (or refreshed) entry at the end def __getitem__(self, key: K) -> V: - self._cache.move_to_end(key) # raise KeyError if not found - return self._cache[key] + with self._lock: + self._cache.move_to_end(key) # raise KeyError if not found + return self._cache[key] def __delitem__(self, key: K) -> None: - self._dispose(self._cache.pop(key)) + with self._lock: + self._dispose(self._cache.pop(key)) def __contains__(self, key: object) -> bool: return key in self._cache @@ -52,6 +69,18 @@ def __contains__(self, key: object) -> bool: def __len__(self) -> int: return len(self._cache) + def pop(self, key: K, default: Any = _MISSING) -> Any: + """Remove *key* and return its value, without disposing it.""" + # MutableMapping would implement this as __getitem__ + __delitem__, which is not + # atomic (and would dispose the value). + with self._lock: + try: + return self._cache.pop(key) + except KeyError: + if default is _MISSING: + raise + return default + def replace(self, key: K, value: V) -> None: """Replace the value of an entry that is present, without disposing the old value. @@ -61,13 +90,15 @@ def replace(self, key: K, value: V) -> None: is in the LRU order, i.e. it does not count as a use. Everything else should just assign to the cache. """ - assert key in self._cache, "Unexpected attempt to update a non-existing item." - self._cache[key] = value + with self._lock: + assert key in self._cache, "Unexpected attempt to update a non-existing item." + self._cache[key] = value def clear(self) -> None: - for value in self._cache.values(): - self._dispose(value) - self._cache.clear() + with self._lock: + for value in self._cache.values(): + self._dispose(value) + self._cache.clear() def __iter__(self) -> Iterator[K]: return iter(self._cache) diff --git a/src/borg/testsuite/helpers/lrucache_test.py b/src/borg/testsuite/helpers/lrucache_test.py index e152a3189f..1462098bfa 100644 --- a/src/borg/testsuite/helpers/lrucache_test.py +++ b/src/borg/testsuite/helpers/lrucache_test.py @@ -1,3 +1,5 @@ +import sys +import threading from tempfile import TemporaryFile import pytest @@ -81,3 +83,50 @@ def test_dispose(self): c.clear() assert c.items() == set() assert f3.closed + + def test_pop(self): + disposed = [] + c = LRUCache(2, dispose=disposed.append) + c["a"] = 1 + assert c.pop("a") == 1 # pop hands the value to the caller, so it is not disposed + assert disposed == [] + assert "a" not in c + assert c.pop("a", "default") == "default" + with pytest.raises(KeyError): + c.pop("a") + + def test_threaded_access(self): + # several threads sharing one cache must not corrupt it or raise: they hit the same + # keys, so they race on inserting, refreshing, evicting and removing the same entries. + # e.g. popping via MutableMapping (get, then delete) raises KeyError here when + # another thread removes the key in between. + c: LRUCache = LRUCache(8) + keys = list(range(16)) # more keys than capacity, so entries keep getting evicted + errors: list[Exception] = [] + start = threading.Barrier(8) + + def worker(n): + start.wait() + try: + for _round in range(2000): + for key in keys: + c[key] = n + c.get(key) + c.pop(key, None) + except Exception as e: + errors.append(e) + + # the default switch interval (5 ms) is longer than these loops, so threads would + # hardly ever switch inside one cache operation and the races would not show up. + previous_interval = sys.getswitchinterval() + sys.setswitchinterval(1e-6) + try: + threads = [threading.Thread(target=worker, args=(n,)) for n in range(8)] + for t in threads: + t.start() + for t in threads: + t.join() + finally: + sys.setswitchinterval(previous_interval) + assert errors == [] + assert len(c) <= 8