ignite-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From sboi...@apache.org
Subject [01/20] incubator-ignite git commit: #ignite-965: map and reduce are executed on server.
Date Tue, 23 Jun 2015 17:47:27 GMT
Repository: incubator-ignite
Updated Branches:
  refs/heads/ignite-965 01c9c3acb -> 47276a81e


#ignite-965: map and reduce are executed on server.


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

Branch: refs/heads/ignite-965
Commit: d481ea8c76f9de8e0e94e6a2ee7a46d3788c1442
Parents: 01c9c3a
Author: ivasilinets <ivasilinets@gridgain.com>
Authored: Tue Jun 23 18:08:27 2015 +0300
Committer: ivasilinets <ivasilinets@gridgain.com>
Committed: Tue Jun 23 18:08:27 2015 +0300

----------------------------------------------------------------------
 .../IgniteComputeTaskCommandHandler.java        | 102 ++++++++++++++-----
 .../rest/request/RestComputeTaskRequest.java    |  38 +++++--
 .../scripting/IgniteScriptProcessor.java        |  22 +++-
 modules/nodejs/src/main/js/apache-ignite.js     |   3 +-
 modules/nodejs/src/main/js/compute-task.js      |  38 -------
 modules/nodejs/src/main/js/compute.js           |  75 ++------------
 modules/nodejs/src/test/js/test-compute.js      |  37 ++++---
 modules/nodejs/src/test/js/test-utils.js        |  10 +-
 .../http/jetty/GridJettyRestHandler.java        |  17 +---
 9 files changed, 161 insertions(+), 181 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/d481ea8c/modules/core/src/main/java/org/apache/ignite/internal/processors/rest/handlers/compute/IgniteComputeTaskCommandHandler.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/rest/handlers/compute/IgniteComputeTaskCommandHandler.java
b/modules/core/src/main/java/org/apache/ignite/internal/processors/rest/handlers/compute/IgniteComputeTaskCommandHandler.java
index a3708ab..7329a5b 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/rest/handlers/compute/IgniteComputeTaskCommandHandler.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/rest/handlers/compute/IgniteComputeTaskCommandHandler.java
@@ -25,12 +25,11 @@ import org.apache.ignite.internal.processors.rest.*;
 import org.apache.ignite.internal.processors.rest.handlers.*;
 import org.apache.ignite.internal.processors.rest.request.*;
 import org.apache.ignite.internal.util.future.*;
-import org.apache.ignite.internal.util.typedef.T3;
 import org.apache.ignite.internal.util.typedef.internal.*;
 import org.apache.ignite.resources.*;
 import org.jetbrains.annotations.*;
 
-import javax.script.ScriptException;
+import javax.script.*;
 import java.util.*;
 
 import static org.apache.ignite.internal.processors.rest.GridRestCommand.*;
@@ -64,25 +63,28 @@ public class IgniteComputeTaskCommandHandler extends GridRestCommandHandlerAdapt
 
         final RestComputeTaskRequest req0 = (RestComputeTaskRequest) req;
 
