ignite-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From sboi...@apache.org
Subject [50/50] incubator-ignite git commit: #ignite-834: Change size and clear to jobs.
Date Tue, 05 May 2015 12:24:56 GMT
#ignite-834: Change size and clear to jobs.


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

Branch: refs/heads/ignite-834
Commit: 863d6bf69f75c2bf1a7571ffa76259544baf0539
Parents: 086553e
Author: ivasilinets <ivasilinets@gridgain.com>
Authored: Tue May 5 15:23:41 2015 +0300
Committer: ivasilinets <ivasilinets@gridgain.com>
Committed: Tue May 5 15:23:41 2015 +0300

----------------------------------------------------------------------
 .../processors/cache/GridCacheAdapter.java      | 331 +++++--------------
 .../GridDistributedCacheAdapter.java            |  28 +-
 2 files changed, 103 insertions(+), 256 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/863d6bf6/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 d59e9cc..596705e 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
@@ -25,7 +25,6 @@ import org.apache.ignite.compute.*;
 import org.apache.ignite.configuration.*;
 import org.apache.ignite.internal.*;
 import org.apache.ignite.internal.cluster.*;
-import org.apache.ignite.internal.compute.*;
 import org.apache.ignite.internal.processors.affinity.*;
 import org.apache.ignite.internal.processors.cache.affinity.*;
 import org.apache.ignite.internal.processors.cache.distributed.*;
@@ -1123,32 +1122,31 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K,
V
      * @param clearCall Global clear callable object.
      * @throws IgniteCheckedException In case of cache could not be cleared on any of the
nodes.
      */
