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

Timestamp:

2017-05-18T20:38:56+12:00 (7 years ago)

Author:

ak19

Message:

Major changes to SafeProcess and classes of the download package, by bringing the final GLI (and final Greenstone) class DownloadJob over to use SafeProcess. Significant changes to SafeProcess: 1. Introduction of cancelRunningProcess as a new method, so that callers don't need to know how cancelling a process is implemented (as an interrupt) nor do they need to know what thread they ought to interrupt (which should be the thread that launched SafeProcess), nor do they need to know of the complexity surrounding the join() blocking call which should not be interrupted. 2. Introduction of the SafeProcess.MainHandler interface that provides methods that allow implementers to write hooks to various stages of the SafeProcess' internal process' life cycle. 3. moved process cleanUp() code into a reusable method within SafeProcess. Significant changes to DownloadJob and its associated DownloadProgressBar and DownloadScrollPane classes: 1. Unused member vars removed or commented out and those that can be declared final are now declared so, as a programming pricinple for thread safety, since DownloadJob and the other download classes will have to interact with multiple threads and could be called by different threads. 2. Replaced existing actionPerformed() and callDownload() of DownloadJob with SafeProcess and implemented the necessary Hooks in the SafeProcess.MainHandler() interface to ensure that perl is still told to terminate wget on a cancel action. 3. Replaced DownloadScrollPane's deleteDownloadJob() with a new version that now more responsively removes its DownloadProgressBar (with controls) for a DownloadJob. It's called by the SafeProcess.MainHandler interface hooks implemented by DownloadJob, so DownloadScrollPane no longer needs to wait() to be notify()ed when a process has cleaned up on premature termination by a cancel action. 4. Made the necessary methods in DownloadProgressBar synchronized for thread safety. 5. GShell now uses SafeProcess' new cancelRunningProcess() method in place of directly calling interrupt on the (GShell) thread that launched SafeProcess.

File:

• main/trunk/gli/src/org/greenstone/gatherer/util/SafeProcess.java (modified) (23 diffs)

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

@@ -10 +10 @@
 import java.io.OutputStream;
 import java.io.OutputStreamWriter;
+import java.net.Socket;
 import java.util.Arrays;
 import java.util.Scanner;
 import java.util.Stack;
+import javax.swing.SwingUtilities;
+
 
 import com.sun.jna.*;
@@ -39 +42 @@
     public static final int STDIN = 2;
     // can't make this variable final and init in a static block, because it needs to use other SafeProcess static methods which rely on this in turn:
-    public static String WIN_KILL_CMD; 
-        
+    public static String WIN_KILL_CMD;
+
+    /**
+     * Boolean interruptible is used to mark any sections of blocking code that should not be interrupted
+     * with an InterruptedExceptions. At present only the cancelRunningProcess() attempts to do such a thing
+     * and avoids doing so when interruptible is false.
+     * Note that interruptible is also used as a lock, so remember to synchronize on it when using it!
+    */
+    public Boolean interruptible = Boolean.TRUE; 
     
     // charset for reading process stderr and stdout streams
@@ -55 +65 @@
     private Process process = null;
     private boolean forciblyTerminateProcess = false;
+
+    /** a ref to the thread in which the Process is being executed (the thread wherein Runtime.exec() is called) */
+    private Thread theProcessThread = null;
     
     // output from running SafeProcess.runProcess()
@@ -64 +77 @@
     // allow callers to process exceptions of the main process thread if they want
     private ExceptionHandler exceptionHandler = null;
+    /** allow callers to implement hooks that get called during the main phases of the internal
+     * process' life cycle, such as before and after process.destroy() gets called
+    */
+    private MainProcessHandler mainHandler = null;
 
     // whether std/err output should be split at new lines
@@ -110 +127 @@
     }
 
+    /** to set a handler that will handle the main (SafeProcess) thread,
+     * implementing the hooks that will get called during the internal process' life cycle,
+     * such as before and after process.destroy() is called */
+    public void setMainHandler(MainProcessHandler handler) {
+    this.mainHandler = handler;
+    }
+
     // set if you want the std output or err output to have \n at each newline read from the stream
     public void setSplitStdOutputNewLines(boolean split) {
@@ -118 +142 @@
     }
 
