Răsfoiți Sursa

HUE-2864 [livy] Capture stderr in separate thread and mix it in later

Since we're parsing the output to see when we should send the
next command, it's dangerous to mix stdout and stderr together
because they could stomp on each other. This captures stderr in a
separate thread and blends it into the output.
Erick Tryzelaar 10 ani în urmă
părinte
comite
13aab92d20

+ 41 - 3
apps/spark/java/livy-repl/src/main/scala/com/cloudera/hue/livy/repl/sparkr/SparkRInterpreter.scala

@@ -18,12 +18,15 @@
 
 
 package com.cloudera.hue.livy.repl.sparkr
 package com.cloudera.hue.livy.repl.sparkr
 
 
+import java.util.concurrent.locks.ReentrantLock
+
 import com.cloudera.hue.livy.repl.process.ProcessInterpreter
 import com.cloudera.hue.livy.repl.process.ProcessInterpreter
 import org.json4s.jackson.JsonMethods._
 import org.json4s.jackson.JsonMethods._
 import org.json4s.jackson.Serialization.write
 import org.json4s.jackson.Serialization.write
 import org.json4s.{JValue, _}
 import org.json4s.{JValue, _}
 
 
 import scala.annotation.tailrec
 import scala.annotation.tailrec
+import scala.io.Source
 
 
 private object SparkRInterpreter {
 private object SparkRInterpreter {
   val LIVY_END_MARKER = "# ----LIVY_END_OF_COMMAND----"
   val LIVY_END_MARKER = "# ----LIVY_END_OF_COMMAND----"
@@ -37,10 +40,11 @@ private class SparkRInterpreter(process: Process)
 
 
   implicit val formats = DefaultFormats
   implicit val formats = DefaultFormats
 
 
-  private var executionCount = 0
+  private[this] var executionCount = 0
 
 
   final override protected def waitUntilReady(): Unit = {
   final override protected def waitUntilReady(): Unit = {
-    readTo("\n> ")
+    sendExecuteRequest("")
+    executionCount = 0
   }
   }
 
 
   override protected def sendExecuteRequest(commands: String): Option[JValue] = synchronized {
   override protected def sendExecuteRequest(commands: String): Option[JValue] = synchronized {
@@ -63,7 +67,7 @@ private class SparkRInterpreter(process: Process)
           "status" -> "ok",
           "status" -> "ok",
           "execution_count" -> (executionCount - 1),
           "execution_count" -> (executionCount - 1),
           "data" -> Map(
           "data" -> Map(
-            "text/plain" -> output
+            "text/plain" -> (output + takeErrorLines())
           )
           )
         ))))
         ))))
       case (false, output) =>
       case (false, output) =>
@@ -93,4 +97,38 @@ private class SparkRInterpreter(process: Process)
       }
       }
     }
     }
   }
   }
+
+  private[this] val _lock = new ReentrantLock()
+  private[this] var stderrLines = Seq[String]()
+
+  private def takeErrorLines(): String = {
+    var lines: Seq[String] = null
+    _lock.lock()
+    try {
+      lines = stderrLines
+      stderrLines = Seq[String]()
+    } finally {
+      _lock.unlock()
+    }
+
+    lines.mkString("\n")
+  }
+
+  private[this] val stderrThread = new Thread("sparkr stderr thread") {
+    override def run() = {
+      val lines = Source.fromInputStream(process.getErrorStream).getLines()
+
+      for (line <- lines) {
+        _lock.lock()
+        try {
+          stderrLines :+= line
+        } finally {
+          _lock.unlock()
+        }
+      }
+    }
+  }
+
+  stderrThread.setDaemon(true)
+  stderrThread.start()
 }
 }

+ 1 - 1
apps/spark/java/livy-repl/src/main/scala/com/cloudera/hue/livy/repl/sparkr/SparkRSession.scala

@@ -39,7 +39,7 @@ object SparkRSession {
     env.put("SPARK_HOME", sys.env.getOrElse("SPARK_HOME", "."))
     env.put("SPARK_HOME", sys.env.getOrElse("SPARK_HOME", "."))
     env.put("SPARKR_DRIVER_R", createFakeShell().toString)
     env.put("SPARKR_DRIVER_R", createFakeShell().toString)
 
 
-    builder.redirectErrorStream(true)
+    builder.redirectError(Redirect.PIPE)
 
 
     val process = builder.start()
     val process = builder.start()