synology_apm_repo.sdk.dedup.chunk_walk module

The shared chunk-map walking engine behind bucket-major (physical-order) export.

plan_chunks_windowed (Pass 1) groups each DATA chunk placement in a range by the (stream_id, bucket_id) it physically lives in, as compact ChunkRuns rather than per-chunk entries; exec_chunks (Pass 2) visits those groups bucket-major and decodes each unique chunk once, writing through a caller-supplied on_run(dest_offset, data) callback rather than a hardcoded os.pwrite().

This module is the one deliberate exception to ``DedupFile._extents()`` being private: every call here carries its own # noqa: SLF001 because bucket-major planning genuinely needs the chunk-native fields (addr/map_num/repeat) that DedupFile.read deliberately never exposes. Any other module reaching for _extents() wants plaintext for a byte window (read/stream), not a raw chunk-map record.

class synology_apm_repo.sdk.dedup.chunk_walk.ChunkRun(chunk_idx_start, length, dest_offset_start)

Bases: object

A maximal run of physically-contiguous chunks within one bucket: chunk_idx_start, chunk_idx_start + 1, …, chunk_idx_start + length - 1, mapping to dest_offset_start, dest_offset_start + FIXED_CHUNK_LENGTH, … in destination order.

Preserves a ChunkMapKind.MAPPING record’s own compactness (an address template of map_num chunks replayed 1 + repeat times, FORMAT-SPEC.md: ChunkMapRecord) instead of expanding every replay into a per-chunk entry — collapses millions of chunk placements into a run count several orders of magnitude smaller. Plain @dataclass, not a hot-path NamedTuple like ChunkAddress — there are only thousands of these per export, not millions.

chunk_idx_start: int

length: int

dest_offset_start: int

class synology_apm_repo.sdk.dedup.chunk_walk.ChunkPlan(groups, holes, zeros)

Bases: object

Pass 1’s output: every DATA chunk placement in the walked range, grouped by the (stream_id, bucket_id) it actually lives in — ready for exec_chunks to visit bucket-major. holes/ zeros are byte totals only (their zero-fill, if wanted, is already written by plan_chunks_windowed itself during the same extents() walk that groups the DATA chunks, rather than deferred to a separate pass that would have to re-walk or re-store every ZERO/HOLE extent for no benefit).

groups: dict[tuple[StreamId, BucketId], list[ChunkRun]]

holes: int

zeros: int

synology_apm_repo.sdk.dedup.chunk_walk.DEFAULT_WINDOW_ENTRIES = 1048576

One window bounds max_entries total chunks (not runs — see plan_chunks_windowed), regardless of how few ChunkRuns that expands to — a run-based plan for a real 32 GiB VM’s ~2.5M chunks costs a few hundred KB across every bucket group, not the ~20 MB a flat per-chunk array would, but is still unbounded in principle for however large a range a caller hands it. plan_chunks_windowed bounds peak plan memory to one window’s worth of chunks regardless of total range size.

async synology_apm_repo.sdk.dedup.chunk_walk.plan_chunks_windowed(base, start, end, window_start, *, write_zero_fill, max_entries=1048576)

One extents() walk: DATA chunks are grouped as ChunkRuns for exec_chunks; ZERO/HOLE regions are resolved immediately — calling write_zero_fill(local_offset, length) if given (the caller’s job to decide whether that means writing actual zero bytes or doing nothing, e.g. a sparse destination that’s already zero-filled), or just counted if write_zero_fill is None. Yields a ChunkPlan every max_entries chunks (counting each ChunkRun’s own length, not 1 per run) so peak plan memory is bounded regardless of how compactly the runs themselves happen to pack. Always yields at least one plan, possibly empty (a range that is entirely HOLE/ZERO). A window boundary may fall mid-run: a run that would cross max_entries is split at the boundary, its own remainder carried into the next window.

ZERO/HOLE extents are clipped to [start, end); a boundary DATA extent is not, because chunks are the atomic decode unit — any chunk overlapping [start, end) is decoded in full, only a chunk entirely outside it is skipped (same first_k/last_k bound as _fill_data_extent).

window_start must be a multiple of FIXED_CHUNK_LENGTH and <= start, or this raises ValueError — a non-aligned window_start would silently misalign every dest_offset, and one past start makes an included chunk’s dest_offset negative.

async synology_apm_repo.sdk.dedup.chunk_walk.count_planned_bytes(base, start, end)

How many bytes plan_chunks_windowed would plan across [start, end) — i.e. the DATA extents’ total expanded (repeat-multiplied) chunk count times FIXED_CHUNK_LENGTH, the exact same number exec_chunks computes as its own planned_total from a single (unwindowed) plan’s groups.

A second, O(1)-memory extents() walk purely for this count (never building any placement array). Windowed planning’s own per-window planned_total resets at every window boundary, so a progress bar built on it would look like it restarts partway through; the progress denominator has to be the planned work across the whole range. This extra walk is cheap relative to the decode pass it makes accurate.

async synology_apm_repo.sdk.dedup.chunk_walk.exec_chunks(plan, *, pool, on_run, size, export_cache=None, progress=None, max_concurrent_opens=1, max_concurrent_reads=1)

Pass 2: visits buckets ascending, chunks ascending within each, decoding each unique chunk once. Two independent merges happen around decode: the source side merges adjacent chunks’ still-compressed byte ranges into one ObjectStore.read per BucketReader.read_chunks; the destination side merges consecutive dest_offset chunks into one buffer per on_run call (see _exec_one_bucket_group).

Parameters:

• max_concurrent_reads (int) – The single knob for every kind of read concurrency this function produces, cross-bucket or in-bucket (default 1: serial) — one asyncio.Semaphore sized to it is shared by the cross-bucket dispatch loop below and by each bucket’s own in-bucket fan-out inside BucketReader.read_chunks, where the caller’s pre-acquired permit covers that call’s first merged run and only an additional run within the same bucket acquires a further permit from the same pool. on_run must tolerate concurrent, out-of-order calls at scattered offsets when this is > 1 (_ExportSink’s os.pwrite already does).

• max_concurrent_opens (int) – Runs _prefetch_bucket_opens in the background to warm export_cache ahead of the dispatch loop (default 1: off) — orthogonal to max_concurrent_reads, since it only ever touches bucket opens, never on_run/decode/write.

• export_cache (BucketReaderCache | None) – Default None creates one here (bounded to DEFAULT_BUCKET_CACHE_SIZE, the same 16-bucket cap Pool itself defaults to, reused here as one shared constant instead of a separately hardcoded copy), discarded at the end of this call — share one explicitly across a whole batch via export_to instead. Safe to pass to every concurrent _exec_one_bucket_group call unmodified: plan.groups keys each task on its own distinct (stream_id, bucket_id), so two concurrent groups never touch the same cache entry.

Note

Cancellation is native asyncio.CancelledError, arriving at the next await; with max_concurrent_reads > 1 and more than one bucket group in plan, a TaskGroup supervises the concurrent groups, so a caller-visible exception then arrives wrapped in an ExceptionGroup rather than as the original exception type — a real shape difference from the single-group/ max_concurrent_reads=1 path that an opted-in caller must be ready for.