Permalink
Browse files

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.
  • Loading branch information...
1 parent 92dbc00 commit 538aa22ec8144430e778a34bc749eccb6f5dbf4c @som-snytt som-snytt committed Apr 30, 2013
@@ -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()
}
@@ -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()
@@ -1 +0,0 @@
-Success
View
@@ -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"
+ }
}

0 comments on commit 538aa22

Please sign in to comment.