+    
+    /*
+    public boolean canInterrupt() {
+    boolean canInterrupt;
+    synchronized(interruptible) {
+        canInterrupt = interruptible.booleanValue();
+    }
+    return canInterrupt;
+    }
+    */
+
+    /**
+     * Call this method when you want to prematurely and safely terminate any process
+     * that SafeProcess may be running.
+     * You may want to implement the SafeProcess.MainHandler interface to write code
+     * for any hooks that will get called during the process' life cycle.
+     * @return false if process has already terminated or if it was already terminating
+     * when cancel was called. In such cases no interrupt is sent. Returns boolean sentInterrupt.
+     */
+    public synchronized boolean cancelRunningProcess() {
+    // on interrupt:
+    // - forciblyTerminate will be changed to true if the interrupt came in when the process was
+    // still running (and so before the process' streams were being joined)
+    // - and forciblyTerminate will still remain false if the interrupt happens when the process'
+    // streams are being/about to be joined (hence after the process naturally terminated).
+    // So we don't touch the value of this.forciblyTerminate here.
+    // The value of forciblyTerminate determines whether Process.destroy() and its associated before
+    // and after handlers are called or not: we don't bother destroying the process if it came to
+    // a natural end.
+
+    // no process to interrupt, so we're done
+    if(this.process == null) {
+        log("@@@ No Java process to interrupt.");
+        return false;
+    }
+    
+    boolean sentInterrupt = false;
+
+    // can't interrupt when SafeProcess is joining (cleanly terminating) worker threads
+    // have to wait until afterward     
+    if (interruptible) {
+        // either way, we can now interrupt the thread - if we have one (we should)
+        if(this.theProcessThread != null) { // we're told which thread should be interrupted
+        this.theProcessThread.interrupt();
+        log("@@@ Successfully sent interrupt to process.");
+        sentInterrupt = true;
+        }
+    }
+    else { // wait for join()s to finish.
+        // During and after joining(), there's no need to interrupt any more anyway: no calls
+        // subsequent to joins() block, so everything thereafter is insensitive to InterruptedExceptions.
+
+        if(SwingUtilities.isEventDispatchThread()) {
+        log("#### Event Dispatch thread, returning");
+        return false;
+        }
+
+        while(!interruptible) {
+
+        log("######### Waiting for process to become interruptible...");
+        
+        // https://docs.oracle.com/javase/tutorial/essential/concurrency/guardmeth.html
+        // wait will release lock on this object, and regain it when loop condition interruptible is true
+        try {
+            this.wait(); // can't interrupt when SafeProcess is joining (cleanly terminating) worker threads, so wait
+        } catch(Exception e) {
+            log("@@@ Interrupted exception while waiting for SafeProcess' worker threads to finish joining on cancelling process");
+        }
+        }
+
+        // now the process is sure to have ended as the worker threads would have been joined
+    }
+
+    return sentInterrupt;
+    }
+
+    
     // In future, think of changing the method doRuntimeExec() over to using ProcessBuilder
     // instead of Runtime.exec(). ProcessBuilder seems to have been introduced from Java 5.
     // https://docs.oracle.com/javase/7/docs/api/java/lang/ProcessBuilder.html
-    // https://zeroturnaround.com/rebellabs/how-to-deal-with-subprocesses-in-java/
+    // See also https://zeroturnaround.com/rebellabs/how-to-deal-with-subprocesses-in-java/
     // which suggests using Apache Common Exec to launch processes and says what will be forthcoming in Java 9
     
@@ -159 +260 @@
     }
 
+    this.theProcessThread = Thread.currentThread(); // store a ref to the thread wherein the Process is being run
     return prcs;
     }
@@ -173 +275 @@
     outputGobbler.start();
     