-    private void clearRemotes(long timeout, GlobalClearCallable clearCall) throws IgniteCheckedException
{
-        try {
-            // Send job to remote nodes only.
-            Collection<ClusterNode> nodes =
-                ctx.grid().cluster().forCacheNodes(name(), true, true, false).forRemotes().nodes();
+    private void clearRemotes(long timeout, final TopologyVersionAwareJob clearCall) throws
IgniteCheckedException {
+        // Send job to remote nodes only.
+        ClusterGroup nodes = ctx.grid().cluster().forCacheNodes(name(), true, true, false).forRemotes();
 
-            IgniteInternalFuture<Object> fut = null;
+        if (!nodes.nodes().isEmpty()) {
+            ctx.kernalContext().task().setThreadContext(TC_TIMEOUT, timeout);
 
-            if (!nodes.isEmpty()) {
-                ctx.kernalContext().task().setThreadContext(TC_TIMEOUT, timeout);
+            ctx.grid().context().task().setThreadContext(TC_SUBGRID, nodes);
 
-                fut = ctx.closures().callAsyncNoFailover(BROADCAST, clearCall, nodes, true);
-            }
+            ctx.grid().context().task().execute(new ComputeTaskAdapter<Object, Object>()
{
+                /** {@inheritDoc} */
+                @Nullable @Override public Map<? extends ComputeJob, ClusterNode> map(List<ClusterNode>
subgrid,
+                    @Nullable Object arg) throws IgniteException {
+                    Map<ComputeJob, ClusterNode> jobs = new HashMap();
+                    for (ClusterNode node : subgrid)
+                        jobs.put(clearCall, node);
 
-            if (fut != null)
-                fut.get();
-        }
-        catch (ClusterGroupEmptyCheckedException ignore) {
-            if (log.isDebugEnabled())
-                log.debug("All remote nodes left while cache clearLocally [cacheName=" +
name() + "]");
-        }
-        catch (ComputeTaskTimeoutCheckedException e) {
-            U.warn(log, "Timed out waiting for remote nodes to finish cache clear (consider
increasing " +
-                "'networkTimeout' configuration property) [cacheName=" + name() + "]");
+                    return jobs;
+                }
 
-            throw e;
+                /** {@inheritDoc} */
+                @Nullable @Override public Object reduce(List<ComputeJobResult> results)
throws IgniteException {
+                    return null;
+                }
+            }, null).get();
         }
     }
 
@@ -1161,26 +1159,28 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K,
V
      * @param clearCall Global clear callable object.
      * @return Future.
      */
-    private IgniteInternalFuture<?> clearAsync(GlobalClearCallable clearCall) {
+    private IgniteInternalFuture<?> clearAsync(final TopologyVersionAwareJob clearCall)
{
         Collection<ClusterNode> nodes = ctx.grid().cluster().forCacheNodes(name(),
true, true, false).nodes();
 
         if (!nodes.isEmpty()) {
-            IgniteInternalFuture<Object> fut =
-                ctx.closures().callAsyncNoFailover(BROADCAST, clearCall, nodes, true);
+            ctx.grid().context().task().setThreadContext(TC_SUBGRID, nodes);
 
-            return fut.chain(new CX1<IgniteInternalFuture<Object>, Object>()
{
-                @Override public Object applyx(IgniteInternalFuture<Object> fut) throws
IgniteCheckedException {
-                    try {
-                        return fut.get();
-                    }
-                    catch (ClusterGroupEmptyCheckedException ignore) {
-                        if (log.isDebugEnabled())
-                            log.debug("All remote nodes left while cache clearLocally [cacheName="
+ name() + "]");
+            return ctx.grid().context().task().execute(new ComputeTaskAdapter<Object,
Object>() {
+                /** {@inheritDoc} */
+                @Nullable @Override public Map<? extends ComputeJob, ClusterNode> map(List<ClusterNode>
subgrid,
+                                                                  @Nullable Object arg) throws
IgniteException {
+                    Map<ComputeJob, ClusterNode> jobs = new HashMap();
+                    for (ClusterNode node : subgrid)
+                        jobs.put(clearCall, node);
 
-                        return null;
-                    }
+                    return jobs;
                 }
-            });
+
+                /** {@inheritDoc} */
+                @Nullable @Override public Object reduce(List<ComputeJobResult> results)
throws IgniteException {
+                    return null;
+                }
+            }, null);
         }
         else
             return new GridFinishedFuture<>();
@@ -3567,7 +3567,7 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K,
V
     }
 
     /** {@inheritDoc} */
-    @Override public IgniteInternalFuture<Integer> sizeAsync(CachePeekMode[] peekModes)
{
+    @Override public IgniteInternalFuture<Integer> sizeAsync(final CachePeekMode[]
peekModes) {
         assert peekModes != null;
 
         PeekModes modes = parsePeekModes(peekModes, true);
@@ -3581,23 +3581,32 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K,
V
         if (nodes.isEmpty())
             return new GridFinishedFuture<>(0);
 
-        IgniteInternalFuture<Collection<Integer>> fut =
-            ctx.closures().broadcastNoFailover(
-                new SizeCallable(ctx.name(), ctx.affinity().affinityTopologyVersion(), peekModes),
null, nodes);
+        ctx.grid().context().task().setThreadContext(TC_SUBGRID, nodes);
+
+        return ctx.grid().context().task().execute(new ComputeTaskAdapter<Object, Integer>()
{
+
+            /** {@inheritDoc} */
+            @Nullable @Override public Map<? extends ComputeJob, ClusterNode> map(List<ClusterNode>
subgrid,
+                @Nullable Object arg) throws IgniteException {
+                Map<ComputeJob, ClusterNode> jobs = new HashMap();
+                for (ClusterNode node : subgrid)
+                    jobs.put(new SizeCallable(ctx.name(), ctx.affinity().affinityTopologyVersion(),
peekModes), node);
 
-        return fut.chain(new CX1<IgniteInternalFuture<Collection<Integer>>,
Integer>() {
-            @Override public Integer applyx(IgniteInternalFuture<Collection<Integer>>
fut)
-            throws IgniteCheckedException {
-                Collection<Integer> res = fut.get();
+                return jobs;
+            }
 
-                int totalSize = 0;
+            /** {@inheritDoc} */
+            @Nullable @Override public Integer reduce(List<ComputeJobResult> results)
throws IgniteException {
+                int size = 0;
 
-                for (Integer size : res)
-                    totalSize += size;
+                for (ComputeJobResult res : results) {
+                    if (res.getException() == null)
+                        size += res.<Integer>getData();
+                }
 
-                return totalSize;
+                return size;
             }
-        });
+        }, null);
     }
 
     /** {@inheritDoc} */
@@ -3915,51 +3924,6 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K,
V
     }
 
     /**
-     * Gets cache global size (with or without backups).
-     *
-     * @param primaryOnly {@code True} if only primary sizes should be included.
-     * @return Global size.
-     * @throws IgniteCheckedException If internal task execution failed.
-     */
-    private int globalSize(boolean primaryOnly) throws IgniteCheckedException {
-        try {
-            // Send job to remote nodes only.
-            Collection<ClusterNode> nodes = ctx.grid().cluster().forCacheNodes(name()).forRemotes().nodes();
-
-            IgniteInternalFuture<Collection<Integer>> fut = null;
-
-            if (!nodes.isEmpty()) {
-                ctx.kernalContext().task().setThreadContext(TC_TIMEOUT, gridCfg.getNetworkTimeout());
-
-                fut = ctx.closures().broadcastNoFailover(
-                    new GlobalSizeCallable(name(), ctx.affinity().affinityTopologyVersion(),
primaryOnly), null, nodes);
-            }
-
-            // Get local value.
-            int globalSize = primaryOnly ? primarySize() : size();
-
-            if (fut != null) {
-                for (Integer i : fut.get())
-                    globalSize += i;
-            }
-
-            return globalSize;
-        }
-        catch (ClusterGroupEmptyCheckedException ignore) {
-            if (log.isDebugEnabled())
-                log.debug("All remote nodes left while cache clearLocally [cacheName=" +
name() + "]");
-
-            return primaryOnly ? primarySize() : size();
-        }
-        catch (ComputeTaskTimeoutCheckedException e) {
-            U.warn(log, "Timed out waiting for remote nodes to finish cache clear (consider
increasing " +
-                "'networkTimeout' configuration property) [cacheName=" + name() + "]");
-
-            throw e;
-        }
-    }
-
-    /**
      * @param op Cache operation.
      * @param <T> Return type.
      * @return Operation result.
@@ -4900,41 +4864,10 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K,
V
     }
 
     /**
-     * Internal callable which performs clear operation on a cache with the given name.
-     */
-    @GridInternal
-    private static abstract class GlobalClearCallable extends DelayedCallable implements
Callable<Object> {
-        /**
-         * Empty constructor for serialization.
-         */
-        public GlobalClearCallable() {
-            // No-op.
-        }
-
-        /**
-         * @param cacheName Cache name.
-         * @param topVer Affinity topology version.
-         */
-        protected GlobalClearCallable(String cacheName, AffinityTopologyVersion topVer) {
-            super(cacheName, topVer);
-        }
-
-        /** {@inheritDoc} */
-        @Override public void writeExternal(ObjectOutput out) throws IOException {
-            super.writeExternal(out);
-        }
-
-        /** {@inheritDoc} */
-        @Override public void readExternal(ObjectInput in) throws IOException, ClassNotFoundException
{
-            super.readExternal(in);
-        }
-    }
-
-    /**
      * Global clear all.
      */
     @GridInternal
-    private static class GlobalClearAllCallable extends GlobalClearCallable {
+    private static class GlobalClearAllCallable extends TopologyVersionAwareJob {
         /** */
         private static final long serialVersionUID = 0L;
 
@@ -4954,9 +4887,7 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K,
V
         }
 
         /** {@inheritDoc} */
-        @Override public Object call() throws Exception {
-            waitAffinityReadyFuture();
-
+        @Nullable @Override public Object localExecute() {
             ((IgniteEx)ignite).cachex(cacheName).clearLocally();
 
             return null;
@@ -4967,7 +4898,7 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K,
V
      * Global clear keys.
      */
     @GridInternal
-    private static class GlobalClearKeySetCallable<K, V> extends GlobalClearCallable
{
+    private static class GlobalClearKeySetCallable<K, V> extends TopologyVersionAwareJob
{
         /** */
         private static final long serialVersionUID = 0L;
 
@@ -4993,34 +4924,18 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K,
V
         }
 
         /** {@inheritDoc} */
-        @Override public Object call() throws Exception {
-            waitAffinityReadyFuture();
-
+        @Nullable @Override public Object localExecute() {
             ((IgniteEx)ignite).<K, V>cachex(cacheName).clearLocallyAll(keys);
 
             return null;
         }
-
-        /** {@inheritDoc} */
-        @Override public void writeExternal(ObjectOutput out) throws IOException {
-            super.writeExternal(out);
-
-            out.writeObject(keys);
-        }
-
-        /** {@inheritDoc} */
-        @Override public void readExternal(ObjectInput in) throws IOException, ClassNotFoundException
{
-            super.readExternal(in);
-
-            keys = (Set<K>) in.readObject();
-        }
     }
 
     /**
      * Internal callable for global size calculation.
      */
     @GridInternal
-    private static class SizeCallable extends DelayedCallable implements IgniteClosure<Object,
Integer> {
+    private static class SizeCallable extends TopologyVersionAwareJob {
         /** */
         private static final long serialVersionUID = 0L;
 
@@ -5046,9 +4961,7 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K,
V
         }
 
         /** {@inheritDoc} */
-        @Override public Integer apply(Object o) {
-            waitAffinityReadyFuture();
-
+        @Nullable @Override public Object localExecute() {
             IgniteInternalCache<Object, Object> cache = ((IgniteEx)ignite).cachex(cacheName);
 
             assert cache != null : cacheName;
@@ -5062,89 +4975,12 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K,
V
         }
 
         /** {@inheritDoc} */
-        @SuppressWarnings("ForLoopReplaceableByForEach")
-        @Override public void writeExternal(ObjectOutput out) throws IOException {
-            super.writeExternal(out);
-
-            out.writeInt(peekModes.length);
-
-            for (int i = 0; i < peekModes.length; i++)
-                U.writeEnum(out, peekModes[i]);
-        }
-
-        /** {@inheritDoc} */
-        @Override public void readExternal(ObjectInput in) throws IOException, ClassNotFoundException
{
-            super.readExternal(in);
-
-            int len = in.readInt();
-
-            peekModes = new CachePeekMode[len];
-
-            for (int i = 0; i < len; i++)
-                peekModes[i] = CachePeekMode.fromOrdinal(in.readByte());
-        }
-
-        /** {@inheritDoc} */
         public String toString() {
             return S.toString(SizeCallable.class, this);
         }
     }
 
     /**
-     * Internal callable which performs {@link IgniteInternalCache#size()} or {@link IgniteInternalCache#primarySize()}
-     * operation on a cache with the given name.
-     */
-    @GridInternal
-    private static class GlobalSizeCallable extends DelayedCallable implements IgniteClosure<Object,
Integer>, Externalizable {
-        /** */
-        private static final long serialVersionUID = 0L;
-
-        /** Primary only flag. */
-        private boolean primaryOnly;
-
-        /**
-         * Empty constructor for serialization.
-         */
-        public GlobalSizeCallable() {
-            // No-op.
-        }
-
-        /**
-         * @param cacheName Cache name.
-         * @param topVer Affinity topology version.
-         * @param primaryOnly Primary only flag.
-         */
-        private GlobalSizeCallable(String cacheName, AffinityTopologyVersion topVer, boolean
primaryOnly) {
-            super(cacheName, topVer);
-
-            this.primaryOnly = primaryOnly;
-        }
-
-        /** {@inheritDoc} */
-        @Override public Integer apply(Object o) {
-            waitAffinityReadyFuture();
-
-            IgniteInternalCache<Object, Object> cache = ((IgniteEx)ignite).cachex(cacheName);
-
-            return primaryOnly ? cache.primarySize() : cache.size();
-        }
-
-        /** {@inheritDoc} */
-        @Override public void writeExternal(ObjectOutput out) throws IOException {
-            super.writeExternal(out);
-
-            out.writeBoolean(primaryOnly);
-        }
-
-        /** {@inheritDoc} */
-        @Override public void readExternal(ObjectInput in) throws IOException, ClassNotFoundException
{
-            super.readExternal(in);
-
-            primaryOnly = in.readBoolean();
-        }
-    }
-
-    /**
      * Holder for last async operation future.
      */
     protected static class FutureHolder {
@@ -5709,7 +5545,7 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K,
V
     /**
      * Delayed callable class.
      */
-    protected static abstract class DelayedCallable<K, V> implements Externalizable
{
+    protected static abstract class TopologyVersionAwareJob<K, V> extends ComputeJobAdapter
{
         /** */
         private static final long serialVersionUID = 0L;
 
@@ -5730,22 +5566,37 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K,
V
         /**
          * Empty constructor for serialization.
          */
-        public DelayedCallable() {
+        public TopologyVersionAwareJob() {
             // No-op.
         }
 
         /**
+         * @param cacheName Cache name.
          * @param topVer Affinity topology version.
          */
-        public DelayedCallable(String cacheName, AffinityTopologyVersion topVer) {
+        public TopologyVersionAwareJob(String cacheName, AffinityTopologyVersion topVer)
{
             this.cacheName = cacheName;
             this.topVer = topVer;
         }
 
+        /** {@inheritDoc} */
+        @Nullable @Override public final Object execute() {
+            waitAffinityReadyFuture();
+
+            localExecute();
+
+            return null;
+        }
+
+        /**
+         * @return Local execution result.
+         */
+        @Nullable protected abstract Object localExecute();
+
         /**
          * Hold (suspend) job execution until our cache version becomes equal to remote cache's
version.
          */
-        public void waitAffinityReadyFuture() {
+        private void waitAffinityReadyFuture() {
             GridCacheProcessor cacheProc = ((IgniteKernal) ignite).context().cache();
 
             GridCacheAdapter<K, V> cacheAdapter = cacheProc.internalCache(cacheName);
@@ -5768,21 +5619,5 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K,
V
                 }
             }
         }
-
-        /** {@inheritDoc} */
-        @Override public void writeExternal(ObjectOutput out) throws IOException {
-            U.writeString(out, cacheName);
-
-            topVer.writeExternal(out);
-        }
-
-        /** {@inheritDoc} */
-        @Override public void readExternal(ObjectInput in) throws IOException, ClassNotFoundException
{
-            cacheName = U.readString(in);
-
-            topVer = new AffinityTopologyVersion();
-
-            topVer.readExternal(in);
-        }
     }
 }

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/863d6bf6/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 661df87..3a685cc 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
@@ -32,6 +32,7 @@ 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.*;
 import org.apache.ignite.transactions.*;
 import org.jetbrains.annotations.*;
 
@@ -230,13 +231,23 @@ public abstract class GridDistributedCacheAdapter<K, V> extends
GridCacheAdapter
      * operation on a cache with the given name.
      */
     @GridInternal
-    private static class GlobalRemoveAllCallable<K,V> extends DelayedCallable<K,
V> implements Callable<Object> {
+    private static class GlobalRemoveAllCallable<K,V> implements Callable<Object>,
Externalizable {
         /** */
         private static final long serialVersionUID = 0L;
 
+        /** Cache name. */
+        private String cacheName;
+
+        /** Topology version. */
+        private AffinityTopologyVersion topVer;
+
         /** Skip store flag. */
         private boolean skipStore;
 
+        /** Injected grid instance. */
+        @IgniteInstanceResource
+        private Ignite ignite;
+
         /**
          * Empty constructor for serialization.
          */
@@ -250,7 +261,8 @@ public abstract class GridDistributedCacheAdapter<K, V> extends
GridCacheAdapter
          * @param skipStore Skip store flag.
          */
         private GlobalRemoveAllCallable(String cacheName, @NotNull AffinityTopologyVersion
topVer, boolean skipStore) {
-            super(cacheName, topVer);
+            this.cacheName = cacheName;
+            this.topVer = topVer;
             this.skipStore = skipStore;
         }
 
@@ -258,11 +270,11 @@ public abstract class GridDistributedCacheAdapter<K, V> extends
GridCacheAdapter
          * {@inheritDoc}
          */
         @Override public Object call() throws Exception {
-            GridCacheAdapter<K, V> cacheAdapter = ((IgniteKernal) ignite).context().cache().internalCache(cacheName);
+            GridCacheAdapter<K, V> cacheAdapter = ((IgniteKernal)ignite).context().cache().internalCache(cacheName);
 
             final GridCacheContext<K, V> ctx = cacheAdapter.context();
 
-            waitAffinityReadyFuture();
+            ctx.affinity().affinityReadyFuture(topVer).get();
 
             ctx.gate().enter();
 
@@ -326,15 +338,15 @@ public abstract class GridDistributedCacheAdapter<K, V> extends
GridCacheAdapter
 
         /** {@inheritDoc} */
         @Override public void writeExternal(ObjectOutput out) throws IOException {
-            super.writeExternal(out);
-
+            U.writeString(out, cacheName);
+            out.writeObject(topVer);
             out.writeBoolean(skipStore);
         }
 
         /** {@inheritDoc} */
         @Override public void readExternal(ObjectInput in) throws IOException, ClassNotFoundException
{
-            super.readExternal(in);
-
+            cacheName = U.readString(in);
+            topVer = (AffinityTopologyVersion)in.readObject();
             skipStore = in.readBoolean();
         }
     }


Mime
View raw message