diff --git a/src/borg/archive.py b/src/borg/archive.py index 4af2d47f18..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 @@ -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): """ @@ -356,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.") @@ -365,7 +395,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 @@ -1244,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: @@ -1253,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 da7ea854e5..817bde427c 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, zeros 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,91 @@ 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) + # 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: + 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 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)