diff --git a/src/borg/archiver/lock_cmds.py b/src/borg/archiver/lock_cmds.py index 971fb653e8..10c2751e1f 100644 --- a/src/borg/archiver/lock_cmds.py +++ b/src/borg/archiver/lock_cmds.py @@ -3,8 +3,9 @@ from ._common import with_repository from ..cache import Cache from ..constants import * # NOQA -from ..helpers import prepare_subprocess_env, set_ec, CommandError, ThreadRunner +from ..helpers import prepare_subprocess_env, set_ec, CommandError from ..helpers.argparsing import ArgumentParser, REMAINDER +from ..storelocking import LockRefresher from ..logger import create_logger @@ -17,7 +18,7 @@ def do_with_lock(self, args, repository): """Runs a user-specified command with the repository lock held.""" # the repository lock needs to get refreshed regularly, or it will be killed as stale. # refreshing the lock is not part of the repository API, so we do it indirectly via repository.info. - lock_refreshing_thread = ThreadRunner(sleep_interval=60, target=repository.info) + lock_refreshing_thread = LockRefresher(repository.info, sleep_interval=60) lock_refreshing_thread.start() env = prepare_subprocess_env(system=True) try: diff --git a/src/borg/fuse.py b/src/borg/fuse.py index 2f7eb9836f..cdfba58e17 100644 --- a/src/borg/fuse.py +++ b/src/borg/fuse.py @@ -12,6 +12,11 @@ asynchronous manner. However, as long as we do not use any asynchronous operations (like using "await") in this code, it is still de facto synchronous, and we are safe. + +The only exception is a background thread that periodically refreshes the +repository lock while the mount is idle (see #9872). borgstore connections are +not thread-safe, so all repository access from the FUSE handlers and the refresh +thread is serialized via self._repo_lock. """ import errno @@ -22,6 +27,7 @@ import struct import sys import tempfile +import threading import time from collections import defaultdict, Counter from signal import SIGINT @@ -65,6 +71,7 @@ def async_wrapper(fn): from .helpers import daemonizing, signal_handler, format_file_size, bin_to_hex, Error from .helpers import HardLinkManager from .helpers import msgpack +from .storelocking import LockRefresher from .helpers.lrucache import LRUCache from .item import Item from .platform import uid2user, gid2group @@ -296,6 +303,9 @@ def __init__(self, manifest, args, repository): # Archives to be loaded when first accessed, mapped by their placeholder inode self.pending_archives = {} self.cache = ItemCache(repository, self.repo_objs) + # serializes all repository access (FUSE handlers and the background lock-refresh + # thread), because borgstore connections are not thread-safe. see _lock_refresh. + self._repo_lock = threading.RLock() self.allow_damaged_files = False self.versions = False self.uid_forced = None @@ -334,7 +344,8 @@ def get_item(self, inode): return self._items[inode] except KeyError: # while self.cache does some internal caching, it has still quite some overhead, so we cache the result. - item = self.cache.get(inode) + with self._repo_lock: + item = self.cache.get(inode) self._inode_cache[inode] = item return item @@ -342,7 +353,8 @@ def check_pending_archive(self, inode): # Check if this is an archive we need to load archive_info = self.pending_archives.pop(inode, None) if archive_info is not None: - self._process_archive(archive_info.id, [os.fsencode(self.archive_root_dir[archive_info.id])]) + with self._repo_lock: + self._process_archive(archive_info.id, [os.fsencode(self.archive_root_dir[archive_info.id])]) def _allocate_inode(self): self.inode_count += 1 @@ -602,12 +614,17 @@ def pop_option(options, key, present, not_present, wanted_type, int_base=0): # job - seeing the mountpoint empty, rsync would delete everything in the # mirror. umount = False + # keep the repository lock of an idle mount alive, so it is not killed as stale (see #9872). + # started here (after a possible daemonizing fork, as threads do not survive fork()). + lock_refreshing_thread = LockRefresher(self.repository_uncached.info, sleep_interval=60, lock=self._repo_lock) + lock_refreshing_thread.start() try: with signal_handler("SIGUSR1", self.sig_info_handler), signal_handler("SIGINFO", self.sig_info_handler): signal = fuse_main() # no crash and no signal (or it's ^C and we're in the foreground) -> umount request umount = signal is None or (signal == SIGINT and foreground) finally: + lock_refreshing_thread.terminate() llfuse.close(umount) @async_wrapper @@ -721,7 +738,8 @@ def read(self, fh, offset, size): del self.data_cache[id] else: try: - cdata = self.repository_uncached.get(id) + with self._repo_lock: + cdata = self.repository_uncached.get(id) except Repository.ObjectNotFound: if self.allow_damaged_files: data = zeros[:s] diff --git a/src/borg/hlfuse.py b/src/borg/hlfuse.py index b5f106d0de..0e963327e4 100644 --- a/src/borg/hlfuse.py +++ b/src/borg/hlfuse.py @@ -3,6 +3,7 @@ import hashlib import os import stat +import threading import time from collections import Counter from typing import TYPE_CHECKING @@ -27,6 +28,7 @@ from .helpers import daemonizing, signal_handler, bin_to_hex, Error from .helpers import HardLinkManager from .helpers import msgpack +from .storelocking import LockRefresher from .helpers.lrucache import LRUCache from .item import Item from .platform import uid2user, gid2group @@ -87,6 +89,9 @@ def __init__(self, manifest, args, repository): self._manifest = manifest self.repo_objs = manifest.repo_objs self.repository = repository + # serializes all repository access (FUSE handlers and the background lock-refresh + # thread), because borgstore connections are not thread-safe. see _lock_refresh. + self._repo_lock = threading.RLock() self.default_uid = os.getuid() self.default_gid = os.getgid() @@ -171,7 +176,8 @@ def _create_filesystem(self): def check_pending_archive(self, node): archive_info = self.pending_archives.pop(node, None) if archive_info is not None: - self._process_archive(archive_info.id, node) + with self._repo_lock: + self._process_archive(archive_info.id, node) def _iter_archive_items(self, archive_item_ids, filter=None): unpacker = msgpack.Unpacker() @@ -532,9 +538,16 @@ def pop_option(options, key, present, not_present, wanted_type, int_base=0): logger.debug("fuse: mount repo, going to background: migrating lock.") self.repository.migrate_lock(old_id, new_id) - # Run the FUSE main loop in foreground (we might be daemonized already or not) - with signal_handler("SIGUSR1", self.sig_info_handler), signal_handler("SIGINFO", self.sig_info_handler): - hlfuse.FUSE(self, mountpoint, options, foreground=True, use_ino=True) + # keep the repository lock of an idle mount alive, so it is not killed as stale (see #9872). + # started here (after a possible daemonizing fork, as threads do not survive fork()). + lock_refreshing_thread = LockRefresher(self.repository.info, sleep_interval=60, lock=self._repo_lock) + lock_refreshing_thread.start() + try: + # Run the FUSE main loop in foreground (we might be daemonized already or not) + with signal_handler("SIGUSR1", self.sig_info_handler), signal_handler("SIGINFO", self.sig_info_handler): + hlfuse.FUSE(self, mountpoint, options, foreground=True, use_ino=True) + finally: + lock_refreshing_thread.terminate() def statfs(self, path): debug_log(f"statfs(path={path!r})") @@ -645,7 +658,8 @@ def read(self, path, size, offset, fi): else: try: # Direct repository access - cdata = self.repository.get(id) + with self._repo_lock: + cdata = self.repository.get(id) except Repository.ObjectNotFound: if self.allow_damaged_files: data = zeros[:s] diff --git a/src/borg/storelocking.py b/src/borg/storelocking.py index b407f77785..f781862383 100644 --- a/src/borg/storelocking.py +++ b/src/borg/storelocking.py @@ -2,6 +2,7 @@ import hashlib import json import random +import threading import time from borgstore.store import ObjectNotFound @@ -296,3 +297,57 @@ def refresh(self): logger.debug("LOCK-REFRESH: our lock was killed while refreshing it, no safe way to continue.") self._delete_lock(new_key, ignore_not_found=True, update_last_refresh=True) raise LockTimeout(str(self.store)) + + +class LockRefresher: + def __init__(self, refresh_callable, sleep_interval=60, lock=None): + """ + Periodically refreshes the repository lock in a background thread. + + :param refresh_callable: Callable (e.g. repository.info) to refresh the lock. + :param sleep_interval: Frequency of refresh attempts (in seconds). + :param lock: Optional reentrant lock (RLock) to serialize repository access. + """ + self.refresh_callable = refresh_callable + self.sleep_interval = sleep_interval + self.lock = lock + self._thread = None + self._keep_running = threading.Event() + self._keep_running.set() + + def _run(self): + while self._keep_running.is_set(): + try: + if self.lock is not None: + with self.lock: + self.refresh_callable() + else: + self.refresh_callable() + except LockTimeout: + logger.error("Lock refresh thread: lock has been killed/timed out. Terminating refresh.") + break + except Exception as e: + logger.warning(f"Lock refresh thread: temporary error during lock refresh: {e}. Will retry.") + + # sleep up to self.sleep_interval in small steps to allow quick shutdown + count = 1000 + micro_sleep = float(self.sleep_interval) / count + while self._keep_running.is_set() and count > 0: + time.sleep(micro_sleep) + count -= 1 + + def start(self): + # daemon=True so a refresh that is stuck in a blocking refresh_callable() (e.g. network + # I/O to a remote repo that never returns) can never keep the process alive on exit. + self._thread = threading.Thread(target=self._run, name="LockRefresher", daemon=True) + self._thread.start() + + def terminate(self): + if self._thread is not None: + self._keep_running.clear() + # bounded join: if refresh_callable() is wedged (e.g. a hung remote-repo request), + # don't let it stall shutdown - e.g. a FUSE unmount waiting for this. the thread is a + # daemon, so it is torn down at interpreter exit even if it is still alive here. + self._thread.join(timeout=5.0) + if self._thread.is_alive(): + logger.warning("Lock refresh thread did not stop within 5s; continuing shutdown.") diff --git a/src/borg/testsuite/archiver/mount_cmds_test.py b/src/borg/testsuite/archiver/mount_cmds_test.py index 2fcc91bba8..94ea24bd4e 100644 --- a/src/borg/testsuite/archiver/mount_cmds_test.py +++ b/src/borg/testsuite/archiver/mount_cmds_test.py @@ -380,3 +380,36 @@ def wrapper(self, old_id, new_id): finally: # Undecorate Lock.migrate_lock = Lock.migrate_lock.__wrapped__ + + +def test_fuse_lock_refresh_calls_repository_info(): + # regression test for #9872: an idle mount must keep its repository lock alive via a + # background refresh. LockRefresher periodically calls repository.info() while holding + # the serialization lock (the FUSE handlers use the same lock), so repo access from the + # refresh thread and the FUSE handlers never runs concurrently. + import threading + + from ...storelocking import LockRefresher + + # a plain Lock is enough here and, unlike RLock, has .locked() on all Python versions. + # in the FUSE code the lock is an RLock (it needs to be reentrant), but LockRefresher + # only ever acquires/releases it, so a plain Lock exercises the same code path. + lock = threading.Lock() + called = threading.Event() + held_while_calling = [] + + class FakeRepository: + def info(self): + held_while_calling.append(lock.locked()) # the lock must be held while we run + called.set() + return dict(id=b"\x00" * 32, version=3) + + refresher = LockRefresher(FakeRepository().info, sleep_interval=60, lock=lock) + refresher.start() + try: + assert called.wait(timeout=10), "LockRefresher did not call repository.info()" + finally: + refresher.terminate() + + assert held_while_calling and all(held_while_calling), "repository.info() must run while holding the lock" + assert not lock.locked(), "lock must be released again after refreshing"