SI-6488 Interrupt i/o threads on process destroy

The previous fix uses Thread.stop to stop the threads which
drain outputs, but should do something orderly.

This commit interrupts the threads, which must check isInterrupted
before attempting further i/o.

The reading thread will suppress either the interruption or the
IOException cited in the ticket.

Similarly, i/o handlers must preserve and communicate interrupted
status.

The modest goal of this fix is to ameliorate any stack traces
when the process is destroyed.

The test runs itself as a sample process with output.

4 files changed, 75 insertions, 15 deletions

diff --git a/src/library/scala/sys/process/BasicIO.scala b/src/library/scala/sys/process/BasicIO.scala
index 58517de402..b31bbf0540 100644
--- a/src/library/scala/sys/process/BasicIO.scala
+++ b/src/library/scala/sys/process/BasicIO.scala
@@ -162,21 +162,29 @@ object BasicIO {
     */
   def processFully(processLine: String => Unit): InputStream => Unit = in => {
     val reader = new BufferedReader(new InputStreamReader(in))
-    processLinesFully(processLine)(reader.readLine)
-    reader.close()
+    try processLinesFully(processLine)(reader.readLine)
+    finally reader.close()
   }
 
   /** Calls `processLine` with the result of `readLine` until the latter returns
-    * `null`.
-    */
+   *  `null` or the current thread is interrupted.
+   */
   def processLinesFully(processLine: String => Unit)(readLine: () => String) {
-    def readFully() {
-      val line = readLine()
-      if (line != null) {
-        processLine(line)
-        readFully()
+    def working = (Thread.currentThread.isInterrupted == false)
+    def halting = { Thread.currentThread.interrupt(); null }
+    def readFully(): Unit =
+      if (working) {
+        val line =
+          try readLine()
+          catch {
+            case _: InterruptedException    => halting
+            case e: IOException if !working => halting
+          }
+        if (line != null) {
+          processLine(line)
+          readFully()
+        }
       }
-    }
     readFully()
   }
 
diff --git a/src/library/scala/sys/process/ProcessImpl.scala b/src/library/scala/sys/process/ProcessImpl.scala
index 7a5fc4ef9b..2b7fcdeb73 100644
--- a/src/library/scala/sys/process/ProcessImpl.scala
+++ b/src/library/scala/sys/process/ProcessImpl.scala
@@ -223,8 +223,8 @@ private[process] trait ProcessImpl {
       p.exitValue()
     }
     override def destroy() = {
-      try{
-        outputThreads foreach (_.stop())
+      try {
+        outputThreads foreach (_.interrupt()) // on destroy, don't bother consuming any more output
         p.destroy()
       }
       finally inputThread.interrupt()
diff --git a/test/files/run/t6488.check b/test/files/run/t6488.check
deleted file mode 100644
index 35821117c8..0000000000
--- a/test/files/run/t6488.check
+++ /dev/null
@@ -1 +0,0 @@
-Success
diff --git a/test/files/run/t6488.scala b/test/files/run/t6488.scala
index 487614ecfd..e234876fbe 100644
--- a/test/files/run/t6488.scala
+++ b/test/files/run/t6488.scala
@@ -1,11 +1,64 @@
-import sys.process._
+import scala.sys.process._
+import scala.util.Try
+import scala.util.Properties.{ javaHome, javaClassPath }
+import java.io.{ File, IOException }
+import java.util.concurrent.CountDownLatch
+import java.util.concurrent.TimeUnit._
+import java.util.concurrent.atomic._
+
 object Test {
+  /*
   // Program that prints "Success" if the command was successfully run then destroyed
   // It will silently pass if the command "/bin/ls" does not exist
-  // It will fail due to the uncatchable exception in t6488 race condition
+  // It will fail due to the uncatchable exception in t6488 race condition,
+  // i.e., if any uncaught exceptions on spawned threads are printed.
   def main(args: Array[String]) {
     try Process("/bin/ls").run(ProcessLogger { _ => () }).destroy
     catch { case _ => () }
     println("Success")
   }
+  */
+
+  // Show that no uncaught exceptions are thrown on spawned I/O threads
+  // when the process is destroyed.  The default handler will print
+  // stack traces in the failing case.
+  def main(args: Array[String]) {
+    if (args.nonEmpty && args(0) == "data")
+      data()
+    else
+      test()          // args(0) == "jvm"
+  }
+
+  // fork the data spewer, wait for input, then destroy the process
+  def test() {
+    val f = new File(javaHome, "bin").listFiles.sorted filter (_.getName startsWith "java") find (_.canExecute) getOrElse {
+      // todo signal test runner that test is skipped
+      new File("/bin/ls")  // innocuous
+    }
+    //Process(f.getAbsolutePath).run(ProcessLogger { _ => () }).destroy
+    val reading = new CountDownLatch(1)
+    val count   = new AtomicInteger
+    def counted = count.get
+    val command = s"${f.getAbsolutePath} -classpath ${javaClassPath} Test data"
+    Try {
+      Process(command) run ProcessLogger { (s: String) =>
+        //Console println s"[[$s]]"     // java help
+        count.getAndIncrement
+        reading.countDown
+        Thread.`yield`()
+      }
+    } foreach { (p: Process) =>
+      val ok = reading.await(10, SECONDS)
+      if (!ok) Console println "Timed out waiting for process output!"
+      p.destroy()
+    }
+    //Console println s"Read count $counted lines"
+  }
+
+  // spew something
+  def data() {
+    def filler = "." * 100
+    for (i <- 1 to 1000)
+      Console println s"Outputting data line $i $filler"
+  }
 }