ignite-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From sboi...@apache.org
Subject [20/50] [abbrv] incubator-ignite git commit: # ignite-157 wait for 'preparing' transactions in 'processCheckPreparedTxRequest'
Date Tue, 28 Apr 2015 15:01:05 GMT
# ignite-157 wait for 'preparing' transactions in 'processCheckPreparedTxRequest'


Project: http://git-wip-us.apache.org/repos/asf/incubator-ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-ignite/commit/7b5f91a2
Tree: http://git-wip-us.apache.org/repos/asf/incubator-ignite/tree/7b5f91a2
Diff: http://git-wip-us.apache.org/repos/asf/incubator-ignite/diff/7b5f91a2

Branch: refs/heads/ignite-286
Commit: 7b5f91a2d7723a7c4f4c87bbdc8b2b5772253651
Parents: 62d8053
Author: sboikov <sboikov@gridgain.com>
Authored: Fri Apr 24 15:08:04 2015 +0300
Committer: sboikov <sboikov@gridgain.com>
Committed: Fri Apr 24 15:43:52 2015 +0300

----------------------------------------------------------------------
 .../cache/transactions/IgniteTxManager.java     | 22 +++--
 ...xOriginatingNodeFailureAbstractSelfTest.java | 91 +++++++++++++++++++-
 ...itionedTxOriginatingNodeFailureSelfTest.java | 86 ++++++++++++++++--
 3 files changed, 184 insertions(+), 15 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/7b5f91a2/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 d139afd..e4cf28b 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
