Repository: ignite Updated Branches: refs/heads/master 2dc0d9f75 -> 03dd9eb78
IGNITE-10434: MVCC: Fixed races on asynchronous TX rollback. This closes #5569. Project: http://git-wip-us.apache.org/repos/asf/ignite/repo Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/03dd9eb7 Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/03dd9eb7 Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/03dd9eb7 Branch: refs/heads/master Commit: 03dd9eb7813a1361715e08dc90d64c15eb3dca64 Parents: 2dc0d9f Author: Igor Seliverstov <[email protected]> Authored: Fri Dec 28 14:34:05 2018 +0300 Committer: Igor Seliverstov <[email protected]> Committed: Fri Dec 28 14:34:05 2018 +0300 ---------------------------------------------------------------------- .../GridDistributedTxRemoteAdapter.java | 19 +----- .../dht/GridDhtTxAbstractEnlistFuture.java | 2 + .../dht/NearTxQueryEnlistResultHandler.java | 2 +- .../near/GridNearTxAbstractEnlistFuture.java | 10 +-- .../near/GridNearTxQueryEnlistFuture.java | 3 +- .../cache/transactions/IgniteInternalTx.java | 3 +- .../cache/transactions/IgniteTxAdapter.java | 23 ++++++- .../cache/transactions/IgniteTxHandler.java | 66 ++++++++------------ .../transactions/IgniteTxLocalAdapter.java | 13 +--- .../cache/transactions/TxRollbackAsyncTest.java | 3 - 10 files changed, 62 insertions(+), 82 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/ignite/blob/03dd9eb7/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedTxRemoteAdapter.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedTxRemoteAdapter.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedTxRemoteAdapter.java index a915478..c4255fb 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedTxRemoteAdapter.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedTxRemoteAdapter.java @@ -121,9 +121,6 @@ public abstract class GridDistributedTxRemoteAdapter extends IgniteTxAdapter @GridToStringInclude protected IgniteTxRemoteState txState; - /** {@code True} if tx should skip adding itself to completed version map on finish. */ - private boolean skipCompletedVers; - /** Transaction label. */ @GridToStringInclude @Nullable private String txLbl; @@ -918,7 +915,7 @@ public abstract class GridDistributedTxRemoteAdapter extends IgniteTxAdapter // Note that we don't evict near entries here - // they will be deleted by their corresponding transactions. if (state(ROLLING_BACK) || state() == UNKNOWN) { - cctx.tm().rollbackTx(this, false, skipCompletedVers); + cctx.tm().rollbackTx(this, false, skipCompletedVersions()); TxCounters counters = txCounters(false); @@ -964,20 +961,6 @@ public abstract class GridDistributedTxRemoteAdapter extends IgniteTxAdapter } /** - * @return {@code True} if tx should skip adding itself to completed version map on finish. - */ - public boolean skipCompletedVersions() { - return skipCompletedVers; - } - - /** - * @param skipCompletedVers {@code True} if tx should skip adding itself to completed version map on finish. - */ - public void skipCompletedVersions(boolean skipCompletedVers) { - this.skipCompletedVers = skipCompletedVers; - } - - /** * Adds explicit version if there is one. * * @param e Transaction entry. http://git-wip-us.apache.org/repos/asf/ignite/blob/03dd9eb7/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxAbstractEnlistFuture.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxAbstractEnlistFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxAbstractEnlistFuture.java index 2864712..936758d 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxAbstractEnlistFuture.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxAbstractEnlistFuture.java @@ -898,6 +898,8 @@ public abstract class GridDhtTxAbstractEnlistFuture<T> extends GridCacheFutureAd m.put(node.id(), mapping = new GridDistributedTxMapping(node)); mapping.markQueryUpdate(); + + checkCompleted(); } /** */ http://git-wip-us.apache.org/repos/asf/ignite/blob/03dd9eb7/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/NearTxQueryEnlistResultHandler.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/NearTxQueryEnlistResultHandler.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/NearTxQueryEnlistResultHandler.java index 00289ca..72f0173 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/NearTxQueryEnlistResultHandler.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/NearTxQueryEnlistResultHandler.java @@ -124,7 +124,7 @@ public final class NearTxQueryEnlistResultHandler implements CI1<IgniteInternalF GridNearTxQueryEnlistResponse res = createResponse(fut); if (res.removeMapping()) { - tx.forceSkipCompletedVersions(); + tx.skipCompletedVersions(true); tx.rollbackDhtLocalAsync().listen(new CI1<IgniteInternalFuture<IgniteInternalTx>>() { @Override public void apply(IgniteInternalFuture<IgniteInternalTx> fut0) { http://git-wip-us.apache.org/repos/asf/ignite/blob/03dd9eb7/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxAbstractEnlistFuture.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxAbstractEnlistFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxAbstractEnlistFuture.java index e93834b..b52d440 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxAbstractEnlistFuture.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxAbstractEnlistFuture.java @@ -275,6 +275,8 @@ public abstract class GridNearTxAbstractEnlistFuture<T> extends GridCacheCompoun if (node.isLocal()) tx.colocatedLocallyMapped(true); + + checkCompleted(); } /** @@ -395,15 +397,15 @@ public abstract class GridNearTxAbstractEnlistFuture<T> extends GridCacheCompoun tx.setRollbackOnly(); synchronized (this) { - boolean done = super.onDone(res, err, cancelled); - - assert done; - GridDhtTxAbstractEnlistFuture localFuture0 = localEnlistFuture; if (localFuture0 != null && (err != null || cancelled)) localFuture0.onDone(cancelled ? new IgniteFutureCancelledCheckedException("Future was cancelled: " + localFuture0) : err); + boolean done = super.onDone(res, err, cancelled); + + assert done; + // Clean up. cctx.mvcc().removeVersionedFuture(this); http://git-wip-us.apache.org/repos/asf/ignite/blob/03dd9eb7/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxQueryEnlistFuture.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxQueryEnlistFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxQueryEnlistFuture.java index 245ef39..b0a83dc 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxQueryEnlistFuture.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxQueryEnlistFuture.java @@ -37,6 +37,7 @@ 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 static org.apache.ignite.internal.managers.communication.GridIoPolicy.QUERY_POOL; import static org.apache.ignite.internal.processors.cache.distributed.dht.NearTxQueryEnlistResultHandler.createResponse; /** @@ -243,7 +244,7 @@ public class GridNearTxQueryEnlistFuture extends GridNearTxQueryAbstractEnlistFu IgniteInternalFuture<?> txSync = cctx.tm().awaitFinishAckAsync(nodeId, tx.threadId()); if (txSync == null || txSync.isDone()) - cctx.io().send(nodeId, req, cctx.ioPolicy()); + cctx.io().send(nodeId, req, QUERY_POOL); // Process query requests in query pool. else txSync.listen(new CI1<IgniteInternalFuture<?>>() { @Override public void apply(IgniteInternalFuture<?> f) { http://git-wip-us.apache.org/repos/asf/ignite/blob/03dd9eb7/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteInternalTx.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteInternalTx.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteInternalTx.java index 7588fb0..bb27ff0 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteInternalTx.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteInternalTx.java @@ -38,6 +38,7 @@ import org.apache.ignite.lang.IgniteUuid; import org.apache.ignite.transactions.TransactionConcurrency; import org.apache.ignite.transactions.TransactionIsolation; import org.apache.ignite.transactions.TransactionState; +import org.jetbrains.annotations.Contract; import org.jetbrains.annotations.Nullable; /** @@ -662,5 +663,5 @@ public interface IgniteInternalTx { * @return Transaction counters. * @param createIfAbsent {@code True} if non-null instance is needed. */ - @Nullable public TxCounters txCounters(boolean createIfAbsent); + @Nullable @Contract("true -> !null;") public TxCounters txCounters(boolean createIfAbsent); } http://git-wip-us.apache.org/repos/asf/ignite/blob/03dd9eb7/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 8be90dd..b392feb 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 @@ -17,8 +17,6 @@ package org.apache.ignite.internal.processors.cache.transactions; -import javax.cache.expiry.ExpiryPolicy; -import javax.cache.processor.EntryProcessor; import java.io.Externalizable; import java.io.IOException; import java.io.ObjectInput; @@ -39,6 +37,8 @@ import java.util.UUID; import java.util.concurrent.atomic.AtomicLong; import java.util.concurrent.atomic.AtomicReference; import java.util.concurrent.atomic.AtomicReferenceFieldUpdater; +import javax.cache.expiry.ExpiryPolicy; +import javax.cache.processor.EntryProcessor; import org.apache.ignite.IgniteCheckedException; import org.apache.ignite.IgniteException; import org.apache.ignite.IgniteLogger; @@ -278,6 +278,9 @@ public abstract class IgniteTxAdapter extends GridMetadataAwareAdapter implement @GridToStringInclude protected volatile MvccSnapshot mvccSnapshot; + /** {@code True} if tx should skip adding itself to completed version map on finish. */ + private boolean skipCompletedVers; + /** Rollback finish future. */ @GridToStringExclude private volatile IgniteInternalFuture rollbackFut; @@ -428,6 +431,20 @@ public abstract class IgniteTxAdapter extends GridMetadataAwareAdapter implement } /** + * @return {@code True} if tx should skip adding itself to completed version map on finish. + */ + public boolean skipCompletedVersions() { + return skipCompletedVers; + } + + /** + * @param skipCompletedVers {@code True} if tx should skip adding itself to completed version map on finish. + */ + public void skipCompletedVersions(boolean skipCompletedVers) { + this.skipCompletedVers = skipCompletedVers; + } + + /** * @return Shared cache context. */ public GridCacheSharedContext<?, ?> context() { @@ -1988,7 +2005,7 @@ public abstract class IgniteTxAdapter extends GridMetadataAwareAdapter implement public abstract void addActiveCache(GridCacheContext cacheCtx, boolean recovery) throws IgniteCheckedException; /** {@inheritDoc} */ - @Nullable @Override public TxCounters txCounters(boolean createIfAbsent) { + @Override public TxCounters txCounters(boolean createIfAbsent) { if (createIfAbsent && txCounters == null) TX_COUNTERS_UPD.compareAndSet(this, null, new TxCounters()); http://git-wip-us.apache.org/repos/asf/ignite/blob/03dd9eb7/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 319073b..a9e6e47 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 @@ -1179,9 +1179,6 @@ public class IgniteTxHandler { nearTx = !F.isEmpty(req.nearWrites()) ? startNearRemoteTx(ctx.deploy().globalLoader(), nodeId, req) : null; dhtTx = startRemoteTx(nodeId, req, res); - if (dhtTx != null && req.updateCounters() != null) // Remember update counters on prepare state. - dhtTx.txCounters(true).updateCounters(req.updateCounters()); - // Set evicted keys from near transaction. if (nearTx != null) res.nearEvicted(nearTx.evicted()); @@ -1335,23 +1332,28 @@ public class IgniteTxHandler { return; } - final GridDhtTxRemote dhtTx = ctx.tm().tx(req.version()); + // Always add version to rollback history to prevent races with rollbacks. + if (!req.commit()) + ctx.tm().addRolledbackTx(null, req.version()); + + GridDhtTxRemote dhtTx = ctx.tm().tx(req.version()); GridNearTxRemote nearTx = ctx.tm().nearTx(req.version()); - final GridCacheVersion nearTxId = - (dhtTx != null ? dhtTx.nearXidVersion() : (nearTx != null ? nearTx.nearXidVersion() : null)); + IgniteInternalTx anyTx = U.<IgniteInternalTx>firstNotNull(dhtTx, nearTx); - if (txFinishMsgLog.isDebugEnabled()) { - txFinishMsgLog.debug("Received dht finish request [txId=" + nearTxId + - ", dhtTxId=" + req.version() + + final GridCacheVersion nearTxId = anyTx != null ? anyTx.nearXidVersion() : null; + + if (txFinishMsgLog.isDebugEnabled()) + txFinishMsgLog.debug("Received dht finish request [txId=" + nearTxId + ", dhtTxId=" + req.version() + ", node=" + nodeId + ']'); - } - // Safety - local transaction will finish explicitly. - if (nearTx != null && nearTx.local()) - nearTx = null; + if (anyTx == null && req.commit()) + ctx.tm().addCommittedTx(null, req.version(), null); - finish(nodeId, dhtTx, req); + if (dhtTx != null) + finish(nodeId, dhtTx, req); + else + applyPartitionsUpdatesCounters(req.updateCounters()); if (nearTx != null) finish(nodeId, nearTx, req); @@ -1404,31 +1406,7 @@ public class IgniteTxHandler { IgniteTxRemoteEx tx, GridDhtTxFinishRequest req ) { - // We don't allow explicit locks for transactions and - // therefore immediately return if transaction is null. - // However, we may decide to relax this restriction in - // future. - if (tx == null) { - if (req.commit()) - // Must be some long time duplicate, but we add it anyway. - ctx.tm().addCommittedTx(tx, req.version(), null); - else - ctx.tm().addRolledbackTx(tx, req.version()); - - if (log.isDebugEnabled()) - log.debug("Received finish request for non-existing transaction (added to completed set) " + - "[senderNodeId=" + nodeId + ", res=" + req + ']'); - - return; - } - else { - if (req.updateCounters() != null) - tx.txCounters(true).updateCounters(req.updateCounters()); - - if (log.isDebugEnabled()) - log.debug("Received finish request for transaction [senderNodeId=" + nodeId + ", req=" + req + - ", tx=" + tx + ']'); - } + assert tx != null; req.txState(tx.txState()); @@ -1448,6 +1426,9 @@ public class IgniteTxHandler { tx.commitRemoteTx(); } else { + if (tx.dht() && req.updateCounters() != null) + tx.txCounters(true).updateCounters(req.updateCounters()); + tx.doneRemote(req.baseVersion(), null, null, null); tx.mvccSnapshot(req.mvccSnapshot()); tx.rollbackRemoteTx(); @@ -1715,6 +1696,8 @@ public class IgniteTxHandler { if (log.isDebugEnabled()) log.debug("Attempt to start a completed transaction (will ignore): " + tx); + applyPartitionsUpdatesCounters(req.updateCounters()); + return null; } @@ -1725,6 +1708,8 @@ public class IgniteTxHandler { ctx.tm().uncommitTx(tx); + applyPartitionsUpdatesCounters(req.updateCounters()); + return null; } } @@ -1733,6 +1718,9 @@ public class IgniteTxHandler { tx.transactionNodes(req.transactionNodes()); } + if (req.updateCounters() != null) + tx.txCounters(true).updateCounters(req.updateCounters()); + if (!tx.isSystemInvalidate()) { int idx = 0; http://git-wip-us.apache.org/repos/asf/ignite/blob/03dd9eb7/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 c829d45..0d78017 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 @@ -162,9 +162,6 @@ public abstract class IgniteTxLocalAdapter extends IgniteTxAdapter implements Ig /** */ private volatile boolean qryEnlisted; - /** Whether to skip update of completed versions map during rollback caused by empty update set in MVCC TX. */ - private boolean forceSkipCompletedVers; - /** * Empty constructor required for {@link Externalizable}. */ @@ -1112,7 +1109,7 @@ public abstract class IgniteTxLocalAdapter extends IgniteTxAdapter implements Ig } if (DONE_FLAG_UPD.compareAndSet(this, 0, 1)) { - cctx.tm().rollbackTx(this, clearThreadMap, forceSkipCompletedVers); + cctx.tm().rollbackTx(this, clearThreadMap, skipCompletedVersions()); cctx.mvccCaching().onTxFinished(this, false); @@ -1133,14 +1130,6 @@ public abstract class IgniteTxLocalAdapter extends IgniteTxAdapter implements Ig } /** - * Forces transaction to skip update of completed versions map during rollback caused by empty update set - * in MVCC TX. - */ - public void forceSkipCompletedVersions() { - forceSkipCompletedVers = true; - } - - /** * @param ctx Cache context. * @param key Key. * @param expiryPlc Expiry policy. http://git-wip-us.apache.org/repos/asf/ignite/blob/03dd9eb7/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/transactions/TxRollbackAsyncTest.java ---------------------------------------------------------------------- diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/transactions/TxRollbackAsyncTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/transactions/TxRollbackAsyncTest.java index c480c2d..bdc5a37 100644 --- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/transactions/TxRollbackAsyncTest.java +++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/transactions/TxRollbackAsyncTest.java @@ -650,9 +650,6 @@ public class TxRollbackAsyncTest extends GridCommonAbstractTest { */ @Test public void testMixedAsyncRollbackTypes() throws Exception { - if (MvccFeatureChecker.forcedMvcc()) - fail("https://issues.apache.org/jira/browse/IGNITE-10434";); - final Ignite client = startClient(); final AtomicBoolean stop = new AtomicBoolean();
