You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by av...@apache.org on 2018/07/26 19:56:30 UTC
[42/50] [abbrv] ignite git commit: ignite-6467 Fixed
GridDhtInvalidPartitionException in lockAllAsync for client requests (tried
to create cache entries before topology check)
ignite-6467 Fixed GridDhtInvalidPartitionException in lockAllAsync for client requests (tried to create cache entries before topology check)
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/7d4c46b9
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/7d4c46b9
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/7d4c46b9
Branch: refs/heads/ignite-8446
Commit: 7d4c46b9bfd502712143d6e05d40546a5c3420fb
Parents: f13d2ed
Author: sboikov <sb...@gridgain.com>
Authored: Wed Jul 25 22:42:23 2018 +0300
Committer: sboikov <sb...@gridgain.com>
Committed: Wed Jul 25 22:42:23 2018 +0300
----------------------------------------------------------------------
.../dht/GridDhtTransactionalCacheAdapter.java | 176 ++++++++-----------
...niteCacheClientNodeChangingTopologyTest.java | 95 ++++++++++
2 files changed, 165 insertions(+), 106 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/7d4c46b9/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTransactionalCacheAdapter.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTransactionalCacheAdapter.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTransactionalCacheAdapter.java
index fe1aee9..161c542 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTransactionalCacheAdapter.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTransactionalCacheAdapter.java
@@ -917,39 +917,85 @@ public abstract class GridDhtTransactionalCacheAdapter<K, V> extends GridDhtCach
GridDhtLockFuture fut = null;
- if (!req.inTx()) {
- GridDhtPartitionTopology top = null;
+ GridDhtPartitionTopology top = null;
- if (req.firstClientRequest()) {
- assert CU.clientNode(nearNode);
+ if (req.firstClientRequest()) {
+ assert CU.clientNode(nearNode);
- top = topology();
+ top = topology();
- top.readLock();
+ top.readLock();
- if (!top.topologyVersionFuture().isDone()) {
- top.readUnlock();
+ if (!top.topologyVersionFuture().isDone()) {
+ top.readUnlock();
- return null;
- }
+ return null;
}
+ }
- try {
- if (top != null && needRemap(req.topologyVersion(), top.readyTopologyVersion())) {
- if (log.isDebugEnabled()) {
- log.debug("Client topology version mismatch, need remap lock request [" +
+ try {
+ if (top != null && needRemap(req.topologyVersion(), top.readyTopologyVersion())) {
+ if (log.isDebugEnabled()) {
+ log.debug("Client topology version mismatch, need remap lock request [" +
"reqTopVer=" + req.topologyVersion() +
", locTopVer=" + top.readyTopologyVersion() +
", req=" + req + ']');
- }
+ }
- GridNearLockResponse res = sendClientLockRemapResponse(nearNode,
- req,
- top.lastTopologyChangeVersion());
+ GridNearLockResponse res = sendClientLockRemapResponse(nearNode,
+ req,
+ top.lastTopologyChangeVersion());
- return new GridFinishedFuture<>(res);
- }
+ return new GridFinishedFuture<>(res);
+ }
+
+ if (req.inTx()) {
+ if (tx == null) {
+ tx = new GridDhtTxLocal(
+ ctx.shared(),
+ req.topologyVersion(),
+ nearNode.id(),
+ req.version(),
+ req.futureId(),
+ req.miniId(),
+ req.threadId(),
+ /*implicitTx*/false,
+ /*implicitSingleTx*/false,
+ ctx.systemTx(),
+ false,
+ ctx.ioPolicy(),
+ PESSIMISTIC,
+ req.isolation(),
+ req.timeout(),
+ req.isInvalidate(),
+ !req.skipStore(),
+ false,
+ req.txSize(),
+ null,
+ req.subjectId(),
+ req.taskNameHash());
+
+ if (req.syncCommit())
+ tx.syncMode(FULL_SYNC);
+
+ tx = ctx.tm().onCreated(null, tx);
+ if (tx == null || !tx.init()) {
+ String msg = "Failed to acquire lock (transaction has been completed): " +
+ req.version();
+
+ U.warn(log, msg);
+
+ if (tx != null)
+ tx.rollbackDhtLocal();
+
+ return new GridDhtFinishedFuture<>(new IgniteTxRollbackCheckedException(msg));
+ }
+
+ tx.topologyVersion(req.topologyVersion());
+ }
+ }
+ else {
fut = new GridDhtLockFuture(ctx,
nearNode.id(),
req.version(),
@@ -970,10 +1016,10 @@ public abstract class GridDhtTransactionalCacheAdapter<K, V> extends GridDhtCach
if (!ctx.mvcc().addFuture(fut))
throw new IllegalStateException("Duplicate future ID: " + fut);
}
- finally {
- if (top != null)
- top.readUnlock();
- }
+ }
+ finally {
+ if (top != null)
+ top.readUnlock();
}
boolean timedOut = false;
@@ -1022,88 +1068,6 @@ public abstract class GridDhtTransactionalCacheAdapter<K, V> extends GridDhtCach
// Handle implicit locks for pessimistic transactions.
if (req.inTx()) {
- if (tx == null) {
- GridDhtPartitionTopology top = null;
-
- if (req.firstClientRequest()) {
- assert CU.clientNode(nearNode);
-
- top = topology();
-
- top.readLock();
-
- if (!top.topologyVersionFuture().isDone()) {
- top.readUnlock();
-
- return null;
- }
- }
-
- try {
- if (top != null && needRemap(req.topologyVersion(), top.readyTopologyVersion())) {
- if (log.isDebugEnabled()) {
- log.debug("Client topology version mismatch, need remap lock request [" +
- "reqTopVer=" + req.topologyVersion() +
- ", locTopVer=" + top.readyTopologyVersion() +
- ", req=" + req + ']');
- }
-
- GridNearLockResponse res = sendClientLockRemapResponse(nearNode,
- req,
- top.lastTopologyChangeVersion());
-
- return new GridFinishedFuture<>(res);
- }
-
- tx = new GridDhtTxLocal(
- ctx.shared(),
- req.topologyVersion(),
- nearNode.id(),
- req.version(),
- req.futureId(),
- req.miniId(),
- req.threadId(),
- /*implicitTx*/false,
- /*implicitSingleTx*/false,
- ctx.systemTx(),
- false,
- ctx.ioPolicy(),
- PESSIMISTIC,
- req.isolation(),
- req.timeout(),
- req.isInvalidate(),
- !req.skipStore(),
- false,
- req.txSize(),
- null,
- req.subjectId(),
- req.taskNameHash());
-
- if (req.syncCommit())
- tx.syncMode(FULL_SYNC);
-
- tx = ctx.tm().onCreated(null, tx);
-
- if (tx == null || !tx.init()) {
- String msg = "Failed to acquire lock (transaction has been completed): " +
- req.version();
-
- U.warn(log, msg);
-
- if (tx != null)
- tx.rollbackDhtLocal();
-
- return new GridDhtFinishedFuture<>(new IgniteTxRollbackCheckedException(msg));
- }
-
- tx.topologyVersion(req.topologyVersion());
- }
- finally {
- if (top != null)
- top.readUnlock();
- }
- }
-
ctx.tm().txContext(tx);
if (log.isDebugEnabled())
http://git-wip-us.apache.org/repos/asf/ignite/blob/7d4c46b9/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/IgniteCacheClientNodeChangingTopologyTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/IgniteCacheClientNodeChangingTopologyTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/IgniteCacheClientNodeChangingTopologyTest.java
index c82b4b9..10c5f37 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/IgniteCacheClientNodeChangingTopologyTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/IgniteCacheClientNodeChangingTopologyTest.java
@@ -45,6 +45,7 @@ import org.apache.ignite.IgniteTransactions;
import org.apache.ignite.cache.CacheAtomicityMode;
import org.apache.ignite.cache.affinity.Affinity;
import org.apache.ignite.cache.affinity.AffinityFunction;
+import org.apache.ignite.cache.affinity.rendezvous.RendezvousAffinityFunction;
import org.apache.ignite.cluster.ClusterNode;
import org.apache.ignite.configuration.CacheConfiguration;
import org.apache.ignite.configuration.IgniteConfiguration;
@@ -504,6 +505,7 @@ public class IgniteCacheClientNodeChangingTopologyTest extends GridCommonAbstrac
checkData(map, null, cache, 4);
}
+
/**
* @throws Exception If failed.
*/
@@ -928,6 +930,87 @@ public class IgniteCacheClientNodeChangingTopologyTest extends GridCommonAbstrac
}
/**
+ * @return Cache configuration.
+ */
+ private CacheConfiguration testPessimisticTx3Cfg() {
+ CacheConfiguration ccfg = new CacheConfiguration(DEFAULT_CACHE_NAME);
+
+ ccfg.setCacheMode(PARTITIONED);
+ ccfg.setBackups(0);
+ ccfg.setAtomicityMode(TRANSACTIONAL);
+ ccfg.setWriteSynchronizationMode(FULL_SYNC);
+ ccfg.setRebalanceMode(SYNC);
+ ccfg.setAffinity(new RendezvousAffinityFunction(false, 16));
+
+ return ccfg;
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testPessimisticTx3() throws Exception {
+ for (int iter = 0; iter < 5; iter++) {
+ info("Iteration: " + iter);
+
+ ccfg = testPessimisticTx3Cfg();
+
+ IgniteEx ignite0 = startGrid(0);
+
+ Map<Integer, Integer> map = new HashMap<>();
+
+ final IgniteCache<Integer, Integer> cache0 = ignite0.cache(DEFAULT_CACHE_NAME);
+
+ for (int i = 0; i < 10000; i++) {
+ cache0.put(i, i);
+ map.put(i, i + 1);
+ }
+
+ client = true;
+
+ ccfg = testPessimisticTx3Cfg();
+
+ final Ignite ignite3 = startGrid(3);
+
+ final IgniteCache<Integer, Integer> cache = ignite3.cache(DEFAULT_CACHE_NAME);
+
+ TestCommunicationSpi spi = (TestCommunicationSpi)ignite3.configuration().getCommunicationSpi();
+ spi.blockMessages(GridNearLockRequest.class, ignite0.localNode().id());
+
+ IgniteInternalFuture putFut = GridTestUtils.runAsync(new Callable<Object>() {
+ @Override public Object call() throws Exception {
+ Thread.currentThread().setName("put-thread");
+
+ try (Transaction tx = ignite3.transactions().txStart(PESSIMISTIC, REPEATABLE_READ)) {
+ cache.putAll(map);
+
+ tx.commit();
+ }
+
+ return null;
+ }
+ });
+
+ spi.waitForBlocked();
+
+ client = false;
+
+ ccfg = testPessimisticTx3Cfg();
+
+ startGrid(1);
+
+ // Want provoke case when client req is processed when target partition is RENTING,
+ // there is no easy way to do it, so just try sleep.
+ U.sleep(ThreadLocalRandom.current().nextInt(1000) + 100);
+
+ spi.stopBlock();
+
+ putFut.get();
+
+ stopAllGrids();
+ }
+ }
+
+ /**
* @throws Exception If failed.
*/
public void testOptimisticSerializableTx() throws Exception {
@@ -2004,6 +2087,8 @@ public class IgniteCacheClientNodeChangingTopologyTest extends GridCommonAbstrac
blockedMsgs.add(new T2<>(node, (GridIoMessage)msg));
+ notifyAll();
+
return;
}
}
@@ -2072,6 +2157,16 @@ public class IgniteCacheClientNodeChangingTopologyTest extends GridCommonAbstrac
blockedMsgs.clear();
}
}
+
+ /**
+ * @throws InterruptedException If interrupted.
+ */
+ public void waitForBlocked() throws InterruptedException {
+ synchronized (this) {
+ while (blockedMsgs.isEmpty())
+ wait();
+ }
+ }
}
/**