| /* |
| * 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.uima.ducc.cli; |
| |
| import java.io.File; |
| import java.io.FileInputStream; |
| import java.io.FileOutputStream; |
| import java.io.IOException; |
| import java.io.OutputStreamWriter; |
| import java.lang.management.ManagementFactory; |
| import java.net.InetAddress; |
| import java.util.ArrayList; |
| import java.util.Arrays; |
| import java.util.Iterator; |
| import java.util.List; |
| import java.util.Map; |
| import java.util.Properties; |
| import java.util.concurrent.CountDownLatch; |
| import java.util.regex.Matcher; |
| import java.util.regex.Pattern; |
| |
| import org.apache.uima.ducc.common.IDucc; |
| import org.apache.uima.ducc.common.crypto.Crypto; |
| import org.apache.uima.ducc.common.utils.DuccProperties; |
| import org.apache.uima.ducc.common.utils.DuccPropertiesResolver; |
| import org.apache.uima.ducc.common.utils.Utils; |
| import org.apache.uima.ducc.transport.dispatcher.IDuccEventDispatcher; |
| import org.apache.uima.ducc.transport.event.AbstractDuccOrchestratorEvent; |
| import org.apache.uima.ducc.transport.event.IDuccContext.DuccContext; |
| |
| /** |
| * Define common methods and data used by all the DUCC API and CLI. |
| */ |
| public abstract class CliBase |
| implements IUiOptions |
| { |
| |
| private String myClassName = "N/A"; |
| private boolean init_done = false; |
| protected String ducc_home; |
| protected IDuccEventDispatcher dispatcher; |
| |
| protected CommandLine commandLine; |
| |
| protected long friendlyId = -1; |
| protected int returnCode = 0; |
| |
| protected DuccProperties cli_props; |
| protected ArrayList<String> errors = new ArrayList<String>(); |
| protected ArrayList<String> warnings = new ArrayList<String>(); |
| protected ArrayList<String> messages = new ArrayList<String>(); |
| |
| protected boolean debug; |
| private boolean load_defaults = true; |
| |
| protected ConsoleListener console_listener = null; |
| protected boolean suppress_console_log; |
| protected String host_address = "N/A"; |
| protected boolean console_attach = false; |
| protected IDuccCallback consoleCb = null; |
| |
| protected MonitorListener monitor_listener = null; |
| |
| CountDownLatch waiter = null; |
| |
| protected Properties userSpecifiedProperties; |
| |
| /** |
| * All extenders must implement execute - this method does whatever processing on the input |
| * is needed and passes the CLI request to the internal DUCC processes. |
| * |
| * @return Return true if execution works, and false otherwise. |
| * @throws java.lang.Exception The specific exception is a function of the implementor. |
| */ |
| public abstract boolean execute() throws Exception; |
| |
| protected void inhibitDefaults() |
| { |
| this.load_defaults = false; |
| } |
| |
| /* |
| * Get log directory or employ default log directory if not specified |
| * UIMA-4617 Make it relative to the run-time working directory, not HOME |
| */ |
| String getLogDirectory() |
| { |
| |
| String log_directory = cli_props.getProperty(UiOption.LogDirectory.pname()); |
| if(log_directory == null) { |
| // no log directory was specified - default to user's home + "/ducc/logs" |
| log_directory = System.getProperty("user.home") + IDucc.userLogsSubDirectory; |
| } |
| |
| File f; |
| if (log_directory.startsWith(File.separator)) { |
| f = new File(log_directory); |
| } else { |
| // Make the log directory relative to the run-time working directory |
| // NOTE: the working-directory is ALWAYS present when the logging-directory is specified |
| String working_directory = cli_props.getProperty(UiOption.WorkingDirectory.pname()); |
| f = new File(working_directory, log_directory); |
| try { |
| log_directory = f.getCanonicalPath(); |
| } catch (IOException e) { |
| throw new IllegalArgumentException("getLogDirectory: Cannot get full name of log directory " + log_directory); |
| } |
| } |
| |
| cli_props.setProperty(UiOption.LogDirectory.pname(), log_directory); |
| |
| /* |
| * make sure the logdir is actually legal. |
| * JD may also be creating it so to reduce any race or NFS delay blindly create and then test |
| */ |
| f.mkdirs(); |
| if ( ! f.exists() ) { |
| throw new IllegalArgumentException("getLogDirectory: Cannot create log directory " + log_directory); |
| } |
| |
| if ( ! f.isDirectory() ) { |
| throw new IllegalArgumentException("Specified log_directory is not a directory: " + log_directory); |
| } |
| |
| if ( ! f.canWrite() ) { |
| throw new IllegalArgumentException("Log directory exists but cannot be written: " + f); |
| } |
| |
| if ( ! f.canExecute() ) { |
| throw new IllegalArgumentException("Log directory exists but cannot be accessed (must be writable and executable): " + f); |
| } |
| |
| return log_directory; |
| } |
| |
| /* |
| * If the workong directory has been defined (or defaulted) make sure it is absolute |
| */ |
| void setWorkingDirectory() throws IOException |
| { |
| String working_directory = cli_props.getProperty(UiOption.WorkingDirectory.pname()); |
| if (working_directory == null) { |
| return; // Not valid for this request |
| } |
| File f = new File(working_directory); |
| if ( ! f.exists() ) { |
| throw new IllegalArgumentException("Working directory " + working_directory + " does not exist."); |
| } |
| if ( ! f.isAbsolute() ) { |
| cli_props.setProperty(UiOption.WorkingDirectory.pname(), f.getCanonicalPath()); |
| } |
| } |
| |
| /* |
| * Check the syntax & if a service refers to itself -- place-holders already resolved |
| * Strip any broker URL decorations |
| */ |
| boolean check_service_dependencies(String endpoint) |
| { |
| String deps = cli_props.getProperty(UiOption.ServiceDependency.pname()); |
| try { |
| String dependencies = DuccUiUtilities.check_service_dependencies(endpoint, deps); |
| if (dependencies != null) { |
| cli_props.setProperty(UiOption.ServiceDependency.pname(), dependencies); |
| } |
| return true; |
| } catch ( Throwable t ) { |
| message("ERROR:", t.toString()); |
| return false; |
| } |
| } |
| |
| /* |
| * Check if -Xmx value is >= memory size ... if both are specified |
| */ |
| void check_heap_size(String argsOption) { |
| String jvmArgs = cli_props.getProperty(argsOption); |
| String memSize = cli_props.getProperty(UiOption.ProcessMemorySize.pname()); |
| if (jvmArgs == null || memSize == null) { |
| return; |
| } |
| |
| // The numbers may be terminated by a units factor, white-space, or the end of the string |
| // The units factor may be any of kKmMgG ... if omitted is bytes |
| // Match -Xmx###[units-flag] and take the last one specified (IBM & Oracle JREs do this) |
| String xmxRegex = "-Xmx([0-9]+)($|[\\skKmMgG])"; |
| Pattern patn = Pattern.compile(xmxRegex); |
| Matcher matcher = patn.matcher(jvmArgs); |
| Long size = null; |
| String unit = null; |
| while (matcher.find()) { |
| size = Long.valueOf(matcher.group(1)); |
| unit = matcher.group(2); |
| } |
| if (size == null) { |
| return; |
| } |
| if (unit.isEmpty()) { // Was last option in list |
| unit = " "; |
| } |
| char factor = unit.toLowerCase().charAt(0); |
| int shift = "gmk".indexOf(factor); // Number of 1024's to divide size by to get GB |
| if (shift < 0) { // No explicit unit factor ... white-space => bytes |
| shift = 3; |
| } |
| long sizeGB = size >> (10 * shift); // Shift 10 bits per K |
| int memGB = Integer.valueOf(memSize); |
| if (sizeGB >= memGB) { |
| String text = "WARNING - process_memory_size is " + memSize + "G but the max heap is " + size + unit + " --- swapping may occur"; |
| message(text); |
| } |
| } |
| |
| void setUser() |
| throws Exception |
| { |
| /* |
| * marshal user |
| */ |
| String user = DuccUiUtilities.getUser(); |
| cli_props.setProperty(UiOption.User.pname(), user); |
| String property = DuccPropertiesResolver.getInstance().getProperty(DuccPropertiesResolver.ducc_signature_required); |
| if(property != null) { |
| String signatureRequiredProperty = property.trim().toLowerCase(); |
| if(signatureRequiredProperty.equals("on")) { |
| Crypto crypto = new Crypto(user, true); |
| byte[] cypheredMessage = crypto.getSignature(); |
| cli_props.put(UiOption.Signature.pname(), cypheredMessage); |
| } |
| } |
| } |
| |
| /** |
| * Standard init for all except the Service calls that are sent to the SM |
| */ |
| |
| protected synchronized void init(String myClassName, UiOption[] opts, String[] args, DuccProperties cli_props, |
| IDuccCallback consoleCb) throws Exception { |
| this.init (myClassName, opts, args, null, cli_props, consoleCb, "orchestrator"); |
| } |
| |
| protected synchronized void init(String myClassName, UiOption[] opts, Properties props, DuccProperties cli_props, |
| IDuccCallback consoleCb) throws Exception { |
| this.init (myClassName, opts, null, props, cli_props, consoleCb, "orchestrator"); |
| } |
| |
| /** |
| * |
| * @param myClassName Name of the class invoking me, for help string |
| * @param uiOpts Array of IUioptions permitted for this command |
| * @param args Arguments from the command line (or null) |
| * @param props Properties passed in from the API (or null) |
| * @param cli_props (Initially) empty properties file to be filled in |
| * @param consoleCb Console callback object (optional) |
| * @param servlet The name of the http servlet that will serve this request |
| * @throws Exception |
| */ |
| protected synchronized void init(String myClassName, IUiOption[] uiOpts, String[] args, Properties props, |
| DuccProperties cli_props, IDuccCallback consoleCb, String servlet) |
| throws Exception |
| { |
| |
| // Either args or props passed in, not both |
| if (args != null) { |
| CliFixups.cleanupArgs(args, myClassName); |
| } else { |
| CliFixups.cleanupProps(props, myClassName); |
| } |
| |
| if ( init_done ) return; |
| |
| if ( consoleCb == null ) { |
| this.consoleCb = new DefaultCallback(); |
| } else { |
| this.consoleCb = consoleCb; |
| } |
| |
| this.myClassName = myClassName; |
| ducc_home = Utils.findDuccHome(); |
| |
| this.cli_props = cli_props; |
| commandLine = new CommandLine(args, uiOpts, props); |
| try { |
| commandLine.parse(); |
| } catch (Exception e) { |
| usage(e.getMessage()); |
| } |
| |
| if ( commandLine.contains(UiOption.Help)) { |
| usage(null); |
| } |
| |
| debug = commandLine.contains(UiOption.Debug); |
| |
| // Load the specification file, if given on the command line. Note that registration |
| // bypasses the somewhat redundant --specification option so we check two options. |
| // Cannot have both as --specification && --register are never both valid. |
| String fname = null; |
| for (IUiOption spec : new IUiOption[]{ UiOption.Specification, UiOption.Register }) { |
| if ( commandLine.isOption(spec) && commandLine.contains(spec)) { // legal for this command, and also specified? |
| fname = commandLine.get(spec); |
| if (fname.length() == 0) { // Check if --register has no value |
| fname = null; |
| } |
| break; |
| } |
| } |
| // If have a specification file re-parse using it for default values |
| if ( fname != null ) { |
| FileInputStream fis = new FileInputStream(new File(fname)); |
| Properties defaults = new Properties(); |
| defaults.load(fis); |
| fis.close(); |
| CliFixups.cleanupProps(defaults, myClassName); // May correct or drop deprecated options |
| |
| // If invoked with overriding properties add to or replace the defaults |
| if (props != null) { |
| defaults.putAll(props); |
| } |
| commandLine = new CommandLine(args, uiOpts, defaults); |
| commandLine.parse(); |
| } |
| commandLine.verify(); // Insure all the rules specified by the IUiOpts are enforced |
| |
| // Copy options into cli_props |
| setOptions(uiOpts); |
| |
| // Save a copy of the user-specified ones by cloning the underlying properties |
| userSpecifiedProperties = (Properties)((Properties)cli_props).clone(); |
| |
| // May need to suppress logging in console listener, or in the DUCC process. |
| suppress_console_log = cli_props.containsKey(UiOption.SuppressConsoleLog.pname()); |
| |
| // Apply defaults for and fixup the environment if needed |
| // -- unless default loading is inhibited, as it must be for modify operations |
| // What this routine does is fill in all the options that weren't specified |
| // on the command line with their defaults. For 'modify' we want to bypass |
| // this because ONLY the options from the command line should be set. |
| if ( load_defaults ) { |
| setDefaults(uiOpts, suppress_console_log); |
| } |
| |
| // This is not used by DUCC ... allows ducc-mon to display the origin of a job |
| cli_props.setProperty(UiOption.SubmitPid.pname(), ManagementFactory.getRuntimeMXBean().getName()); |
| |
| // First set working-directory as the log-directory may be relative to it |
| setWorkingDirectory(); |
| if ( load_defaults && (getLogDirectory() == null) ) { |
| throw new IllegalArgumentException("Cannot access log directory."); |
| } |
| setUser(); |
| |
| //NodeIdentity ni = new NodeIdentity(); UIMA-3899, use getHostAddress() directly. jrc |
| host_address = InetAddress.getLocalHost().getHostAddress(); |
| |
| initConsoleListener(); |
| |
| // AllInOne doesn't dispatch requests (and local doesn't need a running DUCC!) |
| if (!cli_props.containsKey(UiOption.AllInOne.pname())) { |
| dispatcher = DispatcherFactory.create(cli_props, servlet); |
| } |
| |
| init_done = true; |
| } |
| |
| /* |
| * Save options as properties after resolving any ${..} placeholders |
| */ |
| void setOptions(IUiOption[] uiOpts) |
| throws Exception |
| { |
| // Find the environment variables that are always propagated |
| List<String> envNameList; |
| String envNames = DuccPropertiesResolver.get(DuccPropertiesResolver.ducc_environment_propagated); |
| if (envNames != null) { |
| envNameList = Arrays.asList(envNames.split("\\s+")); |
| } else { |
| envNameList = new ArrayList<String>(0); |
| } |
| |
| Map<IUiOption, String> parsed = commandLine.allOptions(); |
| for (IUiOption opt : parsed.keySet() ) { |
| // If a "flexible" boolean that accepts various true/false values, add it only if true |
| String val; |
| if (opt.optargs() && "true".equals(opt.deflt())) { |
| boolean bval = commandLine.getBoolean(opt); |
| if (bval) { |
| val = ""; |
| } else { |
| if (debug) System.out.println("CLI omitted boolean " + opt.pname() + " = '" + parsed.get(opt) + "'"); |
| continue; |
| } |
| } else { |
| val = parsed.get(opt); |
| if (val == null) { // Should only happen for no-arg options |
| val = ""; |
| } else { |
| if (val.contains("${")) { |
| val = resolvePlaceholders(val, envNameList); |
| } |
| } |
| } |
| val = val.trim(); |
| cli_props.put(opt.pname(), val); |
| if (debug) System.out.println("CLI set " + opt.pname() + " = '" + val + "'"); |
| } |
| } |
| |
| /* |
| * Check for missing required options, set defaults, and validate where possible |
| * Also fixup the environment for all that use it. |
| */ |
| void setDefaults(IUiOption[] uiOpts, boolean suppress_console) throws Exception { |
| ArrayList<String> envNameList = new ArrayList<String>(0); // Why this when are resolving against use caller's environment? |
| for (IUiOption uiopt : uiOpts) { |
| if (!cli_props.containsKey(uiopt.pname())) { |
| // |
| // here deal with stuff that wasn't given explicitly in the command |
| // |
| // our convention - optargs() implies boolean, but it does't have to. |
| // If the arg is not expllicitly specified, we assume |
| // it is (boolean,false) for the sake of dealing with defaults. |
| // -- and then just leave it out -- |
| // similarly - noargs() is definitely boolean, same treatement |
| // |
| if ( (! uiopt.optargs()) && (! uiopt.noargs() ) && uiopt.deflt() != null) { |
| String deflt = uiopt.deflt(); |
| if (deflt.startsWith("$$")) { // Lookup default in ducc.properties |
| deflt = DuccPropertiesResolver.get(deflt.substring(2)); |
| if (deflt == null) { |
| throw new IllegalArgumentException("Invalid default (undefined property) for " + uiopt.pname()); |
| } |
| } else if (deflt.contains("${")) { |
| deflt = resolvePlaceholders(deflt, envNameList); |
| } |
| if (debug) System.out.println("CLI set default: " + uiopt.pname() + " = " + deflt); |
| cli_props.put(uiopt.pname(), deflt); |
| } |
| } else { |
| // |
| // here clean up stuff that was specified but we want to validate it |
| // |
| if (uiopt == UiOption.ProcessMemorySize || uiopt == UiOption.ReservationMemorySize) { |
| String val = cli_props.getStringProperty(uiopt.pname()); |
| if (!val.matches("^\\d+$")) { |
| throw new IllegalArgumentException("Invalid non-numeric value for " + uiopt.pname() + ": " + val); |
| } |
| } |
| } |
| // If this request accepts the --environment option may need to augment it by |
| // renaming LD_LIBRARY_PATH & propagating some user values |
| if (uiopt == UiOption.Environment) { |
| String environment = cli_props.getProperty(uiopt.pname()); |
| String allInOne = cli_props.getProperty(UiOption.AllInOne.pname()); |
| environment = DuccUiUtilities.fixupEnvironment(environment, allInOne); |
| cli_props.setProperty(uiopt.pname(), environment); |
| } |
| } |
| } |
| |
| /* |
| * Resolve any ${..} placeholders against user's system properties and environment |
| * NOTE - this resolves against the caller's sys-props & environment ... the one in DuccUiUtilities |
| * resolves against the process JVM args to match what is done by Spring in UIMA-AS. |
| * 2.0: Leave unresolved entries as is & warn if not one of the always-propagated ones |
| */ |
| private String resolvePlaceholders(String contents, List<String> envNameList) { |
| // Placeholders syntax ${<placeholder>} |
| Pattern pattern = Pattern.compile("\\$\\{(.*?)\\}"); // Stops on first '}' |
| Matcher matcher = pattern.matcher(contents); |
| |
| StringBuffer sb = new StringBuffer(); |
| while (matcher.find()) { |
| final String key = matcher.group(1); |
| String value = System.getProperty(key); |
| if (value == null) { |
| value = System.getenv(key); |
| } |
| if (value != null) { |
| matcher.appendReplacement(sb, value); |
| } else { |
| matcher.appendReplacement(sb, ""); // Can't include the value as it looks like a group specification |
| value = "${" + key + "}"; |
| sb.append(value); |
| if (!envNameList.contains(key)) { |
| message("WARN: undefined placeholder", value, "not replaced"); |
| } |
| } |
| } |
| matcher.appendTail(sb); |
| return sb.toString(); |
| } |
| |
| void saveSpec(String name, DuccProperties props) |
| throws Exception |
| { |
| String directory = props.getProperty("log_directory") + File.separator + friendlyId; |
| String fileName = directory + File.separator + name; |
| File f = new File(directory); |
| |
| f.mkdirs(); |
| if ( ! f.exists() ) { |
| throw new IllegalStateException("saveSpec: Cannot create log directory: " + f.toString()); |
| } |
| |
| // Save the specification (but exclude the 'signature' entry) |
| String comments = null; |
| OutputStreamWriter out = new OutputStreamWriter(new FileOutputStream(fileName)); |
| String key = UiOption.Signature.pname(); |
| if ( props.containsKey(key) ) { |
| Object value = props.remove(key); |
| props.store(out, comments); |
| props.put(key, value); |
| } else { |
| props.store(out, comments); |
| } |
| out.close(); |
| |
| // Also save just the values the user provided |
| fileName = directory + File.separator + DuccUiConstants.user_specified_properties; |
| out = new OutputStreamWriter(new FileOutputStream(fileName)); |
| userSpecifiedProperties.store(out, comments); |
| out.close(); |
| } |
| |
| /** |
| * Extract messages and job pid from reply. This sets messages and errors into the appropriate |
| * structures for the API, and extracts the numeric id of the [job, ducclet, reservation, service] |
| * returned by the Orchestrator. |
| * |
| * @return true if the action succeeded and false otherwise. The action in this case, is whatever |
| * the Orchestrator was asked to do: submit something, cancel something, etc. |
| */ |
| boolean extractReply(AbstractDuccOrchestratorEvent reply) |
| { |
| /* |
| * process reply |
| */ |
| boolean rc = true; |
| Properties properties = reply.getProperties(); |
| @SuppressWarnings("unchecked") |
| ArrayList<String> value_submit_warnings = (ArrayList<String>) properties.get(UiOption.SubmitWarnings.pname()); |
| if(value_submit_warnings != null) { |
| message("Job warnings:"); |
| Iterator<String> reasons = value_submit_warnings.iterator(); |
| while(reasons.hasNext()) { |
| message("WARN:", reasons.next()); |
| } |
| } |
| @SuppressWarnings("unchecked") |
| ArrayList<String> value_submit_errors = (ArrayList<String>) properties.get(UiOption.SubmitErrors.pname()); |
| if(value_submit_errors != null) { |
| message("Job errors:"); |
| Iterator<String> reasons = value_submit_errors.iterator(); |
| while(reasons.hasNext()) { |
| message("ERROR:", reasons.next()); |
| } |
| rc = false; |
| } |
| |
| String pid = reply.getProperties().getProperty(UiOption.JobId.pname()); |
| if (pid == null ) { |
| message("ERROR: Request ID not found in reply"); |
| rc = false; |
| } else { |
| friendlyId = Long.parseLong(pid); |
| if ( friendlyId < 0 ) { |
| message("ERROR: Invalid Request ID", pid); |
| rc = false; |
| } |
| } |
| |
| return rc; |
| } |
| |
| void usage(String message) |
| { |
| if ( message != null ) { |
| System.out.println(message); |
| } |
| System.out.println(commandLine.formatHelp(myClassName)); |
| System.exit(1); |
| } |
| |
| /** |
| * Set a property via the API. This method allows the API user to |
| * build up or override properties after the initial API object is constructed. |
| * |
| * @param key This is the property name. |
| * @param value This is the value of the property. |
| * |
| * @return true if the property is set. Returns false if the property is not legal for this API. |
| */ |
| public boolean setProperty(String key, String value) |
| { |
| |
| if ( ! commandLine.isOptionName(key)) { |
| return false; |
| } |
| cli_props.setProperty(key, value); |
| return true; |
| } |
| |
| protected IDuccCallback getCallback() |
| { |
| return consoleCb; |
| } |
| |
| /** |
| * NOTE: We do NOT want to be intentionally throwing from the CLI. Pls pass e.getMessage() or |
| * e.toString() to this instead of throwing. |
| */ |
| synchronized void message(String ... e ) |
| { |
| if ( e.length > 1 ) { |
| StringBuffer sb = new StringBuffer(); |
| int i = 0; |
| for (i = 0; i < e.length - 1; i++) { |
| sb.append(e[i]); |
| sb.append(' '); |
| } |
| sb.append(e[i]); |
| consoleCb.status(sb.toString()); |
| } else { |
| consoleCb.status(e[0]); |
| } |
| } |
| |
| /** |
| * This returns the return code from the execution of the requested work. Return code is only |
| * available when the monitor wait completes ... if not waiting then assume success. |
| * |
| * @return The exit code of the job, process, etc. |
| */ |
| public int getReturnCode() |
| { |
| waitForCompletion(); |
| return returnCode; |
| } |
| |
| /** |
| * This returns the unique numeric id for the requested work. For submissions (job, reservation, etc) |
| * this is the newly assigned id. |
| * @return The unique numeric id of the job, reservation, etc. |
| */ |
| synchronized public long getDuccId() |
| { |
| return friendlyId; |
| } |
| |
| synchronized void consoleExits() |
| { |
| if ( waiter != null ) waiter.countDown(); |
| } |
| |
| synchronized void monitorExits(int rc) |
| { |
| this.returnCode = rc; |
| if ( waiter != null ) waiter.countDown(); |
| if ( console_listener != null ) { |
| console_listener.shutdown(); |
| } |
| } |
| |
| // TODO TODO TODO - do we have to support lots of these for multi-threaded stuff? Hope not ... |
| protected synchronized void startMonitors(boolean start_stdin, DuccContext context) |
| throws Exception |
| { |
| int wait_count = 0; |
| |
| if ( console_listener != null ) { |
| wait_count++; |
| } |
| |
| boolean monitor_attach = |
| ( |
| cli_props.containsKey(UiOption.WaitForCompletion.pname()) || |
| cli_props.containsKey(UiOption.CancelOnInterrupt.pname()) |
| ); |
| |
| if ( monitor_attach ) { |
| wait_count++; |
| } |
| |
| // Probably over-cautious but create the waiter before starting the threads that use it |
| if ( wait_count > 0 ) { |
| waiter = new CountDownLatch(wait_count); |
| if ( console_listener != null ) { |
| startConsoleListener(start_stdin); |
| } |
| if ( monitor_attach ) { |
| startMonitor(context); |
| } |
| } |
| } |
| |
| protected synchronized void startMonitor(DuccContext context) |
| { |
| monitor_listener = new MonitorListener(this, friendlyId, cli_props, context); |
| Thread mlt = new Thread(monitor_listener); //MonitorListenerThread |
| mlt.start(); |
| } |
| |
| /** |
| * Needs to be done before submitting the job because the job needs the ports. We'll |
| * just define the listener, but not start it until the job monitor starts, in case the |
| * submission fails. |
| */ |
| protected void initConsoleListener() throws Exception { |
| String value; |
| |
| console_attach = cli_props.containsKey(UiOption.AttachConsole.pname()); |
| if (console_attach) { |
| console_listener = new ConsoleListener(this, consoleCb); |
| value = console_listener.getConsoleHostAddress(); |
| if (myClassName.equals(DuccManagedReservationSubmit.class.getName())) { |
| value += "?splitstreams"; // Add a query string so APs have separate streams |
| } |
| } else if (suppress_console_log) { |
| value = "suppress"; |
| } else { |
| return; |
| } |
| // Set the console "suppress" flag or the host:port for the console listener into the env |
| String key = UiOption.Environment.pname(); |
| String env = cli_props.getProperty(key); |
| if (env == null) { |
| env = "DUCC_CONSOLE_LISTENER=" + value; |
| } else { |
| env += " DUCC_CONSOLE_LISTENER=" + value; |
| } |
| cli_props.setProperty(key, env); |
| } |
| |
| /** |
| * Be sure to call this BEFORE submission, to insure the callback address is set in properties. |
| */ |
| protected synchronized void startConsoleListener(boolean start_stdin) |
| throws Exception |
| { |
| if ( console_attach ) { |
| console_listener.startStdin(start_stdin); |
| Thread t = new Thread(console_listener); |
| t.start(); |
| } else { |
| message("WARN: Attermpt to start console but no console listener is defined."); |
| } |
| } |
| |
| protected synchronized void stopListeners() |
| { |
| if ( console_listener != null ) { |
| console_listener.shutdown(); |
| console_listener = null; |
| } |
| |
| if ( monitor_listener != null ) { |
| monitor_listener.shutdown(); |
| monitor_listener = null; |
| } |
| } |
| |
| /** |
| * This is used to find if the remote console is redirected to the local process, and if so, is it still |
| * active. |
| * @return True if the console is still attached to the remote process, false otherwise. |
| */ |
| public boolean isConsoleAttached() |
| { |
| return ( (console_listener != null ) && ( !console_listener.isShutdown())); |
| } |
| |
| /** |
| * Wait for the listeners - maybe a console listener, maybe a job listener, maybe both. |
| * |
| * @return true if a monitor wait was done, false otherwise. A monitor wait |
| * results in a return code from the process. In all other cases |
| * the return code is spurious. |
| */ |
| public boolean waitForCompletion() |
| { |
| try { |
| if ( waiter != null ) { |
| waiter.await(); |
| return true; |
| } |
| } catch (InterruptedException e) { |
| // TODO Auto-generated catch block |
| e.printStackTrace(); |
| } |
| return false; |
| } |
| |
| } |