@@ -2054,7 +2054,7 @@ public class IgniteTxManager extends GridCacheSharedManagerAdapter {
                     log.debug("Processing node failed event [locNodeId=" + cctx.localNodeId()
+
                         ", failedNodeId=" + evtNodeId + ']');
 
-                for (IgniteInternalTx tx : txs()) {
+                for (final IgniteInternalTx tx : txs()) {
                     if ((tx.near() && !tx.local()) || (tx.storeUsed() &&
tx.masterNodeIds().contains(evtNodeId))) {
                         // Invalidate transactions.
                         salvageTx(tx, false, RECOVERY_FINISH);
@@ -2062,12 +2062,24 @@ public class IgniteTxManager extends GridCacheSharedManagerAdapter
{
                     else {
                         // Check prepare only if originating node ID failed. Otherwise parent
node will finish this tx.
                         if (tx.originatingNodeId().equals(evtNodeId)) {
-                            if (tx.state() == PREPARED)
+                            if (tx.optimistic() && tx.state() == PREPARED)
                                 commitIfPrepared(tx);
                             else {
-                                if (tx.setRollbackOnly())
-                                    tx.rollbackAsync();
-                                // If we could not mark tx as rollback, it means that transaction
is being committed.
+                                IgniteInternalFuture<IgniteInternalTx> prepFut = tx.currentPrepareFuture();
+
+                                if (prepFut != null) {
+                                    prepFut.listen(new CI1<IgniteInternalFuture<IgniteInternalTx>>()
{
+                                        @Override public void apply(IgniteInternalFuture<IgniteInternalTx>
fut) {
+                                            if (tx.setRollbackOnly())
+                                                tx.rollbackAsync();
+                                        }
+                                    });
+                                }
+                                else {
+                                    // If we could not mark tx as rollback, it means that
transaction is being committed.
+                                    if (tx.setRollbackOnly())
+                                        tx.rollbackAsync();
+                                }
                             }
                         }
                     }

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/7b5f91a2/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/IgniteTxOriginatingNodeFailureAbstractSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/IgniteTxOriginatingNodeFailureAbstractSelfTest.java
b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/IgniteTxOriginatingNodeFailureAbstractSelfTest.java
index ea8c60b..865f1f6 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/IgniteTxOriginatingNodeFailureAbstractSelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/IgniteTxOriginatingNodeFailureAbstractSelfTest.java
@@ -37,7 +37,9 @@ import org.apache.ignite.testframework.*;
 import java.util.*;
 import java.util.concurrent.*;
 
+import static org.apache.ignite.cache.CachePeekMode.*;
 import static org.apache.ignite.transactions.TransactionConcurrency.*;
+import static org.apache.ignite.transactions.TransactionIsolation.*;
 
 /**
  * Abstract test for originating node failure.
@@ -67,6 +69,18 @@ public abstract class IgniteTxOriginatingNodeFailureAbstractSelfTest extends
Gri
     /**
      * @throws Exception If failed.
      */
+    public void testPessimisticManyKeysCommit() throws Exception {
+        Collection<Integer> keys = new ArrayList<>(200);
+
+        for (int i = 0; i < 200; i++)
+            keys.add(i);
+
+        testPessimisticTxOriginatingNodeFails(keys);
+    }
+
+    /**
+     * @throws Exception If failed.
+     */
     public void testManyKeysRollback() throws Exception {
         Collection<Integer> keys = new ArrayList<>(200);
 
@@ -103,6 +117,75 @@ public abstract class IgniteTxOriginatingNodeFailureAbstractSelfTest
extends Gri
 
     /**
      * @param keys Keys to update.
+     * @throws Exception If failed.
+     */
+    protected void testPessimisticTxOriginatingNodeFails(Collection<Integer> keys)
throws Exception {
+        final Map<Integer, String> map = new HashMap<>();
+
+        final String initVal = "initialValue";
+
+        for (Integer key : keys) {
+            grid(originatingNode()).cache(null).put(key, initVal);
+
+            map.put(key, String.valueOf(key));
+        }
+
+        ClusterNode txNode = grid(originatingNode()).localNode();
+
+        final Ignite txIgniteNode = G.ignite(txNode.id());
+
+        info("Starting pessimistic tx " +
+            "[values=" + map + ", topVer=" + (grid(1)).context().discovery().topologyVersion()
+ ']');
+
+        GridTestUtils.runAsync(new Callable<Object>() {
+            @Override public Object call() throws Exception {
+                IgniteCache<Integer, String> cache = txIgniteNode.cache(null);
+
+                assertNotNull(cache);
+
+                TransactionProxyImpl tx =
+                    (TransactionProxyImpl)txIgniteNode.transactions().txStart(PESSIMISTIC,
REPEATABLE_READ);
+
+                IgniteInternalTx txEx = GridTestUtils.getFieldValue(tx, "tx");
+
+                assertTrue(txEx.pessimistic());
+
+                cache.putAll(map);
+
+                return null;
+            }
+        }).get();
+
+        info("Stopping originating node " + txNode);
+
+        G.stop(G.ignite(txNode.id()).name(), true);
+
+        info("Stopped grid, waiting for transactions to complete.");
+
+        boolean txFinished = GridTestUtils.waitForCondition(new GridAbsPredicate() {
+            @Override public boolean apply() {
+                for (Ignite ignite : G.allGrids()) {
+                    IgniteKernal g = (IgniteKernal)ignite;
+
+                    GridCacheSharedContext<Object, Object> ctx = g.context().cache().context();
+
+                    int txNum = ctx.tm().idMapSize();
+
+                    if (txNum != 0)
+                        return false;
+                }
+
+                return true;
+            }
+        }, 10_000);
+
+        assertTrue(txFinished);
+
+        info("Transactions finished.");
+    }
+
+    /**
+     * @param keys Keys to update.
      * @param partial Flag indicating whether to simulate partial prepared state.
      * @throws Exception If failed.
      */
@@ -140,8 +223,8 @@ public abstract class IgniteTxOriginatingNodeFailureAbstractSelfTest extends
Gri
             nodeMap.put(key, nodes);
         }
 
-        info("Starting tx [values=" + map + ", topVer=" +
-            (grid(1)).context().discovery().topologyVersion() + ']');
+        info("Starting optimistic tx " +
+            "[values=" + map + ", topVer=" + (grid(1)).context().discovery().topologyVersion()
+ ']');
 
         if (partial)
             ignoreMessages(grid(1).localNode().id(), ignoreMessageClass());
@@ -158,6 +241,8 @@ public abstract class IgniteTxOriginatingNodeFailureAbstractSelfTest extends
Gri
 
                 IgniteInternalTx txEx = GridTestUtils.getFieldValue(tx, "tx");
 
+                assertTrue(txEx.optimistic());
+
                 cache.putAll(map);
 
                 try {
@@ -214,7 +299,7 @@ public abstract class IgniteTxOriginatingNodeFailureAbstractSelfTest extends
Gri
 
                         assertNotNull(cache);
 
-                        assertEquals(partial ? initVal : val, cache.localPeek(key, CachePeekMode.ONHEAP));
+                        assertEquals(partial ? initVal : val, cache.localPeek(key, ONHEAP));
 
                         return null;
                     }

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/7b5f91a2/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridCachePartitionedTxOriginatingNodeFailureSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridCachePartitionedTxOriginatingNodeFailureSelfTest.java
b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridCachePartitionedTxOriginatingNodeFailureSelfTest.java
index d4e84a5..fc1c403 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridCachePartitionedTxOriginatingNodeFailureSelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridCachePartitionedTxOriginatingNodeFailureSelfTest.java
@@ -20,13 +20,13 @@ package org.apache.ignite.internal.processors.cache.distributed.dht;
 import org.apache.ignite.cache.*;
 import org.apache.ignite.cluster.*;
 import org.apache.ignite.configuration.*;
-import org.apache.ignite.internal.*;
-import org.apache.ignite.internal.processors.cache.*;
 import org.apache.ignite.internal.processors.cache.distributed.*;
 import org.apache.ignite.internal.processors.cache.distributed.near.*;
 
 import java.util.*;
 
+import static org.apache.ignite.cache.CacheMode.*;
+
 /**
  * Tests transaction consistency when originating node fails.
  */
@@ -37,7 +37,7 @@ public class GridCachePartitionedTxOriginatingNodeFailureSelfTest extends
 
     /** {@inheritDoc} */
     @Override protected CacheMode cacheMode() {
-        return CacheMode.PARTITIONED;
+        return PARTITIONED;
     }
 
     /** {@inheritDoc} */
@@ -58,6 +58,21 @@ public class GridCachePartitionedTxOriginatingNodeFailureSelfTest extends
      * @throws Exception If failed.
      */
     public void testTxFromPrimary() throws Exception {
+        txFromPrimary(true);
+    }
+
+    /**
+     * @throws Exception If failed.
+     */
+    public void testPessimisticTxFromPrimary() throws Exception {
+        txFromPrimary(false);
+    }
+
+    /**
+     * @param optimistic If {@code true} tests optimistic transaction.
+     * @throws Exception If failed.
+     */
+    private void txFromPrimary(boolean optimistic) throws Exception {
         ClusterNode txNode = grid(originatingNode()).localNode();
 
         Integer key = null;
@@ -72,13 +87,31 @@ public class GridCachePartitionedTxOriginatingNodeFailureSelfTest extends
 
         assertNotNull(key);
 
-        testTxOriginatingNodeFails(Collections.singleton(key), false);
+        if (optimistic)
+            testTxOriginatingNodeFails(Collections.singleton(key), false);
+        else
+            testPessimisticTxOriginatingNodeFails(Collections.singleton(key));
     }
 
     /**
      * @throws Exception If failed.
      */
     public void testTxFromBackup() throws Exception {
+        txFromBackup(true);
+    }
+
+    /**
+     * @throws Exception If failed.
+     */
+    public void testPessimisticTxFromBackup() throws Exception {
+        txFromBackup(false);
+    }
+
+    /**
+     * @param optimistic If {@code true} tests optimistic transaction.
+     * @throws Exception If failed.
+     */
+    private void txFromBackup(boolean optimistic) throws Exception {
         ClusterNode txNode = grid(originatingNode()).localNode();
 
         Integer key = null;
@@ -93,13 +126,31 @@ public class GridCachePartitionedTxOriginatingNodeFailureSelfTest extends
 
         assertNotNull(key);
 
-        testTxOriginatingNodeFails(Collections.singleton(key), false);
+        if (optimistic)
+            testTxOriginatingNodeFails(Collections.singleton(key), false);
+        else
+            testPessimisticTxOriginatingNodeFails(Collections.singleton(key));
     }
 
     /**
      * @throws Exception If failed.
      */
     public void testTxFromNotColocated() throws Exception {
+        txFromNotColocated(true);
+    }
+
+    /**
+     * @throws Exception If failed.
+     */
+    public void testPessimisticTxFromNotColocated() throws Exception {
+        txFromNotColocated(false);
+    }
+
+    /**
+     * @param optimistic If {@code true} tests optimistic transaction.
+     * @throws Exception If failed.
+     */
+    private void txFromNotColocated(boolean optimistic) throws Exception {
         ClusterNode txNode = grid(originatingNode()).localNode();
 
         Integer key = null;
@@ -115,13 +166,31 @@ public class GridCachePartitionedTxOriginatingNodeFailureSelfTest extends
 
         assertNotNull(key);
 
-        testTxOriginatingNodeFails(Collections.singleton(key), false);
+        if (optimistic)
+            testTxOriginatingNodeFails(Collections.singleton(key), false);
+        else
+            testPessimisticTxOriginatingNodeFails(Collections.singleton(key));
     }
 
     /**
      * @throws Exception If failed.
      */
     public void testTxAllNodes() throws Exception {
+        txAllNodes(true);
+    }
+
+    /**
+     * @throws Exception If failed.
+     */
+    public void testPessimisticTxAllNodes() throws Exception {
+        txAllNodes(false);
+    }
+
+    /**
+     * @param optimistic If {@code true} tests optimistic transaction.
+     * @throws Exception If failed.
+     */
+    private void txAllNodes(boolean optimistic) throws Exception {
         List<ClusterNode> allNodes = new ArrayList<>(GRID_CNT);
 
         for (int i = 0; i < GRID_CNT; i++)
@@ -145,6 +214,9 @@ public class GridCachePartitionedTxOriginatingNodeFailureSelfTest extends
 
         assertEquals(GRID_CNT, keys.size());
 
-        testTxOriginatingNodeFails(keys, false);
+        if (optimistic)
+            testTxOriginatingNodeFails(keys, false);
+        else
+            testPessimisticTxOriginatingNodeFails(keys);
     }
 }


Mime
View raw message