fixed chunker: read directly into per-chunk buffers

src/borg/chunkers/fixed.py

@@ -4,7 +4,8 @@
 
 import time
 
-from .reader import FileReader
+from .reader import FileReader, Chunk
+from ..constants import CH_DATA, CH_ALLOC, zeros
 
 
 class ChunkerFixed:
@@ -26,6 +27,11 @@ class ChunkerFixed:
 
     Note: the last block of a data or hole range may be less than the block size,
           this is supported and not considered to be an error.
+
+    Chunks are yielded as memoryviews over per-chunk buffers that the file data is
+    read into directly - each byte is copied exactly once, from the reader's block
+    buffer into the chunk (all-zero chunks are yielded as allocation CH_ALLOC with
+    data None). Call release_chunk_data() after consuming a chunk.
     """
 
     def __init__(self, block_size: int, header_size: int = 0, sparse: bool = False) -> None:
@@ -48,24 +54,25 @@ def chunkify(self, fd: BinaryIO | None = None, fh: int = -1, fmap: list | None =
         # Initialize the reader with the file descriptors
         self.reader = FileReader(fd=fd, fh=fh, read_size=self.reader_block_size, sparse=self.sparse, fmap=fmap)
 
-        # Handle header if present
-        if self.header_size > 0:
-            # Read the header block using read
-            started_chunking = time.monotonic()
-            header_chunk = self.reader.read(self.header_size)
-            self.chunking_time += time.monotonic() - started_chunking
-
-            if header_chunk.meta["size"] > 0:
-                # Yield the header chunk
-                yield header_chunk
-
-        # Process the rest of the file using read
+        # Read the header block first, if there is one; after it, read block_size sized blocks.
+        wanted = self.header_size if self.header_size > 0 else self.block_size
         while True:
             started_chunking = time.monotonic()
-            chunk = self.reader.read(self.block_size)
+            # The file data is read directly into a fresh per-chunk buffer (zero-filled
+            # for ranges stemming from holes), so each byte is copied only once, instead
+            # of being assembled in an intermediate buffer and copied out of it again.
+            buf = bytearray(wanted)
+            got = self.reader.readinto(buf, wanted)
             self.chunking_time += time.monotonic() - started_chunking
-            size = chunk.meta["size"]
-            if size == 0:
+            if got == 0:
                 break  # EOF
-            assert size <= self.block_size
-            yield chunk
+            assert got <= wanted
+            data = memoryview(buf)[:got]
+            if zeros.startswith(data):
+                # All-zero chunk: either read as zeros from the file or stemming from
+                # a hole in a sparse file - we can not distinguish that here.
+                data.release()
+                yield Chunk(None, size=got, allocation=CH_ALLOC)
+            else:
+                yield Chunk(data, size=got, allocation=CH_DATA)
+            wanted = self.block_size

src/borg/chunkers/reader.pyi

@@ -4,10 +4,10 @@ from typing import Any, BinaryIO, NamedTuple
 has_seek_hole: bool
 
 class _Chunk(NamedTuple):
-    data: bytes | None
+    data: bytes | memoryview | None
     meta: dict[str, Any]
 
-def Chunk(data: bytes | None, **meta) -> type[_Chunk]: ...
+def Chunk(data: bytes | memoryview | None, **meta) -> type[_Chunk]: ...
 def release_chunk_data(data: bytes | memoryview | None) -> None: ...
 
 fmap_entry = tuple[int, int, bool]
@@ -38,4 +38,6 @@ class FileReader:
         fmap: list[fmap_entry] = None,
     ) -> None: ...
     def _fill_buffer(self) -> bool: ...
+    def _readinto_direct(self, tv: memoryview, size: int) -> int: ...
     def read(self, size: int) -> type[_Chunk]: ...
+    def readinto(self, target: bytearray | memoryview, size: int) -> int: ...

src/borg/chunkers/reader.pyx

@@ -15,6 +15,9 @@ from ..constants import CH_DATA, CH_ALLOC, CH_HOLE, zeros
 # because the FS also needs to support this.
 has_seek_hole = hasattr(os, 'SEEK_DATA') and hasattr(os, 'SEEK_HOLE')
 
+# os.readv is POSIX; on platforms without it (win32) we fall back to os.read + copy.
+has_readv = hasattr(os, 'readv')
+
 _Chunk = namedtuple('_Chunk', 'meta data')
 _Chunk.__doc__ = """\
     Chunk namedtuple
@@ -234,6 +237,12 @@ class FileReader:
         self.fd = fd
         self.fh = fh
         self.fmap = fmap
+        # Without sparse processing and without a given fmap there are no ranges to
+        # consider - the file is read start to end. readinto() then reads directly
+        # from the file into the caller's buffer (see there), instead of going
+        # through the block reader.
+        self.direct = not sparse and fmap is None
+        self.direct_offset = 0  # bytes read so far via the direct path (for fadvise)
 
     def _fill_buffer(self):
         """
@@ -358,22 +367,59 @@ class FileReader:
             # Otherwise, all chunks were CH_ALLOC
             return Chunk(None, size=bytes_read, allocation=CH_ALLOC)
 
+    def _readinto_direct(self, tv, size):
+        """Read up to 'size' bytes from the file directly into 'tv' (a writable memoryview)."""
+        pos = 0
+        while pos < size:
+            if self.fh >= 0:
+                if has_readv:
+                    got = os.readv(self.fh, [tv[pos:size]])
+                else:
+                    data = os.read(self.fh, size - pos)
+                    got = len(data)
+                    tv[pos:pos + got] = data
+                if got > 0:
+                    safe_fadvise(self.fh, self.direct_offset, got, "DONTNEED")
+            else:
+                try:
+                    got = self.fd.readinto(tv[pos:size])
+                except AttributeError:
+                    # file-like object without readinto: fall back to read + copy
+                    data = self.fd.read(size - pos)
+                    got = len(data)
+                    tv[pos:pos + got] = data
+            if not got:
+                break  # EOF
+            pos += got
+            self.direct_offset += got
+        return pos
+
     def readinto(self, target, size):
         """
         Read up to 'size' bytes from the file directly into 'target' (a writable
         buffer, e.g. a memoryview over the caller's scan buffer).
 
-        Unlike read(), this does not allocate or combine intermediate byte
-        objects: each byte is copied exactly once, from the buffered file block
-        into 'target'. Ranges stemming from holes / all-zero blocks are written
-        as zero bytes ('target' may contain stale data). The caller detects
-        all-zero chunks itself at chunk granularity, so no allocation type is
-        returned.
+        Fast path (no sparse processing, no fmap given): the file data is read
+        by the OS directly into 'target' - zero copies in user space and one
+        syscall per request instead of one per block.
+
+        Otherwise, unlike read(), this does not allocate or combine intermediate
+        byte objects: each byte is copied exactly once, from the buffered file
+        block into 'target'. Ranges stemming from holes / all-zero blocks are
+        written as zero bytes ('target' may contain stale data). The caller
+        detects all-zero chunks itself at chunk granularity, so no allocation
+        type is returned.
 
         :param target: writable buffer, len(target) >= size
         :param size: number of bytes to read
         :return: number of bytes written to target (0 at EOF).
         """
+        if self.direct and self.blockify_gen is None:
+            # blockify_gen check: if read() was used on this reader before, keep
+            # using the buffered path, for consistent file position and buffer state.
+            with memoryview(target) as tv:
+                return self._readinto_direct(tv, size)
+
         # Initialize if not already done
         if self.blockify_gen is None:
             self.buffer = []