1 /* Copyright (c) 2009 Peter Troshin
\r
3 * JAva Bioinformatics Analysis Web Services (JABAWS) @version: 1.0
\r
5 * This library is free software; you can redistribute it and/or modify it under the terms of the
\r
6 * Apache License version 2 as published by the Apache Software Foundation
\r
8 * This library is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY; without
\r
9 * even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the Apache
\r
10 * License for more details.
\r
12 * A copy of the license is in apache_license.txt. It is also available here:
\r
13 * @see: http://www.apache.org/licenses/LICENSE-2.0.txt
\r
15 * Any republication or derived work distributed in source code form
\r
16 * must include this copyright and license notice.
\r
19 package compbio.engine;
\r
21 import java.io.File;
\r
22 import java.security.InvalidParameterException;
\r
23 import java.util.List;
\r
25 import org.apache.log4j.Logger;
\r
27 import compbio.data.sequence.FastaSequence;
\r
28 import compbio.engine.client.ConfExecutable;
\r
29 import compbio.engine.client.ConfiguredExecutable;
\r
30 import compbio.engine.client.Executable;
\r
31 import compbio.engine.client.PathValidator;
\r
32 import compbio.engine.client.EngineUtil;
\r
33 import compbio.engine.cluster.drmaa.AsyncClusterRunner;
\r
34 import compbio.engine.cluster.drmaa.ClusterRunner;
\r
35 import compbio.engine.conf.DirectoryManager;
\r
36 import compbio.engine.conf.PropertyHelperManager;
\r
37 import compbio.engine.local.AsyncLocalRunner;
\r
38 import compbio.engine.local.LocalRunner;
\r
39 import compbio.metadata.JobSubmissionException;
\r
40 import compbio.util.PropertyHelper;
\r
41 import compbio.util.SysPrefs;
\r
42 import compbio.util.Util;
\r
44 public class Configurator {
\r
46 private static Logger log = Logger.getLogger(Configurator.class);
\r
47 private static final PropertyHelper ph = PropertyHelperManager.getPropertyHelper();
\r
49 public static final boolean IS_LOCAL_ENGINE_ENABLED = initBooleanValue("engine.local.enable");
\r
50 public static final boolean IS_CLUSTER_ENGINE_ENABLED = initBooleanValue("engine.cluster.enable");
\r
51 public final static String LOCAL_WORK_DIRECTORY = initLocalDirectory();
\r
52 public final static String CLUSTER_WORK_DIRECTORY = initClusterWorkDirectory();
\r
54 private static boolean initBooleanValue(String key) {
\r
56 String status = ph.getProperty(key);
\r
57 log.debug("Loading property: " + key + " with value: " + status);
\r
58 if (Util.isEmpty(status)) {
\r
61 return new Boolean(status.trim()).booleanValue();
\r
64 private static String initClusterWorkDirectory() {
\r
65 String tmpDir = null;
\r
66 if (IS_CLUSTER_ENGINE_ENABLED) {
\r
67 tmpDir = ph.getProperty("cluster.tmp.directory");
\r
68 if (!Util.isEmpty(tmpDir)) {
\r
69 tmpDir = tmpDir.trim();
\r
71 throw new RuntimeException("Cluster work directory must be provided! ");
\r
73 if (LOCAL_WORK_DIRECTORY != null && LOCAL_WORK_DIRECTORY.equals(CLUSTER_WORK_DIRECTORY)) {
\r
74 throw new InvalidParameterException("Cluster engine output directory must be different of that for local engine!");
\r
80 private static String initLocalDirectory() {
\r
81 String tmp_dir = ph.getProperty("local.tmp.directory");
\r
82 // Use system temporary directory if local.tmp.directory is not defined
\r
83 if (Util.isEmpty(tmp_dir)) {
\r
84 tmp_dir = SysPrefs.getSystemTmpDir();
\r
85 log.debug("local.tmp.directory is not defined using system tmp: " + tmp_dir);
\r
87 if (!PathValidator.isAbsolutePath(tmp_dir)) {
\r
88 log.debug("local.tmp.directory path is relative! " + tmp_dir);
\r
89 tmp_dir = EngineUtil.convertToAbsolute(tmp_dir);
\r
90 log.debug("local.tmp.directory path changed to absolute: " + tmp_dir);
\r
92 return tmp_dir.trim();
\r
96 * Depending on the values defined in the properties
\r
97 * (engine.cluster.enable=true and engine.local.enable=true) return either
\r
98 * Cluster job submission engine {@link #JobRunner} or local job submission
\r
99 * engine {@link #LocalRunner} If both engines enabled than ask
\r
100 * {@link LoadBalancer} for an engine. This method will fall back and return
\r
103 * 1) No engines are defined in the properties or they have been defined incorrectly
\r
105 * 2) Execution environment is Windows as the system cannot really run
\r
106 * cluster submission from windows
\r
108 * @param executable
\r
109 * @return SyncExecutor backed up by either cluster or local engines
\r
110 * @throws JobSubmissionException
\r
112 static Executable.ExecProvider getExecProvider(ConfiguredExecutable<?> executable, List<FastaSequence> dataSet)
\r
113 throws JobSubmissionException {
\r
114 // Look where executable claims to be executed
\r
115 Executable.ExecProvider provider = executable.getSupportedRuntimes();
\r
116 if (!IS_CLUSTER_ENGINE_ENABLED && !IS_LOCAL_ENGINE_ENABLED) {
\r
117 // Both engines disabled!
\r
118 throw new RuntimeException("Both engines are disabled! "
\r
119 + "Check conf/Engine.cluster.properties and conf/Engine.local.properties. At least one engine must be enabled!");
\r
121 if (provider == Executable.ExecProvider.Local) {
\r
122 if (IS_LOCAL_ENGINE_ENABLED) {
\r
123 return Executable.ExecProvider.Local;
\r
125 throw new JobSubmissionException("Executable can be executed only on locally, but local engine is disabled!");
\r
128 if (provider == Executable.ExecProvider.Cluster) {
\r
129 if (IS_CLUSTER_ENGINE_ENABLED) {
\r
130 return Executable.ExecProvider.Cluster;
\r
132 throw new JobSubmissionException("Executable can be executed only on the cluster, but cluster engine is disabled!");
\r
135 // We are here if executable can be executed on both Cluster and Local
\r
136 // engines i.e. provider = Any
\r
137 // If we still here executable supports All exec environments
\r
138 // Check whether we work on windows
\r
139 if (SysPrefs.isWindows) {
\r
140 // no matter what the settings are, we cannot send tasks to the
\r
141 // cluster from windows env
\r
142 return Executable.ExecProvider.Local;
\r
144 // Now if both engines are configured that load balance them
\r
145 if (IS_CLUSTER_ENGINE_ENABLED && IS_LOCAL_ENGINE_ENABLED) {
\r
146 // If the dataset is NULL than base a decision on local engine load
\r
148 if (dataSet == null) {
\r
149 return LoadBalancer.getEngine(executable);
\r
151 // If the dataset is provided, consider it
\r
152 // This should be the main root for any load balancing
\r
154 return LoadBalancer.getEngine(executable, dataSet);
\r
155 } else if (IS_CLUSTER_ENGINE_ENABLED) {
\r
156 return Executable.ExecProvider.Cluster;
\r
158 // If we are here, than local engine is enabled or one of the two will
\r
159 // happen (1) exception is thrown if both engines are disabled
\r
160 // or (2) previous statement will return the cluster engine
\r
161 return Executable.ExecProvider.Local;
\r
164 public static <T> ConfiguredExecutable<T> configureExecutable(Executable<T> executable) throws JobSubmissionException {
\r
166 ConfExecutable<T> confExec = new ConfExecutable<T>(executable, DirectoryManager.getTaskDirectory(executable.getClass()));
\r
167 Executable.ExecProvider provider = getExecProvider(confExec, null);
\r
168 confExec.setExecProvider(provider);
\r
169 setupWorkDirectory(confExec, provider);
\r
173 public static <T> ConfiguredExecutable<T> configureExecutable(Executable<T> executable, List<FastaSequence> dataSet)
\r
174 throws JobSubmissionException {
\r
176 ConfExecutable<T> confExec = new ConfExecutable<T>(executable, DirectoryManager.getTaskDirectory(executable.getClass()));
\r
177 Executable.ExecProvider provider = getExecProvider(confExec, dataSet);
\r
178 confExec.setExecProvider(provider);
\r
179 setupWorkDirectory(confExec, provider);
\r
183 static <T> void setupWorkDirectory(ConfExecutable<T> confExec, Executable.ExecProvider provider) {
\r
184 assert provider != null && provider != Executable.ExecProvider.Any;
\r
185 String workDir = "";
\r
186 if (provider == Executable.ExecProvider.Local) {
\r
187 workDir = Configurator.LOCAL_WORK_DIRECTORY + File.separator + confExec.getTaskId();
\r
189 workDir = Configurator.CLUSTER_WORK_DIRECTORY + File.separator + confExec.getTaskId();
\r
191 // Create working directory for the task
\r
192 File wdir = new File(workDir);
\r
194 log.info("Creating working directory for the task in: " + wdir.getAbsolutePath());
\r
195 // Tell the executable where to get the results
\r
196 confExec.setWorkDirectory(workDir);
\r
199 public static <T> ConfiguredExecutable<T> configureExecutable(Executable<T> executable, Executable.ExecProvider provider)
\r
200 throws JobSubmissionException {
\r
201 if (executable == null) {
\r
202 throw new InvalidParameterException("Executable must be provided!");
\r
204 ConfExecutable<T> confExec = new ConfExecutable<T>(executable, DirectoryManager.getTaskDirectory(executable.getClass()));
\r
205 if (provider == Executable.ExecProvider.Cluster && !IS_CLUSTER_ENGINE_ENABLED) {
\r
206 throw new JobSubmissionException("Cluster engine is disabled or not configured!");
\r
208 if (provider == Executable.ExecProvider.Local && !IS_LOCAL_ENGINE_ENABLED) {
\r
209 throw new JobSubmissionException("Local engine is disabled or not configured!");
\r
211 confExec.setExecProvider(provider);
\r
212 setupWorkDirectory(confExec, provider);
\r
216 public static AsyncExecutor getAsyncEngine(ConfiguredExecutable<?> executable, Executable.ExecProvider provider) {
\r
218 assert provider != Executable.ExecProvider.Any && provider != null;
\r
219 if (provider == Executable.ExecProvider.Cluster) {
\r
220 return new AsyncClusterRunner();
\r
222 return new AsyncLocalRunner();
\r
225 public static SyncExecutor getSyncEngine(ConfiguredExecutable<?> executable, Executable.ExecProvider provider)
\r
226 throws JobSubmissionException {
\r
228 assert provider != Executable.ExecProvider.Any && provider != null;
\r
229 if (provider == Executable.ExecProvider.Cluster) {
\r
230 return ClusterRunner.getInstance(executable);
\r
232 return new LocalRunner(executable);
\r
235 public static AsyncExecutor getAsyncEngine(ConfiguredExecutable<?> executable) {
\r
236 if (isTargetedForLocalExecution(executable)) {
\r
237 return new AsyncLocalRunner();
\r
239 return new AsyncClusterRunner();
\r
242 public static AsyncExecutor getAsyncEngine(String taskId) {
\r
243 if (isLocal(taskId)) {
\r
244 return new AsyncLocalRunner();
\r
246 return new AsyncClusterRunner();
\r
249 public static SyncExecutor getSyncEngine(ConfiguredExecutable<?> executable) throws JobSubmissionException {
\r
250 if (isTargetedForLocalExecution(executable)) {
\r
251 return new LocalRunner(executable);
\r
253 return ClusterRunner.getInstance(executable);
\r
256 static boolean isTargetedForLocalExecution(ConfiguredExecutable<?> executable) {
\r
257 // In the uncommon case that the cluster and local execution temporary
\r
258 // directories are the same, in this case the method return true anyway
\r
261 * Could have done this String taskDir = executable.getWorkDirectory();
\r
262 * int idx = taskDir.lastIndexOf(File.separator); String workDir =
\r
263 * taskDir.substring(0, idx); assert
\r
264 * !(workDir.equals(CLUSTER_WORK_DIRECTORY) && workDir
\r
265 * .equals(LOCAL_WORK_DIRECTORY)) :
\r
266 * "Could not determine executable target!"; if
\r
267 * (workDir.equals(LOCAL_WORK_DIRECTORY)) { return true; }
\r
269 String taskDir = executable.getTaskId();
\r
270 return isLocal(taskDir);
\r
273 static boolean isLocal(String taskId) {
\r
274 if (Util.isEmpty(taskId)) {
\r
275 throw new NullPointerException("TaskId must be provided!");
\r
277 if (!EngineUtil.isValidJobId(taskId)) {
\r
278 throw new InvalidParameterException("TaskId is not valid!");
\r
280 return !taskId.startsWith(ConfExecutable.CLUSTER_TASK_ID_PREFIX);
\r
283 public static String getWorkDirectory(String taskId) {
\r
284 assert !compbio.util.Util.isEmpty(taskId);
\r
285 assert EngineUtil.isValidJobId(taskId);
\r
286 log.info("Getting workdirectory for TaskID: " + taskId);
\r
287 if (taskId.startsWith(ConfExecutable.CLUSTER_TASK_ID_PREFIX)) {
\r
288 return CLUSTER_WORK_DIRECTORY + File.separator + taskId;
\r
290 return LOCAL_WORK_DIRECTORY + File.separator + taskId;
\r