ignite-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From sboi...@apache.org
Subject [1/5] incubator-ignite git commit: Ignite-176 fixed GridClosureProcessor routing
Date Mon, 09 Feb 2015 14:25:18 GMT
Repository: incubator-ignite
Updated Branches:
  refs/heads/ignite-176 [created] 577fdf5d8


Ignite-176 fixed GridClosureProcessor routing


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

Branch: refs/heads/ignite-176
Commit: 46d4d2608e1fa96081535beda6b4493572895523
Parents: f5901bc
Author: avinogradov <avinogradov@gridgain.com>
Authored: Mon Feb 9 14:04:50 2015 +0300
Committer: avinogradov <avinogradov@gridgain.com>
Committed: Mon Feb 9 14:04:50 2015 +0300

----------------------------------------------------------------------
 .../closure/GridClosureProcessor.java           | 286 ++-----------------
 1 file changed, 18 insertions(+), 268 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/46d4d260/modules/core/src/main/java/org/apache/ignite/internal/processors/closure/GridClosureProcessor.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/closure/GridClosureProcessor.java
b/modules/core/src/main/java/org/apache/ignite/internal/processors/closure/GridClosureProcessor.java
index c5dda9a..0ca6355 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/closure/GridClosureProcessor.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/closure/GridClosureProcessor.java
@@ -416,9 +416,11 @@ public class GridClosureProcessor extends GridProcessorAdapter {
             // In case cache key is passed instead of affinity key.
             final Object affKey0 = ctx.affinity().affinityKey(cacheName, affKey);
 
+            final ClusterNode node = ctx.affinity().mapKeyToNode(affKey0);
+
             ctx.task().setThreadContext(TC_SUBGRID, nodes);
 
-            return ctx.task().execute(new T5<>(cacheName, affKey0, job), null, false);
+            return ctx.task().execute(new T5<>(node, job), null, false);
         }
         catch (IgniteCheckedException e) {
             return ComputeTaskInternalFuture.finishedFuture(ctx, T5.class, e);
@@ -446,9 +448,11 @@ public class GridClosureProcessor extends GridProcessorAdapter {
             // In case cache key is passed instead of affinity key.
             final Object affKey0 = ctx.affinity().affinityKey(cacheName, affKey);
 
+            final ClusterNode node = ctx.affinity().mapKeyToNode(affKey0);
+
             ctx.task().setThreadContext(TC_SUBGRID, nodes);
 
-            return ctx.task().execute(new T4(cacheName, affKey0, job), null, false);
+            return ctx.task().execute(new T4(node, job), null, false);
         }
         catch (IgniteCheckedException e) {
             return ComputeTaskInternalFuture.finishedFuture(ctx, T4.class, e);
@@ -1006,21 +1010,6 @@ public class GridClosureProcessor extends GridProcessorAdapter {
     /**
      * Converts given closure to a grid job.
      *
-     * @param c Closure to convert to grid job.
-     * @param cacheName Cache name.
-     * @param affKey Affinity key.
-     * @return Grid job made out of closure.
-     */
-    private static <R> ComputeJob job(final Callable<R> c, @Nullable final String
cacheName, final Object affKey) {
-        A.notNull(c, "job");
-
-        return c instanceof ComputeJobMasterLeaveAware ? new C3MLA<>(c, cacheName,
affKey) :
-            new C3<>(c, cacheName, affKey);
-    }
-
-    /**
-     * Converts given closure to a grid job.
-     *
      * @param r Closure to convert to grid job.
      * @return Grid job made out of closure.
      */
@@ -1030,20 +1019,6 @@ public class GridClosureProcessor extends GridProcessorAdapter {
        return r instanceof ComputeJobMasterLeaveAware ? new C4MLA(r) : new C4(r);
     }
 
-    /**
-     * Converts given closure to a grid job.
-     *
-     * @param r Closure to convert to grid job.
-     * @param cacheName Cache name.
-     * @param affKey Affinity key.
-     * @return Grid job made out of closure.
-     */
-    private static ComputeJob job(final Runnable r, @Nullable final String cacheName, final
Object affKey) {
-        A.notNull(r, "job");
-
-        return r instanceof ComputeJobMasterLeaveAware ? new C5MLA(r, cacheName, affKey)
: new C5(r, cacheName, affKey);
-    }
-
     /** */
     private class JobMapper {
         /** */
@@ -1232,36 +1207,27 @@ public class GridClosureProcessor extends GridProcessorAdapter {
         private static final long serialVersionUID = 0L;
 
         /** */
-        private final String cacheName;
-
-        /** */
-        private Object affKey;
+        private ClusterNode node;
 
         /** */
         private Runnable job;
 
-        /** */
-        @LoadBalancerResource
-        private ComputeLoadBalancer lb;
-
         /**
-         * @param cacheName Cache name.
-         * @param affKey Affinity key.
+         * @param node Cluster node.
          * @param job Job.
          */
-        private T4(@Nullable String cacheName, Object affKey, Runnable job) {
+        private T4(ClusterNode node, Runnable job) {
             super(U.peerDeployAware0(job));
 
-            this.cacheName = cacheName;
-            this.affKey = affKey;
+            this.node = node;
             this.job = job;
         }
 
         /** {@inheritDoc} */
         @Override public Map<? extends ComputeJob, ClusterNode> map(List<ClusterNode>
subgrid, @Nullable Void arg) {
-            ComputeJob job = job(this.job, cacheName, affKey);
+            ComputeJob job = job(this.job);
 
-            return Collections.singletonMap(job, lb.getBalancedNode(job, null));
+            return Collections.singletonMap(job, node);
         }
     }
 
@@ -1272,36 +1238,27 @@ public class GridClosureProcessor extends GridProcessorAdapter {
         private static final long serialVersionUID = 0L;
 
         /** */
-        private final String cacheName;
-
-        /** */
-        private Object affKey;
+        private ClusterNode node;
 
         /** */
         private Callable<R> job;
 
-        /** */
-        @LoadBalancerResource
-        private ComputeLoadBalancer lb;
-
         /**
-         * @param cacheName Cache name.
-         * @param affKey Affinity key.
+         * @param node Cluster node.
          * @param job Job.
          */
-        private T5(@Nullable String cacheName, Object affKey, Callable<R> job) {
+        private T5(ClusterNode node, Callable<R> job) {
             super(U.peerDeployAware0(job));
 
-            this.cacheName = cacheName;
-            this.affKey = affKey;
+            this.node = node;
             this.job = job;
         }
 
         /** {@inheritDoc} */
         @Override public Map<? extends ComputeJob, ClusterNode> map(List<ClusterNode>
subgrid, @Nullable Void arg) {
-            ComputeJob job = job(this.job, cacheName, affKey);
+            ComputeJob job = job(this.job);
 
-            return Collections.singletonMap(job, lb.getBalancedNode(job, null));
+            return Collections.singletonMap(job, node);
         }
 
         /** {@inheritDoc} */
@@ -1792,112 +1749,6 @@ public class GridClosureProcessor extends GridProcessorAdapter {
     }
 
     /**
-     *
-     */
-    private static class C3<R> implements ComputeJob, Externalizable {
-        /** */
-        private static final long serialVersionUID = 0L;
-
-        /** */
-        @CacheName
-        private String cn;
-
-        /** */
-        @CacheAffinityKeyMapped
-        private Object ak;
-
-
-        /** */
-        protected Callable<R> c;
-
-        /**
-         *
-         */
-        public C3(){
-            // No-op.
-        }
-
-        /**
-         * @param c Callable.
-         * @param cacheName Cache name.
-         * @param affKey Affinity key.
-         */
-        public C3(Callable<R> c, @Nullable String cacheName, Object affKey) {
-            this.cn = cacheName;
-            this.ak = affKey;
-            this.c = c;
-        }
-
-        /** {@inheritDoc} */
-        @Override public Object execute() {
-            try {
-                return c.call();
-            }
-            catch (Exception e) {
-                throw new IgniteException(e);
-            }
-        }
-
-        /** {@inheritDoc} */
-        @Override public void cancel() {
-            // No-op.
-        }
-
-        /** {@inheritDoc} */
-        @Override public void writeExternal(ObjectOutput out) throws IOException {
-            out.writeObject(cn);
-            out.writeObject(ak);
-            out.writeObject(c);
-        }
-
-        /** {@inheritDoc} */
-        @Override public void readExternal(ObjectInput in) throws IOException, ClassNotFoundException
{
-            cn = (String)in.readObject();
-            ak = in.readObject();
-            c = (Callable<R>)in.readObject();
-        }
-
-        /** {@inheritDoc} */
-        @Override public String toString() {
-            return S.toString(C3.class, this);
-        }
-    }
-
-    /**
-     *
-     */
-    private static class C3MLA<R> extends C3<R> implements ComputeJobMasterLeaveAware{
-        /** */
-        private static final long serialVersionUID = 0L;
-
-        /**
-         *
-         */
-        public C3MLA() {
-            super();
-        }
-
-        /**
-         * @param c Callable.
-         * @param cacheName Cache name.
-         * @param affKey Affinity key.
-         */
-        public C3MLA(Callable<R> c, @Nullable String cacheName, Object affKey) {
-            super(c, cacheName, affKey);
-        }
-
-        /** {@inheritDoc} */
-        @Override public void onMasterNodeLeft(ComputeTaskSession ses) {
-            ((ComputeJobMasterLeaveAware)c).onMasterNodeLeft(ses);
-        }
-
-        /** {@inheritDoc} */
-        @Override public String toString() {
-            return S.toString(C3MLA.class, this, super.toString());
-        }
-    }
-
-    /**
      */
     private static class C4 implements ComputeJob, Externalizable {
         /** */
@@ -1979,105 +1830,4 @@ public class GridClosureProcessor extends GridProcessorAdapter {
             return S.toString(C4MLA.class, this, super.toString());
         }
     }
-
-    /**
-     */
-    private static class C5 implements ComputeJob, Externalizable {
-        /** */
-        private static final long serialVersionUID = 0L;
-
-        /** */
-        @CacheName
-        private String cn;
-
-        /** */
-        @CacheAffinityKeyMapped
-        private Object ak;
-
-        /** */
-        protected Runnable r;
-
-        /**
-         *
-         */
-        public C5(){
-            // No-op.
-        }
-
-        /**
-         * @param r Runnable.
-         * @param cacheName Cache name.
-         * @param affKey Affinity key.
-         */
-        public C5(Runnable r, @Nullable String cacheName, Object affKey) {
-            this.cn = cacheName;
-            this.ak = affKey;
-            this.r = r;
-        }
-
-        /** {@inheritDoc} */
-        @Nullable @Override public Object execute() {
-            r.run();
-
-            return null;
-        }
-
-        /** {@inheritDoc} */
-        @Override public void cancel() {
-            // No-op.
-        }
-
-        /** {@inheritDoc} */
-        @Override public void writeExternal(ObjectOutput out) throws IOException {
-            out.writeObject(cn);
-            out.writeObject(ak);
-            out.writeObject(r);
-        }
-
-        /** {@inheritDoc} */
-        @Override public void readExternal(ObjectInput in) throws IOException, ClassNotFoundException
{
-            cn = (String)in.readObject();
-            ak = in.readObject();
-            r = (Runnable)in.readObject();
-        }
-
-        /** {@inheritDoc} */
-        @Override public String toString() {
-            return S.toString(C5.class, this, super.toString());
-        }
-    }
-
-    /**
-     *
-     */
-    private static class C5MLA extends C5 implements ComputeJobMasterLeaveAware{
-        /** */
-        private static final long serialVersionUID = 0L;
-
-        /**
-         *
-         */
-        public C5MLA() {
-            super();
-        }
-
-        /**
-         * @param r Runnable.
-         * @param cacheName Cache name.
-         * @param affKey Affinity key.
-         */
-        public C5MLA(Runnable r, @Nullable String cacheName, Object affKey) {
-            super(r, cacheName, affKey);
-        }
-
-        /** {@inheritDoc} */
-        @Override public void onMasterNodeLeft(ComputeTaskSession ses) {
-            ((ComputeJobMasterLeaveAware)r).onMasterNodeLeft(ses);
-        }
-
-        /** {@inheritDoc} */
-        @Override public String toString() {
-            return S.toString(C5MLA.class, this, super.toString());
-        }
-    }
 }


Mime
View raw message