Changeset 31639 for main

Timestamp:

2017-05-01T20:07:17+12:00 (7 years ago)

Author:

ak19

Message:

1. Correcting switch statement in SafeProcess and tidying up some debugging. 2. GShell needs to have its own LineByLineHandlers not CustomProcessHandlers, since most of the err/out stream processing code should be the default provided by SafeProcess. 3. GShell.gotException wasn't being called during an expected InterruptedException on Cancel, so corrected that to make the code write a message to the DebugStream as the original code did when a build operation had been cancelled.

Location:

main/trunk/gli/src/org/greenstone/gatherer

Files:

• shell/GShell.java (modified) (8 diffs)
• util/SafeProcess.java (modified) (17 diffs)

main/trunk/gli/src/org/greenstone/gatherer/shell/GShell.java

@@ -152 +152 @@
        } */
 
+    // This method is only used by old_runLocal(). If that is removed, then remove this too
     protected StringBuffer get_stream_char(InputStreamReader isr, StringBuffer line_buffer,
                    BufferedOutputStream bos) throws IOException 
@@ -224 +225 @@
     // old_runLocal uses a potentially unsafe way of running a process and an inefficient way of reading
     // from the process stdout and stderr streams. Replaced by new runLocal() which uses SafeProcess and
-    // the new CustomProcessHandler inner class instantiated for each of the 2 process output streams.
+    // the new LineByLineHandler inner class instantiated for each of the 2 process output streams.
+    // If this method is ever removed from the file, can remove its helper get_stream_char() too.
     private void old_runLocal(String[] args, BufferedOutputStream bos) 
     {
@@ -354 +356 @@
     
     prcs = new SafeProcess(args);
-    SafeProcess.CustomProcessHandler processOutHandler
-        = new SynchronizedProcessHandler(bos, SafeProcess.STDOUT);
-    SafeProcess.CustomProcessHandler processErrHandler
-        = new SynchronizedProcessHandler(bos, SafeProcess.STDERR);
+    SafeProcess.LineByLineHandler processOutLineHandler
+        = new SynchronizedLineByLineHandler(bos, SafeProcess.STDOUT);
+    SafeProcess.LineByLineHandler processErrLineHandler
+        = new SynchronizedLineByLineHandler(bos, SafeProcess.STDERR);
     
     prcs.setExceptionHandler(this); 
-    int exitValue = prcs.runProcess(null, processOutHandler, processErrHandler); // use default procIn handling
+    int exitValue = prcs.runProcess(processOutLineHandler, processErrLineHandler); // use default procIn handling
     
     if(exitValue == 0) {
-        System.err.println("*** Exited normally");
+        //System.err.println("*** Exitted normally");
         status = OK;
         fireMessage(type, typeAsString(type) + "> " + Dictionary.get("GShell.Success"), status, null);
     }
     else {
-        System.err.println("*** Exited abnormally with exit value " + exitValue);
+        //System.err.println("*** Exitted abnormally with exit value " + exitValue);
         status = ERROR;
         fireMessage(type, typeAsString(type) + "> " + Dictionary.get("GShell.Failure"), status, null);
@@ -385 +387 @@
 
         this.interrupt(); // interrupt this thread which is running the SafeProcess prcs
-        // this will propagate the CANCEL status
+        // this will propagate the CANCEL status to any worker threads launched by the SafeProcess
         prcs = null;        
     }
-    hasSignalledStop(); // synchronized. Will set status to CANCELLED in thread-safe manner.    
+    setStatus(CANCELLED); //or: hasSignalledStop(); // synchronized. Either will set status to CANCELLED in thread-safe manner. 
     }
     