-    // any error???
     try {
-        this.exitValue = process.waitFor(); // can throw an InterruptedException if process did not terminate
+        this.exitValue = process.waitFor(); // can throw an InterruptedException if process was cancelled/prematurely terminated
     } catch(InterruptedException ie) {
         log("*** Process interrupted (InterruptedException). Expected to be a Cancel operation.");
@@ -185 +286 @@
         // propagate interrupts to worker threads here
         // unless the interrupt emanated from any of them in any join(),
-        // which will be caught by caller's catch on InterruptedException.
+        // which will be caught by the calling method's own catch on InterruptedException.
         // Only if the thread that SafeProcess runs in was interrupted
         // should we propagate the interrupt to the worker threads.
@@ -215 +316 @@
         
         //log("Process exitValue: " + exitValue);
+        ///log("@@@@ Before join phase. Forcibly terminating: " + this.forciblyTerminateProcess);
         
         // From the comments of 
@@ -225 +327 @@
         // Any of these can throw InterruptedExceptions too
         // and will be processed by the calling function's catch on InterruptedException.
-        // However, no one besides us will interrupting these threads I think...
-        // and we won't be throwing the InterruptedException from within the threads...
-        // So if any streamgobbler.join() call throws an InterruptedException, that would be unexpected
-
-        outputGobbler.join(); 
+
+
+        // Thread.joins() below are blocking calls, same as Process.waitFor(), and a cancel action could
+        // send an interrupt during any Join: the InterruptedException ensuing will then break out of the
+        // joins() section. We don't want that to happen: by the time the joins() start happening, the
+        // actual process has finished in some way (naturally terminated or interrupted), and nothing
+        // should interrupt the joins() (nor ideally any potential p.destroy after that).
+        // So we mark the join() section as an un-interruptible section, and make anyone who's seeking
+        // to interrupt just then first wait for this Thread (in which SafeProcess runs) to become
+        // interruptible again. Thos actually assumes anything interruptible can still happen thereafter
+        // when in reality, none of the subsequent actions after the joins() block. So they nothing
+        // thereafter, which is the cleanup phase, will actually respond to an InterruptedException.
+
+        
+        if(this.mainHandler != null) {
+        // this method can unset forcible termination flag
+        // if the process had already naturally terminated by this stage:
+        this.forciblyTerminateProcess = mainHandler.beforeWaitingForStreamsToEnd(this.forciblyTerminateProcess);
+        }
+
+        ///log("@@@@ After beforeJoin Handler. Forcibly terminating: " + this.forciblyTerminateProcess);
+
+        // Anyone could interrupt/cancel during waitFor() above,
+        // but no one should interrupt while the worker threads come to a clean close,
+        // so make anyone wanting to cancel the process at this stage wait()
+        // until we're done with the join()s:
+        synchronized(interruptible) {
+        interruptible = Boolean.FALSE;
+        }
+        //Thread.sleep(5000); // Uncomment to test this uninterruptible section, also comment out block checking for
+            // EventDispatchThread in cancelRunningProcess() and 2 calls to progress.enableCancelJob() in DownloadJob.java
+        outputGobbler.join();
         errorGobbler.join();
-        inputGobbler.join();
-        
-        
+        inputGobbler.join(); 
+
+        synchronized(interruptible) {
+        interruptible = Boolean.TRUE;
+        }
+
+        ///log("@@@@ Join phase done...");
+
+        // notify any of those waiting to interrupt this thread, that they may feel free to do so again
+        // https://docs.oracle.com/javase/tutorial/essential/concurrency/guardmeth.html
+        synchronized(this) {
+        this.notify();
+        }
+
         // set the variables that the code which created a SafeProcess object may want to inspect
         this.outputStr = outputGobbler.getOutput();
-        this.errorStr = errorGobbler.getOutput();       
+        this.errorStr = errorGobbler.getOutput();
+        
+        // call the after join()s hook
+        if(this.mainHandler != null) {
+        this.forciblyTerminateProcess = mainHandler.afterStreamsEnded(this.forciblyTerminateProcess);
+        }
     }
 
@@ -283 +428 @@
         
         Thread.currentThread().interrupt();
-    } finally { 
-
-        if( this.forciblyTerminateProcess ) {
-        destroyProcess(process); // see runProcess() below      
-        }
-        process = null;
-        this.forciblyTerminateProcess = false; // reset
+    } finally {
+        
+        cleanUp("SafeProcess.runBasicProcess");
     }
     return this.exitValue;
@@ -347 +488 @@
 
 
-            // 3. kick off the stream gobblers
+        // 3. kick off the stream gobblers
         this.exitValue = waitForWithStreams(inputGobbler, outputGobbler, errorGobbler);
        
@@ -365 +506 @@
         log("@@@@ Unexpected InterruptedException when waiting for process stream gobblers to die");
         } else {
-        log("*** Unexpected InterruptException when waiting for process stream gobblers to die:" + ie.getMessage(), ie);
+        log("*** Unexpected InterruptException when waiting for process stream gobblers to die: " + ie.getMessage(), ie);
         }
 
@@ -371 +512 @@
         Thread.currentThread().interrupt();
     
