Thread.kill / Thread.raise do not ever stop threads that spawn sub processes

commit 74dc1032735102de870a0ac248cce86f1fa9be3a
Author: Charles Oliver Nutter <headius@headius.com>
Date:   Mon Jul 2 19:42:26 2012 -0500

    Partial fix for JRUBY-6681.

diff --git a/src/org/jruby/RubyKernel.java b/src/org/jruby/RubyKernel.java
index 60ee679..0d7d981 100644
--- a/src/org/jruby/RubyKernel.java
+++ b/src/org/jruby/RubyKernel.java
@@ -55,6 +55,7 @@ import org.jruby.common.IRubyWarnings.ID;
 import org.jruby.evaluator.ASTInterpreter;
 import org.jruby.exceptions.MainExitException;
 import org.jruby.exceptions.RaiseException;
+import org.jruby.exceptions.ThreadKill;
 import org.jruby.internal.runtime.methods.CallConfiguration;
 import org.jruby.internal.runtime.methods.JavaMethod.JavaMethodNBlock;
 import org.jruby.javasupport.util.RuntimeHelpers;
@@ -1485,6 +1486,12 @@ public class RubyKernel {
         try {
             // NOTE: not searching executable path before invoking args
             tuple = ShellLauncher.runAndWaitPid(runtime, args, output, false);
+        } catch (ThreadKill tk) {
+            // propagate thread.kill
+            throw tk;
+        } catch (RaiseException re) {
+            // propagate thread.raise and other Ruby exceptions during process launch
+            throw re;
         } catch (Exception e) {
             tuple = new long[] {127, -1};
         }
diff --git a/src/org/jruby/internal/runtime/ThreadService.java b/src/org/jruby/internal/runtime/ThreadService.java
index 8360a9e..0bd2a2f 100644
--- a/src/org/jruby/internal/runtime/ThreadService.java
+++ b/src/org/jruby/internal/runtime/ThreadService.java
@@ -329,6 +329,10 @@ public class ThreadService {
             this.type = type;
             this.exception = exception;
         }
+
+        public String toString() {
+            return type.toString();
+        }
     }
 
     public synchronized void deliverEvent(Event event) {
diff --git a/src/org/jruby/threading/PumpedProcessBlockingTask.java b/src/org/jruby/threading/PumpedProcessBlockingTask.java
new file mode 100644
index 0000000..801588a
--- /dev/null
+++ b/src/org/jruby/threading/PumpedProcessBlockingTask.java
@@ -0,0 +1,61 @@
+package org.jruby.threading;
+
+import org.jruby.Ruby;
+import org.jruby.RubyThread;
+import org.jruby.util.ShellLauncher;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+
+/**
+ * An implementation of BlockingTask that (optionally) pumps a process's
+ * streams, waits for its result, and if interrupted destroys the subprocess.
+ */
+public class PumpedProcessBlockingTask implements RubyThread.BlockingTask {
+    public PumpedProcessBlockingTask(Ruby runtime, Process aProcess, InputStream input, OutputStream output, OutputStream error) {
+        this.runtime = runtime;
+        this.aProcess = aProcess;
+        this.input = input;
+        this.output = output;
+        this.error = error;
+        pump = true;
+    }
+
+    public PumpedProcessBlockingTask(Ruby runtime, Process aProcess) {
+        this.runtime = runtime;
+        this.aProcess = aProcess;
+        this.input = null;
+        this.output = null;
+        this.error = null;
+        pump = false;
+    }
+
+    public void run() throws InterruptedException {
+        if (pump) {
+            try {
+                ShellLauncher.handleStreams(runtime, aProcess, input, output, error);
+            } catch (IOException e) {
+                throw runtime.newIOErrorFromException(e);
+            }
+        }
+        result = aProcess.waitFor();
+        finished = true;
+    }
+
+    public void wakeup() {
+        aProcess.destroy();
+    }
+
+    public long result() {return result;}
+    public boolean finished() {return finished;};
+
+    final Ruby runtime;
+    final Process aProcess;
+    final InputStream input;
+    final OutputStream output;
+    final OutputStream error;
+    final boolean pump;
+    long result;
+    boolean finished;
+}
diff --git a/src/org/jruby/util/ShellLauncher.java b/src/org/jruby/util/ShellLauncher.java
index 8f26dde..eb22111 100644
--- a/src/org/jruby/util/ShellLauncher.java
+++ b/src/org/jruby/util/ShellLauncher.java
@@ -53,14 +53,7 @@ import java.util.regex.Matcher;
 import java.util.regex.Pattern;
 
 import jnr.posix.util.ProcessMaker;
-import org.jruby.Main;
-import org.jruby.Ruby;
-import org.jruby.RubyArray;
-import org.jruby.RubyHash;
-import org.jruby.RubyIO;
-import org.jruby.RubyInstanceConfig;
-import org.jruby.RubyModule;
-import org.jruby.RubyString;
+import org.jruby.*;
 import jnr.posix.POSIX;
 import jnr.posix.util.FieldAccess;
 import jnr.posix.util.Platform;
@@ -68,6 +61,7 @@ import org.jruby.javasupport.util.RuntimeHelpers;
 import org.jruby.ext.rbconfig.RbConfigLibrary;
 import org.jruby.runtime.ThreadContext;
 import org.jruby.runtime.builtin.IRubyObject;
+import org.jruby.threading.PumpedProcessBlockingTask;
 import org.jruby.util.io.IOOptions;
 
 /**
@@ -512,13 +506,20 @@ public class ShellLauncher {
         return (int)runAndWaitPid(runtime, rawArgs, output, doExecutableSearch)[0];
     }
 
-    public static long[] runAndWaitPid(Ruby runtime, IRubyObject[] rawArgs, OutputStream output, boolean doExecutableSearch) {
-        OutputStream error = runtime.getErrorStream();
-        InputStream input = runtime.getInputStream();
+    public static long[] runAndWaitPid(final Ruby runtime, IRubyObject[] rawArgs, final OutputStream output, boolean doExecutableSearch) {
+        ThreadContext context = runtime.getCurrentContext();
+        final OutputStream error = runtime.getErrorStream();
+        final InputStream input = runtime.getInputStream();
         try {
-            Process aProcess = run(runtime, rawArgs, doExecutableSearch);
-            handleStreams(runtime, aProcess, input, output, error);
-            return new long[] {aProcess.waitFor(), getPidFromProcess(aProcess)};
+            final Process aProcess = run(runtime, rawArgs, doExecutableSearch);
+            long pid = getPidFromProcess(aProcess);
+            final long[] result = {Long.MAX_VALUE};
+
+            RubyThread.BlockingTask waiter = new PumpedProcessBlockingTask(runtime, aProcess, input, output, error);
+            context.getThread().executeBlockingTask(waiter);
+            context.pollThreadEvents();
+
+            return new long[] {result[0], pid};
         } catch (IOException e) {
             throw runtime.newIOErrorFromException(e);
         } catch (InterruptedException e) {
@@ -1431,7 +1432,7 @@ public class ShellLauncher {
         }
     }
 
-    private static void handleStreams(Ruby runtime, Process p, InputStream in, OutputStream out, OutputStream err) throws IOException {
+    public static void handleStreams(Ruby runtime, Process p, InputStream in, OutputStream out, OutputStream err) throws IOException {
         InputStream pOut = p.getInputStream();
         InputStream pErr = p.getErrorStream();
         OutputStream pIn = p.getOutputStream();

[t192zcs@vahdrtvapp05 tmp]$ ruby nap.rb
Good morning -- Wed Sep 05 12:47:41 -0500 2012
Backtick nap starting
Killing the main thread at Wed Sep 05 12:47:42 -0500 2012!
Terminated at Wed Sep 05 12:47:42 -0500 2012
[t192zcs@vahdrtvapp05 tmp]$ ps -ef | grep sleep
t192zcs   9581     1  0 12:47 pts/1    00:00:00 sleep 100
t192zcs   9651  7184  0 12:48 pts/1    00:00:00 grep sleep

[t192zcs@vahdrtvapp05 tmp]$ java -jar jruby-complete-1.6.7.2.jar nap.rb
Good morning -- Wed Sep 05 12:49:47 -0500 2012
Backtick nap starting
Killing the main thread at Wed Sep 05 12:49:48 -0500 2012!
Terminated at Wed Sep 05 12:49:48 -0500 2012
[t192zcs@vahdrtvapp05 tmp]$ ps -ef | grep sleep
t192zcs   9816  7184  0 12:49 pts/1    00:00:00 grep sleep
[t192zcs@vahdrtvapp05 tmp]$