ignite-656: passing skipStore flag to higher API
Project: http://git-wip-us.apache.org/repos/asf/incubator-ignite/repo Commit: http://git-wip-us.apache.org/repos/asf/incubator-ignite/commit/63c32301 Tree: http://git-wip-us.apache.org/repos/asf/incubator-ignite/tree/63c32301 Diff: http://git-wip-us.apache.org/repos/asf/incubator-ignite/diff/63c32301 Branch: refs/heads/ignite-709_1 Commit: 63c32301be6e003a4eba18c1b3ad5e78b41d2ed7 Parents: 1361d1c Author: Denis Magda <dma...@gridgain.com> Authored: Tue Apr 21 16:16:40 2015 +0300 Committer: Denis Magda <dma...@gridgain.com> Committed: Tue Apr 21 16:16:40 2015 +0300 ---------------------------------------------------------------------- .../processors/cache/GridCacheAdapter.java | 4 +- .../cache/distributed/dht/GridDhtGetFuture.java | 6 +- .../distributed/dht/GridDhtTxLocalAdapter.java | 5 +- .../dht/colocated/GridDhtColocatedCache.java | 7 +- .../near/GridNearTransactionalCache.java | 11 ++-- .../transactions/IgniteTxLocalAdapter.java | 69 +++++++++++++------- .../cache/transactions/IgniteTxLocalEx.java | 4 +- .../cache/GridCacheAbstractFullApiSelfTest.java | 2 + 8 files changed, 69 insertions(+), 39 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/63c32301/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 38ebfe4..450e8a2 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 @@ -1744,7 +1744,7 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V } } - if (!skipVals && misses != null && readThrough && ctx.readThrough() && !ctx.skipStore()) { + if (!skipVals && misses != null && readThrough && ctx.readThrough()) { final Map<KeyCacheObject, GridCacheVersion> loadKeys = misses; final IgniteTxLocalAdapter tx0 = tx; @@ -1891,7 +1891,7 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V else { return asyncOp(tx, new AsyncOp<Map<K1, V1>>(keys) { @Override public IgniteInternalFuture<Map<K1, V1>> op(IgniteTxLocalAdapter tx) { - return tx.getAllAsync(ctx, keys, null, deserializePortable, skipVals, false); + return tx.getAllAsync(ctx, keys, null, deserializePortable, skipVals, false, readThrough); } }); } http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/63c32301/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtGetFuture.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtGetFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtGetFuture.java index 76ae5a1..ed5414d 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtGetFuture.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtGetFuture.java @@ -355,7 +355,8 @@ public final class GridDhtGetFuture<K, V> extends GridCompoundIdentityFuture<Col null, /*deserialize portable*/false, skipVals, - /*keep cache objects*/true); + /*keep cache objects*/true, + /*skip store*/readThrough); } } } @@ -391,7 +392,8 @@ public final class GridDhtGetFuture<K, V> extends GridCompoundIdentityFuture<Col null, /*deserialize portable*/false, skipVals, - /*keep cache objects*/true); + /*keep cache objects*/true, + /*skip store*/readThrough); } } } http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/63c32301/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxLocalAdapter.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxLocalAdapter.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxLocalAdapter.java index 79f1f96..65ca0ad 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxLocalAdapter.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxLocalAdapter.java @@ -593,9 +593,8 @@ public abstract class GridDhtTxLocalAdapter extends IgniteTxLocalAdapter { false, -1L, -1L, - null); - - txEntry.skipStore(skipStore); + null, + skipStore); if (read) txEntry.ttl(accessTtl); http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/63c32301/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 65fe785..cafd268 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 @@ -164,6 +164,8 @@ public class GridDhtColocatedCache<K, V> extends GridDhtTransactionalCacheAdapte IgniteTxLocalAdapter tx = ctx.tm().threadLocalTx(ctx); + CacheOperationContext opCtx = ctx.operationContextPerCall(); + if (tx != null && !tx.implicit() && !skipTx) { return asyncOp(tx, new AsyncOp<Map<K, V>>(keys) { @Override public IgniteInternalFuture<Map<K, V>> op(IgniteTxLocalAdapter tx) { @@ -172,15 +174,14 @@ public class GridDhtColocatedCache<K, V> extends GridDhtTransactionalCacheAdapte entry, deserializePortable, skipVals, - false); + false, + opCtx != null && opCtx.skipStore()); } }); } AffinityTopologyVersion topVer = tx == null ? ctx.affinity().affinityTopologyVersion() : tx.topologyVersion(); - CacheOperationContext opCtx = ctx.operationContextPerCall(); - subjId = ctx.subjectIdPerCall(subjId, opCtx); return loadAsync( http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/63c32301/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTransactionalCache.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTransactionalCache.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTransactionalCache.java index 661a289..00b18a4 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTransactionalCache.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTransactionalCache.java @@ -113,6 +113,10 @@ public class GridNearTransactionalCache<K, V> extends GridNearCacheAdapter<K, V> IgniteTxLocalAdapter tx = ctx.tm().threadLocalTx(ctx); + CacheOperationContext opCtx = ctx.operationContextPerCall(); + + boolean skipStore = opCtx != null && opCtx.skipStore(); + if (tx != null && !tx.implicit() && !skipTx) { return asyncOp(tx, new AsyncOp<Map<K, V>>(keys) { @Override public IgniteInternalFuture<Map<K, V>> op(IgniteTxLocalAdapter tx) { @@ -121,13 +125,12 @@ public class GridNearTransactionalCache<K, V> extends GridNearCacheAdapter<K, V> entry, deserializePortable, skipVals, - false); + false, + skipStore); } }); } - CacheOperationContext opCtx = ctx.operationContextPerCall(); - subjId = ctx.subjectIdPerCall(subjId, opCtx); return loadAsync(null, @@ -139,7 +142,7 @@ public class GridNearTransactionalCache<K, V> extends GridNearCacheAdapter<K, V> deserializePortable, skipVals ? null : opCtx != null ? opCtx.expiry() : null, skipVals, - opCtx != null && opCtx.skipStore()); + skipStore); } /** http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/63c32301/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxLocalAdapter.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxLocalAdapter.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxLocalAdapter.java index 4591c58..8f66d62 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxLocalAdapter.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxLocalAdapter.java @@ -1104,7 +1104,7 @@ public abstract class IgniteTxLocalAdapter extends IgniteTxAdapter /** * Checks if there is a cached or swapped value for - * {@link #getAllAsync(GridCacheContext, Collection, GridCacheEntryEx, boolean, boolean, boolean)} method. + * {@link #getAllAsync(GridCacheContext, Collection, GridCacheEntryEx, boolean, boolean, boolean, boolean)} method. * * @param cacheCtx Cache context. * @param keys Key to enlist. @@ -1116,6 +1116,7 @@ public abstract class IgniteTxLocalAdapter extends IgniteTxAdapter * @param deserializePortable Deserialize portable flag. * @param skipVals Skip values flag. * @param keepCacheObjects Keep cache objects flag. + * @param skipStore Skip store flag. * @throws IgniteCheckedException If failed. * @return Enlisted keys. */ @@ -1130,7 +1131,8 @@ public abstract class IgniteTxLocalAdapter extends IgniteTxAdapter int keysCnt, boolean deserializePortable, boolean skipVals, - boolean keepCacheObjects + boolean keepCacheObjects, + boolean skipStore ) throws IgniteCheckedException { assert !F.isEmpty(keys); assert keysCnt == keys.size(); @@ -1294,7 +1296,8 @@ public abstract class IgniteTxLocalAdapter extends IgniteTxAdapter true, -1L, -1L, - null); + null, + skipStore); if (groupLock()) txEntry.groupLockEntry(true); @@ -1329,7 +1332,8 @@ public abstract class IgniteTxLocalAdapter extends IgniteTxAdapter false, -1L, -1L, - null); + null, + skipStore); // Mark as checked immediately for non-pessimistic. if (val != null && !pessimistic()) @@ -1388,7 +1392,7 @@ public abstract class IgniteTxLocalAdapter extends IgniteTxAdapter /** * Loads all missed keys for - * {@link #getAllAsync(GridCacheContext, Collection, GridCacheEntryEx, boolean, boolean, boolean)} method. + * {@link #getAllAsync(GridCacheContext, Collection, GridCacheEntryEx, boolean, boolean, boolean, boolean)} method. * * @param cacheCtx Cache context. * @param map Return map. @@ -1396,6 +1400,7 @@ public abstract class IgniteTxLocalAdapter extends IgniteTxAdapter * @param redos Keys to retry. * @param deserializePortable Deserialize portable flag. * @param keepCacheObjects Keep cache objects flag. + * @param skipStore Skip store flag. * @return Loaded key-value pairs. */ private <K, V> IgniteInternalFuture<Map<K, V>> checkMissed( @@ -1405,7 +1410,8 @@ public abstract class IgniteTxLocalAdapter extends IgniteTxAdapter @Nullable final Collection<KeyCacheObject> redos, final boolean deserializePortable, final boolean skipVals, - final boolean keepCacheObjects + final boolean keepCacheObjects, + final boolean skipStore ) { assert redos != null || pessimistic(); @@ -1455,7 +1461,7 @@ public abstract class IgniteTxLocalAdapter extends IgniteTxAdapter }, loadMissing( cacheCtx, - !cacheCtx.skipStore(), + !skipStore, false, missedMap.keySet(), deserializePortable, @@ -1597,7 +1603,8 @@ public abstract class IgniteTxLocalAdapter extends IgniteTxAdapter @Nullable GridCacheEntryEx cached, final boolean deserializePortable, final boolean skipVals, - final boolean keepCacheObjects) { + final boolean keepCacheObjects, + final boolean skipStore) { if (F.isEmpty(keys)) return new GridFinishedFuture<>(Collections.<K, V>emptyMap()); @@ -1627,7 +1634,8 @@ public abstract class IgniteTxLocalAdapter extends IgniteTxAdapter keysCnt, deserializePortable, skipVals, - keepCacheObjects); + keepCacheObjects, + skipStore); if (single && missed.isEmpty()) return new GridFinishedFuture<>(retMap); @@ -1748,7 +1756,8 @@ public abstract class IgniteTxLocalAdapter extends IgniteTxAdapter null, deserializePortable, skipVals, - keepCacheObjects); + keepCacheObjects, + skipStore); } return new GridFinishedFuture<>(Collections.<K, V>emptyMap()); @@ -1812,7 +1821,8 @@ public abstract class IgniteTxLocalAdapter extends IgniteTxAdapter return new GridEmbeddedFuture<>( // First future. - checkMissed(cacheCtx, retMap, missed, redos, deserializePortable, skipVals, keepCacheObjects), + checkMissed(cacheCtx, retMap, missed, redos, deserializePortable, skipVals, keepCacheObjects, + skipStore), // Closure that returns another future, based on result from first. new PMC<Map<K, V>>() { @Override public IgniteInternalFuture<Map<K, V>> postMiss(Map<K, V> map) { @@ -1829,7 +1839,8 @@ public abstract class IgniteTxLocalAdapter extends IgniteTxAdapter null, deserializePortable, skipVals, - true); + true, + skipStore); } }, // Finalize. @@ -1964,6 +1975,7 @@ public abstract class IgniteTxLocalAdapter extends IgniteTxAdapter * @param enlisted Collection of keys enlisted into this transaction. * @param drPutMap DR put map (optional). * @param drRmvMap DR remove map (optional). + * @param skipStore Skip store flag. * @return Future with skipped keys (the ones that didn't pass filter for pessimistic transactions). */ protected <K, V> IgniteInternalFuture<Set<KeyCacheObject>> enlistWrite( @@ -1981,7 +1993,8 @@ public abstract class IgniteTxLocalAdapter extends IgniteTxAdapter final GridCacheReturn ret, Collection<KeyCacheObject> enlisted, @Nullable Map<KeyCacheObject, GridCacheDrInfo> drPutMap, - @Nullable Map<KeyCacheObject, GridCacheVersion> drRmvMap + @Nullable Map<KeyCacheObject, GridCacheVersion> drRmvMap, + boolean skipStore ) { assert cached == null || keys.size() == 1; assert cached == null || F.first(keys).equals(cached.key()); @@ -2131,7 +2144,8 @@ public abstract class IgniteTxLocalAdapter extends IgniteTxAdapter false, -1L, -1L, - null); + null, + skipStore); txEntry.markValid(); } @@ -2155,7 +2169,8 @@ public abstract class IgniteTxLocalAdapter extends IgniteTxAdapter true, drTtl, drExpireTime, - drVer); + drVer, + skipStore); if (!implicit() && readCommitted()) cacheCtx.evicts().touch(entry, topologyVersion()); @@ -2247,7 +2262,8 @@ public abstract class IgniteTxLocalAdapter extends IgniteTxAdapter true, drTtl, drExpireTime, - drVer); + drVer, + skipStore); enlisted.add(cacheKey); @@ -2273,7 +2289,7 @@ public abstract class IgniteTxLocalAdapter extends IgniteTxAdapter if (missedForLoad != null) { IgniteInternalFuture<Boolean> fut = loadMissing( cacheCtx, - /*read through*/cacheCtx.config().isLoadPreviousValue() && !cacheCtx.skipStore(), + /*read through*/cacheCtx.config().isLoadPreviousValue() && !skipStore, /*async*/true, missedForLoad, deserializePortables(cacheCtx), @@ -2602,7 +2618,8 @@ public abstract class IgniteTxLocalAdapter extends IgniteTxAdapter ret, enlisted, drMap, - null); + null, + opCtx != null && opCtx.skipStore()); if (pessimistic() && !groupLock()) { // Loose all skipped. @@ -2793,11 +2810,11 @@ public abstract class IgniteTxLocalAdapter extends IgniteTxAdapter try { Collection<KeyCacheObject> enlisted = new ArrayList<>(); + CacheOperationContext opCtx = cacheCtx.operationContextPerCall(); + ExpiryPolicy plc; if (!F.isEmpty(filter)) { - CacheOperationContext opCtx = cacheCtx.operationContextPerCall(); - plc = opCtx != null ? opCtx.expiry() : null; } else @@ -2818,7 +2835,8 @@ public abstract class IgniteTxLocalAdapter extends IgniteTxAdapter ret, enlisted, null, - drMap + drMap, + opCtx != null && opCtx.skipStore() ); if (log.isDebugEnabled()) @@ -3022,7 +3040,8 @@ public abstract class IgniteTxLocalAdapter extends IgniteTxAdapter ret, enlisted, null, - null + null, + cacheCtx.skipStore() ).get(); // No keys should be skipped with empty filter. @@ -3146,6 +3165,7 @@ public abstract class IgniteTxLocalAdapter extends IgniteTxAdapter * @param drTtl DR TTL (if any). * @param drExpireTime DR expire time (if any). * @param drVer DR version. + * @param skipStore Skip store flag. * @return Transaction entry. */ protected final IgniteTxEntry addEntry(GridCacheOperation op, @@ -3158,7 +3178,8 @@ public abstract class IgniteTxLocalAdapter extends IgniteTxAdapter boolean filtersSet, long drTtl, long drExpireTime, - @Nullable GridCacheVersion drVer) { + @Nullable GridCacheVersion drVer, + boolean skipStore) { assert invokeArgs == null || op == TRANSFORM; IgniteTxKey key = entry.txKey(); @@ -3228,7 +3249,7 @@ public abstract class IgniteTxLocalAdapter extends IgniteTxAdapter entry, filter, drVer, - entry.context().skipStore()); + skipStore); txEntry.conflictExpireTime(drExpireTime); http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/63c32301/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxLocalEx.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxLocalEx.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxLocalEx.java index 9f1a70e..61041e1 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxLocalEx.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxLocalEx.java @@ -70,6 +70,7 @@ public interface IgniteTxLocalEx extends IgniteInternalTx { * @param deserializePortable Deserialize portable flag. * @param skipVals Skip values flag. * @param keepCacheObjects Keep cache objects + * @param skipStore Skip store flag. * @return Future for this get. */ public <K, V> IgniteInternalFuture<Map<K, V>> getAllAsync( @@ -78,7 +79,8 @@ public interface IgniteTxLocalEx extends IgniteInternalTx { @Nullable GridCacheEntryEx cached, boolean deserializePortable, boolean skipVals, - boolean keepCacheObjects); + boolean keepCacheObjects, + boolean skipStore); /** * @param cacheCtx Cache context. http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/63c32301/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheAbstractFullApiSelfTest.java ---------------------------------------------------------------------- diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheAbstractFullApiSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheAbstractFullApiSelfTest.java index 8e7e6b1..88c1cd8 100644 --- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheAbstractFullApiSelfTest.java +++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheAbstractFullApiSelfTest.java @@ -4534,6 +4534,8 @@ public abstract class GridCacheAbstractFullApiSelfTest extends GridCacheAbstract tx.commit(); + assert map.size() == 0; + // cacheSkipStore putAll(..)/removeAll(..) check tx = txs.txStart(txConcurrency, txIsolation);
