release chunk data memoryviews after use, see #1755

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
 

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 [
             (

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

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]
 

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: