Source code

001/**
002 * Licensed to the Apache Software Foundation (ASF) under one
003 * or more contributor license agreements.  See the NOTICE file
004 * distributed with this work for additional information
005 * regarding copyright ownership.  The ASF licenses this file
006 * to you under the Apache License, Version 2.0 (the
007 * "License"); you may not use this file except in compliance
008 * with the License.  You may obtain a copy of the License at
009 *
010 *   http://www.apache.org/licenses/LICENSE-2.0
011 *
012 * Unless required by applicable law or agreed to in writing,
013 * software distributed under the License is distributed on an
014 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
015 * KIND, either express or implied.  See the License for the
016 * specific language governing permissions and limitations
017 * under the License.
018 */
019package org.apache.reef.examples.retained_evalCLR;
020
021import org.apache.reef.client.ClientConfiguration;
022import org.apache.reef.runtime.local.client.LocalRuntimeConfiguration;
023import org.apache.reef.runtime.yarn.client.YarnClientConfiguration;
024import org.apache.reef.tang.Configuration;
025import org.apache.reef.tang.Injector;
026import org.apache.reef.tang.JavaConfigurationBuilder;
027import org.apache.reef.tang.Tang;
028import org.apache.reef.tang.annotations.Name;
029import org.apache.reef.tang.annotations.NamedParameter;
030import org.apache.reef.tang.exceptions.BindException;
031import org.apache.reef.tang.exceptions.InjectionException;
032import org.apache.reef.tang.formats.AvroConfigurationSerializer;
033import org.apache.reef.tang.formats.CommandLine;
034import org.apache.reef.tang.formats.ConfigurationModule;
035import org.apache.reef.tang.formats.OptionalParameter;
036
037import java.io.File;
038import java.io.IOException;
039import java.util.Arrays;
040import java.util.logging.Level;
041import java.util.logging.Logger;
042
043/**
044 * Retained Evaluators example - main class.
045 */
046public final class Launch {
047
048  /**
049   * Number of REEF worker threads in local mode.
050   */
051  private static final int NUM_LOCAL_THREADS = JobDriver.totalEvaluators;
052  /**
053   * Standard Java logger
054   */
055  private static final Logger LOG = Logger.getLogger(Launch.class.getName());
056
057  /**
058   * This class should not be instantiated.
059   */
060  private Launch() {
061    throw new RuntimeException("Do not instantiate this class!");
062  }
063
064  /**
065   * Parse the command line arguments.
066   *
067   * @param args command line arguments, as passed to main()
068   * @return Configuration object.
069   * @throws BindException configuration error.
070   * @throws IOException   error reading the configuration.
071   */
072  private static Configuration parseCommandLine(final String[] args)
073      throws BindException, IOException {
074    final JavaConfigurationBuilder confBuilder = Tang.Factory.getTang().newConfigurationBuilder();
075    final CommandLine cl = new CommandLine(confBuilder);
076    cl.registerShortNameOfClass(Local.class);
077    cl.registerShortNameOfClass(Command.class);
078    cl.registerShortNameOfClass(NumRuns.class);
079    cl.processCommandLine(args);
080    return confBuilder.build();
081  }
082
083  private static Configuration cloneCommandLineConfiguration(final Configuration commandLineConf)
084      throws InjectionException, BindException {
085    final Injector injector = Tang.Factory.getTang().newInjector(commandLineConf);
086    final JavaConfigurationBuilder cb = Tang.Factory.getTang().newConfigurationBuilder();
087    cb.bindNamedParameter(Command.class, injector.getNamedInstance(Command.class));
088    cb.bindNamedParameter(NumRuns.class, String.valueOf(injector.getNamedInstance(NumRuns.class)));
089    return cb.build();
090  }
091
092  /**
093   * Parse command line arguments and create TANG configuration ready to be submitted to REEF.
094   *
095   * @param args Command line arguments, as passed into main().
096   * @return (immutable) TANG Configuration object.
097   * @throws BindException      if configuration commandLineInjector fails.
098   * @throws InjectionException if configuration commandLineInjector fails.
099   * @throws IOException        error reading the configuration.
100   */
101  private static Configuration getClientConfiguration(final String[] args)
102      throws BindException, InjectionException, IOException {
103
104    final Configuration commandLineConf = parseCommandLine(args);
105
106    final Configuration clientConfiguration = ClientConfiguration.CONF
107        .set(ClientConfiguration.ON_JOB_RUNNING, JobClient.RunningJobHandler.class)
108        .set(ClientConfiguration.ON_JOB_MESSAGE, JobClient.JobMessageHandler.class)
109        .set(ClientConfiguration.ON_JOB_COMPLETED, JobClient.CompletedJobHandler.class)
110        .set(ClientConfiguration.ON_JOB_FAILED, JobClient.FailedJobHandler.class)
111        .set(ClientConfiguration.ON_RUNTIME_ERROR, JobClient.RuntimeErrorHandler.class)
112        .build();
113
114    // TODO: Remove the injector, have stuff injected.
115    final Injector commandLineInjector = Tang.Factory.getTang().newInjector(commandLineConf);
116    final boolean isLocal = commandLineInjector.getNamedInstance(Local.class);
117    final Configuration runtimeConfiguration;
118    if (isLocal) {
119      LOG.log(Level.INFO, "Running on the local runtime");
120      runtimeConfiguration = LocalRuntimeConfiguration.CONF
121          .set(LocalRuntimeConfiguration.NUMBER_OF_THREADS, NUM_LOCAL_THREADS)
122          .build();
123    } else {
124      LOG.log(Level.INFO, "Running on YARN");
125      runtimeConfiguration = YarnClientConfiguration.CONF.build();
126    }
127
128    return Tang.Factory.getTang()
129        .newConfigurationBuilder(runtimeConfiguration, clientConfiguration,
130            cloneCommandLineConfiguration(commandLineConf))
131        .build();
132  }
133
134  private static ConfigurationModule addAll(final ConfigurationModule conf, final OptionalParameter<String> param, final File folder) {
135    ConfigurationModule result = conf;
136    for (final File f : folder.listFiles()) {
137      if (f.canRead() && f.exists() && f.isFile()) {
138        result = result.set(param, f.getAbsolutePath());
139      }
140    }
141    return result;
142  }
143
144  /**
145   * Main method that starts the Retained Evaluators job.
146   *
147   * @param args command line parameters.
148   */
149  public static void main(final String[] args) {
150    try {
151      final File dotNetFolder = new File(args[0]).getAbsoluteFile();
152      String[] removedArgs = Arrays.copyOfRange(args, 1, args.length);
153
154      final Configuration config = getClientConfiguration(removedArgs);
155      LOG.log(Level.INFO, "Configuration:\n--\n{0}--",
156          new AvroConfigurationSerializer().toString(config));
157      final Injector injector = Tang.Factory.getTang().newInjector(config);
158      final JobClient client = injector.getInstance(JobClient.class);
159      client.submit(dotNetFolder);
160      client.waitForCompletion();
161      LOG.info("Done!");
162    } catch (final BindException | InjectionException | IOException ex) {
163      LOG.log(Level.SEVERE, "Job configuration error", ex);
164    }
165  }
166
167  /**
168   * Command line parameter: a command to run. e.g. "echo Hello REEF"
169   */
170  @NamedParameter(doc = "The shell command", short_name = "cmd", default_value = "*INTERACTIVE*")
171  public static final class Command implements Name<String> {
172  }
173
174  /**
175   * Command line parameter: number of experiments to run.
176   */
177  @NamedParameter(doc = "Number of times to run the command",
178      short_name = "num_runs", default_value = "1")
179  public static final class NumRuns implements Name<Integer> {
180  }
181
182  /**
183   * Command line parameter = true to run locally, or false to run on YARN.
184   */
185  @NamedParameter(doc = "Whether or not to run on the local runtime",
186      short_name = "local", default_value = "true")
187  public static final class Local implements Name<Boolean> {
188  }
189}