-    } finally { 
-        //String cmd = (this.command == null) ? Arrays.toString(this.command_args) : this.command;
-        //log("*** In finally of SafeProcess.runProcess(3 params): " + cmd);
-
-        if( this.forciblyTerminateProcess ) {
-        log("*** Going to call process.destroy 2");
-        destroyProcess(process);                
-        log("*** Have called process.destroy 2");
-        }
-        process = null;
-        this.forciblyTerminateProcess = false; // reset
+    } finally {
+        
+        cleanUp("SafeProcess.runProcess(3 params)");
     }
     
@@ -424 +557 @@
 
 
-        // 4. kick off the stream gobblers
+        // 3. kick off the stream gobblers
         this.exitValue = waitForWithStreams(inputGobbler, outputGobbler, errorGobbler);
        
@@ -452 +585 @@
         Thread.currentThread().interrupt(); // re-interrupt the thread - which thread? Infinite loop?
     
-    } finally { 
-
-        // Moved into here from GS2PerlConstructor and GShell.runLocal() which said
-        // "I need to somehow kill the child process. Unfortunately Thread.stop() and Process.destroy() both fail to do this. But now, thankx to the magic of Michaels 'close the stream suggestion', it works fine (no it doesn't!)"
-        // http://steveliles.github.io/invoking_processes_from_java.html
-        // 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( this.forciblyTerminateProcess ) {
-        log("*** Going to call process.destroy 1");
-        destroyProcess(process);    
-        log("*** Have called process.destroy 1");
-        }
-        process = null;
-        this.forciblyTerminateProcess = false; //reset      
+    } finally {
+        
+        cleanUp("SafeProcess.runProcess(2 params)");    
     }
     
     return this.exitValue;
+    }
+
+    private void cleanUp(String callingMethod) {
+    
+    // Moved into here from GS2PerlConstructor and GShell.runLocal() which said
+    // "I need to somehow kill the child process. Unfortunately Thread.stop() and Process.destroy() both fail to do this. But now, thankx to the magic of Michaels 'close the stream suggestion', it works fine (no it doesn't!)"
+    // http://steveliles.github.io/invoking_processes_from_java.html
+    // 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 " + callingMethod + ": " + cmd);
+
+    // if we're forcibly terminating the process, call the before- and afterDestroy hooks
+    // besides actually destroying the process
+    if( this.forciblyTerminateProcess ) {
+        log("*** Going to call process.destroy from " + callingMethod);
+
+        if(mainHandler != null) mainHandler.beforeProcessDestroy();
+        SafeProcess.destroyProcess(process); // see runProcess(2 args/3 args)
+        if(mainHandler != null) mainHandler.afterProcessDestroy();
+
+        log("*** Have called process.destroy from " + callingMethod);
+    }   
+
+    process = null;
+    this.theProcessThread = null; // let the process thread ref go too
+    boolean wasForciblyTerminated = this.forciblyTerminateProcess;
+    this.forciblyTerminateProcess = false; // reset
+    
+    if(mainHandler != null) mainHandler.doneCleanup(wasForciblyTerminated);
     }
 
