From acf6ce4054ac257bde7d586d02183efe8ce74a06 Mon Sep 17 00:00:00 2001 From: lkdvos Date: Fri, 2 Oct 2026 15:29:54 -0400 Subject: [PATCH 1/6] Make ClockCache hits lock-free ClockCache is now an open-addressing table of immutable entries, published atomically and read with acquire loads, so hits take no lock and write no shared cache line (hits and misses are counted per thread). Misses and all other mutations still take the lock. Shared hits stay at ~15-25 ns at 8 threads (was ~330-620 ns) and are 2x faster single-threaded. Entries keep their key type, so concretely typed caches store keys unboxed. Task-local caches keep the previous locked implementation, now the internal LocalClockCache (a SlotCache), to avoid the per-thread counters in every task's cache. Adds concurrency stress tests and benchmark/contention.jl. Co-Authored-By: Claude Opus 5.5 --- CLAUDE.md | 1 + benchmark/Project.toml | 1 + benchmark/contention.jl | 134 ++++++++++++++++ docs/src/eviction.md | 5 +- docs/src/index.md | 3 +- src/MemoizationKit.jl | 1 + src/cachestyle.jl | 4 +- src/call.jl | 5 +- src/containers/clock.jl | 301 ++++++++++++++++++++++++++++++++--- src/containers/localclock.jl | 36 +++++ src/containers/slots.jl | 16 +- test/Project.toml | 2 + test/test_concurrency.jl | 117 ++++++++++++++ test/test_containers.jl | 8 +- 14 files changed, 597 insertions(+), 37 deletions(-) create mode 100644 benchmark/contention.jl create mode 100644 src/containers/localclock.jl create mode 100644 test/test_concurrency.jl diff --git a/CLAUDE.md b/CLAUDE.md index 6496343..cdd9dca 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -40,6 +40,7 @@ Keep Markdown prose to one sentence per line. - `@cached` preserves the method signature and moves the body to `MemoizationKit.implementation`. - The default global cache shares one size limit across a function's cached methods, with a separate cache per container type. - Preserve inferred return types and allocation-free RAM hits for concrete keys with the built-in containers. +- `ClockCache` hits are lock-free; keep the rules listed in `src/containers/clock.jl`. Task-local caches use the locked `LocalClockCache`. - Built-in RAM containers distinguish key types and use `hash` and `isequal`. - `MemoizationKit.cachekey` may merge equivalent inputs; merged calls must accept the same result and return type. - Compute misses outside cache locks so recursion works; concurrent misses may compute twice. diff --git a/benchmark/Project.toml b/benchmark/Project.toml index 66ef505..f5ac05c 100644 --- a/benchmark/Project.toml +++ b/benchmark/Project.toml @@ -4,6 +4,7 @@ MemoizationKit = "1b238080-9255-4fe9-b224-89eb24efe93b" LRUCache = "8ac3fa9e-de4c-5943-b1dc-09c6b5f20637" Printf = "de0858da-6303-5e67-8744-51eddeeeb8d7" Random = "9a3f8284-a2c9-5f02-9a11-845980a1fd5c" +Statistics = "10745b16-79ce-11e8-11f9-7d13ad32a3b2" [sources] MemoizationKit = {path = ".."} diff --git a/benchmark/contention.jl b/benchmark/contention.jl new file mode 100644 index 0000000..cf13700 --- /dev/null +++ b/benchmark/contention.jl @@ -0,0 +1,134 @@ +# Cost of a lookup when all threads share one cache, against task-local caches. +# "locked Clock" is the `ClockCache` of task-local caches (`MemoizationKit.LocalClockCache`), whose hits +# take the lock, shared here for comparison. +# +# for t in 1 2 4 8; do julia --heap-size-hint=4G --project=benchmark -t $t benchmark/contention.jl; done +# +# One task per thread, each doing `M` lookups from its own key sequence. Reports the median over +# `REPS` runs of the time per lookup within a task (ns/lookup, lower is better; perfect scaling +# keeps it constant as threads are added), and the bytes allocated per lookup. + +using Random, Printf, Statistics +using MemoizationKit: MemoizationKit, ClockCache, LocalClockCache, LRU, @cached, GlobalCache, TaskLocalCache + +const M = 200_000 +const REPS = 7 + +const CONTAINERS = [ + "ClockCache" => n -> ClockCache{Any, Any}(; maxsize = n), + "locked Clock" => n -> LocalClockCache{Any, Any}(; maxsize = n), + "LRU" => n -> LRU{Any, Any}(; maxsize = n), + "task-local Clock" => nothing, # one locked Clock per task, uncontended +] + +function zipf(rng, n, s, len) + cdf = cumsum(1 ./ (1:n) .^ s) + cdf ./= cdf[end] + return [searchsortedfirst(cdf, rand(rng)) for _ in 1:len] +end + +tuplekey(i) = (i, i % 7, :leg, Float64(i)) +compute(k::Int) = 3k +compute(k::Tuple) = k[1] + +# name => (key sequence per task, cache capacity, prefill keys) +function workloads(ntasks) + rngs = [Xoshiro(t) for t in 1:ntasks] + return [ + "hits, Int keys (1k)" => ([rand(r, 1:1_000, M) for r in rngs], 10_000, 1:1_000), + "hits, 4-tuple keys (1k)" => ([tuplekey.(rand(r, 1:1_000, M)) for r in rngs], 10_000, tuplekey.(1:1_000)), + "hits, one hot Int key" => ([fill(7, M) for r in rngs], 10_000, 7:7), + "zipf 1.0 (100k keys, cap 10k)" => ([zipf(r, 100_000, 1.0, M) for r in rngs], 10_000, 1:0), + "uniform (20k keys, cap 10k)" => ([rand(r, 1:20_000, M) for r in rngs], 10_000, 1:0), + ] +end + +function run!(c, seq) + acc = 0 + for k in seq + acc += get!(() -> compute(k), c, k)::Int + end + return acc +end + +# ns per lookup in each task, and total bytes allocated +function threaded(make, seqs, cap, prefill) + shared = make === nothing ? nothing : make(cap) + caches = [shared === nothing ? LocalClockCache{Any, Any}(; maxsize = cap) : shared for _ in seqs] + for c in unique(caches), k in prefill + get!(() -> compute(k), c, k) + end + times = zeros(length(seqs)) + start = Threads.Atomic{Int}(0) + bytes = @allocated @sync for (t, seq) in enumerate(seqs) + Threads.@spawn begin + Threads.atomic_add!(start, 1) + while start[] < length(seqs) # start together + ccall(:jl_cpu_pause, Cvoid, ()) + end + t0 = time_ns() + run!(caches[t], seq) + times[t] = (time_ns() - t0) / length(seq) + end + end + return mean(times), bytes / sum(length, seqs) +end + +# ns per hit through `@cached` functions using each cache style, Int and 4-tuple keys +@cached f_clock(x) = compute(x) +@cached f_locked(x) = compute(x) +@cached f_lru(x) = compute(x) +@cached f_local(x) = compute(x) +MemoizationKit.CacheStyle(::typeof(f_clock), x) = GlobalCache{ClockCache}() +MemoizationKit.CacheStyle(::typeof(f_locked), x) = GlobalCache{LocalClockCache}() +MemoizationKit.CacheStyle(::typeof(f_lru), x) = GlobalCache{LRU}() +MemoizationKit.CacheStyle(::typeof(f_local), x) = TaskLocalCache{ClockCache}() +const CACHED = ["ClockCache" => f_clock, "locked Clock" => f_locked, "LRU" => f_lru, "task-local Clock" => f_local] + +function callall(f::F, seq) where {F} + acc = 0 + for k in seq + acc += f(k)::Int + end + return acc +end + +function threaded_cached(f, seqs, keys) + times = zeros(length(seqs)) + bytes = @allocated @sync for (t, seq) in enumerate(seqs) + Threads.@spawn begin + callall(f, keys) # fill this task's cache (task-local) + t0 = time_ns() + callall(f, seq) + times[t] = (time_ns() - t0) / length(seq) + end + end + return mean(times), bytes / sum(length, seqs) +end + +nt = Threads.nthreads() +println("Julia ", VERSION, ", ", nt, " threads (", nt, " tasks), ", M, " lookups per task, median of ", REPS, " runs") +println("load average: ", read(`cat /proc/loadavg`, String)) +@printf("%-30s %-17s %10s %10s %8s\n", "workload", "container", "ns/lookup", "(min)", "B/lookup") +for (wname, (seqs, cap, prefill)) in workloads(nt) + for (cname, make) in CONTAINERS + threaded(make, seqs, cap, prefill) # compile + rs = [threaded(make, seqs, cap, prefill) for _ in 1:REPS] + @printf( + "%-30s %-17s %10.1f %10.1f %8.2f\n", wname, cname, + median(first.(rs)), minimum(first.(rs)), median(last.(rs)) + ) + end +end +for (kname, keys) in ("@cached hits, Int" => 1:1_000, "@cached hits, 4-tuple" => tuplekey.(1:1_000)) + seqs = [rand(Xoshiro(t), keys, M) for t in 1:nt] + for (cname, f) in CACHED + foreach(f, keys) + threaded_cached(f, seqs, keys) + rs = [threaded_cached(f, seqs, keys) for _ in 1:REPS] + @printf( + "%-30s %-17s %10.1f %10.1f %8.2f\n", kname, cname, + median(first.(rs)), minimum(first.(rs)), median(last.(rs)) + ) + end +end diff --git a/docs/src/eviction.md b/docs/src/eviction.md index c548435..21ab6f7 100644 --- a/docs/src/eviction.md +++ b/docs/src/eviction.md @@ -34,6 +34,7 @@ Unlike LRU, Clock does not track the exact order of hits. A recently inserted entry that has never been hit again can be evicted before an older entry that has been reused. Choose Clock for approximate recency with little bookkeeping on hits: a hit updates a bit rather than a recency list. +A shared `ClockCache` is read without a lock, so hits from many threads do not wait on each other. Eviction may scan several slots to find a victim. ## LRU: exact recency @@ -59,7 +60,7 @@ sort!(collect(keys(cache))) # B was evicted Unlike Clock, LRU always retains the exact recency order, even after several hits between evictions. Choose LRU when that ordering suits the workload. -Maintaining the order requires more bookkeeping on hits, but choosing the next victim does not require a scan. +Maintaining the order requires more bookkeeping on hits, under a lock, but choosing the next victim does not require a scan. Neither policy guarantees a better hit rate for every workload; use the [dashboard](dashboard.md) or [timing](timing.md) to evaluate yours. ## Selecting a policy @@ -83,5 +84,5 @@ Custom eviction policies can be supplied through the [cache interface](interface Both policies evict as many entries as needed to fit a new value or a reduced limit. A value larger than the entire limit is returned without being stored. Byte limits measure cached values, not keys or container overhead. -Shared caches use locks under either policy; computations run outside those locks. +Shared caches take a lock to insert and evict entries, and `LRU` also on hits; computations run outside those locks. Disk caches have no eviction policy; see [Disk caching](disk.md) for cleanup. diff --git a/docs/src/index.md b/docs/src/index.md index 65d76bc..22f662f 100644 --- a/docs/src/index.md +++ b/docs/src/index.md @@ -167,6 +167,7 @@ See [Configuration](configuration.md) for persistent defaults and custom size me Changes to external state or method definitions (including through Revise) do not automatically invalidate results. Clear the global cache with `empty_caches!`; for disk results, update [`MemoizationKit.diskversion`](@ref). - Built-in shared caches are thread-safe. + [`ClockCache`](@ref) hits take no lock, so they scale with the number of threads. Computations run outside the cache lock, allowing recursion; simultaneous misses may compute the same key more than once. Exceptions are propagated without storing a result. @@ -183,7 +184,7 @@ Strategies, settings, persistence, and monitoring work together under one functi | [Memoize.jl](https://github.com/JuliaCollections/Memoize.jl#usage) | You want a small `@memoize` API with a dictionary of your choice. | Limits and statistics depend on the container; persistence, a dashboard, and package settings require additional integration. | | [Memoization.jl](https://github.com/marius311/Memoization.jl#readme) | You need to memoize individual calls or closures, as well as method definitions. | Its default cache is unbounded and not thread-safe; custom containers can address those needs, but management and persistence tools require additional integration. | | [LRUCache.jl](https://github.com/JuliaCollections/LRUCache.jl#readme) | You want a thread-safe cache dictionary with limits, statistics, resizing, and eviction callbacks. | You write the function wrapper and management code, or combine it with a memoization macro. | -| MemoizationKit.jl | You want bounded caches, per-type strategies, disk persistence, and live monitoring under one API. | The macro requires a method definition; shared caches use locks, and disk results need explicit versioning and cleanup. | +| MemoizationKit.jl | You want bounded caches, per-type strategies, disk persistence, and live monitoring under one API. | The macro requires a method definition; shared caches lock to insert entries (and `LRU` also on hits), and disk results need explicit versioning and cleanup. | Memoize.jl and Memoization.jl both accept LRUCache.jl containers, so bounded memory and statistics are available without MemoizationKit. Memoization.jl also preserves return-type inference. diff --git a/src/MemoizationKit.jl b/src/MemoizationKit.jl index b353060..29005ee 100644 --- a/src/MemoizationKit.jl +++ b/src/MemoizationKit.jl @@ -26,6 +26,7 @@ include("containers/interface.jl") include("containers/slots.jl") include("containers/lru.jl") include("containers/clock.jl") +include("containers/localclock.jl") include("registry.jl") include("preferences.jl") diff --git a/src/cachestyle.jl b/src/cachestyle.jl index 50df162..b12e6f4 100644 --- a/src/cachestyle.jl +++ b/src/cachestyle.jl @@ -54,7 +54,9 @@ Strategy that stores results in caches local to the current task, of container t `C` is either an `AbstractCache` or an `AbstractDict` type such as `Dict`; if it is not already concrete, it is completed to `C{K, V}`. Without `C`, the container is the one set by the `container` preference. Task-local caches need no locking but are not shared, not bounded -unless `C` is, and are not visible to [`cache_info`](@ref). +unless `C` is, and are not visible to [`cache_info`](@ref). `TaskLocalCache{ClockCache}` uses +an internal variant of [`ClockCache`](@ref) whose hits take the lock, which costs nothing +uncontended and saves the memory of the per-thread counters. """ struct TaskLocalCache{C} <: CacheStyle end diff --git a/src/call.jl b/src/call.jl index 7d994a0..d07349f 100644 --- a/src/call.jl +++ b/src/call.jl @@ -80,7 +80,10 @@ end # One untyped cache per function for the containers of MemoizationKit; other dictionaries (e.g. `Dict`) # would box untyped keys, so they get one typed dictionary per key and value type. -_localtype(C, K, V) = C isa UnionAll ? (C <: AbstractCache ? C{Any, Any} : C{K, V}) : C +_localtype(C, K, V) = C isa UnionAll ? (C <: AbstractCache ? _localcache(C){Any, Any} : C{K, V}) : C +# A task-local cache is used by one task only, so lock-free hits gain nothing there, while the +# per-thread counters of `ClockCache` would cost memory in every task. +_localcache(C) = C === ClockCache ? LocalClockCache : C # Task-local caches follow the size settings of `f`, but are not registered. function _newlocal(f, ::Type{T}) where {T <: AbstractCache} fc = @lock REGISTRY.lock _functioncaches!(f) diff --git a/src/containers/clock.jl b/src/containers/clock.jl index ac257f7..a939591 100644 --- a/src/containers/clock.jl +++ b/src/containers/clock.jl @@ -1,3 +1,16 @@ +# An entry of a `ClockCache`: immutable apart from its reference bit, see the rules below. +mutable struct Entry{K, V} + const key::K + const hash::UInt + const val::V + const size::Int + const pos::Int # slot in `ring` + @atomic ref::Bool +end + +struct Tombstone end +const TOMBSTONE = Tombstone() + """ ClockCache{K, V}(; maxsize::Integer = 10_000, by = nothing) @@ -8,37 +21,281 @@ never hit again are evicted first. `maxsize` limits the number of entries, or, when `by` is given, the sum of `by(value)` over all entries. + +Lookups take no lock and write no shared memory, so hits scale with the number of threads that +share the cache. Insertions, evictions and the other changes take a lock. A lookup that races +with a removal may still return the removed value, and one that races with an insertion may +miss it; a returned value always belongs to its key. """ -mutable struct ClockCache{K, V} <: SlotCache{K, V} - const slots::Slots{V} - const ref::Vector{Bool} - const live::Vector{Bool} - hand::Int # last inspected slot +mutable struct ClockCache{K, V} <: AbstractCache{K, V} + @atomic table::Vector{Any} # open-addressing hash table of `Entry{K, V}`s, see `_find` + used::Int # entries and tombstones in `table` + const ring::Vector{Union{Nothing, Entry{K, V}}} # the clock, with `nothing` for a free slot + const free::Vector{Int} # free slots of `ring` + hand::Int # last inspected slot of `ring` + count::Int + currentsize::Int + maxsize::Int + const by::Any + const counts::Vector{Int} # hits and misses per thread, see `_count!` + const lock::ReentrantLock end -ClockCache{K, V}(; maxsize::Integer = 10_000, by = nothing) where {K, V} = - ClockCache{K, V}(Slots{V}(maxsize, by), Bool[], Bool[], 0) +# Lock-free lookups rest on three rules, which every method below keeps: +# +# 1. An `Entry` is immutable apart from its reference bit, so a reader that loads one sees a key +# with its own value. +# 2. A published table never changes length: it is replaced as a whole, by a release store to +# `c.table`. Readers load the table and then each slot with acquire ordering, so they see +# fully constructed entries, and their indices are always in bounds. +# 3. In a published table, a slot only changes from `nothing` to an entry, from an entry to +# `TOMBSTONE`, or from `TOMBSTONE` to an entry, all under the lock; or to `nothing` when the +# next slot is `nothing`, so that no probe passes it anyway. Entries never move, so a probe +# for a key that is present and not concurrently removed always finds it. +# +# Writers store slots with `setindex!`, which is a release store with a GC write barrier on every +# supported Julia version. (Storing through a pointer would skip the barrier, and the GC could +# free the entry.) Removed entries are dropped from the table rather than unset, so readers +# never see an undefined slot, and the GC collects them once no reader holds them. -function admit!(c::ClockCache, i::Int) - i > length(c.ref) && (push!(c.ref, false); push!(c.live, false)) - c.ref[i] = false - c.live[i] = true - return c +const MINTABLE = 16 +_emptytable(n::Int = MINTABLE) = fill!(Vector{Any}(undef, n), nothing) + +const COUNTSTRIDE = 8 # `Int`s per 64-byte cache line +const HIT, MISS = 1, 2 + +function ClockCache{K, V}(; maxsize::Integer = 10_000, by = nothing) where {K, V} + maxsize >= 0 || throw(ArgumentError("maxsize must be non-negative")) + return ClockCache{K, V}( + _emptytable(), 0, Union{Nothing, Entry{K, V}}[], Int[], 0, 0, 0, maxsize, by, + zeros(Int, COUNTSTRIDE * Threads.maxthreadid()), ReentrantLock() + ) +end + +_entrysize(c::ClockCache, v) = c.by === nothing ? 1 : Int(c.by(v))::Int + +@inline _loadslot(t::Vector{Any}, i::Int) = GC.@preserve t unsafe_load(pointer(t, i), :acquire) + +# The entry for the probe `p` in table `t`, or `nothing`. Linear probing from the hash; a table +# is at most half full, so the probe ends at an empty slot. +@inline function _find(t::Vector{Any}, p::Key{P}, ::Type{E}) where {P, E <: Entry} + mask = length(t) - 1 + i = p.hash % Int + for _ in 0:mask # bounds the loop also if slots change while probing + e = _loadslot(t, (i & mask) + 1) + e === nothing && return nothing + if e isa E && e.hash == p.hash + k = e.key + k isa P && isequal(p.key, k::P) && return e + end + i += 1 + end + return nothing +end +_find(c::ClockCache{K, V}, p::Key) where {K, V} = _find((@atomic :acquire c.table), p, Entry{K, V}) + +_lookup(c::ClockCache, key) = (p = _probe(c, key); p === nothing ? nothing : _find(c, p)) + +# Hits and misses are counted per thread, in separate cache lines, with relaxed loads and stores: +# a shared counter would bounce between cores on every hit. Two tasks on one thread cannot +# interleave between the load and the store, since neither yields; threads adopted after the +# cache was created share counters, which may then lose a few counts. +@inline function _count!(c::ClockCache, which::Int) + v = c.counts + i = mod(Threads.threadid() - 1, length(v) ÷ COUNTSTRIDE) * COUNTSTRIDE + which + GC.@preserve v begin + p = pointer(v, i) + unsafe_store!(p, unsafe_load(p, :monotonic) + 1, :monotonic) + end + return nothing end +function _counted(c::ClockCache, which::Int) + v = c.counts + return GC.@preserve v sum(i -> unsafe_load(pointer(v, i), :monotonic), which:COUNTSTRIDE:length(v)) +end + +# Hits only write the reference bit if it is clear, so hot entries are not written at all. +@inline _touch!(e::Entry) = ((@atomic :monotonic e.ref) || (@atomic :monotonic e.ref = true); e) + +# `default::F` makes Julia specialize on it, which it may skip for a function only passed along. +function Base.get!(default::F, c::ClockCache{K, V}, key) where {F <: Base.Callable, K, V} + p = _newkey(c, key) + e = _find(c, p) + if e !== nothing + _count!(c, HIT) + return _touch!(e).val + end + _count!(c, MISS) + v = convert(V, default())::V + sz = _entrysize(c, v) + @lock c.lock begin + # another task may have filled the key while we were computing + e = _find(c, p) + e === nothing || return e.val + _insert!(c, p, v, sz) + end + return v +end + +function Base.get(c::ClockCache, key, default) + e = _lookup(c, key) + return e === nothing ? default : _touch!(e).val +end +function Base.getindex(c::ClockCache, key) + e = _lookup(c, key) + e === nothing && throw(KeyError(key)) + return _touch!(e).val +end +Base.haskey(c::ClockCache, key) = _lookup(c, key) !== nothing +Base.length(c::ClockCache) = @lock c.lock c.count +Base.isempty(c::ClockCache) = length(c) == 0 -touch!(c::ClockCache, i::Int) = (c.ref[i] = true; c) +# Mutations, all with the lock held. -forget!(c::ClockCache, i::Int) = (c.live[i] = false; c) +# Insert a new key (not currently present), evicting until it fits. +# Values larger than the whole cache are not stored. +function _insert!(c::ClockCache{K, V}, p::Key, v, sz::Int) where {K, V} + sz > c.maxsize && return c + while c.currentsize + sz > c.maxsize && c.count > 0 + _evict_one!(c) + end + 2 * (c.used + 1) > length(c.table) && _rebuild!(c, c.count + 1) + isempty(c.free) && push!(c.ring, nothing) + pos = isempty(c.free) ? length(c.ring) : pop!(c.free) + e = Entry{K, V}(p.key, p.hash, v, sz, pos, false) + c.ring[pos] = e + _place!(c.table, e) === nothing && (c.used += 1) + c.count += 1 + c.currentsize += sz + return c +end -# Terminates within two sweeps, since the first sweep clears every reference bit. -function victim(c::ClockCache) - n = length(c.ref) +# Store `e` in the first free slot of its probe sequence; returns what was there. +function _place!(t::Vector{Any}, e::Entry) + mask = length(t) - 1 + i = e.hash % Int while true - c.hand = c.hand >= n ? 1 : c.hand + 1 - i = c.hand - c.live[i] || continue - c.ref[i] || return i - c.ref[i] = false + j = (i & mask) + 1 + old = t[j] + if !(old isa Entry) + t[j] = e + return old + end + i += 1 end return end + +# Publish a new table without tombstones, with room for `n` entries at a quarter load. +function _rebuild!(c::ClockCache, n::Int) + t = _emptytable(max(MINTABLE, nextpow(2, 4n))) + for e in c.ring + e === nothing || _place!(t, e) + end + c.used = c.count + @atomic :release c.table = t + return c +end + +function _remove!(c::ClockCache, e::Entry) + t = c.table + mask = length(t) - 1 + i = e.hash % Int + while t[(i & mask) + 1] !== e + i += 1 + end + if t[((i + 1) & mask) + 1] === nothing + # No probe continues past the next slot, so this slot, and the tombstones before it, + # can end probes as well: they become empty, which saves rebuilds. + while true + t[(i & mask) + 1] = nothing + c.used -= 1 + i -= 1 + t[(i & mask) + 1] === TOMBSTONE || break + end + else + t[(i & mask) + 1] = TOMBSTONE + end + c.ring[e.pos] = nothing + push!(c.free, e.pos) + c.count -= 1 + c.currentsize -= e.size + return c +end + +# Readers may set reference bits while the hand clears them, so after two sweeps without +# finding a clear bit the entry under the hand is evicted anyway. +function _evict_one!(c::ClockCache) + n = length(c.ring) + for sweep in 1:(2n + 1) + c.hand = c.hand >= n ? 1 : c.hand + 1 + e = c.ring[c.hand] + e === nothing && continue + if sweep <= 2n && (@atomic :monotonic e.ref) + @atomic :monotonic e.ref = false + else + return _remove!(c, e) + end + end + error("unreachable: no entry to evict") +end + +function Base.setindex!(c::ClockCache{K, V}, v, key) where {K, V} + p = _newkey(c, key) + val = convert(V, v)::V + sz = _entrysize(c, val) + @lock c.lock begin + e = _find(c, p) + e === nothing || _remove!(c, e) + _insert!(c, p, val, sz) + end + return c +end + +function Base.delete!(c::ClockCache, key) + @lock c.lock begin + e = _lookup(c, key) + e === nothing || _remove!(c, e) + end + return c +end + +function Base.resize!(c::ClockCache; maxsize::Integer) + maxsize >= 0 || throw(ArgumentError("maxsize must be non-negative")) + @lock c.lock begin + c.maxsize = maxsize + while c.currentsize > c.maxsize && c.count > 0 + _evict_one!(c) + end + end + return c +end + +function Base.empty!(c::ClockCache) + @lock c.lock begin + @atomic :release c.table = _emptytable() + empty!(c.ring) + empty!(c.free) + c.used = c.hand = c.count = c.currentsize = 0 + end + return c +end + +function cache_stats(c::ClockCache) + return @lock c.lock (; + hits = _counted(c, HIT), misses = _counted(c, MISS), length = c.count, c.currentsize, c.maxsize, c.by, + ) +end + +# Iteration walks a snapshot copied under the lock, in slot order; see the `SlotCache` methods. +Base.IteratorSize(::Type{<:ClockCache}) = Base.SizeUnknown() +function Base.iterate(c::ClockCache{K, V}) where {K, V} + snapshot = @lock c.lock Pair{K, V}[e.key => e.val for e in c.ring if e !== nothing] + return iterate(c, (snapshot, 1)) +end +function Base.iterate(::ClockCache, (snapshot, i)::Tuple{Vector, Int}) + return i > length(snapshot) ? nothing : (snapshot[i], (snapshot, i + 1)) +end + +# The entries below the compact header. +Base.show(io::IO, m::MIME"text/plain", c::ClockCache) = invoke(show, Tuple{IO, MIME"text/plain", AbstractDict}, io, m, c) diff --git a/src/containers/localclock.jl b/src/containers/localclock.jl new file mode 100644 index 0000000..3f5ed9b --- /dev/null +++ b/src/containers/localclock.jl @@ -0,0 +1,36 @@ +# `ClockCache` for a single task, where lock-free hits gain nothing: every operation takes the +# (uncontended) lock, and the hit and miss counters are plain fields. `TaskLocalCache{ClockCache}` +# uses it, see `_localtype`; it evicts in the same order as `ClockCache`. +mutable struct LocalClockCache{K, V} <: SlotCache{K, V} + const slots::Slots{V} + const ref::Vector{Bool} + const live::Vector{Bool} + hand::Int # last inspected slot +end + +LocalClockCache{K, V}(; maxsize::Integer = 10_000, by = nothing) where {K, V} = + LocalClockCache{K, V}(Slots{V}(maxsize, by), Bool[], Bool[], 0) + +function admit!(c::LocalClockCache, i::Int) + i > length(c.ref) && (push!(c.ref, false); push!(c.live, false)) + c.ref[i] = false + c.live[i] = true + return c +end + +touch!(c::LocalClockCache, i::Int) = (c.ref[i] = true; c) + +forget!(c::LocalClockCache, i::Int) = (c.live[i] = false; c) + +# Terminates within two sweeps, since the first sweep clears every reference bit. +function victim(c::LocalClockCache) + n = length(c.ref) + while true + c.hand = c.hand >= n ? 1 : c.hand + 1 + i = c.hand + c.live[i] || continue + c.ref[i] || return i + c.ref[i] = false + end + return +end diff --git a/src/containers/slots.jl b/src/containers/slots.jl index 0abd341..481aa39 100644 --- a/src/containers/slots.jl +++ b/src/containers/slots.jl @@ -51,19 +51,23 @@ function victim end function forget! end # The probe for a key that is inserted: converted to `K`, which throws if that is not possible. -_newkey(::SlotCache{K}, key) where {K} = Key(key isa K ? key : convert(K, key)::K) +_newkey(::AbstractCache{K}, key) where {K} = Key(key isa K ? key : convert(K, key)::K) -# The slot of `key` in `c`, or 0. A key that cannot be converted to `K` is not in the cache. -function _slot(c::SlotCache{K}, key) where {K} - key isa K && return get(c.slots.index, Key(key), 0) +# The probe for a key that is looked up, or `nothing`: a key that cannot be converted to `K` is +# not in the cache. +function _probe(::AbstractCache{K}, key) where {K} + key isa K && return Key(key) k = try convert(K, key)::K catch - return 0 + return nothing end - return get(c.slots.index, Key(k), 0) + return Key(k) end +# The slot of `key` in `c`, or 0. +_slot(c::SlotCache, key) = (p = _probe(c, key); p === nothing ? 0 : get(c.slots.index, p, 0)) + _entrysize(s::Slots, v) = s.by === nothing ? 1 : Int(s.by(v))::Int # Insert a new key (not currently present), evicting until it fits. diff --git a/test/Project.toml b/test/Project.toml index 3479116..d073d2f 100644 --- a/test/Project.toml +++ b/test/Project.toml @@ -3,6 +3,7 @@ Aqua = "4c88cf16-eb10-579e-8560-4a9242c79595" MemoizationKit = "1b238080-9255-4fe9-b224-89eb24efe93b" ParallelTestRunner = "d3525ed8-44d0-4b2c-a655-542cee43accc" Preferences = "21216c6a-2e73-6563-6e65-726566657250" +Random = "9a3f8284-a2c9-5f02-9a11-845980a1fd5c" SQLite = "0aa819cd-b072-5ff4-a722-6bc24af294d9" Serialization = "9e88b42a-f829-5b0c-bbe9-9e923198166b" Tachikoma = "468859d6-42d8-48b7-8ad9-1d312e0e3b0a" @@ -15,6 +16,7 @@ MemoizationKit = {path = ".."} [compat] Aqua = "0.8.9" ParallelTestRunner = "2.2.0" +Random = "1.10" SQLite = "1.6" Serialization = "1" Tachikoma = "2.7" diff --git a/test/test_concurrency.jl b/test/test_concurrency.jl new file mode 100644 index 0000000..3e7789b --- /dev/null +++ b/test/test_concurrency.jl @@ -0,0 +1,117 @@ +using Test +using Random +using MemoizationKit +using MemoizationKit: cache_stats, LocalClockCache + +# Every value is a function of its key, so any value returned for the wrong key is detected. +expected(k::Int) = 3k +expected(k::NTuple{4, Int}) = (sum(k), k[1]) +expected(k::String) = length(k) +expected(k::Tuple{Vector{Int}, Symbol}) = (sum(k[1]), k[2]) + +function randkey(rng, n) + i = rand(rng, 1:n) + r = rand(rng) + return r < 0.4 ? i : r < 0.7 ? (i, i + 1, -i, 2i) : r < 0.85 ? string(i) : ([i, 1], :s) +end + +# Many tasks doing every operation at once, checking each returned value; returns the number +# of `get!` calls and whether all values were correct. +function stress!(c, ntasks, niter; nkeys = 500, maxsizes = (8, 64, 256)) + gets = Threads.Atomic{Int}(0) + ok = Threads.Atomic{Bool}(true) + @sync for t in 1:ntasks + Threads.@spawn begin + rng = Xoshiro(t) + n = 0 + for _ in 1:niter + k = randkey(rng, nkeys) + r = rand(rng) + if r < 0.8 + n += 1 + isequal(get!(() -> expected(k), c, k), expected(k)) || (ok[] = false) + elseif r < 0.9 + v = get(c, k, nothing) + v === nothing || isequal(v, expected(k)) || (ok[] = false) + elseif r < 0.95 + c[k] = expected(k) + elseif r < 0.99 + delete!(c, k) + elseif r < 0.995 + resize!(c; maxsize = rand(rng, maxsizes)) + elseif r < 0.997 + empty!(c) + else + all(((k, v),) -> isequal(v, expected(k)), c) || (ok[] = false) + end + end + Threads.atomic_add!(gets, n) + end + end + return gets[], ok[] +end + +const NTASKS = 4 * Threads.nthreads() + +@testset "$C: concurrent stress" for C in (LRU, ClockCache, LocalClockCache) + for by in (nothing, x -> 1 + (x isa Tuple ? 1 : 0)) + c = C{Any, Any}(; maxsize = 64, by) + gets, ok = stress!(c, NTASKS, 20_000) + @test ok + s = cache_stats(c) + @test s.hits + s.misses == gets + @test s.currentsize <= s.maxsize + @test length(c) == length(collect(c)) == s.length + @test all(((k, v),) -> isequal(v, expected(k)), c) + @test s.currentsize == (by === nothing ? length(c) : sum(by(v) for (k, v) in c; init = 0)) + end +end + +# Many readers of a few hot keys while a writer churns the rest of the cache, so that the +# lock-free table is rebuilt and entries are evicted under the readers. +@testset "ClockCache: hits during rebuilds" begin + c = ClockCache{Any, Any}(; maxsize = 2_000) + ok = Threads.Atomic{Bool}(true) + done = Threads.Atomic{Bool}(false) + @sync begin + for t in 1:max(1, Threads.nthreads() - 1) + Threads.@spawn while !done[] + k = mod(t, 8) + get!(() -> 3k, c, k) == 3k || (ok[] = false) + v = get(c, (k, k + 1, -k, 2k), nothing) + v === nothing || v == expected((k, k + 1, -k, 2k)) || (ok[] = false) + yield() # lets the writer run when there is a single thread + end + end + Threads.@spawn begin + rng = Xoshiro(0) + for i in 1:100_000 + k = (i, i + 1, -i, 2i) + get!(() -> expected(k), c, k) + i % 25_000 == 0 && empty!(c) + i % 7_919 == 0 && resize!(c; maxsize = rand(rng, 100:3_000)) + end + done[] = true + end + end + @test ok[] + @test all(((k, v),) -> isequal(v, expected(k)), c) +end + +@cached shared(x::Int) = x^2 +MemoizationKit.CacheStyle(::typeof(shared), ::Int) = GlobalCache{ClockCache}() +@cached tasklocal(x::Int) = x^2 +MemoizationKit.CacheStyle(::typeof(tasklocal), ::Int) = TaskLocalCache{ClockCache}() + +@testset "@cached with $f" for f in (shared, tasklocal) + @test @inferred(f(3)) == 9 + f(4) + @test (@allocated f(4)) == 0 +end + +@testset "containers of @cached" begin + c = only(cache_info(shared)).second + @test c isa ClockCache{Any, Any} + @test cache_stats(c).misses == 2 && cache_stats(c).hits >= 1 + @test MemoizationKit._container(tasklocal, TaskLocalCache{ClockCache}(), (1,), Int) isa LocalClockCache{Any, Any} +end diff --git a/test/test_containers.jl b/test/test_containers.jl index 349a10a..2b832f0 100644 --- a/test/test_containers.jl +++ b/test/test_containers.jl @@ -1,8 +1,8 @@ using Test using MemoizationKit -using MemoizationKit: AbstractCache, cache_stats +using MemoizationKit: AbstractCache, cache_stats, LocalClockCache -const CACHETYPES = (LRU, ClockCache) +const CACHETYPES = (LRU, ClockCache, LocalClockCache) @testset "$C: basic interface" for C in CACHETYPES c = C{Int, String}(; maxsize = 3) @@ -64,8 +64,8 @@ end @test !haskey(c, 1) && haskey(c, 2) end -@testset "ClockCache: second chance" begin - c = ClockCache{Int, Int}(; maxsize = 3) +@testset "$C: second chance" for C in (ClockCache, LocalClockCache) + c = C{Int, Int}(; maxsize = 3) for i in 1:3 c[i] = i end From 4f3012d32c22454c1e25a37b608911c63fec89b9 Mon Sep 17 00:00:00 2001 From: lkdvos Date: Wed, 14 Oct 2026 19:06:04 -0400 Subject: [PATCH 2/6] Simplify ClockCache and use AtomicMemory on Julia 1.12+ - Slots are read and written through `@atomic` indexing of an `AtomicMemory` on 1.12+. On 1.10 and 1.11 the pointer load and `setindex!` stay, with a test that the store compiles to a release store with a GC write barrier. - Fewer single-use helpers and descriptive names without leading underscores; the fields of `Entry` read on a hit come first. - The hand no longer errors when, after two sweeps of bits re-set by readers, it lands on a free position: it moves on to the next entry. - Remove benchmark/contention.jl. Co-Authored-By: Claude Opus 5.5 --- benchmark/contention.jl | 134 ------------------ src/containers/clock.jl | 298 +++++++++++++++++++-------------------- src/containers/slots.jl | 2 +- test/Project.toml | 2 + test/test_concurrency.jl | 11 ++ 5 files changed, 158 insertions(+), 289 deletions(-) delete mode 100644 benchmark/contention.jl diff --git a/benchmark/contention.jl b/benchmark/contention.jl deleted file mode 100644 index cf13700..0000000 --- a/benchmark/contention.jl +++ /dev/null @@ -1,134 +0,0 @@ -# Cost of a lookup when all threads share one cache, against task-local caches. -# "locked Clock" is the `ClockCache` of task-local caches (`MemoizationKit.LocalClockCache`), whose hits -# take the lock, shared here for comparison. -# -# for t in 1 2 4 8; do julia --heap-size-hint=4G --project=benchmark -t $t benchmark/contention.jl; done -# -# One task per thread, each doing `M` lookups from its own key sequence. Reports the median over -# `REPS` runs of the time per lookup within a task (ns/lookup, lower is better; perfect scaling -# keeps it constant as threads are added), and the bytes allocated per lookup. - -using Random, Printf, Statistics -using MemoizationKit: MemoizationKit, ClockCache, LocalClockCache, LRU, @cached, GlobalCache, TaskLocalCache - -const M = 200_000 -const REPS = 7 - -const CONTAINERS = [ - "ClockCache" => n -> ClockCache{Any, Any}(; maxsize = n), - "locked Clock" => n -> LocalClockCache{Any, Any}(; maxsize = n), - "LRU" => n -> LRU{Any, Any}(; maxsize = n), - "task-local Clock" => nothing, # one locked Clock per task, uncontended -] - -function zipf(rng, n, s, len) - cdf = cumsum(1 ./ (1:n) .^ s) - cdf ./= cdf[end] - return [searchsortedfirst(cdf, rand(rng)) for _ in 1:len] -end - -tuplekey(i) = (i, i % 7, :leg, Float64(i)) -compute(k::Int) = 3k -compute(k::Tuple) = k[1] - -# name => (key sequence per task, cache capacity, prefill keys) -function workloads(ntasks) - rngs = [Xoshiro(t) for t in 1:ntasks] - return [ - "hits, Int keys (1k)" => ([rand(r, 1:1_000, M) for r in rngs], 10_000, 1:1_000), - "hits, 4-tuple keys (1k)" => ([tuplekey.(rand(r, 1:1_000, M)) for r in rngs], 10_000, tuplekey.(1:1_000)), - "hits, one hot Int key" => ([fill(7, M) for r in rngs], 10_000, 7:7), - "zipf 1.0 (100k keys, cap 10k)" => ([zipf(r, 100_000, 1.0, M) for r in rngs], 10_000, 1:0), - "uniform (20k keys, cap 10k)" => ([rand(r, 1:20_000, M) for r in rngs], 10_000, 1:0), - ] -end - -function run!(c, seq) - acc = 0 - for k in seq - acc += get!(() -> compute(k), c, k)::Int - end - return acc -end - -# ns per lookup in each task, and total bytes allocated -function threaded(make, seqs, cap, prefill) - shared = make === nothing ? nothing : make(cap) - caches = [shared === nothing ? LocalClockCache{Any, Any}(; maxsize = cap) : shared for _ in seqs] - for c in unique(caches), k in prefill - get!(() -> compute(k), c, k) - end - times = zeros(length(seqs)) - start = Threads.Atomic{Int}(0) - bytes = @allocated @sync for (t, seq) in enumerate(seqs) - Threads.@spawn begin - Threads.atomic_add!(start, 1) - while start[] < length(seqs) # start together - ccall(:jl_cpu_pause, Cvoid, ()) - end - t0 = time_ns() - run!(caches[t], seq) - times[t] = (time_ns() - t0) / length(seq) - end - end - return mean(times), bytes / sum(length, seqs) -end - -# ns per hit through `@cached` functions using each cache style, Int and 4-tuple keys -@cached f_clock(x) = compute(x) -@cached f_locked(x) = compute(x) -@cached f_lru(x) = compute(x) -@cached f_local(x) = compute(x) -MemoizationKit.CacheStyle(::typeof(f_clock), x) = GlobalCache{ClockCache}() -MemoizationKit.CacheStyle(::typeof(f_locked), x) = GlobalCache{LocalClockCache}() -MemoizationKit.CacheStyle(::typeof(f_lru), x) = GlobalCache{LRU}() -MemoizationKit.CacheStyle(::typeof(f_local), x) = TaskLocalCache{ClockCache}() -const CACHED = ["ClockCache" => f_clock, "locked Clock" => f_locked, "LRU" => f_lru, "task-local Clock" => f_local] - -function callall(f::F, seq) where {F} - acc = 0 - for k in seq - acc += f(k)::Int - end - return acc -end - -function threaded_cached(f, seqs, keys) - times = zeros(length(seqs)) - bytes = @allocated @sync for (t, seq) in enumerate(seqs) - Threads.@spawn begin - callall(f, keys) # fill this task's cache (task-local) - t0 = time_ns() - callall(f, seq) - times[t] = (time_ns() - t0) / length(seq) - end - end - return mean(times), bytes / sum(length, seqs) -end - -nt = Threads.nthreads() -println("Julia ", VERSION, ", ", nt, " threads (", nt, " tasks), ", M, " lookups per task, median of ", REPS, " runs") -println("load average: ", read(`cat /proc/loadavg`, String)) -@printf("%-30s %-17s %10s %10s %8s\n", "workload", "container", "ns/lookup", "(min)", "B/lookup") -for (wname, (seqs, cap, prefill)) in workloads(nt) - for (cname, make) in CONTAINERS - threaded(make, seqs, cap, prefill) # compile - rs = [threaded(make, seqs, cap, prefill) for _ in 1:REPS] - @printf( - "%-30s %-17s %10.1f %10.1f %8.2f\n", wname, cname, - median(first.(rs)), minimum(first.(rs)), median(last.(rs)) - ) - end -end -for (kname, keys) in ("@cached hits, Int" => 1:1_000, "@cached hits, 4-tuple" => tuplekey.(1:1_000)) - seqs = [rand(Xoshiro(t), keys, M) for t in 1:nt] - for (cname, f) in CACHED - foreach(f, keys) - threaded_cached(f, seqs, keys) - rs = [threaded_cached(f, seqs, keys) for _ in 1:REPS] - @printf( - "%-30s %-17s %10.1f %10.1f %8.2f\n", kname, cname, - median(first.(rs)), minimum(first.(rs)), median(last.(rs)) - ) - end -end diff --git a/src/containers/clock.jl b/src/containers/clock.jl index a939591..a2eabc0 100644 --- a/src/containers/clock.jl +++ b/src/containers/clock.jl @@ -1,16 +1,44 @@ -# An entry of a `ClockCache`: immutable apart from its reference bit, see the rules below. +# An entry of a `ClockCache`; only its reference bit changes after construction. The fields read +# on a hit come first. mutable struct Entry{K, V} - const key::K const hash::UInt + const key::K const val::V + @atomic used::Bool # the reference bit const size::Int - const pos::Int # slot in `ring` - @atomic ref::Bool + const position::Int # in `ring` end struct Tombstone end const TOMBSTONE = Tombstone() +# The table is an open-addressing hash table with linear probing, at most half full, so that +# every probe ends at an empty slot. Readers take no lock, which is safe because of three rules: +# +# 1. Entries do not change after construction, apart from their reference bit, so a reader +# sees a key together with its own value. +# 2. A published table never changes length: it is replaced as a whole, by a release store to +# `c.table`, so the indices of a reader stay in bounds. +# 3. Under the lock, a slot only changes from `nothing` to an entry, from an entry to +# `TOMBSTONE`, or from `TOMBSTONE` to an entry; or to `nothing` when the next slot is +# `nothing`, so that no probe passes it anyway. Entries never move, so a probe for a key that +# is present and not concurrently removed always finds it. +# +# Slots are loaded with acquire and stored with release ordering, so a reader sees fully +# constructed entries. On Julia 1.10 and 1.11, which lack atomic indexing, a `Vector{Any}` is +# read through a pointer and written with `setindex!`, which compiles to a release store with a +# GC write barrier there (checked by the tests). Storing through a pointer would skip the +# barrier, and the GC could free the entry. +@static if VERSION >= v"1.12" + const SlotTable = AtomicMemory{Any} + loadslot(t::SlotTable, i::Int) = @atomic :acquire t[i] + storeslot!(t::SlotTable, i::Int, x) = (@atomic :release t[i] = x; t) +else + const SlotTable = Vector{Any} + loadslot(t::SlotTable, i::Int) = GC.@preserve t unsafe_load(pointer(t, i), :acquire) + storeslot!(t::SlotTable, i::Int, x) = (t[i] = x; t) +end + """ ClockCache{K, V}(; maxsize::Integer = 10_000, by = nothing) @@ -28,234 +56,196 @@ with a removal may still return the removed value, and one that races with an in miss it; a returned value always belongs to its key. """ mutable struct ClockCache{K, V} <: AbstractCache{K, V} - @atomic table::Vector{Any} # open-addressing hash table of `Entry{K, V}`s, see `_find` - used::Int # entries and tombstones in `table` - const ring::Vector{Union{Nothing, Entry{K, V}}} # the clock, with `nothing` for a free slot - const free::Vector{Int} # free slots of `ring` - hand::Int # last inspected slot of `ring` + @atomic table::SlotTable # entries, read without the lock + occupied::Int # slots of `table` holding an entry or a tombstone + const ring::Vector{Union{Nothing, Entry{K, V}}} # the clock, `nothing` for a free position + const freepositions::Vector{Int} + hand::Int # last inspected position of `ring` count::Int currentsize::Int maxsize::Int const by::Any - const counts::Vector{Int} # hits and misses per thread, see `_count!` + const counts::Vector{Int} # hits and misses per thread, see `countlookup!` const lock::ReentrantLock end -# Lock-free lookups rest on three rules, which every method below keeps: -# -# 1. An `Entry` is immutable apart from its reference bit, so a reader that loads one sees a key -# with its own value. -# 2. A published table never changes length: it is replaced as a whole, by a release store to -# `c.table`. Readers load the table and then each slot with acquire ordering, so they see -# fully constructed entries, and their indices are always in bounds. -# 3. In a published table, a slot only changes from `nothing` to an entry, from an entry to -# `TOMBSTONE`, or from `TOMBSTONE` to an entry, all under the lock; or to `nothing` when the -# next slot is `nothing`, so that no probe passes it anyway. Entries never move, so a probe -# for a key that is present and not concurrently removed always finds it. -# -# Writers store slots with `setindex!`, which is a release store with a GC write barrier on every -# supported Julia version. (Storing through a pointer would skip the barrier, and the GC could -# free the entry.) Removed entries are dropped from the table rather than unset, so readers -# never see an undefined slot, and the GC collects them once no reader holds them. - -const MINTABLE = 16 -_emptytable(n::Int = MINTABLE) = fill!(Vector{Any}(undef, n), nothing) - -const COUNTSTRIDE = 8 # `Int`s per 64-byte cache line -const HIT, MISS = 1, 2 +function emptytable(n::Int = 16) + t = SlotTable(undef, n) + foreach(i -> storeslot!(t, i, nothing), 1:n) + return t +end function ClockCache{K, V}(; maxsize::Integer = 10_000, by = nothing) where {K, V} maxsize >= 0 || throw(ArgumentError("maxsize must be non-negative")) - return ClockCache{K, V}( - _emptytable(), 0, Union{Nothing, Entry{K, V}}[], Int[], 0, 0, 0, maxsize, by, - zeros(Int, COUNTSTRIDE * Threads.maxthreadid()), ReentrantLock() - ) + counts = zeros(Int, 8 * Threads.maxthreadid()) + return ClockCache{K, V}(emptytable(), 0, Union{Nothing, Entry{K, V}}[], Int[], 0, 0, 0, maxsize, by, counts, ReentrantLock()) end -_entrysize(c::ClockCache, v) = c.by === nothing ? 1 : Int(c.by(v))::Int +# Slot indices run from 1 to `length(t)`, a power of two; the probe for `hash` starts at +# `(hash % Int) & mask + 1`, and moves on with `i & mask + 1`, wrapping around at the end. -@inline _loadslot(t::Vector{Any}, i::Int) = GC.@preserve t unsafe_load(pointer(t, i), :acquire) - -# The entry for the probe `p` in table `t`, or `nothing`. Linear probing from the hash; a table -# is at most half full, so the probe ends at an empty slot. -@inline function _find(t::Vector{Any}, p::Key{P}, ::Type{E}) where {P, E <: Entry} +# The entry of the probe `p` (see `Key`), or `nothing`. +findentry(::ClockCache, ::Nothing) = nothing +@inline function findentry(c::ClockCache{K, V}, p::Key{P}) where {K, V, P} + t = @atomic :acquire c.table mask = length(t) - 1 - i = p.hash % Int - for _ in 0:mask # bounds the loop also if slots change while probing - e = _loadslot(t, (i & mask) + 1) + i = (p.hash % Int) & mask + 1 + for _ in 0:mask # bounds the probe also while slots change + e = loadslot(t, i) e === nothing && return nothing - if e isa E && e.hash == p.hash - k = e.key - k isa P && isequal(p.key, k::P) && return e - end - i += 1 + e isa Entry{K, V} && e.hash == p.hash && e.key isa P && isequal(p.key, e.key::P) && return e + i = i & mask + 1 end return nothing end -_find(c::ClockCache{K, V}, p::Key) where {K, V} = _find((@atomic :acquire c.table), p, Entry{K, V}) -_lookup(c::ClockCache, key) = (p = _probe(c, key); p === nothing ? nothing : _find(c, p)) +# A hit sets the reference bit only if it is clear, so hot entries are not written to at all. +@inline function usevalue!(e::Entry) + (@atomic :monotonic e.used) || (@atomic :monotonic e.used = true) + return e.val +end -# Hits and misses are counted per thread, in separate cache lines, with relaxed loads and stores: -# a shared counter would bounce between cores on every hit. Two tasks on one thread cannot -# interleave between the load and the store, since neither yields; threads adopted after the -# cache was created share counters, which may then lose a few counts. -@inline function _count!(c::ClockCache, which::Int) - v = c.counts - i = mod(Threads.threadid() - 1, length(v) ÷ COUNTSTRIDE) * COUNTSTRIDE + which - GC.@preserve v begin - p = pointer(v, i) +# Hits and misses are counted per thread, a cache line (8 `Int`s) apart, since a shared counter +# would bounce between cores on every hit. Two tasks on one thread cannot interleave between the +# load and the store, since neither yields; threads adopted after the cache was created share +# counters, and may then lose a few counts. +const HIT, MISS = 1, 2 +@inline function countlookup!(c::ClockCache, kind::Int) + counts = c.counts + i = 8 * mod(Threads.threadid() - 1, length(counts) ÷ 8) + kind + GC.@preserve counts begin + p = pointer(counts, i) unsafe_store!(p, unsafe_load(p, :monotonic) + 1, :monotonic) end return nothing end -function _counted(c::ClockCache, which::Int) - v = c.counts - return GC.@preserve v sum(i -> unsafe_load(pointer(v, i), :monotonic), which:COUNTSTRIDE:length(v)) -end - -# Hits only write the reference bit if it is clear, so hot entries are not written at all. -@inline _touch!(e::Entry) = ((@atomic :monotonic e.ref) || (@atomic :monotonic e.ref = true); e) -# `default::F` makes Julia specialize on it, which it may skip for a function only passed along. function Base.get!(default::F, c::ClockCache{K, V}, key) where {F <: Base.Callable, K, V} - p = _newkey(c, key) - e = _find(c, p) + probe = _newkey(c, key) + e = findentry(c, probe) if e !== nothing - _count!(c, HIT) - return _touch!(e).val + countlookup!(c, HIT) + return usevalue!(e) end - _count!(c, MISS) + countlookup!(c, MISS) v = convert(V, default())::V sz = _entrysize(c, v) @lock c.lock begin # another task may have filled the key while we were computing - e = _find(c, p) + e = findentry(c, probe) e === nothing || return e.val - _insert!(c, p, v, sz) + addentry!(c, probe, v, sz) end return v end function Base.get(c::ClockCache, key, default) - e = _lookup(c, key) - return e === nothing ? default : _touch!(e).val + e = findentry(c, _probe(c, key)) + return e === nothing ? default : usevalue!(e) end function Base.getindex(c::ClockCache, key) - e = _lookup(c, key) + e = findentry(c, _probe(c, key)) e === nothing && throw(KeyError(key)) - return _touch!(e).val + return usevalue!(e) end -Base.haskey(c::ClockCache, key) = _lookup(c, key) !== nothing +Base.haskey(c::ClockCache, key) = findentry(c, _probe(c, key)) !== nothing Base.length(c::ClockCache) = @lock c.lock c.count Base.isempty(c::ClockCache) = length(c) == 0 -# Mutations, all with the lock held. +# The mutations below are called with the lock held. -# Insert a new key (not currently present), evicting until it fits. -# Values larger than the whole cache are not stored. -function _insert!(c::ClockCache{K, V}, p::Key, v, sz::Int) where {K, V} +# Insert a key that is not present, evicting until it fits; a value larger than the whole cache +# is not stored. +function addentry!(c::ClockCache{K, V}, p::Key, v, sz::Int) where {K, V} sz > c.maxsize && return c while c.currentsize + sz > c.maxsize && c.count > 0 - _evict_one!(c) + evict!(c) end - 2 * (c.used + 1) > length(c.table) && _rebuild!(c, c.count + 1) - isempty(c.free) && push!(c.ring, nothing) - pos = isempty(c.free) ? length(c.ring) : pop!(c.free) - e = Entry{K, V}(p.key, p.hash, v, sz, pos, false) - c.ring[pos] = e - _place!(c.table, e) === nothing && (c.used += 1) + if 2 * (c.occupied + 1) > length(c.table) + # publish a new table without tombstones, at most a quarter full + t = emptytable(max(16, nextpow(2, 4 * (c.count + 1)))) + foreach(e -> e === nothing || placeentry!(t, e), c.ring) + c.occupied = c.count + @atomic :release c.table = t + end + position = isempty(c.freepositions) ? length(push!(c.ring, nothing)) : pop!(c.freepositions) + e = Entry{K, V}(p.hash, p.key, v, false, sz, position) + c.ring[position] = e + placeentry!(c.table, e) === nothing && (c.occupied += 1) c.count += 1 c.currentsize += sz return c end -# Store `e` in the first free slot of its probe sequence; returns what was there. -function _place!(t::Vector{Any}, e::Entry) +# Store `e` in the first slot of its probe without an entry, and return what was there. +function placeentry!(t::SlotTable, e::Entry) mask = length(t) - 1 - i = e.hash % Int - while true - j = (i & mask) + 1 - old = t[j] - if !(old isa Entry) - t[j] = e - return old - end - i += 1 + i = (e.hash % Int) & mask + 1 + while (old = loadslot(t, i)) isa Entry + i = i & mask + 1 end - return + storeslot!(t, i, e) + return old end -# Publish a new table without tombstones, with room for `n` entries at a quarter load. -function _rebuild!(c::ClockCache, n::Int) - t = _emptytable(max(MINTABLE, nextpow(2, 4n))) - for e in c.ring - e === nothing || _place!(t, e) - end - c.used = c.count - @atomic :release c.table = t - return c -end - -function _remove!(c::ClockCache, e::Entry) +function removeentry!(c::ClockCache, e::Entry) t = c.table mask = length(t) - 1 - i = e.hash % Int - while t[(i & mask) + 1] !== e - i += 1 + i = (e.hash % Int) & mask + 1 + while loadslot(t, i) !== e + i = i & mask + 1 end - if t[((i + 1) & mask) + 1] === nothing - # No probe continues past the next slot, so this slot, and the tombstones before it, - # can end probes as well: they become empty, which saves rebuilds. + if loadslot(t, i & mask + 1) === nothing + # No probe continues past the next slot, so this slot and the tombstones before it can + # end probes as well: they become empty, which saves rebuilding the table. while true - t[(i & mask) + 1] = nothing - c.used -= 1 - i -= 1 - t[(i & mask) + 1] === TOMBSTONE || break + storeslot!(t, i, nothing) + c.occupied -= 1 + i = (i - 2) & mask + 1 + loadslot(t, i) === TOMBSTONE || break end else - t[(i & mask) + 1] = TOMBSTONE + storeslot!(t, i, TOMBSTONE) end - c.ring[e.pos] = nothing - push!(c.free, e.pos) + c.ring[e.position] = nothing + push!(c.freepositions, e.position) c.count -= 1 c.currentsize -= e.size return c end -# Readers may set reference bits while the hand clears them, so after two sweeps without -# finding a clear bit the entry under the hand is evicted anyway. -function _evict_one!(c::ClockCache) +# The hand clears set reference bits until it finds a clear one. Readers may set bits again +# meanwhile, so after two sweeps the hand evicts the next entry anyway. Requires `c.count > 0`. +function evict!(c::ClockCache) n = length(c.ring) - for sweep in 1:(2n + 1) - c.hand = c.hand >= n ? 1 : c.hand + 1 + steps = 0 + while true + c.hand = c.hand % n + 1 + steps += 1 e = c.ring[c.hand] e === nothing && continue - if sweep <= 2n && (@atomic :monotonic e.ref) - @atomic :monotonic e.ref = false - else - return _remove!(c, e) - end + (steps > 2n || !(@atomic :monotonic e.used)) && return removeentry!(c, e) + @atomic :monotonic e.used = false end - error("unreachable: no entry to evict") + return end function Base.setindex!(c::ClockCache{K, V}, v, key) where {K, V} - p = _newkey(c, key) + probe = _newkey(c, key) val = convert(V, v)::V sz = _entrysize(c, val) @lock c.lock begin - e = _find(c, p) - e === nothing || _remove!(c, e) - _insert!(c, p, val, sz) + e = findentry(c, probe) + e === nothing || removeentry!(c, e) + addentry!(c, probe, val, sz) end return c end function Base.delete!(c::ClockCache, key) @lock c.lock begin - e = _lookup(c, key) - e === nothing || _remove!(c, e) + e = findentry(c, _probe(c, key)) + e === nothing || removeentry!(c, e) end return c end @@ -265,7 +255,7 @@ function Base.resize!(c::ClockCache; maxsize::Integer) @lock c.lock begin c.maxsize = maxsize while c.currentsize > c.maxsize && c.count > 0 - _evict_one!(c) + evict!(c) end end return c @@ -273,21 +263,21 @@ end function Base.empty!(c::ClockCache) @lock c.lock begin - @atomic :release c.table = _emptytable() + @atomic :release c.table = emptytable() empty!(c.ring) - empty!(c.free) - c.used = c.hand = c.count = c.currentsize = 0 + empty!(c.freepositions) + c.occupied = c.hand = c.count = c.currentsize = 0 end return c end function cache_stats(c::ClockCache) - return @lock c.lock (; - hits = _counted(c, HIT), misses = _counted(c, MISS), length = c.count, c.currentsize, c.maxsize, c.by, - ) + counts = c.counts + total(kind) = GC.@preserve counts sum(i -> unsafe_load(pointer(counts, i), :monotonic), kind:8:length(counts)) + return @lock c.lock (; hits = total(HIT), misses = total(MISS), length = c.count, c.currentsize, c.maxsize, c.by) end -# Iteration walks a snapshot copied under the lock, in slot order; see the `SlotCache` methods. +# Iteration walks a snapshot copied under the lock, in insertion order, like that of `SlotCache`. Base.IteratorSize(::Type{<:ClockCache}) = Base.SizeUnknown() function Base.iterate(c::ClockCache{K, V}) where {K, V} snapshot = @lock c.lock Pair{K, V}[e.key => e.val for e in c.ring if e !== nothing] diff --git a/src/containers/slots.jl b/src/containers/slots.jl index 481aa39..bd94fa4 100644 --- a/src/containers/slots.jl +++ b/src/containers/slots.jl @@ -68,7 +68,7 @@ end # The slot of `key` in `c`, or 0. _slot(c::SlotCache, key) = (p = _probe(c, key); p === nothing ? 0 : get(c.slots.index, p, 0)) -_entrysize(s::Slots, v) = s.by === nothing ? 1 : Int(s.by(v))::Int +_entrysize(c, v) = c.by === nothing ? 1 : Int(c.by(v))::Int # `c` is a `Slots` or a `ClockCache` # Insert a new key (not currently present), evicting until it fits. # Values larger than the whole cache are not stored. diff --git a/test/Project.toml b/test/Project.toml index d073d2f..fc5d33e 100644 --- a/test/Project.toml +++ b/test/Project.toml @@ -1,5 +1,6 @@ [deps] Aqua = "4c88cf16-eb10-579e-8560-4a9242c79595" +InteractiveUtils = "b77e0a4c-d291-57a0-90e8-8db25a27a240" MemoizationKit = "1b238080-9255-4fe9-b224-89eb24efe93b" ParallelTestRunner = "d3525ed8-44d0-4b2c-a655-542cee43accc" Preferences = "21216c6a-2e73-6563-6e65-726566657250" @@ -15,6 +16,7 @@ MemoizationKit = {path = ".."} [compat] Aqua = "0.8.9" +InteractiveUtils = "1.10" ParallelTestRunner = "2.2.0" Random = "1.10" SQLite = "1.6" diff --git a/test/test_concurrency.jl b/test/test_concurrency.jl index 3e7789b..c4c3a44 100644 --- a/test/test_concurrency.jl +++ b/test/test_concurrency.jl @@ -1,5 +1,6 @@ using Test using Random +using InteractiveUtils: code_llvm using MemoizationKit using MemoizationKit: cache_stats, LocalClockCache @@ -98,6 +99,16 @@ end @test all(((k, v),) -> isequal(v, expected(k)), c) end +# Without `AtomicMemory`, lock-free readers rely on `setindex!` into a `Vector{Any}` compiling to +# a release store with a GC write barrier. +if VERSION < v"1.12" + @testset "ClockCache: slot stores are release stores" begin + ir = sprint(code_llvm, MemoizationKit.storeslot!, (Vector{Any}, Int, Any)) + @test occursin(r"store atomic .* release", ir) + @test occursin("gc_queue_root", ir) + end +end + @cached shared(x::Int) = x^2 MemoizationKit.CacheStyle(::typeof(shared), ::Int) = GlobalCache{ClockCache}() @cached tasklocal(x::Int) = x^2 From d2bc8ded9f8396837cac17a9f2a71ac892515c14 Mon Sep 17 00:00:00 2001 From: lkdvos Date: Thu, 15 Oct 2026 11:09:48 -0400 Subject: [PATCH 3/6] Define the ClockCache constructor inside the struct The default constructor had unbound type parameters, which Aqua flags on Julia 1.10, since K and V only occur in the Union element type of ring. Co-Authored-By: Claude Opus 5.5 --- src/containers/clock.jl | 12 ++++++------ 1 file changed, 6 insertions(+), 6 deletions(-) diff --git a/src/containers/clock.jl b/src/containers/clock.jl index a2eabc0..96a19c7 100644 --- a/src/containers/clock.jl +++ b/src/containers/clock.jl @@ -67,6 +67,12 @@ mutable struct ClockCache{K, V} <: AbstractCache{K, V} const by::Any const counts::Vector{Int} # hits and misses per thread, see `countlookup!` const lock::ReentrantLock + + function ClockCache{K, V}(; maxsize::Integer = 10_000, by = nothing) where {K, V} + maxsize >= 0 || throw(ArgumentError("maxsize must be non-negative")) + counts = zeros(Int, 8 * Threads.maxthreadid()) + return new{K, V}(emptytable(), 0, Union{Nothing, Entry{K, V}}[], Int[], 0, 0, 0, maxsize, by, counts, ReentrantLock()) + end end function emptytable(n::Int = 16) @@ -75,12 +81,6 @@ function emptytable(n::Int = 16) return t end -function ClockCache{K, V}(; maxsize::Integer = 10_000, by = nothing) where {K, V} - maxsize >= 0 || throw(ArgumentError("maxsize must be non-negative")) - counts = zeros(Int, 8 * Threads.maxthreadid()) - return ClockCache{K, V}(emptytable(), 0, Union{Nothing, Entry{K, V}}[], Int[], 0, 0, 0, maxsize, by, counts, ReentrantLock()) -end - # Slot indices run from 1 to `length(t)`, a power of two; the probe for `hash` starts at # `(hash % Int) & mask + 1`, and moves on with `i & mask + 1`, wrapping around at the end. From cb8ecb7c20ead2d864b5c692728298b4b91fe358 Mon Sep 17 00:00:00 2001 From: lkdvos Date: Thu, 15 Oct 2026 11:13:40 -0400 Subject: [PATCH 4/6] Find the per-thread counters without a division The counter array holds a power of two of cache lines, so a hit finds its thread's line with a mask instead of mod. Hits and misses are an @enum, and the hand wraps with a comparison. Co-Authored-By: Claude Opus 5.5 --- src/containers/clock.jl | 17 +++++++++-------- 1 file changed, 9 insertions(+), 8 deletions(-) diff --git a/src/containers/clock.jl b/src/containers/clock.jl index 96a19c7..135e46e 100644 --- a/src/containers/clock.jl +++ b/src/containers/clock.jl @@ -70,7 +70,7 @@ mutable struct ClockCache{K, V} <: AbstractCache{K, V} function ClockCache{K, V}(; maxsize::Integer = 10_000, by = nothing) where {K, V} maxsize >= 0 || throw(ArgumentError("maxsize must be non-negative")) - counts = zeros(Int, 8 * Threads.maxthreadid()) + counts = zeros(Int, 8 * nextpow(2, Threads.maxthreadid())) return new{K, V}(emptytable(), 0, Union{Nothing, Entry{K, V}}[], Int[], 0, 0, 0, maxsize, by, counts, ReentrantLock()) end end @@ -106,13 +106,14 @@ end end # Hits and misses are counted per thread, a cache line (8 `Int`s) apart, since a shared counter -# would bounce between cores on every hit. Two tasks on one thread cannot interleave between the -# load and the store, since neither yields; threads adopted after the cache was created share +# would bounce between cores on every hit. There is a power of two of such lines, so a thread +# finds its own without a division. Two tasks on one thread cannot interleave between the load +# and the store, since neither yields; threads adopted after the cache was created share # counters, and may then lose a few counts. -const HIT, MISS = 1, 2 -@inline function countlookup!(c::ClockCache, kind::Int) +@enum LookupOutcome HIT = 1 MISS = 2 +@inline function countlookup!(c::ClockCache, outcome::LookupOutcome) counts = c.counts - i = 8 * mod(Threads.threadid() - 1, length(counts) ÷ 8) + kind + i = 8 * ((Threads.threadid() - 1) & (length(counts) ÷ 8 - 1)) + Int(outcome) GC.@preserve counts begin p = pointer(counts, i) unsafe_store!(p, unsafe_load(p, :monotonic) + 1, :monotonic) @@ -220,7 +221,7 @@ function evict!(c::ClockCache) n = length(c.ring) steps = 0 while true - c.hand = c.hand % n + 1 + c.hand = c.hand >= n ? 1 : c.hand + 1 steps += 1 e = c.ring[c.hand] e === nothing && continue @@ -273,7 +274,7 @@ end function cache_stats(c::ClockCache) counts = c.counts - total(kind) = GC.@preserve counts sum(i -> unsafe_load(pointer(counts, i), :monotonic), kind:8:length(counts)) + total(outcome) = GC.@preserve counts sum(i -> unsafe_load(pointer(counts, i), :monotonic), Int(outcome):8:length(counts)) return @lock c.lock (; hits = total(HIT), misses = total(MISS), length = c.count, c.currentsize, c.maxsize, c.by) end From cfe7869200c7a4ad2644b16be8af7116acf3be9c Mon Sep 17 00:00:00 2001 From: lkdvos Date: Thu, 15 Oct 2026 11:32:20 -0400 Subject: [PATCH 5/6] Shorten ClockCache and share snapshot iteration with SlotCache - DictCache, the supertype of SlotCache and ClockCache, holds their snapshot iteration and text/plain display; getindex uses the AbstractDict fallback. - The clock ring is a Vector{Entry{K, V}} with free positions unassigned. - nextslot replaces the repeated probe arithmetic; shorter comments. Co-Authored-By: Claude Opus 5.5 --- src/containers/clock.jl | 115 +++++++++++++++------------------------- src/containers/slots.jl | 39 ++++++-------- 2 files changed, 59 insertions(+), 95 deletions(-) diff --git a/src/containers/clock.jl b/src/containers/clock.jl index 135e46e..5269d6a 100644 --- a/src/containers/clock.jl +++ b/src/containers/clock.jl @@ -1,7 +1,6 @@ -# An entry of a `ClockCache`; only its reference bit changes after construction. The fields read -# on a hit come first. +# An entry of a `ClockCache`; only its reference bit changes after construction. mutable struct Entry{K, V} - const hash::UInt + const hash::UInt # the fields read on a hit come first const key::K const val::V @atomic used::Bool # the reference bit @@ -12,23 +11,16 @@ end struct Tombstone end const TOMBSTONE = Tombstone() -# The table is an open-addressing hash table with linear probing, at most half full, so that -# every probe ends at an empty slot. Readers take no lock, which is safe because of three rules: -# -# 1. Entries do not change after construction, apart from their reference bit, so a reader -# sees a key together with its own value. -# 2. A published table never changes length: it is replaced as a whole, by a release store to -# `c.table`, so the indices of a reader stay in bounds. -# 3. Under the lock, a slot only changes from `nothing` to an entry, from an entry to -# `TOMBSTONE`, or from `TOMBSTONE` to an entry; or to `nothing` when the next slot is -# `nothing`, so that no probe passes it anyway. Entries never move, so a probe for a key that -# is present and not concurrently removed always finds it. -# -# Slots are loaded with acquire and stored with release ordering, so a reader sees fully -# constructed entries. On Julia 1.10 and 1.11, which lack atomic indexing, a `Vector{Any}` is -# read through a pointer and written with `setindex!`, which compiles to a release store with a -# GC write barrier there (checked by the tests). Storing through a pointer would skip the -# barrier, and the GC could free the entry. +# The table of entries is an open-addressing hash table with linear probing, at most half full, +# read without the lock. This is safe because: +# 1. entries do not change after construction, apart from their reference bit; +# 2. a published table never changes length, but is replaced as a whole; +# 3. a slot only changes, under the lock, from `nothing` or `TOMBSTONE` to an entry, from an entry +# to `TOMBSTONE`, or to `nothing` if the next slot is `nothing`. Entries never move, so a probe +# finds every entry that is present. +# Slots are loaded with acquire and stored with release ordering. Before Julia 1.12, which added +# atomic indexing, `setindex!` into a `Vector{Any}` is a release store with a GC write barrier +# (checked by the tests); storing through a pointer would skip the barrier. @static if VERSION >= v"1.12" const SlotTable = AtomicMemory{Any} loadslot(t::SlotTable, i::Int) = @atomic :acquire t[i] @@ -55,10 +47,10 @@ share the cache. Insertions, evictions and the other changes take a lock. A look with a removal may still return the removed value, and one that races with an insertion may miss it; a returned value always belongs to its key. """ -mutable struct ClockCache{K, V} <: AbstractCache{K, V} - @atomic table::SlotTable # entries, read without the lock +mutable struct ClockCache{K, V} <: DictCache{K, V} + @atomic table::SlotTable occupied::Int # slots of `table` holding an entry or a tombstone - const ring::Vector{Union{Nothing, Entry{K, V}}} # the clock, `nothing` for a free position + const ring::Vector{Entry{K, V}} # the clock; free positions are unassigned const freepositions::Vector{Int} hand::Int # last inspected position of `ring` count::Int @@ -71,7 +63,7 @@ mutable struct ClockCache{K, V} <: AbstractCache{K, V} function ClockCache{K, V}(; maxsize::Integer = 10_000, by = nothing) where {K, V} maxsize >= 0 || throw(ArgumentError("maxsize must be non-negative")) counts = zeros(Int, 8 * nextpow(2, Threads.maxthreadid())) - return new{K, V}(emptytable(), 0, Union{Nothing, Entry{K, V}}[], Int[], 0, 0, 0, maxsize, by, counts, ReentrantLock()) + return new{K, V}(emptytable(), 0, Entry{K, V}[], Int[], 0, 0, 0, maxsize, by, counts, ReentrantLock()) end end @@ -81,35 +73,33 @@ function emptytable(n::Int = 16) return t end -# Slot indices run from 1 to `length(t)`, a power of two; the probe for `hash` starts at -# `(hash % Int) & mask + 1`, and moves on with `i & mask + 1`, wrapping around at the end. +# The slot after `i`, wrapping around (tables have a power-of-two length); a probe for `hash` +# starts at `nextslot(t, hash % Int)`. +nextslot(t::SlotTable, i::Int) = i & (length(t) - 1) + 1 -# The entry of the probe `p` (see `Key`), or `nothing`. +# The entry for the probe `p` (see `Key`), or `nothing`. findentry(::ClockCache, ::Nothing) = nothing @inline function findentry(c::ClockCache{K, V}, p::Key{P}) where {K, V, P} t = @atomic :acquire c.table - mask = length(t) - 1 - i = (p.hash % Int) & mask + 1 - for _ in 0:mask # bounds the probe also while slots change + i = nextslot(t, p.hash % Int) + for _ in 1:length(t) # bounds the probe also while slots change e = loadslot(t, i) e === nothing && return nothing e isa Entry{K, V} && e.hash == p.hash && e.key isa P && isequal(p.key, e.key::P) && return e - i = i & mask + 1 + i = nextslot(t, i) end return nothing end -# A hit sets the reference bit only if it is clear, so hot entries are not written to at all. +# Sets the reference bit only if it is clear, so hot entries are not written to. @inline function usevalue!(e::Entry) (@atomic :monotonic e.used) || (@atomic :monotonic e.used = true) return e.val end -# Hits and misses are counted per thread, a cache line (8 `Int`s) apart, since a shared counter -# would bounce between cores on every hit. There is a power of two of such lines, so a thread -# finds its own without a division. Two tasks on one thread cannot interleave between the load -# and the store, since neither yields; threads adopted after the cache was created share -# counters, and may then lose a few counts. +# Each thread counts in its own cache line (8 `Int`s), since a shared counter would bounce +# between cores on every hit; a power of two of lines avoids a division. Tasks on one thread +# cannot interleave between the load and the store, as neither yields. @enum LookupOutcome HIT = 1 MISS = 2 @inline function countlookup!(c::ClockCache, outcome::LookupOutcome) counts = c.counts @@ -144,11 +134,6 @@ function Base.get(c::ClockCache, key, default) e = findentry(c, _probe(c, key)) return e === nothing ? default : usevalue!(e) end -function Base.getindex(c::ClockCache, key) - e = findentry(c, _probe(c, key)) - e === nothing && throw(KeyError(key)) - return usevalue!(e) -end Base.haskey(c::ClockCache, key) = findentry(c, _probe(c, key)) !== nothing Base.length(c::ClockCache) = @lock c.lock c.count Base.isempty(c::ClockCache) = length(c) == 0 @@ -162,16 +147,16 @@ function addentry!(c::ClockCache{K, V}, p::Key, v, sz::Int) where {K, V} while c.currentsize + sz > c.maxsize && c.count > 0 evict!(c) end - if 2 * (c.occupied + 1) > length(c.table) - # publish a new table without tombstones, at most a quarter full + if 2 * (c.occupied + 1) > length(c.table) # publish a new table without tombstones t = emptytable(max(16, nextpow(2, 4 * (c.count + 1)))) - foreach(e -> e === nothing || placeentry!(t, e), c.ring) + for i in eachindex(c.ring) + isassigned(c.ring, i) && placeentry!(t, c.ring[i]) + end c.occupied = c.count @atomic :release c.table = t end - position = isempty(c.freepositions) ? length(push!(c.ring, nothing)) : pop!(c.freepositions) - e = Entry{K, V}(p.hash, p.key, v, false, sz, position) - c.ring[position] = e + position = isempty(c.freepositions) ? length(resize!(c.ring, length(c.ring) + 1)) : pop!(c.freepositions) + e = c.ring[position] = Entry{K, V}(p.hash, p.key, v, false, sz, position) placeentry!(c.table, e) === nothing && (c.occupied += 1) c.count += 1 c.currentsize += sz @@ -180,10 +165,9 @@ end # Store `e` in the first slot of its probe without an entry, and return what was there. function placeentry!(t::SlotTable, e::Entry) - mask = length(t) - 1 - i = (e.hash % Int) & mask + 1 + i = nextslot(t, e.hash % Int) while (old = loadslot(t, i)) isa Entry - i = i & mask + 1 + i = nextslot(t, i) end storeslot!(t, i, e) return old @@ -191,24 +175,22 @@ end function removeentry!(c::ClockCache, e::Entry) t = c.table - mask = length(t) - 1 - i = (e.hash % Int) & mask + 1 + i = nextslot(t, e.hash % Int) while loadslot(t, i) !== e - i = i & mask + 1 + i = nextslot(t, i) end - if loadslot(t, i & mask + 1) === nothing - # No probe continues past the next slot, so this slot and the tombstones before it can - # end probes as well: they become empty, which saves rebuilding the table. + if loadslot(t, nextslot(t, i)) === nothing + # no probe passes this slot or the tombstones before it, so they can all become empty while true storeslot!(t, i, nothing) c.occupied -= 1 - i = (i - 2) & mask + 1 + i = nextslot(t, i - 2) loadslot(t, i) === TOMBSTONE || break end else storeslot!(t, i, TOMBSTONE) end - c.ring[e.position] = nothing + Base._unsetindex!(c.ring, e.position) # internal, but without the fallback of `_unset!`: `isassigned` must be exact push!(c.freepositions, e.position) c.count -= 1 c.currentsize -= e.size @@ -223,8 +205,8 @@ function evict!(c::ClockCache) while true c.hand = c.hand >= n ? 1 : c.hand + 1 steps += 1 + isassigned(c.ring, c.hand) || continue e = c.ring[c.hand] - e === nothing && continue (steps > 2n || !(@atomic :monotonic e.used)) && return removeentry!(c, e) @atomic :monotonic e.used = false end @@ -278,15 +260,6 @@ function cache_stats(c::ClockCache) return @lock c.lock (; hits = total(HIT), misses = total(MISS), length = c.count, c.currentsize, c.maxsize, c.by) end -# Iteration walks a snapshot copied under the lock, in insertion order, like that of `SlotCache`. -Base.IteratorSize(::Type{<:ClockCache}) = Base.SizeUnknown() -function Base.iterate(c::ClockCache{K, V}) where {K, V} - snapshot = @lock c.lock Pair{K, V}[e.key => e.val for e in c.ring if e !== nothing] - return iterate(c, (snapshot, 1)) +function snapshot(c::ClockCache{K, V}) where {K, V} + return @lock c.lock Pair{K, V}[c.ring[i].key => c.ring[i].val for i in eachindex(c.ring) if isassigned(c.ring, i)] end -function Base.iterate(::ClockCache, (snapshot, i)::Tuple{Vector, Int}) - return i > length(snapshot) ? nothing : (snapshot[i], (snapshot, i + 1)) -end - -# The entries below the compact header. -Base.show(io::IO, m::MIME"text/plain", c::ClockCache) = invoke(show, Tuple{IO, MIME"text/plain", AbstractDict}, io, m, c) diff --git a/src/containers/slots.jl b/src/containers/slots.jl index bd94fa4..e2ecf4a 100644 --- a/src/containers/slots.jl +++ b/src/containers/slots.jl @@ -12,10 +12,21 @@ Base.hash(k::Key, h::UInt) = hash(k.hash, h) Base.isequal(p::Key{K}, s::Key{Any}) where {K} = p.hash == s.hash && s.key isa K && isequal(p.key, s.key::K) Base.isequal(a::Key{Any}, b::Key{Any}) = a.hash == b.hash && typeof(a.key) === typeof(b.key) && isequal(a.key, b.key) -# The shared implementation of `LRU` and `ClockCache`: every method of a dict-like +# Caches that iterate over `snapshot(c)`, the entries copied under the lock, so that iteration is +# safe while other tasks use the cache (holding the lock across `iterate` calls would deadlock on +# an early `break`). The length may change meanwhile, so `collect` must not rely on it. +abstract type DictCache{K, V} <: AbstractCache{K, V} end +Base.IteratorSize(::Type{<:DictCache}) = Base.SizeUnknown() +function Base.iterate(c::DictCache, (entries, i) = (snapshot(c), 1)) + return i > length(entries) ? nothing : (entries[i], (entries, i + 1)) +end +# The entries below the compact header. +Base.show(io::IO, m::MIME"text/plain", c::DictCache) = invoke(show, Tuple{IO, MIME"text/plain", AbstractDict}, io, m, c) + +# The shared implementation of `LRU` and `LocalClockCache`: every method of a dict-like # `AbstractCache`, on the storage in the field `slots::Slots{V}`. The subtypes are eviction # policies over slot numbers, through the hooks below, which are called with the lock held. -abstract type SlotCache{K, V} <: AbstractCache{K, V} end +abstract type SlotCache{K, V} <: DictCache{K, V} end # The entries in numbered slots (reused once freed), the key index, the size accounting, the # statistics and the lock. @@ -145,30 +156,13 @@ function Base.get(c::SlotCache, key, default) end end -function Base.getindex(c::SlotCache, key) - return @lock c.slots.lock begin - i = _slot(c, key) - i == 0 && throw(KeyError(key)) - touch!(c, i) - c.slots.vals[i] - end -end - Base.haskey(c::SlotCache, key) = @lock c.slots.lock _slot(c, key) != 0 Base.length(c::SlotCache) = @lock c.slots.lock length(c.slots.index) Base.isempty(c::SlotCache) = @lock c.slots.lock isempty(c.slots.index) -# Iteration walks a snapshot copied under the lock, so it is safe while other tasks use the -# cache (holding the lock across `iterate` calls would deadlock on an early `break`). -# The length may change between `length` and `iterate`, so `collect` must not rely on it. -Base.IteratorSize(::Type{<:SlotCache}) = Base.SizeUnknown() -function Base.iterate(c::SlotCache{K, V}) where {K, V} +function snapshot(c::SlotCache{K, V}) where {K, V} s = c.slots - snapshot = @lock s.lock Pair{K, V}[k.key::K => s.vals[i] for (k, i) in s.index] - return iterate(c, (snapshot, 1)) -end -function Base.iterate(::SlotCache, (snapshot, i)::Tuple{Vector, Int}) - return i > length(snapshot) ? nothing : (snapshot[i], (snapshot, i + 1)) + return @lock s.lock Pair{K, V}[k.key::K => s.vals[i] for (k, i) in s.index] end function Base.setindex!(c::SlotCache{K, V}, v, key) where {K, V} @@ -214,6 +208,3 @@ function cache_stats(c::SlotCache) s = c.slots return @lock s.lock (; s.hits, s.misses, length = length(s.index), s.currentsize, s.maxsize, s.by) end - -# The entries below the compact header. -Base.show(io::IO, m::MIME"text/plain", c::SlotCache) = invoke(show, Tuple{IO, MIME"text/plain", AbstractDict}, io, m, c) From c8e300cc72ea68e725ad5be4fd156b261a76a016 Mon Sep 17 00:00:00 2001 From: lkdvos Date: Thu, 15 Oct 2026 11:35:22 -0400 Subject: [PATCH 6/6] Call Base._unsetindex! directly, without a silent fallback If the internal function disappears, freeing a slot should fail loudly rather than keep the reference. Co-Authored-By: Claude Opus 5.5 --- src/containers/clock.jl | 2 +- src/containers/slots.jl | 14 +++----------- 2 files changed, 4 insertions(+), 12 deletions(-) diff --git a/src/containers/clock.jl b/src/containers/clock.jl index 5269d6a..c3f73f3 100644 --- a/src/containers/clock.jl +++ b/src/containers/clock.jl @@ -190,7 +190,7 @@ function removeentry!(c::ClockCache, e::Entry) else storeslot!(t, i, TOMBSTONE) end - Base._unsetindex!(c.ring, e.position) # internal, but without the fallback of `_unset!`: `isassigned` must be exact + Base._unsetindex!(c.ring, e.position) # internal push!(c.freepositions, e.position) c.count -= 1 c.currentsize -= e.size diff --git a/src/containers/slots.jl b/src/containers/slots.jl index e2ecf4a..ce2467f 100644 --- a/src/containers/slots.jl +++ b/src/containers/slots.jl @@ -106,26 +106,18 @@ function _insert!(c::SlotCache, p::Key, v, sz::Int) end # Free slot `i`: drop it from the index and clear its key and value so they can be -# garbage-collected while the slot sits on the free list. +# garbage-collected while the slot sits on the free list (`Base._unsetindex!` is internal). function _free!(c::SlotCache, i::Int) forget!(c, i) s = c.slots delete!(s.index, s.keys[i]) - _unset!(s.keys, i) - _unset!(s.vals, i) + Base._unsetindex!(s.keys, i) + Base._unsetindex!(s.vals, i) s.currentsize -= s.sizes[i] push!(s.free, i) return c end -# `Base._unsetindex!` is internal, so fall back to keeping the reference if it ever disappears. -function _unset!(v::Vector, i::Int) - @static if isdefined(Base, :_unsetindex!) - Base._unsetindex!(v, i) - end - return v -end - function Base.get!(default::Base.Callable, c::SlotCache{K, V}, key) where {K, V} p = _newkey(c, key) s = c.slots