Repository: ignite Updated Branches: refs/heads/ignite-4932 c5c1720d3 -> 1838fdbf5
ignite-4932 WIP Project: http://git-wip-us.apache.org/repos/asf/ignite/repo Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/1838fdbf Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/1838fdbf Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/1838fdbf Branch: refs/heads/ignite-4932 Commit: 1838fdbf5fb46e7a958bc04b82c69173d651a746 Parents: c5c1720 Author: sboikov <[email protected]> Authored: Thu Apr 13 13:37:41 2017 +0300 Committer: sboikov <[email protected]> Committed: Thu Apr 13 13:46:46 2017 +0300 ---------------------------------------------------------------------- .../processors/cache/GridCacheAdapter.java | 2 +- .../dht/GridPartitionedGetFuture.java | 2 +- .../dht/GridPartitionedSingleGetFuture.java | 2 +- .../dht/atomic/GridDhtAtomicCache.java | 2 +- .../dht/colocated/GridDhtColocatedCache.java | 2 +- .../local/atomic/GridLocalAtomicCache.java | 2 +- .../cache/IgniteCacheNoSyncForGetTest.java | 95 +++++++++++++++++--- 7 files changed, 88 insertions(+), 19 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/ignite/blob/1838fdbf/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheAdapter.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheAdapter.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheAdapter.java index 05b4761..b954ea0 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheAdapter.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheAdapter.java @@ -1929,7 +1929,7 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V long expireTime = swapEntry.expireTime(); if (expireTime != 0) { - if (expireTime - U.currentTimeMillis() > 0) { + if (expireTime > U.currentTimeMillis()) { res = new EntryGetWithTtlResult(swapEntry.value(), swapEntry.version(), false, http://git-wip-us.apache.org/repos/asf/ignite/blob/1838fdbf/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridPartitionedGetFuture.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridPartitionedGetFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridPartitionedGetFuture.java index b4b4494..6f3f776 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridPartitionedGetFuture.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridPartitionedGetFuture.java @@ -455,7 +455,7 @@ public class GridPartitionedGetFuture<K, V> extends CacheDistributedGetFutureAda if (swapEntry != null) { long expireTime = swapEntry.expireTime(); - if (expireTime == 0 || expireTime < U.currentTimeMillis()) { + if (expireTime == 0 || expireTime > U.currentTimeMillis()) { skipEntry = true; v = swapEntry.value(); http://git-wip-us.apache.org/repos/asf/ignite/blob/1838fdbf/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridPartitionedSingleGetFuture.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridPartitionedSingleGetFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridPartitionedSingleGetFuture.java index 80b307f..85ccd6b 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridPartitionedSingleGetFuture.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridPartitionedSingleGetFuture.java @@ -377,7 +377,7 @@ public class GridPartitionedSingleGetFuture extends GridFutureAdapter<Object> im if (swapEntry != null) { long expireTime = swapEntry.expireTime(); - if (expireTime == 0 || expireTime < U.currentTimeMillis()) { + if (expireTime == 0 || expireTime > U.currentTimeMillis()) { skipEntry = true; v = swapEntry.value(); http://git-wip-us.apache.org/repos/asf/ignite/blob/1838fdbf/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicCache.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicCache.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicCache.java index c6bceef..5497980 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicCache.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicCache.java @@ -1584,7 +1584,7 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> { if (swapEntry != null) { long expireTime = swapEntry.expireTime(); - if (expireTime == 0 || expireTime < U.currentTimeMillis()) { + if (expireTime == 0 || expireTime > U.currentTimeMillis()) { ctx.addResult(locVals, key, swapEntry.value(), http://git-wip-us.apache.org/repos/asf/ignite/blob/1838fdbf/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/colocated/GridDhtColocatedCache.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/colocated/GridDhtColocatedCache.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/colocated/GridDhtColocatedCache.java index 19e001f..c09ad9b 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/colocated/GridDhtColocatedCache.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/colocated/GridDhtColocatedCache.java @@ -467,7 +467,7 @@ public class GridDhtColocatedCache<K, V> extends GridDhtTransactionalCacheAdapte if (swapEntry != null) { long expireTime = swapEntry.expireTime(); - if (expireTime == 0 || expireTime < U.currentTimeMillis()) { + if (expireTime == 0 || expireTime > U.currentTimeMillis()) { if (locVals == null) locVals = U.newHashMap(keys.size()); http://git-wip-us.apache.org/repos/asf/ignite/blob/1838fdbf/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/local/atomic/GridLocalAtomicCache.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/local/atomic/GridLocalAtomicCache.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/local/atomic/GridLocalAtomicCache.java index 74eaf17..b181e16 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/local/atomic/GridLocalAtomicCache.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/local/atomic/GridLocalAtomicCache.java @@ -414,7 +414,7 @@ public class GridLocalAtomicCache<K, V> extends GridLocalCache<K, V> { if (swapEntry != null) { long expireTime = swapEntry.expireTime(); - if (expireTime == 0 || expireTime < U.currentTimeMillis()) { + if (expireTime == 0 || expireTime > U.currentTimeMillis()) { skipEntry = true; ctx.addResult(vals, http://git-wip-us.apache.org/repos/asf/ignite/blob/1838fdbf/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheNoSyncForGetTest.java ---------------------------------------------------------------------- diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheNoSyncForGetTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheNoSyncForGetTest.java index e4c37f4..d1b97c5 100644 --- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheNoSyncForGetTest.java +++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteCacheNoSyncForGetTest.java @@ -23,6 +23,8 @@ import java.util.Set; import java.util.concurrent.Callable; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; +import javax.cache.expiry.Duration; +import javax.cache.expiry.ModifiedExpiryPolicy; import javax.cache.processor.EntryProcessorException; import javax.cache.processor.MutableEntry; import org.apache.ignite.Ignite; @@ -43,12 +45,14 @@ import org.apache.ignite.testframework.GridTestUtils; import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest; import static org.apache.ignite.cache.CacheAtomicityMode.ATOMIC; +import static org.apache.ignite.cache.CacheAtomicityMode.TRANSACTIONAL; import static org.apache.ignite.cache.CacheMemoryMode.OFFHEAP_TIERED; import static org.apache.ignite.cache.CacheWriteSynchronizationMode.FULL_SYNC; /** * */ +@SuppressWarnings("unchecked") public class IgniteCacheNoSyncForGetTest extends GridCommonAbstractTest { /** IP finder. */ private static final TcpDiscoveryIpFinder IP_FINDER = new TcpDiscoveryVmIpFinder(true); @@ -95,22 +99,52 @@ public class IgniteCacheNoSyncForGetTest extends GridCommonAbstractTest { * @throws Exception If failed. */ public void testAtomicGetOffheap() throws Exception { - doGet(ATOMIC, OFFHEAP_TIERED, false); + boolean getAll[] = {true, false}; + boolean cfgExpiryPlc[] = {true, false}; + boolean withExpiryPlc[] = {true, false}; + + for (boolean getAll0 : getAll) { + for (boolean expiryPlc0 : cfgExpiryPlc) + for (boolean withExpiryPlc0 : withExpiryPlc) + doGet(ATOMIC, OFFHEAP_TIERED, getAll0, expiryPlc0, withExpiryPlc0); + } + } - doGet(ATOMIC, OFFHEAP_TIERED, true); + /** + * @throws Exception If failed. + */ + public void testTxGetOffheap() throws Exception { + boolean getAll[] = {true, false}; + boolean cfgExpiryPlc[] = {true, false}; + boolean withExpiryPlc[] = {true, false}; + + for (boolean getAll0 : getAll) { + for (boolean expiryPlc0 : cfgExpiryPlc) + for (boolean withExpiryPlc0 : withExpiryPlc) + doGet(TRANSACTIONAL, OFFHEAP_TIERED, getAll0, expiryPlc0, withExpiryPlc0); + } } /** + * @param atomicityMode Cache atomicity mode. + * @param memoryMode Cache memory mode. + * @param getAll Test getAll flag. + * @param cfgExpiryPlc Configured expiry policy flag. + * @param withExpiryPlc Custom expiry policy flag. * @throws Exception If failed. */ private void doGet(CacheAtomicityMode atomicityMode, CacheMemoryMode memoryMode, - final boolean getAll) throws Exception { + final boolean getAll, + final boolean cfgExpiryPlc, + final boolean withExpiryPlc) throws Exception { + log.info("Test get [getAll=" + getAll + ", cfgExpiryPlc=" + cfgExpiryPlc + ']'); + Ignite srv = ignite(0); Ignite client = ignite(1); - final IgniteCache cache = client.createCache(cacheConfiguration(atomicityMode, memoryMode)); + final IgniteCache cache = client.createCache(cacheConfiguration(atomicityMode, memoryMode, cfgExpiryPlc)); final Map<Object, Object> data = new HashMap<>(); @@ -143,11 +177,11 @@ public class IgniteCacheNoSyncForGetTest extends GridCommonAbstractTest { if (getAll) { assertEquals(data, client.compute().affinityCall(cache.getName(), 1, - new GetAllClosure(data.keySet(), cache.getName()))); + new GetAllClosure(data.keySet(), cache.getName(), withExpiryPlc))); } else { assertEquals(1, client.compute().affinityCall(cache.getName(), 1, - new GetClosure(1, cache.getName()))); + new GetClosure(1, cache.getName(), withExpiryPlc))); } hangLatch.countDown(); @@ -182,10 +216,15 @@ public class IgniteCacheNoSyncForGetTest extends GridCommonAbstractTest { assertTrue(wait); + IgniteCache srvCache = srv.cache(cache.getName()); + + if (withExpiryPlc) + srvCache = srvCache.withExpiryPolicy(ModifiedExpiryPolicy.factoryOf(Duration.FIVE_MINUTES).create()); + if (getAll) - assertEquals(data, srv.cache(cache.getName()).getAll(data.keySet())); + assertEquals(data, srvCache.getAll(data.keySet())); else - assertEquals(1, srv.cache(cache.getName()).get(1)); + assertEquals(1, srvCache.get(1)); hangLatch.countDown(); @@ -202,9 +241,14 @@ public class IgniteCacheNoSyncForGetTest extends GridCommonAbstractTest { } /** + * @param atomicityMode Atomicity mode. + * @param memoryMode Memory mode. + * @param expiryPlc Expiry policy flag. * @return Cache configuration. */ - private CacheConfiguration cacheConfiguration(CacheAtomicityMode atomicityMode, CacheMemoryMode memoryMode) { + private CacheConfiguration cacheConfiguration(CacheAtomicityMode atomicityMode, + CacheMemoryMode memoryMode, + boolean expiryPlc) { CacheConfiguration ccfg = new CacheConfiguration(); ccfg.setAtomicityMode(atomicityMode); @@ -212,6 +256,9 @@ public class IgniteCacheNoSyncForGetTest extends GridCommonAbstractTest { ccfg.setWriteSynchronizationMode(FULL_SYNC); ccfg.setName("testCache"); + if (expiryPlc) + ccfg.setExpiryPolicyFactory(ModifiedExpiryPolicy.factoryOf(Duration.FIVE_MINUTES)); + return ccfg; } @@ -256,17 +303,28 @@ public class IgniteCacheNoSyncForGetTest extends GridCommonAbstractTest { /** */ private final String cacheName; + /** */ + private final boolean withExpiryPlc; + /** * @param key Key. + * @param cacheName Cache name. + * @param withExpiryPlc Custom expiry policy flag. */ - GetClosure(int key, String cacheName) { + GetClosure(int key, String cacheName, boolean withExpiryPlc) { this.key = key; this.cacheName = cacheName; + this.withExpiryPlc = withExpiryPlc; } /** {@inheritDoc} */ @Override public Object call() throws Exception { - return ignite.cache(cacheName).get(key); + IgniteCache cache = ignite.cache(cacheName); + + if (withExpiryPlc) + cache = cache.withExpiryPolicy(ModifiedExpiryPolicy.factoryOf(Duration.FIVE_MINUTES).create()); + + return cache.get(key); } } @@ -284,17 +342,28 @@ public class IgniteCacheNoSyncForGetTest extends GridCommonAbstractTest { /** */ private final String cacheName; + /** */ + private final boolean withExpiryPlc; + /** * @param keys Keys. + * @param cacheName Cache name. + * @param withExpiryPlc Custom expiry policy flag. */ - GetAllClosure(Set<Object> keys, String cacheName) { + GetAllClosure(Set<Object> keys, String cacheName, boolean withExpiryPlc) { this.keys = keys; this.cacheName = cacheName; + this.withExpiryPlc = withExpiryPlc; } /** {@inheritDoc} */ @Override public Object call() throws Exception { - return ignite.cache(cacheName).getAll(keys); + IgniteCache cache = ignite.cache(cacheName); + + if (withExpiryPlc) + cache = cache.withExpiryPolicy(ModifiedExpiryPolicy.factoryOf(Duration.FIVE_MINUTES).create()); + + return cache.getAll(keys); } } }
