From ebfdaca99652d78d25e4e5c23a059e402716cd02 Mon Sep 17 00:00:00 2001 From: Thomas Waldmann Date: Thu, 30 Jul 2026 19:03:41 +0200 Subject: [PATCH 1/2] fetch_many: cache recently parsed chunks, #1678 A content data stream may reference the same chunk many times (e.g. the all-zero chunks of a sparse file converge on the same ids). Previously, every repetition was decrypted, authenticated and decompressed again. Add a small LRU cache of recently parsed chunks to DownloadPipeline, keyed on (id, ro_type), so repeated chunks are served from the cache. The repository is still asked for every occurrence (a cheap cached-pack slice for local repos, and it keeps the legacy remote repos' pipelining in borg transfer intact), only the parsing work is skipped. Extracting a file with a 256 MiB hole now parses the all-zero chunk once instead of 63 times. Co-Authored-By: Claude Fable 5 --- src/borg/archive.py | 13 +++++++- src/borg/testsuite/archive_test.py | 51 +++++++++++++++++++++++++++++- 2 files changed, 62 insertions(+), 2 deletions(-) diff --git a/src/borg/archive.py b/src/borg/archive.py index 4af2d47f18..8f4454a536 100644 --- a/src/borg/archive.py +++ b/src/borg/archive.py @@ -318,9 +318,16 @@ def OsOpen(*, flags, path=None, parent_fd=None, name=None, noatime=False, op="op class DownloadPipeline: + # A content data stream may reference the same chunk many times (e.g. the all-zero + # chunks of a sparse file, see issue #1678), thus cache the most recently parsed + # chunks, so repeated chunks do not get decrypted, authenticated and decompressed + # again. Chunks can be up to MAX_DATA_SIZE bytes, thus keep the cache small. + PARSED_CACHE_SIZE = 4 + def __init__(self, repository, repo_objs): self.repository = repository self.repo_objs = repo_objs + self.parsed_cache = LRUCache(capacity=self.PARSED_CACHE_SIZE) # (id, ro_type) -> data def unpack_many(self, ids, *, filter=None): """ @@ -365,7 +372,11 @@ def fetch_many(self, chunks, ro_type=None, replacement_chunk=True): logger.error(f"repository object {bin_to_hex(id)} missing, returning None.") data = None else: - _, data = self.repo_objs.parse(id, cdata, ro_type=ro_type) + try: + data = self.parsed_cache[(id, ro_type)] + except KeyError: + _, data = self.repo_objs.parse(id, cdata, ro_type=ro_type) + self.parsed_cache[(id, ro_type)] = data assert size is None or len(data) == size yield data diff --git a/src/borg/testsuite/archive_test.py b/src/borg/testsuite/archive_test.py index da7ea854e5..69f4b96ede 100644 --- a/src/borg/testsuite/archive_test.py +++ b/src/borg/testsuite/archive_test.py @@ -8,10 +8,14 @@ import pytest from . import rejected_dotdot_paths +from ..cache import ChunkListEntry +from ..constants import ROBJ_FILE_STREAM from ..crypto.key import PlaintextKey -from ..archive import Archive, CacheChunkBuffer, RobustUnpacker, valid_msgpacked_dict, ITEM_KEYS, Statistics +from ..archive import Archive, CacheChunkBuffer, DownloadPipeline, RobustUnpacker, valid_msgpacked_dict +from ..archive import ITEM_KEYS, Statistics from ..archive import BackupOSError, backup_io, backup_io_iter, get_item_uid_gid from ..helpers import msgpack +from ..repoobj import RepoObj from ..item import Item, ArchiveItem from ..manifest import Archives, Manifest from ..platform import uid2user, gid2group, is_win32 @@ -193,6 +197,51 @@ def test_partial_cache_chunk_buffer(): assert data == [Item(internal_dict=d) for d in unpacker] +class MockFetchRepo: + """serve repo objects from a dict, recording all requested ids.""" + + def __init__(self, objects): + self.objects = objects # id -> cdata + self.requested_ids = [] + + def get_many(self, ids, read_data=True, raise_missing=True): + for id in ids: + self.requested_ids.append(id) + yield self.objects[id] + + +def test_download_pipeline_parsed_cache(): + # a content data stream may reference the same chunk many times (e.g. the all-zero + # chunks of a sparse file): repeated chunks shall be parsed (decrypted, authenticated, + # decompressed) only once, see issue #1678. + key = PlaintextKey(None) + repo_objs = RepoObj(key) + chunks_data = [b"foobar" * 100, b"\0" * 1000, b"barbaz" * 100] + entries = [] + objects = {} + for data in chunks_data: + id = repo_objs.id_hash(data) + objects[id] = repo_objs.format(id, {}, data, ro_type=ROBJ_FILE_STREAM) + entries.append(ChunkListEntry(id, len(data))) + # reference the second chunk many times, interleaved with the other chunks + chunk_list = [entries[0]] + [entries[1]] * 5 + [entries[2]] + [entries[1]] * 5 + repository = MockFetchRepo(objects) + pipeline = DownloadPipeline(repository, repo_objs) + parsed_ids = [] + orig_parse = repo_objs.parse + + def counting_parse(id, cdata, **kw): + parsed_ids.append(id) + return orig_parse(id, cdata, **kw) + + repo_objs.parse = counting_parse + result = list(pipeline.fetch_many(chunk_list, ro_type=ROBJ_FILE_STREAM)) + assert result == [chunks_data[0]] + [chunks_data[1]] * 5 + [chunks_data[2]] + [chunks_data[1]] * 5 + # each distinct chunk was parsed only once, the repetitions were served from the cache + assert len(parsed_ids) == 3 + assert len(set(parsed_ids)) == 3 + + def make_chunks(items): return b"".join(msgpack.packb({"path": item}) for item in items) From c76f2cea8536d78ef0f5c0c8fce256e88100ac09 Mon Sep 17 00:00:00 2001 From: Thomas Waldmann Date: Thu, 30 Jul 2026 19:09:08 +0200 Subject: [PATCH 2/2] fetch_many: serve all-zero chunks without repository access, #1678 The holes of a sparse file produce long runs of references to the same all-zero chunk. Instead of fetching that chunk from the repository over and over, detect it and serve it directly from the zeros constant. Detection compares the chunk id against the id of an all-zero chunk of the same size, memoized in the already existing zero_chunk_ids mapping (now shared with the create side via the new zero_chunk_id() function). To bound the memoized id computations to a few chunk sizes, they are only done for ids occurring repeatedly within the requested stream - a repeated id means repeating plaintext, which usually is a run of zeros. Unique ids are only compared against already memoized sizes. Extracting a file with a 256 MiB hole now does not access the repository at all for the hole (before: 63 object fetches). Co-Authored-By: Claude Fable 5 --- src/borg/archive.py | 44 +++++++++++++++++++++++++----- src/borg/testsuite/archive_test.py | 44 ++++++++++++++++++++++++++++-- 2 files changed, 79 insertions(+), 9 deletions(-) diff --git a/src/borg/archive.py b/src/borg/archive.py index 8f4454a536..4f41cdf41d 100644 --- a/src/borg/archive.py +++ b/src/borg/archive.py @@ -6,7 +6,7 @@ import stat import sys import time -from collections import defaultdict +from collections import Counter, defaultdict from contextlib import contextmanager from datetime import timedelta from functools import partial @@ -363,7 +363,30 @@ def fetch_many(self, chunks, ro_type=None, replacement_chunk=True): sizes = [None] * len(ids) else: raise TypeError(f"unsupported or mixed element types: {chunks}") - for id, size, cdata in zip(ids, sizes, self.repository.get_many(ids, raise_missing=False)): + # All-zero chunks can be served directly from the zeros constant, without repository + # access, by comparing against the (memoized) id of an all-zero chunk of same size. + # Only compute that id for ids occurring repeatedly within this stream: a repeated id + # means repeating plaintext, which usually is a run of zeros (e.g. the "holes" of a + # sparse file, see issue #1678) - and the repetition also keeps the memoization + # effective, as it bounds the computations to a few chunk sizes. + id_hash = self.repo_objs.key.id_hash + counts = Counter(ids) + zero_flags = [] + for id, size in zip(ids, sizes): + if size is None or not 0 < size <= len(zeros): + zero_flags.append(False) + elif counts[id] > 1: + zero_flags.append(id == zero_chunk_id(id_hash, size)) + else: + # unique id: only compare against already memoized zero chunk ids (cheap). + zero_flags.append(id == zero_chunk_ids.get((id_hash, size))) + fetch_ids = [id for id, zero in zip(ids, zero_flags) if not zero] + fetched = self.repository.get_many(fetch_ids, raise_missing=False) + for id, size, zero in zip(ids, sizes, zero_flags): + if zero: + yield zeros[:size] + continue + cdata = next(fetched) if cdata is None: if replacement_chunk and size is not None: logger.error(f"repository object {bin_to_hex(id)} missing, returning {size} zero bytes.") @@ -1255,6 +1278,17 @@ def stat_attrs(self, st, path, fd=None): zero_chunk_ids = LRUCache(10) # type: ignore[var-annotated] +def zero_chunk_id(id_hash, size): + """return the id of an all-zero chunk of length *size* (memoized).""" + assert 0 < size <= len(zeros) + try: + return zero_chunk_ids[(id_hash, size)] + except KeyError: + chunk_id = id_hash(memoryview(zeros)[:size]) + zero_chunk_ids[(id_hash, size)] = chunk_id + return chunk_id + + def cached_hash(chunk, id_hash): allocation = chunk.meta["allocation"] if allocation == CH_DATA: @@ -1264,11 +1298,7 @@ def cached_hash(chunk, id_hash): size = chunk.meta["size"] assert size <= len(zeros) data = memoryview(zeros)[:size] - try: - chunk_id = zero_chunk_ids[(id_hash, size)] - except KeyError: - chunk_id = id_hash(data) - zero_chunk_ids[(id_hash, size)] = chunk_id + chunk_id = zero_chunk_id(id_hash, size) else: raise ValueError("unexpected allocation type") return chunk_id, data diff --git a/src/borg/testsuite/archive_test.py b/src/borg/testsuite/archive_test.py index 69f4b96ede..817bde427c 100644 --- a/src/borg/testsuite/archive_test.py +++ b/src/borg/testsuite/archive_test.py @@ -9,7 +9,7 @@ from . import rejected_dotdot_paths from ..cache import ChunkListEntry -from ..constants import ROBJ_FILE_STREAM +from ..constants import ROBJ_FILE_STREAM, zeros from ..crypto.key import PlaintextKey from ..archive import Archive, CacheChunkBuffer, DownloadPipeline, RobustUnpacker, valid_msgpacked_dict from ..archive import ITEM_KEYS, Statistics @@ -216,7 +216,8 @@ def test_download_pipeline_parsed_cache(): # decompressed) only once, see issue #1678. key = PlaintextKey(None) repo_objs = RepoObj(key) - chunks_data = [b"foobar" * 100, b"\0" * 1000, b"barbaz" * 100] + # note: repeated, but not all-zero data, so it is not served via the zeros shortcut + chunks_data = [b"foobar" * 100, b"idletone" * 125, b"barbaz" * 100] entries = [] objects = {} for data in chunks_data: @@ -242,6 +243,45 @@ def counting_parse(id, cdata, **kw): assert len(set(parsed_ids)) == 3 +def test_download_pipeline_zero_chunks_served_locally(): + # repeated all-zero chunks (e.g. from the holes of a sparse file) shall be served + # directly from the zeros constant, without repository access, see issue #1678. + key = PlaintextKey(None) + repo_objs = RepoObj(key) + data = b"foobar" * 100 + data_id = repo_objs.id_hash(data) + objects = {data_id: repo_objs.format(data_id, {}, data, ro_type=ROBJ_FILE_STREAM)} + zero_size = 1000 + zero_id = repo_objs.id_hash(zeros[:zero_size]) + # note: the all-zero chunk is intentionally NOT in the repository objects, + # thus serving it can only work without repository access. + chunk_list = [ + ChunkListEntry(zero_id, zero_size), + ChunkListEntry(data_id, len(data)), + ChunkListEntry(zero_id, zero_size), + ChunkListEntry(zero_id, zero_size), + ] + repository = MockFetchRepo(objects) + pipeline = DownloadPipeline(repository, repo_objs) + result = list(pipeline.fetch_many(chunk_list, ro_type=ROBJ_FILE_STREAM)) + assert result == [zeros[:zero_size], data, zeros[:zero_size], zeros[:zero_size]] + assert repository.requested_ids == [data_id] + # now that the zero chunk id of this size is known, even a single occurrence + # is served locally: + repository.requested_ids.clear() + result = list(pipeline.fetch_many([ChunkListEntry(zero_id, zero_size)], ro_type=ROBJ_FILE_STREAM)) + assert result == [zeros[:zero_size]] + assert repository.requested_ids == [] + # but a single occurrence of an all-zero chunk of an unknown size is still + # fetched from the repository: + other_size = 500 + other_id = repo_objs.id_hash(zeros[:other_size]) + objects[other_id] = repo_objs.format(other_id, {}, zeros[:other_size], ro_type=ROBJ_FILE_STREAM) + result = list(pipeline.fetch_many([ChunkListEntry(other_id, other_size)], ro_type=ROBJ_FILE_STREAM)) + assert result == [zeros[:other_size]] + assert repository.requested_ids == [other_id] + + def make_chunks(items): return b"".join(msgpack.packb({"path": item}) for item in items)