diff --git a/docs/changes.rst b/docs/changes.rst index 2c2069e6e..9e32e9209 100644 --- a/docs/changes.rst +++ b/docs/changes.rst @@ -220,6 +220,10 @@ Fixes: Other changes: - crypto: start a new session after encrypting 2TiB with one aes256-ocb session key, #6501 +- chunkers: refactor the shared machinery (buffering, min/max clamping, normalized + chunking, iterator protocol) into a common ChunkerBase class instead of keeping a + copy per chunker; fastcdc also shares the window-less scan loop with the AES + chunkers via a _scan() hook. Cut points stay bit-identical for all chunkers. - removed some global options (they were difficult to use and spammed the help output): - --remote-path -> BORG_REMOTE_PATH diff --git a/scripts/borg.exe.spec b/scripts/borg.exe.spec index d43398545..a705db35b 100644 --- a/scripts/borg.exe.spec +++ b/scripts/borg.exe.spec @@ -14,9 +14,11 @@ if is_win32: else: hiddenimports = ['borg.platform.posix', 'borghash', 'rich._unicode_data.unicode17-0-0'] -# The *-aes chunkers cimport their shared base class (ChunkerPHTE), so they import -# that module at C level when initializing - PyInstaller can not see this by -# static analysis of the Python sources. +# The chunkers cimport their shared base classes (ChunkerBase, and ChunkerPHTE +# for the *-aes chunkers), so they import those modules at C level when +# initializing - PyInstaller can not see this by static analysis of the Python +# sources. +hiddenimports.append('borg.chunkers.base') hiddenimports.append('borg.chunkers.phte_chunker') block_cipher = None diff --git a/setup.py b/setup.py index d05adf30d..4d790e47d 100644 --- a/setup.py +++ b/setup.py @@ -54,6 +54,7 @@ if not is_win32: compress_source = "src/borg/compress.pyx" crypto_ll_source = "src/borg/crypto/low_level.pyx" crypto_legacy_ll_source = "src/borg/legacy/crypto/low_level.pyx" +chunker_base_source = "src/borg/chunkers/base.pyx" buzhash_source = "src/borg/chunkers/buzhash.pyx" buzhash64_source = "src/borg/chunkers/buzhash64.pyx" buzhash64_impl_source = "src/borg/chunkers/buzhash64_impl.c" @@ -81,6 +82,7 @@ cython_sources = [ compress_source, crypto_ll_source, crypto_legacy_ll_source, + chunker_base_source, buzhash_source, buzhash64_source, fastcdc_source, @@ -223,6 +225,7 @@ if not on_rtd: Extension("borg.compress", **compress_ext_kwargs), Extension("borg.hashindex", [hashindex_source], extra_compile_args=cflags), Extension("borg.item", [item_source], extra_compile_args=cflags), + Extension("borg.chunkers.base", [chunker_base_source], extra_compile_args=cflags), Extension("borg.chunkers.buzhash", [buzhash_source], extra_compile_args=cflags), Extension( "borg.chunkers.buzhash64", diff --git a/src/borg/chunkers/base.pxd b/src/borg/chunkers/base.pxd new file mode 100644 index 000000000..a99f040eb --- /dev/null +++ b/src/borg/chunkers/base.pxd @@ -0,0 +1,34 @@ +# cython: language_level=3 + +from libc.stdint cimport uint8_t, uint64_t, int64_t + + +cdef class ChunkerBase: + # chunking parameters + cdef str algo # chunker name, for error messages + cdef uint64_t chunk_mask + cdef uint64_t mask_s, mask_l # normalized chunking: strict / loose masks + cdef size_t normal_size # chunk length at which we switch mask_s -> mask_l + cdef int nc_level # normalized chunking level (0 = disabled) + + # buffer and stream state + cdef uint8_t* data + cdef object _fd + cdef int fh + cdef int done, eof + cdef size_t min_size, buf_size, remaining, position, last + cdef long long bytes_read, bytes_yielded + cdef readonly float chunking_time + cdef object file_reader + cdef size_t reader_block_size + cdef bint sparse + + cdef int _setup_common(self, str algo, int chunk_min_exp, int chunk_max_exp, int hash_mask_bits, + int nc_level, size_t normal_size, bint high_masks, bint sparse) except 0 + cdef int fill(self) except 0 + cdef object process(self) + + # hooks used by the shared (window-less) process(); overridden by subclasses. + # window-based chunkers (buzhash64) override process() itself instead. + cdef int64_t _scan(self, const uint8_t *p, size_t n, uint64_t *digest, uint64_t mask) noexcept + cdef uint64_t _restart(self) noexcept diff --git a/src/borg/chunkers/base.pyx b/src/borg/chunkers/base.pyx new file mode 100644 index 000000000..fb10da9f6 --- /dev/null +++ b/src/borg/chunkers/base.pyx @@ -0,0 +1,260 @@ +# cython: language_level=3 + +import time + +from libc.stdint cimport uint8_t, uint64_t, int64_t +from libc.stdlib cimport malloc, free +from libc.string cimport memmove + +from ..constants import CH_DATA, CH_ALLOC, zeros +from .reader import FileReader, Chunk + +# Common driver for all content-defined chunkers: buffering, min/max clamping, +# normalized chunking parameters, sparse/hole handling and the iterator +# protocol are implemented here once. +# +# Subclasses either use the shared window-less process() by overriding the +# _scan() / _restart() hooks (fastcdc, and ChunkerPHTE for the AES chunkers), +# or override process() as a whole (window-based chunkers like buzhash64). +# The hooks are Cython cdef methods, i.e. one vtable-dispatched call per +# *scan span* (up to a whole chunk of data) and per chunk start - never per +# byte, so the indirection is not measurable. + + +cdef inline uint64_t _mask_bits(int bits, bint high): + """A mask with one-bits, in the least or most significant positions.""" + cdef uint64_t low + if bits <= 0: + return 0 + if bits >= 64: + return 0xFFFFFFFFFFFFFFFF + low = (1 << bits) - 1 + return (low << (64 - bits)) if high else low + + +cdef class ChunkerBase: + """ + Content-Defined Chunker base class: buffer and stream handling. + + Subclasses implement the cut decision, either via the _scan()/_restart() + hooks (using the shared process()) or by overriding process(). + """ + + cdef int _setup_common(self, str algo, int chunk_min_exp, int chunk_max_exp, int hash_mask_bits, + int nc_level, size_t normal_size, bint high_masks, bint sparse) except 0: + """Set up chunking parameters and the buffer (called by the subclass's __cinit__). + + high_masks selects where the cut-decision mask one-bits are placed: the low bits + (hashes with uniform output, e.g. buzhash64 or the AES chunkers) or the high bits + (Gear-style hashes, which accumulate information in their high bits). + """ + min_size = 1 << chunk_min_exp + max_size = 1 << chunk_max_exp + assert max_size <= len(zeros) + assert min_size + 1 <= max_size, "too small max_size" + + self.algo = algo + self.chunk_mask = _mask_bits(hash_mask_bits, high_masks) + self.min_size = min_size + # Normalized chunking (FastCDC-style): use a stricter mask (lower cut probability) + # until the chunk reaches its expected/normal size, then a looser mask (higher cut + # probability). This concentrates chunk sizes around the target and reduces + # chunk-size variance. nc_level == 0 disables it, keeping behavior byte-identical + # to the single-mask chunker. + assert nc_level >= 0 + assert hash_mask_bits - nc_level >= 1, "nc_level too large for hash_mask_bits" + assert hash_mask_bits + nc_level <= 48, "nc_level too large for hash_mask_bits" + self.nc_level = nc_level + if nc_level: + self.mask_s = _mask_bits(hash_mask_bits + nc_level, high_masks) + self.mask_l = _mask_bits(hash_mask_bits - nc_level, high_masks) + # normal_size is the chunk length at which we switch from the strict to the loose + # mask; it dominates the mean chunk size. The default is the nominal target size + # (1ULL << hash_mask_bits) minus the expected loose-phase tail (1ULL << (bits - nc_level)), + # which lands the mean close to the target instead of overshooting it. Pass an + # explicit normal_size to tune it further. + self.normal_size = normal_size if normal_size else ((1ULL << hash_mask_bits) - (1ULL << (hash_mask_bits - nc_level))) + else: + self.mask_s = self.chunk_mask + self.mask_l = self.chunk_mask + self.normal_size = 0 + + self.buf_size = max_size + self.data = malloc(self.buf_size) + if self.data == NULL: + raise MemoryError("Failed to allocate chunker buffer") + self.fh = -1 + self.done = 0 + self.eof = 0 + self.remaining = 0 + self.position = 0 + self.last = 0 + self.bytes_read = 0 + self.bytes_yielded = 0 + self._fd = None + self.chunking_time = 0.0 + self.reader_block_size = 1024 * 1024 + self.sparse = sparse + return 1 + + def __dealloc__(self): + if self.data != NULL: + free(self.data) + self.data = NULL + + cdef int64_t _scan(self, const uint8_t *p, size_t n, uint64_t *digest, uint64_t mask) noexcept: + """Scan n positions for a cut point. Implemented by subclasses using the shared process().""" + return -1 + + cdef uint64_t _restart(self) noexcept: + """Hash/digest state at the first scan position of a new chunk (at self.position, + which is min_size into the chunk). Window-less hashes restart from 0 (the default); + windowed hashes warm up over the preceding bytes.""" + return 0 + + cdef int fill(self) except 0: + """Fill the chunker's buffer with more data.""" + cdef ssize_t n + + # Move remaining data to the beginning of the buffer + with nogil: + memmove(self.data, self.data + self.last, self.position + self.remaining - self.last) + self.position -= self.last + self.last = 0 + n = self.buf_size - self.position - self.remaining + + if self.eof or n == 0: + return 1 + + if n > 0: + # zero-copy path: the reader writes file data (and zeros for holes) + # directly into the scan buffer - one memcpy per byte instead of + # slice/join/copy chains through intermediate bytes objects. + n = self.file_reader.readinto( + (self.data + self.position + self.remaining), n) + if n > 0: + self.remaining += n + self.bytes_read += n + else: + self.eof = 1 + return 1 + + cdef object process(self): + """Process the chunker's buffer and return the next chunk.""" + cdef uint64_t digest = 0, mask, mask_s = self.mask_s, mask_l = self.mask_l + cdef int nc_level = self.nc_level + cdef size_t n, old_last, min_size = self.min_size + cdef size_t normal_size = self.normal_size, normal_pos, chunk_len, span, did + cdef int64_t r + cdef uint8_t* p + cdef uint8_t* stop + + if self.done: + if self.bytes_read == self.bytes_yielded: + raise StopIteration + else: + raise Exception("chunkifier byte count mismatch") + + # ensure at least min_size + 1 bytes are buffered, or we are at eof + while self.remaining < min_size + 1 and not self.eof: + self.fill() + + # at eof with only a remainder (< min_size + 1): emit it as the final chunk + if self.eof and self.remaining < min_size + 1: + self.done = 1 + if self.remaining: + old_last = self.last + self.position += self.remaining + self.last = self.position + n = self.last - old_last + self.remaining = 0 + self.bytes_yielded += n + return memoryview((self.data + old_last)[:n]) + else: + if self.bytes_read == self.bytes_yielded: + raise StopIteration + else: + raise Exception("chunkifier byte count mismatch") + + # skip the sub-minimum region (no cut allowed below min_size), then scan + self.position += min_size + self.remaining -= min_size + digest = self._restart() + + while True: + chunk_len = self.position - self.last + mask = mask_s if (nc_level and chunk_len < normal_size) else mask_l + + if self.remaining == 0: + if self.eof: + break # cut at end of data + self.fill() + if self.remaining == 0: + break # buffer full -> chunk reached max_size -> forced cut + continue + + p = self.data + self.position + stop = p + self.remaining + if nc_level and chunk_len < normal_size: + # do not scan past the strict->loose transition; re-evaluate the mask there + normal_pos = self.last + normal_size + if (self.data + normal_pos) < stop: + stop = self.data + normal_pos + + span = stop - p + r = self._scan(p, span, &digest, mask) + if r == -2: + raise Exception(f"{self.algo}: OpenSSL EVP encryption failed") + if r >= 0: + did = r + 1 # cut right after the position that matched + self.position += did + self.remaining -= did + break + else: + self.position += span + self.remaining -= span + + old_last = self.last + self.last = self.position + n = self.last - old_last + self.bytes_yielded += n + return memoryview((self.data + old_last)[:n]) + + def chunkify(self, fd, fh=-1, fmap=None): + """ + Cut a file into chunks. + + :param fd: Python file object + :param fh: OS-level file handle (if available), + defaults to -1 which means not to use OS-level fd. + :param fmap: a file map, same format as generated by sparsemap + """ + self._fd = fd + self.fh = fh + self.file_reader = FileReader(fd=fd, fh=fh, read_size=self.reader_block_size, sparse=self.sparse, fmap=fmap) + self.done = 0 + self.remaining = 0 + self.bytes_read = 0 + self.bytes_yielded = 0 + self.position = 0 + self.last = 0 + self.eof = 0 + return self + + def __iter__(self): + return self + + def __next__(self): + started_chunking = time.monotonic() + data = self.process() + got = len(data) + # we do not know whether the data came from a hole in a sparse file, + # but we can just check if it was all-zero (and either came from a hole + # or from stored zeros - we can not detect that here). + if zeros.startswith(data): + data = None + allocation = CH_ALLOC + else: + allocation = CH_DATA + self.chunking_time += time.monotonic() - started_chunking + return Chunk(data, size=got, allocation=allocation) diff --git a/src/borg/chunkers/buzhash64.pyx b/src/borg/chunkers/buzhash64.pyx index a3608cc02..610e80f83 100644 --- a/src/borg/chunkers/buzhash64.pyx +++ b/src/borg/chunkers/buzhash64.pyx @@ -5,17 +5,14 @@ import os import cython -import time from cpython.bytes cimport PyBytes_AsString from libc.stdint cimport uint8_t, uint64_t from libc.stdlib cimport malloc, free -from libc.string cimport memcpy, memmove, memset from ..crypto.low_level import CSPRNG -from ..constants import CH_DATA, CH_ALLOC, CH_HOLE, zeros -from .reader import FileReader, Chunk +from .base cimport ChunkerBase cdef extern from "buzhash64_impl.h": size_t bz64_scan(const uint64_t *table, const uint64_t *table_rot, @@ -104,7 +101,7 @@ cdef uint64_t _buzhash64_update(uint64_t sum, unsigned char remove, unsigned cha return BARREL_SHIFT64(sum, 1) ^ BARREL_SHIFT64(h[remove], lenmod) ^ h[add] -cdef class ChunkerBuzHash64: +cdef class ChunkerBuzHash64(ChunkerBase): """ Content-Defined Chunker, variable chunk sizes. @@ -115,61 +112,24 @@ cdef class ChunkerBuzHash64: window contents. If the last n bits of the rolling hash are 0, a chunk is cut. Additionally it obeys some more criteria, like a minimum and maximum chunk size. It also uses a per-repo random seed to avoid some chunk length fingerprinting attacks. + + Buffering and iteration live in ChunkerBase; the window-based scan loop needs its + own process() (the shared one is for window-less hashes). """ - cdef uint64_t chunk_mask - cdef uint64_t mask_s, mask_l # normalized chunking: strict / loose masks - cdef size_t normal_size # chunk length at which we switch mask_s -> mask_l - cdef int nc_level # normalized chunking level (0 = disabled) cdef uint64_t* table cdef uint64_t* table_rot cdef int force_scalar - cdef uint8_t* data - cdef object _fd # Python object for file descriptor - cdef int fh - cdef int done, eof - cdef size_t min_size, buf_size, window_size, remaining, position, last - cdef long long bytes_read, bytes_yielded # off_t in C, using long long for compatibility - cdef readonly float chunking_time - cdef object file_reader # FileReader instance - cdef size_t reader_block_size - cdef bint sparse + cdef size_t window_size def __cinit__(self, bytes key, int chunk_min_exp, int chunk_max_exp, int hash_mask_bits, int hash_window_size, int nc_level=0, size_t normal_size=0, bint sparse=False): cdef int i_rot cdef uint64_t lenmod self.table = NULL self.table_rot = NULL - self.data = NULL - min_size = 1 << chunk_min_exp - max_size = 1 << chunk_max_exp - assert max_size <= len(zeros) - # see chunker_process, first while loop condition, first term must be able to get True: - assert hash_window_size + min_size + 1 <= max_size, "too small max_size" + # see process, first while loop condition, first term must be able to get True: + assert hash_window_size + (1 << chunk_min_exp) + 1 <= (1 << chunk_max_exp), "too small max_size" self.window_size = hash_window_size - self.chunk_mask = (1ULL << hash_mask_bits) - 1 - self.min_size = min_size - # Normalized chunking (FastCDC-style): use a stricter mask (lower cut probability) until - # the chunk reaches its expected/normal size, then a looser mask (higher cut probability). - # This concentrates chunk sizes around the target and reduces chunk-size variance. - # nc_level == 0 disables it, keeping behavior byte-identical to the single-mask chunker. - assert nc_level >= 0 - assert hash_mask_bits - nc_level >= 1, "nc_level too large for hash_mask_bits" - assert hash_mask_bits + nc_level <= 48, "nc_level too large for hash_mask_bits" - self.nc_level = nc_level - if nc_level: - self.mask_s = (1ULL << (hash_mask_bits + nc_level)) - 1 - self.mask_l = (1ULL << (hash_mask_bits - nc_level)) - 1 - # normal_size is the chunk length at which we switch from the strict to the loose - # mask; it dominates the mean chunk size. The default is the nominal target size - # (1ULL << hash_mask_bits) minus the expected loose-phase tail (1ULL << (bits - nc_level)), - # which lands the mean close to the target instead of overshooting it. Pass an - # explicit normal_size to tune it further. - self.normal_size = normal_size if normal_size else ((1ULL << hash_mask_bits) - (1ULL << (hash_mask_bits - nc_level))) - else: - self.mask_s = self.chunk_mask - self.mask_l = self.chunk_mask - self.normal_size = 0 self.table = buzhash64_init_table(key) # precomputed ROTL(table[b], window_size % 64): saves one rotate per byte # in the scan kernels (bit-identical, see buzhash64_impl.c) @@ -180,22 +140,9 @@ cdef class ChunkerBuzHash64: for i_rot in range(256): self.table_rot[i_rot] = BARREL_SHIFT64(self.table[i_rot], lenmod) self.force_scalar = 1 if os.environ.get("BORG_BUZHASH64_FORCE_SCALAR", "") not in ("", "0") else 0 - self.buf_size = max_size - self.data = malloc(self.buf_size) - if self.data == NULL: - raise MemoryError("Failed to allocate chunker buffer") - self.fh = -1 - self.done = 0 - self.eof = 0 - self.remaining = 0 - self.position = 0 - self.last = 0 - self.bytes_read = 0 - self.bytes_yielded = 0 - self._fd = None - self.chunking_time = 0.0 - self.reader_block_size = 1024 * 1024 - self.sparse = sparse + # buzhash64 output is uniform, so contiguous low-bit masks are used (high_masks=False) + self._setup_common("buzhash64", chunk_min_exp, chunk_max_exp, hash_mask_bits, + nc_level, normal_size, False, sparse) def __dealloc__(self): """Free the chunker's resources.""" @@ -205,44 +152,13 @@ cdef class ChunkerBuzHash64: if self.table_rot != NULL: free(self.table_rot) self.table_rot = NULL - if self.data != NULL: - free(self.data) - self.data = NULL @property def kernel(self): """Which scan kernel this chunker uses: 'neon', 'avx2', 'blocked' or 'scalar'.""" return (bz64_kernel_name(self.force_scalar)).decode("ascii") - cdef int fill(self) except 0: - """Fill the chunker's buffer with more data.""" - cdef ssize_t n - - # Move remaining data to the beginning of the buffer - with nogil: - memmove(self.data, self.data + self.last, self.position + self.remaining - self.last) - self.position -= self.last - self.last = 0 - n = self.buf_size - self.position - self.remaining - - if self.eof or n == 0: - return 1 - - if n > 0: - # zero-copy path: the reader writes file data (and zeros for holes) - # directly into the scan buffer - one memcpy per byte instead of - # slice/join/copy chains through intermediate bytes objects. - n = self.file_reader.readinto( - (self.data + self.position + self.remaining), n) - if n > 0: - self.remaining += n - self.bytes_read += n - else: - self.eof = 1 - - return 1 - - cdef object process(self) except *: + cdef object process(self): """Process the chunker's buffer and return the next chunk.""" cdef uint64_t sum, mask cdef uint64_t mask_s = self.mask_s, mask_l = self.mask_l @@ -332,45 +248,6 @@ cdef class ChunkerBuzHash64: # Return a memory view of the chunk return memoryview((self.data + old_last)[:n]) - def chunkify(self, fd, fh=-1, fmap=None): - """ - Cut a file into chunks. - - :param fd: Python file object - :param fh: OS-level file handle (if available), - defaults to -1 which means not to use OS-level fd. - :param fmap: a file map, same format as generated by sparsemap - """ - self._fd = fd - self.fh = fh - self.file_reader = FileReader(fd=fd, fh=fh, read_size=self.reader_block_size, sparse=self.sparse, fmap=fmap) - self.done = 0 - self.remaining = 0 - self.bytes_read = 0 - self.bytes_yielded = 0 - self.position = 0 - self.last = 0 - self.eof = 0 - return self - - def __iter__(self): - return self - - def __next__(self): - started_chunking = time.monotonic() - data = self.process() - got = len(data) - # we do not have SEEK_DATA/SEEK_HOLE support in chunker_process C code, - # but we can just check if data was all-zero (and either came from a hole - # or from stored zeros - we can not detect that here). - if zeros.startswith(data): - data = None - allocation = CH_ALLOC - else: - allocation = CH_DATA - self.chunking_time += time.monotonic() - started_chunking - return Chunk(data, size=got, allocation=allocation) - def buzhash64(data, bytes key): cdef uint64_t *table diff --git a/src/borg/chunkers/fastcdc.pyx b/src/borg/chunkers/fastcdc.pyx index 30c40fb69..7172fe94f 100644 --- a/src/borg/chunkers/fastcdc.pyx +++ b/src/borg/chunkers/fastcdc.pyx @@ -3,17 +3,14 @@ import os import cython -import time from cpython.bytes cimport PyBytes_AsString from libc.stdint cimport uint8_t, uint64_t, int64_t from libc.stdlib cimport malloc, free -from libc.string cimport memcpy, memmove, memset from ..crypto.low_level import CSPRNG -from ..constants import CH_DATA, CH_ALLOC, CH_HOLE, zeros -from .reader import FileReader, Chunk +from .base cimport ChunkerBase cdef extern from "fastcdc_impl.h": int64_t fc_scan(const uint64_t *gear, const uint8_t *p, size_t n, uint64_t *fp, uint64_t mask, int force_scalar) nogil @@ -32,6 +29,8 @@ cdef extern from "fastcdc_impl.h": # # It implements the same FastCDC techniques the buzhash64 chunker uses: sub-minimum cut-point # skipping, normalized chunking (strict/loose mask around a "normal" size), and min/max clamping. +# All of that - buffering, the scan loop, sparse/hole handling - lives in ChunkerBase; this +# class only provides the keyed Gear table and the _scan() hook calling the C kernel. # # The inner scan runs in a C kernel (fastcdc_impl.c) with SIMD implementations (NEON on # aarch64, AVX2 on x86-64, blocked scalar elsewhere) that are bit-identical to the plain @@ -59,230 +58,41 @@ cdef uint64_t* fastcdc_init_gear(bytes key) except NULL: return gear -cdef inline uint64_t _high_mask(int bits): - """A mask with one-bits in the most significant positions (Gear's strong bits).""" - if bits <= 0: - return 0 - if bits >= 64: - return 0xFFFFFFFFFFFFFFFF - return ((1 << bits) - 1) << (64 - bits) - - -cdef class ChunkerFastCDC: +cdef class ChunkerFastCDC(ChunkerBase): """ FastCDC content-defined chunker, variable chunk sizes, keyed Gear hash. Unlike the buzhash chunkers, Gear is window-less, so there is no hash_window_size parameter. """ - cdef uint64_t chunk_mask - cdef uint64_t mask_s, mask_l # normalized chunking: strict / loose masks - cdef size_t normal_size # chunk length at which we switch mask_s -> mask_l - cdef int nc_level # normalized chunking level (0 = disabled) cdef uint64_t* gear - cdef uint8_t* data - cdef object _fd - cdef int fh - cdef int done, eof - cdef size_t min_size, buf_size, remaining, position, last - cdef long long bytes_read, bytes_yielded - cdef readonly float chunking_time - cdef object file_reader - cdef size_t reader_block_size - cdef bint sparse cdef int force_scalar def __cinit__(self, bytes key, int chunk_min_exp, int chunk_max_exp, int hash_mask_bits, int nc_level=0, size_t normal_size=0, bint sparse=False): self.gear = NULL - self.data = NULL - min_size = 1 << chunk_min_exp - max_size = 1 << chunk_max_exp - assert max_size <= len(zeros) - assert min_size + 1 <= max_size, "too small max_size" - - self.chunk_mask = _high_mask(hash_mask_bits) - self.min_size = min_size - # Normalized chunking, identical structure to the buzhash64 chunker (see there), but with - # the mask one-bits placed in the high bits of the Gear hash. - assert nc_level >= 0 - assert hash_mask_bits - nc_level >= 1, "nc_level too large for hash_mask_bits" - assert hash_mask_bits + nc_level <= 48, "nc_level too large for hash_mask_bits" - self.nc_level = nc_level - if nc_level: - self.mask_s = _high_mask(hash_mask_bits + nc_level) - self.mask_l = _high_mask(hash_mask_bits - nc_level) - self.normal_size = normal_size if normal_size else ((1ULL << hash_mask_bits) - (1ULL << (hash_mask_bits - nc_level))) - else: - self.mask_s = self.chunk_mask - self.mask_l = self.chunk_mask - self.normal_size = 0 self.force_scalar = 1 if os.environ.get("BORG_FASTCDC_FORCE_SCALAR", "") not in ("", "0") else 0 self.gear = fastcdc_init_gear(key) - self.buf_size = max_size - self.data = malloc(self.buf_size) - if self.data == NULL: - raise MemoryError("Failed to allocate chunker buffer") - self.fh = -1 - self.done = 0 - self.eof = 0 - self.remaining = 0 - self.position = 0 - self.last = 0 - self.bytes_read = 0 - self.bytes_yielded = 0 - self._fd = None - self.chunking_time = 0.0 - self.reader_block_size = 1024 * 1024 - self.sparse = sparse + # Gear accumulates information in its high bits, so the cut-decision + # masks must use the high bits of the hash (high_masks=True). The Gear + # hash restarts from 0 at each chunk (window-less), so the default + # _restart() is correct. + self._setup_common("fastcdc", chunk_min_exp, chunk_max_exp, hash_mask_bits, + nc_level, normal_size, True, sparse) def __dealloc__(self): if self.gear != NULL: free(self.gear) self.gear = NULL - if self.data != NULL: - free(self.data) - self.data = NULL @property def kernel(self): """Which scan kernel this chunker uses: 'neon', 'avx2', 'blocked' or 'scalar'.""" return (fc_kernel_name(self.force_scalar)).decode("ascii") - cdef int fill(self) except 0: - """Fill the chunker's buffer with more data.""" - cdef ssize_t n - - with nogil: - memmove(self.data, self.data + self.last, self.position + self.remaining - self.last) - self.position -= self.last - self.last = 0 - n = self.buf_size - self.position - self.remaining - - if self.eof or n == 0: - return 1 - - if n > 0: - # zero-copy path: the reader writes file data (and zeros for holes) - # directly into the scan buffer - one memcpy per byte instead of - # slice/join/copy chains through intermediate bytes objects. - n = self.file_reader.readinto( - (self.data + self.position + self.remaining), n) - if n > 0: - self.remaining += n - self.bytes_read += n - else: - self.eof = 1 - return 1 - - cdef object process(self) except *: - """Process the chunker's buffer and return the next chunk.""" - cdef uint64_t fp = 0, mask, mask_s = self.mask_s, mask_l = self.mask_l - cdef int nc_level = self.nc_level - cdef size_t n, old_last, min_size = self.min_size - cdef size_t normal_size = self.normal_size, normal_pos, chunk_len, did, span + cdef int64_t _scan(self, const uint8_t *p, size_t n, uint64_t *digest, uint64_t mask) noexcept: cdef int64_t r - cdef int force_scalar = self.force_scalar - cdef uint8_t* p - cdef uint8_t* stop - cdef uint64_t* gear = self.gear - - if self.done: - if self.bytes_read == self.bytes_yielded: - raise StopIteration - else: - raise Exception("chunkifier byte count mismatch") - - # ensure at least min_size + 1 bytes are buffered, or we are at eof - while self.remaining < min_size + 1 and not self.eof: - self.fill() - - # at eof with only a remainder (< min_size + 1): emit it as the final chunk - if self.eof and self.remaining < min_size + 1: - self.done = 1 - if self.remaining: - old_last = self.last - self.position += self.remaining - self.last = self.position - n = self.last - old_last - self.remaining = 0 - self.bytes_yielded += n - return memoryview((self.data + old_last)[:n]) - else: - if self.bytes_read == self.bytes_yielded: - raise StopIteration - else: - raise Exception("chunkifier byte count mismatch") - - # skip the sub-minimum region (no cut allowed below min_size), then gear-scan - self.position += min_size - self.remaining -= min_size - fp = 0 - - while True: - chunk_len = self.position - self.last - mask = mask_s if (nc_level and chunk_len < normal_size) else mask_l - - if self.remaining == 0: - if self.eof: - break # cut at end of data - self.fill() - if self.remaining == 0: - break # buffer full -> chunk reached max_size -> forced cut - continue - - p = self.data + self.position - stop = p + self.remaining - if nc_level and chunk_len < normal_size: - # do not scan past the strict->loose transition; re-evaluate the mask there - normal_pos = self.last + normal_size - if (self.data + normal_pos) < stop: - stop = self.data + normal_pos - - span = stop - p - with nogil: - r = fc_scan(gear, p, span, &fp, mask, force_scalar) - - if r >= 0: - did = r + 1 # cut right after the byte that triggered the boundary - self.position += did - self.remaining -= did - break - else: - self.position += span - self.remaining -= span - - old_last = self.last - self.last = self.position - n = self.last - old_last - self.bytes_yielded += n - return memoryview((self.data + old_last)[:n]) - - def chunkify(self, fd, fh=-1, fmap=None): - self._fd = fd - self.fh = fh - self.file_reader = FileReader(fd=fd, fh=fh, read_size=self.reader_block_size, sparse=self.sparse, fmap=fmap) - self.done = 0 - self.remaining = 0 - self.bytes_read = 0 - self.bytes_yielded = 0 - self.position = 0 - self.last = 0 - self.eof = 0 - return self - - def __iter__(self): - return self - - def __next__(self): - started_chunking = time.monotonic() - data = self.process() - got = len(data) - if zeros.startswith(data): - data = None - allocation = CH_ALLOC - else: - allocation = CH_DATA - self.chunking_time += time.monotonic() - started_chunking - return Chunk(data, size=got, allocation=allocation) + with nogil: + r = fc_scan(self.gear, p, n, digest, mask, self.force_scalar) + return r def fastcdc_get_gear_table(bytes key): diff --git a/src/borg/chunkers/phte_chunker.pxd b/src/borg/chunkers/phte_chunker.pxd index eadca13fa..476b8d07a 100644 --- a/src/borg/chunkers/phte_chunker.pxd +++ b/src/borg/chunkers/phte_chunker.pxd @@ -1,34 +1,15 @@ # cython: language_level=3 -from libc.stdint cimport uint8_t, uint64_t, int64_t +from libc.stdint cimport uint8_t, uint64_t + +from .base cimport ChunkerBase -cdef class ChunkerPHTE: - # chunking parameters - cdef uint64_t chunk_mask - cdef uint64_t mask_s, mask_l # normalized chunking: strict / loose masks - cdef size_t normal_size # chunk length at which we switch mask_s -> mask_l - cdef int nc_level # normalized chunking level (0 = disabled) - cdef str algo # chunker name, for error messages +cdef class ChunkerPHTE(ChunkerBase): cdef str kernel_str # scan kernel the subclass selected - # buffer and stream state - cdef uint8_t* data - cdef object _fd - cdef int fh - cdef int done, eof - cdef size_t min_size, buf_size, remaining, position, last - cdef long long bytes_read, bytes_yielded - cdef readonly float chunking_time - cdef object file_reader - cdef size_t reader_block_size - cdef bint sparse - cdef int _setup(self, str algo, int chunk_min_exp, int chunk_max_exp, int hash_mask_bits, int nc_level, size_t normal_size, bint sparse) except 0 - cdef int fill(self) except 0 - cdef object process(self) # implemented by the subclasses, calling into their C kernel - cdef int64_t _scan(self, const uint8_t *p, size_t n, uint64_t *digest, uint64_t mask) noexcept cdef uint64_t _digest64(self, const uint8_t *q) noexcept diff --git a/src/borg/chunkers/phte_chunker.pyx b/src/borg/chunkers/phte_chunker.pyx index 7a969b6bf..102d392bf 100644 --- a/src/borg/chunkers/phte_chunker.pyx +++ b/src/borg/chunkers/phte_chunker.pyx @@ -1,21 +1,15 @@ # cython: language_level=3 -import time +from libc.stdint cimport uint8_t, uint64_t -from cpython.bytes cimport PyBytes_AsString -from libc.stdint cimport uint8_t, uint64_t, int64_t -from libc.stdlib cimport malloc, free -from libc.string cimport memcpy, memmove, memset - -from ..constants import CH_DATA, CH_ALLOC, CH_HOLE, zeros -from .reader import FileReader, Chunk +from .base cimport ChunkerBase # Shared chunker driver for the "UHF-then-PRF" (Chk-PHTE) chunkers: rabin-aes, # goldilocks-aes and toeplitz-aes. They differ only in their rolling universal # hash, which lives in their C kernels (see phte_core.h / phte_scan.h) and in -# the key derivation in their own modules. Everything else - buffering, -# min/max clamping, normalized chunking, the scan loop, sparse/hole handling - -# is identical and implemented here once. +# the key derivation in their own modules. Buffering, min/max clamping, +# normalized chunking and the scan loop live in ChunkerBase; this class adds +# the 64-byte-window digest warm-up and the kernel-name plumbing. # # Subclasses implement _scan() and _digest64() by calling their C kernel. Both # are Cython cdef methods, i.e. one vtable-dispatched call per *scan span* @@ -23,7 +17,7 @@ from .reader import FileReader, Chunk # indirection is not measurable. -cdef class ChunkerPHTE: +cdef class ChunkerPHTE(ChunkerBase): """ Content-Defined Chunker, variable chunk sizes, UHF-then-PRF cut decision. @@ -36,210 +30,25 @@ cdef class ChunkerPHTE: cdef int _setup(self, str algo, int chunk_min_exp, int chunk_max_exp, int hash_mask_bits, int nc_level, size_t normal_size, bint sparse) except 0: """Set up chunking parameters and the buffer (called by the subclass).""" - min_size = 1 << chunk_min_exp - max_size = 1 << chunk_max_exp - assert max_size <= len(zeros) - assert min_size + 1 <= max_size, "too small max_size" # the rolling window needs 64 bytes of in-chunk history at the first # possible cut position (min_size), so the minimum chunk size must be # at least the window size: assert chunk_min_exp >= 6, "chunk_min_exp must be >= 6 (2^6 = 64 = window size)" - - self.algo = algo self.kernel_str = "unknown" - self.chunk_mask = (1ULL << hash_mask_bits) - 1 - self.min_size = min_size - # Normalized chunking, identical structure to the fastcdc chunker (see there); - # the AES output is uniform, so contiguous low-bit masks are fine. - assert nc_level >= 0 - assert hash_mask_bits - nc_level >= 1, "nc_level too large for hash_mask_bits" - assert hash_mask_bits + nc_level <= 48, "nc_level too large for hash_mask_bits" - self.nc_level = nc_level - if nc_level: - self.mask_s = (1ULL << (hash_mask_bits + nc_level)) - 1 - self.mask_l = (1ULL << (hash_mask_bits - nc_level)) - 1 - self.normal_size = normal_size if normal_size else ((1ULL << hash_mask_bits) - (1ULL << (hash_mask_bits - nc_level))) - else: - self.mask_s = self.chunk_mask - self.mask_l = self.chunk_mask - self.normal_size = 0 - - self.buf_size = max_size - self.data = malloc(self.buf_size) - if self.data == NULL: - raise MemoryError("Failed to allocate chunker buffer") - self.fh = -1 - self.done = 0 - self.eof = 0 - self.remaining = 0 - self.position = 0 - self.last = 0 - self.bytes_read = 0 - self.bytes_yielded = 0 - self._fd = None - self.chunking_time = 0.0 - self.reader_block_size = 1024 * 1024 - self.sparse = sparse - return 1 - - def __dealloc__(self): - if self.data != NULL: - free(self.data) - self.data = NULL - - cdef int64_t _scan(self, const uint8_t *p, size_t n, uint64_t *digest, uint64_t mask) noexcept: - """Scan n positions, see the kernel's scan(). Implemented by subclasses.""" - return -1 + # the AES output is uniform, so contiguous low-bit masks are fine + return self._setup_common(algo, chunk_min_exp, chunk_max_exp, hash_mask_bits, + nc_level, normal_size, False, sparse) cdef uint64_t _digest64(self, const uint8_t *q) noexcept: """Digest of the 64 bytes at q (window warm-up). Implemented by subclasses.""" return 0 + cdef uint64_t _restart(self) noexcept: + # warm up the rolling digest over the 64 bytes preceding the first + # scan position (they are within the current chunk, since min_size >= 64). + return self._digest64(self.data + self.position - 64) + @property def kernel(self): """Which scan kernel this chunker uses: 'aes-arm64', 'aes-ni' or 'evp'.""" return self.kernel_str - - cdef int fill(self) except 0: - """Fill the chunker's buffer with more data.""" - cdef ssize_t n - - memmove(self.data, self.data + self.last, self.position + self.remaining - self.last) - self.position -= self.last - self.last = 0 - n = self.buf_size - self.position - self.remaining - - if self.eof or n == 0: - return 1 - - if n > 0: - # zero-copy path: the reader writes file data (and zeros for holes) - # directly into the scan buffer - one memcpy per byte instead of - # slice/join/copy chains through intermediate bytes objects. - n = self.file_reader.readinto( - (self.data + self.position + self.remaining), n) - if n > 0: - self.remaining += n - self.bytes_read += n - else: - self.eof = 1 - return 1 - - cdef object process(self): - """Process the chunker's buffer and return the next chunk.""" - cdef uint64_t digest = 0, mask, mask_s = self.mask_s, mask_l = self.mask_l - cdef int nc_level = self.nc_level - cdef size_t n, old_last, min_size = self.min_size - cdef size_t normal_size = self.normal_size, normal_pos, chunk_len, span, did - cdef int64_t r - cdef uint8_t* p - cdef uint8_t* stop - - if self.done: - if self.bytes_read == self.bytes_yielded: - raise StopIteration - else: - raise Exception("chunkifier byte count mismatch") - - # ensure at least min_size + 1 bytes are buffered, or we are at eof - while self.remaining < min_size + 1 and not self.eof: - self.fill() - - # at eof with only a remainder (< min_size + 1): emit it as the final chunk - if self.eof and self.remaining < min_size + 1: - self.done = 1 - if self.remaining: - old_last = self.last - self.position += self.remaining - self.last = self.position - n = self.last - old_last - self.remaining = 0 - self.bytes_yielded += n - return memoryview((self.data + old_last)[:n]) - else: - if self.bytes_read == self.bytes_yielded: - raise StopIteration - else: - raise Exception("chunkifier byte count mismatch") - - # skip the sub-minimum region (no cut allowed below min_size), then scan. - # warm up the rolling digest over the 64 bytes preceding the first - # scan position (they are within the current chunk, since min_size >= 64). - self.position += min_size - self.remaining -= min_size - digest = self._digest64(self.data + self.position - 64) - - while True: - chunk_len = self.position - self.last - mask = mask_s if (nc_level and chunk_len < normal_size) else mask_l - - if self.remaining == 0: - if self.eof: - break # cut at end of data - self.fill() - if self.remaining == 0: - break # buffer full -> chunk reached max_size -> forced cut - continue - - p = self.data + self.position - stop = p + self.remaining - if nc_level and chunk_len < normal_size: - # do not scan past the strict->loose transition; re-evaluate the mask there - normal_pos = self.last + normal_size - if (self.data + normal_pos) < stop: - stop = self.data + normal_pos - - span = stop - p - r = self._scan(p, span, &digest, mask) - if r == -2: - raise Exception(f"{self.algo}: OpenSSL EVP encryption failed") - if r >= 0: - did = r + 1 # cut right after the position that matched - self.position += did - self.remaining -= did - break - else: - self.position += span - self.remaining -= span - - old_last = self.last - self.last = self.position - n = self.last - old_last - self.bytes_yielded += n - return memoryview((self.data + old_last)[:n]) - - def chunkify(self, fd, fh=-1, fmap=None): - """ - Cut a file into chunks. - - :param fd: Python file object - :param fh: OS-level file handle (if available), - defaults to -1 which means not to use OS-level fd. - :param fmap: a file map, same format as generated by sparsemap - """ - self._fd = fd - self.fh = fh - self.file_reader = FileReader(fd=fd, fh=fh, read_size=self.reader_block_size, sparse=self.sparse, fmap=fmap) - self.done = 0 - self.remaining = 0 - self.bytes_read = 0 - self.bytes_yielded = 0 - self.position = 0 - self.last = 0 - self.eof = 0 - return self - - def __iter__(self): - return self - - def __next__(self): - started_chunking = time.monotonic() - data = self.process() - got = len(data) - if zeros.startswith(data): - data = None - allocation = CH_ALLOC - else: - allocation = CH_DATA - self.chunking_time += time.monotonic() - started_chunking - return Chunk(data, size=got, allocation=allocation)