upstream/borgbackup
1
0
Fork 0
mirror of https://github.com/borgbackup/borg.git synced 2026-07-15 21:12:50 -04:00
Code Issues Projects Releases Packages Wiki Activity Actions 1

repository: cache whole packs in get_many, replacing adjacent-read grouping

Browse source 
Keep up to PACK_READER_CACHE_SIZE whole packs in an LRU so repeated and out-of-order ids read from memory instead of refetching from the store.
This commit is contained in:
Mrityunjay Raj 2026-07-03 00:06:23 +05:30
parent d076788dc5
commit f6712364df
2 changed files with 66 additions and 55 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

77
src/borg/repository.py
View file

@ -18,6 +18,7 @@ from .helpers import Location
from .helpers import bin_to_hex, hex_to_bin
from .helpers import get_cache_dir
from .helpers import ProgressIndicatorPercent
from .helpers.lrucache import LRUCache
from .storelocking import Lock
from .logger import create_logger
from .manifest import NoManifestError
@ -206,7 +207,7 @@ class PackReader:
self.key = "packs/" + bin_to_hex(pack_id) if pack_id is not None else None
self.pack_contents = pack_contents
def _read(self, offset, size):
def read(self, offset, size):
# read from the in-memory pack if we have it, else range-read from the store
if self.pack_contents is not None:
return self.pack_contents[offset : offset + size]
@ -221,7 +222,7 @@ class PackReader:
hdr_size = RepoObj.obj_header.size
offset = 0
while True:
hdr_data = self._read(offset, hdr_size)
hdr_data = self.read(offset, hdr_size)
if len(hdr_data) < hdr_size:
break # clean EOF, or trailing partial bytes
hdr = RepoObj.ObjHeader(*RepoObj.obj_header.unpack(hdr_data))
@ -303,9 +304,9 @@ class Repository:
exit_mcode = 21
# Max bytes one grouped read in get_many() covers; a run reaching this size is split into
# multiple reads.
GET_MANY_GROUP_SIZE_LIMIT = 64 * 1024 * 1024
# Whole packs kept in memory for reads; the least recently used is evicted first.
# Memory use is this count times the pack size.
PACK_READER_CACHE_SIZE = 3
def __init__(
self,
@ -391,6 +392,7 @@ class Repository:
self.exclusive = exclusive
self._pack_writer = None
self._chunks = None # ChunkIndex; loaded lazily on first access to .chunks
self._pack_cache = LRUCache(capacity=self.PACK_READER_CACHE_SIZE) # pack_id -> PackReader
def __repr__(self):
return f"<{self.__class__.__name__} {self._location}>"
@ -598,6 +600,7 @@ class Repository:
self.store.close()
self.store_opened = False
self.opened = False
self._pack_cache.clear()
def info(self):
"""return some infos about the repo (must be opened first)"""
@ -801,6 +804,16 @@ class Repository:
else:
return None
def _cached_pack_reader(self, pack_id):
"""Return a PackReader holding the whole pack, fetching it from the store on a cache miss."""
reader = self._pack_cache.get(pack_id)
if reader is not None:
return reader
key = "packs/" + bin_to_hex(pack_id)
reader = PackReader(pack_id=pack_id, pack_contents=self.store.load(key))
self._pack_cache[pack_id] = reader
return reader
def get_many(self, ids, read_data=True, raise_missing=True):
if not read_data:
# read_data=False returns only each object's header+meta, sized per object by get().
@ -808,57 +821,21 @@ class Repository:
yield self.get(id_, read_data=read_data, raise_missing=raise_missing)
return
# A pack stores its objects back-to-back. Fetch a run of ids that are byte-contiguous in
# one pack (each obj_offset == the previous obj_offset + obj_size) with one store.load and
# slice each object out of it. Objects are yielded in ids order.
run = [] # (id, obj_size) of the objects in the current run, in read order
run_key = None # store key "packs/<pack_id>" the run reads from
run_start = 0 # obj_offset of the run's first object
run_end = 0 # obj_offset just past the run's last object
def flush_run():
# read the whole run with one store.load and yield each object's slice
if not run:
return
try:
blob = self.store.load(run_key, offset=run_start, size=run_end - run_start)
except StoreObjectNotFound:
if raise_missing:
raise self.ObjectNotFound(run[0][0], str(self._location)) from None
for _ in run:
yield None
return
pos = 0
for _id, obj_size in run:
yield blob[pos : pos + obj_size]
pos += obj_size
for id_ in ids:
self._lock_refresh()
entry = self.chunks.get(id_)
if entry is None or self.chunks.is_pending(id_):
# not a flushed pack object: end the run and let get() read it or raise.
yield from flush_run()
run, run_key = [], None
# id is unknown or still buffered, not yet in a pack: read it with get()
yield self.get(id_, read_data=True, raise_missing=raise_missing)
continue
key = "packs/" + bin_to_hex(entry.pack_id)
contiguous = (
run
and key == run_key
and entry.obj_offset == run_end
and (run_end - run_start) + entry.obj_size <= self.GET_MANY_GROUP_SIZE_LIMIT
)
if contiguous:
run.append((id_, entry.obj_size))
run_end += entry.obj_size
else:
yield from flush_run()
run = [(id_, entry.obj_size)]
run_key = key
run_start = entry.obj_offset
run_end = entry.obj_offset + entry.obj_size
yield from flush_run()
try:
reader = self._cached_pack_reader(entry.pack_id)
except StoreObjectNotFound:
if raise_missing:
raise self.ObjectNotFound(id_, str(self._location)) from None
yield None
continue
yield reader.read(entry.obj_offset, entry.obj_size)
def put(self, id, data):
"""put a repo object

44
src/borg/testsuite/repository_test.py
View file

@ -205,9 +205,8 @@ def test_multi_object_pack_roundtrip(repo_fixtures, request):
def test_get_many_one_load_per_pack(repo_fixtures, request):
# Ids that are byte-contiguous in a pack -- each id's obj_offset equals the previous id's
# obj_offset + obj_size -- are read with one store.load. stats["load_calls"] counts store.load()
# calls.
# get_many loads each pack once as a whole and slices every object out of the cached bytes.
# stats["load_calls"] counts store.load() calls.
objects = [(H(i), fchunk(b"payload-%02d" % i, chunk_id=H(i))) for i in range(6)]
with get_repository_from_fixture(repo_fixtures, request) as repository:
repository._pack_writer.max_count = 3 # three objects per pack -> two packs for the six objects
@ -224,7 +223,7 @@ def test_get_many_one_load_per_pack(repo_fixtures, request):
def test_get_many_keeps_request_order(repo_fixtures, request):
# Ids requested out of their stored order come back in the requested order with their own bytes.
# Ids whose objects are not consecutive in a pack each take their own store.load.
# Interleaving two packs still loads each pack once: the pack cache serves the second visit.
objects = [(H(i), fchunk(b"payload-%02d" % i, chunk_id=H(i))) for i in range(6)]
with get_repository_from_fixture(repo_fixtures, request) as repository:
repository._pack_writer.max_count = 3 # two packs: {H0,H1,H2} and {H3,H4,H5}
@ -235,7 +234,7 @@ def test_get_many_keeps_request_order(repo_fixtures, request):
loads_before = repository.store.stats["load_calls"]
assert list(repository.get_many(ids)) == one_by_one # same order, same bytes
assert repository.store.stats["load_calls"] - loads_before == 6 # each id takes its own store.load
assert repository.store.stats["load_calls"] - loads_before == 2 # each pack loaded once
def test_get_many_missing_id_yields_none(repo_fixtures, request):
@ -254,6 +253,41 @@ def test_get_many_missing_id_yields_none(repo_fixtures, request):
assert result[2] == repository.get(H(2))
def test_get_many_repeated_ids(repo_fixtures, request):
# A dedup'd item repeats chunk ids, e.g. [A, A, B, C, B]. Each repeat returns the right bytes
# and each pack is loaded once: the cached pack serves the later visits.
objects = [(H(i), fchunk(b"payload-%02d" % i, chunk_id=H(i))) for i in range(4)]
with get_repository_from_fixture(repo_fixtures, request) as repository:
repository._pack_writer.max_count = 2 # two packs: {H0,H1} and {H2,H3}
for chunk_id, chunk in objects:
repository.put(chunk_id, chunk)
assert len({repository.chunks[H(i)].pack_id for i in range(4)}) == 2
ids = [H(0), H(0), H(1), H(2), H(1)] # A A B C B, A/B in one pack, C in the other
one_by_one = [repository.get(chunk_id) for chunk_id in ids]
loads_before = repository.store.stats["load_calls"]
assert list(repository.get_many(ids)) == one_by_one # every position, including repeats
assert repository.store.stats["load_calls"] - loads_before == 2 # two packs, one load each
def test_get_many_evicts_least_recently_used(repo_fixtures, request):
# Visiting more packs than the cache holds evicts the oldest; revisiting an evicted pack reloads it.
objects = [(H(i), fchunk(b"payload-%02d" % i, chunk_id=H(i))) for i in range(8)]
with get_repository_from_fixture(repo_fixtures, request) as repository:
repository._pack_writer.max_count = 2 # four packs: {H0,H1} {H2,H3} {H4,H5} {H6,H7}
for chunk_id, chunk in objects:
repository.put(chunk_id, chunk)
assert len({repository.chunks[H(i)].pack_id for i in range(8)}) == 4
assert repository.PACK_READER_CACHE_SIZE == 3 # cache holds three of the four packs
# touch packs 0,1,2 (fills cache), then 3 (evicts pack 0), then pack 0 again (reload).
ids = [H(0), H(2), H(4), H(6), H(0)]
one_by_one = [repository.get(chunk_id) for chunk_id in ids]
loads_before = repository.store.stats["load_calls"]
assert list(repository.get_many(ids)) == one_by_one
assert repository.store.stats["load_calls"] - loads_before == 5 # 4 distinct packs + 1 reload
def build_one_pack(repository, objects):
with repository:
repository._pack_writer.max_count = len(objects) + 1 # prevent per-put flush; one pack on flush()