You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@reef.apache.org by we...@apache.org on 2014/10/23 02:03:50 UTC
[19/51] [abbrv] [partial] Initial merge of Wake,
Tang and REEF into one repository and project
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/50444bba/reef-examples/index2.html
----------------------------------------------------------------------
diff --git a/reef-examples/index2.html b/reef-examples/index2.html
deleted file mode 100644
index 0c640de..0000000
--- a/reef-examples/index2.html
+++ /dev/null
@@ -1,212 +0,0 @@
-<!--
-
- Copyright (C) 2014 Microsoft Corporation
-
- Licensed 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.
-
--->
-<!--
-
- Copyright (C) 2013 Microsoft Corporation
-
- Licensed 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.
-
--->
-<!DOCTYPE html>
-
-<html lang="en" xmlns="http://www.w3.org/1999/xhtml">
-<head>
- <meta charset="utf-8">
- <title>Wake profile</title>
- <script type="text/javascript" src="d3.v2.js"></script>
- <script type="text/javascript" src="reef-sankey.js"></script>
- <!-- <script type="text/javascript" src="http://d3js.org/d3.v2.min.js?2.9.1"></script>-->
- <!-- http://mbostock.github.com/d3/d3.js?2.6.0"></script>
- <script type="text/javascript" src="http://mbostock.github.com/d3/d3.layout.js?2.6.0"></script>
- <script type="text/javascript" src="http://mbostock.github.com/d3/d3.geom.js?2.6.0"></script> -->
- <!-- <script type="text/javascript" src="./sankey.js"></script>
- chart {
- width: 1900px;
- }
- -->
- <!--.chart {
- width: 100%;
- background: url(wateradasd.jpg);
- background-size: 160%;
- font-stretch: extra-condensed;
- color: white;
- }-->
- <style>
-
- body {
-
- overflow: hidden;
- margin: 10px;
- font-size: 10pt;
- font-family: "Segoe UI", "Comic Sans MS", Verdana, "Helvetica Neue", Helvetica;
- font-weight: 200;
- }
-
- .node rect {
- cursor: move;
- fill-opacity: .9;
- shape-rendering: crispEdges;
- }
-
- .node text {
- pointer-events: none;
- }
-
- .link {
- fill: none;
- stroke: #0066dd;
- stroke-opacity: .2;
- }
-
- .back-link {
- fill: none;
- stroke: #990;
- stroke-opacity: .2;
- }
-
-
- .link:hover {
- stroke-opacity: .5;
- }
-
- .back-link:hover {
- stroke-opacity: .5;
- }
-
- </style>
-
-</head>
-<body>
-
-
-<p id="chart" class="chart"></p>
-
-
-<script type="text/javascript" charset="utf-8">
- var mycolors = [
- "#ffffff",
- "#eeeeee",
- "#dddddd",
- "#cccccc",
- "#bbbbbb",
- "#aaaaaa",
- "#999999",
- "#888888",
- "#777777",
- "#666666",
- "#555555",
- "#444444",
- "#333333",
- "#222222",
- "#111111",
- "#000000"];
-
- var margin = { top: 10, right: 10, bottom: 10, left: 10 },
- width = 1200- margin.left - margin.right,
- height = 650 - margin.top - margin.bottom;
-
- var formatNumber = d3.format(",.0f"),
- format = function (d) { return formatNumber(d); },
- color = d3.scale.category20();
-
- var svg = d3.select("#chart").append("svg")
- .attr("width", width + margin.left + margin.right)
- .attr("height", height + margin.top + margin.bottom)
- .append("g")
- .attr("transform", "translate(" + margin.left + "," + margin.top + ")");
-
- var sankey = d3.sankey()
- .nodeWidth(15)
- .nodePadding(20)
- .size([width, height]);
-
- var path = sankey.link();
- //d3.json("profile-BGD.json", function (energy) {
- d3.json("profile-Node-2.json", function (energy) {
- // d3.json("profile-Node-1.json", function (energy) {
- // d3.json("profile-distributed-shell-1377046500540.json", function (energy) {
-
- sankey
- .nodes(energy.nodes)
- .links(energy.links)
- .layout(128);
-
- var link = svg.append("g").selectAll(".link")
- .data(energy.links)
- .enter().append("path")
- .attr("class", function (d) { return d.back ? "back-link" : "link" })
- .attr("d", path)
- .style("stroke-width", function (d) { return Math.max(1, d.dy); })
- .sort(function (a, b) { return b.dy - a.dy; });
-
- link.append("title")
- .text(function (d) { return d.source.name + " → " + d.target.name; });// + "\n" + format(d.value); });
-
- var node = svg.append("g").selectAll(".node")
- .data(energy.nodes)
- .enter().append("g")
- .attr("class", "node")
- .attr("transform", function (d) { return "translate(" + d.x + "," + d.y + ")"; })
- .call(d3.behavior.drag()
- .origin(function (d) { return d; })
- .on("dragstart", function () { this.parentNode.appendChild(this); })
- .on("drag", dragmove));
-
- node.append("rect")
- .attr("height", function (d) { return d.dy; })
- .attr("width", sankey.nodeWidth())
- .style("fill", function (d) { if (d.count) { return d.color = mycolors[Math.round(Math.log(d.latency)+1)] } else {
- return mycolors[0]; } })//d.name.replace(/ .*/, "")); })
- .style("stroke", function (d) { return d3.rgb(d.color).darker(2); })
- .append("title")
- .text(function (d) { if (d.latency) { return "event count " + format(d.count) + " mean latency " + format(d.latency)
- + "ms"; } else { return d.name; }});
-
- node.append("text")
- .attr("x", -6)
- .attr("y", function (d) { return d.dy / 2; })
- .attr("dy", ".35em")
- .attr("text-anchor", "end")
- .attr("transform", null)
- .text(function (d) { return d.name; })
- .filter(function (d) { return d.x < width / 2; })
- .attr("x", 6 + sankey.nodeWidth())
- .attr("text-anchor", "start");
-
- function dragmove(d) {
- d3.select(this).attr("transform", "translate(" + d.x + "," + (d.y = Math.max(0, Math.min(height - d.dy,
- d3.event.y))) + ")");
- sankey.relayout();
- link.attr("d", path);
- }
- });
-
-
-</script>
-</body>
-</html>
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/50444bba/reef-examples/pom.xml
----------------------------------------------------------------------
diff --git a/reef-examples/pom.xml b/reef-examples/pom.xml
index ffce683..2375020 100644
--- a/reef-examples/pom.xml
+++ b/reef-examples/pom.xml
@@ -1,11 +1,12 @@
<?xml version="1.0"?>
-<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+<project xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns="http://maven.apache.org/POM/4.0.0"
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<artifactId>reef-examples</artifactId>
<name>REEF Examples</name>
<parent>
- <groupId>com.microsoft.reef</groupId>
+ <groupId>org.apache.reef</groupId>
<artifactId>reef-project</artifactId>
<version>0.10-SNAPSHOT</version>
</parent>
@@ -126,13 +127,13 @@
<executable>java</executable>
<arguments>
<argument>-classpath</argument>
- <classpath />
- <argument>-Djava.util.logging.config.class=com.microsoft.reef.util.logging.Config
+ <classpath/>
+ <argument>-Djava.util.logging.config.class=org.apache.reef.util.logging.Config
</argument>
<!-- <argument>-Dlog4j.debug=true</argument> -->
<argument>-Dcom.microsoft.reef.runtime.local.folder=${project.build.directory}
</argument>
- <argument>com.microsoft.reef.examples.hello.HelloREEF</argument>
+ <argument>org.apache.reef.examples.hello.HelloREEF</argument>
</arguments>
</configuration>
</plugin>
@@ -151,13 +152,13 @@
<executable>java</executable>
<arguments>
<argument>-classpath</argument>
- <classpath />
- <argument>-Djava.util.logging.config.class=com.microsoft.reef.util.logging.Config
+ <classpath/>
+ <argument>-Djava.util.logging.config.class=org.apache.reef.util.logging.Config
</argument>
<!-- <argument>-Dlog4j.debug=true</argument> -->
<argument>-Dcom.microsoft.reef.runtime.local.folder=${project.build.directory}
</argument>
- <argument>com.microsoft.reef.examples.hellohttp.HelloREEFHttp</argument>
+ <argument>org.apache.reef.examples.hellohttp.HelloREEFHttp</argument>
</arguments>
</configuration>
</plugin>
@@ -176,13 +177,13 @@
<executable>java</executable>
<arguments>
<argument>-classpath</argument>
- <classpath />
- <argument>-Djava.util.logging.config.class=com.microsoft.reef.util.logging.Config
+ <classpath/>
+ <argument>-Djava.util.logging.config.class=org.apache.reef.util.logging.Config
</argument>
<!-- <argument>-Dlog4j.debug=true</argument> -->
<argument>-Dcom.microsoft.reef.runtime.local.folder=${project.build.directory}
</argument>
- <argument>com.microsoft.reef.examples.hello.HelloREEFNoClient</argument>
+ <argument>org.apache.reef.examples.hello.HelloREEFNoClient</argument>
</arguments>
</configuration>
</plugin>
@@ -201,12 +202,12 @@
<executable>java</executable>
<arguments>
<argument>-classpath</argument>
- <classpath />
- <argument>-Djava.util.logging.config.class=com.microsoft.reef.util.logging.Config
+ <classpath/>
+ <argument>-Djava.util.logging.config.class=org.apache.reef.util.logging.Config
</argument>
<argument>-Dcom.microsoft.reef.runtime.local.folder=${project.build.directory}
</argument>
- <argument>com.microsoft.reef.examples.groupcomm.matmul.MatMultREEF</argument>
+ <argument>org.apache.reef.examples.groupcomm.matmul.MatMultREEF</argument>
</arguments>
</configuration>
</plugin>
@@ -226,12 +227,12 @@
<executable>java</executable>
<arguments>
<argument>-classpath</argument>
- <classpath />
- <argument>-Djava.util.logging.config.class=com.microsoft.reef.util.logging.Config
+ <classpath/>
+ <argument>-Djava.util.logging.config.class=org.apache.reef.util.logging.Config
</argument>
<argument>-Dcom.microsoft.reef.runtime.local.folder=${project.build.directory}
</argument>
- <argument>com.microsoft.reef.examples.retained_eval.Launch</argument>
+ <argument>org.apache.reef.examples.retained_eval.Launch</argument>
<!-- <argument>-cmd</argument>
<argument>date</argument>
<argument>-num_runs</argument>
@@ -256,10 +257,10 @@
<executable>java</executable>
<arguments>
<argument>-classpath</argument>
- <classpath />
- <argument>-Djava.util.logging.config.class=com.microsoft.reef.util.logging.Config
+ <classpath/>
+ <argument>-Djava.util.logging.config.class=org.apache.reef.util.logging.Config
</argument>
- <argument>com.microsoft.reef.examples.retained_eval.Launch</argument>
+ <argument>org.apache.reef.examples.retained_eval.Launch</argument>
<argument>-cmd</argument>
<argument>date</argument>
<argument>-num_runs</argument>
@@ -285,12 +286,12 @@
<executable>java</executable>
<arguments>
<argument>-classpath</argument>
- <classpath />
+ <classpath/>
<argument>-Dcom.microsoft.reef.runtime.local.folder=${project.build.directory}
</argument>
- <argument>-Djava.util.logging.config.class=com.microsoft.reef.util.logging.Config
+ <argument>-Djava.util.logging.config.class=org.apache.reef.util.logging.Config
</argument>
- <argument>com.microsoft.reef.examples.suspend.Launch</argument>
+ <argument>org.apache.reef.examples.suspend.Launch</argument>
<argument>-delay</argument>
<argument>1</argument>
<argument>-cycles</argument>
@@ -316,12 +317,12 @@
<executable>java</executable>
<arguments>
<argument>-classpath</argument>
- <classpath />
+ <classpath/>
<argument>-Dcom.microsoft.reef.runtime.local.folder=${project.build.directory}
</argument>
- <argument>-Djava.util.logging.config.class=com.microsoft.reef.util.logging.Config
+ <argument>-Djava.util.logging.config.class=org.apache.reef.util.logging.Config
</argument>
- <argument>com.microsoft.reef.examples.pool.Launch</argument>
+ <argument>org.apache.reef.examples.pool.Launch</argument>
<argument>-evaluators</argument>
<argument>4</argument>
<argument>-tasks</argument>
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/50444bba/reef-examples/reef-sankey.js
----------------------------------------------------------------------
diff --git a/reef-examples/reef-sankey.js b/reef-examples/reef-sankey.js
deleted file mode 100644
index d15d60e..0000000
--- a/reef-examples/reef-sankey.js
+++ /dev/null
@@ -1,325 +0,0 @@
-/*
- * Copyright (C) 2014 Microsoft Corporation
- *
- * Licensed 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.
- */
-d3.sankey = function () {
- var sankey = {},
- nodeWidth = 24,
- nodePadding = 8,
- size = [1, 1],
- nodes = [],
- links = [];
-
- sankey.nodeWidth = function (_) {
- if (!arguments.length) return nodeWidth;
- nodeWidth = +_;
- return sankey;
- };
-
- sankey.nodePadding = function (_) {
- if (!arguments.length) return nodePadding;
- nodePadding = +_;
- return sankey;
- };
-
- sankey.nodes = function (_) {
- if (!arguments.length) return nodes;
- nodes = _;
- return sankey;
- };
-
- sankey.links = function (_) {
- if (!arguments.length) return links;
- links = _;
- return sankey;
- };
-
- sankey.size = function (_) {
- if (!arguments.length) return size;
- size = _;
- return sankey;
- };
-
- sankey.layout = function (iterations) {
- computeNodeLinks();
- computeNodeValues();
- computeNodeBreadths();
- computeNodeDepths(iterations);
- computeLinkDepths();
- return sankey;
- };
-
- sankey.relayout = function () {
- computeLinkDepths();
- return sankey;
- };
-
- sankey.link = function () {
- var curvature = 0.6;
-
- function link(d) {
- var x0 = d.source.x + d.source.dx,
- x1 = d.target.x,
- xi = d3.interpolateNumber(x0, x1),
- x2 = xi(curvature),
- x3 = xi(1 - curvature),
- y0 = d.source.y + d.sy + d.dy / 2,
- y1 = d.target.y + d.ty + d.dy / 2;
- return "M" + x0 + "," + y0
- + "C" + x2 + "," + y0
- + " " + x3 + "," + y1
- + " " + x1 + "," + y1;
- }
-
- link.curvature = function (_) {
- if (!arguments.length) return curvature;
- curvature = +_;
- return link;
- };
-
- return link;
- };
-
- // Populate the sourceLinks and targetLinks for each node.
- // Also, if the source and target are not objects, assume they are indices.
- function computeNodeLinks() {
- nodes.forEach(function (node) {
- node.sourceLinks = [];
- node.targetLinks = [];
- });
- links.forEach(function (link) {
- var source = link.source,
- target = link.target;
- if (typeof source === "number") source = link.source = nodes[link.source];
- if (typeof target === "number") target = link.target = nodes[link.target];
- source.sourceLinks.push(link);
- target.targetLinks.push(link);
- });
- }
-
- // Compute the value (size) of each node by summing the associated links.
- function computeNodeValues() {
- nodes.forEach(function (node) {
- // RCS alt sizing
- if (node.count) {
- node.value = Math.log(node.count+1)/Math.LN2;// / Math.LN2;
- } else {
- node.value = 0.2;
- } // node.value = Math.max(
- //d3.sum(node.sourceLinks, value),
- //d3.sum(node.targetLinks, value)
- //);
- });
- }
-
- // Iteratively assign the breadth (x-position) for each node.
- // Nodes are assigned the maximum breadth of incoming neighbors plus one;
- // nodes with no incoming links are assigned breadth zero, while
- // nodes with no outgoing links are assigned the maximum breadth.
- function computeNodeBreadths() {
- var remainingNodes = nodes,
- nextNodes,
- x = 0;
-
- while (remainingNodes.length) {
- nextNodes = [];
- remainingNodes.forEach(function (node) {
- node.x = x;
- node.dx = nodeWidth;
- node.sourceLinks.forEach(function (link) {
- nextNodes.push(link.target);
- });
- });
- remainingNodes = nextNodes;
- ++x;
- }
-
- moveSourcesRight();
-// moveSinksRight(x);
- scaleNodeBreadths((size[0] - nodeWidth) / (x - 1));
- }
-
- function moveSourcesRight() {
- nodes.forEach(function (node) {
- if (!node.targetLinks.length) {
- node.x = d3.min(node.sourceLinks, function (d) { return d.target.x; }) - 1;
- }
- });
- }
-
- function moveSinksRight(x) {
- nodes.forEach(function (node) {
- if (!node.sourceLinks.length) {
- node.x = x - 1;
- }
- });
- }
-
- function scaleNodeBreadths(kx) {
- nodes.forEach(function (node) {
- node.x *= kx;
- });
- }
-
- function computeNodeDepths(iterations) {
- var nodesByBreadth = d3.nest()
- .key(function (d) { return d.x; })
- .sortKeys(d3.ascending)
- .entries(nodes)
- .map(function (d) { return d.values; });
-
- //
- initializeNodeDepth();
- resolveCollisions();
- for (var alpha = 1; iterations > 0; --iterations) {
- relaxRightToLeft(alpha *= .99);
- resolveCollisions();
- relaxLeftToRight(alpha);
- resolveCollisions();
- }
-
- function initializeNodeDepth() {
- var ky = d3.min(nodesByBreadth, function (nodes) {
- return (size[1] - (nodes.length - 1) * nodePadding) / d3.sum(nodes, value);
- });
-
- nodesByBreadth.forEach(function (nodes) {
- nodes.forEach(function (node, i) {
- node.y = i;
- node.dy = node.value * ky;
- });
- });
-
- links.forEach(function (link) {
- link.dy = 3;//0.1 * ky; // RCS alt scaling link.value * ky;
- });
- }
-
- function relaxLeftToRight(alpha) {
- nodesByBreadth.forEach(function (nodes, breadth) {
- nodes.forEach(function (node) {
- if (node.targetLinks.length) {
- var y = d3.sum(node.targetLinks, weightedSource) / d3.sum(node.targetLinks, value);
- node.y += (y - center(node)) * alpha;
- }
- });
- });
-
- function weightedSource(link) {
- return center(link.source) * link.value;
- }
- }
-
- function relaxRightToLeft(alpha) {
- nodesByBreadth.slice().reverse().forEach(function (nodes) {
- nodes.forEach(function (node) {
- if (node.sourceLinks.length) {
- var y = d3.sum(node.sourceLinks, weightedTarget) / d3.sum(node.sourceLinks, value);
- node.y += (y - center(node)) * alpha;
- }
- });
- });
-
- function weightedTarget(link) {
- return center(link.target) * link.value;
- }
- }
-
- function resolveCollisions() {
- nodesByBreadth.forEach(function (nodes) {
- var node,
- dy,
- y0 = 0,
- n = nodes.length,
- i;
-
- // Push any overlapping nodes down.
- nodes.sort(ascendingDepth);
- for (i = 0; i < n; ++i) {
- node = nodes[i];
- dy = y0 - node.y;
- if (dy > 0) node.y += dy;
- y0 = node.y + node.dy + nodePadding;
- }
-
- // If the bottommost node goes outside the bounds, push it back up.
- dy = y0 - nodePadding - size[1];
- if (dy > 0) {
- y0 = node.y -= dy;
-
- // Push any overlapping nodes back up.
- for (i = n - 2; i >= 0; --i) {
- node = nodes[i];
- dy = node.y + node.dy + nodePadding - y0;
- if (dy > 0) node.y -= dy;
- y0 = node.y;
- }
- }
- });
- }
-
- function ascendingDepth(a, b) {
- return a.y - b.y;
- }
- }
-
- function computeLinkDepths() {
- nodes.forEach(function (node) {
- node.sourceLinks.sort(ascendingTargetDepth);
- node.targetLinks.sort(ascendingSourceDepth);
- });
- nodes.forEach(function (node) {
- // RCS alt sizing
- var sdy = (node.dy)/(Math.max(1, node.sourceLinks.length));
- var tdy = (node.dy)/(Math.max(1, node.targetLinks.length));
- var sy = -1.5+sdy/2, ty = -1.5+tdy/2; //node.dy / 2.0, ty = node.dy / 2.0;
- node.sourceLinks.forEach(function (link) {
- link.sy = sy;
- sy += sdy;
- });
- node.targetLinks.forEach(function (link) {
- link.ty = ty;
- ty += tdy;
- });
- /*var sy = 0, ty = 0;
- node.sourceLinks.forEach(function (link) {
- link.sy = sy;
- sy += link.dy;
- });
- node.targetLinks.forEach(function (link) {
- link.ty = ty;
- ty += link.dy;
- });*/
- });
-
- function ascendingSourceDepth(a, b) {
- return a.source.y - b.source.y;
- }
-
- function ascendingTargetDepth(a, b) {
- return a.target.y - b.target.y;
- }
- }
-
- function center(node) {
- return node.y + node.dy / 2;
- }
-
- function value(link) {
- return link.value;
- }
-
- return sankey;
-};
-
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/50444bba/reef-examples/src/main/java/com/microsoft/reef/examples/data/loading/DataLoadingREEF.java
----------------------------------------------------------------------
diff --git a/reef-examples/src/main/java/com/microsoft/reef/examples/data/loading/DataLoadingREEF.java b/reef-examples/src/main/java/com/microsoft/reef/examples/data/loading/DataLoadingREEF.java
deleted file mode 100644
index 67f4fa1..0000000
--- a/reef-examples/src/main/java/com/microsoft/reef/examples/data/loading/DataLoadingREEF.java
+++ /dev/null
@@ -1,125 +0,0 @@
-/**
- * Copyright (C) 2014 Microsoft Corporation
- *
- * Licensed 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.
- */
-package com.microsoft.reef.examples.data.loading;
-
-import com.microsoft.reef.annotations.audience.ClientSide;
-import com.microsoft.reef.client.DriverConfiguration;
-import com.microsoft.reef.client.DriverLauncher;
-import com.microsoft.reef.client.LauncherStatus;
-import com.microsoft.reef.driver.evaluator.EvaluatorRequest;
-import com.microsoft.reef.io.data.loading.api.DataLoadingRequestBuilder;
-import com.microsoft.reef.runtime.local.client.LocalRuntimeConfiguration;
-import com.microsoft.reef.runtime.yarn.client.YarnClientConfiguration;
-import com.microsoft.reef.util.EnvironmentUtils;
-import com.microsoft.tang.Configuration;
-import com.microsoft.tang.Injector;
-import com.microsoft.tang.JavaConfigurationBuilder;
-import com.microsoft.tang.Tang;
-import com.microsoft.tang.annotations.Name;
-import com.microsoft.tang.annotations.NamedParameter;
-import com.microsoft.tang.exceptions.BindException;
-import com.microsoft.tang.exceptions.InjectionException;
-import com.microsoft.tang.formats.CommandLine;
-import org.apache.hadoop.mapred.TextInputFormat;
-
-import java.io.IOException;
-import java.util.logging.Level;
-import java.util.logging.Logger;
-
-/**
- * Client for the data loading demo app
- */
-@ClientSide
-public class DataLoadingREEF {
-
- private static final Logger LOG = Logger.getLogger(DataLoadingREEF.class.getName());
-
- private static final int NUM_LOCAL_THREADS = 16;
- private static final int NUM_SPLITS = 6;
- private static final int NUM_COMPUTE_EVALUATORS = 2;
-
- /**
- * Command line parameter = true to run locally, or false to run on YARN.
- */
- @NamedParameter(doc = "Whether or not to run on the local runtime",
- short_name = "local", default_value = "true")
- public static final class Local implements Name<Boolean> {
- }
-
- @NamedParameter(doc = "Number of minutes before timeout",
- short_name = "timeout", default_value = "2")
- public static final class TimeOut implements Name<Integer> {
- }
-
- @NamedParameter(short_name = "input")
- public static final class InputDir implements Name<String> {
- }
-
- public static void main(final String[] args)
- throws InjectionException, BindException, IOException {
-
- final Tang tang = Tang.Factory.getTang();
-
- final JavaConfigurationBuilder cb = tang.newConfigurationBuilder();
-
- new CommandLine(cb)
- .registerShortNameOfClass(Local.class)
- .registerShortNameOfClass(TimeOut.class)
- .registerShortNameOfClass(DataLoadingREEF.InputDir.class)
- .processCommandLine(args);
-
- final Injector injector = tang.newInjector(cb.build());
-
- final boolean isLocal = injector.getNamedInstance(Local.class);
- final int jobTimeout = injector.getNamedInstance(TimeOut.class) * 60 * 1000;
- final String inputDir = injector.getNamedInstance(DataLoadingREEF.InputDir.class);
-
- final Configuration runtimeConfiguration;
- if (isLocal) {
- LOG.log(Level.INFO, "Running Data Loading demo on the local runtime");
- runtimeConfiguration = LocalRuntimeConfiguration.CONF
- .set(LocalRuntimeConfiguration.NUMBER_OF_THREADS, NUM_LOCAL_THREADS)
- .build();
- } else {
- LOG.log(Level.INFO, "Running Data Loading demo on YARN");
- runtimeConfiguration = YarnClientConfiguration.CONF.build();
- }
-
- final EvaluatorRequest computeRequest = EvaluatorRequest.newBuilder()
- .setNumber(NUM_COMPUTE_EVALUATORS)
- .setMemory(512)
- .setNumberOfCores(1)
- .build();
-
- final Configuration dataLoadConfiguration = new DataLoadingRequestBuilder()
- .setMemoryMB(1024)
- .setInputFormatClass(TextInputFormat.class)
- .setInputPath(inputDir)
- .setNumberOfDesiredSplits(NUM_SPLITS)
- .setComputeRequest(computeRequest)
- .setDriverConfigurationModule(DriverConfiguration.CONF
- .set(DriverConfiguration.GLOBAL_LIBRARIES, EnvironmentUtils.getClassLocation(LineCounter.class))
- .set(DriverConfiguration.ON_CONTEXT_ACTIVE, LineCounter.ContextActiveHandler.class)
- .set(DriverConfiguration.ON_TASK_COMPLETED, LineCounter.TaskCompletedHandler.class)
- .set(DriverConfiguration.DRIVER_IDENTIFIER, "DataLoadingREEF"))
- .build();
-
- final LauncherStatus state =
- DriverLauncher.getLauncher(runtimeConfiguration).run(dataLoadConfiguration, jobTimeout);
-
- LOG.log(Level.INFO, "REEF job completed: {0}", state);
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/50444bba/reef-examples/src/main/java/com/microsoft/reef/examples/data/loading/LineCounter.java
----------------------------------------------------------------------
diff --git a/reef-examples/src/main/java/com/microsoft/reef/examples/data/loading/LineCounter.java b/reef-examples/src/main/java/com/microsoft/reef/examples/data/loading/LineCounter.java
deleted file mode 100644
index 8b47514..0000000
--- a/reef-examples/src/main/java/com/microsoft/reef/examples/data/loading/LineCounter.java
+++ /dev/null
@@ -1,124 +0,0 @@
-/**
- * Copyright (C) 2014 Microsoft Corporation
- *
- * Licensed 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.
- */
-package com.microsoft.reef.examples.data.loading;
-
-import java.util.concurrent.atomic.AtomicInteger;
-import java.util.logging.Level;
-import java.util.logging.Logger;
-
-import javax.inject.Inject;
-
-import com.microsoft.reef.annotations.audience.DriverSide;
-import com.microsoft.reef.driver.context.ActiveContext;
-import com.microsoft.reef.driver.context.ContextConfiguration;
-import com.microsoft.reef.driver.task.CompletedTask;
-import com.microsoft.reef.driver.task.TaskConfiguration;
-import com.microsoft.reef.io.data.loading.api.DataLoadingService;
-import com.microsoft.reef.poison.PoisonedConfiguration;
-import com.microsoft.tang.Configuration;
-import com.microsoft.tang.Tang;
-import com.microsoft.tang.annotations.Unit;
-import com.microsoft.tang.exceptions.BindException;
-import com.microsoft.wake.EventHandler;
-
-/**
- * Driver side for the line counting demo that uses the data loading service.
- */
-@DriverSide
-@Unit
-public class LineCounter {
-
- private static final Logger LOG = Logger.getLogger(LineCounter.class.getName());
-
- private final AtomicInteger ctrlCtxIds = new AtomicInteger();
- private final AtomicInteger lineCnt = new AtomicInteger();
- private final AtomicInteger completedDataTasks = new AtomicInteger();
-
- private final DataLoadingService dataLoadingService;
-
- @Inject
- public LineCounter(final DataLoadingService dataLoadingService) {
- this.dataLoadingService = dataLoadingService;
- this.completedDataTasks.set(dataLoadingService.getNumberOfPartitions());
- }
-
- public class ContextActiveHandler implements EventHandler<ActiveContext> {
-
- @Override
- public void onNext(final ActiveContext activeContext) {
-
- final String contextId = activeContext.getId();
- LOG.log(Level.FINER, "Context active: {0}", contextId);
-
- if (dataLoadingService.isDataLoadedContext(activeContext)) {
-
- final String lcContextId = "LineCountCtxt-" + ctrlCtxIds.getAndIncrement();
- LOG.log(Level.FINEST, "Submit LineCount context {0} to: {1}",
- new Object[] { lcContextId, contextId });
-
- final Configuration poisonedConfiguration = PoisonedConfiguration.CONTEXT_CONF
- .set(PoisonedConfiguration.CRASH_PROBABILITY, "0.4")
- .set(PoisonedConfiguration.CRASH_TIMEOUT, "1")
- .build();
-
- activeContext.submitContext(Tang.Factory.getTang()
- .newConfigurationBuilder(poisonedConfiguration,
- ContextConfiguration.CONF.set(ContextConfiguration.IDENTIFIER, lcContextId).build())
- .build());
-
- } else if (activeContext.getId().startsWith("LineCountCtxt")) {
-
- final String taskId = "LineCountTask-" + ctrlCtxIds.getAndIncrement();
- LOG.log(Level.FINEST, "Submit LineCount task {0} to: {1}", new Object[] { taskId, contextId });
-
- try {
- activeContext.submitTask(TaskConfiguration.CONF
- .set(TaskConfiguration.IDENTIFIER, taskId)
- .set(TaskConfiguration.TASK, LineCountingTask.class)
- .build());
- } catch (final BindException ex) {
- LOG.log(Level.SEVERE, "Configuration error in " + contextId, ex);
- throw new RuntimeException("Configuration error in " + contextId, ex);
- }
- } else {
- LOG.log(Level.FINEST, "Line count Compute Task {0} -- Closing", contextId);
- activeContext.close();
- }
- }
- }
-
- public class TaskCompletedHandler implements EventHandler<CompletedTask> {
- @Override
- public void onNext(final CompletedTask completedTask) {
-
- final String taskId = completedTask.getId();
- LOG.log(Level.FINEST, "Completed Task: {0}", taskId);
-
- final byte[] retBytes = completedTask.get();
- final String retStr = retBytes == null ? "No RetVal": new String(retBytes);
- LOG.log(Level.FINE, "Line count from {0} : {1}", new String[] { taskId, retStr });
-
- lineCnt.addAndGet(Integer.parseInt(retStr));
-
- if (completedDataTasks.decrementAndGet() <= 0) {
- LOG.log(Level.INFO, "Total line count: {0}", lineCnt.get());
- }
-
- LOG.log(Level.FINEST, "Releasing Context: {0}", taskId);
- completedTask.getActiveContext().close();
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/50444bba/reef-examples/src/main/java/com/microsoft/reef/examples/data/loading/LineCountingTask.java
----------------------------------------------------------------------
diff --git a/reef-examples/src/main/java/com/microsoft/reef/examples/data/loading/LineCountingTask.java b/reef-examples/src/main/java/com/microsoft/reef/examples/data/loading/LineCountingTask.java
deleted file mode 100644
index 0b08545..0000000
--- a/reef-examples/src/main/java/com/microsoft/reef/examples/data/loading/LineCountingTask.java
+++ /dev/null
@@ -1,56 +0,0 @@
-/**
- * Copyright (C) 2014 Microsoft Corporation
- *
- * Licensed 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.
- */
-package com.microsoft.reef.examples.data.loading;
-
-import com.microsoft.reef.annotations.audience.TaskSide;
-import com.microsoft.reef.io.data.loading.api.DataSet;
-import com.microsoft.reef.io.network.util.Pair;
-import com.microsoft.reef.task.Task;
-import org.apache.hadoop.io.LongWritable;
-import org.apache.hadoop.io.Text;
-
-import javax.inject.Inject;
-import java.util.logging.Level;
-import java.util.logging.Logger;
-
-/**
- * The task that iterates over the data set to count the number of records.
- * Assumes TextInputFormat and that records represent lines.
- */
-@TaskSide
-public class LineCountingTask implements Task {
-
- private static final Logger LOG = Logger.getLogger(LineCountingTask.class.getName());
-
- private final DataSet<LongWritable, Text> dataSet;
-
- @Inject
- public LineCountingTask(final DataSet<LongWritable, Text> dataSet) {
- this.dataSet = dataSet;
- }
-
- @Override
- public byte[] call(final byte[] memento) throws Exception {
- LOG.log(Level.FINER, "LineCounting task started");
- int numEx = 0;
- for (final Pair<LongWritable, Text> keyValue : dataSet) {
- // LOG.log(Level.FINEST, "Read line: {0}", keyValue);
- ++numEx;
- }
- LOG.log(Level.FINER, "LineCounting task finished: read {0} lines", numEx);
- return Integer.toString(numEx).getBytes();
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/50444bba/reef-examples/src/main/java/com/microsoft/reef/examples/hello/HelloDriver.java
----------------------------------------------------------------------
diff --git a/reef-examples/src/main/java/com/microsoft/reef/examples/hello/HelloDriver.java b/reef-examples/src/main/java/com/microsoft/reef/examples/hello/HelloDriver.java
deleted file mode 100644
index a018508..0000000
--- a/reef-examples/src/main/java/com/microsoft/reef/examples/hello/HelloDriver.java
+++ /dev/null
@@ -1,81 +0,0 @@
-/**
- * Copyright (C) 2014 Microsoft Corporation
- *
- * Licensed 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.
- */
-package com.microsoft.reef.examples.hello;
-
-import com.microsoft.reef.driver.evaluator.AllocatedEvaluator;
-import com.microsoft.reef.driver.evaluator.EvaluatorRequest;
-import com.microsoft.reef.driver.evaluator.EvaluatorRequestor;
-import com.microsoft.reef.driver.task.TaskConfiguration;
-import com.microsoft.tang.Configuration;
-import com.microsoft.tang.annotations.Unit;
-import com.microsoft.wake.EventHandler;
-import com.microsoft.wake.time.event.StartTime;
-
-import javax.inject.Inject;
-import java.util.logging.Level;
-import java.util.logging.Logger;
-
-/**
- * The Driver code for the Hello REEF Application
- */
-@Unit
-public final class HelloDriver {
-
- private static final Logger LOG = Logger.getLogger(HelloDriver.class.getName());
-
- private final EvaluatorRequestor requestor;
-
- /**
- * Job driver constructor - instantiated via TANG.
- *
- * @param requestor evaluator requestor object used to create new evaluator containers.
- */
- @Inject
- public HelloDriver(final EvaluatorRequestor requestor) {
- this.requestor = requestor;
- LOG.log(Level.FINE, "Instantiated 'HelloDriver'");
- }
-
- /**
- * Handles the StartTime event: Request as single Evaluator.
- */
- public final class StartHandler implements EventHandler<StartTime> {
- @Override
- public void onNext(final StartTime startTime) {
- HelloDriver.this.requestor.submit(EvaluatorRequest.newBuilder()
- .setNumber(1)
- .setMemory(64)
- .setNumberOfCores(1)
- .build());
- LOG.log(Level.INFO, "Requested Evaluator.");
- }
- }
-
- /**
- * Handles AllocatedEvaluator: Submit the HelloTask
- */
- public final class EvaluatorAllocatedHandler implements EventHandler<AllocatedEvaluator> {
- @Override
- public void onNext(final AllocatedEvaluator allocatedEvaluator) {
- LOG.log(Level.INFO, "Submitting HelloREEF task to AllocatedEvaluator: {0}", allocatedEvaluator);
- final Configuration taskConfiguration = TaskConfiguration.CONF
- .set(TaskConfiguration.IDENTIFIER, "HelloREEFTask")
- .set(TaskConfiguration.TASK, HelloTask.class)
- .build();
- allocatedEvaluator.submitTask(taskConfiguration);
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/50444bba/reef-examples/src/main/java/com/microsoft/reef/examples/hello/HelloREEF.java
----------------------------------------------------------------------
diff --git a/reef-examples/src/main/java/com/microsoft/reef/examples/hello/HelloREEF.java b/reef-examples/src/main/java/com/microsoft/reef/examples/hello/HelloREEF.java
deleted file mode 100644
index 6642390..0000000
--- a/reef-examples/src/main/java/com/microsoft/reef/examples/hello/HelloREEF.java
+++ /dev/null
@@ -1,74 +0,0 @@
-/**
- * Copyright (C) 2014 Microsoft Corporation
- *
- * Licensed 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.
- */
-package com.microsoft.reef.examples.hello;
-
-import com.microsoft.reef.client.DriverConfiguration;
-import com.microsoft.reef.client.DriverLauncher;
-import com.microsoft.reef.client.LauncherStatus;
-import com.microsoft.reef.runtime.local.client.LocalRuntimeConfiguration;
-import com.microsoft.reef.util.EnvironmentUtils;
-import com.microsoft.tang.Configuration;
-import com.microsoft.tang.exceptions.BindException;
-import com.microsoft.tang.exceptions.InjectionException;
-
-import java.util.logging.Level;
-import java.util.logging.Logger;
-
-/**
- * The Client for Hello REEF example.
- */
-public final class HelloREEF {
-
- private static final Logger LOG = Logger.getLogger(HelloREEF.class.getName());
-
- /**
- * Number of milliseconds to wait for the job to complete.
- */
- private static final int JOB_TIMEOUT = 10000; // 10 sec.
-
- /**
- * @return the configuration of the HelloREEF driver.
- */
- public static Configuration getDriverConfiguration() {
- return DriverConfiguration.CONF
- .set(DriverConfiguration.GLOBAL_LIBRARIES, EnvironmentUtils.getClassLocation(HelloDriver.class))
- .set(DriverConfiguration.DRIVER_IDENTIFIER, "HelloREEF")
- .set(DriverConfiguration.ON_DRIVER_STARTED, HelloDriver.StartHandler.class)
- .set(DriverConfiguration.ON_EVALUATOR_ALLOCATED, HelloDriver.EvaluatorAllocatedHandler.class)
- .build();
- }
-
- public static LauncherStatus runHelloReef(final Configuration runtimeConf, final int timeOut)
- throws BindException, InjectionException {
- final Configuration driverConf = getDriverConfiguration();
- return DriverLauncher.getLauncher(runtimeConf).run(driverConf, timeOut);
- }
-
- /**
- * Start Hello REEF job. Runs method runHelloReef().
- *
- * @param args command line parameters.
- * @throws BindException configuration error.
- * @throws InjectionException configuration error.
- */
- public static void main(final String[] args) throws BindException, InjectionException {
- final Configuration runtimeConfiguration = LocalRuntimeConfiguration.CONF
- .set(LocalRuntimeConfiguration.NUMBER_OF_THREADS, 2)
- .build();
- final LauncherStatus status = runHelloReef(runtimeConfiguration, JOB_TIMEOUT);
- LOG.log(Level.INFO, "REEF job completed: {0}", status);
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/50444bba/reef-examples/src/main/java/com/microsoft/reef/examples/hello/HelloREEFNoClient.java
----------------------------------------------------------------------
diff --git a/reef-examples/src/main/java/com/microsoft/reef/examples/hello/HelloREEFNoClient.java b/reef-examples/src/main/java/com/microsoft/reef/examples/hello/HelloREEFNoClient.java
deleted file mode 100644
index 7da6546..0000000
--- a/reef-examples/src/main/java/com/microsoft/reef/examples/hello/HelloREEFNoClient.java
+++ /dev/null
@@ -1,61 +0,0 @@
-/**
- * Copyright (C) 2014 Microsoft Corporation
- *
- * Licensed 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.
- */
-package com.microsoft.reef.examples.hello;
-
-import com.microsoft.reef.client.DriverConfiguration;
-import com.microsoft.reef.client.REEF;
-import com.microsoft.reef.runtime.local.client.LocalRuntimeConfiguration;
-import com.microsoft.reef.util.EnvironmentUtils;
-import com.microsoft.tang.Configuration;
-import com.microsoft.tang.Tang;
-import com.microsoft.tang.exceptions.BindException;
-import com.microsoft.tang.exceptions.InjectionException;
-
-import java.util.logging.Level;
-import java.util.logging.Logger;
-
-/**
- * A main() for running hello REEF without a persistent client connection.
- */
-public final class HelloREEFNoClient {
-
- private static final Logger LOG = Logger.getLogger(HelloREEFNoClient.class.getName());
-
- public static void runHelloReefWithoutClient(
- final Configuration runtimeConf) throws InjectionException {
-
- final REEF reef = Tang.Factory.getTang().newInjector(runtimeConf).getInstance(REEF.class);
-
- final Configuration driverConf = DriverConfiguration.CONF
- .set(DriverConfiguration.GLOBAL_LIBRARIES, EnvironmentUtils.getClassLocation(HelloDriver.class))
- .set(DriverConfiguration.DRIVER_IDENTIFIER, "HelloREEF")
- .set(DriverConfiguration.ON_DRIVER_STARTED, HelloDriver.StartHandler.class)
- .set(DriverConfiguration.ON_EVALUATOR_ALLOCATED, HelloDriver.EvaluatorAllocatedHandler.class)
- .build();
-
- reef.submit(driverConf);
- }
-
- public static void main(final String[] args) throws BindException, InjectionException {
-
- final Configuration runtimeConfiguration = LocalRuntimeConfiguration.CONF
- .set(LocalRuntimeConfiguration.NUMBER_OF_THREADS, 2)
- .build();
-
- runHelloReefWithoutClient(runtimeConfiguration);
- LOG.log(Level.INFO, "Job Submitted");
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/50444bba/reef-examples/src/main/java/com/microsoft/reef/examples/hello/HelloReefYarn.java
----------------------------------------------------------------------
diff --git a/reef-examples/src/main/java/com/microsoft/reef/examples/hello/HelloReefYarn.java b/reef-examples/src/main/java/com/microsoft/reef/examples/hello/HelloReefYarn.java
deleted file mode 100644
index a5b6870..0000000
--- a/reef-examples/src/main/java/com/microsoft/reef/examples/hello/HelloReefYarn.java
+++ /dev/null
@@ -1,67 +0,0 @@
-/**
- * Copyright (C) 2014 Microsoft Corporation
- *
- * Licensed 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.
- */
-package com.microsoft.reef.examples.hello;
-
-import com.microsoft.reef.client.DriverConfiguration;
-import com.microsoft.reef.client.DriverLauncher;
-import com.microsoft.reef.client.LauncherStatus;
-import com.microsoft.reef.runtime.yarn.client.YarnClientConfiguration;
-import com.microsoft.tang.Configuration;
-import com.microsoft.tang.exceptions.InjectionException;
-
-import java.util.logging.Level;
-import java.util.logging.Logger;
-
-/**
- * The Client for Hello REEF example.
- */
-public final class HelloReefYarn {
-
- private static final Logger LOG = Logger.getLogger(HelloReefYarn.class.getName());
-
- /**
- * Number of milliseconds to wait for the job to complete.
- */
- private static final int JOB_TIMEOUT = 30000; // 30 sec.
-
-
- /**
- * @return the configuration of the HelloREEF driver.
- */
- private static Configuration getDriverConfiguration() {
- return DriverConfiguration.CONF
- .set(DriverConfiguration.GLOBAL_LIBRARIES, HelloReefYarn.class.getProtectionDomain().getCodeSource().getLocation().getFile())
- .set(DriverConfiguration.DRIVER_IDENTIFIER, "HelloREEF")
- .set(DriverConfiguration.ON_DRIVER_STARTED, HelloDriver.StartHandler.class)
- .set(DriverConfiguration.ON_EVALUATOR_ALLOCATED, HelloDriver.EvaluatorAllocatedHandler.class)
- .build();
- }
-
- /**
- * Start Hello REEF job. Runs method runHelloReef().
- *
- * @param args command line parameters.
- * @throws com.microsoft.tang.exceptions.BindException configuration error.
- * @throws com.microsoft.tang.exceptions.InjectionException configuration error.
- */
- public static void main(final String[] args) throws InjectionException {
-
- final LauncherStatus status = DriverLauncher
- .getLauncher(YarnClientConfiguration.CONF.build())
- .run(getDriverConfiguration(), JOB_TIMEOUT);
- LOG.log(Level.INFO, "REEF job completed: {0}", status);
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/50444bba/reef-examples/src/main/java/com/microsoft/reef/examples/hello/HelloTask.java
----------------------------------------------------------------------
diff --git a/reef-examples/src/main/java/com/microsoft/reef/examples/hello/HelloTask.java b/reef-examples/src/main/java/com/microsoft/reef/examples/hello/HelloTask.java
deleted file mode 100644
index a210a26..0000000
--- a/reef-examples/src/main/java/com/microsoft/reef/examples/hello/HelloTask.java
+++ /dev/null
@@ -1,36 +0,0 @@
-/**
- * Copyright (C) 2014 Microsoft Corporation
- *
- * Licensed 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.
- */
-package com.microsoft.reef.examples.hello;
-
-import com.microsoft.reef.task.Task;
-
-import javax.inject.Inject;
-
-/**
- * A 'hello REEF' Task.
- */
-public final class HelloTask implements Task {
-
- @Inject
- HelloTask() {
- }
-
- @Override
- public final byte[] call(final byte[] memento) {
- System.out.println("Hello, REEF!");
- return null;
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/50444bba/reef-examples/src/main/java/com/microsoft/reef/examples/hello/package-info.java
----------------------------------------------------------------------
diff --git a/reef-examples/src/main/java/com/microsoft/reef/examples/hello/package-info.java b/reef-examples/src/main/java/com/microsoft/reef/examples/hello/package-info.java
deleted file mode 100644
index c481238..0000000
--- a/reef-examples/src/main/java/com/microsoft/reef/examples/hello/package-info.java
+++ /dev/null
@@ -1,19 +0,0 @@
-/**
- * Copyright (C) 2014 Microsoft Corporation
- *
- * Licensed 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.
- */
-/**
- * The Hello REEF example.
- */
-package com.microsoft.reef.examples.hello;
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/50444bba/reef-examples/src/main/java/com/microsoft/reef/examples/hellohttp/Command.java
----------------------------------------------------------------------
diff --git a/reef-examples/src/main/java/com/microsoft/reef/examples/hellohttp/Command.java b/reef-examples/src/main/java/com/microsoft/reef/examples/hellohttp/Command.java
deleted file mode 100644
index b429de7..0000000
--- a/reef-examples/src/main/java/com/microsoft/reef/examples/hellohttp/Command.java
+++ /dev/null
@@ -1,27 +0,0 @@
-/**
- * Copyright (C) 2014 Microsoft Corporation
- *
- * Licensed 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.
- */
-
-package com.microsoft.reef.examples.hellohttp;
-
-import com.microsoft.tang.annotations.Name;
-import com.microsoft.tang.annotations.NamedParameter;
-
-/**
- * Command line parameter: a command to run. e.g. "echo Hello REEF"
- */
-@NamedParameter(doc = "The shell command", short_name = "cmd", default_value = "*INTERACTIVE*")
-final class Command implements Name<String> {
-}
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/50444bba/reef-examples/src/main/java/com/microsoft/reef/examples/hellohttp/HelloREEFHttp.java
----------------------------------------------------------------------
diff --git a/reef-examples/src/main/java/com/microsoft/reef/examples/hellohttp/HelloREEFHttp.java b/reef-examples/src/main/java/com/microsoft/reef/examples/hellohttp/HelloREEFHttp.java
deleted file mode 100644
index 5bf4290..0000000
--- a/reef-examples/src/main/java/com/microsoft/reef/examples/hellohttp/HelloREEFHttp.java
+++ /dev/null
@@ -1,111 +0,0 @@
-/**
- * Copyright (C) 2014 Microsoft Corporation
- *
- * Licensed 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.
- */
-
-package com.microsoft.reef.examples.hellohttp;
-
-import com.microsoft.reef.client.DriverConfiguration;
-import com.microsoft.reef.client.DriverLauncher;
-import com.microsoft.reef.client.DriverServiceConfiguration;
-import com.microsoft.reef.client.LauncherStatus;
-import com.microsoft.reef.runtime.local.client.LocalRuntimeConfiguration;
-import com.microsoft.reef.util.EnvironmentUtils;
-import com.microsoft.reef.webserver.HttpHandlerConfiguration;
-import com.microsoft.reef.webserver.HttpServerReefEventHandler;
-import com.microsoft.reef.webserver.ReefEventStateManager;
-import com.microsoft.tang.Configuration;
-import com.microsoft.tang.Configurations;
-import com.microsoft.tang.exceptions.BindException;
-import com.microsoft.tang.exceptions.InjectionException;
-
-import java.util.logging.Logger;
-
-/**
- * Example to run HelloREEF with a webserver.
- */
-public final class HelloREEFHttp {
- private static final Logger LOG = Logger.getLogger(HelloREEFHttp.class.getName());
-
- /**
- * Number of milliseconds to wait for the job to complete.
- */
- public static final int JOB_TIMEOUT = 60 * 1000; // 60 sec.
-
- /**
- * @return the driver-side configuration to be merged into the DriverConfiguration to enable the HTTP server.
- */
- public static Configuration getHTTPConfiguration() {
- final Configuration httpHandlerConfiguration = HttpHandlerConfiguration.CONF
- .set(HttpHandlerConfiguration.HTTP_HANDLERS, HttpServerReefEventHandler.class)
- .set(HttpHandlerConfiguration.HTTP_HANDLERS, HttpServerShellCmdtHandler.class)
- .build();
- final Configuration driverConfigurationForHttpServer = DriverServiceConfiguration.CONF
- .set(DriverServiceConfiguration.ON_EVALUATOR_ALLOCATED, ReefEventStateManager.AllocatedEvaluatorStateHandler.class)
- .set(DriverServiceConfiguration.ON_CONTEXT_ACTIVE, ReefEventStateManager.ActiveContextStateHandler.class)
- .set(DriverServiceConfiguration.ON_TASK_RUNNING, ReefEventStateManager.TaskRunningStateHandler.class)
- .set(DriverServiceConfiguration.ON_DRIVER_STARTED, ReefEventStateManager.StartStateHandler.class)
- .set(DriverServiceConfiguration.ON_DRIVER_STOP, ReefEventStateManager.StopStateHandler.class)
- .build();
- return Configurations.merge(httpHandlerConfiguration, driverConfigurationForHttpServer);
- }
-
- /**
- * @return the configuration of the HelloREEF driver.
- */
- public static Configuration getDriverConfiguration() {
- return DriverConfiguration.CONF
- .set(DriverConfiguration.GLOBAL_LIBRARIES, EnvironmentUtils.getClassLocation(HttpShellJobDriver.class))
- .set(DriverConfiguration.DRIVER_IDENTIFIER, "HelloHTTP")
- .set(DriverConfiguration.ON_DRIVER_STARTED, HttpShellJobDriver.StartHandler.class)
- .set(DriverConfiguration.ON_EVALUATOR_ALLOCATED, HttpShellJobDriver.AllocatedEvaluatorHandler.class)
- .set(DriverConfiguration.ON_EVALUATOR_FAILED, HttpShellJobDriver.FailedEvaluatorHandler.class)
- .set(DriverConfiguration.ON_CONTEXT_ACTIVE, HttpShellJobDriver.ActiveContextHandler.class)
- .set(DriverConfiguration.ON_CONTEXT_CLOSED, HttpShellJobDriver.ClosedContextHandler.class)
- .set(DriverConfiguration.ON_CONTEXT_FAILED, HttpShellJobDriver.FailedContextHandler.class)
- .set(DriverConfiguration.ON_TASK_COMPLETED, HttpShellJobDriver.CompletedTaskHandler.class)
- .set(DriverConfiguration.ON_CLIENT_MESSAGE, HttpShellJobDriver.ClientMessageHandler.class)
- .set(DriverConfiguration.ON_CLIENT_CLOSED, HttpShellJobDriver.HttpClientCloseHandler.class)
- .set(DriverConfiguration.ON_DRIVER_STOP, HttpShellJobDriver.StopHandler.class)
- .build();
- }
-
- /**
- * Run Hello Reef with merged configuration
- *
- * @param runtimeConf
- * @param timeOut
- * @return
- * @throws BindException
- * @throws InjectionException
- */
- public static LauncherStatus runHelloReef(final Configuration runtimeConf, final int timeOut)
- throws BindException, InjectionException {
- final Configuration driverConf = Configurations.merge(HelloREEFHttp.getDriverConfiguration(), getHTTPConfiguration());
- return DriverLauncher.getLauncher(runtimeConf).run(driverConf, timeOut);
- }
-
- /**
- * main program
- *
- * @param args
- * @throws InjectionException
- */
- public static void main(final String[] args) throws InjectionException {
- final Configuration runtimeConfiguration = LocalRuntimeConfiguration.CONF
- .set(LocalRuntimeConfiguration.NUMBER_OF_THREADS, 3)
- .build();
- final LauncherStatus status = runHelloReef(runtimeConfiguration, HelloREEFHttp.JOB_TIMEOUT);
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/50444bba/reef-examples/src/main/java/com/microsoft/reef/examples/hellohttp/HelloREEFHttpYarn.java
----------------------------------------------------------------------
diff --git a/reef-examples/src/main/java/com/microsoft/reef/examples/hellohttp/HelloREEFHttpYarn.java b/reef-examples/src/main/java/com/microsoft/reef/examples/hellohttp/HelloREEFHttpYarn.java
deleted file mode 100644
index 44abf39..0000000
--- a/reef-examples/src/main/java/com/microsoft/reef/examples/hellohttp/HelloREEFHttpYarn.java
+++ /dev/null
@@ -1,49 +0,0 @@
-/**
- * Copyright (C) 2014 Microsoft Corporation
- *
- * Licensed 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.
- */
-package com.microsoft.reef.examples.hellohttp;
-
-import com.microsoft.reef.client.LauncherStatus;
-import com.microsoft.reef.runtime.yarn.client.YarnClientConfiguration;
-import com.microsoft.tang.Configuration;
-import com.microsoft.tang.exceptions.BindException;
-import com.microsoft.tang.exceptions.InjectionException;
-
-import java.io.IOException;
-import java.util.logging.Level;
-import java.util.logging.Logger;
-
-/**
- * HelloREEFHttp for running on Yarn
- */
-public class HelloREEFHttpYarn {
-
- private static final Logger LOG = Logger.getLogger(HelloREEFHttpYarn.class.getName());
-
- /**
- * Start Hello REEF job. Runs method runHelloReef().
- *
- * @param args command line parameters.
- * @throws com.microsoft.tang.exceptions.BindException configuration error.
- * @throws com.microsoft.tang.exceptions.InjectionException configuration error.
- */
- public static void main(final String[] args) throws BindException, InjectionException, IOException {
-
- final Configuration runtimeConfiguration = YarnClientConfiguration.CONF.build();
-
- final LauncherStatus status = HelloREEFHttp.runHelloReef(runtimeConfiguration, HelloREEFHttp.JOB_TIMEOUT);
- LOG.log(Level.INFO, "REEF job completed: {0}", status);
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/50444bba/reef-examples/src/main/java/com/microsoft/reef/examples/hellohttp/HttpServerShellCmdtHandler.java
----------------------------------------------------------------------
diff --git a/reef-examples/src/main/java/com/microsoft/reef/examples/hellohttp/HttpServerShellCmdtHandler.java b/reef-examples/src/main/java/com/microsoft/reef/examples/hellohttp/HttpServerShellCmdtHandler.java
deleted file mode 100644
index 81b3c96..0000000
--- a/reef-examples/src/main/java/com/microsoft/reef/examples/hellohttp/HttpServerShellCmdtHandler.java
+++ /dev/null
@@ -1,164 +0,0 @@
-/**
- * Copyright (C) 2014 Microsoft Corporation
- *
- * Licensed 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.
- */
-
-package com.microsoft.reef.examples.hellohttp;
-
-import com.microsoft.reef.util.CommandUtils;
-import com.microsoft.reef.webserver.HttpHandler;
-import com.microsoft.reef.webserver.ParsedHttpRequest;
-import com.microsoft.tang.InjectionFuture;
-import com.microsoft.tang.annotations.Unit;
-import com.microsoft.wake.EventHandler;
-import javax.inject.Inject;
-import javax.servlet.ServletException;
-import javax.servlet.http.HttpServletRequest;
-import javax.servlet.http.HttpServletResponse;
-import java.io.IOException;
-import java.nio.charset.Charset;
-import java.util.List;
-import java.util.Map;
-import java.util.logging.Level;
-import java.util.logging.Logger;
-
-/**
- * Http Event handler for Shell Command
- */
-@Unit
-class HttpServerShellCmdtHandler implements HttpHandler {
- /**
- * Standard Java logger.
- */
- private static final Logger LOG = Logger.getLogger(HttpServerShellCmdtHandler.class.getName());
-
- private static final int WAIT_TIMEOUT = 10 * 1000;
-
- private static final int WAIT_TIME = 50;
-
- /**
- * ClientMessageHandler
- */
- private final InjectionFuture<HttpShellJobDriver.ClientMessageHandler> messageHandler;
-
- /**
- * uri specification
- */
- private String uriSpecification = "Command";
-
- /**
- * output for command
- */
- private String cmdOutput = null;
-
- /**
- * HttpServerDistributedShellEventHandler constructor.
- */
- @Inject
- public HttpServerShellCmdtHandler(final InjectionFuture<HttpShellJobDriver.ClientMessageHandler> messageHandler) {
- this.messageHandler = messageHandler;
- }
-
- /**
- * returns URI specification for the handler
- *
- * @return
- */
- @Override
- public String getUriSpecification() {
- return uriSpecification;
- }
-
- /**
- * set URI specification
- * @param s
- */
- public void setUriSpecification(final String s) {
- uriSpecification = s;
- }
-
- /**
- * it is called when receiving a http request
- *
- * @param parsedHttpRequest
- * @param response
- */
- @Override
- public final synchronized void onHttpRequest(final ParsedHttpRequest parsedHttpRequest, final HttpServletResponse response) throws IOException, ServletException {
- LOG.log(Level.INFO, "HttpServeShellCmdtHandler in webserver onHttpRequest is called: {0}", parsedHttpRequest.getRequestUri());
- final Map<String, List<String>> queries = parsedHttpRequest.getQueryMap();
- final String queryStr = parsedHttpRequest.getQueryString();
-
- if (parsedHttpRequest.getTargetEntity().equalsIgnoreCase("Evaluators")) {
- final byte[] b = HttpShellJobDriver.CODEC.encode(queryStr);
- LOG.log(Level.INFO, "HttpServeShellCmdtHandler call HelloDriver onCommand(): {0}", queryStr);
- messageHandler.get().onNext(b);
-
- notify();
-
- final long endTime = System.currentTimeMillis() + WAIT_TIMEOUT;
- while (cmdOutput == null) {
- final long waitTime = endTime - System.currentTimeMillis();
- if (waitTime <= 0) {
- break;
- }
-
- try {
- wait(WAIT_TIME);
- } catch (final InterruptedException e) {
- LOG.log(Level.WARNING, "HttpServeShellCmdtHandler onHttpRequest InterruptedException: {0}", e);
- }
- }
- response.getOutputStream().write(cmdOutput.getBytes(Charset.forName("UTF-8")));
- cmdOutput = null;
- } else if (parsedHttpRequest.getTargetEntity().equalsIgnoreCase("Driver")) {
- final String cmdOutput = CommandUtils.runCommand(queryStr);
- response.getOutputStream().write(cmdOutput.getBytes(Charset.forName("UTF-8")));
- }
- }
-
- /**
- * called after shell command is completed
- * @param message
- */
- public final synchronized void onHttpCallback(byte[] message) {
- final long endTime = System.currentTimeMillis() + WAIT_TIMEOUT;
- while (cmdOutput != null) {
- final long waitTime = endTime - System.currentTimeMillis();
- if (waitTime <= 0) {
- break;
- }
-
- try {
- wait(WAIT_TIME);
- } catch(final InterruptedException e) {
- LOG.log(Level.WARNING, "HttpServeShellCmdtHandler onHttpCallback InterruptedException: {0}", e);
- }
- }
- LOG.log(Level.INFO, "HttpServeShellCmdtHandler OnCallback: {0}", HttpShellJobDriver.CODEC.decode(message));
- cmdOutput = HttpShellJobDriver.CODEC.decode(message);
-
- notify();
- }
-
- /**
- * Handler for client to call back
- */
- final class ClientCallBackHandler implements EventHandler<byte[]> {
- @Override
- public void onNext(final byte[] message) {
- HttpServerShellCmdtHandler.this.onHttpCallback(message);
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/50444bba/reef-examples/src/main/java/com/microsoft/reef/examples/hellohttp/HttpShellJobDriver.java
----------------------------------------------------------------------
diff --git a/reef-examples/src/main/java/com/microsoft/reef/examples/hellohttp/HttpShellJobDriver.java b/reef-examples/src/main/java/com/microsoft/reef/examples/hellohttp/HttpShellJobDriver.java
deleted file mode 100644
index 459a4d4..0000000
--- a/reef-examples/src/main/java/com/microsoft/reef/examples/hellohttp/HttpShellJobDriver.java
+++ /dev/null
@@ -1,369 +0,0 @@
-/**
- * Copyright (C) 2014 Microsoft Corporation
- *
- * Licensed 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.
- */
-
-package com.microsoft.reef.examples.hellohttp;
-
-import com.microsoft.reef.driver.context.ActiveContext;
-import com.microsoft.reef.driver.context.ClosedContext;
-import com.microsoft.reef.driver.context.ContextConfiguration;
-import com.microsoft.reef.driver.context.FailedContext;
-import com.microsoft.reef.driver.evaluator.AllocatedEvaluator;
-import com.microsoft.reef.driver.evaluator.EvaluatorRequest;
-import com.microsoft.reef.driver.evaluator.EvaluatorRequestor;
-import com.microsoft.reef.driver.evaluator.FailedEvaluator;
-import com.microsoft.reef.driver.task.CompletedTask;
-import com.microsoft.reef.driver.task.TaskConfiguration;
-import com.microsoft.tang.JavaConfigurationBuilder;
-import com.microsoft.tang.Tang;
-import com.microsoft.tang.annotations.Unit;
-import com.microsoft.tang.exceptions.BindException;
-import com.microsoft.wake.EventHandler;
-import com.microsoft.wake.remote.impl.ObjectSerializableCodec;
-import com.microsoft.wake.time.event.StartTime;
-import com.microsoft.wake.time.event.StopTime;
-
-import javax.inject.Inject;
-import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.logging.Level;
-import java.util.logging.Logger;
-
-/**
- * The Driver code for the Hello REEF Http Distributed Shell Application
- */
-@Unit
-public final class HttpShellJobDriver {
-
- private static final Logger LOG = Logger.getLogger(HttpShellJobDriver.class.getName());
-
- /**
- * Possible states of the job driver. Can be one of:
- * <dl>
- * <du><code>INIT</code></du><dd>initial state, ready to request the evaluators.</dd>
- * <du><code>WAIT_EVALUATORS</code></du><dd>Wait for requested evaluators to initialize.</dd>
- * <du><code>READY</code></du><dd>Ready to submitTask a new task.</dd>
- * <du><code>WAIT_TASKS</code></du><dd>Wait for tasks to complete.</dd>
- * </dl>
- */
- private enum State {
- INIT, WAIT_EVALUATORS, READY, WAIT_TASKS
- }
-
- /**
- * Job driver state.
- */
- private State state = State.INIT;
-
- /**
- * String codec is used to encode the results
- * before passing them back to the client.
- */
- public static final ObjectSerializableCodec<String> CODEC = new ObjectSerializableCodec<>();
-
- /**
- * First command to execute. Sometimes client can send us the first command
- * before Evaluators are available; we need to store this command here.
- */
- private String cmd;
-
- /**
- * Evaluator Requester
- */
- private final EvaluatorRequestor evaluatorRequestor;
-
- /**
- * Number of Evalutors to request (default is 1).
- */
- private final int numEvaluators = 2;
-
- /**
- * Shell execution results from each Evaluator.
- */
- private final List<String> results = new ArrayList<>();
-
- /**
- * Map from context ID to running evaluator context.
- */
- private final Map<String, ActiveContext> contexts = new HashMap<>();
-
- /**
- * Number of evaluators/tasks to complete.
- */
- private int expectCount = 0;
-
- /**
- * Callback handler for http return message
- */
- private HttpServerShellCmdtHandler.ClientCallBackHandler httpCallbackHandler;
-
- /**
- * Job Driver Constructor
- *
- * @param requestor
- * @param clientCallBackHandler
- */
- @Inject
- public HttpShellJobDriver(final EvaluatorRequestor requestor, final HttpServerShellCmdtHandler.ClientCallBackHandler clientCallBackHandler) {
- this.evaluatorRequestor = requestor;
- this.httpCallbackHandler = clientCallBackHandler;
- LOG.log(Level.FINE, "Instantiated 'HelloDriver'");
- }
-
- /**
- * Receive notification that an Evaluator had been allocated,
- * and submitTask a new Task in that Evaluator.
- */
- final class AllocatedEvaluatorHandler implements EventHandler<AllocatedEvaluator> {
- @Override
- public void onNext(final AllocatedEvaluator eval) {
- synchronized (HttpShellJobDriver.this) {
- LOG.log(Level.INFO, "Allocated Evaluator: {0} expect {1} running {2}",
- new Object[]{eval.getId(), HttpShellJobDriver.this.expectCount, HttpShellJobDriver.this.contexts.size()});
- assert (HttpShellJobDriver.this.state == State.WAIT_EVALUATORS);
- try {
- eval.submitContext(ContextConfiguration.CONF.set(
- ContextConfiguration.IDENTIFIER, eval.getId() + "_context").build());
- } catch (final BindException ex) {
- LOG.log(Level.SEVERE, "Failed to submit a context to evaluator: " + eval.getId(), ex);
- throw new RuntimeException(ex);
- }
- }
- }
- }
-
- /**
- * Receive notification that the entire Evaluator had failed.
- * Stop other jobs and pass this error to the job observer on the client.
- */
- final class FailedEvaluatorHandler implements EventHandler<FailedEvaluator> {
- @Override
- public void onNext(final FailedEvaluator eval) {
- synchronized (HttpShellJobDriver.this) {
- LOG.log(Level.SEVERE, "FailedEvaluator", eval);
- for (final FailedContext failedContext : eval.getFailedContextList()) {
- HttpShellJobDriver.this.contexts.remove(failedContext.getId());
- }
- throw new RuntimeException("Failed Evaluator: ", eval.getEvaluatorException());
- }
- }
- }
-
- /**
- * Receive notification that a new Context is available.
- * Submit a new Distributed Shell Task to that Context.
- */
- final class ActiveContextHandler implements EventHandler<ActiveContext> {
- @Override
- public void onNext(final ActiveContext context) {
- synchronized (HttpShellJobDriver.this) {
- LOG.log(Level.INFO, "Context available: {0} expect {1} state {2}",
- new Object[]{context.getId(), HttpShellJobDriver.this.expectCount, HttpShellJobDriver.this.state});
- assert (HttpShellJobDriver.this.state == State.WAIT_EVALUATORS);
- HttpShellJobDriver.this.contexts.put(context.getId(), context);
- if (--HttpShellJobDriver.this.expectCount <= 0) {
- HttpShellJobDriver.this.state = State.READY;
- if (HttpShellJobDriver.this.cmd == null) {
- LOG.log(Level.INFO, "All evaluators ready; waiting for command. State: {0}",
- HttpShellJobDriver.this.state);
- } else {
- HttpShellJobDriver.this.submit(HttpShellJobDriver.this.cmd);
- }
- }
- }
- }
- }
-
- /**
- * Receive notification that the Context had completed.
- * Remove context from the list of active context.
- */
- final class ClosedContextHandler implements EventHandler<ClosedContext> {
- @Override
- public void onNext(final ClosedContext context) {
- LOG.log(Level.INFO, "Completed Context: {0}", context.getId());
- synchronized (HttpShellJobDriver.this) {
- HttpShellJobDriver.this.contexts.remove(context.getId());
- }
- }
- }
-
- final class HttpClientCloseHandler implements EventHandler<Void> {
- @Override
- public void onNext(final Void aVoid) throws RuntimeException {
- LOG.log(Level.INFO, "Received a close message from the client. You can put code here to properly close drivers and evaluators.");
- for (final ActiveContext c : contexts.values()) {
- c.close();
- }
- }
- }
-
- /**
- * Receive notification that the Context had failed.
- * Remove context from the list of active context and notify the client.
- */
- final class FailedContextHandler implements EventHandler<FailedContext> {
- @Override
- public void onNext(final FailedContext context) {
- LOG.log(Level.SEVERE, "FailedContext", context);
- synchronized (HttpShellJobDriver.this) {
- HttpShellJobDriver.this.contexts.remove(context.getId());
- }
- throw new RuntimeException("Failed context: ", context.asError());
- }
- }
-
- /**
- * Receive notification that the Task has completed successfully.
- */
- final class CompletedTaskHandler implements EventHandler<CompletedTask> {
- @Override
- public void onNext(final CompletedTask task) {
- LOG.log(Level.INFO, "Completed task: {0}", task.getId());
- // Take the message returned by the task and add it to the running result.
- final String result = CODEC.decode(task.get());
- synchronized (HttpShellJobDriver.this) {
- HttpShellJobDriver.this.results.add(task.getId() + " :: " + result);
- LOG.log(Level.INFO, "Task {0} result {1}: {2} state: {3}", new Object[]{
- task.getId(), HttpShellJobDriver.this.results.size(), result, HttpShellJobDriver.this.state});
- if (--HttpShellJobDriver.this.expectCount <= 0) {
- HttpShellJobDriver.this.returnResults();
- HttpShellJobDriver.this.state = State.READY;
- if (HttpShellJobDriver.this.cmd != null) {
- HttpShellJobDriver.this.submit(HttpShellJobDriver.this.cmd);
- }
- }
- }
- }
- }
-
- /**
- * Construct the final result and forward it to the Client.
- */
- private void returnResults() {
- final StringBuilder sb = new StringBuilder();
- for (final String result : this.results) {
- sb.append(result);
- }
- this.results.clear();
- LOG.log(Level.INFO, "Return results to the client:\n{0}", sb);
- httpCallbackHandler.onNext(CODEC.encode(sb.toString()));
- }
-
- /**
- * Receive notification from the client.
- */
- final class ClientMessageHandler implements EventHandler<byte[]> {
- @Override
- public void onNext(final byte[] message) {
- synchronized (HttpShellJobDriver.this) {
- final String command = CODEC.decode(message);
- LOG.log(Level.INFO, "Client message: {0} state: {1}",
- new Object[]{command, HttpShellJobDriver.this.state});
- assert (HttpShellJobDriver.this.cmd == null);
- if (HttpShellJobDriver.this.state == State.READY) {
- HttpShellJobDriver.this.submit(command);
- } else {
- // not ready yet - save the command for better times.
- assert (HttpShellJobDriver.this.state == State.WAIT_EVALUATORS);
- HttpShellJobDriver.this.cmd = command;
- }
- }
- }
- }
-
- /**
- * Submit command to all available evaluators.
- *
- * @param command shell command to execute.
- */
- private void submit(final String command) {
- LOG.log(Level.INFO, "Submit command {0} to {1} evaluators. state: {2}",
- new Object[]{command, this.contexts.size(), this.state});
- assert (this.state == State.READY);
- this.expectCount = this.contexts.size();
- this.state = State.WAIT_TASKS;
- this.cmd = null;
- for (final ActiveContext context : this.contexts.values()) {
- this.submit(context, command);
- }
- }
-
- /**
- * Submit a Task that execute the command to a single Evaluator.
- * This method is called from <code>submitTask(cmd)</code>.
- */
- private void submit(final ActiveContext context, final String command) {
- try {
- LOG.log(Level.INFO, "Send command {0} to context: {1}", new Object[]{command, context});
- final JavaConfigurationBuilder cb = Tang.Factory.getTang().newConfigurationBuilder();
- cb.addConfiguration(
- TaskConfiguration.CONF
- .set(TaskConfiguration.IDENTIFIER, context.getId() + "_task")
- .set(TaskConfiguration.TASK, ShellTask.class)
- .build()
- );
- cb.bindNamedParameter(com.microsoft.reef.examples.hellohttp.Command.class, command);
- context.submitTask(cb.build());
- } catch (final BindException ex) {
- LOG.log(Level.SEVERE, "Bad Task configuration for context: " + context.getId(), ex);
- context.close();
- throw new RuntimeException(ex);
- }
- }
-
- /**
- * Job Driver is ready and the clock is set up: request the evaluators.
- */
- final class StartHandler implements EventHandler<StartTime> {
- @Override
- public void onNext(final StartTime startTime) {
- LOG.log(Level.INFO, "{0} StartTime: {1}", new Object[]{state, startTime});
- assert (state == State.INIT);
- requestEvaluators();
- }
- }
-
- /**
- * Request the evaluators.
- */
- private synchronized void requestEvaluators() {
- assert (this.state == State.INIT);
- LOG.log(Level.INFO, "Schedule on {0} Evaluators.", this.numEvaluators);
- this.evaluatorRequestor.submit(
- EvaluatorRequest.newBuilder()
- .setMemory(128)
- .setNumberOfCores(1)
- .setNumber(this.numEvaluators).build()
- );
- this.state = State.WAIT_EVALUATORS;
- this.expectCount = this.numEvaluators;
- }
-
- /**
- * Shutting down the job driver: close the evaluators.
- */
- final class StopHandler implements EventHandler<StopTime> {
- @Override
- public void onNext(final StopTime time) {
- LOG.log(Level.INFO, "{0} StopTime: {1}", new Object[]{state, time});
- for (final ActiveContext context : contexts.values()) {
- context.close();
- }
- }
- }
-}