diff --git a/src/borg/archive.py b/src/borg/archive.py index 4af2d47f18..d5e5e73432 100644 --- a/src/borg/archive.py +++ b/src/borg/archive.py @@ -22,7 +22,7 @@ logger = create_logger() from . import xattr -from .chunkers import get_chunker, Chunk +from .chunkers import get_chunker, Chunk, release_chunk_data from .cache import ChunkListEntry, build_chunkindex_from_repo, delete_chunkindex_from_repo from .crypto.key import key_factory, UnsupportedPayloadError from .constants import * # NOQA @@ -402,6 +402,7 @@ def flush(self, flush=False): alloc = chunk.meta["allocation"] if alloc == CH_DATA: data = bytes(chunk.data) + release_chunk_data(chunk.data) elif alloc in (CH_ALLOC, CH_HOLE): data = zeros[: chunk.meta["size"]] else: @@ -1279,7 +1280,10 @@ def chunk_processor(chunk): started_hashing = time.monotonic() chunk_id, data = cached_hash(chunk, self.key.id_hash) stats.hashing_time += time.monotonic() - started_hashing - chunk_entry = cache.add_chunk(chunk_id, {}, data, stats=stats, ro_type=ROBJ_FILE_STREAM) + try: + chunk_entry = cache.add_chunk(chunk_id, {}, data, stats=stats, ro_type=ROBJ_FILE_STREAM) + finally: + release_chunk_data(data) return chunk_entry item.chunks = [] @@ -2379,9 +2383,12 @@ def process_chunks(self, archive, target, item): def chunk_processor(self, target, chunk): chunk_id, data = cached_hash(chunk, self.key.id_hash) size = len(data) - if chunk_id in self.seen_chunks: - return self.cache.reuse_chunk(chunk_id, size, target.stats) - chunk_entry = self.cache.add_chunk(chunk_id, {}, data, stats=target.stats, ro_type=ROBJ_FILE_STREAM) + try: + if chunk_id in self.seen_chunks: + return self.cache.reuse_chunk(chunk_id, size, target.stats) + chunk_entry = self.cache.add_chunk(chunk_id, {}, data, stats=target.stats, ro_type=ROBJ_FILE_STREAM) + finally: + release_chunk_data(data) self.seen_chunks.add(chunk_entry.id) return chunk_entry diff --git a/src/borg/archiver/benchmark_cmd.py b/src/borg/archiver/benchmark_cmd.py index 2f58ba44e3..bb8694237f 100644 --- a/src/borg/archiver/benchmark_cmd.py +++ b/src/borg/archiver/benchmark_cmd.py @@ -166,7 +166,7 @@ def do_benchmark_cpu(self, args): key_96 = os.urandom(12) import io - from ..chunkers import get_chunker # noqa + from ..chunkers import get_chunker, release_chunk_data # noqa if not args.json: print("Chunkers =======================================================") @@ -176,8 +176,8 @@ def do_benchmark_cpu(self, args): def chunkit(ch): with io.BytesIO(random_10M) as data_file: - for _ in ch.chunkify(fd=data_file): - pass + for chunk in ch.chunkify(fd=data_file): + release_chunk_data(chunk.data) for spec, setup, func, vars in [ ( diff --git a/src/borg/archiver/transfer_cmd.py b/src/borg/archiver/transfer_cmd.py index d51e6261f0..59671b7c32 100644 --- a/src/borg/archiver/transfer_cmd.py +++ b/src/borg/archiver/transfer_cmd.py @@ -1,6 +1,6 @@ from ._common import with_repository, with_other_repository, Highlander from ..archive import Archive, cached_hash, DownloadPipeline -from ..chunkers import get_chunker +from ..chunkers import get_chunker, release_chunk_data from ..constants import * # NOQA from ..crypto.key import uses_same_id_hash, uses_same_chunker_secret from ..helpers import Error @@ -55,19 +55,23 @@ def transfer_chunks( if not dry_run: chunk_id, data = cached_hash(chunk, archive.key.id_hash) size = len(data) - # Check if the chunk is already in the repository - chunk_present = cache.seen_chunk(chunk_id, size) - if chunk_present: - chunk_entry = cache.reuse_chunk(chunk_id, size, archive.stats) - present += size - else: - # Add the new chunk to the repository - chunk_entry = cache.add_chunk(chunk_id, {}, data, stats=archive.stats, ro_type=ROBJ_FILE_STREAM) - transfer += size + try: + # Check if the chunk is already in the repository + chunk_present = cache.seen_chunk(chunk_id, size) + if chunk_present: + chunk_entry = cache.reuse_chunk(chunk_id, size, archive.stats) + present += size + else: + # Add the new chunk to the repository + chunk_entry = cache.add_chunk(chunk_id, {}, data, stats=archive.stats, ro_type=ROBJ_FILE_STREAM) + transfer += size + finally: + release_chunk_data(data) chunks.append(chunk_entry) else: # In dry-run mode, just estimate the size - size = len(chunk.data) if chunk.data is not None else chunk.size + size = len(chunk.data) if chunk.data is not None else chunk.meta["size"] + release_chunk_data(chunk.data) transfer += size else: # Original implementation without re-chunking diff --git a/src/borg/chunkers/reader.pyi b/src/borg/chunkers/reader.pyi index 2d05edf155..b1e6a143c5 100644 --- a/src/borg/chunkers/reader.pyi +++ b/src/borg/chunkers/reader.pyi @@ -8,6 +8,7 @@ class _Chunk(NamedTuple): meta: dict[str, Any] def Chunk(data: bytes | None, **meta) -> type[_Chunk]: ... +def release_chunk_data(data: bytes | memoryview | None) -> None: ... fmap_entry = tuple[int, int, bool] diff --git a/src/borg/chunkers/reader.pyx b/src/borg/chunkers/reader.pyx index 7d9e35e8ee..3453fa9949 100644 --- a/src/borg/chunkers/reader.pyx +++ b/src/borg/chunkers/reader.pyx @@ -32,12 +32,29 @@ _Chunk.__doc__ = """\ all-zero chunk from a HOLE range of a file (from a sparse hole): meta = {'allocation' = CH_HOLE, 'size' = size_of_chunk } data = None + + Attention: after consuming a chunk, call release_chunk_data(chunk.data)! + Otherwise, the memory backing a memoryview is not freed timely (CPython) + or even never freed (pypy: dropped, but unreleased memoryviews over + C-API-created bytes objects are never reclaimed by the GC there). """ def Chunk(data, **meta): return _Chunk(meta, data) +def release_chunk_data(data): + """Release chunk data (as yielded by a chunker) after use. + + Explicitly releasing the memoryview frees the underlying buffer timely. + This is especially important on pypy: there, a dropped (but not released) + memoryview over a C-API-created bytes object pins the buffer via cpyext + and the memory is never reclaimed, not even by gc.collect(). + """ + if isinstance(data, memoryview): + data.release() + + def dread(offset, size, fd=None, fh=-1): use_fh = fh >= 0 if use_fh: