ignite-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From vkuliche...@apache.org
Subject [16/40] incubator-ignite git commit: # ignite-57 async support for removeAll()
Date Mon, 09 Feb 2015 23:22:01 GMT
# ignite-57 async support for removeAll()


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

Branch: refs/heads/ignite-61
Commit: 9140a08747366186778382a09674c0e95b03d53a
Parents: ed55c48
Author: sboikov <semen.boikov@inria.fr>
Authored: Fri Feb 6 21:43:11 2015 +0300
Committer: sboikov <semen.boikov@inria.fr>
Committed: Fri Feb 6 21:43:11 2015 +0300

----------------------------------------------------------------------
 .../apache/ignite/cache/CacheProjection.java    |  5 ++
 .../cache/GridCacheProjectionImpl.java          |  5 ++
 .../processors/cache/GridCacheProxyImpl.java    | 12 ++++
 .../processors/cache/IgniteCacheProxy.java      |  5 +-
 .../GridDistributedCacheAdapter.java            | 60 ++++++++++++++++--
 .../distributed/near/GridNearAtomicCache.java   |  5 ++
 .../processors/cache/local/GridLocalCache.java  | 12 ++++
 .../local/atomic/GridLocalAtomicCache.java      | 11 ++++
 .../cache/GridCacheAbstractFullApiSelfTest.java | 64 +++++++++++++++-----
 9 files changed, 159 insertions(+), 20 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/9140a087/modules/core/src/main/java/org/apache/ignite/cache/CacheProjection.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/cache/CacheProjection.java b/modules/core/src/main/java/org/apache/ignite/cache/CacheProjection.java
index 84c2839..bca1ae4 100644
--- a/modules/core/src/main/java/org/apache/ignite/cache/CacheProjection.java
+++ b/modules/core/src/main/java/org/apache/ignite/cache/CacheProjection.java
@@ -1678,6 +1678,11 @@ public interface CacheProjection<K, V> extends Iterable<CacheEntry<K,
V>> {
     public void removeAll() throws IgniteCheckedException;
 
     /**
+     * @return Remove future.
+     */
+    public IgniteInternalFuture<?> removeAllAsync();
+
+    /**
      * Asynchronously removes mappings from cache for entries for which the optionally passed
in filters do
      * pass. If passed in filters are {@code null}, then all entries in cache will be enrolled
      * into transaction.

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/9140a087/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheProjectionImpl.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheProjectionImpl.java
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheProjectionImpl.java
index 80b7341..b4ac195 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheProjectionImpl.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheProjectionImpl.java
@@ -1206,6 +1206,11 @@ public class GridCacheProjectionImpl<K, V> implements GridCacheProjectionEx<K,
V
     }
 
     /** {@inheritDoc} */
+    @Override public IgniteInternalFuture<?> removeAllAsync() {
+        return cache.removeAllAsync();
+    }
+
+    /** {@inheritDoc} */
     @Override public IgniteInternalFuture<?> removeAllAsync(@Nullable IgnitePredicate<CacheEntry<K,
V>>... filter) {
         return cache.removeAllAsync(and(filter, false));
     }

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/9140a087/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheProxyImpl.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheProxyImpl.java
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheProxyImpl.java
index 8fd125e..ee8a89f 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheProxyImpl.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheProxyImpl.java
@@ -1614,6 +1614,18 @@ public class GridCacheProxyImpl<K, V> implements GridCacheProxy<K,
V>, Externali
     }
 
     /** {@inheritDoc} */
+    @Override public IgniteInternalFuture<?> removeAllAsync() {
+        GridCacheProjectionImpl<K, V> prev = gate.enter(prj);
+
+        try {
+            return delegate.removeAllAsync();
+        }
+        finally {
+            gate.leave(prev);
+        }
+    }
+
+    /** {@inheritDoc} */
     @Override public IgniteInternalFuture<?> removeAllAsync(@Nullable IgnitePredicate<CacheEntry<K,
V>>[] filter) {
         GridCacheProjectionImpl<K, V> prev = gate.enter(prj);
 

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/9140a087/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/IgniteCacheProxy.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/IgniteCacheProxy.java
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/IgniteCacheProxy.java
index c6166af..a733ed2 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/IgniteCacheProxy.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/IgniteCacheProxy.java
@@ -736,7 +736,10 @@ public class IgniteCacheProxy<K, V> extends IgniteAsyncSupportAdapter<IgniteCach
         GridCacheProjectionImpl<K, V> prev = gate.enter(prj);
 
         try {
-            delegate.removeAll();
+            if (isAsync())
+                setFuture(delegate.removeAllAsync());
+            else
+                delegate.removeAll();
         }
         catch (IgniteCheckedException e) {
             throw cacheException(e);

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/9140a087/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedCacheAdapter.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedCacheAdapter.java
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedCacheAdapter.java
index 5365ec3..c0ec962 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedCacheAdapter.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedCacheAdapter.java
@@ -29,6 +29,7 @@ import org.apache.ignite.internal.processors.cache.distributed.near.*;
 import org.apache.ignite.internal.processors.cache.transactions.*;
 import org.apache.ignite.internal.processors.cache.version.*;
 import org.apache.ignite.internal.processors.task.*;
+import org.apache.ignite.internal.util.future.*;
 import org.apache.ignite.internal.util.typedef.internal.*;
 import org.apache.ignite.lang.*;
 import org.apache.ignite.resources.*;
@@ -141,7 +142,7 @@ public abstract class GridDistributedCacheAdapter<K, V> extends
GridCacheAdapter
             do {
                 topVer = ctx.affinity().affinityTopologyVersion();
 
-                // Send job to all nodes.
+                // Send job to all data nodes.
                 Collection<ClusterNode> nodes = ctx.grid().forDataNodes(name()).nodes();
 
                 if (!nodes.isEmpty()) {
@@ -155,12 +156,61 @@ public abstract class GridDistributedCacheAdapter<K, V> extends
GridCacheAdapter
             if (log.isDebugEnabled())
                 log.debug("All remote nodes left while cache remove [cacheName=" + name()
+ "]");
         }
-        catch (ComputeTaskTimeoutCheckedException e) {
-            U.warn(log, "Timed out waiting for remote nodes to finish cache remove (consider
increasing " +
-                "'networkTimeout' configuration property) [cacheName=" + name() + "]");
+    }
+
+    /** {@inheritDoc} */
+    @Override public IgniteInternalFuture<?> removeAllAsync() {
+        GridFutureAdapter<Void> opFut = new GridFutureAdapter<>();
+
+        long topVer = ctx.affinity().affinityTopologyVersion();
+
+        removeAllAsync(opFut, topVer);
+
+        return opFut;
+    }
+
+    /**
+     * @param opFut Future.
+     * @param topVer Topology version.
+     */
+    private void removeAllAsync(final GridFutureAdapter<Void> opFut, final long topVer)
{
+        Collection<ClusterNode> nodes = ctx.grid().forDataNodes(name()).nodes();
 
-            throw e;
+        if (!nodes.isEmpty()) {
+            IgniteInternalFuture<?> rmvFut = ctx.closures().callAsyncNoFailover(BROADCAST,
+                    new GlobalRemoveAllCallable<>(name(), topVer), nodes, true);
+
+            rmvFut.listenAsync(new IgniteInClosure<IgniteInternalFuture<?>>()
{
+                @Override public void apply(IgniteInternalFuture<?> fut) {
+                    try {
+                        fut.get();
+
+                        long topVer0 = ctx.affinity().affinityTopologyVersion();
+
+                        if (topVer0 == topVer)
+                            opFut.onDone();
+                        else
+                            removeAllAsync(opFut, topVer0);
+                    }
+                    catch (ClusterGroupEmptyCheckedException ignore) {
+                        if (log.isDebugEnabled())
+                            log.debug("All remote nodes left while cache remove [cacheName="
+ name() + "]");
+
+                        opFut.onDone();
+                    }
+                    catch (IgniteCheckedException e) {
+                        opFut.onDone(e);
+                    }
+                    catch (Error e) {
+                        opFut.onDone(e);
+
+                        throw e;
+                    }
+                }
+            });
         }
+        else
+            opFut.onDone();
     }
 
     /** {@inheritDoc} */

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/9140a087/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearAtomicCache.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearAtomicCache.java
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearAtomicCache.java
index 335f268..62a59a7 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearAtomicCache.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearAtomicCache.java
@@ -635,6 +635,11 @@ public class GridNearAtomicCache<K, V> extends GridNearCacheAdapter<K,
V> {
     }
 
     /** {@inheritDoc} */
+    @Override public IgniteInternalFuture<?> removeAllAsync() {
+        return dht.removeAllAsync();
+    }
+
+    /** {@inheritDoc} */
     @Override public IgniteInternalFuture<?> removeAllAsync(IgnitePredicate<CacheEntry<K,
V>>[] filter) {
         return dht.removeAllAsync(keySet(filter));
     }

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/9140a087/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/local/GridLocalCache.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/local/GridLocalCache.java
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/local/GridLocalCache.java
index b4bc3a5..845f697 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/local/GridLocalCache.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/local/GridLocalCache.java
@@ -31,6 +31,7 @@ import org.jetbrains.annotations.*;
 
 import java.io.*;
 import java.util.*;
+import java.util.concurrent.*;
 
 /**
  * Local cache implementation.
@@ -197,6 +198,17 @@ public class GridLocalCache<K, V> extends GridCacheAdapter<K,
V> {
     }
 
     /** {@inheritDoc} */
+    @Override public IgniteInternalFuture<?> removeAllAsync() {
+        return ctx.closures().callLocalSafe(new Callable<Void>() {
+            @Override public Void call() throws Exception {
+                removeAll();
+
+                return null;
+            }
+        });
+    }
+
+    /** {@inheritDoc} */
     @Override public void onDeferredDelete(GridCacheEntryEx<K, V> entry, GridCacheVersion
ver) {
         assert false : "Should not be called";
     }

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/9140a087/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/local/atomic/GridLocalAtomicCache.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/local/atomic/GridLocalAtomicCache.java
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/local/atomic/GridLocalAtomicCache.java
index 4a268d4..109667e 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/local/atomic/GridLocalAtomicCache.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/local/atomic/GridLocalAtomicCache.java
@@ -472,6 +472,17 @@ public class GridLocalAtomicCache<K, V> extends GridCacheAdapter<K,
V> {
     }
 
     /** {@inheritDoc} */
+    @Override public IgniteInternalFuture<?> removeAllAsync() {
+        return ctx.closures().callLocalSafe(new Callable<Void>() {
+            @Override public Void call() throws Exception {
+                removeAll();
+
+                return null;
+            }
+        });
+    }
+
+    /** {@inheritDoc} */
     @Override public IgniteInternalFuture<?> removeAllAsync(IgnitePredicate<CacheEntry<K,
V>>[] filter) {
         return removeAllAsync(keySet(filter), filter);
     }

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/9140a087/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheAbstractFullApiSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheAbstractFullApiSelfTest.java
b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheAbstractFullApiSelfTest.java
index 0d66b15..ce06330 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheAbstractFullApiSelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheAbstractFullApiSelfTest.java
@@ -2098,14 +2098,36 @@ public abstract class GridCacheAbstractFullApiSelfTest extends GridCacheAbstract
     /**
      * @throws Exception In case of error.
      */
-    public void testRemoveAll() throws Exception {
-        cache().put("key1", 1);
-        cache().put("key2", 2);
-        cache().put("key3", 3);
+    public void testGlobalRemoveAll() throws Exception {
+        globalRemoveAll(false);
+    }
+
+    /**
+     * @throws Exception In case of error.
+     */
+    public void testGlobalRemoveAllAsync() throws Exception {
+        globalRemoveAll(true);
+    }
+
+    /**
+     * @throws Exception In case of error.
+     */
+    private void globalRemoveAll(boolean async) throws Exception {
+        jcache().put("key1", 1);
+        jcache().put("key2", 2);
+        jcache().put("key3", 3);
 
         checkSize(F.asSet("key1", "key2", "key3"));
 
-        cache().removeAll(F.asList("key1", "key2"));
+        IgniteCache<String, Integer> asyncCache = jcache().withAsync();
+
+        if (async) {
+            asyncCache.removeAll(F.asSet("key1", "key2"));
+
+            asyncCache.future().get();
+        }
+        else
+            jcache().removeAll(F.asSet("key1", "key2"));
 
         checkSize(F.asSet("key3"));
 
@@ -2114,24 +2136,38 @@ public abstract class GridCacheAbstractFullApiSelfTest extends GridCacheAbstract
         checkContainsKey(true, "key3");
 
         // Put values again.
-        cache().put("key1", 1);
-        cache().put("key2", 2);
-        cache().put("key3", 3);
+        jcache().put("key1", 1);
+        jcache().put("key2", 2);
+        jcache().put("key3", 3);
+
+        if (async) {
+            IgniteCache<String, Integer> asyncCache0 = jcache(gridCount() > 1 ?
1 : 0).withAsync();
 
-        cache(gridCount() > 1 ? 1 : 0).removeAll();
+            asyncCache0.removeAll();
+
+            asyncCache0.future().get();
+        }
+        else
+            jcache(gridCount() > 1 ? 1 : 0).removeAll();
 
         assert cache().isEmpty();
-        long entryCount = hugeRemoveAllEntryCount();
+        long entryCnt = hugeRemoveAllEntryCount();
 
-        for (int i = 0; i < entryCount; i++)
+        for (int i = 0; i < entryCnt; i++)
             cache().put(String.valueOf(i), i);
 
-        for (int i = 0; i < entryCount; i++)
+        for (int i = 0; i < entryCnt; i++)
             assertEquals(Integer.valueOf(i), cache().get(String.valueOf(i)));
 
-        cache().removeAll();
+        if (async) {
+            asyncCache.removeAll();
+
+            asyncCache.future().get();
+        }
+        else
+            cache().removeAll();
 
-        for (int i = 0; i < entryCount; i++)
+        for (int i = 0; i < entryCnt; i++)
             assertNull(cache().get(String.valueOf(i)));
     }
 


Mime
View raw message