@@ -691 +693 @@
 
     if(e instanceof InterruptedException) {
-        SafeProcess.log("*** InterruptedException in GShell.runLocal()");
+        DebugStream.println("We've been asked to stop.");
+        SafeProcess.log("@@@ Interruption to SafeProcess run by GShell.runLocal()");
         setStatus(CANCELLED); // expected exception
     } else {
@@ -706 +709 @@
     // Each instance of this class is run in its own thread by class SafeProcess.InputGobbler.
     // This class deals with each incoming line from the perl process' stderr or stdout streams. One
-    // instance of this class for each stream. However, since multiple instances of this CustomProcessHandler
-    // could be (and in fact, are) writing to the same file in their own threads, several objects, not just
-    // the bufferedwriter object, needed to be made threadsafe.
-    // This class also handles exceptions during the running of the perl process.
-    // The runPerlCommand code originally would do a sendProcessStatus on each exception, so we ensure
+    // instance of this class for each stream. However, since multiple instances of this LineByLineHandler
+    // are firing off events to the same outputstream object (bos) in their own threads, several objects,
+    // not just bost, needed to be made threadsafe. So gotLine needs to be synchronized, and created a
+    // synchronized setStatus() method above too.
+    // This class also handles exceptions that may occur when reading from the perl process' stderr or stout.
+    // The old GShell.runLocal() code would set the GShell.status to ERROR on each exception, so we ensure
     // we do that here too, to continue original behaviour. These calls are also synchronized to make their
     // use of the EventListeners threadsafe.
-    protected class SynchronizedProcessHandler extends SafeProcess.CustomProcessHandler
+    protected class SynchronizedLineByLineHandler extends SafeProcess.LineByLineHandler
     {
     private final BufferedOutputStream bos; // needs to be final to be able to synchronize on the shared object
     
-    public SynchronizedProcessHandler(BufferedOutputStream bos, int src) {
+    public SynchronizedLineByLineHandler(BufferedOutputStream bos, int src) {
         super(src); // will set this.source to STDERR or STDOUT
         this.bos = bos; // caller will close bw, since many more than one
-                            // SynchronizedProcessHandlers are using it
+                            // SynchronizedLineByLineHandlers are using it
     }
 
     // trying to keep synchronized methods as short as possible
-    private void logException(Exception e) {
-        String stream = (this.source == SafeProcess.STDERR) ? "stderr" : "stdout";
-        String msg = "Got exception when processing the perl process' " + stream + " stream.";
+    public void gotException(Exception e) {
+        String msg = "Got exception when processing the perl process' " + SafeProcess.streamToString(this.source) + " stream.";
 
         DebugStream.println(msg); // method is already synchronized
@@ -732 +735 @@
         SafeProcess.log("*** " + msg, e);
 
-        GShell.this.setStatus(GShell.ERROR);
+        GShell.this.setStatus(GShell.ERROR); // synchronized
     }
 
@@ -740 +743 @@
     }
 
-    public void run(Closeable inputStream) {
-        InputStream is = (InputStream) inputStream;
-
-        BufferedReader br = null;
-        try {       
-
-        br = new BufferedReader(new InputStreamReader(is, "UTF-8"));
-        String line=null;
-
-        ///while ( !GShell.this.prcs.processRunning() || hasSignalledStop()) { // need to sync on prcs
-        while ( !Thread.currentThread().isInterrupted() && (line = br.readLine()) != null ) {
-            this.gotLine(line); // synchronized         
-        }       
-
-        } catch (Exception e) {
-        logException(e);
-        } finally {
-        System.err.println("*********");
-        if(Thread.currentThread().isInterrupted()) {
-            log("We've been asked to stop.");
-            
-        }
-        SafeProcess.closeResource(br);
-        String stream = (this.source == SafeProcess.STDERR) ? "stderr" : "stdout";
-        System.err.println("*** In finally of " + stream + " stream");
-        System.err.println("*********");
-        }
-    }
-
-    protected synchronized void gotLine(String line) {
+    // every time we read a line from the SafeProcess' stderr or stdout stream we come here.
+    // a different isntance of SynchronizedL
+    public synchronized void gotLine(String line) {
         fireMessage(GShell.this.type,   // not final, needs synchro
             GShell.this.typeAsString(type) + "> " + line,

main/trunk/gli/src/org/greenstone/gatherer/util/SafeProcess.java

@@ -21 +21 @@
 
 public class SafeProcess {
+    //public static int DEBUG = 0;
 
     public static final int STDERR = 0;
@@ -99 +100 @@
     public void setSplitStdErrorNewLines(boolean split) {
     splitStdErrorNewLines = split;
-    }
-
-    // logger and DebugStream print commands are synchronized, therefore thread safe.
-    public static void log(String msg) {
-    //logger.info(msg);
-
-    System.err.println(msg);
-
-    //DebugStream.println(msg);
-    }
-
-    public static void log(String msg, Exception e) { // Print stack trace on the exception
-    //logger.error(msg, e);
-
-    System.err.println(msg);
-    e.printStackTrace();
-
-    //DebugStream.println(msg);
-    //DebugStream.printStackTrace(e);
-    }
-
-    public static void log(Exception e) {
-    //logger.error(e);
-
-    e.printStackTrace();
-
-    //DebugStream.printStackTrace(e);
-    }
-
-    public static void log(String msg, Exception e, boolean printStackTrace) {
-    if(printStackTrace) { 
-        log(msg, e);
-    } else {
-        log(msg);
-    }
     }
 
@@ -184 +150 @@
         this.exitValue = process.waitFor(); // can throw an InterruptedException if process did not terminate
     } catch(InterruptedException ie) {
-        
         log("*** Process interrupted (InterruptedException). Expected to be a Cancel operation.");
             // don't print stacktrace: an interrupt here is not an error, it's expected to be a cancel action
+        if(exceptionHandler != null) {      
+        exceptionHandler.gotException(ie);
+        }           
         
         // propagate interrupts to worker threads here
@@ -204 +172 @@
         // to swallow the interrupt this time and not let it propagate by commenting out the next line?
         //Thread.currentThread().interrupt(); // re-interrupt the thread
-        
+
         inputGobbler.interrupt();
         errorGobbler.interrupt();
         outputGobbler.interrupt();
-        
+
+        // even after the interrupts, we want to proceed to calling join() on all the worker threads
+        // in order to wait for each of them to die before attempting to destroy the process if it
+        // still hasn't terminated after all that.
     } finally {     
         
@@ -358 +329 @@
     
     } finally { 
-
-        log("*** In finally of SafeProcess.runProcess(3 params)");
+        //String cmd = (this.command == null) ? Arrays.toString(this.command_args) : this.command;
+        //log("*** In finally of SafeProcess.runProcess(3 params): " + cmd);
 
         if( process != null ) {
@@ -440 +411 @@
         // http://www.javaworld.com/article/2071275/core-java/when-runtime-exec---won-t.html?page=2
         // http://mark.koli.ch/leaky-pipes-remember-to-close-your-streams-when-using-javas-runtimegetruntimeexec        
+        
+        //String cmd = (this.command == null) ? Arrays.toString(this.command_args) : this.command;
+        //log("*** In finally of SafeProcess.runProcess(2 params): " + cmd);
+
         if( process != null ) {
         log("*** Going to call process.destroy 1");
@@ -446 +421 @@
         log("*** Have called process.destroy 1");
         }
-        log("*** In finally of SafeProcess.runProcess(2 params)");
     }
     
@@ -484 +458 @@
     protected CustomProcessHandler(int src) {
     this.source = src; // STDERR or STDOUT or STDIN
-
-    // modify threadname to prefix stream src (stdin/stderr/stdout)
-    // Useful for debugging if thread is named clearly
-    String stream;
-    switch(src) {
-    case SafeProcess.STDERR:
-       stream = "stderr"; 
-    case SafeProcess.STDOUT:
-        stream = "stdout";
-    default:
-        stream = "stdin";
-    }
-    Thread.currentThread().setName(stream + Thread.currentThread().getName());
+    }
+
+    public String getThreadNamePrefix() {
+    return SafeProcess.streamToString(this.source); 
     }
     
@@ -510 +475 @@
     protected LineByLineHandler(int src) {
     this.source = src; // STDERR or STDOUT
+    }
+
+    public String getThreadNamePrefix() {
+    return SafeProcess.streamToString(this.source); 
     }
 
@@ -530 +499 @@
     private LineByLineHandler lineByLineHandler = null;
 
+    protected InputStreamGobbler() {
+    super("InputStreamGobbler");
+    }
+
     public InputStreamGobbler(InputStream is)
     {
-    super("InputStreamGobbler"); // thread name
+    this(); // sets thread name
     this.is = is;
     this.split_newlines = false;
@@ -539 +512 @@
     public InputStreamGobbler(InputStream is, boolean split_newlines)
     {
-    super("InputStreamGobbler"); // thread name
+    this(); // sets thread name
     this.is = is;
     this.split_newlines = split_newlines;
@@ -546 +519 @@
     public InputStreamGobbler(InputStream is, CustomProcessHandler customHandler)
     {
-    super("InputStreamGobbler"); // thread name
+    this(); // thread name
     this.is = is;
     this.customHandler = customHandler;
+    this.adjustThreadName(customHandler.getThreadNamePrefix());
+    }
+
+    
+    private void adjustThreadName(String prefix) {  
+    this.setName(prefix + this.getName());
     }
 
     public void setLineByLineHandler(LineByLineHandler lblHandler) {
     this.lineByLineHandler = lblHandler;
+    this.adjustThreadName(lblHandler.getThreadNamePrefix());
     }
 
@@ -579 +559 @@
         lineByLineHandler.gotException(ioe);
         } else {
-        log("Exception when reading from a process' stdout/stderr stream: ", ioe);
+        log("Exception when reading process stream with " + this.getName() + ": ", ioe);
         }
     } finally {
-        log("*********");
         if(this.isInterrupted()) {
-        log("We've been asked to stop.");       
+        log("@@@ Successfully interrupted " + this.getName() + ".");
         }
         SafeProcess.closeResource(br);
-        log("*** In finally of " + this.getName());
-        log("*********");
     }
     }
@@ -619 +596 @@
     private CustomProcessHandler customHandler = null;
 
+    protected OutputStreamGobbler() {
+    super("stdinOutputStreamGobbler"); // thread name
+    }
+
     public OutputStreamGobbler(OutputStream os) {
-    super("OutputStreamGobbler"); // thread name
+    this(); // set thread name
     this.os = os;
     }
@@ -626 +607 @@
     public OutputStreamGobbler(OutputStream os, String inputstr)
     {
-    super("OutputStreamGobbler"); // thread name
+    this(); // set thread name
     this.os = os;
     this.inputstr = inputstr;
@@ -632 +613 @@
 
     public OutputStreamGobbler(OutputStream os, CustomProcessHandler customHandler) {
-    super("OutputStreamGobbler"); // thread name
+    this(); // set thread name
     this.os = os;
     this.customHandler = customHandler;
@@ -692 +673 @@
 } // end static inner class OutputStreamGobbler
 
+//**************** Static methods **************//
+
+
+    // logger and DebugStream print commands are synchronized, therefore thread safe.
+    public static void log(String msg) {
+    //logger.info(msg);
+
+    System.err.println(msg);
+
+    //DebugStream.println(msg);
+    }
+
+    public static void log(String msg, Exception e) { // Print stack trace on the exception
+    //logger.error(msg, e);
+
+    System.err.println(msg);
+    e.printStackTrace();
+
+    //DebugStream.println(msg);
+    //DebugStream.printStackTrace(e);
+    }
+
+    public static void log(Exception e) {
+    //logger.error(e);
+
+    e.printStackTrace();
+
+    //DebugStream.printStackTrace(e);
+    }
+
+    public static void log(String msg, Exception e, boolean printStackTrace) {
+    if(printStackTrace) { 
+        log(msg, e);
+    } else {
+        log(msg);
+    }
+    }
+
+    public static String streamToString(int src) {
+    String stream;
+    switch(src) {
+    case STDERR:
+        stream = "stderr"; 
+        break;
+    case STDOUT:
+        stream = "stdout";
+        break;
+    default:
+        stream = "stdin";
+    }
+    return stream;
+    }
 
 //**************** Useful static methods. Copied from GLI's Utility.java ******************