extract: avoid refetching/reparsing repeated chunks, fixes #1678

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

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)