Source code

001/*
002 * Copyright (c) 2004-2010 The Regents of the University of California.
003 * All rights reserved.
004 *
005 * '$Author: jianwu $'
006 * '$Date: 2012-09-14 22:48:44 +0000 (Fri, 14 Sep 2012) $' 
007 * '$Revision: 30678 $'
008 * 
009 * Permission is hereby granted, without written agreement and without
010 * license or royalty fees, to use, copy, modify, and distribute this
011 * software and its documentation for any purpose, provided that the above
012 * copyright notice and the following two paragraphs appear in all copies
013 * of this software.
014 *
015 * IN NO EVENT SHALL THE UNIVERSITY OF CALIFORNIA BE LIABLE TO ANY PARTY
016 * FOR DIRECT, INDIRECT, SPECIAL, INCIDENTAL, OR CONSEQUENTIAL DAMAGES
017 * ARISING OUT OF THE USE OF THIS SOFTWARE AND ITS DOCUMENTATION, EVEN IF
018 * THE UNIVERSITY OF CALIFORNIA HAS BEEN ADVISED OF THE POSSIBILITY OF
019 * SUCH DAMAGE.
020 *
021 * THE UNIVERSITY OF CALIFORNIA SPECIFICALLY DISCLAIMS ANY WARRANTIES,
022 * INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF
023 * MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE. THE SOFTWARE
024 * PROVIDED HEREUNDER IS ON AN "AS IS" BASIS, AND THE UNIVERSITY OF
025 * CALIFORNIA HAS NO OBLIGATION TO PROVIDE MAINTENANCE, SUPPORT, UPDATES,
026 * ENHANCEMENTS, OR MODIFICATIONS.
027 *
028 */
029
030package org.kepler.actor.job;
031
032import org.apache.commons.logging.Log;
033import org.apache.commons.logging.LogFactory;
034import org.kepler.job.Job;
035import org.kepler.job.JobException;
036import org.kepler.job.JobFactory;
037
038import ptolemy.actor.TypedAtomicActor;
039import ptolemy.actor.TypedIOPort;
040import ptolemy.actor.parameters.PortParameter;
041import ptolemy.data.ArrayToken;
042import ptolemy.data.BooleanToken;
043import ptolemy.data.ObjectToken;
044import ptolemy.data.StringToken;
045import ptolemy.data.expr.Parameter;
046import ptolemy.data.type.BaseType;
047import ptolemy.kernel.CompositeEntity;
048import ptolemy.kernel.util.IllegalActionException;
049import ptolemy.kernel.util.NameDuplicationException;
050
051//////////////////////////////////////////////////////////////////////////
052//// JobCreator
053
054/**
055 * <p>
056 * Create a job. It does not execute it!
057 * </p>
058 * <p>
059 * This actor uses the Job and JobFactory classes to create a job that can be
060 * submitted with JobSubmitter.
061 * </p>
062 * 
063 * <p>
064 * The cmdFile should be the path to the job submission file that you have
065 * created before. This is jobmanager specific! It can be empty, if you want the
066 * JobManager to create the submission file for the (later) selected JobManager.
067 * Note: this capability of the job managers classes is not implemented yet, so
068 * you have to feed here the job submission file (prepared in advance). The file
069 * can either local or remote. Indicate in cmdFileLocal which case holds.
070 * </p>
071 * 
072 * <p>
073 * Executable is the path to the executable if it is located on this local
074 * machine. It will be staged to the remote site (into the job's working
075 * directory) before job submission. If the executable is located on the remote
076 * site, do not give it here, and use its absolute path in your submission file.
077 * </p>
078 * 
079 * <p>
080 * The working directory is the path to the remote working directory in which
081 * you want the job's files to be created. It will be expanded with the job's
082 * unique id assigned by the JobFactory.
083 * </p>
084 * 
085 * <p>
086 * The array of inputfiles is the list of (local) inputfiles to be staged to the
087 * remote site before job submission.
088 * </p>
089 * 
090 * <p>
091 * The array of remotefiles is the list of remote input files that are to be
092 * copied into the job's working directory before job submission.
093 * </p>
094 * 
095 * <p>
096 * The output is the created job (of type Object) that can be used later to
097 * submit the job in JobSubmitter and query its status in JobStatus actor. It is
098 * null if job creation failed.
099 * </p>
100 * 
101 * @author Norbert Podhorszki
102 * @version $Id: JobCreator.java 30678 2012-09-14 22:48:44Z jianwu $
103 * @since Ptolemy II 5.0.1
104 */
105public class JobCreator extends TypedAtomicActor {
106        /**
107         * Construct an actor with the given container and name.
108         * 
109         * @param container
110         *            The container.
111         * @param name
112         *            The name of this actor.
113         * @exception IllegalActionException
114         *                If the actor cannot be contained by the proposed
115         *                container.
116         * @exception NameDuplicationException
117         *                If the container already has an actor with this name.
118         */
119        public JobCreator(CompositeEntity container, String name)
120                        throws NameDuplicationException, IllegalActionException {
121                super(container, name);
122
123                // Uncomment the next line to see debugging statements
124                // addDebugListener(new ptolemy.kernel.util.StreamListener());
125
126                // submission file parameter & port
127                cmdFile = new PortParameter(this, "cmdFile", new StringToken(
128                                "/path/to/job.submit"));
129                new Parameter(cmdFile.getPort(), "_showName", BooleanToken.TRUE);
130                cmdFile.setStringMode(true);
131
132                // local/remote submission file flag parameter
133                cmdFileLocal = new Parameter(this, "cmdFileLocal", new BooleanToken(
134                                true));
135                cmdFileLocal.setTypeEquals(BaseType.BOOLEAN);
136
137                // executable file's name parameter & port
138                executable = new PortParameter(this, "executable", new StringToken(""));
139                new Parameter(executable.getPort(), "_showName", BooleanToken.TRUE);
140                executable.setStringMode(true);
141
142                // working dir name parameter & port
143                workdir = new PortParameter(this, "workdir", new StringToken(
144                                ".kepler-hpcc"));
145                workdir.setStringMode(true);
146
147                new Parameter(workdir.getPort(), "_showName", BooleanToken.TRUE);
148
149                // list of input files' names parameter & port
150                inputfiles = new PortParameter(this, "inputfiles", new ArrayToken(
151                                BaseType.STRING));
152
153                new Parameter(inputfiles.getPort(), "_showName", BooleanToken.TRUE);
154
155                // list of remote input files' names parameter & port
156                remotefiles = new PortParameter(this, "remotefiles", new ArrayToken(
157                                BaseType.STRING));
158
159                new Parameter(remotefiles.getPort(), "_showName", BooleanToken.TRUE);
160
161                // Output: job object reference to the submitted job
162                job = new TypedIOPort(this, "job", false, true);
163                job.setTypeEquals(BaseType.OBJECT);
164                new Parameter(job, "_showName", BooleanToken.FALSE);
165        }
166
167        /***********************************************************
168         * ports and parameters
169         */
170
171        /**
172         * The submit file to be used at job submission. Absolute (or relative to
173         * current dir of Java) file path should be provided. Currently you have to
174         * prepare your submit file.
175         * 
176         * <p>
177         * This parameter is read each time in fire().
178         * </p>
179         */
180        public PortParameter cmdFile;
181
182        /**
183         * Specifying whether the cmdFile is locally stored or on the remote target.
184         * 
185         * <p>
186         * This parameter is read each time in fire().
187         * </p>
188         */
189        public Parameter cmdFileLocal;
190
191        /**
192         * The executable file to be used at job submission. Absolute path names, or
193         * relative to current dir of the running java virtual machine, should be
194         * provided. If it is "" then it is considered to be already at the remote
195         * site, otherwise the actor will look for it locally and stage it to the
196         * <i>workdir</i> before job submission.
197         * 
198         * <p>
199         * This parameter is read each time in fire().
200         * </p>
201         */
202        public PortParameter executable;
203
204        /**
205         * The working directory in which the actual job submission command will be
206         * executed (on the remote machine if the job manager is a remote
207         * jobmanager).
208         * 
209         * <p>
210         * It should be an absolute path, or a relative one. In the latter case on
211         * remote machine, the directory path will be relative to the user's home
212         * directory (coming from the use of ssh)
213         * </p>
214         * 
215         * <p>
216         * This parameter is read each time in fire().
217         * </p>
218         */
219        public PortParameter workdir;
220
221        /**
222         * The string array of inputfiles. Absolute path names, or relative to
223         * current dir of the running java virtual machine, should be provided.
224         * 
225         * <p>
226         * This parameter is read each time in fire().
227         * </p>
228         */
229        public PortParameter inputfiles;
230
231        /**
232         * The string array of remote input files. Absolute path names, or relative
233         * to the user home dir on the remote host should be provided.
234         * 
235         * <p>
236         * This parameter is read each time in fire().
237         * </p>
238         */
239        public PortParameter remotefiles;
240
241        /**
242         * The job object. It will be null if the job submission fails.
243         */
244        public TypedIOPort job;
245
246        /***********************************************************
247         * public methods
248         */
249
250        /**
251         * fire
252         * 
253         * @exception IllegalActionException
254         *                Not thrown.
255         */
256        public void fire() throws IllegalActionException {
257                super.fire();
258
259                cmdFile.update();
260                executable.update();
261                workdir.update();
262                inputfiles.update();
263                remotefiles.update();
264
265                String strCmdFile = ((StringToken) cmdFile.getToken()).stringValue();
266                boolean bCmdFileLocal = ((BooleanToken) cmdFileLocal.getToken())
267                                .booleanValue();
268
269                String strExecutable = ((StringToken) executable.getToken())
270                                .stringValue();
271                String strWorkdir = ((StringToken) workdir.getToken()).stringValue();
272                
273                //back compatibility, remove the double quotes at the very beginning and at the very last.
274                strWorkdir  = strWorkdir.replaceAll("^\"|\"$", "");
275                strExecutable  = strExecutable.replaceAll("^\"|\"$", "");
276                strCmdFile  = strCmdFile.replaceAll("^\"|\"$", "");
277
278                // Process the inputfiles parameter.
279                ArrayToken inputTokens = (ArrayToken) inputfiles.getToken();
280                String[] inputArray = null;
281                if (inputTokens.length() >= 1) {
282                        inputArray = new String[inputTokens.length()];
283                        int i;
284                        for (i = 0; i < inputTokens.length(); i++) {
285                                inputArray[i] = (((StringToken) inputTokens.getElement(i))
286                                                .stringValue());
287                        }
288                        // process empty array
289                        if (i == 0 || inputArray[0] == "")
290                                inputArray = null;
291                }
292
293                // Process the remotefiles parameter.
294                ArrayToken remoteTokens = (ArrayToken) remotefiles.getToken();
295                String[] remoteArray = null;
296                if (remoteTokens.length() >= 1) {
297                        remoteArray = new String[remoteTokens.length()];
298                        int i;
299                        for (i = 0; i < remoteTokens.length(); i++) {
300                                remoteArray[i] = (((StringToken) remoteTokens.getElement(i))
301                                                .stringValue());
302                        }
303                        // process empty array
304                        if (i == 0 || remoteArray[0] == "")
305                                remoteArray = null;
306                }
307
308                // create job
309                String strJobID = JobFactory.create();
310                Job _job = JobFactory.get(strJobID);
311
312                try {
313                        // set _job's executable, working dir and input files
314                        if (strExecutable != null && strExecutable.trim().length() > 0)
315                                _job.setExecutable(strExecutable, true, "");
316
317                        if (strWorkdir != null && strWorkdir.trim().length() > 0)
318                                _job.setWorkdir(strWorkdir);
319
320                        if (strCmdFile != null && strCmdFile.trim().length() > 0)
321                                _job.setSubmitFile(strCmdFile, bCmdFileLocal);
322
323                        if (inputArray != null) {
324                                for (int i = 0; i < inputArray.length; i++) {
325                                        if (inputArray[i] != null
326                                                        && inputArray[i].trim().length() > 0)
327                                                _job.setInputFile(inputArray[i], true);
328                                }
329                        }
330
331                        if (remoteArray != null) {
332                                for (int i = 0; i < remoteArray.length; i++) {
333                                        if (remoteArray[i] != null
334                                                        && remoteArray[i].trim().length() > 0)
335                                                _job.setInputFile(remoteArray[i], false);
336                                }
337                        }
338
339                } catch (JobException ex) {
340                        log.error(ex);
341                        JobFactory.remove(strJobID);
342                        strJobID = "";
343                        _job = null;
344                        ex.printStackTrace();
345                        throw new IllegalActionException("JobCreator Error: " + ex.toString());
346                }
347
348                job.send(0, new ObjectToken(_job));
349        }
350
351        private static final Log log = LogFactory
352                        .getLog(JobCreator.class.getName());
353        private static final boolean isDebugging = log.isDebugEnabled();
354
355}