http://git-wip-us.apache.org/repos/asf/ignite/blob/3ff71fd7/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 2c7bf8a..758f82c 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 @@ -66,9 +66,7 @@ import org.apache.ignite.internal.processors.dr.GridDrType; import org.apache.ignite.internal.transactions.IgniteTxHeuristicCheckedException; import org.apache.ignite.internal.transactions.IgniteTxRollbackCheckedException; import org.apache.ignite.internal.transactions.IgniteTxTimeoutCheckedException; -import org.apache.ignite.internal.util.F0; import org.apache.ignite.internal.util.GridLeanMap; -import org.apache.ignite.internal.util.GridLeanSet; import org.apache.ignite.internal.util.future.GridEmbeddedFuture; import org.apache.ignite.internal.util.future.GridFinishedFuture; import org.apache.ignite.internal.util.future.GridFutureAdapter; @@ -123,18 +121,6 @@ public abstract class IgniteTxLocalAdapter extends IgniteTxAdapter /** */ private static final long serialVersionUID = 0L; - /** Per-transaction read map. */ - @GridToStringInclude - protected Map<IgniteTxKey, IgniteTxEntry> txMap; - - /** Read view on transaction map. */ - @GridToStringExclude - protected IgniteTxMap readView; - - /** Write view on transaction map. */ - @GridToStringExclude - protected IgniteTxMap writeView; - /** Minimal version encountered (either explicit lock or XID of this transaction). */ protected GridCacheVersion minVer; @@ -156,9 +142,6 @@ public abstract class IgniteTxLocalAdapter extends IgniteTxAdapter /** Commit error. */ protected AtomicReference<Throwable> commitErr = new AtomicReference<>(); - /** Active cache IDs. */ - protected Set<Integer> activeCacheIds = new HashSet<>(); - /** Need return value. */ protected boolean needRetVal; @@ -168,6 +151,10 @@ public abstract class IgniteTxLocalAdapter extends IgniteTxAdapter /** Flag indicating whether deployment is enabled for caches from this transaction or not. */ private boolean depEnabled; + /** */ + @GridToStringInclude + protected IgniteTxLocalState txState; + /** * Empty constructor required for {@link Externalizable}. */ @@ -209,7 +196,6 @@ public abstract class IgniteTxLocalAdapter extends IgniteTxAdapter cctx, xidVer, implicit, - implicitSingle, /*local*/true, sys, plc, @@ -225,6 +211,13 @@ public abstract class IgniteTxLocalAdapter extends IgniteTxAdapter ); minVer = xidVer; + + txState = implicitSingle ? new IgniteTxImplicitSingleStateImpl() : new IgniteTxStateImpl(); + } + + /** {@inheritDoc} */ + @Override public IgniteTxState txState() { + return txState; } /** @@ -246,7 +239,7 @@ public abstract class IgniteTxLocalAdapter extends IgniteTxAdapter /** {@inheritDoc} */ @Override public boolean empty() { - return txMap.isEmpty(); + return txState.empty(); } /** {@inheritDoc} */ @@ -267,16 +260,8 @@ public abstract class IgniteTxLocalAdapter extends IgniteTxAdapter /** {@inheritDoc} */ @Override public boolean onOwnerChanged(GridCacheEntryEx entry, GridCacheMvccCandidate owner) { assert false; - return false; - } - /** - * Gets collection of active cache IDs for this transaction. - * - * @return Collection of active cache IDs. - */ - @Override public Collection<Integer> activeCacheIds() { - return activeCacheIds; + return false; } /** {@inheritDoc} */ @@ -284,77 +269,70 @@ public abstract class IgniteTxLocalAdapter extends IgniteTxAdapter return depEnabled; } + /** + * @param depEnabled Flag indicating whether deployment is enabled for caches from this transaction or not. + */ + public void activeCachesDeploymentEnabled(boolean depEnabled) { + this.depEnabled = depEnabled; + } + /** {@inheritDoc} */ @Override public boolean isStarted() { - return txMap != null; + return txState.initialized(); } /** {@inheritDoc} */ @Override public boolean hasWriteKey(IgniteTxKey key) { - return writeView.containsKey(key); + return txState.hasWriteKey(key); } /** * @return Transaction read set. */ @Override public Set<IgniteTxKey> readSet() { - return txMap == null ? Collections.<IgniteTxKey>emptySet() : readView.keySet(); + return txState.readSet(); } /** * @return Transaction write set. */ @Override public Set<IgniteTxKey> writeSet() { - return txMap == null ? Collections.<IgniteTxKey>emptySet() : writeView.keySet(); - } - - /** {@inheritDoc} */ - @Override public boolean removed(IgniteTxKey key) { - if (txMap == null) - return false; - - IgniteTxEntry e = txMap.get(key); - - return e != null && e.op() == DELETE; + return txState.writeSet(); } /** {@inheritDoc} */ @Override public Map<IgniteTxKey, IgniteTxEntry> readMap() { - return readView == null ? Collections.<IgniteTxKey, IgniteTxEntry>emptyMap() : readView; + return txState.readMap(); } /** {@inheritDoc} */ @Override public Map<IgniteTxKey, IgniteTxEntry> writeMap() { - return writeView == null ? Collections.<IgniteTxKey, IgniteTxEntry>emptyMap() : writeView; + return txState.writeMap(); } /** {@inheritDoc} */ @Override public Collection<IgniteTxEntry> allEntries() { - return txMap == null ? Collections.<IgniteTxEntry>emptySet() : txMap.values(); + return txState.allEntries(); } /** {@inheritDoc} */ @Override public Collection<IgniteTxEntry> readEntries() { - return readView == null ? Collections.<IgniteTxEntry>emptyList() : readView.values(); + return txState.readEntries(); } /** {@inheritDoc} */ @Override public Collection<IgniteTxEntry> writeEntries() { - return writeView == null ? Collections.<IgniteTxEntry>emptyList() : writeView.values(); + return txState.writeEntries(); } /** {@inheritDoc} */ @Nullable @Override public IgniteTxEntry entry(IgniteTxKey key) { - return txMap == null ? null : txMap.get(key); + return txState.entry(key); } /** {@inheritDoc} */ @Override public void seal() { - if (readView != null) - readView.seal(); - - if (writeView != null) - writeView.seal(); + txState.seal(); } /** {@inheritDoc} */ @@ -409,7 +387,7 @@ public abstract class IgniteTxLocalAdapter extends IgniteTxAdapter KeyCacheObject key, CacheEntryPredicate[] filter ) throws GridCacheFilterFailedException { - IgniteTxEntry e = txMap == null ? null : txMap.get(cacheCtx.txKey(key)); + IgniteTxEntry e = entry(cacheCtx.txKey(key)); if (e != null) { // We should look at tx entry previous value. If this is a user peek then previous @@ -652,7 +630,7 @@ public abstract class IgniteTxLocalAdapter extends IgniteTxAdapter if (!storeEnabled() || internal()) return; - Collection<CacheStoreManager> stores = stores(); + Collection<CacheStoreManager> stores = txState.stores(cctx); if (stores == null || stores.isEmpty()) return; @@ -854,7 +832,9 @@ public abstract class IgniteTxLocalAdapter extends IgniteTxAdapter checkValid(); - boolean empty = F.isEmpty(near() ? txMap : writeMap()); + Collection<IgniteTxEntry> commitEntries = near() ? allEntries() : writeEntries(); + + boolean empty = F.isEmpty(commitEntries); // Register this transaction as completed prior to write-phase to // ensure proper lock ordering for removed entries. @@ -874,7 +854,7 @@ public abstract class IgniteTxLocalAdapter extends IgniteTxAdapter /* * Commit to cache. Note that for 'near' transaction we loop through all the entries. */ - for (IgniteTxEntry txEntry : (near() ? allEntries() : writeEntries())) { + for (IgniteTxEntry txEntry : commitEntries) { GridCacheContext cacheCtx = txEntry.context(); GridDrType drType = cacheCtx.isDrEnabled() ? DR_PRIMARY : DR_NONE; @@ -1282,7 +1262,7 @@ public abstract class IgniteTxLocalAdapter extends IgniteTxAdapter cctx.tm().rollbackTx(this); if (!internal()) { - Collection<CacheStoreManager> stores = stores(); + Collection<CacheStoreManager> stores = txState.stores(cctx); if (stores != null && !stores.isEmpty()) { assert isWriteToStoreFromDhtValid(stores) : @@ -1582,25 +1562,6 @@ public abstract class IgniteTxLocalAdapter extends IgniteTxAdapter } /** - * Adds skipped key. - * - * @param skipped Skipped set (possibly {@code null}). - * @param key Key to add. - * @return Skipped set. - */ - private Set<KeyCacheObject> skip(Set<KeyCacheObject> skipped, KeyCacheObject key) { - if (skipped == null) - skipped = new GridLeanSet<>(); - - skipped.add(key); - - if (log.isDebugEnabled()) - log.debug("Added key to skipped set: " + key); - - return skipped; - } - - /** * Loads all missed keys for * {@link #getAllAsync(GridCacheContext, Collection, boolean, boolean, boolean, boolean)} method. * @@ -1954,6 +1915,24 @@ public abstract class IgniteTxLocalAdapter extends IgniteTxAdapter } /** {@inheritDoc} */ + @Override public <K, V> IgniteInternalFuture<GridCacheReturn> putAsync( + GridCacheContext cacheCtx, + K key, + V val, + boolean retval, + CacheEntryPredicate[] filter) { + return putAsync0(cacheCtx, key, val, null, null, retval, filter); + } + + /** {@inheritDoc} */ + @Override public <K, V> IgniteInternalFuture<GridCacheReturn> invokeAsync(GridCacheContext cacheCtx, + K key, + EntryProcessor<K, V, Object> entryProcessor, + Object... invokeArgs) { + return (IgniteInternalFuture)putAsync0(cacheCtx, key, null, entryProcessor, invokeArgs, true, null); + } + + /** {@inheritDoc} */ @Override public IgniteInternalFuture<?> putAllDrAsync( GridCacheContext cacheCtx, Map<KeyCacheObject, GridCacheDrInfo> drMap @@ -2009,12 +1988,88 @@ public abstract class IgniteTxLocalAdapter extends IgniteTxAdapter } /** + * @param cacheCtx Cache context. + * @param cacheKey Key to enlist. + * @param val Value. + * @param expiryPlc Explicitly specified expiry policy for entry. + * @param entryProcessor Entry processor (for invoke operation). + * @param invokeArgs Optional arguments for EntryProcessor. + * @param retval Flag indicating whether a value should be returned. + * @param lockOnly If {@code true}, then entry will be enlisted as noop. + * @param filter User filters. + * @param ret Return value. + * @param skipStore Skip store flag. + * @param singleRmv {@code True} for single key remove operation ({@link Cache#remove(Object)}. + * @return Future for entry values loading. + */ + private <K, V> IgniteInternalFuture<Void> enlistWrite( + final GridCacheContext cacheCtx, + KeyCacheObject cacheKey, + Object val, + @Nullable ExpiryPolicy expiryPlc, + @Nullable EntryProcessor<K, V, Object> entryProcessor, + @Nullable Object[] invokeArgs, + final boolean retval, + boolean lockOnly, + final CacheEntryPredicate[] filter, + final GridCacheReturn ret, + boolean skipStore, + final boolean singleRmv) { + try { + addActiveCache(cacheCtx); + + final boolean hasFilters = !F.isEmptyOrNulls(filter) && !F.isAlwaysTrue(filter); + final boolean needVal = singleRmv || retval || hasFilters; + final boolean needReadVer = needVal && (serializable() && optimistic()); + + if (entryProcessor != null) + transform = true; + + boolean loadMissed = enlistWriteEntry(cacheCtx, + cacheKey, + val, + entryProcessor, + invokeArgs, + expiryPlc, + retval, + lockOnly, + filter, + /*drVer*/null, + /*drTtl*/-1L, + /*drExpireTime*/-1L, + ret, + /*enlisted*/null, + skipStore, + singleRmv, + hasFilters, + needVal, + needReadVer); + + if (loadMissed) { + return loadMissing(cacheCtx, + Collections.singleton(cacheKey), + filter, + ret, + needReadVer, + singleRmv, + hasFilters, + skipStore, + retval); + } + + return new GridFinishedFuture<>(); + } + catch (IgniteCheckedException e) { + return new GridFinishedFuture<>(e); + } + } + + /** * Internal routine for <tt>putAll(..)</tt> * * @param cacheCtx Cache context. * @param keys Keys to enlist. * @param expiryPlc Explicitly specified expiry policy for entry. - * @param implicit Implicit flag. * @param lookup Value lookup map ({@code null} for remove). * @param invokeMap Map with entry processors for invoke operation. * @param invokeArgs Optional arguments for EntryProcessor. @@ -2027,13 +2082,12 @@ public abstract class IgniteTxLocalAdapter extends IgniteTxAdapter * @param drRmvMap DR remove map (optional). * @param skipStore Skip store flag. * @param singleRmv {@code True} for single key remove operation ({@link Cache#remove(Object)}. - * @return Future with skipped keys (the ones that didn't pass filter for pessimistic transactions). + * @return Future for missing values loading. */ - private <K, V> IgniteInternalFuture<Set<KeyCacheObject>> enlistWrite( + private <K, V> IgniteInternalFuture<Void> enlistWrite( final GridCacheContext cacheCtx, Collection<?> keys, @Nullable ExpiryPolicy expiryPlc, - boolean implicit, @Nullable Map<?, ?> lookup, @Nullable Map<?, EntryProcessor<K, V, Object>> invokeMap, @Nullable Object[] invokeArgs, @@ -2056,8 +2110,6 @@ public abstract class IgniteTxLocalAdapter extends IgniteTxAdapter return new GridFinishedFuture<>(e); } - Set<KeyCacheObject> skipped = null; - boolean rmv = lookup == null && invokeMap == null; Set<KeyCacheObject> missedForLoad = null; @@ -2115,345 +2167,441 @@ public abstract class IgniteTxLocalAdapter extends IgniteTxAdapter KeyCacheObject cacheKey = cacheCtx.toCacheKeyObject(key); - IgniteTxKey txKey = cacheCtx.txKey(cacheKey); - - IgniteTxEntry txEntry = entry(txKey); - - // First time access. - if (txEntry == null) { - while (true) { - GridCacheEntryEx entry = entryEx(cacheCtx, txKey, topologyVersion()); - - try { - entry.unswap(false); - - // Check if lock is being explicitly acquired by the same thread. - if (!implicit && cctx.kernalContext().config().isCacheSanityCheckEnabled() && - entry.lockedByThread(threadId, xidVer)) - throw new IgniteCheckedException("Cannot access key within transaction if lock is " + - "externally held [key=" + key + ", entry=" + entry + ", xidVer=" + xidVer + - ", threadId=" + threadId + - ", locNodeId=" + cctx.localNodeId() + ']'); - - CacheObject old = null; - GridCacheVersion readVer = null; + boolean loadMissed = enlistWriteEntry(cacheCtx, + cacheKey, + val, + entryProcessor, + invokeArgs, + expiryPlc, + retval, + lockOnly, + filter, + drVer, + drTtl, + drExpireTime, + ret, + enlisted, + skipStore, + singleRmv, + hasFilters, + needVal, + needReadVer); + + if (loadMissed) { + if (missedForLoad == null) + missedForLoad = new HashSet<>(); + + missedForLoad.add(cacheKey); + } + } - if (optimistic() && !implicit()) { - try { - if (needReadVer) { - T2<CacheObject, GridCacheVersion> res = primaryLocal(entry) ? - entry.innerGetVersioned(this, - /*swap*/false, - /*unmarshal*/retval, - /*metrics*/retval, - /*events*/retval, - CU.subjectId(this, cctx), - entryProcessor, - resolveTaskName(), - null) : null; + if (missedForLoad != null) { + return loadMissing(cacheCtx, + missedForLoad, + filter, + ret, + needReadVer, + singleRmv, + hasFilters, + skipStore, + retval); + } - if (res != null) { - old = res.get1(); - readVer = res.get2(); - } - } - else { - old = entry.innerGet(this, - /*swap*/false, - /*read-through*/false, - /*fail-fast*/false, - /*unmarshal*/retval, - /*metrics*/retval, - /*events*/retval, - /*temporary*/false, - CU.subjectId(this, cctx), - entryProcessor, - resolveTaskName(), - null); - } - } - catch (ClusterTopologyCheckedException e) { - entry.context().evicts().touch(entry, topologyVersion()); + return new GridFinishedFuture<>(); + } + catch (IgniteCheckedException e) { + return new GridFinishedFuture<>(e); + } + } - throw e; - } - } - else - old = retval ? entry.rawGetOrUnmarshal(false) : entry.rawGet(); + /** + * @param cacheCtx Cache context. + * @param keys Keys to load. + * @param ret Return value. + * @param needReadVer Read version flag. + * @param singleRmv {@code True} for single remove operation. + * @param hasFilters {@code True} if filters not empty. + * @param skipStore Skip store flag. + * @param retval Return value flag. + * @return Load future. + */ + private IgniteInternalFuture<Void> loadMissing( + final GridCacheContext cacheCtx, + final Set<KeyCacheObject> keys, + final CacheEntryPredicate[] filter, + final GridCacheReturn ret, + final boolean needReadVer, + final boolean singleRmv, + final boolean hasFilters, + final boolean skipStore, + final boolean retval) { + GridInClosure3<KeyCacheObject, Object, GridCacheVersion> c = + new GridInClosure3<KeyCacheObject, Object, GridCacheVersion>() { + @Override public void apply(KeyCacheObject key, + @Nullable Object val, + @Nullable GridCacheVersion loadVer) { + if (log.isDebugEnabled()) + log.debug("Loaded value from remote node [key=" + key + ", val=" + val + ']'); - if (old != null && hasFilters && !filter(entry.context(), cacheKey, old, filter)) { - skipped = skip(skipped, cacheKey); + IgniteTxEntry e = entry(new IgniteTxKey(key, cacheCtx.cacheId())); - ret.set(cacheCtx, old, false); + assert e != null; - if (!readCommitted()) { - // Enlist failed filters as reads for non-read-committed mode, - // so future ops will get the same values. - txEntry = addEntry(READ, - old, - null, - null, - entry, - null, - CU.empty0(), - false, - -1L, - -1L, - null, - skipStore); + if (needReadVer) { + assert loadVer != null; - txEntry.markValid(); + e.serializableReadVersion(singleRmv && val != null ? SER_READ_NOT_EMPTY_VER : loadVer); + } - if (needReadVer) { - assert readVer != null; + if (singleRmv) { + assert !hasFilters && !retval; + assert val == null || Boolean.TRUE.equals(val) : val; - txEntry.serializableReadVersion(singleRmv ? SER_READ_NOT_EMPTY_VER : readVer); - } - } + ret.set(cacheCtx, null, val != null); + } + else { + CacheObject cacheVal = cacheCtx.toCacheObject(val); - if (readCommitted()) - cacheCtx.evicts().touch(entry, topologyVersion()); + if (e.op() == TRANSFORM) { + GridCacheVersion ver; - break; // While. + try { + ver = e.cached().version(); } + catch (GridCacheEntryRemovedException ex) { + assert optimistic() : e; - final GridCacheOperation op = lockOnly ? NOOP : rmv ? DELETE : - entryProcessor != null ? TRANSFORM : old != null ? UPDATE : CREATE; + if (log.isDebugEnabled()) + log.debug("Failed to get entry version: [msg=" + ex.getMessage() + ']'); - txEntry = addEntry(op, - cacheCtx.toCacheObject(val), - entryProcessor, - invokeArgs, - entry, - expiryPlc, - filter, - true, - drTtl, - drExpireTime, - drVer, - skipStore); + ver = null; + } - if (!implicit() && readCommitted() && !cacheCtx.offheapTiered()) - cacheCtx.evicts().touch(entry, topologyVersion()); + addInvokeResult(e, cacheVal, ret, ver); + } + else { + boolean success = !hasFilters || isAll(e.context(), key, cacheVal, filter); - enlisted.add(cacheKey); + ret.set(cacheCtx, cacheVal, success); + } + } + } + }; - if (!pessimistic() && !implicit()) { - txEntry.markValid(); + return loadMissing( + cacheCtx, + /*read through*/cacheCtx.config().isLoadPreviousValue() && !skipStore, + /*async*/true, + keys, + /*skipVals*/singleRmv, + needReadVer, + c); + } - if (old == null) { - if (needVal) { - if (missedForLoad == null) - missedForLoad = new HashSet<>(); + /** + * @param cacheCtx Cache context. + * @param cacheKey Key. + * @param val Value. + * @param entryProcessor Entry processor. + * @param invokeArgs Optional arguments for EntryProcessor. + * @param expiryPlc Explicitly specified expiry policy for entry. + * @param retval Return value flag. + * @param lockOnly + * @param filter Filter. + * @param drVer DR version. + * @param drTtl DR ttl. + * @param drExpireTime DR expire time. + * @param ret Return value. + * @param enlisted Enlisted keys collection. + * @param skipStore Skip store flag. + * @param singleRmv {@code True} for single remove operation. + * @param hasFilters {@code True} if filters not empty. + * @param needVal {@code True} if value is needed. + * @param needReadVer {@code True} if need read entry version. + * @return {@code True} if entry value should be loaded. + * @throws IgniteCheckedException If failed. + */ + private boolean enlistWriteEntry(GridCacheContext cacheCtx, + final KeyCacheObject cacheKey, + final @Nullable Object val, + final @Nullable EntryProcessor<?, ?, ?> entryProcessor, + final @Nullable Object[] invokeArgs, + final @Nullable ExpiryPolicy expiryPlc, + final boolean retval, + final boolean lockOnly, + final CacheEntryPredicate[] filter, + final GridCacheVersion drVer, + final long drTtl, + long drExpireTime, + final GridCacheReturn ret, + @Nullable final Collection<KeyCacheObject> enlisted, + boolean skipStore, + boolean singleRmv, + boolean hasFilters, + final boolean needVal, + boolean needReadVer + ) throws IgniteCheckedException { + boolean loadMissed = false; - missedForLoad.add(cacheKey); - } - else { - assert !implicit() || !transform : this; - assert txEntry.op() != TRANSFORM : txEntry; + final boolean rmv = val == null && entryProcessor == null; - if (retval) - ret.set(cacheCtx, null, true); - else - ret.success(true); - } - } - else { - if (needReadVer) { - assert readVer != null; + IgniteTxKey txKey = cacheCtx.txKey(cacheKey); - txEntry.serializableReadVersion(singleRmv ? SER_READ_NOT_EMPTY_VER : readVer); - } + IgniteTxEntry txEntry = entry(txKey); - if (retval && !transform) - ret.set(cacheCtx, old, true); - else { - if (txEntry.op() == TRANSFORM) { - GridCacheVersion ver; + // First time access. + if (txEntry == null) { + while (true) { + GridCacheEntryEx entry = entryEx(cacheCtx, txKey, topologyVersion()); - try { - ver = entry.version(); - } - catch (GridCacheEntryRemovedException ex) { - assert optimistic() : txEntry; + try { + entry.unswap(false); + + // Check if lock is being explicitly acquired by the same thread. + if (!implicit && cctx.kernalContext().config().isCacheSanityCheckEnabled() && + entry.lockedByThread(threadId, xidVer)) { + throw new IgniteCheckedException("Cannot access key within transaction if lock is " + + "externally held [key=" + CU.value(cacheKey, cacheCtx, false) + + ", entry=" + entry + + ", xidVer=" + xidVer + + ", threadId=" + threadId + + ", locNodeId=" + cctx.localNodeId() + ']'); + } - if (log.isDebugEnabled()) - log.debug("Failed to get entry version " + - "[err=" + ex.getMessage() + ']'); + CacheObject old = null; + GridCacheVersion readVer = null; - ver = null; - } + if (optimistic() && !implicit()) { + try { + if (needReadVer) { + T2<CacheObject, GridCacheVersion> res = primaryLocal(entry) ? + entry.innerGetVersioned(this, + /*swap*/false, + /*unmarshal*/retval, + /*metrics*/retval, + /*events*/retval, + CU.subjectId(this, cctx), + entryProcessor, + resolveTaskName(), + null) : null; - addInvokeResult(txEntry, old, ret, ver); - } - else - ret.success(true); - } + if (res != null) { + old = res.get1(); + readVer = res.get2(); } } - // Pessimistic. else { - if (retval && !transform) - ret.set(cacheCtx, old, true); - else - ret.success(true); + old = entry.innerGet(this, + /*swap*/false, + /*read-through*/false, + /*fail-fast*/false, + /*unmarshal*/retval, + /*metrics*/retval, + /*events*/retval, + /*temporary*/false, + CU.subjectId(this, cctx), + entryProcessor, + resolveTaskName(), + null); } - - break; // While. } - catch (GridCacheEntryRemovedException ignore) { - if (log.isDebugEnabled()) - log.debug("Got removed entry in transaction putAll0 method: " + entry); + catch (ClusterTopologyCheckedException e) { + entry.context().evicts().touch(entry, topologyVersion()); + + throw e; } } - } - else { - if (entryProcessor == null && txEntry.op() == TRANSFORM) - throw new IgniteCheckedException("Failed to enlist write value for key (cannot have update value in " + - "transaction after EntryProcessor is applied): " + key); + else + old = retval ? entry.rawGetOrUnmarshal(false) : entry.rawGet(); - GridCacheEntryEx entry = txEntry.cached(); + if (old != null && hasFilters && !filter(entry.context(), cacheKey, old, filter)) { + ret.set(cacheCtx, old, false); - CacheObject v = txEntry.value(); - - boolean del = txEntry.op() == DELETE && rmv; + if (!readCommitted()) { + // Enlist failed filters as reads for non-read-committed mode, + // so future ops will get the same values. + txEntry = addEntry(READ, + old, + null, + null, + entry, + null, + CU.empty0(), + false, + -1L, + -1L, + null, + skipStore); - if (!del) { - if (hasFilters && !filter(entry.context(), cacheKey, v, filter)) { - skipped = skip(skipped, cacheKey); + txEntry.markValid(); - ret.set(cacheCtx, v, false); + if (needReadVer) { + assert readVer != null; - continue; + txEntry.serializableReadVersion(singleRmv ? SER_READ_NOT_EMPTY_VER : readVer); + } } - GridCacheOperation op = rmv ? DELETE : entryProcessor != null ? TRANSFORM : - v != null ? UPDATE : CREATE; + if (readCommitted()) + cacheCtx.evicts().touch(entry, topologyVersion()); - txEntry = addEntry(op, - cacheCtx.toCacheObject(val), - entryProcessor, - invokeArgs, - entry, - expiryPlc, - filter, - true, - drTtl, - drExpireTime, - drVer, - skipStore); + break; // While. + } + + final GridCacheOperation op = lockOnly ? NOOP : rmv ? DELETE : + entryProcessor != null ? TRANSFORM : old != null ? UPDATE : CREATE; + + txEntry = addEntry(op, + cacheCtx.toCacheObject(val), + entryProcessor, + invokeArgs, + entry, + expiryPlc, + filter, + true, + drTtl, + drExpireTime, + drVer, + skipStore); + if (!implicit() && readCommitted() && !cacheCtx.offheapTiered()) + cacheCtx.evicts().touch(entry, topologyVersion()); + + if (enlisted != null) enlisted.add(cacheKey); - if (txEntry.op() == TRANSFORM) { - GridCacheVersion ver; + if (!pessimistic() && !implicit()) { + txEntry.markValid(); - try { - ver = entry.version(); - } - catch (GridCacheEntryRemovedException e) { - assert optimistic() : txEntry; + if (old == null) { + if (needVal) + loadMissed = true; + else { + assert !implicit() || !transform : this; + assert txEntry.op() != TRANSFORM : txEntry; - if (log.isDebugEnabled()) - log.debug("Failed to get entry version: [msg=" + e.getMessage() + ']'); + if (retval) + ret.set(cacheCtx, null, true); + else + ret.success(true); + } + } + else { + if (needReadVer) { + assert readVer != null; - ver = null; + txEntry.serializableReadVersion(singleRmv ? SER_READ_NOT_EMPTY_VER : readVer); } - addInvokeResult(txEntry, txEntry.value(), ret, ver); - } - } + if (retval && !transform) + ret.set(cacheCtx, old, true); + else { + if (txEntry.op() == TRANSFORM) { + GridCacheVersion ver; - if (!pessimistic()) { - txEntry.markValid(); + try { + ver = entry.version(); + } + catch (GridCacheEntryRemovedException ex) { + assert optimistic() : txEntry; + + if (log.isDebugEnabled()) + log.debug("Failed to get entry version " + + "[err=" + ex.getMessage() + ']'); + + ver = null; + } + addInvokeResult(txEntry, old, ret, ver); + } + else + ret.success(true); + } + } + } + // Pessimistic. + else { if (retval && !transform) - ret.set(cacheCtx, v, true); + ret.set(cacheCtx, old, true); else ret.success(true); } - } + + break; // While. + } + catch (GridCacheEntryRemovedException ignore) { + if (log.isDebugEnabled()) + log.debug("Got removed entry in transaction putAll0 method: " + entry); + } } } - catch (IgniteCheckedException e) { - return new GridFinishedFuture<>(e); - } - - if (missedForLoad != null) { - final boolean skipVals = singleRmv; - - IgniteInternalFuture<Void> fut = loadMissing( - cacheCtx, - /*read through*/cacheCtx.config().isLoadPreviousValue() && !skipStore, - /*async*/true, - missedForLoad, - skipVals, - needReadVer, - new GridInClosure3<KeyCacheObject, Object, GridCacheVersion>() { - @Override public void apply(KeyCacheObject key, - @Nullable Object val, - @Nullable GridCacheVersion loadVer) { - if (log.isDebugEnabled()) - log.debug("Loaded value from remote node [key=" + key + ", val=" + val + ']'); + else { + if (entryProcessor == null && txEntry.op() == TRANSFORM) + throw new IgniteCheckedException("Failed to enlist write value for key (cannot have update value in " + + "transaction after EntryProcessor is applied): " + CU.value(cacheKey, cacheCtx, false)); - IgniteTxEntry e = entry(new IgniteTxKey(key, cacheCtx.cacheId())); + GridCacheEntryEx entry = txEntry.cached(); - assert e != null; + CacheObject v = txEntry.value(); - if (needReadVer) { - assert loadVer != null; + boolean del = txEntry.op() == DELETE && rmv; - e.serializableReadVersion(singleRmv && val != null ? SER_READ_NOT_EMPTY_VER : loadVer); - } + if (!del) { + if (hasFilters && !filter(entry.context(), cacheKey, v, filter)) { + ret.set(cacheCtx, v, false); - if (singleRmv) { - assert !hasFilters && !retval; - assert val == null || Boolean.TRUE.equals(val) : val; + return loadMissed; + } - ret.set(cacheCtx, null, val != null); - } - else { - CacheObject cacheVal = cacheCtx.toCacheObject(val); + GridCacheOperation op = rmv ? DELETE : entryProcessor != null ? TRANSFORM : + v != null ? UPDATE : CREATE; - if (e.op() == TRANSFORM) { - GridCacheVersion ver; + txEntry = addEntry(op, + cacheCtx.toCacheObject(val), + entryProcessor, + invokeArgs, + entry, + expiryPlc, + filter, + true, + drTtl, + drExpireTime, + drVer, + skipStore); - try { - ver = e.cached().version(); - } - catch (GridCacheEntryRemovedException ex) { - assert optimistic() : e; + if (enlisted != null) + enlisted.add(cacheKey); - if (log.isDebugEnabled()) - log.debug("Failed to get entry version: [msg=" + ex.getMessage() + ']'); + if (txEntry.op() == TRANSFORM) { + GridCacheVersion ver; - ver = null; - } + try { + ver = entry.version(); + } + catch (GridCacheEntryRemovedException e) { + assert optimistic() : txEntry; - addInvokeResult(e, cacheVal, ret, ver); - } - else { - boolean success = !hasFilters || isAll(e.context(), key, cacheVal, filter); + if (log.isDebugEnabled()) + log.debug("Failed to get entry version: [msg=" + e.getMessage() + ']'); - ret.set(cacheCtx, cacheVal, success); - } - } + ver = null; } - }); - return new GridEmbeddedFuture<>( - new C2<Void, Exception, Set<KeyCacheObject>>() { - @Override public Set<KeyCacheObject> apply(Void b, Exception e) { - if (e != null) - throw new GridClosureException(e); + addInvokeResult(txEntry, txEntry.value(), ret, ver); + } + } - return Collections.emptySet(); - } - }, fut - ); + if (!pessimistic()) { + txEntry.markValid(); + + if (retval && !transform) + ret.set(cacheCtx, v, true); + else + ret.success(true); + } } - return new GridFinishedFuture<>(skipped); + return loadMissed; } /** @@ -2486,22 +2634,19 @@ public abstract class IgniteTxLocalAdapter extends IgniteTxAdapter * * @param cacheCtx Context. * @param keys Keys. - * @param failed Collection of potentially failed keys (need to populate in this method). * @param ret Return value. * @param rmv {@code True} if remove. * @param retval Flag to return value or not. * @param read {@code True} if read. * @param accessTtl TTL for read operation. * @param filter Filter to check entries. - * @return Failed keys. * @throws IgniteCheckedException If error. * @param computeInvoke If {@code true} computes return value for invoke operation. */ @SuppressWarnings("unchecked") - protected Set<KeyCacheObject> postLockWrite( + protected final void postLockWrite( GridCacheContext cacheCtx, Iterable<KeyCacheObject> keys, - Set<KeyCacheObject> failed, GridCacheReturn ret, boolean rmv, boolean retval, @@ -2606,8 +2751,6 @@ public abstract class IgniteTxLocalAdapter extends IgniteTxAdapter log.debug("Filter passed in post lock for key: " + k); } else { - failed = skip(failed, k); - // Revert operation to previous. (if no - NOOP, so entry will be unlocked). txEntry.setAndMarkValid(txEntry.previousOperation(), cacheCtx.toCacheObject(ret.value())); txEntry.filters(CU.empty0()); @@ -2638,11 +2781,6 @@ public abstract class IgniteTxLocalAdapter extends IgniteTxAdapter } } } - - if (log.isDebugEnabled()) - log.debug("Entries that failed after lock filter check: " + failed); - - return failed; } /** @@ -2696,6 +2834,144 @@ public abstract class IgniteTxLocalAdapter extends IgniteTxAdapter } /** + * @param cacheCtx Cache context. + * @param retval Return value flag. + * @throws IgniteCheckedException If failed. + */ + private void beforePut(GridCacheContext cacheCtx, boolean retval) throws IgniteCheckedException { + checkUpdatesAllowed(cacheCtx); + + cacheCtx.checkSecurity(SecurityPermission.CACHE_PUT); + + if (retval) + needReturnValue(true); + + checkValid(); + + init(); + } + + /** + * Internal method for single update operation. + * + * @param cacheCtx Cache context. + * @param key Key. + * @param val Value. + * @param entryProcessor Entry processor. + * @param invokeArgs Optional arguments for EntryProcessor. + * @param retval Return value flag. + * @param filter Filter. + * @return Operation future. + */ + private <K, V> IgniteInternalFuture putAsync0( + final GridCacheContext cacheCtx, + K key, + @Nullable V val, + @Nullable EntryProcessor entryProcessor, + @Nullable final Object[] invokeArgs, + final boolean retval, + @Nullable final CacheEntryPredicate[] filter + ) { + assert key != null; + + try { + beforePut(cacheCtx, retval); + + final GridCacheReturn ret = new GridCacheReturn(localResult(), false); + + CacheOperationContext opCtx = cacheCtx.operationContextPerCall(); + + KeyCacheObject cacheKey = cacheCtx.toCacheKeyObject(key); + + final IgniteInternalFuture<Void> loadFut = enlistWrite( + cacheCtx, + cacheKey, + val, + opCtx != null ? opCtx.expiry() : null, + entryProcessor, + invokeArgs, + retval, + /*lockOnly*/false, + filter, + ret, + opCtx != null && opCtx.skipStore(), + /*singleRmv*/false); + + if (pessimistic()) { + assert loadFut == null || loadFut.isDone() : loadFut; + + final Collection<KeyCacheObject> enlisted = Collections.singleton(cacheKey); + + if (log.isDebugEnabled()) + log.debug("Before acquiring transaction lock for put on key: " + enlisted); + + IgniteInternalFuture<Boolean> fut = cacheCtx.cache().txLockAsync(enlisted, + lockTimeout(), + this, + false, + retval, + isolation, + isInvalidate(), + -1L); + + PLC1<GridCacheReturn> plc1 = new PLC1<GridCacheReturn>(ret) { + @Override public GridCacheReturn postLock(GridCacheReturn ret) + throws IgniteCheckedException + { + if (log.isDebugEnabled()) + log.debug("Acquired transaction lock for put on keys: " + enlisted); + + postLockWrite(cacheCtx, + enlisted, + ret, + /*remove*/false, + retval, + /*read*/false, + -1L, + filter, + /*computeInvoke*/true); + + return ret; + } + }; + + if (fut.isDone()) { + try { + return nonInterruptable(plc1.apply(fut.get(), null)); + } + catch (GridClosureException e) { + return new GridFinishedFuture<>(e.unwrap()); + } + catch (IgniteCheckedException e) { + try { + return nonInterruptable(plc1.apply(false, e)); + } + catch (Exception e1) { + return new GridFinishedFuture<>(e1); + } + } + } + else { + return nonInterruptable(new GridEmbeddedFuture<>( + fut, + plc1 + )); + } + } + else + return optimisticPutFuture(loadFut, ret); + } + catch (IgniteCheckedException e) { + return new GridFinishedFuture(e); + } + catch (RuntimeException e) { + onException(); + + throw e; + } + } + + /** * Internal method for all put and transform operations. Only one of {@code map}, {@code transformMap} * maps must be non-null. * @@ -2721,17 +2997,12 @@ public abstract class IgniteTxLocalAdapter extends IgniteTxAdapter assert filter == null || invokeMap == null; try { - checkUpdatesAllowed(cacheCtx); + beforePut(cacheCtx, retval); } catch (IgniteCheckedException e) { return new GridFinishedFuture(e); } - cacheCtx.checkSecurity(SecurityPermission.CACHE_PUT); - - if (retval) - needReturnValue(true); - // Cached entry may be passed only from entry wrapper. final Map<?, ?> map0; final Map<?, EntryProcessor<K, V, Object>> invokeMap0; @@ -2757,15 +3028,6 @@ public abstract class IgniteTxLocalAdapter extends IgniteTxAdapter assert map0 != null || invokeMap0 != null; - try { - checkValid(); - } - catch (IgniteCheckedException e) { - return new GridFinishedFuture<>(e); - } - - init(); - final GridCacheReturn ret = new GridCacheReturn(localResult(), false); if (F.isEmpty(map0) && F.isEmpty(invokeMap0)) { @@ -2783,15 +3045,14 @@ public abstract class IgniteTxLocalAdapter extends IgniteTxAdapter try { Set<?> keySet = map0 != null ? map0.keySet() : invokeMap0.keySet(); - Collection<KeyCacheObject> enlisted = new ArrayList<>(); + final Collection<KeyCacheObject> enlisted = new ArrayList<>(); CacheOperationContext opCtx = cacheCtx.operationContextPerCall(); - final IgniteInternalFuture<Set<KeyCacheObject>> loadFut = enlistWrite( + final IgniteInternalFuture<Void> loadFut = enlistWrite( cacheCtx, keySet, opCtx != null ? opCtx.expiry() : null, - implicit, map0, invokeMap0, invokeArgs, @@ -2806,15 +3067,12 @@ public abstract class IgniteTxLocalAdapter extends IgniteTxAdapter false); if (pessimistic()) { - // Loose all skipped. - final Set<KeyCacheObject> loaded = loadFut.get(); - - final Collection<KeyCacheObject> keys = F.view(enlisted, F0.notIn(loaded)); + assert loadFut == null || loadFut.isDone() : loadFut; if (log.isDebugEnabled()) - log.debug("Before acquiring transaction lock for put on keys: " + keys); + log.debug("Before acquiring transaction lock for put on keys: " + enlisted); - IgniteInternalFuture<Boolean> fut = cacheCtx.cache().txLockAsync(keys, + IgniteInternalFuture<Boolean> fut = cacheCtx.cache().txLockAsync(enlisted, lockTimeout(), this, false, @@ -2828,11 +3086,10 @@ public abstract class IgniteTxLocalAdapter extends IgniteTxAdapter throws IgniteCheckedException { if (log.isDebugEnabled()) - log.debug("Acquired transaction lock for put on keys: " + keys); + log.debug("Acquired transaction lock for put on keys: " + enlisted); postLockWrite(cacheCtx, - keys, - loaded, + enlisted, ret, /*remove*/false, retval, @@ -2861,64 +3118,79 @@ public abstract class IgniteTxLocalAdapter extends IgniteTxAdapter } } } - else + else { return nonInterruptable(new GridEmbeddedFuture<>( fut, plc1 )); + } } - else { - if (implicit()) { - // Should never load missing values for implicit transaction as values will be returned - // with prepare response, if required. - assert loadFut.isDone(); + else + return optimisticPutFuture(loadFut, ret); + } + catch (RuntimeException e) { + onException(); - try { - loadFut.get(); - } - catch (IgniteCheckedException e) { - return new GridFinishedFuture<>(e); - } + throw e; + } + } - return nonInterruptable(commitAsync().chain(new CX1<IgniteInternalFuture<IgniteInternalTx>, GridCacheReturn>() { - @Override public GridCacheReturn applyx(IgniteInternalFuture<IgniteInternalTx> txFut) throws IgniteCheckedException { - try { - txFut.get(); + /** + * @param loadFut Missing keys load future. + * @param ret Future result. + * @return Future. + */ + private IgniteInternalFuture optimisticPutFuture(IgniteInternalFuture<Void> loadFut, final GridCacheReturn ret) { + if (implicit()) { + // Should never load missing values for implicit transaction as values will be returned + // with prepare response, if required. + assert loadFut.isDone(); - return implicitRes; - } - catch (IgniteCheckedException | RuntimeException e) { - rollbackAsync(); + try { + loadFut.get(); + } + catch (IgniteCheckedException e) { + return new GridFinishedFuture<>(e); + } - throw e; - } + return nonInterruptable(commitAsync().chain( + new CX1<IgniteInternalFuture<IgniteInternalTx>, GridCacheReturn>() { + @Override public GridCacheReturn applyx(IgniteInternalFuture<IgniteInternalTx> txFut) + throws IgniteCheckedException { + try { + txFut.get(); + + return implicitRes; } - })); - } - else - return nonInterruptable(loadFut.chain(new CX1<IgniteInternalFuture<Set<KeyCacheObject>>, GridCacheReturn>() { - @Override public GridCacheReturn applyx(IgniteInternalFuture<Set<KeyCacheObject>> f) throws IgniteCheckedException { - f.get(); + catch (IgniteCheckedException | RuntimeException e) { + rollbackAsync(); - return ret; + throw e; } - })); - } + } + } + )); } - catch (RuntimeException e) { - for (IgniteTxEntry txEntry : txMap.values()) { - GridCacheEntryEx cached0 = txEntry.cached(); - - if (cached0 != null) - txEntry.context().evicts().touch(cached0, topologyVersion()); - } + else { + return nonInterruptable(loadFut.chain(new CX1<IgniteInternalFuture<Void>, GridCacheReturn>() { + @Override public GridCacheReturn applyx(IgniteInternalFuture<Void> f) throws IgniteCheckedException { + f.get(); - throw e; + return ret; + } + })); } - catch (IgniteCheckedException e) { - setRollbackOnly(); + } - return new GridFinishedFuture<>(e); + /** + * + */ + private void onException() { + for (IgniteTxEntry txEntry : allEntries()) { + GridCacheEntryEx cached0 = txEntry.cached(); + + if (cached0 != null) + txEntry.context().evicts().touch(cached0, topologyVersion()); } } @@ -2974,9 +3246,10 @@ public abstract class IgniteTxLocalAdapter extends IgniteTxAdapter assert keys0 != null; - if (log.isDebugEnabled()) + if (log.isDebugEnabled()) { log.debug("Called removeAllAsync(...) [tx=" + this + ", keys=" + keys0 + ", implicit=" + implicit + ", retval=" + retval + "]"); + } try { checkValid(); @@ -3002,140 +3275,131 @@ public abstract class IgniteTxLocalAdapter extends IgniteTxAdapter init(); - try { - Collection<KeyCacheObject> enlisted = new ArrayList<>(); - - CacheOperationContext opCtx = cacheCtx.operationContextPerCall(); + final Collection<KeyCacheObject> enlisted = new ArrayList<>(); - ExpiryPolicy plc; + CacheOperationContext opCtx = cacheCtx.operationContextPerCall(); - if (!F.isEmpty(filter)) - plc = opCtx != null ? opCtx.expiry() : null; - else - plc = null; + ExpiryPolicy plc; - final IgniteInternalFuture<Set<KeyCacheObject>> loadFut = enlistWrite( - cacheCtx, - keys0, - plc, - implicit, - /** lookup map */null, - /** invoke map */null, - /** invoke arguments */null, - retval, - /** lock only */false, - filter, - ret, - enlisted, - null, - drMap, - opCtx != null && opCtx.skipStore(), - singleRmv - ); + if (!F.isEmpty(filter)) + plc = opCtx != null ? opCtx.expiry() : null; + else + plc = null; - if (log.isDebugEnabled()) - log.debug("Remove keys: " + enlisted); + final IgniteInternalFuture<Void> loadFut = enlistWrite( + cacheCtx, + keys0, + plc, + /** lookup map */null, + /** invoke map */null, + /** invoke arguments */null, + retval, + /** lock only */false, + filter, + ret, + enlisted, + null, + drMap, + opCtx != null && opCtx.skipStore(), + singleRmv + ); - // Acquire locks only after having added operation to the write set. - // Otherwise, during rollback we will not know whether locks need - // to be rolled back. - if (pessimistic()) { - // Loose all skipped. - final Collection<KeyCacheObject> passedKeys = F.view(enlisted, F0.notIn(loadFut.get())); + if (log.isDebugEnabled()) + log.debug("Remove keys: " + enlisted); - if (log.isDebugEnabled()) - log.debug("Before acquiring transaction lock for remove on keys: " + passedKeys); + // Acquire locks only after having added operation to the write set. + // Otherwise, during rollback we will not know whether locks need + // to be rolled back. + if (pessimistic()) { + assert loadFut.isDone() : loadFut; - IgniteInternalFuture<Boolean> fut = cacheCtx.cache().txLockAsync(passedKeys, - lockTimeout(), - this, - false, - retval, - isolation, - isInvalidate(), - -1L); + if (log.isDebugEnabled()) + log.debug("Before acquiring transaction lock for remove on keys: " + enlisted); - PLC1<GridCacheReturn> plc1 = new PLC1<GridCacheReturn>(ret) { - @Override protected GridCacheReturn postLock(GridCacheReturn ret) - throws IgniteCheckedException - { - if (log.isDebugEnabled()) - log.debug("Acquired transaction lock for remove on keys: " + passedKeys); + IgniteInternalFuture<Boolean> fut = cacheCtx.cache().txLockAsync(enlisted, + lockTimeout(), + this, + false, + retval, + isolation, + isInvalidate(), + -1L); + + PLC1<GridCacheReturn> plc1 = new PLC1<GridCacheReturn>(ret) { + @Override protected GridCacheReturn postLock(GridCacheReturn ret) + throws IgniteCheckedException + { + if (log.isDebugEnabled()) + log.debug("Acquired transaction lock for remove on keys: " + enlisted); - postLockWrite(cacheCtx, - passedKeys, - loadFut.get(), - ret, + postLockWrite(cacheCtx, + enlisted, + ret, /*remove*/true, - retval, + retval, /*read*/false, - -1L, - filter, + -1L, + filter, /*computeInvoke*/false); - return ret; - } - }; + return ret; + } + }; - if (fut.isDone()) { + if (fut.isDone()) { + try { + return nonInterruptable(plc1.apply(fut.get(), null)); + } + catch (GridClosureException e) { + return new GridFinishedFuture<>(e.unwrap()); + } + catch (IgniteCheckedException e) { try { - return nonInterruptable(plc1.apply(fut.get(), null)); + return nonInterruptable(plc1.apply(false, e)); } - catch (GridClosureException e) { - return new GridFinishedFuture<>(e.unwrap()); - } - catch (IgniteCheckedException e) { - try { - return nonInterruptable(plc1.apply(false, e)); - } - catch (Exception e1) { - return new GridFinishedFuture<>(e1); - } + catch (Exception e1) { + return new GridFinishedFuture<>(e1); } } - else - return nonInterruptable(new GridEmbeddedFuture<>( - fut, - plc1 - )); } - else { - if (implicit()) { - // Should never load missing values for implicit transaction as values will be returned - // with prepare response, if required. - assert loadFut.isDone(); - - return nonInterruptable(commitAsync().chain(new CX1<IgniteInternalFuture<IgniteInternalTx>, GridCacheReturn>() { - @Override public GridCacheReturn applyx(IgniteInternalFuture<IgniteInternalTx> txFut) - throws IgniteCheckedException { - try { - txFut.get(); + else + return nonInterruptable(new GridEmbeddedFuture<>( + fut, + plc1 + )); + } + else { + if (implicit()) { + // Should never load missing values for implicit transaction as values will be returned + // with prepare response, if required. + assert loadFut.isDone(); - return implicitRes; - } - catch (IgniteCheckedException | RuntimeException e) { - rollbackAsync(); + return nonInterruptable(commitAsync().chain(new CX1<IgniteInternalFuture<IgniteInternalTx>, GridCacheReturn>() { + @Override public GridCacheReturn applyx(IgniteInternalFuture<IgniteInternalTx> txFut) + throws IgniteCheckedException { + try { + txFut.get(); - throw e; - } + return implicitRes; } - })); - } - else - return nonInterruptable(loadFut.chain(new CX1<IgniteInternalFuture<Set<KeyCacheObject>>, GridCacheReturn>() { - @Override public GridCacheReturn applyx(IgniteInternalFuture<Set<KeyCacheObject>> f) - throws IgniteCheckedException { - f.get(); + catch (IgniteCheckedException | RuntimeException e) { + rollbackAsync(); - return ret; + throw e; } - })); + } + })); } - } - catch (IgniteCheckedException e) { - setRollbackOnly(); + else { + return nonInterruptable(loadFut.chain(new CX1<IgniteInternalFuture<Void>, GridCacheReturn>() { + @Override public GridCacheReturn applyx(IgniteInternalFuture<Void> f) + throws IgniteCheckedException { + f.get(); - return new GridFinishedFuture<>(e); + return ret; + } + })); + } } } @@ -3169,16 +3433,8 @@ public abstract class IgniteTxLocalAdapter extends IgniteTxAdapter * @return {@code True} if transaction was successfully started. */ public boolean init() { - if (txMap == null) { - txMap = new LinkedHashMap<>(txSize > 0 ? txSize : 16, 1.0f); + return !txState.init(txSize) || cctx.tm().onStarted(this); - readView = new IgniteTxMap(txMap, CU.reads()); - writeView = new IgniteTxMap(txMap, CU.writes()); - - return cctx.tm().onStarted(this); - } - - return true; } /** @@ -3188,38 +3444,8 @@ public abstract class IgniteTxLocalAdapter extends IgniteTxAdapter * @throws IgniteCheckedException If caches already enlisted in this transaction are not compatible with given * cache (e.g. they have different stores). */ - protected void addActiveCache(GridCacheContext cacheCtx) throws IgniteCheckedException { - int cacheId = cacheCtx.cacheId(); - - // Check if we can enlist new cache to transaction. - if (!activeCacheIds.contains(cacheId)) { - String err = cctx.verifyTxCompatibility(this, activeCacheIds, cacheCtx); - - if (err != null) { - StringBuilder cacheNames = new StringBuilder(); - - int idx = 0; - - for (Integer activeCacheId : activeCacheIds) { - cacheNames.append(cctx.cacheContext(activeCacheId).name()); - - if (idx++ < activeCacheIds.size() - 1) - cacheNames.append(", "); - } - - throw new IgniteCheckedException("Failed to enlist new cache to existing transaction (" + - err + - ") [activeCaches=[" + cacheNames + "]" + - ", cacheName=" + cacheCtx.name() + - ", cacheSystem=" + cacheCtx.systemTx() + - ", txSystem=" + system() + ']'); - } - else - activeCacheIds.add(cacheId); - - if (activeCacheIds.size() == 1) - depEnabled = cacheCtx.deploymentEnabled(); - } + protected final void addActiveCache(GridCacheContext cacheCtx) throws IgniteCheckedException { + txState.addActiveCache(cacheCtx, this); } /** @@ -3294,7 +3520,7 @@ public abstract class IgniteTxLocalAdapter extends IgniteTxAdapter "Invalid tx state for adding entry [op=" + op + ", val=" + val + ", entry=" + entry + ", filter=" + Arrays.toString(filter) + ", txCtx=" + cctx.tm().txContextVersion() + ", tx=" + this + ']'; - IgniteTxEntry old = txMap.get(key); + IgniteTxEntry old = entry(key); // Keep old filter if already have one (empty filter is always overridden). if (!filtersSet || !F.isEmptyOrNulls(filter)) { @@ -3358,7 +3584,7 @@ public abstract class IgniteTxLocalAdapter extends IgniteTxAdapter if (!hasDrTtl) txEntry.expiry(expiryPlc); - txMap.put(key, txEntry); + txState.addEntry(txEntry); if (log.isDebugEnabled()) log.debug("Created transaction entry: " + txEntry); @@ -3420,7 +3646,7 @@ public abstract class IgniteTxLocalAdapter extends IgniteTxAdapter /** {@inheritDoc} */ @Override public String toString() { return GridToStringBuilder.toString(IgniteTxLocalAdapter.class, this, "super", super.toString(), - "size", (txMap == null ? 0 : txMap.size())); + "size", allEntries().size()); } /**
http://git-wip-us.apache.org/repos/asf/ignite/blob/3ff71fd7/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 0d83338..5dc3338 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 @@ -93,9 +93,37 @@ public interface IgniteTxLocalEx extends IgniteInternalTx { /** * @param cacheCtx Cache context. + * @param key Key. + * @param val Value. + * @param retval Return value flag. + * @param filter Filter. + * @return Future for put operation. + */ + public <K, V> IgniteInternalFuture<GridCacheReturn> putAsync( + GridCacheContext cacheCtx, + K key, + V val, + boolean retval, + CacheEntryPredicate[] filter); + + /** + * @param cacheCtx Cache context. + * @param key Key. + * @param entryProcessor Entry processor. + * @param invokeArgs Optional arguments for entry processor. + * @return Operation future. + */ + public <K, V> IgniteInternalFuture<GridCacheReturn> invokeAsync( + GridCacheContext cacheCtx, + K key, + EntryProcessor<K, V, Object> entryProcessor, + Object... invokeArgs); + + /** + * @param cacheCtx Cache context. * @param map Entry processors map. * @param invokeArgs Optional arguments for entry processor. - * @return Transform operation future. + * @return Operation future. */ public <K, V, T> IgniteInternalFuture<GridCacheReturn> invokeAsync( GridCacheContext cacheCtx, http://git-wip-us.apache.org/repos/asf/ignite/blob/3ff71fd7/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxLocalState.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxLocalState.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxLocalState.java new file mode 100644 index 0000000..123d396 --- /dev/null +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxLocalState.java @@ -0,0 +1,44 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.ignite.internal.processors.cache.transactions; + +/** + * + */ +public interface IgniteTxLocalState extends IgniteTxState { + /** + * @param entry Entry. + */ + public void addEntry(IgniteTxEntry entry); + + /** + * @param txSize Transaction size. + * @return {@code True} if transaction was successfully started. + */ + public boolean init(int txSize); + + /** + * @return {@code True} if init method was called. + */ + public boolean initialized(); + + /** + * + */ + public void seal(); +} http://git-wip-us.apache.org/repos/asf/ignite/blob/3ff71fd7/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxLocalStateAdapter.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxLocalStateAdapter.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxLocalStateAdapter.java new file mode 100644 index 0000000..cde5203 --- /dev/null +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxLocalStateAdapter.java @@ -0,0 +1,41 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.ignite.internal.processors.cache.transactions; + +import org.apache.ignite.internal.processors.cache.GridCacheContext; +import org.apache.ignite.internal.util.typedef.internal.U; + +/** + * + */ +public abstract class IgniteTxLocalStateAdapter implements IgniteTxLocalState { + /** + * @param cacheCtx Cache context. + * @param tx Transaction. + * @param commit {@code False} if transaction rolled back. + */ + protected final void onTxEnd(GridCacheContext cacheCtx, IgniteInternalTx tx, boolean commit) { + if (cacheCtx.cache().configuration().isStatisticsEnabled()) { + // Convert start time from ms to ns. + if (commit) + cacheCtx.cache().metrics0().onTxCommit((U.currentTimeMillis() - tx.startTime()) * 1000); + else + cacheCtx.cache().metrics0().onTxRollback((U.currentTimeMillis() - tx.startTime()) * 1000); + } + } +} http://git-wip-us.apache.org/repos/asf/ignite/blob/3ff71fd7/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxManager.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxManager.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxManager.java index c2e7dea..ccccca0 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxManager.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxManager.java @@ -1092,13 +1092,7 @@ public class IgniteTxManager extends GridCacheSharedManagerAdapter { if (!tx.system()) cctx.txMetrics().onTxCommit(); - for (int cacheId : tx.activeCacheIds()) { - GridCacheContext cacheCtx = cctx.cacheContext(cacheId); - - if (cacheCtx.cache().configuration().isStatisticsEnabled()) - // Convert start time from ms to ns. - cacheCtx.cache().metrics0().onTxCommit((U.currentTimeMillis() - tx.startTime()) * 1000); - } + tx.txState().onTxEnd(cctx, tx, true); } if (slowTxWarnTimeout > 0 && tx.local() && @@ -1163,13 +1157,7 @@ public class IgniteTxManager extends GridCacheSharedManagerAdapter { if (!tx.system()) cctx.txMetrics().onTxRollback(); - for (int cacheId : tx.activeCacheIds()) { - GridCacheContext cacheCtx = cctx.cacheContext(cacheId); - - if (cacheCtx.cache().configuration().isStatisticsEnabled()) - // Convert start time from ms to ns. - cacheCtx.cache().metrics0().onTxRollback((U.currentTimeMillis() - tx.startTime()) * 1000); - } + tx.txState().onTxEnd(cctx, tx, false); } if (log.isDebugEnabled()) @@ -1233,7 +1221,7 @@ public class IgniteTxManager extends GridCacheSharedManagerAdapter { if (!tx.system()) threadMap.remove(tx.threadId(), tx); else { - Integer cacheId = F.first(tx.activeCacheIds()); + Integer cacheId = tx.txState().firstCacheId(); if (cacheId != null) sysThreadMap.remove(new TxThreadKey(tx.threadId(), cacheId), tx); http://git-wip-us.apache.org/repos/asf/ignite/blob/3ff71fd7/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxMap.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxMap.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxMap.java index 6408573..429c995 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxMap.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxMap.java @@ -170,8 +170,7 @@ public class IgniteTxMap extends AbstractMap<IgniteTxKey, IgniteTxEntry> impleme } /** {@inheritDoc} */ - @Nullable - @Override public IgniteTxEntry get(Object key) { + @Nullable @Override public IgniteTxEntry get(Object key) { IgniteTxEntry e = txMap.get(key); return e == null ? null : filter.apply(e) ? e : null; http://git-wip-us.apache.org/repos/asf/ignite/blob/3ff71fd7/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxRemoteEx.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxRemoteEx.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxRemoteEx.java index 9660e4e..b80909f 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxRemoteEx.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxRemoteEx.java @@ -25,22 +25,13 @@ import org.apache.ignite.internal.processors.cache.version.GridCacheVersion; */ public interface IgniteTxRemoteEx extends IgniteInternalTx { /** - * @return Remote thread ID. - */ - public long remoteThreadId(); - - /** * @param baseVer Base version. * @param committedVers Committed version. * @param rolledbackVers Rolled back version. * @param pendingVers Pending versions. */ - public void doneRemote(GridCacheVersion baseVer, Collection<GridCacheVersion> committedVers, - Collection<GridCacheVersion> rolledbackVers, Collection<GridCacheVersion> pendingVers); - - /** - * @param e Sets write value for pessimistic transactions. - * @return {@code True} if entry was found. - */ - public boolean setWriteValue(IgniteTxEntry e); + public void doneRemote(GridCacheVersion baseVer, + Collection<GridCacheVersion> committedVers, + Collection<GridCacheVersion> rolledbackVers, + Collection<GridCacheVersion> pendingVers); } \ No newline at end of file
