ignite-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From sboi...@apache.org
Subject [07/15] incubator-ignite git commit: # Wait for next topology version before retry, retries for async tx operations
Date Mon, 17 Aug 2015 11:05:28 GMT
# Wait for next topology version before retry, retries for async tx operations


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

Branch: refs/heads/ignite-gg-10606
Commit: 122a9dbf337d5c1128be32d4efee1e0f1dc683f5
Parents: 47895da
Author: sboikov <sboikov@gridgain.com>
Authored: Fri Aug 14 12:50:06 2015 +0300
Committer: sboikov <sboikov@gridgain.com>
Committed: Fri Aug 14 13:57:19 2015 +0300

----------------------------------------------------------------------
 .../processors/cache/GridCacheAdapter.java      | 143 +++++++++++++++++--
 .../dht/atomic/GridNearAtomicUpdateFuture.java  |  25 +++-
 .../IgniteCachePutRetryAbstractSelfTest.java    |  94 ++++++++++--
 3 files changed, 237 insertions(+), 25 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/122a9dbf/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheAdapter.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheAdapter.java
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheAdapter.java
index 47ede5b..91af352 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheAdapter.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheAdapter.java
@@ -2283,8 +2283,7 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K,
V
      * @param filter Filter.
      * @return Put future.
      */
-    public IgniteInternalFuture<Boolean> putAsync(K key, V val,
-                                                  @Nullable CacheEntryPredicate... filter)
{
+    public IgniteInternalFuture<Boolean> putAsync(K key, V val, @Nullable CacheEntryPredicate...
filter) {
         final boolean statsEnabled = ctx.config().isStatisticsEnabled();
 
         final long start = statsEnabled ? System.nanoTime() : 0L;
@@ -3975,8 +3974,15 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K,
V
                         U.addLastCause(e, e1, log);
                     }
 
-                    if (X.hasCause(e, ClusterTopologyCheckedException.class) && i
!= retries - 1)
+                    if (X.hasCause(e, ClusterTopologyCheckedException.class) && i
!= retries - 1) {
+                        AffinityTopologyVersion topVer = tx.topologyVersion();
+
+                        assert topVer != null && topVer.topologyVersion() > 0
: tx;
+
+                        ctx.affinity().affinityReadyFuture(topVer.topologyVersion() + 1).get();
+
                         continue;
+                    }
 
                     throw e;
                 }
@@ -4014,18 +4020,36 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K,
V
 
         IgniteTxLocalAdapter tx = ctx.tm().threadLocalTx(ctx);
 
-        if (tx == null || tx.implicit())
-            tx = ctx.tm().newTx(
-                true,
-                op.single(),
-                ctx.systemTx() ? ctx : null,
-                OPTIMISTIC,
-                READ_COMMITTED,
-                ctx.kernalContext().config().getTransactionConfiguration().getDefaultTxTimeout(),
-                !ctx.skipStore(),
-                0);
+        if (tx == null || tx.implicit()) {
+            boolean skipStore = ctx.skipStore(); // Save value of thread-local flag.
+
+            CacheOperationContext opCtx = ctx.operationContextPerCall();
+
+            int retries = opCtx != null && opCtx.noRetries() ? 1 : MAX_RETRIES;
 
-        return asyncOp(tx, op);
+            if (retries == 1) {
+                tx = ctx.tm().newTx(
+                    true,
+                    op.single(),
+                    ctx.systemTx() ? ctx : null,
+                    OPTIMISTIC,
+                    READ_COMMITTED,
+                    ctx.kernalContext().config().getTransactionConfiguration().getDefaultTxTimeout(),
+                    !skipStore,
+                    0);
+
+                return asyncOp(tx, op);
+            }
+            else {
+                AsyncOpRetryFuture<T> fut = new AsyncOpRetryFuture<>(op, skipStore,
retries);
+
+                fut.execute();
+
+                return fut;
+            }
+        }
+        else
+            return asyncOp(tx, op);
     }
 
     /**
@@ -4624,6 +4648,97 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K,
V
     /**
      *
      */
+    private class AsyncOpRetryFuture<T> extends GridFutureAdapter<T> {
+        /** */
+        private AsyncOp<T> op;
+
+        /** */
+        private boolean skipStore;
+
+        /** */
+        private int retries;
+
+        /** */
+        private IgniteTxLocalAdapter tx;
+
+        /**
+         * @param op Operation.
+         * @param skipStore Skip store flag.
+         * @param retries Number of retries.
+         */
+        public AsyncOpRetryFuture(AsyncOp<T> op,
+            boolean skipStore,
+            int retries) {
+            assert retries > 1 : retries;
+
+            this.op = op;
+            this.tx = null;
+            this.skipStore = skipStore;
+            this.retries = retries;
+        }
+
+        /**
+         * @param tx Transaction.
+         */
+        public void execute() {
+            tx = ctx.tm().newTx(
+                true,
+                op.single(),
+                ctx.systemTx() ? ctx : null,
+                OPTIMISTIC,
+                READ_COMMITTED,
+                ctx.kernalContext().config().getTransactionConfiguration().getDefaultTxTimeout(),
+                !skipStore,
+                0);
+
+            IgniteInternalFuture<T> fut = asyncOp(tx, op);
+
+            fut.listen(new IgniteInClosure<IgniteInternalFuture<T>>() {
+                @Override public void apply(IgniteInternalFuture<T> fut) {
+                    try {
+                        T res = fut.get();
+
+                        onDone(res);
+                    }
+                    catch (IgniteCheckedException e) {
+                        if (X.hasCause(e, ClusterTopologyCheckedException.class) &&
--retries > 0) {
+                            IgniteTxLocalAdapter tx = AsyncOpRetryFuture.this.tx;
+
+                            assert tx != null;
+
+                            AffinityTopologyVersion topVer = tx.topologyVersion();
+
+                            assert topVer != null && topVer.topologyVersion() >
0 : tx;
+
+                            IgniteInternalFuture<?> topFut =
+                                ctx.affinity().affinityReadyFuture(topVer.topologyVersion()
+ 1);
+
+                            topFut.listen(new IgniteInClosure<IgniteInternalFuture<?>>()
{
+                                @Override public void apply(IgniteInternalFuture<?>
topFut) {
+                                    try {
+                                        topFut.get();
+
+                                        execute();
+                                    }
+                                    catch (IgniteCheckedException e) {
+                                        onDone(e);
+                                    }
+                                }
+                            });
+
+                            return;
+                        }
+
+                        onDone(e);
+                    }
+                }
+            });
+        }
+    }
+
+    /**
+     *
+     */
     private static class PeekModes {
         /** */
         private boolean near;

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/122a9dbf/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridNearAtomicUpdateFuture.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridNearAtomicUpdateFuture.java
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridNearAtomicUpdateFuture.java
index 751c9ba..0498839 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridNearAtomicUpdateFuture.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridNearAtomicUpdateFuture.java
@@ -409,9 +409,15 @@ public class GridNearAtomicUpdateFuture extends GridFutureAdapter<Object>
 
         GridFutureAdapter<Void> fut0;
 
+        long nextTopVer;
+
         synchronized (this) {
             mappings = new ConcurrentHashMap8<>(keys.size(), 1.0f);
 
+            assert topVer != null && topVer.topologyVersion() > 0 : this;
+
+            nextTopVer = topVer.topologyVersion() + 1;
+
             topVer = AffinityTopologyVersion.ZERO;
 
             fut0 = topCompleteFut;
@@ -428,7 +434,24 @@ public class GridNearAtomicUpdateFuture extends GridFutureAdapter<Object>
         updVer = null;
         topLocked = false;
 
-        map();
+        IgniteInternalFuture<?> fut = cctx.affinity().affinityReadyFuture(nextTopVer);
+
+        fut.listen(new CI1<IgniteInternalFuture<?>>() {
+            @Override public void apply(final IgniteInternalFuture<?> fut) {
+                cctx.kernalContext().closure().runLocalSafe(new Runnable() {
+                    @Override public void run() {
+                        try {
+                            fut.get();
+
+                            map();
+                        }
+                        catch (IgniteCheckedException e) {
+                            onDone(e);
+                        }
+                    }
+                });
+            }
+        });
     }
 
     /** {@inheritDoc} */

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/122a9dbf/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/dht/IgniteCachePutRetryAbstractSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/dht/IgniteCachePutRetryAbstractSelfTest.java
b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/dht/IgniteCachePutRetryAbstractSelfTest.java
index dcba325..9abc5c8 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/dht/IgniteCachePutRetryAbstractSelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/dht/IgniteCachePutRetryAbstractSelfTest.java
@@ -27,6 +27,7 @@ import org.apache.ignite.internal.util.typedef.*;
 import org.apache.ignite.internal.util.typedef.internal.*;
 import org.apache.ignite.testframework.*;
 
+import java.util.*;
 import java.util.concurrent.*;
 import java.util.concurrent.atomic.*;
 
@@ -76,11 +77,25 @@ public abstract class IgniteCachePutRetryAbstractSelfTest extends GridCacheAbstr
     protected CacheAtomicWriteOrderMode writeOrderMode() {
         return CLOCK;
     }
-
     /**
      * @throws Exception If failed.
      */
     public void testPut() throws Exception {
+        checkPut(false);
+    }
+
+    /**
+     * @throws Exception If failed.
+     */
+    public void testPutAsync() throws Exception {
+        checkPut(true);
+    }
+
+    /**
+     * @param async If {@code true} tests asynchronous put.
+     * @throws Exception If failed.
+     */
+    private void checkPut(boolean async) throws Exception {
         final AtomicBoolean finished = new AtomicBoolean();
 
         IgniteInternalFuture<Object> fut = GridTestUtils.runAsync(new Callable<Object>()
{
@@ -104,20 +119,67 @@ public abstract class IgniteCachePutRetryAbstractSelfTest extends GridCacheAbstr
         if (atomicityMode() == ATOMIC)
             assertEquals(writeOrderMode(), cache.getConfiguration(CacheConfiguration.class).getAtomicWriteOrderMode());
 
-        for (int i = 0; i < keysCnt; i++)
-            cache.put(i, i);
+        int iter = 0;
+
+        long stopTime = System.currentTimeMillis() + 60_000;
+
+        if (async) {
+            IgniteCache<Object, Object> cache0 = cache.withAsync();
+
+            while (System.currentTimeMillis() < stopTime) {
+                Integer val = ++iter;
+
+                for (int i = 0; i < keysCnt; i++) {
+                    cache0.put(i, val);
+
+                    cache0.future().get();
+                }
+
+                for (int i = 0; i < keysCnt; i++) {
+                    cache0.get(i);
+
+                    assertEquals(val, cache0.future().get());
+                }
+            }
+        }
+        else {
+            while (System.currentTimeMillis() < stopTime) {
+                Integer val = ++iter;
+
+                for (int i = 0; i < keysCnt; i++)
+                    cache.put(i, val);
+
+                for (int i = 0; i < keysCnt; i++)
+                    assertEquals(val, cache.get(i));
+            }
+        }
 
         finished.set(true);
         fut.get();
 
         for (int i = 0; i < keysCnt; i++)
-            assertEquals(i, ignite(0).cache(null).get(i));
+            assertEquals(iter, cache.get(i));
+    }
+
+    /**
+     * @throws Exception If failed.
+     */
+    public void testFailsWithNoRetries() throws Exception {
+        checkFailsWithNoRetries(false);
+    }
+
+    /**
+     * @throws Exception If failed.
+     */
+    public void testFailsWithNoRetriesAsync() throws Exception {
+        checkFailsWithNoRetries(true);
     }
 
     /**
+     * @param async If {@code true} tests asynchronous put.
      * @throws Exception If failed.
      */
-    public void testFailWithNoRetries() throws Exception {
+    private void checkFailsWithNoRetries(boolean async) throws Exception {
         final AtomicBoolean finished = new AtomicBoolean();
 
         IgniteInternalFuture<Object> fut = GridTestUtils.runAsync(new Callable<Object>()
{
@@ -136,22 +198,34 @@ public abstract class IgniteCachePutRetryAbstractSelfTest extends GridCacheAbstr
 
         int keysCnt = keysCount();
 
-        boolean exceptionThrown = false;
+        boolean eThrown = false;
+
+        IgniteCache<Object, Object> cache = ignite(0).cache(null).withNoRetries();
+
+        if (async)
+            cache = cache.withAsync();
 
         for (int i = 0; i < keysCnt; i++) {
             try {
-                ignite(0).cache(null).withNoRetries().put(i, i);
+                if (async) {
+                    cache.put(i, i);
+
+                    cache.future().get();
+                }
+                else
+                    cache.put(i, i);
             }
             catch (Exception e) {
-                assertTrue("Invalid exception: " + e, X.hasCause(e, ClusterTopologyCheckedException.class)
|| X.hasCause(e, CachePartialUpdateException.class));
+                assertTrue("Invalid exception: " + e,
+                    X.hasCause(e, ClusterTopologyCheckedException.class, CachePartialUpdateException.class));
 
-                exceptionThrown = true;
+                eThrown = true;
 
                 break;
             }
         }
 
-        assertTrue(exceptionThrown);
+        assertTrue(eThrown);
 
         finished.set(true);
         fut.get();


Mime
View raw message