diff --git a/src/borg/chunkers/fixed.py b/src/borg/chunkers/fixed.py index 74db7311c2..5db28ed635 100644 --- a/src/borg/chunkers/fixed.py +++ b/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 diff --git a/src/borg/chunkers/reader.pyi b/src/borg/chunkers/reader.pyi index b1e6a143c5..7870952f61 100644 --- a/src/borg/chunkers/reader.pyi +++ b/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: ... diff --git a/src/borg/chunkers/reader.pyx b/src/borg/chunkers/reader.pyx index c3ba7c8405..f6866de03f 100644 --- a/src/borg/chunkers/reader.pyx +++ b/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 = []