ignite-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From sboi...@apache.org
Subject incubator-ignite git commit: #ignite-373: Use tasks instead of callable in removeAll.
Date Tue, 12 May 2015 13:32:37 GMT
Repository: incubator-ignite
Updated Branches:
  refs/heads/ignite-373 ceb283542 -> e50632cad


#ignite-373: Use tasks instead of callable in 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/e50632ca
Tree: http://git-wip-us.apache.org/repos/asf/incubator-ignite/tree/e50632ca
Diff: http://git-wip-us.apache.org/repos/asf/incubator-ignite/diff/e50632ca

Branch: refs/heads/ignite-373
Commit: e50632cad41ddd076d1e5bf65ec87d1aa3d06485
Parents: ceb2835
Author: ivasilinets <ivasilinets@gridgain.com>
Authored: Tue May 12 16:29:39 2015 +0300
Committer: ivasilinets <ivasilinets@gridgain.com>
Committed: Tue May 12 16:29:39 2015 +0300

----------------------------------------------------------------------
 .../processors/cache/GridCacheAdapter.java      |  19 ++-
 .../GridDistributedCacheAdapter.java            | 153 +++++++++++--------
 2 files changed, 99 insertions(+), 73 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/e50632ca/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 3826bfa..0b7fa91 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
@@ -4842,7 +4842,9 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K,
V
         }
 
         /** {@inheritDoc} */
-        @Nullable @Override public Object localExecute(@Nullable IgniteInternalCache cache)
{
+        @Nullable @Override public Object localExecute() {
+            IgniteInternalCache cache = ((IgniteKernal)ignite).context().cache().cache(cacheName);
+
             if (cache != null)
                 cache.clearLocally();
 
@@ -4880,7 +4882,9 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K,
V
         }
 
         /** {@inheritDoc} */
-        @Nullable @Override public Object localExecute(@Nullable IgniteInternalCache cache)
{
+        @Nullable @Override public Object localExecute() {
+            IgniteInternalCache cache = ((IgniteKernal)ignite).context().cache().cache(cacheName);
+
             if (cache != null)
                 cache.clearLocallyAll(keys);
 
@@ -4918,7 +4922,9 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K,
V
         }
 
         /** {@inheritDoc} */
-        @Nullable @Override public Object localExecute(@Nullable IgniteInternalCache cache)
{
+        @Nullable @Override public Object localExecute() {
+            IgniteInternalCache cache = ((IgniteKernal)ignite).context().cache().cache(cacheName);
+
             if (cache == null)
                 return 0;
 
@@ -5541,16 +5547,13 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K,
V
         @Nullable @Override public final Object execute() {
             waitAffinityReadyFuture();
 
-            IgniteInternalCache cache = ((IgniteKernal)ignite).context().cache().cache(cacheName);
-
-            return localExecute(cache);
+            return localExecute();
         }
 
         /**
-         * @param cache Cache.
          * @return Local execution result.
          */
-        @Nullable protected abstract Object localExecute(@Nullable IgniteInternalCache cache);
+        @Nullable protected abstract Object localExecute();
 
         /**
          * Holds (suspends) job execution until our cache version becomes equal to remote
cache's version.

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/e50632ca/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 d10ab56..cafa8b3 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
@@ -19,6 +19,7 @@ package org.apache.ignite.internal.processors.cache.distributed;
 
 import org.apache.ignite.*;
 import org.apache.ignite.cluster.*;
+import org.apache.ignite.compute.*;
 import org.apache.ignite.internal.*;
 import org.apache.ignite.internal.cluster.*;
 import org.apache.ignite.internal.processors.affinity.*;
@@ -33,16 +34,14 @@ import org.apache.ignite.internal.util.future.*;
 import org.apache.ignite.internal.util.lang.*;
 import org.apache.ignite.internal.util.typedef.internal.*;
 import org.apache.ignite.lang.*;
-import org.apache.ignite.resources.*;
 import org.apache.ignite.transactions.*;
 import org.jetbrains.annotations.*;
 
 import java.io.*;
 import java.util.*;
-import java.util.concurrent.*;
 
-import static org.apache.ignite.internal.GridClosureCallMode.*;
 import static org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtPartitionState.*;
+import static org.apache.ignite.internal.processors.task.GridTaskThreadContextKey.*;
 
 /**
  * Distributed cache implementation.
@@ -144,10 +143,10 @@ public abstract class GridDistributedCacheAdapter<K, V> extends
GridCacheAdapter
         try {
             AffinityTopologyVersion topVer;
 
-            boolean removedAll;
+            Boolean rmvAll;
 
             do {
-                removedAll = true;
+                rmvAll = true;
 
                 topVer = ctx.affinity().affinityTopologyVersion();
 
@@ -155,19 +154,12 @@ public abstract class GridDistributedCacheAdapter<K, V> extends
GridCacheAdapter
                 Collection<ClusterNode> nodes = ctx.grid().cluster().forDataNodes(name()).nodes();
 
                 if (!nodes.isEmpty()) {
-                    CacheOperationContext opCtx = ctx.operationContextPerCall();
+                    ctx.kernalContext().task().setThreadContext(TC_SUBGRID, nodes);
 
-                    Collection<Boolean> results = ctx.closures().callAsyncNoFailover(BROADCAST,
-                        Collections.singleton(new GlobalRemoveAllCallable<>(name(),
topVer,
-                        opCtx != null && opCtx.skipStore())), nodes, true).get();
-
-                    for (Boolean res : results) {
-                        if (res != null && !res)
-                            removedAll = false;
-                    }
+                    rmvAll = ctx.kernalContext().task().execute(new RemoveAllTask(ctx), null).get();
                 }
             }
-            while (ctx.affinity().affinityTopologyVersion().compareTo(topVer) != 0 || !removedAll);
+            while (ctx.affinity().affinityTopologyVersion().compareTo(topVer) != 0 || rmvAll
== null || !rmvAll);
         }
         catch (ClusterGroupEmptyCheckedException ignore) {
             if (log.isDebugEnabled())
@@ -194,19 +186,18 @@ public abstract class GridDistributedCacheAdapter<K, V> extends
GridCacheAdapter
         Collection<ClusterNode> nodes = ctx.grid().cluster().forDataNodes(name()).nodes();
 
         if (!nodes.isEmpty()) {
-            CacheOperationContext opCtx = ctx.operationContextPerCall();
+            ctx.kernalContext().task().setThreadContext(TC_SUBGRID, nodes);
 
-            IgniteInternalFuture<?> rmvFut = ctx.closures().callAsyncNoFailover(BROADCAST,
-                    new GlobalRemoveAllCallable<>(name(), topVer, opCtx != null &&
opCtx.skipStore()), nodes, true);
+            IgniteInternalFuture<Boolean> rmvAll = ctx.kernalContext().task().execute(new
RemoveAllTask(ctx), null);
 
-            rmvFut.listen(new IgniteInClosure<IgniteInternalFuture<?>>() {
-                @Override public void apply(IgniteInternalFuture<?> fut) {
+            rmvAll.listen(new IgniteInClosure<IgniteInternalFuture<Boolean>>()
{
+                @Override public void apply(IgniteInternalFuture<Boolean> fut) {
                     try {
-                        fut.get();
+                        Boolean res = fut.get();
 
                         AffinityTopologyVersion topVer0 = ctx.affinity().affinityTopologyVersion();
 
-                        if (topVer0.equals(topVer))
+                        if (topVer0.equals(topVer) && res != null && res)
                             opFut.onDone();
                         else
                             removeAllAsync(opFut, topVer0);
@@ -238,31 +229,78 @@ public abstract class GridDistributedCacheAdapter<K, V> extends
GridCacheAdapter
     }
 
     /**
-     * Internal callable which performs remove all primary key mappings
-     * operation on a cache with the given name.
+     * Remove task.
      */
     @GridInternal
-    private static class GlobalRemoveAllCallable<K,V> implements Callable<Boolean>,
Externalizable {
+    private static class RemoveAllTask extends ComputeTaskAdapter<Object, Boolean>
{
         /** */
         private static final long serialVersionUID = 0L;
 
-        /** Cache name. */
-        private String cacheName;
+        /** Cache context. */
+        private GridCacheContext ctx;
+
+        /**
+         * Empty constructor for serialization.
+         */
+        public RemoveAllTask(){
+            // No-op.
+        }
+
+        /**
+         * @param ctx Cache context.
+         */
+        public RemoveAllTask(GridCacheContext ctx) {
+            this.ctx = ctx;
+        }
+
+        /** {@inheritDoc} */
+        @Nullable @Override public Map<? extends ComputeJob, ClusterNode> map(List<ClusterNode>
subgrid,
+            @Nullable Object arg) throws IgniteException {
+            Map<ComputeJob, ClusterNode> jobs = new HashMap();
 
-        /** Topology version. */
-        private AffinityTopologyVersion topVer;
+            CacheOperationContext opCtx = ctx.operationContextPerCall();
+
+            boolean skipStore = opCtx != null && opCtx.skipStore();
+
+            for (ClusterNode node : subgrid) {
+                jobs.put(new GlobalRemoveAllJob(ctx.name(), ctx.affinity().affinityTopologyVersion(),
skipStore),
+                    node);
+            }
+
+            return jobs;
+        }
+
+        /** {@inheritDoc} */
+        @Override public ComputeJobResultPolicy result(ComputeJobResult res, List<ComputeJobResult>
rcvd) {
+            return ComputeJobResultPolicy.WAIT;
+        }
+
+        /** {@inheritDoc} */
+        @Nullable @Override public Boolean reduce(List<ComputeJobResult> results) throws
IgniteException {
+            for (ComputeJobResult locRes : results) {
+                if (locRes != null && (locRes.getException() != null || !locRes.<Boolean>getData()))
+                    return false;
+            }
+
+            return true;
+        }
+    }
+    /**
+     * Internal job which performs remove all primary key mappings
+     * operation on a cache with the given name.
+     */
+    @GridInternal
+    private static class GlobalRemoveAllJob<K,V>  extends TopologyVersionAwareJob {
+        /** */
+        private static final long serialVersionUID = 0L;
 
         /** Skip store flag. */
         private boolean skipStore;
 
-        /** Injected grid instance. */
-        @IgniteInstanceResource
-        private Ignite ignite;
-
         /**
          * Empty constructor for serialization.
          */
-        public GlobalRemoveAllCallable() {
+        public GlobalRemoveAllJob() {
             // No-op.
         }
 
@@ -271,24 +309,20 @@ public abstract class GridDistributedCacheAdapter<K, V> extends
GridCacheAdapter
          * @param topVer Topology version.
          * @param skipStore Skip store flag.
          */
-        private GlobalRemoveAllCallable(String cacheName, @NotNull AffinityTopologyVersion
topVer, boolean skipStore) {
-            this.cacheName = cacheName;
-            this.topVer = topVer;
+        private GlobalRemoveAllJob(String cacheName, @NotNull AffinityTopologyVersion topVer,
boolean skipStore) {
+            super(cacheName, topVer);
+
             this.skipStore = skipStore;
         }
 
-        /**
-         * {@inheritDoc}
-         */
-        @Override public Boolean call() throws Exception {
-            GridCacheAdapter<K, V> cacheAdapter = ((IgniteKernal)ignite).context().cache().internalCache(cacheName);
+        /** {@inheritDoc} */
+        @Nullable @Override public Object localExecute() {
+            GridCacheAdapter cache = ((IgniteKernal) ignite).context().cache().internalCache(cacheName);
 
-            if (cacheAdapter == null)
+            if (cache == null)
                 return false;
 
-            final GridCacheContext<K, V> ctx = cacheAdapter.context();
-
-            ctx.affinity().affinityReadyFuture(topVer).get();
+            final GridCacheContext<K, V> ctx = cache.context();
 
             ctx.gate().enter();
 
@@ -299,16 +333,16 @@ public abstract class GridDistributedCacheAdapter<K, V> extends
GridCacheAdapter
                 GridDhtCacheAdapter<K, V> dht;
                 GridNearCacheAdapter<K, V> near = null;
 
-                if (cacheAdapter instanceof GridNearCacheAdapter) {
-                    near = ((GridNearCacheAdapter<K, V>)cacheAdapter);
+                if (cache instanceof GridNearCacheAdapter) {
+                    near = ((GridNearCacheAdapter<K, V>) cache);
                     dht = near.dht();
                 }
                 else
-                    dht = (GridDhtCacheAdapter<K, V>)cacheAdapter;
+                    dht = (GridDhtCacheAdapter<K, V>) cache;
 
                 try (DataStreamerImpl<KeyCacheObject, Object> dataLdr =
-                         (DataStreamerImpl)ignite.dataStreamer(cacheName)) {
-                    ((DataStreamerImpl)dataLdr).maxRemapCount(0);
+                         (DataStreamerImpl) ignite.dataStreamer(cacheName)) {
+                    ((DataStreamerImpl) dataLdr).maxRemapCount(0);
 
                     dataLdr.skipStore(skipStore);
 
@@ -351,25 +385,14 @@ public abstract class GridDistributedCacheAdapter<K, V> extends
GridCacheAdapter
                     }
                 }
             }
+            catch (IgniteCheckedException e) {
+                throw U.convertException(e);
+            }
             finally {
                 ctx.gate().leave();
             }
 
             return true;
         }
-
-        /** {@inheritDoc} */
-        @Override public void writeExternal(ObjectOutput out) throws IOException {
-            U.writeString(out, cacheName);
-            out.writeObject(topVer);
-            out.writeBoolean(skipStore);
-        }
-
-        /** {@inheritDoc} */
-        @Override public void readExternal(ObjectInput in) throws IOException, ClassNotFoundException
{
-            cacheName = U.readString(in);
-            topVer = (AffinityTopologyVersion)in.readObject();
-            skipStore = in.readBoolean();
-        }
     }
 }


Mime
View raw message