-        List<T3<String, String, String>> mapping =  req0.mapping();
-
-        Object res = ctx.grid().compute().execute(new JsTask(mapping, ctx), null);
+        Object res = ctx.grid().compute().execute(new JsTask(req0.mapFunc(), req0.argument(),
req0.reduceFunc(), ctx), null);
 
         return new GridFinishedFuture<>(new GridRestResponse(res));
     }
 
     private static class JsTask extends ComputeTaskAdapter<String, Object> {
-        /** Mapping. */
-        private List<T3<String, String, String>> mapping;
+        /** Mapping function. */
+        private String mapFunc;
+
+        private String reduceFunc;
 
         /** Grid kernal context. */
         private GridKernalContext ctx;
 
+        private String arg;
+
         /**
-         * @param mapping Task mapping.
          */
-        public JsTask(List<T3<String, String, String>> mapping, GridKernalContext
ctx) {
-            this.mapping = mapping;
+        public JsTask(String mapFunc, String arg, String reduceFunc, GridKernalContext ctx)
{
+            this.mapFunc = mapFunc;
+            this.reduceFunc = reduceFunc;
+            this.arg = arg;
             this.ctx = ctx;
         }
 
@@ -90,27 +92,68 @@ public class IgniteComputeTaskCommandHandler extends GridRestCommandHandlerAdapt
         @Override public Map<? extends ComputeJob, ClusterNode> map(List<ClusterNode>
nodes, String arg) {
             Map<ComputeJob, ClusterNode> map = new HashMap<>();
 
-            for (final T3<String, String, String> job : mapping) {
-                UUID nodeId = UUID.fromString(job.get2());
+            String nodesIds = "[";
 
-                ClusterNode node = ctx.grid().cluster().node(nodeId);
+            for (ClusterNode node : nodes)
+                nodesIds += "\""  + node.id().toString() + "\"" + ",";
 
-                map.put(new ComputeJobAdapter() {
-                    /** Ignite. */
-                    @IgniteInstanceResource
-                    private Ignite ignite;
+            nodesIds = nodesIds.substring(0, nodesIds.length() - 1) + "]";
 
-                    @Override public Object execute() throws IgniteException {
-                        System.out.println("Compute job on node " + ignite.cluster().localNode().id());
+            try {
+                String newMap = new String("function () {\n" +
+                    "   var res = [];\n" +
+                    "   var resCont = function(f, args, nodeId) {\n" +
+                    "       res.push([f.toString(), args, nodeId])\n" +
+                    "   }\n" +
+                    "   var locF = " + mapFunc + "; \n locF(" +
+                        nodesIds + ", " +
+                    "\"" + this.arg + "\"" +
+                    ", resCont.bind(null)" + ");\n" +
+                    "   return res;\n" +
+                    "}");
 
-                        try {
-                            return ((IgniteKernal)ignite).context().scripting().runJS(job.get1(),
job.get3());
-                        }
-                        catch (ScriptException e) {
-                            throw new IgniteException(e);
+                List mapRes = (List)ctx.scripting().runJS(newMap);
+
+                for (Object arr : mapRes) {
+                    Object[] nodeTask = ((List)arr).toArray();
+
+                    final String func = (String)nodeTask[0];
+
+                    final List argv = (List) nodeTask[1];
+
+                    String nodeIdStr = (String) nodeTask[2];
+
+                    UUID nodeId = UUID.fromString(nodeIdStr);
+
+                    ClusterNode node = ctx.grid().cluster().node(nodeId);
+
+                    map.put(new ComputeJobAdapter() {
+                        /** Ignite. */
+                        @IgniteInstanceResource
+                        private Ignite ignite;
+
+                        @Override public Object execute() throws IgniteException {
+                            System.out.println("Compute job on node " + ignite.cluster().localNode().id());
+                            try {
+                                String[] argv1 = new String[argv.size()];
+
+                                for (int i = 0; i < argv1.length; ++i)
+                                    argv1[i] = "\"" + argv.get(i).toString() + "\"";
+
+                                return ctx.scripting().runJS(func, argv1);
+                            }
+                            catch (Exception e) {
+                                throw new IgniteException(e);
+                            }
                         }
-                    }
-                }, node);
+                    }, node);
+
+                }
+            }
+            catch (ScriptException e) {
+                throw new IgniteException(e);
+            }
+            finally {
             }
 
             return map;
@@ -123,7 +166,12 @@ public class IgniteComputeTaskCommandHandler extends GridRestCommandHandlerAdapt
             for (ComputeJobResult res : results)
                 data.add(res.getData());
 
-            return data;
+            try {
+                return ctx.scripting().runJS(reduceFunc, new String[] {data.toString()});
+            }
+            catch (ScriptException e) {
+                throw new IgniteException(e);
+            }
         }
     }
 }

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/d481ea8c/modules/core/src/main/java/org/apache/ignite/internal/processors/rest/request/RestComputeTaskRequest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/rest/request/RestComputeTaskRequest.java
b/modules/core/src/main/java/org/apache/ignite/internal/processors/rest/request/RestComputeTaskRequest.java
index bae7ce5..89e04da 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/rest/request/RestComputeTaskRequest.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/rest/request/RestComputeTaskRequest.java
@@ -26,19 +26,35 @@ import java.util.*;
  */
 public class RestComputeTaskRequest extends GridRestRequest {
     /** Mapping tasks to nodes. */
-    private List<T3<String, String, String>> mapping;
+    private String mapFunc;
 
-    /**
-     * @param mapping Mapping tasks to nodes.
-     */
-    public void mapping(List<T3<String, String, String>> mapping) {
-        this.mapping = mapping;
+    /** Function argument. */
+    private String arg;
+
+    private String reduceFunc;
+
+    public void reduceFunc(String reduceFunc) {
+        this.reduceFunc = reduceFunc;
+    }
+
+    public String reduceFunc() {
+
+        return reduceFunc;
+    }
+
+    public String mapFunc() {
+        return mapFunc;
+    }
+
+    public String argument() {
+        return arg;
+    }
+
+    public void mapFunc(String mapFunc) {
+        this.mapFunc = mapFunc;
     }
 
-    /**
-     * @return Mapping tasks to nodes.
-     */
-    public List<T3<String, String, String>> mapping() {
-        return mapping;
+    public void argument(String arg) {
+        this.arg = arg;
     }
 }

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/d481ea8c/modules/core/src/main/java/org/apache/ignite/internal/processors/scripting/IgniteScriptProcessor.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/scripting/IgniteScriptProcessor.java
b/modules/core/src/main/java/org/apache/ignite/internal/processors/scripting/IgniteScriptProcessor.java
index 54004de..99c206e 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/scripting/IgniteScriptProcessor.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/scripting/IgniteScriptProcessor.java
@@ -20,8 +20,10 @@ package org.apache.ignite.internal.processors.scripting;
 import org.apache.ignite.*;
 import org.apache.ignite.internal.*;
 import org.apache.ignite.internal.processors.*;
+import org.apache.ignite.internal.util.typedef.*;
 
 import javax.script.*;
+import java.util.*;
 
 /**
  * Ignite scripting manager.
@@ -46,7 +48,7 @@ public class IgniteScriptProcessor extends GridProcessorAdapter {
      * @throws ScriptException If script failed.
      */
     public Object run(String engName, String script) throws ScriptException {
-        if (engName.equals(JAVA_SCRIPT_ENGINE_NAME))
+        if (!engName.equals(JAVA_SCRIPT_ENGINE_NAME))
             throw new IgniteException("Engine is not supported. [engName=" + engName + "]");
 
         return runJS(script);
@@ -54,9 +56,11 @@ public class IgniteScriptProcessor extends GridProcessorAdapter {
 
     /**
      * @param script Script.
+     * @param args Arguments.
+     * @return Script result.
      * @throws ScriptException If script failed.
      */
-    public Object runJS(String script, String args) throws ScriptException {
+    public Object runJS(String script, String[] args) throws ScriptException {
         ScriptEngine engine = factory.getEngineByName("JavaScript");
 
         Bindings b = engine.createBindings();
@@ -65,13 +69,23 @@ public class IgniteScriptProcessor extends GridProcessorAdapter {
 
         engine.setBindings(b, ScriptContext.ENGINE_SCOPE);
 
-        script = "(" + script + ")(" + args + ");";
+        script = "(" + script + ")(" ;
+
+        for (int i = 0; i < args.length; ++i)
+            script += args[i] + (i < args.length - 1 ? "," : "");
+
+        script += ");";
 
         return engine.eval(script);
     }
 
+    /**
+     * @param script Script.
+     * @return Script result.
+     * @throws ScriptException If script failed.
+     */
     public Object runJS(String script) throws ScriptException {
-        return runJS(script, "");
+        return runJS(script, new String[]{""});
     }
 
     /**

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/d481ea8c/modules/nodejs/src/main/js/apache-ignite.js
----------------------------------------------------------------------
diff --git a/modules/nodejs/src/main/js/apache-ignite.js b/modules/nodejs/src/main/js/apache-ignite.js
index f90cf68..0df6160 100644
--- a/modules/nodejs/src/main/js/apache-ignite.js
+++ b/modules/nodejs/src/main/js/apache-ignite.js
@@ -20,6 +20,5 @@ module.exports = {
   Ignition : require('./ignition.js').Ignition,
   Server : require('./server.js').Server,
   Ignite : require('./ignite.js').Ignite,
-  Compute : require('./compute.js').Compute,
-  ComputeJob: require('./compute.js').ComputeJob
+  Compute : require('./compute.js').Compute
 }
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/d481ea8c/modules/nodejs/src/main/js/compute-task.js
----------------------------------------------------------------------
diff --git a/modules/nodejs/src/main/js/compute-task.js b/modules/nodejs/src/main/js/compute-task.js
deleted file mode 100644
index d13f361..0000000
--- a/modules/nodejs/src/main/js/compute-task.js
+++ /dev/null
@@ -1,38 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements.  See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License.  You may obtain a copy of the License at
- *
- *      http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-/**
- * @constructor
- * @this {ComputeTask}
- */
-function ComputeTask() {
-}
-
-/**
- * @param {string[]} nodes Nodes id
- * @param {string} arg Argument
- * @returns {Object.<string, Cache~onGet>} Map of grid jobs assigned to subgrid node.
Unless {@link ComputeTaskContinuousMapper} is
- * injected into task, if {@code null} or empty map is returned, exception will be thrown
- */
-ComputeTask.prototype.map = function(nodes, arg) {
-}
-
-/**
- * @param {string[]} results Results
- */
-ComputeTask.prototype.reduce = function(results) {
-}
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/d481ea8c/modules/nodejs/src/main/js/compute.js
----------------------------------------------------------------------
diff --git a/modules/nodejs/src/main/js/compute.js b/modules/nodejs/src/main/js/compute.js
index fb38e7a..7a5f808 100644
--- a/modules/nodejs/src/main/js/compute.js
+++ b/modules/nodejs/src/main/js/compute.js
@@ -16,7 +16,6 @@
  */
 
 var Server = require("./server").Server;
-var ComputeTask = require("./compute-task").ComputeTask;
 
 /**
  * @constructor
@@ -72,76 +71,14 @@ Compute.prototype._escape = function(f) {
  * @param {string} arg  Argument
  * @param {} callback Callback
  */
-Compute.prototype.execute = function(task, arg, callback) {
-  this._nodes(this._onNodesExecute.bind(this, task, arg, callback));
-}
-
-Compute.prototype._nodes = function(callback) {
-  this._server.runCommand("top", [Server.pair("mtr", "false"), Server.pair("attr", "false")],
-    this._onNodes.bind(this, callback))
-}
-
-Compute.prototype._onNodes = function(callback, error, results) {
-  if (error) {
-    callback.call(null, error, null);
-
-    return;
-  }
-
-  var nodes = [];
-
-  for (var res of results) {
-    nodes.push(res["nodeId"])
-  }
-
-  callback.call(null, null, nodes);
-}
-
-Compute.prototype._onNodesExecute = function(task, arg, callback, err, nodes) {
-  if (err) {
-      callback.call(null, error, null);
-
-      return;
-  }
-
-  var computeJobList = task.map(nodes, arg);
+Compute.prototype.execute = function(map, reduce, arg, callback) {
+   var params = [];
 
-  var params = [];
-  var i = 1;
+    params.push(Server.pair("map", this._escape(map)));
+    params.push(Server.pair("reduce", this._escape(reduce)));
+    params.push(Server.pair("arg", this._escape(arg)));
 
-  console.log("TASK" + computeJobList);
-  for (var job of computeJobList) {
-    params.push(Server.pair("f" + i, this._escape(job.func)));
-    params.push(Server.pair("args" + i,  JSON.stringify(job.args)));
-    params.push(Server.pair("n" + i, job.node));
-    i++;
-  }
-
-  this._server.runCommand("exectask", params, this._onResExecute.bind(this, task, callback));
-}
-
-
-Compute.prototype._onResExecute = function(task, callback, err, results) {
-  if (err) {
-    callback.call(null, err, null);
-
-    return;
-  }
-
-  console.log("ON RES EXEC = " + results);
-
-  var res = task.reduce(results);
-
-  callback.call(null, null, res);
+    this._server.runCommand("exectask", params, callback);
 }
 
 exports.Compute = Compute
-
-
-function ComputeJob(func, args, node) {
-    this.func = func;
-    this.args = args;
-    this.node = node;
-}
-
-exports.ComputeJob = ComputeJob;
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/d481ea8c/modules/nodejs/src/test/js/test-compute.js
----------------------------------------------------------------------
diff --git a/modules/nodejs/src/test/js/test-compute.js b/modules/nodejs/src/test/js/test-compute.js
index 87a284c..c444380 100644
--- a/modules/nodejs/src/test/js/test-compute.js
+++ b/modules/nodejs/src/test/js/test-compute.js
@@ -17,10 +17,6 @@
 
 var TestUtils = require("./test-utils").TestUtils;
 
-var Apache = require(TestUtils.scriptPath());
-var Cache = Apache.Cache;
-var Server = Apache.Server;
-
 var assert = require("assert");
 
 testComputeAffinityRun = function() {
@@ -32,11 +28,7 @@ testComputeAffinityCall = function() {
 }
 
 testComputeExecute = function() {
-  var CharacterCountTask = require("./simple-compute-task").CharacterCountTask
-
-  var task = new CharacterCountTask();
-
-  TestUtils.startIgniteNode(onStart1.bind(null, task));
+  TestUtils.startIgniteNode(onStart1);
 }
 
 function onStart(locOnPut, error, ignite) {
@@ -92,10 +84,31 @@ function onError1(error, res) {
   TestUtils.testDone();
 }
 
-function onStart1(task, error, ignite) {
-  var comp = ignite.compute();
+function onStart1(error, ignite) {
+  var map = function(nodes, arg, emit) {
+    var words = arg.split(" ");
+
+    for (var i = 0; i < words.length; i++) {
+      var f = function (word) {
+        println(">>> Printing " + word);
+
+        return word.length;
+      };
+
+      emit(f, [words[i]], nodes[i %  nodes.length]);
+    }
+  };
+
+  var reduce = function(results) {
+    var sum = 0;
+
+    for (var i = 0; i < results.length; ++i)
+     sum += parseInt(results[i], 10);
+
+    return sum;
+  };
 
-  comp.execute(task, "Hi Alice", onComputeResult);
+  ignite.compute().execute(map, reduce, "Hi Alice", onComputeResult);
 }
 
 function onComputeResult(error, res) {

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/d481ea8c/modules/nodejs/src/test/js/test-utils.js
----------------------------------------------------------------------
diff --git a/modules/nodejs/src/test/js/test-utils.js b/modules/nodejs/src/test/js/test-utils.js
index 135da8c..ee83d7f 100644
--- a/modules/nodejs/src/test/js/test-utils.js
+++ b/modules/nodejs/src/test/js/test-utils.js
@@ -129,8 +129,9 @@ TestUtils.testDone = function() {
  * @param {Ignition~onStart} callback Called on connect
  */
 TestUtils.startIgniteNode = function(callback) {
-  var Apache = require(TestUtils.scriptPath());
-  var Ignition = Apache.Ignition;
+  var Ignite = require(TestUtils.scriptPath());
+  var Ignition = Ignite.Ignition;
+
   Ignition.start(['127.0.0.1:9095'], null, callback);
 }
 
@@ -141,8 +142,9 @@ TestUtils.startIgniteNode = function(callback) {
  * @param {Ignition~onStart} callback Called on connect
  */
 TestUtils.startIgniteNodeWithKey = function(secretKey, callback) {
-  var Apache = require(TestUtils.scriptPath());
-  var Ignition = Apache.Ignition;
+  var Ignite = require(TestUtils.scriptPath());
+  var Ignition = Ignite.Ignition;
+
   Ignition.start(['127.0.0.1:9095'], secretKey, callback);
 }
 

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/d481ea8c/modules/rest-http/src/main/java/org/apache/ignite/internal/processors/rest/protocols/http/jetty/GridJettyRestHandler.java
----------------------------------------------------------------------
diff --git a/modules/rest-http/src/main/java/org/apache/ignite/internal/processors/rest/protocols/http/jetty/GridJettyRestHandler.java
b/modules/rest-http/src/main/java/org/apache/ignite/internal/processors/rest/protocols/http/jetty/GridJettyRestHandler.java
index 5fe4cd9..7d71530 100644
--- a/modules/rest-http/src/main/java/org/apache/ignite/internal/processors/rest/protocols/http/jetty/GridJettyRestHandler.java
+++ b/modules/rest-http/src/main/java/org/apache/ignite/internal/processors/rest/protocols/http/jetty/GridJettyRestHandler.java
@@ -462,20 +462,9 @@ public class GridJettyRestHandler extends AbstractHandler {
             case EXECUTE_TASK: {
                 RestComputeTaskRequest restReq0 = new RestComputeTaskRequest();
 
-                List<Object> funcs = values("f", params);
-                List<Object> nodes = values("n", params);
-
-                List<Object> args = values("args", params);
-
-                assert funcs.size() == nodes.size();
-
-                List<T3<String, String, String>> mapping = new ArrayList<>();
-
-
-                for (int i = 0; i < funcs.size(); ++i)
-                    mapping.add(new T3((String) funcs.get(i), (String)nodes.get(i), (String)args.get(i)));
-
-                restReq0.mapping(mapping);
+                restReq0.mapFunc((String)params.get("map"));
+                restReq0.argument((String) params.get("arg"));
+                restReq0.reduceFunc((String) params.get("reduce"));
 
                 restReq = restReq0;
 


Mime
View raw message