Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
ISPN-5607 Preemptively invalidate near cache after writes
* Near cache stale reads can't be fully guaranteed since there's always the possibility of a read to come in when the server has already executed a write operation but it's in process of sending back the response. * However, we can't at least guarantee that within a single thread, a read after a write will read the written value. The change proposed in this PR addresses this, by preemptively invalidating data after a successful write.
- Loading branch information
Showing
9 changed files
with
277 additions
and
24 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
111 changes: 111 additions & 0 deletions
111
...od-client/src/main/java/org/infinispan/client/hotrod/impl/InvalidatedNearRemoteCache.java
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 |
|---|---|---|
| @@ -0,0 +1,111 @@ | ||
| package org.infinispan.client.hotrod.impl; | ||
|
|
||
| import org.infinispan.client.hotrod.Flag; | ||
| import org.infinispan.client.hotrod.RemoteCacheManager; | ||
| import org.infinispan.client.hotrod.VersionedValue; | ||
| import org.infinispan.client.hotrod.near.NearCacheService; | ||
|
|
||
| import java.util.Map; | ||
| import java.util.concurrent.TimeUnit; | ||
|
|
||
| /** | ||
| * Near {@link org.infinispan.client.hotrod.RemoteCache} implementation | ||
| * enabling | ||
| * | ||
| * @param <K> | ||
| * @param <V> | ||
| */ | ||
| public class InvalidatedNearRemoteCache<K, V> extends RemoteCacheImpl<K, V> { | ||
|
|
||
| private final NearCacheService<K, V> nearcache; | ||
|
|
||
| public InvalidatedNearRemoteCache(RemoteCacheManager rcm, String name, NearCacheService<K, V> nearcache) { | ||
| super(rcm, name); | ||
| this.nearcache = nearcache; | ||
| } | ||
|
|
||
| @Override | ||
| public V get(Object key) { | ||
| VersionedValue<V> versioned = getVersioned((K) key); | ||
| return versioned != null ? versioned.getValue() : null; | ||
| } | ||
|
|
||
| @Override | ||
| public VersionedValue<V> getVersioned(K key) { | ||
| VersionedValue<V> nearValue = nearcache.get(key); | ||
| if (nearValue == null) { | ||
| VersionedValue<V> remoteValue = super.getVersioned(key); | ||
| if (remoteValue != null) | ||
| nearcache.putIfAbsent(key, remoteValue); | ||
|
|
||
| return remoteValue; | ||
| } | ||
|
|
||
| return nearValue; | ||
| } | ||
|
|
||
| @Override | ||
| public V put(K key, V value, long lifespan, TimeUnit lifespanUnit, long maxIdleTime, TimeUnit maxIdleTimeUnit) { | ||
| V ret = super.put(key, value, lifespan, lifespanUnit, maxIdleTime, maxIdleTimeUnit); | ||
| nearcache.remove(key); // Eager invalidation to avoid race | ||
| return ret; | ||
| } | ||
|
|
||
| @Override | ||
| public void putAll(Map<? extends K, ? extends V> map, long lifespan, TimeUnit lifespanUnit, long maxIdleTime, TimeUnit maxIdleTimeUnit) { | ||
| super.putAll(map, lifespan, lifespanUnit, maxIdleTime, maxIdleTimeUnit); | ||
| map.keySet().forEach(nearcache::remove); | ||
| } | ||
|
|
||
| @Override | ||
| public V replace(K key, V value, long lifespan, TimeUnit lifespanUnit, long maxIdleTime, TimeUnit maxIdleTimeUnit) { | ||
| boolean hasForceReturnValue = operationsFactory.hasFlag(Flag.FORCE_RETURN_VALUE); | ||
| V prev = super.replace(key, value, lifespan, lifespanUnit, maxIdleTime, maxIdleTimeUnit); | ||
| invalidateNearCacheIfNeeded(hasForceReturnValue, key, prev); | ||
| return prev; | ||
| } | ||
|
|
||
| @Override | ||
| public boolean replaceWithVersion(K key, V newValue, long version, long lifespan, TimeUnit lifespanTimeUnit, long maxIdle, TimeUnit maxIdleTimeUnit) { | ||
| boolean replaced = super.replaceWithVersion(key, newValue, version, lifespan, lifespanTimeUnit, maxIdle, maxIdleTimeUnit); | ||
| if (replaced) nearcache.remove(key); | ||
| return replaced; | ||
| } | ||
|
|
||
| @Override | ||
| public V remove(Object key) { | ||
| boolean hasForceReturnValue = operationsFactory.hasFlag(Flag.FORCE_RETURN_VALUE); | ||
| V prev = super.remove(key); | ||
| invalidateNearCacheIfNeeded(hasForceReturnValue, key, prev); | ||
| return prev; | ||
| } | ||
|
|
||
| @Override | ||
| public boolean removeWithVersion(K key, long version) { | ||
| boolean removed = super.removeWithVersion(key, version); | ||
| if (removed) nearcache.remove(key); // Eager invalidation to avoid race | ||
| return removed; | ||
| } | ||
|
|
||
| @Override | ||
| public void clear() { | ||
| super.clear(); | ||
| nearcache.clear(); // Clear near cache too | ||
| } | ||
|
|
||
| @SuppressWarnings("unchecked") | ||
| void invalidateNearCacheIfNeeded(boolean hasForceReturnValue, Object key, Object prev) { | ||
| if (!hasForceReturnValue || prev != null) | ||
| nearcache.remove((K) key); | ||
| } | ||
|
|
||
| @Override | ||
| public void start() { | ||
| nearcache.start(this); | ||
| } | ||
|
|
||
| @Override | ||
| public void stop() { | ||
| nearcache.stop(this); | ||
| } | ||
| } |
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
109 changes: 109 additions & 0 deletions
109
...-client/src/test/java/org/infinispan/client/hotrod/near/AvoidStaleNearCacheReadsTest.java
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 |
|---|---|---|
| @@ -0,0 +1,109 @@ | ||
| package org.infinispan.client.hotrod.near; | ||
|
|
||
| import org.infinispan.client.hotrod.RemoteCache; | ||
| import org.infinispan.client.hotrod.RemoteCacheManager; | ||
| import org.infinispan.client.hotrod.VersionedValue; | ||
| import org.infinispan.client.hotrod.configuration.ConfigurationBuilder; | ||
| import org.infinispan.client.hotrod.configuration.NearCacheMode; | ||
| import org.infinispan.client.hotrod.test.SingleHotRodServerTest; | ||
| import org.infinispan.commons.util.concurrent.NotifyingFuture; | ||
| import org.testng.annotations.AfterMethod; | ||
| import org.testng.annotations.Test; | ||
|
|
||
| import java.util.HashMap; | ||
| import java.util.Map; | ||
| import java.util.concurrent.ExecutionException; | ||
| import java.util.concurrent.Future; | ||
| import java.util.concurrent.TimeUnit; | ||
| import java.util.concurrent.TimeoutException; | ||
| import java.util.function.BiConsumer; | ||
| import java.util.function.Consumer; | ||
| import java.util.stream.IntStream; | ||
|
|
||
| import static org.testng.AssertJUnit.assertEquals; | ||
| import static org.testng.AssertJUnit.assertNull; | ||
|
|
||
| @Test(groups = "functional", testName = "client.hotrod.near.AvoidStaleNearCacheReadsTest") | ||
| public class AvoidStaleNearCacheReadsTest extends SingleHotRodServerTest { | ||
|
|
||
| @AfterMethod(alwaysRun=true) | ||
| @Override | ||
| protected void clearContent() { | ||
| super.clearContent(); | ||
| remoteCacheManager.getCache().clear(); // Clear the near cache too | ||
| } | ||
|
|
||
| @Override | ||
| protected RemoteCacheManager getRemoteCacheManager() { | ||
| ConfigurationBuilder builder = new ConfigurationBuilder(); | ||
| builder.addServer().host("127.0.0.1").port(hotrodServer.getPort()); | ||
| builder.nearCache().mode(NearCacheMode.INVALIDATED).maxEntries(-1); | ||
| return new RemoteCacheManager(builder.build()); | ||
| } | ||
|
|
||
| public void testAvoidStaleReadsAfterPutRemove() { | ||
| repeated((i, remote) -> { | ||
| String value = "v" + i; | ||
| remote.put(1, value); | ||
| assertEquals(value, remote.get(1)); | ||
| remote.remove(1); | ||
| assertNull(remote.get(1)); | ||
| }); | ||
| } | ||
|
|
||
| public void testAvoidStaleReadsAfterPutAll() { | ||
| repeated((i, remote) -> { | ||
| String value = "v" + i; | ||
| Map<Integer, String> map = new HashMap<>(); | ||
| map.put(1, value); | ||
| remote.putAll(map); | ||
| assertEquals(value, remote.get(1)); | ||
| }); | ||
| } | ||
|
|
||
| public void testAvoidStaleReadsAfterReplace() { | ||
| repeated((i, remote) -> { | ||
| String value = "v" + i; | ||
| remote.replace(1, value); | ||
| VersionedValue<String> versioned = remote.getVersioned(1); | ||
| assertEquals(value, versioned.getValue()); | ||
| }); | ||
| } | ||
|
|
||
| public void testAvoidStaleReadsAfterReplaceWithVersion() { | ||
| repeated((i, remote) -> { | ||
| String value = "v" + i; | ||
| VersionedValue<String> versioned = remote.getVersioned(1); | ||
| remote.replaceWithVersion(1, value, versioned.getVersion()); | ||
| assertEquals(value, remote.get(1)); | ||
| }); | ||
| } | ||
|
|
||
| public void testAvoidStaleReadsAfterPutAsyncRemoveVersioned() { | ||
| repeated((i, remote) -> { | ||
| String value = "v" + i; | ||
| await(remote.putAsync(1, value)); | ||
| VersionedValue<String> versioned = remote.getVersioned(1); | ||
| assertEquals(value, versioned.getValue()); | ||
| remote.removeWithVersion(1, versioned.getVersion()); | ||
| assertNull(remote.get(1)); | ||
| }); | ||
| } | ||
|
|
||
| private void repeated(BiConsumer<Integer, RemoteCache<Integer, String>> c) { | ||
| RemoteCache<Integer, String> remote = remoteCacheManager.getCache(); | ||
| remote.putIfAbsent(1, "v0"); | ||
| IntStream.range(1, 1000).forEach(i -> { | ||
| c.accept(i, remote); | ||
| }); | ||
| } | ||
|
|
||
| static <T> T await(Future<T> f) { | ||
| try { | ||
| return f.get(10000, TimeUnit.SECONDS); | ||
| } catch (InterruptedException | ExecutionException | TimeoutException e ) { | ||
| throw new AssertionError(e); | ||
| } | ||
| } | ||
|
|
||
| } |
Oops, something went wrong.