You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by ag...@apache.org on 2018/04/20 17:45:56 UTC
[3/6] ignite git commit: IGNITE-7910 Transaction rollback on PME
http://git-wip-us.apache.org/repos/asf/ignite/blob/ee5fbce3/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxFinishFuture.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxFinishFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxFinishFuture.java
index b6a8855..da8c6c8 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxFinishFuture.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxFinishFuture.java
@@ -39,6 +39,7 @@ import org.apache.ignite.internal.processors.cache.GridCacheFuture;
import org.apache.ignite.internal.processors.cache.GridCacheReturn;
import org.apache.ignite.internal.processors.cache.GridCacheReturnCompletableWrapper;
import org.apache.ignite.internal.processors.cache.GridCacheSharedContext;
+import org.apache.ignite.internal.processors.cache.GridCacheVersionedFuture;
import org.apache.ignite.internal.processors.cache.distributed.GridDistributedTxMapping;
import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtTxFinishRequest;
import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtTxFinishResponse;
@@ -55,6 +56,7 @@ import org.apache.ignite.internal.util.typedef.F;
import org.apache.ignite.internal.util.typedef.internal.CU;
import org.apache.ignite.internal.util.typedef.internal.S;
import org.apache.ignite.internal.util.typedef.internal.U;
+import org.apache.ignite.lang.IgniteInClosure;
import org.apache.ignite.lang.IgniteUuid;
import org.apache.ignite.transactions.TransactionRollbackException;
@@ -120,14 +122,8 @@ public final class GridNearTxFinishFuture<K, V> extends GridCacheCompoundIdentit
futId = IgniteUuid.randomUuid();
- CacheWriteSynchronizationMode syncMode;
-
if (tx.explicitLock())
- syncMode = FULL_SYNC;
- else
- syncMode = tx.syncMode();
-
- tx.syncMode(syncMode);
+ tx.syncMode(FULL_SYNC);
if (log == null) {
msgLog = cctx.txFinishMessageLogger();
@@ -399,9 +395,10 @@ public final class GridNearTxFinishFuture<K, V> extends GridCacheCompoundIdentit
*
* @param commit Commit flag.
* @param clearThreadMap If {@code true} removes {@link GridNearTxLocal} from thread map.
+ * @param onTimeout If {@code true} called from timeout handler.
*/
@SuppressWarnings("ForLoopReplaceableByForEach")
- public void finish(boolean commit, boolean clearThreadMap) {
+ public void finish(final boolean commit, final boolean clearThreadMap, final boolean onTimeout) {
if (tx.onNeedCheckBackup()) {
assert tx.onePhaseCommit();
@@ -414,6 +411,58 @@ public final class GridNearTxFinishFuture<K, V> extends GridCacheCompoundIdentit
return;
}
+ if (!commit && !clearThreadMap)
+ tryRollbackAsync(onTimeout); // Asynchronous rollback.
+ else
+ doFinish(commit, clearThreadMap);
+ }
+
+ /**
+ * Does async rollback when it's safe.
+ * If current future is not lock future (enlist future) waits until completion and tries again.
+ * Else terminates or waits for lock future depending on rollback mode.
+ *
+ * @param onTimeout If {@code true} called from timeout handler.
+ */
+ private void tryRollbackAsync(boolean onTimeout) {
+ IgniteInternalFuture<?> curFut = tx.tryRollbackAsync();
+
+ if (curFut == null) { // Safe to rollback.
+ doFinish(false, false);
+
+ return;
+ }
+
+ if (curFut instanceof GridCacheVersionedFuture && !onTimeout) {
+ try {
+ curFut.cancel(); // Force cancellation.
+ }
+ catch (IgniteCheckedException e) {
+ log.error("Failed to cancel lock for the transaction: " + CU.txString(tx), e);
+ }
+ }
+
+ curFut.listen(new IgniteInClosure<IgniteInternalFuture<?>>() {
+ @Override public void apply(IgniteInternalFuture<?> fut) {
+ try {
+ fut.get();
+
+ tryRollbackAsync(onTimeout);
+ }
+ catch (IgniteCheckedException e) {
+ doFinish(false, false);
+ }
+ }
+ });
+ }
+
+ /**
+ * Finishes a transaction.
+ *
+ * @param commit Commit.
+ * @param clearThreadMap Clear thread map.
+ */
+ private void doFinish(boolean commit, boolean clearThreadMap) {
try {
if (tx.localFinish(commit, clearThreadMap) || (!commit && tx.state() == UNKNOWN)) {
if ((tx.onePhaseCommit() && needFinishOnePhase(commit)) || (!tx.onePhaseCommit() && mappings != null)) {
@@ -423,17 +472,18 @@ public final class GridNearTxFinishFuture<K, V> extends GridCacheCompoundIdentit
if (mapping != null) {
assert !hasFutures() : futures();
- finish(1, mapping, commit);
+ finish(1, mapping, commit, !clearThreadMap);
}
}
else
- finish(mappings.mappings(), commit);
+ finish(mappings.mappings(), commit, !clearThreadMap);
}
markInitialized();
}
else
- onDone(new IgniteCheckedException("Failed to commit transaction: " + CU.txString(tx)));
+ onDone(new IgniteCheckedException("Failed to " + (commit ? "commit" : "rollback") +
+ " transaction: " + CU.txString(tx)));
}
catch (Error | RuntimeException e) {
onDone(e);
@@ -679,26 +729,28 @@ public final class GridNearTxFinishFuture<K, V> extends GridCacheCompoundIdentit
/**
* @param mappings Mappings.
* @param commit Commit flag.
+ * @param {@code true} If need to add completed version on finish.
*/
- private void finish(Iterable<GridDistributedTxMapping> mappings, boolean commit) {
+ private void finish(Iterable<GridDistributedTxMapping> mappings, boolean commit, boolean useCompletedVer) {
assert !hasFutures() : futures();
int miniId = 0;
// Create mini futures.
for (GridDistributedTxMapping m : mappings)
- finish(++miniId, m, commit);
+ finish(++miniId, m, commit, useCompletedVer);
}
/**
* @param miniId Mini future ID.
* @param m Mapping.
* @param commit Commit flag.
+ * @param useCompletedVer {@code True} if need to add completed version on finish.
*/
- private void finish(int miniId, GridDistributedTxMapping m, boolean commit) {
+ private void finish(int miniId, GridDistributedTxMapping m, boolean commit, boolean useCompletedVer) {
ClusterNode n = m.primary();
- assert !m.empty() : m;
+ assert !m.empty() : m + " " + tx.state();
CacheWriteSynchronizationMode syncMode = tx.syncMode();
@@ -706,7 +758,7 @@ public final class GridNearTxFinishFuture<K, V> extends GridCacheCompoundIdentit
syncMode = FULL_SYNC;
// Version to be added in completed versions on primary node.
- GridCacheVersion completedVer = !commit && tx.timeout() > 0 ? tx.xidVersion() : null;
+ GridCacheVersion completedVer = !commit && useCompletedVer ? tx.xidVersion() : null;
GridNearTxFinishRequest req = new GridNearTxFinishRequest(
futId,
@@ -746,7 +798,7 @@ public final class GridNearTxFinishFuture<K, V> extends GridCacheCompoundIdentit
add(fut); // Append new future.
- if (tx.pessimistic())
+ if (tx.pessimistic() && !useCompletedVer)
cctx.tm().beforeFinishRemote(n.id(), tx.threadId());
try {
@@ -1000,7 +1052,15 @@ public final class GridNearTxFinishFuture<K, V> extends GridCacheCompoundIdentit
*/
void onNearFinishResponse(GridNearTxFinishResponse res) {
if (res.error() != null)
- onDone(res.error());
+ if (res.error() instanceof IgniteTxRollbackCheckedException) {
+ // This exception is expected on asynchronous rollback.
+ if (log.isDebugEnabled())
+ log.debug("Transaction was rolled back: " + tx);
+
+ onDone(tx);
+ }
+ else
+ onDone(res.error());
else
onDone(tx);
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/ee5fbce3/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxLocal.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxLocal.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxLocal.java
index e339750..286a35d 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxLocal.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxLocal.java
@@ -34,7 +34,6 @@ import javax.cache.CacheException;
import javax.cache.expiry.ExpiryPolicy;
import javax.cache.processor.EntryProcessor;
import org.apache.ignite.IgniteCheckedException;
-import org.apache.ignite.IgniteException;
import org.apache.ignite.cluster.ClusterNode;
import org.apache.ignite.internal.IgniteInternalFuture;
import org.apache.ignite.internal.NodeStoppingException;
@@ -67,6 +66,7 @@ import org.apache.ignite.internal.processors.cache.transactions.IgniteTxEntry;
import org.apache.ignite.internal.processors.cache.transactions.IgniteTxKey;
import org.apache.ignite.internal.processors.cache.transactions.TransactionProxy;
import org.apache.ignite.internal.processors.cache.transactions.TransactionProxyImpl;
+import org.apache.ignite.internal.processors.cache.transactions.TransactionProxyRollbackOnlyImpl;
import org.apache.ignite.internal.processors.cache.version.GridCacheVersion;
import org.apache.ignite.internal.processors.timeout.GridTimeoutObject;
import org.apache.ignite.internal.transactions.IgniteTxOptimisticCheckedException;
@@ -166,17 +166,24 @@ public class GridNearTxLocal extends GridDhtTxLocalAdapter implements GridTimeou
protected boolean transform;
/** */
- private final boolean trackTimeout;
+ private boolean trackTimeout;
/** */
@GridToStringExclude
private TransactionProxyImpl proxy;
+ /** */
+ @GridToStringExclude
+ private TransactionProxyImpl rollbackOnlyProxy;
+
+ /** Tx label. */
+ private @Nullable String lb;
+
/**
* Empty constructor required for {@link Externalizable}.
*/
public GridNearTxLocal() {
- this.trackTimeout = false;
+ // No-op.
}
/**
@@ -192,6 +199,7 @@ public class GridNearTxLocal extends GridDhtTxLocalAdapter implements GridTimeou
* @param txSize Transaction size.
* @param subjId Subject ID.
* @param taskNameHash Task name hash code.
+ * @param lb Label.
*/
public GridNearTxLocal(
GridCacheSharedContext ctx,
@@ -205,7 +213,8 @@ public class GridNearTxLocal extends GridDhtTxLocalAdapter implements GridTimeou
boolean storeEnabled,
int txSize,
@Nullable UUID subjId,
- int taskNameHash
+ int taskNameHash,
+ @Nullable String lb
) {
super(
ctx,
@@ -224,6 +233,7 @@ public class GridNearTxLocal extends GridDhtTxLocalAdapter implements GridTimeou
txSize,
subjId,
taskNameHash);
+ this.lb = lb;
mappings = implicitSingle ? new IgniteTxMappingsSingleImpl() : new IgniteTxMappingsImpl();
@@ -343,6 +353,24 @@ public class GridNearTxLocal extends GridDhtTxLocalAdapter implements GridTimeou
}
/** {@inheritDoc} */
+ @Override public long timeout(long timeout) {
+ long old = super.timeout(timeout);
+
+ if (old == timeout)
+ return old;
+
+ if (trackTimeout) {
+ if (!cctx.time().removeTimeoutObject(this))
+ return old; // Do nothing, because transaction is started to roll back.
+ }
+
+ if (timeout() > 0) // Method can be called only for explicit transactions.
+ trackTimeout = cctx.time().addTimeoutObject(this);
+
+ return old;
+ }
+
+ /** {@inheritDoc} */
@Override public boolean ownsLock(GridCacheEntryEx entry) throws GridCacheEntryRemovedException {
return entry.detached() || super.ownsLock(entry);
}
@@ -550,23 +578,28 @@ public class GridNearTxLocal extends GridDhtTxLocalAdapter implements GridTimeou
opCtx != null && opCtx.recovery(),
dataCenterId);
- if (pessimistic()) {
- assert loadFut == null || loadFut.isDone() : loadFut;
+ try {
+ loadFut.get();
+ }
+ catch (IgniteCheckedException e) {
+ return new GridFinishedFuture(e);
+ }
+
+ long timeout = remainingTime();
+
+ if (timeout == -1)
+ return new GridFinishedFuture<>(timeoutException());
- if (loadFut != null)
- loadFut.get();
+ if (isRollbackOnly())
+ return new GridFinishedFuture<>(rollbackException());
+ if (pessimistic()) {
final Collection<KeyCacheObject> enlisted = Collections.singleton(cacheKey);
if (log.isDebugEnabled())
log.debug("Before acquiring transaction lock for put on key: " + enlisted);
- long timeout = remainingTime();
-
- if (timeout == -1)
- return new GridFinishedFuture<>(timeoutException());
-
- IgniteInternalFuture<Boolean> fut = cacheCtx.cache().txLockAsync(enlisted,
+ IgniteInternalFuture<Boolean>fut = cacheCtx.cache().txLockAsync(enlisted,
timeout,
this,
/*read*/entryProcessor != null, // Needed to force load from store.
@@ -726,25 +759,24 @@ public class GridNearTxLocal extends GridDhtTxLocalAdapter implements GridTimeou
opCtx != null && opCtx.recovery(),
dataCenterId);
- if (pessimistic()) {
- assert loadFut == null || loadFut.isDone() : loadFut;
+ try {
+ loadFut.get();
+ }
+ catch (IgniteCheckedException e) {
+ return new GridFinishedFuture(e);
+ }
- if (loadFut != null) {
- try {
- loadFut.get();
- }
- catch (IgniteCheckedException e) {
- return new GridFinishedFuture(e);
- }
- }
+ long timeout = remainingTime();
- if (log.isDebugEnabled())
- log.debug("Before acquiring transaction lock for put on keys: " + enlisted);
+ if (timeout == -1)
+ return new GridFinishedFuture<>(timeoutException());
- long timeout = remainingTime();
+ if (isRollbackOnly())
+ return new GridFinishedFuture<>(rollbackException());
- if (timeout == -1)
- return new GridFinishedFuture<>(timeoutException());
+ if (pessimistic()) {
+ if (log.isDebugEnabled())
+ log.debug("Before acquiring transaction lock for put on keys: " + enlisted);
IgniteInternalFuture<Boolean> fut = cacheCtx.cache().txLockAsync(enlisted,
timeout,
@@ -842,7 +874,12 @@ public class GridNearTxLocal extends GridDhtTxLocalAdapter implements GridTimeou
boolean keepBinary,
boolean recovery,
Byte dataCenterId) {
+ GridFutureAdapter<Void> enlistFut = new GridFutureAdapter<>();
+
try {
+ if (!updateLockFuture(null, enlistFut))
+ return finishFuture(enlistFut, timedOut() ? timeoutException() : rollbackException(), false);
+
addActiveCache(cacheCtx, recovery);
final boolean hasFilters = !F.isEmptyOrNulls(filter) && !F.isAlwaysTrue(filter);
@@ -883,7 +920,7 @@ public class GridNearTxLocal extends GridDhtTxLocalAdapter implements GridTimeou
if (topVer == null)
topVer = entryTopVer;
- return loadMissing(cacheCtx,
+ IgniteInternalFuture<Void> loadFut = loadMissing(cacheCtx,
topVer != null ? topVer : topologyVersion(),
Collections.singleton(cacheKey),
filter,
@@ -896,12 +933,29 @@ public class GridNearTxLocal extends GridDhtTxLocalAdapter implements GridTimeou
keepBinary,
recovery,
expiryPlc);
+
+ loadFut.listen(new IgniteInClosure<IgniteInternalFuture<Void>>() {
+ @Override public void apply(IgniteInternalFuture<Void> fut) {
+ try {
+ fut.get();
+
+ finishFuture(enlistFut, null, true);
+ }
+ catch (IgniteCheckedException e) {
+ finishFuture(enlistFut, e, true);
+ }
+ }
+ });
+
+ return enlistFut;
}
- return new GridFinishedFuture<>();
+ finishFuture(enlistFut, null, true);
+
+ return enlistFut;
}
catch (IgniteCheckedException e) {
- return new GridFinishedFuture<>(e);
+ return finishFuture(enlistFut, e, true);
}
}
@@ -925,7 +979,7 @@ public class GridNearTxLocal extends GridDhtTxLocalAdapter implements GridTimeou
* @param singleRmv {@code True} for single key remove operation ({@link Cache#remove(Object)}.
* @param keepBinary Keep binary flag.
* @param dataCenterId Optional data center ID.
- * @return Future for missing values loading.
+ * @return Future for enlisting writes.
*/
private <K, V> IgniteInternalFuture<Void> enlistWrite(
final GridCacheContext cacheCtx,
@@ -950,11 +1004,16 @@ public class GridNearTxLocal extends GridDhtTxLocalAdapter implements GridTimeou
) {
assert retval || invokeMap == null;
+ GridFutureAdapter<Void> enlistFut = new GridFutureAdapter<>();
+
+ if (!updateLockFuture(null, enlistFut))
+ return finishFuture(enlistFut, timedOut() ? timeoutException() : rollbackException(), false);
+
try {
addActiveCache(cacheCtx, recovery);
}
catch (IgniteCheckedException e) {
- return new GridFinishedFuture<>(e);
+ return finishFuture(enlistFut, e, false);
}
boolean rmv = lookup == null && invokeMap == null;
@@ -971,6 +1030,9 @@ public class GridNearTxLocal extends GridDhtTxLocalAdapter implements GridTimeou
Set<KeyCacheObject> missedForLoad = null;
for (Object key : keys) {
+ if (isRollbackOnly())
+ return finishFuture(enlistFut, timedOut() ? timeoutException() : rollbackException(), false);
+
if (key == null) {
rollback();
@@ -1056,7 +1118,7 @@ public class GridNearTxLocal extends GridDhtTxLocalAdapter implements GridTimeou
if (topVer == null)
topVer = entryTopVer;
- return loadMissing(cacheCtx,
+ IgniteInternalFuture<Void> loadFut = loadMissing(cacheCtx,
topVer != null ? topVer : topologyVersion(),
missedForLoad,
filter,
@@ -1069,12 +1131,27 @@ public class GridNearTxLocal extends GridDhtTxLocalAdapter implements GridTimeou
keepBinary,
recovery,
expiryPlc);
+
+ loadFut.listen(new IgniteInClosure<IgniteInternalFuture<Void>>() {
+ @Override public void apply(IgniteInternalFuture<Void> fut) {
+ try {
+ fut.get();
+
+ finishFuture(enlistFut, null, true);
+ }
+ catch (IgniteCheckedException e) {
+ finishFuture(enlistFut, e, true);
+ }
+ }
+ });
+
+ return enlistFut;
}
- return new GridFinishedFuture<>();
+ return finishFuture(enlistFut, null, true);
}
catch (IgniteCheckedException e) {
- return new GridFinishedFuture<>(e);
+ return finishFuture(enlistFut, e, true);
}
}
@@ -1537,6 +1614,21 @@ public class GridNearTxLocal extends GridDhtTxLocalAdapter implements GridTimeou
dataCenterId
);
+ try {
+ loadFut.get();
+ }
+ catch (IgniteCheckedException e) {
+ return new GridFinishedFuture(e);
+ }
+
+ long timeout = remainingTime();
+
+ if (timeout == -1)
+ return new GridFinishedFuture<>(timeoutException());
+
+ if (isRollbackOnly())
+ return new GridFinishedFuture<>(rollbackException());
+
if (log.isDebugEnabled())
log.debug("Remove keys: " + enlisted);
@@ -1544,25 +1636,9 @@ public class GridNearTxLocal extends GridDhtTxLocalAdapter implements GridTimeou
// Otherwise, during rollback we will not know whether locks need
// to be rolled back.
if (pessimistic()) {
- assert loadFut == null || loadFut.isDone() : loadFut;
-
- if (loadFut != null) {
- try {
- loadFut.get();
- }
- catch (IgniteCheckedException e) {
- return new GridFinishedFuture<>(e);
- }
- }
-
if (log.isDebugEnabled())
log.debug("Before acquiring transaction lock for remove on keys: " + enlisted);
- long timeout = remainingTime();
-
- if (timeout == -1)
- return new GridFinishedFuture<>(timeoutException());
-
IgniteInternalFuture<Boolean> fut = cacheCtx.cache().txLockAsync(enlisted,
timeout,
this,
@@ -1684,6 +1760,11 @@ public class GridNearTxLocal extends GridDhtTxLocalAdapter implements GridTimeou
try {
checkValid();
+ GridFutureAdapter<?> enlistFut = new GridFutureAdapter<>();
+
+ if (!updateLockFuture(null, enlistFut))
+ return new GridFinishedFuture<>(timedOut() ? timeoutException() : rollbackException());
+
final Map<K, V> retMap = new GridLeanMap<>(keysCnt);
final Map<KeyCacheObject, GridCacheVersion> missed = new GridLeanMap<>(pessimistic() ? keysCnt : 0);
@@ -1692,19 +1773,32 @@ public class GridNearTxLocal extends GridDhtTxLocalAdapter implements GridTimeou
ExpiryPolicy expiryPlc = opCtx != null ? opCtx.expiry() : null;
- final Collection<KeyCacheObject> lockKeys = enlistRead(cacheCtx,
- entryTopVer,
- keys,
- expiryPlc,
- retMap,
- missed,
- keysCnt,
- deserializeBinary,
- skipVals,
- keepCacheObjects,
- skipStore,
- recovery,
- needVer);
+ final Collection<KeyCacheObject> lockKeys;
+
+ try {
+ lockKeys = enlistRead(cacheCtx,
+ entryTopVer,
+ keys,
+ expiryPlc,
+ retMap,
+ missed,
+ keysCnt,
+ deserializeBinary,
+ skipVals,
+ keepCacheObjects,
+ skipStore,
+ recovery,
+ needVer);
+ }
+ catch (IgniteCheckedException e) {
+ return new GridFinishedFuture<>(e);
+ }
+ finally {
+ finishFuture(enlistFut, null, true);
+ }
+
+ if (isRollbackOnly())
+ return new GridFinishedFuture<>(timedOut() ? timeoutException() : rollbackException());
if (single && missed.isEmpty())
return new GridFinishedFuture<>(retMap);
@@ -1996,6 +2090,9 @@ public class GridNearTxLocal extends GridDhtTxLocalAdapter implements GridTimeou
// Transactions that are pessimistic and not read-committed are covered
// outside of this loop.
for (KeyCacheObject key : keys) {
+ if (isRollbackOnly())
+ throw timedOut() ? timeoutException() : rollbackException();
+
if ((pessimistic() || needReadVer) && !readCommitted() && !skipVals)
addActiveCache(cacheCtx, recovery);
@@ -2383,13 +2480,6 @@ public class GridNearTxLocal extends GridDhtTxLocalAdapter implements GridTimeou
// with prepare response, if required.
assert loadFut.isDone();
- try {
- loadFut.get();
- }
- catch (IgniteCheckedException e) {
- return new GridFinishedFuture<>(e);
- }
-
return nonInterruptable(commitNearTxLocalAsync().chain(
new CX1<IgniteInternalFuture<IgniteInternalTx>, GridCacheReturn>() {
@Override public GridCacheReturn applyx(IgniteInternalFuture<IgniteInternalTx> txFut)
@@ -2805,7 +2895,7 @@ public class GridNearTxLocal extends GridDhtTxLocalAdapter implements GridTimeou
/**
* @return DHT map.
*/
- IgniteTxMappings mappings() {
+ public IgniteTxMappings mappings() {
return mappings;
}
@@ -2887,16 +2977,17 @@ public class GridNearTxLocal extends GridDhtTxLocalAdapter implements GridTimeou
* @throws IgniteCheckedException If the transaction is in an incorrect state, or timed out.
*/
public void resume() throws IgniteCheckedException {
- resume(true);
+ resume(true, Thread.currentThread().getId());
}
/**
* Resumes transaction (possibly in another thread) if it was previously suspended.
*
* @param checkTimeout Whether timeout should be checked.
+ * @param threadId Thread id to restore.
* @throws IgniteCheckedException If the transaction is in an incorrect state, or timed out.
*/
- private void resume(boolean checkTimeout) throws IgniteCheckedException {
+ private void resume(boolean checkTimeout, long threadId) throws IgniteCheckedException {
if (log.isDebugEnabled())
log.debug("Resume near local tx: " + this);
@@ -2906,7 +2997,7 @@ public class GridNearTxLocal extends GridDhtTxLocalAdapter implements GridTimeou
synchronized (this) {
checkValid(checkTimeout);
- cctx.tm().resumeTx(this);
+ cctx.tm().resumeTx(this, threadId);
}
}
@@ -3215,15 +3306,15 @@ public class GridNearTxLocal extends GridDhtTxLocalAdapter implements GridTimeou
NearTxFinishFuture fut = finishFut;
if (fut != null)
- return chainFinishFuture(fut, true);
+ return chainFinishFuture(fut, true, true, false);
if (fastFinish()) {
GridNearTxFastFinishFuture fut0;
- if (!FINISH_FUT_UPD.compareAndSet(this, null, fut0 = new GridNearTxFastFinishFuture(this, true, false)))
- return chainFinishFuture(finishFut, true);
+ if (!FINISH_FUT_UPD.compareAndSet(this, null, fut0 = new GridNearTxFastFinishFuture(this, true)))
+ return chainFinishFuture(finishFut, true, true, false);
- fut0.finish();
+ fut0.finish(false);
return fut0;
}
@@ -3231,7 +3322,7 @@ public class GridNearTxLocal extends GridDhtTxLocalAdapter implements GridTimeou
final GridNearTxFinishFuture fut0;
if (!FINISH_FUT_UPD.compareAndSet(this, null, fut0 = new GridNearTxFinishFuture<>(cctx, this, true)))
- return chainFinishFuture(finishFut, true);
+ return chainFinishFuture(finishFut, true, true, false);
cctx.mvcc().addFuture(fut0, fut0.futureId());
@@ -3243,14 +3334,12 @@ public class GridNearTxLocal extends GridDhtTxLocalAdapter implements GridTimeou
// Make sure that here are no exceptions.
prepareFut.get();
- TransactionState state = state();
-
- fut0.finish(state == PREPARED || state == COMMITTING || state == COMMITTED, true);
+ fut0.finish(true, true, false);
}
catch (Error | RuntimeException e) {
COMMIT_ERR_UPD.compareAndSet(GridNearTxLocal.this, null, e);
- fut0.finish(false, true);
+ fut0.finish(false, true, false);
throw e;
}
@@ -3258,7 +3347,7 @@ public class GridNearTxLocal extends GridDhtTxLocalAdapter implements GridTimeou
COMMIT_ERR_UPD.compareAndSet(GridNearTxLocal.this, null, e);
if (!(e instanceof NodeStoppingException))
- fut0.finish(false, true);
+ fut0.finish(false, true, true);
}
}
});
@@ -3282,14 +3371,20 @@ public class GridNearTxLocal extends GridDhtTxLocalAdapter implements GridTimeou
* @return Rollback future.
*/
public IgniteInternalFuture<IgniteInternalTx> rollbackNearTxLocalAsync() {
- return rollbackNearTxLocalAsync(false);
+ return rollbackNearTxLocalAsync(true, false);
}
/**
- * @param onTimeout {@code True} if rolled back asynchronously on timeout.
+ * @param clearThreadMap {@code True} if needed to clear thread map.
+ * @param onTimeout {@code True} if called from timeout handler.
+ * Note: For async rollbacks (from timeouts or another thread) should not clear thread map
+ * since thread started tx still should be able to see this tx to prevent subsequent operations.
+ *
* @return Rollback future.
*/
- public IgniteInternalFuture<IgniteInternalTx> rollbackNearTxLocalAsync(final boolean onTimeout) {
+ @SuppressWarnings("ThrowableResultOfMethodCallIgnored")
+ public IgniteInternalFuture<IgniteInternalTx> rollbackNearTxLocalAsync(final boolean clearThreadMap,
+ final boolean onTimeout) {
if (log.isDebugEnabled())
log.debug("Rolling back near tx: " + this);
@@ -3299,15 +3394,16 @@ public class GridNearTxLocal extends GridDhtTxLocalAdapter implements GridTimeou
NearTxFinishFuture fut = finishFut;
if (fut != null)
- return chainFinishFuture(finishFut, false, !onTimeout);
+ return chainFinishFuture(finishFut, false, clearThreadMap, onTimeout);
- if (fastFinish()) {
+ // Enable fast finish only from tx thread.
+ if (clearThreadMap && fastFinish()) {
GridNearTxFastFinishFuture fut0;
- if (!FINISH_FUT_UPD.compareAndSet(this, null, fut0 = new GridNearTxFastFinishFuture(this, false, onTimeout)))
- return chainFinishFuture(finishFut, false, !onTimeout);
+ if (!FINISH_FUT_UPD.compareAndSet(this, null, fut0 = new GridNearTxFastFinishFuture(this, false)))
+ return chainFinishFuture(finishFut, false, true, onTimeout);
- fut0.finish();
+ fut0.finish(true);
return fut0;
}
@@ -3315,7 +3411,7 @@ public class GridNearTxLocal extends GridDhtTxLocalAdapter implements GridTimeou
final GridNearTxFinishFuture fut0;
if (!FINISH_FUT_UPD.compareAndSet(this, null, fut0 = new GridNearTxFinishFuture<>(cctx, this, false)))
- return chainFinishFuture(finishFut, false, !onTimeout);
+ return chainFinishFuture(finishFut, false, clearThreadMap, onTimeout);
cctx.mvcc().addFuture(fut0, fut0.futureId());
@@ -3332,9 +3428,9 @@ public class GridNearTxLocal extends GridDhtTxLocalAdapter implements GridTimeou
log.debug("Got optimistic tx failure [tx=" + this + ", err=" + e + ']');
}
- fut0.finish(false, !onTimeout);
+ fut0.finish(false, clearThreadMap, onTimeout);
}
- else if (!onTimeout) {
+ else {
prepFut.listen(new CI1<IgniteInternalFuture<?>>() {
@Override public void apply(IgniteInternalFuture<?> f) {
try {
@@ -3346,11 +3442,10 @@ public class GridNearTxLocal extends GridDhtTxLocalAdapter implements GridTimeou
log.debug("Got optimistic tx failure [tx=" + this + ", err=" + e + ']');
}
- fut0.finish(false, true);
+ fut0.finish(false, clearThreadMap, onTimeout);
}
});
- } else
- fut0.finish(false, false);
+ }
return fut0;
}
@@ -3360,17 +3455,14 @@ public class GridNearTxLocal extends GridDhtTxLocalAdapter implements GridTimeou
return rollbackNearTxLocalAsync();
}
- /** */
- private IgniteInternalFuture<IgniteInternalTx> chainFinishFuture(final NearTxFinishFuture fut, final boolean commit) {
- return chainFinishFuture(fut, commit, true);
- }
-
/**
* @param fut Already started finish future.
* @param commit Commit flag.
+ * @param clearThreadMap Clear thread map.
* @return Finish future.
*/
- private IgniteInternalFuture<IgniteInternalTx> chainFinishFuture(final NearTxFinishFuture fut, final boolean commit, final boolean clearThreadMap) {
+ private IgniteInternalFuture<IgniteInternalTx> chainFinishFuture(final NearTxFinishFuture fut, final boolean commit,
+ final boolean clearThreadMap, final boolean onTimeout) {
assert fut != null;
if (fut.commit() != commit) {
@@ -3394,9 +3486,23 @@ public class GridNearTxLocal extends GridDhtTxLocalAdapter implements GridTimeou
if (!cctx.mvcc().addFuture(rollbackFut, rollbackFut.futureId()))
return;
- rollbackFut.finish(false, clearThreadMap);
+ rollbackFut.finish(false, clearThreadMap, onTimeout);
}
}
+ else {
+ finishFut.listen(new IgniteInClosure<IgniteInternalFuture<IgniteInternalTx>>() {
+ @Override public void apply(IgniteInternalFuture<IgniteInternalTx> fut) {
+ try {
+ fut.get();
+
+ rollbackFut.markInitialized();
+ }
+ catch (IgniteCheckedException e) {
+ rollbackFut.onDone(e);
+ }
+ }
+ });
+ }
}
});
@@ -3411,7 +3517,7 @@ public class GridNearTxLocal extends GridDhtTxLocalAdapter implements GridTimeou
fut0.onDone(new IgniteTxTimeoutCheckedException("Failed to commit transaction, " +
"transaction is concurrently rolled back on timeout: " + tx));
else
- fut0.onDone(new IgniteCheckedException("Failed to commit transaction, " +
+ fut0.onDone(new IgniteTxRollbackCheckedException("Failed to commit transaction, " +
"transaction is concurrently rolled back: " + tx));
}
});
@@ -3442,13 +3548,11 @@ public class GridNearTxLocal extends GridDhtTxLocalAdapter implements GridTimeou
if (state() != PREPARING) {
if (timeout == -1)
- return new GridFinishedFuture<>(
- new IgniteTxTimeoutCheckedException("Transaction timed out: " + this));
+ return new GridFinishedFuture<>(timeoutException());
setRollbackOnly();
- return new GridFinishedFuture<>(
- new IgniteCheckedException("Invalid transaction state for prepare [state=" + state() + ", tx=" + this + ']'));
+ return new GridFinishedFuture<>(rollbackException());
}
if (timeout == -1)
@@ -3483,19 +3587,6 @@ public class GridNearTxLocal extends GridDhtTxLocalAdapter implements GridTimeou
setRollbackOnly();
fut.onError(new IgniteTxRollbackCheckedException("Failed to prepare transaction: " + this, e));
-
- try {
- rollback();
- }
- catch (IgniteTxOptimisticCheckedException e1) {
- if (log.isDebugEnabled())
- log.debug("Failed optimistically to prepare transaction [tx=" + this + ", e=" + e1 + ']');
-
- fut.onError(e);
- }
- catch (IgniteCheckedException e1) {
- U.error(log, "Failed to rollback transaction: " + this, e1);
- }
}
return chainOnePhasePrepare(fut);
@@ -3788,17 +3879,26 @@ public class GridNearTxLocal extends GridDhtTxLocalAdapter implements GridTimeou
/** {@inheritDoc} */
@Override public void close() throws IgniteCheckedException {
+ close(true);
+ }
+
+ /**
+ * @param clearThreadMap Clear thread map.
+ */
+ public void close(boolean clearThreadMap) throws IgniteCheckedException {
TransactionState state = state();
try {
if (state == COMMITTED || state == ROLLED_BACK)
return;
+ boolean rmv = false;
+
if (trackTimeout)
- removeTimeoutHandler();
+ rmv = removeTimeoutHandler();
- if (state != ROLLING_BACK && state != COMMITTING)
- rollback();
+ if (state != COMMITTING && state != ROLLING_BACK && (!trackTimeout || rmv))
+ rollbackNearTxLocalAsync(clearThreadMap, false).get();
synchronized (this) {
try {
@@ -3815,10 +3915,7 @@ public class GridNearTxLocal extends GridDhtTxLocalAdapter implements GridTimeou
}
}
finally {
- // It is possible tx was rolled back asynchronously on timeout and thread map is not cleared yet.
- boolean cleanup = state == ROLLED_BACK && timedOut();
-
- if (cleanup)
+ if (clearThreadMap)
cctx.tm().clearThreadMap(this);
if (accessMap != null) {
@@ -3891,6 +3988,16 @@ public class GridNearTxLocal extends GridDhtTxLocalAdapter implements GridTimeou
}
/**
+ * @return Public API proxy.
+ */
+ public TransactionProxy rollbackOnlyProxy() {
+ if (rollbackOnlyProxy == null)
+ rollbackOnlyProxy = new TransactionProxyRollbackOnlyImpl<>(this, cctx, false);
+
+ return rollbackOnlyProxy;
+ }
+
+ /**
* @param cacheCtx Cache context.
* @param topVer Topology version.
* @param map Return map.
@@ -3954,14 +4061,6 @@ public class GridNearTxLocal extends GridDhtTxLocalAdapter implements GridTimeou
expiryPlc,
new GridInClosure3<KeyCacheObject, Object, GridCacheVersion>() {
@Override public void apply(KeyCacheObject key, Object val, GridCacheVersion loadVer) {
- if (isRollbackOnly()) {
- if (log.isDebugEnabled())
- log.debug("Ignoring loaded value for read because transaction was rolled back: " +
- GridNearTxLocal.this);
-
- return;
- }
-
CacheObject cacheVal = cacheCtx.toCacheObject(val);
CacheObject visibleVal = cacheVal;
@@ -4163,11 +4262,18 @@ public class GridNearTxLocal extends GridDhtTxLocalAdapter implements GridTimeou
return startTime() + timeout();
}
+ /**
+ * @return Tx label.
+ */
+ public String label() {
+ return lb;
+ }
+
/** {@inheritDoc} */
@Override public void onTimeout() {
if (state() == SUSPENDED) {
try {
- resume(false);
+ resume(false, threadId());
}
catch (IgniteCheckedException e) {
log.warning("Error resuming suspended transaction on timeout: " + this, e);
@@ -4185,9 +4291,10 @@ public class GridNearTxLocal extends GridDhtTxLocalAdapter implements GridTimeou
@Override public void run() {
// Note: if rollback asynchronously on timeout should not clear thread map
// since thread started tx still should be able to see this tx.
- rollbackNearTxLocalAsync(true);
+ rollbackNearTxLocalAsync(false, true);
- U.warn(log, "Transaction was rolled back because the timeout is reached: " + GridNearTxLocal.this);
+ U.warn(log, "The transaction was forcibly rolled back because a timeout is reached: " +
+ CU.txString(GridNearTxLocal.this));
}
});
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/ee5fbce3/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/local/GridLocalLockFuture.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/local/GridLocalLockFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/local/GridLocalLockFuture.java
index 93d82c9..18a8431 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/local/GridLocalLockFuture.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/local/GridLocalLockFuture.java
@@ -36,20 +36,21 @@ import org.apache.ignite.internal.processors.cache.GridCacheFutureAdapter;
import org.apache.ignite.internal.processors.cache.GridCacheMvccCandidate;
import org.apache.ignite.internal.processors.cache.GridCacheVersionedFuture;
import org.apache.ignite.internal.processors.cache.KeyCacheObject;
+import org.apache.ignite.internal.processors.cache.distributed.near.GridNearTxLocal;
import org.apache.ignite.internal.processors.cache.transactions.IgniteTxKey;
import org.apache.ignite.internal.processors.cache.transactions.IgniteTxLocalEx;
import org.apache.ignite.internal.processors.cache.transactions.TxDeadlock;
import org.apache.ignite.internal.processors.cache.version.GridCacheVersion;
import org.apache.ignite.internal.processors.timeout.GridTimeoutObjectAdapter;
import org.apache.ignite.internal.transactions.IgniteTxTimeoutCheckedException;
+import org.apache.ignite.internal.util.typedef.internal.CU;
+import org.apache.ignite.transactions.TransactionDeadlockException;
import org.apache.ignite.internal.util.tostring.GridToStringExclude;
import org.apache.ignite.internal.util.tostring.GridToStringInclude;
-import org.apache.ignite.internal.util.typedef.F;
import org.apache.ignite.internal.util.typedef.internal.S;
import org.apache.ignite.internal.util.typedef.internal.U;
import org.apache.ignite.lang.IgniteInClosure;
import org.apache.ignite.lang.IgniteUuid;
-import org.apache.ignite.transactions.TransactionDeadlockException;
import org.jetbrains.annotations.Nullable;
/**
@@ -155,6 +156,12 @@ public final class GridLocalLockFuture<K, V> extends GridCacheFutureAdapter<Bool
if (log == null)
log = U.logger(cctx.kernalContext(), logRef, GridLocalLockFuture.class);
+
+ if (tx != null && tx instanceof GridNearTxLocal && !((GridNearTxLocal)tx).updateLockFuture(null, this)) {
+ GridNearTxLocal tx0 = (GridNearTxLocal)tx;
+
+ onError(tx0.timedOut() ? tx0.timeoutException() : tx0.rollbackException());
+ }
}
/**
@@ -457,6 +464,9 @@ public final class GridLocalLockFuture<K, V> extends GridCacheFutureAdapter<Bool
if (!success)
undoLocks();
+ if (tx != null && success)
+ ((GridNearTxLocal)tx).clearLockFuture(this);
+
if (onDone(success, err)) {
if (log.isDebugEnabled())
log.debug("Completing future: " + this);
@@ -490,47 +500,51 @@ public final class GridLocalLockFuture<K, V> extends GridCacheFutureAdapter<Bool
if (log.isDebugEnabled())
log.debug("Timed out waiting for lock response: " + this);
- if (inTx() && cctx.tm().deadlockDetectionEnabled()) {
- Set<IgniteTxKey> keys = new HashSet<>();
+ if (inTx()) {
+ if (cctx.tm().deadlockDetectionEnabled()) {
+ Set<IgniteTxKey> keys = new HashSet<>();
- List<GridLocalCacheEntry> entries = entries();
+ List<GridLocalCacheEntry> entries = entries();
- for (int i = 0; i < entries.size(); i++) {
- GridLocalCacheEntry e = entries.get(i);
+ for (int i = 0; i < entries.size(); i++) {
+ GridLocalCacheEntry e = entries.get(i);
- List<GridCacheMvccCandidate> mvcc = e.mvccAllLocal();
+ List<GridCacheMvccCandidate> mvcc = e.mvccAllLocal();
- if (mvcc == null)
- continue;
+ if (mvcc == null)
+ continue;
- GridCacheMvccCandidate cand = mvcc.get(0);
+ GridCacheMvccCandidate cand = mvcc.get(0);
- if (cand.owner() && cand.tx() && !cand.version().equals(tx.xidVersion()))
- keys.add(e.txKey());
- }
+ if (cand.owner() && cand.tx() && !cand.version().equals(tx.xidVersion()))
+ keys.add(e.txKey());
+ }
- IgniteInternalFuture<TxDeadlock> fut = cctx.tm().detectDeadlock(tx, keys);
+ IgniteInternalFuture<TxDeadlock> fut = cctx.tm().detectDeadlock(tx, keys);
- fut.listen(new IgniteInClosure<IgniteInternalFuture<TxDeadlock>>() {
- @Override public void apply(IgniteInternalFuture<TxDeadlock> fut) {
- try {
- TxDeadlock deadlock = fut.get();
+ fut.listen(new IgniteInClosure<IgniteInternalFuture<TxDeadlock>>() {
+ @Override public void apply(IgniteInternalFuture<TxDeadlock> fut) {
+ try {
+ TxDeadlock deadlock = fut.get();
- if (deadlock != null)
- ERR_UPD.compareAndSet(GridLocalLockFuture.this, null,
- new IgniteTxTimeoutCheckedException("Failed to acquire lock within provided timeout for " +
- "transaction [timeout=" + tx.timeout() + ", tx=" + tx + ']',
- new TransactionDeadlockException(deadlock.toString(cctx.shared()))));
- }
- catch (IgniteCheckedException e) {
- U.warn(log, "Failed to detect deadlock.", e);
+ err = new IgniteTxTimeoutCheckedException("Failed to acquire lock within provided " +
+ "timeout for transaction [timeout=" + tx.timeout() + ", tx=" + CU.txString(tx) + ']',
+ deadlock != null ? new TransactionDeadlockException(deadlock.toString(cctx.shared())) :
+ null);
+ }
+ catch (IgniteCheckedException e) {
+ err = e;
- ERR_UPD.compareAndSet(GridLocalLockFuture.this, null, e);
- }
+ U.warn(log, "Failed to detect deadlock.", e);
+ }
- onComplete(false);
- }
- });
+ onComplete(false);
+ }
+ });
+ }
+ else
+ err = new IgniteTxTimeoutCheckedException("Failed to acquire lock within provided " +
+ "timeout for transaction [timeout=" + tx.timeout() + ", tx=" + CU.txString(tx) + ']');
}
else
onComplete(false);
@@ -541,4 +555,4 @@ public final class GridLocalLockFuture<K, V> extends GridCacheFutureAdapter<Bool
return S.toString(LockTimeoutObject.class, this);
}
}
-}
\ No newline at end of file
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/ee5fbce3/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/GridCacheDatabaseSharedManager.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/GridCacheDatabaseSharedManager.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/GridCacheDatabaseSharedManager.java
index f24f5d8..d5da199 100755
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/GridCacheDatabaseSharedManager.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/GridCacheDatabaseSharedManager.java
@@ -2369,7 +2369,6 @@ public class GridCacheDatabaseSharedManager extends IgniteCacheDatabaseSharedMan
if (stateId == GridDhtPartitionState.OWNING.ordinal()
|| (stateId == GridDhtPartitionState.MOVING.ordinal()
-
&& part.initialUpdateCounter() < restore.get2())) {
part.initialUpdateCounter(restore.get2());
http://git-wip-us.apache.org/repos/asf/ignite/blob/ee5fbce3/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/store/GridCacheStoreManagerAdapter.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/store/GridCacheStoreManagerAdapter.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/store/GridCacheStoreManagerAdapter.java
index 6c5922a..64a052d 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/store/GridCacheStoreManagerAdapter.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/store/GridCacheStoreManagerAdapter.java
@@ -1431,6 +1431,11 @@ public abstract class GridCacheStoreManagerAdapter extends GridCacheManagerAdapt
}
/** {@inheritDoc} */
+ @Nullable @Override public String label() {
+ return null;
+ }
+
+ /** {@inheritDoc} */
@Override public long timeout() {
return tx.timeout();
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/ee5fbce3/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTransactionsImpl.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTransactionsImpl.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTransactionsImpl.java
index 12f655a..25ba849 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTransactionsImpl.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTransactionsImpl.java
@@ -17,13 +17,18 @@
package org.apache.ignite.internal.processors.cache.transactions;
+import java.util.Collection;
+import org.apache.ignite.IgniteTransactions;
import org.apache.ignite.configuration.TransactionConfiguration;
import org.apache.ignite.internal.IgniteTransactionsEx;
import org.apache.ignite.internal.processors.cache.GridCacheContext;
import org.apache.ignite.internal.processors.cache.GridCacheSharedContext;
import org.apache.ignite.internal.processors.cache.distributed.near.GridNearTxLocal;
+import org.apache.ignite.internal.util.typedef.F;
import org.apache.ignite.internal.util.typedef.internal.A;
import org.apache.ignite.internal.util.typedef.internal.CU;
+import org.apache.ignite.lang.IgniteClosure;
+import org.apache.ignite.lang.IgnitePredicate;
import org.apache.ignite.transactions.Transaction;
import org.apache.ignite.transactions.TransactionConcurrency;
import org.apache.ignite.transactions.TransactionIsolation;
@@ -38,11 +43,15 @@ public class IgniteTransactionsImpl<K, V> implements IgniteTransactionsEx {
/** Cache shared context. */
private GridCacheSharedContext<K, V> cctx;
+ /** Label. */
+ private String lb;
+
/**
* @param cctx Cache shared context.
*/
- public IgniteTransactionsImpl(GridCacheSharedContext<K, V> cctx) {
+ public IgniteTransactionsImpl(GridCacheSharedContext<K, V> cctx, @Nullable String lb) {
this.cctx = cctx;
+ this.lb = lb;
}
/** {@inheritDoc} */
@@ -166,7 +175,8 @@ public class IgniteTransactionsImpl<K, V> implements IgniteTransactionsEx {
isolation,
timeout,
true,
- txSize
+ txSize,
+ lb
);
assert tx != null;
@@ -195,6 +205,26 @@ public class IgniteTransactionsImpl<K, V> implements IgniteTransactionsEx {
cctx.resetTxMetrics();
}
+ /** {@inheritDoc} */
+ @Override public Collection<Transaction> localActiveTransactions() {
+ return F.viewReadOnly(cctx.tm().activeTransactions(), new IgniteClosure<IgniteInternalTx, Transaction>() {
+ @Override public Transaction apply(IgniteInternalTx tx) {
+ return ((GridNearTxLocal)tx).rollbackOnlyProxy();
+ }
+ }, new IgnitePredicate<IgniteInternalTx>() {
+ @Override public boolean apply(IgniteInternalTx tx) {
+ return tx.local() && tx.near();
+ }
+ });
+ }
+
+ /** {@inheritDoc} */
+ @Override public IgniteTransactions withLabel(String lb) {
+ A.notNull(lb, "label should not be empty.");
+
+ return new IgniteTransactionsImpl<>(cctx, lb);
+ }
+
/**
* @param ctx Cache context.
*/
@@ -202,4 +232,4 @@ public class IgniteTransactionsImpl<K, V> implements IgniteTransactionsEx {
if (!ctx.transactional())
throw new TransactionException("Failed to start transaction on non-transactional cache: " + ctx.name());
}
-}
\ No newline at end of file
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/ee5fbce3/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxAdapter.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxAdapter.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxAdapter.java
index 945ef48..3cf1146 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxAdapter.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxAdapter.java
@@ -65,6 +65,7 @@ import org.apache.ignite.internal.processors.cache.version.GridCacheLazyPlainVer
import org.apache.ignite.internal.processors.cache.version.GridCacheVersion;
import org.apache.ignite.internal.processors.cache.version.GridCacheVersionConflictContext;
import org.apache.ignite.internal.processors.cache.version.GridCacheVersionedEntryEx;
+import org.apache.ignite.internal.transactions.IgniteTxRollbackCheckedException;
import org.apache.ignite.internal.processors.cluster.BaselineTopology;
import org.apache.ignite.internal.transactions.IgniteTxTimeoutCheckedException;
import org.apache.ignite.internal.util.GridSetWrapper;
@@ -706,7 +707,15 @@ public abstract class IgniteTxAdapter extends GridMetadataAwareAdapter implement
*/
public final IgniteCheckedException timeoutException() {
return new IgniteTxTimeoutCheckedException("Failed to acquire lock within provided timeout " +
- "for transaction [timeout=" + timeout() + ", tx=" + this + ']');
+ "for transaction [timeout=" + timeout() + ", tx=" + CU.txString(this) + ']');
+ }
+
+ /**
+ * @return Rollback exception.
+ */
+ public final IgniteCheckedException rollbackException() {
+ return new IgniteTxRollbackCheckedException("Failed to finish transaction because it has been rolled back " +
+ "[timeout=" + timeout() + ", tx=" + CU.txString(this) + ']');
}
/** {@inheritDoc} */
@@ -993,9 +1002,9 @@ public abstract class IgniteTxAdapter extends GridMetadataAwareAdapter implement
*
* @param state State to set.
* @param timedOut Timeout flag.
+ *
* @return {@code True} if state changed.
*/
- @SuppressWarnings({"TooBroadScope"})
protected final boolean state(TransactionState state, boolean timedOut) {
boolean valid = false;
@@ -1058,7 +1067,7 @@ public abstract class IgniteTxAdapter extends GridMetadataAwareAdapter implement
}
case MARKED_ROLLBACK: {
- valid = prev == ACTIVE || prev == PREPARING || prev == PREPARED || prev == SUSPENDED;
+ valid = prev == ACTIVE || prev == PREPARING || prev == PREPARED || prev == SUSPENDED;
break;
}
@@ -1099,11 +1108,9 @@ public abstract class IgniteTxAdapter extends GridMetadataAwareAdapter implement
if (state != ACTIVE && state != SUSPENDED)
seal();
- if (cctx.wal() != null && cctx.tm().logTxRecords()) {
+ if (cctx.wal() != null && cctx.tm().logTxRecords() && txNodes != null) {
// Log tx state change to WAL.
if (state == PREPARED || state == COMMITTED || state == ROLLED_BACK) {
- assert txNodes != null || state == ROLLED_BACK : "txNodes=" + txNodes + " state=" + state;
-
BaselineTopology baselineTop = cctx.kernalContext().state().clusterState().baselineTopology();
Map<Short, Collection<Short>> participatingNodes = consistentIdMapper
http://git-wip-us.apache.org/repos/asf/ignite/blob/ee5fbce3/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxHandler.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxHandler.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxHandler.java
index 38c877b..44482ab 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxHandler.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxHandler.java
@@ -256,9 +256,6 @@ public class IgniteTxHandler {
IgniteInternalFuture<GridNearTxPrepareResponse> fut = locTx.prepareAsyncLocal(req);
- if (locTx.isRollbackOnly())
- locTx.rollbackNearTxLocalAsync();
-
return fut.chain(new C1<IgniteInternalFuture<GridNearTxPrepareResponse>, GridNearTxPrepareResponse>() {
@Override public GridNearTxPrepareResponse apply(IgniteInternalFuture<GridNearTxPrepareResponse> f) {
try {
@@ -885,7 +882,7 @@ public class IgniteTxHandler {
req.threadId(),
req.futureId(),
req.miniId(),
- new IgniteCheckedException("Transaction has been already completed."));
+ new IgniteTxRollbackCheckedException("Transaction has been already completed or not started yet."));
try {
ctx.io().send(nodeId, res, req.policy());
@@ -1698,6 +1695,8 @@ public class IgniteTxHandler {
res.invalidPartitionsByCacheId(tx.invalidPartitions());
if (tx.empty() && req.last()) {
+ tx.skipCompletedVersions(req.skipCompletedVersion());
+
tx.rollbackRemoteTx();
return null;
http://git-wip-us.apache.org/repos/asf/ignite/blob/ee5fbce3/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 7a7f65f..54b7ae2 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
@@ -18,7 +18,6 @@
package org.apache.ignite.internal.processors.cache.transactions;
import java.io.Externalizable;
-import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.Map;
@@ -963,7 +962,7 @@ public abstract class IgniteTxLocalAdapter extends IgniteTxAdapter implements Ig
if (commit)
cctx.tm().commitTx(this);
else
- cctx.tm().rollbackTx(this, clearThreadMap);
+ cctx.tm().rollbackTx(this, clearThreadMap, false);
}
state(commit ? COMMITTED : ROLLED_BACK);
@@ -1028,7 +1027,7 @@ public abstract class IgniteTxLocalAdapter extends IgniteTxAdapter implements Ig
}
if (DONE_FLAG_UPD.compareAndSet(this, 0, 1)) {
- cctx.tm().rollbackTx(this, clearThreadMap);
+ cctx.tm().rollbackTx(this, clearThreadMap, false);
if (!internal()) {
Collection<CacheStoreManager> stores = txState.stores(cctx);
@@ -1310,7 +1309,7 @@ public abstract class IgniteTxLocalAdapter extends IgniteTxAdapter implements Ig
if (isRollbackOnly()) {
if (remainingTime() == -1)
- throw new IgniteTxTimeoutCheckedException("Cache transaction timed out: " + this);
+ throw new IgniteTxTimeoutCheckedException("Cache transaction timed out: " + CU.txString(this));
TransactionState state = state();
@@ -1322,7 +1321,7 @@ public abstract class IgniteTxLocalAdapter extends IgniteTxAdapter implements Ig
throw new IgniteTxHeuristicCheckedException("Cache transaction is in unknown state " +
"(remote transactions will be invalidated): " + this);
- throw new IgniteCheckedException("Cache transaction marked as rollback-only: " + this);
+ throw rollbackException();
}
}
@@ -1367,12 +1366,6 @@ public abstract class IgniteTxLocalAdapter extends IgniteTxAdapter implements Ig
checkInternal(key);
- TransactionState state = state();
-
- assert state == TransactionState.ACTIVE || remainingTime() == -1 :
- "Invalid tx state for adding entry [op=" + op + ", val=" + val + ", entry=" + entry + ", filter=" +
- Arrays.toString(filter) + ", txCtx=" + cctx.tm().txContextVersion() + ", tx=" + this + ']';
-
IgniteTxEntry old = entry(key);
// Keep old filter if already have one (empty filter is always overridden).
@@ -1692,7 +1685,8 @@ public abstract class IgniteTxLocalAdapter extends IgniteTxAdapter implements Ig
final GridClosureException ex = new GridClosureException(
new IgniteTxTimeoutCheckedException("Failed to acquire lock within provided timeout " +
- "for transaction [timeout=" + timeout() + ", tx=" + this + ']', deadlockErr)
+ "for transaction [timeout=" + timeout() + ", tx=" + CU.txString(IgniteTxLocalAdapter.this) +
+ ']', deadlockErr)
);
if (commit && commitAfterLock())
@@ -1773,7 +1767,7 @@ public abstract class IgniteTxLocalAdapter extends IgniteTxAdapter implements Ig
if (!locked)
throw new GridClosureException(new IgniteTxTimeoutCheckedException("Failed to acquire lock " +
"within provided timeout for transaction [timeout=" + timeout() +
- ", tx=" + IgniteTxLocalAdapter.this + ']'));
+ ", tx=" + CU.txString(IgniteTxLocalAdapter.this) + ']'));
IgniteInternalFuture<T> fut = postLock();
http://git-wip-us.apache.org/repos/asf/ignite/blob/ee5fbce3/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 7fa31bf..dc8bc46 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
@@ -294,6 +294,22 @@ public class IgniteTxManager extends GridCacheSharedManagerAdapter {
}
/**
+ * Rollback transactions blocking partition map exchange.
+ *
+ * @param topVer Initial exchange version.
+ */
+ public void rollbackOnTopologyChange(AffinityTopologyVersion topVer) {
+ for (IgniteInternalTx tx : activeTransactions()) {
+ if (tx.local() && tx.near() && needWaitTransaction(tx, topVer)) {
+ U.warn(log, "The transaction was forcibly rolled back on partition map exchange because a timeout is " +
+ "reached: [tx=" + CU.txString(tx) + ", topVer=" + topVer + ']');
+
+ ((GridNearTxLocal)tx).rollbackNearTxLocalAsync(false, false);
+ }
+ }
+ }
+
+ /**
* @param cacheId Cache ID.
* @param txMap Transactions map.
*/
@@ -303,7 +319,7 @@ public class IgniteTxManager extends GridCacheSharedManagerAdapter {
for (IgniteTxEntry entry : tx.allEntries()) {
if (entry.cacheId() == cacheId) {
- rollbackTx(tx, false);
+ rollbackTx(tx, false, false);
break;
}
@@ -316,9 +332,9 @@ public class IgniteTxManager extends GridCacheSharedManagerAdapter {
txFinishSync.onDisconnected(reconnectFut);
for (IgniteInternalTx tx : idMap.values())
- rollbackTx(tx, true);
+ rollbackTx(tx, true, false);
for (IgniteInternalTx tx : nearIdMap.values())
- rollbackTx(tx, true);
+ rollbackTx(tx, true, false);
IgniteClientDisconnectedException err =
new IgniteClientDisconnectedException(reconnectFut, "Client node disconnected.");
@@ -414,8 +430,9 @@ public class IgniteTxManager extends GridCacheSharedManagerAdapter {
private boolean isCompleted(IgniteInternalTx tx) {
boolean completed = completedVersHashMap.containsKey(tx.xidVersion());
- // Need check that for tx with timeout rollback message was not received before lock.
- if (!completed && tx.local() && tx.dht() && tx.timeout() > 0)
+ // Need check that for tx rollback message was not received before lock.
+ // This could happen on timeout or async rollback.
+ if (!completed && tx.local() && tx.dht())
return completedVersHashMap.containsKey(tx.nearXidVersion());
return completed;
@@ -428,6 +445,7 @@ public class IgniteTxManager extends GridCacheSharedManagerAdapter {
* @param isolation Isolation.
* @param timeout transaction timeout.
* @param txSize Expected transaction size.
+ * @param lb Label.
* @return New transaction.
*/
public GridNearTxLocal newTx(
@@ -438,7 +456,8 @@ public class IgniteTxManager extends GridCacheSharedManagerAdapter {
TransactionIsolation isolation,
long timeout,
boolean storeEnabled,
- int txSize
+ int txSize,
+ @Nullable String lb
) {
assert sysCacheCtx == null || sysCacheCtx.systemTx();
@@ -458,7 +477,8 @@ public class IgniteTxManager extends GridCacheSharedManagerAdapter {
storeEnabled,
txSize,
subjId,
- taskNameHash);
+ taskNameHash,
+ lb);
if (tx.system()) {
AffinityTopologyVersion topVer = cctx.tm().lockedTopologyVersion(Thread.currentThread().getId(), tx);
@@ -560,10 +580,9 @@ public class IgniteTxManager extends GridCacheSharedManagerAdapter {
}
});
- for (IgniteInternalTx tx : txs()) {
- if (needWaitTransaction(tx, topVer)) {
+ for (IgniteInternalTx tx : activeTransactions()) {
+ if (needWaitTransaction(tx, topVer))
res.add(tx.finishFuture());
- }
}
res.markInitialized();
@@ -960,7 +979,7 @@ public class IgniteTxManager extends GridCacheSharedManagerAdapter {
/**
* @param tx Committed transaction.
- * @return If transaction was not already present in committed set.
+ * @return If transaction was not already present in completed set.
*/
public boolean addRolledbackTx(IgniteInternalTx tx) {
return addRolledbackTx(tx, tx.xidVersion());
@@ -1293,15 +1312,17 @@ public class IgniteTxManager extends GridCacheSharedManagerAdapter {
*
* @param tx Transaction to rollback.
* @param clearThreadMap {@code True} if need remove tx from thread map.
+ * @param skipCompletedVers {@code True} if tx should skip adding itself to completed versions map on finish.
*/
- public void rollbackTx(IgniteInternalTx tx, boolean clearThreadMap) {
+ public void rollbackTx(IgniteInternalTx tx, boolean clearThreadMap, boolean skipCompletedVers) {
assert tx != null;
if (log.isDebugEnabled())
log.debug("Rolling back from TM [locNodeId=" + cctx.localNodeId() + ", tx=" + tx + ']');
// 1. Record transaction version to avoid duplicates.
- addRolledbackTx(tx);
+ if (!skipCompletedVers)
+ addRolledbackTx(tx);
ConcurrentMap<GridCacheVersion, IgniteInternalTx> txIdMap = transactionMap(tx);
@@ -1355,10 +1376,11 @@ public class IgniteTxManager extends GridCacheSharedManagerAdapter {
*
* @param tx Transaction to finish.
* @param commit {@code True} if transaction is committed, {@code false} if rolled back.
+ * @param clearThreadMap {@code True} if need remove tx from thread map.
*/
public void fastFinishTx(GridNearTxLocal tx, boolean commit, boolean clearThreadMap) {
assert tx != null;
- assert tx.writeMap().isEmpty();
+ tx.writeMap().isEmpty();
assert tx.optimistic() || tx.readMap().isEmpty();
ConcurrentMap<GridCacheVersion, IgniteInternalTx> txIdMap = transactionMap(tx);
@@ -1683,7 +1705,14 @@ public class IgniteTxManager extends GridCacheSharedManagerAdapter {
private void txUnlock(IgniteInternalTx tx, IgniteTxEntry txEntry) {
while (true) {
try {
- txEntry.cached().txUnlock(tx);
+ GridCacheEntryEx entry = txEntry.cached();
+
+ assert entry != null;
+
+ if (entry.detached())
+ break;
+
+ entry.txUnlock(tx);
break;
}
@@ -1691,14 +1720,7 @@ public class IgniteTxManager extends GridCacheSharedManagerAdapter {
if (log.isDebugEnabled())
log.debug("Got removed entry in TM txUnlock(..) method (will retry): " + txEntry);
- try {
- txEntry.cached(txEntry.context().cache().entryEx(txEntry.key(), tx.topologyVersion()));
- }
- catch (GridDhtInvalidPartitionException e) {
- tx.addInvalidPartition(txEntry.context(), e.partition());
-
- break;
- }
+ txEntry.cached(txEntry.context().cache().entryEx(txEntry.key(), tx.topologyVersion()));
}
}
}
@@ -1708,31 +1730,8 @@ public class IgniteTxManager extends GridCacheSharedManagerAdapter {
* @param entries Entries to unlock.
*/
private void unlockMultiple(IgniteInternalTx tx, Iterable<IgniteTxEntry> entries) {
- for (IgniteTxEntry txEntry : entries) {
- GridCacheContext cacheCtx = txEntry.context();
-
- while (true) {
- try {
- GridCacheEntryEx entry = txEntry.cached();
-
- assert entry != null;
-
- if (entry.detached())
- break;
-
- entry.txUnlock(tx);
-
- break;
- }
- catch (GridCacheEntryRemovedException ignored) {
- if (log.isDebugEnabled())
- log.debug("Got removed entry in TM unlockMultiple(..) method (will retry): " + txEntry);
-
- // Renew cache entry.
- txEntry.cached(cacheCtx.cache().entryEx(txEntry.key(), tx.topologyVersion()));
- }
- }
- }
+ for (IgniteTxEntry txEntry : entries)
+ txUnlock(tx, txEntry);
}
/**
@@ -1772,13 +1771,6 @@ public class IgniteTxManager extends GridCacheSharedManagerAdapter {
}
/**
- * @return All transactions.
- */
- public Collection<IgniteInternalTx> txs() {
- return F.concat(false, idMap.values(), nearIdMap.values());
- }
-
- /**
* @return Slow tx warn timeout.
*/
public int slowTxWarnTimeout() {
@@ -1859,12 +1851,12 @@ public class IgniteTxManager extends GridCacheSharedManagerAdapter {
* @return Finish future for related remote transactions.
*/
@SuppressWarnings("unchecked")
- public IgniteInternalFuture<IgniteInternalTx> remoteTxFinishFuture(GridCacheVersion nearVer) {
- GridCompoundFuture<IgniteInternalTx, IgniteInternalTx> fut = new GridCompoundFuture<>();
+ public IgniteInternalFuture<?> remoteTxFinishFuture(GridCacheVersion nearVer) {
+ GridCompoundFuture<Void, Void> fut = new GridCompoundFuture<>();
- for (final IgniteInternalTx tx : txs()) {
+ for (final IgniteInternalTx tx : activeTransactions()) {
if (!tx.local() && nearVer.equals(tx.nearXidVersion()))
- fut.add(tx.finishFuture());
+ fut.add((IgniteInternalFuture) tx.finishFuture());
}
fut.markInitialized();
@@ -1884,7 +1876,7 @@ public class IgniteTxManager extends GridCacheSharedManagerAdapter {
@Nullable GridFutureAdapter<Boolean> fut,
@Nullable Collection<GridCacheVersion> processedVers)
{
- for (final IgniteInternalTx tx : txs()) {
+ for (final IgniteInternalTx tx : activeTransactions()) {
if (nearVer.equals(tx.nearXidVersion())) {
IgniteInternalFuture<?> prepFut = tx.currentPrepareFuture();
@@ -2308,7 +2300,7 @@ public class IgniteTxManager extends GridCacheSharedManagerAdapter {
*
* @param tx Transaction to be suspended.
*
- * @see #resumeTx(GridNearTxLocal)
+ * @see #resumeTx(GridNearTxLocal, long)
* @see GridNearTxLocal#suspend()
* @see GridNearTxLocal#resume()
* @throws IgniteCheckedException If failed to suspend transaction.
@@ -2331,13 +2323,14 @@ public class IgniteTxManager extends GridCacheSharedManagerAdapter {
* Please don't use directly. Use tx.resume() instead.
*
* @param tx Transaction to be resumed.
+ * @param threadId Thread id to restore.
*
* @see #suspendTx(GridNearTxLocal)
* @see GridNearTxLocal#suspend()
* @see GridNearTxLocal#resume()
* @throws IgniteCheckedException If failed to resume tx.
*/
- public void resumeTx(GridNearTxLocal tx) throws IgniteCheckedException {
+ public void resumeTx(GridNearTxLocal tx, long threadId) throws IgniteCheckedException {
assert tx != null && !tx.system() : tx;
if (!tx.state(ACTIVE)) {
@@ -2349,8 +2342,6 @@ public class IgniteTxManager extends GridCacheSharedManagerAdapter {
assert !transactionMap(tx).containsValue(tx) : tx;
assert !haveSystemTxForThread(Thread.currentThread().getId());
- long threadId = Thread.currentThread().getId();
-
if (threadMap.putIfAbsent(threadId, tx) != null)
throw new IgniteCheckedException("Thread already has started a transaction.");
@@ -2420,7 +2411,7 @@ public class IgniteTxManager extends GridCacheSharedManagerAdapter {
log.debug("Processing node failed event [locNodeId=" + cctx.localNodeId() +
", failedNodeId=" + evtNodeId + ']');
- for (final IgniteInternalTx tx : txs()) {
+ for (final IgniteInternalTx tx : activeTransactions()) {
if ((tx.near() && !tx.local()) || (tx.storeWriteThrough() && tx.masterNodeIds().contains(evtNodeId))) {
// Invalidate transactions.
salvageTx(tx, RECOVERY_FINISH);
http://git-wip-us.apache.org/repos/asf/ignite/blob/ee5fbce3/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/TransactionProxyImpl.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/TransactionProxyImpl.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/TransactionProxyImpl.java
index f25fc36..628b813 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/TransactionProxyImpl.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/TransactionProxyImpl.java
@@ -43,6 +43,7 @@ import org.apache.ignite.transactions.Transaction;
import org.apache.ignite.transactions.TransactionConcurrency;
import org.apache.ignite.transactions.TransactionIsolation;
import org.apache.ignite.transactions.TransactionState;
+import org.jetbrains.annotations.Nullable;
import static org.apache.ignite.transactions.TransactionState.SUSPENDED;
@@ -233,6 +234,14 @@ public class TransactionProxyImpl<K, V> implements TransactionProxy, Externaliza
}
/** {@inheritDoc} */
+ @Nullable @Override public String label() {
+ if (async)
+ save(tx.label());
+
+ return tx.label();
+ }
+
+ /** {@inheritDoc} */
@Override public long timeout(long timeout) {
return tx.timeout(timeout);
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/ee5fbce3/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/TransactionProxyRollbackOnlyImpl.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/TransactionProxyRollbackOnlyImpl.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/TransactionProxyRollbackOnlyImpl.java
new file mode 100644
index 0000000..74735f9
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/TransactionProxyRollbackOnlyImpl.java
@@ -0,0 +1,80 @@
+/*
+ * 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 java.io.Externalizable;
+import org.apache.ignite.IgniteException;
+import org.apache.ignite.internal.processors.cache.GridCacheSharedContext;
+import org.apache.ignite.internal.processors.cache.distributed.near.GridNearTxLocal;
+import org.apache.ignite.lang.IgniteFuture;
+
+/**
+ * Cache transaction proxy which support only rollback or close operations and getters.
+ */
+@SuppressWarnings("unchecked")
+public class TransactionProxyRollbackOnlyImpl<K, V> extends TransactionProxyImpl<K, V> {
+ /** */
+ private static final long serialVersionUID = 0L;
+
+ /**
+ * Empty constructor required for {@link Externalizable}.
+ */
+ public TransactionProxyRollbackOnlyImpl() {
+ // No-op.
+ }
+
+ /**
+ * @param tx Tx.
+ * @param cctx Cctx.
+ * @param async Async.
+ */
+ public TransactionProxyRollbackOnlyImpl(GridNearTxLocal tx,
+ GridCacheSharedContext<K, V> cctx, boolean async) {
+ super(tx, cctx, async);
+ }
+
+ /** {@inheritDoc} */
+ @Override public boolean setRollbackOnly() {
+ throw new UnsupportedOperationException();
+ }
+
+ /** {@inheritDoc} */
+ @Override public void commit() {
+ throw new UnsupportedOperationException();
+ }
+
+ /** {@inheritDoc} */
+ @Override public IgniteFuture<Void> commitAsync() throws IgniteException {
+ throw new UnsupportedOperationException();
+ }
+
+ /** {@inheritDoc} */
+ @Override public void suspend() throws IgniteException {
+ throw new UnsupportedOperationException();
+ }
+
+ /** {@inheritDoc} */
+ @Override public void resume() throws IgniteException {
+ throw new UnsupportedOperationException();
+ }
+
+ /** {@inheritDoc} */
+ @Override public long timeout(long timeout) {
+ throw new UnsupportedOperationException();
+ }
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/ee5fbce3/modules/core/src/main/java/org/apache/ignite/internal/util/GridPartitionStateMap.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/util/GridPartitionStateMap.java b/modules/core/src/main/java/org/apache/ignite/internal/util/GridPartitionStateMap.java
index 173ddbe..f5893c0 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/util/GridPartitionStateMap.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/util/GridPartitionStateMap.java
@@ -52,7 +52,7 @@ public class GridPartitionStateMap extends AbstractMap<Integer, GridDhtPartition
@Override public Set<Entry<Integer, GridDhtPartitionState>> entrySet() {
return new AbstractSet<Entry<Integer, GridDhtPartitionState>>() {
@Override public Iterator<Entry<Integer, GridDhtPartitionState>> iterator() {
- final int size = states.isEmpty() ? 0 : (states.length() - 1)/ BITS + 1;
+ final int size = states.isEmpty() ? 0 : (states.length() - 1) / BITS + 1;
return new Iterator<Entry<Integer, GridDhtPartitionState>>() {
private int next;
http://git-wip-us.apache.org/repos/asf/ignite/blob/ee5fbce3/modules/core/src/main/java/org/apache/ignite/internal/visor/tx/VisorTxInfo.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/visor/tx/VisorTxInfo.java b/modules/core/src/main/java/org/apache/ignite/internal/visor/tx/VisorTxInfo.java
new file mode 100644
index 0000000..c51d43f
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/visor/tx/VisorTxInfo.java
@@ -0,0 +1,130 @@
+/*
+ * 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.visor.tx;
+
+import java.io.Serializable;
+import java.util.Collection;
+import java.util.UUID;
+import org.apache.ignite.lang.IgniteUuid;
+import org.apache.ignite.transactions.TransactionConcurrency;
+import org.apache.ignite.transactions.TransactionIsolation;
+import org.apache.ignite.transactions.TransactionState;
+
+/**
+ */
+public class VisorTxInfo implements Serializable {
+ /** */
+ private static final long serialVersionUID = 0L;
+
+ /** */
+ private final IgniteUuid xid;
+
+ /** */
+ private final long duration;
+
+ /** */
+ private final TransactionIsolation isolation;
+
+ /** */
+ private final TransactionConcurrency concurrency;
+
+ /** */
+ private final long timeout;
+
+ /** */
+ private final String lb;
+
+ /** */
+ private final Collection<UUID> primaryNodes;
+
+ /** */
+ private final TransactionState state;
+
+ /** */
+ private int size;
+
+ /**
+ * @param xid Xid.
+ * @param duration Duration.
+ * @param isolation Isolation.
+ * @param concurrency Concurrency.
+ * @param timeout Timeout.
+ * @param lb Label.
+ * @param primaryNodes Primary nodes.
+ * @param state State.
+ * @param size Size.
+ */
+ public VisorTxInfo(IgniteUuid xid, long duration, TransactionIsolation isolation,
+ TransactionConcurrency concurrency, long timeout, String lb, Collection<UUID> primaryNodes,
+ TransactionState state, int size) {
+ this.xid = xid;
+ this.duration = duration;
+ this.isolation = isolation;
+ this.concurrency = concurrency;
+ this.timeout = timeout;
+ this.lb = lb;
+ this.primaryNodes = primaryNodes;
+ this.state = state;
+ this.size = size;
+ }
+
+ /** */
+ public IgniteUuid getXid() {
+ return xid;
+ }
+
+ /** */
+ public long getDuration() {
+ return duration;
+ }
+
+ /** */
+ public TransactionIsolation getIsolation() {
+ return isolation;
+ }
+
+ /** */
+ public TransactionConcurrency getConcurrency() {
+ return concurrency;
+ }
+
+ /** */
+ public long getTimeout() {
+ return timeout;
+ }
+
+ /** */
+ public String getLabel() {
+ return lb;
+ }
+
+ /** */
+ public Collection<UUID> getPrimaryNodes() {
+ return primaryNodes;
+ }
+
+ /** */
+ public TransactionState getState() {
+ return state;
+ }
+
+ /** */
+ public int getSize() {
+ return size;
+ }
+}