You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@reef.apache.org by bg...@apache.org on 2015/08/04 08:47:01 UTC
[2/2] incubator-reef git commit: [REEF-491] Initial Vortex
implementation This addressed the issue by * Creating a Java ThreadPool-like
user interface * Enabling distributed execution of user-submitted tasks *
Developing a mechanism for handling
[REEF-491] Initial Vortex implementation
This addressed the issue by
* Creating a Java ThreadPool-like user interface
* Enabling distributed execution of user-submitted tasks
* Developing a mechanism for handling of resource preemption events
* Developing simple task scheduling mechanisms
* Creating simple unit/integration tests to verify correctness.
JIRA:
[REEF-491](https://issues.apache.org/jira/browse/REEF-491)
Pull Request:
Closes #310
Project: http://git-wip-us.apache.org/repos/asf/incubator-reef/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-reef/commit/2c994ec9
Tree: http://git-wip-us.apache.org/repos/asf/incubator-reef/tree/2c994ec9
Diff: http://git-wip-us.apache.org/repos/asf/incubator-reef/diff/2c994ec9
Branch: refs/heads/master
Commit: 2c994ec9714c2ce30f237c1b808e6335019e754b
Parents: 0181e2c
Author: John Yang <jo...@apache.org>
Authored: Tue Jun 16 10:46:43 2015 +0800
Committer: Byung-Gon Chun <bg...@apache.org>
Committed: Tue Aug 4 15:41:07 2015 +0900
----------------------------------------------------------------------
lang/java/reef-applications/pom.xml | 8 -
.../reef-applications/reef-vortex/README.md | 99 +++++++
lang/java/reef-applications/reef-vortex/pom.xml | 62 ++++-
.../org/apache/reef/vortex/VortexDriver.java | 60 -----
.../apache/reef/vortex/api/VortexFunction.java | 43 ++++
.../apache/reef/vortex/api/VortexFuture.java | 106 ++++++++
.../org/apache/reef/vortex/api/VortexStart.java | 34 +++
.../reef/vortex/api/VortexThreadPool.java | 50 ++++
.../apache/reef/vortex/api/package-info.java | 22 ++
.../vortex/common/TaskletExecutionRequest.java | 68 +++++
.../vortex/common/TaskletFailureReport.java | 61 +++++
.../reef/vortex/common/TaskletResultReport.java | 64 +++++
.../reef/vortex/common/VortexRequest.java | 41 +++
.../apache/reef/vortex/common/WorkerReport.java | 42 +++
.../apache/reef/vortex/common/package-info.java | 22 ++
.../reef/vortex/driver/DefaultVortexMaster.java | 108 ++++++++
.../vortex/driver/PendingTaskletScheduler.java | 59 +++++
.../reef/vortex/driver/PendingTasklets.java | 51 ++++
.../reef/vortex/driver/RunningWorkers.java | 257 +++++++++++++++++++
.../org/apache/reef/vortex/driver/Tasklet.java | 96 +++++++
.../reef/vortex/driver/VortexConfHelper.java | 66 +++++
.../apache/reef/vortex/driver/VortexDriver.java | 193 ++++++++++++++
.../reef/vortex/driver/VortexLauncher.java | 62 +++++
.../apache/reef/vortex/driver/VortexMaster.java | 67 +++++
.../reef/vortex/driver/VortexMasterConf.java | 98 +++++++
.../reef/vortex/driver/VortexRequestor.java | 50 ++++
.../reef/vortex/driver/VortexStartExecutor.java | 52 ++++
.../reef/vortex/driver/VortexWorkerConf.java | 53 ++++
.../reef/vortex/driver/VortexWorkerManager.java | 95 +++++++
.../apache/reef/vortex/driver/package-info.java | 22 ++
.../reef/vortex/evaluator/VortexWorker.java | 160 ++++++++++++
.../reef/vortex/evaluator/package-info.java | 22 ++
.../reef/vortex/examples/addone/AddOne.java | 36 +++
.../vortex/examples/addone/AddOneFunction.java | 34 +++
.../vortex/examples/addone/AddOneStart.java | 67 +++++
.../vortex/examples/addone/package-info.java | 22 ++
.../reef/vortex/examples/hello/HelloVortex.java | 36 +++
.../examples/hello/HelloVortexFunction.java | 37 +++
.../vortex/examples/hello/HelloVortexStart.java | 48 ++++
.../vortex/examples/hello/package-info.java | 22 ++
.../org/apache/reef/vortex/package-info.java | 2 +-
.../vortex/driver/DefaultVortexMasterTest.java | 158 ++++++++++++
.../reef/vortex/driver/RunningWorkersTest.java | 64 +++++
.../org/apache/reef/vortex/driver/TestUtil.java | 66 +++++
.../apache/reef/vortex/driver/package-info.java | 22 ++
lang/java/reef-tests/pom.xml | 5 +
.../org/apache/reef/tests/AllTestsSuite.java | 4 +-
.../applications/ApplicationTestSuite.java | 30 +++
.../reef/tests/applications/package-info.java | 22 ++
.../applications/vortex/VortexTestSuite.java | 30 +++
.../vortex/addone/AddOneFunction.java | 34 +++
.../applications/vortex/addone/AddOneTest.java | 63 +++++
.../vortex/addone/AddOneTestStart.java | 67 +++++
.../vortex/addone/package-info.java | 22 ++
.../tests/applications/vortex/package-info.java | 22 ++
55 files changed, 3135 insertions(+), 71 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/2c994ec9/lang/java/reef-applications/pom.xml
----------------------------------------------------------------------
diff --git a/lang/java/reef-applications/pom.xml b/lang/java/reef-applications/pom.xml
index 828ace3..b7ddb2c 100644
--- a/lang/java/reef-applications/pom.xml
+++ b/lang/java/reef-applications/pom.xml
@@ -31,14 +31,6 @@ under the License.
<relativePath>../../..</relativePath>
</parent>
- <dependencies>
- <dependency>
- <groupId>${project.groupId}</groupId>
- <artifactId>reef-common</artifactId>
- <version>${project.version}</version>
- </dependency>
- </dependencies>
-
<modules>
<module>reef-vortex</module>
</modules>
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/2c994ec9/lang/java/reef-applications/reef-vortex/README.md
----------------------------------------------------------------------
diff --git a/lang/java/reef-applications/reef-vortex/README.md b/lang/java/reef-applications/reef-vortex/README.md
new file mode 100644
index 0000000..0ea6518
--- /dev/null
+++ b/lang/java/reef-applications/reef-vortex/README.md
@@ -0,0 +1,99 @@
+Vortex
+========================
+Vortex is a runtime for low-priority jobs to effectively cope
+with non-cooperative(thus low-latency) preemptions with the following goals:
+
+* Enable high-priority jobs such as latency-sensitive online serving jobs to reclaim resources quickly
+* Enable low-priority jobs such as offline batch data analytics jobs to efficiently utilize volatile resources
+
+Vortex is currently under [active development](https://issues.apache.org/jira/browse/REEF-364)
+
+Key Components
+========================
+* ThreadPool API
+ * Users can submit custom functions and inputs to Vortex and retrieve results
+ * This API hides the details of distributed execution on unreliable resources(e.g. handling of preemption)
+* Tasklet
+ * User-submitted function and input are translated into Tasklet(s) in VortexMaster
+ * A Tasklet is then scheduled to and executed by a VortexWorker
+* Queue
+ * VortexMaster Queue(in REEF Driver): Keeps track of Tasklets waiting to be scheduled/enqueued to a VortexWorker Queue
+ * VortexWorker Queue(in REEF Evaluator): Keeps track of Tasklets waiting to be executed
+
+Example: Vector Calculation on Vortex
+====================
+```java
+/**
+ * User's main function.
+ */
+public final class AddOne {
+ private AddOne() {
+ }
+
+ public static void main(final String[] args) {
+ VortexLauncher.launchLocal("Vortex_Example_AddOne", AddOneStart.class, 2, 1024, 4);
+ }
+}
+```
+
+* Through `VortexLauncher` user launches a Vortex job, passing following arguments
+ * Name of the job
+ * Implementation of VortexStart(`AddOneStart`)
+ * Amount of resources to use
+
+```java
+/**
+ * AddOne User Code Example.
+ */
+final class AddOneStart implements VortexStart {
+ @Inject
+ public AddOneStart() {
+ }
+
+ @Override
+ public void start(final VortexThreadPool vortexThreadPool) {
+ final Vector<Integer> inputVector = new Vector<>();
+ for (int i = 0; i < 1000; i++) {
+ inputVector.add(i);
+ }
+
+ final List<VortexFuture<Integer>> futures = new ArrayList<>();
+ final AddOneFunction addOneFunction = new AddOneFunction();
+ for (final int i : inputVector) {
+ futures.add(vortexThreadPool.submit(addOneFunction, i));
+ }
+
+ final Vector<Integer> outputVector = new Vector<>();
+ for (final VortexFuture<Integer> future : futures) {
+ try {
+ outputVector.add(future.get());
+ } catch (InterruptedException | ExecutionException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ System.out.println("RESULT:");
+ System.out.println(outputVector);
+ }
+}
+```
+
+* Through `VortexThreadPool#submit`, user submits a custom function(`AddOneFunction`) and its input to be executed on Vortex runtime
+* Using returned `VortexFuture`, user retrieves execution results
+
+```java
+/**
+ * Outputs input + 1.
+ */
+final class AddOneFunction implements VortexFunction<Integer, Integer> {
+ @Override
+ public Integer call(final Integer input) throws Exception {
+ return input + 1;
+ }
+}
+```
+* User implementation of VortexFunction, takes an integer as the input and outputs input+1
+
+
+
+
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/2c994ec9/lang/java/reef-applications/reef-vortex/pom.xml
----------------------------------------------------------------------
diff --git a/lang/java/reef-applications/reef-vortex/pom.xml b/lang/java/reef-applications/reef-vortex/pom.xml
index 175c4a5..2f47ab2 100644
--- a/lang/java/reef-applications/reef-vortex/pom.xml
+++ b/lang/java/reef-applications/reef-vortex/pom.xml
@@ -20,7 +20,7 @@ under the License.
<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">
<modelVersion>4.0.0</modelVersion>
- <artifactId>reef-vortex</artifactId>
+ <artifactId>vortex</artifactId>
<name>REEF Vortex</name>
<parent>
@@ -28,4 +28,64 @@ under the License.
<artifactId>reef-applications</artifactId>
<version>0.12.0-incubating-SNAPSHOT</version>
</parent>
+
+ <dependencies>
+ <dependency>
+ <groupId>${project.groupId}</groupId>
+ <artifactId>reef-common</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>junit</groupId>
+ <artifactId>junit</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.mockito</groupId>
+ <artifactId>mockito-core</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>${project.groupId}</groupId>
+ <artifactId>reef-runtime-local</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+ </dependencies>
+
+ <build>
+ <plugins>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-shade-plugin</artifactId>
+ <executions>
+ <execution>
+ <phase>package</phase>
+ <goals>
+ <goal>shade</goal>
+ </goals>
+ </execution>
+ </executions>
+ <configuration>
+ <outputFile>
+ ${project.build.directory}/${project.artifactId}-${project.version}-shaded.jar
+ </outputFile>
+ </configuration>
+ </plugin>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-checkstyle-plugin</artifactId>
+ <version>${maven-checkstyle-plugin.version}</version>
+ <executions>
+ <execution>
+ <id>validate</id>
+ <phase>validate</phase>
+ <goals>
+ <goal>check</goal>
+ <goal>checkstyle</goal>
+ </goals>
+ </execution>
+ </executions>
+ </plugin>
+ </plugins>
+ </build>
</project>
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/2c994ec9/lang/java/reef-applications/reef-vortex/src/main/java/org/apache/reef/vortex/VortexDriver.java
----------------------------------------------------------------------
diff --git a/lang/java/reef-applications/reef-vortex/src/main/java/org/apache/reef/vortex/VortexDriver.java b/lang/java/reef-applications/reef-vortex/src/main/java/org/apache/reef/vortex/VortexDriver.java
deleted file mode 100644
index d3a4c70..0000000
--- a/lang/java/reef-applications/reef-vortex/src/main/java/org/apache/reef/vortex/VortexDriver.java
+++ /dev/null
@@ -1,60 +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.
- */
-package org.apache.reef.vortex;
-
-import org.apache.reef.driver.evaluator.AllocatedEvaluator;
-import org.apache.reef.driver.evaluator.EvaluatorRequest;
-import org.apache.reef.driver.evaluator.EvaluatorRequestor;
-import org.apache.reef.tang.annotations.Unit;
-import org.apache.reef.wake.EventHandler;
-import org.apache.reef.wake.time.event.StartTime;
-
-import javax.inject.Inject;
-import java.util.logging.Level;
-import java.util.logging.Logger;
-
-@Unit
-public final class VortexDriver {
- private static final Logger LOG = Logger.getLogger(VortexDriver.class.getName());
- private final EvaluatorRequestor requestor;
-
- @Inject
- public VortexDriver(final EvaluatorRequestor requestor) {
- this.requestor = requestor;
- }
-
- public final class StartHandler implements EventHandler<StartTime> {
- @Override
- public void onNext(final StartTime startTime) {
- requestor.submit(EvaluatorRequest.newBuilder()
- .setNumber(1)
- .setMemory(64)
- .setNumberOfCores(1)
- .build());
- }
- }
-
- public final class EvaluatorAllocatedHandler implements EventHandler<AllocatedEvaluator> {
- @Override
- public void onNext(final AllocatedEvaluator allocatedEvaluator) {
- LOG.log(Level.INFO, "Evaluator Allocated: {0}", allocatedEvaluator);
- allocatedEvaluator.close();
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/2c994ec9/lang/java/reef-applications/reef-vortex/src/main/java/org/apache/reef/vortex/api/VortexFunction.java
----------------------------------------------------------------------
diff --git a/lang/java/reef-applications/reef-vortex/src/main/java/org/apache/reef/vortex/api/VortexFunction.java b/lang/java/reef-applications/reef-vortex/src/main/java/org/apache/reef/vortex/api/VortexFunction.java
new file mode 100644
index 0000000..96e47b6
--- /dev/null
+++ b/lang/java/reef-applications/reef-vortex/src/main/java/org/apache/reef/vortex/api/VortexFunction.java
@@ -0,0 +1,43 @@
+/*
+ * 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.
+ */
+package org.apache.reef.vortex.api;
+
+import org.apache.reef.annotations.Unstable;
+
+import java.io.Serializable;
+
+/**
+ * Typed user function.
+ * Implement your functions using this interface.
+ * TODO[REEF-504]: Clean up Serializable in Vortex.
+ *
+ * @param <TInput> input type
+ * @param <TOutput> output type
+ */
+@Unstable
+public interface VortexFunction<TInput extends Serializable, TOutput extends Serializable> extends Serializable {
+ /**
+ * @param input of the function
+ * @return output of the function
+ * @throws Exception thrown here will bubble up in VortexFuture#get as ExecutionException
+ * Exception should be thrown only after all resources are released as they cannot be cleared by Vortex
+ * For example if threads are spawned here, shut them down before throwing an exception
+ */
+ TOutput call(TInput input) throws Exception;
+}
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/2c994ec9/lang/java/reef-applications/reef-vortex/src/main/java/org/apache/reef/vortex/api/VortexFuture.java
----------------------------------------------------------------------
diff --git a/lang/java/reef-applications/reef-vortex/src/main/java/org/apache/reef/vortex/api/VortexFuture.java b/lang/java/reef-applications/reef-vortex/src/main/java/org/apache/reef/vortex/api/VortexFuture.java
new file mode 100644
index 0000000..09c9caf
--- /dev/null
+++ b/lang/java/reef-applications/reef-vortex/src/main/java/org/apache/reef/vortex/api/VortexFuture.java
@@ -0,0 +1,106 @@
+/*
+ * 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.
+ */
+package org.apache.reef.vortex.api;
+
+import org.apache.reef.annotations.Unstable;
+
+import java.util.concurrent.*;
+
+/**
+ * The interface between user code and submitted task.
+ * TODO[REEF-505]: Callback features for VortexFuture.
+ */
+@Unstable
+public final class VortexFuture<TOutput> implements Future<TOutput> {
+ private TOutput userResult;
+ private Exception userException;
+ private final CountDownLatch countDownLatch = new CountDownLatch(1);
+
+ /**
+ * TODO[REEF-502]: Support Vortex Tasklet(s) cancellation by user.
+ */
+ @Override
+ public boolean cancel(final boolean mayInterruptIfRunning) {
+ throw new UnsupportedOperationException("Cancel not yet supported");
+ }
+
+ /**
+ * TODO[REEF-502]: Support Vortex Tasklet(s) cancellation by user.
+ */
+ @Override
+ public boolean isCancelled() {
+ throw new UnsupportedOperationException("Cancel not yet supported");
+ }
+
+ /**
+ * @return true it the task completed, false if not.
+ */
+ @Override
+ public boolean isDone() {
+ return countDownLatch.getCount() == 0;
+ }
+
+ /**
+ * Infinitely wait for the result of the task.
+ */
+ @Override
+ public TOutput get() throws InterruptedException, ExecutionException {
+ countDownLatch.await();
+ if (userResult != null) {
+ return userResult;
+ } else {
+ assert(userException != null);
+ throw new ExecutionException(userException);
+ }
+ }
+
+ /**
+ * Wait a certain period of time for the result of the task.
+ */
+ @Override
+ public TOutput get(final long timeout, final TimeUnit unit)
+ throws InterruptedException, ExecutionException, TimeoutException {
+ if (!countDownLatch.await(timeout, unit)) {
+ throw new TimeoutException();
+ }
+
+ if (userResult != null) {
+ return userResult;
+ } else {
+ assert(userException != null);
+ throw new ExecutionException(userException);
+ }
+ }
+
+ /**
+ * Called by VortexMaster to let the user know that the task completed.
+ */
+ public void completed(final TOutput result) {
+ this.userResult = result;
+ this.countDownLatch.countDown();
+ }
+
+ /**
+ * Called by VortexMaster to let the user know that the task threw an exception.
+ */
+ public void threwException(final Exception exception) {
+ this.userException = exception;
+ this.countDownLatch.countDown();
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/2c994ec9/lang/java/reef-applications/reef-vortex/src/main/java/org/apache/reef/vortex/api/VortexStart.java
----------------------------------------------------------------------
diff --git a/lang/java/reef-applications/reef-vortex/src/main/java/org/apache/reef/vortex/api/VortexStart.java b/lang/java/reef-applications/reef-vortex/src/main/java/org/apache/reef/vortex/api/VortexStart.java
new file mode 100644
index 0000000..50f9308
--- /dev/null
+++ b/lang/java/reef-applications/reef-vortex/src/main/java/org/apache/reef/vortex/api/VortexStart.java
@@ -0,0 +1,34 @@
+/*
+ * 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.
+ */
+package org.apache.reef.vortex.api;
+
+import org.apache.reef.annotations.Unstable;
+
+/**
+ * The starting point of Vortex user thread.
+ */
+@Unstable
+public interface VortexStart {
+ /**
+ * Implement this method, using VortexThreadPool to run jobs on Vortex.
+ * Your implementation of this interface will be instantiated by Tang
+ * and this method will be called upon the start of REEF Driver.
+ */
+ void start(final VortexThreadPool vortexThreadPool);
+}
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/2c994ec9/lang/java/reef-applications/reef-vortex/src/main/java/org/apache/reef/vortex/api/VortexThreadPool.java
----------------------------------------------------------------------
diff --git a/lang/java/reef-applications/reef-vortex/src/main/java/org/apache/reef/vortex/api/VortexThreadPool.java b/lang/java/reef-applications/reef-vortex/src/main/java/org/apache/reef/vortex/api/VortexThreadPool.java
new file mode 100644
index 0000000..25c5c90
--- /dev/null
+++ b/lang/java/reef-applications/reef-vortex/src/main/java/org/apache/reef/vortex/api/VortexThreadPool.java
@@ -0,0 +1,50 @@
+/*
+ * 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.
+ */
+package org.apache.reef.vortex.api;
+
+import org.apache.reef.annotations.Unstable;
+import org.apache.reef.vortex.driver.VortexMaster;
+
+import javax.inject.Inject;
+import java.io.Serializable;
+
+/**
+ * Distributed thread pool.
+ */
+@Unstable
+public final class VortexThreadPool {
+ private final VortexMaster vortexMaster;
+
+ @Inject
+ private VortexThreadPool(final VortexMaster vortexMaster) {
+ this.vortexMaster = vortexMaster;
+ }
+
+ /**
+ * @param function to run on Vortex
+ * @param input of the function
+ * @param <TInput> input type
+ * @param <TOutput> output type
+ * @return VortexFuture for tracking execution progress
+ */
+ public <TInput extends Serializable, TOutput extends Serializable> VortexFuture<TOutput>
+ submit(final VortexFunction<TInput, TOutput> function, final TInput input) {
+ return vortexMaster.enqueueTasklet(function, input);
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/2c994ec9/lang/java/reef-applications/reef-vortex/src/main/java/org/apache/reef/vortex/api/package-info.java
----------------------------------------------------------------------
diff --git a/lang/java/reef-applications/reef-vortex/src/main/java/org/apache/reef/vortex/api/package-info.java b/lang/java/reef-applications/reef-vortex/src/main/java/org/apache/reef/vortex/api/package-info.java
new file mode 100644
index 0000000..9a7633e
--- /dev/null
+++ b/lang/java/reef-applications/reef-vortex/src/main/java/org/apache/reef/vortex/api/package-info.java
@@ -0,0 +1,22 @@
+/*
+ * 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.
+ */
+/**
+ * Vortex user api for running jobs on Vortex.
+ */
+package org.apache.reef.vortex.api;
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/2c994ec9/lang/java/reef-applications/reef-vortex/src/main/java/org/apache/reef/vortex/common/TaskletExecutionRequest.java
----------------------------------------------------------------------
diff --git a/lang/java/reef-applications/reef-vortex/src/main/java/org/apache/reef/vortex/common/TaskletExecutionRequest.java b/lang/java/reef-applications/reef-vortex/src/main/java/org/apache/reef/vortex/common/TaskletExecutionRequest.java
new file mode 100644
index 0000000..2c60c8a
--- /dev/null
+++ b/lang/java/reef-applications/reef-vortex/src/main/java/org/apache/reef/vortex/common/TaskletExecutionRequest.java
@@ -0,0 +1,68 @@
+/*
+ * 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.
+ */
+package org.apache.reef.vortex.common;
+
+import org.apache.reef.annotations.Unstable;
+import org.apache.reef.vortex.api.VortexFunction;
+
+import java.io.Serializable;
+
+/**
+ * Request to execute a tasklet.
+ */
+@Unstable
+public final class TaskletExecutionRequest<TInput extends Serializable, TOutput extends Serializable>
+ implements VortexRequest {
+ private final int taskletId;
+ private final VortexFunction<TInput, TOutput> userFunction;
+ private final TInput input;
+
+ /**
+ * @return the type of this VortexRequest.
+ */
+ @Override
+ public RequestType getType() {
+ return RequestType.ExecuteTasklet;
+ }
+
+ /**
+ * Vortex Master -> Vortex Worker request to execute a tasklet.
+ */
+ public TaskletExecutionRequest(final int taskletId,
+ final VortexFunction<TInput, TOutput> userFunction,
+ final TInput input) {
+ this.taskletId = taskletId;
+ this.userFunction = userFunction;
+ this.input = input;
+ }
+
+ /**
+ * Execute the function using the input.
+ */
+ public TOutput execute() throws Exception {
+ return userFunction.call(input);
+ }
+
+ /**
+ * Get id of the tasklet.
+ */
+ public int getTaskletId() {
+ return taskletId;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/2c994ec9/lang/java/reef-applications/reef-vortex/src/main/java/org/apache/reef/vortex/common/TaskletFailureReport.java
----------------------------------------------------------------------
diff --git a/lang/java/reef-applications/reef-vortex/src/main/java/org/apache/reef/vortex/common/TaskletFailureReport.java b/lang/java/reef-applications/reef-vortex/src/main/java/org/apache/reef/vortex/common/TaskletFailureReport.java
new file mode 100644
index 0000000..dc847e7
--- /dev/null
+++ b/lang/java/reef-applications/reef-vortex/src/main/java/org/apache/reef/vortex/common/TaskletFailureReport.java
@@ -0,0 +1,61 @@
+/*
+ * 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.
+ */
+package org.apache.reef.vortex.common;
+
+import org.apache.reef.annotations.Unstable;
+
+/**
+ * Report of a tasklet exception.
+ */
+@Unstable
+public final class TaskletFailureReport implements WorkerReport {
+ private final int taskletId;
+ private final Exception exception;
+
+ /**
+ * @param taskletId of the failed tasklet.
+ * @param exception that caused the tasklet failure.
+ */
+ public TaskletFailureReport(final int taskletId, final Exception exception) {
+ this.taskletId = taskletId;
+ this.exception = exception;
+ }
+
+ /**
+ * @return the type of this WorkerReport.
+ */
+ @Override
+ public WorkerReportType getType() {
+ return WorkerReportType.TaskletFailure;
+ }
+
+ /**
+ * @return the id of the tasklet.
+ */
+ public int getTaskletId() {
+ return taskletId;
+ }
+
+ /**
+ * @return the exception that caused the tasklet failure.
+ */
+ public Exception getException() {
+ return exception;
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/2c994ec9/lang/java/reef-applications/reef-vortex/src/main/java/org/apache/reef/vortex/common/TaskletResultReport.java
----------------------------------------------------------------------
diff --git a/lang/java/reef-applications/reef-vortex/src/main/java/org/apache/reef/vortex/common/TaskletResultReport.java b/lang/java/reef-applications/reef-vortex/src/main/java/org/apache/reef/vortex/common/TaskletResultReport.java
new file mode 100644
index 0000000..0a6fab3
--- /dev/null
+++ b/lang/java/reef-applications/reef-vortex/src/main/java/org/apache/reef/vortex/common/TaskletResultReport.java
@@ -0,0 +1,64 @@
+/*
+ * 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.
+ */
+package org.apache.reef.vortex.common;
+
+import org.apache.reef.annotations.Unstable;
+
+import java.io.Serializable;
+
+/**
+ * Report of a tasklet execution result.
+ */
+@Unstable
+public final class TaskletResultReport<TOutput extends Serializable> implements WorkerReport {
+ private final int taskletId;
+ private final TOutput result;
+
+ /**
+ * @param taskletId of the tasklet.
+ * @param result of the tasklet execution.
+ */
+ public TaskletResultReport(final int taskletId, final TOutput result) {
+ this.taskletId = taskletId;
+ this.result = result;
+ }
+
+ /**
+ * @return the type of this WorkerReport.
+ */
+ @Override
+ public WorkerReportType getType() {
+ return WorkerReportType.TaskletResult;
+ }
+
+ /**
+ * @return the id of the tasklet.
+ */
+ public int getTaskletId() {
+ return taskletId;
+ }
+
+ /**
+ * @return the result of the tasklet execution.
+ */
+ public TOutput getResult() {
+ return result;
+ }
+
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/2c994ec9/lang/java/reef-applications/reef-vortex/src/main/java/org/apache/reef/vortex/common/VortexRequest.java
----------------------------------------------------------------------
diff --git a/lang/java/reef-applications/reef-vortex/src/main/java/org/apache/reef/vortex/common/VortexRequest.java b/lang/java/reef-applications/reef-vortex/src/main/java/org/apache/reef/vortex/common/VortexRequest.java
new file mode 100644
index 0000000..967ab98
--- /dev/null
+++ b/lang/java/reef-applications/reef-vortex/src/main/java/org/apache/reef/vortex/common/VortexRequest.java
@@ -0,0 +1,41 @@
+/*
+ * 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.
+ */
+package org.apache.reef.vortex.common;
+
+import org.apache.reef.annotations.Unstable;
+
+import java.io.Serializable;
+
+/**
+ * Master -> Worker protocol.
+ */
+@Unstable
+public interface VortexRequest extends Serializable {
+ /**
+ * Type of Request.
+ */
+ enum RequestType {
+ ExecuteTasklet
+ }
+
+ /**
+ * @return the type of this VortexRequest.
+ */
+ RequestType getType();
+}
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/2c994ec9/lang/java/reef-applications/reef-vortex/src/main/java/org/apache/reef/vortex/common/WorkerReport.java
----------------------------------------------------------------------
diff --git a/lang/java/reef-applications/reef-vortex/src/main/java/org/apache/reef/vortex/common/WorkerReport.java b/lang/java/reef-applications/reef-vortex/src/main/java/org/apache/reef/vortex/common/WorkerReport.java
new file mode 100644
index 0000000..4444f3c
--- /dev/null
+++ b/lang/java/reef-applications/reef-vortex/src/main/java/org/apache/reef/vortex/common/WorkerReport.java
@@ -0,0 +1,42 @@
+/*
+ * 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.
+ */
+package org.apache.reef.vortex.common;
+
+import org.apache.reef.annotations.Unstable;
+
+import java.io.Serializable;
+
+/**
+ * Worker -> Master protocol.
+ */
+@Unstable
+public interface WorkerReport extends Serializable {
+ /**
+ * Type of WorkerReport.
+ */
+ enum WorkerReportType {
+ TaskletResult,
+ TaskletFailure
+ }
+
+ /**
+ * @return the type of this WorkerReport.
+ */
+ WorkerReportType getType();
+}
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/2c994ec9/lang/java/reef-applications/reef-vortex/src/main/java/org/apache/reef/vortex/common/package-info.java
----------------------------------------------------------------------
diff --git a/lang/java/reef-applications/reef-vortex/src/main/java/org/apache/reef/vortex/common/package-info.java b/lang/java/reef-applications/reef-vortex/src/main/java/org/apache/reef/vortex/common/package-info.java
new file mode 100644
index 0000000..35cf59e
--- /dev/null
+++ b/lang/java/reef-applications/reef-vortex/src/main/java/org/apache/reef/vortex/common/package-info.java
@@ -0,0 +1,22 @@
+/*
+ * 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.
+ */
+/**
+ * Vortex Code used both in Vortex Driver and Vortex Evaluator.
+ */
+package org.apache.reef.vortex.common;
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/2c994ec9/lang/java/reef-applications/reef-vortex/src/main/java/org/apache/reef/vortex/driver/DefaultVortexMaster.java
----------------------------------------------------------------------
diff --git a/lang/java/reef-applications/reef-vortex/src/main/java/org/apache/reef/vortex/driver/DefaultVortexMaster.java b/lang/java/reef-applications/reef-vortex/src/main/java/org/apache/reef/vortex/driver/DefaultVortexMaster.java
new file mode 100644
index 0000000..f1a55b7
--- /dev/null
+++ b/lang/java/reef-applications/reef-vortex/src/main/java/org/apache/reef/vortex/driver/DefaultVortexMaster.java
@@ -0,0 +1,108 @@
+/*
+ * 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.
+ */
+package org.apache.reef.vortex.driver;
+
+import net.jcip.annotations.ThreadSafe;
+import org.apache.reef.annotations.audience.DriverSide;
+import org.apache.reef.vortex.api.VortexFunction;
+import org.apache.reef.vortex.api.VortexFuture;
+
+import javax.inject.Inject;
+import java.io.Serializable;
+import java.util.*;
+import java.util.concurrent.atomic.AtomicInteger;
+
+/**
+ * Default implementation of VortexMaster.
+ * Uses two thread-safe data structures(pendingTasklets, runningWorkers) in implementing VortexMaster interface.
+ */
+@ThreadSafe
+@DriverSide
+final class DefaultVortexMaster implements VortexMaster {
+ private final AtomicInteger taskletIdCounter = new AtomicInteger();
+ private final RunningWorkers runningWorkers;
+ private final PendingTasklets pendingTasklets;
+
+ /**
+ * @param runningWorkers for managing all running workers.
+ */
+ @Inject
+ DefaultVortexMaster(final RunningWorkers runningWorkers,
+ final PendingTasklets pendingTasklets) {
+ this.runningWorkers = runningWorkers;
+ this.pendingTasklets = pendingTasklets;
+ }
+
+ /**
+ * Add a new tasklet to pendingTasklets.
+ */
+ @Override
+ public <TInput extends Serializable, TOutput extends Serializable> VortexFuture<TOutput>
+ enqueueTasklet(final VortexFunction<TInput, TOutput> function, final TInput input) {
+ // TODO[REEF-500]: Simple duplicate Vortex Tasklet launch.
+ final VortexFuture<TOutput> vortexFuture = new VortexFuture<>();
+ this.pendingTasklets.addLast(new Tasklet<>(taskletIdCounter.getAndIncrement(), function, input, vortexFuture));
+ return vortexFuture;
+ }
+
+ /**
+ * Add a new worker to runningWorkers.
+ */
+ @Override
+ public void workerAllocated(final VortexWorkerManager vortexWorkerManager) {
+ runningWorkers.addWorker(vortexWorkerManager);
+ }
+
+ /**
+ * Remove the worker from runningWorkers and add back the lost tasklets to pendingTasklets.
+ */
+ @Override
+ public void workerPreempted(final String id) {
+ final Collection<Tasklet> preemptedTasklets = runningWorkers.removeWorker(id);
+ for (final Tasklet tasklet : preemptedTasklets) {
+ pendingTasklets.addFirst(tasklet);
+ }
+ }
+
+ /**
+ * Notify task completion to runningWorkers.
+ */
+ @Override
+ public void taskletCompleted(final String workerId,
+ final int taskletId,
+ final Serializable result) {
+ runningWorkers.completeTasklet(workerId, taskletId, result);
+ }
+
+ /**
+ * Notify task failure to runningWorkers.
+ */
+ @Override
+ public void taskletErrored(final String workerId, final int taskletId, final Exception exception) {
+ runningWorkers.errorTasklet(workerId, taskletId, exception);
+ }
+
+ /**
+ * Terminate the job.
+ */
+ @Override
+ public void terminate() {
+ runningWorkers.terminate();
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/2c994ec9/lang/java/reef-applications/reef-vortex/src/main/java/org/apache/reef/vortex/driver/PendingTaskletScheduler.java
----------------------------------------------------------------------
diff --git a/lang/java/reef-applications/reef-vortex/src/main/java/org/apache/reef/vortex/driver/PendingTaskletScheduler.java b/lang/java/reef-applications/reef-vortex/src/main/java/org/apache/reef/vortex/driver/PendingTaskletScheduler.java
new file mode 100644
index 0000000..6a0b8b0
--- /dev/null
+++ b/lang/java/reef-applications/reef-vortex/src/main/java/org/apache/reef/vortex/driver/PendingTaskletScheduler.java
@@ -0,0 +1,59 @@
+/*
+ * 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.
+ */
+package org.apache.reef.vortex.driver;
+
+import org.apache.reef.annotations.audience.DriverSide;
+import org.apache.reef.wake.EventHandler;
+
+import javax.inject.Inject;
+import java.util.logging.Level;
+import java.util.logging.Logger;
+
+/**
+ * Pending Tasklet Scheduler.
+ */
+@DriverSide
+final class PendingTaskletScheduler implements EventHandler<Integer> {
+ private static final Logger LOG = Logger.getLogger(PendingTaskletScheduler.class.getName());
+
+ private final RunningWorkers runningWorkers;
+ private final PendingTasklets pendingTasklets;
+
+ @Inject
+ private PendingTaskletScheduler(final RunningWorkers runningWorkers,
+ final PendingTasklets pendingTasklets) {
+ this.runningWorkers = runningWorkers;
+ this.pendingTasklets = pendingTasklets;
+ }
+
+ /**
+ * Repeatedly take a tasklet from the pending queue and schedule/launch it via RunningWorkers.
+ */
+ @Override
+ public void onNext(final Integer integer) {
+ while (!runningWorkers.isTerminated()) {
+ try {
+ final Tasklet tasklet = pendingTasklets.takeFirst(); // blocks when no tasklet exists
+ runningWorkers.launchTasklet(tasklet); // blocks when no worker exists
+ } catch (InterruptedException e) {
+ LOG.log(Level.INFO, "Interrupted upon termination");
+ }
+ }
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/2c994ec9/lang/java/reef-applications/reef-vortex/src/main/java/org/apache/reef/vortex/driver/PendingTasklets.java
----------------------------------------------------------------------
diff --git a/lang/java/reef-applications/reef-vortex/src/main/java/org/apache/reef/vortex/driver/PendingTasklets.java b/lang/java/reef-applications/reef-vortex/src/main/java/org/apache/reef/vortex/driver/PendingTasklets.java
new file mode 100644
index 0000000..eac4fc6
--- /dev/null
+++ b/lang/java/reef-applications/reef-vortex/src/main/java/org/apache/reef/vortex/driver/PendingTasklets.java
@@ -0,0 +1,51 @@
+/*
+ * 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.
+ */
+package org.apache.reef.vortex.driver;
+
+import net.jcip.annotations.ThreadSafe;
+import org.apache.reef.annotations.audience.DriverSide;
+
+import javax.inject.Inject;
+import java.util.concurrent.BlockingDeque;
+import java.util.concurrent.LinkedBlockingDeque;
+
+/**
+ * Keep tracks of all pending tasklets.
+ */
+@ThreadSafe
+@DriverSide
+final class PendingTasklets {
+ private final BlockingDeque<Tasklet> pendingTasklets = new LinkedBlockingDeque<>();
+
+ @Inject
+ PendingTasklets() {
+ }
+
+ void addLast(final Tasklet tasklet) {
+ pendingTasklets.addLast(tasklet);
+ }
+
+ void addFirst(final Tasklet tasklet) {
+ pendingTasklets.addFirst(tasklet);
+ }
+
+ Tasklet takeFirst() throws InterruptedException {
+ return pendingTasklets.takeFirst();
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/2c994ec9/lang/java/reef-applications/reef-vortex/src/main/java/org/apache/reef/vortex/driver/RunningWorkers.java
----------------------------------------------------------------------
diff --git a/lang/java/reef-applications/reef-vortex/src/main/java/org/apache/reef/vortex/driver/RunningWorkers.java b/lang/java/reef-applications/reef-vortex/src/main/java/org/apache/reef/vortex/driver/RunningWorkers.java
new file mode 100644
index 0000000..f712786
--- /dev/null
+++ b/lang/java/reef-applications/reef-vortex/src/main/java/org/apache/reef/vortex/driver/RunningWorkers.java
@@ -0,0 +1,257 @@
+/*
+ * 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.
+ */
+package org.apache.reef.vortex.driver;
+
+import net.jcip.annotations.ThreadSafe;
+import org.apache.reef.annotations.audience.DriverSide;
+
+import javax.inject.Inject;
+import java.io.Serializable;
+import java.util.*;
+import java.util.concurrent.locks.Condition;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
+
+/**
+ * Keeps track of all running VortexWorkers and Tasklets.
+ * Upon Tasklet launch request, randomly schedules it to a VortexWorkerManager.
+ */
+@ThreadSafe
+@DriverSide
+final class RunningWorkers {
+ // RunningWorkers and its locks
+ private final HashMap<String, VortexWorkerManager> runningWorkers = new HashMap<>(); // Running workers/tasklets
+ private final ReentrantReadWriteLock rwLock = new ReentrantReadWriteLock();
+ private final Lock readLock = rwLock.readLock(); // Need to acquire this to launch/complete tasklets
+ private final Lock writeLock = rwLock.writeLock(); // Need to acquire this to add/remove worker
+ private final Condition noRunningWorker = writeLock.newCondition(); // When there's no running worker
+
+ // To keep track of workers that are preempted before acknowledged
+ private final Set<String> removedBeforeAddedWorkers = new HashSet<>();
+
+ // Terminated
+ private volatile boolean terminated = false;
+
+ /**
+ * RunningWorkers constructor.
+ */
+ @Inject
+ RunningWorkers() {
+ }
+
+ /**
+ * Concurrency: Called by multiple threads.
+ * Parameter: Called exactly once per vortexWorkerManager.
+ */
+ void addWorker(final VortexWorkerManager vortexWorkerManager) {
+ if (!terminated) {
+ writeLock.lock();
+ try {
+ if (!terminated) { // Make sure it is really terminated
+ if (!removedBeforeAddedWorkers.contains(vortexWorkerManager.getId())) {
+ this.runningWorkers.put(vortexWorkerManager.getId(), vortexWorkerManager);
+
+ // Notify (possibly) waiting scheduler
+ if (runningWorkers.size() == 1) {
+ noRunningWorker.signalAll();
+ }
+ }
+ return;
+ }
+ } finally {
+ writeLock.unlock();
+ }
+ }
+
+ // Terminate the worker
+ vortexWorkerManager.terminate();
+ }
+
+ /**
+ * Concurrency: Called by multiple threads.
+ * Parameter: Called exactly once per id.
+ */
+ Collection<Tasklet> removeWorker(final String id) {
+ if (!terminated) {
+ writeLock.lock();
+ try {
+ if (!terminated) { // Make sure it is really terminated
+ final VortexWorkerManager vortexWorkerManager = this.runningWorkers.remove(id);
+ if (vortexWorkerManager != null) {
+ return vortexWorkerManager.removed();
+ } else {
+ // Called before addWorker (e.g. RM preempted the resource before the Evaluator started)
+ removedBeforeAddedWorkers.add(id);
+ return new ArrayList<>(0);
+ }
+ }
+ } finally {
+ writeLock.unlock();
+ }
+ }
+
+ // No need to return anything since it is terminated
+ return new ArrayList<>(0);
+ }
+
+ /**
+ * Concurrency: Called by single scheduler thread.
+ * Parameter: Same tasklet can be launched multiple times.
+ */
+ void launchTasklet(final Tasklet tasklet) {
+ if (!terminated) {
+ readLock.lock();
+ try {
+ if (!terminated) { // Make sure it is really terminated
+ // Wait until there is a running worker
+ if (runningWorkers.isEmpty()) {
+ readLock.unlock();
+ writeLock.lock();
+ try {
+ while (runningWorkers.isEmpty()) {
+ try {
+ noRunningWorker.await();
+ } catch (InterruptedException e) {
+ throw new RuntimeException(e);
+ }
+ }
+ readLock.lock();
+ } finally {
+ writeLock.unlock();
+ }
+ }
+
+ final VortexWorkerManager vortexWorkerManager = randomlyChooseWorker();
+ vortexWorkerManager.launchTasklet(tasklet);
+ return;
+ }
+ } finally {
+ readLock.unlock();
+ }
+ }
+ }
+
+ /**
+ * Concurrency: Called by multiple threads.
+ * Parameter: Same arguments can come in multiple times.
+ * (e.g. preemption message coming before tasklet completion message multiple times)
+ */
+ void completeTasklet(final String workerId,
+ final int taskletId,
+ final Serializable result) {
+ if (!terminated) {
+ readLock.lock();
+ try {
+ if (!terminated) { // Make sure it is really terminated
+ if (runningWorkers.containsKey(workerId)) { // Preemption can come before
+ this.runningWorkers.get(workerId).taskletCompleted(taskletId, result);
+ }
+ return;
+ }
+ } finally {
+ readLock.unlock();
+ }
+ }
+ }
+
+ /**
+ * Concurrency: Called by multiple threads.
+ * Parameter: Same arguments can come in multiple times.
+ * (e.g. preemption message coming before tasklet error message multiple times)
+ */
+ void errorTasklet(final String workerId,
+ final int taskletId,
+ final Exception exception) {
+ if (!terminated) {
+ readLock.lock();
+ try {
+ if (!terminated) { // Make sure it is really terminated
+ if (runningWorkers.containsKey(workerId)) { // Preemption can come before
+ this.runningWorkers.get(workerId).taskletThrewException(taskletId, exception);
+ }
+ return;
+ }
+ } finally {
+ readLock.unlock();
+ }
+ }
+ }
+
+ void terminate() {
+ if (!terminated) {
+ writeLock.lock();
+ try {
+ if (!terminated) { // Make sure it is really terminated
+ terminated = true;
+ for (final VortexWorkerManager vortexWorkerManager : runningWorkers.values()) {
+ vortexWorkerManager.terminate();
+ }
+ runningWorkers.clear();
+ return;
+ }
+ } finally {
+ writeLock.unlock();
+ }
+ }
+ throw new RuntimeException("Attempting to terminate an already terminated RunningWorkers");
+ }
+
+ boolean isTerminated() {
+ return terminated;
+ }
+
+ private VortexWorkerManager randomlyChooseWorker() {
+ final Collection<VortexWorkerManager> workers = runningWorkers.values();
+ final int index = new Random().nextInt(workers.size());
+ int i = 0;
+ for (final VortexWorkerManager vortexWorkerManager : workers) {
+ if (i == index) {
+ return vortexWorkerManager;
+ } else {
+ i++;
+ }
+ }
+ throw new RuntimeException("Bad Index");
+ }
+
+ ///////////////////////////////////////// For Tests Only
+
+ /**
+ * For unit tests to check whether the worker is running.
+ */
+ boolean isWorkerRunning(final String workerId) {
+ return runningWorkers.containsKey(workerId);
+ }
+
+ /**
+ * For unit tests to see where a tasklet is scheduled to.
+ * @param taskletId id of the tasklet in question
+ * @return id of the worker (null if the tasklet was not scheduled to any worker)
+ */
+ String getWhereTaskletWasScheduledTo(final int taskletId) {
+ for (final Map.Entry<String, VortexWorkerManager> entry : runningWorkers.entrySet()) {
+ final String workerId = entry.getKey();
+ final VortexWorkerManager vortexWorkerManager = entry.getValue();
+ if (vortexWorkerManager.containsTasklet(taskletId)) {
+ return workerId;
+ }
+ }
+ return null;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/2c994ec9/lang/java/reef-applications/reef-vortex/src/main/java/org/apache/reef/vortex/driver/Tasklet.java
----------------------------------------------------------------------
diff --git a/lang/java/reef-applications/reef-vortex/src/main/java/org/apache/reef/vortex/driver/Tasklet.java b/lang/java/reef-applications/reef-vortex/src/main/java/org/apache/reef/vortex/driver/Tasklet.java
new file mode 100644
index 0000000..5d64c57
--- /dev/null
+++ b/lang/java/reef-applications/reef-vortex/src/main/java/org/apache/reef/vortex/driver/Tasklet.java
@@ -0,0 +1,96 @@
+/*
+ * 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.
+ */
+package org.apache.reef.vortex.driver;
+
+import org.apache.reef.annotations.audience.DriverSide;
+import org.apache.reef.vortex.api.VortexFunction;
+import org.apache.reef.vortex.api.VortexFuture;
+
+import java.io.Serializable;
+
+/**
+ * Representation of user task in Driver.
+ */
+@DriverSide
+class Tasklet<TInput extends Serializable, TOutput extends Serializable> implements Serializable {
+ private final int taskletId;
+ private final VortexFunction<TInput, TOutput> userTask;
+ private final TInput input;
+ private final VortexFuture<TOutput> vortexFuture;
+
+ Tasklet(final int taskletId,
+ final VortexFunction<TInput, TOutput> userTask,
+ final TInput input,
+ final VortexFuture<TOutput> vortexFuture) {
+ this.taskletId = taskletId;
+ this.userTask = userTask;
+ this.input = input;
+ this.vortexFuture = vortexFuture;
+ }
+
+ /**
+ * @return id of the tasklet
+ */
+ int getId() {
+ return taskletId;
+ }
+
+ /**
+ * @return the input of the tasklet
+ */
+ TInput getInput() {
+ return input;
+ }
+
+ /**
+ * @return the user function of the tasklet
+ */
+ VortexFunction<TInput, TOutput> getUserFunction() {
+ return userTask;
+ }
+
+ /**
+ * Called by VortexMaster to let the user know that the task completed.
+ */
+ void completed(final TOutput result) {
+ vortexFuture.completed(result);
+ }
+
+ /**
+ * Called by VortexMaster to let the user know that the task threw an exception.
+ */
+ void threwException(final Exception exception) {
+ vortexFuture.threwException(exception);
+ }
+
+ /**
+ * For tests.
+ */
+ boolean isCompleted() {
+ return vortexFuture.isDone();
+ }
+
+ /**
+ * @return description of the tasklet in string.
+ */
+ @Override
+ public String toString() {
+ return "Tasklet: " + taskletId;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/2c994ec9/lang/java/reef-applications/reef-vortex/src/main/java/org/apache/reef/vortex/driver/VortexConfHelper.java
----------------------------------------------------------------------
diff --git a/lang/java/reef-applications/reef-vortex/src/main/java/org/apache/reef/vortex/driver/VortexConfHelper.java b/lang/java/reef-applications/reef-vortex/src/main/java/org/apache/reef/vortex/driver/VortexConfHelper.java
new file mode 100644
index 0000000..a7b9106
--- /dev/null
+++ b/lang/java/reef-applications/reef-vortex/src/main/java/org/apache/reef/vortex/driver/VortexConfHelper.java
@@ -0,0 +1,66 @@
+/*
+ * 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.
+ */
+package org.apache.reef.vortex.driver;
+
+import org.apache.reef.annotations.Unstable;
+import org.apache.reef.client.DriverConfiguration;
+import org.apache.reef.tang.Configuration;
+import org.apache.reef.tang.Configurations;
+import org.apache.reef.util.EnvironmentUtils;
+import org.apache.reef.vortex.api.VortexStart;
+
+/**
+ * Helper class for building a configuration for Vortex.
+ */
+@Unstable
+public final class VortexConfHelper {
+ private VortexConfHelper() {
+ }
+
+ private static final int DEFAULT_NUM_OF_VORTEX_START_THERAD = 1;
+
+ /**
+ * @return Configuration for Vortex job.
+ */
+ public static Configuration getVortexConf(final String jobName,
+ final Class<? extends VortexStart> vortexStart,
+ final int numOfWorkers,
+ final int workerMemory,
+ final int workerCores) {
+ final Configuration vortexDriverConf = DriverConfiguration.CONF
+ .set(DriverConfiguration.GLOBAL_LIBRARIES, EnvironmentUtils.getClassLocation(VortexDriver.class))
+ .set(DriverConfiguration.ON_DRIVER_STARTED, VortexDriver.StartHandler.class)
+ .set(DriverConfiguration.ON_EVALUATOR_ALLOCATED, VortexDriver.AllocatedEvaluatorHandler.class)
+ .set(DriverConfiguration.ON_TASK_RUNNING, VortexDriver.RunningTaskHandler.class)
+ .set(DriverConfiguration.ON_TASK_MESSAGE, VortexDriver.TaskMessageHandler.class)
+ .set(DriverConfiguration.ON_EVALUATOR_FAILED, VortexDriver.FailedEvaluatorHandler.class)
+ .set(DriverConfiguration.DRIVER_IDENTIFIER, jobName)
+ .build();
+
+ final Configuration vortexMasterConf = VortexMasterConf.CONF
+ .set(VortexMasterConf.WORKER_NUM, numOfWorkers)
+ .set(VortexMasterConf.WORKER_MEM, workerMemory)
+ .set(VortexMasterConf.WORKER_CORES, workerCores)
+ .set(VortexMasterConf.VORTEX_START, vortexStart)
+ .set(VortexMasterConf.NUM_OF_VORTEX_START_THERAD, DEFAULT_NUM_OF_VORTEX_START_THERAD) // fixed to 1 for now
+ .build();
+
+ return Configurations.merge(vortexDriverConf, vortexMasterConf);
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/2c994ec9/lang/java/reef-applications/reef-vortex/src/main/java/org/apache/reef/vortex/driver/VortexDriver.java
----------------------------------------------------------------------
diff --git a/lang/java/reef-applications/reef-vortex/src/main/java/org/apache/reef/vortex/driver/VortexDriver.java b/lang/java/reef-applications/reef-vortex/src/main/java/org/apache/reef/vortex/driver/VortexDriver.java
new file mode 100644
index 0000000..4dedb1d
--- /dev/null
+++ b/lang/java/reef-applications/reef-vortex/src/main/java/org/apache/reef/vortex/driver/VortexDriver.java
@@ -0,0 +1,193 @@
+/*
+ * 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.
+ */
+package org.apache.reef.vortex.driver;
+
+import org.apache.commons.lang.SerializationUtils;
+import org.apache.reef.annotations.audience.DriverSide;
+import org.apache.reef.driver.evaluator.*;
+import org.apache.reef.driver.task.RunningTask;
+import org.apache.reef.driver.task.TaskConfiguration;
+import org.apache.reef.driver.task.TaskMessage;
+import org.apache.reef.tang.Configuration;
+import org.apache.reef.tang.Configurations;
+import org.apache.reef.tang.annotations.Parameter;
+import org.apache.reef.tang.annotations.Unit;
+import org.apache.reef.vortex.api.VortexStart;
+import org.apache.reef.vortex.common.TaskletFailureReport;
+import org.apache.reef.vortex.common.TaskletResultReport;
+import org.apache.reef.vortex.common.WorkerReport;
+import org.apache.reef.vortex.evaluator.VortexWorker;
+import org.apache.reef.wake.EStage;
+import org.apache.reef.wake.EventHandler;
+import org.apache.reef.wake.impl.SingleThreadStage;
+import org.apache.reef.wake.time.event.StartTime;
+
+import javax.inject.Inject;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.logging.Level;
+import java.util.logging.Logger;
+
+/**
+ * REEF Driver for Vortex.
+ */
+@Unit
+@DriverSide
+final class VortexDriver {
+ private static final Logger LOG = Logger.getLogger(VortexDriver.class.getName());
+ private static final int MAX_NUM_OF_FAILURES = 5;
+ private static final int SCHEDULER_EVENT = 0; // Dummy number to comply with onNext() interface
+
+ private final AtomicInteger numberOfFailures = new AtomicInteger(0);
+ private final EvaluatorRequestor evaluatorRequestor; // for requesting resources
+ private final VortexMaster vortexMaster; // Vortex Master
+ private final VortexRequestor vortexRequestor; // For sending Commands to remote workers
+
+ // Resource configuration for single thread pool
+ private final int evalMem;
+ private final int evalNum;
+ private final int evalCores;
+
+ private final EStage<VortexStart> vortexStartEStage;
+ private final VortexStart vortexStart;
+ private final EStage<Integer> pendingTaskletSchedulerEStage;
+
+ @Inject
+ private VortexDriver(final EvaluatorRequestor evaluatorRequestor,
+ final VortexRequestor vortexRequestor,
+ final VortexMaster vortexMaster,
+ final EStage<VortexStart> vortexStartEStage,
+ final VortexStart vortexStart,
+ final PendingTaskletScheduler pendingTaskletScheduler,
+ @Parameter(VortexMasterConf.WorkerMem.class) final int workerMem,
+ @Parameter(VortexMasterConf.WorkerNum.class) final int workerNum,
+ @Parameter(VortexMasterConf.WorkerCores.class) final int workerCores) {
+ this.vortexStartEStage = vortexStartEStage;
+ this.vortexStart = vortexStart;
+ this.pendingTaskletSchedulerEStage = new SingleThreadStage<>(pendingTaskletScheduler, 1);
+ this.evaluatorRequestor = evaluatorRequestor;
+ this.vortexMaster = vortexMaster;
+ this.vortexRequestor = vortexRequestor;
+ this.evalMem = workerMem;
+ this.evalNum = workerNum;
+ this.evalCores = workerCores;
+ }
+
+ /**
+ * Driver started.
+ */
+ final class StartHandler implements EventHandler<StartTime> {
+ @Override
+ public void onNext(final StartTime startTime) {
+ // Initial Evaluator Request
+ evaluatorRequestor.submit(EvaluatorRequest.newBuilder()
+ .setNumber(evalNum)
+ .setMemory(evalMem)
+ .setNumberOfCores(evalCores)
+ .build());
+
+ // Run Vortex Start
+ vortexStartEStage.onNext(vortexStart);
+
+ // Run Scheduler
+ pendingTaskletSchedulerEStage.onNext(SCHEDULER_EVENT);
+ }
+ }
+
+ /**
+ * Container allocated.
+ */
+ final class AllocatedEvaluatorHandler implements EventHandler<AllocatedEvaluator> {
+ @Override
+ public void onNext(final AllocatedEvaluator allocatedEvaluator) {
+ LOG.log(Level.INFO, "Container allocated");
+ final String workerId = allocatedEvaluator.getId() + "_vortex_worker";
+
+ final Configuration workerConfiguration = VortexWorkerConf.CONF
+ .set(VortexWorkerConf.NUM_OF_THREADS, evalCores) // NUM_OF_THREADS = evalCores
+ .build();
+
+ final Configuration taskConfiguration = TaskConfiguration.CONF
+ .set(TaskConfiguration.IDENTIFIER, workerId)
+ .set(TaskConfiguration.TASK, VortexWorker.class)
+ .set(TaskConfiguration.ON_SEND_MESSAGE, VortexWorker.class)
+ .set(TaskConfiguration.ON_MESSAGE, VortexWorker.DriverMessageHandler.class)
+ .set(TaskConfiguration.ON_CLOSE, VortexWorker.TaskCloseHandler.class)
+ .build();
+
+ allocatedEvaluator.submitTask(Configurations.merge(workerConfiguration, taskConfiguration));
+ }
+ }
+
+ /**
+ * Evaluator up and running.
+ */
+ final class RunningTaskHandler implements EventHandler<RunningTask> {
+ @Override
+ public void onNext(final RunningTask reefTask) {
+ LOG.log(Level.INFO, "Worker up and running");
+ vortexMaster.workerAllocated(new VortexWorkerManager(vortexRequestor, reefTask));
+ }
+ }
+
+ /**
+ * Message received.
+ */
+ final class TaskMessageHandler implements EventHandler<TaskMessage> {
+ @Override
+ public void onNext(final TaskMessage taskMessage) {
+ final String workerId = taskMessage.getId();
+ final WorkerReport workerReport= (WorkerReport)SerializationUtils.deserialize(taskMessage.get());
+ switch (workerReport.getType()) {
+ case TaskletResult:
+ final TaskletResultReport taskletResultReport = (TaskletResultReport)workerReport;
+ vortexMaster.taskletCompleted(workerId, taskletResultReport.getTaskletId(), taskletResultReport.getResult());
+ break;
+ case TaskletFailure:
+ final TaskletFailureReport taskletFailureReport = (TaskletFailureReport)workerReport;
+ vortexMaster.taskletErrored(workerId, taskletFailureReport.getTaskletId(), taskletFailureReport.getException());
+ break;
+ default:
+ throw new RuntimeException("Unknown Report");
+ }
+ }
+ }
+
+ /**
+ * Evaluator preempted.
+ * TODO[REEF-501]: Distinguish different types of FailedEvaluator in Vortex.
+ */
+ final class FailedEvaluatorHandler implements EventHandler<FailedEvaluator> {
+ @Override
+ public void onNext(final FailedEvaluator failedEvaluator) {
+ LOG.log(Level.INFO, "Evaluator preempted");
+ if (numberOfFailures.incrementAndGet() >= MAX_NUM_OF_FAILURES) {
+ throw new RuntimeException("Exceeded max number of failures");
+ } else {
+ // We request a new evaluator to take the place of the preempted one
+ evaluatorRequestor.submit(EvaluatorRequest.newBuilder()
+ .setNumber(1)
+ .setMemory(evalMem)
+ .setNumberOfCores(evalCores)
+ .build());
+
+ vortexMaster.workerPreempted(failedEvaluator.getFailedTask().get().getId());
+ }
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/2c994ec9/lang/java/reef-applications/reef-vortex/src/main/java/org/apache/reef/vortex/driver/VortexLauncher.java
----------------------------------------------------------------------
diff --git a/lang/java/reef-applications/reef-vortex/src/main/java/org/apache/reef/vortex/driver/VortexLauncher.java b/lang/java/reef-applications/reef-vortex/src/main/java/org/apache/reef/vortex/driver/VortexLauncher.java
new file mode 100644
index 0000000..b288824
--- /dev/null
+++ b/lang/java/reef-applications/reef-vortex/src/main/java/org/apache/reef/vortex/driver/VortexLauncher.java
@@ -0,0 +1,62 @@
+/*
+ * 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.
+ */
+package org.apache.reef.vortex.driver;
+
+import org.apache.reef.annotations.Unstable;
+import org.apache.reef.client.DriverLauncher;
+import org.apache.reef.client.LauncherStatus;
+import org.apache.reef.runtime.local.client.LocalRuntimeConfiguration;
+import org.apache.reef.tang.Configuration;
+import org.apache.reef.tang.exceptions.InjectionException;
+import org.apache.reef.vortex.api.VortexStart;
+
+/**
+ * Launches a Vortex Job.
+ */
+@Unstable
+public final class VortexLauncher {
+ private VortexLauncher() {
+ }
+
+ private static final int MAX_NUMBER_OF_EVALUATORS = 10;
+
+ /**
+ * Launch a Vortex job using the local runtime.
+ */
+ public static LauncherStatus launchLocal(final String jobName,
+ final Class<? extends VortexStart> vortexUserCode,
+ final int numOfWorkers,
+ final int workerMemory,
+ final int workerCores) {
+ final Configuration runtimeConf = LocalRuntimeConfiguration.CONF
+ .set(LocalRuntimeConfiguration.MAX_NUMBER_OF_EVALUATORS, MAX_NUMBER_OF_EVALUATORS)
+ .build();
+ final Configuration vortexConf =
+ VortexConfHelper.getVortexConf(jobName, vortexUserCode, numOfWorkers, workerMemory, workerCores);
+ return launch(runtimeConf, vortexConf);
+ }
+
+ private static LauncherStatus launch(final Configuration runtimeConf, final Configuration vortexConf) {
+ try {
+ return DriverLauncher.getLauncher(runtimeConf).run(vortexConf);
+ } catch (InjectionException e) {
+ throw new RuntimeException(e);
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/2c994ec9/lang/java/reef-applications/reef-vortex/src/main/java/org/apache/reef/vortex/driver/VortexMaster.java
----------------------------------------------------------------------
diff --git a/lang/java/reef-applications/reef-vortex/src/main/java/org/apache/reef/vortex/driver/VortexMaster.java b/lang/java/reef-applications/reef-vortex/src/main/java/org/apache/reef/vortex/driver/VortexMaster.java
new file mode 100644
index 0000000..c9da30d
--- /dev/null
+++ b/lang/java/reef-applications/reef-vortex/src/main/java/org/apache/reef/vortex/driver/VortexMaster.java
@@ -0,0 +1,67 @@
+/*
+ * 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.
+ */
+package org.apache.reef.vortex.driver;
+
+import org.apache.reef.annotations.Unstable;
+import org.apache.reef.annotations.audience.DriverSide;
+import org.apache.reef.tang.annotations.DefaultImplementation;
+import org.apache.reef.vortex.api.VortexFunction;
+import org.apache.reef.vortex.api.VortexFuture;
+
+import java.io.Serializable;
+
+/**
+ * The heart of Vortex.
+ * Processes various tasklet related events/requests coming from different components of the system.
+ */
+@Unstable
+@DriverSide
+@DefaultImplementation(DefaultVortexMaster.class)
+public interface VortexMaster {
+ /**
+ * Submit a new Tasklet to be run sometime in the future.
+ */
+ <TInput extends Serializable, TOutput extends Serializable> VortexFuture<TOutput>
+ enqueueTasklet(final VortexFunction<TInput, TOutput> vortexFunction, final TInput input);
+
+ /**
+ * Call this when a new worker is up and running.
+ */
+ void workerAllocated(final VortexWorkerManager vortexWorkerManager);
+
+ /**
+ * Call this when a worker is preempted.
+ */
+ void workerPreempted(final String id);
+
+ /**
+ * Call this when a Tasklet is completed.
+ */
+ void taskletCompleted(final String workerId, final int taskletId, final Serializable result);
+
+ /**
+ * Call this when a Tasklet errored.
+ */
+ void taskletErrored(final String workerId, final int taskletId, final Exception exception);
+
+ /**
+ * Release all resources and shut down.
+ */
+ void terminate();
+}
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/2c994ec9/lang/java/reef-applications/reef-vortex/src/main/java/org/apache/reef/vortex/driver/VortexMasterConf.java
----------------------------------------------------------------------
diff --git a/lang/java/reef-applications/reef-vortex/src/main/java/org/apache/reef/vortex/driver/VortexMasterConf.java b/lang/java/reef-applications/reef-vortex/src/main/java/org/apache/reef/vortex/driver/VortexMasterConf.java
new file mode 100644
index 0000000..41b7ab2
--- /dev/null
+++ b/lang/java/reef-applications/reef-vortex/src/main/java/org/apache/reef/vortex/driver/VortexMasterConf.java
@@ -0,0 +1,98 @@
+/*
+ * 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.
+ */
+package org.apache.reef.vortex.driver;
+
+import org.apache.reef.annotations.Unstable;
+import org.apache.reef.annotations.audience.DriverSide;
+import org.apache.reef.tang.annotations.Name;
+import org.apache.reef.tang.annotations.NamedParameter;
+import org.apache.reef.tang.formats.ConfigurationModule;
+import org.apache.reef.tang.formats.ConfigurationModuleBuilder;
+import org.apache.reef.tang.formats.RequiredImpl;
+import org.apache.reef.tang.formats.RequiredParameter;
+import org.apache.reef.vortex.api.VortexStart;
+import org.apache.reef.wake.EStage;
+import org.apache.reef.wake.StageConfiguration;
+import org.apache.reef.wake.impl.ThreadPoolStage;
+
+/**
+ * Vortex Master configuration.
+ */
+@Unstable
+@DriverSide
+public final class VortexMasterConf extends ConfigurationModuleBuilder {
+ /**
+ * Number of Workers.
+ */
+ @NamedParameter(doc = "Number of Workers")
+ final class WorkerNum implements Name<Integer> {
+ }
+
+ /**
+ * Worker Memory.
+ */
+ @NamedParameter(doc = "Worker Memory")
+ final class WorkerMem implements Name<Integer> {
+ }
+
+ /**
+ * Worker Cores.
+ */
+ @NamedParameter(doc = "Worker Cores")
+ final class WorkerCores implements Name<Integer> {
+ }
+
+ /**
+ * Number of Workers.
+ */
+ public static final RequiredParameter<Integer> WORKER_NUM = new RequiredParameter<>();
+
+ /**
+ * Worker Memory.
+ */
+ public static final RequiredParameter<Integer> WORKER_MEM = new RequiredParameter<>();
+
+ /**
+ * Worker Cores.
+ */
+ public static final RequiredParameter<Integer> WORKER_CORES = new RequiredParameter<>();
+
+ /**
+ * Vortex Start.
+ */
+ public static final RequiredImpl<VortexStart> VORTEX_START = new RequiredImpl<>();
+
+ /**
+ * Number of Vortex Start threads.
+ */
+ public static final RequiredParameter<Integer> NUM_OF_VORTEX_START_THERAD = new RequiredParameter<>();
+
+ /**
+ * Vortex Master configuration.
+ */
+ public static final ConfigurationModule CONF = new VortexMasterConf()
+ .bindNamedParameter(WorkerNum.class, WORKER_NUM)
+ .bindNamedParameter(WorkerMem.class, WORKER_MEM)
+ .bindNamedParameter(WorkerCores.class, WORKER_CORES)
+ .bindImplementation(VortexStart.class, VORTEX_START)
+ .bindNamedParameter(StageConfiguration.NumberOfThreads.class, NUM_OF_VORTEX_START_THERAD)
+ .bindNamedParameter(StageConfiguration.StageHandler.class, VortexStartExecutor.class)
+ .bindImplementation(EStage.class, ThreadPoolStage.class)
+ .build();
+}
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/2c994ec9/lang/java/reef-applications/reef-vortex/src/main/java/org/apache/reef/vortex/driver/VortexRequestor.java
----------------------------------------------------------------------
diff --git a/lang/java/reef-applications/reef-vortex/src/main/java/org/apache/reef/vortex/driver/VortexRequestor.java b/lang/java/reef-applications/reef-vortex/src/main/java/org/apache/reef/vortex/driver/VortexRequestor.java
new file mode 100644
index 0000000..2c02d6e
--- /dev/null
+++ b/lang/java/reef-applications/reef-vortex/src/main/java/org/apache/reef/vortex/driver/VortexRequestor.java
@@ -0,0 +1,50 @@
+/*
+ * 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.
+ */
+package org.apache.reef.vortex.driver;
+
+import org.apache.commons.lang.SerializationUtils;
+import org.apache.reef.annotations.audience.DriverSide;
+import org.apache.reef.driver.task.RunningTask;
+import org.apache.reef.vortex.common.VortexRequest;
+
+import javax.inject.Inject;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+
+/**
+ * Takes the serialization work from the scheduler thread.
+ */
+@DriverSide
+class VortexRequestor {
+ private final ExecutorService executorService = Executors.newCachedThreadPool();
+
+ @Inject
+ VortexRequestor() {
+ }
+
+ void send(final RunningTask reefTask, final VortexRequest vortexRequest) {
+ executorService.execute(new Runnable() {
+ @Override
+ public void run() {
+ // Possible race condition with VortexWorkerManager#terminate is addressed by the global lock in VortexMaster
+ reefTask.send(SerializationUtils.serialize(vortexRequest));
+ }
+ });
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-reef/blob/2c994ec9/lang/java/reef-applications/reef-vortex/src/main/java/org/apache/reef/vortex/driver/VortexStartExecutor.java
----------------------------------------------------------------------
diff --git a/lang/java/reef-applications/reef-vortex/src/main/java/org/apache/reef/vortex/driver/VortexStartExecutor.java b/lang/java/reef-applications/reef-vortex/src/main/java/org/apache/reef/vortex/driver/VortexStartExecutor.java
new file mode 100644
index 0000000..6a5ae4b
--- /dev/null
+++ b/lang/java/reef-applications/reef-vortex/src/main/java/org/apache/reef/vortex/driver/VortexStartExecutor.java
@@ -0,0 +1,52 @@
+/*
+ * 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.
+ */
+package org.apache.reef.vortex.driver;
+
+import org.apache.reef.annotations.audience.DriverSide;
+import org.apache.reef.vortex.api.VortexStart;
+import org.apache.reef.vortex.api.VortexThreadPool;
+import org.apache.reef.wake.EventHandler;
+
+import javax.inject.Inject;
+
+/**
+ * Thread pool that executes VortexStart.
+ */
+@DriverSide
+final class VortexStartExecutor implements EventHandler<VortexStart> {
+ private final VortexThreadPool vortexThreadPool; // ThreadPool-like API for VortexStart to use
+ private final VortexMaster vortexMaster;
+
+ @Inject
+ private VortexStartExecutor(final VortexThreadPool vortexThreadPool,
+ final VortexMaster vortexMaster) {
+ this.vortexThreadPool = vortexThreadPool;
+ this.vortexMaster = vortexMaster;
+ }
+
+ /**
+ * Execute the passed VortexStart and terminate right after.
+ * @param vortexStart to execute
+ */
+ @Override
+ public void onNext(final VortexStart vortexStart) {
+ vortexStart.start(vortexThreadPool);
+ vortexMaster.terminate();
+ }
+}
\ No newline at end of file