borgbackup/borg
1
0
Fork 0
mirror of https://github.com/borgbackup/borg.git synced 2026-10-12 02:11:36 +00:00
Code Issues Packages Projects Releases Wiki Activity

chunkers: refactor shared machinery into ChunkerBase

Browse Source 
The CDC chunkers each carried a private copy of the same buffering
machinery: fill(), chunkify(), __iter__/__next__, the buffer/stream
state, the normalized-chunking parameter setup and the buffer part of
__dealloc__ - about 340 near-identical lines across buzhash, buzhash64,
fastcdc and ChunkerPHTE. Every buffering change (e.g. the recent
zero-copy fill) had to be made in four places.

Promote the ChunkerPHTE shape into a general ChunkerBase (base.pyx +
base.pxd for cross-module cdef inheritance):

- state + _setup_common() (sizes, buffer, counters, NC masks - with a
  high_masks flag choosing high-bit (Gear) or low-bit (uniform hash)
  mask placement), fill(), chunkify(), __iter__(), __next__(), and the
  buffer __dealloc__ live in ChunkerBase, once.
- the window-less process() also moves into the base, parameterized by
  the _scan() / _restart() hooks (one virtual call per scan span and
  per chunk start, never per byte).
- ChunkerFastCDC becomes a ChunkerBase subclass and drops its own
  process() entirely: it only keeps the keyed Gear table, the kernel
  property and a _scan() hook (295 -> 105 lines).
- ChunkerPHTE shrinks to the 64-byte digest warm-up (_restart ->
  _digest64) and kernel-name plumbing (245 -> 54 lines); rabin-aes,
  goldilocks-aes and toeplitz-aes are unchanged.
- ChunkerBuzHash64 inherits everything but keeps its window-based
  process() (400 -> 277 lines).
- buzhash (32-bit) is left untouched: it is frozen borg 1.x
  compatible legacy.

The only C-level difference: the memmove in what was ChunkerPHTE's
fill() now runs with the GIL released, like the other copies already
did.

Net -210 lines. No behavior change: chunk cut points verified
bit-identical per-chunker over 1 GiB of synthetic data for all six CDC
chunkers, chunker test suite passes. Throughput unchanged (the hook
indirection is per-span, not per-byte).

Co-Authored-By: Claude Opus 5 <[email protected]>
This commit is contained in:
Thomas Waldmann
2026-08-05 01:50:24 +02:00
co-authored by Claude Opus 5
parent ac96b1de72
commit cc9e5a7828
9 changed files with 350 additions and 570 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/changes.rst
+4
View File
@@ -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
scripts/borg.exe.spec
+5 -3
View File
@@ -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
setup.py
+3
View File
@@ -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",
src/borg/chunkers/base.pxd
+34
View File
@@ -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
src/borg/chunkers/base.pyx
+260
View File
@@ -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 <bits> one-bits, in the least or most significant positions."""
cdef uint64_t low
if bits <= 0:
return 0
if bits >= 64:
return <uint64_t>0xFFFFFFFFFFFFFFFF
low = (<uint64_t>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 = <uint8_t*>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(
<uint8_t[:n]>(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 = <size_t>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)
src/borg/chunkers/buzhash64.pyx
+12 -135
View File
@@ -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 = <uint8_t*>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 (<bytes>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(
<uint8_t[:n]>(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
src/borg/chunkers/fastcdc.pyx
+14 -204
View File
@@ -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 <bits> one-bits in the most significant positions (Gear's strong bits)."""
if bits <= 0:
return 0
if bits >= 64:
return <uint64_t>0xFFFFFFFFFFFFFFFF
return ((<uint64_t>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 = <uint8_t*>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 (<bytes>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(
<uint8_t[:n]>(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 = <size_t>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):
src/borg/chunkers/phte_chunker.pxd
+4 -23
View File
@@ -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
src/borg/chunkers/phte_chunker.pyx
+14 -205
View File
@@ -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 = <uint8_t*>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 <prefix>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(
<uint8_t[:n]>(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 = <size_t>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)