diff --git a/docs/design/capacity_compaction.md b/docs/design/capacity_compaction.md new file mode 100644 index 00000000..e19a6e15 --- /dev/null +++ b/docs/design/capacity_compaction.md @@ -0,0 +1,338 @@ +# Capacity Compaction for AdaptiveArrayPools (`compact!`) + +- **Date:** 2026-06-28 +- **Status:** Implemented (CPU + Metal + CUDA), Julia 1.12+. Shipped on top of the + `trim!` work. The Timer-driven auto mode (§9) remains future work. +- **Scope:** Manual `compact!` API that shrinks over-allocated backing buffers in + place. CPU + Metal + CUDA parity; an optional Timer-driven auto mode is deferred. +- **Author:** design session + +--- + +## 1. Motivation + +`trim!` (the sibling feature) drops the pool's references to **inactive** slots — +it reduces the *number of retained slots*. It deliberately never touches active +slots, so it cannot help when the bloated buffer is one the caller is still +holding. + +The remaining gap: a slot's backing buffer keeps the **high-water-mark capacity** +it ever needed. A rare large input grows a slot to (say) 1,000,000 elements; every +later `acquire!` on that slot only uses 100, but the buffer stays at 1,000,000 +because `resize!` down never frees capacity (verified: `resize!(v, 100)` leaves +`capacity == 1_000_000`). That over-allocation is invisible to `trim!` whenever +the slot is active. + +`compact!` fills that gap: + +> For buffers whose allocated capacity is far larger than the size currently in +> use, shrink the capacity back down to a small multiple of the used size — +> reclaiming the outlier bloat while keeping a right-sized buffer for reuse. + +This is a **cold-path** operation: called rarely (manually, or eventually on a +slow timer), not in the hot `acquire!`/`rewind!` loop. The design therefore +favors simplicity and safety over maximal reclamation. + +## 2. Relationship to `trim!` — two orthogonal axes + +| | `trim!` (exists) | `compact!` (this doc) | +| --- | --- | --- | +| Reduces | number of retained slots (structural) | over-allocated capacity of a slot's buffer | +| Touches | inactive slots only | retained buffers, **including active ones** | +| Safety | always safe (no live references) | must re-sync live wrappers (see §7) | +| Verb family | `reset!` / `empty!` / `trim!` | adds `compact!` | + +They are independent: `trim!` removes the tail; `compact!` compacts what remains. +Keeping them as **separate functions** (rather than a `trim!(; shrink_factor=…)` +kwarg) keeps each operation single-purpose and preserves `trim!`'s "always safe" +guarantee. The opt-in is layered: calling `compact!` opts into capacity compaction +of inactive slots (Tier 1, safe); `compact!(…; active=true)` further opts into +compacting active slots (Tier 2, §7). + +## 3. Core concept: capacity vs logical length + +For a backing vector `v = tp.vectors[slot]`: + +- **used** = the slot's **current logical extent**, recorded per-slot in + `tp.slot_extents[slot]` by `_claim_slot!` (`_slot_used`). This is the size of the + most recent claim and covers **both** acquisition paths — `acquire!` (Array + wrapper) *and* `acquire_view!` (which returns an **uncached** `SubArray`/ + `ReshapedArray`). **Not** `length(v)`: `_claim_slot!` only ever *grows* the backing + (`if length(vec) < n; resize!`, `acquire.jl:73`), so `length(v)` is the **high-water + mark**. An earlier design read `used` from the cached wrappers only; that + **under-counted a live view's extent** (views cache no wrapper) and let + `compact!(active=true)` shrink a backing below a held view → OOB. `slot_extents` + fixes this (and, being a `Vector{Int}`, makes `_slot_used` return a concrete `Int`, + keeping the gate arithmetic allocation-free). (Verified by TDD.) +- **capacity** = allocated buffer length (`length(getfield(v, :ref).mem)` on 1.12; + wrapped in `_slot_capacity`). + +`compact!` shrinks `capacity` toward `used`; it never changes `used` (held wrappers +keep their `:size`, and `target ≥ used` so every wrapper stays in bounds). + +A slot is a `compact!` candidate when **both**: + +``` +capacity ≥ factor × used # ratio gate: "≥10× bloated" +(capacity − target) × sizeof(T) ≥ min_bytes # absolute gate: real reclaim + where target = ceil(shrink_to × used) +``` + +The absolute byte gate prevents pointless reallocation of small buffers (e.g. a +1000-capacity / 50-used buffer is 20× bloated but only reclaims ~7.6 KB). + +## 4. API (as implemented) + +Mirror `trim!`'s surface for consistency: + +```julia +compact!(pool::AdaptiveArrayPool; + factor::Real = 10, # shrink only if capacity ≥ factor × used + shrink_to::Real = 1.5, # new capacity = ceil(shrink_to × used) + min_bytes::Int = 2^20, # …and only if ≥ this many bytes are reclaimed + active::Bool = true, # default: also compact ACTIVE slots (§7); false = inactive-only + force_gc::Bool = false) +compact!(pool::AdaptiveArrayPool, ::Type{T}; kwargs...) # single type +compact!(pool::AdaptiveArrayPool, types::Type...; kwargs...) # varargs (mirrors trim!) +compact!(; kwargs...) # task-local CPU pool +``` + +GPU extensions, `DisabledPool`, and the legacy (< 1.12) path mirror `trim!`'s +overload set exactly (legacy = warn-once no-op). Returns the same shape of +summary as `trim!`, plus a compaction-specific field: + +```julia +(; slots_compacted, bytes_reclaimed, gc_triggered) +``` + +(Reuse the `_trim_summary` pattern — a single `_compact_summary` helper — so every +overload returns one concrete, inferrable `NamedTuple`. See the `trim!` PR's +type-stability work for why this matters.) + +## 5. Shrink mechanism — in-place backing shrink (preserve object identity) + +**Decision (revised):** shrink the backing vector **in place** — keep the same +`Vector` object and swap only its internal `Memory` for a new, smaller one. Do +**not** replace the slot object (`tp.vectors[slot] = nv`). + +Use an **explicit `Memory` swap**, not `sizehint!`. `sizehint!` is documented as an +*advisory hint* — it is not guaranteed to shrink the buffer (the amount is +implementation/version-defined), so it cannot be the mechanism for a reclamation +API. The explicit swap is deterministic: + +```julia +target = ceil(Int, shrink_to * used) # target ≥ used +nv = Vector{T}(undef, target) # new small buffer (guaranteed size) +copyto!(nv, 1, v, 1, used) # preserve the live `used` elements +setfield!(v, :size, (used,)) # logical length stays `used` +setfield!(v, :ref, getfield(nv, :ref)) # v keeps identity; its Memory is now nv's +# then re-sync every cached wrapper for this slot (see §7) +``` + +This is the same `setfield!(:ref/:size)` the pool already uses for wrapper reuse, +so it is 1.12+-only (legacy < 1.12 → no-op, like `trim!`). + +Why keep the object and swap `:ref`, rather than "allocate a new vector and assign +the slot" (the earlier draft)? A smoke test +(`scratchpad/manual_refswap.jl`) confirmed the difference for live +`acquire_view!` views: + +- **Replace object** (`tp.vectors[slot] = nv`): a `SubArray`'s `parent` still points + at the *old* vector object → the view does **not** follow → broken. +- **Swap `:ref` in place**: the vector object keeps its identity and only its `:ref` + changes; a `SubArray` reads through `parent` on every access, so it **follows the + new buffer automatically** (verified: identity kept, `capacity 1M→5`, view reads + the new buffer, old `Memory` GC-collected). + +So **both** wrappers (re-synced, §7) **and** views (follow `parent`) survive. + +A bounds note: a view is only valid up to the slot's current `used` length, so +shrinking capacity toward `used` never invalidates a view that was valid at the +current size (a stale wider view was already out of bounds before compaction). + +**Does the old buffer actually get freed?** Yes — verified. The §5 `Memory` swap +re-points `v.ref` at a new smaller `Memory`; a `WeakRef` to the old `Memory` is dead +after `GC.gc()`, confirming it is reclaimed. One catch: +each `Array` **wrapper carries its own `:ref`** still pointing at the old `Memory`, +which pins it. So full reclamation requires **both** the in-place shrink **and** the +wrapper re-sync (§7.1) — skipping the re-sync leaves the wrapper stale *and* keeps +the old buffer alive. (`acquire_view!` views do not pin the old buffer: they read +through `parent`, which now points at the new one.) + +Note this is a different operation from `trim!`. `trim!` shrinks the *outer slot +array* (`resize!(tp.vectors, n_active)`), dropping whole inactive backing-`Vector` +objects — it reduces the slot **count** and returns those buffers 100%. `compact!` +keeps the slot and shrinks its buffer's **capacity** in place. They are +complementary (remove unused slots vs. right-size the buffers of slots still in +use), not two spellings of one thing. + +## 6. Modular layering (for the future auto-manager) + +The reusable unit is the per-slot primitive; everything else composes it: + +``` +_maybe_compact_slot!(tp, slot, factor, shrink_to, min_bytes) -> reclaimed::Int + │ one slot: gate check → (if candidate) in-place shrink + wrapper re-sync + │ returns bytes reclaimed (0 if skipped) + ▼ +compact!(pool; …) -> summary + │ scan all typed pools' retained slots, call the primitive, accumulate + ▼ +auto-manager (Phase C): + Timer fires → sets a "compaction requested" flag (does NOT touch buffers) + → the next pool touch point on the user's task runs the deferred compaction +``` + +The auto-manager writes **no new reclamation logic** — it only *schedules* a call +to `compact!`/`_maybe_compact_slot!`, which then runs synchronously on the user's +task at the next touch point (§7.3). The `factor`/`min_bytes` gates make it +naturally "leave the common case alone, only touch true outliers." + +That touch point may have `n_active > 0`, so the auto path **can** compact active +slots (Tier 2), not just inactive ones — safely, because the in-place swap + +re-sync runs single-threaded at a moment the user is inside a pool call rather than +mid-array-access. (This is the whole point of cooperative deferral: it lets the +timer reach the long-lived *active* outliers the user cares about, without the data +race a direct background mutation would cause.) + +## 7. Safety + +### 7.1 Two tiers — and why all the risk lives in Tier 2 + +The entire safety story splits cleanly by slot state: + +- **Tier 1 — inactive slots (default).** No live user reference exists. Shrinking + is unconditionally safe; the only bookkeeping is clearing the slot's cached + wrappers (they get rebuilt on the next `acquire!`). A plain `compact!(pool)` does + only this. +- **Tier 2 — active slots (opt-in, `active=true`).** The user is currently holding + the array from this slot, so the swap must re-sync the wrapper (§7.1) and is only + done at a point where the user is not mid-access — either an explicit mid-scope + manual call, or the auto path's cooperative deferral (§7.3). With the in-place + strategy (§5) both wrappers and views stay valid, so the gating concern here is + **concurrency, not references**. + +**Why Tier 2 is possible at all** (the non-obvious part): the array `acquire!` +returns is **not a copy** — it *is* the pool's own cached wrapper object, the same +one stored in `tp.arr_wrappers[N][slot]`. (Verified: `a === arr_wrappers[1][1]`.) +So the pool is not blind to "what the user holds" — it handed out its own objects +and still references every one of them. The wrapper is a thin handle that only +*points* at the backing buffer via a mutable `:ref`; the data lives in the buffer. +So after the in-place backing shrink (§5), the pool re-points the wrapper: + +```julia +# v = tp.vectors[slot]; after the §5 in-place Memory swap +for wrappers in tp.arr_wrappers # one per dimensionality N + w = wrappers === nothing ? continue : wrappers[slot] + w === nothing && continue + setfield!(w, :ref, getfield(v, :ref)) # re-point to v's new (smaller) Memory + # :size unchanged — logical dims are preserved +end +``` + +The wrapper carries its **own** `:ref` (a copy taken at creation), so it does not +follow `v`'s in-place change on its own and must be re-synced. The user holds the +*same* `w` object, so repointing its `:ref` redirects the user's array — no +reassignment in user code. This is the **same `setfield!(:ref)` the pool already +performs on every `acquire!`**, so the mechanism is load-bearing, not new. + +### 7.2 Views survive too — under the in-place strategy + +With the §5 in-place shrink, escaped `acquire_view!` results are **no longer a +hazard**. A `SubArray` / `ReshapedArray` holds its `parent` — the backing **vector +object** — and reads through it on every access. Because in-place shrink keeps that +object's identity and only swaps its `:ref`, the view transparently follows to the +new buffer (verified in the smoke test). Both reference kinds survive: + +| held reference | how it survives an in-place compaction | +| --- | --- | +| `acquire!` / `reshape!` `Array` wrapper | pool re-syncs its own `:ref` (§7.1) | +| `acquire_view!` `SubArray` / `ReshapedArray` | follows `parent`'s in-place `:ref` swap automatically | + +This is exactly why §5 chose in-place over replace-object: the *replace-object* +alternative (`tp.vectors[slot] = nv`) strands views on the old object (stale read + +pinned old buffer). Staying in-place is a **correctness requirement** for Tier 2, +not a stylistic preference. + +The only residual view caveat is bounds: never shrink capacity below the slot's +current `used` length — which the policy never does (`target ≥ used`). + +### 7.3 Concurrency — the real constraint for auto (Phase C) + +References are solved (§7.1–7.2). The genuine wall for **active** compaction is +concurrency: a `Timer` callback that reallocates a buffer while the user's task is +reading/writing it is a data race no reference trick can fix — lost writes, +use-after-free on the freed old `Memory`, a torn `:ref`. This is independent of +wrappers vs views. + +**The only safe auto design is cooperative deferral:** + +- The `Timer` does **not** touch buffers. It sets a "compaction requested" flag + (plus the policy). +- The actual shrink runs **synchronously on the user's task** at the next pool + touch point (`acquire!` / `checkpoint!` / `rewind!`, or an explicit + `maybe_compact!(pool)` the user can sprinkle into a long compute). At that instant + the user is inside a pool call, not mid-array-access, so the in-place swap + + re-sync is single-threaded-safe — and it can compact **active** slots, not just + inactive ones. +- Honest limitation: a buffer held across a long compute with **no** intervening + pool call cannot be compacted until the next touch point. Reclamation is delayed, + never unsafe. With `Threads.nthreads() > 1`, stricter synchronization is required + if pool calls can race across threads. + +So the user's goal — "a timer that reclaims oversized **active** outliers" — is +achievable: the timer **schedules** the compaction for the next safe touch point +rather than performing it directly. + +## 8. Backend parity + +- **CPU (`Vector{T}`):** the explicit `Memory` swap of §5. CPU's native `resize!` + *never* reallocates on shrink (capacity is retained forever), so the explicit + swap is the only way to actually return the memory. +- **CUDA / Metal:** backing is `CuVector` / `MtlArray`. Same shape — allocate a + smaller **device** buffer, copy `used` elements device-to-device, swap the + `:data`/DataRef + `:maxsize` + `:dims` (keeping the array object's identity, as on + CPU), then re-sync the GPU wrappers. Mind the GPU refcount rules the `trim!` work + established (Metal passes the `DataRef` directly; CUDA's constructor takes + ownership). Crucially, this must **not** route through the pool's + `_resize_to_fit!`, which deliberately *preserves* the GPU allocation + (`maxsize`) for reuse — `compact!` is exactly the op that overrides that and frees + it. Metal is hardware-testable on Apple Silicon; CUDA via CI. + + Note the symmetry: CUDA.jl's native `resize!` (v5.10+) already reallocates a + device buffer when it shrinks below `capacity ÷ 4` (`acquire.jl:18`). That is the + same idea as `compact!`'s `factor` gate — "realloc once `used` drops below a + fraction of capacity" — just with a fixed 25% threshold instead of our + configurable `factor` (e.g. `factor=10` ⇒ realloc below 10%). `compact!` brings + that capacity-aware behavior to CPU (which has none) and makes the threshold a + policy knob on both backends. +- **Legacy (< 1.12):** defined no-op that warns once and returns a zero summary, + exactly like legacy `trim!` (the `setfield!(:ref)` mechanism is 1.12+-only). + +## 9. Phasing + +1. **Phase A — CPU `compact!`** + the `_maybe_compact_slot!` primitive, full TDD: + capacity actually reclaimed, `used` data preserved, live wrapper still valid + after compaction, gates respected (ratio + min_bytes), no-op when nothing is + bloated, type-stable summary, varargs/per-type/DisabledPool/legacy parity. +2. **Phase B — Metal + CUDA parity** (Metal validated on hardware). +3. **Phase C — auto mode (deferred; out of scope for the first PR).** When taken + up: an external policy decides *when* and flips a "compaction requested" flag; + the actual compaction runs at a safe pool-scope boundary — concretely, the end + of a `@with_pool` block (the `rewind!`/scope-finalization the macro already + emits), not an arbitrary instant. That point is single-threaded-safe and may have + `n_active > 0`, so it can also reach active outliers (§7.3). Driving it from a + `Timer` is one possible policy source, but the **trigger is decoupled from the + execution point** — the timer only flags; the macro boundary executes. + +## 10. Open questions / decisions still to make + +- **Default `min_bytes`** — `2^20` (1 MiB) is a starting guess; tune against a + benchmark before settling. +- **Default `factor` / `shrink_to`** — `factor=10`, `shrink_to=1.5` are starting + guesses; CUDA.jl's native 25% (`÷4`) is a reference point for `factor`. +- **Whether `compact!` should also drop fully-inactive slots** (i.e. subsume + `trim!`) or stay strictly capacity-only and compose with `trim!`. Current lean: + stay capacity-only; users call `trim!(pool); compact!(pool)` for both axes. +- **Per-type policy** — whether `factor`/`shrink_to` should ever be settable + per element type, or stay global per call. Default: global per call (simpler). diff --git a/ext/AdaptiveArrayPoolsCUDAExt/acquire.jl b/ext/AdaptiveArrayPoolsCUDAExt/acquire.jl index 564f4ee0..0ed23880 100644 --- a/ext/AdaptiveArrayPoolsCUDAExt/acquire.jl +++ b/ext/AdaptiveArrayPoolsCUDAExt/acquire.jl @@ -96,6 +96,7 @@ CUDA.jl's resize! unnecessarily (especially after safety invalidation sets dims= idx = tp.n_active if idx > length(tp.vectors) push!(tp.vectors, allocate_vector(tp, total_len)) + push!(tp.slot_extents, total_len) # parallel to `vectors` (for compact!/_slot_used) _check_pool_growth(tp, idx) else # _resize_to_fit! handles all cases: @@ -103,6 +104,7 @@ CUDA.jl's resize! unnecessarily (especially after safety invalidation sets dims= # - n != length: setfield!(:dims) — restores length after safety invalidation # - n == length: no-op (hot path) _resize_to_fit!(@inbounds(tp.vectors[idx]), total_len) + @inbounds tp.slot_extents[idx] = total_len # current logical extent end return idx end @@ -119,7 +121,10 @@ the wrapper points to a different array's memory via `setfield!(:data)`. idx = tp.n_active if idx > length(tp.vectors) push!(tp.vectors, CuVector{T}(undef, 0)) + push!(tp.slot_extents, 0) # placeholder slot: wrapper points elsewhere (reshape!) _check_pool_growth(tp, idx) + else + @inbounds tp.slot_extents[idx] = 0 end return idx end diff --git a/ext/AdaptiveArrayPoolsCUDAExt/state.jl b/ext/AdaptiveArrayPoolsCUDAExt/state.jl index f2a7542a..1bf7ab67 100644 --- a/ext/AdaptiveArrayPoolsCUDAExt/state.jl +++ b/ext/AdaptiveArrayPoolsCUDAExt/state.jl @@ -317,6 +317,7 @@ CUDA.reclaim() function Base.empty!(tp::CuTypedPool) empty!(tp.vectors) empty!(tp.arr_wrappers) + empty!(tp.slot_extents) tp.n_active = 0 # Restore sentinel values empty!(tp._checkpoint_n_active) @@ -429,3 +430,122 @@ end return AdaptiveArrayPools._trim_summary(counts, force_gc) end end + +# ============================================================================== +# compact! for CuAdaptiveArrayPool (parity with CPU / Metal) +# ============================================================================== +# +# Shrinks an over-allocated device buffer in place. Like Metal (and unlike CPU), +# CUDA has no uncached views — acquire_view! == acquire! → cached CuArray wrapper — +# so a slot's live references are exactly its cached wrappers; compaction swaps the +# backing buffer's DataRef and re-syncs those wrappers via the same refcount helper +# used on grow (`_update_cuda_wrapper_data!`). `slot_extents` (set in +# `_cuda_claim_slot!`) gives the live extent, matching the CPU `_slot_used` model. +# Device capacity uses `_aligned_sizeof(T)` (CUDA buffer alignment), like `_resize_to_fit!`. + +# Device capacity (elements) of a CUDA backing buffer. +@inline AdaptiveArrayPools._slot_capacity(v::CuVector{T}) where {T} = + Int(getfield(v, :maxsize) ÷ _aligned_sizeof(T)) + +# Current logical extent of a slot (records both acquire!/acquire_view! since CUDA +# routes both through `_cuda_claim_slot!`). `::Int` keeps the gate allocation-free. +@inline function AdaptiveArrayPools._slot_used(tp::CuTypedPool, slot::Int)::Int + ext = tp.slot_extents + return slot <= length(ext) ? (@inbounds ext[slot]) : 0 +end + +# Keep `slot_extents` parallel to `vectors` when `trim!` truncates a CUDA pool. +@inline function AdaptiveArrayPools._trim_slot_extents!(tp::CuTypedPool, keep::Int) + length(tp.slot_extents) > keep && resize!(tp.slot_extents, keep) + return nothing +end + +# Shrink one slot's device buffer to capacity `target` in place: allocate a smaller +# CuVector, copy the live `used` elements, swap the backing's DataRef (keeping the +# backing object's identity), re-sync cached wrappers, and drop the temporary's own +# reference. All buffer ownership goes through `_update_cuda_wrapper_data!` +# (unsafe_free! old + copy/retain new), so the old device buffer's refcount reaches +# zero and CUDA frees it. +function AdaptiveArrayPools._compact_slot!(tp::CuTypedPool{T}, slot::Int, target::Int, used::Int) where {T} + v = @inbounds tp.vectors[slot] + nv = CuVector{T}(undef, target) + used > 0 && copyto!(nv, 1, v, 1, used) + _update_cuda_wrapper_data!(v, nv) # backing takes nv's buffer (refcounted) + setfield!(v, :dims, (target,)) + for wrappers in tp.arr_wrappers + wrappers === nothing && continue + slot <= length(wrappers) || continue + w = @inbounds wrappers[slot] + w === nothing && continue + _update_cuda_wrapper_data!(w, v) # held wrapper re-points to new buffer (dims kept) + end + unsafe_free!(getfield(nv, :data)) # drop the temporary's own reference + return nothing +end + +# Gate + compact one CUDA slot. Returns reclaimed device bytes (0 if skipped). +function AdaptiveArrayPools._maybe_compact_slot!(tp::CuTypedPool{T}, slot::Int, factor::Real, shrink_to::Real, min_bytes::Int) where {T} + used = AdaptiveArrayPools._slot_used(tp, slot) + used == 0 && return 0 + cap = AdaptiveArrayPools._slot_capacity(@inbounds tp.vectors[slot]) + cap >= factor * used || return 0 + target = max(used, ceil(Int, min(Float64(cap), shrink_to * used))) # clamp to [used, cap] + reclaim = (cap - target) * _aligned_sizeof(T) + reclaim >= min_bytes || return 0 + AdaptiveArrayPools._compact_slot!(tp, slot, target, used) + return reclaim +end + +# Guard + compact for one element type on CUDA (mirrors `_trim_one_counts!`): never +# creates a pool for a never-used fallback type. +@inline function AdaptiveArrayPools._compact_one_counts!(pool::CuAdaptiveArrayPool, ::Type{T}, factor::Real, shrink_to::Real, min_bytes::Int, active::Bool)::NTuple{2, Int} where {T} + (!(T <: _CUDA_FIXED_TYPES) && !haskey(pool.others, T)) && return (0, 0) + return AdaptiveArrayPools._compact_counts!(AdaptiveArrayPools.get_typed_pool!(pool, T), factor, shrink_to, min_bytes, active) +end + +function AdaptiveArrayPools.compact!( + pool::CuAdaptiveArrayPool; + factor::Real = 10, shrink_to::Real = 1.5, min_bytes::Int = 2^20, + active::Bool = true, force_gc::Bool = false, + ) + # Mirror the CUDA `trim!`: Ref accumulators keep the `foreach_fixed_slot` closure + # box-free over the CUDA fixed-slot set, so the returned summary stays concrete. + slots = Ref(0) + bytes = Ref(0) + AdaptiveArrayPools.foreach_fixed_slot(pool) do tp + c = AdaptiveArrayPools._compact_counts!(tp, factor, shrink_to, min_bytes, active) + slots[] += c[1] + bytes[] += c[2] + end + oc = AdaptiveArrayPools._compact_others_counts!(values(pool.others), factor, shrink_to, min_bytes, active) + force_gc && (GC.gc(); CUDA.reclaim()) + return AdaptiveArrayPools._compact_summary((slots[] + oc[1], bytes[] + oc[2]), force_gc) +end + +@inline function AdaptiveArrayPools.compact!( + pool::CuAdaptiveArrayPool, ::Type{T}; + factor::Real = 10, shrink_to::Real = 1.5, min_bytes::Int = 2^20, + active::Bool = true, force_gc::Bool = false, + ) where {T} + counts = AdaptiveArrayPools._compact_one_counts!(pool, T, factor, shrink_to, min_bytes, active) + force_gc && (GC.gc(); CUDA.reclaim()) + return AdaptiveArrayPools._compact_summary(counts, force_gc) +end + +@generated function AdaptiveArrayPools.compact!( + pool::CuAdaptiveArrayPool, types::Type...; + factor::Real = 10, shrink_to::Real = 1.5, min_bytes::Int = 2^20, + active::Bool = true, force_gc::Bool = false, + ) + n = length(types) + syms = [Symbol(:c, i) for i in 1:n] + assigns = [:($(syms[i]) = AdaptiveArrayPools._compact_one_counts!(pool, types[$i], factor, shrink_to, min_bytes, active)) for i in 1:n] + counts_expr = n == 0 ? :((0, 0)) : + Expr(:tuple, (Expr(:call, :+, (:($(s)[$j]) for s in syms)...) for j in 1:2)...) + return quote + $(assigns...) + counts = $counts_expr + force_gc && (GC.gc(); CUDA.reclaim()) + return AdaptiveArrayPools._compact_summary(counts, force_gc) + end +end diff --git a/ext/AdaptiveArrayPoolsCUDAExt/types.jl b/ext/AdaptiveArrayPoolsCUDAExt/types.jl index 5a437161..f3908540 100644 --- a/ext/AdaptiveArrayPoolsCUDAExt/types.jl +++ b/ext/AdaptiveArrayPoolsCUDAExt/types.jl @@ -36,6 +36,11 @@ mutable struct CuTypedPool{T} <: AbstractTypedPool{T, CuVector{T}} # --- N-D Wrapper Cache (setfield!-based reuse, matches CPU TypedPool) --- arr_wrappers::Vector{Union{Nothing, Vector{Any}}} # index=N (dimensionality), value=per-slot CuArray{T,N} + # --- Per-slot current logical extent (parallel to `vectors`, matches CPU TypedPool) --- + # Size of each slot's most recent `_cuda_claim_slot!`; `compact!` reads it + # (`_slot_used`) to know how much of an over-allocated device buffer is in use. + slot_extents::Vector{Int} + # --- State Management (1-based sentinel pattern) --- n_active::Int _checkpoint_n_active::Vector{Int} @@ -46,6 +51,7 @@ function CuTypedPool{T}() where {T} return CuTypedPool{T}( CuVector{T}[], # vectors Union{Nothing, Vector{Any}}[], # arr_wrappers (indexed by N) + Int[], # slot_extents (parallel to vectors) 0, [0], [0] # State (1-based sentinel) ) end diff --git a/ext/AdaptiveArrayPoolsMetalExt/acquire.jl b/ext/AdaptiveArrayPoolsMetalExt/acquire.jl index 93a9a6cc..2c2a0560 100644 --- a/ext/AdaptiveArrayPoolsMetalExt/acquire.jl +++ b/ext/AdaptiveArrayPoolsMetalExt/acquire.jl @@ -80,9 +80,11 @@ Metal.jl's resize! unnecessarily (especially after safety invalidation sets dims idx = tp.n_active if idx > length(tp.vectors) push!(tp.vectors, allocate_vector(tp, total_len)) + push!(tp.slot_extents, total_len) # parallel to `vectors` (for compact!/_slot_used) _check_pool_growth(tp, idx) else _resize_to_fit!(@inbounds(tp.vectors[idx]), total_len) + @inbounds tp.slot_extents[idx] = total_len # current logical extent end return idx end @@ -99,7 +101,10 @@ the wrapper points to a different array's memory via `setfield!(:data)`. idx = tp.n_active if idx > length(tp.vectors) push!(tp.vectors, MtlArray{T, 1, S}(undef, 0)) + push!(tp.slot_extents, 0) # placeholder slot: wrapper points elsewhere (reshape!) _check_pool_growth(tp, idx) + else + @inbounds tp.slot_extents[idx] = 0 end return idx end diff --git a/ext/AdaptiveArrayPoolsMetalExt/state.jl b/ext/AdaptiveArrayPoolsMetalExt/state.jl index de2ceb81..5d5e5083 100644 --- a/ext/AdaptiveArrayPoolsMetalExt/state.jl +++ b/ext/AdaptiveArrayPoolsMetalExt/state.jl @@ -299,6 +299,7 @@ Actual VRAM release depends on GC + Metal.jl's memory pool. function Base.empty!(tp::MetalTypedPool) empty!(tp.vectors) empty!(tp.arr_wrappers) + empty!(tp.slot_extents) tp.n_active = 0 # Restore sentinel values empty!(tp._checkpoint_n_active) @@ -403,3 +404,122 @@ end return AdaptiveArrayPools._trim_summary(counts, force_gc) end end + +# ============================================================================== +# compact! for MetalAdaptiveArrayPool (parity with CPU) +# ============================================================================== +# +# Shrinks an over-allocated device buffer in place. Unlike CPU, Metal has no +# uncached views (`acquire_view!` == `acquire!` → cached `MtlArray` wrapper), so a +# slot's live references are exactly its cached wrappers; compaction swaps the +# backing buffer's `DataRef` and re-syncs those wrappers via the same refcount +# helper used on grow (`_update_metal_wrapper_data!`). `slot_extents` (set in +# `_metal_claim_slot!`) gives the live extent, matching the CPU `_slot_used` model. + +# Device capacity (elements) of a Metal backing buffer. +@inline AdaptiveArrayPools._slot_capacity(v::MtlArray{T}) where {T} = + Int(getfield(v, :maxsize) ÷ sizeof(T)) + +# Current logical extent of a slot (records both acquire!/acquire_view! since Metal +# routes both through `_metal_claim_slot!`). `::Int` keeps the gate allocation-free. +@inline function AdaptiveArrayPools._slot_used(tp::MetalTypedPool, slot::Int)::Int + ext = tp.slot_extents + return slot <= length(ext) ? (@inbounds ext[slot]) : 0 +end + +# Keep `slot_extents` parallel to `vectors` when `trim!` truncates a Metal pool. +@inline function AdaptiveArrayPools._trim_slot_extents!(tp::MetalTypedPool, keep::Int) + length(tp.slot_extents) > keep && resize!(tp.slot_extents, keep) + return nothing +end + +# Shrink one slot's device buffer to capacity `target` in place: allocate a smaller +# MtlArray, copy the live `used` elements, swap the backing's `DataRef` (keeping the +# backing object's identity), re-sync cached wrappers, and drop the temporary's own +# reference. All buffer ownership goes through `_update_metal_wrapper_data!` +# (unsafe_free! old + copy/retain new), so the old device buffer's refcount reaches +# zero and Metal frees it. +function AdaptiveArrayPools._compact_slot!(tp::MetalTypedPool{T, S}, slot::Int, target::Int, used::Int) where {T, S} + v = @inbounds tp.vectors[slot] + nv = MtlArray{T, 1, S}(undef, target) + used > 0 && copyto!(nv, 1, v, 1, used) + _update_metal_wrapper_data!(v, nv) # backing takes nv's buffer (refcounted) + setfield!(v, :dims, (target,)) + for wrappers in tp.arr_wrappers + wrappers === nothing && continue + slot <= length(wrappers) || continue + w = @inbounds wrappers[slot] + w === nothing && continue + _update_metal_wrapper_data!(w, v) # held wrapper re-points to new buffer (dims kept) + end + unsafe_free!(getfield(nv, :data)) # drop the temporary's own reference + return nothing +end + +# Gate + compact one Metal slot. Returns reclaimed device bytes (0 if skipped). +function AdaptiveArrayPools._maybe_compact_slot!(tp::MetalTypedPool{T, S}, slot::Int, factor::Real, shrink_to::Real, min_bytes::Int) where {T, S} + used = AdaptiveArrayPools._slot_used(tp, slot) + used == 0 && return 0 + cap = AdaptiveArrayPools._slot_capacity(@inbounds tp.vectors[slot]) + cap >= factor * used || return 0 + target = max(used, ceil(Int, min(Float64(cap), shrink_to * used))) # clamp to [used, cap] + reclaim = (cap - target) * sizeof(T) + reclaim >= min_bytes || return 0 + AdaptiveArrayPools._compact_slot!(tp, slot, target, used) + return reclaim +end + +# Guard + compact for one element type on Metal (mirrors `_trim_one_counts!`): never +# creates a pool for a never-used fallback type. +@inline function AdaptiveArrayPools._compact_one_counts!(pool::MetalAdaptiveArrayPool, ::Type{T}, factor::Real, shrink_to::Real, min_bytes::Int, active::Bool)::NTuple{2, Int} where {T} + (!(T <: _METAL_FIXED_TYPES) && !haskey(pool.others, T)) && return (0, 0) + return AdaptiveArrayPools._compact_counts!(AdaptiveArrayPools.get_typed_pool!(pool, T), factor, shrink_to, min_bytes, active) +end + +function AdaptiveArrayPools.compact!( + pool::MetalAdaptiveArrayPool; + factor::Real = 10, shrink_to::Real = 1.5, min_bytes::Int = 2^20, + active::Bool = true, force_gc::Bool = false, + ) + # Mirror the Metal `trim!`: Ref accumulators keep the `foreach_fixed_slot` + # closure box-free over the Metal-specific fixed-slot set, so the returned + # summary stays a concrete NamedTuple. + slots = Ref(0) + bytes = Ref(0) + AdaptiveArrayPools.foreach_fixed_slot(pool) do tp + c = AdaptiveArrayPools._compact_counts!(tp, factor, shrink_to, min_bytes, active) + slots[] += c[1] + bytes[] += c[2] + end + oc = AdaptiveArrayPools._compact_others_counts!(values(pool.others), factor, shrink_to, min_bytes, active) + force_gc && GC.gc() + return AdaptiveArrayPools._compact_summary((slots[] + oc[1], bytes[] + oc[2]), force_gc) +end + +@inline function AdaptiveArrayPools.compact!( + pool::MetalAdaptiveArrayPool, ::Type{T}; + factor::Real = 10, shrink_to::Real = 1.5, min_bytes::Int = 2^20, + active::Bool = true, force_gc::Bool = false, + ) where {T} + counts = AdaptiveArrayPools._compact_one_counts!(pool, T, factor, shrink_to, min_bytes, active) + force_gc && GC.gc() + return AdaptiveArrayPools._compact_summary(counts, force_gc) +end + +@generated function AdaptiveArrayPools.compact!( + pool::MetalAdaptiveArrayPool, types::Type...; + factor::Real = 10, shrink_to::Real = 1.5, min_bytes::Int = 2^20, + active::Bool = true, force_gc::Bool = false, + ) + n = length(types) + syms = [Symbol(:c, i) for i in 1:n] + assigns = [:($(syms[i]) = AdaptiveArrayPools._compact_one_counts!(pool, types[$i], factor, shrink_to, min_bytes, active)) for i in 1:n] + counts_expr = n == 0 ? :((0, 0)) : + Expr(:tuple, (Expr(:call, :+, (:($(s)[$j]) for s in syms)...) for j in 1:2)...) + return quote + $(assigns...) + counts = $counts_expr + force_gc && GC.gc() + return AdaptiveArrayPools._compact_summary(counts, force_gc) + end +end diff --git a/ext/AdaptiveArrayPoolsMetalExt/types.jl b/ext/AdaptiveArrayPoolsMetalExt/types.jl index a04e196c..ab6156b6 100644 --- a/ext/AdaptiveArrayPoolsMetalExt/types.jl +++ b/ext/AdaptiveArrayPoolsMetalExt/types.jl @@ -34,6 +34,11 @@ mutable struct MetalTypedPool{T, S} <: AbstractTypedPool{T, MtlArray{T, 1, S}} # --- N-D Wrapper Cache (setfield!-based reuse, matches CPU TypedPool) --- arr_wrappers::Vector{Union{Nothing, Vector{Any}}} # index=N (dimensionality), value=per-slot MtlArray{T,N,S} + # --- Per-slot current logical extent (parallel to `vectors`, matches CPU TypedPool) --- + # Size of each slot's most recent `_metal_claim_slot!`; `compact!` reads it + # (`_slot_used`) to know how much of an over-allocated device buffer is in use. + slot_extents::Vector{Int} + # --- State Management (1-based sentinel pattern) --- n_active::Int _checkpoint_n_active::Vector{Int} @@ -44,6 +49,7 @@ function MetalTypedPool{T, S}() where {T, S} return MetalTypedPool{T, S}( MtlArray{T, 1, S}[], # vectors Union{Nothing, Vector{Any}}[], # arr_wrappers (indexed by N) + Int[], # slot_extents (parallel to vectors) 0, [0], [0], # State (1-based sentinel) ) end diff --git a/src/AdaptiveArrayPools.jl b/src/AdaptiveArrayPools.jl index 959d241e..53bd61c2 100644 --- a/src/AdaptiveArrayPools.jl +++ b/src/AdaptiveArrayPools.jl @@ -16,7 +16,7 @@ export Bit # Sentinel type for BitArray (use with acquire!, trues!, falses!) export @with_pool, @maybe_with_pool, @safe_with_pool, @safe_maybe_with_pool export STATIC_POOLING, MAYBE_POOLING, RUNTIME_CHECK export PoolEscapeError, EscapePoint, PoolMutationError, MutationPoint -export checkpoint!, rewind!, reset!, trim! +export checkpoint!, rewind!, reset!, trim!, compact! export get_task_local_cuda_pool, get_task_local_cuda_pools # CUDA (stubs, overridden by extension) export get_task_local_metal_pool, get_task_local_metal_pools # Metal (stubs, overridden by extension) diff --git a/src/acquire.jl b/src/acquire.jl index 067a90b8..a02c03e0 100644 --- a/src/acquire.jl +++ b/src/acquire.jl @@ -67,12 +67,14 @@ Returns the slot index. This is the shared primitive for all acquisition paths idx = tp.n_active if idx > length(tp.vectors) push!(tp.vectors, allocate_vector(tp, n)) + push!(tp.slot_extents, n) # parallel to `vectors`; records this slot's extent _check_pool_growth(tp, idx) else @inbounds vec = tp.vectors[idx] if length(vec) < n resize!(vec, n) end + @inbounds tp.slot_extents[idx] = n # current logical extent (for compact!/_slot_used) end return idx end @@ -89,7 +91,10 @@ the wrapper points to a different array's memory via `setfield!(:ref)`. idx = tp.n_active if idx > length(tp.vectors) push!(tp.vectors, Vector{T}(undef, 0)) + push!(tp.slot_extents, 0) # placeholder slot: backing points elsewhere (reshape!) _check_pool_growth(tp, idx) + else + @inbounds tp.slot_extents[idx] = 0 end return idx end diff --git a/src/legacy/state.jl b/src/legacy/state.jl index c8253e98..7993fd2a 100644 --- a/src/legacy/state.jl +++ b/src/legacy/state.jl @@ -790,3 +790,66 @@ trim!(; force_gc::Bool = false) = trim!(get_task_local_pool(); force_gc = force_ trim!(::DisabledPool; force_gc::Bool = false) = _ZERO_TRIM_SUMMARY trim!(::DisabledPool, ::Type{T}; force_gc::Bool = false) where {T} = _ZERO_TRIM_SUMMARY trim!(::DisabledPool, types::Type...; force_gc::Bool = false) = _ZERO_TRIM_SUMMARY + +# ============================================================================== +# compact! — no-op on the legacy path (Julia < 1.12) +# ============================================================================== +# +# Modern `compact!` shrinks over-allocated backing capacity in place via +# `setfield!(:ref/:size)`, which only exists on Julia 1.12+. The legacy pool uses a +# different N-way cache layout, so `compact!` is a defined-and-exported NO-OP here +# (returns a zero summary, warns once) to keep the public API callable across the +# full supported Julia range. Upgrade to Julia 1.12+ for actual capacity compaction. + +const _ZERO_COMPACT_SUMMARY = (; + slots_compacted = 0, bytes_reclaimed = 0, gc_triggered = false, +) + +function compact!( + pool::AdaptiveArrayPool; + factor::Real = 10, shrink_to::Real = 1.5, min_bytes::Int = 2^20, + active::Bool = true, force_gc::Bool = false + ) + @warn "compact! is a no-op on Julia < 1.12 (legacy pool architecture). " * + "Upgrade to Julia 1.12+ for capacity compaction." maxlog = 1 + return _ZERO_COMPACT_SUMMARY +end + +compact!( + pool::AdaptiveArrayPool, ::Type{T}; + factor::Real = 10, shrink_to::Real = 1.5, min_bytes::Int = 2^20, + active::Bool = true, force_gc::Bool = false +) where {T} = + compact!(pool; factor = factor, shrink_to = shrink_to, min_bytes = min_bytes, active = active, force_gc = force_gc) + +compact!( + pool::AdaptiveArrayPool, types::Type...; + factor::Real = 10, shrink_to::Real = 1.5, min_bytes::Int = 2^20, + active::Bool = true, force_gc::Bool = false +) = + compact!(pool; factor = factor, shrink_to = shrink_to, min_bytes = min_bytes, active = active, force_gc = force_gc) + +compact!(; + factor::Real = 10, shrink_to::Real = 1.5, min_bytes::Int = 2^20, + active::Bool = true, force_gc::Bool = false +) = + compact!( + get_task_local_pool(); + factor = factor, shrink_to = shrink_to, min_bytes = min_bytes, active = active, force_gc = force_gc +) + +compact!( + ::DisabledPool; + factor::Real = 10, shrink_to::Real = 1.5, min_bytes::Int = 2^20, + active::Bool = true, force_gc::Bool = false +) = _ZERO_COMPACT_SUMMARY +compact!( + ::DisabledPool, ::Type{T}; + factor::Real = 10, shrink_to::Real = 1.5, min_bytes::Int = 2^20, + active::Bool = true, force_gc::Bool = false +) where {T} = _ZERO_COMPACT_SUMMARY +compact!( + ::DisabledPool, types::Type...; + factor::Real = 10, shrink_to::Real = 1.5, min_bytes::Int = 2^20, + active::Bool = true, force_gc::Bool = false +) = _ZERO_COMPACT_SUMMARY diff --git a/src/state.jl b/src/state.jl index 34fad009..48a8891c 100644 --- a/src/state.jl +++ b/src/state.jl @@ -508,6 +508,7 @@ Restores sentinel values for 1-based sentinel pattern. function Base.empty!(tp::TypedPool) empty!(tp.vectors) empty!(tp.arr_wrappers) + empty!(tp.slot_extents) tp.n_active = 0 # Restore sentinel values (1-based sentinel pattern) empty!(tp._checkpoint_n_active) @@ -786,6 +787,14 @@ function _inactive_storage_bytes(tp::AbstractTypedPool, first::Int, last::Int) return total end +# Truncate `tp.slot_extents` parallel to `tp.vectors` when trimming. `TypedPool` carries +# the per-slot extent record; other pools (e.g. `BitTypedPool`) do not — no-op for them. +_trim_slot_extents!(::AbstractTypedPool, ::Int) = nothing +@inline function _trim_slot_extents!(tp::TypedPool, keep::Int) + length(tp.slot_extents) > keep && resize!(tp.slot_extents, keep) + return nothing +end + # Trim one typed pool: drop inactive backing-vector and wrapper references by # truncating `tp.vectors` and the wrapper caches to the active count. Returns the # raw `(slots, wrappers, bytes)` counts (bytes measured before truncation). The @@ -800,6 +809,7 @@ function _trim_counts!(tp::AbstractTypedPool)::NTuple{3, Int} if released > 0 bytes = _inactive_storage_bytes(tp, keep + 1, old_len) resize!(tp.vectors, keep) + _trim_slot_extents!(tp, keep) # keep the extent record parallel to `vectors` end return (max(0, released), wrappers_released, bytes) end @@ -839,6 +849,249 @@ function _trim_others_counts!(others)::NTuple{3, Int} return (slots, wrappers, bytes) end +# ============================================================================== +# State Management - compact! primitive (in-place capacity compaction) +# ============================================================================== +# +# A slot's backing `Vector` length is the *high-water mark* (the largest size ever +# acquired — `_claim_slot!` only grows it). The *current logical size* is recorded +# per-slot in `tp.slot_extents` by `_claim_slot!`. So a slot is "bloated" when its +# backing capacity is far larger than the size actually in use. `compact!` returns +# that excess by swapping the backing's `Memory` for a right-sized one in place +# (keeping the `Vector` object's identity, so views following `parent` stay valid) +# and re-syncing the cached wrappers' `:ref`. + +# Allocated backing capacity (Memory length) for a CPU backing vector. +@inline _slot_capacity(v::Vector) = length(getfield(v, :ref).mem) + +# Current logical extent of a slot = the size compaction must preserve. This is the +# size of the slot's most recent `_claim_slot!`, recorded in `tp.slot_extents` for +# BOTH the `acquire!` (Array wrapper) and `acquire_view!` (uncached SubArray / +# ReshapedArray) paths — so a live view's extent is never under-counted (the bug a +# wrapper-only scan had: `acquire_view!` caches no wrapper). Returns `::Int` — a +# concrete `Vector{Int}` read, keeping the gate arithmetic in `_maybe_compact_slot!` +# allocation-free (a `length(::Any)` wrapper scan boxed it). 0 if unrecorded. +@inline function _slot_used(tp::TypedPool, slot::Int)::Int + ext = tp.slot_extents + return slot <= length(ext) ? (@inbounds ext[slot]) : 0 +end + +# Shrink one slot's backing to capacity `target` in place: allocate a smaller +# `Memory`, copy the live `used` elements, swap the backing `Vector`'s `:ref`/`:size` +# (identity preserved → views follow), then re-sync the slot's cached wrappers. +function _compact_slot!(tp::TypedPool{T}, slot::Int, target::Int, used::Int) where {T} + v = @inbounds tp.vectors[slot] + nv = Vector{T}(undef, target) + copyto!(nv, 1, v, 1, used) + setfield!(v, :size, (target,)) + setfield!(v, :ref, getfield(nv, :ref)) + for wrappers in tp.arr_wrappers + wrappers === nothing && continue + slot <= length(wrappers) || continue + w = @inbounds wrappers[slot] + w === nothing && continue + setfield!(w, :ref, getfield(v, :ref)) # :size unchanged + end + return nothing +end + +# Gate + compact one slot. Compacts iff backing capacity is `≥ factor × used` AND +# the reclaim is `≥ min_bytes`. Returns bytes reclaimed (0 if skipped). The generic +# fallback (e.g. `BitTypedPool`, whose backing is not a `Vector`) is a no-op. +_maybe_compact_slot!(::AbstractTypedPool, ::Int, ::Real, ::Real, ::Int) = 0 +function _maybe_compact_slot!(tp::TypedPool{T}, slot::Int, factor::Real, shrink_to::Real, min_bytes::Int) where {T} + used = _slot_used(tp, slot) + used == 0 && return 0 + cap = _slot_capacity(@inbounds tp.vectors[slot]) + cap >= factor * used || return 0 + # Clamp the shrink goal to `[used, cap]`: never below the live size, and never + # above capacity (shrinking to ≥ cap is a no-op). Capping the Float before `ceil` + # also guards against an absurd `shrink_to` overflowing `Int` (InexactError). + target = max(used, ceil(Int, min(Float64(cap), shrink_to * used))) + reclaim = (cap - target) * sizeof(T) + reclaim >= min_bytes || return 0 + _compact_slot!(tp, slot, target, used) + return reclaim +end + +# Public summary NamedTuple from raw `(slots, bytes)` counts and the gc flag. Single +# definition point for the `compact!` return shape — every overload funnels through +# here, so they all return the identical concrete +# `@NamedTuple{slots_compacted::Int, bytes_reclaimed::Int, gc_triggered::Bool}`. +@inline _compact_summary(c::NTuple{2, Int}, gc::Bool) = (; + slots_compacted = c[1], bytes_reclaimed = c[2], gc_triggered = gc, +) + +# Zero summary: returned by no-op compactions (DisabledPool, nothing bloated). +const _ZERO_COMPACT_SUMMARY = _compact_summary((0, 0), false) + +# Compact one typed pool's slots, gating each via `_maybe_compact_slot!`. Returns the +# raw `(slots_compacted, bytes_reclaimed)` counts; the `::NTuple{2, Int}` return keeps +# callers concrete even through the `others` dynamic dispatch. +# +# `active=true` (default): scan ALL slots (`1 : end`), so even slots the user is still +# holding are compacted. This is safe because `_compact_slot!` re-syncs every cached +# wrapper's `:ref` (and the backing keeps its `Vector` identity, so views follow), and +# `_maybe_compact_slot!`'s `target ≥ used` guarantee never drops the live elements. +# `active=false`: scan only INACTIVE slots (`n_active+1 : end`), the opt-out. +function _compact_counts!(tp::AbstractTypedPool, factor::Real, shrink_to::Real, min_bytes::Int, active::Bool)::NTuple{2, Int} + slots = 0 + bytes = 0 + start = active ? 1 : (tp.n_active + 1) + for slot in start:length(tp.vectors) + r = _maybe_compact_slot!(tp, slot, factor, shrink_to, min_bytes) + if r > 0 + slots += 1 + bytes += r + end + end + return (slots, bytes) +end + +# Sum `_compact_counts!` over every fixed slot, unrolled at compile time — same +# anti-boxing rationale as `_trim_fixed_counts!`: a closure fold would box the +# accumulators (Core.Box) and widen the summary to `Any`; this stays `NTuple{2, Int}`. +@generated function _compact_fixed_counts!(pool::AdaptiveArrayPool, factor::Real, shrink_to::Real, min_bytes::Int, active::Bool) + syms = [Symbol(:c, i) for i in 1:length(FIXED_SLOT_FIELDS)] + assigns = [ + :($(syms[i]) = _compact_counts!(getfield(pool, $(QuoteNode(f))), factor, shrink_to, min_bytes, active)) + for (i, f) in enumerate(FIXED_SLOT_FIELDS) + ] + sums = [Expr(:call, :+, [:($(s)[$j]) for s in syms]...) for j in 1:2] + return quote + Base.@_inline_meta + $(assigns...) + ($(sums...),) + end +end + +# Sum `_compact_counts!` over the `others` (non-fixed-slot) typed pools. A plain loop +# (no closure) avoids boxing; the `::NTuple{2, Int}` assertion keeps the running total +# concrete despite dynamic dispatch over the `Any`-typed iterable. +function _compact_others_counts!(others, factor::Real, shrink_to::Real, min_bytes::Int, active::Bool)::NTuple{2, Int} + slots = 0 + bytes = 0 + for tp in others + c = _compact_counts!(tp, factor, shrink_to, min_bytes, active)::NTuple{2, Int} + slots += c[1] + bytes += c[2] + end + return (slots, bytes) +end + +# Guard + compact for one element type, returning raw `(slots, bytes)`. Like +# `_trim_one_counts!`, never *creates* a pool: a never-used fallback type is skipped +# so a reclamation call does not surprise-register a fallback pool in `pool.others`. +function _compact_one_counts!(pool::AdaptiveArrayPool, ::Type{T}, factor::Real, shrink_to::Real, min_bytes::Int, active::Bool)::NTuple{2, Int} where {T} + (!(T <: _FIXED_SLOT_TYPES) && !haskey(pool.others, T)) && return (0, 0) + return _compact_counts!(get_typed_pool!(pool, T), factor, shrink_to, min_bytes, active) +end + +""" + compact!(pool::AdaptiveArrayPool; + factor::Real = 10, shrink_to::Real = 1.5, min_bytes::Int = 2^20, + active::Bool = true, force_gc::Bool = false) + +Shrink the **over-allocated capacity** of backing buffers in place. A slot's backing +length is the high-water mark (the largest size ever acquired); its current logical +size is recorded per-slot in `slot_extents` (set by the shared claim path, so it covers +both `acquire!` and `acquire_view!`). A slot is *bloated* when its retained capacity is +`≥ factor ×` that logical size; `compact!` swaps such a backing's `Memory` for one of +size `ceil(shrink_to × used)` (keeping the `Vector`'s identity so views following +`parent` stay valid) and re-syncs the cached wrappers' `:ref`. + +Orthogonal to [`trim!`](@ref): `trim!` drops whole inactive *slots*, `compact!` shrinks +the *capacity* of retained ones. + +- **`active=true`** (default): also compact **active** slots — buffers the caller is still + holding — not just inactive ones. This is `compact!`'s reason for existing over `trim!`: + it reclaims peak capacity from arrays still in use. The held wrapper and any `view` of it + follow the in-place swap (wrapper `:ref` re-synced; backing identity preserved), and the + `target ≥ used` guarantee never drops live elements. The `factor` gate (below) means a + normally-sized live array is never touched — only grossly over-allocated ones. Call it at + a synchronous point where no other task is mid-access to this pool's arrays. +- **`active=false`**: opt out — touch only **inactive** slots (`n_active+1 : end`), i.e. + buffers with no live array references. + +A slot is compacted only if both gates pass: `capacity ≥ factor × used` **and** the +reclaim `(capacity − target) × sizeof(T) ≥ min_bytes` (default 1 MiB), so tiny buffers +and reasonably-sized live arrays are not churned. Returns +`(; slots_compacted, bytes_reclaimed, gc_triggered)`. + +`force_gc=true` runs `GC.gc()` after the swaps to make the detached buffers collectable. + +!!! note "Julia version" + Requires Julia 1.12+ (uses `setfield!(:ref/:size)`). On older Julia (the legacy + pool architecture) `compact!` is a defined no-op that returns a zero summary and + warns once, so dependent packages compile across the supported Julia range. + +See also: [`trim!`](@ref), [`reset!`](@ref), [`empty!`](@ref). +""" +function compact!( + pool::AdaptiveArrayPool; + factor::Real = 10, shrink_to::Real = 1.5, min_bytes::Int = 2^20, + active::Bool = true, force_gc::Bool = false + ) + counts = _compact_fixed_counts!(pool, factor, shrink_to, min_bytes, active) .+ + _compact_others_counts!(pool._others_values, factor, shrink_to, min_bytes, active) + force_gc && GC.gc() + return _compact_summary(counts, force_gc) +end + +""" + compact!(pool::AdaptiveArrayPool, ::Type{T}; kwargs...) + +Compact slots for a single element type `T` only. See [`compact!`](@ref). +""" +function compact!( + pool::AdaptiveArrayPool, ::Type{T}; + factor::Real = 10, shrink_to::Real = 1.5, min_bytes::Int = 2^20, + active::Bool = true, force_gc::Bool = false + ) where {T} + counts = _compact_one_counts!(pool, T, factor, shrink_to, min_bytes, active) + force_gc && GC.gc() + return _compact_summary(counts, force_gc) +end + +""" + compact!(pool::AdaptiveArrayPool, types::Type...; kwargs...) + +Compact slots for several element types at once, running `GC.gc()` (when `force_gc=true`) +a single time after all listed types are compacted. Returns one combined summary. Uses +`@generated` for zero-overhead compile-time unrolling, mirroring [`trim!`](@ref). +See [`compact!`](@ref). +""" +@generated function compact!( + pool::AdaptiveArrayPool, types::Type...; + factor::Real = 10, shrink_to::Real = 1.5, min_bytes::Int = 2^20, + active::Bool = true, force_gc::Bool = false + ) + n = length(types) + syms = [Symbol(:c, i) for i in 1:n] + assigns = [:($(syms[i]) = _compact_one_counts!(pool, types[$i], factor, shrink_to, min_bytes, active)) for i in 1:n] + counts_expr = n == 0 ? :((0, 0)) : + Expr(:tuple, (Expr(:call, :+, (:($(s)[$j]) for s in syms)...) for j in 1:2)...) + return quote + $(assigns...) + counts = $counts_expr + force_gc && GC.gc() + return _compact_summary(counts, force_gc) + end +end + +""" + compact!(; kwargs...) + +Compact slots of the current task-local pool (`get_task_local_pool()`). +""" +compact!(; + factor::Real = 10, shrink_to::Real = 1.5, min_bytes::Int = 2^20, + active::Bool = true, force_gc::Bool = false, +) = compact!( + get_task_local_pool(); + factor = factor, shrink_to = shrink_to, min_bytes = min_bytes, active = active, force_gc = force_gc +) + """ trim!(pool::AdaptiveArrayPool; force_gc::Bool = false) @@ -945,3 +1198,19 @@ Base.empty!(::DisabledPool) = nothing trim!(::DisabledPool; force_gc::Bool = false) = _ZERO_TRIM_SUMMARY trim!(::DisabledPool, ::Type{T}; force_gc::Bool = false) where {T} = _ZERO_TRIM_SUMMARY trim!(::DisabledPool, types::Type...; force_gc::Bool = false) = _ZERO_TRIM_SUMMARY + +compact!( + ::DisabledPool; + factor::Real = 10, shrink_to::Real = 1.5, min_bytes::Int = 2^20, + active::Bool = true, force_gc::Bool = false +) = _ZERO_COMPACT_SUMMARY +compact!( + ::DisabledPool, ::Type{T}; + factor::Real = 10, shrink_to::Real = 1.5, min_bytes::Int = 2^20, + active::Bool = true, force_gc::Bool = false +) where {T} = _ZERO_COMPACT_SUMMARY +compact!( + ::DisabledPool, types::Type...; + factor::Real = 10, shrink_to::Real = 1.5, min_bytes::Int = 2^20, + active::Bool = true, force_gc::Bool = false +) = _ZERO_COMPACT_SUMMARY diff --git a/src/types.jl b/src/types.jl index 5f030c5c..d4016c39 100644 --- a/src/types.jl +++ b/src/types.jl @@ -113,6 +113,14 @@ mutable struct TypedPool{T} <: AbstractTypedPool{T, Vector{T}} # --- N-D Wrapper Cache (setfield!-based reuse) --- arr_wrappers::Vector{Union{Nothing, Vector{Any}}} # index=N (dimensionality), value=per-slot Array{T,N} + # --- Per-slot current logical extent (parallel to `vectors`) --- + # The size of each slot's most recent `_claim_slot!`, recording the live extent + # for BOTH the `acquire!` (Array) and `acquire_view!` (uncached SubArray/Reshaped) + # paths. `compact!` reads this (`_slot_used`) to know how much of an over-allocated + # backing buffer is actually in use — the backing `Vector`'s own length is a + # capacity high-water mark, not the current extent. + slot_extents::Vector{Int} + # --- State Management (1-based sentinel pattern) --- n_active::Int _checkpoint_n_active::Vector{Int} # Saved n_active at each checkpoint @@ -124,6 +132,8 @@ TypedPool{T}() where {T} = TypedPool{T}( Vector{T}[], # N-D Wrapper Cache Union{Nothing, Vector{Any}}[], + # Per-slot current logical extent (parallel to `vectors`) + Int[], # State Management (1-based sentinel pattern: guaranteed non-empty) 0, # n_active [0], # _checkpoint_n_active: sentinel (n_active=0 at depth=0) diff --git a/test/cuda/runtests.jl b/test/cuda/runtests.jl index df466279..85f94e40 100644 --- a/test/cuda/runtests.jl +++ b/test/cuda/runtests.jl @@ -50,5 +50,6 @@ else include("test_runtime_mutation.jl") include("test_scope_depth_validation.jl") include("test_trim.jl") + include("test_compact.jl") end end diff --git a/test/cuda/test_compact.jl b/test/cuda/test_compact.jl new file mode 100644 index 00000000..8c845c11 --- /dev/null +++ b/test/cuda/test_compact.jl @@ -0,0 +1,81 @@ +# CUDA compact! Tests +# Parity with CPU/Metal test_compact.jl: shrink an over-allocated device buffer in place. +# +# Like Metal, CUDA has NO uncached views — acquire_view! == acquire! (returns a cached +# CuArray wrapper) — so the CPU view-extent bug is structurally absent. We verify +# slot_extents-based compaction shrinks the device buffer and re-syncs the held wrapper +# (the DataRef swap + refcount path). Uses Float32/Int32 (aligned_sizeof == sizeof for +# these, so the capacity helper below is exact). + +@testset "CUDA compact!" begin + + # Device capacity (elements). For Float32/Int32, _aligned_sizeof == sizeof. + _ccap(v) = Int(getfield(v, :maxsize) ÷ sizeof(eltype(v))) + Summary = @NamedTuple{slots_compacted::Int, bytes_reclaimed::Int, gc_triggered::Bool} + ZERO = (; slots_compacted = 0, bytes_reclaimed = 0, gc_triggered = false) + + @testset "inactive: compact!(pool) shrinks the device buffer" begin + pool = CuAdaptiveArrayPool{0}() + checkpoint!(pool); a = acquire!(pool, Float32, 1_000_000); fill!(a, 0.0f0); rewind!(pool) + checkpoint!(pool); b = acquire!(pool, Float32, 100); fill!(b, 0.0f0); rewind!(pool) + tp = pool.float32 + @test tp.n_active == 0 + @test _ccap(tp.vectors[1]) >= 1_000_000 + + s = compact!(pool) + + @test s.slots_compacted == 1 + @test s.bytes_reclaimed > 0 + @test s.gc_triggered == false + @test _ccap(tp.vectors[1]) < 1000 # device buffer shrunk in place + end + + @testset "active=true shrinks a HELD slot in place; data intact + writable" begin + pool = CuAdaptiveArrayPool{0}() + checkpoint!(pool); a = acquire!(pool, Float32, 1_000_000); fill!(a, 0.0f0); rewind!(pool) + checkpoint!(pool) + b = acquire!(pool, Float32, 100) # slot 1 ACTIVE (b held) + copyto!(b, Float32.(1:100)) + tp = pool.float32 + @test tp.n_active == 1 + cap0 = _ccap(tp.vectors[1]) + @test cap0 >= 1_000_000 + + @test compact!(pool; active = false).slots_compacted == 0 # leaves held slot alone + @test _ccap(tp.vectors[1]) == cap0 + + s = compact!(pool; active = true) + @test s.slots_compacted == 1 + @test _ccap(tp.vectors[1]) < 1000 + @test Array(b) == Float32.(1:100) # held wrapper re-synced: data intact + copyto!(b, fill(9.0f0, 100)) # ...and still writable into the new buffer + @test Array(b)[1] == 9.0f0 + rewind!(pool) + end + + @testset "no-op + force_gc + type stability + DisabledPool" begin + pool = CuAdaptiveArrayPool{0}() + @test compact!(pool) == ZERO + @test compact!(pool; force_gc = true).gc_triggered == true + @test only(Base.return_types(compact!, (typeof(pool),))) == Summary + @test (@inferred compact!(pool)) isa Summary + @test compact!(ext.DISABLED_CUDA) == ZERO + @test compact!(ext.DISABLED_CUDA, Float32) == ZERO + end + + @testset "per-type + varargs (CUDA fixed types)" begin + pool = CuAdaptiveArrayPool{0}() + checkpoint!(pool); acquire!(pool, Float32, 1_000_000); rewind!(pool) + checkpoint!(pool); acquire!(pool, Float32, 100); rewind!(pool) + checkpoint!(pool); acquire!(pool, Int32, 1_000_000); rewind!(pool) + checkpoint!(pool); acquire!(pool, Int32, 100); rewind!(pool) + + s = compact!(pool, Float32, Int32) + @test s.slots_compacted == 2 + @test _ccap(pool.float32.vectors[1]) < 1000 + @test _ccap(pool.int32.vectors[1]) < 1000 + @test only(Base.return_types(compact!, (typeof(pool), Type{Float32}, Type{Int32}))) == Summary + @test compact!(ext.DISABLED_CUDA, Float32, Int32) == ZERO + end + +end diff --git a/test/metal/runtests.jl b/test/metal/runtests.jl index 70c8ba1a..63404b7d 100644 --- a/test/metal/runtests.jl +++ b/test/metal/runtests.jl @@ -52,5 +52,6 @@ else include("test_reshape.jl") include("test_task_local_pool.jl") include("test_trim.jl") + include("test_compact.jl") end end diff --git a/test/metal/test_compact.jl b/test/metal/test_compact.jl new file mode 100644 index 00000000..4a2fd693 --- /dev/null +++ b/test/metal/test_compact.jl @@ -0,0 +1,84 @@ +# Metal compact! Tests +# Parity with CPU test_compact.jl: shrink an over-allocated device buffer in place. +# +# Metal has NO uncached views — acquire_view! == acquire! (returns a cached MtlArray +# wrapper) — so the CPU view-extent bug is structurally absent here. We still verify +# slot_extents-based compaction shrinks the device buffer and re-syncs the held +# wrapper (the DataRef swap + refcount path) on real hardware. Uses Float32/Int32 +# (Metal's fixed types; no Float64 on Metal). + +@testset "Metal compact!" begin + + # Device capacity (elements) of a Metal backing buffer = maxsize bytes / sizeof(T). + _mcap(v) = Int(getfield(v, :maxsize) ÷ sizeof(eltype(v))) + Summary = @NamedTuple{slots_compacted::Int, bytes_reclaimed::Int, gc_triggered::Bool} + ZERO = (; slots_compacted = 0, bytes_reclaimed = 0, gc_triggered = false) + + @testset "inactive: compact!(pool) shrinks the device buffer" begin + pool = MetalAdaptiveArrayPool{0, METAL_STORAGE}() + # slot 1: small last-use (100) over a large high-water device buffer (1M) + checkpoint!(pool); a = acquire!(pool, Float32, 1_000_000); fill!(a, 0.0f0); rewind!(pool) + checkpoint!(pool); b = acquire!(pool, Float32, 100); fill!(b, 0.0f0); rewind!(pool) + tp = pool.float32 + @test tp.n_active == 0 + @test _mcap(tp.vectors[1]) >= 1_000_000 + + s = compact!(pool) + + @test s.slots_compacted == 1 + @test s.bytes_reclaimed > 0 + @test s.gc_triggered == false + @test _mcap(tp.vectors[1]) < 1000 # device buffer shrunk in place + end + + @testset "active=true shrinks a HELD slot in place; data intact + writable" begin + pool = MetalAdaptiveArrayPool{0, METAL_STORAGE}() + checkpoint!(pool); a = acquire!(pool, Float32, 1_000_000); fill!(a, 0.0f0); rewind!(pool) + checkpoint!(pool) + b = acquire!(pool, Float32, 100) # slot 1 ACTIVE (b held) + copyto!(b, Float32.(1:100)) + tp = pool.float32 + @test tp.n_active == 1 + cap0 = _mcap(tp.vectors[1]) + @test cap0 >= 1_000_000 + + # default (active=false) leaves the held slot untouched + @test compact!(pool; active = false).slots_compacted == 0 + @test _mcap(tp.vectors[1]) == cap0 + + # active=true shrinks the device buffer the user is still holding + s = compact!(pool; active = true) + @test s.slots_compacted == 1 + @test _mcap(tp.vectors[1]) < 1000 + @test Array(b) == Float32.(1:100) # held wrapper re-synced: data intact + copyto!(b, fill(9.0f0, 100)) # ...and still writable into the new buffer + @test Array(b)[1] == 9.0f0 + rewind!(pool) + end + + @testset "no-op + force_gc + type stability + DisabledPool" begin + pool = MetalAdaptiveArrayPool{0, METAL_STORAGE}() + @test compact!(pool) == ZERO + @test compact!(pool; force_gc = true).gc_triggered == true + @test only(Base.return_types(compact!, (typeof(pool),))) == Summary + @test (@inferred compact!(pool)) isa Summary + @test compact!(ext.DISABLED_METAL) == ZERO + @test compact!(ext.DISABLED_METAL, Float32) == ZERO + end + + @testset "per-type + varargs (Metal fixed types)" begin + pool = MetalAdaptiveArrayPool{0, METAL_STORAGE}() + checkpoint!(pool); acquire!(pool, Float32, 1_000_000); rewind!(pool) + checkpoint!(pool); acquire!(pool, Float32, 100); rewind!(pool) + checkpoint!(pool); acquire!(pool, Int32, 1_000_000); rewind!(pool) + checkpoint!(pool); acquire!(pool, Int32, 100); rewind!(pool) + + s = compact!(pool, Float32, Int32) + @test s.slots_compacted == 2 + @test _mcap(pool.float32.vectors[1]) < 1000 + @test _mcap(pool.int32.vectors[1]) < 1000 + @test only(Base.return_types(compact!, (typeof(pool), Type{Float32}, Type{Int32}))) == Summary + @test compact!(ext.DISABLED_METAL, Float32, Int32) == ZERO + end + +end diff --git a/test/runtests.jl b/test/runtests.jl index 0e3bb653..a638f433 100644 --- a/test/runtests.jl +++ b/test/runtests.jl @@ -23,6 +23,7 @@ else include("test_basic.jl") include("test_state.jl") include("test_trim.jl") + include("test_compact.jl") include("test_multidimensional.jl") include("test_macros.jl") include("test_task_local_pool.jl") @@ -54,6 +55,7 @@ else include("test_basic.jl") include("test_state.jl") include("test_trim_legacy.jl") + include("test_compact_legacy.jl") include("test_multidimensional.jl") include("test_macros.jl") include("test_task_local_pool.jl") diff --git a/test/test_compact.jl b/test/test_compact.jl new file mode 100644 index 00000000..3f6ec136 --- /dev/null +++ b/test/test_compact.jl @@ -0,0 +1,404 @@ +using Test +using AdaptiveArrayPools +using AdaptiveArrayPools: checkpoint!, rewind! + +# ============================================================================== +# compact! — capacity compaction (shrink over-allocated backing buffers in place) +# Design: docs/design/capacity_compaction.md | Plan: docs/plans/PLAN_compact.md +# +# Model: a slot's backing Vector length is the HIGH-WATER mark (largest ever +# acquired; _claim_slot! only grows). The CURRENT use is recorded per-slot in +# `tp.slot_extents` by _claim_slot! — covering BOTH acquire! (Array) and acquire_view! +# (uncached SubArray/ReshapedArray). So a slot is bloated when backing capacity ≫ +# slot_extents[slot]; compact! shrinks the backing to ~shrink_to×(that extent) in +# place and re-syncs wrappers (held views follow via preserved Vector identity). +# ============================================================================== + +# Allocated backing capacity (Memory length). +_cap(v) = length(getfield(v, :ref).mem) + +# Build slot 1 (Float64) bloated: acquire `big` (grows backing to high-water), rewind, +# re-acquire `small` on the same slot (backing stays high-water; wrapper size = small). +# Returns (pool, wrapper-of-size-small). +function _bloated_pool(big::Int, small::Int) + pool = AdaptiveArrayPool{0}() + checkpoint!(pool) + a = acquire!(pool, Float64, big) + a .= 0.0 + rewind!(pool) + checkpoint!(pool) + b = acquire!(pool, Float64, small) + return pool, b +end + +# Build slot 1 (Float64) bloated AND INACTIVE: grow backing to `big` (high-water), +# rewind, re-acquire `small` (this sets the cached wrapper's size to `small`), then +# rewind again so the slot is inactive (n_active == 0) while its backing capacity +# stays at the `big` high-water mark. An inactive slot is only "bloated" when its +# LAST logical size (the cached wrapper size) is far below its retained capacity — +# `acquire(big) → rewind` alone leaves wrapper size == capacity (not bloated). +function _inactive_bloated_pool(big::Int, small::Int) + pool = AdaptiveArrayPool{0}() + checkpoint!(pool) + a = acquire!(pool, Float64, big) + a .= 0.0 + rewind!(pool) + checkpoint!(pool) + b = acquire!(pool, Float64, small) + b .= 0.0 + rewind!(pool) + return pool +end + +# Concrete public summary shape (mirror of the `trim!` summary test). +const _COMPACT_SUMMARY = @NamedTuple{ + slots_compacted::Int, bytes_reclaimed::Int, gc_triggered::Bool, +} +const _ZERO_COMPACT = (; slots_compacted = 0, bytes_reclaimed = 0, gc_triggered = false) + +@testset "compact!" begin + + @testset "primitive: in-place shrink + wrapper re-sync + view follow + GC" begin + pool, b = _bloated_pool(1_000_000, 100) + b .= 1.0:100.0 + tp = pool.float64 + @test length(b) == 100 # wrapper = current use + @test length(tp.vectors[1]) == 1_000_000 # backing = high-water (the bloat) + @test _cap(tp.vectors[1]) >= 1_000_000 + + vw = view(tp.vectors[1], 1:100) # view into the slot's backing object + + reclaimed = AdaptiveArrayPools._maybe_compact_slot!(tp, 1, 10, 1.5, 2^20) + + @test reclaimed > 0 + @test length(tp.vectors[1]) == 150 # backing shrunk to ceil(1.5 * 100) + @test _cap(tp.vectors[1]) == 150 + @test b == collect(1.0:100.0) # wrapper re-synced: data intact + b[1] = 99.0 + @test tp.vectors[1][1] == 99.0 # wrapper writes into the new buffer + @test collect(vw)[2] == 2.0 # view follows parent to the new buffer + rewind!(pool) + end + + # GC reclaim in its own function scope so intermediate strong refs (the wrapper, + # the captured Memory, the view) leave scope before collection — otherwise + # conservative stack roots keep the old buffer alive and the test is flaky. + function _old_mem_weakref() + pool, b = _bloated_pool(1_000_000, 100) + b .= 1.0:100.0 + tp = pool.float64 + wr = WeakRef(getfield(tp.vectors[1], :ref).mem) + AdaptiveArrayPools._maybe_compact_slot!(tp, 1, 10, 1.5, 2^20) + return wr, pool # keep `pool` alive; only the OLD buffer should die + end + + @testset "primitive: old backing buffer is GC-reclaimed after compaction" begin + wr, pool = _old_mem_weakref() + GC.gc(true); GC.gc(true) + @test wr.value === nothing # old 1M Memory reclaimed + @test pool.float64.n_active == 1 # pool still alive (not a trivial pass) + end + + @testset "primitive gates: ratio and absolute-bytes" begin + # ratio gate: capacity only 5x the wrapper size (< factor 10) -> skip + # (min_bytes = 0 isolates the ratio gate). + pool, _ = _bloated_pool(500, 100) + tp = pool.float64 + cap0 = _cap(tp.vectors[1]) + @test AdaptiveArrayPools._maybe_compact_slot!(tp, 1, 10, 1.5, 0) == 0 + @test _cap(tp.vectors[1]) == cap0 # untouched + rewind!(pool) + + # bytes gate: 100x bloated but reclaim ~8 KB < min_bytes (1 MiB) -> skip + pool2, _ = _bloated_pool(1000, 10) + tp2 = pool2.float64 + cap2 = _cap(tp2.vectors[1]) + @test AdaptiveArrayPools._maybe_compact_slot!(tp2, 1, 10, 1.5, 2^20) == 0 + @test _cap(tp2.vectors[1]) == cap2 # untouched + rewind!(pool2) + end + + # ── Phase 2: public compact! — Tier 1 (inactive slots only) ──────────────── + + @testset "Tier 1: compact!(pool) reclaims inactive bloat" begin + pool = _inactive_bloated_pool(1_000_000, 100) + tp = pool.float64 + @test tp.n_active == 0 # slot 1 is inactive… + @test _cap(tp.vectors[1]) >= 1_000_000 # …but retains high-water capacity + + s = compact!(pool) + + @test s.slots_compacted == 1 + @test s.bytes_reclaimed >= (1_000_000 - 150) * sizeof(Float64) + @test s.gc_triggered == false + @test _cap(tp.vectors[1]) == 150 # shrunk to ceil(1.5 * 100) + end + + @testset "default (active=true) compacts active bloat; active=false opts out" begin + # compact!'s purpose is to reclaim from arrays still in use, so the default + # reaches active slots too. The factor gate (10×) keeps it from touching + # normally-sized arrays (covered by the next testset). + pool, b = _bloated_pool(1_000_000, 100) # slot 1 ACTIVE (b held), cap 1M, used 100 + b .= 1.0:100.0 + tp = pool.float64 + @test tp.n_active == 1 + cap0 = _cap(tp.vectors[1]) + + # active=false explicitly opts out → active slot untouched + s_off = compact!(pool; active = false) + @test s_off.slots_compacted == 0 + @test _cap(tp.vectors[1]) == cap0 + + # default compacts the held active slot in place; held wrapper follows + s = compact!(pool) + @test s.slots_compacted == 1 + @test _cap(tp.vectors[1]) == 150 + @test b == collect(1.0:100.0) # data intact + re-synced + b[1] = 99.0 + @test tp.vectors[1][1] == 99.0 # writes into the new buffer + rewind!(pool) + end + + @testset "default leaves a normally-sized active array alone (factor gate)" begin + # Safety net for the active=true default: a slot whose capacity ≈ its use + # (not 10× bloated) is never compacted, so live arrays are not churned. + pool = AdaptiveArrayPool{0}() + checkpoint!(pool) + a = acquire!(pool, Float64, 100) # capacity ≈ use, NOT bloated + a .= 5.0 + cap0 = _cap(pool.float64.vectors[1]) + + s = compact!(pool) # default active=true, but factor gate skips + + @test s.slots_compacted == 0 + @test _cap(pool.float64.vectors[1]) == cap0 + @test a == fill(5.0, 100) # untouched, still valid + rewind!(pool) + end + + @testset "per-type compact!(pool, T) targets only T; guard never creates a pool" begin + pool = _inactive_bloated_pool(1_000_000, 100) + # also build an inactive-bloated Int64 slot (Int64 is a fixed slot) + checkpoint!(pool); ai = acquire!(pool, Int64, 1_000_000); ai .= 0; rewind!(pool) + checkpoint!(pool); bi = acquire!(pool, Int64, 100); bi .= 0; rewind!(pool) + + s = compact!(pool, Float64) # compact only Float64 + @test s.slots_compacted == 1 + @test _cap(pool.float64.vectors[1]) == 150 + @test _cap(pool.int64.vectors[1]) >= 1_000_000 # Int64 untouched + + # Guard: a never-used fallback type must NOT be registered in `others`. + @test !haskey(pool.others, UInt8) + n_before = length(pool.others) + s2 = compact!(pool, UInt8) + @test s2 == _ZERO_COMPACT + @test length(pool.others) == n_before + @test !haskey(pool.others, UInt8) # still not created + end + + @testset "compact! return value is type-stable (no Any-typed fields)" begin + # Mirrors the trim! stability test: the summary is a public return value, so + # its fields must infer concrete even though the full-pool form folds over + # fixed slots (@generated) AND the `others` Vector{Any} (dynamic dispatch). + pool = AdaptiveArrayPool{0}() + checkpoint!(pool) + acquire!(pool, Float64, 100) + acquire!(pool, Int64, 50) + acquire!(pool, UInt8, 10) # exercises the `others` path + rewind!(pool) + + rt = only(Base.return_types(compact!, (typeof(pool),))) + @test isconcretetype(rt) + @test rt == _COMPACT_SUMMARY + @test only(Base.return_types(compact!, (typeof(pool), Type{Float64}))) == _COMPACT_SUMMARY + + @test (@inferred compact!(pool)) isa _COMPACT_SUMMARY + @test (@inferred compact!(pool, Float64)) isa _COMPACT_SUMMARY + end + + @testset "no-op when nothing is bloated; force_gc sets the flag" begin + pool = AdaptiveArrayPool{0}() + s = compact!(pool) + @test s == _ZERO_COMPACT + @test compact!(pool; force_gc = true).gc_triggered == true + @test compact!(pool; force_gc = false).gc_triggered == false + end + + @testset "DisabledPool zero summary; compact!() uses the task-local pool" begin + s = compact!(DISABLED_CPU) + @test s == _ZERO_COMPACT + @test compact!(DISABLED_CPU, Float64) == _ZERO_COMPACT + + # No-arg form routes to the task-local pool: on an empty one it returns the + # zero summary, and force_gc threads through the no-arg overload. + pool = get_task_local_pool() + empty!(pool) + @test compact!() == _ZERO_COMPACT + @test compact!(; force_gc = true).gc_triggered == true + empty!(pool) # cleanup (task-local is shared) + end + + @testset "self-heal: re-acquire after compacting an inactive slot works" begin + pool = _inactive_bloated_pool(1_000_000, 100) + s = compact!(pool) + @test s.slots_compacted == 1 + @test _cap(pool.float64.vectors[1]) == 150 + + # Re-acquire the compacted slot within the shrunk capacity (reuse), then at a + # size that forces a regrow — both must yield correct, writable arrays whose + # writes land in the backing (wrapper :ref re-synced through the grow). + checkpoint!(pool) + a = acquire!(pool, Float64, 120) # 120 <= 150: reuse shrunk buffer + a .= 3.0 + @test length(a) == 120 + @test all(a .== 3.0) + rewind!(pool) + + checkpoint!(pool) + b = acquire!(pool, Float64, 5_000) # 5000 > 150: backing must regrow + b .= 7.0 + @test length(b) == 5_000 + @test all(b .== 7.0) + b[1] = 9.0 + @test pool.float64.vectors[1][1] == 9.0 # write lands in the regrown backing + rewind!(pool) + end + + # ── Phase 3: Tier 2 (active=true opt-in) + Type... varargs ───────────────── + + @testset "Tier 2: active=true compacts a slot the user is still holding" begin + pool, b = _bloated_pool(1_000_000, 100) # slot 1 is ACTIVE (b held), cap 1M + b .= 1.0:100.0 + tp = pool.float64 + @test tp.n_active == 1 + @test _cap(tp.vectors[1]) >= 1_000_000 + + vw = view(b, 1:50) # view INTO the held active wrapper + + # active=false (default) must leave the active slot alone. + s0 = compact!(pool; active = false) + @test s0.slots_compacted == 0 + @test _cap(tp.vectors[1]) >= 1_000_000 + + # active=true shrinks the held slot in place. + s = compact!(pool; active = true) + @test s.slots_compacted == 1 + @test s.bytes_reclaimed >= (1_000_000 - 150) * sizeof(Float64) + @test _cap(tp.vectors[1]) == 150 # ceil(1.5 * 100) + + # The user's held wrapper survives: data intact, re-synced, still writable. + @test b == collect(1.0:100.0) + b[1] = 99.0 + @test tp.vectors[1][1] == 99.0 # write lands in the new buffer + @test collect(vw)[2] == 2.0 # view of the held wrapper follows + vw[1] = -1.0 + @test b[1] == -1.0 # view write reaches the wrapper + rewind!(pool) + end + + @testset "active=true also still compacts inactive slots (scan = 1:end)" begin + # active=true is "compact everything"; inactive bloat is included too. + pool = _inactive_bloated_pool(1_000_000, 100) + s = compact!(pool; active = true) + @test s.slots_compacted == 1 + @test _cap(pool.float64.vectors[1]) == 150 + end + + @testset "varargs compact!(pool, T...) compacts each listed type" begin + pool = _inactive_bloated_pool(1_000_000, 100) # Float64 inactive bloat + checkpoint!(pool); ai = acquire!(pool, Int64, 1_000_000); ai .= 0; rewind!(pool) + checkpoint!(pool); bi = acquire!(pool, Int64, 100); bi .= 0; rewind!(pool) + + s = compact!(pool, Float64, Int64) + @test s.slots_compacted == 2 + @test _cap(pool.float64.vectors[1]) == 150 + @test _cap(pool.int64.vectors[1]) == 150 + + # Type-stable: the varargs form returns the same concrete summary. + @test only(Base.return_types(compact!, (typeof(pool), Type{Float64}, Type{Int64}))) == + _COMPACT_SUMMARY + @test (@inferred compact!(pool, Float64, Int64)) isa _COMPACT_SUMMARY # 2nd call: no-op + + # A never-used fallback type in the list is skipped, never created. + pool2 = AdaptiveArrayPool{0}() + @test !haskey(pool2.others, UInt16) + s2 = compact!(pool2, Float64, UInt16) + @test s2.slots_compacted == 0 + @test !haskey(pool2.others, UInt16) + end + + @testset "varargs honors active=true and force_gc" begin + pool, b = _bloated_pool(1_000_000, 100) # active Float64 bloat + b .= 0.0 + s = compact!(pool, Float64; active = true, force_gc = true) + @test s.slots_compacted == 1 + @test s.gc_triggered == true + @test _cap(pool.float64.vectors[1]) == 150 + rewind!(pool) + end + + @testset "DisabledPool varargs is a zero-summary no-op" begin + @test compact!(DISABLED_CPU, Float64, Int64) == _ZERO_COMPACT + @test compact!(DISABLED_CPU, Float64, Int64; active = true) == _ZERO_COMPACT + end + + # ── Regression: active=true must not shrink below a live acquire_view! extent ── + # A slot can carry a SMALL cached Array wrapper (from a prior acquire!) AND a + # LARGER live view (from acquire_view!, which returns an uncached SubArray / + # ReshapedArray). `_slot_used` only scans the cached Array wrappers, so it + # under-reports the live extent and `compact!(active=true)` could shrink the + # backing below the view → out-of-bounds. Assertions are length-only (never index + # the view) so a buggy run reports a failure instead of segfaulting the runner. + @testset "active=true respects a live 1-D acquire_view! extent" begin + pool = AdaptiveArrayPool{0}() + # slot 1: small cached Array wrapper (100) over a large high-water backing (1M) + checkpoint!(pool); a = acquire!(pool, Float64, 1_000_000); a .= 0.0; rewind!(pool) + checkpoint!(pool); a2 = acquire!(pool, Float64, 100); a2 .= 0.0; rewind!(pool) + tp = pool.float64 + @test length(tp.arr_wrappers[1][1]) == 100 # small cached wrapper + @test _cap(tp.vectors[1]) >= 1_000_000 + + checkpoint!(pool) + v = acquire_view!(pool, Float64, 1000) # large uncached view on the slot + @test length(v) == 1000 + @test tp.n_active == 1 # slot 1 is ACTIVE + + compact!(pool; active = true) + + # SAFE INVARIANT: the backing must still cover the live view (no OOB). + @test _cap(tp.vectors[1]) >= length(v) + @test length(tp.vectors[1]) >= length(v) + rewind!(pool) + end + + @testset "active=true respects a live N-D acquire_view! (ReshapedArray) extent" begin + pool = AdaptiveArrayPool{0}() + checkpoint!(pool); a = acquire!(pool, Float64, 1_000_000); a .= 0.0; rewind!(pool) + checkpoint!(pool); a2 = acquire!(pool, Float64, 100); a2 .= 0.0; rewind!(pool) + tp = pool.float64 + + checkpoint!(pool) + vw = acquire_view!(pool, Float64, 40, 25) # ReshapedArray, extent 1000 + @test length(vw) == 1000 + @test tp.n_active == 1 + + compact!(pool; active = true) + + @test _cap(tp.vectors[1]) >= length(vw) + @test length(tp.vectors[1]) >= length(vw) + rewind!(pool) + end + + @testset "extreme kwargs never throw (shrink target clamped to capacity)" begin + pool = _inactive_bloated_pool(1_000_000, 100) + # shrink_to so large that ceil(Int, shrink_to*used) would overflow Int — + # must be a safe no-op (target clamped to capacity → reclaim 0 → skipped), + # not an InexactError. + s = compact!(pool; shrink_to = 1.0e300) + @test s.slots_compacted == 0 + @test _cap(pool.float64.vectors[1]) >= 1_000_000 # untouched + # shrink_to larger than capacity/used ratio (but not overflowing) also no-ops. + @test compact!(pool; shrink_to = 1.0e6).slots_compacted == 0 + end + +end diff --git a/test/test_compact_legacy.jl b/test/test_compact_legacy.jl new file mode 100644 index 00000000..b4e7e171 --- /dev/null +++ b/test/test_compact_legacy.jl @@ -0,0 +1,36 @@ +using Test +using AdaptiveArrayPools +using AdaptiveArrayPools: compact!, get_task_local_pool + +# Legacy path (Julia < 1.12): compact! is a defined, exported NO-OP that returns a +# zero summary and warns once. This keeps the public API portable across the full +# supported Julia range (dependents can import/call compact! on any version). +# Mirrors test_trim_legacy.jl. + +@testset "compact! (legacy no-op, Julia < 1.12)" begin + pool = get_task_local_pool() + + # First call warns once and returns a zero summary. + s = @test_logs (:warn,) match_mode = :any compact!(pool) + @test s.slots_compacted == 0 + @test s.bytes_reclaimed == 0 + @test s.gc_triggered == false + + # Same zero-summary shape for the other entry points (single-type, varargs, no-arg). + @test compact!(pool, Float64).slots_compacted == 0 + @test compact!(pool, Float64, Int64).slots_compacted == 0 + @test compact!().slots_compacted == 0 + + # Legacy no-op ignores the active / force_gc kwargs (no reclamation to do). + @test compact!(pool; active = true).slots_compacted == 0 + @test compact!(pool; active = false).slots_compacted == 0 + @test compact!(pool; force_gc = true).gc_triggered == false + + # DisabledPool: zero-summary no-op on the legacy path too. + zd = compact!(DISABLED_CPU) + @test zd.slots_compacted == 0 + @test zd.bytes_reclaimed == 0 + @test zd.gc_triggered == false + @test compact!(DISABLED_CPU, Float64).slots_compacted == 0 + @test compact!(DISABLED_CPU, Float64, Int64).slots_compacted == 0 +end diff --git a/test/test_trim_legacy.jl b/test/test_trim_legacy.jl index 0fd9de42..dc31c5ff 100644 --- a/test/test_trim_legacy.jl +++ b/test/test_trim_legacy.jl @@ -16,8 +16,9 @@ using AdaptiveArrayPools: trim!, get_task_local_pool @test s.estimated_bytes_released == 0 @test s.gc_triggered == false - # Same zero-summary shape for the other entry points. + # Same zero-summary shape for the other entry points (single-type, varargs, no-arg). @test trim!(pool, Float64).slots_released == 0 + @test trim!(pool, Float64, Int64).slots_released == 0 @test trim!().slots_released == 0 # Legacy no-op ignores force_gc (no reclamation to do). @@ -30,4 +31,5 @@ using AdaptiveArrayPools: trim!, get_task_local_pool @test zd.estimated_bytes_released == 0 @test zd.gc_triggered == false @test trim!(DISABLED_CPU, Float64).slots_released == 0 + @test trim!(DISABLED_CPU, Float64, Int64).slots_released == 0 end