-
Notifications
You must be signed in to change notification settings - Fork 1.6k
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Fix async refresh from prematurely replacing the value
When a refresh-after-write is triggered the entry should be reloaded in the background, the present value available for reads, and atomically replaced when the new value has been loaded. This operation is atomic with other writes to that entry and is blocking (rather than clobbering) if another write is attempted. This should work the same for synchronous and asynchronous caches, but unfortunately it wasn't. For an asycn cache a new, incomplete future was immediately put into the cache and available to be consumed by the next request. Due to layering, the custom reloadAsync was not called and the operation delegated to load instead. This was of course wrong and not the intended (or expected) behavior, so it is now fixed. Thanks to Etienne Houle @ Stingray for notifying me of this problem.
- Loading branch information
Showing
9 changed files
with
146 additions
and
33 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -74,7 +74,7 @@ | |
| * for a given configuration are used. | ||
| * | ||
| * @author [email protected] (Ben Manes) | ||
| * @param <K> the type of keys maintained by this map | ||
| * @param <K> the type of keys maintained by this cache | ||
| * @param <V> the type of mapped values | ||
| */ | ||
| @ThreadSafe | ||
|
|
@@ -727,8 +727,6 @@ void afterRead(Node<K, V> node, long now, boolean recordHit) { | |
| } | ||
| node.setAccessTime(now); | ||
|
|
||
| // fastpath is disabled due to unfavorable benchmarks | ||
| // boolean delayable = canFastpath(node) || (readBuffer.offer(node) != Buffer.FULL); | ||
| boolean delayable = skipReadBuffer() || (readBuffer.offer(node) != Buffer.FULL); | ||
| if (shouldDrainBuffers(delayable)) { | ||
| scheduleDrainBuffers(); | ||
|
|
@@ -751,20 +749,34 @@ void refreshIfNeeded(Node<K, V> node, long now) { | |
| if (!refreshAfterWrite()) { | ||
| return; | ||
| } | ||
| long writeTime = node.getWriteTime(); | ||
| if (((now - writeTime) > refreshAfterWriteNanos()) && node.casWriteTime(writeTime, now)) { | ||
| long oldWriteTime = node.getWriteTime(); | ||
| long refreshWriteTime = isAsync ? Long.MAX_VALUE : now; | ||
| if (((now - oldWriteTime) > refreshAfterWriteNanos()) | ||
| && node.casWriteTime(oldWriteTime, refreshWriteTime)) { | ||
| try { | ||
| executor().execute(() -> { | ||
| K key = node.getKey(); | ||
| if ((key != null) && node.isAlive()) { | ||
| BiFunction<? super K, ? super V, ? extends V> refreshFunction = (k, v) -> { | ||
| if (node.getWriteTime() != now) { | ||
| if (node.getWriteTime() != refreshWriteTime) { | ||
| return v; | ||
| } | ||
| try { | ||
| if (isAsync) { | ||
| @SuppressWarnings("unchecked") | ||
| V oldValue = ((CompletableFuture<V>) v).join(); | ||
| CompletableFuture<V> future = | ||
| cacheLoader.asyncReload(key, oldValue, Runnable::run); | ||
| if (future.join() == null) { | ||
| return null; | ||
| } | ||
| @SuppressWarnings("unchecked") | ||
| V castFuture = (V) future; | ||
| return castFuture; | ||
| } | ||
| return cacheLoader.reload(k, v); | ||
| } catch (Exception e) { | ||
| node.setWriteTime(writeTime); | ||
| node.setWriteTime(oldWriteTime); | ||
| return LocalCache.throwUnchecked(e); | ||
| } | ||
| }; | ||
|
|
@@ -2916,26 +2928,40 @@ static final class BoundedLocalAsyncLoadingCache<K, V> | |
|
|
||
| @SuppressWarnings("unchecked") | ||
| BoundedLocalAsyncLoadingCache(Caffeine<K, V> builder, AsyncCacheLoader<? super K, V> loader) { | ||
| super(LocalCacheFactory.newBoundedLocalCache((Caffeine<K, CompletableFuture<V>>) builder, | ||
| asyncLoader(loader, builder), true), loader); | ||
| super((BoundedLocalCache<K, CompletableFuture<V>>) LocalCacheFactory.newBoundedLocalCache( | ||
| builder, asyncLoader(loader, builder), true), loader); | ||
| isWeighted = builder.isWeighted(); | ||
| } | ||
|
|
||
| private static <K, V> CacheLoader<? super K, CompletableFuture<V>> asyncLoader( | ||
| private static <K, V> CacheLoader<K, V> asyncLoader( | ||
| AsyncCacheLoader<? super K, V> loader, Caffeine<?, ?> builder) { | ||
| Executor executor = builder.getExecutor(); | ||
| return key -> loader.asyncLoad(key, executor); | ||
| return new CacheLoader<K, V>() { | ||
| @Override public V load(K key) { | ||
| @SuppressWarnings("unchecked") | ||
| V newValue = (V) loader.asyncLoad(key, executor); | ||
| return newValue; | ||
| } | ||
| @Override public V reload(K key, V oldValue) { | ||
| @SuppressWarnings("unchecked") | ||
| V newValue = (V) loader.asyncReload(key, oldValue, executor); | ||
| return newValue; | ||
| } | ||
| @Override public CompletableFuture<V> asyncReload(K key, V oldValue, Executor executor) { | ||
| return loader.asyncReload(key, oldValue, executor); | ||
| } | ||
| }; | ||
| } | ||
|
|
||
| @Override | ||
| protected Policy<K, V> policy() { | ||
| if (policy == null) { | ||
| @SuppressWarnings("unchecked") | ||
| BoundedLocalCache<K, V> castedCache = (BoundedLocalCache<K, V>) cache; | ||
| BoundedLocalCache<K, V> castCache = (BoundedLocalCache<K, V>) cache; | ||
| Function<CompletableFuture<V>, V> transformer = Async::getIfReady; | ||
| @SuppressWarnings("unchecked") | ||
| Function<V, V> castedTransformer = (Function<V, V>) transformer; | ||
| policy = new BoundedPolicy<>(castedCache, castedTransformer, isWeighted); | ||
| Function<V, V> castTransformer = (Function<V, V>) transformer; | ||
| policy = new BoundedPolicy<>(castCache, castTransformer, isWeighted); | ||
| } | ||
| return policy; | ||
| } | ||
|
|
||
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters