ignite-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From sboi...@apache.org
Subject [16/50] incubator-ignite git commit: IGNITE-621 - Fixing remap logic.
Date Mon, 06 Jul 2015 08:49:55 GMT
IGNITE-621 - Fixing remap logic.


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

Branch: refs/heads/ignite-752
Commit: c2c90b52972bec53919d97ec07d2aeab4d0d55e8
Parents: a9d0662
Author: Alexey Goncharuk <agoncharuk@gridgain.com>
Authored: Thu Jun 25 17:06:31 2015 -0700
Committer: Alexey Goncharuk <agoncharuk@gridgain.com>
Committed: Thu Jun 25 17:06:31 2015 -0700

----------------------------------------------------------------------
 .../apache/ignite/IgniteSystemProperties.java   |  3 +
 .../processors/cache/GridCacheAdapter.java      |  2 +-
 .../processors/cache/GridCacheAtomicFuture.java | 12 ++-
 .../processors/cache/GridCacheMvccManager.java  |  8 +-
 .../dht/atomic/GridDhtAtomicUpdateFuture.java   | 12 ++-
 .../dht/atomic/GridNearAtomicUpdateFuture.java  | 88 ++++++++++++++++++--
 .../communication/tcp/TcpCommunicationSpi.java  |  2 +-
 ...eAtomicInvalidPartitionHandlingSelfTest.java |  7 +-
 8 files changed, 110 insertions(+), 24 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/c2c90b52/modules/core/src/main/java/org/apache/ignite/IgniteSystemProperties.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/IgniteSystemProperties.java b/modules/core/src/main/java/org/apache/ignite/IgniteSystemProperties.java
index 542fa30..40fc873 100644
--- a/modules/core/src/main/java/org/apache/ignite/IgniteSystemProperties.java
+++ b/modules/core/src/main/java/org/apache/ignite/IgniteSystemProperties.java
@@ -343,6 +343,9 @@ public final class IgniteSystemProperties {
     /** Maximum size for affinity assignment history. */
     public static final String IGNITE_AFFINITY_HISTORY_SIZE = "IGNITE_AFFINITY_HISTORY_SIZE";
 
+    /** Number of cache operation retries in case of topology exceptions. */
+    public static final String IGNITE_CACHE_RETRIES_COUNT = "IGNITE_CACHE_RETRIES_COUNT";
+
     /**
      * Enforces singleton.
      */

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/c2c90b52/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 f993527..e138520 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
@@ -79,7 +79,7 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K,
V
     public static final int CLEAR_ALL_SPLIT_THRESHOLD = 10000;
 
     /** Maximum number of retries when topology changes. */
-    public static final int MAX_RETRIES = 100;
+    public static final int MAX_RETRIES = IgniteSystemProperties.getInteger(IGNITE_CACHE_RETRIES_COUNT,
100);
 
     /** Deserialization stash. */
     private static final ThreadLocal<IgniteBiTuple<String, String>> stash = new
ThreadLocal<IgniteBiTuple<String,

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/c2c90b52/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheAtomicFuture.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheAtomicFuture.java
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheAtomicFuture.java
index 35d3ec5..8724d3a 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheAtomicFuture.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheAtomicFuture.java
@@ -17,6 +17,7 @@
 
 package org.apache.ignite.internal.processors.cache;
 
+import org.apache.ignite.internal.*;
 import org.apache.ignite.internal.processors.affinity.*;
 
 import java.util.*;
@@ -26,14 +27,17 @@ import java.util.*;
  */
 public interface GridCacheAtomicFuture<R> extends GridCacheFuture<R> {
     /**
-     * @return {@code True} if partition exchange should wait for this future to complete.
+     * @return Future topology version.
      */
-    public boolean waitForPartitionExchange();
+    public AffinityTopologyVersion topologyVersion();
 
     /**
-     * @return Future topology version.
+     * Gets future that will be completed when it is safe when update is finished on the
given version of topology.
+     *
+     * @param topVer Topology version to finish.
+     * @return Future or {@code null} if no need to wait.
      */
-    public AffinityTopologyVersion topologyVersion();
+    public IgniteInternalFuture<Void> completeFuture(AffinityTopologyVersion topVer);
 
     /**
      * @return Future keys.

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/c2c90b52/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheMvccManager.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheMvccManager.java
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheMvccManager.java
index c528e08..f24cf01 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheMvccManager.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheMvccManager.java
@@ -338,7 +338,7 @@ public class GridCacheMvccManager extends GridCacheSharedManagerAdapter
{
     public void addAtomicFuture(GridCacheVersion futVer, GridCacheAtomicFuture<?> fut)
{
         IgniteInternalFuture<?> old = atomicFuts.put(futVer, fut);
 
-        assert old == null;
+        assert old == null : "Old future is not null [futVer=" + futVer + ", fut=" + fut
+ ", old=" + old + ']';
     }
 
     /**
@@ -1002,8 +1002,10 @@ public class GridCacheMvccManager extends GridCacheSharedManagerAdapter
{
         res.ignoreChildFailures(ClusterTopologyCheckedException.class, CachePartialUpdateCheckedException.class);
 
         for (GridCacheAtomicFuture<?> fut : atomicFuts.values()) {
-            if (fut.waitForPartitionExchange() && fut.topologyVersion().compareTo(topVer)
< 0)
-                res.add((IgniteInternalFuture<Object>)fut);
+            IgniteInternalFuture<Void> complete = fut.completeFuture(topVer);
+
+            if (complete != null)
+                res.add((IgniteInternalFuture)complete);
         }
 
         res.markInitialized();

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/c2c90b52/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicUpdateFuture.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicUpdateFuture.java
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicUpdateFuture.java
index ff8454e..37b57e6 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicUpdateFuture.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicUpdateFuture.java
@@ -20,6 +20,7 @@ package org.apache.ignite.internal.processors.cache.distributed.dht.atomic;
 import org.apache.ignite.*;
 import org.apache.ignite.cache.*;
 import org.apache.ignite.cluster.*;
+import org.apache.ignite.internal.*;
 import org.apache.ignite.internal.cluster.*;
 import org.apache.ignite.internal.processors.affinity.*;
 import org.apache.ignite.internal.processors.cache.*;
@@ -170,13 +171,16 @@ public class GridDhtAtomicUpdateFuture extends GridFutureAdapter<Void>
     }
 
     /** {@inheritDoc} */
-    @Override public boolean waitForPartitionExchange() {
-        return waitForExchange;
+    @Override public AffinityTopologyVersion topologyVersion() {
+        return updateReq.topologyVersion();
     }
 
     /** {@inheritDoc} */
-    @Override public AffinityTopologyVersion topologyVersion() {
-        return updateReq.topologyVersion();
+    @Override public IgniteInternalFuture<Void> completeFuture(AffinityTopologyVersion
topVer) {
+        if (waitForExchange && topologyVersion().compareTo(topVer) < 0)
+            return this;
+
+        return null;
     }
 
     /** {@inheritDoc} */

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/c2c90b52/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 536eb40..ea9b335 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
@@ -105,7 +105,10 @@ public class GridNearAtomicUpdateFuture extends GridFutureAdapter<Object>
     private final ExpiryPolicy expiryPlc;
 
     /** Future map topology version. */
-    private AffinityTopologyVersion topVer = AffinityTopologyVersion.ZERO;
+    private volatile AffinityTopologyVersion topVer = AffinityTopologyVersion.ZERO;
+
+    /** Completion future for a particular topology version. */
+    private GridFutureAdapter<Void> topCompleteFut;
 
     /** Optional filter. */
     private final CacheEntryPredicate[] filter;
@@ -246,8 +249,10 @@ public class GridNearAtomicUpdateFuture extends GridFutureAdapter<Object>
         return F.view(F.viewReadOnly(mappings.keySet(), U.id2Node(cctx.kernalContext())),
F.notNull());
     }
 
-    /** {@inheritDoc} */
-    @Override public boolean waitForPartitionExchange() {
+    /**
+     * @return {@code True} if this future should block partition map exchange.
+     */
+    private boolean waitForPartitionExchange() {
         // Wait fast-map near atomic update futures in CLOCK mode.
         return fastMap;
     }
@@ -323,13 +328,36 @@ public class GridNearAtomicUpdateFuture extends GridFutureAdapter<Object>
         else {
             topLocked = true;
 
-            // Cannot remap.
-            remapCnt.set(1);
+            synchronized (this) {
+                this.topVer = topVer;
+
+                // Cannot remap.
+                remapCnt.set(1);
+            }
 
             map0(topVer, null, false, null);
         }
     }
 
+    /** {@inheritDoc} */
+    @Override public IgniteInternalFuture<Void> completeFuture(AffinityTopologyVersion
topVer) {
+        if (waitForPartitionExchange() && topologyVersion().compareTo(topVer) <
0) {
+            synchronized (this) {
+                if (this.topVer == AffinityTopologyVersion.ZERO)
+                    return null;
+
+                if (this.topVer.compareTo(topVer) < 0) {
+                    if (topCompleteFut == null)
+                        topCompleteFut = new GridFutureAdapter<>();
+
+                    return topCompleteFut;
+                }
+            }
+        }
+
+        return null;
+    }
+
     /**
      * @param failed Keys to remap.
      */
@@ -339,14 +367,20 @@ public class GridNearAtomicUpdateFuture extends GridFutureAdapter<Object>
 
         Collection<Object> remapKeys = new ArrayList<>(failed.size());
         Collection<Object> remapVals = vals != null ? new ArrayList<>(failed.size())
: null;
+        Collection<GridCacheDrInfo> remapConflictPutVals = conflictPutVals != null
? new ArrayList<GridCacheDrInfo>(failed.size()) : null;
+        Collection<GridCacheVersion> remapConflictRmvVals = conflictRmvVals != null
? new ArrayList<GridCacheVersion>(failed.size()) : null;
 
         Iterator<?> keyIt = keys.iterator();
         Iterator<?> valsIt = vals != null ? vals.iterator() : null;
+        Iterator<GridCacheDrInfo> conflictPutValsIt = conflictPutVals != null ? conflictPutVals.iterator()
: null;
+        Iterator<GridCacheVersion> conflictRmvValsIt = conflictRmvVals != null ? conflictRmvVals.iterator()
: null;
 
         for (Object key : failed) {
             while (keyIt.hasNext()) {
                 Object nextKey = keyIt.next();
                 Object nextVal = valsIt != null ? valsIt.next() : null;
+                GridCacheDrInfo nextConflictPutVal = conflictPutValsIt != null ? conflictPutValsIt.next()
: null;
+                GridCacheVersion nextConflictRmvVal = conflictRmvValsIt != null ? conflictRmvValsIt.next()
: null;
 
                 if (F.eq(key, nextKey)) {
                     remapKeys.add(nextKey);
@@ -354,6 +388,12 @@ public class GridNearAtomicUpdateFuture extends GridFutureAdapter<Object>
                     if (remapVals != null)
                         remapVals.add(nextVal);
 
+                    if (remapConflictPutVals != null)
+                        remapConflictPutVals.add(nextConflictPutVal);
+
+                    if (remapConflictRmvVals != null)
+                        remapConflictRmvVals.add(nextConflictRmvVal);
+
                     break;
                 }
             }
@@ -361,13 +401,29 @@ public class GridNearAtomicUpdateFuture extends GridFutureAdapter<Object>
 
         keys = remapKeys;
         vals = remapVals;
+        conflictPutVals = remapConflictPutVals;
+        conflictRmvVals = remapConflictRmvVals;
 
-        mappings = new ConcurrentHashMap8<>(keys.size(), 1.0f);
         single = null;
         futVer = null;
         err = null;
         opRes = null;
-        topVer = AffinityTopologyVersion.ZERO;
+
+        GridFutureAdapter<Void> fut0;
+
+        synchronized (this) {
+            mappings = new ConcurrentHashMap8<>(keys.size(), 1.0f);
+
+            topVer = AffinityTopologyVersion.ZERO;
+
+            fut0 = topCompleteFut;
+
+            topCompleteFut = null;
+        }
+
+        if (fut0 != null)
+            fut0.onDone();
+
         singleNodeId = null;
         singleReq = null;
         fastMapRemap = false;
@@ -405,6 +461,15 @@ public class GridNearAtomicUpdateFuture extends GridFutureAdapter<Object>
             if (futVer != null)
                 cctx.mvcc().removeAtomicFuture(version());
 
+            GridFutureAdapter<Void> fut0;
+
+            synchronized (this) {
+                fut0 = topCompleteFut;
+            }
+
+            if (fut0 != null)
+                fut0.onDone();
+
             return true;
         }
 
@@ -544,6 +609,10 @@ public class GridNearAtomicUpdateFuture extends GridFutureAdapter<Object>
 
                 return;
             }
+
+            synchronized (this) {
+                this.topVer = topVer;
+            }
         }
         finally {
             cache.topology().readUnlock();
@@ -559,7 +628,8 @@ public class GridNearAtomicUpdateFuture extends GridFutureAdapter<Object>
         boolean remap = false;
 
         synchronized (this) {
-            if ((syncMode == FULL_ASYNC && cctx.config().getAtomicWriteOrderMode()
== PRIMARY) || mappings.isEmpty()) {
+            if (topVer != AffinityTopologyVersion.ZERO &&
+                ((syncMode == FULL_ASYNC && cctx.config().getAtomicWriteOrderMode()
== PRIMARY) || mappings.isEmpty())) {
                 CachePartialUpdateCheckedException err0 = err;
 
                 if (err0 != null)
@@ -1040,7 +1110,7 @@ public class GridNearAtomicUpdateFuture extends GridFutureAdapter<Object>
         if (err0 == null)
             err0 = this.err = new CachePartialUpdateCheckedException("Failed to update keys
(retry update if possible).");
 
-        List<Object> keys = new ArrayList<>(failedKeys.size());
+        Collection<Object> keys = new ArrayList<>(failedKeys.size());
 
         for (KeyCacheObject key : failedKeys)
             keys.add(key.value(cctx.cacheObjectContext(), false));

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/c2c90b52/modules/core/src/main/java/org/apache/ignite/spi/communication/tcp/TcpCommunicationSpi.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/spi/communication/tcp/TcpCommunicationSpi.java
b/modules/core/src/main/java/org/apache/ignite/spi/communication/tcp/TcpCommunicationSpi.java
index addf243d..4ca2995 100644
--- a/modules/core/src/main/java/org/apache/ignite/spi/communication/tcp/TcpCommunicationSpi.java
+++ b/modules/core/src/main/java/org/apache/ignite/spi/communication/tcp/TcpCommunicationSpi.java
@@ -210,7 +210,7 @@ public class TcpCommunicationSpi extends IgniteSpiAdapter
     public static final int DFLT_ACK_SND_THRESHOLD = 16;
 
     /** Default socket write timeout. */
-    public static final long DFLT_SOCK_WRITE_TIMEOUT = 200;
+    public static final long DFLT_SOCK_WRITE_TIMEOUT = 2000;
 
     /** No-op runnable. */
     private static final IgniteRunnable NOOP = new IgniteRunnable() {

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/c2c90b52/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridCacheAtomicInvalidPartitionHandlingSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridCacheAtomicInvalidPartitionHandlingSelfTest.java
b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridCacheAtomicInvalidPartitionHandlingSelfTest.java
index 054a110..b255558 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridCacheAtomicInvalidPartitionHandlingSelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridCacheAtomicInvalidPartitionHandlingSelfTest.java
@@ -41,6 +41,7 @@ import org.jsr166.*;
 import java.util.*;
 import java.util.concurrent.*;
 import java.util.concurrent.atomic.*;
+import java.util.concurrent.locks.*;
 
 import static org.apache.ignite.cache.CacheAtomicWriteOrderMode.*;
 import static org.apache.ignite.cache.CacheMode.*;
@@ -236,6 +237,8 @@ public class GridCacheAtomicInvalidPartitionHandlingSelfTest extends GridCommonA
 
             System.err.println("FINISHED PUTS");
 
+            GridCacheMapEntry.debug = true;
+
             // Start put threads.
             IgniteInternalFuture<?> fut = multithreadedAsync(new Callable<Object>()
{
                 @Override public Object call() throws Exception {
@@ -340,12 +343,12 @@ public class GridCacheAtomicInvalidPartitionHandlingSelfTest extends
GridCommonA
                         }
                         catch (AssertionError e) {
                             if (r == 9) {
-                                System.err.println("Failed to verify cache contents: " +
e.getMessage());
+                                info("Failed to verify cache contents: " + e.getMessage());
 
                                 throw e;
                             }
 
-                            System.err.println("Failed to verify cache contents, will retry:
" + e.getMessage());
+                            info("Failed to verify cache contents, will retry: " + e.getMessage());
 
                             // Give some time to finish async updates.
                             U.sleep(1000);


Mime
View raw message