@@ -608 +757 @@
 // e.g. full-import.pl may be terminated with p.destroy(), but it launches import.pl which is left running until it naturally terminates.
 static void destroyProcess(Process p) {
+    log("### in SafeProcess.destroyProcess(Process p)");
+
     // If it isn't windows, process.destroy() terminates any child processes too
     if(!Utility.isWindows()) {
@@ -626 +777 @@
     // so we can use it to find the pids of any subprocesses it launched in order to terminate those too.
     
-    long processID = SafeProcess.getProcessID(p);       
-    log("Attempting to terminate sub processes of Windows process with pid " + processID);
-    terminateSubProcessesRecursively(processID, p);
+    long processID = SafeProcess.getProcessID(p);
+    if(processID == -1) { // the process doesn't exist or no longer exists (terminated naturally?)
+    p.destroy(); // minimum step, do this anyway, at worst there's no process and this won't have any effect
+    } else {
+    log("Attempting to terminate sub processes of Windows process with pid " + processID);
+    terminateSubProcessesRecursively(processID, p);
+    }
     
 }
@@ -706 +861 @@
 private static String getWinProcessKillCmd(Long processID) {
     // check if we first need to init WIN_KILL_CMD. We do this only once, but can't do it in a static codeblock
-    
+    // because of a cyclical dependency regarding this during static initialization
+
     if(WIN_KILL_CMD == null) {      
     if(SafeProcess.isAvailable("wmic")) {
@@ -734 +890 @@
 // where is not part of winbin. where is a system command on windows, but only since 2003, https://ss64.com/nt/where.html
 // There is no `where` on Linux/Mac, must use which for them.
-// On windows, "which tskill" fails but "which" succeeds on taskkill|wmic|browser names.
+// On windows, "which tskill" fails (and "where tskill" works), but "which" succeeds on taskkill|wmic|browser names.
 public static boolean isAvailable(String program) {     
     try {
@@ -764 +920 @@
 public static interface ExceptionHandler {
 
-    // when implementing ExceptionHandler.gotException(), if it manipulates anything that's
-    // not threadsafe, declare gotException() as a synchronized method to ensure thread safety
-    public void gotException(Exception e); // can't declare as synchronized in interface method declaration
+    /**
+     * Called whenever an exception occurs during the execution of the main thread of SafeProcess
+     * (the thread in which the Process is run).
+     * Since this method can't be declared as synchronized in this interface method declaration,
+     * when implementing ExceptionHandler.gotException(), if it manipulates anything that's
+     * not threadsafe, declare gotException() as a synchronized method to ensure thread safety
+     */
+    public void gotException(Exception e);
+}
+
+/** On interrupting (cancelling) a process, 
+ * if the class that uses SafeProcess wants to do special handling 
+ * either before and after join() is called on all the worker threads,
+ * or, only on forcible termination, before and after process.destroy() is to be called,
+ * then that class can implement this MainProcessHandler interface
+ */
+public static interface MainProcessHandler {
+    /** 
+     * Called before the streamgobbler join()s.
+     * If not overriding, the default implementation should be:
+     * public boolean beforeWaitingForStreamsToEnd(boolean forciblyTerminating) { return forciblyTerminating; }
+     * When overriding:
+     * @param forciblyTerminating is true if currently it's been decided that the process needs to be
+     * forcibly terminated. Return false if you don't want it to be. For a basic implementation,
+     * return the parameter.
+     * @return true if the process is still running and therefore still needs to be destroyed, or if
+     * you can't determine whether it's still running or not. Process.destroy() will then be called.
+     * @return false if the process has already naturally terminated by this stage. Process.destroy()
+     * won't be called, and neither will the before- and after- processDestroy methods of this class.
+    */
+    public boolean beforeWaitingForStreamsToEnd(boolean forciblyTerminating);
+    /** 
+     * Called after the streamgobbler join()s have finished.
+     * If not overriding, the default implementation should be:
+     * public boolean afterStreamsEnded(boolean forciblyTerminating) { return forciblyTerminating; }
+     * When overriding:
+     * @param forciblyTerminating is true if currently it's been decided that the process needs to be
+     * forcibly terminated. Return false if you don't want it to be. For a basic implementation,
+     * return the parameter (usual case).
+     * @return true if the process is still running and therefore still needs to be destroyed, or if
+     * can't determine whether it's still running or not. Process.destroy() will then be called.
+     * @return false if the process has already naturally terminated by this stage. Process.destroy()
+     * won't be called, and neither will the before- and after- processDestroy methods of this class.
+    */
+    public boolean afterStreamsEnded(boolean forciblyTerminating);
+    /**
+     * called after join()s and before process.destroy()/destroyProcess(Process), iff forciblyTerminating
+     */
+    public void beforeProcessDestroy(); 
+    /**
+     * Called after process.destroy()/destroyProcess(Process), iff forciblyTerminating
+     */
+    public void afterProcessDestroy();
+
+    /** 
+     * Always called after process ended: whether it got destroyed or not
+     */
+    public void doneCleanup(boolean wasForciblyTerminated);
 }
 
@@ -1080 +1291 @@
     }
 
+    // in Java 6, Sockets don't yet implement Closeable
+    public static boolean closeSocket(Socket resourceHandle) {
+    boolean success = false;
+    try {
+        if(resourceHandle != null) {
+        resourceHandle.close();
+        resourceHandle = null;
+        success = true;
+        }
+    } catch(Exception e) {
+        log("Exception closing resource: " + e.getMessage(), e);
+        resourceHandle = null;
+        success = false;
+    } finally {
+        return success;
+    }
+    }
+
     public static boolean closeProcess(Process prcs) {
     boolean success = true;