mirror of
https://github.com/borgbackup/borg.git
synced 2026-10-12 02:11:36 +00:00
read item metadata chunks with gather_many, refs #10054
An archive's item metadata chunks are written into the same packs as the file content chunks (one ChunkBuffer flush per 8 MiB of items). Reading them via get_many loaded each whole 50 MB pack just to take a few MB of item metadata out of it: e.g. "borg list" of an archive with 3 MB of item metadata loaded 49.6 MB. DownloadPipeline.fetch_many now uses gather_many for ROBJ_ARCHIVE_STREAM (unpack_many / iter_items: list, diff, extract, mount, compact, recreate, transfer, analyze, ...), as do archive_get_items for the item_ptrs and the archives check for the item metadata chunks. File content is still read via get_many (whole packs). The legacy repositories get a gather_many that just calls get_many, so transfer from a borg 1.x repository still works. Co-Authored-By: Claude Opus 5.5 <[email protected]>
This commit is contained in:
co-authored by
Claude Opus 5.5
parent
4f94a6d8ca
commit
f2c40bcfa2
@@ -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.
|
||||
|
||||
+8
-3
@@ -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)
|
||||
|
||||
@@ -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"""
|
||||
|
||||
@@ -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
|
||||
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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
|
||||
|
||||
Reference in New Issue
Block a user