diff --git a/docs/internals/packs.rst b/docs/internals/packs.rst index 3433ed206..381cf2eab 100644 --- a/docs/internals/packs.rst +++ b/docs/internals/packs.rst @@ -242,7 +242,9 @@ reads only the blob header and the metadata slot. Two methods read many blobs: most blobs of a pack. - ``gather_many()`` reads the ranges of up to 1000 blobs (or about 16 MiB) from any number of packs with one ``store.gather`` call, for many small blobs spread over - many packs, like the archive metadata objects. + many packs, like the archive metadata objects and the item metadata chunks of an + archive (stored in the same packs as the file content chunks). Reading an archive's + items uses it, so it does not load the whole packs around them. With ``BORG_STORE_CACHE``, borgstore loads a whole pack into a local cache directory on a cache miss and serves later reads of that pack from there. diff --git a/src/borg/archive.py b/src/borg/archive.py index 4f9fe20fc..060e80801 100644 --- a/src/borg/archive.py +++ b/src/borg/archive.py @@ -434,7 +434,12 @@ class DownloadPipeline: # All-zero chunks can be served directly from the zeros constant, without repository access. zero_flags = zero_chunk_flags(ids, sizes, self.repo_objs.key.id_hash) fetch_ids = [id for id, zero in zip(ids, zero_flags) if not zero] - fetched = self.repository.get_many(fetch_ids, raise_missing=False) + if ro_type == ROBJ_ARCHIVE_STREAM: + # item metadata chunks are small and spread over packs that mostly hold file content chunks, + # so read just their byte ranges instead of loading whole packs. + fetched = self.repository.gather_many(fetch_ids, raise_missing=False) + else: + 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] @@ -547,7 +552,7 @@ def archive_get_items(metadata, *, repo_objs, repository): if "item_ptrs" in metadata: # looks like a v2+ archive assert "items" not in metadata items = [] - for id, cdata in zip(metadata.item_ptrs, repository.get_many(metadata.item_ptrs)): + for id, cdata in zip(metadata.item_ptrs, repository.gather_many(metadata.item_ptrs)): _, data = repo_objs.parse(id, cdata, ro_type=ROBJ_ARCHIVE_CHUNKIDS) ids = msgpack.unpackb(data) items.extend(ids) @@ -2770,7 +2775,7 @@ class ArchiveChecker: continue if state > 0: unpacker.resync() - for chunk_id, cdata in zip(items, self.repository.get_many(items)): + for chunk_id, cdata in zip(items, self.repository.gather_many(items)): try: _, data = self.repo_objs.parse(chunk_id, cdata, ro_type=ROBJ_ARCHIVE_STREAM) unpacker.feed(data) diff --git a/src/borg/legacy/remote.py b/src/borg/legacy/remote.py index df373aedd..262621ef1 100644 --- a/src/borg/legacy/remote.py +++ b/src/borg/legacy/remote.py @@ -619,6 +619,10 @@ class LegacyRemoteRepository: # note: legacy remote protocol does not support raise_missing parameter, so we ignore it here yield from self.call_many("get", [{"id": id, "read_data": read_data} for id in ids]) + def gather_many(self, ids, raise_missing=True): + # a legacy repository has no packs, so this is just get_many (same interface as Repository). + return self.get_many(ids, raise_missing=raise_missing) + @api(since=parse_version("1.0.0")) def put(self, id, data, wait=True): """actual remoting is done via self.call in the @api decorator""" diff --git a/src/borg/legacy/repository.py b/src/borg/legacy/repository.py index 274f1fe00..bafe2c15c 100644 --- a/src/borg/legacy/repository.py +++ b/src/borg/legacy/repository.py @@ -1152,6 +1152,10 @@ class LegacyRepository: for id_ in ids: yield self.get(id_, read_data=read_data, raise_missing=raise_missing) + def gather_many(self, ids, raise_missing=True): + # a legacy repository has no packs, so this is just get_many (same interface as Repository). + return self.get_many(ids, raise_missing=raise_missing) + def put(self, id, data, wait=True): """put a repo object diff --git a/src/borg/repository.py b/src/borg/repository.py index 52ec596c6..a58469d49 100644 --- a/src/borg/repository.py +++ b/src/borg/repository.py @@ -2019,7 +2019,8 @@ class Repository: One store.gather call reads the byte ranges of up to GATHER_MAX_COUNT objects (or about GATHER_MAX_SIZE bytes) from any number of packs, so a backend that supports it (e.g. REST) needs one roundtrip per batch instead of one per object. This suits many small objects spread over many packs, like the archive - metadata objects (each usually in a tiny pack of its own). get_many() loads whole packs instead, which + metadata objects (each usually in a tiny pack of its own) and the item metadata chunks of an archive + (written into the packs between file content chunks). get_many() loads whole packs instead, which suits reading most of the objects of a pack. raise_missing: like for get(). Ids whose objects cannot be gathered (unknown or still buffered ids, and diff --git a/src/borg/testsuite/archive_test.py b/src/borg/testsuite/archive_test.py index f98d3e161..816cfa5ec 100644 --- a/src/borg/testsuite/archive_test.py +++ b/src/borg/testsuite/archive_test.py @@ -7,9 +7,9 @@ from unittest.mock import Mock import pytest -from . import rejected_dotdot_paths, is_utime_fully_supported, make_test_key +from . import rejected_dotdot_paths, is_utime_fully_supported, make_test_key, set_test_key_on_open from ..cache import ChunkListEntry -from ..constants import ROBJ_FILE_STREAM, zeros +from ..constants import ROBJ_ARCHIVE_STREAM, ROBJ_FILE_STREAM, zeros from ..archive import Archive, CacheChunkBuffer, DownloadPipeline, RobustUnpacker, valid_msgpacked_dict from ..archive import ITEM_KEYS, Statistics from ..archive import zero_chunk_flags, zero_chunk_id, zero_chunk_ids @@ -19,6 +19,7 @@ from ..helpers import msgpack from ..repoobj import RepoObj from ..item import Item, ArchiveItem from ..manifest import Archives, Manifest +from ..repository import Repository from ..platform import uid2user, gid2group, is_win32 @@ -353,6 +354,41 @@ def test_download_pipeline_zero_chunks_served_locally(): assert repository.requested_ids == [other_id] +def test_download_pipeline_item_stream_reads_ranges(tmp_path, monkeypatch): + # item metadata chunks share packs with file content chunks: reading an archive's items gathers just + # their byte ranges, while file content is still read by loading the whole pack. + set_test_key_on_open(monkeypatch) # the index/ objects need a key, see Repository.set_key + key = make_test_key(None) + repo_objs = RepoObj(key) + items = [Item(path=f"file{i}", mode=0o100644) for i in range(3)] + stream = b"".join(msgpack.packb(item.as_dict()) for item in items) + item_chunks = [stream[:20], stream[20:]] + content = os.urandom(10000) + with Repository(os.fspath(tmp_path / "repository"), exclusive=True, create=True) as repository: + content_id = repo_objs.id_hash(content) + repository.put(content_id, repo_objs.format(content_id, {}, content, ro_type=ROBJ_FILE_STREAM)) + item_ids = [] + for data in item_chunks: + item_id = repo_objs.id_hash(data) + repository.put(item_id, repo_objs.format(item_id, {}, data, ro_type=ROBJ_ARCHIVE_STREAM)) + item_ids.append(item_id) + repository.flush() + assert len({repository.chunks[id].pack_id for id in item_ids + [content_id]}) == 1 # all in one pack + pipeline = DownloadPipeline(repository, repo_objs) + + loads_before, gathers_before = repository.store.stats["load_calls"], repository.store.stats["gather_calls"] + assert [item.path for item in pipeline.unpack_many(item_ids)] == [item.path for item in items] + assert repository.store.stats["gather_calls"] - gathers_before == 1 + assert repository.store.stats["load_calls"] == loads_before # no whole-pack load + + loads_before = repository.store.stats["load_calls"] + assert list(pipeline.fetch_many([ChunkListEntry(content_id, len(content))], ro_type=ROBJ_FILE_STREAM)) == [ + content + ] + assert repository.store.stats["load_calls"] - loads_before == 1 # the whole pack + assert len(repository._pack_cache) == 1 + + def test_zero_chunk_flags(): # cheap all-zero chunk detection from the chunk ids/sizes alone: the zero chunk id # is computed for ids occurring repeatedly, while unique ids are only compared