From 0063d92a76324e4404fc9b4f09ddd5b19f1ede71 Mon Sep 17 00:00:00 2001 From: Thomas Waldmann Date: Sun, 2 Aug 2026 14:24:56 +0200 Subject: [PATCH] Store: serialize all operations with an internal lock, fixes #206 A Store instance can now be shared between threads. Neither the backends (e.g. one paramiko sftp session, one requests session for rest) nor the Store's own bookkeeping (the _stats Counter, the writethrough cache accounting) are safe under concurrent calls, so all operations now take a store-level RLock (reentrant, because operations nest: create_levels uses "with self:", load/store/... call find). list() stays a lazy generator: the lock is only held while fetching the next item, not across the whole iteration, so other threads' operations interleave with a long listing and the iterating thread itself can do store operations inside its listing loop without deadlocking. Serialization is per operation: multi-operation sequences that need to be atomic against other threads must still be coordinated by the caller. The uncontended lock costs ~100ns per operation, noise compared to any backend call, so the locking is unconditional rather than opt-in. This is needed by borgbackup/borg#9988: borg2's PackWriter gets a background store-thread that stores full packs while the main thread assembles the next pack - and keeps using the same Store (lock refresh, reads of already stored packs, index writes) in the meantime. Tests: a serialization-asserting backend wrapper (fails when two backend calls overlap - verified to trigger without the lock), a list() laziness / interleaving test, and a stats lost-update test. --- docs/changes.rst | 8 ++ src/borgstore/store.py | 63 ++++++++++++++- tests/test_threading.py | 169 ++++++++++++++++++++++++++++++++++++++++ 3 files changed, 238 insertions(+), 2 deletions(-) create mode 100644 tests/test_threading.py diff --git a/docs/changes.rst b/docs/changes.rst index 0067058..a867cfc 100644 --- a/docs/changes.rst +++ b/docs/changes.rst @@ -6,6 +6,14 @@ Version 0.5.6 (not released yet) New features: +- Store: thread safety, #206. A Store instance can now be shared between threads: + all operations are serialized by an internal lock, so the backend (e.g. one + sftp/rest session) and the store's own bookkeeping (stats, cache) never see + concurrent calls. list() stays a lazy generator: the lock is only held while + fetching the next item, so other threads' operations interleave with a long + listing. Serialization is per operation; multi-operation atomicity remains the + caller's responsibility. This enables callers like borg to call into the store + from a background thread (e.g. borgbackup/borg#9988's pack store-thread). - hash / defrag: support the "blake3" algorithm (in addition to all hashlib algorithms). Needs the optional "blake3" package: pip install 'borgstore[blake3]'. For backends that hash server-side, it needs to be installed on the server. diff --git a/src/borgstore/store.py b/src/borgstore/store.py index b0f227f..1986361 100644 --- a/src/borgstore/store.py +++ b/src/borgstore/store.py @@ -7,14 +7,17 @@ - configurable nesting - recursive list method - soft deletion +- thread safety (one operation at a time, see Store docstring) """ from binascii import hexlify from collections import Counter from contextlib import contextmanager import enum +from functools import wraps import logging import os +import threading import time from typing import Iterator, NamedTuple, Optional @@ -83,7 +86,31 @@ def get_backend(url, permissions=None, quota=None): return backend +def _locked(method): + """Decorator: run the Store method while holding the store's lock, see Store docstring.""" + + @wraps(method) + def wrapper(self, *args, **kwargs): + with self._lock: + return method(self, *args, **kwargs) + + return wrapper + + class Store: + """ + High-level key/value store, using a backend for the actual storage. + + Thread safety: a Store instance may be shared between threads, all operations are + serialized by an internal lock (backends and the Store's own bookkeeping - stats, + cache - are not thread-safe themselves, e.g. one sftp/rest session), #206. + list() is special: it stays a lazy generator, the lock is only held while fetching + the next item, so other threads' operations can interleave with a long listing + (and the listing thread itself can do store operations inside its loop). + Serialization is per operation - multi-operation sequences that need to be atomic + against other threads must be coordinated by the caller. + """ + def __init__( self, url: Optional[str] = None, @@ -94,6 +121,10 @@ def __init__( cache_url: Optional[str] = None, cache_backend: Optional[BackendBase] = None, ): + # serializes all operations of this store, see the class docstring. + # reentrant, because operations nest (e.g. create_levels uses "with self:", + # load/store/... call find). created first: some @_locked methods run in __init__. + self._lock = threading.RLock() self.url = url if backend is None and url is not None: backend = get_backend(url, permissions=permissions) @@ -176,6 +207,7 @@ def _cache_policy_for(self, name: str) -> CachePolicy: return policy return CachePolicy(mode=CacheMode.C_OFF, max_age=None, size=None) + @_locked def set_levels(self, levels: dict, create: bool = False) -> None: if not levels or not isinstance(levels, dict): raise ValueError("No or invalid levels configuration given.") @@ -184,6 +216,7 @@ def set_levels(self, levels: dict, create: bool = False) -> None: if create: self.create_levels() + @_locked def create_levels(self): """creating any needed namespaces / directory in advance""" # doing that saves a lot of ad-hoc mkdir calls, which is especially important @@ -216,6 +249,7 @@ def create_levels(self): else: raise ValueError(f"Invalid levels: {namespace}: {levels}") + @_locked def create(self) -> None: self.backend.create() if self.cache_backend is not None and not self._cache_disabled: @@ -223,6 +257,7 @@ def create(self) -> None: if self.backend.precreate_dirs: self.create_levels() + @_locked def destroy(self) -> None: self.backend.destroy() if self.cache_backend is not None: @@ -236,6 +271,7 @@ def __exit__(self, exc_type, exc_val, exc_tb): self.close() return False + @_locked def open(self) -> None: self.backend.open() if self.cache_backend is not None and not self._cache_disabled: @@ -247,6 +283,7 @@ def open(self) -> None: else: self._cache_cleanup_expired() + @_locked def close(self) -> None: self.backend.close() if self.cache_backend is not None: @@ -257,6 +294,7 @@ def close(self) -> None: except Exception as err: logger.warning(f"borgstore: cache close failed: {err!r}") + @_locked def quota(self) -> dict: return self.backend.quota() @@ -298,6 +336,7 @@ def _stats_get_volume(self, key): return self._stats.get(f"{key}_volume", 0) @property + @_locked def stats(self): """ Return statistics such as method call counters, overall time [s], overall data volume, and overall throughput. @@ -347,6 +386,7 @@ def _get_levels(self, name): # Store.create_levels requires all namespaces to be configured in self.levels. raise KeyError(f"no matching namespace found for: {name}") + @_locked def find(self, name: str, *, deleted=False) -> str: """ Find an item checking all supported nesting levels and return its nested name: @@ -376,6 +416,7 @@ def find(self, name: str, *, deleted=False) -> str: break return nested_name + @_locked def info(self, name: str, *, deleted=False) -> ItemInfo: with self._stats_updater("info", f"info({name!r}, deleted={deleted})"): return self._backend_call(lambda: self.backend.info(self.find(name, deleted=deleted)), volume=0) @@ -397,6 +438,7 @@ def _cache_load(self, nested_name: str, *, size=None, offset=0) -> Optional[byte self._stats["cache_load_volume"] += len(value) return value + @_locked def load(self, name: str, *, size=None, offset=0, deleted=False) -> bytes: with self._stats_updater("load", f"load({name!r}, offset={offset}, size={size}, deleted={deleted})"): cache_policy = self._cache_policy_for(name) @@ -444,6 +486,7 @@ def _cache_store(self, nested_name: str, value: StoreValue) -> None: logger.warning(f"borgstore: cache store failed for {nested_name!r}: {err!r}") self._stats["cache_errors"] += 1 + @_locked def store(self, name: str, value: StoreValue) -> None: """ store into item . @@ -476,6 +519,7 @@ def _cache_delete(self, nested_name: str) -> None: logger.warning(f"borgstore: cache delete failed for {nested_name!r}: {err!r}") self._stats["cache_errors"] += 1 + @_locked def delete(self, name: str, *, deleted=False) -> None: """ Really and immediately deletes an item. @@ -488,6 +532,7 @@ def delete(self, name: str, *, deleted=False) -> None: if self._cache_policy_for(name).mode in {CacheMode.C_WRITETHROUGH, CacheMode.C_MIRROR}: self._cache_delete(nested_name) + @_locked def cache_invalidate(self, name: str, *, deleted: bool = False) -> None: """ Invalidate cached items. @@ -534,6 +579,7 @@ def _cache_move(self, old_nested: str, new_nested: str) -> None: logger.warning(f"borgstore: cache move failed for {old_nested!r}->{new_nested!r}: {err!r}") self._stats["cache_errors"] += 1 + @_locked def move( self, name: str, @@ -596,13 +642,24 @@ def list(self, name: str, deleted: bool = False) -> Iterator[ItemInfo]: Note: list bypasses the cache and always queries the primary backend to ensure we only return items that really exist there, even if other clients have updated or deleted items directly in the primary backend. + + Note: the store's lock is only held while fetching the next item, not across the + whole iteration, so other threads' operations (and the iterating thread's own + operations inside its loop) interleave with a long listing, see the class docstring. """ # we need this wrapper due to the recursion - we only want to increment list_calls once: logger.debug(f"borgstore: list_start({name!r}, deleted={deleted})") - self._stats["list_calls"] += 1 + with self._lock: + self._stats["list_calls"] += 1 + inner = self._list(name, deleted=deleted) count = 0 try: - for info in self._list(name, deleted=deleted): + while True: + with self._lock: + try: + info = next(inner) + except StopIteration: + break count += 1 yield info finally: @@ -641,6 +698,7 @@ def _list(self, name: str, deleted: bool = False) -> Iterator[ItemInfo]: elif not deleted and not is_deleted: yield info + @_locked def hash(self, name: str, algorithm: str = "sha256", *, deleted: bool = False) -> str: """ compute the hex digest of the content of item using . @@ -654,6 +712,7 @@ def hash(self, name: str, algorithm: str = "sha256", *, deleted: bool = False) - lambda: self.backend.hash(self.find(name, deleted=deleted), algorithm=algorithm), volume=0 ) + @_locked def defrag(self, sources, *, target=None, algorithm=None, namespace=None, deleted=False) -> str: """ efficiently create a new item (target) by combining blocks from existing items (sources) diff --git a/tests/test_threading.py b/tests/test_threading.py new file mode 100644 index 0000000..02c3ff9 --- /dev/null +++ b/tests/test_threading.py @@ -0,0 +1,169 @@ +""" +Tests for Store thread safety (#206): all operations of a Store shared between threads +are serialized by an internal lock, so the (not thread-safe) backend never sees +concurrent calls. list() stays lazy: the lock is only held per fetched item. +""" + +import threading +import time + +import pytest + +from . import key +from borgstore.backends.posixfs import PosixFS +from borgstore.store import Store + +CONFIG = {"zero/": {"levels": [0]}} + + +class SerializationAssertingBackend: + """Wraps a backend; every call checks that no other backend call is in progress. + + The Store's lock must make concurrent backend calls impossible; without it, the small + sleep inside each call makes overlapping calls from multiple threads very likely. + """ + + def __init__(self, backend): + self._backend = backend + self._busy = False + self.violations = 0 + self.calls = 0 + + def _enter(self): + if self._busy: + self.violations += 1 + self._busy = True + self.calls += 1 + time.sleep(0.0005) # widen the race window, so unserialized calls actually overlap + + def _exit(self): + self._busy = False + + def __getattr__(self, name): + attr = getattr(self._backend, name) + if not callable(attr): + return attr # e.g. precreate_dirs + + if name == "list": + # the backend list generator runs stepwise; guard each step, not the whole iteration. + def guarded_list(*args, **kwargs): + inner = attr(*args, **kwargs) + while True: + self._enter() + try: + info = next(inner) + except StopIteration: + break + finally: + self._exit() + yield info + + return guarded_list + + def guarded(*args, **kwargs): + self._enter() + try: + return attr(*args, **kwargs) + finally: + self._exit() + + return guarded + + +@pytest.fixture() +def asserting_store(tmp_path): + backend = SerializationAssertingBackend(PosixFS(tmp_path / "store")) + store = Store(backend=backend, config=CONFIG) + store.create() + with store: + yield store, backend + store.destroy() + + +def test_concurrent_ops_are_serialized(asserting_store): + store, backend = asserting_store + errors = [] + + def hammer(thread_no): + try: + for i in range(20): + k = f"zero/{key(thread_no * 1000 + i)}" + value = f"{thread_no}-{i}".encode() + store.store(k, value) + assert store.load(k) == value + store.info(k) + if i % 5 == 0: + list(store.list("zero")) + store.delete(k) + except Exception as exc: # raising in a thread would go unnoticed by pytest + errors.append(exc) + + threads = [threading.Thread(target=hammer, args=(n,)) for n in range(4)] + for t in threads: + t.start() + for t in threads: + t.join() + assert not errors + assert backend.calls > 0 + assert backend.violations == 0 # no backend call overlapped another one + + +def test_list_is_lazy_and_interleaves(asserting_store): + store, backend = asserting_store + for i in range(50): + store.store(f"zero/{key(i)}", b"value") + + # the iterating thread itself can do store operations inside its listing loop + # (the lock is not held while the generator is suspended): + seen = 0 + for info in store.list("zero"): + seen += 1 + store.info(f"zero/{info.name}") + assert seen == 50 + + # ... and another thread's operations interleave with a long listing: + listing_started = threading.Event() + other_done = [] + + def other_thread(): + listing_started.wait() + for i in range(10): + k = f"zero/{key(100 + i)}" + store.store(k, b"other") + store.delete(k) + other_done.append(True) + + t = threading.Thread(target=other_thread) + t.start() + seen = 0 + for _info in store.list("zero"): + seen += 1 + if seen == 1: + listing_started.set() + time.sleep(0.001) # keep the listing running while the other thread works + t.join() + assert seen >= 50 # the other thread's short-lived items may or may not be seen + assert other_done == [True] + assert backend.violations == 0 + + +def test_stats_safe_under_concurrency(asserting_store): + store, backend = asserting_store + n_threads, n_ops = 4, 25 + + def hammer(thread_no): + for i in range(n_ops): + k = f"zero/{key(thread_no * 1000 + i)}" + store.store(k, b"x") + store.load(k) + + threads = [threading.Thread(target=hammer, args=(n,)) for n in range(n_threads)] + for t in threads: + t.start() + for t in threads: + t.join() + stats = store.stats + # without serialization, the lost-update races on the Counter would lose increments. + assert stats["store_calls"] == n_threads * n_ops + assert stats["load_calls"] == n_threads * n_ops + assert backend.violations == 0