rebuild cache on MapRef, fix cancellation defects - #369
Conversation
stasimus
changed the title
are the following the above-mentioned defects:
|
|
Yes, exactly those four, declared and asserted in |
|
I am at the beginning of review and I understood that I fail to comprehend what happens in new functionality. ScalaDocs are really required! With explicitly explaining |
|
Added scaladocs on LoadingCache, EntryMap, EntryState, the cache operations and the ExpiringCache eviction routine, describing the algorithm and the four defects this change fixes. P1, contention on shared state. The state used to be one Ref[F, Map[K, EntryRef]], so every insert or removal of any key CAS-ed the same Ref, and a getOrUpdate of one key lost its CAS whenever an unrelated key was written. Sustained writes elsewhere starved it, which is what MaxRetries = 10000 and IllegalStateException("extreme contention") were guarding, i.e. contention turned into a user visible failure. Every write also copied the whole map. Now each key has its own Ref over a ConcurrentHashMap, unrelated keys never interfere and the retry limit is gone. P2, cancellation poisoned the key. The value computation ran unmasked inside the retry loop with no onCancel, so cancelling getOrUpdate left Loading(deferred) in the map with the deferred never completed: the key stayed unusable and every waiter blocked forever, and a value computed just as cancellation hit was leaked. Now state transitions are masked, and cancellation unlinks the key, completes the deferred with CancelledError and releases the value if one was produced. P3, expiration ignored Loading. removeExpiredAndCheckSize only inspected Value states, so an entry whose load never completes was never evicted. Now loads that run longer than the expiration interval are evicted and their waiters get ExpiredError. P4, a stuck load blocked finalization, since clear runs on resource release and waits on Loading entries. P2 and P3 remove both ways of getting stuck there. CacheDefectsSpec covers all four. |
where is the code for benchmarks? |
|
@stasimus, could we get the PR with failing (and ignored or explicitly waiting for failures with corresponding comments and verbose printouts) unit-tests against "current" (as in I tried to do the quick rollback of implementation in the branch, but new unit-tests do not compile against old implementation. It also means that this PR might change public API in incompatible way - we must provide the instructions on how to migrate from "old" to "new" APIs! |
|
Benchmarks are in a new benchmark module now, JMH based, at benchmark/src/main/scala/com/evolution/scache/bench/CacheBenchmark.scala. It carries a frozen copy of the pre-MapRef implementation under com.evolution.scache.v1, so old and new are measured in the same run: impl=v1 against impl=v2, across flavor=single, partitioned and expiring, over get, get1, getOrUpdate, put, modify, remove, contains and foldMap. Run it with: sbt "benchmark/Jmh/run", or narrow it down, e.g. sbt "benchmark/Jmh/run -p impl=v1,v2 -p flavor=single .getOrUpdateHitRandomKeys.". I will refresh the numbers in the description once the full suite has run on a quiet machine. |
|
Ran the JMH suite twice back to back on the same machine, once on 7c9fa9f and once on this branch, and put the before and after table in the README under Benchmarks, Results. Raw JMH output of both runs is committed in benchmark/results. Biggest gains are where the old code had to CAS the shared map: put of distinct keys 1.66 to 9.24 M ops/s on the unpartitioned cache, modify of distinct keys 1.88 to 11.44, remove and put 0.84 to 3.87. With partitioning the same operations gain 1.6x to 1.9x, reads gain around 1.2x. foldMap is 2 to 3 percent slower, which is the price of walking a ConcurrentHashMap instead of an atomic snapshot. |
📝 WalkthroughWalkthroughThe cache now uses per-key entry maps and ChangesCache storage and public API
Regression validation
Benchmark suite
Estimated code review effort: 5 (Critical) | ~90 minutes Sequence Diagram(s)sequenceDiagram
participant Loader
participant ExpiringCache
participant EntryMap
participant WaitingFiber
ExpiringCache->>EntryMap: track loading entry
ExpiringCache->>EntryMap: evict after loadingTimeout
EntryMap-->>WaitingFiber: complete with ExpiredError
EntryMap-->>Loader: complete with ExpiredError
Suggested reviewers: 🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 passed)
✨ Finishing Touches🧪 Generate unit tests (beta)
Comment |
stasimus
changed the title
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 4
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
scache/src/test/scala/com/evolution/scache/CacheSpec.scala (1)
1075-1090: 🩺 Stability & Availability | 🟡 Minor | ⚡ Quick winThe
cancellation propertest can block on the loadingDeferred.
fiber.cancel.startruns the cancellation concurrently. The followingcache.get(0)is not synchronized against it. Ifgetobserves the entry still inLoadingstate, it awaits the entry'sDeferred. ThatDeferredis completed by the cancellation cleanup withCancelledError, so the test recovers, but only after the cleanup runs.If the cancellation cleanup has not yet marked the entry,
getreturns the loaded value instead ofnone, andresult shouldEqual nonefails.Join the cancellation before asserting, or poll
getuntil it returnsnone.💚 Proposed fix
fiber <- cache.getOrUpdateEnsure(0)(deferred.get) - fiber <- fiber.cancel.start - result <- cache.get(0) - _ <- IO { result shouldEqual none } - _ <- deferred.complete(0) - _ <- fiber.joinWithNever + cancelling <- fiber.cancel.start + _ <- cancelling.joinWithNever + result <- cache.get(0) + _ <- IO { result shouldEqual none } + _ <- deferred.complete(0)🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@scache/src/test/scala/com/evolution/scache/CacheSpec.scala` around lines 1075 - 1090, Update the `cancellation proper` test to synchronize cancellation before asserting the cache state: join the cancellation fiber created by `fiber.cancel.start` before calling `cache.get(0)` and checking `result shouldEqual none`. Preserve the existing deferred completion and metrics assertions.
🧹 Nitpick comments (4)
README.md (1)
148-157: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueUse the
shellfence language for the sbt commands.The block contains shell commands, not Scala code. The comparison block at Line 204 already uses
shell. Align the two blocks.📝 Proposed fix
-```scala +```shell // everything, around 10 minutes sbt "benchmark/Jmh/run"🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@README.md` around lines 148 - 157, Change the fenced code language for the sbt command block in the README from scala to shell, matching the existing shell fence used by the comparison block while leaving the commands unchanged.scache/src/main/scala/com/evolution/scache/ExpiringCache.scala (2)
442-447: 🚀 Performance & Scalability | 🔵 Trivial | ⚡ Quick win
foldMapPardelegates tocache.foldMap.The method name promises parallel folding, but the body calls
cache.foldMap. The underlyingLoadingCache.foldMapPardoes run in parallel. This predates the change, and only the signature line changed here.Change the delegation to
cache.foldMapParif the parallel behavior is intended.♻️ Proposed fix
def foldMapPar[A: CommutativeMonoid](f: (K, Either[F[V], V]) => F[A]): F[A] = { - cache.foldMap { + cache.foldMapPar { case (k, Right(v)) => f(k, v.value.asRight) case (k, Left(v)) => f(k, v.map { _.value }.asLeft) } }Note: this needs
Parallel[F]onapply, which currently only requiresMonadThrow: Clock.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@scache/src/main/scala/com/evolution/scache/ExpiringCache.scala` around lines 442 - 447, Update ExpiringCache.foldMapPar to delegate to cache.foldMapPar so it preserves the promised parallel behavior, and add the required Parallel[F] constraint to the surrounding apply or relevant construction context alongside the existing MonadThrow and Clock requirements.
41-51: 🚀 Performance & Scalability | 🔵 TrivialConsider the scan cost when
loadingTimeoutis much shorter than the expiration.
expireIntervalis bounded byloadingTimeoutMs / 2. One run walks every entry of the cache. A shortloadingTimeoutnext to a large cache therefore schedules a full scan very often, for example every 50 ms for a 100 ms timeout.The comment describes the trade-off, but nothing bounds the resulting work. Consider documenting the cost in
Config.loadingTimeout, or tracking loading entries separately so that the loading sweep does not need a full traversal.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@scache/src/main/scala/com/evolution/scache/ExpiringCache.scala` around lines 41 - 51, Address the full-cache scan cost in the expireInterval logic around Config.loadingTimeout: avoid scheduling whole-cache traversals solely from a short loadingTimeout, preferably by tracking loading entries separately so loading cleanup does not require scanning every cache entry; otherwise document the scan-cost trade-off in Config.loadingTimeout and preserve expiration cleanup behavior.scache/src/main/scala/com/evolution/scache/Cache.scala (1)
511-511: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winUpdate the context-bound documentation for the new
Asyncrequirement.The signature now requires
Async[F]. The scaladoc above still explains the bounds in terms ofSyncandConcurrent, and theexpiringscaladoc still namesTemporal. Readers of the API docs will see the old requirements.Add a line describing why
Asyncis needed, matching the README migration note.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@scache/src/main/scala/com/evolution/scache/Cache.scala` at line 511, Update the Scaladoc for loading to document the Async[F] context bound and its purpose, replacing the outdated Sync/Concurrent explanation and matching the README migration wording. Also update the expiring Scaladoc to reflect its current Async requirement instead of naming Temporal.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@benchmark/src/main/scala/com/evolution/scache/bench/CacheBenchmark.scala`:
- Around line 207-213: Add method-level `@OperationsPerInvocation`(320000) to
removeAndPutRandomKeys, overriding the class-level operation count, and
regenerate the recorded benchmark results.
In `@scache/src/main/scala/com/evolution/scache/ExpiringCache.scala`:
- Around line 104-129: Update evictLoading so the loading deferred is completed
with ExpiredError before the entry is marked Removed and unlinked, preventing a
concurrent load from publishing a value after eviction. Preserve the existing
deferred-identity check and no-op behavior when the entry is no longer the
matching load.
In `@scache/src/main/scala/com/evolution/scache/LoadingCache.scala`:
- Around line 676-701: Update the LoadingCache put path at
LoadingCache.scala:676-701 and the corresponding modify path at
LoadingCache.scala:825-850 so deferred publication and the EntryState transition
use distinct, balanced ownership. Retain or transfer ownership when complete
publishes entry, release the producer’s ownership exactly once when set/setRef
fails after publication, preserve the waiter-owned reference, and release the
unpublished entry when complete fails before retrying or exiting.
In `@scache/src/test/scala/com/evolution/scache/CacheDefectsSpec.scala`:
- Around line 211-240: Increase the loadingTimeout in the “a new load generation
does not inherit the previous generation's stuck-timer” test and adjust the
second sleep in the result block to remain well below the new timeout,
preserving the presence assertion while providing sufficient scheduler-timing
margin.
---
Outside diff comments:
In `@scache/src/test/scala/com/evolution/scache/CacheSpec.scala`:
- Around line 1075-1090: Update the `cancellation proper` test to synchronize
cancellation before asserting the cache state: join the cancellation fiber
created by `fiber.cancel.start` before calling `cache.get(0)` and checking
`result shouldEqual none`. Preserve the existing deferred completion and metrics
assertions.
---
Nitpick comments:
In `@README.md`:
- Around line 148-157: Change the fenced code language for the sbt command block
in the README from scala to shell, matching the existing shell fence used by the
comparison block while leaving the commands unchanged.
In `@scache/src/main/scala/com/evolution/scache/Cache.scala`:
- Line 511: Update the Scaladoc for loading to document the Async[F] context
bound and its purpose, replacing the outdated Sync/Concurrent explanation and
matching the README migration wording. Also update the expiring Scaladoc to
reflect its current Async requirement instead of naming Temporal.
In `@scache/src/main/scala/com/evolution/scache/ExpiringCache.scala`:
- Around line 442-447: Update ExpiringCache.foldMapPar to delegate to
cache.foldMapPar so it preserves the promised parallel behavior, and add the
required Parallel[F] constraint to the surrounding apply or relevant
construction context alongside the existing MonadThrow and Clock requirements.
- Around line 41-51: Address the full-cache scan cost in the expireInterval
logic around Config.loadingTimeout: avoid scheduling whole-cache traversals
solely from a short loadingTimeout, preferably by tracking loading entries
separately so loading cleanup does not require scanning every cache entry;
otherwise document the scan-cost trade-off in Config.loadingTimeout and preserve
expiration cleanup behavior.
🪄 Autofix
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro Plus
Run ID: c0b2a606-d294-4b09-acd4-cc3e8476ca8b
📒 Files selected for processing (15)
README.mdbenchmark/results/mapref.jsonbenchmark/results/master-7c9fa9f.jsonbenchmark/src/main/scala/com/evolution/scache/bench/CacheBenchmark.scalabuild.sbtproject/plugins.sbtscache/src/main/scala/com/evolution/scache/Cache.scalascache/src/main/scala/com/evolution/scache/CancelledError.scalascache/src/main/scala/com/evolution/scache/ExpiredError.scalascache/src/main/scala/com/evolution/scache/ExpiringCache.scalascache/src/main/scala/com/evolution/scache/LoadingCache.scalascache/src/main/scala/com/evolution/scache/SerialMap.scalascache/src/test/scala/com/evolution/scache/CacheDefectsSpec.scalascache/src/test/scala/com/evolution/scache/CacheSpec.scalascache/src/test/scala/com/evolution/scache/SerialMapSpec.scala
| @Benchmark | ||
| def removeAndPutRandomKeys(state: PopulatedCacheState): Unit = { | ||
| parRun { (fiber, i) => | ||
| val k = key(fiber, i) | ||
| state.cache.remove(k).flatten *> state.cache.put(k, i).flatten.void | ||
| }.unsafeRunSync() | ||
| } |
There was a problem hiding this comment.
🎯 Functional Correctness | 🟡 Minor | ⚡ Quick win
🧩 Analysis chain
🌐 Web query:
For JMH 1.37, does a method-level @OperationsPerInvocation override a class-level annotation for one benchmark method?
💡 Result:
Yes, in JMH 1.37 (and generally across all versions), a method-level @OperationsPerInvocation annotation overrides any @OperationsPerInvocation annotation defined at the class level [1]. JMH annotation processing follows a hierarchy where the most specific annotation—the one closest to the benchmark method—takes precedence [2]. Specifically: 1. Class-level annotations set a default configuration for all @Benchmark methods within that class [1]. 2. Method-level annotations are treated as more specific and will override the class-level settings for that specific benchmark [1]. This behavior is consistent with how JMH handles other metadata annotations, where the "closest" annotation in the hierarchy wins [2]. When you define @OperationsPerInvocation on a specific @Benchmark method, JMH uses that value to calculate the individual operation cost for that method, ignoring any value provided at the class level for that same method [1].
Citations:
- 1: https://self-learning-java-tutorial.blogspot.com/2022/08/jmh-get-individual-operations-cost-with.html
- 2: https://hg.openjdk.org/code-tools/jmh/file/cb9aa824b55a/jmh-samples/src/main/java/org/openjdk/jmh/samples/JMHSample_24_Inheritance.java
🏁 Script executed:
#!/bin/bash
set -eu
file="benchmark/src/main/scala/com/evolution/scache/bench/CacheBenchmark.scala"
printf '%s\n' '--- annotation and benchmark definitions ---'
rg -n -C 5 '`@OperationsPerInvocation`|removeAndPutRandomKeys|Fibers|OpsPerFiber|parRun' "$file"
printf '%s\n' '--- imports ---'
sed -n '1,80p' "$file"
printf '%s\n' '--- relevant body ---'
sed -n '190,220p' "$file"Repository: evolution-gaming/scache
Length of output: 9430
🏁 Script executed:
#!/bin/bash
set -eu
printf '%s\n' '--- JMH version and benchmark result artifacts ---'
rg -n -C 3 'jmh|Jmh|OperationsPerInvocation|removeAndPutRandomKeys' \
build.sbt project benchmark README.md .github 2>/dev/null || true
printf '%s\n' '--- matching tracked files ---'
git ls-files | rg -i 'benchmark|jmh|result|csv|json|txt' | head -200
printf '%s\n' '--- annotation usage ---'
rg -n -C 2 '`@OperationsPerInvocation`' .Repository: evolution-gaming/scache
Length of output: 43424
🌐 Web query:
JMH 1.37 OperationsPerInvocation annotation method level class level precedence official source BenchmarkGenerator
💡 Result:
JMH’s @OperationsPerInvocation has this level precedence: 1) Method-level @OperationsPerInvocation on a specific @Benchmark method overrides any class-level @OperationsPerInvocation. 2) If a benchmark method has no method-level @OperationsPerInvocation, the enclosing class’ @OperationsPerInvocation (if present) is used. 3) If neither is present, the default is 1 operation per single @Benchmark invocation. This is how JMH’s BenchmarkGenerator wires the chosen operations-per-invocation value into generated benchmark entries (it uses the resolved operationsPerInvocation from the method group during generation). [1] Operational definition: “value” = number of operations per single Benchmark call; default 1. [2][3] Notes: - @OperationsPerInvocation can be put at Benchmark method level or at the enclosing class level to affect one method or all methods in the class. [4] Sources: [2][4][3][1]
Citations:
- 1: https://github.com/openjdk/jmh/blob/master/jmh-core/src/main/java/org/openjdk/jmh/generators/core/BenchmarkGenerator.java
- 2: https://javadoc.io/doc/org.openjdk.jmh/jmh-core/1.21/org/openjdk/jmh/annotations/OperationsPerInvocation.html
- 3: https://stackoverflow.com/questions/32527223/what-exactly-is-number-of-operations-in-jmh
- 4: https://self-learning-java-tutorial.blogspot.com/2022/08/jmh-get-individual-operations-cost-with.html
🏁 Script executed:
#!/bin/bash
set -eu
python3 - <<'PY'
from pathlib import Path
import re
p = Path("benchmark/src/main/scala/com/evolution/scache/bench/CacheBenchmark.scala")
s = p.read_text()
fibers = int(re.search(r"final val Fibers = (\d+)", s).group(1))
ops_per_fiber = int(re.search(r"final val OpsPerFiber = (\d+)", s).group(1))
method = re.search(
r"def removeAndPutRandomKeys\b.*?(?=\n `@Benchmark`|\Z)", s, re.S
).group(0)
remove_calls = len(re.findall(r"\.cache\.remove\s*\(", method))
put_calls = len(re.findall(r"\.cache\.put\s*\(", method))
print({
"Fibers": fibers,
"OpsPerFiber": ops_per_fiber,
"invocations_per_benchmark_call": fibers * ops_per_fiber,
"remove_calls_per_invocation": fibers * ops_per_fiber * remove_calls,
"put_calls_per_invocation": fibers * ops_per_fiber * put_calls,
"cache_calls_per_invocation": fibers * ops_per_fiber * (remove_calls + put_calls),
})
PYRepository: evolution-gaming/scache
Length of output: 352
Correct the operation count for removeAndPutRandomKeys.
This benchmark performs 160,000 remove calls and 160,000 put calls per invocation. Add method-level @OperationsPerInvocation(320000) to override the class-level value, then regenerate the recorded results.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@benchmark/src/main/scala/com/evolution/scache/bench/CacheBenchmark.scala`
around lines 207 - 213, Add method-level `@OperationsPerInvocation`(320000) to
removeAndPutRandomKeys, overriding the class-level operation count, and
regenerate the recorded benchmark results.
| def evictLoading( | ||
| key: K, | ||
| entryRef: LoadingCache.EntryRef[F, TimestampedValue], | ||
| deferred: LoadingDeferred, | ||
| ): F[Unit] = { | ||
| entryRef | ||
| .modify { | ||
| case state: EntryState.Loading[F, TimestampedValue] if state.deferred == deferred => | ||
| (EntryState.Removed, true) | ||
| case state => | ||
| (state, false) | ||
| } | ||
| .flatMap { | ||
| case true => | ||
| entryMap | ||
| .ref(key) | ||
| .update { | ||
| case Some(`entryRef`) => none | ||
| case other => other | ||
| } | ||
| .productR { deferred.complete(ExpiredError.asLeft).void } | ||
| case false => | ||
| ().pure[F] | ||
| } | ||
| .uncancelable | ||
| } |
There was a problem hiding this comment.
🎯 Functional Correctness | 🟡 Minor | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
# Description: Trace what a load does when it observes EntryState.Removed after eviction.
set -euo pipefail
fd -t f 'LoadingCache.scala' --exec rg -n -C 10 'tryPutNewValue|EntryState.Removed' {}
fd -t f 'ExpiringCacheSpec.scala' --exec rg -n -C 6 'ExpiredError|loadingTimeout' {}Repository: evolution-gaming/scache
Length of output: 14714
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo '--- candidate files ---'
fd -t f '(LoadingCache|ExpiringCache|README|Spec)' .
echo '--- LoadingCache load and remove paths ---'
file=$(fd -t f 'LoadingCache.scala' | head -n 1)
sed -n '280,480p' "$file"
sed -n '640,715p' "$file"
sed -n '930,985p' "$file"
echo '--- ExpiringCache and documentation ---'
expiring=$(fd -t f 'ExpiringCache.scala' | head -n 1)
sed -n '45,145p' "$expiring"
rg -n -C 4 'detached|ExpiredError|expire|loading' README.md "$expiring" $(fd -t f 'ExpiringCacheSpec.scala' || true) || true
echo '--- tests covering removal during loading ---'
fd -t f -e scala . | xargs rg -n -C 5 'remove.*load|load.*remove|ExpiredError|loadingTimeout|detached' || trueRepository: evolution-gaming/scache
Length of output: 50381
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo '--- public remove contract ---'
rg -n -C 8 'def remove|Removes loading|load.*outlive|puts the key back' \
scache/src/main/scala/com/evolution/scache/LoadingCache.scala \
scache/src/main/scala/com/evolution/scache/Cache.scala \
README.md
echo '--- timeout-related tests ---'
rg -n -C 12 'loadingTimeout|ExpiredError|remove.*loading|loading.*remove|detached|outlive' \
scache/src/test/scala/com/evolution/scache/ExpiringCacheSpec.scala \
scache/src/test/scala/com/evolution/scache/CacheSpec.scala \
scache/src/test/scala/com/evolution/scache/CacheDefectsSpec.scala
echo '--- concise source assertions ---'
python3 - <<'PY'
from pathlib import Path
loading = Path('scache/src/main/scala/com/evolution/scache/LoadingCache.scala').read_text()
expiring = Path('scache/src/main/scala/com/evolution/scache/ExpiringCache.scala').read_text()
readme = Path('README.md').read_text()
checks = {
'load retries insertion after Removed': 'case (EntryState.Removed, _) =>' in loading and 'case EntryState.Removed =>\\n tryPutNewValue' in loading,
'remove documents reinsertion': 'stores its value\\n * under it' in loading,
'ExpiringCache documents detached load': 'only detached from the cache' in expiring,
'README documents detached load': 'only detached from the cache' in readme,
'eviction completes ExpiredError': 'deferred.complete(ExpiredError.asLeft).void' in expiring,
}
for name, result in checks.items():
print(f'{name}: {result}')
PY
echo '--- minimal transition model from the documented branches ---'
python3 - <<'PY'
class EntryRef:
def __init__(self, state):
self.state = state
def load_after_eviction(entry_ref, map_ref, key, value):
if entry_ref.state == 'Removed':
new_entry_ref = EntryRef(('Value', value))
if map_ref.get(key) is None:
map_ref[key] = new_entry_ref
return 'inserted'
return 'existing-entry-wins'
return 'different-state'
entry_ref = EntryRef('Removed')
entry_map = {}
print(load_after_eviction(entry_ref, entry_map, 'k', 'v'))
print(entry_map['k'].state)
PYRepository: evolution-gaming/scache
Length of output: 22832
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo '--- complete load race and publication branch ---'
sed -n '320,485p' scache/src/main/scala/com/evolution/scache/LoadingCache.scala
echo '--- eviction test through post-eviction completion ---'
sed -n '181,210p' scache/src/test/scala/com/evolution/scache/CacheDefectsSpec.scala
echo '--- dependency and race API references ---'
rg -n -C 3 'cats.effect|race1|racePair' build.sbt project scache/build.sbt scache/src/main scache/src/test 2>/dev/null || true
echo '--- deterministic interleaving model ---'
python3 - <<'PY'
# Model only the state transitions visible in the source:
# eviction marks Removed, unlinks the key, and completes the Deferred;
# load either completes the Deferred with its value first or observes ExpiredError first.
def outcome(load_wins):
entry_map = {'k': 'loading'}
state = 'Loading'
deferred = None
state = 'Removed'
entry_map.pop('k', None)
if load_wins:
deferred = 'value'
if state == 'Removed' and 'k' not in entry_map:
entry_map['k'] = 'value'
else:
deferred = 'ExpiredError'
# race1 returns the Deferred result and does not enter tryPutNewValue
return deferred, entry_map.get('k')
for winner in (True, False):
print(f'{"load" if winner else "eviction"} wins Deferred: {outcome(winner)}')
PYRepository: evolution-gaming/scache
Length of output: 37286
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo '--- race1 cancellation and result precedence ---'
sed -n '1258,1315p' scache/src/main/scala/com/evolution/scache/LoadingCache.scala
echo '--- exact eviction ordering ---'
sed -n '99,129p' scache/src/main/scala/com/evolution/scache/ExpiringCache.scala
echo '--- exact post-eviction publication comments ---'
sed -n '40,66p' scache/src/main/scala/com/evolution/scache/LoadingCache.scala
sed -n '533,540p' scache/src/main/scala/com/evolution/scache/ExpiringCache.scalaRepository: evolution-gaming/scache
Length of output: 5659
Prevent an evicted load from winning the publication race
evictLoading marks the entry Removed and unlinks the key before completing the deferred with ExpiredError. The load can complete the deferred with its value in this gap, observe Removed, run tryPutNewValue, and repopulate the key. Complete the deferred before publishing Removed, or add a tombstone that prevents publication after eviction.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@scache/src/main/scala/com/evolution/scache/ExpiringCache.scala` around lines
104 - 129, Update evictLoading so the loading deferred is completed with
ExpiredError before the entry is marked Removed and unlinked, preventing a
concurrent load from publishing a value after eviction. Preserve the existing
deferred-identity check and no-op behavior when the entry is no longer the
matching load.
| case (state: EntryState.Loading[F, V], set) => | ||
| state | ||
| .deferred | ||
| .complete(entry.asRight) | ||
| .flatMap { | ||
| // We successfully completed the deferred, now trying to set the value. | ||
| case true => | ||
| set(EntryState.Value(entry)).flatMap { | ||
| // We successfully replaced the entry with our value, so we are done. | ||
| case true => | ||
| set(EntryState.Value(entry)).flatMap { | ||
| // We successfully replaced the entry with our value, so we are done. | ||
| case true => | ||
| none[V] | ||
| .pure[F] | ||
| .asRight[Int] | ||
| .pure[F] | ||
| // Another fiber placed their new value before us | ||
| // so we just release our value and exit. | ||
| case false => | ||
| entry | ||
| .release | ||
| .traverse { _.start } // Start releasing and forget | ||
| .as { | ||
| none[V] | ||
| .pure[F] | ||
| .asRight[Int] | ||
| } | ||
| } | ||
| // Someone just completed the deferred we saw | ||
| none[V] | ||
| .pure[F] | ||
| .asRight[Unit] | ||
| .pure[F] | ||
|
|
||
| // Another fiber placed their new value before us | ||
| // so we just release our value and exit. | ||
| case false => | ||
| entry | ||
| .release | ||
| .traverse { _.start } // Start releasing and forget | ||
| .as { | ||
| none[V] | ||
| .pure[F] | ||
| .asRight[Int] | ||
| } | ||
| releaseAndExit | ||
| } | ||
|
|
||
| // The key was just removed from the map, so just release the value and exit. | ||
| case (EntryState.Removed, _) => | ||
| entry | ||
| .release | ||
| .traverse { _.start } // Start releasing and forget | ||
| .as { | ||
| none[V] | ||
| .pure[F] | ||
| .asRight[Int] | ||
| } | ||
| } | ||
| .uncancelable | ||
| // Someone just completed the deferred we saw | ||
| // so we just release our value and exit. | ||
| case false => | ||
| releaseAndExit | ||
| } |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟠 Major | 🏗️ Heavy lift
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
# Description: Inspect every site that completes a Loading deferred and then commits the entry state.
set -euo pipefail
fd -t f 'LoadingCache.scala' --exec rg -n -C 15 '\.complete\(entry\.asRight\)' {}Repository: evolution-gaming/scache
Length of output: 5206
🏁 Script executed:
#!/bin/bash
set -euo pipefail
printf '%s\n' '--- LoadingCache.scala structure ---'
ast-grep outline scache/src/main/scala/com/evolution/scache/LoadingCache.scala
printf '%s\n' '--- put implementation around both completion sites ---'
sed -n '300,390p' scache/src/main/scala/com/evolution/scache/LoadingCache.scala
sed -n '620,715p' scache/src/main/scala/com/evolution/scache/LoadingCache.scala
printf '%s\n' '--- modify implementation around the completion site ---'
sed -n '770,870p' scache/src/main/scala/com/evolution/scache/LoadingCache.scala
printf '%s\n' '--- EntryState, entry construction, and release helpers ---'
rg -n -C 8 'sealed|EntryState|def entryOf|releaseAndExit|setRef|set\(' scache/src/main/scala/com/evolution/scache/LoadingCache.scalaRepository: evolution-gaming/scache
Length of output: 43114
🏁 Script executed:
#!/bin/bash
set -euo pipefail
printf '%s\n' '--- Relevant source ranges ---'
sed -n '300,390p' scache/src/main/scala/com/evolution/scache/LoadingCache.scala
sed -n '620,715p' scache/src/main/scala/com/evolution/scache/LoadingCache.scala
sed -n '770,870p' scache/src/main/scala/com/evolution/scache/LoadingCache.scala
printf '%s\n' '--- Definitions and all related call sites ---'
rg -n -C 10 'EntryState|entryOf|releaseAndExit|setRef|complete\(entry\.asRight\)' scache/src/main/scala/com/evolution/scache/LoadingCache.scalaRepository: evolution-gaming/scache
Length of output: 46332
🏁 Script executed:
#!/bin/bash
set -euo pipefail
python3 - <<'PY'
from pathlib import Path
p = Path("scache/src/main/scala/com/evolution/scache/LoadingCache.scala")
lines = p.read_text().splitlines()
for lo, hi in [(250, 390), (560, 715), (770, 870)]:
print(f"--- {p}:{lo}-{hi} ---")
for n in range(lo, min(hi, len(lines)) + 1):
print(f"{n:4}: {lines[n-1]}")
PY
printf '%s\n' '--- EntryState and release definitions in the repository ---'
rg -n -C 12 'sealed (trait|abstract class) EntryState|object EntryState|case class (Loading|Value|Removed)|def release1|def releaseAndReturnValue|def entryOf' scache/src/main/scala scache/src/testRepository: evolution-gaming/scache
Length of output: 34522
🌐 Web query:
evolution-gaming/scache LoadingCache.scala EntryState deferred complete entry.asRight
💡 Result:
The code fragment entry.deferred.complete(value.asRight) in evolution-gaming/scache is part of the library's internal mechanism for managing asynchronous value loading [1][2]. In the context of the scache library—a Scala library built on top of cats-effect for concurrent caching—the LoadingCache component handles cache misses by triggering an asynchronous load operation [3]. To prevent redundant concurrent loads for the same key, the library uses a coordination pattern involving a Deferred (from cats-effect) [1]. 1. Deferred State: When a value is requested and not yet present, the cache creates an Entry.Loading state containing a Deferred [1]. This Deferred acts as a synchronization point for any subsequent requests for the same key. 2. Completing the Deferred: When the actual loading process succeeds, the result (a loaded entry) is wrapped as a Right (via.asRight) and used to complete the Deferred [1][4]. This unblocks any concurrent requests waiting on the same key, allowing them to receive the loaded value. 3. Handling Errors: Conversely, if the loading process fails, the error is typically lifted into the Left side of an Either (via.asLeft) and used to complete the Deferred [1], signaling to waiting callers that the load operation failed. This pattern ensures that only one load operation executes for a given key, while others efficiently await the single result [3]. The.asRight syntax is standard cats/cats-effect usage to create an Either type, specifically signaling the success case of the asynchronous operation.
Citations:
🏁 Script executed:
#!/bin/bash
set -euo pipefail
python3 - <<'PY'
import urllib.request
url = "https://raw.githubusercontent.com/evolution-gaming/scache/master/scache/src/main/scala/com/evolution/scache/LoadingCache.scala"
try:
data = urllib.request.urlopen(url, timeout=10).read().decode()
except Exception as e:
print(type(e).__name__, e)
else:
lines = data.splitlines()
for lo, hi in [(300, 390), (620, 715), (770, 870)]:
print(f"--- remote source {lo}-{hi} ---")
for n in range(lo, min(hi, len(lines)) + 1):
print(f"{n:4}: {lines[n-1]}")
PYRepository: evolution-gaming/scache
Length of output: 305
🏁 Script executed:
#!/bin/bash
set -euo pipefail
python3 - <<'PY'
from dataclasses import dataclass
`@dataclass`
class Entry:
released: bool = False
class Model:
def __init__(self):
self.state = "Loading"
self.deferred = None
self.waiter_entry = None
def complete(self, entry):
if self.deferred is not None:
return False
self.deferred = entry
self.waiter_entry = entry
return True
def remove(self):
self.state = "Removed"
def set_value(self, entry):
if self.state != "Loading":
return False
self.state = "Value"
return True
def release(self, entry):
entry.released = True
# put: publication succeeds, removal wins before the state commit.
m = Model()
put_entry = Entry()
assert m.complete(put_entry) is True
m.remove()
assert m.set_value(put_entry) is False
m.release(put_entry)
assert m.waiter_entry is put_entry and m.waiter_entry.released
print("put: waiter receives the published entry after its release starts")
# modify: publication succeeds, removal wins before the state commit,
# and the retry path does not release the published entry.
m = Model()
modify_entry = Entry()
assert m.complete(modify_entry) is True
m.remove()
assert m.set_value(modify_entry) is False
assert m.waiter_entry is modify_entry and not m.waiter_entry.released
print("modify: retry leaves the published entry unreleased")
PYRepository: evolution-gaming/scache
Length of output: 279
Make deferred publication and EntryState transition share entry ownership. At LoadingCache.scala:676-701 and 825-850, a competing Removed transition can make set/setRef fail after complete publishes entry. put then releases an entry that waiters still hold, while modify retries without releasing it. Ensure every published entry has exactly one owner on both commit-success and commit-failure paths.
📍 Affects 1 file
scache/src/main/scala/com/evolution/scache/LoadingCache.scala#L676-L701(this comment)scache/src/main/scala/com/evolution/scache/LoadingCache.scala#L825-L850
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@scache/src/main/scala/com/evolution/scache/LoadingCache.scala` around lines
676 - 701, Update the LoadingCache put path at LoadingCache.scala:676-701 and
the corresponding modify path at LoadingCache.scala:825-850 so deferred
publication and the EntryState transition use distinct, balanced ownership.
Retain or transfer ownership when complete publishes entry, release the
producer’s ownership exactly once when set/setRef fails after publication,
preserve the waiter-owned reference, and release the unpublished entry when
complete fails before retrying or exiting.
| test("a new load generation does not inherit the previous generation's stuck-timer") { | ||
| val config = ExpiringCache.Config[IO, Int, Int]( | ||
| expireAfterRead = 1.minute, | ||
| loadingTimeout = 200.millis.some, | ||
| ) | ||
| val io = ExpiringCache.of[IO, Int, Int](config).use { cache => | ||
| for { | ||
| started1 <- Deferred[IO, Unit] | ||
| gate1 <- Deferred[IO, Unit] | ||
| loader1 <- cache.getOrUpdate(0) { started1.complete(()) *> gate1.get.as(1) }.start | ||
| _ <- started1.get | ||
| _ <- IO.sleep(150.millis) | ||
| _ <- gate1.complete(()) | ||
| _ <- loader1.join | ||
| _ <- cache.remove(0).flatten | ||
| started2 <- Deferred[IO, Unit] | ||
| gate2 <- Deferred[IO, Unit] | ||
| loader2 <- cache.getOrUpdate(0) { started2.complete(()) *> gate2.get.as(2) }.start | ||
| _ <- started2.get | ||
| result <- { | ||
| for { | ||
| _ <- IO.sleep(150.millis) | ||
| present <- cache.contains(0) | ||
| _ = present shouldEqual true | ||
| } yield () | ||
| }.guarantee { gate2.complete(()) *> loader2.join.void } | ||
| } yield result | ||
| } | ||
| io.run() | ||
| } |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟡 Minor | ⚡ Quick win
Increase the timing margin of the generation test.
The test sets loadingTimeout = 200.millis, sleeps 150 ms after the second load starts, and then asserts that key 0 is still present. The cleanup interval for this config is max(min(60000/10, 200/2), 10) = 100 ms, so two cleanup runs can occur inside the 150 ms window. On a loaded CI machine the second load can be evicted and the assertion fails.
Raise loadingTimeout relative to the sleep, for example a 2 s timeout with a 200 ms sleep, so that the margin does not depend on scheduler jitter.
💚 Proposed fix
val config = ExpiringCache.Config[IO, Int, Int](
expireAfterRead = 1.minute,
- loadingTimeout = 200.millis.some,
+ loadingTimeout = 2.seconds.some,
)and
- _ <- IO.sleep(150.millis)
+ _ <- IO.sleep(300.millis)
_ <- gate1.complete(())with the second sleep inside the result block adjusted to stay well below the new timeout.
📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| test("a new load generation does not inherit the previous generation's stuck-timer") { | |
| val config = ExpiringCache.Config[IO, Int, Int]( | |
| expireAfterRead = 1.minute, | |
| loadingTimeout = 200.millis.some, | |
| ) | |
| val io = ExpiringCache.of[IO, Int, Int](config).use { cache => | |
| for { | |
| started1 <- Deferred[IO, Unit] | |
| gate1 <- Deferred[IO, Unit] | |
| loader1 <- cache.getOrUpdate(0) { started1.complete(()) *> gate1.get.as(1) }.start | |
| _ <- started1.get | |
| _ <- IO.sleep(150.millis) | |
| _ <- gate1.complete(()) | |
| _ <- loader1.join | |
| _ <- cache.remove(0).flatten | |
| started2 <- Deferred[IO, Unit] | |
| gate2 <- Deferred[IO, Unit] | |
| loader2 <- cache.getOrUpdate(0) { started2.complete(()) *> gate2.get.as(2) }.start | |
| _ <- started2.get | |
| result <- { | |
| for { | |
| _ <- IO.sleep(150.millis) | |
| present <- cache.contains(0) | |
| _ = present shouldEqual true | |
| } yield () | |
| }.guarantee { gate2.complete(()) *> loader2.join.void } | |
| } yield result | |
| } | |
| io.run() | |
| } | |
| test("a new load generation does not inherit the previous generation's stuck-timer") { | |
| val config = ExpiringCache.Config[IO, Int, Int]( | |
| expireAfterRead = 1.minute, | |
| loadingTimeout = 2.seconds.some, | |
| ) | |
| val io = ExpiringCache.of[IO, Int, Int](config).use { cache => | |
| for { | |
| started1 <- Deferred[IO, Unit] | |
| gate1 <- Deferred[IO, Unit] | |
| loader1 <- cache.getOrUpdate(0) { started1.complete(()) *> gate1.get.as(1) }.start | |
| _ <- started1.get | |
| _ <- IO.sleep(300.millis) | |
| _ <- gate1.complete(()) | |
| _ <- loader1.join | |
| _ <- cache.remove(0).flatten | |
| started2 <- Deferred[IO, Unit] | |
| gate2 <- Deferred[IO, Unit] | |
| loader2 <- cache.getOrUpdate(0) { started2.complete(()) *> gate2.get.as(2) }.start | |
| _ <- started2.get | |
| result <- { | |
| for { | |
| _ <- IO.sleep(150.millis) | |
| present <- cache.contains(0) | |
| _ = present shouldEqual true | |
| } yield () | |
| }.guarantee { gate2.complete(()) *> loader2.join.void } | |
| } yield result | |
| } | |
| io.run() | |
| } |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@scache/src/test/scala/com/evolution/scache/CacheDefectsSpec.scala` around
lines 211 - 240, Increase the loadingTimeout in the “a new load generation does
not inherit the previous generation's stuck-timer” test and adjust the second
sleep in the result block to remain well below the new timeout, preserving the
presence assertion while providing sufficient scheduler-timing margin.
2 participants
Rebuilds LoadingCache on cats.effect.std.MapRef: per-key CAS instead of whole-map Ref, cancelable loads with cleanup, ExpiringCache evicts stale Loading entries. Fixes four proven defects; defect spec now asserts expected behavior and passes. Adds load-test bench (Test/runMain CacheLoadTest). Creation bounds widened Concurrent to Async.
Bench: 8 fibers x 100k ops/fiber, keySpace 10k, median of 3 runs, ops/s.
Summary by CodeRabbit
New Features
Documentation
Tests