From 8aa6a53ebc5733048fbd6ff7bdcf589403dc1064 Mon Sep 17 00:00:00 2001 From: Thomas Waldmann Date: Thu, 30 Jul 2026 21:21:12 +0200 Subject: [PATCH 1/2] release chunk data memoryviews after use, see #1755 On pypy, a memoryview over a C-API-created bytes object pins the buffer via cpyext: if the consumer drops the memoryview without calling .release(), the memory is never reclaimed - not even by gc.collect(). Since the chunkers wrap every yielded chunk in a memoryview, borg create leaked roughly 2-3x the processed data volume on pypy (a 20 GB create grew to a 43 GB memory footprint and got killed by the OS). Add release_chunk_data() to chunkers and call it at all places that consume chunker output: ChunkBuffer.flush, ChunksProcessor.process_file_chunks, ArchiveRecreater.chunk_processor, transfer --rechunk and the chunker benchmark. Explicitly releasing also frees the buffer timely on CPython instead of relying on refcounting of the last reference. Document the consumer obligation in the Chunk docstring. Memory now stays flat for large creates on pypy (~160 MB for a multi-GB create instead of linear growth). Also fix a pre-existing AttributeError in transfer --rechunk --dry-run: Chunk namedtuples have no .size attribute, all-zero chunks need chunk.meta["size"]. Co-Authored-By: Claude Fable 5 --- src/borg/archive.py | 17 ++++++++++++----- src/borg/archiver/benchmark_cmd.py | 6 +++--- src/borg/archiver/transfer_cmd.py | 26 +++++++++++++++----------- src/borg/chunkers/reader.pyx | 17 +++++++++++++++++ 4 files changed, 47 insertions(+), 19 deletions(-) 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.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: From 5ce324c075149f75e6bc871fd506ea9d1cc162c4 Mon Sep 17 00:00:00 2001 From: Thomas Waldmann Date: Thu, 30 Jul 2026 21:35:57 +0200 Subject: [PATCH 2/2] add release_chunk_data to reader type stub Co-Authored-By: Claude Fable 5 --- src/borg/chunkers/reader.pyi | 1 + 1 file changed, 1 insertion(+) 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]