Repository: reef Updated Branches: refs/heads/master 57a44fe0d -> b1a555ed8
[REEF-1000] Allow more configuration in VortexLauncher This addressed the issue by * Exposing the Master configuration in the Launcher code * Modifying examples where users can set the Application variables JIRA: [REEF-1000](https://issues.apache.org/jira/browse/REEF-1000) Pull Request: Closes #716 Project: http://git-wip-us.apache.org/repos/asf/reef/repo Commit: http://git-wip-us.apache.org/repos/asf/reef/commit/b1a555ed Tree: http://git-wip-us.apache.org/repos/asf/reef/tree/b1a555ed Diff: http://git-wip-us.apache.org/repos/asf/reef/diff/b1a555ed Branch: refs/heads/master Commit: b1a555ed8d9f6d34d3ca4ade2a8ef662f5ca90f5 Parents: 57a44fe Author: Yunseong Lee <[email protected]> Authored: Tue Dec 8 01:40:41 2015 +0800 Committer: Andrew Chung <[email protected]> Committed: Thu Dec 10 22:34:46 2015 -0800 ---------------------------------------------------------------------- .../reef/vortex/driver/VortexConfHelper.java | 71 ----------- .../reef/vortex/driver/VortexJobConf.java | 125 +++++++++++++++++++ .../reef/vortex/driver/VortexLauncher.java | 17 +-- .../reef/vortex/driver/VortexMasterConf.java | 4 +- .../reef/vortex/examples/addone/AddOne.java | 30 ++++- .../vortex/examples/addone/AddOneStart.java | 8 +- .../reef/vortex/examples/hello/HelloVortex.java | 18 ++- .../examples/matmul/IdentityMatMulStart.java | 59 +++++---- .../reef/vortex/examples/matmul/MatMul.java | 40 +++++- .../applications/vortex/addone/AddOneTest.java | 38 ++++-- .../cancellation/TaskletCancellationTest.java | 21 +++- .../vortex/exception/VortexExceptionTest.java | 21 +++- 12 files changed, 316 insertions(+), 136 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/reef/blob/b1a555ed/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 deleted file mode 100644 index 6117e4e..0000000 --- a/lang/java/reef-applications/reef-vortex/src/main/java/org/apache/reef/vortex/driver/VortexConfHelper.java +++ /dev/null @@ -1,71 +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.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_THREAD = 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 int workerCapacity) { - 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.WORKER_CAPACITY, workerCapacity) - .set(VortexMasterConf.VORTEX_START, vortexStart) - .set(VortexMasterConf.NUM_OF_VORTEX_START_THREAD, DEFAULT_NUM_OF_VORTEX_START_THREAD) // fixed to 1 for now - .build(); - - // TODO[JIRA REEF-1000]: Consider exposing VortexMasterConf.FUTURE_CALLBACK_THREAD_POOL_SIZE. - // For now, use default value defined in the NamedParameter. - - return Configurations.merge(vortexDriverConf, vortexMasterConf); - } -} http://git-wip-us.apache.org/repos/asf/reef/blob/b1a555ed/lang/java/reef-applications/reef-vortex/src/main/java/org/apache/reef/vortex/driver/VortexJobConf.java ---------------------------------------------------------------------- diff --git a/lang/java/reef-applications/reef-vortex/src/main/java/org/apache/reef/vortex/driver/VortexJobConf.java b/lang/java/reef-applications/reef-vortex/src/main/java/org/apache/reef/vortex/driver/VortexJobConf.java new file mode 100644 index 0000000..91522db --- /dev/null +++ b/lang/java/reef-applications/reef-vortex/src/main/java/org/apache/reef/vortex/driver/VortexJobConf.java @@ -0,0 +1,125 @@ +/* + * 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.Private; +import org.apache.reef.client.DriverConfiguration; +import org.apache.reef.tang.Configuration; +import org.apache.reef.tang.Configurations; +import org.apache.reef.util.BuilderUtils; +import org.apache.reef.util.EnvironmentUtils; +import org.apache.reef.util.Optional; + +/** + * Helper class for building a configuration for Vortex. + */ +@Unstable +public final class VortexJobConf { + private final Configuration conf; + + private VortexJobConf(final Configuration conf) { + this.conf = conf; + } + + /** + * Create a Builder object for Vortex job configuration. + */ + public static Builder newBuilder() { + return new VortexJobConf.Builder(); + } + + /** + * Convert to the Tang Configuration. + */ + @Private + public Configuration getConfiguration() { + return conf; + } + + /** + * Builder object to create a {@link VortexJobConf}. + */ + public static final class Builder implements org.apache.reef.util.Builder<VortexJobConf> { + private String jobName; + private Configuration vortexMasterConf; + private Optional<Configuration> userConf = Optional.empty(); + + private Builder() { + } + + /** + * @param vortexMasterConf Configuration for the Vortex Master, which can be built via {@link VortexMasterConf}. + */ + public Builder setVortexMasterConf(final Configuration vortexMasterConf) { + this.vortexMasterConf = vortexMasterConf; + return this; + } + + /** + * @param userConf Configuration set by user (e.g., Parameters in {@link org.apache.reef.vortex.api.VortexStart} + */ + public Builder setUserConf(final Configuration userConf) { + this.userConf = Optional.of(userConf); + return this; + } + + /** + * @param jobName Name of the job which is assigned to the Driver. + */ + public Builder setJobName(final String jobName) { + this.jobName = jobName; + return this; + } + + /** + * Instantiate a {@link VortexJobConf} object, where a Configuration is built by Tang internally. + * + * {@link IllegalArgumentException} will be thrown if required parameters are not set + * (See {@link #setJobName(String)} and {@link #setVortexMasterConf(Configuration)}). + * + * Also, {@link org.apache.reef.tang.exceptions.BindException} can be thrown while merging the configurations. + * + * @return An instance of VortexJobConf object. + */ + @Override + public VortexJobConf build() { + BuilderUtils.notNull(jobName); + BuilderUtils.notNull(vortexMasterConf); + + 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 jobConf; + if (userConf.isPresent()) { + jobConf = Configurations.merge(vortexDriverConf, vortexMasterConf, userConf.get()); + } else { + jobConf = Configurations.merge(vortexDriverConf, vortexMasterConf); + } + return new VortexJobConf(jobConf); + } + } +} http://git-wip-us.apache.org/repos/asf/reef/blob/b1a555ed/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 index c7461e5..92a500e 100644 --- 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 @@ -24,7 +24,6 @@ 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. @@ -39,23 +38,11 @@ public final class VortexLauncher { /** * 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 int workerCapacity) { + public static LauncherStatus launchLocal(final VortexJobConf vortexConf) { 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, - workerCapacity); - return launch(runtimeConf, vortexConf); + return launch(runtimeConf, vortexConf.getConfiguration()); } private static LauncherStatus launch(final Configuration runtimeConf, final Configuration vortexConf) { http://git-wip-us.apache.org/repos/asf/reef/blob/b1a555ed/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 index da7c4ad..43a4d44 100644 --- 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 @@ -62,7 +62,7 @@ public final class VortexMasterConf extends ConfigurationModuleBuilder { /** * Number of Vortex Start Threads. */ - @NamedParameter(doc = "Number of Vortex Start Threads") + @NamedParameter(doc = "Number of Vortex Start Threads", default_value = "1") final class NumberOfVortexStartThreads implements Name<Integer> { } @@ -101,7 +101,7 @@ public final class VortexMasterConf extends ConfigurationModuleBuilder { /** * Number of Vortex Start threads. */ - public static final RequiredParameter<Integer> NUM_OF_VORTEX_START_THREAD = new RequiredParameter<>(); + public static final OptionalParameter<Integer> NUM_OF_VORTEX_START_THREAD = new OptionalParameter<>(); /** * Size of threadpool for callbacks on VortexFuture. http://git-wip-us.apache.org/repos/asf/reef/blob/b1a555ed/lang/java/reef-applications/reef-vortex/src/main/java/org/apache/reef/vortex/examples/addone/AddOne.java ---------------------------------------------------------------------- diff --git a/lang/java/reef-applications/reef-vortex/src/main/java/org/apache/reef/vortex/examples/addone/AddOne.java b/lang/java/reef-applications/reef-vortex/src/main/java/org/apache/reef/vortex/examples/addone/AddOne.java index 25573fc..4af8875 100644 --- a/lang/java/reef-applications/reef-vortex/src/main/java/org/apache/reef/vortex/examples/addone/AddOne.java +++ b/lang/java/reef-applications/reef-vortex/src/main/java/org/apache/reef/vortex/examples/addone/AddOne.java @@ -18,7 +18,13 @@ */ package org.apache.reef.vortex.examples.addone; +import org.apache.reef.tang.Configuration; +import org.apache.reef.tang.Tang; +import org.apache.reef.tang.annotations.Name; +import org.apache.reef.tang.annotations.NamedParameter; +import org.apache.reef.vortex.driver.VortexJobConf; import org.apache.reef.vortex.driver.VortexLauncher; +import org.apache.reef.vortex.driver.VortexMasterConf; /** * User's main function. @@ -31,6 +37,28 @@ final class AddOne { * Launch the vortex job, passing appropriate arguments. */ public static void main(final String[] args) { - VortexLauncher.launchLocal("Vortex_Example_AddOne", AddOneStart.class, 2, 1024, 4, 2000); + final Configuration vortexMasterConf = VortexMasterConf.CONF + .set(VortexMasterConf.WORKER_NUM, 2) + .set(VortexMasterConf.WORKER_MEM, 1024) + .set(VortexMasterConf.WORKER_CORES, 4) + .set(VortexMasterConf.WORKER_CAPACITY, 2000) + .set(VortexMasterConf.VORTEX_START, AddOneStart.class) + .build(); + + final Configuration userConf = Tang.Factory.getTang().newConfigurationBuilder() + .bindNamedParameter(Dimension.class, "1000") + .build(); + + final VortexJobConf vortexJobConf = VortexJobConf.newBuilder() + .setJobName("Vortex_Example_AddOne") + .setVortexMasterConf(vortexMasterConf) + .setUserConf(userConf) + .build(); + + VortexLauncher.launchLocal(vortexJobConf); + } + + @NamedParameter(doc = "dimension of input vector") + public static class Dimension implements Name<Integer> { } } http://git-wip-us.apache.org/repos/asf/reef/blob/b1a555ed/lang/java/reef-applications/reef-vortex/src/main/java/org/apache/reef/vortex/examples/addone/AddOneStart.java ---------------------------------------------------------------------- diff --git a/lang/java/reef-applications/reef-vortex/src/main/java/org/apache/reef/vortex/examples/addone/AddOneStart.java b/lang/java/reef-applications/reef-vortex/src/main/java/org/apache/reef/vortex/examples/addone/AddOneStart.java index 510cb14..7a7db46 100644 --- a/lang/java/reef-applications/reef-vortex/src/main/java/org/apache/reef/vortex/examples/addone/AddOneStart.java +++ b/lang/java/reef-applications/reef-vortex/src/main/java/org/apache/reef/vortex/examples/addone/AddOneStart.java @@ -18,6 +18,7 @@ */ package org.apache.reef.vortex.examples.addone; +import org.apache.reef.tang.annotations.Parameter; import org.apache.reef.vortex.api.VortexFuture; import org.apache.reef.vortex.api.VortexThreadPool; import org.apache.reef.vortex.api.VortexStart; @@ -32,8 +33,11 @@ import java.util.concurrent.ExecutionException; * AddOne User Code Example. */ final class AddOneStart implements VortexStart { + private final int dimension; + @Inject - private AddOneStart() { + private AddOneStart(@Parameter(AddOne.Dimension.class) final int dimension) { + this.dimension = dimension; } /** @@ -42,7 +46,7 @@ final class AddOneStart implements VortexStart { @Override public void start(final VortexThreadPool vortexThreadPool) { final Vector<Integer> inputVector = new Vector<>(); - for (int i = 0; i < 1000; i++) { + for (int i = 0; i < dimension; i++) { inputVector.add(i); } http://git-wip-us.apache.org/repos/asf/reef/blob/b1a555ed/lang/java/reef-applications/reef-vortex/src/main/java/org/apache/reef/vortex/examples/hello/HelloVortex.java ---------------------------------------------------------------------- diff --git a/lang/java/reef-applications/reef-vortex/src/main/java/org/apache/reef/vortex/examples/hello/HelloVortex.java b/lang/java/reef-applications/reef-vortex/src/main/java/org/apache/reef/vortex/examples/hello/HelloVortex.java index c4e3b6f..a6c05af 100644 --- a/lang/java/reef-applications/reef-vortex/src/main/java/org/apache/reef/vortex/examples/hello/HelloVortex.java +++ b/lang/java/reef-applications/reef-vortex/src/main/java/org/apache/reef/vortex/examples/hello/HelloVortex.java @@ -18,7 +18,10 @@ */ package org.apache.reef.vortex.examples.hello; +import org.apache.reef.tang.Configuration; +import org.apache.reef.vortex.driver.VortexJobConf; import org.apache.reef.vortex.driver.VortexLauncher; +import org.apache.reef.vortex.driver.VortexMasterConf; /** * User's main function. @@ -31,6 +34,19 @@ final class HelloVortex { * Launch the vortex job, passing appropriate arguments. */ public static void main(final String[] args) { - VortexLauncher.launchLocal("Vortex_Example_HelloVortex", HelloVortexStart.class, 1, 1024, 1, 2000); + final Configuration vortexMasterConf = VortexMasterConf.CONF + .set(VortexMasterConf.WORKER_NUM, 1) + .set(VortexMasterConf.WORKER_MEM, 1024) + .set(VortexMasterConf.WORKER_CORES, 1) + .set(VortexMasterConf.WORKER_CAPACITY, 2000) + .set(VortexMasterConf.VORTEX_START, HelloVortexStart.class) + .build(); + + final VortexJobConf vortexJobConf = VortexJobConf.newBuilder() + .setVortexMasterConf(vortexMasterConf) + .setJobName("HelloVortex") + .build(); + + VortexLauncher.launchLocal(vortexJobConf); } } http://git-wip-us.apache.org/repos/asf/reef/blob/b1a555ed/lang/java/reef-applications/reef-vortex/src/main/java/org/apache/reef/vortex/examples/matmul/IdentityMatMulStart.java ---------------------------------------------------------------------- diff --git a/lang/java/reef-applications/reef-vortex/src/main/java/org/apache/reef/vortex/examples/matmul/IdentityMatMulStart.java b/lang/java/reef-applications/reef-vortex/src/main/java/org/apache/reef/vortex/examples/matmul/IdentityMatMulStart.java index 092d64d..d02f904 100644 --- a/lang/java/reef-applications/reef-vortex/src/main/java/org/apache/reef/vortex/examples/matmul/IdentityMatMulStart.java +++ b/lang/java/reef-applications/reef-vortex/src/main/java/org/apache/reef/vortex/examples/matmul/IdentityMatMulStart.java @@ -18,6 +18,7 @@ */ package org.apache.reef.vortex.examples.matmul; +import org.apache.reef.tang.annotations.Parameter; import org.apache.reef.vortex.api.FutureCallback; import org.apache.reef.vortex.api.VortexStart; import org.apache.reef.vortex.api.VortexThreadPool; @@ -38,12 +39,18 @@ import java.util.logging.Logger; */ final class IdentityMatMulStart implements VortexStart { private static final Logger LOG = Logger.getLogger(IdentityMatMulStart.class.getName()); - private static final int DIVIDE_FACTOR = 10000; - private static final int NUM_ROWS = 100000; - private static final int NUM_COLUMNS = 10; + + private final int divideFactor; + private final int numRows; + private final int numColumns; @Inject - private IdentityMatMulStart() { + private IdentityMatMulStart(@Parameter(MatMul.DivideFactor.class) final int divideFactor, + @Parameter(MatMul.NumRows.class) final int numRows, + @Parameter(MatMul.NumColumns.class) final int numColumns) { + this.divideFactor = divideFactor; + this.numRows = numRows; + this.numColumns = numColumns; } /** @@ -51,14 +58,14 @@ final class IdentityMatMulStart implements VortexStart { */ @Override public void start(final VortexThreadPool vortexThreadPool) { - final List<Matrix<Double>> leftSplits = generateMatrixSplits(NUM_ROWS, NUM_COLUMNS, DIVIDE_FACTOR); - final Matrix<Double> right = generateIdentityMatrix(NUM_COLUMNS); + final List<Matrix<Double>> leftSplits = generateMatrixSplits(numRows, numColumns, divideFactor); + final Matrix<Double> right = generateIdentityMatrix(numColumns); // Measure job finish time starting from here.. final double start = System.currentTimeMillis(); // Define callback that is invoked when Tasklets finish. - final CountDownLatch latch = new CountDownLatch(DIVIDE_FACTOR); + final CountDownLatch latch = new CountDownLatch(divideFactor); final FutureCallback<MatMulOutput> callback = new FutureCallback<MatMulOutput>() { @Override public void onSuccess(final MatMulOutput output) { @@ -80,7 +87,7 @@ final class IdentityMatMulStart implements VortexStart { // Submit Tasklets and register callback. final MatMulFunction matMulFunction = new MatMulFunction(); - for (int i = 0; i < DIVIDE_FACTOR; i++) { + for (int i = 0; i < divideFactor; i++) { vortexThreadPool.submit(matMulFunction, new MatMulInput(i, leftSplits.get(i), right), callback); } @@ -95,16 +102,16 @@ final class IdentityMatMulStart implements VortexStart { /** * Generate a matrix with random values. - * @param numRows number of matrix's rows. - * @param numColumns number of matrix's columns. + * @param nRows number of matrix's rows. + * @param nColumns number of matrix's columns. * @return Matrix that consists of random values. */ - private Matrix<Double> generateRandomMatrix(final int numRows, final int numColumns) { - final List<List<Double>> rows = new ArrayList<>(numRows); + private Matrix<Double> generateRandomMatrix(final int nRows, final int nColumns) { + final List<List<Double>> rows = new ArrayList<>(nRows); final Random random = new Random(); - for (int i = 0; i < numRows; i++) { - final List<Double> row = new ArrayList<>(numColumns); - for (int j = 0; j < numColumns; j++) { + for (int i = 0; i < nRows; i++) { + final List<Double> row = new ArrayList<>(nColumns); + for (int j = 0; j < nColumns; j++) { row.add(random.nextDouble()); } rows.add(row); @@ -131,22 +138,22 @@ final class IdentityMatMulStart implements VortexStart { } /** - * Generate sub-matrices which splits a matrix as many as {@param divideFactor}. + * Generate sub-matrices which splits a matrix as many as {@param nSplits}. * Note that the matrix is split in row-wise, so the number of columns remain same while - * the number of rows is divided by {@param divideFactor}. - * @param numRows Number of rows of the original Matrix. - * @param numColumns Number of columns of the original Matrix. - * @param divideFactor Number of partitions to split the matrix into. + * the number of rows is divided by {@param nSplits}. + * @param nRows Number of rows of the original Matrix. + * @param nColumns Number of columns of the original Matrix. + * @param nSplits Number of partitions to split the matrix into. * @return List of matrices divided into multiple sub-matrices. */ - private List<Matrix<Double>> generateMatrixSplits(final int numRows, final int numColumns, final int divideFactor) { - final List<Matrix<Double>> splits = new ArrayList<>(divideFactor); + private List<Matrix<Double>> generateMatrixSplits(final int nRows, final int nColumns, final int nSplits) { + final List<Matrix<Double>> splits = new ArrayList<>(nSplits); - int remainingNumSplits = divideFactor; - int remainingNumRows = numRows; - for (int i = 0; i < divideFactor; i++) { + int remainingNumSplits = nSplits; + int remainingNumRows = nRows; + for (int i = 0; i < nSplits; i++) { final int splitNumRows = (remainingNumRows + remainingNumSplits - 1) / remainingNumSplits; - splits.add(generateRandomMatrix(splitNumRows, numColumns)); + splits.add(generateRandomMatrix(splitNumRows, nColumns)); remainingNumRows -= splitNumRows; remainingNumSplits--; http://git-wip-us.apache.org/repos/asf/reef/blob/b1a555ed/lang/java/reef-applications/reef-vortex/src/main/java/org/apache/reef/vortex/examples/matmul/MatMul.java ---------------------------------------------------------------------- diff --git a/lang/java/reef-applications/reef-vortex/src/main/java/org/apache/reef/vortex/examples/matmul/MatMul.java b/lang/java/reef-applications/reef-vortex/src/main/java/org/apache/reef/vortex/examples/matmul/MatMul.java index a09bf59..db4cdfd 100644 --- a/lang/java/reef-applications/reef-vortex/src/main/java/org/apache/reef/vortex/examples/matmul/MatMul.java +++ b/lang/java/reef-applications/reef-vortex/src/main/java/org/apache/reef/vortex/examples/matmul/MatMul.java @@ -18,7 +18,13 @@ */ package org.apache.reef.vortex.examples.matmul; +import org.apache.reef.tang.Configuration; +import org.apache.reef.tang.Tang; +import org.apache.reef.tang.annotations.Name; +import org.apache.reef.tang.annotations.NamedParameter; +import org.apache.reef.vortex.driver.VortexJobConf; import org.apache.reef.vortex.driver.VortexLauncher; +import org.apache.reef.vortex.driver.VortexMasterConf; /** * User's main function. @@ -31,6 +37,38 @@ final class MatMul { * Launch the vortex job, passing appropriate arguments. */ public static void main(final String[] args) { - VortexLauncher.launchLocal("Vortex_Example_MatMul", IdentityMatMulStart.class, 2, 1024, 4, 2000); + final Configuration vortexMasterConf = VortexMasterConf.CONF + .set(VortexMasterConf.WORKER_NUM, 2) + .set(VortexMasterConf.WORKER_MEM, 1024) + .set(VortexMasterConf.WORKER_CORES, 4) + .set(VortexMasterConf.WORKER_CAPACITY, 2000) + .set(VortexMasterConf.VORTEX_START, IdentityMatMulStart.class) + .build(); + + final Configuration userConf = Tang.Factory.getTang().newConfigurationBuilder() + .bindNamedParameter(DivideFactor.class, String.valueOf("10000")) + .bindNamedParameter(NumRows.class, String.valueOf("100000")) + .bindNamedParameter(NumColumns.class, String.valueOf("10")) + .build(); + + final VortexJobConf vortexJobConf = VortexJobConf.newBuilder() + .setVortexMasterConf(vortexMasterConf) + .setJobName("Vortex_Example_MatMul") + .setUserConf(userConf) + .build(); + + VortexLauncher.launchLocal(vortexJobConf); + } + + @NamedParameter(doc = "Number of splits the matrix is divided into") + public static class DivideFactor implements Name<Integer> { + } + + @NamedParameter(doc = "Number of rows of the original matrix") + public static class NumRows implements Name<Integer> { + } + + @NamedParameter(doc = "Number of columns of the original matrix") + public static class NumColumns implements Name<Integer> { } } http://git-wip-us.apache.org/repos/asf/reef/blob/b1a555ed/lang/java/reef-tests/src/test/java/org/apache/reef/tests/applications/vortex/addone/AddOneTest.java ---------------------------------------------------------------------- diff --git a/lang/java/reef-tests/src/test/java/org/apache/reef/tests/applications/vortex/addone/AddOneTest.java b/lang/java/reef-tests/src/test/java/org/apache/reef/tests/applications/vortex/addone/AddOneTest.java index f6d0098..4dfa9e7 100644 --- a/lang/java/reef-tests/src/test/java/org/apache/reef/tests/applications/vortex/addone/AddOneTest.java +++ b/lang/java/reef-tests/src/test/java/org/apache/reef/tests/applications/vortex/addone/AddOneTest.java @@ -22,7 +22,8 @@ import org.apache.reef.client.LauncherStatus; import org.apache.reef.tang.Configuration; import org.apache.reef.tests.TestEnvironment; import org.apache.reef.tests.TestEnvironmentFactory; -import org.apache.reef.vortex.driver.VortexConfHelper; +import org.apache.reef.vortex.driver.VortexJobConf; +import org.apache.reef.vortex.driver.VortexMasterConf; import org.junit.After; import org.junit.Assert; import org.junit.Before; @@ -55,9 +56,21 @@ public final class AddOneTest { */ @Test public void testVortexAddOne() { - final Configuration conf = - VortexConfHelper.getVortexConf("TEST_Vortex_AddOneTest", AddOneTestStart.class, 2, 64, 4, 2000); - final LauncherStatus status = this.testEnvironment.run(conf); + + final Configuration vortexMasterConf = VortexMasterConf.CONF + .set(VortexMasterConf.WORKER_NUM, 2) + .set(VortexMasterConf.WORKER_MEM, 64) + .set(VortexMasterConf.WORKER_CORES, 4) + .set(VortexMasterConf.WORKER_CAPACITY, 2000) + .set(VortexMasterConf.VORTEX_START, AddOneTestStart.class) + .build(); + + final VortexJobConf vortexJobConf = VortexJobConf.newBuilder() + .setJobName("TEST_Vortex_AddOneTest") + .setVortexMasterConf(vortexMasterConf) + .build(); + + final LauncherStatus status = this.testEnvironment.run(vortexJobConf.getConfiguration()); Assert.assertTrue("Job state after execution: " + status, status.isSuccess()); } @@ -67,9 +80,20 @@ public final class AddOneTest { */ @Test public void testVortexAddOneCallback() { - final Configuration conf = - VortexConfHelper.getVortexConf("TEST_Vortex_AddOneCallbackTest", AddOneCallbackTestStart.class, 2, 64, 4, 2000); - final LauncherStatus status = this.testEnvironment.run(conf); + final Configuration vortexMasterConf = VortexMasterConf.CONF + .set(VortexMasterConf.WORKER_NUM, 2) + .set(VortexMasterConf.WORKER_MEM, 64) + .set(VortexMasterConf.WORKER_CORES, 4) + .set(VortexMasterConf.WORKER_CAPACITY, 2000) + .set(VortexMasterConf.VORTEX_START, AddOneTestStart.class) + .build(); + + final VortexJobConf vortexJobConf = VortexJobConf.newBuilder() + .setJobName("TEST_Vortex_AddOneCallbackTest") + .setVortexMasterConf(vortexMasterConf) + .build(); + + final LauncherStatus status = this.testEnvironment.run(vortexJobConf.getConfiguration()); Assert.assertTrue("Job state after execution: " + status, status.isSuccess()); } } http://git-wip-us.apache.org/repos/asf/reef/blob/b1a555ed/lang/java/reef-tests/src/test/java/org/apache/reef/tests/applications/vortex/cancellation/TaskletCancellationTest.java ---------------------------------------------------------------------- diff --git a/lang/java/reef-tests/src/test/java/org/apache/reef/tests/applications/vortex/cancellation/TaskletCancellationTest.java b/lang/java/reef-tests/src/test/java/org/apache/reef/tests/applications/vortex/cancellation/TaskletCancellationTest.java index d8ea09c..0d86154 100644 --- a/lang/java/reef-tests/src/test/java/org/apache/reef/tests/applications/vortex/cancellation/TaskletCancellationTest.java +++ b/lang/java/reef-tests/src/test/java/org/apache/reef/tests/applications/vortex/cancellation/TaskletCancellationTest.java @@ -23,7 +23,8 @@ import org.apache.reef.client.LauncherStatus; import org.apache.reef.tang.Configuration; import org.apache.reef.tests.TestEnvironment; import org.apache.reef.tests.TestEnvironmentFactory; -import org.apache.reef.vortex.driver.VortexConfHelper; +import org.apache.reef.vortex.driver.VortexJobConf; +import org.apache.reef.vortex.driver.VortexMasterConf; import org.junit.After; import org.junit.Assert; import org.junit.Before; @@ -53,10 +54,20 @@ public final class TaskletCancellationTest { @Test public void testVortexTaskletCancellation() { - final Configuration conf = - VortexConfHelper.getVortexConf( - "TEST_Vortex_TaskletCancellationTest", TaskletCancellationTestStart.class, 2, 64, 4, 2000); - final LauncherStatus status = this.testEnvironment.run(conf); + final Configuration vortexMasterConf = VortexMasterConf.CONF + .set(VortexMasterConf.WORKER_NUM, 2) + .set(VortexMasterConf.WORKER_MEM, 64) + .set(VortexMasterConf.WORKER_CORES, 4) + .set(VortexMasterConf.WORKER_CAPACITY, 2000) + .set(VortexMasterConf.VORTEX_START, TaskletCancellationTestStart.class) + .build(); + + final VortexJobConf vortexJobConf = VortexJobConf.newBuilder() + .setJobName("TEST_Vortex_TaskletCancellationTest") + .setVortexMasterConf(vortexMasterConf) + .build(); + + final LauncherStatus status = this.testEnvironment.run(vortexJobConf.getConfiguration()); Assert.assertTrue("Job state after execution: " + status, status.isSuccess()); } } http://git-wip-us.apache.org/repos/asf/reef/blob/b1a555ed/lang/java/reef-tests/src/test/java/org/apache/reef/tests/applications/vortex/exception/VortexExceptionTest.java ---------------------------------------------------------------------- diff --git a/lang/java/reef-tests/src/test/java/org/apache/reef/tests/applications/vortex/exception/VortexExceptionTest.java b/lang/java/reef-tests/src/test/java/org/apache/reef/tests/applications/vortex/exception/VortexExceptionTest.java index 948fc04..980c1c1 100644 --- a/lang/java/reef-tests/src/test/java/org/apache/reef/tests/applications/vortex/exception/VortexExceptionTest.java +++ b/lang/java/reef-tests/src/test/java/org/apache/reef/tests/applications/vortex/exception/VortexExceptionTest.java @@ -22,7 +22,8 @@ import org.apache.reef.client.LauncherStatus; import org.apache.reef.tang.Configuration; import org.apache.reef.tests.TestEnvironment; import org.apache.reef.tests.TestEnvironmentFactory; -import org.apache.reef.vortex.driver.VortexConfHelper; +import org.apache.reef.vortex.driver.VortexJobConf; +import org.apache.reef.vortex.driver.VortexMasterConf; import org.junit.After; import org.junit.Assert; import org.junit.Before; @@ -55,10 +56,20 @@ public final class VortexExceptionTest { */ @Test public void testVortexExceptionCallback() { - final Configuration conf = - VortexConfHelper.getVortexConf("TEST_Vortex_ExceptionCallbackTest", - ExceptionCallbackTestStart.class, 2, 64, 4, 2000); - final LauncherStatus status = this.testEnvironment.run(conf); + final Configuration vortexMasterConf = VortexMasterConf.CONF + .set(VortexMasterConf.WORKER_NUM, 2) + .set(VortexMasterConf.WORKER_MEM, 64) + .set(VortexMasterConf.WORKER_CORES, 4) + .set(VortexMasterConf.WORKER_CAPACITY, 2000) + .set(VortexMasterConf.VORTEX_START, ExceptionCallbackTestStart.class) + .build(); + + final VortexJobConf vortexJobConf = VortexJobConf.newBuilder() + .setJobName("TEST_Vortex_AddOneCallbackTest") + .setVortexMasterConf(vortexMasterConf) + .build(); + + final LauncherStatus status = this.testEnvironment.run(vortexJobConf.getConfiguration()); Assert.assertTrue("Job state after execution: " + status, status.isSuccess()); } }
